Adding predicate to protocol and metadata log replay query (#336)

Pass a predicate hint to `Engine` in `read_metadata` so that log files
are filtered to contain either metadata or protocol columns. The motivation of this change is to filter log data at the engine level instead of needlessly passing EngineData back to the kernel

kernel/src/engine/sync/json.rs

@@ -41,9 +41,9 @@ impl JsonHandler for SyncJsonHandler {
         &self,
         files: &[FileMeta],
         schema: SchemaRef,
-        _predicate: Option<Expression>,
+        predicate: Option<Expression>,
     ) -> DeltaResult<FileDataReadResultIterator> {
-        debug!("Reading json files: {:#?}", files);
+        debug!("Reading json files: {files:#?} with predicate {predicate:#?}");
         if files.is_empty() {
             return Ok(Box::new(std::iter::empty()));
         }

kernel/src/engine/sync/parquet.rs

@@ -38,9 +38,9 @@ impl ParquetHandler for SyncParquetHandler {
         &self,
         files: &[FileMeta],
         schema: SchemaRef,
-        _predicate: Option<Expression>,
+        predicate: Option<Expression>,
     ) -> DeltaResult<FileDataReadResultIterator> {
-        debug!("Reading parquet files: {files:#?} with schema {schema:#?}");
+        debug!("Reading parquet files: {files:#?} with schema {schema:#?} and predicate {predicate:#?}");
         if files.is_empty() {
             return Ok(Box::new(std::iter::empty()));
         }

kernel/src/scan/mod.rs

@@ -627,6 +627,10 @@ mod tests {
 
     #[test]
     fn test_scan_data() {
+        use tracing_subscriber::EnvFilter;
+        tracing_subscriber::fmt()
+            .with_env_filter(EnvFilter::from_default_env())
+            .init();
         let path =
             std::fs::canonicalize(PathBuf::from("./tests/data/table-without-dv-small/")).unwrap();
         let url = url::Url::from_directory_path(path).unwrap();

kernel/src/snapshot.rs

@@ -3,6 +3,7 @@
 //!
 
 use std::cmp::Ordering;
+use std::ops::Not;
 use std::sync::Arc;
 
 use itertools::Itertools;
@@ -71,9 +72,14 @@ impl LogSegment {
 
     fn read_metadata(&self, engine: &dyn Engine) -> DeltaResult<Option<(Metadata, Protocol)>> {
         let schema = get_log_schema().project(&[PROTOCOL_NAME, METADATA_NAME])?;
+        // filter out log files that do not contain metadata or protocol information
+        use Expression as Expr;
+        let filter = Some(Expr::or(
+            Expr::not(Expr::is_null(Expr::column("metaData.id"))),
+            Expr::not(Expr::is_null(Expr::column("protocol.min_reader_version"))),
+        ));
         // read the same protocol and metadata schema for both commits and checkpoints
-        // TODO add metadata.table_id is not null and protocol.something_required is not null
-        let data_batches = self.replay(engine, schema.clone(), schema, None)?;
+        let data_batches = self.replay(engine, schema.clone(), schema, filter)?;
         let mut metadata_opt: Option<Metadata> = None;
         let mut protocol_opt: Option<Protocol> = None;
         for batch in data_batches {