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 ecbf5806 [core] Add captured mutation and index build plans with
structured row-id conflicts (#1202)
ecbf5806 is described below
commit ecbf5806c2daa3c013752db46526085b0f4747f4
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 23 08:47:39 2026 +0800
[core] Add captured mutation and index build plans with structured row-id
conflicts (#1202)
---
.../python/python/pypaimon_rust/datafusion.pyi | 58 +++
bindings/python/src/context.rs | 3 +
bindings/python/src/error.rs | 153 +++++++-
bindings/python/src/index_build.rs | 106 +++++-
bindings/python/src/read.rs | 56 ++-
bindings/python/src/write.rs | 243 ++++++++++++
crates/paimon/src/error.rs | 52 +++
crates/paimon/src/table/authorized_update_tests.rs | 184 +++++++++
crates/paimon/src/table/data_evolution_writer.rs | 160 +++++++-
.../paimon/src/table/global_index_build_builder.rs | 42 +++
.../src/table/global_index_build_common/vector.rs | 2 +-
crates/paimon/src/table/global_index_build_plan.rs | 151 ++++++++
crates/paimon/src/table/mod.rs | 4 +-
crates/paimon/src/table/source.rs | 12 +-
crates/paimon/src/table/table_commit.rs | 107 +++++-
.../table/table_commit/row_id_conflict_tests.rs | 222 +++++++++++
crates/paimon/src/table/table_update.rs | 49 ++-
crates/paimon/src/table/table_update_by_row_id.rs | 101 ++++-
crates/paimon/src/table/table_update_plan_tests.rs | 418 +++++++++++++++++++++
.../paimon/src/table/vindex_index_build_builder.rs | 149 +++++---
.../src/table/vindex_index_build_builder/tests.rs | 151 ++++++++
21 files changed, 2343 insertions(+), 80 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 427e9810..f50293ef 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -57,6 +57,9 @@ class Split:
def serialize(self) -> bytes: ...
class Plan:
+ def __init__(self, snapshot_id: Optional[int], splits: Iterable[Split]) ->
None:
+ """Capture typed split metadata; constructing a plan does not
authorize execution."""
+ ...
def snapshot_id(self) -> Optional[int]:
"""Selected snapshot, preserved for empty plans; None if no snapshot
was selected."""
...
@@ -66,6 +69,16 @@ class Plan:
...
def __len__(self) -> int: ...
+class RowIdPlan:
+ def __init__(self, plan: Plan) -> None:
+ """Group complete data-evolution files from plan_all without reading
manifests."""
+ ...
+ def snapshot_id(self) -> Optional[int]: ...
+ def groups(self) -> List[Tuple[int, int, Split]]:
+ """Inclusive row-ID bounds and all physical files of each logical
group."""
+ ...
+ def plan(self) -> Plan: ...
+
class ScanStats:
"""Read-only counters from normal metadata planning."""
@property
@@ -417,6 +430,14 @@ class Table:
"""
...
+class GlobalIndexBuildPlan:
+ def snapshot_id(self) -> Optional[int]: ...
+ def shards(self) -> List["GlobalIndexBuildShard"]: ...
+
+class GlobalIndexBuildShard:
+ def snapshot_id(self) -> int: ...
+ def row_range(self) -> Tuple[int, int]: ...
+
class GlobalIndexBuildBuilder:
@staticmethod
def supports_index_type(index_type: str) -> bool: ...
@@ -425,6 +446,16 @@ class GlobalIndexBuildBuilder:
def with_index_type(self, index_type: str) -> "GlobalIndexBuildBuilder":
...
def with_options(self, options: Dict[str, str]) ->
"GlobalIndexBuildBuilder": ...
def with_partition_filter(self, predicate: Dict[str, Any]) ->
"GlobalIndexBuildBuilder": ...
+ def plan(self) -> GlobalIndexBuildPlan:
+ """Capture independently executable vindex shards at one snapshot."""
+ ...
+ def build_shard(self, shard: GlobalIndexBuildShard) ->
List["CommitMessage"]:
+ """Build one captured shard with fresh worker authorization.
+
+ The shard supplies its schema, index configuration and options;
+ this builder supplies the worker table and I/O environment.
+ """
+ ...
def build(self) -> List["CommitMessage"]:
"""Build unpublished index files; the caller decides whether to commit
or abort."""
...
@@ -477,6 +508,17 @@ class BatchTableWrite:
...
class BatchTableUpdate:
+ def with_plan(self, plan: Plan) -> "BatchTableUpdate":
+ """Reuse complete plan_all file groups for row-ID updates and deletes.
+
+ Workers reauthorize against the table; a plan is metadata, not a grant.
+ Data manifests are not replanned. Deletes still read the captured index
+ manifest to preserve deletion vectors for every file in the bucket.
+ Assign all deletes for one partition/bucket to the same worker.
+ Selected row ranges, scores and streaming splits are not mutation
plans.
+ Other update operations reject this configuration before consuming
input.
+ """
+ ...
def with_blob_uri_reader_factory(
self, factory: Optional[UriReaderFactory]
) -> "BatchTableUpdate": ...
@@ -540,6 +582,17 @@ class TableUpdateByRowId:
def _abort(self) -> None: ...
class StreamTableUpdate:
+ def with_plan(self, plan: Plan) -> "StreamTableUpdate":
+ """Reuse complete plan_all file groups for row-ID updates and deletes.
+
+ Workers reauthorize against the table; a plan is metadata, not a grant.
+ Data manifests are not replanned. Deletes still read the captured index
+ manifest to preserve deletion vectors for every file in the bucket.
+ Assign all deletes for one partition/bucket to the same worker.
+ Selected row ranges, scores and streaming splits are not mutation
plans.
+ Other update operations reject this configuration before consuming
input.
+ """
+ ...
def with_blob_uri_reader_factory(
self, factory: Optional[UriReaderFactory]
) -> "StreamTableUpdate": ...
@@ -707,6 +760,11 @@ class BatchTableCommit:
) -> None:
"""Commit once, using the builder's overwrite configuration and batch
identifier."""
...
+ def check_row_id_conflicts(
+ self, messages: Iterable[CommitMessage], latest_snapshot_id: int,
check_from_snapshot: int
+ ) -> None:
+ """Check captured DML history without publishing; compactions are
ignored."""
+ ...
def truncate_table(self) -> None:
"""Truncate all data; shares the one-time guard with commit()."""
...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 7e56ceda..33fdd265 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -993,6 +993,8 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::table::PyTable>()?;
this.add_class::<crate::system_table::PySystemRead>()?;
this.add_class::<crate::index_build::PyGlobalIndexBuildBuilder>()?;
+ this.add_class::<crate::index_build::PyGlobalIndexBuildPlan>()?;
+ this.add_class::<crate::index_build::PyGlobalIndexBuildShard>()?;
this.add_class::<crate::index_build::PyGlobalIndexDropBuilder>()?;
this.add_class::<crate::full_text_search::PyFullTextSearchBuilder>()?;
this.add_class::<crate::full_text_search::PyFullTextScan>()?;
@@ -1019,6 +1021,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::read::PyTableScan>()?;
this.add_class::<crate::read::PyStreamTableScan>()?;
this.add_class::<crate::read::PyPlan>()?;
+ this.add_class::<crate::read::PyRowIdPlan>()?;
this.add_class::<crate::read::PyFileScanPlan>()?;
this.add_class::<crate::read::PyScanStats>()?;
this.add_class::<crate::read::PyTableRead>()?;
diff --git a/bindings/python/src/error.rs b/bindings/python/src/error.rs
index 4e9b277a..d5570ccd 100644
--- a/bindings/python/src/error.rs
+++ b/bindings/python/src/error.rs
@@ -32,9 +32,20 @@ pyo3::create_exception!(
"A catalog failure with code, resource_type, resource_name and message
attributes."
);
+pyo3::create_exception!(
+ pypaimon_rust,
+ RowIdExistenceConflict,
+ PyRuntimeError,
+ "A staged file no longer matches the current row-ID layout."
+);
+
pub fn register_module(py: Python<'_>, module: &Bound<'_, PyModule>) ->
PyResult<()> {
module.add("ForkSafetyError", py.get_type::<ForkSafetyError>())?;
- module.add("CatalogError", py.get_type::<CatalogError>())
+ module.add("CatalogError", py.get_type::<CatalogError>())?;
+ module.add(
+ "RowIdExistenceConflict",
+ py.get_type::<RowIdExistenceConflict>(),
+ )
}
/// Preserve the catalog error contract across the FFI without parsing display
strings.
@@ -134,6 +145,28 @@ pub fn to_py_err(err: paimon::Error) -> PyErr {
if err.is_process_fork_unsupported() {
return ForkSafetyError::new_err(err.to_string());
}
+ if let paimon::Error::RowIdExistenceConflict {
+ file_name,
+ first_row_id,
+ row_count,
+ bucket,
+ current_first_row_id,
+ current_row_count,
+ } = &err
+ {
+ return Python::attach(|py| {
+ let exception = RowIdExistenceConflict::new_err(err.to_string());
+ let value = exception.value(py);
+ value.setattr("file_name", file_name)?;
+ value.setattr("first_row_id", first_row_id)?;
+ value.setattr("row_count", row_count)?;
+ value.setattr("bucket", bucket)?;
+ value.setattr("current_first_row_id", current_first_row_id)?;
+ value.setattr("current_row_count", current_row_count)?;
+ Ok::<_, PyErr>(exception)
+ })
+ .unwrap_or_else(|error| error);
+ }
match err {
// Unimplemented scan semantics: distinct from malformed input so upper
// layers can catch NotImplementedError and decide on a fallback.
@@ -170,6 +203,124 @@ pub fn df_to_py_err(err:
datafusion::error::DataFusionError) -> PyErr {
mod tests {
use super::*;
+ #[test]
+ fn row_id_existence_error_preserves_structured_layout() {
+ Python::attach(|py| {
+ let error = to_py_err(paimon::Error::RowIdExistenceConflict {
+ file_name: "opaque '(4): file.parquet".into(),
+ first_row_id: Some(4),
+ row_count: 8,
+ bucket: 3,
+ current_first_row_id: None,
+ current_row_count: Some(9),
+ });
+ assert!(error.is_instance_of::<RowIdExistenceConflict>(py));
+ let value = error.value(py);
+ assert_eq!(
+ value
+ .getattr("file_name")
+ .unwrap()
+ .extract::<String>()
+ .unwrap(),
+ "opaque '(4): file.parquet"
+ );
+ assert_eq!(
+ value
+ .getattr("first_row_id")
+ .unwrap()
+ .extract::<i64>()
+ .unwrap(),
+ 4
+ );
+ assert_eq!(
+ value
+ .getattr("row_count")
+ .unwrap()
+ .extract::<i64>()
+ .unwrap(),
+ 8
+ );
+ assert_eq!(
+ value.getattr("bucket").unwrap().extract::<i32>().unwrap(),
+ 3
+ );
+ assert!(value.getattr("current_first_row_id").unwrap().is_none());
+ assert_eq!(
+ value
+ .getattr("current_row_count")
+ .unwrap()
+ .extract::<i64>()
+ .unwrap(),
+ 9
+ );
+ });
+ }
+
+ #[test]
+ fn row_id_existence_error_survives_python_worker_serialization() {
+ Python::attach(|py| {
+ let modules = py
+ .import("sys")
+ .unwrap()
+ .getattr("modules")
+ .unwrap()
+ .cast_into::<pyo3::types::PyDict>()
+ .unwrap();
+ let previous = modules.get_item("pypaimon_rust").unwrap();
+ let module = PyModule::new(py, "pypaimon_rust").unwrap();
+ register_module(py, &module).unwrap();
+ modules.set_item("pypaimon_rust", &module).unwrap();
+ let result = (|| -> PyResult<()> {
+ let error = to_py_err(paimon::Error::RowIdExistenceConflict {
+ file_name: "worker:staged.parquet".into(),
+ first_row_id: Some(4),
+ row_count: 8,
+ bucket: 3,
+ current_first_row_id: None,
+ current_row_count: Some(9),
+ });
+ let pickle = py.import("pickle")?;
+ let bytes = pickle.call_method1("dumps", (error.value(py),))?;
+ let restored = pickle.call_method1("loads", (bytes,))?;
+ assert!(restored.is_instance_of::<RowIdExistenceConflict>());
+ assert_eq!(
+ restored.getattr("file_name")?.extract::<String>()?,
+ "worker:staged.parquet"
+ );
+
assert_eq!(restored.getattr("first_row_id")?.extract::<i64>()?, 4);
+ assert_eq!(restored.getattr("row_count")?.extract::<i64>()?,
8);
+ assert_eq!(restored.getattr("bucket")?.extract::<i32>()?, 3);
+ assert!(restored.getattr("current_first_row_id")?.is_none());
+
assert_eq!(restored.getattr("current_row_count")?.extract::<i64>()?, 9);
+ Ok(())
+ })();
+ match previous {
+ Some(previous) => modules.set_item("pypaimon_rust",
previous).unwrap(),
+ None => modules.del_item("pypaimon_rust").unwrap(),
+ }
+ result.unwrap();
+ });
+ }
+
+ #[test]
+ fn unknown_commit_outcome_does_not_expose_a_rewritable_conflict() {
+ Python::attach(|py| {
+ let error = to_py_err(paimon::Error::UnexpectedError {
+ message: "Commit outcome may be unknown; retain prepared
files".into(),
+ source: Some(Box::new(paimon::Error::RowIdExistenceConflict {
+ file_name: "staged.parquet".into(),
+ first_row_id: Some(0),
+ row_count: 10,
+ bucket: 0,
+ current_first_row_id: None,
+ current_row_count: None,
+ })),
+ });
+ assert!(error.is_instance_of::<PyValueError>(py));
+ assert!(!error.is_instance_of::<RowIdExistenceConflict>(py));
+ });
+ }
+
#[test]
fn process_fork_error_has_distinct_python_type() {
Python::attach(|py| {
diff --git a/bindings/python/src/index_build.rs
b/bindings/python/src/index_build.rs
index ff5079c6..b9e016a6 100644
--- a/bindings/python/src/index_build.rs
+++ b/bindings/python/src/index_build.rs
@@ -21,10 +21,14 @@ use std::collections::HashMap;
use std::sync::Arc;
use paimon::spec::Predicate;
-use paimon::table::{GlobalIndexBuildBuilder, GlobalIndexDropBuilder, Table};
+use paimon::table::{
+ GlobalIndexBuildBuilder, GlobalIndexBuildPlan, GlobalIndexBuildShard,
GlobalIndexDropBuilder,
+ Table,
+};
use paimon_datafusion::runtime::runtime;
+use pyo3::exceptions::PyValueError;
use pyo3::prelude::*;
-use pyo3::types::PyDict;
+use pyo3::types::{PyBytes, PyDict};
use crate::error::to_py_err;
use crate::predicate::dict_to_table_predicate;
@@ -35,6 +39,84 @@ enum PyIndexColumns {
Columns(Vec<String>),
}
+#[pyclass(
+ name = "GlobalIndexBuildPlan",
+ module = "pypaimon_rust.datafusion",
+ frozen
+)]
+pub struct PyGlobalIndexBuildPlan {
+ inner: GlobalIndexBuildPlan,
+}
+
+#[pyclass(
+ name = "GlobalIndexBuildShard",
+ module = "pypaimon_rust.datafusion",
+ frozen
+)]
+pub struct PyGlobalIndexBuildShard {
+ inner: GlobalIndexBuildShard,
+}
+
+// Runtime handles stay in the worker's Table; pickle carries domain metadata.
+macro_rules! build_plan_pickle {
+ ($class:ident, { $($methods:item)* }) => {
+ #[pymethods]
+ impl $class {
+ $($methods)*
+
+ fn _serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py,
PyBytes>> {
+ let bytes = serde_json::to_vec(&self.inner)
+ .map_err(|error|
PyValueError::new_err(error.to_string()))?;
+ Ok(PyBytes::new(py, &bytes))
+ }
+
+ #[staticmethod]
+ fn _deserialize(state: &Bound<'_, PyBytes>) -> PyResult<Self> {
+ serde_json::from_slice(state.as_bytes())
+ .map(|inner| Self { inner })
+ .map_err(|error| {
+ PyValueError::new_err(format!(
+ "Invalid global index build metadata: {error}"
+ ))
+ })
+ }
+
+ fn __reduce__<'py>(
+ slf: &Bound<'py, Self>,
+ py: Python<'py>,
+ ) -> PyResult<(Py<PyAny>, (Py<PyBytes>,))> {
+ let bytes = slf.borrow()._serialize(py)?.unbind();
+ Ok((slf.get_type().getattr("_deserialize")?.unbind(),
(bytes,)))
+ }
+ }
+ };
+}
+
+build_plan_pickle!(PyGlobalIndexBuildPlan, {
+ fn snapshot_id(&self) -> Option<i64> {
+ self.inner.snapshot_id()
+ }
+
+ fn shards(&self) -> Vec<PyGlobalIndexBuildShard> {
+ self.inner
+ .shards()
+ .into_iter()
+ .map(|inner| PyGlobalIndexBuildShard { inner })
+ .collect()
+ }
+});
+
+build_plan_pickle!(PyGlobalIndexBuildShard, {
+ fn snapshot_id(&self) -> i64 {
+ self.inner.snapshot_id()
+ }
+
+ fn row_range(&self) -> (i64, i64) {
+ let range = self.inner.row_range();
+ (range.from(), range.to())
+ }
+});
+
#[pyclass(name = "GlobalIndexBuildBuilder", module =
"pypaimon_rust.datafusion")]
pub struct PyGlobalIndexBuildBuilder {
table: Arc<Table>,
@@ -225,6 +307,26 @@ impl PyGlobalIndexBuildBuilder {
Ok(messages.into_iter().map(PyCommitMessage::new).collect())
}
+ fn plan(&self, py: Python<'_>) -> PyResult<PyGlobalIndexBuildPlan> {
+ let builder = self.builder().map_err(to_py_err)?;
+ let inner = py
+ .detach(|| runtime().block_on(builder.plan()))
+ .map_err(to_py_err)?;
+ Ok(PyGlobalIndexBuildPlan { inner })
+ }
+
+ fn build_shard(
+ &self,
+ py: Python<'_>,
+ shard: &PyGlobalIndexBuildShard,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ let builder = self.table.new_global_index_build_builder();
+ let messages = py
+ .detach(|| runtime().block_on(builder.build_shard(&shard.inner)))
+ .map_err(to_py_err)?;
+ Ok(messages.into_iter().map(PyCommitMessage::new).collect())
+ }
+
fn execute(&self, py: Python<'_>) -> PyResult<usize> {
let builder = self.builder().map_err(to_py_err)?;
py.detach(|| runtime().block_on(builder.execute()))
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 05996167..3545ff39 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -1170,7 +1170,7 @@ impl PyRecordBatchReader {
#[pyclass(name = "Plan", module = "pypaimon_rust.datafusion")]
pub struct PyPlan {
- inner: paimon::table::Plan,
+ pub(crate) inner: paimon::table::Plan,
}
#[pyclass(name = "FileScanPlan", module = "pypaimon_rust.datafusion", frozen)]
@@ -1197,6 +1197,20 @@ impl From<paimon::table::Plan> for PyPlan {
#[pymethods]
impl PyPlan {
+ /// Restore captured planning metadata using typed splits. Execution
+ /// authorizes the transported plan independently.
+ #[new]
+ #[pyo3(signature = (snapshot_id, splits))]
+ fn new(snapshot_id: Option<i64>, splits: &Bound<'_, PyAny>) ->
PyResult<Self> {
+ let splits = splits
+ .try_iter()?
+ .map(|split| Ok(split?.extract::<PyRef<'_,
PySplit>>()?.inner.clone()))
+ .collect::<PyResult<Vec<_>>>()?;
+ Ok(Self {
+ inner: paimon::table::Plan::from_splits(snapshot_id, splits),
+ })
+ }
+
/// Snapshot selected by the scan, even when pruning produces no splits.
/// `None` means no snapshot was selected, including snapshot-free format
tables.
fn snapshot_id(&self) -> Option<i64> {
@@ -1223,6 +1237,46 @@ impl PyPlan {
}
}
+/// Row-tracking routing metadata; constructing it performs no I/O.
+#[pyclass(name = "RowIdPlan", module = "pypaimon_rust.datafusion")]
+pub struct PyRowIdPlan {
+ inner: paimon::table::RowIdPlan,
+}
+
+#[pymethods]
+impl PyRowIdPlan {
+ #[new]
+ fn new(plan: &PyPlan) -> PyResult<Self> {
+ Ok(Self {
+ inner:
paimon::table::RowIdPlan::from_plan(plan.inner.clone()).map_err(to_py_err)?,
+ })
+ }
+
+ fn snapshot_id(&self) -> Option<i64> {
+ self.inner.snapshot_id()
+ }
+
+ fn groups(&self) -> Vec<(i64, i64, PySplit)> {
+ self.inner
+ .groups()
+ .iter()
+ .map(|(first, last, split)| {
+ (
+ *first,
+ *last,
+ PySplit {
+ inner: split.clone(),
+ },
+ )
+ })
+ .collect()
+ }
+
+ fn plan(&self) -> PyPlan {
+ self.inner.plan().clone().into()
+ }
+}
+
#[pyclass(name = "Split", module = "pypaimon_rust.datafusion")]
pub struct PySplit {
pub(crate) inner: DataSplit,
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 299730b0..e5688b9b 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -616,12 +616,24 @@ fn arrow_upsert_batches(input: &Bound<'_, PyAny>) ->
PyResult<Vec<RecordBatch>>
struct UpdateContext {
inner: TableUpdate,
+ planned: bool,
table: Arc<Table>,
table_location: String,
commit_user: String,
}
impl UpdateContext {
+ // Reject a mismatched operation before consuming a Python input iterator
+ // or converting callbacks. Core independently enforces the same contract.
+ fn ensure_unplanned(&self) -> PyResult<()> {
+ if self.planned {
+ return Err(PyValueError::new_err(
+ "with_plan supports only row-ID update and delete operations",
+ ));
+ }
+ Ok(())
+ }
+
fn with_blob_uri_reader_factory(
&mut self,
factory: Option<Py<PyAny>>,
@@ -634,6 +646,7 @@ impl UpdateContext {
fn new(context: &WriteContext) -> PyResult<Self> {
Ok(Self {
+ planned: false,
inner: context
.table
.new_write_builder()
@@ -670,6 +683,7 @@ impl UpdateContext {
when_matched: &Bound<'_, PyAny>,
when_not_matched: &Bound<'_, PyAny>,
) -> PyResult<Vec<PyCommitMessage>> {
+ self.ensure_unplanned()?;
let source = match source {
Some(source) => match
source.extract::<PyRef<crate::table::PyTable>>() {
Ok(table) =>
paimon::table::MergeSource::Table(table.inner.clone()),
@@ -706,6 +720,7 @@ impl UpdateContext {
assignments: &Bound<'_, PyDict>,
read_columns: Option<Vec<String>>,
) -> PyResult<Vec<PyCommitMessage>> {
+ self.ensure_unplanned()?;
let predicate = predicate
.map(|predicate| dict_to_table_predicate(predicate,
self.table.schema(), true))
.transpose()?;
@@ -795,6 +810,7 @@ impl UpdateContext {
count: usize,
read_type: Option<&str>,
) -> PyResult<crate::shard_update::PyShardTableUpdator> {
+ self.ensure_unplanned()?;
let read_type = read_type
.map(|json| {
serde_json::from_str::<RowType>(json)
@@ -829,6 +845,7 @@ impl UpdateContext {
input: &Bound<'_, PyAny>,
keys: Vec<String>,
) -> PyResult<Vec<PyCommitMessage>> {
+ self.ensure_unplanned()?;
let batches = arrow_upsert_batches(input)?;
let messages = py
.detach(||
runtime().block_on(self.inner.upsert_by_arrow_with_key(batches, keys)))
@@ -845,6 +862,7 @@ impl UpdateContext {
py: Python<'_>,
predicate: Option<&Bound<'_, PyDict>>,
) -> PyResult<Vec<PyCommitMessage>> {
+ self.ensure_unplanned()?;
let predicate = predicate
.map(|predicate| dict_to_table_predicate(predicate,
self.table.schema(), true))
.transpose()?;
@@ -1064,6 +1082,18 @@ pub struct PyStreamTableUpdate {
#[pymethods]
impl PyStreamTableUpdate {
+ fn with_plan<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ plan: &crate::read::PyPlan,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ slf.context
+ .inner
+ .with_plan(plan.inner.clone())
+ .map_err(to_py_err)?;
+ slf.context.planned = true;
+ Ok(slf)
+ }
+
#[pyo3(signature = (factory))]
fn with_blob_uri_reader_factory(
mut slf: PyRefMut<'_, Self>,
@@ -1184,6 +1214,18 @@ impl PyStreamTableUpdate {
#[pymethods]
impl PyBatchTableUpdate {
+ fn with_plan<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ plan: &crate::read::PyPlan,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ slf.context
+ .inner
+ .with_plan(plan.inner.clone())
+ .map_err(to_py_err)?;
+ slf.context.planned = true;
+ Ok(slf)
+ }
+
#[pyo3(signature = (predicate))]
fn delete_by_predicate(
&self,
@@ -1615,6 +1657,27 @@ impl PyBatchTableCommit {
.map_err(to_callback_error)
}
+ /// Validate logical DML conflicts without publishing or deleting any
files.
+ fn check_row_id_conflicts(
+ &self,
+ py: Python<'_>,
+ messages: &Bound<'_, PyAny>,
+ latest_snapshot_id: i64,
+ check_from_snapshot: i64,
+ ) -> PyResult<()> {
+ let messages = self
+ .context
+ .messages(messages, "check_row_id_conflicts", false)?;
+ py.detach(|| {
+ runtime().block_on(self.context.inner.check_row_id_conflicts(
+ &messages,
+ latest_snapshot_id,
+ check_from_snapshot,
+ ))
+ })
+ .map_err(to_py_err)
+ }
+
fn truncate_table(&mut self, py: Python<'_>) -> PyResult<()> {
self.check_committed()?;
py.detach(|| runtime().block_on(self.context.inner.truncate_table()))
@@ -1784,6 +1847,186 @@ mod tests {
use paimon::spec::{BigIntType, IntType, Schema, TableSchema, VarCharType};
use pyo3::ffi::c_str;
+ #[test]
+ fn
captured_plan_rejects_other_operations_before_python_input_consumption() {
+ Python::attach(|py| {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "planned_binding_boundary"),
+ "memory:/planned_binding_boundary".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let locals = PyDict::new(py);
+ locals
+ .set_item(
+ "batch",
+ Py::new(py,
PyBatchWriteBuilder::new(table.clone())).unwrap(),
+ )
+ .unwrap();
+ locals
+ .set_item(
+ "stream",
+ Py::new(py, PyStreamWriteBuilder::new(table)).unwrap(),
+ )
+ .unwrap();
+ locals
+ .set_item("Plan", py.get_type::<crate::read::PyPlan>())
+ .unwrap();
+ py.run(c_str!(r#"
+class ExplodingInput:
+ def __iter__(self):
+ raise AssertionError('the supplied plan must reject input before
iteration')
+
+for builder, identifier in [(batch, None), (stream, 7)]:
+ update = builder.new_update()
+ assert update.with_plan(Plan(None, [])) is update
+ assert update.with_plan(Plan(None, [])) is update
+ options = {} if identifier is None else {'commit_identifier': identifier}
+ operations = [
+ lambda: update.upsert_by_arrow_with_key(ExplodingInput(), ['id'],
**options),
+ lambda: update.merge_into(ExplodingInput(), on=[], when_matched=[],
when_not_matched=[], **options),
+ ]
+ for operation in operations:
+ try:
+ operation()
+ except ValueError as error:
+ assert 'with_plan supports only row-ID' in str(error), str(error)
+ else:
+ raise AssertionError('captured plan was silently ignored')
+"#),Some(&locals),Some(&locals)).unwrap();
+ });
+ }
+
+ #[test]
+ fn row_id_conflict_check_does_not_publish_or_consume_batch_commit() {
+ use arrow::array::{Array, Int32Array, Int64Array};
+ use futures::TryStreamExt;
+
+ Python::attach(|py| {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Arc::new(Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "conflict_check_binding"),
+ "memory:/conflict_check_binding".into(),
+ TableSchema::new(0, &schema),
+ None,
+ ));
+ let messages = py.detach(|| {
+ runtime().block_on(async {
+ let mut writer =
table.new_write_builder().new_write().unwrap();
+ let batch = RecordBatch::try_new(
+
paimon::arrow::build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![Arc::new(Int32Array::from(vec![1, 2]))],
+ )
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ writer.close().await;
+ let updates = RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![0])) as
arrow::array::ArrayRef,
+ ),
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![9])) as
arrow::array::ArrayRef,
+ ),
+ ])
+ .unwrap();
+ table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .update_by_arrow_with_row_id(vec![updates])
+ .await
+ .unwrap()
+ })
+ });
+ let messages = messages
+ .into_iter()
+ .map(PyCommitMessage::new)
+ .collect::<Vec<_>>();
+ let builder = PyBatchWriteBuilder::new(table.clone());
+ let commit = Py::new(py, builder.new_commit().unwrap()).unwrap();
+ let locals = PyDict::new(py);
+ locals.set_item("commit", commit).unwrap();
+ locals.set_item("messages", messages).unwrap();
+ py.run(
+ c_str!("commit.check_row_id_conflicts(messages, 1, 1)"),
+ Some(&locals),
+ Some(&locals),
+ )
+ .unwrap();
+ let snapshot = py
+ .detach(||
runtime().block_on(table.snapshot_manager().get_latest_snapshot()))
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.id(), 1, "the independent check must not
publish");
+ py.run(
+ c_str!("commit.commit(messages)"),
+ Some(&locals),
+ Some(&locals),
+ )
+ .unwrap();
+ py.detach(|| {
+ runtime().block_on(async {
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(
+ snapshot.id(),
+ 2,
+ "the check must not consume the batch boundary"
+ );
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let mut values = batches
+ .iter()
+ .flat_map(|batch| {
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(values.null_count(), 0);
+ values.values().to_vec()
+ })
+ .collect::<Vec<_>>();
+ values.sort_unstable();
+ assert_eq!(values, [2, 9]);
+ })
+ });
+ });
+ }
+
#[test]
fn public_uri_factory_and_commit_user_setters_are_chainable() {
Python::attach(|py| {
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 01f877aa..f4eec826 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -31,6 +31,30 @@ pub enum Error {
/// A configured resource budget could not admit the requested reservation.
#[snafu(display("Paimon resource exhausted: {}", message))]
ResourceExhausted { message: String },
+ /// A staged row-ID range no longer matches the current data-file layout.
+ ///
+ /// `current_row_count` is present only for a DELETE whose physical file
+ /// still exists with different row-ID metadata. Logical DML conflicts and
+ /// uncertain publication outcomes must not be classified as this error.
+ #[snafu(display(
+ "{}",
+ row_id_existence_conflict_message(
+ file_name,
+ first_row_id,
+ row_count,
+ bucket,
+ current_first_row_id,
+ current_row_count
+ )
+ ))]
+ RowIdExistenceConflict {
+ file_name: String,
+ first_row_id: Option<i64>,
+ row_count: i64,
+ bucket: i32,
+ current_first_row_id: Option<i64>,
+ current_row_count: Option<i64>,
+ },
#[snafu(whatever, display("Paimon data invalid for {}: {:?}", message,
source))]
DataInvalid {
message: String,
@@ -151,6 +175,34 @@ pub enum Error {
},
}
+fn row_id_existence_conflict_message(
+ file_name: &str,
+ first_row_id: &Option<i64>,
+ row_count: &i64,
+ bucket: &i32,
+ current_first_row_id: &Option<i64>,
+ current_row_count: &Option<i64>,
+) -> String {
+ let first_row_id = first_row_id.map_or_else(|| "null".into(), |value|
value.to_string());
+ if let Some(current_row_count) = current_row_count {
+ let current_first_row_id =
+ current_first_row_id.map_or_else(|| "null".into(), |value|
value.to_string());
+ format!(
+ "Row ID existence conflict: DELETE for file '{file_name}'
references \
+ firstRowId={first_row_id}, rowCount={row_count} in bucket
{bucket}, but the current \
+ file has firstRowId={current_first_row_id},
rowCount={current_row_count}. The file \
+ may have been reassigned by a concurrent commit."
+ )
+ } else {
+ format!(
+ "Row ID existence conflict: file '{file_name}' references
firstRowId={first_row_id}, \
+ rowCount={row_count} in bucket {bucket}, but no matching file
exists in the current \
+ snapshot. The referenced file may have been rewritten by a
concurrent compaction \
+ or removed by an overwrite."
+ )
+ }
+}
+
impl From<opendal::Error> for Error {
fn from(source: opendal::Error) -> Self {
Error::from_opendal_with_context(source, "IO operation failed on
underlying storage")
diff --git a/crates/paimon/src/table/authorized_update_tests.rs
b/crates/paimon/src/table/authorized_update_tests.rs
index eb24ea09..6de0fdc4 100644
--- a/crates/paimon/src/table/authorized_update_tests.rs
+++ b/crates/paimon/src/table/authorized_update_tests.rs
@@ -26,6 +26,190 @@ const EVOLUTION_OPTIONS: &[(&str, &str)] = &[
("deletion-vectors.enabled", "true"),
];
+#[tokio::test]
+async fn
transported_row_id_plans_keep_the_rest_snapshot_after_concurrent_append() {
+ for delete in [false, true] {
+ let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+ fixture.append("seed", &[(1, "a", 10)], false).await;
+ fixture.state.deny_query.store(false, Ordering::SeqCst);
+ let plan = fixture
+ .table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(1));
+ fixture.append("concurrent", &[(2, "a", 20)], false).await;
+ let requests = fixture.state.query_requests.load(Ordering::SeqCst);
+ let mut update =
fixture.table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let messages = if delete {
+ update.delete_by_row_id(vec![0]).await.unwrap()
+ } else {
+ update
+ .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+ .await
+ .unwrap()
+ };
+ assert_eq!(messages[0].check_from_snapshot, Some(1));
+ assert!(fixture.state.query_requests.load(Ordering::SeqCst) >
requests);
+ fixture
+ .table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ assert_eq!(
+ fixture.stored_rows().await,
+ if delete {
+ vec![(2, "a".into(), 20)]
+ } else {
+ vec![(1, "a".into(), 99), (2, "a".into(), 20)]
+ }
+ );
+ fixture
+ .table
+ .snapshot_manager()
+ .delete_snapshot(1)
+ .await
+ .unwrap();
+ let error = if delete {
+ update.delete_by_row_id(vec![0]).await.unwrap_err()
+ } else {
+ update
+ .update_by_arrow_with_row_id(vec![row_update(0, 100)])
+ .await
+ .unwrap_err()
+ };
+ assert!(matches!(
+ error,
+ crate::Error::SnapshotNotExist { snapshot_id: 1 }
+ ));
+ }
+}
+
+#[tokio::test]
+async fn
transported_row_id_plans_refresh_whole_table_authorization_without_snapshot_files()
{
+ for delete in [false, true] {
+ for response in [None]
+ .into_iter()
+ .chain(restricted_responses().into_iter().map(Some))
+ {
+ let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+ fixture.append("seed", &[(1, "a", 10)], false).await;
+ fixture.state.deny_query.store(false, Ordering::SeqCst);
+ let plan = fixture
+ .table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let requests = fixture.state.query_requests.load(Ordering::SeqCst);
+ if let Some(response) = response {
+ *fixture.state.query_response.lock().await = response;
+ } else {
+ fixture.state.deny_query.store(true, Ordering::SeqCst);
+ }
+ let before = fixture
+ .table
+ .file_io()
+ .list_status_recursive(fixture.table.location())
+ .await
+ .unwrap()
+ .len();
+ let mut update =
fixture.table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let result = if delete {
+ update.delete_by_row_id(vec![0]).await
+ } else {
+ update
+ .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+ .await
+ };
+ assert!(
+ result.is_err(),
+ "captured grant must not survive a policy change"
+ );
+ assert!(fixture.state.query_requests.load(Ordering::SeqCst) >
requests);
+ assert_eq!(
+ fixture
+ .table
+ .file_io()
+ .list_status_recursive(fixture.table.location())
+ .await
+ .unwrap()
+ .len(),
+ before
+ );
+ }
+ }
+ let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+ fixture.append("seed", &[(1, "a", 10)], false).await;
+ fixture.state.deny_query.store(false, Ordering::SeqCst);
+ let plan = fixture
+ .table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ fixture
+ .table
+ .file_io()
+ .delete_dir(&format!("{}/snapshot", fixture.table.location()))
+ .await
+ .unwrap();
+ assert!(fixture
+ .table
+ .snapshot_manager()
+ .get_snapshot(1)
+ .await
+ .is_err());
+ let mut update = fixture.table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+ .await
+ .unwrap();
+ assert_eq!(messages[0].check_from_snapshot, Some(1));
+ assert_eq!(
+ update.delete_by_row_id(vec![0]).await.unwrap()[0].check_from_snapshot,
+ Some(1)
+ );
+}
+
+#[tokio::test]
+async fn
restricted_read_plans_cannot_be_repurposed_as_mutation_plans_after_policy_relaxes()
{
+ let fixture = RestTable::new(false, EVOLUTION_OPTIONS).await;
+ fixture.append("seed", &[(1, "a", 10)], false).await;
+ fixture.state.deny_query.store(false, Ordering::SeqCst);
+ *fixture.state.query_response.lock().await =
restricted_responses().remove(0);
+ let plan = fixture
+ .table
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap();
+ *fixture.state.query_response.lock().await = serde_json::json!({});
+ let mut update = fixture.table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let error = update
+ .update_by_arrow_with_row_id(vec![row_update(0, 99)])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("filtered or masked read plan"));
+ assert!(update
+ .delete_by_row_id(vec![0])
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("filtered or masked read plan"));
+}
+
fn row_update(row_id: i64, value: i32) -> RecordBatch {
RecordBatch::try_from_iter([
(
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 2df6a785..470f7aa0 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -39,7 +39,7 @@ use crate::table::index_file_path::IndexFileLocation;
use crate::table::source::{data_evolution_anchor_file,
is_data_evolution_normal_file};
use crate::table::stats_filter::group_by_overlapping_row_id;
use crate::table::Table;
-use crate::table::{DataSplit, RowRange};
+use crate::table::{DataSplit, Plan, RowRange};
use crate::Result;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch, StructArray};
use arrow_buffer::NullBuffer;
@@ -355,6 +355,7 @@ impl DataEvolutionWriter {
)?;
let file_matches = group_matched_rows_by_file(&self.matched_batches,
file_index)?;
self.validate_partition_values(file_index, &file_matches)?;
+ index.validate_live_rows(&file_matches).await?;
// 3. For each affected file: read original columns, apply updates,
write partial files
let mut writer = DataEvolutionPartialWriter::new(&self.table,
self.update_columns.clone())?;
@@ -598,7 +599,15 @@ impl DataEvolutionDeleteWriter {
}
#[must_use = "commit messages must be passed to TableCommit"]
- pub async fn prepare_commit(mut self) -> Result<Vec<CommitMessage>> {
+ pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> {
+ self.prepare_commit_inner(None).await
+ }
+
+ pub(super) async fn prepare_commit_with_plan(self, plan: &Plan) ->
Result<Vec<CommitMessage>> {
+ self.prepare_commit_inner(Some(plan)).await
+ }
+
+ async fn prepare_commit_inner(mut self, captured: Option<&Plan>) ->
Result<Vec<CommitMessage>> {
dedup_i64_in_place(&mut self.row_ids);
if self.row_ids.is_empty() {
self.table
@@ -610,9 +619,18 @@ impl DataEvolutionDeleteWriter {
}
// A row filter or mask cannot authorize a physical file mutation.
- let snapshot =
super::time_travel::resolve_snapshot(&self.table).await?;
- let scan_table =
self.table.copy_with_pinned_snapshot(snapshot.as_ref());
- let plan = scan_table.new_read_builder().new_scan().plan_all().await?;
+ let (snapshot, plan) = match captured {
+ Some(plan) => {
+ let (_, snapshot, plan) = authorize_row_id_plan(&self.table,
plan).await?;
+ (snapshot, plan)
+ }
+ None => {
+ let snapshot =
super::time_travel::resolve_snapshot(&self.table).await?;
+ let scan_table =
self.table.copy_with_pinned_snapshot(snapshot.as_ref());
+ let plan =
scan_table.new_read_builder().new_scan().plan_all().await?;
+ (snapshot, plan)
+ }
+ };
let snapshot = snapshot.ok_or_else(|| crate::Error::DataInvalid {
message: "No files with row tracking found in target table".into(),
source: None,
@@ -1249,6 +1267,11 @@ pub(super) struct RowIdFileIndex {
}
impl RowIdFileIndex {
+ pub(super) async fn load_plan(table: &Table, plan: &Plan) -> Result<Self> {
+ let (read_table, _, plan) = authorize_row_id_plan(table, plan).await?;
+ Self::from_splits(read_table, plan.splits())
+ }
+
pub(super) async fn load(table: &Table, snapshot_id: Option<i64>) ->
Result<Self> {
// Authorize before resolving an empty target. The original handle's
// session is retained when the read view pins the selected snapshot.
@@ -1362,11 +1385,136 @@ impl RowIdFileIndex {
.map(|&index| self.files[index].first_row_id)
.collect())
}
+
+ async fn validate_live_rows(&self, matches: &HashMap<usize,
Vec<MatchedRow>>) -> Result<()> {
+ for (&file_pos, rows) in matches {
+ let group = &self.files[file_pos];
+ let anchor = data_evolution_anchor_file(&group.files)?;
+ let Some(deletion_file) =
group.split.deletion_file_for_data_file(anchor) else {
+ continue;
+ };
+ let deletion_vector =
+ DeletionVectorFactory::read(self.read_table.file_io(),
deletion_file).await?;
+ let first = anchor.first_row_id.expect("validated row-ID anchor");
+ for row in rows {
+ let row_id = group.first_row_id + row.offset as i64;
+ if deletion_vector.is_deleted((row_id - first) as u64) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot update deleted _ROW_ID
{row_id}"),
+ source: None,
+ });
+ }
+ }
+ }
+ Ok(())
+ }
+}
+
+/// Validate transported planner metadata and mint a worker-local mutation
grant.
+/// The caller supplies complete planner groups; manifests are deliberately not
+/// re-read to rediscover metadata already captured by the coordinator.
+async fn authorize_row_id_plan(
+ table: &Table,
+ plan: &Plan,
+) -> Result<(Table, Option<Snapshot>, Plan)> {
+ let required = table.schema().core_options().query_auth_enabled()
+ || plan.splits().iter().any(DataSplit::query_auth_required);
+ let grant = table.authorize_read(required, None).await?;
+ if required && grant.is_none() {
+ return Err(super::query_auth::unsupported(
+ "transported row-ID plans require REST authorization",
+ ));
+ }
+ if grant
+ .as_ref()
+ .is_some_and(|grant| !grant.is_unrestricted_whole_table())
+ {
+ return Err(super::query_auth::unsupported(
+ "row-ID updates require an unrestricted whole-table grant",
+ ));
+ }
+ super::RowIdPlan::from_plan(plan.clone())?;
+ if plan.splits().iter().any(|split| {
+ split
+ .query_auth_response()
+ .is_some_and(|response| !response.is_unrestricted())
+ }) {
+ return Err(super::query_auth::unsupported(
+ "a filtered or masked read plan cannot authorize row-ID mutations",
+ ));
+ }
+ let snapshot = match plan.snapshot_id() {
+ None => None,
+ Some(id) => {
+ let snapshot = if let Some(snapshot) = table.travel_snapshot() {
+ snapshot.clone()
+ } else if table.rest_env().is_some() {
+ match table.snapshot_manager().get_latest_snapshot().await? {
+ Some(snapshot) if snapshot.id() == id => snapshot,
+ // A concurrent append does not invalidate the captured
+ // plan. Resolve its exact snapshot rather than replacing
+ // the worker's conflict baseline with the latest one.
+ _ => table.snapshot_manager().get_snapshot(id).await?,
+ }
+ } else {
+ table.snapshot_manager().get_snapshot(id).await?
+ };
+ if snapshot.id() != id || snapshot.schema_id() >
table.schema().id() {
+ return Err(crate::Error::DataInvalid {
+ message: "The row-ID plan does not match the table's
captured snapshot/schema"
+ .into(),
+ source: None,
+ });
+ }
+ Some(snapshot)
+ }
+ };
+ let options = table.schema().core_options();
+ let computer = if table.schema().partition_keys().is_empty() {
+ None
+ } else {
+ Some(PartitionComputer::new(
+ table.schema().partition_keys(),
+ table.schema().fields(),
+ options.partition_default_name(),
+ options.legacy_partition_name(),
+ )?)
+ };
+ for split in plan.splits() {
+ let expected = bucket_path(
+ &table.data_file_location(),
+ computer.as_ref(),
+ split.partition(),
+ split.bucket(),
+ )?;
+ if split.bucket_path() != expected {
+ return Err(crate::Error::DataInvalid {
+ message: "The row-ID split bucket path does not belong to the
target table".into(),
+ source: None,
+ });
+ }
+ for file in split.data_files() {
+ let (first, last) = file.row_id_range().expect("validated row-ID
range");
+ if first < 0
+ || snapshot.as_ref().is_none_or(|snapshot| {
+ file.schema_id > snapshot.schema_id()
+ || snapshot.next_row_id().is_some_and(|next| last >=
next)
+ })
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "The row-ID split contains file metadata outside
its snapshot".into(),
+ source: None,
+ });
+ }
+ }
+ }
+ let read_table = table.copy_with_pinned_snapshot(snapshot.as_ref());
+ Ok((read_table, snapshot, plan.clone().planned(grant)))
}
/// Dedicated column files can cross normal-file boundaries. They provide
/// values, but must not merge anchors or introduce otherwise absent row IDs.
-fn group_by_normal_row_id(files: Vec<DataFileMeta>) -> Vec<Vec<DataFileMeta>> {
+pub(super) fn group_by_normal_row_id(files: Vec<DataFileMeta>) ->
Vec<Vec<DataFileMeta>> {
let (normal, dedicated): (Vec<_>, Vec<_>) =
files.into_iter().partition(is_data_evolution_normal_file);
group_by_overlapping_row_id(normal)
diff --git a/crates/paimon/src/table/global_index_build_builder.rs
b/crates/paimon/src/table/global_index_build_builder.rs
index cb203619..ab46595c 100644
--- a/crates/paimon/src/table/global_index_build_builder.rs
+++ b/crates/paimon/src/table/global_index_build_builder.rs
@@ -21,6 +21,7 @@ use std::time::Instant;
#[cfg(feature = "fulltext")]
use super::full_text_index_build_builder::FullTextIndexBuildBuilder;
use super::global_index_build_common::{resolve_index_fields, IndexColumns};
+use super::global_index_build_plan::{GlobalIndexBuildPlan,
GlobalIndexBuildShard};
use super::global_index_types::{
normalize_global_index_type, normalize_queryable_global_index_type,
BTREE_GLOBAL_INDEX_TYPE,
FULL_TEXT_GLOBAL_INDEX_TYPE,
@@ -125,6 +126,47 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
Ok(self.prepare().await?.messages)
}
+ /// Capture independently executable vector-index shards without writing
files.
+ ///
+ /// The plan owns its snapshot, schema and build options. Workers execute a
+ /// shard rather than replanning the table or publishing its commit
messages.
+ pub async fn plan(&self) -> Result<GlobalIndexBuildPlan> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ self.table.ensure_not_branch_reference_for_write()?;
+ let index_type =
normalize_global_index_type(self.index_type.trim()).ok_or_else(|| {
+ Error::Unsupported {
+ message: format!("Unsupported global index type: {}",
self.index_type),
+ }
+ })?;
+ if !crate::vindex::is_vindex_index_type(index_type) {
+ return Err(Error::Unsupported {
+ message: "Global index build plans currently support only
vindex indexes".into(),
+ });
+ }
+ let columns = self.columns.as_ref().ok_or_else(|| Error::DataInvalid {
+ message: "Global index column is required".into(),
+ source: None,
+ })?;
+ let fields = resolve_index_fields(self.table, columns, index_type)?;
+ let mut builder = VindexIndexBuildBuilder::new(self.table, index_type);
+ builder
+ .with_index_column(fields[0].name())
+ .with_options(self.options.clone());
+ builder.partition_filter = self.partition_filter.clone();
+ builder.plan().await
+ }
+
+ /// Build only the files and physical row range captured by this shard.
+ /// Builder settings do not replace the plan's schema or index options.
+ pub async fn build_shard(&self, shard: &GlobalIndexBuildShard) ->
Result<Vec<CommitMessage>> {
+ Ok(shard
+ .build(self.table)
+ .await?
+ .into_iter()
+ .map(|(message, _)| message)
+ .collect())
+ }
+
/// Prepare and commit, rejecting changes to the snapshot used for the
build.
pub async fn execute(&self) -> Result<usize> {
let prepared = self.prepare().await?;
diff --git a/crates/paimon/src/table/global_index_build_common/vector.rs
b/crates/paimon/src/table/global_index_build_common/vector.rs
index a226597d..97ed5b71 100644
--- a/crates/paimon/src/table/global_index_build_common/vector.rs
+++ b/crates/paimon/src/table/global_index_build_common/vector.rs
@@ -26,7 +26,7 @@ use crate::table::{RowRange, Table};
use crate::{Error, Result};
use std::collections::HashMap;
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub(crate) struct VectorIndexShard {
pub(crate) partition: BinaryRow,
pub(crate) partition_bytes: Vec<u8>,
diff --git a/crates/paimon/src/table/global_index_build_plan.rs
b/crates/paimon/src/table/global_index_build_plan.rs
new file mode 100644
index 00000000..9f8536d2
--- /dev/null
+++ b/crates/paimon/src/table/global_index_build_plan.rs
@@ -0,0 +1,151 @@
+// 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.
+
+//! Snapshot-bound index tasks, independent of their scheduling framework.
+
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+use super::global_index_build_common::vector::VectorIndexShard;
+use super::vindex_index_build_builder::timing::VectorIndexBuildTiming;
+use super::vindex_index_build_builder::VindexIndexBuildBuilder;
+use super::{CommitMessage, RowRange, Table};
+use crate::spec::{Snapshot, TableSchema};
+use crate::{Error, Result};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct BuildContext {
+ location: String,
+ identifier: String,
+ branch: String,
+ schema: TableSchema,
+ snapshot: Option<Snapshot>,
+ column: String,
+ index_type: String,
+ options: HashMap<String, String>,
+}
+
+/// An immutable global-index build over one captured table snapshot.
+/// Contains metadata only: clients, FileIO and runtime handles stay in Table.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GlobalIndexBuildPlan {
+ context: BuildContext,
+ shards: Vec<VectorIndexShard>,
+}
+
+impl GlobalIndexBuildPlan {
+ pub(crate) fn new(
+ table: &Table,
+ snapshot: Option<Snapshot>,
+ column: String,
+ index_type: String,
+ options: HashMap<String, String>,
+ shards: Vec<VectorIndexShard>,
+ ) -> Self {
+ Self {
+ context: BuildContext {
+ location: table.location().to_string(),
+ identifier: table.identifier().full_name(),
+ branch: table.branch().to_string(),
+ schema: table.schema().clone(),
+ snapshot,
+ column,
+ index_type,
+ options,
+ },
+ shards,
+ }
+ }
+
+ pub fn snapshot_id(&self) -> Option<i64> {
+ self.context.snapshot.as_ref().map(Snapshot::id)
+ }
+
+ /// Each task carries the same captured schema, options and snapshot.
+ pub fn shards(&self) -> Vec<GlobalIndexBuildShard> {
+ self.shards
+ .iter()
+ .cloned()
+ .map(|shard| GlobalIndexBuildShard {
+ context: self.context.clone(),
+ shard,
+ })
+ .collect()
+ }
+}
+
+/// A complete physical row range and its files, executable on another worker.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GlobalIndexBuildShard {
+ context: BuildContext,
+ shard: VectorIndexShard,
+}
+
+impl GlobalIndexBuildShard {
+ pub fn snapshot_id(&self) -> i64 {
+ self.shard.snapshot_id
+ }
+
+ pub fn row_range(&self) -> RowRange {
+ RowRange::new(self.shard.row_range_start, self.shard.row_range_end)
+ }
+
+ pub(in crate::table) async fn build(
+ &self,
+ table: &Table,
+ ) -> Result<Option<(CommitMessage, Option<VectorIndexBuildTiming>)>> {
+ // A serialized task grants no authorization. Check the worker's table
+ // before replacing its schema with the captured read view.
+ table.schema().core_options().ensure_read_authorized()?;
+ table.ensure_not_branch_reference_for_write()?;
+ if table.location() != self.context.location
+ || table.identifier().full_name() != self.context.identifier
+ || table.branch() != self.context.branch
+ {
+ return Err(Error::DataInvalid {
+ message: "Global index build shard belongs to a different
table or branch".into(),
+ source: None,
+ });
+ }
+ let snapshot = self
+ .context
+ .snapshot
+ .as_ref()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Global index build shard requires a captured
snapshot".into(),
+ source: None,
+ })?;
+ if snapshot.id() != self.shard.snapshot_id
+ || self.shard.row_range_start < 0
+ || self.shard.row_range_end < self.shard.row_range_start
+ {
+ return Err(Error::DataInvalid {
+ message: "Invalid global index build shard snapshot or row
range".into(),
+ source: None,
+ });
+ }
+ let table = table
+ .copy_with_resolved_schema(self.context.schema.clone(),
&self.context.branch)?
+ .copy_with_pinned_snapshot(Some(snapshot));
+ let mut builder = VindexIndexBuildBuilder::new(&table,
&self.context.index_type);
+ builder
+ .with_index_column(&self.context.column)
+ .with_options(self.context.options.clone());
+ builder.build_planned_shard(&self.shard).await
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 87e89f56..1f95c5af 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -79,6 +79,7 @@ mod full_text_index_build_builder;
mod full_text_search_builder;
mod global_index_build_builder;
pub(crate) mod global_index_build_common;
+mod global_index_build_plan;
mod global_index_drop_builder;
mod global_index_freshness;
mod global_index_result;
@@ -222,6 +223,7 @@ pub use full_text_search_builder::{
};
use futures::stream::BoxStream;
pub use global_index_build_builder::GlobalIndexBuildBuilder;
+pub use global_index_build_plan::{GlobalIndexBuildPlan, GlobalIndexBuildShard};
pub use global_index_drop_builder::GlobalIndexDropBuilder;
pub use global_index_result::{
GlobalIndexResult, GlobalIndexResultReadBuilder, PrimaryKeySearchPosition,
@@ -263,7 +265,7 @@ pub use table_merge_into::{
pub use table_read::{AuditLogRead, TableRead};
pub use table_scan::{AuditLogScan, FileScanPlan, TableScan};
pub use table_update::TableUpdate;
-pub use table_update_by_row_id::TableUpdateByRowId;
+pub use table_update_by_row_id::{RowIdPlan, TableUpdateByRowId};
pub use table_write::TableWrite;
pub use tag_manager::TagManager;
pub use update_assignment::UpdateAssignment;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 2becb42f..01b292aa 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -1672,7 +1672,7 @@ impl Default for DataSplitBuilder {
/// Read plan: splits and the snapshot they were planned from.
///
/// Reference:
[org.apache.paimon.table.source.PlanImpl](https://github.com/apache/paimon/blob/release-1.3/paimon-core/src/main/java/org/apache/paimon/table/source/PlanImpl.java)
-#[derive(Debug)]
+#[derive(Debug, Clone)]
pub struct Plan {
splits: Vec<DataSplit>,
snapshot_id: Option<i64>,
@@ -1680,6 +1680,16 @@ pub struct Plan {
}
impl Plan {
+ /// Restore a captured plan from its snapshot and typed splits.
+ ///
+ /// This transports planning metadata; it does not authorize access to the
+ /// referenced files. Readers and mutation workers authorize independently.
+ pub fn from_splits(snapshot_id: Option<i64>, splits: Vec<DataSplit>) ->
Self {
+ let mut plan = Self::new(splits);
+ plan.snapshot_id = snapshot_id;
+ plan
+ }
+
pub fn new(splits: Vec<DataSplit>) -> Self {
let mut plan = Self {
splits,
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 87879367..72e7ec2b 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -289,6 +289,70 @@ impl TableCommit {
self
}
+ /// Check logical row-ID/column conflicts before rebasing staged updates.
+ ///
+ /// Both snapshot IDs are resolved through this table's snapshot manager.
+ /// The captured interval must not cross a schema change, and this
+ /// committer's resolved schema must still be current. A newer schema may
+ /// exist before the next data snapshot is published.
+ /// Like Java's `checkRowIdFromSnapshot(..., false)`, COMPACT snapshots are
+ /// excluded: the caller may rewrite physical layouts only after this check
+ /// has rejected concurrent logical writes. This method publishes nothing
+ /// and never deletes prepared files, including on an uncertain outcome.
+ pub async fn check_row_id_conflicts(
+ &self,
+ messages: &[CommitMessage],
+ latest_snapshot_id: i64,
+ check_from_snapshot: i64,
+ ) -> Result<()> {
+ if check_from_snapshot <= 0 || check_from_snapshot >
latest_snapshot_id {
+ return Err(crate::Error::DataInvalid {
+ message:
+ "Row-ID conflict checks require 0 < check-from snapshot <=
latest snapshot"
+ .into(),
+ source: None,
+ });
+ }
+ if Self::check_from_snapshot(messages)? != Some(check_from_snapshot) {
+ return Err(crate::Error::DataInvalid {
+ message: "Row-ID conflict check baseline must match the commit
messages".into(),
+ source: None,
+ });
+ }
+ super::write_builder::ensure_table_write_authorized(&self.table)?;
+ if !self.data_evolution_enabled {
+ return Err(crate::Error::Unsupported {
+ message: "Row-ID conflict checks require a data-evolution
table".into(),
+ });
+ }
+ reject_compact_increment(messages)?;
+ let latest = self
+ .snapshot_manager
+ .get_snapshot(latest_snapshot_id)
+ .await?;
+ let baseline = if check_from_snapshot == latest_snapshot_id {
+ latest.clone()
+ } else {
+ self.snapshot_manager
+ .get_snapshot(check_from_snapshot)
+ .await?
+ };
+ if baseline.schema_id() != latest.schema_id()
+ || self.latest_schema_id().await? != self.table.schema().id()
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "Row-ID conflict checks require an unchanged, current
table schema".into(),
+ source: None,
+ });
+ }
+ self.check_row_id_from_snapshot(
+ Some(&latest),
+ &self.messages_to_entries(messages),
+ Some(check_from_snapshot),
+ )
+ .await
+ }
+
/// Attach caller properties to snapshots published by this committer.
///
/// Like Java's ManifestCommittable properties, these describe the current
@@ -2756,6 +2820,32 @@ impl TableCommit {
delta_entries: &[ManifestEntry],
next_row_id: Option<i64>,
) -> Result<()> {
+ // File identity deliberately excludes row IDs. Check DELETE metadata
+ // before identifier-based merging can cancel a reassigned current ADD.
+ let deleted = delta_entries
+ .iter()
+ .filter(|entry| *entry.kind() == FileKind::Delete)
+ .map(|entry| (commit_file_identifier(entry), entry))
+ .collect::<HashMap<_, _>>();
+ for current in base_entries
+ .iter()
+ .filter(|entry| *entry.kind() == FileKind::Add)
+ {
+ if let Some(deleted) =
deleted.get(&commit_file_identifier(current)) {
+ if current.file().first_row_id != deleted.file().first_row_id
+ || current.file().row_count != deleted.file().row_count
+ {
+ return Err(crate::Error::RowIdExistenceConflict {
+ file_name: deleted.file().file_name.clone(),
+ first_row_id: deleted.file().first_row_id,
+ row_count: deleted.file().row_count,
+ bucket: deleted.bucket(),
+ current_first_row_id: current.file().first_row_id,
+ current_row_count: Some(current.file().row_count),
+ });
+ }
+ }
+ }
let Some(next_row_id) = next_row_id else {
return Ok(());
};
@@ -2818,15 +2908,13 @@ impl TableCommit {
entry.file().row_count,
);
if !existing_index.contains(&key) {
- return Err(crate::Error::DataInvalid {
- message: format!(
- "Row ID existence conflict: file '{}' references
first_row_id={}, row_count={} in bucket {}, but no matching file exists in the
current snapshot.",
- entry.file().file_name,
- first_row_id,
- entry.file().row_count,
- entry.bucket(),
- ),
- source: None,
+ return Err(crate::Error::RowIdExistenceConflict {
+ file_name: entry.file().file_name.clone(),
+ first_row_id: Some(first_row_id),
+ row_count: entry.file().row_count,
+ bucket: entry.bucket(),
+ current_first_row_id: None,
+ current_row_count: None,
});
}
}
@@ -3711,6 +3799,7 @@ mod tests {
include!("table_commit/property_tests.rs");
include!("table_commit/callback_tests.rs");
include!("table_commit/index_delete_tests.rs");
+ include!("table_commit/row_id_conflict_tests.rs");
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs
b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs
new file mode 100644
index 00000000..99e90a4f
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/row_id_conflict_tests.rs
@@ -0,0 +1,222 @@
+// 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.
+
+#[test]
+fn row_id_delete_must_match_the_current_file_layout() {
+ let commit = setup_data_evolution_commit(&test_file_io(),
"memory:/row_id_delete_layout");
+ let mut file = test_data_file("data-reassigned.parquet", 10);
+ file.first_row_id = Some(10);
+ let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1,
file.clone(), 2);
+ file.first_row_id = Some(0);
+ let stale_delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1,
file, 2);
+ let error = commit
+ .check_row_id_existence(&[current], &[stale_delete], Some(20))
+ .expect_err("a DELETE must not cancel the current file after row-ID
reassignment");
+ assert!(error.to_string().contains("Row ID existence conflict"));
+ assert!(matches!(
+ error,
+ crate::Error::RowIdExistenceConflict {
+ first_row_id: Some(0),
+ row_count: 10,
+ bucket: 0,
+ current_first_row_id: Some(10),
+ current_row_count: Some(10),
+ ..
+ }
+ ));
+}
+
+#[test]
+fn row_id_existence_preserves_missing_range_details() {
+ let commit = setup_data_evolution_commit(&test_file_io(),
"memory:/missing_row_id_range");
+ let mut file = test_data_file("opaque '(4): file.parquet", 4);
+ file.first_row_id = Some(7);
+ let entry = ManifestEntry::new(FileKind::Add, vec![], 3, -1, file, 2);
+ let error = commit
+ .check_row_id_existence(&[], &[entry], Some(20))
+ .unwrap_err();
+ assert!(matches!(error, crate::Error::RowIdExistenceConflict {
+ ref file_name, first_row_id: Some(7), row_count: 4, bucket: 3,
+ current_first_row_id: None, current_row_count: None
+ } if file_name == "opaque '(4): file.parquet"));
+}
+
+#[test]
+fn row_id_delete_metadata_check_does_not_require_next_row_id() {
+ let commit = setup_data_evolution_commit(&test_file_io(),
"memory:/delete_without_counter");
+ for (current_id, current_count) in [(None, 10), (Some(0), 11)] {
+ let mut file = test_data_file("same.parquet", current_count);
+ file.first_row_id = current_id;
+ let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1, file,
2);
+ let mut old = test_data_file("same.parquet", 10);
+ old.first_row_id = Some(0);
+ let old = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, old, 2);
+ assert!(
+ matches!(commit.check_row_id_existence(&[current], &[old], None),
+ Err(crate::Error::RowIdExistenceConflict { current_first_row_id,
current_row_count: Some(count), .. })
+ if current_first_row_id == current_id && count == current_count)
+ );
+ }
+ let file = test_data_file("ordinary.parquet", 10);
+ let current = ManifestEntry::new(FileKind::Add, vec![], 0, -1,
file.clone(), 2);
+ let delete = ManifestEntry::new(FileKind::Delete, vec![], 0, -1, file, 2);
+ commit
+ .check_row_id_existence(&[current], &[delete], None)
+ .unwrap();
+ commit.check_row_id_existence(&[], &[], None).unwrap();
+}
+
+fn row_id_check_message(name: &str, column: &str) -> CommitMessage {
+ let mut file = test_data_file(name, 10);
+ file.first_row_id = Some(0);
+ file.file_source = Some(0);
+ file.write_cols = Some(vec![column.into()]);
+ let mut message = CommitMessage::new(EMPTY_SERIALIZED_ROW.clone(), 0,
vec![file]);
+ message.check_from_snapshot = Some(1);
+ message
+}
+
+async fn row_id_check_history(path: &str) -> TableCommit {
+ let io = test_file_io();
+ setup_dirs(&io, path).await;
+ let commit = setup_data_evolution_commit(&io, path);
+ let mut file = test_data_file("base.parquet", 10);
+ file.file_source = Some(0);
+ commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ EMPTY_SERIALIZED_ROW.clone(),
+ 0,
+ vec![file],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+ commit
+
.commit_with_identifier(vec![row_id_check_message("concurrent.parquet",
"name")], 2)
+ .await
+ .unwrap();
+ commit
+}
+
+#[tokio::test]
+async fn
standalone_row_id_check_rejects_logical_changes_without_deleting_staged_files()
{
+ let commit = row_id_check_history("memory:/standalone_row_id_check").await;
+ let io = commit.table.file_io();
+ let path = "memory:/standalone_row_id_check/bucket-0/staged.parquet";
+ io.new_output(path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared"))
+ .await
+ .unwrap();
+ let error = commit
+ .check_row_id_conflicts(&[row_id_check_message("staged.parquet",
"name")], 2, 1)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, crate::Error::DataInvalid { .. }));
+ assert!(error.to_string().contains("multiple MERGE INTO"));
+ assert!(
+ io.exists(path).await.unwrap(),
+ "a failed check must retain prepared files"
+ );
+ assert_eq!(
+ commit
+ .snapshot_manager
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(2)
+ );
+ commit
+ .check_row_id_conflicts(&[row_id_check_message("id-only.parquet",
"id")], 2, 1)
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn standalone_row_id_check_skips_compaction_history() {
+ let commit =
row_id_check_history("memory:/standalone_compact_history").await;
+ // Replace only the fixture's commit kind. The same overlapping delta must
+ // be ignored when it describes physical compaction rather than a DML
write.
+ let snapshot = commit.snapshot_manager.get_snapshot(2).await.unwrap();
+ let mut json = serde_json::to_value(snapshot).unwrap();
+ json["commitKind"] = "COMPACT".into();
+ let path = format!("{}/snapshot-2",
commit.snapshot_manager.snapshot_dir());
+ commit
+ .table
+ .file_io()
+ .new_output(&path)
+ .unwrap()
+ .write(serde_json::to_vec(&json).unwrap().into())
+ .await
+ .unwrap();
+ commit
+ .check_row_id_conflicts(&[row_id_check_message("staged.parquet",
"name")], 2, 1)
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn
standalone_row_id_check_validates_snapshot_baselines_and_current_schema() {
+ let commit =
row_id_check_history("memory:/standalone_check_validation").await;
+ let message = row_id_check_message("staged.parquet", "name");
+ for (latest, baseline) in [(2, 0), (1, 2), (0, 1), (2, -1), (2, 2)] {
+ assert!(matches!(
+ commit
+ .check_row_id_conflicts(std::slice::from_ref(&message),
latest, baseline)
+ .await,
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ }
+ assert!(matches!(
+ commit.check_row_id_conflicts(&[], 2, 1).await,
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ assert!(commit
+ .check_row_id_conflicts(std::slice::from_ref(&message), 3, 1)
+ .await
+ .is_err());
+ save_schema(
+ commit.table.file_io(),
+ "memory:/standalone_check_validation",
+ 1,
+ )
+ .await;
+ let error = commit
+ .check_row_id_conflicts(&[message], 2, 1)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("current table schema"));
+}
+
+#[tokio::test]
+async fn
standalone_row_id_check_fails_closed_for_missing_baseline_and_assembled_authorization()
{
+ let commit =
row_id_check_history("memory:/standalone_missing_baseline").await;
+ commit.snapshot_manager.delete_snapshot(1).await.unwrap();
+ assert!(commit
+ .check_row_id_conflicts(&[row_id_check_message("staged.parquet",
"name")], 2, 1)
+ .await
+ .is_err());
+ let guarded = TableCommit::new(crate::table::query_auth_table(),
"guarded".into());
+ let error = guarded
+ .check_row_id_conflicts(&[row_id_check_message("staged.parquet",
"name")], 2, 1)
+ .await
+ .unwrap_err();
+ assert!(matches!(error, crate::Error::Unsupported { .. }));
+ assert!(error.to_string().contains("query-auth.enabled"));
+}
diff --git a/crates/paimon/src/table/table_update.rs
b/crates/paimon/src/table/table_update.rs
index 1eac4063..d9257a48 100644
--- a/crates/paimon/src/table/table_update.rs
+++ b/crates/paimon/src/table/table_update.rs
@@ -56,6 +56,7 @@ pub struct TableUpdate {
table: Table,
commit_user: String,
update_cols: Option<Vec<String>>,
+ plan: Option<super::Plan>,
blob_uri_reader_factory: Option<std::sync::Arc<dyn
crate::io::UriReaderFactory>>,
}
@@ -65,10 +66,39 @@ impl TableUpdate {
table: table.clone(),
commit_user,
update_cols: None,
+ plan: None,
blob_uri_reader_factory: None,
}
}
+ /// Reuse a captured `plan_all` result for row-ID updates and deletes.
+ ///
+ /// Each split must contain complete logical file groups, including their
+ /// partial-column files and deletion vectors. Splits may be distributed
+ /// among workers; the supplied row IDs must belong to that worker's
groups.
+ /// Arbitrarily trimmed or incremental splits are not mutation plans.
+ /// Workers authorize the whole table again and retain the captured
snapshot
+ /// as the conflict baseline without replanning its data manifests. Deletes
+ /// still read the captured index manifest to preserve all deletion vectors
+ /// when replacing a bucket's shared index file. Assign all deletes for one
+ /// partition/bucket to the same worker.
+ /// Predicate, key-based, MERGE and shard operations reject this setting;
+ /// use a separately configured updater for those operations.
+ pub fn with_plan(&mut self, plan: super::Plan) -> crate::Result<&mut Self>
{
+ super::RowIdPlan::from_plan(plan.clone())?;
+ self.plan = Some(plan);
+ Ok(self)
+ }
+
+ fn ensure_unplanned(&self) -> crate::Result<()> {
+ if self.plan.is_some() {
+ return Err(crate::Error::Unsupported {
+ message: "with_plan supports only row-ID update and delete
operations".into(),
+ });
+ }
+ Ok(())
+ }
+
/// Configure readers for Blob input references. The same factory supplies
/// matched updates and appended rows in upsert and MERGE operations.
pub fn with_blob_uri_reader_factory(
@@ -120,6 +150,7 @@ impl TableUpdate {
shard_count: usize,
read_type: Option<Vec<DataField>>,
) -> crate::Result<super::ShardTableUpdater> {
+ self.ensure_unplanned()?;
let mut updater = super::ShardTableUpdater::new(
&self.table,
read_type,
@@ -170,7 +201,10 @@ impl TableUpdate {
/// Create a row-ID updater sharing one snapshot across per-call columns.
pub async fn new_update_by_row_id(&self) ->
crate::Result<TableUpdateByRowId> {
- let mut writer = TableUpdateByRowId::new(&self.table).await?;
+ let mut writer = match &self.plan {
+ Some(plan) => TableUpdateByRowId::with_plan(&self.table,
plan).await?,
+ None => TableUpdateByRowId::new(&self.table).await?,
+ };
writer.with_blob_uri_reader_factory(self.blob_uri_reader_factory.clone());
Ok(writer)
}
@@ -233,6 +267,7 @@ impl TableUpdate {
assignments: Vec<(String, super::UpdateAssignment)>,
read_columns: Vec<String>,
) -> crate::Result<Vec<CommitMessage>> {
+ self.ensure_unplanned()?;
super::table_update_predicate::update(
&self.table,
predicate,
@@ -261,6 +296,7 @@ impl TableUpdate {
batches: Vec<RecordBatch>,
upsert_keys: Vec<String>,
) -> crate::Result<Vec<CommitMessage>> {
+ self.ensure_unplanned()?;
let mut writer = super::table_upsert::TableUpsert::new(
&self.table,
self.commit_user.clone(),
@@ -286,6 +322,7 @@ impl TableUpdate {
when_matched: Vec<super::WhenMatched>,
when_not_matched: Vec<super::WhenNotMatched>,
) -> crate::Result<Vec<CommitMessage>> {
+ self.ensure_unplanned()?;
super::table_merge_into::merge_into(
&self.table,
&self.commit_user,
@@ -311,6 +348,7 @@ impl TableUpdate {
&self,
predicate: Option<crate::spec::Predicate>,
) -> crate::Result<Vec<CommitMessage>> {
+ self.ensure_unplanned()?;
super::table_update_delete::delete(&self.table, predicate).await
}
@@ -323,10 +361,17 @@ impl TableUpdate {
// ordinary write builder continues rejecting historical handles.
let mut writer = super::DataEvolutionDeleteWriter::new(&self.table)?;
writer.add_row_ids(row_ids)?;
- writer.prepare_commit().await
+ match &self.plan {
+ Some(plan) => writer.prepare_commit_with_plan(plan).await,
+ None => writer.prepare_commit().await,
+ }
}
}
fn same_columns(left: &[String], right: &[String]) -> bool {
left.len() == right.len() && left.iter().all(|column|
right.contains(column))
}
+
+#[cfg(test)]
+#[path = "table_update_plan_tests.rs"]
+mod plan_tests;
diff --git a/crates/paimon/src/table/table_update_by_row_id.rs
b/crates/paimon/src/table/table_update_by_row_id.rs
index afec5f00..acf5eba5 100644
--- a/crates/paimon/src/table/table_update_by_row_id.rs
+++ b/crates/paimon/src/table/table_update_by_row_id.rs
@@ -17,8 +17,8 @@
//! Row-ID updates sharing one target snapshot and file index.
-use super::data_evolution_writer::RowIdFileIndex;
-use crate::table::{CommitMessage, DataEvolutionWriter, Table};
+use super::data_evolution_writer::{group_by_normal_row_id, RowIdFileIndex};
+use crate::table::{CommitMessage, DataEvolutionWriter, DataSplit, Plan, Table};
use arrow_array::RecordBatch;
use std::collections::{HashMap, HashSet};
@@ -29,6 +29,99 @@ fn invalid(message: impl Into<String>) -> crate::Error {
}
}
+/// Routing metadata for complete row-tracked file groups in a captured plan.
+///
+/// Constructing this object performs no I/O and grants no file access. The
+/// reader or mutation executing the retained plan must authorize
independently.
+#[derive(Debug, Clone)]
+pub struct RowIdPlan {
+ plan: Plan,
+ groups: Vec<(i64, i64, DataSplit)>,
+}
+
+impl RowIdPlan {
+ pub fn from_plan(plan: Plan) -> crate::Result<Self> {
+ if plan.snapshot_id().is_none() && !plan.splits().is_empty() {
+ return Err(invalid("A row-ID plan requires its captured snapshot
ID"));
+ }
+ let mut groups = Vec::new();
+ for split in plan.splits() {
+ if Some(split.snapshot_id()) != plan.snapshot_id()
+ || split.is_streaming()
+ || split.row_ranges().is_some()
+ || split.scores().is_some()
+ {
+ return Err(invalid(
+ "A row-ID plan requires full snapshot file groups without
row selection",
+ ));
+ }
+ if split.data_files().is_empty() {
+ return Err(invalid("A row-ID split must contain data files"));
+ }
+ if split
+ .data_deletion_files()
+ .is_some_and(|files| files.len() != split.data_files().len())
+ {
+ return Err(invalid(
+ "Row-ID deletion metadata must match the captured data
files",
+ ));
+ }
+ let mut names = HashSet::new();
+ for file in split.data_files() {
+ if file.row_id_range().is_none() ||
!names.insert(&file.file_name) {
+ return Err(invalid(
+ "A row-ID split contains invalid or duplicate file
metadata",
+ ));
+ }
+ }
+ let file_groups =
group_by_normal_row_id(split.data_files().to_vec());
+ if file_groups.is_empty() {
+ return Err(invalid("A row-ID split requires a normal anchor
file"));
+ }
+ let mut grouped = HashSet::new();
+ for files in file_groups {
+ let anchor =
super::source::data_evolution_anchor_file(&files)?;
+ let (first, last) = anchor
+ .row_id_range()
+ .ok_or_else(|| invalid("Invalid row-ID anchor range"))?;
+ for file in &files {
+ grouped.insert(file.file_name.clone());
+ if super::source::is_data_evolution_normal_file(file)
+ && file.row_id_range() != Some((first, last))
+ {
+ return Err(invalid(
+ "Overlapping row-ID anchors must have identical
ranges",
+ ));
+ }
+ }
+ groups.push((first, last,
split.for_data_evolution_files(files)));
+ }
+ if grouped.len() != names.len() {
+ return Err(invalid(
+ "A row-ID split contains a dedicated file without its
normal anchor",
+ ));
+ }
+ }
+ groups.sort_by_key(|(first, _, _)| *first);
+ if groups.windows(2).any(|pair| pair[0].1 >= pair[1].0) {
+ return Err(invalid("A row-ID plan contains overlapping file
groups"));
+ }
+ Ok(Self { plan, groups })
+ }
+
+ pub fn snapshot_id(&self) -> Option<i64> {
+ self.plan.snapshot_id()
+ }
+
+ pub fn groups(&self) -> &[(i64, i64, DataSplit)] {
+ &self.groups
+ }
+
+ pub fn plan(&self) -> &Plan {
+ &self.plan
+ }
+}
+
/// Update columns selected per input table, as in PyPaimon.
///
/// Each call to `update_columns` stages its files immediately and returns all
@@ -47,6 +140,10 @@ impl TableUpdateByRowId {
Self::with_index(table, RowIdFileIndex::load(table, None).await?)
}
+ pub(super) async fn with_plan(table: &Table, plan: &Plan) ->
crate::Result<Self> {
+ Self::with_index(table, RowIdFileIndex::load_plan(table, plan).await?)
+ }
+
pub(super) fn with_index(table: &Table, index: RowIdFileIndex) ->
crate::Result<Self> {
// Validate table layout without fixing any update columns.
let _ = DataEvolutionWriter::new(table, Vec::new())?;
diff --git a/crates/paimon/src/table/table_update_plan_tests.rs
b/crates/paimon/src/table/table_update_plan_tests.rs
new file mode 100644
index 00000000..b4fb6ddd
--- /dev/null
+++ b/crates/paimon/src/table/table_update_plan_tests.rs
@@ -0,0 +1,418 @@
+// 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.
+
+use super::*;
+use crate::catalog::Identifier;
+use crate::io::FileIOBuilder;
+use crate::spec::{DataType, IntType, Schema, TableSchema};
+use crate::table::{DataSplitBuilder, Plan, RowRange};
+use arrow_array::{ArrayRef, Int32Array, Int64Array};
+use std::sync::Arc;
+
+async fn table(name: &str) -> Table {
+ let schema = Schema::builder()
+ .column("value", DataType::Int(IntType::new()))
+ .option("bucket", "-1")
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("deletion-vectors.enabled", "true")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", name),
+ format!("memory:/{name}"),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ for value in [10, 20] {
+ append(&table, value).await;
+ }
+ table
+}
+
+async fn append(table: &Table, value: i32) {
+ let batch = RecordBatch::try_from_iter(vec![(
+ "value",
+ Arc::new(Int32Array::from(vec![value])) as ArrayRef,
+ )])
+ .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();
+}
+
+fn update_batch(row_id: i64) -> RecordBatch {
+ RecordBatch::try_from_iter(vec![
+ (
+ "_ROW_ID",
+ Arc::new(Int64Array::from(vec![row_id])) as ArrayRef,
+ ),
+ ("value", Arc::new(Int32Array::from(vec![99])) as ArrayRef),
+ ])
+ .unwrap()
+}
+
+#[tokio::test]
+async fn
planned_updates_and_deletes_keep_the_captured_baseline_without_manifest_rescan()
{
+ let table = table("planned_mutation_baseline").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ append(&table, 30).await;
+ table
+ .file_io()
+ .delete_dir(&format!("{}/manifest", table.location()))
+ .await
+ .unwrap();
+
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let messages = update
+ .update_by_arrow_with_row_id(vec![update_batch(0)])
+ .await
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].check_from_snapshot, Some(2));
+ let deletes = update.delete_by_row_id(vec![1]).await.unwrap();
+ assert_eq!(deletes.len(), 1);
+ assert_eq!(deletes[0].check_from_snapshot, Some(2));
+ let error = update
+ .update_by_arrow_with_row_id(vec![update_batch(2)])
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("No file found for _ROW_ID 2"));
+}
+
+#[tokio::test]
+async fn
planned_mutations_reject_foreign_paths_and_selected_splits_before_writing() {
+ let table = table("planned_mutation_validation").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let split = &plan.splits()[0];
+ for (path, snapshot_id, ranges, streaming) in [
+ ("memory:/another/bucket-0", 2, None, false),
+ (split.bucket_path(), 1, None, false),
+ (
+ split.bucket_path(),
+ 2,
+ Some(vec![RowRange::new(0, 0)]),
+ false,
+ ),
+ (split.bucket_path(), 2, None, true),
+ ] {
+ let mut builder = DataSplitBuilder::new()
+ .with_snapshot(snapshot_id)
+ .with_partition(split.partition().clone())
+ .with_bucket(split.bucket())
+ .with_bucket_path(path.to_string())
+ .with_data_files(split.data_files().to_vec())
+ .with_streaming(streaming);
+ if let Some(ranges) = ranges {
+ builder = builder.with_row_ranges(ranges);
+ }
+ let split = builder.build().unwrap();
+ let mut update = table.new_write_builder().new_update().unwrap();
+ if update
+ .with_plan(Plan::from_splits(Some(2), vec![split]))
+ .is_err()
+ {
+ continue;
+ }
+ assert!(update.new_update_by_row_id().await.is_err());
+ assert!(update.delete_by_row_id(vec![0]).await.is_err());
+ }
+}
+
+#[tokio::test]
+async fn planned_update_rejects_captured_deleted_rows_but_allows_live_rows() {
+ let table = table("planned_update_deletion_vector").await;
+ let update = table.new_write_builder().new_update().unwrap();
+ let deletes = update.delete_by_row_id(vec![0]).await.unwrap();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletes)
+ .await
+ .unwrap();
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ assert!(plan
+ .splits()
+ .iter()
+ .any(|split| split.data_deletion_files().is_some()));
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ let mut writer = update.new_update_by_row_id().await.unwrap();
+ let error = writer
+ .update_columns(vec![update_batch(0)], vec!["value".into()])
+ .await
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains("Cannot update deleted _ROW_ID 0"));
+ assert!(writer.commit_messages().is_empty());
+ let messages = writer
+ .update_columns(vec![update_batch(1)], vec!["value".into()])
+ .await
+ .unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].check_from_snapshot, Some(3));
+}
+
+#[tokio::test]
+async fn planned_delete_preserves_unselected_bucket_deletion_vectors() {
+ use futures::TryStreamExt;
+
+ let table = table("planned_delete_bucket_deletion_vectors").await;
+ let deletes = table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![0])
+ .await
+ .unwrap();
+ let previous_index = deletes[0].new_index_files[0].clone();
+ let previous_anchor = previous_index
+ .deletion_vectors_ranges
+ .as_ref()
+ .unwrap()
+ .keys()
+ .next()
+ .unwrap()
+ .clone();
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(deletes)
+ .await
+ .unwrap();
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let routing = crate::table::RowIdPlan::from_plan(plan.clone()).unwrap();
+ // Route only the second anchor. The first anchor's existing deletion
+ // vector must survive replacement of the bucket's shared index file.
+ let selected = routing
+ .groups()
+ .iter()
+ .find(|(first, _, _)| *first == 1)
+ .unwrap()
+ .2
+ .clone();
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update
+ .with_plan(Plan::from_splits(plan.snapshot_id(), vec![selected]))
+ .unwrap();
+ let messages = update.delete_by_row_id(vec![1]).await.unwrap();
+ assert_eq!(messages[0].check_from_snapshot, Some(3));
+ assert_eq!(messages[0].deleted_index_files, vec![previous_index]);
+ let ranges = messages[0].new_index_files[0]
+ .deletion_vectors_ranges
+ .as_ref()
+ .unwrap();
+ assert_eq!(ranges.len(), 2);
+ assert_eq!(ranges[&previous_anchor].cardinality, Some(1));
+ table
+ .new_write_builder()
+ .new_commit()
+ .commit(messages)
+ .await
+ .unwrap();
+ 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();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(), 0);
+}
+
+#[tokio::test]
+async fn row_id_plan_routes_complete_groups_and_rejects_ambiguous_ranges() {
+ let table = table("planned_row_id_routing").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let routing = crate::table::RowIdPlan::from_plan(plan.clone()).unwrap();
+ assert_eq!(routing.snapshot_id(), Some(2));
+ assert_eq!(
+ routing
+ .groups()
+ .iter()
+ .map(|(first, last, _)| (*first, *last))
+ .collect::<Vec<_>>(),
+ vec![(0, 0), (1, 1)]
+ );
+ let selected = routing.groups()[0].2.clone();
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update
+ .with_plan(Plan::from_splits(Some(2), vec![selected.clone()]))
+ .unwrap();
+ assert!(update
+ .delete_by_row_id(vec![1])
+ .await
+ .unwrap_err()
+ .to_string()
+ .contains("No file found for _ROW_ID 1"));
+ assert!(crate::table::RowIdPlan::from_plan(Plan::from_splits(
+ Some(2),
+ vec![selected.clone(), selected]
+ ))
+ .unwrap_err()
+ .to_string()
+ .contains("overlapping file groups"));
+ assert_eq!(routing.plan().splits(), plan.splits());
+}
+
+#[tokio::test]
+async fn row_id_plan_keeps_full_partial_columns_and_dedicated_subranges() {
+ let table = table("planned_row_id_partial_ranges").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let split = &plan.splits()[0];
+ let mut base = split.data_files()[0].clone();
+ base.first_row_id = Some(0);
+ base.row_count = 6;
+ base.max_sequence_number = 0;
+ let mut partial = base.clone();
+ partial.file_name = "partial-column.parquet".into();
+ partial.max_sequence_number = 1;
+ partial.write_cols = Some(vec!["value".into()]);
+ let mut left = partial.clone();
+ left.file_name = "left.blob".into();
+ left.row_count = 2;
+ left.max_sequence_number = 1;
+ left.write_cols = Some(vec!["value".into()]);
+ let mut right = left.clone();
+ right.file_name = "right.vector.store".into();
+ right.first_row_id = Some(2);
+ right.row_count = 4;
+ let files = vec![right, base.clone(), left.clone(), partial];
+ let split = split.for_data_evolution_files(files.clone());
+ let routing =
+ crate::table::RowIdPlan::from_plan(Plan::from_splits(Some(2),
vec![split.clone()]))
+ .unwrap();
+ assert_eq!(routing.groups().len(), 1);
+ assert_eq!((routing.groups()[0].0, routing.groups()[0].1), (0, 5));
+ assert_eq!(routing.groups()[0].2.data_files().len(), 4);
+ assert!(files
+ .iter()
+ .all(|file| routing.groups()[0].2.data_files().contains(file)));
+ left.file_name = "invalid-short-column.parquet".into();
+ let invalid = split.for_data_evolution_files(vec![base, left]);
+ assert!(
+ crate::table::RowIdPlan::from_plan(Plan::from_splits(Some(2),
vec![invalid]))
+ .unwrap_err()
+ .to_string()
+ .contains("identical ranges")
+ );
+}
+
+#[tokio::test]
+async fn
captured_plan_rejects_other_operation_families_before_io_or_callbacks() {
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
+ let table = table("planned_operation_family").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let mut update = table.new_write_builder().new_update().unwrap();
+ update.with_plan(plan).unwrap();
+ for directory in ["manifest", "snapshot"] {
+ table
+ .file_io()
+ .delete_dir(&format!("{}/{directory}", table.location()))
+ .await
+ .unwrap();
+ }
+ let calls = Arc::new(AtomicUsize::new(0));
+ let count = calls.clone();
+ let callback = crate::table::UpdateAssignment::Function(Arc::new(move |_| {
+ count.fetch_add(1, Ordering::SeqCst);
+ Ok(Vec::new())
+ }));
+ let errors = vec![
+ update
+ .new_shard_updater(0, 1, None)
+ .await
+ .err()
+ .unwrap()
+ .to_string(),
+ update
+ .update_by_predicate(None, vec![("value".into(), callback)],
vec![])
+ .await
+ .unwrap_err()
+ .to_string(),
+ update
+ .upsert_by_arrow_with_key(vec![], vec!["value".into()])
+ .await
+ .unwrap_err()
+ .to_string(),
+ update
+ .merge_into(crate::table::MergeSource::SelfTable, vec![], vec![],
vec![])
+ .await
+ .unwrap_err()
+ .to_string(),
+ update
+ .delete_by_predicate(None)
+ .await
+ .unwrap_err()
+ .to_string(),
+ ];
+ assert!(
+ errors
+ .iter()
+ .all(|error| error.contains("with_plan supports only row-ID")),
+ "{errors:?}"
+ );
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+}
diff --git a/crates/paimon/src/table/vindex_index_build_builder.rs
b/crates/paimon/src/table/vindex_index_build_builder.rs
index bfc1ecf6..b85a2539 100644
--- a/crates/paimon/src/table/vindex_index_build_builder.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder.rs
@@ -26,6 +26,8 @@ use planning::plan_vindex_shards;
use validation::{checked_i32, find_index_field, validate_vector_field};
use crate::spec::{CoreOptions, Predicate};
+use crate::table::global_index_build_common::vector::VectorIndexShard;
+use crate::table::global_index_build_plan::GlobalIndexBuildPlan;
use crate::table::{CommitMessage, RowRange, Table};
use crate::vindex::{is_vindex_index_type, VindexVectorIndexOptions};
use crate::{Error, Result};
@@ -60,13 +62,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
self
}
- pub(super) async fn prepare(
- &self,
- ) -> Result<(
- Option<i64>,
- Vec<CommitMessage>,
- Vec<timing::VectorIndexBuildTiming>,
- )> {
+ pub(super) async fn plan(&self) -> Result<GlobalIndexBuildPlan> {
// Building the index scans the table's rows.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
@@ -92,29 +88,30 @@ impl<'a> VindexIndexBuildBuilder<'a> {
let core_options = CoreOptions::new(&merged_options);
super::global_index_build_common::validate_build_table(self.table)?;
let rows_per_shard = core_options.global_index_row_count_per_shard()?;
- let parallelism = core_options.global_index_build_parallelism()?;
+ core_options.global_index_build_parallelism()?;
let index_field = find_index_field(self.table, index_column)?;
validate_vector_field(index_field)?;
- let vindex_options = VindexVectorIndexOptions::new(
+ let options = VindexVectorIndexOptions::new(
self.table.schema().options(),
&self.options,
&self.index_type,
index_field,
)?;
- let dimension = checked_i32(
- vindex_options.dimension() as u64,
+ checked_i32(
+ options.dimension() as u64,
"vindex dimension is too large for Rust builder",
)?;
- let index_meta =
- serde_json::to_vec(&vindex_options.native_options).map_err(|e|
Error::DataInvalid {
- message: format!("Failed to serialize vindex options metadata:
{e}"),
- source: Some(Box::new(e)),
- })?;
-
let snapshot_manager = self.table.snapshot_manager();
let Some(snapshot) = snapshot_manager.get_latest_snapshot().await?
else {
- return Ok((None, vec![], vec![]));
+ return Ok(GlobalIndexBuildPlan::new(
+ self.table,
+ None,
+ index_column.into(),
+ self.index_type.clone(),
+ self.options.clone(),
+ vec![],
+ ));
};
let mut read_builder = self.table.new_read_builder();
@@ -144,48 +141,93 @@ impl<'a> VindexIndexBuildBuilder<'a> {
rows_per_shard,
&indexed,
)?;
- if shards.is_empty() {
- return Ok((Some(snapshot.id()), vec![], vec![]));
+
+ if !shards.is_empty() {
+
crate::table::global_index_build_common::validate_existing_index_overlap(
+ self.table,
+ snapshot.index_manifest(),
+ &self.index_type,
+ index_field.id(),
+ None,
+ &shards
+ .iter()
+ .map(|shard| RowRange::new(shard.row_range_start,
shard.row_range_end))
+ .collect::<Vec<_>>(),
+ )
+ .await?;
}
-
crate::table::global_index_build_common::validate_existing_index_overlap(
+ Ok(GlobalIndexBuildPlan::new(
self.table,
- snapshot.index_manifest(),
+ Some(snapshot),
+ index_column.into(),
+ self.index_type.clone(),
+ self.options.clone(),
+ shards,
+ ))
+ }
+
+ pub(super) async fn build_planned_shard(
+ &self,
+ shard: &VectorIndexShard,
+ ) -> Result<Option<(CommitMessage,
Option<timing::VectorIndexBuildTiming>)>> {
+ let index_column = self
+ .index_column
+ .as_deref()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "vindex index column is required".into(),
+ source: None,
+ })?;
+ let index_field = find_index_field(self.table, index_column)?;
+ validate_vector_field(index_field)?;
+ let options = VindexVectorIndexOptions::new(
+ self.table.schema().options(),
+ &self.options,
&self.index_type,
- index_field.id(),
- None,
- &shards
- .iter()
- .map(|shard| RowRange::new(shard.row_range_start,
shard.row_range_end))
- .collect::<Vec<_>>(),
- )
- .await?;
+ index_field,
+ )?;
+ let dimension = checked_i32(
+ options.dimension() as u64,
+ "vindex dimension is too large for Rust builder",
+ )?;
+ let index_meta =
+ serde_json::to_vec(&options.native_options).map_err(|error|
Error::DataInvalid {
+ message: format!("Failed to serialize vindex options metadata:
{error}"),
+ source: Some(Box::new(error)),
+ })?;
+ let built = self
+ .build_index_file(
+ shard,
+ index_column,
+ dimension,
+ index_field.id(),
+ &options,
+ index_meta,
+ )
+ .await?;
+ Ok(built.map(|built| {
+ let mut message =
CommitMessage::new(shard.partition_bytes.clone(), 0, vec![]);
+ message.new_index_files = vec![built.meta];
+ (message, built.timing)
+ }))
+ }
+ pub(super) async fn prepare(
+ &self,
+ ) -> Result<(
+ Option<i64>,
+ Vec<CommitMessage>,
+ Vec<timing::VectorIndexBuildTiming>,
+ )> {
+ let plan = self.plan().await?;
+ let mut options = self.table.schema().options().clone();
+ options.extend(self.options.clone());
+ let parallelism =
CoreOptions::new(&options).global_index_build_parallelism()?;
let prepared =
super::global_index_build_common::preparation::prepare_shards(
self.table,
- shards,
+ plan.shards(),
parallelism,
- |shard| {
- let vindex_options = &vindex_options;
- let index_meta = index_meta.clone();
- async move {
- let built = self
- .build_index_file(
- &shard,
- index_column,
- dimension,
- index_field.id(),
- vindex_options,
- index_meta,
- )
- .await?;
- Ok(built.map(|built| {
- let mut message =
CommitMessage::new(shard.partition_bytes, 0, vec![]);
- message.new_index_files = vec![built.meta];
- (message, built.timing)
- }))
- }
- },
+ |shard| async move { shard.build(self.table).await },
)
.await?;
let mut messages = Vec::with_capacity(prepared.len());
@@ -196,8 +238,7 @@ impl<'a> VindexIndexBuildBuilder<'a> {
timings.push(timing);
}
}
-
- Ok((Some(snapshot.id()), messages, timings))
+ Ok((plan.snapshot_id(), messages, timings))
}
}
diff --git a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
index 617a8dbc..cba9244d 100644
--- a/crates/paimon/src/table/vindex_index_build_builder/tests.rs
+++ b/crates/paimon/src/table/vindex_index_build_builder/tests.rs
@@ -481,6 +481,157 @@ async fn write_vectors(table: &Table, ids: Vec<i32>,
vectors: Vec<Vec<f32>>) {
.unwrap();
}
+#[tokio::test]
+async fn planned_global_index_validates_dimension_before_empty_snapshot() {
+ let path = "memory:/planned_global_index_empty_dimension";
+ let table = vindex_e2e_table(path, "2");
+ setup_dirs(table.file_io(), path).await;
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_type(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .with_options(HashMap::from([(
+ "ivf-flat.dimension".into(),
+ (i32::MAX as u64 + 1).to_string(),
+ )]));
+ for error in [
+ builder.plan().await.unwrap_err(),
+ builder.build().await.unwrap_err(),
+ ] {
+ assert!(error.to_string().contains("dimension is too large"));
+ }
+}
+
+#[tokio::test]
+async fn
planned_global_index_shards_keep_snapshot_schema_and_physical_coverage() {
+ let path = "memory:/planned_global_index_shards";
+ let mut options = vindex_e2e_options("2");
+ options.insert("ivf-flat.nlist".into(), "1".into());
+ options.insert("deletion-vectors.enabled".into(), "true".into());
+ let table = test_table_with_io(
+ FileIOBuilder::new("memory").build().unwrap(),
+ path,
+ vindex_schema_builder(options).build().unwrap(),
+ );
+ setup_dirs(table.file_io(), path).await;
+ write_vectors(&table, vec![1, 2, 3, 4], vec![vec![1., 0.]; 4]).await;
+ let write = table.new_write_builder();
+ write
+ .new_commit()
+ .commit(
+ write
+ .new_update()
+ .unwrap()
+ .delete_by_row_id(vec![1])
+ .await
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let mut builder = table.new_global_index_build_builder();
+ builder
+ .with_index_type(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding");
+ let plan = builder.plan().await.unwrap();
+ let snapshot_id = plan.snapshot_id().unwrap();
+ let plan: crate::table::GlobalIndexBuildPlan =
+ serde_json::from_slice(&serde_json::to_vec(&plan).unwrap()).unwrap();
+ let shards = plan.shards();
+ assert_eq!(
+ shards.iter().map(|s| s.row_range()).collect::<Vec<_>>(),
+ vec![RowRange::new(0, 1), RowRange::new(2, 3)]
+ );
+ assert!(shards
+ .iter()
+ .all(|shard| shard.snapshot_id() == snapshot_id));
+ write_vectors(&table, vec![5, 6], vec![vec![0., 1.]; 2]).await;
+ // Changing the builder after planning cannot change a worker's source or
field.
+ builder
+ .with_index_column("id")
+ .with_options(HashMap::from([("ivf-flat.dimension".into(),
"999".into())]));
+ let messages = builder.build_shard(&shards[0]).await.unwrap();
+ assert_eq!(messages.len(), 1);
+ let file = &messages[0].new_index_files[0];
+ assert_eq!(
+ file.row_count, 2,
+ "deleted vectors remain in physical build coverage"
+ );
+ let meta = file.global_index_meta.as_ref().unwrap();
+ assert_eq!(
+ (
+ meta.row_range_start,
+ meta.row_range_end,
+ meta.index_field_id
+ ),
+ (0, 1, 1)
+ );
+ assert_eq!(
+ latest_vindex_index_files(&table).await.len(),
+ 0,
+ "workers do not commit"
+ );
+ assert_eq!(
+ table
+ .file_io()
+ .list_status(&format!("{path}/index/"))
+ .await
+ .unwrap()
+ .iter()
+ .filter(|status|
status.path.contains("vector-ivf-flat-global-index-"))
+ .count(),
+ 1,
+ "one task must not build another shard or later append"
+ );
+ write.new_commit().commit(messages).await.unwrap();
+ let plan = table
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(
+ plan.shards()
+ .iter()
+ .map(|s| s.row_range())
+ .collect::<Vec<_>>(),
+ vec![RowRange::new(2, 3), RowRange::new(4, 5)]
+ );
+}
+
+#[tokio::test]
+async fn
planned_global_index_shards_recheck_worker_authorization_and_table_identity() {
+ let path = "memory:/planned_global_index_authorization";
+ let table = vindex_e2e_table(path, "2");
+ setup_dirs(table.file_io(), path).await;
+ write_vectors(&table, vec![1, 2], vec![vec![1., 0.]; 2]).await;
+ let plan = table
+ .new_global_index_build_builder()
+ .with_index_type(IVF_FLAT_IDENTIFIER)
+ .with_index_column("embedding")
+ .plan()
+ .await
+ .unwrap();
+ let shard = &plan.shards()[0];
+ let denied = table.copy_with_options(HashMap::from([(
+ "query-auth.enabled".into(),
+ "true".into(),
+ )]));
+ let error = denied
+ .new_global_index_build_builder()
+ .build_shard(shard)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("query-auth"));
+ let other = vindex_e2e_table("memory:/different_table", "2");
+ let error = other
+ .new_global_index_build_builder()
+ .build_shard(shard)
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("different table"));
+}
+
#[tokio::test]
async fn vector_build_with_deletion_vectors_preserves_physical_coverage() {
for granule in [false, true] {
