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 ea6a591 fix(datafusion): keep dynamic options out of show create
(#468)
ea6a591 is described below
commit ea6a591174cf4872ed16f0d8b4d6958db3d9b7d9
Author: shyjsarah <[email protected]>
AuthorDate: Tue Jul 7 15:02:30 2026 +0800
fix(datafusion): keep dynamic options out of show create (#468)
---
crates/integrations/datafusion/src/catalog.rs | 21 ++--
crates/integrations/datafusion/src/sql_context.rs | 35 +++++-
crates/integrations/datafusion/src/table/mod.rs | 25 ++++-
.../datafusion/tests/sql_context_tests.rs | 122 ++++++++++++++++++++-
4 files changed, 187 insertions(+), 16 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 8808ab5..e74d3ba 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -383,21 +383,26 @@ impl SchemaProvider for PaimonSchemaProvider {
match catalog.get_table(&identifier).await {
Ok(table) => {
let opts = dynamic_options.read().unwrap().clone();
- let table = if opts.is_empty() {
- table
+ let provider = if opts.is_empty() {
+ PaimonTableProvider::try_new_with_blob_reader_registry(
+ table,
+ blob_reader_registry,
+ )?
} else {
+ let table_definition =
crate::table::build_table_definition(&table).ok();
// Dynamic options may select a historical snapshot
// (e.g. `SET 'paimon.scan.version'`); switch to its
// schema so planning sees the snapshot's columns.
- table
+ let table = table
.copy_with_time_travel(opts)
.await
- .map_err(to_datafusion_error)?
+ .map_err(to_datafusion_error)?;
+
PaimonTableProvider::try_new_with_blob_reader_registry_and_definition(
+ table,
+ blob_reader_registry,
+ table_definition,
+ )?
};
- let provider =
PaimonTableProvider::try_new_with_blob_reader_registry(
- table,
- blob_reader_registry,
- )?;
Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
}
Err(paimon::Error::TableNotExist { .. }) => Ok(None),
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 4b6adc5..ecfa252 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -51,8 +51,8 @@ use datafusion::prelude::{DataFrame, SessionContext};
use datafusion::sql::sqlparser::ast::{
AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
CreateTable, CreateTableOptions,
CreateView, Delete, Expr as SqlExpr, FromTable, Insert, Merge, ObjectName,
ObjectType,
- RenameTableNameKind, Reset, ResetStatement, Set, SqlOption, Statement,
TableFactor,
- TableObject, Truncate, Update, Value as SqlValue,
+ RenameTableNameKind, Reset, ResetStatement, Set, ShowCreateObject,
SqlOption, Statement,
+ TableFactor, TableObject, Truncate, Update, Value as SqlValue,
};
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::parser::Parser;
@@ -368,6 +368,10 @@ impl SQLContext {
.await
}
}
+ Statement::ShowCreate {
+ obj_type: ShowCreateObject::Table,
+ obj_name,
+ } => self.handle_show_create_table(sql, obj_name).await,
Statement::AlterTable(alter_table) => {
let (catalog, _catalog_name, _) =
self.resolve_catalog_and_table(&alter_table.name)?;
@@ -861,6 +865,33 @@ impl SQLContext {
ok_result(&self.ctx)
}
+ async fn handle_show_create_table(&self, sql: &str, name: &ObjectName) ->
DFResult<DataFrame> {
+ let (catalog, catalog_name, identifier) =
self.resolve_catalog_and_table(name)?;
+ let table = match catalog.get_table(&identifier).await {
+ Ok(table) => table,
+ Err(paimon::Error::TableNotExist { .. }) => return
self.ctx.sql(sql).await,
+ Err(e) => return Err(to_datafusion_error(e)),
+ };
+ let definition = crate::table::build_table_definition(&table)?;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("table_catalog", ArrowDataType::Utf8, false),
+ Field::new("table_schema", ArrowDataType::Utf8, false),
+ Field::new("table_name", ArrowDataType::Utf8, false),
+ Field::new("definition", ArrowDataType::Utf8, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec![catalog_name])),
+
Arc::new(StringArray::from(vec![identifier.database().to_string()])),
+
Arc::new(StringArray::from(vec![identifier.object().to_string()])),
+ Arc::new(StringArray::from(vec![definition])),
+ ],
+ )?;
+ self.ctx.read_batch(batch)
+ }
+
async fn handle_alter_table(
&self,
catalog: &Arc<dyn Catalog>,
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index 311adf5..ba0a119 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -69,7 +69,7 @@ pub(crate) fn datafusion_read_fields(table: &Table) ->
Vec<DataField> {
pub struct PaimonTableProvider {
table: Table,
schema: ArrowSchemaRef,
- table_definition: String,
+ table_definition: Option<String>,
}
impl PaimonTableProvider {
@@ -77,10 +77,17 @@ impl PaimonTableProvider {
///
/// Loads the table schema and converts it to Arrow for DataFusion.
pub fn try_new(table: Table) -> DFResult<Self> {
+ let table_definition = build_table_definition(&table)?;
+ Self::try_new_with_table_definition(table, Some(table_definition))
+ }
+
+ fn try_new_with_table_definition(
+ table: Table,
+ table_definition: Option<String>,
+ ) -> DFResult<Self> {
let fields = datafusion_read_fields(&table);
let schema =
paimon::arrow::build_target_arrow_schema(&fields).map_err(to_datafusion_error)?;
- let table_definition = build_table_definition(&table)?;
Ok(Self {
table,
schema,
@@ -97,6 +104,16 @@ impl PaimonTableProvider {
Self::try_new(table)
}
+ pub(crate) fn try_new_with_blob_reader_registry_and_definition(
+ table: Table,
+ blob_reader_registry: BlobReaderRegistry,
+ table_definition: Option<String>,
+ ) -> DFResult<Self> {
+ blob_reader_registry
+ .register_if_absent(table.location().to_string(),
table.file_io().clone());
+ Self::try_new_with_table_definition(table, table_definition)
+ }
+
pub fn table(&self) -> &Table {
&self.table
}
@@ -106,7 +123,7 @@ impl PaimonTableProvider {
///
/// Mirrors the syntax accepted by `SQLContext::handle_create_table`:
/// `CREATE TABLE <db>.<table> (<col> <type>, ..., PRIMARY KEY (...))
[PARTITIONED BY (...)] [WITH ('k'='v', ...)]`.
-fn build_table_definition(table: &Table) -> DFResult<String> {
+pub(crate) fn build_table_definition(table: &Table) -> DFResult<String> {
let identifier = table.identifier();
let schema = table.schema();
let mut ddl = String::new();
@@ -322,7 +339,7 @@ impl TableProvider for PaimonTableProvider {
}
fn get_table_definition(&self) -> Option<&str> {
- Some(&self.table_definition)
+ self.table_definition.as_deref()
}
async fn scan(
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index 14e6ae4..1d5ea34 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -24,8 +24,8 @@ use datafusion::datasource::MemTable;
use paimon::catalog::Identifier;
use paimon::spec::{
ArrayType, BinaryType, BlobType, CharType, DataType, FloatType, IntType,
- LocalZonedTimestampType, MapType, MultisetType, TimeType, VarBinaryType,
VarCharType,
- VectorType,
+ LocalZonedTimestampType, MapType, MultisetType, SchemaChange, TimeType,
VarBinaryType,
+ VarCharType, VectorType,
};
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::{PaimonCatalogProvider, SQLContext};
@@ -1380,6 +1380,119 @@ async fn
test_show_create_table_with_partition_and_options() {
);
}
+#[tokio::test]
+async fn test_show_create_table_excludes_session_dynamic_options() {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog).await;
+
+ sql_context
+ .sql("CREATE SCHEMA paimon.test_db")
+ .await
+ .expect("CREATE SCHEMA should succeed");
+ sql_context
+ .sql(
+ "CREATE TABLE paimon.test_db.t (id INT, name VARCHAR, pt INT) \
+ PARTITIONED BY (pt) WITH ('file.format' = 'parquet')",
+ )
+ .await
+ .expect("CREATE TABLE should succeed");
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t VALUES (1, 'one', 1)")
+ .await
+ .expect("INSERT should plan")
+ .collect()
+ .await
+ .expect("INSERT should execute");
+ sql_context
+ .sql("CALL sys.create_tag(table => 'test_db.t', tag => 'before_age')")
+ .await
+ .expect("CREATE TAG should succeed");
+ sql_context
+ .sql("ALTER TABLE paimon.test_db.t ADD COLUMN age INT")
+ .await
+ .expect("ALTER TABLE should succeed");
+ sql_context
+ .sql("SET 'paimon.scan.version' = 'before_age'")
+ .await
+ .expect("SET scan.version should succeed");
+ sql_context
+ .sql("SET 'paimon.blob-as-descriptor' = 'true'")
+ .await
+ .expect("SET blob-as-descriptor should succeed");
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t").await;
+ assert!(
+ definition.contains("'file.format' = 'parquet'"),
+ "definition should keep persisted table options, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"age\" INT"),
+ "definition should use current persisted schema, got: {definition}"
+ );
+ for dynamic_option in ["scan.version", "blob-as-descriptor"] {
+ assert!(
+ !definition.contains(dynamic_option),
+ "definition should not contain session dynamic option
{dynamic_option}, got: {definition}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn test_dynamic_scan_ignores_current_show_create_unsupported_type() {
+ let (_tmp, catalog) = create_test_env();
+ let sql_context = create_sql_context(catalog.clone()).await;
+
+ sql_context
+ .sql("CREATE SCHEMA paimon.test_db")
+ .await
+ .expect("CREATE SCHEMA should succeed");
+ sql_context
+ .sql("CREATE TABLE paimon.test_db.t (id INT)")
+ .await
+ .expect("CREATE TABLE should succeed");
+ sql_context
+ .sql("INSERT INTO paimon.test_db.t VALUES (1)")
+ .await
+ .expect("INSERT should plan")
+ .collect()
+ .await
+ .expect("INSERT should execute");
+ sql_context
+ .sql("CALL sys.create_tag(table => 'test_db.t', tag => 'before_time')")
+ .await
+ .expect("CREATE TAG should succeed");
+
+ let identifier = Identifier::new("test_db", "t");
+ catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::add_column(
+ "unsupported_col".to_string(),
+ DataType::Time(TimeType::new(3).unwrap()),
+ )],
+ false,
+ )
+ .await
+ .expect("ALTER TABLE should add unsupported SHOW CREATE type");
+
+ sql_context
+ .sql("SET 'paimon.scan.version' = 'before_time'")
+ .await
+ .expect("SET scan.version should succeed");
+
+ let rows = sql_context
+ .sql("SELECT * FROM paimon.test_db.t")
+ .await
+ .expect("dynamic scan should plan with historical schema")
+ .collect()
+ .await
+ .expect("dynamic scan should execute");
+ assert_eq!(rows[0].schema().fields().len(), 1);
+ assert_eq!(rows[0].schema().field(0).name(), "id");
+ let row_count: usize = rows.iter().map(|batch| batch.num_rows()).sum();
+ assert_eq!(row_count, 1);
+}
+
#[tokio::test]
async fn test_show_create_table_various_types() {
let (_tmp, catalog) = create_test_env();
@@ -1704,6 +1817,11 @@ async fn
test_show_create_table_rejects_non_round_trippable_types() {
.await
.expect("table should be created");
+ sql_context
+ .sql("SET 'paimon.blob-as-descriptor' = 'true'")
+ .await
+ .expect("SET blob-as-descriptor should succeed");
+
let err = sql_context
.sql(&format!("SHOW CREATE TABLE paimon.test_db.{table_name}"))
.await
