fix(data-evolution): require row tracking and row IDs

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();

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);

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]

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");

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);
 }
 

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"}]}}]}'
         )

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