diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 586ad12b0..6ea5ae241 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/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, + added_snapshot_id: Option, +} + async fn snapshot_manifest_locations( metadata_backend: &B, bucket: &str, @@ -3040,7 +3056,7 @@ async fn snapshot_manifest_locations( table: &crate::table_catalog::IdentifierSegment, entry: &crate::table_catalog::TableEntry, snapshot: &serde_json::Value, -) -> S3Result> +) -> S3Result> 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) -> 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)]) -> Vec { + 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)], + ) { + 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::>(); + seed_test_manifest_data_files(backend, bucket, &data_files).await; + } + async fn seed_test_manifest( backend: &TestTableCatalogObjectBackend, bucket: &str, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index f10e8012a..e21b4ae77 100644 --- a/rustfs/src/table_catalog.rs +++ b/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, - table: &'a str, + namespace: &'a Namespace, + 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, - table: &'a str, + namespace: &'a Namespace, + 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 { 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::(self.catalog_bucket(), &table_config_path) + .read_entry_unlocked::(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::(self.catalog_bucket(), &bucket_config_path) + .read_entry_unlocked::(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> { + let job_path = self.table_metadata_maintenance_report_path(table_bucket, namespace, table, table_id, job_id); + self.read_entry_unlocked::(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), - }; - self.read_entry::(self.catalog_bucket(), &job_path) - .await - .map(|entry| entry.map(|(report, _)| table_maintenance_report_with_recommended_actions(report))) + .table_maintenance_job_path(table_bucket, namespace, table, table_id, job_id), + } } 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 - .table_metadata_maintenance_scheduler_preflight(table_bucket, &namespace_name, &table_name, &scheduler_id, now) - .await? - { - 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, - }); - } + 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?; + self.table_metadata_maintenance_scheduler_preflight( + TableMaintenancePreflightContext { + 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, - namespace: &str, - table: &str, + context: TableMaintenancePreflightContext<'_>, scheduler_id: &str, now: OffsetDateTime, + effective: TableMaintenanceEffectiveConfig, ) -> TableCatalogStoreResult { - 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, - namespace, - table, + table_bucket: context.table_bucket, + namespace: context.namespace, + 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, - namespace, - table, + table_bucket: context.table_bucket, + namespace: context.namespace, + 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, - namespace: &str, - table: &str, + context: TableMaintenancePreflightContext<'_>, worker_id: &str, now: OffsetDateTime, + effective: TableMaintenanceEffectiveConfig, ) -> TableCatalogStoreResult { - 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, - namespace, - table, + table_bucket: context.table_bucket, + namespace: context.namespace, + 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, - namespace, - table, + table_bucket: context.table_bucket, + namespace: context.namespace, + 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 - .get_table_metadata_maintenance_report( + let Some((entry, _)) = self + .read_table_with_etag_unlocked(heartbeat.table_bucket, &namespace, &table) + .await? + else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", heartbeat.table_bucket, - &namespace.public_name(), - table.as_str(), + namespace.public_name(), + 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 { - 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::(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(), - table: table.as_str().to_string(), - table_id: entry.table_id, + namespace: control.namespace.public_name(), + table: control.table.as_str().to_string(), + 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 { - 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::(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(), - table: table.as_str().to_string(), - table_id: entry.table_id, + namespace: control.namespace.public_name(), + table: control.table.as_str().to_string(), + 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, diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 778b2c9c7..b1f786231 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/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): diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 5153a6a14..48432504e 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/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):