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 af72cf06 Support captured row ID routing and read selection (#1209)
af72cf06 is described below
commit af72cf06451f3716e9a73e033cc271517871e64c
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 23 20:35:20 2026 +0800
Support captured row ID routing and read selection (#1209)
---
bindings/python/README.md | 23 ++
.../python/python/pypaimon_rust/datafusion.pyi | 18 ++
bindings/python/src/read.rs | 136 +++++++++++
crates/paimon/src/table/table_update_by_row_id.rs | 100 +++++++-
crates/paimon/src/table/table_update_plan_tests.rs | 256 ++++++++++++++++++++-
5 files changed, 528 insertions(+), 5 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index fc9cc1d2..d793d2d3 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -204,6 +204,29 @@ Combined incremental scans expose
the same API for their selected Delta or Changelog window. Snapshot metadata is
available from `plan.snapshot_id()`.
+### Captured row-ID routing
+
+`RowIdPlan` provides pure routing for distributed Data Evolution operations:
+
+```python
+from pypaimon_rust.datafusion import RowIdPlan
+
+routing = RowIdPlan(table.new_read_builder().new_scan().plan_all())
+first_row_ids = routing.route(row_ids) # row_ids is an Arrow Int64Array
+delete_groups = routing.route_by_bucket(row_ids)
+selected = routing.select(row_ids)
+```
+
+Routing preserves input order and duplicates, and rejects null or foreign IDs.
+Read/update groups use their anchor's first row ID. Delete groups share one
+routing ID per partition and bucket so each worker can publish a complete
+deletion-vector index. These IDs are local to the captured plan.
+
+`select` retains complete normal and dedicated file groups, deletion metadata,
+authorization and the captured snapshot, and adds sorted, deduplicated row
ranges
+for reading. It does not check row liveness or authorize file access; readers
+and mutation workers retain those checks. The selected plan is only for
reading.
+
### Shard column updates
The batch and stream update objects expose
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 073f583d..92964a99 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -78,6 +78,24 @@ class RowIdPlan:
"""Inclusive row-ID bounds and all physical files of each logical
group."""
...
def plan(self) -> Plan: ...
+ def route(self, row_ids: pyarrow.Int64Array) -> pyarrow.Int64Array:
+ """Return each row's owning file-group first row ID, preserving input
order."""
+ ...
+ def route_by_bucket(self, row_ids: pyarrow.Int64Array) ->
pyarrow.Int64Array:
+ """Return one opaque routing ID per captured partition/bucket.
+
+ IDs are stable within this plan. Multiple file groups in the same
+ partition/bucket share an ID so one worker can publish their DV index.
+ """
+ ...
+ def select(self, row_ids: pyarrow.Int64Array) -> Plan:
+ """Build a read-only plan with sorted, deduplicated row selections.
+
+ Complete file groups, deletion metadata, grants, paths and the captured
+ snapshot are retained. Readers authorize execution independently. This
+ selected plan must not be used as a mutation plan.
+ """
+ ...
class MergeAssignment:
"""An immutable source reference, target reference, literal or callback."""
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 3545ff39..4d9b9c32 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -20,6 +20,7 @@ use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
+use arrow::array::{Array, Int64Array};
use arrow::datatypes::SchemaRef;
use arrow::pyarrow::ToPyArrow;
use arrow::record_batch::RecordBatch;
@@ -1243,6 +1244,16 @@ pub struct PyRowIdPlan {
inner: paimon::table::RowIdPlan,
}
+impl PyRowIdPlan {
+ fn row_ids(value: &Bound<'_, PyAny>) -> PyResult<Int64Array> {
+ crate::arrow_input::array_from_pyarrow(value)?
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .cloned()
+ .ok_or_else(|| PyTypeError::new_err("Row-ID routing requires an
Int64Array"))
+ }
+}
+
#[pymethods]
impl PyRowIdPlan {
#[new]
@@ -1275,6 +1286,32 @@ impl PyRowIdPlan {
fn plan(&self) -> PyPlan {
self.inner.plan().clone().into()
}
+
+ /// Return the owning file group's first row ID in input order.
+ fn route(&self, py: Python<'_>, row_ids: &Bound<'_, PyAny>) ->
PyResult<Py<PyAny>> {
+ let row_ids = Self::row_ids(row_ids)?;
+ let groups = py
+ .detach(|| self.inner.route(&row_ids))
+ .map_err(to_py_err)?;
+ groups.to_data().to_pyarrow(py).map(Bound::unbind)
+ }
+
+ /// Return one routing ID per captured partition/bucket, preserving order.
+ fn route_by_bucket(&self, py: Python<'_>, row_ids: &Bound<'_, PyAny>) ->
PyResult<Py<PyAny>> {
+ let row_ids = Self::row_ids(row_ids)?;
+ let groups = py
+ .detach(|| self.inner.route_by_bucket(&row_ids))
+ .map_err(to_py_err)?;
+ groups.to_data().to_pyarrow(py).map(Bound::unbind)
+ }
+
+ /// Select rows for reading while retaining complete groups and their
grants.
+ fn select(&self, py: Python<'_>, row_ids: &Bound<'_, PyAny>) ->
PyResult<PyPlan> {
+ let row_ids = Self::row_ids(row_ids)?;
+ py.detach(|| self.inner.select(&row_ids))
+ .map(Into::into)
+ .map_err(to_py_err)
+ }
}
#[pyclass(name = "Split", module = "pypaimon_rust.datafusion")]
@@ -1363,6 +1400,105 @@ mod tests {
use super::*;
+ #[test]
+ fn row_id_plan_binding_routes_arrow_and_keeps_read_selection_separate() {
+ Python::attach(|py| {
+ 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", "row_id_binding"),
+ "memory:/row-id-binding".into(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let plan = py.detach(|| {
+ runtime().block_on(async {
+ for values in [vec![10, 11], vec![12]] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ let batch = RecordBatch::try_from_iter(vec![(
+ "value",
+ Arc::new(arrow::array::Int32Array::from(values))
+ as arrow::array::ArrayRef,
+ )])
+ .unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ builder
+ .new_commit()
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ }
+ table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap()
+ })
+ });
+ let locals = PyDict::new(py);
+ locals.set_item("plan", PyPlan::from(plan)).unwrap();
+ locals
+ .set_item("RowIdPlan", py.get_type::<PyRowIdPlan>())
+ .unwrap();
+ locals.set_item("Plan", py.get_type::<PyPlan>()).unwrap();
+ py.run(
+ pyo3::ffi::c_str!(
+ r#"
+import pyarrow as pa
+
+routing = RowIdPlan(plan)
+assert routing.snapshot_id() == 2
+ids = pa.array([999, 2, 0, 1, 0, 999], type=pa.int64()).slice(1, 4)
+assert routing.route(ids).to_pylist() == [2, 0, 0, 0]
+assert routing.route_by_bucket(ids).to_pylist() == [0, 0, 0, 0]
+selected = routing.select(ids)
+assert selected.snapshot_id() == 2 and len(selected) == 2
+assert sum(split.row_count() for split in selected.splits()) == 3
+assert routing.route(pa.array([2], type=pa.int64())).to_pylist() == [2]
+try:
+ RowIdPlan(selected)
+except ValueError as error:
+ assert 'without row selection' in str(error)
+else:
+ raise AssertionError('A read selection must not become a mutation plan')
+
+for operation in [routing.route, routing.route_by_bucket, routing.select]:
+ for invalid, error_type in [(pa.array([None], type=pa.int64()),
ValueError),
+ (pa.array([-1], type=pa.int64()), ValueError),
+ (pa.array([3], type=pa.int64()), ValueError),
+ (pa.array([0], type=pa.int32()), TypeError)]:
+ try:
+ operation(invalid)
+ except error_type:
+ pass
+ else:
+ raise AssertionError('Invalid row IDs were accepted')
+
+empty_ids = pa.array([], type=pa.int64())
+empty = RowIdPlan(Plan(7, []))
+assert empty.route(empty_ids).to_pylist() == []
+assert empty.route_by_bucket(empty_ids).to_pylist() == []
+assert empty.select(empty_ids).snapshot_id() == 7
+assert len(empty.select(empty_ids)) == 0
+assert routing.plan().snapshot_id() == 2
+"#
+ ),
+ Some(&locals),
+ None,
+ )
+ .unwrap();
+ });
+ }
+
#[test]
fn
file_planning_binding_keeps_complete_entries_and_independent_query_filters() {
Python::attach(|py| {
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 acf5eba5..630da1d9 100644
--- a/crates/paimon/src/table/table_update_by_row_id.rs
+++ b/crates/paimon/src/table/table_update_by_row_id.rs
@@ -18,9 +18,9 @@
//! Row-ID updates sharing one target snapshot and file index.
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};
+use crate::table::{CommitMessage, DataEvolutionWriter, DataSplit, Plan,
RowRange, Table};
+use arrow_array::{Array, Int64Array, RecordBatch};
+use std::collections::{BTreeMap, HashMap, HashSet};
fn invalid(message: impl Into<String>) -> crate::Error {
crate::Error::DataInvalid {
@@ -37,6 +37,7 @@ fn invalid(message: impl Into<String>) -> crate::Error {
pub struct RowIdPlan {
plan: Plan,
groups: Vec<(i64, i64, DataSplit)>,
+ bucket_first_row_ids: Vec<i64>,
}
impl RowIdPlan {
@@ -106,7 +107,22 @@ impl RowIdPlan {
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 })
+ let mut buckets = HashMap::new();
+ let bucket_first_row_ids = groups
+ .iter()
+ .map(|(first, _, split)| {
+ // Groups are sorted, so the first anchor encountered is a
+ // deterministic representative for its partition and bucket.
+ *buckets
+ .entry((split.partition().to_serialized_bytes(),
split.bucket()))
+ .or_insert(*first)
+ })
+ .collect();
+ Ok(Self {
+ plan,
+ groups,
+ bucket_first_row_ids,
+ })
}
pub fn snapshot_id(&self) -> Option<i64> {
@@ -120,6 +136,82 @@ impl RowIdPlan {
pub fn plan(&self) -> &Plan {
&self.plan
}
+
+ /// Route each row ID to its normal-file group's first row ID.
+ ///
+ /// Output retains input order and duplicates. NULLs, negative IDs and IDs
+ /// outside the captured groups are rejected. This is metadata routing:
+ /// deletion vectors and row existence are checked by the eventual reader
+ /// or mutation, which must authorize independently.
+ pub fn route(&self, row_ids: &Int64Array) -> crate::Result<Int64Array> {
+ let groups = self.group_indices(row_ids)?;
+ Ok(Int64Array::from_iter_values(
+ groups.into_iter().map(|index| self.groups[index].0),
+ ))
+ }
+
+ /// Route each row ID to a stable representative of its partition/bucket.
+ ///
+ /// The representative is the smallest anchor first row ID in that bucket
+ /// across this captured plan. Deletion-vector mutations must assign all
+ /// rows for one partition/bucket to the same worker, including rows from
+ /// different normal-file groups sharing that bucket's index file.
+ pub fn route_by_bucket(&self, row_ids: &Int64Array) ->
crate::Result<Int64Array> {
+ let groups = self.group_indices(row_ids)?;
+ Ok(Int64Array::from_iter_values(
+ groups
+ .into_iter()
+ .map(|index| self.bucket_first_row_ids[index]),
+ ))
+ }
+
+ /// Select row IDs for reading from the captured snapshot.
+ ///
+ /// Keep every file, deletion vector, exact path and authorization grant in
+ /// each selected logical group. Only the read's absolute row-ID ranges are
+ /// narrowed; input duplicates and order do not duplicate or reorder rows.
+ /// The returned plan is a read selection, not a complete mutation plan.
+ pub fn select(&self, row_ids: &Int64Array) -> crate::Result<Plan> {
+ let groups = self.group_indices(row_ids)?;
+ let mut selections: BTreeMap<usize, Vec<RowRange>> = BTreeMap::new();
+ for (index, row_id) in groups.into_iter().zip(row_ids.values()) {
+ selections
+ .entry(index)
+ .or_default()
+ .push(RowRange::new(*row_id, *row_id));
+ }
+ let splits = selections
+ .into_iter()
+ .map(|(index, ranges)| {
+ self.groups[index]
+ .2
+ .clone()
+
.with_selected_row_ranges(super::source::merge_row_ranges(ranges))
+ })
+ .collect();
+ self.plan.clone().map_splits(|_| Ok(splits))
+ }
+
+ fn group_indices(&self, row_ids: &Int64Array) -> crate::Result<Vec<usize>>
{
+ if row_ids.null_count() != 0 {
+ return Err(invalid("_ROW_ID must not be null"));
+ }
+ row_ids
+ .values()
+ .iter()
+ .map(|row_id| {
+ let position = self.groups.partition_point(|(first, _, _)|
first <= row_id);
+ position
+ .checked_sub(1)
+ .filter(|index| *row_id >= 0 && *row_id <=
self.groups[*index].1)
+ .ok_or_else(|| {
+ invalid(format!(
+ "_ROW_ID {row_id} does not belong to any valid
range in the captured plan"
+ ))
+ })
+ })
+ .collect()
+ }
}
/// Update columns selected per input table, as in PyPaimon.
diff --git a/crates/paimon/src/table/table_update_plan_tests.rs
b/crates/paimon/src/table/table_update_plan_tests.rs
index b4fb6ddd..a86c4fd8 100644
--- a/crates/paimon/src/table/table_update_plan_tests.rs
+++ b/crates/paimon/src/table/table_update_plan_tests.rs
@@ -19,7 +19,9 @@ use super::*;
use crate::catalog::Identifier;
use crate::io::FileIOBuilder;
use crate::spec::{DataType, IntType, Schema, TableSchema};
-use crate::table::{DataSplitBuilder, Plan, RowRange};
+use crate::table::{
+ DataSplit, DataSplitBuilder, DeletionFile, Plan, RowIdPlan, RowRange,
ScanStats,
+};
use arrow_array::{ArrayRef, Int32Array, Int64Array};
use std::sync::Arc;
@@ -306,6 +308,258 @@ async fn
row_id_plan_routes_complete_groups_and_rejects_ambiguous_ranges() {
assert_eq!(routing.plan().splits(), plan.splits());
}
+fn routing_split(
+ source: &DataSplit,
+ first: i64,
+ count: i64,
+ partition: i32,
+ bucket: i32,
+) -> DataSplit {
+ let mut file = source.data_files()[0].clone();
+ file.file_name = format!("anchor-{first}.parquet");
+ file.first_row_id = Some(first);
+ file.row_count = count;
+ file.write_cols = None;
+ let mut partition_row = crate::spec::BinaryRowBuilder::new(1);
+ partition_row.write_int(0, partition);
+ DataSplitBuilder::new()
+ .with_snapshot(source.snapshot_id())
+ .with_partition(partition_row.build())
+ .with_bucket(bucket)
+
.with_bucket_path(format!("memory:/partition-{partition}/bucket-{bucket}"))
+ .with_data_files(vec![file])
+ .build()
+ .unwrap()
+}
+
+#[tokio::test]
+async fn row_id_plan_routes_sliced_ids_by_anchor_and_partition_bucket() {
+ let table = table("row_id_arrow_routing").await;
+ let plan = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let source = &plan.splits()[0];
+ let splits = vec![
+ routing_split(source, 30, 5, 1, 1),
+ routing_split(source, i64::MAX - 1, 2, 1, 0),
+ routing_split(source, 5, 4, 1, 0),
+ routing_split(source, 20, 2, 2, 0),
+ routing_split(source, 0, 3, 1, 0),
+ ];
+ let routing = RowIdPlan::from_plan(Plan::from_splits(Some(2),
splits)).unwrap();
+ // Slices, repeated IDs and arbitrary order must retain their coordinates.
+ // The same bucket number in another partition must remain independent.
+ let ids =
+ Int64Array::from(vec![-99, i64::MAX, 8, 20, 2, 30, 5, i64::MAX - 1, 8,
-99]).slice(1, 8);
+ assert_eq!(
+ routing.route(&ids).unwrap().values(),
+ &[i64::MAX - 1, 5, 20, 0, 30, 5, i64::MAX - 1, 5]
+ );
+ assert_eq!(
+ routing.route_by_bucket(&ids).unwrap().values(),
+ &[0, 0, 20, 0, 30, 0, 0, 0]
+ );
+ // Bucket representatives come from the complete plan, not only the IDs
+ // present in this input batch or the order of the input splits.
+ assert_eq!(
+ routing
+ .route_by_bucket(&Int64Array::from(vec![7]))
+ .unwrap()
+ .values(),
+ &[0]
+ );
+ let mut reversed = routing.plan().splits().to_vec();
+ reversed.reverse();
+ let reordered = RowIdPlan::from_plan(Plan::from_splits(Some(2),
reversed)).unwrap();
+ assert_eq!(routing.route(&ids).unwrap(), reordered.route(&ids).unwrap());
+ assert_eq!(
+ routing.route_by_bucket(&ids).unwrap(),
+ reordered.route_by_bucket(&ids).unwrap()
+ );
+
+ for invalid_id in [-1, 3, 4, 9, 19, 22, 35] {
+ let ids = Int64Array::from(vec![0, invalid_id]);
+ for error in [
+ routing.route(&ids).unwrap_err(),
+ routing.route_by_bucket(&ids).unwrap_err(),
+ routing.select(&ids).unwrap_err(),
+ ] {
+ assert!(error.to_string().contains(&format!(
+ "_ROW_ID {invalid_id} does not belong to any valid range"
+ )));
+ }
+ }
+ let null_ids = Int64Array::from(vec![Some(0), None]);
+ for error in [
+ routing.route(&null_ids).unwrap_err(),
+ routing.route_by_bucket(&null_ids).unwrap_err(),
+ routing.select(&null_ids).unwrap_err(),
+ ] {
+ assert!(error.to_string().contains("_ROW_ID must not be null"));
+ }
+ // A failed routing request never changes the captured metadata.
+ assert_eq!(routing.route(&ids).unwrap().len(), ids.len());
+}
+
+#[test]
+fn row_id_plan_empty_inputs_keep_the_captured_snapshot() {
+ for snapshot_id in [None, Some(7)] {
+ let routing = RowIdPlan::from_plan(Plan::from_splits(snapshot_id,
Vec::new())).unwrap();
+ let empty = Int64Array::from(Vec::<i64>::new());
+ assert!(routing.route(&empty).unwrap().is_empty());
+ assert!(routing.route_by_bucket(&empty).unwrap().is_empty());
+ let selected = routing.select(&empty).unwrap();
+ assert_eq!(selected.snapshot_id(), snapshot_id);
+ assert!(selected.splits().is_empty());
+ assert!(routing.route(&Int64Array::from(vec![0])).is_err());
+ assert!(routing.route_by_bucket(&Int64Array::from(vec![0])).is_err());
+ assert!(routing.select(&Int64Array::from(vec![0])).is_err());
+ assert!(routing
+ .select(&Int64Array::from(vec![None::<i64>]))
+ .is_err());
+ }
+}
+
+#[tokio::test]
+async fn
row_id_plan_read_selection_retains_complete_groups_paths_deletions_and_grants()
{
+ let table = table("row_id_read_selection").await;
+ let initial = table
+ .new_read_builder()
+ .new_scan()
+ .plan_all()
+ .await
+ .unwrap();
+ let source = &initial.splits()[0];
+ let mut first = routing_split(source, 10, 6, 1, 0);
+ let mut anchor = first.data_files()[0].clone();
+ anchor.external_path = Some("s3://external/anchor-10.parquet".into());
+ let mut partial = anchor.clone();
+ partial.file_name = "partial-column.parquet".into();
+ partial.external_path =
Some("s3://external/partial-column.parquet".into());
+ partial.max_sequence_number += 1;
+ partial.write_cols = Some(vec!["value".into()]);
+ let mut blob = partial.clone();
+ blob.file_name = "partial-payload.blob".into();
+ blob.external_path = Some("s3://external/partial-payload.blob".into());
+ blob.first_row_id = Some(11);
+ blob.row_count = 2;
+ let deletions = vec![
+ None,
+ Some(DeletionFile::new(
+ "s3://external/deletions".into(),
+ 5,
+ 20,
+ Some(1),
+ )),
+ None,
+ ];
+ first = DataSplitBuilder::new()
+ .with_snapshot(2)
+ .with_partition(first.partition().clone())
+ .with_bucket(first.bucket())
+ .with_bucket_path(first.bucket_path().to_string())
+ .with_data_files(vec![blob, anchor, partial])
+ .with_data_deletion_files(deletions)
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+ let grant = Arc::new(
+ crate::table::query_auth::QueryAuthGrant::parse(
+ &crate::api::AuthTableQueryResponse::default(),
+ 17,
+ None,
+ table.schema().fields(),
+ )
+ .unwrap(),
+ );
+ let stats = ScanStats {
+ manifest_entries_read: 123,
+ split_candidates_built: 3,
+ ..Default::default()
+ };
+ let plan = Plan::from_splits(
+ Some(2),
+ vec![
+ routing_split(source, 60, 2, 1, 0),
+ first,
+ routing_split(source, 30, 5, 1, 0),
+ ],
+ )
+ .planned(Some(grant.clone()))
+ .with_scan_stats(stats);
+ let routing = RowIdPlan::from_plan(plan).unwrap();
+ let ids = Int64Array::from(vec![999, 15, 10, 12, 12, 30, 13, 34, 11,
999]).slice(1, 8);
+ let selected = routing.select(&ids).unwrap();
+ assert_eq!(selected.snapshot_id(), Some(2));
+ assert_eq!(selected.splits().len(), 2);
+ assert_eq!(selected.stats().manifest_entries_read, 123);
+ assert_eq!(selected.stats().split_candidates_built, 3);
+ assert_eq!(selected.stats().final_splits, 2);
+ assert_eq!(selected.stats().final_files, 4);
+ assert_eq!(
+ selected.splits()[0].row_ranges(),
+ Some([RowRange::new(10, 13), RowRange::new(15, 15)].as_slice())
+ );
+ assert_eq!(
+ selected.splits()[1].row_ranges(),
+ Some([RowRange::new(30, 30), RowRange::new(34, 34)].as_slice())
+ );
+ for (selected, (_, _, original)) in
selected.splits().iter().zip(routing.groups()) {
+ assert_eq!(selected.data_files(), original.data_files());
+ assert_eq!(
+ selected.data_deletion_files(),
+ original.data_deletion_files()
+ );
+ assert_eq!(selected.partition(), original.partition());
+ assert_eq!(selected.bucket(), original.bucket());
+ assert_eq!(selected.bucket_path(), original.bucket_path());
+ assert_eq!(selected.raw_convertible(), original.raw_convertible());
+ assert!(selected.query_auth_required());
+ assert!(Arc::ptr_eq(selected.query_auth_grant().unwrap(), &grant));
+ assert_eq!(
+ selected.query_auth_response(),
+ original.query_auth_response()
+ );
+
+ // Wire transport preserves the read selection and auth requirement,
+ // but cannot turn serialized policy metadata back into a live grant.
+ let restored =
+
DataSplit::deserialize_split_v1(&selected.serialize_split_v1().unwrap()).unwrap();
+ assert_eq!(restored.row_ranges(), selected.row_ranges());
+ assert_eq!(restored.data_files(), selected.data_files());
+ assert_eq!(
+ restored.data_deletion_files(),
+ selected.data_deletion_files()
+ );
+ assert!(restored.query_auth_required());
+ assert!(restored.query_auth_grant().is_none());
+ }
+ assert_eq!(routing.plan().splits().len(), 3);
+ assert!(routing
+ .groups()
+ .iter()
+ .all(|(_, _, split)| split.row_ranges().is_none()));
+ // A read selection is intentionally not accepted as a mutation baseline.
+ assert!(RowIdPlan::from_plan(selected.clone()).is_err());
+ assert!(table
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .with_plan(selected)
+ .is_err());
+
+ let empty = routing
+ .select(&Int64Array::from(Vec::<i64>::new()))
+ .unwrap();
+ assert_eq!(empty.snapshot_id(), Some(2));
+ assert!(empty.splits().is_empty());
+ assert_eq!(empty.stats().manifest_entries_read, 123);
+ assert_eq!(empty.stats().final_files, 0);
+}
+
#[tokio::test]
async fn row_id_plan_keeps_full_partial_columns_and_dedicated_subranges() {
let table = table("planned_row_id_partial_ranges").await;
