Adding predicate to protocol and metadata log replay query

kernel/Cargo.toml

@@ -123,3 +123,4 @@ tracing-subscriber = { version = "0.3", default-features = false, features = [
   "env-filter",
   "fmt",
 ] }
+env_logger = "0.11.5"

kernel/src/engine/sync/json.rs

@@ -41,9 +41,12 @@ 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: {:#?} with predicate {:#?}",
+            files, 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,7 @@ mod tests {
 
     #[test]
     fn test_scan_data() {
+        env_logger::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,16 @@ 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".into()))),
+            Expr::not(Expr::is_null(Expr::Column(
+                "protocol.min_reader_version".into(),
+            ))),
+        ));
         // 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 {