fix(table-catalog): pass PyIceberg live smoke (#4637)

* fix(table-catalog): pass PyIceberg live smoke

* fix(table-catalog): satisfy live smoke clippy

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

rustfs/src/admin/handlers/table_catalog.rs

@@ -3015,7 +3015,7 @@ where
             namespace,
             table,
             entry,
-            &manifest_location,
+            &manifest_location.manifest_path,
             crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestFile,
         )?;
         let manifest_object = metadata_backend
@@ -3025,7 +3025,16 @@ where
             .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest object is missing"))?;
         let file_references =
             crate::table_catalog::data_file_references_from_manifest_avro(&manifest_object.data).map_err(catalog_store_error)?;
-        for reference in file_references {
+        for mut reference in file_references {
+            if reference.snapshot_id.is_none() {
+                reference.snapshot_id = manifest_location.added_snapshot_id;
+            }
+            if reference.sequence_number.is_none() {
+                reference.sequence_number = manifest_location.sequence_number;
+            }
+            if reference.file_sequence_number.is_none() {
+                reference.file_sequence_number = manifest_location.sequence_number;
+            }
             validate_manifest_data_file_reference(metadata_backend, bucket, namespace, table, entry, &reference).await?;
             references.push(reference);
         }
@@ -3033,6 +3042,13 @@ where
     Ok(references)
 }
 
+#[derive(Debug)]
+struct SnapshotManifestLocation {
+    manifest_path: String,
+    sequence_number: Option<i64>,
+    added_snapshot_id: Option<i64>,
+}
+
 async fn snapshot_manifest_locations<B>(
     metadata_backend: &B,
     bucket: &str,
@@ -3040,7 +3056,7 @@ async fn snapshot_manifest_locations<B>(
     table: &crate::table_catalog::IdentifierSegment,
     entry: &crate::table_catalog::TableEntry,
     snapshot: &serde_json::Value,
-) -> S3Result<Vec<String>>
+) -> S3Result<Vec<SnapshotManifestLocation>>
 where
     B: crate::table_catalog::TableCatalogObjectBackend,
 {
@@ -3063,7 +3079,14 @@ where
         if references.is_empty() {
             return Err(s3_error!(InvalidRequest, "snapshot manifest-list must reference at least one manifest"));
         }
-        return Ok(references.into_iter().map(|reference| reference.manifest_path).collect());
+        return Ok(references
+            .into_iter()
+            .map(|reference| SnapshotManifestLocation {
+                manifest_path: reference.manifest_path,
+                sequence_number: reference.sequence_number,
+                added_snapshot_id: reference.added_snapshot_id,
+            })
+            .collect());
     }
 
     let Some(manifests) = snapshot.get("manifests").and_then(serde_json::Value::as_array) else {
@@ -3078,7 +3101,11 @@ where
             manifest
                 .as_str()
                 .filter(|manifest| !manifest.is_empty())
-                .map(str::to_string)
+                .map(|manifest| SnapshotManifestLocation {
+                    manifest_path: manifest.to_string(),
+                    sequence_number: None,
+                    added_snapshot_id: None,
+                })
                 .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest location must be a string"))
         })
         .collect()
@@ -8124,6 +8151,53 @@ mod tests {
         assert_eq!(commit.metadata["last-sequence-number"], 1);
     }
 
