From 8b7828a4e6aba8b932aabe1ec46bc3d0aad62aa1 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Thu, 15 Oct 2026 03:33:04 +0300 Subject: [PATCH] Avoid V2 manifest reuse after unpartitioned spec evolution --- iceberg-rust/src/table/manifest_list.rs | 3 +- iceberg-rust/src/table/transaction/append.rs | 16 +- .../src/table/transaction/operation.rs | 146 ++++++++++++++++++ .../src/table/transaction/overwrite.rs | 27 +++- 4 files changed, 180 insertions(+), 12 deletions(-) diff --git a/iceberg-rust/src/table/manifest_list.rs b/iceberg-rust/src/table/manifest_list.rs index aad2ae66..4ea3606d 100644 --- a/iceberg-rust/src/table/manifest_list.rs +++ b/iceberg-rust/src/table/manifest_list.rs @@ -870,7 +870,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { delete_manifest, file_count_all_entries, } = if partition_column_names.is_empty() { - select_manifest_unpartitioned(manifest_list_reader, &mut writer, None)? + select_manifest_unpartitioned(manifest_list_reader, &mut writer, None, table_metadata)? } else { select_manifest_partitioned( manifest_list_reader, @@ -1014,6 +1014,7 @@ impl<'schema, 'metadata> ManifestListWriter<'schema, 'metadata> { &mut writer, row_id_assigner.as_mut(), manifests_to_overwrite, + table_metadata, )? } else { select_manifest_without_overwrites_partitioned( diff --git a/iceberg-rust/src/table/transaction/append.rs b/iceberg-rust/src/table/transaction/append.rs index 7c063ceb..64194452 100644 --- a/iceberg-rust/src/table/transaction/append.rs +++ b/iceberg-rust/src/table/transaction/append.rs @@ -236,7 +236,9 @@ pub(crate) fn select_manifest_unpartitioned( manifest_list_reader: ManifestListReader<&[u8]>, manifest_list_writer: &mut apache_avro::Writer>, mut row_id_assigner: Option<&mut RowIdAssigner>, + table_metadata: &TableMetadata, ) -> Result { + let mut compatibility = ManifestRewriteCompatibility::new(table_metadata)?; let mut selected_data_state = None; let mut selected_delete_state = None; let mut file_count_all_entries = 0; @@ -246,6 +248,15 @@ pub(crate) fn select_manifest_unpartitioned( let row_count = manifest.added_rows_count; file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; + if !compatibility.allows(&manifest, table_metadata) { + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; + continue; + } + match manifest.content { iceberg_rust_spec::manifest_list::Content::Data => { let Some((selected_row_count, _)) = &selected_data_state else { @@ -315,11 +326,8 @@ pub(crate) fn select_manifest_unpartitioned( } } } - let (_, data_manifest) = - selected_data_state.ok_or(Error::NotFound("Manifest for insert".to_owned()))?; - Ok(SelectedManifest { - data_manifest: Some(data_manifest), + data_manifest: selected_data_state.map(|(_, manifest)| manifest), delete_manifest: selected_delete_state.map(|(_, x)| x), file_count_all_entries, }) diff --git a/iceberg-rust/src/table/transaction/operation.rs b/iceberg-rust/src/table/transaction/operation.rs index 4e34a633..7f57ee99 100644 --- a/iceberg-rust/src/table/transaction/operation.rs +++ b/iceberg-rust/src/table/transaction/operation.rs @@ -1910,6 +1910,152 @@ mod tests { ); } + #[tokio::test] + async fn v2_unpartitioned_spec_does_not_reuse_old_partitioned_manifest() { + let mut metadata = sample_metadata(&[], None, &[]); + metadata.schemas.insert( + 0, + SchemaBuilder::default() + .with_schema_id(0) + .with_struct_field(StructField { + id: 1, + name: "partition_source".to_string(), + required: true, + field_type: Type::Primitive(PrimitiveType::Int), + doc: None, + initial_default: None, + write_default: None, + }) + .build() + .unwrap(), + ); + metadata.partition_specs.insert( + 0, + PartitionSpec::builder() + .with_spec_id(0) + .with_partition_field(PartitionField::new( + 1, + 1000, + "partition_source", + Transform::Identity, + )) + .build() + .unwrap(), + ); + let store = Arc::new(InMemory::new()); + let (_, updates) = Operation::Append { + branch: None, + data_files: vec![ + partitioned_data_file("old.parquet", Value::Int(42)), + partitioned_data_file("survivor.parquet", Value::Int(43)), + ], + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let old_snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let old_bytes = store + .get(&strip_prefix(old_snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let old_manifest = ManifestListReader::new(&old_bytes[..], &metadata) + .unwrap() + .next() + .unwrap() + .unwrap(); + + metadata.partition_specs.insert( + 1, + PartitionSpec::builder() + .with_spec_id(1) + .with_fields(Vec::new()) + .build() + .unwrap(), + ); + metadata.default_spec_id = 1; + let (_, updates) = Operation::Append { + branch: None, + data_files: (0..8) + .map(|index| data_file(&format!("new-{index}.parquet"), 1)) + .collect(), + delete_files: Vec::new(), + additional_summary: None, + } + .execute(&metadata, store.clone()) + .await + .unwrap(); + crate::catalog::commit::apply_table_updates(&mut metadata, updates).unwrap(); + let snapshot = metadata.current_snapshot(None).unwrap().unwrap(); + let bytes = store + .get(&strip_prefix(snapshot.manifest_list()).into()) + .await + .unwrap() + .bytes() + .await + .unwrap(); + let manifests = ManifestListReader::new(&bytes[..], &metadata) + .unwrap() + .collect::, _>>() + .unwrap(); + assert!(manifests.len() > 2, "append must use the split path"); + assert!(manifests.iter().any(|entry| { + entry.manifest_path == old_manifest.manifest_path && entry.partition_spec_id == 0 + })); + assert!(manifests.iter().any(|entry| entry.partition_spec_id == 1)); + let mut file_paths = crate::table::datafiles(store.clone(), &manifests, None, (None, None)) + .await + .unwrap() + .map_ok(|(_, entry)| entry.data_file().file_path().clone()) + .try_collect::>() + .await + .unwrap(); + file_paths.sort(); + let mut expected = (0..8) + .map(|index| format!("new-{index}.parquet")) + .chain(["old.parquet".to_string(), "survivor.parquet".to_string()]) + .collect::>(); + expected.sort(); + assert_eq!(file_paths, expected); + let replacement = data_file("replacement.parquet", 1); + let (selector, affected) = ManifestListWriter::from_existing_without_overwrites( + &bytes, + std::iter::once(&replacement), + &HashSet::from([old_manifest.manifest_path.clone()]), + manifest_list_schema_v2(), + &metadata, + ) + .unwrap(); + assert_eq!( + selector + .selected_data_manifest() + .map(|entry| entry.partition_spec_id), + Some(1) + ); + assert_eq!(affected, vec![old_manifest.clone()]); + + let error = Operation::Overwrite { + branch: None, + data_files: vec![replacement], + files_to_overwrite: HashMap::from([( + old_manifest.manifest_path, + vec!["old.parquet".to_string()], + )]), + additional_summary: None, + } + .execute(&metadata, store) + .await + .unwrap_err(); + assert!( + matches!(error, Error::NotSupported(message) if message == "COW rewrite across partition-spec evolution") + ); + } + async fn clear_current_v1_manifest_list_counts( metadata: &TableMetadata, store: &Arc, diff --git a/iceberg-rust/src/table/transaction/overwrite.rs b/iceberg-rust/src/table/transaction/overwrite.rs index f5a34498..654290f3 100644 --- a/iceberg-rust/src/table/transaction/overwrite.rs +++ b/iceberg-rust/src/table/transaction/overwrite.rs @@ -100,7 +100,9 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( manifest_list_writer: &mut apache_avro::Writer>, mut row_id_assigner: Option<&mut RowIdAssigner>, overwrites: &HashSet, + table_metadata: &TableMetadata, ) -> Result { + let mut compatibility = ManifestRewriteCompatibility::new(table_metadata)?; let mut selected_state = None; let mut file_count_all_entries = 0; let mut manifests_to_overwrite = Vec::new(); @@ -110,6 +112,19 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( let row_count = manifest.added_rows_count; file_count_all_entries += manifest.added_files_count.unwrap_or(0) as usize; + if !compatibility.allows(&manifest, table_metadata) { + if overwrites.contains(&manifest.manifest_path) { + manifests_to_overwrite.push(manifest); + } else { + append_manifest( + manifest_list_writer, + row_id_assigner.as_deref_mut(), + manifest, + )?; + } + continue; + } + let Some((selected_row_count, _)) = &selected_state else { selected_state = Some((row_count, manifest)); continue; @@ -144,11 +159,9 @@ pub(crate) fn select_manifest_without_overwrites_unpartitioned( continue; } } - selected_state - .map(|(_, entry)| OverwriteManifest { - manifest: Some(entry), - file_count_all_entries, - manifests_to_overwrite, - }) - .ok_or(Error::NotFound("Manifest for insert".to_owned())) + Ok(OverwriteManifest { + manifest: selected_state.map(|(_, entry)| entry), + file_count_all_entries, + manifests_to_overwrite, + }) }