Skip to content

Support Java-style blob field directives #490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 1 commit into from
Jul 9, 2026
Merged

Support Java-style blob field directives #490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 94 additions & 5 deletions crates/integrations/datafusion/src/sql_context.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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,
})
}
Expand Down Expand Up @@ -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() =>
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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());
Expand Down
27 changes: 25 additions & 2 deletions crates/paimon/src/spec/core_options.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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> {
Expand Down Expand Up @@ -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(),
Expand All @@ -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()])
Expand Down
Loading
Loading