From f898d371f4f5e5df0dc73748bbb0af10dbbb5944 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 15 Aug 2026 19:00:26 -0700 Subject: [PATCH 1/7] fix(scan): reject Data Evolution files without row IDs --- crates/paimon/src/table/table_scan.rs | 40 ++++++++++++--------------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index e4cd60ef..ea58d35f 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -336,20 +336,19 @@ fn retain_manifest_entry_row_ranges( fn data_evolution_row_range_groups( data_files: Vec, row_ranges: Option<&[RowRange]>, -) -> (Vec>, usize) { +) -> crate::Result<(Vec>, usize)> { if data_files.is_empty() { - return (Vec::new(), 0); + return Ok((Vec::new(), 0)); + } + if let Some(file) = data_files.iter().find(|file| file.first_row_id.is_none()) { + return Err(crate::Error::DataInvalid { + message: format!("First row id of '{}' should not be null.", file.file_name), + source: None, + }); } let Some(row_ranges) = row_ranges else { - return (group_by_overlapping_row_id(data_files), 0); + return Ok((group_by_overlapping_row_id(data_files), 0)); }; - let all_ranges_known = data_files.iter().all(|file| file.row_id_range().is_some()); - if !all_ranges_known { - // Avoid unchecked row-range arithmetic in downstream grouping and keep - // the whole bucket as one non-raw group. The reader will then fail on - // invalid metadata instead of silently losing dedicated providers. - return (vec![data_files], 0); - } let row_id_groups = group_by_overlapping_row_id(data_files); let groups_before_pruning = row_id_groups.len(); @@ -362,7 +361,7 @@ fn data_evolution_row_range_groups( }) .collect::>(); let pruned = groups_before_pruning - retained.len(); - (retained, pruned) + Ok((retained, pruned)) } fn split_row_ranges_for_files( @@ -1930,7 +1929,7 @@ impl<'a> PaimonTableScan<'a> { // Data-evolution reads merge overlapping row-id groups column-wise. let file_groups: Vec = if data_evolution_enabled { let (row_id_groups, groups_pruned_by_row_ranges) = - data_evolution_row_range_groups(data_files, effective_row_ranges.as_deref()); + data_evolution_row_range_groups(data_files, effective_row_ranges.as_deref())?; if let Some(trace) = trace.as_deref_mut() { trace.data_evolution_groups_before_stats += row_id_groups.len(); trace.data_evolution_groups_pruned_by_row_ranges += groups_pruned_by_row_ranges; @@ -2336,7 +2335,7 @@ mod tests { } #[test] - fn test_data_evolution_row_range_group_pruning_fails_open_on_unknown_range() { + fn test_data_evolution_row_range_group_rejects_unknown_range() { let files = vec![ make_evo_file("unknown-anchor", 1, 6, 0, None), make_evo_file("left.blob", 1, 2, 0, Some(0)), @@ -2344,16 +2343,11 @@ mod tests { ]; let ranges = [RowRange::new(2, 2)]; - let (groups, pruned) = data_evolution_row_range_groups(files, Some(&ranges)); - - assert_eq!(pruned, 0); - let mut names = groups - .into_iter() - .flatten() - .map(|file| file.file_name) - .collect::>(); - names.sort(); - assert_eq!(names, vec!["left.blob", "right.blob", "unknown-anchor"]); + for row_ranges in [None, Some(ranges.as_slice())] { + let error = data_evolution_row_range_groups(files.clone(), row_ranges).unwrap_err(); + assert!(matches!(error, Error::DataInvalid { message, .. } + if message == "First row id of 'unknown-anchor' should not be null.")); + } } #[test] From 01557f699ada3681d16336c36eb70675b55b3521 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 01:05:11 -0700 Subject: [PATCH 2/7] test(scan): enable row tracking for data evolution fixtures --- crates/integrations/datafusion/tests/variant_pushdown.rs | 1 + crates/paimon/src/table/table_write.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/crates/integrations/datafusion/tests/variant_pushdown.rs b/crates/integrations/datafusion/tests/variant_pushdown.rs index f5dc32f0..2fb60653 100644 --- a/crates/integrations/datafusion/tests/variant_pushdown.rs +++ b/crates/integrations/datafusion/tests/variant_pushdown.rs @@ -67,6 +67,7 @@ async fn setup_data_evolution_shredded_variant_table_with_rows() -> (tempfile::T ) WITH ( 'file.format' = 'parquet', 'data-evolution.enabled' = 'true', + 'row-tracking.enabled' = 'true', 'variant.shreddingSchema' = '{"type":"ROW","fields":[{"name":"payload","type":{"type":"ROW","fields":[{"name":"age","type":"INT"},{"name":"city","type":"STRING"}]}}]}' ) diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index d4a7725d..ce04b74e 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -2239,6 +2239,7 @@ pub(in crate::table) mod tests { .column("a", DataType::Int(IntType::new())) .column("b", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); let table = Table::new( From 9b7d4061de1519dfa7b0c0b635b871031233c22e Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 01:30:24 -0700 Subject: [PATCH 3/7] fix(schema): require row tracking for data evolution --- .../datafusion/src/sql_context.rs | 4 +- .../integrations/datafusion/src/table/mod.rs | 1 + crates/paimon/src/spec/schema.rs | 63 +++++++++++++++++- crates/paimon/src/table/cow_writer.rs | 1 + .../paimon/src/table/data_evolution_reader.rs | 65 ++++++++++++++++++- crates/paimon/src/table/read_builder.rs | 2 +- crates/paimon/src/table/table_write.rs | 1 + 7 files changed, 131 insertions(+), 6 deletions(-) diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index 4c39c49a..ae02f0a9 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -6421,7 +6421,7 @@ mod tests { let sql_context = make_sql_context(catalog.clone()).await; sql_context - .sql("CREATE TABLE mydb.t1 (id INT, payload BLOB NOT NULL) WITH ('data-evolution.enabled' = 'true')") + .sql("CREATE TABLE mydb.t1 (id INT, payload BLOB NOT NULL) WITH ('data-evolution.enabled' = 'true', 'row-tracking.enabled' = 'true')") .await .unwrap(); @@ -6451,7 +6451,7 @@ mod tests { photo BYTES COMMENT '__BLOB_FIELD; raw photo', \ thumb BINARY COMMENT '__BLOB_DESCRIPTOR_FIELD', \ preview VARBINARY COMMENT '__BLOB_VIEW_FIELD; preview ref'\ - ) WITH ('data-evolution.enabled' = 'true')", + ) WITH ('data-evolution.enabled' = 'true', 'row-tracking.enabled' = 'true')", ) .await .unwrap(); diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index 9e3007e2..3e4ad496 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -713,6 +713,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("name", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); let table_schema = TableSchema::new(0, &schema); diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index c763f0ba..dfeab2ca 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -1157,6 +1157,7 @@ impl Schema { ) -> crate::Result<()> { validate_no_reserved_field_names(fields)?; Self::validate_key_field_types(fields, primary_keys, options)?; + Self::validate_row_tracking(options)?; Self::validate_blob_fields(fields, partition_keys, options)?; Self::validate_vector_store_fields(fields, partition_keys, options)?; PartialUpdateConfig::new(options).validate_create_mode(!primary_keys.is_empty())?; @@ -1175,6 +1176,16 @@ impl Schema { Ok(()) } + fn validate_row_tracking(options: &HashMap) -> crate::Result<()> { + let core_options = CoreOptions::new(options); + if core_options.data_evolution_enabled() && !core_options.row_tracking_enabled() { + return Err(crate::Error::ConfigInvalid { + message: "Data evolution config must enabled with row-tracking.enabled".to_string(), + }); + } + Ok(()) + } + /// Normalize primary keys: optionally take from table options (`primary-key`), remove from options. /// Corresponds to Java `normalizePrimaryKeys`. fn normalize_primary_keys( @@ -2518,6 +2529,39 @@ mod tests { assert_eq!(schema.primary_keys(), &["a", "b"]); } + #[test] + fn test_data_evolution_requires_row_tracking() { + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .build(), + "row-tracking.enabled", + ); + + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") + .build() + .unwrap(); + + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .build() + .unwrap(), + ); + assert_config_invalid( + table_schema.apply_changes(vec![crate::spec::SchemaChange::set_option( + "data-evolution.enabled".to_string(), + "true".to_string(), + )]), + "row-tracking.enabled", + ); + } + #[test] fn test_blob_schema_validation_requires_data_evolution() { let err = Schema::builder() @@ -2537,6 +2581,7 @@ mod tests { let err = Schema::builder() .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -2553,6 +2598,7 @@ mod tests { .column("payload", DataType::Blob(BlobType::new())) .partition_keys(["payload"]) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -2568,6 +2614,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); @@ -2587,6 +2634,7 @@ mod tests { ) .option("blob-field", "payload") .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); @@ -2606,6 +2654,7 @@ mod tests { Some("__BLOB_FIELD; payload bytes".to_string()), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); @@ -2630,6 +2679,7 @@ mod tests { Some("__BLOB_FIEL; payload bytes".to_string()), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -2652,6 +2702,7 @@ mod tests { Some("__BLOB_FIELD ; payload bytes".to_string()), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -2675,6 +2726,7 @@ mod tests { ) .option("blob-descriptor-field", "preview") .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -2694,6 +2746,7 @@ mod tests { .column("thumb", DataType::Blob(BlobType::new())) .option("blob-descriptor-field", "thumb") .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); @@ -2730,6 +2783,7 @@ mod tests { let schema = Schema::builder() .column("id", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); @@ -2758,6 +2812,7 @@ mod tests { .column("payload", DataType::Int(IntType::new())) .option("blob-field", "payload") .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap_err(); @@ -3339,6 +3394,7 @@ mod tests { DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -3386,6 +3442,7 @@ mod tests { DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -3473,6 +3530,10 @@ mod tests { "data-evolution.enabled".to_string(), "true".to_string(), ), + crate::spec::SchemaChange::set_option( + "row-tracking.enabled".to_string(), + "true".to_string(), + ), crate::spec::SchemaChange::add_column( "payload".to_string(), DataType::Blob(BlobType::new()), @@ -3719,7 +3780,7 @@ mod tests { .unwrap_err(); assert!( matches!(err, crate::Error::ConfigInvalid { ref message } - if message.contains("Row tracking config must enabled")), + if message.contains("row-tracking.enabled")), "dedicated VECTOR storage should require row-tracking.enabled, got {err:?}" ); } diff --git a/crates/paimon/src/table/cow_writer.rs b/crates/paimon/src/table/cow_writer.rs index 1f77c872..d0913e26 100644 --- a/crates/paimon/src/table/cow_writer.rs +++ b/crates/paimon/src/table/cow_writer.rs @@ -576,6 +576,7 @@ mod tests { let schema = Schema::builder() .column("id", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); let table = Table::new( diff --git a/crates/paimon/src/table/data_evolution_reader.rs b/crates/paimon/src/table/data_evolution_reader.rs index cf04829b..f3f10a25 100644 --- a/crates/paimon/src/table/data_evolution_reader.rs +++ b/crates/paimon/src/table/data_evolution_reader.rs @@ -3628,6 +3628,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -3714,6 +3715,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -3813,6 +3815,7 @@ mod tests { DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -3936,6 +3939,7 @@ mod tests { DataType::Array(ArrayType::new(DataType::Blob(BlobType::new()))), ) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4066,6 +4070,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4233,6 +4238,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4243,6 +4249,7 @@ mod tests { .column("payload", DataType::Blob(BlobType::new())) .column("added", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4362,6 +4369,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4466,6 +4474,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4541,6 +4550,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4631,6 +4641,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4729,6 +4740,7 @@ mod tests { .column("payload", DataType::Blob(BlobType::new())) .column("payload2", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -4959,6 +4971,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5079,6 +5092,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5217,6 +5231,7 @@ mod tests { .column("emb1", vector_float_type(2)) .column("emb2", vector_float_type(3)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5347,6 +5362,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5419,6 +5435,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5514,6 +5531,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5624,6 +5642,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5795,6 +5814,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -5925,6 +5945,10 @@ mod tests { Some(vec!["payload"]), )); } + for file in &mut files { + file.first_row_id = None; + file.file_source = Some(0); + } let file_io = FileIOBuilder::new("file").build().unwrap(); let table_schema = TableSchema::new( @@ -5934,6 +5958,7 @@ mod tests { .column("embedding", vector_float_type(2)) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -6043,7 +6068,7 @@ mod tests { let normal_path = bucket_dir.join("data.parquet"); write_int_parquet_file(&normal_path, vec![("id", vec![1, 2, 3, 4, 5, 6])], None); - let files = vec![ + let mut files = vec![ data_file_meta_with_path( "data.parquet", 0, @@ -6060,6 +6085,14 @@ mod tests { 1, Some(vec!["embedding"]), ), + data_file_meta_with_path( + "emb-middle.vector.parquet", + 2, + 2, + 1, + 1, + Some(vec!["embedding"]), + ), data_file_meta_with_path( "emb-right.vector.parquet", 4, @@ -6069,8 +6102,13 @@ mod tests { Some(vec!["embedding"]), ), data_file_meta_with_path("payload-left.blob", 0, 2, 1, 1, Some(vec!["payload"])), + data_file_meta_with_path("payload-middle.blob", 2, 2, 1, 1, Some(vec!["payload"])), data_file_meta_with_path("payload-right.blob", 4, 2, 1, 1, Some(vec!["payload"])), ]; + for file in &mut files { + file.first_row_id = None; + file.file_source = Some(0); + } let file_io = FileIOBuilder::new("file").build().unwrap(); let table_schema = TableSchema::new( @@ -6080,6 +6118,7 @@ mod tests { .column("embedding", vector_float_type(2)) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -6090,7 +6129,8 @@ mod tests { table_schema, None, ); - TableCommit::new(table.clone(), "missing-selected-dedicated-test".to_string()) + let commit = TableCommit::new(table.clone(), "missing-selected-dedicated-test".to_string()); + commit .commit(vec![CommitMessage::new( BinaryRowBuilder::new(0).build_serialized(), 0, @@ -6098,6 +6138,23 @@ mod tests { )]) .await .unwrap(); + let mut delete_middle = + CommitMessage::new(BinaryRowBuilder::new(0).build_serialized(), 0, Vec::new()); + delete_middle.deleted_files = vec![ + data_file_meta_with_path( + "emb-middle.vector.parquet", + 2, + 2, + 1, + 1, + Some(vec!["embedding"]), + ), + data_file_meta_with_path("payload-middle.blob", 2, 2, 1, 1, Some(vec!["payload"])), + ]; + for file in &mut delete_middle.deleted_files { + file.file_source = Some(0); + } + commit.commit(vec![delete_middle]).await.unwrap(); for (field, provider_kind) in [("embedding", "Vector"), ("payload", "Blob")] { let mut builder = table.new_read_builder(); @@ -6226,6 +6283,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("embedding", vector_float_type(2)) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -6507,6 +6565,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("value", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -6536,6 +6595,7 @@ mod tests { .column("payload", DataType::Blob(BlobType::new())) .column("id", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(), ); @@ -6869,6 +6929,7 @@ mod tests { .column("id", DataType::Int(IntType::new())) .column("value", DataType::Int(IntType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .option("read.batch-size", "2") .build() .unwrap(), diff --git a/crates/paimon/src/table/read_builder.rs b/crates/paimon/src/table/read_builder.rs index 9b482d33..ec8ef966 100644 --- a/crates/paimon/src/table/read_builder.rs +++ b/crates/paimon/src/table/read_builder.rs @@ -1233,7 +1233,7 @@ mod tests { "the conjunct must stay so the read can reject it" ); - for (row_tracking, data_evolution) in [(true, false), (false, true), (true, true)] { + for (row_tracking, data_evolution) in [(true, false), (true, true)] { let table = row_id_table(row_tracking, data_evolution); let mut builder = PaimonReadBuilder::new(&table); builder.with_filter(row_id_leaf(PredicateOperator::GtEq, 102)); diff --git a/crates/paimon/src/table/table_write.rs b/crates/paimon/src/table/table_write.rs index ce04b74e..1d13f3b3 100644 --- a/crates/paimon/src/table/table_write.rs +++ b/crates/paimon/src/table/table_write.rs @@ -1083,6 +1083,7 @@ pub(in crate::table) mod tests { .column("id", DataType::Int(IntType::new())) .column("payload", DataType::Blob(BlobType::new())) .option("data-evolution.enabled", "true") + .option("row-tracking.enabled", "true") .build() .unwrap(); TableSchema::new(0, &schema) From ed08edce4e2ab31d45c5bce3acd36f8e5e83bf8c Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 01:53:37 -0700 Subject: [PATCH 4/7] test(datafusion): expect invalid DE schema to fail early --- .../datafusion/tests/merge_into_tests.rs | 28 ++++--------------- 1 file changed, 5 insertions(+), 23 deletions(-) diff --git a/crates/integrations/datafusion/tests/merge_into_tests.rs b/crates/integrations/datafusion/tests/merge_into_tests.rs index 4598b9a7..036c522f 100644 --- a/crates/integrations/datafusion/tests/merge_into_tests.rs +++ b/crates/integrations/datafusion/tests/merge_into_tests.rs @@ -1152,7 +1152,7 @@ async fn test_rejects_partition_column_in_set() { } #[tokio::test] -async fn test_rejects_table_without_row_tracking() { +async fn test_rejects_enabling_data_evolution_without_row_tracking() { let (_tmp, catalog) = create_test_env(); let sql_context = create_sql_context(catalog.clone()).await; @@ -1169,29 +1169,11 @@ async fn test_rejects_table_without_row_tracking() { .await .unwrap(); - sql_context - .sql("INSERT INTO paimon.test_db.no_tracking (id, name) VALUES (1, 'alice')") + let error = sql_context + .sql("ALTER TABLE paimon.test_db.no_tracking SET TBLPROPERTIES('data-evolution.enabled' = 'true')") .await - .unwrap() - .collect() - .await - .unwrap(); - - sql_context.sql("ALTER TABLE paimon.test_db.no_tracking SET TBLPROPERTIES('data-evolution.enabled' = 'true')").await.unwrap(); - - register_source( - &sql_context, - "CREATE TEMPORARY TABLE paimon.test_db.src_nrt AS SELECT * FROM (VALUES (1, 'ALICE')) AS t(id, name)", - ) - .await; - - assert_merge_error( - &sql_context, - "MERGE INTO paimon.test_db.no_tracking t USING paimon.test_db.src_nrt s ON t.id = s.id \ - WHEN MATCHED THEN UPDATE SET name = s.name", - "row-tracking.enabled", - ) - .await; + .unwrap_err(); + assert!(error.to_string().contains("row-tracking.enabled")); } #[tokio::test] From 62349b3ea76f0fb205782d008a9c698d8194cdb1 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 01:55:53 -0700 Subject: [PATCH 5/7] test(datafusion): use valid data evolution DDL --- .../datafusion/tests/sql_context_tests.rs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs b/crates/integrations/datafusion/tests/sql_context_tests.rs index 37b08c20..e308d192 100644 --- a/crates/integrations/datafusion/tests/sql_context_tests.rs +++ b/crates/integrations/datafusion/tests/sql_context_tests.rs @@ -1036,9 +1036,11 @@ async fn test_create_table_with_blob_type() { .sql( "CREATE TABLE paimon.mydb.assets ( id INT NOT NULL, - payload BLOB, - PRIMARY KEY (id) - ) WITH ('data-evolution.enabled' = 'true')", + payload BLOB + ) WITH ( + 'data-evolution.enabled' = 'true', + 'row-tracking.enabled' = 'true' + )", ) .await .expect("CREATE TABLE with BLOB should succeed"); @@ -1049,7 +1051,7 @@ async fn test_create_table_with_blob_type() { .unwrap(); let schema = table.schema(); assert_eq!(schema.fields().len(), 2); - assert_eq!(schema.primary_keys(), &["id"]); + assert!(schema.primary_keys().is_empty()); assert_eq!( *schema.fields()[1].data_type(), DataType::Blob(BlobType::new()) @@ -2512,7 +2514,10 @@ async fn test_show_create_table_various_types() { h DATE, \ i TIMESTAMP(3), \ j BLOB) \ - WITH ('data-evolution.enabled' = 'true')", + WITH (\ + 'data-evolution.enabled' = 'true', \ + 'row-tracking.enabled' = 'true'\ + )", ) .await .expect("CREATE TABLE should succeed"); From 4d119725ceeb3fa5b65079f4e13eda717bdf9c91 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 09:07:53 -0700 Subject: [PATCH 6/7] fix(schema): enforce row tracking table constraints Mirror Java SchemaValidation.validateRowTracking: reject primary keys and bucket != -1 for row tracking tables, and reject clustering.incremental with data evolution. Downstream tests that relied on creating such tables now assert the create-time rejection. --- .../datafusion/tests/merge_into_tests.rs | 23 ++--- .../datafusion/tests/update_tests.rs | 16 ++-- crates/paimon/src/spec/core_options.rs | 9 ++ crates/paimon/src/spec/schema.rs | 84 +++++++++++++++++-- .../tests/incremental_batch_scan_test.rs | 1 - 5 files changed, 98 insertions(+), 35 deletions(-) diff --git a/crates/integrations/datafusion/tests/merge_into_tests.rs b/crates/integrations/datafusion/tests/merge_into_tests.rs index 036c522f..6a21c0d5 100644 --- a/crates/integrations/datafusion/tests/merge_into_tests.rs +++ b/crates/integrations/datafusion/tests/merge_into_tests.rs @@ -1453,7 +1453,7 @@ async fn test_rejects_table_with_primary_keys() { .sql("CREATE SCHEMA paimon.test_db") .await .unwrap(); - sql_context + let error = sql_context .sql( "CREATE TABLE paimon.test_db.pk_target (\ id INT NOT NULL, name STRING, PRIMARY KEY (id)\ @@ -1462,23 +1462,10 @@ async fn test_rejects_table_with_primary_keys() { )", ) .await - .unwrap(); - - register_source( - &sql_context, - "CREATE TEMPORARY TABLE paimon.test_db.src_pk AS SELECT * FROM (VALUES (1, 'ALICE')) AS t(id, name)", - ) - .await; - - sql_context.sql("ALTER TABLE paimon.test_db.pk_target SET TBLPROPERTIES('data-evolution.enabled' = 'true')").await.unwrap(); - - assert_merge_error( - &sql_context, - "MERGE INTO paimon.test_db.pk_target t USING paimon.test_db.src_pk s ON t.id = s.id \ - WHEN MATCHED THEN UPDATE SET name = s.name", - "does not support primary keys", - ) - .await; + .unwrap_err(); + assert!(error + .to_string() + .contains("Cannot define primary-key for row tracking table")); } #[tokio::test] diff --git a/crates/integrations/datafusion/tests/update_tests.rs b/crates/integrations/datafusion/tests/update_tests.rs index edc39260..2c451d8c 100644 --- a/crates/integrations/datafusion/tests/update_tests.rs +++ b/crates/integrations/datafusion/tests/update_tests.rs @@ -507,14 +507,14 @@ async fn test_update_rejects_primary_key_table_without_data_evolution() { } #[tokio::test] -async fn test_update_rejects_primary_key_table_with_data_evolution() { +async fn test_rejects_primary_key_data_evolution_table_at_create() { let (tmp, catalog) = create_test_env(); let sql_context = create_sql_context(catalog).await; sql_context .sql("CREATE SCHEMA paimon.test_db") .await .unwrap(); - sql_context + let error = sql_context .sql( "CREATE TABLE paimon.test_db.pk_de_t (\ id INT NOT NULL, name VARCHAR, PRIMARY KEY (id)\ @@ -525,14 +525,10 @@ async fn test_update_rejects_primary_key_table_with_data_evolution() { )", ) .await - .unwrap(); - - assert_sql_error( - &sql_context, - "UPDATE paimon.test_db.pk_de_t SET name = 'x' WHERE id = 1", - "does not support primary keys", - ) - .await; + .unwrap_err(); + assert!(error + .to_string() + .contains("Cannot define primary-key for row tracking table")); drop(tmp); } diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index 3f806fd9..1d45d49a 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -69,6 +69,7 @@ const DEFAULT_METADATA_STATS_KEEP_FIRST_N_COLUMNS: i32 = -1; const FIELDS_PREFIX: &str = "fields"; const STATS_MODE_SUFFIX: &str = "stats-mode"; const ROW_TRACKING_ENABLED_OPTION: &str = "row-tracking.enabled"; +const CLUSTERING_INCREMENTAL_OPTION: &str = "clustering.incremental"; pub(crate) const TABLE_TYPE_OPTION: &str = "type"; pub(crate) const FORMAT_TABLE_TYPE: &str = "format-table"; pub(crate) const PATH_OPTION: &str = "path"; @@ -975,6 +976,14 @@ impl<'a> CoreOptions<'a> { .unwrap_or(false) } + /// Whether incremental clustering is enabled. Default is false. + pub fn clustering_incremental_enabled(&self) -> bool { + self.options + .get(CLUSTERING_INCREMENTAL_OPTION) + .map(|v| v.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + /// Suggested target size for a manifest file. Default is 8 MiB. /// /// `manifest.target-file-size` is the Java/Python option. The shorter diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index dfeab2ca..3238ccee 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -1157,7 +1157,7 @@ impl Schema { ) -> crate::Result<()> { validate_no_reserved_field_names(fields)?; Self::validate_key_field_types(fields, primary_keys, options)?; - Self::validate_row_tracking(options)?; + Self::validate_row_tracking(primary_keys, options)?; Self::validate_blob_fields(fields, partition_keys, options)?; Self::validate_vector_store_fields(fields, partition_keys, options)?; PartialUpdateConfig::new(options).validate_create_mode(!primary_keys.is_empty())?; @@ -1176,12 +1176,38 @@ impl Schema { Ok(()) } - fn validate_row_tracking(options: &HashMap) -> crate::Result<()> { + fn validate_row_tracking( + primary_keys: &[String], + options: &HashMap, + ) -> crate::Result<()> { let core_options = CoreOptions::new(options); - if core_options.data_evolution_enabled() && !core_options.row_tracking_enabled() { - return Err(crate::Error::ConfigInvalid { - message: "Data evolution config must enabled with row-tracking.enabled".to_string(), - }); + if core_options.row_tracking_enabled() { + if !primary_keys.is_empty() { + return Err(crate::Error::ConfigInvalid { + message: "Cannot define primary-key for row tracking table.".to_string(), + }); + } + if core_options.bucket() != -1 { + return Err(crate::Error::ConfigInvalid { + message: + "Cannot define bucket for row tracking table, it only support bucket = -1" + .to_string(), + }); + } + } + if core_options.data_evolution_enabled() { + if !core_options.row_tracking_enabled() { + return Err(crate::Error::ConfigInvalid { + message: "Data evolution config must enabled with row-tracking.enabled" + .to_string(), + }); + } + if core_options.clustering_incremental_enabled() { + return Err(crate::Error::ConfigInvalid { + message: "Data evolution config must disabled with clustering.incremental" + .to_string(), + }); + } } Ok(()) } @@ -2529,6 +2555,52 @@ mod tests { assert_eq!(schema.primary_keys(), &["a", "b"]); } + #[test] + fn test_row_tracking_rejects_primary_key_and_bucket() { + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .primary_key(vec!["id".to_string()]) + .option("row-tracking.enabled", "true") + .build(), + "primary-key", + ); + + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("row-tracking.enabled", "true") + .option("bucket", "1") + .build(), + "bucket = -1", + ); + + // Combination reported in review: PK + bucket=1 + row tracking + DE. + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .primary_key(vec!["id".to_string()]) + .option("bucket", "1") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .build(), + "row tracking", + ); + } + + #[test] + fn test_data_evolution_rejects_incremental_clustering() { + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("clustering.incremental", "true") + .build(), + "clustering.incremental", + ); + } + #[test] fn test_data_evolution_requires_row_tracking() { assert_config_invalid( diff --git a/crates/paimon/tests/incremental_batch_scan_test.rs b/crates/paimon/tests/incremental_batch_scan_test.rs index 5fe3cdcb..30217e7b 100644 --- a/crates/paimon/tests/incremental_batch_scan_test.rs +++ b/crates/paimon/tests/incremental_batch_scan_test.rs @@ -697,7 +697,6 @@ async fn diff_rejects_row_ranges_instead_of_dropping_them() { ("changelog-producer", "none"), ("merge-engine", "deduplicate"), ("bucket", "1"), - ("row-tracking.enabled", "true"), ("target-file-size", "1b"), ("source.split.target-size", "1b"), ("source.split.open-file-cost", "1b"), From b2a3b882b4a03a161072176959a91ca980a7c3ec Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 16 Aug 2026 09:45:36 -0700 Subject: [PATCH 7/7] fix(table): adapt feature-gated fixtures to row tracking constraints The hybrid global fixture dropped the now-invalid bucket=1 option, and the Lumina primary-key guard moved before option checks so plain primary-key tables (including legacy invalid schemas) keep their dedicated rejection. --- crates/paimon/src/table/hybrid_search_builder.rs | 1 - .../paimon/src/table/lumina_index_build_builder.rs | 12 ++++++------ 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index a276f30d..3f5006cb 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -1725,7 +1725,6 @@ mod pk_hybrid_tests { ) }) .column(TEXT_COLUMN, DataType::VarChar(VarCharType::string_type())) - .option("bucket", "1") .option("row-tracking.enabled", "true") .build() .unwrap(), diff --git a/crates/paimon/src/table/lumina_index_build_builder.rs b/crates/paimon/src/table/lumina_index_build_builder.rs index b64a8e90..e98aedbe 100644 --- a/crates/paimon/src/table/lumina_index_build_builder.rs +++ b/crates/paimon/src/table/lumina_index_build_builder.rs @@ -265,6 +265,11 @@ pub(crate) struct LuminaIndexShard { } fn validate_table_options(table: &Table, core_options: &CoreOptions) -> Result<()> { + if !table.schema().primary_keys().is_empty() { + return Err(Error::Unsupported { + message: "Lumina index build does not support primary-key tables".to_string(), + }); + } if !core_options.row_tracking_enabled() { return Err(Error::DataInvalid { message: "Lumina index build requires 'row-tracking.enabled' = 'true'".to_string(), @@ -283,11 +288,6 @@ fn validate_table_options(table: &Table, core_options: &CoreOptions) -> Result<( source: None, }); } - if !table.schema().primary_keys().is_empty() { - return Err(Error::Unsupported { - message: "Lumina index build does not support primary-key tables".to_string(), - }); - } if core_options.deletion_vectors_enabled() { return Err(Error::Unsupported { message: @@ -1225,7 +1225,7 @@ mod tests { #[tokio::test] async fn test_execute_rejects_primary_key_table() { let table = test_table_with_schema( - vector_schema_builder(table_options("10")) + vector_schema_builder(HashMap::new()) .primary_key(["id"]) .build() .unwrap(),