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 75ffe70 Support Java-style blob field directives (#490)
75ffe70 is described below
commit 75ffe70deb37760a979ceb081df536a6961c28aa
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 9 15:33:26 2026 +0800
Support Java-style blob field directives (#490)
---
crates/integrations/datafusion/src/sql_context.rs | 99 +++++-
crates/paimon/src/spec/core_options.rs | 27 +-
crates/paimon/src/spec/schema.rs | 377 ++++++++++++++++++++--
docs/src/sql.md | 44 ++-
4 files changed, 519 insertions(+), 28 deletions(-)
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index e7b0ba2..da74a06 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<PaimonDataType> {
fn column_def_comment(col: &ColumnDef) -> Option<String> {
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 a28b0fa..fd685ee 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<String> {
+ 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<String> {
@@ -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 cc3f17a..626889e 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:
<https://github.com/apache/paimon/blob/release-0.8.2/paimon-core/src/main/java/org/apache/paimon/schema/TableSchema.java#L47>
@@ -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<String>,
+}
+
+fn parse_blob_comment_directive(
+ comment: Option<&str>,
+) -> crate::Result<Option<BlobCommentDirective>> {
+ 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<String, String>, 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::<Vec<_>>();
+ 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<DataType> {
+ 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<DataField>,
@@ -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<DataField>,
+ mut fields: Vec<DataField>,
partition_keys: Vec<String>,
primary_keys: Vec<String>,
mut options: HashMap<String, String>,
@@ -800,7 +902,8 @@ impl Schema {
) -> crate::Result<Self> {
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<String, String>,
) -> crate::Result<Vec<DataField>> {
let field_names: Vec<String> = 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<String, String>,
+ ) -> 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::<Vec<_>>();
+ 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 c2b11cb..d3b2520 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -90,7 +90,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 | |
@@ -107,6 +107,48 @@ columns; DataFusion reads those as Arrow
`FixedSizeList<Float32>`, 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
