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 aca0281 test(datafusion): cover SQL schema evolution reads (#391)
aca0281 is described below
commit aca02811122b1102104eaaf810012ea5b2479148
Author: Jiwen liu <[email protected]>
AuthorDate: Sat Jun 20 19:11:30 2026 +0800
test(datafusion): cover SQL schema evolution reads (#391)
---
.../integrations/datafusion/tests/read_tables.rs | 145 ++++++++++++++++++++-
1 file changed, 142 insertions(+), 3 deletions(-)
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index 30760d2..5faa232 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -19,6 +19,8 @@ use std::collections::HashMap;
use std::sync::Arc;
use datafusion::arrow::array::{Array, Int32Array, StringArray};
+use datafusion::arrow::record_batch::RecordBatch;
+use datafusion::arrow::util::display::array_value_to_string;
use datafusion::catalog::CatalogProvider;
use datafusion::datasource::TableProvider;
use datafusion::logical_expr::{col, lit, TableProviderFilterPushDown};
@@ -91,13 +93,42 @@ async fn read_rows(table_name: &str) -> Vec<(i32, String)> {
actual_rows
}
-async fn collect_query(
- sql: &str,
-) ->
datafusion::error::Result<Vec<datafusion::arrow::record_batch::RecordBatch>> {
+async fn collect_query(sql: &str) ->
datafusion::error::Result<Vec<RecordBatch>> {
let ctx = create_context().await;
ctx.sql(sql).await?.collect().await
}
+fn collect_rows(batches: &[RecordBatch]) -> Vec<Vec<String>> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ for row_index in 0..batch.num_rows() {
+ let mut row = Vec::with_capacity(batch.num_columns());
+ for column in batch.columns() {
+ if column.is_null(row_index) {
+ row.push("NULL".to_string());
+ } else {
+ row.push(
+ array_value_to_string(column.as_ref(), row_index)
+ .expect("query result value should format"),
+ );
+ }
+ }
+ rows.push(row);
+ }
+ }
+ rows
+}
+
+async fn assert_sql_rows(sql: &str, expected: &[&[&str]]) {
+ let batches = collect_query(sql).await.expect("query should succeed");
+ let actual = collect_rows(&batches);
+ let expected: Vec<Vec<String>> = expected
+ .iter()
+ .map(|row| row.iter().map(|value| value.to_string()).collect())
+ .collect();
+ assert_eq!(actual, expected, "unexpected result for SQL: {sql}");
+}
+
async fn create_physical_plan(sql: &str) -> datafusion::error::Result<Arc<dyn
ExecutionPlan>> {
let ctx = create_context().await;
ctx.sql(sql).await?.create_physical_plan().await
@@ -742,6 +773,114 @@ async fn test_data_evolution_drop_column_null_fill() {
);
}
+#[tokio::test]
+async fn test_sql_read_format_schema_evolution_add_column() {
+ assert_sql_rows(
+ "SELECT id, name, age FROM
paimon.default.format_schema_evolution_add_column ORDER BY id",
+ &[
+ &["1", "alice", "NULL"],
+ &["2", "bob", "NULL"],
+ &["3", "carol", "30"],
+ &["4", "dave", "40"],
+ &["5", "eve", "50"],
+ &["6", "frank", "60"],
+ ],
+ )
+ .await;
+
+ assert_sql_rows(
+ "SELECT id, age FROM paimon.default.format_schema_evolution_add_column
WHERE age IS NULL ORDER BY id",
+ &[&["1", "NULL"], &["2", "NULL"]],
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_sql_read_format_schema_evolution_type_promotion() {
+ assert_sql_rows(
+ "SELECT id, value FROM
paimon.default.format_schema_evolution_type_promotion ORDER BY id",
+ &[
+ &["1", "100"],
+ &["2", "200"],
+ &["3", "3000000000"],
+ &["4", "4000000000"],
+ &["5", "5000000000"],
+ &["6", "6000000000"],
+ ],
+ )
+ .await;
+
+ assert_sql_rows(
+ "SELECT id, value FROM
paimon.default.format_schema_evolution_type_promotion WHERE value > 3000000000
ORDER BY id",
+ &[&["4", "4000000000"], &["5", "5000000000"], &["6", "6000000000"]],
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_sql_read_schema_evolution_rename_column() {
+ assert_sql_rows(
+ "SELECT id, renamed_payload FROM
paimon.default.schema_evolution_rename_column ORDER BY id",
+ &[
+ &["1", "parquet-old"],
+ &["2", "parquet-old-2"],
+ &["3", "orc-new"],
+ &["4", "avro-new"],
+ ],
+ )
+ .await;
+
+ assert_sql_rows(
+ "SELECT id, renamed_payload FROM
paimon.default.schema_evolution_rename_column WHERE renamed_payload LIKE '%new'
ORDER BY id",
+ &[&["3", "orc-new"], &["4", "avro-new"]],
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_sql_read_mixed_format_schema_evolution_drop_column() {
+ assert_sql_rows(
+ "SELECT id, name FROM
paimon.default.mixed_format_schema_evolution_drop_column ORDER BY id",
+ &[
+ &["1", "parquet-alice"],
+ &["2", "parquet-bob"],
+ &["3", "orc-carol"],
+ &["4", "orc-dave"],
+ &["5", "avro-eve"],
+ &["6", "avro-frank"],
+ ],
+ )
+ .await;
+
+ assert_sql_rows(
+ "SELECT id, name FROM
paimon.default.mixed_format_schema_evolution_drop_column WHERE name LIKE
'avro-%' ORDER BY id",
+ &[&["5", "avro-eve"], &["6", "avro-frank"]],
+ )
+ .await;
+}
+
+#[tokio::test]
+async fn test_sql_read_mixed_format_schema_evolution_reorder_move_column() {
+ assert_sql_rows(
+ "SELECT right_value, left_value, id FROM
paimon.default.mixed_format_schema_evolution_reorder_move_column ORDER BY id",
+ &[
+ &["parquet-right-1", "parquet-left-1", "1"],
+ &["parquet-right-2", "parquet-left-2", "2"],
+ &["orc-right-3", "orc-left-3", "3"],
+ &["orc-right-4", "orc-left-4", "4"],
+ &["avro-right-5", "avro-left-5", "5"],
+ &["avro-right-6", "avro-left-6", "6"],
+ ],
+ )
+ .await;
+
+ assert_sql_rows(
+ "SELECT id, right_value FROM
paimon.default.mixed_format_schema_evolution_reorder_move_column WHERE
right_value LIKE 'orc-%' ORDER BY id",
+ &[&["3", "orc-right-3"], &["4", "orc-right-4"]],
+ )
+ .await;
+}
+
// ======================= Complex Type Tests =======================
#[tokio::test]
