Prevent V2 manifest reuse after unpartitioned spec evolution

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(

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<Vec<u8>>,
     mut row_id_assigner: Option<&mut RowIdAssigner>,
+    table_metadata: &TableMetadata,
 ) -> Result<SelectedManifest, Error> {
+    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,
     })

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::<Result<Vec<_>, _>>()
+            .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::<Vec<_>>()
+            .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::<Vec<_>>();
+        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<InMemory>,

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<Vec<u8>>,
     mut row_id_assigner: Option<&mut RowIdAssigner>,
     overwrites: &HashSet<String>,
+    table_metadata: &TableMetadata,
 ) -> Result<OverwriteManifest, Error> {
+    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,
+    })
 }