+    #[tokio::test]
+    async fn row_level_conflict_inherits_manifest_list_sequence_numbers() {
+        let store = TestTableCatalogStore::default();
+        let metadata_backend = TestTableCatalogObjectBackend::default();
+        let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+        let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
+        let table_location = created.metadata["location"]
+            .as_str()
+            .expect("created metadata should have table location");
+        let manifest_list = format!("{table_location}/metadata/snap-10.avro");
+        let manifest = format!("{table_location}/metadata/manifest-10.avro");
+        let data_file = format!("{table_location}/data/part-10.parquet");
+        seed_test_manifest_list(&metadata_backend, "warehouse", &manifest_list, &[&manifest], 1, 10).await;
+        seed_test_manifest_with_nullable_sequences(&metadata_backend, "warehouse", &manifest, &[(&data_file, 0, 1, 10, None)])
+            .await;
+        let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({
+            "updates": [
+                {
+                    "action": "add-snapshot",
+                    "snapshot": {
+                        "snapshot-id": 10,
+                        "sequence-number": 1,
+                        "timestamp-ms": 1234,
+                        "manifest-list": manifest_list,
+                        "summary": {
+                            "operation": "append"
+                        }
+                    }
+                },
+                {
+                    "action": "set-snapshot-ref",
+                    "ref-name": "main",
+                    "snapshot-id": 10,
+                    "type": "branch"
+                }
+            ]
+        }))
+        .expect("append request should parse");
+
+        let commit = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request)
+            .await
+            .expect("manifest entry should inherit sequence numbers from manifest list");
+
+        assert_eq!(commit.metadata["current-snapshot-id"], 10);
+        assert_eq!(commit.metadata["last-sequence-number"], 1);
+    }
+
     #[tokio::test]
     async fn row_level_conflict_allows_add_only_overwrite_snapshot() {
         let store = TestTableCatalogStore::default();
@@ -9498,6 +9572,83 @@ mod tests {
         writer.into_inner().expect("manifest avro bytes should flush")
     }
 
+    fn test_nullable_long(value: Option<i64>) -> apache_avro::types::Value {
+        match value {
+            Some(value) => apache_avro::types::Value::Union(1, Box::new(apache_avro::types::Value::Long(value))),
+            None => apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+        }
+    }
+
+    fn test_manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i32, i64, Option<i64>)]) -> Vec<u8> {
+        let schema = apache_avro::Schema::parse_str(
+            r#"
+            {
+              "type": "record",
+              "name": "manifest_entry",
+              "fields": [
+                {"name": "status", "type": "int"},
+                {"name": "snapshot_id", "type": "long"},
+                {"name": "sequence_number", "type": ["null", "long"], "default": null},
+                {"name": "file_sequence_number", "type": ["null", "long"], "default": null},
+                {
+                  "name": "data_file",
+                  "type": {
+                    "type": "record",
+                    "name": "data_file",
+                    "fields": [
+                      {"name": "content", "type": "int"},
+                      {"name": "file_path", "type": "string"},
+                      {"name": "record_count", "type": "long"},
+                      {"name": "file_size_in_bytes", "type": "long"}
+                    ]
+                  }
+                }
+              ]
+            }
+            "#,
+        )
+        .expect("manifest avro schema should parse");
+        let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+        for (file_path, content, status, snapshot_id, sequence_number) in files {
+            writer
+                .append(apache_avro::types::Value::Record(vec![
+                    ("status".to_string(), apache_avro::types::Value::Int(*status)),
+                    ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                    ("sequence_number".to_string(), test_nullable_long(*sequence_number)),
+                    ("file_sequence_number".to_string(), test_nullable_long(*sequence_number)),
+                    (
+                        "data_file".to_string(),
+                        apache_avro::types::Value::Record(vec![
+                            ("content".to_string(), apache_avro::types::Value::Int(*content)),
+                            ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                            ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                            ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                        ]),
+                    ),
+                ]))
+                .expect("manifest record should append");
+        }
+        writer.into_inner().expect("manifest avro bytes should flush")
+    }
+
+    async fn seed_test_manifest_list(
+        backend: &TestTableCatalogObjectBackend,
+        bucket: &str,
+        manifest_list_location: &str,
+        manifest_locations: &[&str],
+        sequence_number: i64,
+        snapshot_id: i64,
+    ) {
+        let manifest_list_key = test_snapshot_object_key(bucket, manifest_list_location);
+        backend
+            .put_bytes(
+                bucket,
+                &manifest_list_key,
+                test_manifest_list_avro_bytes(manifest_locations, sequence_number, snapshot_id),
+            )
+            .await;
+    }
+
     async fn seed_test_snapshot_manifest(
         backend: &TestTableCatalogObjectBackend,
         bucket: &str,
@@ -9525,6 +9676,25 @@ mod tests {
         seed_test_manifest_data_files(backend, bucket, files).await;
     }
 
+    async fn seed_test_manifest_with_nullable_sequences(
+        backend: &TestTableCatalogObjectBackend,
+        bucket: &str,
+        manifest_location: &str,
+        files: &[(&str, i32, i32, i64, Option<i64>)],
+    ) {
+        let manifest_key = test_snapshot_object_key(bucket, manifest_location);
+        backend
+            .put_bytes(bucket, &manifest_key, test_manifest_avro_bytes_with_nullable_sequences(files))
+            .await;
+        let data_files = files
+            .iter()
+            .map(|(file_path, content, status, snapshot_id, sequence_number)| {
+                (*file_path, *content, *status, *snapshot_id, sequence_number.unwrap_or_default())
+            })
+            .collect::<Vec<_>>();
+        seed_test_manifest_data_files(backend, bucket, &data_files).await;
+    }
+
     async fn seed_test_manifest(
         backend: &TestTableCatalogObjectBackend,
         bucket: &str,

rustfs/src/table_catalog.rs

@@ -1090,10 +1090,18 @@ struct TableMaintenanceHeartbeatRef<'a> {
     worker_id: &'a str,
 }
 
+struct TableMaintenancePreflightContext<'a> {
+    table_bucket: &'a str,
+    namespace: &'a Namespace,
+    table: &'a IdentifierSegment,
+    entry: &'a TableEntry,
+}
+
 struct TableMaintenanceWorkerControlReport<'a> {
     table_bucket: &'a str,
