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 f0f5d964 Support REST authorized stream and combined incremental reads
(#1130)
f0f5d964 is described below
commit f0f5d9643a514939baa6ef2d079e37248dc88c68
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 15 15:37:59 2026 +0800
Support REST authorized stream and combined incremental reads (#1130)
---
.../python/python/pypaimon_rust/datafusion.pyi | 6 +-
crates/paimon/src/table/incremental_scan.rs | 6 +-
crates/paimon/src/table/stream_table_scan.rs | 16 +-
crates/paimon/src/table/table_read.rs | 65 ++--
crates/paimon/src/table/table_scan.rs | 81 +++-
.../rest_catalog/authorized_snapshot_scans.rs | 409 +++++++++++++++++++++
crates/paimon/tests/rest_catalog_test.rs | 4 +
7 files changed, 533 insertions(+), 54 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 8fdf00e5..7b605d68 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -156,7 +156,11 @@ class ReadBuilder:
...
def new_scan(self) -> TableScan: ...
def new_stream_scan(self) -> StreamTableScan:
- """Create a stateful stream scan retaining filters, read type and
ranges."""
+ """Create a stateful stream scan retaining filters, read type and
ranges.
+
+ REST query authorization refreshes on every plan/poll. Returned splits
+ retain that plan's rules; failures do not consume the failing snapshot.
+ """
...
def new_incremental_scan(
self,
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
index acddf033..5eff66b1 100644
--- a/crates/paimon/src/table/incremental_scan.rs
+++ b/crates/paimon/src/table/incremental_scan.rs
@@ -373,7 +373,6 @@ impl<'a> IncrementalScan<'a> {
/// must exist and supplies snapshot metadata. Snapshot deletion vectors
and
/// automatic global-index pruning do not apply to these historical events.
pub async fn plan_combined_delta(&self) -> crate::Result<Plan> {
-
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
let mode = self.resolve_mode();
if mode != IncrementalScanMode::Delta {
return Err(crate::Error::Unsupported {
@@ -388,7 +387,7 @@ impl<'a> IncrementalScan<'a> {
mode: IncrementalScanMode,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
-
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ let scan = self.scan.authorize_snapshot_read().await?;
// Row positions and shuffled chunks refer to one APPEND-delta row
// space. Changelog supports Java's file/bucket shard selection only.
if mode == IncrementalScanMode::Changelog
@@ -438,8 +437,7 @@ impl<'a> IncrementalScan<'a> {
})
.map(str::to_owned)
.collect();
- self.scan
- .plan_incremental_lists_with_trace(&end_snapshot, manifest_lists,
trace)
+ scan.plan_incremental_lists_with_trace(&end_snapshot, manifest_lists,
trace)
.await
}
diff --git a/crates/paimon/src/table/stream_table_scan.rs
b/crates/paimon/src/table/stream_table_scan.rs
index fea63186..96cadba6 100644
--- a/crates/paimon/src/table/stream_table_scan.rs
+++ b/crates/paimon/src/table/stream_table_scan.rs
@@ -121,7 +121,7 @@ impl StreamTableScan {
pub async fn plan(&mut self) -> Result<Option<Plan>> {
// Validate even when the table is empty or no follow-up is available.
- self.scan.validate_snapshot_read()?;
+ let scan = self.scan.authorize_snapshot_read().await?;
if self.next_snapshot_id.is_none() && !self.consumer_restored {
if let Some(consumer_id) = &self.consumer_id {
let next = self
@@ -143,7 +143,7 @@ impl StreamTableScan {
};
let next = snapshot.id() + 1;
let watermark = snapshot.watermark();
- let plan = self.plan_initial_snapshot(snapshot).await?;
+ let plan = self.plan_initial_snapshot(snapshot, scan).await?;
self.next_snapshot_id = Some(next);
self.watermark = watermark;
return Ok(Some(plan));
@@ -186,7 +186,7 @@ impl StreamTableScan {
}
let watermark = snapshot.watermark();
// Keep the failing snapshot retryable if planning or a callback
fails.
- let plan = self.scan.plan_selected_snapshot(snapshot, mode).await?;
+ let plan = scan.plan_selected_snapshot(snapshot, mode).await?;
self.next_snapshot_id = Some(id + 1);
self.watermark = watermark;
if !plan.splits().is_empty() {
@@ -195,9 +195,13 @@ impl StreamTableScan {
}
}
- async fn plan_initial_snapshot(&self, snapshot: Snapshot) -> Result<Plan> {
- let options = self.scan.snapshot_table().schema().options();
- let mut scan = self.scan.clone();
+ async fn plan_initial_snapshot(
+ &self,
+ snapshot: Snapshot,
+ mut scan: TableScan<'static>,
+ ) -> Result<Plan> {
+ let table = scan.snapshot_table().clone();
+ let options = table.schema().options();
let parse = |key, default| -> Result<i32> {
options
.get(key)
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 598e8bc9..4ae7b711 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/crates/paimon/src/table/table_read.rs
@@ -534,16 +534,24 @@ impl<'a> PaimonTableRead<'a> {
&self,
data_splits: &[DataSplit],
) -> crate::Result<ArrowRecordBatchStream> {
- // Streaming primary-key splits are read raw below, so decide here for
all.
- if self
- .ensure_authorized_by_splits(&self.table.schema.core_options(),
data_splits)?
- .is_some()
- {
- return Err(super::query_auth::unsupported(
- "a row-kind read cannot apply a row filter or column masking",
- ));
- }
+ let grant =
+
self.ensure_authorized_by_splits(&self.table.schema.core_options(),
data_splits)?;
let data_splits = self.matching_splits(data_splits)?;
+ if let Some(grant) = grant {
+ let mut read = self.clone();
+ // The metadata column travels with the Arrow batch through filters
+ // and masks, but is never an authorization operand or physical
field.
+ read.read_type.insert(0, row_kind_field());
+ return read.read_restricted(&data_splits, &grant, true);
+ }
+ self.read_row_kind_splits(&data_splits)
+ }
+
+ /// Read the already-authorized file set, preserving physical change kinds.
+ fn read_row_kind_splits(
+ &self,
+ data_splits: &[DataSplit],
+ ) -> crate::Result<ArrowRecordBatchStream> {
let schema = audit_schema_for_read_type(&self.read_type, false)?;
let (streaming, materialized): (Vec<_>, Vec<_>) = data_splits
.iter()
@@ -555,14 +563,14 @@ impl<'a> PaimonTableRead<'a> {
let mut streams = Vec::with_capacity(2);
if !materialized.is_empty() {
streams.push(prepend_insert_row_kind_stream(
- self.to_arrow(&materialized)?,
+ self.read_splits(&materialized,
&self.table.schema.core_options())?,
Arc::clone(&schema),
)?);
}
if !streaming.is_empty() {
if self.table.schema().primary_keys().is_empty() {
streams.push(prepend_insert_row_kind_stream(
- self.to_arrow(&streaming)?,
+ self.read_splits(&streaming,
&self.table.schema.core_options())?,
Arc::clone(&schema),
)?);
} else {
@@ -1122,7 +1130,7 @@ impl<'a> PaimonTableRead<'a> {
let grant = self.ensure_authorized_by_splits(&core_options,
data_splits)?;
let data_splits = self.matching_splits(data_splits)?;
match grant {
- Some(grant) => self.read_restricted(&data_splits, &grant),
+ Some(grant) => self.read_restricted(&data_splits, &grant, false),
None => self.read_splits(&data_splits, &core_options),
}
}
@@ -1132,6 +1140,7 @@ impl<'a> PaimonTableRead<'a> {
&self,
data_splits: &[DataSplit],
grant: &super::query_auth::QueryAuthGrant,
+ include_row_kind: bool,
) -> crate::Result<ArrowRecordBatchStream> {
use super::query_auth::{filter_batch, mask_batch, mask_inputs,
unsupported};
use crate::arrow::format::FilePredicates;
@@ -1240,7 +1249,14 @@ impl<'a> PaimonTableRead<'a> {
]));
inner.table = &unresolved_table;
}
- let stream = inner.read_splits(data_splits,
&inner.table.schema.core_options())?;
+ let stream = if include_row_kind {
+ // The source creates rowkind from _VALUE_KIND (or +I for snapshot
+ // rows); do not ask the physical format reader to load it.
+ inner.read_type.remove(0);
+ inner.read_row_kind_splits(data_splits)?
+ } else {
+ inner.read_splits(data_splits, &inner.table.schema.core_options())?
+ };
let filters = rules.filters.clone();
let batch_fields = physical.clone();
@@ -1801,16 +1817,20 @@ fn prepend_insert_row_kind(
})
}
+fn row_kind_field() -> DataField {
+ DataField::new(
+ ROW_KIND_FIELD_ID,
+ ROW_KIND_FIELD_NAME.to_string(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ )
+}
+
fn audit_schema_for_read_type(
read_type: &[DataField],
include_sequence: bool,
) -> crate::Result<Arc<ArrowSchema>> {
let mut fields = Vec::with_capacity(read_type.len() + 2);
- fields.push(DataField::new(
- ROW_KIND_FIELD_ID,
- ROW_KIND_FIELD_NAME.to_string(),
- DataType::VarChar(crate::spec::VarCharType::string_type()),
- ));
+ fields.push(row_kind_field());
if include_sequence {
fields.push(DataField::new(
SEQUENCE_NUMBER_FIELD_ID,
@@ -3198,12 +3218,11 @@ mod tests {
if message.contains("different rules")),
"one rule set per read, whatever the order"
);
+ assert!(matches!(read.to_arrow_with_row_kind(&splits),
+ Err(crate::Error::Unsupported { ref message }) if
message.contains("different rules")));
}
- // Streaming primary-key splits are read raw.
- assert!(matches!(
- read.to_arrow_with_row_kind(&[restricted]),
- Err(crate::Error::Unsupported { ref message }) if
message.contains("row-kind")
- ));
+ // Row-kind reads preserve the same grant requirement as normal reads.
+ assert!(read.to_arrow_with_row_kind(&[restricted]).is_ok());
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index fa84bf5e..345021b3 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -54,6 +54,7 @@ use crate::table::source::{
use crate::table::ScanTrace;
use futures::{StreamExt, TryStreamExt};
use indexmap::IndexMap;
+use std::borrow::Cow;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -1136,6 +1137,7 @@ impl<'a> TableScan<'a> {
scan_all_files: scan.scan_all_files,
incremental_split_mode: scan.incremental_split_mode,
projected_read_field_ids: scan.projected_read_field_ids,
+ query_auth_grant: scan.query_auth_grant,
})))
}
@@ -1146,6 +1148,19 @@ impl<'a> TableScan<'a> {
&scan.table
}
+ /// One authorization scope for a single plan/poll. The long-lived stream
+ /// retains the original ungranted scan so the next poll refreshes the
rules.
+ pub(super) async fn authorize_snapshot_read(&self) -> crate::Result<Self> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) => Ok(Self(TableScanKind::Paimon(
+ scan.authorized_scan().await?.into_owned(),
+ ))),
+ TableScanKind::Format(_) => Err(crate::Error::Unsupported {
+ message: "Snapshot authorization requires a Paimon
table".into(),
+ }),
+ }
+ }
+
pub(super) fn validate_snapshot_read(&self) -> crate::Result<()> {
let TableScanKind::Paimon(scan) = &self.0 else {
unreachable!("snapshot scans only contain Paimon tables")
@@ -1198,14 +1213,15 @@ impl<'a> TableScan<'a> {
scan.split_selection
.get_or_insert_with(Default::default)
.snapshot_mode = Some(mode);
- match mode {
+ let plan = match mode {
SnapshotScanMode::All => {
let fields = scan.projected_read_field_ids()?;
scan.plan_snapshot(snapshot, fields.as_ref(), None).await
}
SnapshotScanMode::Delta =>
scan.plan_snapshot_delta(&snapshot).await,
SnapshotScanMode::Changelog =>
scan.plan_snapshot_changelog(&snapshot, None).await,
- }
+ }?;
+ scan.finish_plan(plan).await
}
pub(crate) fn new(
@@ -1461,8 +1477,10 @@ impl<'a> TableScan<'a> {
) -> crate::Result<Plan> {
match &self.0 {
TableScanKind::Paimon(scan) => {
- scan.plan_incremental_lists(end_snapshot, manifest_lists,
trace)
- .await
+ let plan = scan
+ .plan_incremental_lists(end_snapshot, manifest_lists,
trace)
+ .await?;
+ scan.finish_plan(plan).await
}
TableScanKind::Format(_) => Err(crate::Error::Unsupported {
message: "Format tables do not support incremental batch
scans".to_string(),
@@ -1590,6 +1608,9 @@ struct PaimonTableScan<'a> {
scan_all_files: bool,
incremental_split_mode: Option<IncrementalSplitMode>,
projected_read_field_ids: Option<HashSet<i32>>,
+ /// Only set on an ephemeral copy after authorization. Never stored back
into
+ /// StreamTableScan, and never reconstructed from serialized split
metadata.
+ query_auth_grant: Option<Arc<super::query_auth::QueryAuthGrant>>,
}
impl<'a> PaimonTableScan<'a> {
@@ -1669,6 +1690,7 @@ impl<'a> PaimonTableScan<'a> {
scan_all_files: false,
incremental_split_mode: None,
projected_read_field_ids: None,
+ query_auth_grant: None,
}
}
@@ -1726,9 +1748,7 @@ impl<'a> PaimonTableScan<'a> {
/// for `scan.version`; the strict selectors mirror Java's typed
/// `scan.snapshot-id` / `scan.tag-name` handling.
pub async fn plan(&self) -> crate::Result<Plan> {
- let grant = self.authorize_query().await?;
- let restricted = grant.as_deref().map(|grant|
self.restricted_by(grant));
- let scan = restricted.as_ref().unwrap_or(self);
+ let scan = self.authorized_scan().await?;
if let Some((start, end)) = scan
.table
.schema()
@@ -1736,8 +1756,7 @@ impl<'a> PaimonTableScan<'a> {
.incremental_timestamp_window()?
{
let plan = scan.plan_timestamp_window(start, end, None).await?;
- scan.check_planned_files(&plan, grant.is_some()).await?;
- return Ok(plan.planned(grant));
+ return Ok(plan);
}
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
let plan = match
super::time_travel::resolve_snapshot(&scan.table).await? {
@@ -1747,15 +1766,12 @@ impl<'a> PaimonTableScan<'a> {
}
None => Plan::new(Vec::new()),
};
- scan.check_planned_files(&plan, grant.is_some()).await?;
- Ok(plan.planned(grant))
+ scan.finish_plan(plan).await
}
/// Plan the full scan and return metadata-pruning trace counters.
pub async fn plan_with_trace(&self) -> crate::Result<(Plan, ScanTrace)> {
- let grant = self.authorize_query().await?;
- let restricted = grant.as_deref().map(|grant|
self.restricted_by(grant));
- let scan = restricted.as_ref().unwrap_or(self);
+ let scan = self.authorized_scan().await?;
let mut trace = ScanTrace {
limit: scan.limit,
..Default::default()
@@ -1771,8 +1787,7 @@ impl<'a> PaimonTableScan<'a> {
.await?;
trace.snapshot_id = plan.snapshot_id();
trace.planned_data_file_bytes = plan.planned_data_file_bytes();
- scan.check_planned_files(&plan, grant.is_some()).await?;
- return Ok((plan.planned(grant), trace));
+ return Ok((plan, trace));
}
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
let plan = match
super::time_travel::resolve_snapshot(&scan.table).await? {
@@ -1790,12 +1805,11 @@ impl<'a> PaimonTableScan<'a> {
}
None => Plan::new(Vec::new()),
};
- scan.check_planned_files(&plan, grant.is_some()).await?;
- Ok((plan.planned(grant), trace))
+ Ok((scan.finish_plan(plan).await?, trace))
}
/// Keep pruning off masked columns and apply limits after row filtering.
- fn restricted_by(&self, grant: &super::query_auth::QueryAuthGrant) -> Self
{
+ fn restricted_by(&self, grant: &Arc<super::query_auth::QueryAuthGrant>) ->
Self {
let rules = grant.rules();
let fields = self.table.schema().fields();
let masked: HashSet<&str> = rules
@@ -1810,6 +1824,7 @@ impl<'a> PaimonTableScan<'a> {
.any(|name| masked.contains(name.as_str()))
};
let mut scan = self.clone();
+ scan.query_auth_grant = Some(Arc::clone(grant));
scan.data_predicates.retain(|p| !touches_masked(p));
if scan.bucket_predicate.as_ref().is_some_and(touches_masked) {
scan.bucket_predicate = None;
@@ -1930,6 +1945,26 @@ impl<'a> PaimonTableScan<'a> {
.await
}
+ async fn authorized_scan(&self) -> crate::Result<Cow<'_, Self>> {
+ // Timestamp-window planning delegates to combined incremental planning
+ // within the same operation. Reuse its scope, not another REST
request.
+ if self.query_auth_grant.is_some() {
+ self.validate_read_options()?;
+ return Ok(Cow::Borrowed(self));
+ }
+ let grant = self.authorize_query().await?;
+ Ok(match grant {
+ Some(grant) => Cow::Owned(self.restricted_by(&grant)),
+ None => Cow::Borrowed(self),
+ })
+ }
+
+ async fn finish_plan(&self, plan: Plan) -> crate::Result<Plan> {
+ self.check_planned_files(&plan, self.query_auth_grant.is_some())
+ .await?;
+ Ok(plan.planned(self.query_auth_grant.clone()))
+ }
+
/// Authorize this scan and return the grant for the caller to stamp.
async fn authorize_query(
&self,
@@ -2016,7 +2051,13 @@ impl<'a> PaimonTableScan<'a> {
/// included.
fn validate_read_options(&self) -> crate::Result<()> {
let options = CoreOptions::new(self.table.schema().options());
- options.ensure_read_authorized()?;
+ if !self
+ .query_auth_grant
+ .as_ref()
+ .is_some_and(|grant| grant.matches_table(&self.table))
+ {
+ options.ensure_read_authorized()?;
+ }
options.validate_data_file_path_directory()
}
diff --git a/crates/paimon/tests/rest_catalog/authorized_snapshot_scans.rs
b/crates/paimon/tests/rest_catalog/authorized_snapshot_scans.rs
new file mode 100644
index 00000000..75213081
--- /dev/null
+++ b/crates/paimon/tests/rest_catalog/authorized_snapshot_scans.rs
@@ -0,0 +1,409 @@
+// 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::*;
+
+async fn append(g: &Guarded, batch: RecordBatch) {
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file://{}", g._tmp.path().display()),
+ );
+ let catalog = FileSystemCatalog::new(options).unwrap();
+ let plain = catalog.get_table(&g.identifier).await.unwrap();
+ write_batch(&plain, batch, "auth-stream-append").await;
+}
+
+async fn read_frame(
+ builder: &paimon::table::ReadBuilder<'_>,
+ plan: &paimon::table::Plan,
+ kinds: bool,
+) -> paimon::Result<Vec<RecordBatch>> {
+ let read = builder.new_read()?;
+ if kinds {
+ read.to_arrow_with_row_kind(plan.splits())?
+ } else {
+ read.to_arrow(plan.splits())?
+ }
+ .try_collect()
+ .await
+}
+
+#[tokio::test]
+async fn
authorized_stream_refreshes_rules_but_old_frames_keep_captured_rules() {
+ let g = written("stream_refresh", people, vec![people_rows(1..=3)]).await;
+ g.set_rules(
+ &[int_leaf(0, "id", "GREATER_THAN", 1)],
+ &[("name", upper(1, "name"))],
+ );
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["id", "name"]).unwrap();
+ let mut stream = builder.new_stream_scan().unwrap();
+ let first = stream.plan().await.unwrap().unwrap();
+ assert_eq!(
+ ids(&read_frame(&builder, &first, false).await.unwrap()),
+ [2, 3]
+ );
+ assert_eq!(stream.checkpoint(), Some(2));
+ g.set_rules(
+ &[int_leaf(0, "id", "GREATER_OR_EQUAL", 5)],
+ &[("name", upper(1, "name"))],
+ );
+ append(&g, people_rows(4..=5)).await;
+ let next = stream.plan().await.unwrap().unwrap();
+ assert_eq!(
+ id_names(&read_frame(&builder, &next, true).await.unwrap()),
+ [(5, Some("ERIN".into()))]
+ );
+ assert_eq!(stream.checkpoint(), Some(3));
+ assert_eq!(
+ id_names(&read_frame(&builder, &first, true).await.unwrap()),
+ [(2, Some("BOB".into())), (3, Some("CAROL".into()))]
+ );
+ assert!(stream.plan().await.unwrap().is_none());
+ g.set_rules(&[], &[]);
+ append(&g, people_rows(6..=6)).await;
+ assert_eq!(
+ ids(
+ &read_frame(&builder, &stream.plan().await.unwrap().unwrap(),
false)
+ .await
+ .unwrap()
+ ),
+ [6]
+ );
+}
+
+#[tokio::test]
+async fn
authorization_failure_preserves_initial_follow_up_and_empty_poll_progress() {
+ let g = written("stream_retry", people, vec![people_rows(1..=2)]).await;
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["id", "name"]).unwrap();
+ let mut stream = builder.new_stream_scan().unwrap();
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_retry", vec!["id".into()]);
+ assert_forbidden(stream.plan().await.unwrap_err());
+ assert_eq!(stream.checkpoint(), None);
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_retry", vec!["id".into(),
"name".into()]);
+ assert_eq!(stream.plan().await.unwrap().unwrap().snapshot_id(), Some(1));
+ append(&g, people_rows(3..=3)).await;
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_retry", vec!["id".into()]);
+ assert_forbidden(stream.plan().await.unwrap_err());
+ assert_eq!(stream.checkpoint(), Some(2));
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_retry", vec!["id".into(),
"name".into()]);
+ assert_eq!(
+ ids(
+ &read_frame(&builder, &stream.plan().await.unwrap().unwrap(),
false)
+ .await
+ .unwrap()
+ ),
+ [3]
+ );
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_retry", vec!["id".into()]);
+ assert_forbidden(stream.plan().await.unwrap_err());
+ assert_eq!(stream.checkpoint(), Some(3));
+}
+
+fn keyed_people(options: &[(&str, &str)]) -> Schema {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::new(255).unwrap()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("changelog-producer", "input");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+}
+
+fn changes() -> RecordBatch {
+ let batch = people_batch(&[(1, "old"), (1, "new"), (2, "deleted")]);
+ let mut fields =
batch.schema().fields().iter().cloned().collect::<Vec<_>>();
+ fields.push(Arc::new(ArrowField::new(
+ "_VALUE_KIND",
+ ArrowDataType::Int8,
+ false,
+ )));
+ let mut columns = batch.columns().to_vec();
+ columns.push(Arc::new(arrow_array::Int8Array::from(vec![1, 2, 3])));
+ RecordBatch::try_new(Arc::new(ArrowSchema::new(fields)), columns).unwrap()
+}
+
+#[tokio::test]
+async fn
authorized_stream_keeps_physical_change_kinds_after_filters_and_masks() {
+ let g = written(
+ "stream_kinds",
+ keyed_people,
+ vec![people_batch(&[(1, "old"), (2, "deleted")]), changes()],
+ )
+ .await;
+ g.set_rules(
+ &[int_leaf(0, "id", "EQUAL", 1)],
+ &[("name", upper(1, "name"))],
+ );
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["name"]).unwrap();
+ let mut stream = builder.new_stream_scan().unwrap();
+ stream.restore(Some(2)).unwrap();
+ let plan = stream.plan().await.unwrap().unwrap();
+ let batches = read_frame(&builder, &plan, true).await.unwrap();
+ let rows = batches
+ .iter()
+ .flat_map(|batch| {
+ let kinds = column::<StringArray>(batch, "rowkind");
+ let names = column::<StringArray>(batch, "name");
+ (0..batch.num_rows())
+ .map(|i| (kinds.value(i).to_string(),
names.value(i).to_string()))
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(
+ rows,
+ [("-U".into(), "OLD".into()), ("+U".into(), "NEW".into())]
+ );
+ assert!(batches.iter().all(|batch| batch.num_columns() == 2));
+ let serialized = plan.splits()[0].serialize_split_v1().unwrap();
+ let detached =
paimon::table::DataSplit::deserialize_split_v1(&serialized).unwrap();
+ assert_refused(
+ builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(&[detached])
+ .err()
+ .expect("wire bytes cannot mint a grant"),
+ );
+ let mut empty_projection = g.table.new_read_builder();
+ empty_projection.with_projection(&[]).unwrap();
+ let batches = read_frame(&empty_projection, &plan, true).await.unwrap();
+ assert_eq!(
+ batches.iter().map(|batch| batch.num_rows()).sum::<usize>(),
+ 2
+ );
+ assert!(batches.iter().all(|batch| batch.num_columns() == 1));
+}
+
+#[tokio::test]
+async fn
authorized_row_kinds_keep_blob_resolution_after_filters_for_both_split_kinds() {
+ let (_source, good, missing) = blob_payloads();
+ for primary_key in [false, true] {
+ let g = blob_mask_rows(primary_key, vec![None, Some(&missing),
Some(&good)]).await;
+ g.set_rules(
+ &[
+ int_leaf(0, "id", "NOT_EQUAL", 2),
+ int_leaf(0, "id", "NOT_EQUAL", 5),
+ ],
+ &[("payload", field_ref_mask(1, "payload", "BLOB"))],
+ );
+ let builder = g.table.new_read_builder();
+ let mut scan = builder.new_stream_scan().unwrap();
+ let initial = scan.plan().await.unwrap().unwrap();
+ append(
+ &g,
+ RecordBatch::try_from_iter([
+ (
+ "id",
+ Arc::new(Int32Array::from(vec![4, 5])) as
arrow_array::ArrayRef,
+ ),
+ (
+ "payload",
+ Arc::new(LargeBinaryArray::from(vec![
+ Some(good.as_slice()),
+ Some(missing.as_slice()),
+ ])) as arrow_array::ArrayRef,
+ ),
+ (
+ "label",
+ Arc::new(StringArray::from(vec!["row4", "row5"])) as
arrow_array::ArrayRef,
+ ),
+ ])
+ .unwrap(),
+ )
+ .await;
+ let follow_up = scan.plan().await.unwrap().unwrap();
+ let splits = [initial.splits(), follow_up.splits()].concat();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(&splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(ids(&batches), [1, 3, 4]);
+ for batch in &batches {
+ assert!(column::<StringArray>(batch, "rowkind")
+ .iter()
+ .all(|kind| kind == Some("+I")));
+ for payload in column::<LargeBinaryArray>(batch, "payload")
+ .iter()
+ .flatten()
+ {
+ assert_eq!(payload, b"good");
+ }
+ }
+ let mut limited = g.table.new_read_builder();
+ limited.with_limit(1);
+ let batches: Vec<RecordBatch> = limited
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(&splits)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
1);
+ }
+}
+
+#[tokio::test]
+async fn
combined_authorized_events_filter_before_limit_and_stamp_every_split() {
+ for keyed in [false, true] {
+ let g = if keyed {
+ written(
+ "events_keyed",
+ keyed_people,
+ vec![people_rows(1..=2), people_rows(3..=4),
people_rows(5..=6)],
+ )
+ .await
+ } else {
+ written(
+ "events_append",
+ people,
+ vec![people_rows(1..=2), people_rows(3..=4),
people_rows(5..=6)],
+ )
+ .await
+ };
+ g.set_rules(
+ &[int_leaf(0, "id", "GREATER_THAN", 4)],
+ &[("name", upper(1, "name"))],
+ );
+ let mut builder = g.table.new_read_builder();
+ builder
+ .with_projection(&["id"])
+ .unwrap()
+ .with_limit(1)
+ .with_filter(
+ PredicateBuilder::new(g.table.schema().fields())
+ .equal("name", Datum::String("FRANK".into()))
+ .unwrap(),
+ );
+ let mode = if keyed {
+ paimon::table::IncrementalScanMode::Changelog
+ } else {
+ paimon::table::IncrementalScanMode::Delta
+ };
+ let plan = builder
+ .new_incremental_scan(mode, 0, 3)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert_eq!(ids(&read_frame(&builder, &plan, false).await.unwrap()),
[6]);
+ assert_eq!(ids(&read_frame(&builder, &plan, true).await.unwrap()),
[6]);
+ let requested = g.ctx.server.auth_selects("default",
g.identifier.object());
+ assert!(requested
+ .iter()
+ .all(|select| select.as_ref().unwrap().len() == 2));
+ // Java's high-level timestamp-window path shares this authorization
scope.
+ let window = g.table.copy_with_options(HashMap::from([(
+ "incremental-between-timestamp".into(),
+ "0,9223372036854775807".into(),
+ )]));
+ let mut window_builder = window.new_read_builder();
+ window_builder.with_projection(&["id"]).unwrap();
+ let plan = window_builder.new_scan().plan().await.unwrap();
+ assert_eq!(
+ ids(&read_frame(&window_builder, &plan, true).await.unwrap()),
+ [5, 6]
+ );
+ // Zero-length ranges must still request column authorization.
+ g.ctx
+ .server
+ .set_column_auth("default", g.identifier.object(),
vec!["id".into()]);
+ assert_forbidden(
+ builder
+ .new_incremental_scan(mode, 3, 3)
+ .plan_combined()
+ .await
+ .unwrap_err(),
+ );
+ }
+}
+
+#[tokio::test]
+async fn stream_authorizes_empty_tables_and_filtered_frames_before_advancing()
{
+ let g = written("stream_empty", people, vec![]).await;
+ let mut builder = g.table.new_read_builder();
+ builder.with_projection(&["id", "name"]).unwrap();
+ let mut stream = builder.new_stream_scan().unwrap();
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_empty", vec!["id".into()]);
+ assert_forbidden(stream.plan().await.unwrap_err());
+ assert_eq!(stream.checkpoint(), None);
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_empty", vec!["id".into(),
"name".into()]);
+ assert!(stream.plan().await.unwrap().is_none());
+ append(&g, people_rows(1..=2)).await;
+ stream.with_bucket_filter(|_| Ok(false));
+ let frame = stream.plan().await.unwrap().unwrap();
+ assert!(frame.splits().is_empty());
+ assert_eq!(stream.checkpoint(), Some(2));
+ append(&g, people_rows(3..=3)).await;
+ assert!(stream.plan().await.unwrap().is_none());
+ assert_eq!(stream.checkpoint(), Some(3));
+ g.ctx
+ .server
+ .set_column_auth("default", "stream_empty", vec!["id".into()]);
+ assert_forbidden(stream.plan().await.unwrap_err());
+ assert_eq!(stream.checkpoint(), Some(3));
+}
+
+#[tokio::test]
+async fn stream_rejects_schema_or_identity_drift_without_consuming_the_frame()
{
+ for recreate in [false, true] {
+ let g = written("stream_drift", people,
vec![people_rows(1..=2)]).await;
+ let mut stream = g.table.new_stream_scan().unwrap();
+ if recreate {
+ g.ctx
+ .server
+ .set_table_uuid("default", "stream_drift", "replacement-uuid");
+ } else {
+ g.ctx
+ .server
+ .set_table_schema_id("default", "stream_drift",
people(GUARDED), 7);
+ }
+ assert_drifted(
+ stream.plan().await.unwrap_err(),
+ if recreate {
+ "now resolves to uuid"
+ } else {
+ "now resolves to schema"
+ },
+ );
+ assert_eq!(stream.checkpoint(), None);
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 8251943b..557ec5b0 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -5830,3 +5830,7 @@ async fn test_query_auth_explicit_partition_selector() {
assert_forbidden(selector_only.new_scan().plan().await.unwrap_err());
}
}
+
+#[cfg(not(windows))]
+#[path = "rest_catalog/authorized_snapshot_scans.rs"]
+mod authorized_snapshot_scans;
