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 91132f56 [core] Complete historical write, update and reader predicate
parity with Java (#1182)
91132f56 is described below
commit 91132f56fb7794380a81bff199f0a9daaabeed13
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 20 15:27:38 2026 +0800
[core] Complete historical write, update and reader predicate parity with
Java (#1182)
---
bindings/python/src/predicate.rs | 85 +++++-
bindings/python/src/write.rs | 272 ++++++++++++++++--
crates/integrations/datafusion/src/sql_context.rs | 14 +-
.../datafusion/tests/time_travel_schema_tests.rs | 85 ++++--
crates/paimon/src/table/branch_write_tests.rs | 28 +-
crates/paimon/src/table/bucket_assigner_cross.rs | 6 +
crates/paimon/src/table/bucket_function.rs | 75 +++++
crates/paimon/src/table/data_evolution_reader.rs | 63 ++--
crates/paimon/src/table/data_file_reader.rs | 2 +-
crates/paimon/src/table/mod.rs | 6 +-
.../src/table/postpone_fixed_bucket_write.rs | 27 +-
crates/paimon/src/table/read_builder.rs | 48 ++--
crates/paimon/src/table/shard_table_updater.rs | 5 +-
.../paimon/src/table/shard_table_updater/tests.rs | 148 ++++++++++
crates/paimon/src/table/table_commit.rs | 64 +++++
crates/paimon/src/table/table_read.rs | 31 +-
crates/paimon/src/table/table_update.rs | 14 +-
crates/paimon/src/table/time_travel.rs | 97 +++++--
crates/paimon/src/table/write_builder.rs | 23 +-
crates/paimon/tests/cross_partition_write_test.rs | 53 ++++
.../paimon/tests/filtered_write_validation_test.rs | 62 ++++
.../paimon/tests/historical_table_update_test.rs | 272 +++++++++++++++---
.../paimon/tests/historical_write_commit_test.rs | 319 +++++++++++++++++++++
crates/paimon/tests/partitioned_blob_read_test.rs | 280 ++++++++++++++++++
crates/paimon/tests/postpone_native_write_test.rs | 43 +++
.../paimon/tests/precomputed_bucket_write_test.rs | 50 ++++
crates/paimon/tests/preplanned_row_id_read_test.rs | 250 ++++++++++++++++
crates/paimon/tests/rest_catalog/merge_sources.rs | 266 +++++++++++++++++
crates/paimon/tests/table_predicate_delete_test.rs | 11 +-
crates/paimon/tests/table_update_test.rs | 12 +-
30 files changed, 2492 insertions(+), 219 deletions(-)
diff --git a/bindings/python/src/predicate.rs b/bindings/python/src/predicate.rs
index cb7726f7..9656dd26 100644
--- a/bindings/python/src/predicate.rs
+++ b/bindings/python/src/predicate.rs
@@ -19,7 +19,7 @@ use crate::error::to_py_err;
use paimon::spec::{
DataField, DataType, Datum, DecimalType, Predicate, PredicateBuilder,
TableSchema,
};
-use pyo3::exceptions::{PyNotImplementedError, PyValueError};
+use pyo3::exceptions::{PyAttributeError, PyNotImplementedError, PyValueError};
use pyo3::prelude::*;
use pyo3::types::PyTzInfoAccess;
use pyo3::types::{
@@ -200,6 +200,22 @@ fn micros_to_millis_nanos(micros: i64) -> (i64, i32) {
)
}
+fn submicrosecond_nanos(value: &Bound<'_, PyAny>) -> PyResult<i32> {
+ // datetime itself has no nanosecond attribute. Arrow's pandas.Timestamp
+ // subclass carries the remaining three digits outside its microseconds.
+ let nanos = match value.getattr("nanosecond") {
+ Ok(nanos) => nanos.extract::<i32>()?,
+ Err(error) if error.is_instance_of::<PyAttributeError>(value.py()) =>
0,
+ Err(error) => return Err(error),
+ };
+ if !(0..1_000).contains(&nanos) {
+ return Err(PyValueError::new_err(
+ "datetime nanosecond remainder must be between 0 and 999",
+ ));
+ }
+ Ok(nanos)
+}
+
/// Convert a naive `datetime.datetime` into `Datum::Timestamp`.
///
/// TIMESTAMP (without time zone) is timezone-less, so timezone-aware datetimes
@@ -219,7 +235,8 @@ fn timestamp_datum(value: &Bound<'_, PyAny>) ->
PyResult<Datum> {
let naive: chrono::NaiveDateTime = value.extract().map_err(|_| {
PyValueError::new_err("expected a datetime.datetime literal for
Timestamp field")
})?;
- let (millis, nanos) =
micros_to_millis_nanos(naive.and_utc().timestamp_micros());
+ let (millis, micros_nanos) =
micros_to_millis_nanos(naive.and_utc().timestamp_micros());
+ let nanos = micros_nanos + submicrosecond_nanos(value)?;
Ok(Datum::Timestamp { millis, nanos })
}
@@ -254,7 +271,8 @@ fn local_zoned_timestamp_datum(value: &Bound<'_, PyAny>) ->
PyResult<Datum> {
let utc = PyTzInfo::utc(py)?;
let as_utc = dt.call_method1(pyo3::intern!(py, "astimezone"), (utc,))?;
let instant: chrono::DateTime<chrono::Utc> = as_utc.extract()?;
- let (millis, nanos) = micros_to_millis_nanos(instant.timestamp_micros());
+ let (millis, micros_nanos) =
micros_to_millis_nanos(instant.timestamp_micros());
+ let nanos = micros_nanos + submicrosecond_nanos(value)?;
Ok(Datum::LocalZonedTimestamp { millis, nanos })
}
@@ -1468,6 +1486,67 @@ mod tests {
});
}
+ #[test]
+ fn timestamp_literals_preserve_nanosecond_tails() {
+ Python::attach(|py| {
+ // Arrow's pandas.Timestamp is a datetime subclass with a
+ // nanosecond remainder. Keep this unit independent of pandas.
+ let module = pyo3::types::PyModule::from_code(
+ py,
+ c"import datetime
+def stamp(ticks, hours):
+ class NanosecondDateTime(datetime.datetime):
+ @property
+ def nanosecond(self):
+ return ticks % 1000
+ base = datetime.datetime(1970, 1, 1) +
datetime.timedelta(microseconds=ticks // 1000)
+ value = NanosecondDateTime(base.year, base.month, base.day, base.hour,
base.minute, base.second, base.microsecond)
+ if hours is not None:
+ value =
value.replace(tzinfo=datetime.timezone.utc).astimezone(datetime.timezone(datetime.timedelta(hours=hours)))
+ return value
+",
+ c"ns_literal.py",
+ c"ns_literal",
+ )
+ .unwrap();
+ for ticks in [
+ i64::MIN + 1,
+ i64::MIN + 807,
+ 1,
+ 2,
+ -1,
+ -1_000_001,
+ 999_999,
+ 1_000_001,
+ i64::MAX,
+ ] {
+ for hours in [None, Some(0), Some(-5), Some(8)] {
+ let value = module
+ .getattr("stamp")
+ .unwrap()
+ .call1((ticks, hours))
+ .unwrap();
+ let millis = ticks.div_euclid(1_000_000);
+ let nanos = ticks.rem_euclid(1_000_000) as i32;
+ let (data_type, expected) = if hours.is_some() {
+ (
+ DataType::LocalZonedTimestamp(
+
paimon::spec::LocalZonedTimestampType::new(9).unwrap(),
+ ),
+ Datum::LocalZonedTimestamp { millis, nanos },
+ )
+ } else {
+ (
+
DataType::Timestamp(paimon::spec::TimestampType::new(9).unwrap()),
+ Datum::Timestamp { millis, nanos },
+ )
+ };
+ assert_eq!(py_to_datum(&value, &data_type).unwrap(),
expected);
+ }
+ }
+ });
+ }
+
#[test]
fn timestamp_field_rejects_aware_datetime() {
Python::attach(|py| {
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 7d850a81..e0dce175 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -327,8 +327,10 @@ impl WriteTarget {
writer.with_blob_consumer(consumer)?;
Ok(())
}
- // Like Java AbstractFileStoreWrite, PK-only writers ignore it.
- Self::PostponeFixed(_) => Ok(()),
+ Self::PostponeFixed(writer) => {
+ writer.with_blob_consumer(consumer)?;
+ Ok(())
+ }
}
}
@@ -609,6 +611,21 @@ impl UpdateContext {
})
}
+ fn merge_source_schema(
+ &self,
+ source: &paimon::table::MergeSource,
+ ) -> PyResult<Arc<ArrowSchema>> {
+ let table = match source {
+ paimon::table::MergeSource::Batches(batches) => return
Ok(batches[0].schema()),
+ paimon::table::MergeSource::Table(table) => table,
+ paimon::table::MergeSource::SelfTable => &self.table,
+ };
+ // Compile conditions against the same metadata schema that core uses
+ // when projecting a table source, including independently pinned
views.
+
paimon::arrow::build_target_arrow_schema(&table.schema().fields_with_read_metadata())
+ .map_err(to_py_err)
+ }
+
fn merge_into(
&self,
py: Python<'_>,
@@ -624,24 +641,7 @@ impl UpdateContext {
},
None => paimon::table::MergeSource::SelfTable,
};
- let source_schema = match &source {
- paimon::table::MergeSource::Batches(batches) =>
batches[0].schema(),
- paimon::table::MergeSource::Table(table) => {
-
paimon::arrow::build_target_arrow_schema(table.schema().fields())
- .map_err(to_py_err)?
- }
- paimon::table::MergeSource::SelfTable => {
- let schema =
paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
- .map_err(to_py_err)?;
- let mut fields = schema.fields().to_vec();
- fields.push(Arc::new(arrow::datatypes::Field::new(
- "_ROW_ID",
- arrow::datatypes::DataType::Int64,
- false,
- )));
- Arc::new(ArrowSchema::new(fields))
- }
- };
+ let source_schema = self.merge_source_schema(&source)?;
let errors = Arc::new(Mutex::new(None));
let (matched, not_matched) = crate::merge_into::from_python(
py,
@@ -1684,9 +1684,239 @@ mod tests {
use super::*;
use paimon::catalog::Identifier;
use paimon::io::FileIOBuilder;
- use paimon::spec::{IntType, Schema, TableSchema, VarCharType};
+ use paimon::spec::{BigIntType, IntType, Schema, TableSchema, VarCharType};
use pyo3::ffi::c_str;
+ #[test]
+ fn merge_table_source_conditions_resolve_read_metadata() {
+ Python::attach(|py| {
+ for metadata in ["_ROW_ID", "_SEQUENCE_NUMBER"] {
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table = Arc::new(Table::new(
+ file_io.clone(),
+ Identifier::new("default", "metadata_merge_target"),
+ "memory:/metadata-merge-target".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let source = Arc::new(Table::new(
+ file_io,
+ Identifier::new("default", "metadata_merge_source"),
+ "memory:/metadata-merge-source".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let source = Py::new(py,
crate::table::PyTable::new(source)).unwrap();
+ let context =
UpdateContext::new(&WriteContext::new(table)).unwrap();
+ let locals = PyDict::new(py);
+ locals.set_item("metadata", metadata).unwrap();
+ py.run(
+ c_str!(
+ r#"
+condition = dict(sql='"s.' + metadata + '" >= 0')
+matched = [dict(assignments=[('value', 'source', 'value')],
condition=condition, delete=False)]
+not_matched = [dict(assignments=[('id', 'source', metadata), ('value',
'source', 'value')],
+ condition=condition)]
+"#
+ ),
+ Some(&locals),
+ Some(&locals),
+ )
+ .unwrap();
+ let messages = context
+ .merge_into(
+ py,
+ Some(source.bind(py).as_any()),
+ vec![("id".into(), metadata.into())],
+ &locals.get_item("matched").unwrap().unwrap(),
+ &locals.get_item("not_matched").unwrap().unwrap(),
+ )
+ .unwrap();
+ // Conditions must compile even when both snapshots are empty;
+ // empty input prepares no mutation files or synthetic rows.
+ assert!(messages.is_empty());
+ }
+ });
+ }
+
+ #[test]
+ fn merge_table_source_metadata_conditions_update_and_insert() {
+ use arrow::array::{Array, Int32Array, Int64Array};
+ use futures::TryStreamExt;
+
+ async fn seed(table: &Table, ids: Vec<i64>, values: Vec<i32>) {
+ let batch = RecordBatch::try_new(
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![
+ Arc::new(Int64Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ writer.close().await;
+ }
+
+ async fn read(table: &Table, projection: &[&str]) -> Vec<RecordBatch> {
+ let mut builder = table.new_read_builder();
+ builder.with_projection(projection).unwrap();
+ let plan = builder.new_scan().plan().await.unwrap();
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap()
+ }
+
+ Python::attach(|py| {
+ for metadata in ["_ROW_ID", "_SEQUENCE_NUMBER"] {
+ let schema = Schema::builder()
+ .column("id", DataType::BigInt(BigIntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let target = Arc::new(Table::new(
+ file_io.clone(),
+ Identifier::new("default", "metadata_data_target"),
+ "memory:/metadata-data-target".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let source = Arc::new(Table::new(
+ file_io,
+ Identifier::new("default", "metadata_data_source"),
+ "memory:/metadata-data-source".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let (first, second) = py.detach(|| {
+ runtime().block_on(async {
+ seed(&source, vec![9], vec![91]).await;
+ seed(&source, vec![8], vec![82]).await;
+ let mut identities = HashMap::new();
+ for batch in read(&source, &[metadata, "value"]).await
{
+ let keys = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(keys.null_count(), 0);
+ for row in 0..batch.num_rows() {
+ identities.insert(values.value(row),
keys.value(row));
+ }
+ }
+ let first = identities[&91];
+ let second = identities[&82];
+ assert_ne!(first, second);
+ seed(&target, vec![first, first + 100], vec![10,
20]).await;
+ (first, second)
+ })
+ });
+ let source = Py::new(py,
crate::table::PyTable::new(source)).unwrap();
+ let context =
UpdateContext::new(&WriteContext::new(target.clone())).unwrap();
+ let locals = PyDict::new(py);
+ locals.set_item("metadata", metadata).unwrap();
+ locals.set_item("first", first).unwrap();
+ locals.set_item("second", second).unwrap();
+ py.run(
+ c_str!(
+ r#"
+matched = [dict(assignments=[('value', 'source', 'value')],
+ condition=dict(sql='"s.' + metadata + '" = ' + str(first)),
delete=False)]
+not_matched = [dict(assignments=[('id', 'source', metadata), ('value',
'source', 'value')],
+ condition=dict(sql='"s.' + metadata + '" = ' +
str(second)))]
+"#
+ ),
+ Some(&locals),
+ Some(&locals),
+ )
+ .unwrap();
+ let messages = context
+ .merge_into(
+ py,
+ Some(source.bind(py).as_any()),
+ vec![("id".into(), metadata.into())],
+ &locals.get_item("matched").unwrap().unwrap(),
+ &locals.get_item("not_matched").unwrap().unwrap(),
+ )
+ .unwrap();
+ assert!(messages
+ .iter()
+ .any(|message| message.inner.check_from_snapshot ==
Some(1)));
+ assert!(messages.iter().any(|message| {
+ message.inner.check_from_snapshot.is_none()
+ && message
+ .inner
+ .new_files
+ .iter()
+ .all(|file| file.first_row_id.is_none())
+ }));
+ let rows = py.detach(|| {
+ runtime().block_on(async {
+ target
+ .new_write_builder()
+ .new_commit()
+ .commit(messages.into_iter().map(|message|
message.inner).collect())
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in read(&target, &["id", "value",
"_ROW_ID"]).await {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let row_ids = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((ids.value(row), values.value(row),
row_ids.value(row)));
+ }
+ }
+ rows.sort_by_key(|row| row.2);
+ rows
+ })
+ });
+ assert_eq!(
+ rows,
+ [(first, 91, 0), (first + 100, 20, 1), (second, 82, 2)]
+ );
+ }
+ });
+ }
+
#[test]
fn update_inputs_preserve_source_nulls_until_core_selects_payloads() {
use arrow::array::{Array, Int32Array};
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index bad1203f..450a0c19 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -1607,15 +1607,13 @@ impl SQLContext {
ok_result(&self.ctx)
}
- /// Reject write statements while a session-level time-travel selector is
- /// active.
+ /// Reject latest-table statement executors while a session-level
+ /// time-travel selector is active.
///
- /// Writes always operate on the latest table state, but in the same
- /// session reads resolve through the time-travelled snapshot schema (and
- /// INSERT through the provider is rejected by the write builder), so
- /// silently ignoring the selector here would be inconsistent. Failing
- /// with a clear message is safer than writing against a different schema
- /// than concurrent reads observe.
+ /// These executors load the latest catalog table, while session reads
+ /// resolve the selected snapshot schema. Silently ignoring the selector
+ /// here would be inconsistent. Ordinary INSERT uses the resolved provider
+ /// and its selected schema, so it does not need this guard.
fn ensure_no_time_travel_for_write(&self, operation: &str) -> DFResult<()>
{
use paimon::spec::{
SCAN_SNAPSHOT_ID_OPTION, SCAN_TAG_NAME_OPTION,
SCAN_TIMESTAMP_MILLIS_OPTION,
diff --git a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
index d5cfed6f..6801a700 100644
--- a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
+++ b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
@@ -225,21 +225,24 @@ async fn test_session_scan_version_uses_snapshot_schema()
{
assert_eq!(column_names(&batches), vec!["id", "name"]);
assert_eq!(total_rows(&batches), 3);
- // Writing through a time-travelled table is rejected.
- let result = sql_context
+ // Ordinary INSERT retains the selected schema, while its commit publishes
+ // to the latest table state. The session's historical reader stays fixed.
+ sql_context
.sql("INSERT INTO paimon.default.t VALUES (6, 'f')")
- .await;
- let err = match result {
- Err(e) => e.to_string(),
- Ok(df) => match df.collect().await {
- Err(e) => e.to_string(),
- Ok(_) => panic!("INSERT into a time-travelled table should fail"),
- },
- };
- assert!(
- err.contains("time-travel option"),
- "error should mention time travel: {err}"
- );
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let historical = sql_context
+ .sql("SELECT * FROM paimon.default.t")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&historical), vec!["id", "name"]);
+ assert_eq!(total_rows(&historical), 3);
// Other write statements are rejected up front instead of silently
// ignoring the active selector and writing to the latest state.
@@ -261,8 +264,8 @@ async fn test_session_scan_version_uses_snapshot_schema() {
);
}
- // A selector resolving to a snapshot with the current schema id pins the
- // read just the same, so INSERT stays rejected.
+ // A selector resolving to the current schema also permits INSERT without
+ // making newly published rows visible through the pinned read.
sql_context
.sql("SET 'paimon.scan.version' = '2'")
.await
@@ -270,20 +273,22 @@ async fn test_session_scan_version_uses_snapshot_schema()
{
.collect()
.await
.unwrap();
- let result = sql_context
- .sql("INSERT INTO paimon.default.t VALUES (6, 'f', 16)")
- .await;
- let err = match result {
- Err(e) => e.to_string(),
- Ok(df) => match df.collect().await {
- Err(e) => e.to_string(),
- Ok(_) => panic!("INSERT should fail while scan.version pins a
same-schema snapshot"),
- },
- };
- assert!(
- err.contains("time-travel option"),
- "error should mention time travel: {err}"
- );
+ sql_context
+ .sql("INSERT INTO paimon.default.t VALUES (7, 'g', 17)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let pinned = sql_context
+ .sql("SELECT * FROM paimon.default.t")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&pinned), vec!["id", "name", "age"]);
+ assert_eq!(total_rows(&pinned), 5);
sql_context
.sql("RESET 'paimon.scan.version'")
@@ -300,6 +305,26 @@ async fn test_session_scan_version_uses_snapshot_schema() {
.await
.unwrap();
assert_eq!(column_names(&batches), vec!["id", "name", "age"]);
+ assert_eq!(total_rows(&batches), 7);
+ let appended = sql_context
+ .sql("SELECT id, age FROM paimon.default.t WHERE id >= 6 ORDER BY id")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let ids = appended[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::Int32Array>()
+ .unwrap();
+ let ages = appended[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<datafusion::arrow::array::Int32Array>()
+ .unwrap();
+ assert_eq!(ids.iter().collect::<Vec<_>>(), vec![Some(6), Some(7)]);
+ assert_eq!(ages.iter().collect::<Vec<_>>(), vec![None, Some(17)]);
}
#[tokio::test]
diff --git a/crates/paimon/src/table/branch_write_tests.rs
b/crates/paimon/src/table/branch_write_tests.rs
index c1538848..0adbcee1 100644
--- a/crates/paimon/src/table/branch_write_tests.rs
+++ b/crates/paimon/src/table/branch_write_tests.rs
@@ -462,25 +462,31 @@ async fn
branch_write_missing_schema_never_creates_a_branch() {
}
#[tokio::test]
-async fn branch_write_historical_handles_remain_read_only() {
+async fn branch_write_historical_handles_retain_their_read_view() {
let (_dir, main) = table(false, &[]).await;
append(&main, "seed", batch(vec![1], vec![10])).await;
let branch = tagged_branch(&main).await;
let snapshot = latest(&branch).await;
let mut historical = branch.copy_with_pinned_snapshot(Some(&snapshot));
- // The pinned snapshot remains read-only even with no selector in its
options.
+ // A captured view is independent of the writer's publication state, even
+ // when the corresponding selector is absent from its options.
historical.schema = historical
.schema
.copy_with_replaced_options(branch.schema().options().clone());
- 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_commit()
- .commit(vec![])
- .await
- .is_err());
- assert_eq!(latest(&branch).await, snapshot);
+ append(&historical, "historical", batch(vec![2], vec![20])).await;
+ assert_eq!(latest(&branch).await.id(), snapshot.id() + 1);
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&historical).await,
+ vec![(1, 10)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&branch).await,
+ vec![(1, 10), (2, 20)]
+ );
+ assert_eq!(
+ table_write::tests::read_id_value_rows(&main).await,
+ vec![(1, 10)]
+ );
}
#[tokio::test]
diff --git a/crates/paimon/src/table/bucket_assigner_cross.rs
b/crates/paimon/src/table/bucket_assigner_cross.rs
index 5abb9c33..b3324e49 100644
--- a/crates/paimon/src/table/bucket_assigner_cross.rs
+++ b/crates/paimon/src/table/bucket_assigner_cross.rs
@@ -81,6 +81,12 @@ impl GlobalPartitionIndex {
// The cross-partition index reads every primary key.
crate::spec::CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
+ // Read selectors describe the caller's query view, not the state that
+ // the next write extends. Java IndexBootstrap forces a latest scan
too.
+ // Pin that state while retaining this writer's schema and field IDs.
+ let snapshot = table.snapshot_manager().get_latest_snapshot().await?;
+ let table = table.copy_with_pinned_snapshot(snapshot.as_ref());
+
let mut key_to_location: HashMap<Vec<u8>, (Vec<u8>, i32)> =
HashMap::new();
let mut partition_bucket_counts: HashMap<Vec<u8>, BTreeMap<i32, i64>>
= HashMap::new();
diff --git a/crates/paimon/src/table/bucket_function.rs
b/crates/paimon/src/table/bucket_function.rs
index b21082f5..599032bd 100644
--- a/crates/paimon/src/table/bucket_function.rs
+++ b/crates/paimon/src/table/bucket_function.rs
@@ -315,6 +315,81 @@ mod tests {
);
}
+ #[test]
+ fn mod_bucket_preserves_integer_boundaries_and_null_slots() {
+ let int_type = DataType::Int(IntType::new());
+ let long_type = DataType::BigInt(crate::spec::BigIntType::new());
+ for (value, expected) in [(i32::MIN, 2), (-17, 3), (-1, 4), (0, 0),
(i32::MAX, 2)] {
+ assert_eq!(
+ bucket_for_datums(
+ &[(Some(&Datum::Int(value)), &int_type)],
+ BucketFunctionType::Mod,
+ 5,
+ )
+ .unwrap(),
+ expected
+ );
+ }
+ for (value, expected) in [(i64::MIN, 2), (-17, 3), (-1, 4), (0, 0),
(i64::MAX, 2)] {
+ assert_eq!(
+ bucket_for_datums(
+ &[(Some(&Datum::Long(value)), &long_type)],
+ BucketFunctionType::Mod,
+ 5,
+ )
+ .unwrap(),
+ expected
+ );
+ }
+ for data_type in [&int_type, &long_type] {
+ assert_eq!(
+ bucket_for_datums(&[(None, data_type)],
BucketFunctionType::Mod, 5).unwrap(),
+ 0
+ );
+ }
+ }
+
+ #[test]
+ fn timestamp_nanoseconds_use_java_default_bucket_ids() {
+ for (ticks, expected) in [
+ (i64::MIN + 1, 4),
+ (i64::MIN + 807, 0),
+ (-1_000_001_i64, 3),
+ (-1, 0),
+ (0, 4),
+ (1, 2),
+ (2, 3),
+ (999_999, 2),
+ (1_000_001, 0),
+ (i64::MAX, 4),
+ ] {
+ let millis = ticks.div_euclid(1_000_000);
+ let nanos = ticks.rem_euclid(1_000_000) as i32;
+ for (data_type, datum) in [
+ (
+
DataType::Timestamp(crate::spec::TimestampType::new(9).unwrap()),
+ Datum::Timestamp { millis, nanos },
+ ),
+ (
+ DataType::LocalZonedTimestamp(
+ crate::spec::LocalZonedTimestampType::new(9).unwrap(),
+ ),
+ Datum::LocalZonedTimestamp { millis, nanos },
+ ),
+ ] {
+ assert_eq!(
+ bucket_for_datums(
+ &[(Some(&datum), &data_type)],
+ BucketFunctionType::Default,
+ 5
+ )
+ .unwrap(),
+ expected
+ );
+ }
+ }
+ }
+
#[test]
fn hive_bucket_matches_java_reference_case() {
let bool_type = DataType::Boolean(BooleanType::new());
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index da0f950c..6b0bff0c 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -335,6 +335,7 @@ impl DataEvolutionReader {
false,
None,
)?
+ .with_partition_fields(self.partition_fields.clone())
.with_batch_size(self.batch_size)
.with_limit(Some(remaining))
.with_parquet_read_budget(self.parquet_read_budget.clone())
@@ -1135,10 +1136,13 @@ impl DataEvolutionReader {
&prepared_group,
&file_infos,
&read_type,
- &table_fields,
&blob_inline_fields,
- row_ranges.is_some(),
- nested_enabled,
+ SourcePlanReadContext {
+ table_fields: &table_fields,
+ partition_fields: &partition_fields,
+ row_id_pushdown: row_ranges.is_some(),
+ nested_enabled,
+ },
)?;
let active_source_indices: Vec<usize> = source_plan
@@ -2506,6 +2510,16 @@ struct SourcePlan {
nested_plan: Vec<Option<super::data_evolution_nested::NestedFieldPlan>>,
}
+/// Logical inputs needed while selecting physical column providers.
+/// Partition columns may have no provider: the merge restores them from the
+/// split, including when the declared partition type is NOT NULL.
+struct SourcePlanReadContext<'a> {
+ table_fields: &'a [DataField],
+ partition_fields: &'a [DataField],
+ row_id_pushdown: bool,
+ nested_enabled: bool,
+}
+
#[cfg(test)]
fn build_source_plan(
prepared_group: &PreparedMergeGroup,
@@ -2517,10 +2531,13 @@ fn build_source_plan(
prepared_group,
file_infos,
read_type,
- read_type,
blob_descriptor_fields,
- false,
- false,
+ SourcePlanReadContext {
+ table_fields: read_type,
+ partition_fields: &[],
+ row_id_pushdown: false,
+ nested_enabled: false,
+ },
)
}
@@ -2528,11 +2545,15 @@ fn build_source_plan_with_row_id_pushdown(
prepared_group: &PreparedMergeGroup,
file_infos: &[ResolvedFileInfo],
read_type: &[DataField],
- table_fields: &[DataField],
blob_descriptor_fields: &HashSet<String>,
- row_id_pushdown: bool,
- nested_enabled: bool,
+ context: SourcePlanReadContext<'_>,
) -> crate::Result<SourcePlan> {
+ let SourcePlanReadContext {
+ table_fields,
+ partition_fields,
+ row_id_pushdown,
+ nested_enabled,
+ } = context;
let mut sources = Vec::new();
let mut normal_providers: HashMap<i32, usize> = HashMap::new(); //
field_id -> source_idx
let mut vector_field_providers: HashMap<i32, usize> = HashMap::new(); //
field_id -> source_idx
@@ -2749,7 +2770,11 @@ fn build_source_plan_with_row_id_pushdown(
let field_offset =
sources[source_idx].add_read_field(source_field.clone());
column_plan.push(Some((source_idx, field_offset)));
nested_plan.push(None);
- } else if !field.data_type().is_nullable() {
+ } else if !field.data_type().is_nullable()
+ && !partition_fields
+ .iter()
+ .any(|partition| partition.id() == field.id())
+ {
return Err(Error::DataInvalid {
message: format!(
"Cannot read non-nullable field '{}' without a provider",
@@ -4571,10 +4596,13 @@ mod tests {
&prepared_group,
&file_infos,
&read_type,
- &read_type,
&HashSet::new(),
- true,
- false,
+ SourcePlanReadContext {
+ table_fields: &read_type,
+ partition_fields: &[],
+ row_id_pushdown: true,
+ nested_enabled: false,
+ },
)
.unwrap_err();
@@ -4660,10 +4688,13 @@ mod tests {
&group,
&[resolved_info(vec![1]), resolved_info(vec![2])],
&[selected],
- std::slice::from_ref(&map),
&HashSet::new(),
- false,
- false,
+ SourcePlanReadContext {
+ table_fields: std::slice::from_ref(&map),
+ partition_fields: &[],
+ row_id_pushdown: false,
+ nested_enabled: false,
+ },
)
.unwrap();
assert_eq!(plan.column_plan, vec![Some((1, 0))]);
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 4ea4c9bd..7c47cfcd 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -626,7 +626,7 @@ impl DataFileReader {
|| (row_tracking_enabled
&& refs
.iter()
- .any(|(name, _)| *name ==
crate::spec::SEQUENCE_NUMBER_FIELD_NAME));
+ .any(|(name, _)|
crate::spec::is_row_tracking_column(name)));
if post_read_filter {
// Read predicate-only columns without decoder filtering. Reconcile
// old-file types and attach original row IDs before exact
filtering.
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 80c7e396..e811ca07 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -301,9 +301,9 @@ pub struct Table {
/// replay a grant.
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`]. Ordinary writes and publication
reject
- /// this copy; row-level updates may use it as their read snapshot.
+ /// True for a time-traveled or captured read view. Scans and row matching
+ /// use its selected snapshot. Writers retain its schema; commits publish
+ /// against latest state.
time_traveled: bool,
/// Captured snapshot or empty view, so scans don't resolve latest again.
/// Cleared when [`Table::copy_with_options`] changes the selector.
diff --git a/crates/paimon/src/table/postpone_fixed_bucket_write.rs
b/crates/paimon/src/table/postpone_fixed_bucket_write.rs
index 4f55a352..a46730c0 100644
--- a/crates/paimon/src/table/postpone_fixed_bucket_write.rs
+++ b/crates/paimon/src/table/postpone_fixed_bucket_write.rs
@@ -39,18 +39,35 @@ pub struct PostponeFixedBucketTableWrite {
}
impl PostponeFixedBucketTableWrite {
+ /// Configure the Blob consumer before accepting input, including input
+ /// buffered for automatic bucket planning. Primary-key writers do not call
+ /// this consumer, just as for ordinary TableWrite.
+ pub fn with_blob_consumer(
+ &mut self,
+ consumer: Option<std::sync::Arc<dyn crate::spec::BlobConsumer>>,
+ ) -> Result<&mut Self> {
+ self.ensure_configurable("with_blob_consumer")?;
+ self.inner.with_blob_consumer(consumer)?;
+ Ok(self)
+ }
+
pub fn with_blob_uri_reader_factory(
&mut self,
factory: Option<std::sync::Arc<dyn crate::io::UriReaderFactory>>,
) -> Result<&mut Self> {
+ self.ensure_configurable("with_blob_uri_reader_factory")?;
+ self.inner.with_blob_uri_reader_factory(factory)?;
+ Ok(self)
+ }
+
+ fn ensure_configurable(&self, method: &str) -> Result<()> {
self.ensure_writable()?;
if self.check_from_snapshot.is_some() {
- return Err(data_invalid(
- "with_blob_uri_reader_factory must be called before any write
operation",
- ));
+ return Err(data_invalid(format!(
+ "{method} must be called before any write operation"
+ )));
}
- self.inner.with_blob_uri_reader_factory(factory)?;
- Ok(self)
+ Ok(())
}
pub(crate) fn new(
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index 76a0b538..275203cf 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -535,11 +535,6 @@ impl<'a> PaimonReadBuilder<'a> {
let combined = Predicate::and(self.filter.data_predicates.clone());
if let Some(ranges) =
super::row_id_predicate::extract_row_id_ranges(&combined) {
self.derived_row_ranges = Some(ranges);
- // Ranges are a superset, so a conjunct leaves the residual only
when
- // they represent it exactly.
- self.filter
- .data_predicates
- .retain(|conjunct|
!super::row_id_predicate::ranges_represent_conjunct(conjunct));
}
}
@@ -652,7 +647,18 @@ impl<'a> PaimonReadBuilder<'a> {
/// ordinary partition conjuncts for safe stats/LIMIT decisions, rebound
from
/// the partition row to the table layout (partition order may differ).
fn scan_data_predicates(&self) -> Vec<Predicate> {
- let mut predicates = self.filter.data_predicates.clone();
+ // Range extraction replaces exact scan conjuncts only. A reader can
+ // consume splits from another plan, so retain its complete predicate.
+ let mut predicates = self
+ .filter
+ .data_predicates
+ .iter()
+ .filter(|conjunct| {
+ self.derived_row_ranges.is_none()
+ ||
!super::row_id_predicate::ranges_represent_conjunct(conjunct)
+ })
+ .cloned()
+ .collect::<Vec<_>>();
if self.partition_filter.is_none() {
return predicates;
}
@@ -1307,7 +1313,7 @@ mod tests {
Some(vec![RowRange::new(1, 1)])
);
assert_eq!(
- builder.filter.data_predicates,
+ builder.scan_data_predicates(),
vec![row_id_leaf(PredicateOperator::NotEq, 1)]
);
@@ -1317,7 +1323,8 @@ mod tests {
exact.effective_row_ranges(),
Some(vec![RowRange::new(10, i64::MAX)])
);
- assert!(exact.filter.data_predicates.is_empty());
+ assert!(exact.scan_data_predicates().is_empty());
+ assert_eq!(exact.new_read().unwrap().data_predicates().len(), 1);
}
#[test]
@@ -1342,7 +1349,7 @@ mod tests {
builder.effective_row_ranges(),
Some(vec![RowRange::new(10, i64::MAX)])
);
- assert_eq!(builder.filter.data_predicates, vec![disjunction]);
+ assert_eq!(builder.scan_data_predicates(), vec![disjunction]);
}
#[test]
@@ -1363,7 +1370,8 @@ mod tests {
builder.effective_row_ranges(),
Some(vec![RowRange::new(7, 7)])
);
- assert!(builder.filter.data_predicates.is_empty());
+ assert!(builder.scan_data_predicates().is_empty());
+ assert_eq!(builder.new_read().unwrap().data_predicates().len(), 1);
}
#[test]
@@ -1413,7 +1421,7 @@ mod tests {
]));
assert_eq!(builder.effective_row_ranges(), Some(Vec::new()));
- assert_eq!(builder.filter.data_predicates, vec![unsatisfiable]);
+ assert_eq!(builder.scan_data_predicates(), vec![unsatisfiable]);
}
#[test]
@@ -1442,7 +1450,8 @@ mod tests {
Some(vec![RowRange::new(102, i64::MAX)]),
"row_tracking={row_tracking} data_evolution={data_evolution}"
);
- assert!(builder.filter.data_predicates.is_empty());
+ assert!(builder.scan_data_predicates().is_empty());
+ assert_eq!(builder.new_read().unwrap().data_predicates().len(), 1);
}
}
@@ -1456,7 +1465,7 @@ mod tests {
row_id_leaf(PredicateOperator::Eq, 8),
]));
assert!(
- builder.filter.data_predicates.is_empty(),
+ builder.scan_data_predicates().is_empty(),
"handed to ranges"
);
@@ -2289,9 +2298,16 @@ mod tests {
assert_eq!(ranges[0].from(), 10);
assert_eq!(ranges[0].to(), 20);
- // _ROW_ID predicates should be removed from data_predicates
- assert!(!inner.filter.data_predicates.is_empty());
- for p in &inner.filter.data_predicates {
+ // Only scan predicates omit exact ranges. Independent readers retain
+ // the full filter even when their supplied splits were not filtered.
+ assert!(inner
+ .filter
+ .data_predicates
+ .iter()
+ .any(super::super::row_id_predicate::references_row_id));
+ let scan_predicates = inner.scan_data_predicates();
+ assert!(!scan_predicates.is_empty());
+ for p in &scan_predicates {
if let Predicate::Leaf { column, .. } = p {
assert_ne!(column, crate::spec::ROW_ID_FIELD_NAME);
}
diff --git a/crates/paimon/src/table/shard_table_updater.rs
b/crates/paimon/src/table/shard_table_updater.rs
index 3c03b624..3872af02 100644
--- a/crates/paimon/src/table/shard_table_updater.rs
+++ b/crates/paimon/src/table/shard_table_updater.rs
@@ -91,7 +91,7 @@ impl ShardTableUpdater {
if write_columns.is_empty() {
return Err(invalid("Shard update columns must not be empty"));
}
- let writer = DataEvolutionWriter::new(table, write_columns.clone())?;
+ let writer = DataEvolutionWriter::for_row_id(table,
write_columns.clone())?;
let snapshot = super::time_travel::resolve_snapshot(table).await?;
let table = table
.copy_with_options(HashMap::from([(
@@ -282,7 +282,8 @@ impl ShardTableUpdater {
));
}
if self.writer.is_none() {
- let mut writer = DataEvolutionWriter::new(&self.table,
self.write_columns.clone())?;
+ let mut writer =
+ DataEvolutionWriter::for_row_id(&self.table,
self.write_columns.clone())?;
writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
self.writer = Some(writer);
}
diff --git a/crates/paimon/src/table/shard_table_updater/tests.rs
b/crates/paimon/src/table/shard_table_updater/tests.rs
index 4f2c2ffe..4126dcc0 100644
--- a/crates/paimon/src/table/shard_table_updater/tests.rs
+++ b/crates/paimon/src/table/shard_table_updater/tests.rs
@@ -233,6 +233,154 @@ async fn
partitioned_shard_accepts_rebatched_values_and_pins_snapshot() {
);
}
+#[tokio::test]
+async fn partitioned_shard_accepts_unchanged_partition_carriers() {
+ use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+
+ for partition_only in [false, true] {
+ let table = table(true);
+ seed(&table, batch(vec!["a", "b"], vec![1, 2], vec![10, 20])).await;
+ let mut configured = table.new_write_builder().new_update().unwrap();
+ if partition_only {
+ configured.with_update_type(vec!["p".into()]).unwrap();
+ }
+ let mut update = configured.new_shard_updater(0, 1,
None).await.unwrap();
+ let inputs: Vec<_> =
update.arrow_reader().unwrap().try_collect().await.unwrap();
+ assert_eq!(inputs.iter().map(RecordBatch::num_rows).sum::<usize>(), 2);
+ // Rebatch across partitions so both the first writer and each writer
+ // created at a file-group boundary use the same carrier validation.
+ let input = arrow_select::concat::concat_batches(&inputs[0].schema(),
&inputs).unwrap();
+ let output = if partition_only {
+ input.project(&[0]).unwrap()
+ } else {
+ let ids = input
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ RecordBatch::try_from_iter([
+ ("p", input.column_by_name("p").unwrap().clone()),
+ ("id", input.column_by_name("id").unwrap().clone()),
+ (
+ "value",
+ Arc::new(Int32Array::from_iter_values(
+ ids.values().iter().map(|id| id * 100),
+ )) as ArrayRef,
+ ),
+ ])
+ .unwrap()
+ };
+ update.update_by_arrow_batch(output).await.unwrap();
+ let messages = update.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 2);
+ let files = table
+ .file_io()
+ .list_status_recursive(table.location())
+ .await
+ .unwrap();
+ let mut ranges = Vec::new();
+ for message in &messages {
+ assert_eq!(message.check_from_snapshot, Some(1));
+ for file in &message.new_files {
+ assert_eq!(file.schema_id, 0);
+ assert_eq!(file.row_count, 1);
+ ranges.push(file.first_row_id.unwrap());
+ assert_eq!(file.write_cols, partition_only.then(||
vec!["p".into()]));
+ let path = &files
+ .iter()
+ .find(|status| status.path.ends_with(&file.file_name))
+ .unwrap()
+ .path;
+ let bytes = table
+ .file_io()
+ .new_input(path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let reader =
ParquetRecordBatchReaderBuilder::try_new(bytes).unwrap();
+ let schema = reader.schema();
+ let names = schema
+ .fields()
+ .iter()
+ .map(|field| field.name().as_str())
+ .collect::<Vec<_>>();
+ assert_eq!(
+ names,
+ if partition_only {
+ vec!["p"]
+ } else {
+ vec!["p", "id", "value"]
+ }
+ );
+ }
+ }
+ ranges.sort_unstable();
+ assert_eq!(ranges, vec![0, 1]);
+ commit(&table, messages).await;
+ assert_eq!(
+ rows(&table).await,
+ if partition_only {
+ vec![(1, 10), (2, 20)]
+ } else {
+ vec![(1, 100), (2, 200)]
+ }
+ );
+ }
+}
+
+#[tokio::test]
+async fn partitioned_shard_rejects_moving_rows_to_another_partition() {
+ let table = table(true);
+ seed(&table, batch(vec!["a", "b"], vec![1, 2], vec![10, 20])).await;
+ let mut update = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .new_shard_updater(0, 1, None)
+ .await
+ .unwrap();
+ let inputs: Vec<_> =
update.arrow_reader().unwrap().try_collect().await.unwrap();
+ let result: crate::Result<()> = async {
+ for input in inputs {
+ let ids = input
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ update
+ .update_by_arrow_batch(batch(
+ vec!["changed"; input.num_rows()],
+ ids.values().to_vec(),
+ vec![99; input.num_rows()],
+ ))
+ .await?;
+ }
+ update.prepare_commit().await?;
+ Ok(())
+ }
+ .await;
+ let error = result.unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains("Cannot change partition column 'p'"),
+ "{error}"
+ );
+ assert!(update.prepare_commit().await.is_err());
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(1)
+ );
+ assert_eq!(rows(&table).await, vec![(1, 10), (2, 20)]);
+}
+
#[tokio::test]
async fn deleted_rows_do_not_shift_positional_assignments() {
let table = table(false);
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 20846ea6..ce9d8804 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -5328,6 +5328,70 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_explicit_index_deletes_only_exempt_the_deleted_indexes() {
+ for action in ["THROW_ERROR", "DROP_PARTITION_INDEX", "IGNORE"] {
+ for deleted in ["affected", "other", "both"] {
+ let file_io = test_file_io();
+ let path =
format!("memory:/explicit_index_delete_{action}_{deleted}");
+ let table = test_table_with_options(
+ &file_io,
+ &path,
+
HashMap::from([("global-index.column-update-action".into(), action.into())]),
+ );
+ let commit = TableCommit::new(table, "test-user".into());
+ let previous: Vec<_> = [("affected.index", 0), ("other.index",
1)]
+ .into_iter()
+ .map(|(name, id)| IndexManifestEntry {
+ kind: FileKind::Add,
+ partition: vec![],
+ bucket: 0,
+ index_file: test_global_index_file(name, id, 0, 9),
+ version: 1,
+ })
+ .collect();
+ let deletions: Vec<_> = previous
+ .iter()
+ .filter(|entry| {
+ deleted == "both"
+ || entry.index_file.file_name ==
format!("{deleted}.index")
+ })
+ .cloned()
+ .map(|mut entry| {
+ entry.kind = FileKind::Delete;
+ entry
+ })
+ .collect();
+ let mut update = test_data_file("update.parquet", 10);
+ update.first_row_id = Some(0);
+ update.write_cols = Some(vec!["id".into()]);
+ let entries =
+ commit.messages_to_entries(&[CommitMessage::new(vec![], 0,
vec![update])]);
+ let result = commit
+ .global_index_update_entries(&previous, &entries,
&deletions)
+ .await;
+ if action == "THROW_ERROR" && deleted == "other" {
+ assert!(
+ result
+ .unwrap_err()
+ .to_string()
+ .contains("globally indexed columns"),
+ "{action}/{deleted}"
+ );
+ } else {
+ let automatic = result.unwrap();
+ if action == "DROP_PARTITION_INDEX" && deleted == "other" {
+ assert_eq!(automatic.len(), 1);
+ assert_eq!(automatic[0].kind, FileKind::Delete);
+ assert_eq!(automatic[0].index_file,
previous[0].index_file);
+ } else {
+ assert!(automatic.is_empty(), "{action}/{deleted}");
+ }
+ }
+ }
+ }
+ }
+
#[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_read.rs
b/crates/paimon/src/table/table_read.rs
index 7fe76656..39efafd6 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -3611,7 +3611,12 @@ mod tests {
#[tokio::test]
async fn test_a_row_id_filter_configured_on_the_read_is_refused() {
- let table = crate::table::rest_query_auth_table().await;
+ let table = crate::table::rest_query_auth_table()
+ .await
+ .copy_with_options(std::collections::HashMap::from([(
+ "row-tracking.enabled".into(),
+ "true".into(),
+ )]));
let row_id = Predicate::Leaf {
index: 0,
column: crate::spec::ROW_ID_FIELD_NAME.to_string(),
@@ -3619,17 +3624,21 @@ mod tests {
op: crate::spec::PredicateOperator::GtEq,
literals: vec![crate::spec::Datum::Long(1)],
};
+ let mut builder = table.new_read_builder();
+ builder.with_filter(row_id.clone());
let read = TableRead::new(&table, table.schema.fields().to_vec(),
vec![row_id]);
- let split = split_with_grant(Some(grant_for(&table, false)));
- let err = match read.to_arrow(&[split]) {
- Ok(_) => panic!("a system-column filter must be refused"),
- Err(err) => err,
- };
- assert!(
- matches!(err, crate::Error::Unsupported { ref message }
- if message.contains("system column '_ROW_ID'")),
- "got {err:?}"
- );
+ for read in [read, builder.new_read().unwrap()] {
+ let split = split_with_grant(Some(grant_for(&table, false)));
+ let err = match read.to_arrow(&[split]) {
+ Ok(_) => panic!("a system-column filter must be refused"),
+ Err(err) => err,
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("system column '_ROW_ID'")),
+ "got {err:?}"
+ );
+ }
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index de07c4d3..1eac4063 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -46,10 +46,11 @@ 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.
+/// Operations may select a historical read view. Matched updates retain its
+/// snapshot as the conflict baseline; upsert and MERGE append unmatched rows
+/// using that view's schema and assign fresh row IDs at publication.
Publication
+/// resolves the latest snapshot and current table schema independently of the
+/// selected read view.
#[derive(Clone)]
pub struct TableUpdate {
table: Table,
@@ -260,8 +261,6 @@ 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(),
@@ -287,9 +286,6 @@ 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,
diff --git a/crates/paimon/src/table/time_travel.rs
b/crates/paimon/src/table/time_travel.rs
index 245a9595..a7e2216d 100644
--- a/crates/paimon/src/table/time_travel.rs
+++ b/crates/paimon/src/table/time_travel.rs
@@ -167,7 +167,7 @@ mod tests {
use crate::catalog::Identifier;
use crate::io::{FileIO, FileIOBuilder};
use crate::spec::{CommitKind, DataType, IntType, Schema, Snapshot,
TableSchema};
- use crate::table::{SnapshotManager, Table, TableCommit, TableWrite,
TagManager};
+ use crate::table::{SnapshotManager, Table, TagManager};
use arrow_array::{Int32Array, RecordBatch};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use std::collections::HashMap;
@@ -250,11 +250,49 @@ mod tests {
}
async fn write_and_commit(table: &Table, batch: &RecordBatch) {
- let mut write = TableWrite::new(table,
"test-user".to_string()).unwrap();
+ let builder = table
+ .new_write_builder()
+ .with_commit_user("test-user")
+ .unwrap();
+ let mut write = builder.new_write().unwrap();
write.write_arrow_batch(batch).await.unwrap();
let messages = write.prepare_commit().await.unwrap();
- let commit = TableCommit::new(table.clone(), "test-user".to_string());
- commit.commit(messages).await.unwrap();
+ builder
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
+ .await
+ .unwrap();
+ }
+
+ async fn read_ids(table: &Table) -> Vec<i32> {
+ use futures::TryStreamExt;
+
+ let builder = 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 ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ ids
}
/// Table with two schema versions and one snapshot per version:
@@ -402,12 +440,6 @@ mod tests {
.unwrap();
assert_eq!(traveled.schema().id(), 0);
assert_eq!(traveled.travel_snapshot().unwrap().id(), 1);
- assert!(traveled.new_write_builder().new_write().is_err());
- assert!(table
- .copy_with_options(opts)
- .new_write_builder()
- .new_write()
- .is_err());
let builder = traveled.new_read_builder();
let plan = builder.new_scan().plan().await.unwrap();
let batches: Vec<RecordBatch> = builder
@@ -519,7 +551,6 @@ mod tests {
super::resolve_snapshot(&pinned).await.unwrap().unwrap(),
snapshot
);
- assert!(pinned.new_write_builder().new_write().is_err());
assert!(table.travel_snapshot().is_none());
}
@@ -589,7 +620,6 @@ mod tests {
.unwrap()
.splits()
.is_empty());
- assert!(view.new_write_builder().new_write().is_err());
assert!(view.is_time_traveled());
assert!(!view.has_resolved_travel_snapshot());
assert!(crate::catalog::list_partitions_from_file_system(&view)
@@ -631,13 +661,12 @@ mod tests {
}
#[tokio::test]
- async fn test_copy_with_time_travel_same_schema_still_rejects_write() {
+ async fn test_copy_with_time_travel_same_schema_can_append_and_commit() {
let (file_io, table_path) = setup_evolved_table().await;
let table = latest_table(&file_io, &table_path);
// Snapshot 2 carries the current schema, so the schema is not
- // switched — but the copy still reads a pinned snapshot, so writing
- // through it is rejected like any other time-travelled copy.
+ // switched. Its read snapshot does not limit append publication.
let traveled = table
.copy_with_time_travel(options(&[("scan.version", "2")]))
.await
@@ -645,7 +674,19 @@ mod tests {
assert_eq!(traveled.schema().id(), 1);
assert!(!traveled.is_time_traveled());
- assert!(traveled.new_write_builder().new_write().is_err());
+ write_and_commit(&traveled, &batch_v1(vec![6], vec![60],
vec![16])).await;
+ assert_eq!(read_ids(&table).await, vec![1, 2, 3, 4, 5, 6]);
+ assert_eq!(read_ids(&traveled).await, vec![1, 2, 3, 4, 5]);
+ assert_eq!(
+ table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .schema_id(),
+ 1
+ );
}
#[tokio::test]
@@ -712,7 +753,7 @@ mod tests {
}
#[tokio::test]
- async fn test_time_traveled_table_rejects_write() {
+ async fn
test_time_traveled_table_appends_with_its_schema_and_publishes_latest() {
let (file_io, table_path) = setup_evolved_table().await;
let table = latest_table(&file_io, &table_path);
@@ -721,17 +762,17 @@ mod tests {
.await
.unwrap();
- let err = match traveled.new_write_builder().new_write() {
- Err(e) => e,
- Ok(_) => panic!("expected write rejection on time-travelled
table"),
- };
- assert!(
- matches!(err, crate::Error::Unsupported { ref message }
- if message.contains("time-travel option")),
- "expected write rejection on time-travelled table, got {err:?}"
- );
- // The latest table is unaffected.
- assert!(table.new_write_builder().new_write().is_ok());
+ write_and_commit(&traveled, &batch_v0(vec![6], vec![60])).await;
+ let latest_snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(latest_snapshot.id(), 3);
+ assert_eq!(latest_snapshot.schema_id(), 1);
+ assert_eq!(read_ids(&table).await, vec![1, 2, 3, 4, 5, 6]);
+ assert_eq!(read_ids(&traveled).await, vec![1, 2, 3]);
}
#[tokio::test]
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 2c3744ed..ae90dc0c 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -264,8 +264,9 @@ impl<'a> PaimonWriteBuilder<'a> {
pub fn new_update(&self) -> crate::Result<TableUpdate> {
// 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.
+ // for conflict detection. Upsert and MERGE append unmatched rows using
+ // the same read schema. As with ordinary writers, publication uses the
+ // latest state of the selected table and branch.
ensure_paimon_table(self.table)?;
if self.overwrite {
return Err(crate::Error::Unsupported {
@@ -292,6 +293,7 @@ impl<'a> PaimonWriteBuilder<'a> {
}
pub(super) fn ensure_table_write_authorized(table: &Table) ->
crate::Result<()> {
+ ensure_table_write_allowed(table)?;
let options = crate::spec::CoreOptions::new(table.schema().options());
// Query authorization protects reads, not INSERT/OVERWRITE/TRUNCATE.
// A catalog-loaded REST table retains its publication identity and uses
@@ -304,22 +306,13 @@ pub(super) fn ensure_table_write_authorized(table:
&Table) -> crate::Result<()>
message: "Writing a table with 'query-auth.enabled' = true
requires a catalog-loaded REST table with its resolved schema".into(),
});
}
- ensure_table_write_allowed(table)
+ Ok(())
}
pub(super) fn ensure_table_write_allowed(table: &Table) -> crate::Result<()> {
- 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();
- if table.is_time_traveled() || !matches!(selector, Ok(None)) {
- return Err(crate::Error::Unsupported {
- message: "Cannot write to a table with a time-travel option set \
- (scan.version / scan.timestamp-millis / scan.timestamp /
scan.watermark / scan.snapshot-id / scan.tag-name)"
- .to_string(),
- });
- }
- Ok(())
+ // Read selectors choose the read view and its schema. Java writers retain
+ // that schema but restore sequence/index state and publish against latest.
+ ensure_paimon_table(table)
}
pub(super) fn ensure_paimon_table(table: &Table) -> crate::Result<()> {
diff --git a/crates/paimon/tests/cross_partition_write_test.rs
b/crates/paimon/tests/cross_partition_write_test.rs
index 36b82988..93edb51a 100644
--- a/crates/paimon/tests/cross_partition_write_test.rs
+++ b/crates/paimon/tests/cross_partition_write_test.rs
@@ -244,6 +244,59 @@ async fn
merge_engines_route_updates_like_java_after_restart() {
}
}
+#[tokio::test]
+async fn historical_read_views_restore_the_latest_global_key_locations() {
+ for engine in ["deduplicate", "first-row", "partial-update",
"aggregation"] {
+ for captured_empty in [false, true] {
+ let mut options = vec![("merge-engine", engine)];
+ if engine == "aggregation" {
+ options.push(("fields.v.aggregate-function", "sum"));
+ }
+ let table = table(true, &options).await;
+ let empty = table.copy_with_pinned_snapshot(None);
+ write_batch(&table, &batch(&[("x", "a", 1, 10)])).await;
+ let first = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let historical = table.copy_with_pinned_snapshot(Some(&first));
+ // This key is absent from either historical view. Its location
must
+ // still be restored before the next write, as in IndexBootstrap's
+ // forced latest scan rather than the writer table's read selector.
+ write_batch(&table, &batch(&[("x", "b", 2, 20)])).await;
+ let view = if captured_empty { &empty } else { &historical };
+ write_batch(view, &batch(&[("x", "c", 2, 30)])).await;
+
+ let (partition, value) = match engine {
+ "deduplicate" => ("c", 30),
+ "first-row" => ("b", 20),
+ "partial-update" => ("b", 30),
+ "aggregation" => ("b", 50),
+ _ => unreachable!(),
+ };
+ assert_eq!(
+ rows(&table).await,
+ vec![
+ ("x".into(), "a".into(), 1, 10),
+ ("x".into(), partition.into(), 2, value)
+ ],
+ "engine={engine}, captured_empty={captured_empty}"
+ );
+ // Restoring the writer must not mutate either public read view.
+ assert_eq!(
+ rows(view).await,
+ if captured_empty {
+ vec![]
+ } else {
+ vec![("x".into(), "a".into(), 1, 10)]
+ }
+ );
+ }
+ }
+}
+
fn with_kinds(rows: &[(&str, &str, i32, i32)], kinds: &[&str]) -> RecordBatch {
let data = batch(rows);
let mut fields = data.schema().fields().to_vec();
diff --git a/crates/paimon/tests/filtered_write_validation_test.rs
b/crates/paimon/tests/filtered_write_validation_test.rs
index 0a78cd0b..4d51a78e 100644
--- a/crates/paimon/tests/filtered_write_validation_test.rs
+++ b/crates/paimon/tests/filtered_write_validation_test.rs
@@ -335,3 +335,65 @@ async fn
empty_and_filtered_rowkind_inputs_preserve_transport_validation() {
assert!(writer.prepare_commit().await.unwrap().is_empty());
}
}
+
+#[tokio::test]
+async fn completely_filtered_batches_do_not_lock_writer_configuration() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("op", DataType::VarChar(VarCharType::string_type()))
+ .option("rowkind.field", "op")
+ .option("ignore-delete", "true")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(
+ "memory:/filtered-configuration",
+ TableSchema::new(0, &schema),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let batch = |kind| {
+ RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![2])) as ArrayRef),
+ ("op", Arc::new(StringArray::from(vec![kind])) as ArrayRef),
+ ])
+ .unwrap()
+ };
+ let mut writer = table.new_write_builder().new_write().unwrap();
+ writer.write_arrow_batch(&batch("-D")).await.unwrap();
+ writer.with_write_type(vec!["id".into()]).unwrap();
+ writer.with_blob_consumer(None).unwrap();
+ writer.with_blob_uri_reader_factory(None).unwrap();
+ writer.write_arrow_batch(&batch("+I")).await.unwrap();
+ assert!(writer.with_write_type(vec!["id".into()]).is_err());
+ assert!(writer.with_blob_consumer(None).is_err());
+ assert!(writer.with_blob_uri_reader_factory(None).is_err());
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .map(|file| file.row_count)
+ .sum::<i64>(),
+ 1
+ );
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ let reader = table.new_read_builder();
+ let plan = reader.new_scan().plan().await.unwrap();
+ let batches: Vec<_> = reader
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert!(batches
+ .iter()
+ .all(|batch| batch.column_by_name("op").unwrap().is_null(0)));
+}
diff --git a/crates/paimon/tests/historical_table_update_test.rs
b/crates/paimon/tests/historical_table_update_test.rs
index 4a2a92eb..2b9466b5 100644
--- a/crates/paimon/tests/historical_table_update_test.rs
+++ b/crates/paimon/tests/historical_table_update_test.rs
@@ -302,48 +302,254 @@ async fn historical_deletes_preserve_newer_rows() {
}
#[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()],
- )
+async fn historical_table_can_publish_ordinary_append_with_old_schema() {
+ let (latest, historical) =
evolved_table("memory:/historical_ordinary_append").await;
+ let builder = historical.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&batch(
+ &["id", "value", "keep"],
+ &[vec![4], vec![40], vec![400]],
+ ))
.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(),
- }],
- )
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == historical.schema().id()));
+ builder
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
.await
- .unwrap_err();
- assert!(merge.to_string().contains("time-travel option"));
+ .unwrap();
+ assert_eq!(
+ latest
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .schema_id(),
+ latest.schema().id()
+ );
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)],
+ vec![Some(4), Some(40), Some(400), None],
]
);
+ assert_eq!(
+ rows(&historical).await,
+ vec![
+ vec![Some(1), Some(10), Some(100)],
+ vec![Some(2), Some(20), Some(200)],
+ ]
+ );
+}
+
+async fn append_mutation(table: &Table, operation: &str, input: RecordBatch)
-> Vec<CommitMessage> {
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["value".into()]).unwrap();
+ if operation == "upsert" {
+ update
+ .upsert_by_arrow_with_key(vec![input], vec!["id".into()])
+ .await
+ .unwrap()
+ } else {
+ update
+ .merge_into(
+ MergeSource::Batches(vec![input]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: ["id", "value", "keep"]
+ .into_iter()
+ .map(|name| (name.into(),
MergeAssignment::SourceColumn(name.into())))
+ .collect(),
+ }],
+ )
+ .await
+ .unwrap()
+ }
+}
+
+#[tokio::test]
+async fn historical_upsert_and_merge_append_with_the_read_schema() {
+ for operation in ["upsert", "merge"] {
+ for only_matched in [false, true] {
+ let (latest, historical) = evolved_table(&format!(
+ "memory:/historical_append_{operation}_{only_matched}"
+ ))
+ .await;
+ let input = || {
+ if only_matched {
+ batch(&["id", "value", "keep"], &[vec![1], vec![11],
vec![101]])
+ } else {
+ batch(
+ &["id", "value", "keep"],
+ &[vec![1, 4], vec![11, 40], vec![101, 400]],
+ )
+ }
+ };
+ let messages = append_mutation(&historical, operation,
input()).await;
+ assert_eq!(messages.len(), if only_matched { 1 } else { 2 });
+ for message in &messages {
+ for file in &message.new_files {
+ assert_eq!(file.schema_id, 0);
+ if message.check_from_snapshot.is_some() {
+ assert_eq!(message.check_from_snapshot, Some(1));
+ assert_eq!(file.first_row_id, Some(0));
+ assert_eq!(file.write_cols,
Some(vec!["value".into()]));
+ } else {
+ // Only publication assigns an appended row's new
identity.
+ assert_eq!(file.first_row_id, None);
+ }
+ }
+ }
+ commit(&latest, messages).await;
+ let mut 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)],
+ ];
+ if !only_matched {
+ expected.push(vec![Some(4), Some(40), Some(400), None]);
+ }
+ assert_eq!(rows(&latest).await, expected);
+ let stale = append_mutation(&historical, operation, input()).await;
+ let error = latest
+ .new_write_builder()
+ .new_commit()
+ .commit(stale)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().to_lowercase().contains("conflict"),
+ "{error}"
+ );
+ assert_eq!(rows(&latest).await, expected);
+ }
+ }
+}
+
+#[tokio::test]
+async fn captured_empty_upsert_and_merge_only_append() {
+ for operation in ["upsert", "merge"] {
+ let (latest, historical) =
+
evolved_table(&format!("memory:/captured_empty_append_{operation}")).await;
+ let empty = historical.copy_with_pinned_snapshot(None);
+ let messages = append_mutation(
+ &empty,
+ operation,
+ batch(&["id", "value", "keep"], &[vec![4], vec![40], vec![400]]),
+ )
+ .await;
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .all(|message| message.check_from_snapshot.is_none()));
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == 0 && file.first_row_id.is_none()));
+ commit(&latest, messages).await;
+ 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)],
+ vec![Some(4), Some(40), Some(400), None],
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn historical_append_mutations_keep_original_field_ids() {
+ for operation in ["upsert", "merge"] {
+ for change in ["rename", "drop_readd", "widen"] {
+ let (table, historical) = evolved_table(&format!(
+ "memory:/historical_append_fields_{operation}_{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 messages = append_mutation(
+ &historical,
+ operation,
+ batch(
+ &["id", "value", "keep"],
+ &[vec![1, 4], vec![11, 40], vec![101, 400]],
+ ),
+ )
+ .await;
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == 0));
+ // The committer's historical read schema must not replace the
+ // latest schema used for publication or conflict validation.
+ commit(&historical, messages).await;
+ let values = if change == "drop_readd" {
+ [None; 4]
+ } else {
+ [Some(11), Some(20), Some(30), Some(40)]
+ };
+ let expected = vec![
+ vec![Some(1), values[0], Some(100), None],
+ vec![Some(2), values[1], Some(200), None],
+ vec![Some(3), values[2], Some(300), Some(999)],
+ vec![Some(4), values[3], Some(400), None],
+ ];
+ assert_eq!(rows(&latest).await, expected, "{operation} {change}");
+ let stale = append_mutation(
+ &historical,
+ operation,
+ batch(
+ &["id", "value", "keep"],
+ &[vec![1, 4], vec![12, 41], vec![101, 400]],
+ ),
+ )
+ .await;
+ let error = historical
+ .new_write_builder()
+ .new_commit()
+ .commit(stale)
+ .await
+ .unwrap_err();
+ assert!(
+ error.to_string().to_lowercase().contains("conflict"),
+ "{operation} {change}: {error}"
+ );
+ assert_eq!(rows(&latest).await, expected);
+ }
+ }
}
#[tokio::test]
@@ -396,7 +602,7 @@ async fn
historical_partial_files_follow_field_ids_after_schema_changes() {
.iter()
.flat_map(|message| &message.new_files)
.all(|file| file.schema_id == 0 && file.write_cols ==
Some(vec!["value".into()])));
- commit(&latest, messages).await;
+ commit(&historical, messages).await;
let expected = if change == "drop_readd" {
[None, None, None]
} else {
diff --git a/crates/paimon/tests/historical_write_commit_test.rs
b/crates/paimon/tests/historical_write_commit_test.rs
new file mode 100644
index 00000000..babf2ef1
--- /dev/null
+++ b/crates/paimon/tests/historical_write_commit_test.rs
@@ -0,0 +1,319 @@
+// 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, RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, Datum, IntType, Schema, TableSchema};
+use paimon::table::{BranchManager, CommitMessage, Table};
+
+fn batch(extra: bool, values: &[[i32; 4]]) -> RecordBatch {
+ let names = if extra {
+ &["id", "pt", "value", "extra"][..]
+ } else {
+ &["id", "pt", "value"][..]
+ };
+ RecordBatch::try_from_iter(names.iter().enumerate().map(|(index, name)| {
+ (
+ *name,
+ Arc::new(Int32Array::from(
+ values.iter().map(|row| row[index]).collect::<Vec<_>>(),
+ )) as ArrayRef,
+ )
+ }))
+ .unwrap()
+}
+
+async fn write(table: &Table, data: RecordBatch, overwrite: bool) ->
Vec<CommitMessage> {
+ let builder = table.new_write_builder();
+ let builder = if overwrite {
+ builder.with_overwrite()
+ } else {
+ builder
+ };
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&data).await.unwrap();
+ writer.prepare_commit().await.unwrap()
+}
+
+async fn publish(table: &Table, messages: Vec<CommitMessage>) {
+ table
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
+ .await
+ .unwrap();
+}
+
+async fn rows(table: &Table) -> Vec<Vec<Option<i32>>> {
+ 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::new();
+ for batch in batches {
+ for row in 0..batch.num_rows() {
+ rows.push(
+ batch
+ .columns()
+ .iter()
+ .map(|column| {
+ let column =
column.as_any().downcast_ref::<Int32Array>().unwrap();
+ column.is_valid(row).then(|| column.value(row))
+ })
+ .collect::<Vec<_>>(),
+ );
+ }
+ }
+ rows.sort();
+ rows
+}
+
+async fn evolved(path: &str, mode: &str) -> (Table, Table, Vec<CommitMessage>)
{
+ let schema = |extra| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("pt", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .partition_keys(["pt"]);
+ if mode == "pk" {
+ builder = builder.primary_key(["id", "pt"]).option("bucket", "1");
+ } else if mode == "de" {
+ builder = builder
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.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;
+ publish(
+ &original,
+ write(&original, batch(false, &[[1, 1, 10, 0]]), false).await,
+ )
+ .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();
+ let mut values = vec![[2, 1, 20, 200], [4, 2, 40, 400]];
+ if mode == "pk" {
+ values.push([1, 1, 25, 250]);
+ }
+ let messages = write(&latest, batch(true, &values), false).await;
+ publish(&latest, messages.clone()).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_eq!(
+ rows(&historical).await,
+ vec![vec![Some(1), Some(1), Some(10)]]
+ );
+ (latest, historical, messages)
+}
+
+#[tokio::test]
+async fn historical_writers_and_committers_use_latest_publication_state() {
+ for mode in ["append", "de", "pk"] {
+ for operation in ["append", "commit", "overwrite", "truncate",
"partitions"] {
+ let (latest, historical, _) = evolved(
+ &format!("memory:/historical_write_{mode}_{operation}"),
+ mode,
+ )
+ .await;
+ let mut expected = rows(&latest).await;
+ if operation == "truncate" {
+ historical
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .truncate_table()
+ .await
+ .unwrap();
+ expected.clear();
+ } else if operation == "partitions" {
+ historical
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .truncate_partitions(vec![HashMap::from([("pt".into(),
Some(Datum::Int(1)))])])
+ .await
+ .unwrap();
+ expected.retain(|row| row[1] != Some(1));
+ } else {
+ let current_input = operation == "commit";
+ let messages = write(
+ if current_input { &latest } else { &historical },
+ batch(current_input, &[[3, 1, 30, 300]]),
+ operation == "overwrite",
+ )
+ .await;
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == i64::from(current_input)));
+ if operation == "overwrite" {
+ historical
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .overwrite(
+ messages,
+ Some(HashMap::from([("pt".into(),
Some(Datum::Int(1)))])),
+ )
+ .await
+ .unwrap();
+ expected.retain(|row| row[1] != Some(1));
+ } else {
+ publish(&historical, messages).await;
+ }
+ expected.push(vec![
+ Some(3),
+ Some(1),
+ Some(30),
+ current_input.then_some(300),
+ ]);
+ expected.sort();
+ }
+ assert_eq!(rows(&latest).await, expected, "{mode} {operation}");
+ let snapshot = latest
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.id(), 3);
+ assert_eq!(snapshot.schema_id(), 1);
+ // Publication does not change the writer handle's selected read
view.
+ assert_eq!(
+ rows(&historical).await,
+ vec![vec![Some(1), Some(1), Some(10)]]
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn historical_primary_key_writer_restores_latest_sequence_numbers() {
+ let (latest, historical, recent) =
evolved("memory:/historical_pk_sequences", "pk").await;
+ let latest_sequence = recent
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .filter(|file| file.max_sequence_number >= 0)
+ .map(|file| file.max_sequence_number)
+ .max()
+ .unwrap();
+ let messages = write(&historical, batch(false, &[[1, 1, 99, 0]]),
false).await;
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.min_sequence_number > latest_sequence));
+ publish(&historical, messages).await;
+ assert!(rows(&latest)
+ .await
+ .contains(&vec![Some(1), Some(1), Some(99), None]));
+}
+
+#[tokio::test]
+async fn captured_empty_writer_keeps_its_empty_read_view() {
+ for mode in ["append", "de", "pk"] {
+ let (latest, historical, _) =
+ evolved(&format!("memory:/captured_empty_write_{mode}"),
mode).await;
+ let empty = historical.copy_with_pinned_snapshot(None);
+ assert!(rows(&empty).await.is_empty());
+ let before = rows(&latest).await.len();
+ publish(
+ &empty,
+ write(&empty, batch(false, &[[3, 1, 30, 0]]), false).await,
+ )
+ .await;
+ assert!(rows(&empty).await.is_empty());
+ assert_eq!(rows(&latest).await.len(), before + 1);
+ assert_eq!(
+ latest
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .schema_id(),
+ 1
+ );
+ }
+}
+
+#[tokio::test]
+async fn historical_branch_publication_keeps_main_isolated() {
+ let (main, _, _) = evolved("memory:/historical_write_branch", "de").await;
+ let snapshot = main
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ main.tag_manager().create("base", &snapshot).await.unwrap();
+ BranchManager::new(main.file_io().clone(), main.location().into())
+ .create_branch_from_tag("dev", "base")
+ .await
+ .unwrap();
+ let branch = main.copy_with_branch("dev").await.unwrap();
+ let historical = branch
+ .copy_with_time_travel(HashMap::from([("scan.snapshot-id".into(),
"2".into())]))
+ .await
+ .unwrap();
+ let main_rows = rows(&main).await;
+ publish(
+ &branch,
+ write(&branch, batch(true, &[[5, 1, 50, 500]]), false).await,
+ )
+ .await;
+ publish(
+ &historical,
+ write(&historical, batch(true, &[[6, 1, 60, 600]]), false).await,
+ )
+ .await;
+ assert_eq!(rows(&main).await, main_rows);
+ assert_eq!(rows(&branch).await.len(), 5);
+ historical
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .truncate_table()
+ .await
+ .unwrap();
+ assert!(rows(&branch).await.is_empty());
+ assert_eq!(rows(&main).await, main_rows);
+}
diff --git a/crates/paimon/tests/partitioned_blob_read_test.rs
b/crates/paimon/tests/partitioned_blob_read_test.rs
new file mode 100644
index 00000000..b4820d17
--- /dev/null
+++ b/crates/paimon/tests/partitioned_blob_read_test.rs
@@ -0,0 +1,280 @@
+// 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, Int64Array, LargeBinaryArray,
RecordBatch, StringArray};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs};
+use futures::TryStreamExt;
+use paimon::spec::{
+ BlobType, DataType, Datum, IntType, Predicate, PredicateBuilder, Schema,
TableSchema,
+ VarCharType,
+};
+use std::sync::Arc;
+
+#[tokio::test]
+async fn
blob_preselection_restores_omitted_partition_columns_before_filtering() {
+ for nullable in [true, false] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(
+ "p",
+ DataType::VarChar(
+ VarCharType::with_nullable(nullable,
VarCharType::MAX_LENGTH).unwrap(),
+ ),
+ )
+ .column("payload", DataType::Blob(BlobType::new()))
+ .partition_keys(["p"])
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .option("read.batch-size", "2")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table(
+ "memory:/partitioned_blob_read",
+ TableSchema::new(0, &schema),
+ );
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let batch = RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef,
+ ),
+ (
+ "p",
+ Arc::new(StringArray::from(vec!["a", "b", "a", "b"])) as
ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ b"one".as_slice(),
+ b"two",
+ b"three",
+ b"four",
+ ])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .with_write_type(vec!["id".into(), "payload".into()])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| {
+ !file
+ .write_cols
+ .as_ref()
+ .unwrap()
+ .iter()
+ .any(|column| column == "p")
+ }));
+ builder.new_commit().commit(messages).await.unwrap();
+ let splits = table.new_read_builder().new_scan().plan().await.unwrap();
+ let pb = PredicateBuilder::new(table.schema().fields());
+ let predicate = Predicate::or(vec![
+ pb.equal("p", Datum::String("a".into())).unwrap(),
+ pb.equal("id", Datum::Int(999)).unwrap(),
+ ]);
+ for limit in [None, Some(1)] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&["id", "payload"])
+ .unwrap()
+ .with_filter(predicate.clone());
+ if let Some(limit) = limit {
+ read.with_limit(limit);
+ }
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(splits.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut ids = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ ids.sort_unstable();
+ assert_eq!(
+ ids,
+ if limit.is_some() { vec![1] } else { vec![1, 3] },
+ "nullable={nullable}, limit={limit:?}"
+ );
+ let payloads = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .iter()
+ .map(|value| value.unwrap().to_vec())
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(
+ payloads,
+ if limit.is_some() {
+ vec![b"one".to_vec()]
+ } else {
+ vec![b"one".to_vec(), b"three".to_vec()]
+ }
+ );
+ }
+ // A projection containing only partition columns still has one row
+ // per visible row, without needing a physical provider for that
column.
+ let mut read = table.new_read_builder();
+ read.with_projection(&["p"]).unwrap();
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(splits.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
4);
+ let mut values = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .iter()
+ .map(|value| value.unwrap().to_string())
+ })
+ .collect::<Vec<_>>();
+ values.sort();
+ assert_eq!(values, ["a", "a", "b", "b"]);
+
+ let mut metadata = table.new_read_builder();
+ metadata.with_projection(&["id", "_ROW_ID"]).unwrap();
+ let rows: Vec<RecordBatch> = metadata
+ .new_read()
+ .unwrap()
+ .to_arrow(splits.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let row_ids = rows
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let row_ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ ids.values()
+ .iter()
+ .copied()
+ .zip(row_ids.values().iter().copied())
+ })
+ .collect::<std::collections::HashMap<_, _>>();
+ let mut update = builder.new_update().unwrap();
+ update.with_update_type(vec!["id".into()]).unwrap();
+ let delta = RecordBatch::try_from_iter([
+ ("id", Arc::new(Int32Array::from(vec![11])) as ArrayRef),
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![row_ids[&1]])) as ArrayRef,
+ ),
+ ])
+ .unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![delta])
+ .await
+ .unwrap();
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.write_cols == Some(vec!["id".to_string()])));
+ builder.new_commit().commit(messages).await.unwrap();
+ let deletions = builder
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![row_ids[&3]])
+ .await
+ .unwrap();
+ builder.new_commit().commit(deletions).await.unwrap();
+
+ let latest = table.new_read_builder().new_scan().plan().await.unwrap();
+ for projection in [vec!["id", "payload"], vec![], vec!["p"]] {
+ let mut read = table.new_read_builder();
+ read.with_projection(&projection)
+ .unwrap()
+ .with_filter(predicate.clone())
+ .with_limit(1);
+ let batches: Vec<RecordBatch> = read
+ .new_read()
+ .unwrap()
+ .to_arrow(latest.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
+ assert!(batches
+ .iter()
+ .all(|batch| batch.num_columns() == projection.len()));
+ if projection == ["id", "payload"] {
+ let batch = batches.iter().find(|batch| batch.num_rows() >
0).unwrap();
+ assert_eq!(
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 11
+ );
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<LargeBinaryArray>()
+ .unwrap()
+ .value(0),
+ b"one"
+ );
+ }
+ }
+ }
+}
diff --git a/crates/paimon/tests/postpone_native_write_test.rs
b/crates/paimon/tests/postpone_native_write_test.rs
index 3e4d7b53..888d4dae 100644
--- a/crates/paimon/tests/postpone_native_write_test.rs
+++ b/crates/paimon/tests/postpone_native_write_test.rs
@@ -22,6 +22,49 @@ use common::incremental_helpers::{make_batch_with_kinds,
memory_table, pk_schema
use paimon::spec::{BinaryRow, POSTPONE_BUCKET};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
+#[tokio::test]
+async fn
automatic_postpone_configuration_locks_on_buffered_not_filtered_input() {
+ let (io, table) = memory_table(
+ "memory:/postpone_configuration",
+ pk_schema(&[("bucket", "-2"), ("ignore-delete", "true")]),
+ );
+ setup_dirs(&io, table.location()).await;
+ let consumer: std::sync::Arc<dyn paimon::spec::BlobConsumer> =
+ std::sync::Arc::new(|_: &str, _:
Option<&paimon::spec::BlobDescriptor>| {
+ panic!("PK consumer must not run")
+ });
+ let mut writer = table
+ .new_postpone_fixed_bucket_write_builder()
+ .unwrap()
+ .new_write()
+ .unwrap();
+ writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![1], vec![10], vec![3]))
+ .await
+ .unwrap();
+ writer.with_blob_consumer(Some(consumer.clone())).unwrap();
+ writer.with_blob_consumer(None).unwrap();
+ writer.with_blob_uri_reader_factory(None).unwrap();
+ // Automatic planning retains this input until prepare_commit. Its inner
+ // writer has no physical output yet, but changing configuration is late.
+ writer
+ .write_arrow_batch(&make_batch_with_kinds(vec![2], vec![20], vec![0]))
+ .await
+ .unwrap();
+ assert!(writer.with_blob_consumer(None).is_err());
+ assert!(writer.with_blob_consumer(Some(consumer)).is_err());
+ assert!(writer.with_blob_uri_reader_factory(None).is_err());
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(
+ messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .map(|file| file.row_count)
+ .sum::<i64>(),
+ 1
+ );
+}
+
#[tokio::test]
async fn postpone_rolls_by_batch_row_limit_and_preserves_unsorted_retracts() {
let (io, table) = memory_table(
diff --git a/crates/paimon/tests/precomputed_bucket_write_test.rs
b/crates/paimon/tests/precomputed_bucket_write_test.rs
index 5a2f58fe..73835de1 100644
--- a/crates/paimon/tests/precomputed_bucket_write_test.rs
+++ b/crates/paimon/tests/precomputed_bucket_write_test.rs
@@ -582,6 +582,56 @@ async fn
prepared_checkpoints_keep_sequence_progress_before_publication() {
assert_eq!(rows(&table).await, vec![(1, 30), (2, 20)]);
}
+#[tokio::test]
+async fn newly_opened_buckets_restore_latest_sequences_after_a_checkpoint() {
+ let table = table(&[("bucket", "2")]).await;
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow(&[make_batch(vec![1], vec![10])], Some(0))
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .commit_with_identifier(writer.prepare_commit().await.unwrap(), 1)
+ .await
+ .unwrap();
+
+ let mut other = table.new_write_builder().new_write().unwrap();
+ other
+ .write_arrow(&[make_batch(vec![2, 2, 2], vec![20, 21, 22])], Some(1))
+ .await
+ .unwrap();
+ commit(&table, other.prepare_commit().await.unwrap()).await;
+
+ writer
+ .write_arrow(&[make_batch(vec![2], vec![99])], Some(1))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].new_files[0].min_sequence_number, 3);
+ builder
+ .new_commit()
+ .commit_with_identifier(messages, 2)
+ .await
+ .unwrap();
+ assert_eq!(rows(&table).await, vec![(1, 10), (2, 99)]);
+
+ writer
+ .write_arrow(&[make_batch(vec![1], vec![11])], Some(0))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages[0].new_files[0].min_sequence_number, 1);
+ builder
+ .new_commit()
+ .commit_with_identifier(messages, 3)
+ .await
+ .unwrap();
+ assert_eq!(rows(&table).await, vec![(1, 11), (2, 99)]);
+}
+
#[tokio::test]
async fn configured_overwrite_can_use_ordinary_and_direct_assignment() {
let table = table(&[("bucket", "-1")]).await;
diff --git a/crates/paimon/tests/preplanned_row_id_read_test.rs
b/crates/paimon/tests/preplanned_row_id_read_test.rs
new file mode 100644
index 00000000..3507d090
--- /dev/null
+++ b/crates/paimon/tests/preplanned_row_id_read_test.rs
@@ -0,0 +1,250 @@
+// 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::{Int32Array, RecordBatch};
+use common::incremental_helpers::{
+ make_batch, memory_table, persist_table_schema, setup_dirs, write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, Datum, IntType, Predicate, PredicateBuilder,
Schema, TableSchema};
+use paimon::table::{DataSplit, DataSplitBuilder, ReadBuilder, RowRange};
+
+async fn ids(builder: &ReadBuilder<'_>, splits: &[DataSplit]) -> Vec<i32> {
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut result = batches
+ .iter()
+ .flat_map(|batch| {
+ assert_eq!(batch.schema().fields().len(), 1);
+ assert_eq!(batch.schema().field(0).name(), "id");
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .to_vec()
+ })
+ .collect::<Vec<_>>();
+ result.sort_unstable();
+ result
+}
+
+#[tokio::test]
+async fn a_reader_applies_its_row_id_filter_to_an_existing_plan() {
+ for evolution in [false, true] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", evolution.to_string())
+ .option("read.batch-size", "2")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/preplanned_row_id",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ write_batch(&table, &make_batch(vec![10, 11, 12], vec![100, 110,
120])).await;
+ write_batch(&table, &make_batch(vec![13, 14, 15], vec![130, 140,
150])).await;
+ let plan = table.new_read_builder().new_scan().plan().await.unwrap();
+ let fields = table.schema().fields_with_read_metadata();
+ let pb = PredicateBuilder::new(&fields);
+ let cases = [
+ (pb.equal("_ROW_ID", Datum::Long(1)).unwrap(), vec![11]),
+ (
+ pb.equal("_SEQUENCE_NUMBER", Datum::Long(2)).unwrap(),
+ vec![13, 14, 15],
+ ),
+ (
+ Predicate::and(vec![
+ pb.greater_or_equal("_ROW_ID", Datum::Long(1)).unwrap(),
+ pb.less_than("_ROW_ID", Datum::Long(5)).unwrap(),
+ pb.greater_than("value", Datum::Int(120)).unwrap(),
+ ]),
+ vec![13, 14],
+ ),
+ (
+ Predicate::or(vec![
+ pb.equal("_ROW_ID", Datum::Long(1)).unwrap(),
+ pb.equal("value", Datum::Int(140)).unwrap(),
+ ]),
+ vec![11, 14],
+ ),
+ (
+ Predicate::or(vec![
+ pb.equal("_SEQUENCE_NUMBER", Datum::Long(1)).unwrap(),
+ pb.equal("_ROW_ID", Datum::Long(4)).unwrap(),
+ ]),
+ vec![10, 11, 12, 14],
+ ),
+ (
+ Predicate::and(vec![
+ pb.equal("_ROW_ID", Datum::Long(1)).unwrap(),
+ pb.equal("_ROW_ID", Datum::Long(2)).unwrap(),
+ ]),
+ vec![],
+ ),
+ ];
+ for (predicate, expected) in cases {
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(predicate.clone());
+ assert_eq!(
+ ids(&builder, plan.splits()).await,
+ expected,
+ "evolution={evolution}, {predicate:?}"
+ );
+ let mut count = builder.clone();
+ count.with_projection(&[]).unwrap();
+ let counted: Vec<_> = count
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert!(counted.iter().all(|batch| batch.num_columns() == 0));
+ assert_eq!(
+ counted.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ expected.len()
+ );
+ if evolution {
+ builder.with_limit(1);
+ let limited = ids(&builder, plan.splits()).await;
+ assert_eq!(limited.len(), expected.len().min(1));
+ assert!(limited.iter().all(|id| expected.contains(id)));
+ }
+ }
+
+ let mut selected = table.new_read_builder();
+ selected.with_row_ranges(vec![RowRange::new(1, 4)]);
+ let selected = selected.new_scan().plan().await.unwrap();
+ let mut reader = table.new_read_builder();
+ reader
+ .with_projection(&["id"])
+ .unwrap()
+ .with_filter(Predicate::or(vec![
+ pb.equal("_ROW_ID", Datum::Long(0)).unwrap(),
+ pb.equal("_ROW_ID", Datum::Long(2)).unwrap(),
+ pb.equal("_ROW_ID", Datum::Long(4)).unwrap(),
+ pb.equal("_ROW_ID", Datum::Long(5)).unwrap(),
+ ]));
+ assert_eq!(ids(&reader, selected.splits()).await, vec![12, 14]);
+ // A new filter replaces its old derived range as well as the residual.
+ reader.with_filter(pb.equal("id", Datum::Int(13)).unwrap());
+ assert_eq!(ids(&reader, plan.splits()).await, vec![13]);
+ }
+}
+
+#[tokio::test]
+async fn physical_slices_use_the_readers_coordinate_system_before_filtering() {
+ for tracked in [false, true] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", tracked.to_string())
+ .option("read.batch-size", "2")
+ .build()
+ .unwrap();
+ let (io, table) = memory_table("memory:/physical_slices",
TableSchema::new(0, &schema));
+ setup_dirs(&io, table.location()).await;
+ persist_table_schema(&io, table.location(), table.schema()).await;
+ let mut seen = std::collections::HashSet::new();
+ let mut files = Vec::new();
+ let mut bucket_path = String::new();
+ for start in [10, 20, 30] {
+ let values = (start..start + 4).collect::<Vec<_>>();
+ write_batch(&table, &make_batch(values.clone(), values)).await;
+ let plan =
table.new_read_builder().new_scan().plan().await.unwrap();
+ for split in plan.splits() {
+ bucket_path = split.bucket_path().to_string();
+ for file in split.data_files() {
+ if seen.insert(file.file_name.clone()) {
+ files.push(file.clone());
+ }
+ }
+ }
+ }
+ assert_eq!(files.len(), 3);
+ // Drop the middle file and reverse the others. Stable row IDs retain
+ // their gaps; raw offsets restart at zero in this supplied split.
+ let selected = vec![files[2].clone(), files[0].clone()];
+ let mut offset = 0;
+ let ranges = selected
+ .iter()
+ .zip([(2, 4), (1, 3)])
+ .map(|(file, (start, end))| {
+ let base = if tracked {
+ file.first_row_id.unwrap()
+ } else {
+ offset
+ };
+ offset += file.row_count;
+ RowRange::new(base + start, base + end - 1)
+ })
+ .collect();
+ let split = DataSplitBuilder::new()
+ .with_snapshot(3)
+ .with_partition(paimon::spec::BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(bucket_path)
+ .with_data_files(selected)
+ .with_raw_convertible(true)
+ .with_row_ranges(ranges)
+ .build()
+ .unwrap();
+ let mut reader = table.new_read_builder();
+ reader.with_projection(&["id"]).unwrap();
+ assert_eq!(
+ ids(&reader, std::slice::from_ref(&split)).await,
+ vec![11, 12, 32, 33],
+ "tracked={tracked}"
+ );
+ reader.with_filter(
+ PredicateBuilder::new(table.schema().fields())
+ .greater_or_equal("id", Datum::Int(12))
+ .unwrap(),
+ );
+ assert_eq!(
+ ids(&reader, std::slice::from_ref(&split)).await,
+ vec![12, 32, 33],
+ "tracked={tracked}"
+ );
+ reader.with_projection(&[]).unwrap();
+ let counted = reader
+ .new_read()
+ .unwrap()
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert!(counted.iter().all(|batch| batch.num_columns() == 0));
+ assert_eq!(counted.iter().map(RecordBatch::num_rows).sum::<usize>(),
3);
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog/merge_sources.rs
b/crates/paimon/tests/rest_catalog/merge_sources.rs
index 011a4ad2..3f2917c5 100644
--- a/crates/paimon/tests/rest_catalog/merge_sources.rs
+++ b/crates/paimon/tests/rest_catalog/merge_sources.rs
@@ -17,6 +17,272 @@
use super::*;
use paimon::table::{MergeAssignment, MergeSource, TableCommit, WhenMatched,
WhenNotMatched};
+use std::sync::atomic::{AtomicUsize, Ordering};
+
+#[tokio::test]
+async fn historical_append_writers_do_not_require_query_read_grants() {
+ for empty in [false, true] {
+ for policy in ["denied", "filter", "mask"] {
+ let g = written(
+ "historical_append_target",
+ |options| {
+ people(
+ &[
+ options,
+ &[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ],
+ ]
+ .concat(),
+ )
+ },
+ vec![people_rows(1..=2), people_rows(3..=3)],
+ )
+ .await;
+ let snapshot =
g.table.snapshot_manager().get_snapshot(1).await.unwrap();
+ let target = g
+ .table
+ .copy_with_pinned_snapshot((!empty).then_some(&snapshot));
+ match policy {
+ "denied" => g.ctx.server.set_column_auth(
+ "default",
+ "historical_append_target",
+ vec!["id".into()],
+ ),
+ "filter" => g.set_rules(&[int_leaf(0, "id", "GREATER_THAN",
1)], &[]),
+ "mask" => g.set_rules(&[], &[("name", upper(1, "name"))]),
+ _ => unreachable!(),
+ }
+ let builder = target.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&people_batch(&[(4, "new")]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert!(!messages.is_empty());
+ assert!(messages
+ .iter()
+ .flat_map(|message| &message.new_files)
+ .all(|file| file.schema_id == target.schema().id() &&
file.first_row_id.is_none()));
+ builder.try_new_commit().unwrap();
+ assert!(g
+ .ctx
+ .server
+ .auth_selects("default", "historical_append_target")
+ .is_empty());
+
+ // The mock REST endpoint serves SELECT authorization, not commit
+ // publication. Publish through its backing filesystem fixture.
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ g._tmp.path().to_string_lossy().replace('\\', "/"),
+ );
+ let publisher = FileSystemCatalog::new(options)
+ .unwrap()
+ .get_table(&g.identifier)
+ .await
+ .unwrap();
+ TableCommit::new(publisher.clone(), "historical-append".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ ids(&read_all(&publisher.new_read_builder()).await.unwrap()),
+ [1, 2, 3, 4]
+ );
+
+ // A supplied schema cannot obtain publication authority merely by
+ // selecting a historical view of the same physical table.
+ let assembled = Table::new(
+ target.file_io().clone(),
+ target.identifier().clone(),
+ target.location().into(),
+ target.schema().clone(),
+ None,
+ )
+ .copy_with_pinned_snapshot((!empty).then_some(&snapshot));
+
assert_refused(assembled.new_write_builder().new_write().err().unwrap());
+ assert_refused(
+ assembled
+ .new_write_builder()
+ .try_new_commit()
+ .err()
+ .unwrap(),
+ );
+ assert_refused(
+ TableCommit::new(assembled, "assembled".into())
+ .commit(Vec::new())
+ .await
+ .unwrap_err(),
+ );
+ assert!(g
+ .ctx
+ .server
+ .auth_selects("default", "historical_append_target")
+ .is_empty());
+ }
+ }
+}
+
+#[tokio::test]
+async fn
historical_insert_mutations_require_whole_target_grants_before_staging() {
+ for operation in ["upsert", "merge"] {
+ for empty in [false, true] {
+ for policy in ["whole", "denied", "filter", "mask"] {
+ let g = written(
+ "historical_insert_target",
+ |options| {
+ people(
+ &[
+ options,
+ &[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ("deletion-vectors.enabled", "true"),
+ ],
+ ]
+ .concat(),
+ )
+ },
+ vec![people_rows(1..=2), people_rows(3..=3)],
+ )
+ .await;
+ let snapshot =
g.table.snapshot_manager().get_snapshot(1).await.unwrap();
+ let target = g
+ .table
+ .copy_with_pinned_snapshot((!empty).then_some(&snapshot));
+ match policy {
+ "denied" => g.ctx.server.set_column_auth(
+ "default",
+ "historical_insert_target",
+ vec!["id".into()],
+ ),
+ "filter" => g.set_rules(&[int_leaf(0, "id",
"GREATER_THAN", 1)], &[]),
+ "mask" => g.set_rules(&[], &[("name", upper(1, "name"))]),
+ _ => g.set_rules(&[], &[]),
+ }
+ let before = target
+ .file_io()
+ .list_status_recursive(target.location())
+ .await
+ .unwrap()
+ .len();
+ let calls = Arc::new(AtomicUsize::new(0));
+ let called = calls.clone();
+ let assignment =
MergeAssignment::Value(paimon::table::UpdateAssignment::Function(
+ Arc::new(move |batches| {
+ called.fetch_add(1, Ordering::SeqCst);
+ Ok(batches
+ .iter()
+ .map(|batch|
batch.column_by_name("s.name").unwrap().clone())
+ .collect())
+ }),
+ ));
+ let mut update =
target.new_write_builder().new_update().unwrap();
+ update.with_update_type(vec!["name".into()]).unwrap();
+ let source = people_batch(&[(1, "changed"), (4, "new")]);
+ let result = if operation == "upsert" {
+ update
+ .upsert_by_arrow_with_key(vec![source],
vec!["id".into()])
+ .await
+ } else {
+ update
+ .merge_into(
+ MergeSource::Batches(vec![source]),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![("name".into(),
assignment.clone())],
+ }],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: vec![
+ ("id".into(),
MergeAssignment::SourceColumn("id".into())),
+ ("name".into(), assignment),
+ ],
+ }],
+ )
+ .await
+ };
+ if policy == "whole" {
+ let messages = result.unwrap();
+ assert!(!messages.is_empty());
+ if operation == "merge" {
+ assert!(calls.load(Ordering::SeqCst) > 0);
+ }
+ assert!(messages
+ .iter()
+ .any(|message| message.check_from_snapshot.is_none()));
+ // This fixture's REST server supplies grants, while its
+ // filesystem catalog owns snapshot publication.
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ g._tmp.path().to_string_lossy().replace('\\', "/"),
+ );
+ let publisher = FileSystemCatalog::new(options)
+ .unwrap()
+ .get_table(&g.identifier)
+ .await
+ .unwrap();
+ TableCommit::new(publisher, "historical-inserts".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let mut expected = vec![
+ (1, Some("changed".into())),
+ (2, Some("bob".into())),
+ (3, Some("carol".into())),
+ (4, Some("new".into())),
+ ];
+ if empty {
+ expected.push((1, Some("alice".into())));
+ expected.sort();
+ }
+ assert_eq!(
+
id_names(&read_all(&g.table.new_read_builder()).await.unwrap()),
+ expected
+ );
+ } else {
+ let error = result.unwrap_err();
+ if policy == "denied" {
+ assert_forbidden(error);
+ } else {
+ assert!(error.to_string().contains("unrestricted"),
"{error}");
+ }
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+ assert_eq!(
+ target
+ .file_io()
+ .list_status_recursive(target.location())
+ .await
+ .unwrap()
+ .len(),
+ before
+ );
+ assert_eq!(
+ target
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(2)
+ );
+ }
+ let selects = g
+ .ctx
+ .server
+ .auth_selects("default", "historical_insert_target");
+ assert!(!selects.is_empty());
+ assert!(selects.iter().all(Option::is_none));
+ }
+ }
+ }
+}
fn target() -> Table {
Table::new(
diff --git a/crates/paimon/tests/table_predicate_delete_test.rs
b/crates/paimon/tests/table_predicate_delete_test.rs
index 1dc8e441..6ccc96e9 100644
--- a/crates/paimon/tests/table_predicate_delete_test.rs
+++ b/crates/paimon/tests/table_predicate_delete_test.rs
@@ -527,7 +527,6 @@ async fn
historical_predicate_delete_keeps_auth_and_field_index_validation() {
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(),
@@ -541,8 +540,14 @@ async fn
historical_predicate_delete_keeps_auth_and_field_index_validation() {
.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;
+ // A historical handle can append a later row and publish the earlier
+ // predicate deletion; the read snapshot still limits deletion selection.
+ write_batch(
+ &historical,
+ &batch(vec![4], vec![Some("a")], vec![50], vec![1]),
+ )
+ .await;
+ commit(&historical, messages).await;
assert_eq!(ids(&table).await, vec![2, 3, 4]);
}
diff --git a/crates/paimon/tests/table_update_test.rs
b/crates/paimon/tests/table_update_test.rs
index cd39b34d..3a78114f 100644
--- a/crates/paimon/tests/table_update_test.rs
+++ b/crates/paimon/tests/table_update_test.rs
@@ -1373,7 +1373,7 @@ async fn
empty_update_type_means_all_columns_for_upsert_only() {
}
#[tokio::test]
-async fn
update_builder_rejects_overwrite_but_allows_historical_partial_updates() {
+async fn
update_builder_rejects_overwrite_but_historical_handle_can_publish_updates() {
let table = evolution_table().await;
assert!(table
.new_write_builder()
@@ -1394,9 +1394,13 @@ async fn
update_builder_rejects_overwrite_but_allows_historical_partial_updates(
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;
+ historical
+ .new_write_builder()
+ .try_new_commit()
+ .unwrap()
+ .commit(messages)
+ .await
+ .unwrap();
assert_eq!(
read_rows(&table).await,
vec![vec![1, 11, 100], vec![2, 20, 200], vec![3, 30, 300]]