-    namespace: &'a str,
+    namespace: &'a Namespace,
-    table: &'a str,
+    table: &'a IdentifierSegment,
+    entry: &'a TableEntry,
     worker_id: String,
     effective: &'a TableMaintenanceEffectiveConfig,
     status: TableMetadataMaintenanceJobStatus,
@@ -1103,8 +1111,9 @@ struct TableMaintenanceWorkerControlReport<'a> {
 
 struct TableMaintenanceSchedulerControlReport<'a> {
     table_bucket: &'a str,
-    namespace: &'a str,
+    namespace: &'a Namespace,
-    table: &'a str,
+    table: &'a IdentifierSegment,
+    entry: &'a TableEntry,
     scheduler_id: String,
     effective: &'a TableMaintenanceEffectiveConfig,
     status: TableMetadataMaintenanceJobStatus,
@@ -4069,11 +4078,22 @@ where
             )));
         };
 
+        self.get_effective_table_maintenance_config_for_entry_unlocked(table_bucket, &namespace, &table, &entry)
+            .await
+    }
+
+    async fn get_effective_table_maintenance_config_for_entry_unlocked(
+        &self,
+        table_bucket: &str,
+        namespace: &Namespace,
+        table: &IdentifierSegment,
+        entry: &TableEntry,
+    ) -> TableCatalogStoreResult<TableMaintenanceEffectiveConfig> {
         let table_config_path = self
             .paths
-            .table_maintenance_config_path(table_bucket, &namespace, &table, &entry.table_id);
+            .table_maintenance_config_path(table_bucket, namespace, table, &entry.table_id);
         if let Some((config, _)) = self
-            .read_entry::<TableMaintenanceConfig>(self.catalog_bucket(), &table_config_path)
+            .read_entry_unlocked::<TableMaintenanceConfig>(self.catalog_bucket(), &table_config_path)
             .await?
         {
             validate_table_maintenance_config(&config)?;
@@ -4085,7 +4105,7 @@ where
 
         let bucket_config_path = self.paths.table_bucket_maintenance_config_path(table_bucket);
         if let Some((config, _)) = self
-            .read_entry::<TableMaintenanceConfig>(self.catalog_bucket(), &bucket_config_path)
+            .read_entry_unlocked::<TableMaintenanceConfig>(self.catalog_bucket(), &bucket_config_path)
             .await?
         {
             validate_table_maintenance_config(&config)?;
@@ -4175,22 +4195,46 @@ where
                 table.as_str()
             )));
         };
