feat: Correctly set fields on Remove action when AMT is enabled (#3318)

## What changes are proposed in this pull request?

As described in the Delta RFC:


https://github.com/delta-io/delta/blob/master/protocol_rfcs/iceberg-v4-metadata.md#remove-file

<!--
**Uncomment** this section if there are any changes affecting public
APIs. Else, **delete** this section.
### This PR affects the following public APIs
If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.
Note that _new_ public APIs are not considered breaking.
-->

## How was this change tested?

Author: Fokko

kernel/src/actions/mod.rs

@@ -1098,14 +1098,19 @@ pub(crate) struct Remove {
     pub(crate) path: String,
 
     /// The time this logical file was created, as milliseconds since the epoch.
+    ///
+    /// Must be null when adaptiveMetadata is enabled on the table since metadata cleanup
+    /// uses tree reachability instead of timestamp-based expiration.
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) deletion_timestamp: Option<i64>,
 
     /// When `false` the logical file must already be present in the table or the records
     /// in the added file must be contained in one or more remove actions in the same version.
     pub(crate) data_change: bool,
 
-    /// When true the fields `partition_values`, `size`, and `tags` are present
+    /// When true, the fields `partition_values` and `size` are present
+    ///
+    /// Must be true when adaptiveMetadata is enabled on the table.
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) extended_file_metadata: Option<bool>,
 
