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.

Author: shyjsarah

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