Validate that predicate is correctly passed to engine

delta-io · Sep 17, 2024 · a9b2375 · a9b2375
1 parent 59c4988
commit a9b2375
Show file tree

Hide file tree

Showing 4 changed files with 8 additions and 2 deletions.
diff --git a/acceptance/Cargo.toml b/acceptance/Cargo.toml
@@ -42,6 +42,7 @@ tracing-subscriber = { version = "0.3", default-features = false, features = [
   "env-filter",
   "fmt",
 ] }
+env_logger = "0.11.5"
 
 [[test]]
 name = "dat_reader"

diff --git a/acceptance/tests/dat_reader.rs b/acceptance/tests/dat_reader.rs
@@ -6,6 +6,7 @@ use delta_kernel::engine::default::executor::tokio::TokioBackgroundExecutor;
 use delta_kernel::engine::default::DefaultEngine;
 
 fn reader_test(path: &Path) -> datatest_stable::Result<()> {
+    let _ = env_logger::try_init();
     let root_dir = format!(
         "{}/{}",
         env!["CARGO_MANIFEST_DIR"],

diff --git a/kernel/src/engine/default/json.rs b/kernel/src/engine/default/json.rs
@@ -14,6 +14,7 @@ use futures::{StreamExt, TryStreamExt};
 use itertools::Itertools;
 use object_store::path::Path;
 use object_store::{DynObjectStore, GetResultPayload};
+use tracing::debug;
 
 use super::executor::TaskExecutor;
 use super::file_stream::{FileOpenFuture, FileOpener, FileStream};
@@ -119,8 +120,9 @@ impl<E: TaskExecutor> JsonHandler for DefaultJsonHandler<E> {
         &self,
         files: &[FileMeta],
         physical_schema: SchemaRef,
-        _predicate: Option<Expression>,
+        predicate: Option<Expression>,
     ) -> DeltaResult<FileDataReadResultIterator> {
+        debug!("Reading json files with predicate: {:?}", predicate);
         if files.is_empty() {
             return Ok(Box::new(std::iter::empty()));
         }

diff --git a/kernel/src/engine/default/parquet.rs b/kernel/src/engine/default/parquet.rs
@@ -10,6 +10,7 @@ use parquet::arrow::arrow_reader::{
     ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
 };
 use parquet::arrow::async_reader::{ParquetObjectReader, ParquetRecordBatchStreamBuilder};
+use tracing::debug;
 
 use super::file_stream::{FileOpenFuture, FileOpener, FileStream};
 use crate::engine::arrow_utils::{generate_mask, get_requested_indices, reorder_struct_array};
@@ -47,8 +48,9 @@ impl<E: TaskExecutor> ParquetHandler for DefaultParquetHandler<E> {
         &self,
         files: &[FileMeta],
         physical_schema: SchemaRef,
-        _predicate: Option<Expression>,
+        predicate: Option<Expression>,
     ) -> DeltaResult<FileDataReadResultIterator> {
+        debug!("Reading parquet files with predicate: {:?}", predicate);
         if files.is_empty() {
             return Ok(Box::new(std::iter::empty()));
         }