@@ -1127,6 +1132,8 @@ pub(crate) struct Remove {
     /// Contains [statistics] (e.g., count, min/max values for columns) about the data in this
     /// logical file encoded as a JSON string.
     ///
+    /// Must be set when adaptiveMetadata is enabled on the table.
+    ///
     /// [statistics]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Per-file-Statistics
     #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub stats: Option<String>,

kernel/src/transaction/mod.rs

@@ -21,11 +21,9 @@ use crate::committer::{
 use crate::crc::{is_incremental_safe_operation, CrcDelta, FileStatsDelta};
 use crate::engine_data::FilteredEngineData;
 use crate::error::Error;
-#[cfg(feature = "adaptive-metadata-in-dev")]
-use crate::expressions::null_lit;
 use crate::expressions::UnaryExpressionOp::ToJson;
 use crate::expressions::{
-    col, column_name, lit, ArrayData, ColumnName, ExpressionStructPatch,
+    col, column_name, lit, null_lit, ArrayData, ColumnName, ExpressionStructPatch,
     ExpressionStructPatchBuilder,
 };
 use crate::log_replay::HasSelectionVector;
@@ -1551,13 +1549,20 @@ impl<S> Transaction<S> {
             .flat_map(|schema| schema.fields().map(|field| field.name().to_owned()))
             .collect();
 
+        // adaptiveMetadata removes must carry a null deletionTimestamp and extendedFileMetadata =
+        // true (see `Remove` and `build_remove_struct_patch`).
+        let adaptive_metadata_enabled = self
+            .effective_table_config
+            .is_feature_enabled(&TableFeature::AdaptiveMetadataPreview);
+
         let make_eval = |coalesce_stats_with_parsed: bool| {
             let columns_to_drop: Vec<_> = columns_to_drop.iter().map(String::as_str).collect();
             let patch = build_remove_struct_patch(
                 self.commit_timestamp,
                 self.data_change,
                 &columns_to_drop,
                 coalesce_stats_with_parsed,
+                adaptive_metadata_enabled,
             )?;
             let expr = Arc::new(Expression::struct_from([Expression::struct_patch(patch)?]));
             evaluation_handler.new_expression_evaluator(
@@ -1603,21 +1608,32 @@ impl<S> Transaction<S> {
 /// - `partitionValues_parsed`: dropped if present. Unlike stats, no reconstruction is needed: the
 ///   Remove action's `partitionValues` is sourced from `fileConstantValues.partitionValues`, which
 ///   scans always populate from `add.partitionValues`.
+///
+/// When `adaptive_metadata_enabled` is set, the RFC requires `deletionTimestamp` to be null
+/// (cleanup uses tree reachability, not timestamp expiry), so it is emitted as a fixed literal
+/// rather than derived from the input.
 fn build_remove_struct_patch(
     commit_timestamp: i64,
     data_change: bool,
     columns_to_drop: &[&str],
     coalesce_stats_with_parsed: bool,
+    adaptive_metadata_enabled: bool,
 ) -> DeltaResult<ExpressionStructPatch> {
+    let deletion_timestamp = if adaptive_metadata_enabled {
+        null_lit(DataType::LONG)
+    } else {
+        lit(commit_timestamp)
+    };
     // Note: The Delta protocol requires `partitionValues`, `size`, and `tags` when
     // `extendedFileMetadata` is true. We require only `partitionValues` and `size` to match Spark.
+    // Under adaptiveMetadata both are guaranteed present
     let extended_file_metadata = Predicate::and_from([
         col!(SIZE_NAME).is_not_null(),
         col!(FILE_CONSTANT_VALUES_NAME, PARTITION_VALUES_NAME).is_not_null(),
     ]);
     let mut patch = ExpressionStructPatchBuilder::new()
         // deletionTimestamp
-        .insert_after("path", lit(commit_timestamp))
+        .insert_after("path", deletion_timestamp)
         // dataChange
         .insert_after("path", lit(data_change))
         // extended_file_metadata
@@ -2097,27 +2113,52 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn test_remove_action_projection_sets_extended_metadata() -> DeltaResult<()> {
+    /// Verifies the Remove projection's `deletionTimestamp` and `extendedFileMetadata` fields.
+    /// `extendedFileMetadata` is always a presence predicate; `deletionTimestamp` derives from the
+    /// input outside adaptiveMetadata and is fixed to null under it per the RFC.
+    #[rstest]
+    #[case::standard(false)]
+    #[case::adaptive_metadata(true)]
+    fn test_remove_action_projection_deletion_timestamp_and_extended_metadata(
+        #[case] adaptive_metadata_enabled: bool,
+    ) -> DeltaResult<()> {
+        let commit_timestamp = 123;
         let patch = build_remove_struct_patch(
-            0,     /* commit_timestamp */
+            commit_timestamp,
             true,  /* data_change */
             &[],   /* columns_to_drop */
             false, /* coalesce_stats_with_parsed */
+            adaptive_metadata_enabled,
         )?;
         let path_patch = patch
             .field_patches
             .get("path")
             .expect("path should have inserted fields");
+        // Insertions preserve `insert_after` call order: deletionTimestamp, dataChange,
+        // extendedFileMetadata, partitionValues.
+        let deletion_timestamp = path_patch
+            .insertions
+            .first()
+            .expect("deletionTimestamp should be the first inserted field");
         let extended_file_metadata = path_patch
             .insertions
             .get(2)
             .expect("extendedFileMetadata should follow deletionTimestamp and dataChange");
-        let expected = Expression::from_pred(Predicate::and_from([
+
+        let expected_deletion_timestamp = if adaptive_metadata_enabled {
+            null_lit(DataType::LONG)
+        } else {
+            lit(commit_timestamp)
+        };
+        let expected_extended_file_metadata = Expression::from_pred(Predicate::and_from([
             col!(SIZE_NAME).is_not_null(),
             col!(FILE_CONSTANT_VALUES_NAME, PARTITION_VALUES_NAME).is_not_null(),
         ]));
-        assert_eq!(extended_file_metadata.as_ref(), &expected);
+        assert_eq!(deletion_timestamp.as_ref(), &expected_deletion_timestamp);
+        assert_eq!(
+            extended_file_metadata.as_ref(),
+            &expected_extended_file_metadata
+        );
         Ok(())
     }
 

kernel/tests/integration/write/remove_dv.rs

@@ -579,6 +579,76 @@ async fn test_remove_files_adds_expected_entries() -> Result<(), Box<dyn std::er
     Ok(())
 }
 
+/// End-to-end check that a Remove committed to an adaptiveMetadata table conforms to the RFC:
+/// `deletionTimestamp` is null (cleanup uses tree reachability, not timestamp expiry) and
+/// `extendedFileMetadata` is true. Outside adaptiveMetadata,
+/// `test_remove_files_adds_expected_entries` covers the timestamped, conditionally-extended shape.
+#[cfg(feature = "adaptive-metadata-in-dev")]
+#[tokio::test]
+async fn remove_on_adaptive_metadata_table_nulls_deletion_timestamp_and_forces_extended_metadata(
+) -> Result<(), Box<dyn std::error::Error>> {
+    let tmp_dir = tempdir()?;
+    let tmp_dir_url = Url::from_directory_path(tmp_dir.path()).unwrap();
+    let (store, engine, table_location) =
+        test_utils::engine_store_setup("adaptive_remove", Some(&tmp_dir_url));
+    let engine = Arc::new(engine);
+    let schema = schema_ref! { nullable "number": INTEGER };
+
+    let table_url = test_utils::create_table_with_column_mapping_mode(
+        store,
+        table_location,
+        schema,
+        &[],  // no partition columns
+        true, // (3, 7) protocol
+        vec!["adaptiveMetadata-preview"],
+        vec![],
+        "id",
+    )
+    .await?;
+
+    // v1: append a data file.
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    insert_data(
+        snapshot,
+        &engine,
+        vec![Arc::new(Int32Array::from(vec![1, 2, 3]))],
+    )
+    .await?
+    .unwrap_committed();
+
+    // v2: remove the file.
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    let scan_files = snapshot
+        .clone()
+        .scan_builder()
+        .build()?
+        .scan_metadata(engine.as_ref())?
+        .next()
+        .expect("one scan-metadata batch")?
+        .scan_files;
+    let mut txn = begin_transaction(snapshot, engine.as_ref())?.with_data_change(true);
+    txn.remove_files(scan_files);
+    txn.ack_row_tracking_preservation();
+    let version = txn
+        .commit(engine.as_ref())?
+        .unwrap_committed()
+        .commit_version();
+
+    let removes = read_actions_from_commit(&table_url, version, "remove")?;
+    assert_eq!(removes.len(), 1, "expected exactly one remove action");
+    let remove = &removes[0];
+    assert!(
+        remove.get("deletionTimestamp").is_none_or(|v| v.is_null()),
+        "deletionTimestamp must be null under adaptiveMetadata, got {remove}"
+    );
+    assert_eq!(
+        remove["extendedFileMetadata"].as_bool(),
+        Some(true),
+        "extendedFileMetadata must be true under adaptiveMetadata, got {remove}"
+    );
+    Ok(())
+}
+
 /// Verifies that `extendedFileMetadata` is true exactly when `size` and `partitionValues` are
 /// present; `tags` does not affect it.
 ///

test-utils/src/lib.rs

@@ -814,13 +814,24 @@ async fn create_table_impl(
         }
     }
 
+    // adaptiveMetadata auto-enables its dependencies (see `enable_adaptive_metadata_dependencies`)
+    // so callers can pass just `adaptiveMetadata-preview` and get a loadable table.
+    let enable_adaptive_metadata = reader_features.contains(&"adaptiveMetadata-preview")
+        || writer_features.contains(&"adaptiveMetadata-preview");
+    if enable_adaptive_metadata {
+        enable_adaptive_metadata_dependencies(&mut reader_features, &mut writer_features);
+    }
+
     // Column mapping requires per-field `id`/`physicalName` metadata, without which snapshot load
-    // fails. Assign it here (with nested ids for iceberg v3); `max_column_id` feeds
-    // `delta.columnMapping.maxColumnId` below.
+    // fails. Assign it here (with nested ids for iceberg v3 / adaptiveMetadata); `max_column_id`
+    // feeds `delta.columnMapping.maxColumnId` below.
     let (schema, max_column_id) = if reader_features.contains(&"columnMapping") {
         let mut max_id = find_max_column_id_in_schema(&schema).unwrap_or(0);
-        let schema =
-            assign_column_mapping_metadata(&schema, &mut max_id, enable_iceberg_compat_v3)?;
+        let schema = assign_column_mapping_metadata(
+            &schema,
+            &mut max_id,
+            enable_iceberg_compat_v3 || enable_adaptive_metadata,
+        )?;
         (Arc::new(schema), max_id)
     } else {
         (schema, 0i64)
@@ -954,6 +965,36 @@ async fn create_table_impl(
     Ok(table_path)
 }
 
+/// Adds the features `adaptiveMetadata-preview` depends on to `reader_features` and
+/// `writer_features` (each only if not already present).
+///
+/// adaptiveMetadata requires column mapping (in `id` mode, set by the caller) plus RowTracking,
+/// DomainMetadata, DeletionVectors, and InCommitTimestamp. The ReaderWriter dependencies are
+/// mirrored into both feature lists; the writer-only dependencies are added to `writer_features`.
+fn enable_adaptive_metadata_dependencies<'a>(
+    reader_features: &mut Vec<&'a str>,
+    writer_features: &mut Vec<&'a str>,
+) {
+    // ReaderWriter features must appear in both reader and writer feature lists.
+    for f in [
+        "adaptiveMetadata-preview",
+        "columnMapping",
+        "deletionVectors",
+    ] {
+        if !reader_features.contains(&f) {
+            reader_features.push(f);
+        }
+        if !writer_features.contains(&f) {
+            writer_features.push(f);
+        }
+    }
+    for f in ["rowTracking", "domainMetadata", "inCommitTimestamp"] {
+        if !writer_features.contains(&f) {
+            writer_features.push(f);
+        }
+    }
+}
+
 /// Returns a copy of `schema` with `CURRENT_DEFAULT` metadata attached to the named top-level
 /// fields.
 ///