-        let job_path = match job_id {
+
+        self.get_table_metadata_maintenance_report_for_entry_unlocked(table_bucket, &namespace, &table, &entry.table_id, job_id)
+            .await
+    }
+
+    async fn get_table_metadata_maintenance_report_for_entry_unlocked(
+        &self,
+        table_bucket: &str,
+        namespace: &Namespace,
+        table: &IdentifierSegment,
+        table_id: &str,
+        job_id: &str,
+    ) -> TableCatalogStoreResult<Option<TableMetadataMaintenanceReport>> {
+        let job_path = self.table_metadata_maintenance_report_path(table_bucket, namespace, table, table_id, job_id);
+        self.read_entry_unlocked::<TableMetadataMaintenanceReport>(self.catalog_bucket(), &job_path)
+            .await
+            .map(|entry| entry.map(|(report, _)| table_maintenance_report_with_recommended_actions(report)))
+    }
+
+    fn table_metadata_maintenance_report_path(
+        &self,
+        table_bucket: &str,
+        namespace: &Namespace,
+        table: &IdentifierSegment,
+        table_id: &str,
+        job_id: &str,
+    ) -> String {
+        match job_id {
             MAINTENANCE_JOB_ALIAS_LATEST => {
                 self.paths
-                    .table_maintenance_latest_job_path(table_bucket, &namespace, &table, &entry.table_id)
+                    .table_maintenance_latest_job_path(table_bucket, namespace, table, table_id)
             }
             MAINTENANCE_JOB_ALIAS_CURRENT => {
                 self.paths
-                    .table_maintenance_current_job_path(table_bucket, &namespace, &table, &entry.table_id)
+                    .table_maintenance_current_job_path(table_bucket, namespace, table, table_id)
             }
             _ => self
                 .paths
-                .table_maintenance_job_path(table_bucket, &namespace, &table, &entry.table_id, job_id),
+                .table_maintenance_job_path(table_bucket, namespace, table, table_id, job_id),
-        };
+        }
-        self.read_entry::<TableMetadataMaintenanceReport>(self.catalog_bucket(), &job_path)
-            .await
-            .map(|entry| entry.map(|(report, _)| table_maintenance_report_with_recommended_actions(report)))
     }
 
     pub(crate) async fn get_table_maintenance_scheduler_report(
@@ -4319,22 +4363,40 @@ where
         let namespace_name = namespace.public_name();
         let table_name = table.as_str().to_string();
 
-        let effective = {
+        let preflight = {
             let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
-            match self
+            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
-                .table_metadata_maintenance_scheduler_preflight(table_bucket, &namespace_name, &table_name, &scheduler_id, now)
+                return Err(TableCatalogStoreError::NotFound(format!(
-                .await?
+                    "table {}/{}/{}",
-            {
+                    table_bucket, namespace_name, table_name
-                TableMaintenanceSchedulerPreflight::Ready(effective) => effective,
+                )));
-                TableMaintenanceSchedulerPreflight::Complete(report) => {
+            };
-                    let scheduler = self
+            let effective = self
-                        .get_table_maintenance_scheduler_report_at(table_bucket, &namespace_name, &table_name, now)
+                .get_effective_table_maintenance_config_for_entry_unlocked(table_bucket, &namespace, &table, &entry)
-                        .await?;
+                .await?;
-                    return Ok(TableMaintenanceSchedulerRunResult {
+            self.table_metadata_maintenance_scheduler_preflight(
-                        report: *report,
+                TableMaintenancePreflightContext {
-                        scheduler,
+                    table_bucket,
-                    });
+                    namespace: &namespace,
-                }
+                    table: &table,
+                    entry: &entry,
+                },
+                &scheduler_id,
+                now,
+                effective,
+            )
+            .await?
+        };
+        let effective = match preflight {
+            TableMaintenanceSchedulerPreflight::Ready(effective) => effective,
+            TableMaintenanceSchedulerPreflight::Complete(report) => {
+                let scheduler = self
+                    .get_table_maintenance_scheduler_report_at(table_bucket, &namespace_name, &table_name, now)
+                    .await?;
+                return Ok(TableMaintenanceSchedulerRunResult {
+                    report: *report,
+                    scheduler,
+                });
             }
         };
 
@@ -4349,8 +4411,27 @@ where
 
         let report = {
             let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
+                return Err(TableCatalogStoreError::NotFound(format!(
+                    "table {}/{}/{}",
+                    table_bucket, namespace_name, table_name
+                )));
+            };
+            let effective = self
+                .get_effective_table_maintenance_config_for_entry_unlocked(table_bucket, &namespace, &table, &entry)
+                .await?;
             match self
-                .table_metadata_maintenance_scheduler_preflight(table_bucket, &namespace_name, &table_name, &scheduler_id, now)
+                .table_metadata_maintenance_scheduler_preflight(
+                    TableMaintenancePreflightContext {
+                        table_bucket,
+                        namespace: &namespace,
+                        table: &table,
+                        entry: &entry,
+                    },
+                    &scheduler_id,
+                    now,
+                    effective,
+                )
                 .await?
             {
                 TableMaintenanceSchedulerPreflight::Ready(effective) => {
@@ -4359,12 +4440,6 @@ where
                             "maintenance config changed before scheduler claim".to_string(),
                         ));
                     }
-                    let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
-                        return Err(TableCatalogStoreError::NotFound(format!(
-                            "table {}/{}/{}",
-                            table_bucket, namespace_name, table_name
-                        )));
-                    };
                     if entry.metadata_location != report.current_metadata_location {
                         return Err(TableCatalogStoreError::Conflict(
                             "current metadata location changed before maintenance scheduler claim".to_string(),
@@ -4444,8 +4519,20 @@ where
                 })?
         } else {
             let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
+                return Err(TableCatalogStoreError::NotFound(format!(
+                    "table {}/{}/{}",
+                    table_bucket, namespace_name, table_name
+                )));
+            };
             let Some(mut report) = self
