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 cb802ee test: add mixed-format data evolution coverage (#405)
cb802ee is described below
commit cb802ee33adf6ababd1434ca9ac78cd1692b735d
Author: Jiwen liu <[email protected]>
AuthorDate: Thu Jun 25 10:42:47 2026 +0800
test: add mixed-format data evolution coverage (#405)
---
crates/integration_tests/tests/read_tables.rs | 149 ++++++++++++++++++++++++++
dev/spark/provision.py | 64 +++++++++++
2 files changed, 213 insertions(+)
diff --git a/crates/integration_tests/tests/read_tables.rs
b/crates/integration_tests/tests/read_tables.rs
index f8fd62c..e03a123 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -1679,6 +1679,98 @@ async fn test_read_data_evolution_type_promotion() {
);
}
+#[tokio::test]
+async fn test_read_data_evolution_mixed_format_add_column() {
+ let table_name = "data_evolution_mixed_format_add_column";
+ let (plan, batches) = scan_and_read_with_fs_catalog(table_name,
None).await;
+ assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
+ assert_plan_has_multiple_schema_ids(&plan, table_name);
+
+ let mut rows: Vec<(i32, String, i32, Option<String>)> = Vec::new();
+ for batch in &batches {
+ let id = batch
+ .column_by_name("id")
+ .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+ .expect("id");
+ let name = batch
+ .column_by_name("name")
+ .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+ .expect("name");
+ let value = batch
+ .column_by_name("value")
+ .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+ .expect("value");
+ let extra = batch
+ .column_by_name("extra")
+ .and_then(|c| c.as_any().downcast_ref::<StringArray>())
+ .expect("extra");
+ for i in 0..batch.num_rows() {
+ rows.push((
+ id.value(i),
+ name.value(i).to_string(),
+ value.value(i),
+ (!extra.is_null(i)).then(|| extra.value(i).to_string()),
+ ));
+ }
+ }
+ rows.sort_by_key(|(id, _, _, _)| *id);
+
+ assert_eq!(
+ rows,
+ vec![
+ (1, "alice".into(), 100, None),
+ (2, "bob".into(), 200, None),
+ (3, "carol".into(), 300, Some("orc-extra".into())),
+ (4, "dave".into(), 400, Some("avro-extra".into())),
+ ],
+ "Mixed-format data evolution should null-fill old extra values and
preserve new values"
+ );
+}
+
+#[tokio::test]
+async fn test_read_data_evolution_mixed_format_type_promotion() {
+ let table_name = "data_evolution_mixed_format_type_promotion";
+ let (plan, batches) = scan_and_read_with_fs_catalog(table_name,
None).await;
+ assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
+ assert_plan_has_multiple_schema_ids(&plan, table_name);
+
+ for batch in &batches {
+ let value_col = batch.column_by_name("value").expect("value column");
+ assert_eq!(
+ value_col.data_type(),
+ &arrow_array::types::Int64Type::DATA_TYPE,
+ "value column should be Int64 after mixed-format data-evolution
type promotion"
+ );
+ }
+
+ let mut rows: Vec<(i32, i64)> = Vec::new();
+ for batch in &batches {
+ let id = batch
+ .column_by_name("id")
+ .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+ .expect("id");
+ let value = batch
+ .column_by_name("value")
+ .and_then(|c| c.as_any().downcast_ref::<Int64Array>())
+ .expect("value as Int64Array");
+ for i in 0..batch.num_rows() {
+ rows.push((id.value(i), value.value(i)));
+ }
+ }
+ rows.sort_by_key(|(id, _)| *id);
+
+ assert_eq!(
+ rows,
+ vec![
+ (1, 100i64),
+ (2, 200i64),
+ (3, 3_000_000_000i64),
+ (4, 4_000_000_000i64),
+ ],
+ "Mixed-format data evolution should promote INT to BIGINT"
+ );
+}
+
/// Test reading a table after ALTER TABLE DROP COLUMN.
/// Old data files have the dropped column; reader should ignore it.
#[tokio::test]
@@ -3168,6 +3260,63 @@ async fn
test_read_data_evolution_table_only_row_id_with_row_ranges() {
);
}
+#[tokio::test]
+async fn test_read_data_evolution_mixed_format_row_id_projection() {
+ let catalog = create_file_system_catalog();
+ let table = get_table_from_catalog(&catalog,
"data_evolution_mixed_format_add_column").await;
+
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_projection(&["_ROW_ID", "id"]);
+ let scan = read_builder.new_scan();
+ let plan = scan.plan().await.expect("Failed to plan scan");
+
+ assert_plan_file_formats(
+ &plan,
+ &["avro", "orc", "parquet"],
+ "data_evolution_mixed_format_add_column",
+ );
+ assert_plan_has_multiple_schema_ids(&plan,
"data_evolution_mixed_format_add_column");
+
+ let read = read_builder.new_read().expect("Failed to create read");
+ let stream = read
+ .to_arrow(plan.splits())
+ .expect("Failed to create arrow stream");
+ let batches: Vec<RecordBatch> = stream
+ .try_collect()
+ .await
+ .expect("Failed to collect batches");
+
+ let mut row_ids = Vec::new();
+ let mut ids = Vec::new();
+ for batch in &batches {
+ let row_id = batch
+ .column_by_name("_ROW_ID")
+ .and_then(|c| c.as_any().downcast_ref::<Int64Array>())
+ .expect("_ROW_ID");
+ let id = batch
+ .column_by_name("id")
+ .and_then(|c| c.as_any().downcast_ref::<Int32Array>())
+ .expect("id");
+ for i in 0..batch.num_rows() {
+ row_ids.push(row_id.value(i));
+ ids.push(id.value(i));
+ }
+ }
+
+ assert_eq!(row_ids.len(), ids.len());
+ assert_eq!(ids.len(), 4, "Expected all fixture rows to be readable");
+ assert!(
+ row_ids.iter().all(|&row_id| row_id >= 0),
+ "All _ROW_ID values should be non-negative"
+ );
+ let unique: HashSet<i64> = row_ids.iter().copied().collect();
+ assert_eq!(
+ unique.len(),
+ row_ids.len(),
+ "_ROW_ID values should be unique"
+ );
+}
+
// ---------------------------------------------------------------------------
// Full types integration tests (parquet + orc + avro)
// ---------------------------------------------------------------------------
diff --git a/dev/spark/provision.py b/dev/spark/provision.py
index 93a80fb..2ce1c72 100644
--- a/dev/spark/provision.py
+++ b/dev/spark/provision.py
@@ -541,6 +541,70 @@ def main():
"INSERT INTO format_schema_evolution_type_promotion VALUES (5,
5000000000), (6, 6000000000)"
)
+ # ===== Mixed-format Data Evolution: Add Column =====
+ # Combines row-tracking/data-evolution with ADD COLUMN and mixed file
formats.
+ # Old Parquet files lack extra; new ORC/Avro files contain extra.
+ spark.sql(
+ """
+ CREATE TABLE IF NOT EXISTS data_evolution_mixed_format_add_column (
+ id INT,
+ name STRING,
+ value INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'file.format' = 'parquet'
+ )
+ """
+ )
+ spark.sql(
+ """
+ INSERT INTO data_evolution_mixed_format_add_column VALUES
+ (1, 'alice', 100),
+ (2, 'bob', 200)
+ """
+ )
+ spark.sql("ALTER TABLE data_evolution_mixed_format_add_column ADD COLUMNS
(extra STRING)")
+ spark.sql("ALTER TABLE data_evolution_mixed_format_add_column SET
TBLPROPERTIES ('file.format' = 'orc')")
+ spark.sql(
+ "INSERT INTO data_evolution_mixed_format_add_column VALUES (3,
'carol', 300, 'orc-extra')"
+ )
+ spark.sql("ALTER TABLE data_evolution_mixed_format_add_column SET
TBLPROPERTIES ('file.format' = 'avro')")
+ spark.sql(
+ "INSERT INTO data_evolution_mixed_format_add_column VALUES (4, 'dave',
400, 'avro-extra')"
+ )
+
+ # ===== Mixed-format Data Evolution: Type Promotion =====
+ # Old Parquet files have INT; new ORC/Avro files have BIGINT.
+ spark.sql(
+ """
+ CREATE TABLE IF NOT EXISTS data_evolution_mixed_format_type_promotion (
+ id INT,
+ value INT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'row-tracking.enabled' = 'true',
+ 'data-evolution.enabled' = 'true',
+ 'file.format' = 'parquet'
+ )
+ """
+ )
+ spark.sql(
+ "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (1,
100), (2, 200)"
+ )
+ spark.sql(
+ "ALTER TABLE data_evolution_mixed_format_type_promotion ALTER COLUMN
value TYPE BIGINT"
+ )
+ spark.sql("ALTER TABLE data_evolution_mixed_format_type_promotion SET
TBLPROPERTIES ('file.format' = 'orc')")
+ spark.sql(
+ "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (3,
3000000000)"
+ )
+ spark.sql("ALTER TABLE data_evolution_mixed_format_type_promotion SET
TBLPROPERTIES ('file.format' = 'avro')")
+ spark.sql(
+ "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (4,
4000000000)"
+ )
+
# ===== Data Evolution + Schema Evolution: Add Column =====
# Combines data-evolution (row-tracking + MERGE INTO) with ALTER TABLE ADD
COLUMNS.
# Old files lack the new column; MERGE INTO produces partial-column files.
