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 ffa4d0c feat(datafusion): support SHOW CREATE TABLE via
get_table_definition (#444)
ffa4d0c is described below
commit ffa4d0c61c799a132f3cdfc229a6591ebcdf98ce
Author: shyjsarah <[email protected]>
AuthorDate: Mon Jul 6 19:44:13 2026 +0800
feat(datafusion): support SHOW CREATE TABLE via get_table_definition (#444)
DataFusion 53 rewrites `SHOW CREATE TABLE` into a query against
`information_schema.views`, whose `definition` column is populated
by `TableProvider::get_table_definition()`. PaimonTableProvider did
not override this method, so the column came back empty for Paimon
tables.
Add a cached DDL string built from the table's identifier, schema
(fields, primary keys, partition keys, options), and a recursive
`data_type_to_sql` renderer covering all DataType variants. Override
`get_table_definition()` to return it.
---
crates/integrations/datafusion/src/sql_context.rs | 215 ++++++++--
crates/integrations/datafusion/src/table/mod.rs | 159 +++++++-
.../datafusion/tests/sql_context_tests.rs | 454 ++++++++++++++++++++-
3 files changed, 791 insertions(+), 37 deletions(-)
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index c550eff..4b6adc5 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -49,20 +49,21 @@ use datafusion::error::{DataFusionError, Result as
DFResult};
use datafusion::execution::SessionStateBuilder;
use datafusion::prelude::{DataFrame, SessionContext};
use datafusion::sql::sqlparser::ast::{
- AlterTableOperation, 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,
+ 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,
};
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::parser::Parser;
use futures::StreamExt;
use paimon::catalog::{Catalog, Identifier};
use paimon::spec::{
- ArrayType as PaimonArrayType, BigIntType, BlobType, BooleanType, DataField
as PaimonDataField,
- DataType as PaimonDataType, DateType, Datum, DecimalType, DoubleType,
FloatType, IntType,
- LocalZonedTimestampType, MapType as PaimonMapType, RowType as
PaimonRowType, SchemaChange,
- SmallIntType, TimestampType, TinyIntType, VarBinaryType, VarCharType,
VariantType,
+ ArrayType as PaimonArrayType, BigIntType, BinaryType, BlobType,
BooleanType, CharType,
+ DataField as PaimonDataField, DataType as PaimonDataType, DateType, Datum,
DecimalType,
+ DoubleType, FloatType, IntType, LocalZonedTimestampType, MapType as
PaimonMapType,
+ RowType as PaimonRowType, SchemaChange, SmallIntType, TimestampType,
TinyIntType,
+ VarBinaryType, VarCharType, VariantType,
};
use crate::error::to_datafusion_error;
@@ -682,7 +683,7 @@ impl SQLContext {
let pk_cols: Vec<String> = pk
.columns
.iter()
- .map(|c| c.column.expr.to_string())
+ .map(|c| primary_key_column_name(&c.column.expr))
.collect();
builder = builder.primary_key(pk_cols);
}
@@ -1532,11 +1533,25 @@ fn parse_partition_column(token: &str) ->
DFResult<String> {
let first = trimmed.as_bytes()[0];
if first == b'"' || first == b'`' {
- let close = if first == b'"' { b'"' } else { b'`' };
- if let Some(end) = trimmed[1..].find(close as char) {
- let after_quote = trimmed[1 + end + 1..].trim();
- if after_quote.is_empty() {
- return Ok(trimmed[1..1 + end].to_string());
+ let mut value = String::new();
+ let mut end = None;
+ let mut chars = trimmed[1..].char_indices().peekable();
+ while let Some((idx, ch)) = chars.next() {
+ if ch == first as char {
+ if chars.peek().is_some_and(|(_, next)| *next == first as
char) {
+ value.push(ch);
+ chars.next();
+ } else {
+ end = Some(1 + idx + ch.len_utf8());
+ break;
+ }
+ } else {
+ value.push(ch);
+ }
+ }
+ if let Some(end) = end {
+ if trimmed[end..].trim().is_empty() {
+ return Ok(value);
}
}
return Err(DataFusionError::Plan(format!(
@@ -1555,6 +1570,38 @@ fn parse_partition_column(token: &str) ->
DFResult<String> {
}
}
+fn split_partition_columns(inner: &str) -> DFResult<Vec<&str>> {
+ let mut columns = Vec::new();
+ let mut start = 0;
+ let mut quote = None;
+ let mut chars = inner.char_indices().peekable();
+ while let Some((idx, ch)) = chars.next() {
+ match quote {
+ Some(q) if ch == q => {
+ if chars.peek().is_some_and(|(_, next)| *next == q) {
+ chars.next();
+ } else {
+ quote = None;
+ }
+ }
+ Some(_) => {}
+ None if ch == '"' || ch == '`' => quote = Some(ch),
+ None if ch == ',' => {
+ columns.push(&inner[start..idx]);
+ start = idx + ch.len_utf8();
+ }
+ None => {}
+ }
+ }
+ if quote.is_some() {
+ return Err(DataFusionError::Plan(
+ "Unterminated quoted identifier in PARTITIONED BY".to_string(),
+ ));
+ }
+ columns.push(&inner[start..]);
+ Ok(columns)
+}
+
/// Extract `PARTITIONED BY (col1, col2, ...)` from SQL before parsing.
///
/// Paimon only allows column references (no types) in PARTITIONED BY.
@@ -1578,17 +1625,28 @@ fn extract_partition_by(sql: &str) -> DFResult<(String,
Vec<String>)> {
let inner_start = paren_start + 1;
let mut depth = 1;
let mut paren_end = None;
- for (i, ch) in sql[inner_start..].char_indices() {
- match ch {
- '(' => depth += 1,
- ')' => {
+ let mut quote = None;
+ let mut chars = sql[inner_start..].char_indices().peekable();
+ while let Some((i, ch)) = chars.next() {
+ match quote {
+ Some(q) if ch == q => {
+ if chars.peek().is_some_and(|(_, next)| *next == q) {
+ chars.next();
+ } else {
+ quote = None;
+ }
+ }
+ Some(_) => {}
+ None if ch == '"' || ch == '`' => quote = Some(ch),
+ None if ch == '(' => depth += 1,
+ None if ch == ')' => {
depth -= 1;
if depth == 0 {
paren_end = Some(inner_start + i);
break;
}
}
- _ => {}
+ None => {}
}
}
let paren_end = paren_end.ok_or_else(|| {
@@ -1603,7 +1661,7 @@ fn extract_partition_by(sql: &str) -> DFResult<(String,
Vec<String>)> {
}
let mut partition_keys = Vec::new();
- for token in inner.split(',') {
+ for token in split_partition_columns(inner)? {
partition_keys.push(parse_partition_column(token)?);
}
@@ -1627,6 +1685,45 @@ fn column_def_to_paimon_type(col: &ColumnDef) ->
DFResult<PaimonDataType> {
sql_data_type_to_paimon_type(&col.data_type, column_def_nullable(col))
}
+fn primary_key_column_name(expr: &SqlExpr) -> String {
+ match expr {
+ SqlExpr::Identifier(ident) => ident.value.clone(),
+ _ => expr.to_string(),
+ }
+}
+
+fn character_length_or_default(
+ length: &Option<CharacterLength>,
+ default_length: u32,
+) -> DFResult<u32> {
+ match length {
+ Some(CharacterLength::IntegerLength { length, .. }) =>
(*length).try_into().map_err(|_| {
+ DataFusionError::Plan(format!("Character length {length} exceeds
supported range"))
+ }),
+ Some(CharacterLength::Max) => Ok(VarCharType::MAX_LENGTH),
+ None => Ok(default_length),
+ }
+}
+
+fn u64_length_or_default(length: Option<u64>, default_length: usize) ->
DFResult<usize> {
+ match length {
+ Some(length) => length.try_into().map_err(|_| {
+ DataFusionError::Plan(format!("Binary length {length} exceeds
supported range"))
+ }),
+ None => Ok(default_length),
+ }
+}
+
+fn binary_length_or_default(length: &Option<BinaryLength>, default_length:
u32) -> DFResult<u32> {
+ match length {
+ Some(BinaryLength::IntegerLength { length }) =>
(*length).try_into().map_err(|_| {
+ DataFusionError::Plan(format!("Binary length {length} exceeds
supported range"))
+ }),
+ Some(BinaryLength::Max) => Ok(VarBinaryType::MAX_LENGTH),
+ None => Ok(default_length),
+ }
+}
+
fn column_def_nullable(col: &ColumnDef) -> bool {
!col.options.iter().any(|opt| {
matches!(
@@ -1668,21 +1765,39 @@ fn sql_data_type_to_paimon_type(
SqlType::Double(_) | SqlType::DoublePrecision => {
Ok(PaimonDataType::Double(DoubleType::with_nullable(nullable)))
}
- SqlType::Varchar(_)
- | SqlType::CharVarying(_)
- | SqlType::Text
- | SqlType::String(_)
- | SqlType::Char(_)
- | SqlType::Character(_) => Ok(PaimonDataType::VarChar(
+ SqlType::Char(length) | SqlType::Character(length) =>
Ok(PaimonDataType::Char(
+ CharType::with_nullable(nullable,
character_length_or_default(length, 1)? as usize)
+ .map_err(to_datafusion_error)?,
+ )),
+ SqlType::Varchar(length)
+ | SqlType::Nvarchar(length)
+ | SqlType::CharVarying(length)
+ | SqlType::CharacterVarying(length) => Ok(PaimonDataType::VarChar(
+ VarCharType::with_nullable(
+ nullable,
+ character_length_or_default(length, VarCharType::MAX_LENGTH)?,
+ )
+ .map_err(to_datafusion_error)?,
+ )),
+ SqlType::Text | SqlType::String(_) => Ok(PaimonDataType::VarChar(
VarCharType::with_nullable(nullable, VarCharType::MAX_LENGTH)
.map_err(to_datafusion_error)?,
)),
- SqlType::Binary(_) | SqlType::Varbinary(_) | SqlType::Bytea => {
- Ok(PaimonDataType::VarBinary(
- VarBinaryType::try_new(nullable, VarBinaryType::MAX_LENGTH)
- .map_err(to_datafusion_error)?,
- ))
- }
+ SqlType::Binary(length) => Ok(PaimonDataType::Binary(
+ BinaryType::with_nullable(nullable, u64_length_or_default(*length,
1)? as usize)
+ .map_err(to_datafusion_error)?,
+ )),
+ SqlType::Varbinary(length) => Ok(PaimonDataType::VarBinary(
+ VarBinaryType::try_new(
+ nullable,
+ binary_length_or_default(length, VarBinaryType::MAX_LENGTH)?,
+ )
+ .map_err(to_datafusion_error)?,
+ )),
+ SqlType::Bytea => Ok(PaimonDataType::VarBinary(
+ VarBinaryType::try_new(nullable, VarBinaryType::MAX_LENGTH)
+ .map_err(to_datafusion_error)?,
+ )),
SqlType::Blob(_) =>
Ok(PaimonDataType::Blob(BlobType::with_nullable(nullable))),
SqlType::Custom(name, modifiers)
if name.to_string().eq_ignore_ascii_case("VARIANT") &&
modifiers.is_empty() =>
@@ -2822,7 +2937,7 @@ mod tests {
#[test]
fn test_sql_type_string_variants() {
- use datafusion::sql::sqlparser::ast::DataType as SqlType;
+ use datafusion::sql::sqlparser::ast::{CharacterLength, DataType as
SqlType};
for sql_type in [SqlType::Varchar(None), SqlType::Text,
SqlType::String(None)] {
assert_sql_type_to_paimon(
sql_type.clone(),
@@ -2831,17 +2946,39 @@ mod tests {
),
);
}
+ assert_sql_type_to_paimon(
+ SqlType::Char(Some(CharacterLength::IntegerLength {
+ length: 7,
+ unit: None,
+ })),
+ PaimonDataType::Char(CharType::with_nullable(true, 7).unwrap()),
+ );
+ assert_sql_type_to_paimon(
+ SqlType::Varchar(Some(CharacterLength::IntegerLength {
+ length: 42,
+ unit: None,
+ })),
+ PaimonDataType::VarChar(VarCharType::with_nullable(true,
42).unwrap()),
+ );
}
#[test]
fn test_sql_type_binary() {
- use datafusion::sql::sqlparser::ast::DataType as SqlType;
+ use datafusion::sql::sqlparser::ast::{BinaryLength, DataType as
SqlType};
assert_sql_type_to_paimon(
SqlType::Bytea,
PaimonDataType::VarBinary(
VarBinaryType::try_new(true,
VarBinaryType::MAX_LENGTH).unwrap(),
),
);
+ assert_sql_type_to_paimon(
+ SqlType::Binary(Some(8)),
+ PaimonDataType::Binary(BinaryType::with_nullable(true,
8).unwrap()),
+ );
+ assert_sql_type_to_paimon(
+ SqlType::Varbinary(Some(BinaryLength::IntegerLength { length: 32
})),
+ PaimonDataType::VarBinary(VarBinaryType::try_new(true,
32).unwrap()),
+ );
}
#[test]
@@ -3558,6 +3695,16 @@ mod tests {
assert_eq!(keys, vec!["order"]);
}
+ #[test]
+ fn
test_extract_partition_by_double_quoted_identifier_with_escaped_quote_and_comma()
{
+ let (_, keys) = extract_partition_by(
+ "CREATE TABLE t (\"a\"\"b,c\" INT, `d``e,f` INT) \
+ PARTITIONED BY (\"a\"\"b,c\", `d``e,f`)",
+ )
+ .unwrap();
+ assert_eq!(keys, vec!["a\"b,c", "d`e,f"]);
+ }
+
#[test]
fn test_extract_partition_by_backtick_quoted_identifier() {
let (_, keys) =
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index c0a9457..311adf5 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -17,6 +17,7 @@
//! Paimon table provider for DataFusion.
+use std::fmt::Write as _;
use std::sync::Arc;
use async_trait::async_trait;
@@ -24,7 +25,7 @@ use datafusion::arrow::datatypes::{Field, Schema, SchemaRef
as ArrowSchemaRef};
use datafusion::catalog::Session;
use datafusion::datasource::sink::DataSinkExec;
use datafusion::datasource::{TableProvider, TableType};
-use datafusion::error::Result as DFResult;
+use datafusion::error::{DataFusionError, Result as DFResult};
use datafusion::logical_expr::dml::InsertOp;
use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
use datafusion::physical_plan::ExecutionPlan;
@@ -68,6 +69,7 @@ pub(crate) fn datafusion_read_fields(table: &Table) ->
Vec<DataField> {
pub struct PaimonTableProvider {
table: Table,
schema: ArrowSchemaRef,
+ table_definition: String,
}
impl PaimonTableProvider {
@@ -78,7 +80,12 @@ impl PaimonTableProvider {
let fields = datafusion_read_fields(&table);
let schema =
paimon::arrow::build_target_arrow_schema(&fields).map_err(to_datafusion_error)?;
- Ok(Self { table, schema })
+ let table_definition = build_table_definition(&table)?;
+ Ok(Self {
+ table,
+ schema,
+ table_definition,
+ })
}
pub fn try_new_with_blob_reader_registry(
@@ -95,6 +102,150 @@ impl PaimonTableProvider {
}
}
+/// Build a `CREATE TABLE` DDL string for a Paimon table.
+///
+/// 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> {
+ let identifier = table.identifier();
+ let schema = table.schema();
+ let mut ddl = String::new();
+ let _ = write!(
+ ddl,
+ "CREATE TABLE {}.{} (",
+ quote_identifier(identifier.database()),
+ quote_identifier(identifier.object())
+ );
+
+ for (i, field) in schema.fields().iter().enumerate() {
+ if i > 0 {
+ ddl.push_str(", ");
+ }
+ // `NOT NULL` is a column constraint; render it here at the column
+ // level rather than inside nested type arguments (see
`data_type_to_sql`).
+ let ty = data_type_to_sql(field.data_type())?;
+ if field.data_type().is_nullable() {
+ let _ = write!(ddl, "{} {}", quote_identifier(field.name()), ty);
+ } else {
+ let _ = write!(ddl, "{} {} NOT NULL",
quote_identifier(field.name()), ty);
+ }
+ }
+
+ let pks = schema.primary_keys();
+ if !pks.is_empty() {
+ ddl.push_str(", PRIMARY KEY (");
+ for (i, pk) in pks.iter().enumerate() {
+ if i > 0 {
+ ddl.push_str(", ");
+ }
+ let _ = write!(ddl, "{}", quote_identifier(pk));
+ }
+ ddl.push(')');
+ }
+ ddl.push(')');
+
+ let partition_keys = schema.partition_keys();
+ if !partition_keys.is_empty() {
+ ddl.push_str(" PARTITIONED BY (");
+ for (i, pk) in partition_keys.iter().enumerate() {
+ if i > 0 {
+ ddl.push_str(", ");
+ }
+ let _ = write!(ddl, "{}", quote_identifier(pk));
+ }
+ ddl.push(')');
+ }
+
+ let mut options: Vec<_> = schema.options().iter().collect();
+ options.sort_by_key(|(left, _)| *left);
+ if !options.is_empty() {
+ ddl.push_str(" WITH (");
+ for (i, (k, v)) in options.iter().enumerate() {
+ if i > 0 {
+ ddl.push_str(", ");
+ }
+ let _ = write!(
+ ddl,
+ "{} = {}",
+ quote_string_literal(k),
+ quote_string_literal(v)
+ );
+ }
+ ddl.push(')');
+ }
+
+ Ok(ddl)
+}
+
+fn quote_identifier(identifier: &str) -> String {
+ format!("\"{}\"", identifier.replace('"', "\"\""))
+}
+
+fn quote_string_literal(text: &str) -> String {
+ format!("'{}'", text.replace('\'', "''"))
+}
+
+/// Render a Paimon [`DataType`] as a SQL type string matching the syntax
+/// accepted by paimon-rust's `CREATE TABLE` parser.
+///
+/// `NOT NULL` is a column constraint, not a type modifier — it is only valid
+/// at the top of a column definition, not nested inside `MAP`, `ARRAY`, or
+/// `STRUCT` arguments. Callers that render a column should append `NOT NULL`
+/// themselves when the field is non-nullable; recursive calls below must not.
+fn data_type_to_sql(data_type: &DataType) -> DFResult<String> {
+ match data_type {
+ DataType::Boolean(_) => Ok("BOOLEAN".to_string()),
+ DataType::TinyInt(_) => Ok("TINYINT".to_string()),
+ DataType::SmallInt(_) => Ok("SMALLINT".to_string()),
+ DataType::Int(_) => Ok("INT".to_string()),
+ DataType::BigInt(_) => Ok("BIGINT".to_string()),
+ DataType::Decimal(t) => Ok(format!("DECIMAL({}, {})", t.precision(),
t.scale())),
+ DataType::Double(_) => Ok("DOUBLE".to_string()),
+ DataType::Float(_) => Ok("FLOAT".to_string()),
+ DataType::Binary(t) => Ok(format!("BINARY({})", t.length())),
+ DataType::VarBinary(t) => Ok(format!("VARBINARY({})", t.length())),
+ DataType::Blob(_) => Ok("BLOB".to_string()),
+ DataType::Char(t) => Ok(format!("CHAR({})", t.length())),
+ DataType::VarChar(t) => Ok(format!("VARCHAR({})", t.length())),
+ DataType::Date(_) => Ok("DATE".to_string()),
+ DataType::Time(_) => Err(unsupported_show_create_table_type("TIME")),
+ DataType::Timestamp(t) => Ok(format!("TIMESTAMP({})", t.precision())),
+ DataType::Variant(_) => Ok("VARIANT".to_string()),
+ DataType::LocalZonedTimestamp(t) => {
+ Ok(format!("TIMESTAMP({}) WITH TIME ZONE", t.precision()))
+ }
+ DataType::Array(t) => Ok(format!("ARRAY<{}>",
data_type_to_sql(t.element_type())?)),
+ DataType::Map(t) => Ok(format!(
+ "MAP({}, {})",
+ data_type_to_sql(t.key_type())?,
+ data_type_to_sql(t.value_type())?
+ )),
+ DataType::Multiset(_) =>
Err(unsupported_show_create_table_type("MULTISET")),
+ DataType::Row(t) => {
+ let inner: Vec<String> = t
+ .fields()
+ .iter()
+ .map(|f| {
+ let ty = data_type_to_sql(f.data_type())?;
+ if f.name().is_empty() {
+ Ok(ty)
+ } else {
+ Ok(format!("{} {}", quote_identifier(f.name()), ty))
+ }
+ })
+ .collect::<DFResult<_>>()?;
+ Ok(format!("STRUCT<{}>", inner.join(", ")))
+ }
+ DataType::Vector(_) =>
Err(unsupported_show_create_table_type("VECTOR")),
+ }
+}
+
+fn unsupported_show_create_table_type(type_name: &str) -> DataFusionError {
+ DataFusionError::NotImplemented(format!(
+ "SHOW CREATE TABLE does not support {type_name} columns because
paimon-rust cannot round-trip this type in CREATE TABLE"
+ ))
+}
+
/// Distribute `items` into `num_buckets` groups using round-robin assignment.
pub(crate) fn bucket_round_robin<T>(items: Vec<T>, num_buckets: usize) ->
Vec<Vec<T>> {
let mut buckets: Vec<Vec<T>> = (0..num_buckets).map(|_|
Vec::new()).collect();
@@ -170,6 +321,10 @@ impl TableProvider for PaimonTableProvider {
TableType::Base
}
+ fn get_table_definition(&self) -> Option<&str> {
+ Some(&self.table_definition)
+ }
+
async fn scan(
&self,
state: &dyn Session,
diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs
b/crates/integrations/datafusion/tests/sql_context_tests.rs
index d4f0c1d..14e6ae4 100644
--- a/crates/integrations/datafusion/tests/sql_context_tests.rs
+++ b/crates/integrations/datafusion/tests/sql_context_tests.rs
@@ -22,7 +22,11 @@ use std::sync::Arc;
use datafusion::catalog::CatalogProvider;
use datafusion::datasource::MemTable;
use paimon::catalog::Identifier;
-use paimon::spec::{ArrayType, BlobType, DataType, IntType, MapType,
VarCharType};
+use paimon::spec::{
+ ArrayType, BinaryType, BlobType, CharType, DataType, FloatType, IntType,
+ LocalZonedTimestampType, MapType, MultisetType, TimeType, VarBinaryType,
VarCharType,
+ VectorType,
+};
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
use paimon_datafusion::{PaimonCatalogProvider, SQLContext};
use tempfile::TempDir;
@@ -1262,3 +1266,451 @@ async fn
test_multiple_temporary_views_in_same_database() {
.sum::<usize>();
assert_eq!(rows2, 3);
}
+
+// ======================= SHOW CREATE TABLE =======================
+
+/// Collect the `definition` column from `SHOW CREATE TABLE` output as a
String.
+async fn collect_definition(sql_context: &SQLContext, table_ref: &str) ->
String {
+ let rows = sql_context
+ .sql(&format!("SHOW CREATE TABLE {}", table_ref))
+ .await
+ .expect("SHOW CREATE TABLE should plan")
+ .collect()
+ .await
+ .expect("SHOW CREATE TABLE should execute");
+ assert_eq!(
+ rows.len(),
+ 1,
+ "SHOW CREATE TABLE should return exactly one row"
+ );
+ let row = &rows[0];
+ assert_eq!(
+ row.num_rows(),
+ 1,
+ "SHOW CREATE TABLE should return exactly one row"
+ );
+ let val = row.column(3); // definition is the 4th column
+ let def = val
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::StringArray>()
+ .expect("definition column should be a StringArray")
+ .value(0);
+ def.to_string()
+}
+
+#[tokio::test]
+async fn test_show_create_table_simple() {
+ 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(100))")
+ .await
+ .expect("CREATE TABLE should succeed");
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t").await;
+ assert!(
+ definition.contains("CREATE TABLE \"test_db\".\"t\""),
+ "definition should start with CREATE TABLE \"test_db\".\"t\", got:
{definition}"
+ );
+ assert!(
+ definition.contains("\"id\" INT"),
+ "definition should contain `\"id\" INT`, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"name\" VARCHAR("),
+ "definition should contain `\"name\" VARCHAR(...)`, got: {definition}"
+ );
+}
+
+#[tokio::test]
+async fn test_show_create_table_with_primary_key() {
+ 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 NOT NULL, name VARCHAR,
PRIMARY KEY (id))")
+ .await
+ .expect("CREATE TABLE should succeed");
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t").await;
+ assert!(
+ definition.contains("PRIMARY KEY (\"id\")"),
+ "definition should contain PRIMARY KEY (\"id\"), got: {definition}"
+ );
+}
+
+#[tokio::test]
+async fn test_show_create_table_with_partition_and_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 ('bucket' = '4', 'file.format' =
'parquet')",
+ )
+ .await
+ .expect("CREATE TABLE should succeed");
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t").await;
+ assert!(
+ definition.contains("PARTITIONED BY (\"pt\")"),
+ "definition should contain PARTITIONED BY (\"pt\"), got: {definition}"
+ );
+ assert!(
+ definition.contains("'bucket' = '4'"),
+ "definition should contain bucket option, got: {definition}"
+ );
+ assert!(
+ definition.contains("'file.format' = 'parquet'"),
+ "definition should contain file.format option, got: {definition}"
+ );
+}
+
+#[tokio::test]
+async fn test_show_create_table_various_types() {
+ 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 (\
+ a BOOLEAN, \
+ b TINYINT, \
+ c SMALLINT, \
+ d BIGINT, \
+ e DECIMAL(10, 2), \
+ f DOUBLE, \
+ g FLOAT, \
+ h DATE, \
+ i TIMESTAMP(3), \
+ j BLOB) \
+ WITH ('data-evolution.enabled' = 'true')",
+ )
+ .await
+ .expect("CREATE TABLE should succeed");
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t").await;
+ for needle in [
+ "\"a\" BOOLEAN",
+ "\"b\" TINYINT",
+ "\"c\" SMALLINT",
+ "\"d\" BIGINT",
+ "\"e\" DECIMAL(10, 2)",
+ "\"f\" DOUBLE",
+ "\"g\" FLOAT",
+ "\"h\" DATE",
+ "\"i\" TIMESTAMP(3)",
+ "\"j\" BLOB",
+ ] {
+ assert!(
+ definition.contains(needle),
+ "definition should contain `{needle}`, got: {definition}"
+ );
+ }
+}
+
+/// Assert that two `TableSchema`s are equivalent for round-trip purposes:
+/// same fields (id, name, type), same primary keys, same partition keys.
+///
+/// We do not compare `options` because the CREATE TABLE path may inject
+/// catalog defaults (e.g. `bucket`) that the user did not specify; the
+/// schema fields and key columns are what the DDL must preserve.
+fn assert_schema_equivalent(left: &paimon::spec::TableSchema, right:
&paimon::spec::TableSchema) {
+ assert_eq!(
+ left.fields().len(),
+ right.fields().len(),
+ "field count mismatch\nleft (original): {:?}\nright (recreated):
{:?}",
+ left.fields(),
+ right.fields()
+ );
+ for (lf, rf) in left.fields().iter().zip(right.fields().iter()) {
+ assert_eq!(
+ lf.id(),
+ rf.id(),
+ "field id mismatch for `{}`: {} vs {}",
+ lf.name(),
+ lf.id(),
+ rf.id()
+ );
+ assert_eq!(
+ lf.name(),
+ rf.name(),
+ "field name mismatch: `{}` vs `{}`",
+ lf.name(),
+ rf.name()
+ );
+ assert_eq!(
+ lf.data_type(),
+ rf.data_type(),
+ "field type mismatch for `{}`: {:?} vs {:?}",
+ lf.name(),
+ lf.data_type(),
+ rf.data_type()
+ );
+ }
+ assert_eq!(
+ left.primary_keys(),
+ right.primary_keys(),
+ "primary keys mismatch: {:?} vs {:?}",
+ left.primary_keys(),
+ right.primary_keys()
+ );
+ assert_eq!(
+ left.partition_keys(),
+ right.partition_keys(),
+ "partition keys mismatch: {:?} vs {:?}",
+ left.partition_keys(),
+ right.partition_keys()
+ );
+}
+
+/// Round-trip test: the DDL returned by `SHOW CREATE TABLE` must be executable
+/// by paimon-rust's own `CREATE TABLE` parser and reproduce an equivalent
+/// schema (fields, primary keys, partition keys).
+///
+/// This guards against regressions where the rendered DDL drifts away from
+/// what the parser accepts (e.g. `ROW<name: type>` vs `STRUCT<name type>`,
+/// `MAP<k: v>` vs `MAP(k, v)`, or dropped `NOT NULL`).
+#[tokio::test]
+async fn test_show_create_table_round_trip() {
+ 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.t1 (\
+ id INT NOT NULL, \
+ name VARCHAR NOT NULL, \
+ tags ARRAY<STRING>, \
+ props MAP(INT, VARCHAR), \
+ addr STRUCT<city VARCHAR, zip VARCHAR>, \
+ meta STRUCT<kv MAP(STRING, STRING), tags ARRAY<INT>>, \
+ PRIMARY KEY (id)) \
+ PARTITIONED BY (name) \
+ WITH ('bucket' = '2', 'file.format' = 'parquet')",
+ )
+ .await
+ .expect("CREATE TABLE should succeed");
+
+ let identifier = Identifier::new("test_db", "t1");
+ let original = catalog.get_table(&identifier).await.expect("table exists");
+ let original_schema = original.schema().clone();
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.t1").await;
+ // The DDL is rendered as `CREATE TABLE test_db.t1 (...)` without the
+ // catalog prefix; paimon is the default catalog so this resolves back
+ // to the same catalog/database.
+ assert!(
+ definition.starts_with("CREATE TABLE \"test_db\".\"t1\""),
+ "definition should start with `CREATE TABLE \"test_db\".\"t1\"`, got:
{definition}"
+ );
+
+ catalog
+ .drop_table(&identifier, false)
+ .await
+ .expect("drop should succeed");
+
+ sql_context
+ .sql(&definition)
+ .await
+ .expect("DDL should re-execute")
+ .collect()
+ .await
+ .expect("DDL should execute");
+
+ let recreated = catalog
+ .get_table(&identifier)
+ .await
+ .expect("recreated table exists");
+ assert_schema_equivalent(&original_schema, recreated.schema());
+}
+
+#[tokio::test]
+async fn
test_show_create_table_round_trip_with_quoted_identifiers_and_options() {
+ 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");
+
+ let identifier = Identifier::new("test_db", "select");
+ let schema = paimon::spec::Schema::builder()
+ .column("group", DataType::Int(IntType::with_nullable(false)))
+ .column("order", DataType::Int(IntType::with_nullable(false)))
+ .column("a\"b,c", DataType::Int(IntType::new()))
+ .column(
+ "nested",
+ DataType::Row(paimon::spec::RowType::new(vec![
+ paimon::spec::DataField::new(
+ 0,
+ "from".to_string(),
+
DataType::VarChar(VarCharType::new(VarCharType::MAX_LENGTH).unwrap()),
+ ),
+ ])),
+ )
+ .column(
+ "ts_ltz",
+
DataType::LocalZonedTimestamp(LocalZonedTimestampType::new(3).unwrap()),
+ )
+ .column("fixed_char", DataType::Char(CharType::new(7).unwrap()))
+ .column(
+ "bounded_varchar",
+ DataType::VarChar(VarCharType::new(42).unwrap()),
+ )
+ .column(
+ "fixed_binary",
+ DataType::Binary(BinaryType::new(8).unwrap()),
+ )
+ .column(
+ "bounded_varbinary",
+ DataType::VarBinary(VarBinaryType::try_new(true, 32).unwrap()),
+ )
+ .primary_key(vec!["group", "order"])
+ .partition_keys(vec!["a\"b,c"])
+ .option("comment", "Bob's table")
+ .build()
+ .expect("schema should build");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .expect("table should be created");
+ let original = catalog.get_table(&identifier).await.expect("table exists");
+ let original_schema = original.schema().clone();
+
+ let definition = collect_definition(&sql_context,
"paimon.test_db.\"select\"").await;
+ assert!(
+ definition.starts_with("CREATE TABLE \"test_db\".\"select\""),
+ "definition should quote table identifiers, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"order\" INT NOT NULL"),
+ "definition should quote column identifiers, got: {definition}"
+ );
+ assert!(
+ definition.contains("PRIMARY KEY (\"group\", \"order\")"),
+ "definition should quote primary key identifiers, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"a\"\"b,c\" INT"),
+ "definition should escape quoted column identifiers, got: {definition}"
+ );
+ assert!(
+ definition.contains("PARTITIONED BY (\"a\"\"b,c\")"),
+ "definition should escape quoted partition identifiers, got:
{definition}"
+ );
+ assert!(
+ definition.contains("STRUCT<\"from\" VARCHAR"),
+ "definition should quote nested struct field identifiers, got:
{definition}"
+ );
+ assert!(
+ definition.contains("'comment' = 'Bob''s table'"),
+ "definition should escape string literals, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"ts_ltz\" TIMESTAMP(3) WITH TIME ZONE"),
+ "definition should render TIMESTAMP WITH TIME ZONE for LTZ, got:
{definition}"
+ );
+ assert!(
+ definition.contains("\"fixed_char\" CHAR(7)"),
+ "definition should preserve CHAR length, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"bounded_varchar\" VARCHAR(42)"),
+ "definition should preserve VARCHAR length, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"fixed_binary\" BINARY(8)"),
+ "definition should preserve BINARY length, got: {definition}"
+ );
+ assert!(
+ definition.contains("\"bounded_varbinary\" VARBINARY(32)"),
+ "definition should preserve VARBINARY length, got: {definition}"
+ );
+
+ catalog
+ .drop_table(&identifier, false)
+ .await
+ .expect("drop should succeed");
+
+ sql_context
+ .sql(&definition)
+ .await
+ .expect("DDL should re-execute")
+ .collect()
+ .await
+ .expect("DDL should execute");
+
+ let recreated = catalog
+ .get_table(&identifier)
+ .await
+ .expect("recreated table exists");
+ assert_schema_equivalent(&original_schema, recreated.schema());
+}
+
+#[tokio::test]
+async fn test_show_create_table_rejects_non_round_trippable_types() {
+ 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");
+
+ for (table_name, data_type, type_name) in [
+ ("time_t", DataType::Time(TimeType::new(3).unwrap()), "TIME"),
+ (
+ "multiset_t",
+
DataType::Multiset(MultisetType::new(DataType::Int(IntType::new()))),
+ "MULTISET",
+ ),
+ (
+ "vector_t",
+ DataType::Vector(VectorType::new(4,
DataType::Float(FloatType::new())).unwrap()),
+ "VECTOR",
+ ),
+ ] {
+ let identifier = Identifier::new("test_db", table_name);
+ let schema = paimon::spec::Schema::builder()
+ .column("unsupported_col", data_type)
+ .build()
+ .expect("schema should build");
+ catalog
+ .create_table(&identifier, schema, false)
+ .await
+ .expect("table should be created");
+
+ let err = sql_context
+ .sql(&format!("SHOW CREATE TABLE paimon.test_db.{table_name}"))
+ .await
+ .expect_err("SHOW CREATE TABLE should reject unsupported type");
+ assert!(
+ err.to_string().contains(type_name),
+ "error should mention {type_name}, got: {err}"
+ );
+ }
+}