-                .get_table_metadata_maintenance_report(table_bucket, &namespace_name, &table_name, MAINTENANCE_JOB_ALIAS_CURRENT)
+                .get_table_metadata_maintenance_report_for_entry_unlocked(
+                    table_bucket,
+                    &namespace,
+                    &table,
+                    &entry.table_id,
+                    MAINTENANCE_JOB_ALIAS_CURRENT,
+                )
                 .await?
             else {
                 return Err(TableCatalogStoreError::NotFound(format!(
@@ -4552,21 +4639,18 @@ where
 
     async fn table_metadata_maintenance_scheduler_preflight(
         &self,
-        table_bucket: &str,
+        context: TableMaintenancePreflightContext<'_>,
-        namespace: &str,
-        table: &str,
         scheduler_id: &str,
         now: OffsetDateTime,
+        effective: TableMaintenanceEffectiveConfig,
     ) -> TableCatalogStoreResult<TableMaintenanceSchedulerPreflight> {
-        let effective = self
-            .get_effective_table_maintenance_config(table_bucket, namespace, table)
-            .await?;
         if !effective.config.background_enabled {
             let report = self
                 .put_table_metadata_maintenance_scheduler_control_report(TableMaintenanceSchedulerControlReport {
-                    table_bucket,
+                    table_bucket: context.table_bucket,
-                    namespace,
+                    namespace: context.namespace,
-                    table,
+                    table: context.table,
+                    entry: context.entry,
                     scheduler_id: scheduler_id.to_string(),
                     effective: &effective,
                     status: TableMetadataMaintenanceJobStatus::Disabled,
@@ -4579,9 +4663,10 @@ where
         if effective.config.worker_paused {
             let report = self
                 .put_table_metadata_maintenance_scheduler_control_report(TableMaintenanceSchedulerControlReport {
-                    table_bucket,
+                    table_bucket: context.table_bucket,
-                    namespace,
+                    namespace: context.namespace,
-                    table,
+                    table: context.table,
+                    entry: context.entry,
                     scheduler_id: scheduler_id.to_string(),
                     effective: &effective,
                     status: TableMetadataMaintenanceJobStatus::Paused,
@@ -4593,7 +4678,13 @@ where
         }
 
         if let Some(current) = self
-            .get_table_metadata_maintenance_report(table_bucket, namespace, table, MAINTENANCE_JOB_ALIAS_CURRENT)
+            .get_table_metadata_maintenance_report_for_entry_unlocked(
+                context.table_bucket,
+                context.namespace,
+                context.table,
+                &context.entry.table_id,
+                MAINTENANCE_JOB_ALIAS_CURRENT,
+            )
             .await?
         {
             if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Running) {
@@ -4653,8 +4744,27 @@ where
 
         let (effective, queued) = {
             let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
+                return Err(TableCatalogStoreError::NotFound(format!(
+                    "table {}/{}/{}",
+                    table_bucket, namespace_name, table_name
+                )));
+            };
+            let effective = self
+                .get_effective_table_maintenance_config_for_entry_unlocked(table_bucket, &namespace, &table, &entry)
+                .await?;
             match self
-                .table_metadata_maintenance_worker_preflight(table_bucket, &namespace_name, &table_name, &worker_id, now)
+                .table_metadata_maintenance_worker_preflight(
+                    TableMaintenancePreflightContext {
+                        table_bucket,
+                        namespace: &namespace,
+                        table: &table,
+                        entry: &entry,
+                    },
+                    &worker_id,
+                    now,
+                    effective,
+                )
                 .await?
             {
                 TableMaintenanceWorkerPreflight::Ready { effective, queued } => (effective, queued),
@@ -4676,8 +4786,27 @@ where
 
         let (report, effective, delete) = {
             let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
+            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
+                return Err(TableCatalogStoreError::NotFound(format!(
+                    "table {}/{}/{}",
+                    table_bucket, namespace_name, table_name
+                )));
+            };
+            let effective = self
+                .get_effective_table_maintenance_config_for_entry_unlocked(table_bucket, &namespace, &table, &entry)
+                .await?;
             let (effective, queued) = match self
-                .table_metadata_maintenance_worker_preflight(table_bucket, &namespace_name, &table_name, &worker_id, now)
+                .table_metadata_maintenance_worker_preflight(
+                    TableMaintenancePreflightContext {
+                        table_bucket,
+                        namespace: &namespace,
+                        table: &table,
+                        entry: &entry,
+                    },
+                    &worker_id,
+                    now,
+                    effective,
+                )
                 .await?
             {
                 TableMaintenanceWorkerPreflight::Ready { effective, queued } => (effective, queued),
@@ -4697,12 +4826,6 @@ where
                     "maintenance config changed before worker claim".to_string(),
                 ));
             }
-            let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else {
-                return Err(TableCatalogStoreError::NotFound(format!(
-                    "table {}/{}/{}",
-                    table_bucket, namespace_name, table_name
-                )));
-            };
             if entry.metadata_location != report.current_metadata_location {
                 return Err(TableCatalogStoreError::Conflict(
                     "current metadata location changed before maintenance worker claim".to_string(),
@@ -4755,21 +4878,18 @@ where
 
     async fn table_metadata_maintenance_worker_preflight(
         &self,
-        table_bucket: &str,
+        context: TableMaintenancePreflightContext<'_>,
-        namespace: &str,
-        table: &str,
         worker_id: &str,
         now: OffsetDateTime,
+        effective: TableMaintenanceEffectiveConfig,
     ) -> TableCatalogStoreResult<TableMaintenanceWorkerPreflight> {
-        let effective = self
-            .get_effective_table_maintenance_config(table_bucket, namespace, table)
-            .await?;
         if !effective.config.background_enabled {
             let report = self
                 .put_table_metadata_maintenance_worker_control_report(TableMaintenanceWorkerControlReport {
-                    table_bucket,
+                    table_bucket: context.table_bucket,
-                    namespace,
+                    namespace: context.namespace,
-                    table,
+                    table: context.table,
+                    entry: context.entry,
                     worker_id: worker_id.to_string(),
                     effective: &effective,
                     status: TableMetadataMaintenanceJobStatus::Disabled,
@@ -4782,9 +4902,10 @@ where
         if effective.config.worker_paused {
             let report = self
                 .put_table_metadata_maintenance_worker_control_report(TableMaintenanceWorkerControlReport {
-                    table_bucket,
+                    table_bucket: context.table_bucket,
-                    namespace,
+                    namespace: context.namespace,
-                    table,
+                    table: context.table,
+                    entry: context.entry,
                     worker_id: worker_id.to_string(),
                     effective: &effective,
                     status: TableMetadataMaintenanceJobStatus::Paused,
@@ -4796,7 +4917,13 @@ where
         }
 
         if let Some(current) = self
-            .get_table_metadata_maintenance_report(table_bucket, namespace, table, MAINTENANCE_JOB_ALIAS_CURRENT)
+            .get_table_metadata_maintenance_report_for_entry_unlocked(
+                context.table_bucket,
+                context.namespace,
+                context.table,
+                &context.entry.table_id,
+                MAINTENANCE_JOB_ALIAS_CURRENT,
+            )
             .await?
         {
             if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Running) {
@@ -4864,11 +4991,23 @@ where
         let table = parse_table_for_store(heartbeat.table)?;
         let table_path = self.paths.table_entry_path(heartbeat.table_bucket, &namespace, &table);
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
-        let Some(mut report) = self
+        let Some((entry, _)) = self
-            .get_table_metadata_maintenance_report(
+            .read_table_with_etag_unlocked(heartbeat.table_bucket, &namespace, &table)
+            .await?
+        else {
+            return Err(TableCatalogStoreError::NotFound(format!(
+                "table {}/{}/{}",
                 heartbeat.table_bucket,
-                &namespace.public_name(),
+                namespace.public_name(),
-                table.as_str(),
+                table.as_str()
+            )));
+        };
+        let Some(mut report) = self
+            .get_table_metadata_maintenance_report_for_entry_unlocked(
+                heartbeat.table_bucket,
+                &namespace,
+                &table,
+                &entry.table_id,
                 MAINTENANCE_JOB_ALIAS_CURRENT,
             )
             .await?
@@ -4940,28 +5079,17 @@ where
         &self,
         control: TableMaintenanceSchedulerControlReport<'_>,
     ) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
-        let namespace = parse_namespace_for_store(control.namespace)?;
-        let table = parse_table_for_store(control.table)?;
-        let table_path = self.paths.table_entry_path(control.table_bucket, &namespace, &table);
-        let Some((entry, _)) = self.read_entry::<TableEntry>(self.catalog_bucket(), &table_path).await? else {
-            return Err(TableCatalogStoreError::NotFound(format!(
-                "table {}/{}/{}",
-                control.table_bucket,
-                namespace.public_name(),
-                table.as_str()
-            )));
-        };
         let timestamp = maintenance_timestamp(control.now);
         let cleanup_watermark_unix_seconds =
             (control.now - Duration::seconds(TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS)).unix_timestamp();
-        let current_metadata_location = entry.metadata_location.clone();
+        let current_metadata_location = control.entry.metadata_location.clone();
         let report = TableMetadataMaintenanceReport {
             job: TableMetadataMaintenanceJob {
                 job_id: Uuid::new_v4().to_string(),
                 table_bucket: control.table_bucket.to_string(),
-                namespace: namespace.public_name(),
+                namespace: control.namespace.public_name(),
-                table: table.as_str().to_string(),
+                table: control.table.as_str().to_string(),
-                table_id: entry.table_id,
+                table_id: control.entry.table_id.clone(),
                 operation: TableMetadataMaintenanceOperation::DryRun,
                 status: control.status,
                 failure_reason: Some(control.reason.to_string()),
@@ -4981,7 +5109,7 @@ where
                 started_at: None,
                 finished_at: Some(timestamp),
                 current_metadata_location: current_metadata_location.clone(),
-                current_generation: entry.generation,
+                current_generation: control.entry.generation,
                 retain_recent_metadata_files: control.effective.config.retain_recent_metadata_files,
                 safety_window_seconds: TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS,
                 cleanup_watermark_unix_seconds,
@@ -5028,28 +5156,17 @@ where
         &self,
         control: TableMaintenanceWorkerControlReport<'_>,
     ) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
-        let namespace = parse_namespace_for_store(control.namespace)?;
-        let table = parse_table_for_store(control.table)?;
-        let table_path = self.paths.table_entry_path(control.table_bucket, &namespace, &table);
-        let Some((entry, _)) = self.read_entry::<TableEntry>(self.catalog_bucket(), &table_path).await? else {
-            return Err(TableCatalogStoreError::NotFound(format!(
-                "table {}/{}/{}",
-                control.table_bucket,
-                namespace.public_name(),
-                table.as_str()
-            )));
-        };
         let timestamp = maintenance_timestamp(control.now);
         let cleanup_watermark_unix_seconds =
             (control.now - Duration::seconds(TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS)).unix_timestamp();
-        let current_metadata_location = entry.metadata_location.clone();
+        let current_metadata_location = control.entry.metadata_location.clone();
         let report = TableMetadataMaintenanceReport {
             job: TableMetadataMaintenanceJob {
                 job_id: Uuid::new_v4().to_string(),
                 table_bucket: control.table_bucket.to_string(),
-                namespace: namespace.public_name(),
+                namespace: control.namespace.public_name(),
-                table: table.as_str().to_string(),
+                table: control.table.as_str().to_string(),
-                table_id: entry.table_id,
+                table_id: control.entry.table_id.clone(),
                 operation: TableMetadataMaintenanceOperation::DryRun,
                 status: control.status,
                 failure_reason: Some(control.reason.to_string()),
@@ -5069,7 +5186,7 @@ where
                 started_at: Some(timestamp.clone()),
                 finished_at: Some(timestamp),
                 current_metadata_location: current_metadata_location.clone(),
-                current_generation: entry.generation,
+                current_generation: control.entry.generation,
                 retain_recent_metadata_files: control.effective.config.retain_recent_metadata_files,
                 safety_window_seconds: TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS,
                 cleanup_watermark_unix_seconds,

scripts/table-catalog/pyiceberg_smoke.py

@@ -1185,6 +1185,15 @@ def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
                 raise
 
 
+def maintenance_job_id(job: object) -> str | None:
+    if not isinstance(job, dict):
+        return None
+    value = job.get("job-id") or job.get("job_id")
+    if not isinstance(value, str) or not value:
+        return None
+    return value
+
+
 def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     config_path = table_endpoint_path(args, "/maintenance/config")
     scheduler_path = table_endpoint_path(args, "/maintenance/scheduler")
@@ -1202,21 +1211,22 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
         {"retain-recent-metadata-files": 1, "delete": False},
     )
     job = report.get("job")
-    if not isinstance(job, dict) or not job.get("job-id"):
+    job_id = maintenance_job_id(job)
+    if job_id is None:
         raise RuntimeError("maintenance metadata endpoint did not return a job id")
     audit_events = report.get("audit-events")
     if not isinstance(audit_events, list) or not audit_events:
         raise RuntimeError("maintenance metadata endpoint did not return audit events")
     if not any(isinstance(event, dict) and event.get("action") == "PLANNED" for event in audit_events):
         raise RuntimeError("maintenance metadata endpoint did not report a planning audit event")
-    job_report = signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}"))
+    job_report = signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job_id}"))
     if not isinstance(job_report.get("audit-events"), list):
         raise RuntimeError("maintenance job endpoint did not return audit events")
     quarantine = signed_rest_request(
         args,
         deps,
         "POST",
-        table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}/quarantine"),
+        table_endpoint_path(args, f"/maintenance/jobs/{job_id}/quarantine"),
         {"action": "INSPECT"},
     )
     if quarantine.get("action") != "INSPECT" or not isinstance(quarantine.get("report"), dict):

scripts/table-catalog/test_pyiceberg_smoke.py

@@ -336,11 +336,11 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
             if (method, path) == ("GET", config_path):
                 return {"version": 1}
             if (method, path) == ("POST", maintenance_path):
-                return {"job": {"job-id": "job-1"}, "audit-events": [{"action": "PLANNED"}]}
+                return {"job": {"job_id": "job-1"}, "audit-events": [{"action": "PLANNED"}]}
             if (method, path) == ("GET", job_path):
                 return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}, "audit-events": [{"action": "PLANNED"}]}
             if (method, path) == ("POST", quarantine_path):
-                return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}}
+                return {"action": "INSPECT", "report": {"job": {"job_id": "job-1"}}}
             if (method, path) == ("GET", scheduler_path):
                 return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1", "audit-events": [{"action": "PLANNED"}]}]}
             if (method, path) == ("POST", scheduler_run_path):
@@ -406,11 +406,11 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
             if (method, path) == ("GET", config_path):
                 return {"version": 1}
             if (method, path) == ("POST", maintenance_path):
-                return {"job": {"job-id": "job-1"}, "audit-events": [{"action": "PLANNED"}]}
+                return {"job": {"job_id": "job-1"}, "audit-events": [{"action": "PLANNED"}]}
             if (method, path) == ("GET", pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")):
                 return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}, "audit-events": [{"action": "PLANNED"}]}
             if (method, path) == ("POST", quarantine_path):
-                return {"action": "INSPECT", "report": {"job": {"job-id": "job-1"}}}
+                return {"action": "INSPECT", "report": {"job": {"job_id": "job-1"}}}
             if (method, path) == ("GET", scheduler_path):
                 return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1", "audit-events": [{"action": "PLANNED"}]}]}
             if (method, path) == ("POST", scheduler_run_path):