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 d53f64a0 Align inline Blob layouts and updates with Java (#1039)
d53f64a0 is described below
commit d53f64a0f8c2cb2065839b1c7f0dab6fe8e85ad5
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 5 18:37:40 2026 +0800
Align inline Blob layouts and updates with Java (#1039)
---
crates/paimon/src/spec/core_options.rs | 44 +-
crates/paimon/src/spec/schema.rs | 147 ++++++-
crates/paimon/src/table/data_evolution_fields.rs | 32 +-
crates/paimon/src/table/data_evolution_reader.rs | 12 +-
crates/paimon/src/table/data_evolution_writer.rs | 25 +-
.../src/table/dedicated_format_file_writer.rs | 17 +-
crates/paimon/src/table/inline_blob.rs | 3 -
crates/paimon/src/table/mod.rs | 57 ++-
crates/paimon/tests/rest_catalog_test.rs | 64 +++
crates/paimon/tests/table_update_test.rs | 446 +++++++++++++++++++++
10 files changed, 805 insertions(+), 42 deletions(-)
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 1775b5b7..ae04d111 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -180,6 +180,7 @@ const DEFAULT_BTREE_INDEX_HIGH_PRIORITY_POOL_RATIO: f64 =
0.1;
const BLOB_AS_DESCRIPTOR_OPTION: &str = "blob-as-descriptor";
pub(crate) const BLOB_FIELD_OPTION: &str = "blob-field";
pub(crate) const BLOB_DESCRIPTOR_FIELD_OPTION: &str = "blob-descriptor-field";
+pub(crate) const BLOB_DESCRIPTOR_FIELD_FALLBACK: &str =
"blob.stored-descriptor-fields";
pub(crate) const BLOB_VIEW_FIELD_OPTION: &str = "blob-view-field";
pub const BLOB_VIEW_RESOLVE_ENABLED_OPTION: &str = "blob-view.resolve.enabled";
const PK_VECTOR_INDEX_COLUMNS_OPTION: &str = "pk-vector.index.columns";
@@ -1788,7 +1789,13 @@ impl<'a> CoreOptions<'a> {
/// 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> {
- self.parse_csv_set(BLOB_DESCRIPTOR_FIELD_OPTION)
+ // Java ConfigOption.withFallbackKeys: presence of the canonical key,
+ // including an empty value, takes precedence over the fallback key.
+ Self::parse_csv_value(
+ self.options
+ .get(BLOB_DESCRIPTOR_FIELD_OPTION)
+ .or_else(|| self.options.get(BLOB_DESCRIPTOR_FIELD_FALLBACK)),
+ )
}
/// Comma-separated BLOB field names stored as serialized BlobViewStruct
@@ -1816,8 +1823,11 @@ impl<'a> CoreOptions<'a> {
}
fn parse_csv_set(&self, option_name: &'static str) -> HashSet<String> {
- self.options
- .get(option_name)
+ Self::parse_csv_value(self.options.get(option_name))
+ }
+
+ fn parse_csv_value(value: Option<&String>) -> HashSet<String> {
+ value
.map(|s| {
s.split(',')
.map(str::trim)
@@ -3606,6 +3616,34 @@ mod tests {
assert!(!CoreOptions::new(&disabled).blob_view_resolve_enabled());
}
+ #[test]
+ fn blob_descriptor_fallback_obeys_canonical_presence() {
+ for (canonical, expected) in [
+ (None, vec!["a", "b"]),
+ (Some("current"), vec!["current"]),
+ (Some(""), vec![]),
+ (Some(" , "), vec![]),
+ ] {
+ let mut options = HashMap::from([
+ ("blob.stored-descriptor-fields".into(), " a, b,a, ".into()),
+ ("blob-view-field".into(), "view".into()),
+ ]);
+ if let Some(value) = canonical {
+ options.insert("blob-descriptor-field".into(), value.into());
+ }
+ let core = CoreOptions::new(&options);
+ let descriptors = expected
+ .into_iter()
+ .map(String::from)
+ .collect::<HashSet<_>>();
+ assert_eq!(core.blob_descriptor_fields(), descriptors);
+ let mut inline = descriptors;
+ inline.insert("view".into());
+ assert_eq!(core.blob_inline_fields(), inline);
+ assert_eq!(core.blob_fields(), inline);
+ }
+ }
+
#[test]
fn test_validate_scan_options_rejects_unsupported() {
for key in [
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 2ff6bde6..c11ebdfa 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -17,10 +17,10 @@
use crate::spec::core_options::{
first_row_supports_changelog_producer, ChangelogProducer, CoreOptions,
MergeEngine,
- BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION,
BUCKET_KEY_OPTION,
- CHANGELOG_PRODUCER_OPTION, INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
POSTPONE_BUCKET,
- QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION,
TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION,
- TABLE_TYPE_OPTION,
+ BLOB_DESCRIPTOR_FIELD_FALLBACK, BLOB_DESCRIPTOR_FIELD_OPTION,
BLOB_FIELD_OPTION,
+ BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, CHANGELOG_PRODUCER_OPTION,
+ INDEX_FILE_IN_DATA_FILE_DIR_OPTION, POSTPONE_BUCKET,
QUERY_AUTH_ENABLED_OPTION,
+ SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION,
TABLE_TYPE_OPTION,
};
use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType,
VarCharType};
use crate::spec::{
@@ -516,6 +516,9 @@ impl TableSchema {
message: "Cannot drop all fields in
table".to_string(),
});
}
+ if fields[idx].data_type().is_blob_file_field() {
+ remove_blob_field_options(&mut new_schema.options,
name);
+ }
fields.remove(idx);
// Drop the column's field-scoped aggregation options so no
// orphaned `fields.<col>.*` keys remain (which would
otherwise
@@ -978,10 +981,46 @@ fn match_blob_comment_directive(comment: &str) ->
Option<(&'static str, &'static
}
fn append_csv_option(options: &mut HashMap<String, String>, key: &'static str,
field_name: &str) {
+ // Java ColumnDirectiveUtils.modifyFieldOptions migrates the fallback
+ // before adding a canonical entry so existing descriptor fields survive.
+ if key == BLOB_DESCRIPTOR_FIELD_OPTION &&
options.get(key).is_none_or(String::is_empty) {
+ if let Some(value) = options.remove(BLOB_DESCRIPTOR_FIELD_FALLBACK) {
+ options.insert(key.to_string(), value);
+ }
+ }
let value = append_csv_field(options.get(key).map(String::as_str),
field_name);
options.insert(key.to_string(), value);
}
+fn remove_blob_field_options(options: &mut HashMap<String, String>,
field_name: &str) {
+ for key in [
+ BLOB_FIELD_OPTION,
+ BLOB_DESCRIPTOR_FIELD_OPTION,
+ BLOB_DESCRIPTOR_FIELD_FALLBACK,
+ BLOB_VIEW_FIELD_OPTION,
+ ] {
+ let Some(value) = options.get(key) else {
+ continue;
+ };
+ // An exact empty canonical value disables fallback keys in Java.
+ // Dropping an unrelated column must preserve that override.
+ if value.is_empty() {
+ continue;
+ }
+ let retained = value
+ .split(',')
+ .map(str::trim)
+ .filter(|name| !name.is_empty() && *name != field_name)
+ .collect::<Vec<_>>()
+ .join(",");
+ if retained.is_empty() {
+ options.remove(key);
+ } else {
+ options.insert(key.to_string(), retained);
+ }
+ }
+}
+
fn append_csv_field(existing: Option<&str>, field_name: &str) -> String {
let mut fields = existing
.into_iter()
@@ -3258,7 +3297,11 @@ mod tests {
);
}
- for option in ["blob-descriptor-field", "blob-view-field"] {
+ for option in [
+ "blob-descriptor-field",
+ "blob.stored-descriptor-fields",
+ "blob-view-field",
+ ] {
let result = Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column(
@@ -3434,6 +3477,100 @@ mod tests {
);
}
+ #[test]
+ fn blob_descriptor_fallback_survives_column_directives_and_drop() {
+ for canonical in [None, Some(""), Some("thumb")] {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("thumb", DataType::Blob(BlobType::new()))
+ .option("blob.stored-descriptor-fields", "thumb")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ if let Some(value) = canonical {
+ builder = builder.option("blob-descriptor-field", value);
+ }
+ let original = TableSchema::new(0, &builder.build().unwrap());
+ let schema = original
+ .apply_changes(vec![crate::spec::SchemaChange::AddColumn {
+ field_names: vec!["preview".into()],
+ data_type: DataType::Blob(BlobType::new()),
+ comment: Some("__BLOB_DESCRIPTOR_FIELD".into()),
+ column_move: None,
+ }])
+ .unwrap();
+ assert_eq!(
+ schema.core_options().blob_descriptor_fields(),
+ HashSet::from(["thumb".into(), "preview".into()])
+ );
+ if canonical != Some("thumb") {
+ assert!(!schema
+ .options()
+ .contains_key("blob.stored-descriptor-fields"));
+ }
+ let schema = schema
+
.apply_changes(vec![crate::spec::SchemaChange::drop_column("thumb".into())])
+ .unwrap();
+ assert_eq!(
+ schema.core_options().blob_descriptor_fields(),
+ HashSet::from(["preview".into()])
+ );
+ assert!(!schema
+ .options()
+ .contains_key("blob.stored-descriptor-fields"));
+ assert_eq!(original.fields()[1].name(), "thumb");
+ assert_eq!(
+ original
+ .options()
+ .get("blob.stored-descriptor-fields")
+ .unwrap(),
+ "thumb"
+ );
+ }
+ }
+
+ #[test]
+ fn blob_descriptor_fallback_promotes_binary_and_rejects_keys() {
+ let builder = || {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "payload",
+ DataType::VarBinary(
+
crate::spec::VarBinaryType::new(crate::spec::VarBinaryType::DEFAULT_LENGTH)
+ .unwrap(),
+ ),
+ )
+ .option("blob.stored-descriptor-fields", "payload")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ };
+ assert!(matches!(
+ builder().build().unwrap().fields()[1].data_type(),
+ DataType::Blob(_)
+ ));
+ let error = builder().partition_keys(["payload"]).build().unwrap_err();
+ assert!(error.to_string().contains("partition keys"), "{error}");
+ }
+
+ #[test]
+ fn dropping_unrelated_blob_keeps_empty_descriptor_override() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("keep", DataType::Blob(BlobType::new()))
+ .column("drop", DataType::Blob(BlobType::new()))
+ .option("blob-descriptor-field", "")
+ .option("blob.stored-descriptor-fields", "keep")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .build()
+ .unwrap();
+ let changed = TableSchema::new(0, &schema)
+
.apply_changes(vec![crate::spec::SchemaChange::drop_column("drop".into())])
+ .unwrap();
+ assert_eq!(changed.options().get("blob-descriptor-field").unwrap(),
"");
+ assert!(changed.core_options().blob_descriptor_fields().is_empty());
+ }
+
#[test]
fn test_blob_comment_directive_rejects_unknown_add_column_directive() {
let schema = Schema::builder()
diff --git a/crates/paimon/src/table/data_evolution_fields.rs
b/crates/paimon/src/table/data_evolution_fields.rs
index 96476d6b..61fc9d33 100644
--- a/crates/paimon/src/table/data_evolution_fields.rs
+++ b/crates/paimon/src/table/data_evolution_fields.rs
@@ -21,7 +21,7 @@
//! of each selected field. A top-level name takes precedence over splitting at
//! a dot, because older schemas can contain literal dotted column names.
-use crate::spec::{DataField, DataType, RowType};
+use crate::spec::{CoreOptions, DataField, DataType, RowType};
use crate::{Error, Result};
use indexmap::IndexMap;
use std::collections::HashSet;
@@ -32,6 +32,36 @@ struct PathSelection {
tails: Vec<String>,
}
+/// Java BaseAppendFileStoreWrite.withWriteType omits a full column list.
+/// With the explicit optimization, it can also omit all non-dedicated fields.
+pub(super) fn can_omit_normal_write_cols(
+ fields: &[DataField],
+ written_columns: &[String],
+ options: &CoreOptions<'_>,
+) -> bool {
+ if written_columns
+ .iter()
+ .map(String::as_str)
+ .eq(fields.iter().map(DataField::name))
+ {
+ return true;
+ }
+ if !options.data_evolution_enabled()
+ || !options.data_evolution_write_cols_optimization_enabled()
+ {
+ return false;
+ }
+ let inline = options.blob_inline_fields();
+ let dedicated_vector = options.vector_file_format().is_some();
+ written_columns.iter().map(String::as_str).eq(fields
+ .iter()
+ .filter(|field| {
+ !(field.data_type().is_blob_file_field() &&
!inline.contains(field.name())
+ || dedicated_vector && matches!(field.data_type(),
DataType::Vector(_)))
+ })
+ .map(DataField::name))
+}
+
pub(super) fn project_by_paths(fields: &[DataField], paths: &[String]) ->
Result<Vec<DataField>> {
let mut selected: IndexMap<String, PathSelection> = IndexMap::new();
for path in paths {
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 3b267121..3e648ca4 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -896,7 +896,12 @@ impl DataEvolutionReader {
Vec::new()
},
true,
- HashSet::new(),
+ // View structs are physically inline even during the unresolved
+ // prescan. They must come from normal Parquet column providers.
+ self.blob_view_fields
+ .union(&self.blob_descriptor_fields)
+ .cloned()
+ .collect(),
HashSet::new(),
false,
None,
@@ -964,7 +969,8 @@ impl DataEvolutionReader {
// whose values would be transformed by resolution.
let read_type = self.wide_file_read_type.clone();
let table_fields = self.table_fields.clone();
- let blob_descriptor_fields = self.blob_descriptor_fields.clone();
+ let mut blob_inline_fields = self.blob_descriptor_fields.clone();
+ blob_inline_fields.extend(self.blob_view_fields.iter().cloned());
let blob_as_descriptor = self.blob_as_descriptor;
let blob_parallelism = self.blob_parallelism;
let batch_size = self.effective_batch_size();
@@ -994,7 +1000,7 @@ impl DataEvolutionReader {
&file_infos,
&read_type,
&table_fields,
- &blob_descriptor_fields,
+ &blob_inline_fields,
row_ranges.is_some(),
nested_enabled,
)?;
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 760f07ba..2dd6697f 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -148,7 +148,9 @@ impl DataEvolutionWriter {
});
}
let partition_keys = schema.partition_keys();
- let blob_descriptor_fields = core_options.blob_descriptor_fields();
+ // Java CoreOptions.updatableBlobFields() includes descriptor and view
+ // fields, both of which store their references in normal data files.
+ let updatable_blob_fields = core_options.blob_inline_fields();
for col in &update_columns {
let top_level =
DataEvolutionPartialWriter::top_level_write_name(col, schema.fields());
if !allow_partition_columns && partition_keys.iter().any(|key| key
== top_level) {
@@ -157,11 +159,11 @@ impl DataEvolutionWriter {
});
}
if let Some(field) = schema.fields().iter().find(|f| f.name() ==
top_level) {
- if field.data_type().is_blob_type() &&
!blob_descriptor_fields.contains(top_level) {
+ if field.data_type().is_blob_type() &&
!updatable_blob_fields.contains(top_level) {
return Err(crate::Error::Unsupported {
message: format!(
"Cannot update raw-data BLOB column '{col}' in
MERGE INTO. \
- Only BLOB columns listed in
'blob-descriptor-field' can be updated"
+ Only inline BLOB columns listed in
'blob-descriptor-field' or 'blob-view-field' can be updated"
),
});
}
@@ -322,10 +324,10 @@ impl DataEvolutionWriter {
// Rewriting an inline Blob column must preserve references for rows
// without an update. Resolving them would replace descriptors with
// payload bytes and unnecessarily require the external files to exist.
- let read_table = index.read_table.copy_with_options(HashMap::from([(
- "blob-as-descriptor".to_string(),
- "true".to_string(),
- )]));
+ let read_table = index.read_table.copy_with_options(HashMap::from([
+ ("blob-as-descriptor".to_string(), "true".to_string()),
+ ("blob-view.resolve.enabled".to_string(), "false".to_string()),
+ ]));
let file_index = &index.files;
if file_index.is_empty() {
return Err(crate::Error::DataInvalid {
@@ -1347,6 +1349,7 @@ struct PartialWriteSet {
file_format: String,
write_fields: Vec<DataField>,
write_columns: Vec<String>,
+ omit_write_cols: bool,
column_indices: Vec<usize>,
schema: Arc<arrow_schema::Schema>,
file_index_options: Option<Arc<FileIndexOptions>>,
@@ -1471,6 +1474,11 @@ impl DataEvolutionPartialWriter {
target_file_size: core_options.target_file_size(),
file_format: core_options.file_format().to_string(),
write_fields: normal_fields,
+ omit_write_cols:
super::data_evolution_fields::can_omit_normal_write_cols(
+ fields,
+ &normal_columns,
+ core_options,
+ ),
write_columns: normal_columns,
column_indices: normal_indices,
schema,
@@ -1487,6 +1495,7 @@ impl DataEvolutionPartialWriter {
file_format: format!("vector.{vector_file_format}"),
write_fields: vector_fields,
write_columns: vector_columns,
+ omit_write_cols: false,
column_indices: vector_indices,
schema,
file_index_options: None,
@@ -1556,7 +1565,7 @@ impl DataEvolutionPartialWriter {
self.format_options.clone(),
Some(0), // file_source: APPEND
Some(first_row_id),
- Some(write_set.write_columns.clone()),
+ (!write_set.omit_write_cols).then(||
write_set.write_columns.clone()),
)?
.with_file_index(write_set.file_index_options.clone());
self.writers.insert(key.clone(), writer);
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 77b24e23..df5cb99c 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -141,7 +141,7 @@ impl AppendDedicatedFormatFileWriter {
}
let normal_schema =
Arc::new(arrow_schema::Schema::new(normal_arrow_fields));
- let normal_field_names = normal_table_fields
+ let normal_field_names: Vec<String> = normal_table_fields
.iter()
.map(|field| field.name().to_string())
.collect();
@@ -180,15 +180,12 @@ impl AppendDedicatedFormatFileWriter {
};
let core_options = CoreOptions::new(format_options);
- // Full append writes include all normal fields in schema order.
Partial
- // column writers must retain their explicit write_cols instead.
- let normal_write_cols = if core_options.data_evolution_enabled()
- && core_options.data_evolution_write_cols_optimization_enabled()
- {
- None
- } else {
- Some(normal_field_names)
- };
+ let normal_write_cols =
(!super::data_evolution_fields::can_omit_normal_write_cols(
+ table_fields,
+ &normal_field_names,
+ &core_options,
+ ))
+ .then_some(normal_field_names);
let normal_index =
super::data_file_index_writer::FileIndexOptions::parse(format_options,
table_fields)?
.and_then(|options|
options.project_to_fields(&normal_table_fields))
diff --git a/crates/paimon/src/table/inline_blob.rs
b/crates/paimon/src/table/inline_blob.rs
index 1cd68c97..0a367fbd 100644
--- a/crates/paimon/src/table/inline_blob.rs
+++ b/crates/paimon/src/table/inline_blob.rs
@@ -26,9 +26,6 @@ pub(super) fn validate_inline_blob_columns(
batch: &RecordBatch,
options: &HashMap<String, String>,
) -> Result<()> {
- if !options.contains_key("blob-descriptor-field") &&
!options.contains_key("blob-view-field") {
- return Ok(());
- }
let options = CoreOptions::new(options);
for (option, fields) in [
("blob-descriptor-field", options.blob_descriptor_fields()),
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 82039211..36e75d58 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -229,7 +229,7 @@ use crate::spec::{
SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_TIMESTAMP_OPTION, SCAN_VERSION_OPTION,
SCAN_WATERMARK_OPTION,
};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
/// Table represents a table in the catalog.
#[derive(Debug, Clone)]
@@ -252,6 +252,8 @@ pub struct Table {
/// options, so scans don't have to resolve the same selector again.
/// Cleared when [`Table::copy_with_options`] changes the selector.
travel_snapshot: Option<Snapshot>,
+ /// Explicit options remain overrides across later time-travel copies.
+ applied_dynamic_option_keys: HashSet<String>,
}
impl Table {
@@ -277,6 +279,7 @@ impl Table {
rest_env,
time_traveled: false,
travel_snapshot: None,
+ applied_dynamic_option_keys: HashSet::new(),
}
}
@@ -319,6 +322,7 @@ impl Table {
rest_env: None,
time_traveled: false,
travel_snapshot: None,
+ applied_dynamic_option_keys: HashSet::new(),
})
}
@@ -566,6 +570,8 @@ impl Table {
/// [`Table::copy_with_time_travel`] when the options may select a
/// historical snapshot whose schema should be used for reading.
pub fn copy_with_options(&self, extra: HashMap<String, String>) -> Self {
+ let mut applied_dynamic_option_keys =
self.applied_dynamic_option_keys.clone();
+ applied_dynamic_option_keys.extend(extra.keys().cloned());
// Changing the time-travel selector invalidates the resolved snapshot
// (a time-travelled schema then has no matching snapshot anymore, and
// scans of such a copy fail until `copy_with_time_travel` re-resolves
@@ -589,6 +595,7 @@ impl Table {
query_auth_session: self.query_auth_session,
rest_env: self.rest_env.clone(),
time_traveled: self.time_traveled,
+ applied_dynamic_option_keys,
travel_snapshot: if selector_changed {
None
} else {
@@ -623,6 +630,7 @@ impl Table {
travel_snapshot: None,
// Not the schema the catalog loaded, so not a handle it
authorizes.
query_auth_session: None,
+ applied_dynamic_option_keys: HashSet::new(),
..self.clone()
})
}
@@ -636,11 +644,9 @@ impl Table {
pub(crate) async fn copy_with_resolved_snapshot(&self, snapshot:
&Snapshot) -> Result<Self> {
let mut table = self.copy_with_pinned_snapshot(snapshot);
if snapshot.schema_id() != self.schema.id() {
- table.schema = self
- .schema_manager
- .schema(snapshot.schema_id())
- .await?
- .copy_with_replaced_options(table.schema.options().clone());
+ let schema =
self.schema_manager.schema(snapshot.schema_id()).await?;
+ let options = table.historical_field_options(&schema);
+ table.schema = schema.copy_with_replaced_options(options);
}
Ok(table)
}
@@ -684,7 +690,8 @@ impl Table {
/// (`scan.version` / `scan.timestamp-millis` / `scan.timestamp` /
`scan.watermark` /
/// `scan.snapshot-id` / `scan.tag-name`) that resolves to a snapshot, the
/// table's fields and keys come from that snapshot's schema while the
- /// options stay the merged ones (Java `TableSchema.copy(newOptions)`).
+ /// runtime options stay merged. Column declarations follow the historical
+ /// schema unless explicitly overridden, as in Java.
/// Like Java, resolution failures fall back silently to the current
/// schema (the `if let Ok` below swallows them); an invalid selector
/// still fails later at scan planning.
@@ -785,8 +792,8 @@ impl Table {
if let Some(snapshot) = snapshot {
if snapshot.schema_id() != table.schema.id() {
let snapshot_schema =
table.schema_manager.schema(snapshot.schema_id()).await?;
- table.schema =
-
snapshot_schema.copy_with_replaced_options(table.schema.options().clone());
+ let options = table.historical_field_options(&snapshot_schema);
+ table.schema =
snapshot_schema.copy_with_replaced_options(options);
table.time_traveled = true;
}
table.travel_snapshot = Some(snapshot);
@@ -794,6 +801,37 @@ impl Table {
Ok(table)
}
+ fn historical_field_options(&self, historical_schema: &TableSchema) ->
HashMap<String, String> {
+ use crate::spec::{
+ BLOB_DESCRIPTOR_FIELD_FALLBACK, BLOB_DESCRIPTOR_FIELD_OPTION,
BLOB_FIELD_OPTION,
+ BLOB_VIEW_FIELD_OPTION,
+ };
+ let mut options = self.schema.options().clone();
+ // Canonical and fallback keys form one override group in Java
+ // AbstractFileStoreTable.excludeCurrentSchemaFieldOptions.
+ for keys in [
+ &["vector-field"][..],
+ &[BLOB_FIELD_OPTION][..],
+ &[BLOB_DESCRIPTOR_FIELD_OPTION,
BLOB_DESCRIPTOR_FIELD_FALLBACK][..],
+ &[BLOB_VIEW_FIELD_OPTION][..],
+ ] {
+ if keys
+ .iter()
+ .any(|key| self.applied_dynamic_option_keys.contains(*key))
+ {
+ continue;
+ }
+ for key in keys {
+ if let Some(value) = historical_schema.options().get(*key) {
+ options.insert((*key).to_string(), value.clone());
+ } else {
+ options.remove(*key);
+ }
+ }
+ }
+ options
+ }
+
pub async fn copy_with_branch(&self, branch_name: &str) -> Result<Self> {
// The branch schema replaces this one wholesale and could drop the
// declared type, so refuse before any branch I/O.
@@ -858,6 +896,7 @@ impl Table {
rest_env: self.rest_env.clone(),
time_traveled: false,
travel_snapshot: None,
+ applied_dynamic_option_keys: HashSet::new(),
})
}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index c0d0c38f..a7180132 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1055,6 +1055,70 @@ async fn
test_blob_view_limit_only_resolves_selected_references() {
(3, "Repeated again".to_string(), Some(b"bob".to_vec())),
]
);
+
+ // Partial files force the multi-provider merge and its View prescan.
+ // The invalid, unmatched reference must be copied without resolving it.
+ let replacement = BlobViewStruct::new(source_id, picture_field_id, 0)
+ .serialize()
+ .unwrap();
+ let builder = view.new_write_builder();
+ let update = builder.new_update().unwrap();
+ let batch = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![0, 1])) as Arc<dyn Array>,
+ ),
+ (
+ "picture",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(replacement.as_slice()),
+ None,
+ ])) as Arc<dyn Array>,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let batch = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![0])) as Arc<dyn Array>,
+ ),
+ (
+ "name",
+ Arc::new(StringArray::from(vec!["Updated"])) as Arc<dyn Array>,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![batch])
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let mut builder = rest_view.new_read_builder();
+ builder.with_limit(3);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(
+ collect_blob_rows(&batches),
+ vec![
+ (1, "Updated".to_string(), Some(b"alice".to_vec())),
+ (2, "Repeated".to_string(), None),
+ (3, "Repeated again".to_string(), Some(b"bob".to_vec())),
+ ]
+ );
}
#[cfg(not(windows))]
diff --git a/crates/paimon/tests/table_update_test.rs
b/crates/paimon/tests/table_update_test.rs
index 4ae4323b..ce85b467 100644
--- a/crates/paimon/tests/table_update_test.rs
+++ b/crates/paimon/tests/table_update_test.rs
@@ -44,6 +44,235 @@ async fn evolution_table() -> Table {
table
}
+async fn inline_blob_table(option: &str) -> Table {
+ let path = "memory:/inline_blob_update";
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(paimon::spec::BlobType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .option(option, "payload")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn inline_reference(option: &str, id: i64) -> Vec<u8> {
+ if option == "blob-view-field" {
+ paimon::spec::BlobViewStruct::new(
+ paimon::catalog::Identifier::new("missing", "upstream"),
+ 5,
+ id,
+ )
+ .serialize()
+ .unwrap()
+ } else {
+ // These files do not exist: staging a partial update must preserve
+ // references without resolving or copying external payloads.
+ paimon::spec::BlobDescriptor::new(format!("memory:/missing/{id}"), 0,
10).serialize()
+ }
+}
+
+fn inline_batch(table: &Table, ids: Vec<i32>, values: &[Option<Vec<u8>>]) ->
RecordBatch {
+ RecordBatch::try_new(
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![
+ Arc::new(Int32Array::from(ids.clone())),
+ Arc::new(arrow_array::LargeBinaryArray::from_iter(
+ values.iter().map(|value| value.as_deref()),
+ )),
+ Arc::new(Int32Array::from(
+ ids.iter().map(|id| id * 10).collect::<Vec<_>>(),
+ )),
+ ],
+ )
+ .unwrap()
+}
+
+async fn inline_rows(table: &Table) -> Vec<(i32, Option<Vec<u8>>, i32)> {
+ use arrow_array::Array;
+ let read_table = table.copy_with_options(HashMap::from([
+ ("blob-as-descriptor".into(), "true".into()),
+ ("blob-view.resolve.enabled".into(), "false".into()),
+ ]));
+ let builder = read_table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let payloads = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<arrow_array::LargeBinaryArray>()
+ .unwrap();
+ let values = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ ids.value(row),
+ payloads.is_valid(row).then(|| payloads.value(row).to_vec()),
+ values.value(row),
+ ));
+ }
+ }
+ rows.sort_by_key(|row| row.0);
+ rows
+}
+
+#[tokio::test]
+async fn inline_blob_row_id_updates_keep_unmatched_references_and_nulls() {
+ for option in [
+ "blob-descriptor-field",
+ "blob.stored-descriptor-fields",
+ "blob-view-field",
+ ] {
+ for incremental in [false, true] {
+ let table = inline_blob_table(option).await;
+ let before = vec![
+ Some(inline_reference(option, 0)),
+ None,
+ Some(inline_reference(option, 2)),
+ Some(inline_reference(option, 3)),
+ ];
+ write_batch(&table, &inline_batch(&table, vec![0, 1, 2, 3],
&before)).await;
+ let changed = Some(inline_reference(option, 20));
+ let input = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![1, 2])) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(arrow_array::LargeBinaryArray::from_iter([
+ None,
+ changed.as_deref(),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let update = table.new_write_builder().new_update().unwrap();
+ let messages = if incremental {
+ let mut writer = update.new_update_by_row_id().await.unwrap();
+ writer
+ .update_columns(vec![input], vec!["payload".into()])
+ .await
+ .unwrap()
+ } else {
+ update
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap()
+ };
+ assert!(!messages.is_empty());
+ for file in messages.iter().flat_map(|message| &message.new_files)
{
+ assert!(file.file_name.ends_with(".parquet"));
+ assert_eq!(file.write_cols, Some(vec!["payload".into()]));
+ assert_eq!(file.first_row_id, Some(0));
+ assert_eq!(file.row_count, 4);
+ }
+ commit(&table, messages).await;
+ assert_eq!(
+ inline_rows(&table).await,
+ vec![
+ (0, before[0].clone(), 0),
+ (1, None, 10),
+ (2, changed, 20),
+ (3, before[3].clone(), 30),
+ ]
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn inline_blob_upsert_and_delete_preserve_reference_layout() {
+ for option in [
+ "blob-descriptor-field",
+ "blob.stored-descriptor-fields",
+ "blob-view-field",
+ ] {
+ let table = inline_blob_table(option).await;
+ let old = inline_reference(option, 0);
+ write_batch(
+ &table,
+ &inline_batch(&table, vec![0, 1], &[Some(old.clone()), None]),
+ )
+ .await;
+ let new = inline_reference(option, 20);
+ let source = inline_batch(&table, vec![0, 2], &[Some(new.clone()),
None]);
+ let update = table.new_write_builder().new_update().unwrap();
+ let messages = update
+ .upsert_by_arrow_with_key(vec![source], vec!["id".into()])
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.file_name.ends_with(".parquet")));
+ commit(&table, messages).await;
+ assert_eq!(
+ inline_rows(&table).await,
+ vec![(0, Some(new), 0), (1, None, 10), (2, None, 20)]
+ );
+ let update = table.new_write_builder().new_update().unwrap();
+ let messages = update.delete_by_row_id(vec![0, 0]).await.unwrap();
+ commit(&table, messages).await;
+ assert_eq!(
+ inline_rows(&table).await,
+ vec![(1, None, 10), (2, None, 20)]
+ );
+ }
+}
+
+#[tokio::test]
+async fn inline_blob_failed_update_does_not_stage_files() {
+ let table = inline_blob_table("blob-view-field").await;
+ write_batch(&table, &inline_batch(&table, vec![0, 1], &[None,
None])).await;
+ let before = parquet_files(&table).await;
+ let update = table.new_write_builder().new_update().unwrap();
+ let input = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![0, 0])) as ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(arrow_array::LargeBinaryArray::from(vec![
+ Some(b"a".as_slice()),
+ Some(b"b".as_slice()),
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let error = update
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("duplicate UPDATE operations"));
+ assert_eq!(parquet_files(&table).await, before);
+ assert_eq!(inline_rows(&table).await, vec![(0, None, 0), (1, None, 10)]);
+}
+
fn batch(columns: &[(&str, Vec<i32>)]) -> RecordBatch {
RecordBatch::try_from_iter(columns.iter().map(|(name, values)| {
(
@@ -54,6 +283,223 @@ fn batch(columns: &[(&str, Vec<i32>)]) -> RecordBatch {
.unwrap()
}
+#[tokio::test]
+async fn inline_blob_alias_rejects_malformed_writes_and_updates() {
+ let table = inline_blob_table("blob.stored-descriptor-fields").await;
+ let invalid = b"not a descriptor".to_vec();
+ let mut write = table.new_write_builder().new_write().unwrap();
+ let error = write
+ .write_arrow_batch(&inline_batch(&table, vec![0],
&[Some(invalid.clone())]))
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("serialized BlobDescriptor"),
+ "{error}"
+ );
+ assert!(parquet_files(&table).await.is_empty());
+ write_batch(&table, &inline_batch(&table, vec![0], &[None])).await;
+ let before = parquet_files(&table).await;
+ let input = RecordBatch::try_from_iter(vec![
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+ (
+ "payload",
+ Arc::new(arrow_array::LargeBinaryArray::from(vec![Some(
+ invalid.as_slice(),
+ )])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let error = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![input])
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("serialized BlobDescriptor"),
+ "{error}"
+ );
+ assert_eq!(parquet_files(&table).await, before);
+ assert_eq!(inline_rows(&table).await, vec![(0, None, 0)]);
+}
+
+#[tokio::test]
+async fn normal_column_update_metadata_matches_java_blob_layout() {
+ for option in [
+ "blob-field",
+ "blob-descriptor-field",
+ "blob.stored-descriptor-fields",
+ "blob-view-field",
+ ] {
+ for optimize in [false, true] {
+ let table = inline_blob_table(option)
+ .await
+ .copy_with_options(HashMap::from([
+ (
+
"data-evolution.write-cols-optimization.enabled".into(),
+ optimize.to_string(),
+ ),
+ ("metadata.stats-mode".into(), "full".into()),
+ ]));
+ persist_table_schema(table.file_io(), table.location(),
table.schema()).await;
+ let reference = if option == "blob-field" {
+ b"raw payload".to_vec()
+ } else {
+ inline_reference(option, 0)
+ };
+ write_batch(&table, &inline_batch(&table, vec![0],
&[Some(reference)])).await;
+ let original = inline_rows(&table).await[0].1.clone();
+ let messages = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![matched(
+ vec![0],
+ &[("id", vec![1]), ("value", vec![100])],
+ )])
+ .await
+ .unwrap();
+ let file = &messages[0].new_files[0];
+ let expected = if optimize && option == "blob-field" {
+ None
+ } else {
+ Some(vec!["id".to_string(), "value".to_string()])
+ };
+ assert_eq!(file.write_cols, expected);
+ assert_eq!(file.first_row_id, Some(0));
+ commit(&table, messages).await;
+ assert_eq!(inline_rows(&table).await, vec![(1, original, 100)]);
+ }
+ }
+}
+
+#[tokio::test]
+async fn inline_blob_history_restores_layout_after_drop_and_add() {
+ for option in [
+ "blob-descriptor-field",
+ "blob.stored-descriptor-fields",
+ "blob-view-field",
+ ] {
+ let table = inline_blob_table(option).await;
+ let reference = inline_reference(option, 0);
+ write_batch(
+ &table,
+ &inline_batch(&table, vec![0, 1], &[Some(reference.clone()),
None]),
+ )
+ .await;
+ let update = table.new_write_builder().new_update().unwrap();
+ commit(
+ &table,
+ update
+ .update_by_arrow_with_row_id(vec![matched(vec![0], &[("value",
vec![100])])])
+ .await
+ .unwrap(),
+ )
+ .await;
+ let expected = inline_rows(&table).await;
+ let schema = table
+ .schema()
+ .apply_changes(vec![
+ paimon::spec::SchemaChange::drop_column("payload".into()),
+ paimon::spec::SchemaChange::AddColumn {
+ field_names: vec!["reference".into()],
+ data_type: DataType::Blob(paimon::spec::BlobType::new()),
+ comment: Some("__BLOB_DESCRIPTOR_FIELD".into()),
+ column_move: None,
+ },
+ ])
+ .unwrap();
+ persist_table_schema(table.file_io(), table.location(), &schema).await;
+ let current = table.copy_with_resolved_schema(schema, "main").unwrap();
+ let historical = current
+ .copy_with_time_travel_strict(HashMap::from([
+ ("scan.snapshot-id".into(), "2".into()),
+ ("read.batch-size".into(), "1".into()),
+ ]))
+ .await
+ .unwrap();
+ assert_eq!(
+ historical.schema().options().get(option).unwrap(),
+ "payload"
+ );
+ assert_eq!(inline_rows(&historical).await, expected);
+ if option == "blob.stored-descriptor-fields" {
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ table
+ .tag_manager()
+ .create("before_drop", &snapshot)
+ .await
+ .unwrap();
+ let branches =
+ paimon::table::BranchManager::new(table.file_io().clone(),
table.location().into());
+ branches
+ .create_branch_from_tag("dev", "before_drop")
+ .await
+ .unwrap();
+ persist_table_schema(
+ table.file_io(),
+ &branches.branch_path("dev"),
+ current.schema(),
+ )
+ .await;
+ let overridden = table
+
.copy_with_options(HashMap::from([("blob-descriptor-field".into(),
"".into())]));
+ let branch = overridden.copy_with_branch("dev").await.unwrap();
+ let history = branch
+ .copy_with_time_travel_strict(HashMap::from([(
+ "scan.snapshot-id".into(),
+ "2".into(),
+ )]))
+ .await
+ .unwrap();
+ assert_eq!(inline_rows(&history).await, expected);
+ let resolved = overridden
+ .copy_with_resolved_schema(current.schema().clone(), "main")
+ .unwrap();
+ let history = resolved
+ .copy_with_time_travel_strict(HashMap::from([(
+ "scan.snapshot-id".into(),
+ "2".into(),
+ )]))
+ .await
+ .unwrap();
+ assert_eq!(inline_rows(&history).await, expected);
+ }
+ // An explicit alias override keeps the canonical/fallback group from
+ // being restored, including across successive copies.
+ let overridden = current
+ .copy_with_options(HashMap::from([(
+ "blob.stored-descriptor-fields".into(),
+ "".into(),
+ )]))
+
.copy_with_time_travel_strict(HashMap::from([("scan.snapshot-id".into(),
"2".into())]))
+ .await
+ .unwrap();
+ assert_eq!(
+ overridden
+ .schema()
+ .options()
+ .get("blob-descriptor-field")
+ .unwrap(),
+ "reference"
+ );
+ assert_eq!(
+ overridden
+ .schema()
+ .options()
+ .get("blob.stored-descriptor-fields")
+ .unwrap(),
+ ""
+ );
+ }
+}
+
fn matched(row_ids: Vec<i64>, columns: &[(&str, Vec<i32>)]) -> RecordBatch {
let input = batch(columns);
let mut named = vec![("_ROW_ID", Arc::new(Int64Array::from(row_ids)) as
ArrayRef)];
