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 b23bde4f Support append row kinds and historical row updates (#1180)
b23bde4f is described below
commit b23bde4f66d8b069494cb8a3106d5c5ff69f8ca9
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 11:06:38 2026 +0800
Support append row kinds and historical row updates (#1180)
---
crates/paimon/src/spec/schema.rs | 48 +-
crates/paimon/src/table/mod.rs | 12 +-
crates/paimon/src/table/row_kind_generator.rs | 56 +-
crates/paimon/src/table/table_commit.rs | 160 +++++-
crates/paimon/src/table/table_merge_input.rs | 3 +-
crates/paimon/src/table/table_update.rs | 17 +-
crates/paimon/src/table/table_update_delete.rs | 2 +-
crates/paimon/src/table/table_write.rs | 70 ++-
crates/paimon/src/table/table_write_precomputed.rs | 5 +-
crates/paimon/src/table/write_builder.rs | 31 +-
crates/paimon/tests/append_row_kind_test.rs | 310 +++++++++++
.../paimon/tests/historical_table_update_test.rs | 581 +++++++++++++++++++++
crates/paimon/tests/table_predicate_delete_test.rs | 18 +-
crates/paimon/tests/table_update_test.rs | 22 +-
14 files changed, 1206 insertions(+), 129 deletions(-)
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 35e152a1..bf257eb7 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -23,7 +23,7 @@ use crate::spec::core_options::{
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::types::{ArrayType, DataType, MapType, MultisetType, RowType};
use crate::spec::{
remove_field_scoped_options, rename_field_scoped_options,
validate_no_aggregation_on_sequence_field, AggregationConfig, BlobType,
ColumnMove,
@@ -1416,7 +1416,7 @@ impl Schema {
Self::validate_first_row_changelog_producer(options)?;
Self::validate_changelog_producer_requires_primary_keys(options,
primary_keys)?;
Self::validate_read_sequence_number_requires_primary_keys(options,
primary_keys)?;
- Self::validate_rowkind_field(options, primary_keys, fields)?;
+ Self::validate_rowkind_field(options, fields)?;
Self::validate_deletion_vectors(options)?;
Self::validate_bucket_keys(options, fields, partition_keys,
primary_keys)?;
Self::validate_sequence_field(options, fields, partition_keys,
primary_keys)?;
@@ -2120,28 +2120,14 @@ impl Schema {
fn validate_rowkind_field(
options: &HashMap<String, String>,
- primary_keys: &[String],
fields: &[DataField],
) -> crate::Result<()> {
let core = CoreOptions::new(options);
let Some(field_name) = core.rowkind_field() else {
return Ok(());
};
-
- if primary_keys.is_empty() {
- return Err(crate::Error::ConfigInvalid {
- message: "rowkind.field requires a primary-key
table".to_string(),
- });
- }
-
- let producer = core
- .try_changelog_producer()
+ core.try_changelog_producer()
.map_err(Self::options_error_to_config_invalid)?;
- if producer == ChangelogProducer::Input {
- return Err(crate::Error::ConfigInvalid {
- message: "rowkind.field cannot be used with
changelog-producer=input".to_string(),
- });
- }
let Some(field) = fields.iter().find(|f| f.name() == field_name) else {
return Err(crate::Error::ConfigInvalid {
@@ -2149,13 +2135,10 @@ impl Schema {
});
};
- if !matches!(
- field.data_type(),
- DataType::VarChar(v) if v.length() == VarCharType::MAX_LENGTH
- ) {
+ if !matches!(field.data_type(), DataType::Char(_) |
DataType::VarChar(_)) {
return Err(crate::Error::ConfigInvalid {
message: format!(
- "rowkind.field '{field_name}' must be STRING (unbounded
VARCHAR) type"
+ "rowkind.field '{field_name}' must be CHARACTER_STRING
(CHAR or VARCHAR) type"
),
});
}
@@ -6591,17 +6574,13 @@ mod tests {
}
#[test]
- fn rowkind_field_requires_primary_key_table() {
- let err = Schema::builder()
+ fn rowkind_field_accepts_append_tables() {
+ Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("op", DataType::VarChar(VarCharType::string_type()))
.option("rowkind.field", "op")
.build()
- .unwrap_err();
- assert!(
- matches!(err, crate::Error::ConfigInvalid { ref message } if
message.contains("rowkind.field")),
- "got {err:?}"
- );
+ .unwrap();
}
#[test]
@@ -6690,20 +6669,15 @@ mod tests {
}
#[test]
- fn rowkind_field_rejects_changelog_producer_input() {
- let err = Schema::builder()
+ fn rowkind_field_accepts_changelog_producer_input() {
+ Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("op", DataType::VarChar(VarCharType::string_type()))
.primary_key(["id"])
.option("changelog-producer", "input")
.option("rowkind.field", "op")
.build()
- .unwrap_err();
- assert!(
- matches!(err, crate::Error::ConfigInvalid { ref message }
- if message.contains("changelog-producer=input")),
- "got {err:?}"
- );
+ .unwrap();
}
#[test]
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index f11c3426..80c7e396 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -302,7 +302,8 @@ pub struct Table {
query_auth_session: Option<u64>,
rest_env: Option<RESTEnv>,
/// True when this table copy was switched to a historical schema by
- /// [`Table::copy_with_time_travel`]. Such a copy is read-only.
+ /// [`Table::copy_with_time_travel`]. Ordinary writes and publication
reject
+ /// this copy; row-level updates may use it as their read snapshot.
time_traveled: bool,
/// Captured snapshot or empty view, so scans don't resolve latest again.
/// Cleared when [`Table::copy_with_options`] changes the selector.
@@ -801,7 +802,7 @@ impl Table {
})
}
- /// Create a read-only copy pinned to an already resolved snapshot.
+ /// Create a read view pinned to an already resolved snapshot.
///
/// Replaces any selector that originally resolved the snapshot with an
/// explicit `scan.snapshot-id`, so every subsequent scan stage observes
the
@@ -817,7 +818,7 @@ impl Table {
Ok(table)
}
- /// Create a read-only copy pinned to a snapshot from this table and
branch,
+ /// Create a read view pinned to a snapshot from this table and branch,
/// preserving the current read schema and options other than scan
selectors.
///
/// Unlike time travel, pinning must not change the fields used by an
already
@@ -1133,8 +1134,9 @@ impl Table {
})
}
- /// Whether this copy is read-only because its schema was time-traveled or
- /// its read view was explicitly pinned (including an empty view).
+ /// Whether this copy has a time-traveled schema or an explicitly pinned
+ /// read view (including an empty view). Ordinary writes and commits reject
+ /// it; row-level updates retain its snapshot as their conflict baseline.
pub fn is_time_traveled(&self) -> bool {
self.time_traveled
}
diff --git a/crates/paimon/src/table/row_kind_generator.rs
b/crates/paimon/src/table/row_kind_generator.rs
index f037dabe..9631182a 100644
--- a/crates/paimon/src/table/row_kind_generator.rs
+++ b/crates/paimon/src/table/row_kind_generator.rs
@@ -19,17 +19,17 @@
//!
//! Reference: Java `org.apache.paimon.table.sink.RowKindGenerator`.
-use arrow_array::{Array, LargeStringArray, RecordBatch, StringArray};
+use arrow_array::{LargeStringArray, RecordBatch, StringArray};
use crate::spec::{RowKind, TableSchema};
pub struct RowKindGenerator {
- index: usize,
+ field_name: String,
}
impl RowKindGenerator {
pub fn create(schema: &TableSchema, field_name: &str) ->
crate::Result<Self> {
- let index = schema
+ schema
.fields()
.iter()
.position(|f| f.name() == field_name)
@@ -40,28 +40,42 @@ impl RowKindGenerator {
),
source: None,
})?;
- Ok(Self { index })
+ Ok(Self {
+ field_name: field_name.to_owned(),
+ })
}
- pub fn generate(&self, batch: &RecordBatch, row: usize) ->
crate::Result<RowKind> {
- let col = batch.column(self.index);
- if col.is_null(row) {
- return Err(crate::Error::DataInvalid {
- message: "Row kind cannot be null.".to_string(),
- source: None,
- });
- }
- let value = if let Some(strings) =
col.as_any().downcast_ref::<StringArray>() {
- strings.value(row)
+ pub fn generate_batch(&self, batch: &RecordBatch) ->
crate::Result<Vec<RowKind>> {
+ // with_write_type may reorder the physical columns. Resolve once for
+ // the batch, then parse values without a per-row field-name search.
+ let col =
+ batch
+ .column_by_name(&self.field_name)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: format!(
+ "Cannot find rowkind field '{}' in write input",
+ self.field_name
+ ),
+ source: None,
+ })?;
+ let parse = |value: Option<&str>| {
+ value
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Row kind cannot be null.".into(),
+ source: None,
+ })
+ .and_then(RowKind::from_short_string)
+ };
+ if let Some(strings) = col.as_any().downcast_ref::<StringArray>() {
+ strings.iter().map(parse).collect()
} else if let Some(strings) =
col.as_any().downcast_ref::<LargeStringArray>() {
- strings.value(row)
+ strings.iter().map(parse).collect()
} else {
- return Err(crate::Error::DataInvalid {
- message: "rowkind.field column must be
Utf8/String".to_string(),
+ Err(crate::Error::DataInvalid {
+ message: "rowkind.field column must be Utf8/String".into(),
source: None,
- });
- };
- RowKind::from_short_string(value)
+ })
+ }
}
}
@@ -104,6 +118,6 @@ mod tests {
],
)
.unwrap();
- assert_eq!(gen.generate(&batch, 0).unwrap(), RowKind::Delete);
+ assert_eq!(gen.generate_batch(&batch).unwrap(), vec![RowKind::Delete]);
}
}
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index df9ec40a..20846ea6 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -1796,11 +1796,10 @@ impl TableCommit {
self.check_global_index_sources(latest_snapshot, entries,
new_index_entries)
.await?;
let mut index_entries = new_index_entries.clone();
- index_entries.extend(self.global_index_update_entries(
- &previous,
- entries,
- new_index_entries,
- )?);
+ index_entries.extend(
+ self.global_index_update_entries(&previous, entries,
new_index_entries)
+ .await?,
+ );
let all = Self::merge_index_entries(&previous, &index_entries,
false)?;
self.check_deletion_vector_references(
latest_snapshot,
@@ -2019,34 +2018,69 @@ impl TableCommit {
Ok(all)
}
- fn global_index_update_entries(
+ async fn global_index_update_entries(
&self,
previous_entries: &[IndexManifestEntry],
commit_entries: &[ManifestEntry],
new_index_entries: &[IndexManifestEntry],
) -> Result<Vec<IndexManifestEntry>> {
+ if !previous_entries
+ .iter()
+ .any(|entry| entry.index_file.global_index_meta.is_some())
+ {
+ return Ok(vec![]);
+ }
+ let action =
+
CoreOptions::new(self.table.schema().options()).global_index_column_update_action()?;
+ if action == GlobalIndexColumnUpdateAction::Ignore {
+ return Ok(vec![]);
+ }
let mut updated_cols = HashSet::new();
+ let mut updated_field_ids = HashSet::new();
let mut written_partitions: Vec<Vec<u8>> = Vec::new();
for entry in commit_entries
.iter()
.filter(|entry| *entry.kind() == FileKind::Add)
{
- let Some(write_cols) = entry.file().write_cols.as_ref() else {
- continue;
- };
+ let file = entry.file();
// Dedicated normal/vector/blob INSERT files also carry write_cols.
// Only files targeting existing row IDs can invalidate an
existing index.
- if entry.file().first_row_id.is_none()
- && !write_cols
- .iter()
- .any(|col| col == crate::spec::ROW_ID_FIELD_NAME)
+ if file.first_row_id.is_none()
+ && !file.write_cols.as_ref().is_some_and(|columns| {
+ columns
+ .iter()
+ .any(|col| col == crate::spec::ROW_ID_FIELD_NAME)
+ })
{
continue;
}
- for col in write_cols {
- if !is_system_field(col) {
- updated_cols.insert(col.clone());
- }
+ // Partial files retain their writer's schema, including historical
+ // names. Index metadata identifies fields by ID; comparing against
+ // today's names would miss renames and confuse dropped/readded
names.
+ let historical_schema = if file.schema_id ==
self.table.schema().id() {
+ None
+ } else {
+ Some(self.table.schema_manager().schema(file.schema_id).await?)
+ };
+ let schema = historical_schema
+ .as_deref()
+ .unwrap_or_else(|| self.table.schema());
+ let fields = match &file.write_cols {
+ Some(columns) =>
super::data_evolution_fields::project_by_paths(
+ schema.fields(),
+ &columns
+ .iter()
+ .filter(|name| !is_system_field(name))
+ .cloned()
+ .collect::<Vec<_>>(),
+ )?,
+ // A complete-row update can omit write_cols, as can a compact
+ // normal file which excludes dedicated Blob/vector columns.
+ None => schema.data_file_fields(None),
+ };
+ for field in fields {
+ updated_field_ids.insert(field.id());
+ updated_cols.insert(field.name().to_string());
}
if !written_partitions
.iter()
@@ -2089,8 +2123,8 @@ impl TableCommit {
}
let matched = indexed_field_ids
.iter()
+ .filter(|field_id| updated_field_ids.contains(field_id))
.filter_map(|field_id| field_by_id.get(field_id))
- .filter(|field_name| updated_cols.contains(*field_name))
.cloned()
.collect::<Vec<_>>();
if !matched.is_empty() {
@@ -2102,7 +2136,7 @@ impl TableCommit {
return Ok(vec![]);
}
- match
CoreOptions::new(self.table.schema().options()).global_index_column_update_action()?
{
+ match action {
GlobalIndexColumnUpdateAction::Ignore => Ok(vec![]),
GlobalIndexColumnUpdateAction::DropPartitionIndex => Ok(affected
.into_iter()
@@ -5206,6 +5240,94 @@ mod tests {
assert_eq!(index_entries[0].index_file.file_name, "lumina-0.index");
}
+ #[tokio::test]
+ async fn test_index_update_policy_uses_file_schema_identity() {
+ for mode in ["rename", "drop_readd", "complete_row"] {
+ for action in ["THROW_ERROR", "DROP_PARTITION_INDEX", "IGNORE"] {
+ for extra_field in [false, true] {
+ let file_io = test_file_io();
+ let path =
format!("memory:/index_identity_{mode}_{action}_{extra_field}");
+ setup_dirs(&file_io, &path).await;
+ let old = test_schema();
+ let mut encoded = serde_json::to_value(&old).unwrap();
+ encoded["id"] = serde_json::json!(1);
+ if mode == "rename" {
+ encoded["fields"][0]["name"] =
serde_json::json!("renamed_id");
+ } else if mode == "drop_readd" {
+ encoded["fields"][0]["id"] = serde_json::json!(2);
+ encoded["highestFieldId"] = serde_json::json!(2);
+ }
+ let current: TableSchema =
serde_json::from_value(encoded).unwrap();
+ let current = current.copy_with_options(HashMap::from([(
+ "global-index.column-update-action".into(),
+ action.into(),
+ )]));
+ let index_id = current.fields()[0].id();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "t"),
+ path.clone(),
+ current,
+ None,
+ );
+ file_io.mkdirs(&format!("{path}/schema/")).await.unwrap();
+ file_io
+ .new_output(&format!("{path}/schema/schema-0"))
+ .unwrap()
+ .write(serde_json::to_vec(&old).unwrap().into())
+ .await
+ .unwrap();
+ let commit = TableCommit::new(table, "test-user".into());
+ let index_file = if extra_field {
+ test_global_index_file_with_extra_fields(
+ "id.index",
+ 1,
+ vec![index_id],
+ 0,
+ 9,
+ )
+ } else {
+ test_global_index_file("id.index", index_id, 0, 9)
+ };
+ let previous = vec![IndexManifestEntry {
+ kind: FileKind::Add,
+ partition: vec![],
+ bucket: 0,
+ index_file,
+ version: 1,
+ }];
+ let mut file = test_data_file("update.parquet", 10);
+ file.first_row_id = Some(0);
+ if mode == "complete_row" {
+ file.schema_id = 1;
+ file.write_cols = None;
+ } else {
+ file.write_cols = Some(vec!["id".into()]);
+ }
+ let entries =
+
commit.messages_to_entries(&[CommitMessage::new(vec![], 0, vec![file])]);
+ let result = commit
+ .global_index_update_entries(&previous, &entries, &[])
+ .await;
+ if mode == "drop_readd" || action == "IGNORE" {
+ assert!(result.unwrap().is_empty(),
"{mode}/{action}/{extra_field}");
+ } else if action == "THROW_ERROR" {
+ let error = result.unwrap_err().to_string();
+ assert!(error.contains("globally indexed columns"),
"{error}");
+ if mode == "rename" {
+ assert!(error.contains("renamed_id"), "{error}");
+ }
+ } else {
+ let deleted = result.unwrap();
+ assert_eq!(deleted.len(), 1,
"{mode}/{action}/{extra_field}");
+ assert_eq!(deleted[0].kind, FileKind::Delete);
+ assert_eq!(deleted[0].index_file,
previous[0].index_file);
+ }
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn test_partial_update_indexed_column_rejects_by_default() {
let file_io = test_file_io();
diff --git a/crates/paimon/src/table/table_merge_input.rs
b/crates/paimon/src/table/table_merge_input.rs
index 4cb83ed3..8c95ee07 100644
--- a/crates/paimon/src/table/table_merge_input.rs
+++ b/crates/paimon/src/table/table_merge_input.rs
@@ -84,7 +84,8 @@ impl MergeInput {
});
}
crate::spec::CoreOptions::new(table.schema().options()).validate_scan_options()?;
- let schema =
crate::arrow::build_target_arrow_schema(table.schema().fields())?;
+ let schema =
+
crate::arrow::build_target_arrow_schema(&table.schema().fields_with_read_metadata())?;
for column in columns {
schema
.field_with_name(column)
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index 49001080..de07c4d3 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -46,6 +46,10 @@ fn invalid(message: impl Into<String>) -> crate::Error {
/// Row-ID updates infer columns from the input unless configured with
/// [`with_update_type`](Self::with_update_type). Key-based upserts accept a
/// column subset including the keys and match keys within each partition.
+/// Row updates, deletes and shard updates may select a historical read view;
+/// their messages retain its snapshot as the conflict baseline. Publish those
+/// messages using a current table. Upsert and MERGE with INSERT clauses
+/// require a current table as ordinary writes do.
#[derive(Clone)]
pub struct TableUpdate {
table: Table,
@@ -256,6 +260,8 @@ impl TableUpdate {
batches: Vec<RecordBatch>,
upsert_keys: Vec<String>,
) -> crate::Result<Vec<CommitMessage>> {
+ // Unlike partial row updates, an upsert may append complete rows.
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
let mut writer = super::table_upsert::TableUpsert::new(
&self.table,
self.commit_user.clone(),
@@ -281,6 +287,9 @@ impl TableUpdate {
when_matched: Vec<super::WhenMatched>,
when_not_matched: Vec<super::WhenNotMatched>,
) -> crate::Result<Vec<CommitMessage>> {
+ if !when_not_matched.is_empty() {
+ super::write_builder::ensure_table_write_allowed(&self.table)?;
+ }
super::table_merge_into::merge_into(
&self.table,
&self.commit_user,
@@ -314,11 +323,9 @@ impl TableUpdate {
if row_ids.is_empty() {
return Ok(Vec::new());
}
- let mut writer = self
- .table
- .new_write_builder()
- .with_commit_user(self.commit_user.clone())?
- .new_delete()?;
+ // The high-level updater selects a read snapshot; the low-level
+ // ordinary write builder continues rejecting historical handles.
+ let mut writer = super::DataEvolutionDeleteWriter::new(&self.table)?;
writer.add_row_ids(row_ids)?;
writer.prepare_commit().await
}
diff --git a/crates/paimon/src/table/table_update_delete.rs
b/crates/paimon/src/table/table_update_delete.rs
index bd2d99e3..d7029350 100644
--- a/crates/paimon/src/table/table_update_delete.rs
+++ b/crates/paimon/src/table/table_update_delete.rs
@@ -30,7 +30,7 @@ pub(super) async fn delete(
table: &Table,
predicate: Option<Predicate>,
) -> crate::Result<Vec<CommitMessage>> {
- super::write_builder::ensure_table_write_allowed(table)?;
+ super::write_builder::ensure_paimon_table(table)?;
// Partition-only deletes use manifest metadata directly, but still need
// the same unrestricted whole-table grant as row-level deletes. Authorize
// before the empty-target and partition-optimization branches.
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 80edf85a..39842497 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -424,9 +424,7 @@ impl TableWrite {
}
let row_kind_generator = match core_options.rowkind_field() {
- Some(field_name) if has_primary_keys => {
- Some(RowKindGenerator::create(schema, field_name)?)
- }
+ Some(field_name) => Some(RowKindGenerator::create(schema,
field_name)?),
_ => None,
};
let row_kind_filter = RowKindFilter::of(core_options);
@@ -784,8 +782,9 @@ impl TableWrite {
return writer.write(batch).await;
}
self.ensure_active()?;
- let input = batch;
- let Some(batch) = self.normalize_write_batch(input)? else {
+ let filtered = self.filter_write_input(batch)?;
+ let input = filtered.as_ref().unwrap_or(batch);
+ let Some(batch) = self.normalize_filtered_write_batch(input)? else {
return Ok(());
};
let routing = self.normalize_automatic_routing_input(input)?;
@@ -807,7 +806,11 @@ impl TableWrite {
pub(super) fn normalize_write_batch(&self, batch: &RecordBatch) ->
Result<Option<RecordBatch>> {
let filtered = self.filter_write_input(batch)?;
- let batch =
self.validate_write_batch_schema(filtered.as_ref().unwrap_or(batch))?;
+ self.normalize_filtered_write_batch(filtered.as_ref().unwrap_or(batch))
+ }
+
+ fn normalize_filtered_write_batch(&self, batch: &RecordBatch) ->
Result<Option<RecordBatch>> {
+ let batch = self.validate_write_batch_schema(batch)?;
// Filtering already precedes value normalization. Generate the system
// column now so every surviving PK batch has the same buffer schema.
let batch = self.generate_rowkind_batch(&batch)?;
@@ -870,16 +873,16 @@ impl TableWrite {
/// its optional RowKindFilter, independently of physical KV nullability.
/// Ignored rows never reach value extraction or nested writers.
fn filter_write_input(&self, batch: &RecordBatch) ->
Result<Option<RecordBatch>> {
- if self.primary_key_indices.is_empty() {
+ // Ordinary Arrow append input has no event header. It represents
+ // INSERTs and needs no kind allocation or pre-routing filter.
+ if self.primary_key_indices.is_empty() &&
self.row_kind_generator.is_none() {
return Ok(None);
}
// Reject malformed schemas even when every event would be ignored.
// Empty arrays check equivalent layouts without examining row values.
self.validate_write_batch_schema(&RecordBatch::new_empty(batch.schema()))?;
let kinds = if let Some(generator) = &self.row_kind_generator {
- (0..batch.num_rows())
- .map(|row| generator.generate(batch, row))
- .collect::<Result<Vec<_>>>()?
+ generator.generate_batch(batch)?
} else if let Some(column) =
batch.column_by_name(VALUE_KIND_FIELD_NAME) {
column
.as_any()
@@ -896,15 +899,18 @@ impl TableWrite {
// still identify the event; other engines and partition migrations
// require every declared top-level NOT NULL field.
let delete_required_fields = self.delete_not_null_fields();
+ let projected = self.project_full_write_input(batch)?;
+ let physical = projected.as_ref().unwrap_or(batch);
for (index, field) in self.write_schema.fields().iter().enumerate() {
- if field.is_nullable() || batch.column(index).null_count() == 0 {
+ if field.is_nullable() || physical.column(index).null_count() == 0
{
continue;
}
let delete_required = delete_required_fields
.as_ref()
.is_none_or(|fields| fields.contains(field.name()));
for (row, kind) in kinds.iter().enumerate() {
- if batch.column(index).is_null(row) && (*kind !=
RowKind::Delete || delete_required)
+ if physical.column(index).is_null(row)
+ && (*kind != RowKind::Delete || delete_required)
{
return Err(crate::Error::DataInvalid {
message: format!("Cannot write null to non-null
column({})", field.name()),
@@ -913,14 +919,25 @@ impl TableWrite {
}
}
}
- let Some(filter) = &self.row_kind_filter else {
- return Ok(None);
- };
- let rows = kinds
- .into_iter()
- .enumerate()
- .filter_map(|(row, kind)| filter.test(kind).then_some(row))
- .collect::<Vec<_>>();
+ let mut rows = Vec::with_capacity(batch.num_rows());
+ for (row, kind) in kinds.into_iter().enumerate() {
+ if self
+ .row_kind_filter
+ .as_ref()
+ .is_some_and(|filter| !filter.test(kind))
+ {
+ continue;
+ }
+ if self.primary_key_indices.is_empty()
+ && !matches!(kind, RowKind::Insert | RowKind::UpdateAfter)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Append only writer cannot accept row
with RowKind {kind:?}"),
+ source: None,
+ });
+ }
+ rows.push(row);
+ }
if rows.len() == batch.num_rows() {
Ok(None)
} else {
@@ -1265,6 +1282,11 @@ impl TableWrite {
}
fn generate_rowkind_batch(&self, batch: &RecordBatch) ->
Result<RecordBatch> {
+ // Append rows have already been filtered and checked. Their events do
+ // not become a physical KV system column.
+ if self.primary_key_indices.is_empty() {
+ return Ok(batch.clone());
+ }
let Some(generator) = &self.row_kind_generator else {
// Every PK batch sent to a buffered writer uses the same schema.
// Mixing default INSERT batches with explicit kinds must neither
@@ -1291,9 +1313,11 @@ impl TableWrite {
});
}
- let kinds = (0..batch.num_rows())
- .map(|row| generator.generate(batch, row).map(|kind|
kind.to_value()))
- .collect::<Result<Vec<_>>>()?;
+ let kinds = generator
+ .generate_batch(batch)?
+ .into_iter()
+ .map(|kind| kind.to_value())
+ .collect();
Self::add_per_row_value_kind_column(batch, kinds)
}
diff --git a/crates/paimon/src/table/table_write_precomputed.rs
b/crates/paimon/src/table/table_write_precomputed.rs
index 35f6ad30..3730d3d0 100644
--- a/crates/paimon/src/table/table_write_precomputed.rs
+++ b/crates/paimon/src/table/table_write_precomputed.rs
@@ -63,8 +63,9 @@ impl TableWrite {
upper_bound - 1,
)));
}
- let input = batch;
- let Some(batch) = self.normalize_write_batch(input)? else {
+ let filtered = self.filter_write_input(batch)?;
+ let input = filtered.as_ref().unwrap_or(batch);
+ let Some(batch) = self.normalize_filtered_write_batch(input)? else {
return Ok(());
};
let partitions = self.precomputed_partition_bytes(input, &batch)?;
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index a073c7ec..2c3744ed 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -262,7 +262,11 @@ impl<'a> PaimonWriteBuilder<'a> {
/// Create a configured table update object.
pub fn new_update(&self) -> crate::Result<TableUpdate> {
- ensure_table_write_allowed(self.table)?;
+ // Row-level updates can select an older read snapshot. Their partial
+ // files retain that schema and their messages carry the snapshot used
+ // for conflict detection. Appending and publishing still require a
+ // current table, including append-capable operations on TableUpdate.
+ ensure_paimon_table(self.table)?;
if self.overwrite {
return Err(crate::Error::Unsupported {
message: "TableUpdate does not support overwrite".to_string(),
@@ -304,16 +308,7 @@ pub(super) fn ensure_table_write_authorized(table: &Table)
-> crate::Result<()>
}
pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> {
- let options = crate::spec::CoreOptions::new(table.schema().options());
- let declared = options.table_type()?;
- if declared.requires_table_engine() {
- return Err(crate::Error::Unsupported {
- message: format!(
- "table '{}' is declared '{declared}' and cannot be written as
a Paimon table",
- table.identifier().full_name()
- ),
- });
- }
+ ensure_paimon_table(table)?;
// A time-travel table may carry a historical schema.
let selector =
crate::spec::CoreOptions::new(table.schema().options()).try_time_travel_selector();
@@ -327,6 +322,20 @@ pub(super) fn ensure_table_write_allowed(table: &Table) ->
crate::Result<()> {
Ok(())
}
+pub(super) fn ensure_paimon_table(table: &Table) -> crate::Result<()> {
+ let options = crate::spec::CoreOptions::new(table.schema().options());
+ let declared = options.table_type()?;
+ if declared.requires_table_engine() {
+ return Err(crate::Error::Unsupported {
+ message: format!(
+ "table '{}' is declared '{declared}' and cannot be written as
a Paimon table",
+ table.identifier().full_name()
+ ),
+ });
+ }
+ Ok(())
+}
+
pub(super) fn validate_commit_user(commit_user: &str) -> crate::Result<()> {
let is_invalid = commit_user.is_empty()
|| commit_user == "."
diff --git a/crates/paimon/tests/append_row_kind_test.rs
b/crates/paimon/tests/append_row_kind_test.rs
new file mode 100644
index 00000000..313e5d40
--- /dev/null
+++ b/crates/paimon/tests/append_row_kind_test.rs
@@ -0,0 +1,310 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{ArrayRef, Int32Array, Int8Array, RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{CharType, DataType, IntType, Schema, TableSchema,
VarCharType};
+use paimon::table::Table;
+use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+use std::sync::Arc;
+
+async fn make_table(evolution: bool, options: &[(&str, &str)]) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .column("pt", DataType::VarChar(VarCharType::string_type()))
+ .partition_keys(["pt"])
+ .option("rowkind.field", "op")
+ .option("data-evolution.enabled", evolution.to_string())
+ .option("row-tracking.enabled", evolution.to_string());
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ let path = "memory:/append-row-kind";
+ let (io, table) = memory_table(path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ table
+}
+
+fn batch(rows: &[(Option<i32>, &str, &str)]) -> RecordBatch {
+ RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from_iter(rows.iter().map(|x| x.0))) as
ArrayRef,
+ ),
+ (
+ "op",
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|x| x.1)))
as ArrayRef,
+ ),
+ (
+ "pt",
+ Arc::new(StringArray::from_iter_values(rows.iter().map(|x| x.2)))
as ArrayRef,
+ ),
+ ])
+ .unwrap()
+}
+
+async fn ids(table: &Table) -> Vec<i32> {
+ let read = table.new_read_builder();
+ let plan = read.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ ids.sort();
+ ids
+}
+
+#[tokio::test]
+async fn append_rejects_retracts_and_filters_only_configured_events() {
+ for evolution in [false, true] {
+ for kind in ["-D", "-U"] {
+ let table = make_table(evolution, &[]).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(&[(Some(1), kind, "bad")]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("Append"), "{error}");
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ writer.close().await;
+ }
+ let table = make_table(evolution, &[("ignore-delete", "true")]).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(&[
+ (Some(1), "-D", "bad"),
+ (Some(2), "+U", "keep"),
+ (Some(3), "-U", "bad"),
+ (Some(4), "+I", "keep"),
+ ]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 2
+ );
+ builder.new_commit().commit(messages).await.unwrap();
+ writer.close().await;
+ assert_eq!(ids(&table).await, vec![2, 4]);
+ }
+}
+
+#[tokio::test]
+async fn
append_checks_required_values_before_filter_and_keeps_ignore_before_distinct() {
+ let table = make_table(false, &[("ignore-delete", "true")]).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(&[(None, "-D", "bad")]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("non-null column(id)"), "{error}");
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ writer.close().await;
+
+ let table = make_table(false, &[("ignore-update-before", "true")]).await;
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(&[(Some(1), "-U", "bad")]))
+ .await
+ .unwrap();
+ let error = writer
+ .write_arrow_batch(&batch(&[(Some(2), "-D", "bad")]))
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("Append"), "{error}");
+ assert!(writer.prepare_commit().await.unwrap().is_empty());
+ writer.close().await;
+}
+
+#[tokio::test]
+async fn append_selected_write_type_filters_the_same_rows_as_routing() {
+ for precomputed in [false, true] {
+ for reordered in [false, true] {
+ let table = make_table(
+ false,
+ &[
+ ("ignore-delete", "true"),
+ ("bucket", "2"),
+ ("bucket-key", "id"),
+ ],
+ )
+ .await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ // Full input retains omitted event/routing fields, whereas a
+ // reordered input resolves the generator at its current position.
+ let names = if reordered {
+ vec!["op", "id", "pt"]
+ } else {
+ vec!["id"]
+ };
+ writer
+
.with_write_type(names.into_iter().map(str::to_owned).collect())
+ .unwrap();
+ let input = batch(&[(Some(1), "-D", "bad"), (Some(2), "+I",
"keep")]);
+ let input = if reordered {
+ input.project(&[1, 0, 2]).unwrap()
+ } else {
+ input
+ };
+ writer
+ .write_arrow(&[input], precomputed.then_some(0))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|m| &m.new_files)
+ .map(|f| f.row_count)
+ .sum::<i64>(),
+ 1
+ );
+ builder.new_commit().commit(messages).await.unwrap();
+ writer.close().await;
+ assert_eq!(ids(&table).await, vec![2]);
+ }
+ }
+}
+
+#[tokio::test]
+async fn generated_primary_key_kinds_preserve_input_changelog_events() {
+ for op_type in [
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::VarChar(VarCharType::new(2).unwrap()),
+ DataType::Char(CharType::new(2).unwrap()),
+ ] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("op", op_type)
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("rowkind.field", "op")
+ .option("changelog-producer", "input")
+ .build()
+ .unwrap();
+ let path = "memory:/generated-input-changelog";
+ let (io, table) = memory_table(path, TableSchema::new(0, &schema));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, table.schema()).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let input = RecordBatch::try_from_iter(vec![
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 1, 2, 2])) as ArrayRef,
+ ),
+ (
+ "value",
+ Arc::new(Int32Array::from(vec![10, 20, 25, 30])) as ArrayRef,
+ ),
+ (
+ "op",
+ Arc::new(StringArray::from(vec!["+I", "-D", "-U", "+U"])) as
ArrayRef,
+ ),
+ ])
+ .unwrap();
+ writer.write_arrow_batch(&input).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ let logs: Vec<_> = messages
+ .iter()
+ .flat_map(|m| &m.new_changelog_files)
+ .collect();
+ assert_eq!(logs.len(), 1);
+ assert_eq!(logs[0].row_count, 4);
+ assert_eq!(logs[0].delete_row_count, Some(2));
+ let bytes = io
+ .new_input(&format!("{path}/bucket-0/{}", logs[0].file_name))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let batches = ParquetRecordBatchReaderBuilder::try_new(bytes)
+ .unwrap()
+ .build()
+ .unwrap()
+ .collect::<Result<Vec<_>, _>>()
+ .unwrap();
+ let values: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ let kinds: Vec<_> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("_VALUE_KIND")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int8Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect();
+ assert_eq!(values, vec![10, 20, 25, 30]);
+ assert_eq!(kinds, vec![0, 3, 1, 2]);
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(ids(&table).await, vec![2]);
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.changelog_record_count(), Some(4));
+ writer.close().await;
+ }
+}
diff --git a/crates/paimon/tests/historical_table_update_test.rs
b/crates/paimon/tests/historical_table_update_test.rs
new file mode 100644
index 00000000..4a2a92eb
--- /dev/null
+++ b/crates/paimon/tests/historical_table_update_test.rs
@@ -0,0 +1,581 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use arrow_array::{Array, ArrayRef, Int32Array, Int64Array, RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, Datum, IntType, PredicateBuilder, Schema,
TableSchema};
+use paimon::table::{
+ CommitMessage, MergeAssignment, MergeSource, Table, UpdateAssignment,
WhenMatched,
+ WhenNotMatched,
+};
+
+fn batch(names: &[&str], values: &[Vec<i32>]) -> RecordBatch {
+ RecordBatch::try_from_iter(names.iter().zip(values).map(|(name, values)| {
+ (
+ *name,
+ Arc::new(Int32Array::from(values.clone())) as ArrayRef,
+ )
+ }))
+ .unwrap()
+}
+
+async fn evolved_table(path: &str) -> (Table, Table) {
+ let schema = |extra| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("keep", DataType::Int(IntType::new()))
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("deletion-vectors.enabled", "true");
+ if extra {
+ builder = builder.column("extra", DataType::Int(IntType::new()));
+ }
+ TableSchema::new(i64::from(extra), &builder.build().unwrap())
+ };
+ let (io, original) = memory_table(path, schema(false));
+ setup_dirs(&io, path).await;
+ persist_table_schema(&io, path, original.schema()).await;
+ write_batch(
+ &original,
+ &batch(
+ &["id", "value", "keep"],
+ &[vec![1, 2], vec![10, 20], vec![100, 200]],
+ ),
+ )
+ .await;
+ let latest_schema = schema(true);
+ persist_table_schema(&io, path, &latest_schema).await;
+ let latest = original
+ .copy_with_resolved_schema(latest_schema, "main")
+ .unwrap();
+ write_batch(
+ &latest,
+ &batch(
+ &["id", "value", "keep", "extra"],
+ &[vec![3], vec![30], vec![300], vec![999]],
+ ),
+ )
+ .await;
+ let historical = latest
+ .copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(),
"1".into())]))
+ .await
+ .unwrap();
+ assert_eq!(historical.schema().id(), 0);
+ assert!(historical.is_time_traveled());
+ (latest, historical)
+}
+
+async fn commit(table: &Table, messages: Vec<CommitMessage>) {
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+async fn rows(table: &Table) -> Vec<Vec<Option<i64>>> {
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows: Vec<Vec<Option<i64>>> = Vec::new();
+ for batch in batches {
+ for row in 0..batch.num_rows() {
+ rows.push(
+ batch
+ .columns()
+ .iter()
+ .map(|array| {
+ array.is_valid(row).then(|| {
+ if let Some(array) =
array.as_any().downcast_ref::<Int32Array>() {
+ i64::from(array.value(row))
+ } else {
+ array
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .value(row)
+ }
+ })
+ })
+ .collect(),
+ );
+ }
+ }
+ rows.sort_by_key(|row| row[0]);
+ rows
+}
+
+fn changed() -> RecordBatch {
+ RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(vec![11])) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+#[tokio::test]
+async fn historical_updates_keep_schema_and_conflict_baseline() {
+ for operation in [
+ "row_id",
+ "incremental",
+ "groups",
+ "predicate",
+ "shard",
+ "merge",
+ ] {
+ let (latest, historical) =
+
evolved_table(&format!("memory:/historical_update_{operation}")).await;
+ let mut update = historical.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ let messages = match operation {
+ "row_id" => update
+ .update_by_arrow_with_row_id(vec![changed()])
+ .await
+ .unwrap(),
+ "groups" => update
+ .update_by_arrow_batches_with_row_id([Ok(vec![changed()])])
+ .await
+ .unwrap(),
+ "incremental" => update
+ .new_update_by_row_id()
+ .await
+ .unwrap()
+ .update_columns(vec![changed()], vec!["value".into()])
+ .await
+ .unwrap(),
+ "predicate" => update
+ .update_by_predicate(
+ Some(
+ PredicateBuilder::new(historical.schema().fields())
+ .equal("id", Datum::Int(1))
+ .unwrap(),
+ ),
+ vec![(
+ "value".into(),
+
UpdateAssignment::Array(vec![Arc::new(Int32Array::from(vec![11]))]),
+ )],
+ Vec::new(),
+ )
+ .await
+ .unwrap(),
+ "merge" => update
+ .merge_into(
+ MergeSource::Batches(vec![batch(&["id", "value"],
&[vec![1], vec![11]])]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap(),
+ "shard" => {
+ let mut shard = update.new_shard_updater(0, 1,
None).await.unwrap();
+ let inputs: Vec<_> =
shard.arrow_reader().unwrap().try_collect().await.unwrap();
+
assert_eq!(inputs.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+ for input in inputs {
+ let ids = input
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ shard
+ .update_by_arrow_batch(batch(
+ &["value"],
+ &[ids
+ .values()
+ .iter()
+ .map(|id| if *id == 1 { 11 } else { 20 })
+ .collect()],
+ ))
+ .await
+ .unwrap();
+ }
+ shard.prepare_commit().await.unwrap()
+ }
+ _ => unreachable!(),
+ };
+ assert!(!messages.is_empty(), "{operation}");
+ for message in &messages {
+ assert_eq!(message.check_from_snapshot, Some(1), "{operation}");
+ for file in &message.new_files {
+ assert_eq!(file.schema_id, 0);
+ assert_eq!(file.write_cols, Some(vec!["value".into()]));
+ }
+ }
+ commit(&latest, messages).await;
+ let expected = vec![
+ vec![Some(1), Some(11), Some(100), None],
+ vec![Some(2), Some(20), Some(200), None],
+ vec![Some(3), Some(30), Some(300), Some(999)],
+ ];
+ assert_eq!(rows(&latest).await, expected, "{operation}");
+ // Selection and check_from_snapshot must both keep snapshot 1. A
+ // second update of the same old file must not silently overwrite it.
+ let stale = update
+ .update_by_arrow_with_row_id(vec![changed()])
+ .await
+ .unwrap();
+ assert!(stale
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ let error = latest
+ .new_write_builder()
+ .new_commit()
+ .commit(stale)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("conflict"),
+ "{operation}: {error}"
+ );
+ assert_eq!(rows(&latest).await, expected);
+ }
+}
+
+#[tokio::test]
+async fn historical_deletes_preserve_newer_rows() {
+ for operation in ["row_id", "predicate"] {
+ let (latest, historical) =
+
evolved_table(&format!("memory:/historical_delete_{operation}")).await;
+ let update = historical.new_write_builder().new_update().unwrap();
+ let messages = if operation == "row_id" {
+ update.delete_by_row_id(vec![0, 0]).await.unwrap()
+ } else {
+ update
+ .delete_by_predicate(Some(
+ PredicateBuilder::new(historical.schema().fields())
+ .equal("id", Datum::Int(1))
+ .unwrap(),
+ ))
+ .await
+ .unwrap()
+ };
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ commit(&latest, messages).await;
+ assert_eq!(
+ rows(&latest).await,
+ vec![
+ vec![Some(2), Some(20), Some(200), None],
+ vec![Some(3), Some(30), Some(300), Some(999)],
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn historical_updater_cannot_append_or_publish() {
+ let (latest, historical) =
evolved_table("memory:/historical_write_guards").await;
+ assert!(historical.new_write_builder().new_write().is_err());
+ assert!(historical.new_write_builder().try_new_commit().is_err());
+ assert!(historical.new_write_builder().new_delete().is_err());
+ assert!(historical
+ .new_write_builder()
+ .new_data_evolution_writer(vec!["value".into()])
+ .is_err());
+ let update = historical.new_write_builder().new_update().unwrap();
+ let upsert = update
+ .upsert_by_arrow_with_key(
+ vec![batch(
+ &["id", "value", "keep"],
+ &[vec![1], vec![11], vec![100]],
+ )],
+ vec!["id".into()],
+ )
+ .await
+ .unwrap_err();
+ assert!(upsert.to_string().contains("time-travel option"));
+ let merge = update
+ .merge_into(
+ MergeSource::Batches(Vec::new()),
+ Vec::new(),
+ Vec::new(),
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: Vec::new(),
+ }],
+ )
+ .await
+ .unwrap_err();
+ assert!(merge.to_string().contains("time-travel option"));
+ assert_eq!(
+ rows(&latest).await,
+ vec![
+ vec![Some(1), Some(10), Some(100), None],
+ vec![Some(2), Some(20), Some(200), None],
+ vec![Some(3), Some(30), Some(300), Some(999)],
+ ]
+ );
+}
+
+#[tokio::test]
+async fn historical_partial_files_follow_field_ids_after_schema_changes() {
+ for change in ["rename", "drop_readd", "widen"] {
+ let (table, historical) =
+
evolved_table(&format!("memory:/historical_fields_{change}")).await;
+ let mut schema = serde_json::to_value(table.schema()).unwrap();
+ schema["id"] = 2.into();
+ match change {
+ "rename" => schema["fields"][1]["name"] = "renamed".into(),
+ "drop_readd" => {
+ schema["fields"][1]["id"] = 4.into();
+ schema["highestFieldId"] = 4.into();
+ }
+ "widen" => schema["fields"][1]["type"] = "BIGINT".into(),
+ _ => unreachable!(),
+ }
+ let schema: TableSchema = serde_json::from_value(schema).unwrap();
+ persist_table_schema(table.file_io(), table.location(), &schema).await;
+ let latest = table.copy_with_resolved_schema(schema, "main").unwrap();
+ let value = if change == "widen" { 1i64 << 40 } else { 40 };
+ let values: ArrayRef = if change == "widen" {
+ Arc::new(Int64Array::from(vec![value]))
+ } else {
+ Arc::new(Int32Array::from(vec![40]))
+ };
+ let input = RecordBatch::try_new(
+
paimon::arrow::build_target_arrow_schema(latest.schema().fields()).unwrap(),
+ vec![
+ Arc::new(Int32Array::from(vec![4])),
+ values,
+ Arc::new(Int32Array::from(vec![400])),
+ Arc::new(Int32Array::from(vec![1000])),
+ ],
+ )
+ .unwrap();
+ write_batch(&latest, &input).await;
+ let messages = historical
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![changed()])
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == 0 && file.write_cols ==
Some(vec!["value".into()])));
+ commit(&latest, messages).await;
+ let expected = if change == "drop_readd" {
+ [None, None, None]
+ } else {
+ [Some(11), Some(20), Some(30)]
+ };
+ assert_eq!(
+ rows(&latest).await,
+ vec![
+ vec![Some(1), expected[0], Some(100), None],
+ vec![Some(2), expected[1], Some(200), None],
+ vec![Some(3), expected[2], Some(300), Some(999)],
+ vec![Some(4), Some(value), Some(400), Some(1000)],
+ ],
+ "{change}"
+ );
+ }
+}
+
+#[tokio::test]
+async fn historical_deletion_vectors_cannot_restore_a_concurrent_deletion() {
+ let (latest, historical) =
evolved_table("memory:/historical_delete_conflict").await;
+ let newer = latest
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![1])
+ .await
+ .unwrap();
+ commit(&latest, newer).await;
+ let stale = historical
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![0])
+ .await
+ .unwrap();
+ assert!(stale
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ let error = latest
+ .new_write_builder()
+ .new_commit()
+ .commit(stale)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().to_ascii_lowercase().contains("conflict"),
+ "{error}"
+ );
+ assert_eq!(
+ rows(&latest).await,
+ vec![
+ vec![Some(1), Some(10), Some(100), None],
+ vec![Some(3), Some(30), Some(300), Some(999)],
+ ]
+ );
+}
+
+#[tokio::test]
+async fn captured_empty_update_views_do_not_observe_later_rows() {
+ let (latest, _) = evolved_table("memory:/captured_empty_update").await;
+ let empty = latest.copy_with_pinned_snapshot(None);
+ let update = empty.new_write_builder().new_update().unwrap();
+ let messages = update
+ .update_by_predicate(
+ None,
+ vec![(
+ "value".into(),
+ UpdateAssignment::Function(Arc::new(|_| panic!("empty view
evaluated a callback"))),
+ )],
+ vec!["value".into()],
+ )
+ .await
+ .unwrap();
+ assert!(messages.is_empty());
+ assert!(update.delete_by_predicate(None).await.unwrap().is_empty());
+ let mut shard = update.new_shard_updater(0, 1, None).await.unwrap();
+ let batches: Vec<_> =
shard.arrow_reader().unwrap().try_collect().await.unwrap();
+ assert!(batches.is_empty());
+ assert!(shard.prepare_commit().await.unwrap().is_empty());
+ let messages = update
+ .merge_into(
+ MergeSource::Batches(vec![batch(&["id", "value"], &[vec![1],
vec![11]])]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ assert!(messages.is_empty());
+ let error = update
+ .update_by_arrow_with_row_id(vec![changed()])
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().contains("No files with row tracking"),
+ "{error}"
+ );
+ assert_eq!(rows(&latest).await.len(), 3);
+}
+
+#[tokio::test]
+async fn row_id_merge_reads_the_independently_selected_source_view() {
+ for source_view in ["current", "snapshot", "empty"] {
+ let (latest, historical) =
+
evolved_table(&format!("memory:/merge_source_view_{source_view}")).await;
+ let changed = RecordBatch::try_from_iter([
+ ("_ROW_ID", Arc::new(Int64Array::from(vec![0])) as ArrayRef),
+ ("value", Arc::new(Int32Array::from(vec![99])) as ArrayRef),
+ ])
+ .unwrap();
+ commit(
+ &latest,
+ latest
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![changed])
+ .await
+ .unwrap(),
+ )
+ .await;
+ let source = match source_view {
+ "snapshot" => latest
+
.copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(), "3".into())]))
+ .await
+ .unwrap(),
+ "empty" => latest.copy_with_pinned_snapshot(None),
+ _ => latest.clone(),
+ };
+ let assignment = if source_view == "empty" {
+ MergeAssignment::Value(UpdateAssignment::Function(Arc::new(|_| {
+ panic!("an empty MERGE source evaluated an assignment")
+ })))
+ } else {
+ MergeAssignment::SourceColumn("value".into())
+ };
+ let messages = historical
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Table(Arc::new(source)),
+ vec![("_ROW_ID".into(), "_ROW_ID".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![("keep".into(), assignment)],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap();
+ let kept = if source_view == "empty" {
+ assert!(messages.is_empty());
+ [100, 200]
+ } else {
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ commit(&latest, messages).await;
+ [99, 20]
+ };
+ assert_eq!(
+ rows(&latest).await,
+ vec![
+ vec![Some(1), Some(99), Some(kept[0]), None],
+ vec![Some(2), Some(20), Some(kept[1]), None],
+ vec![Some(3), Some(30), Some(300), Some(999)],
+ ],
+ "{source_view}"
+ );
+ }
+}
diff --git a/crates/paimon/tests/table_predicate_delete_test.rs
b/crates/paimon/tests/table_predicate_delete_test.rs
index cc30f344..1dc8e441 100644
--- a/crates/paimon/tests/table_predicate_delete_test.rs
+++ b/crates/paimon/tests/table_predicate_delete_test.rs
@@ -495,7 +495,7 @@ async fn row_delete_rejects_concurrent_dv_replacement() {
}
#[tokio::test]
-async fn
predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indices() {
+async fn historical_predicate_delete_keeps_auth_and_field_index_validation() {
let table = table(&[]).await;
seed(&table).await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -516,7 +516,18 @@ async fn
predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indice
.contains("does not match the table schema"));
let historical =
table.copy_with_options(HashMap::from([("scan.snapshot-id".into(),
"1".into())]));
- assert!(historical.new_write_builder().new_update().is_err());
+ let messages = historical
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_predicate(Some(predicate.clone()))
+ .await
+ .unwrap();
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ assert!(historical.new_write_builder().new_write().is_err());
let auth = table.copy_with_options(HashMap::from([(
"query-auth.enabled".into(),
"true".into(),
@@ -530,6 +541,9 @@ async fn
predicate_delete_rejects_time_travel_auth_and_inconsistent_field_indice
.unwrap_err();
assert!(error.to_string().contains("query-auth.enabled"));
assert_eq!(ids(&table).await, vec![0, 1, 2, 3]);
+ write_batch(&table, &batch(vec![4], vec![Some("a")], vec![50],
vec![1])).await;
+ commit(&table, messages).await;
+ assert_eq!(ids(&table).await, vec![2, 3, 4]);
}
#[tokio::test]
diff --git a/crates/paimon/tests/table_update_test.rs
b/crates/paimon/tests/table_update_test.rs
index 8689dd5e..cd39b34d 100644
--- a/crates/paimon/tests/table_update_test.rs
+++ b/crates/paimon/tests/table_update_test.rs
@@ -1373,16 +1373,34 @@ async fn
empty_update_type_means_all_columns_for_upsert_only() {
}
#[tokio::test]
-async fn update_builder_rejects_overwrite_and_time_travel() {
+async fn
update_builder_rejects_overwrite_but_allows_historical_partial_updates() {
let table = evolution_table().await;
assert!(table
.new_write_builder()
.with_overwrite()
.new_update()
.is_err());
+ seed(&table).await;
let historical =
table.copy_with_options(HashMap::from([("scan.snapshot-id".into(),
"1".into())]));
- assert!(historical.new_write_builder().new_update().is_err());
+ let messages = historical
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![matched(vec![0], &[("value",
vec![11])])])
+ .await
+ .unwrap();
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot == Some(1)));
+ assert!(historical.new_write_builder().new_write().is_err());
+ assert!(historical.new_write_builder().try_new_commit().is_err());
+ commit(&table, messages).await;
+ assert_eq!(
+ read_rows(&table).await,
+ vec![vec![1, 11, 100], vec![2, 20, 200], vec![3, 30, 300]]
+ );
}
#[tokio::test]
