chore: Refactor make the concept of schema containing file actions a first class method

kernel/src/actions/mod.rs

@@ -132,6 +132,13 @@ pub(crate) fn get_log_domain_metadata_schema() -> &'static SchemaRef {
     &LOG_DOMAIN_METADATA_SCHEMA
 }
 
+/// Returns true if the schema contains file actions (add or remove)
+/// columns.
+#[internal_api]
+pub(crate) fn schema_contains_file_actions(schema: &SchemaRef) -> bool {
+    schema.contains(ADD_NAME) || schema.contains(REMOVE_NAME)
+}
+
 /// Nest an existing add action schema in an additional [`ADD_NAME`] struct.
 ///
 /// This is useful for JSON conversion, as it allows us to wrap a dynamically maintained add action
@@ -1974,4 +1981,46 @@ mod tests {
         assert!(record_batch.column(2).is_null(0));
         assert!(record_batch.column(3).is_null(0));
     }
+
+    #[test]
+    fn test_schema_contains_file_actions_with_add() {
+        let schema = get_log_schema()
+            .project(&[ADD_NAME, PROTOCOL_NAME])
+            .unwrap();
+        assert!(schema_contains_file_actions(&schema));
+        assert!(schema_contains_file_actions(
+            &schema.project(&[ADD_NAME]).unwrap()
+        ));
+    }
+
+    #[test]
+    fn test_schema_contains_file_actions_with_remove() {
+        let schema = get_log_schema()
+            .project(&[REMOVE_NAME, METADATA_NAME])
+            .unwrap();
+        assert!(schema_contains_file_actions(&schema));
+        assert!(schema_contains_file_actions(
+            &schema.project(&[REMOVE_NAME]).unwrap()
+        ));
+    }
+
+    #[test]
+    fn test_schema_contains_file_actions_with_both() {
+        let schema = get_log_schema().project(&[ADD_NAME, REMOVE_NAME]).unwrap();
+        assert!(schema_contains_file_actions(&schema));
+    }
+
+    #[test]
+    fn test_schema_contains_file_actions_with_neither() {
+        let schema = get_log_schema()
+            .project(&[PROTOCOL_NAME, METADATA_NAME])
+            .unwrap();
+        assert!(!schema_contains_file_actions(&schema));
+    }
+
+    #[test]
+    fn test_schema_contains_file_actions_empty_schema() {
+        let schema = Arc::new(StructType::new_unchecked([]));
+        assert!(!schema_contains_file_actions(&schema));
+    }
 }

kernel/src/log_segment.rs

@@ -5,7 +5,7 @@ use std::sync::{Arc, LazyLock};
 
 use crate::actions::visitors::SidecarVisitor;
 use crate::actions::{
-    get_log_schema, Metadata, Protocol, ADD_NAME, METADATA_NAME, PROTOCOL_NAME, REMOVE_NAME,
+    get_log_schema, schema_contains_file_actions, Metadata, Protocol, METADATA_NAME, PROTOCOL_NAME,
     SIDECAR_NAME,
 };
 use crate::last_checkpoint_hint::LastCheckpointHint;
@@ -369,8 +369,7 @@ impl LogSegment {
         checkpoint_read_schema: SchemaRef,
         meta_predicate: Option<PredicateRef>,
     ) -> DeltaResult<impl Iterator<Item = DeltaResult<ActionsBatch>> + Send> {
-        let need_file_actions = checkpoint_read_schema.contains(ADD_NAME)
-            || checkpoint_read_schema.contains(REMOVE_NAME);
+        let need_file_actions = schema_contains_file_actions(&checkpoint_read_schema);
 
         // Only validate sidecar requirement if we actually have checkpoint files
         if !self.checkpoint_parts.is_empty() {