diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index e7b0ba27..da74a061 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -49,10 +49,10 @@ use datafusion::error::{DataFusionError, Result as DFResult}; use datafusion::execution::SessionStateBuilder; 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, ShowCreateObject, SqlOption, Statement, - TableFactor, TableObject, Truncate, Update, Value as SqlValue, + AlterTableOperation, BinaryLength, CharacterLength, ColumnDef, ColumnOption, CreateTable, + CreateTableOptions, CreateView, Delete, Expr as SqlExpr, FromTable, Insert, Merge, ObjectName, + ObjectType, 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; @@ -1723,7 +1723,7 @@ fn column_def_to_paimon_type(col: &ColumnDef) -> DFResult { fn column_def_comment(col: &ColumnDef) -> Option { col.options.iter().find_map(|opt| match &opt.option { - datafusion::sql::sqlparser::ast::ColumnOption::Comment(comment) => Some(comment.clone()), + ColumnOption::Comment(comment) => Some(comment.clone()), _ => None, }) } @@ -1841,6 +1841,10 @@ fn sql_data_type_to_paimon_type( VarBinaryType::try_new(nullable, VarBinaryType::MAX_LENGTH) .map_err(to_datafusion_error)?, )), + other if other.to_string().eq_ignore_ascii_case("BYTES") => 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() => @@ -3401,6 +3405,61 @@ mod tests { } } + #[tokio::test] + async fn test_create_table_blob_comment_directives() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql( + "CREATE TABLE mydb.t1 (\ + id INT, \ + photo BYTES COMMENT '__BLOB_FIELD; raw photo', \ + thumb BINARY COMMENT '__BLOB_DESCRIPTOR_FIELD', \ + preview VARBINARY COMMENT '__BLOB_VIEW_FIELD; preview ref'\ + ) WITH ('data-evolution.enabled' = 'true')", + ) + .await + .unwrap(); + + let calls = catalog.take_calls(); + assert_eq!(calls.len(), 1); + if let CatalogCall::CreateTable { schema, .. } = &calls[0] { + assert!(matches!( + schema.fields()[1].data_type(), + PaimonDataType::Blob(_) + )); + assert!(matches!( + schema.fields()[2].data_type(), + PaimonDataType::Blob(_) + )); + assert!(matches!( + schema.fields()[3].data_type(), + PaimonDataType::Blob(_) + )); + assert_eq!(schema.fields()[1].description(), Some("raw photo")); + assert_eq!(schema.fields()[2].description(), None); + assert_eq!(schema.fields()[3].description(), Some("preview ref")); + assert_eq!( + schema.options().get("blob-field").map(String::as_str), + Some("photo") + ); + assert_eq!( + schema + .options() + .get("blob-descriptor-field") + .map(String::as_str), + Some("thumb") + ); + assert_eq!( + schema.options().get("blob-view-field").map(String::as_str), + Some("preview") + ); + } else { + panic!("expected CreateTable call"); + } + } + #[tokio::test] async fn test_alter_table_add_column() { let catalog = Arc::new(MockCatalog::new()); @@ -3457,6 +3516,36 @@ mod tests { } } + #[tokio::test] + async fn test_alter_table_add_blob_comment_directive_passes_core_input() { + let catalog = Arc::new(MockCatalog::new()); + let sql_context = make_sql_context(catalog.clone()).await; + + sql_context + .sql("ALTER TABLE mydb.t1 ADD COLUMN preview BYTES COMMENT '__BLOB_DESCRIPTOR_FIELD; preview descriptor'") + .await + .unwrap(); + + let calls = catalog.take_calls(); + assert_eq!(calls.len(), 1); + if let CatalogCall::AlterTable { changes, .. } = &calls[0] { + assert_eq!(changes.len(), 1); + assert!(matches!( + &changes[0], + SchemaChange::AddColumn { + field_names, + data_type, + comment, + .. + } if field_names.first().map(String::as_str) == Some("preview") + && matches!(data_type, PaimonDataType::VarBinary(_)) + && comment.as_deref() == Some("__BLOB_DESCRIPTOR_FIELD; preview descriptor") + )); + } else { + panic!("expected AlterTable call"); + } + } + #[tokio::test] async fn test_alter_table_drop_column() { let catalog = Arc::new(MockCatalog::new()); diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index a28b0fa3..fd685ee1 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -93,8 +93,9 @@ const DEFAULT_DYNAMIC_BUCKET_TARGET_ROW_NUM: i64 = 200_000; const DEFAULT_GLOBAL_INDEX_ROW_COUNT_PER_SHARD: i64 = 100_000; const DEFAULT_GLOBAL_INDEX_FALLBACK_SCAN_MAX_SIZE: i64 = 256 * 1024 * 1024; const BLOB_AS_DESCRIPTOR_OPTION: &str = "blob-as-descriptor"; -const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field"; -const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field"; +pub(crate) const BLOB_FIELD_OPTION: &str = "blob-field"; +pub(crate) const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field"; +pub(crate) const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field"; pub const BLOB_VIEW_RESOLVE_ENABLED_OPTION: &str = "blob-view.resolve.enabled"; /// Merge engine for primary-key tables. @@ -833,6 +834,17 @@ impl<'a> CoreOptions<'a> { .unwrap_or(false) } + /// Comma-separated BLOB field names stored in dedicated `.blob` files. + /// + /// Fields listed in `blob-descriptor-field` or `blob-view-field` are also + /// treated as BLOB fields, matching Java Paimon's `blob-field` semantics. + pub fn blob_fields(&self) -> HashSet { + let mut fields = self.parse_csv_set(BLOB_FIELD_OPTION); + fields.extend(self.blob_descriptor_fields()); + fields.extend(self.blob_view_fields()); + fields + } + /// Comma-separated BLOB field names stored as serialized BlobDescriptor /// bytes inline in normal data files (no .blob files for these fields). pub fn blob_descriptor_fields(&self) -> HashSet { @@ -1448,6 +1460,7 @@ mod tests { #[test] fn test_blob_view_options() { let options = HashMap::from([ + (BLOB_FIELD_OPTION.to_string(), "raw".to_string()), ( BLOB_DESCRIPTOR_FIELD_OPTION.to_string(), "thumb, payload".to_string(), @@ -1459,6 +1472,16 @@ mod tests { ]); let core = CoreOptions::new(&options); + assert_eq!( + core.blob_fields(), + HashSet::from([ + "raw".to_string(), + "thumb".to_string(), + "payload".to_string(), + "image".to_string(), + "video".to_string() + ]) + ); assert_eq!( core.blob_descriptor_fields(), HashSet::from(["thumb".to_string(), "payload".to_string()]) diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index cc3f17aa..626889ef 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -16,18 +16,23 @@ // under the License. use crate::spec::core_options::{ - first_row_supports_changelog_producer, CoreOptions, BUCKET_KEY_OPTION, - QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, + first_row_supports_changelog_producer, CoreOptions, BLOB_DESCRIPTOR_FIELD_OPTION, + BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, QUERY_AUTH_ENABLED_OPTION, + SEQUENCE_FIELD_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType}; use crate::spec::{ - remove_field_scoped_options, rename_field_scoped_options, AggregationConfig, ColumnMove, - ColumnMoveType, PartialUpdateConfig, + remove_field_scoped_options, rename_field_scoped_options, AggregationConfig, BlobType, + ColumnMove, ColumnMoveType, PartialUpdateConfig, }; use serde::{Deserialize, Serialize}; use serde_with::serde_as; use std::collections::{HashMap, HashSet}; +const BLOB_FIELD_DIRECTIVE: &str = "__BLOB_FIELD"; +const BLOB_DESCRIPTOR_FIELD_DIRECTIVE: &str = "__BLOB_DESCRIPTOR_FIELD"; +const BLOB_VIEW_FIELD_DIRECTIVE: &str = "__BLOB_VIEW_FIELD"; + /// The table schema for paimon table. /// /// Impl References: @@ -203,8 +208,8 @@ impl TableSchema { } SchemaChange::AddColumn { field_names, - data_type, - comment, + mut data_type, + mut comment, column_move, } => { let name = top_level_field(&field_names)?; @@ -214,6 +219,11 @@ impl TableSchema { column: name.to_string(), }); } + if let Some(directive) = parse_blob_comment_directive(comment.as_deref())? { + append_csv_option(&mut new_schema.options, directive.option_key, name); + comment = directive.comment; + data_type = normalize_blob_field_type(name, data_type)?; + } // Mirrors Java: an added column has no value for existing // rows, so it must be nullable. if !data_type.is_nullable() { @@ -618,6 +628,98 @@ fn reassign_field_ids(data_type: DataType, next_id: &mut i32) -> DataType { } } +struct BlobCommentDirective { + option_key: &'static str, + comment: Option, +} + +fn parse_blob_comment_directive( + comment: Option<&str>, +) -> crate::Result> { + let Some(comment) = comment else { + return Ok(None); + }; + let comment = comment.trim(); + if !comment.starts_with("__BLOB") { + return Ok(None); + } + let Some((option_key, marker)) = match_blob_comment_directive(comment) else { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Unsupported BLOB comment directive '{comment}'. Supported directives are \ + '{BLOB_FIELD_DIRECTIVE}', '{BLOB_DESCRIPTOR_FIELD_DIRECTIVE}', and \ + '{BLOB_VIEW_FIELD_DIRECTIVE}'." + ), + }); + }; + let real_comment = if comment.len() == marker.len() { + None + } else { + let real_comment = comment[marker.len() + 1..].trim(); + (!real_comment.is_empty()).then(|| real_comment.to_string()) + }; + Ok(Some(BlobCommentDirective { + option_key, + comment: real_comment, + })) +} + +fn match_blob_comment_directive(comment: &str) -> Option<(&'static str, &'static str)> { + [ + (BLOB_VIEW_FIELD_DIRECTIVE, BLOB_VIEW_FIELD_OPTION), + ( + BLOB_DESCRIPTOR_FIELD_DIRECTIVE, + BLOB_DESCRIPTOR_FIELD_OPTION, + ), + (BLOB_FIELD_DIRECTIVE, BLOB_FIELD_OPTION), + ] + .into_iter() + .find_map(|(marker, option_key)| { + if !comment.starts_with(marker) { + return None; + } + if comment.len() == marker.len() || comment.as_bytes().get(marker.len()) == Some(&b';') { + Some((option_key, marker)) + } else { + None + } + }) +} + +fn append_csv_option(options: &mut HashMap, key: &'static str, field_name: &str) { + let value = append_csv_field(options.get(key).map(String::as_str), field_name); + options.insert(key.to_string(), value); +} + +fn append_csv_field(existing: Option<&str>, field_name: &str) -> String { + let mut fields = existing + .into_iter() + .flat_map(|value| value.split(',')) + .map(str::trim) + .filter(|field| !field.is_empty()) + .map(ToString::to_string) + .collect::>(); + if !fields.iter().any(|field| field == field_name) { + fields.push(field_name.to_string()); + } + fields.join(",") +} + +fn normalize_blob_field_type(field_name: &str, data_type: DataType) -> crate::Result { + let nullable = data_type.is_nullable(); + match data_type { + DataType::Blob(_) => Ok(data_type), + DataType::Binary(_) | DataType::VarBinary(_) => { + Ok(DataType::Blob(BlobType::with_nullable(nullable))) + } + other => Err(crate::Error::ConfigInvalid { + message: format!( + "BLOB field option references non-binary column '{field_name}' with type {other:?}" + ), + }), + } +} + /// Insert a brand-new field according to an optional move (used by `AddColumn`). fn insert_field_with_move( fields: &mut Vec, @@ -792,7 +894,7 @@ pub struct Schema { impl Schema { /// Build a schema with validation. Normalizes partition/primary keys from options if present. fn new( - fields: Vec, + mut fields: Vec, partition_keys: Vec, primary_keys: Vec, mut options: HashMap, @@ -800,7 +902,8 @@ impl Schema { ) -> crate::Result { let primary_keys = Self::normalize_primary_keys(&primary_keys, &mut options)?; let partition_keys = Self::normalize_partition_keys(&partition_keys, &mut options)?; - let fields = Self::normalize_fields(&fields, &partition_keys, &primary_keys)?; + Self::normalize_blob_comment_directives(&mut fields, &mut options)?; + let fields = Self::normalize_fields(&fields, &partition_keys, &primary_keys, &options)?; Self::validate_key_field_types(&fields, &primary_keys, &options)?; Self::validate_blob_fields(&fields, &partition_keys, &options)?; Self::validate_vector_store_fields(&fields, &partition_keys, &options)?; @@ -861,12 +964,14 @@ impl Schema { Ok(partition_keys.to_vec()) } - /// Normalize fields: validate (duplicate/subset checks) and make primary key columns non-nullable. + /// Normalize fields: validate duplicate/subset checks, promote configured + /// binary BLOB fields, and make primary key columns non-nullable. /// Corresponds to Java `normalizeFields`. fn normalize_fields( fields: &[DataField], partition_keys: &[String], primary_keys: &[String], + options: &HashMap, ) -> crate::Result> { let field_names: Vec = fields.iter().map(|f| f.name().to_string()).collect(); Self::validate_no_duplicate_fields(&field_names)?; @@ -874,25 +979,34 @@ impl Schema { Self::validate_primary_keys(&field_names, primary_keys)?; Self::validate_primary_keys_not_partition_only(partition_keys, primary_keys)?; - if primary_keys.is_empty() { + let blob_field_names = CoreOptions::new(options).blob_fields(); + for name in &blob_field_names { + if !field_names.iter().any(|field| field == name) { + return Err(crate::Error::ConfigInvalid { + message: format!("BLOB field option references missing column '{name}'"), + }); + } + } + + if primary_keys.is_empty() && blob_field_names.is_empty() { return Ok(fields.to_vec()); } let pk_set: HashSet<&str> = primary_keys.iter().map(String::as_str).collect(); let mut new_fields = Vec::with_capacity(fields.len()); for f in fields { - if pk_set.contains(f.name()) && f.data_type().is_nullable() { - new_fields.push( - DataField::new( - f.id(), - f.name().to_string(), - f.data_type().copy_with_nullable(false)?, - ) - .with_description(f.description().map(|s| s.to_string())), - ); + let mut data_type = if blob_field_names.contains(f.name()) { + normalize_blob_field_type(f.name(), f.data_type().clone())? } else { - new_fields.push(f.clone()); + f.data_type().clone() + }; + if pk_set.contains(f.name()) && data_type.is_nullable() { + data_type = data_type.copy_with_nullable(false)?; } + new_fields.push( + DataField::new(f.id(), f.name().to_string(), data_type) + .with_description(f.description().map(|s| s.to_string())), + ); } Ok(new_fields) } @@ -911,6 +1025,20 @@ impl Schema { } } + fn normalize_blob_comment_directives( + fields: &mut [DataField], + options: &mut HashMap, + ) -> crate::Result<()> { + for field in fields { + let Some(directive) = parse_blob_comment_directive(field.description())? else { + continue; + }; + append_csv_option(options, directive.option_key, field.name()); + *field = field.clone().with_description(directive.comment); + } + Ok(()) + } + /// Partition key constraint must not contain duplicates; all partition keys must be in table columns. fn validate_partition_keys( field_names: &[String], @@ -1029,6 +1157,22 @@ impl Schema { } let core_options = CoreOptions::new(options); + let blob_descriptor_fields = core_options.blob_descriptor_fields(); + let blob_view_fields = core_options.blob_view_fields(); + let mut overlapping_fields = blob_view_fields + .intersection(&blob_descriptor_fields) + .cloned() + .collect::>(); + overlapping_fields.sort(); + if let Some(field) = overlapping_fields.first() { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Field '{field}' in '{BLOB_VIEW_FIELD_OPTION}' can not also be in \ + '{BLOB_DESCRIPTOR_FIELD_OPTION}'." + ), + }); + } + if !core_options.data_evolution_enabled() { return Err(crate::Error::ConfigInvalid { message: "Data evolution config must enabled for table with BLOB type column." @@ -1643,6 +1787,199 @@ mod tests { assert_eq!(schema.fields().len(), 2); } + #[test] + fn test_blob_field_option_promotes_binary_column() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column( + "payload", + DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + ) + .option("blob-field", "payload") + .option("data-evolution.enabled", "true") + .build() + .unwrap(); + + assert!(matches!(schema.fields()[1].data_type(), DataType::Blob(_))); + } + + #[test] + fn test_blob_comment_directive_promotes_column_and_strips_comment() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column_with_description( + "payload", + DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + Some("__BLOB_FIELD; payload bytes".to_string()), + ) + .option("data-evolution.enabled", "true") + .build() + .unwrap(); + + assert!(matches!(schema.fields()[1].data_type(), DataType::Blob(_))); + assert_eq!(schema.fields()[1].description(), Some("payload bytes")); + assert_eq!( + schema.options().get("blob-field").map(String::as_str), + Some("payload") + ); + } + + #[test] + fn test_blob_comment_directive_rejects_unknown_create_directive() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column_with_description( + "payload", + DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + Some("__BLOB_FIEL; payload bytes".to_string()), + ) + .option("data-evolution.enabled", "true") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { message } if message.contains("Unsupported BLOB comment directive") && message.contains("__BLOB_FIEL")), + "unknown __BLOB* comment directive should be rejected" + ); + } + + #[test] + fn test_blob_comment_directive_rejects_malformed_marker_separator() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column_with_description( + "payload", + DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + Some("__BLOB_FIELD ; payload bytes".to_string()), + ) + .option("data-evolution.enabled", "true") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { message } if message.contains("Unsupported BLOB comment directive") && message.contains("__BLOB_FIELD ; payload bytes")), + "BLOB directive marker should be followed immediately by ';'" + ); + } + + #[test] + fn test_blob_comment_directive_rejects_descriptor_view_conflict() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column_with_description( + "preview", + DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + Some("__BLOB_VIEW_FIELD".to_string()), + ) + .option("blob-descriptor-field", "preview") + .option("data-evolution.enabled", "true") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { message } + if message.contains("preview") + && message.contains("blob-view-field") + && message.contains("blob-descriptor-field")), + "field configured as both blob descriptor and blob view should be rejected" + ); + } + + #[test] + fn test_blob_comment_directive_add_column_updates_options() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("thumb", DataType::Blob(BlobType::new())) + .option("blob-descriptor-field", "thumb") + .option("data-evolution.enabled", "true") + .build() + .unwrap(); + + let schema = TableSchema::new(0, &schema) + .apply_changes(vec![crate::spec::SchemaChange::AddColumn { + field_names: vec!["preview".to_string()], + data_type: DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + comment: Some("__BLOB_DESCRIPTOR_FIELD; preview descriptor".to_string()), + column_move: None, + }]) + .unwrap(); + + let preview = schema + .fields() + .iter() + .find(|field| field.name() == "preview") + .unwrap(); + assert!(matches!(preview.data_type(), DataType::Blob(_))); + assert_eq!(preview.description(), Some("preview descriptor")); + assert_eq!( + schema + .options() + .get("blob-descriptor-field") + .map(String::as_str), + Some("thumb,preview") + ); + } + + #[test] + fn test_blob_comment_directive_rejects_unknown_add_column_directive() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .option("data-evolution.enabled", "true") + .build() + .unwrap(); + + let err = TableSchema::new(0, &schema) + .apply_changes(vec![crate::spec::SchemaChange::AddColumn { + field_names: vec!["preview".to_string()], + data_type: DataType::VarBinary( + crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH) + .unwrap(), + ), + comment: Some("__BLOB_UNKNOWN; preview descriptor".to_string()), + column_move: None, + }]) + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { message } if message.contains("Unsupported BLOB comment directive") && message.contains("__BLOB_UNKNOWN")), + "unknown __BLOB* add-column directive should be rejected" + ); + } + + #[test] + fn test_blob_field_option_rejects_non_binary_column() { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("payload", DataType::Int(IntType::new())) + .option("blob-field", "payload") + .option("data-evolution.enabled", "true") + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { message } if message.contains("non-binary column")), + "blob-field on a non-binary column should be rejected" + ); + } + #[test] fn test_partial_update_schema_validation_rejects_unsupported_options() { for (key, value) in [ diff --git a/docs/src/sql.md b/docs/src/sql.md index 86cf6128..32da83f6 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -100,7 +100,7 @@ The following SQL data types are supported in CREATE TABLE and mapped to their c | `FLOAT` / `REAL` | FloatType | | | `DOUBLE` / `DOUBLE PRECISION` | DoubleType | | | `VARCHAR` / `TEXT` / `STRING` / `CHAR` | VarCharType | | -| `BINARY` / `VARBINARY` / `BYTEA` | VarBinaryType | | +| `BINARY` / `VARBINARY` / `BYTEA` / `BYTES` | VarBinaryType | | | `VARIANT` | VariantType | Semi-structured value encoded as value + metadata binary buffers | | `BLOB` | BlobType | Binary large object | | `DATE` | DateType | | @@ -117,6 +117,48 @@ columns; DataFusion reads those as Arrow `FixedSizeList`, and vindex index creation uses `N` as the vector dimension. `SHOW CREATE TABLE` currently does not round-trip `VECTOR` columns. +### Blob Columns + +BLOB columns store large binary values using Paimon's dedicated BLOB layout. +Declare them as top-level columns and enable data evolution: + +```sql +CREATE TABLE paimon.my_db.assets ( + id INT, + picture BLOB +) WITH ( + 'data-evolution.enabled' = 'true' +); +``` + +For Java-compatible DDL, DataFusion also supports the BLOB comment directives +used by Java Paimon. A binary column with one of these directive comments is +normalized to a Paimon BLOB column in the core schema layer: + +```sql +CREATE TABLE paimon.my_db.assets ( + id INT, + picture BYTES COMMENT '__BLOB_FIELD; original image', + thumbnail BYTES COMMENT '__BLOB_DESCRIPTOR_FIELD; descriptor bytes', + picture_ref BYTES COMMENT '__BLOB_VIEW_FIELD; upstream image reference' +) WITH ( + 'data-evolution.enabled' = 'true' +); +``` + +The directive is stripped from the stored column comment; text after the first +semicolon is kept as the real comment. The directives also populate the matching +table options. A comment directive that starts with `__BLOB` but is not one of +the supported directives is rejected. + +| Comment directive | Table option | Storage semantics | +| --- | --- | --- | +| `__BLOB_FIELD` | `blob-field` | Store BLOB bytes in dedicated `.blob` files | +| `__BLOB_DESCRIPTOR_FIELD` | `blob-descriptor-field` | Store serialized `BlobDescriptor` bytes inline | +| `__BLOB_VIEW_FIELD` | `blob-view-field` | Store serialized `BlobViewStruct` bytes inline | + +The same directives are supported by `ALTER TABLE ... ADD COLUMN`. + ### Blob View Blob View stores an inline reference to a BLOB value in another table, using a