This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 64c64a30 fix(data-evolution): require row tracking and row IDs (#718)
64c64a30 is described below
commit 64c64a3096ff991bb41e08fcd2ed8fbef24bbbc2
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 17 09:06:03 2026 +0800
fix(data-evolution): require row tracking and row IDs (#718)
---
crates/integrations/datafusion/src/sql_context.rs | 4 +-
crates/integrations/datafusion/src/table/mod.rs | 1 +
.../datafusion/tests/merge_into_tests.rs | 51 ++------
.../datafusion/tests/sql_context_tests.rs | 15 ++-
.../integrations/datafusion/tests/update_tests.rs | 16 +--
.../datafusion/tests/variant_pushdown.rs | 1 +
crates/paimon/src/spec/core_options.rs | 9 ++
crates/paimon/src/spec/schema.rs | 135 ++++++++++++++++++++-
crates/paimon/src/table/cow_writer.rs | 1 +
crates/paimon/src/table/data_evolution_reader.rs | 65 +++++++++-
crates/paimon/src/table/hybrid_search_builder.rs | 1 -
.../paimon/src/table/lumina_index_build_builder.rs | 12 +-
crates/paimon/src/table/read_builder.rs | 2 +-
crates/paimon/src/table/table_scan.rs | 40 +++---
crates/paimon/src/table/table_write.rs | 2 +
crates/paimon/tests/incremental_batch_scan_test.rs | 1 -
16 files changed, 263 insertions(+), 93 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/integrations/datafusion/tests/merge_into_tests.rs
b/crates/integrations/datafusion/tests/merge_into_tests.rs
index 4598b9a7..6a21c0d5 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')")
- .await
- .unwrap()
- .collect()
+ let error = sql_context
+ .sql("ALTER TABLE paimon.test_db.no_tracking SET
TBLPROPERTIES('data-evolution.enabled' = 'true')")
.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]
@@ -1471,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)\
@@ -1480,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/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");
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/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/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 c763f0ba..3238ccee 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(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())?;
@@ -1175,6 +1176,42 @@ impl Schema {
Ok(())
}
+ fn validate_row_tracking(
+ primary_keys: &[String],
+ options: &HashMap<String, String>,
+ ) -> crate::Result<()> {
+ let core_options = CoreOptions::new(options);
+ 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(())
+ }
+
/// Normalize primary keys: optionally take from table options
(`primary-key`), remove from options.
/// Corresponds to Java `normalizePrimaryKeys`.
fn normalize_primary_keys(
@@ -2518,6 +2555,85 @@ 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(
+ 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 +2653,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 +2670,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 +2686,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 +2706,7 @@ mod tests {
)
.option("blob-field", "payload")
.option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
.build()
.unwrap();
@@ -2606,6 +2726,7 @@ mod tests {
Some("__BLOB_FIELD; payload bytes".to_string()),
)
.option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
.build()
.unwrap();
@@ -2630,6 +2751,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 +2774,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 +2798,7 @@ mod tests {
)
.option("blob-descriptor-field", "preview")
.option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
.build()
.unwrap_err();
@@ -2694,6 +2818,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 +2855,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 +2884,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 +3466,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 +3514,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 +3602,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 +3852,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/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(),
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_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 35b8b1a8..5c121922 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<DataFileMeta>,
row_ranges: Option<&[RowRange]>,
-) -> (Vec<Vec<DataFileMeta>>, usize) {
+) -> crate::Result<(Vec<Vec<DataFileMeta>>, 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::<Vec<_>>();
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<SplitGroup> = 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::<Vec<_>>();
- 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]
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index d4a7725d..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)
@@ -2239,6 +2240,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(
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"),
