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 a2da430e feat(core): add stateful StreamTableScan for Native streaming
reads (#1075)
a2da430e is described below
commit a2da430e89c10593d7ea4fcae8a39ce96916a43f
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 19:19:10 2026 +0800
feat(core): add stateful StreamTableScan for Native streaming reads (#1075)
---
.../python/python/pypaimon_rust/datafusion.pyi | 17 +
bindings/python/src/context.rs | 1 +
bindings/python/src/read.rs | 146 +++++-
bindings/python/src/table.rs | 4 +
bindings/python/tests/test_stream_table_scan.py | 226 ++++++++
crates/paimon/src/lib.rs | 3 +
crates/paimon/src/table/consumer_manager.rs | 60 ++-
crates/paimon/src/table/data_file_reader.rs | 46 +-
crates/paimon/src/table/mod.rs | 7 +
crates/paimon/src/table/read_builder.rs | 5 +
crates/paimon/src/table/snapshot_scan_tests.rs | 566 +++++++++++++++++++++
crates/paimon/src/table/stream_table_scan.rs | 252 +++++++++
crates/paimon/src/table/table_scan.rs | 244 ++++++++-
crates/paimon/tests/stream_table_scan_test.rs | 466 +++++++++++++++++
14 files changed, 2020 insertions(+), 23 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index adfa1c75..ff4453ca 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -80,6 +80,19 @@ class TableScan:
...
def plan(self) -> Plan: ...
+class StreamTableScan:
+ """Stateful continuous scan; checkpoint is the next snapshot to consume."""
+
+ def with_bucket_filter(self, filter: Optional[Callable[[int], bool]] =
None) -> StreamTableScan: ...
+ def with_consumer_id(self, consumer_id: str) -> StreamTableScan: ...
+ def with_shard(self, index: int, count: int) -> StreamTableScan: ...
+ def plan(self) -> Optional[Plan]: ...
+ def checkpoint(self) -> Optional[int]: ...
+ def watermark(self) -> Optional[int]: ...
+ def restore(self, next_snapshot_id: Optional[int] = None) -> None: ...
+ def notify_checkpoint_complete(self, next_snapshot: Optional[int]) ->
None: ...
+
+
class RecordBatchReader:
def __iter__(self) -> "RecordBatchReader": ...
def __next__(self) -> pyarrow.RecordBatch: ...
@@ -130,6 +143,9 @@ class ReadBuilder:
"""Set Data Evolution row ranges. Empty selects no rows; format tables
are unsupported."""
...
def new_scan(self) -> TableScan: ...
+ def new_stream_scan(self) -> StreamTableScan:
+ """Create a stateful stream scan retaining filters, read type and
ranges."""
+ ...
def new_incremental_scan(
self,
start_snapshot_id: int,
@@ -196,6 +212,7 @@ class Table:
def location(self) -> str: ...
def schema(self) -> TableSchema: ...
def new_read_builder(self, options: Optional[Dict[str, str]] = None) ->
ReadBuilder: ...
+ def new_stream_scan(self) -> StreamTableScan: ...
def new_batch_write_builder(self) -> "BatchWriteBuilder": ...
def new_postpone_fixed_bucket_write_builder(self) ->
"PostponeFixedBucketWriteBuilder": ...
def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 3bb58b04..2bc67988 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -669,6 +669,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::table::PyTable>()?;
this.add_class::<crate::read::PyReadBuilder>()?;
this.add_class::<crate::read::PyTableScan>()?;
+ this.add_class::<crate::read::PyStreamTableScan>()?;
this.add_class::<crate::read::PyPlan>()?;
this.add_class::<crate::read::PyTableRead>()?;
this.add_class::<crate::read::PyRecordBatchReader>()?;
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 02bd66d1..deb87eca 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -23,11 +23,14 @@ use arrow::pyarrow::ToPyArrow;
use arrow::record_batch::RecordBatch;
use futures::TryStreamExt;
use paimon::spec::{is_variant_extraction_row, DataField, DataType, Predicate,
RowType};
-use paimon::table::{ArrowRecordBatchStream, DataSplit, IncrementalScanMode,
RowRange, Table};
+use paimon::table::{
+ ArrowRecordBatchStream, DataSplit, IncrementalScanMode, RowRange,
StreamTableScan, Table,
+};
use paimon_datafusion::runtime::runtime;
-use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
+use pyo3::exceptions::{PyBaseException, PyRuntimeError, PyTypeError,
PyValueError};
use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyDict};
+use pyo3::{PyTraverseError, PyVisit};
use tokio::sync::Notify;
use crate::error::to_py_err;
@@ -322,6 +325,18 @@ impl PyReadBuilder {
}
}
+ /// Create a stateful Java-compatible stream scan.
+ pub(crate) fn new_stream_scan(&self) -> PyResult<PyStreamTableScan> {
+ Ok(PyStreamTableScan {
+ scan: self
+ .new_scan()
+ .read_builder()?
+ .new_stream_scan()
+ .map_err(to_py_err)?,
+ bucket_filter:
Arc::new(Mutex::new(PyBucketFilterState::default())),
+ })
+ }
+
/// Plan physical changes in (start_snapshot_id, end_snapshot_id] as one
/// ordinary split plan. Mode is `delta` by default; `changelog` reads
/// changelog manifest files and `auto` follows the table's producer.
@@ -469,6 +484,133 @@ impl PyTableScan {
}
}
+/// A thin wrapper: planning, cursor and consumer progress belong to core.
+#[pyclass(name = "StreamTableScan", module = "pypaimon_rust.datafusion")]
+pub struct PyStreamTableScan {
+ scan: StreamTableScan,
+ bucket_filter: Arc<Mutex<PyBucketFilterState>>,
+}
+
+// The wrapper and core closure share one Python reference per object, so GC
+// can traverse every retained reference exactly once through the wrapper.
+#[derive(Default)]
+struct PyBucketFilterState {
+ filter: Option<Py<PyAny>>,
+ error: Option<Py<PyBaseException>>,
+}
+
+#[pymethods]
+impl PyStreamTableScan {
+ fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> {
+ let state = self.bucket_filter.lock().unwrap();
+ visit.call(&state.filter)?;
+ visit.call(&state.error)
+ }
+
+ fn __clear__(&mut self) {
+ let references = {
+ let mut state = self.bucket_filter.lock().unwrap();
+ (state.filter.take(), state.error.take())
+ };
+ // Drop Python references outside the lock: destructors may re-enter
+ // Python. The core closure sees an empty filter and retains no cycle.
+ drop(references);
+ }
+
+ #[pyo3(signature = (filter = None))]
+ fn with_bucket_filter(
+ mut slf: PyRefMut<'_, Self>,
+ filter: Option<Py<PyAny>>,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ if filter
+ .as_ref()
+ .is_some_and(|filter| !filter.bind(slf.py()).is_callable())
+ {
+ return Err(PyTypeError::new_err("bucket filter must be callable"));
+ }
+ let previous_filter =
+ std::mem::replace(&mut slf.bucket_filter.lock().unwrap().filter,
filter);
+ drop(previous_filter);
+ let state = Arc::clone(&slf.bucket_filter);
+ slf.scan.with_bucket_filter(move |bucket| {
+ Python::attach(|py| {
+ // Release the lock before invoking Python: callbacks may run
+ // GC, which traverses this same state.
+ let filter = state
+ .lock()
+ .unwrap()
+ .filter
+ .as_ref()
+ .map(|filter| filter.clone_ref(py));
+ let Some(filter) = filter else {
+ return Ok(true);
+ };
+ filter
+ .call1(py, (bucket,))
+ .and_then(|result| result.is_truthy(py))
+ .map_err(|error| {
+ let error = error.into_value(py);
+ let previous_error =
state.lock().unwrap().error.replace(error);
+ drop(previous_error);
+ paimon::Error::UnexpectedError {
+ message: "Python bucket filter failed".into(),
+ source: None,
+ }
+ })
+ })
+ });
+ Ok(slf)
+ }
+
+ fn with_consumer_id(
+ mut slf: PyRefMut<'_, Self>,
+ consumer_id: String,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ slf.scan.with_consumer_id(consumer_id).map_err(to_py_err)?;
+ Ok(slf)
+ }
+
+ fn with_shard(
+ mut slf: PyRefMut<'_, Self>,
+ index: usize,
+ count: usize,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ slf.scan.with_shard(index, count).map_err(to_py_err)?;
+ Ok(slf)
+ }
+
+ fn checkpoint(&self) -> Option<i64> {
+ self.scan.checkpoint()
+ }
+
+ fn watermark(&self) -> Option<i64> {
+ self.scan.watermark()
+ }
+
+ #[pyo3(signature = (next_snapshot_id = None))]
+ fn restore(&mut self, next_snapshot_id: Option<i64>) -> PyResult<()> {
+ self.scan.restore(next_snapshot_id).map_err(to_py_err)
+ }
+
+ fn notify_checkpoint_complete(
+ &self,
+ py: Python<'_>,
+ next_snapshot: Option<i64>,
+ ) -> PyResult<()> {
+ py.detach(||
runtime().block_on(self.scan.notify_checkpoint_complete(next_snapshot)))
+ .map_err(to_py_err)
+ }
+
+ fn plan(&mut self, py: Python<'_>) -> PyResult<Option<PyPlan>> {
+ let plan = py.detach(|| runtime().block_on(self.scan.plan()));
+ let error = self.bucket_filter.lock().unwrap().error.take();
+ if let Some(error) = error {
+ return Err(PyErr::from_value(error.into_bound(py).into_any()));
+ }
+ plan.map(|plan| plan.map(PyPlan::from)).map_err(to_py_err)
+ }
+}
+
#[pymethods]
impl PyTableScan {
/// Select a half-open range of append-table row positions.
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index ed19d070..7dbee4d3 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -175,6 +175,10 @@ impl PyTable {
PyBatchWriteBuilder::new(Arc::clone(&self.inner))
}
+ fn new_stream_scan(&self) -> PyResult<crate::read::PyStreamTableScan> {
+ PyReadBuilder::new(Arc::clone(&self.inner)).new_stream_scan()
+ }
+
fn new_postpone_fixed_bucket_write_builder(
&self,
) -> PyResult<PyPostponeFixedBucketWriteBuilder> {
diff --git a/bindings/python/tests/test_stream_table_scan.py
b/bindings/python/tests/test_stream_table_scan.py
new file mode 100644
index 00000000..2e81b69a
--- /dev/null
+++ b/bindings/python/tests/test_stream_table_scan.py
@@ -0,0 +1,226 @@
+# 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.
+
+"""Stateful core streaming through the Python wrapper."""
+
+import gc
+import weakref
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.datafusion import PaimonCatalog, StreamTableScan, SQLContext
+
+
[email protected]
+def table(tmp_path):
+ options = {"warehouse": str(tmp_path)}
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", options)
+ ctx.sql("CREATE SCHEMA paimon.snapdb")
+ ctx.sql("CREATE TABLE paimon.snapdb.t (id INT, value INT, PRIMARY KEY
(id)) "
+ "WITH ('bucket'='4', 'changelog-producer'='input', "
+ "'source.split.target-size'='1 b',
'source.split.open-file-cost'='1 b')")
+ table = PaimonCatalog(options).get_table("snapdb.t")
+ for value in (100, 200):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.record_batch({'id': pa.array(range(32),
type=pa.int32()),
+ 'value': pa.array(range(value, value +
32), type=pa.int32())}))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ return table
+
+
+def _rows(builder, plan):
+ return sorted((row['id'], row['value'])
+ for batch in builder.new_read().read(plan.splits()) for row
in batch.to_pylist())
+
+
[email protected]('follow_up', [False, True])
+def test_cursor_restore_and_initial_vs_follow_up(table, follow_up):
+ builder = table.new_read_builder()
+ scan = builder.new_stream_scan()
+ assert isinstance(scan, StreamTableScan)
+ assert scan.checkpoint() is None
+ if follow_up:
+ scan.restore(1)
+ plan = scan.plan()
+ expected_id = 1 if follow_up else 2
+ assert plan.snapshot_id() == expected_id
+ assert _rows(builder, plan) == [(i, (100 if follow_up else 200) + i) for i
in range(32)]
+ assert all(split.is_streaming() == follow_up for split in plan.splits())
+ assert scan.checkpoint() == expected_id + 1
+ if follow_up:
+ assert scan.plan().snapshot_id() == 2
+ assert scan.plan() is None
+ assert scan.checkpoint() == 3
+ scan.restore(1)
+ assert scan.plan().snapshot_id() == 1
+ scan.restore(None)
+ assert scan.plan().snapshot_id() == 2
+
+
[email protected]('follow_up', [False, True])
+def test_bucket_callback_replacement_clearing_and_shard(table, follow_up):
+ builder = table.new_read_builder()
+ scan = builder.new_stream_scan().with_bucket_filter(lambda bucket: False)
+ scan.restore(2 if follow_up else None)
+ plan = scan.plan()
+ assert plan is None if follow_up else plan.splits() == []
+ scan.with_bucket_filter(lambda bucket: bucket % 2 == 0)
+ scan.restore(2 if follow_up else None)
+ selected_rows = _rows(builder, scan.plan())
+ assert 0 < len(selected_rows) < 32
+ shard = builder.new_stream_scan().with_shard(0, 2)
+ shard.restore(2 if follow_up else None)
+ assert selected_rows == _rows(builder, shard.plan())
+ scan.with_bucket_filter(None)
+ scan.restore(2 if follow_up else None)
+ assert len(_rows(builder, scan.plan())) == 32
+ union = []
+ for index in range(2):
+ shard = builder.new_stream_scan().with_shard(index, 2)
+ shard.restore(2 if follow_up else None)
+ union.extend(_rows(builder, shard.plan()))
+ assert sorted(union) == [(i, 200 + i) for i in range(32)]
+
+
[email protected]('bound_method', [False, True])
+def test_bucket_callback_owner_cycle_is_collected(table, bound_method):
+ class Owner:
+ def __init__(self):
+ self.scan = table.new_stream_scan()
+ self.scan.with_bucket_filter(self.select if bound_method else self)
+
+ def select(self, bucket):
+ return True
+
+ def __call__(self, bucket):
+ return self.select(bucket)
+
+ owner = Owner()
+ reference = weakref.ref(owner)
+ # A live owner must keep the callback and its scan usable during GC.
+ gc.collect()
+ assert owner.scan.plan().snapshot_id() == 2
+ del owner
+ gc.collect()
+ assert reference() is None
+
+
[email protected]('clear', [False, True])
+def test_bucket_callback_owner_is_released_on_replacement(table, clear):
+ class Owner:
+ def select(self, bucket):
+ gc.collect()
+ return True
+
+ owner = Owner()
+ reference = weakref.ref(owner)
+ scan = table.new_stream_scan().with_bucket_filter(owner.select)
+ del owner
+ gc.collect()
+ assert reference() is not None
+ assert scan.plan().snapshot_id() == 2
+ scan.with_bucket_filter(None if clear else lambda bucket: True)
+ gc.collect()
+ assert reference() is None
+ scan.restore(None)
+ assert scan.plan().snapshot_id() == 2
+
+
[email protected]('follow_up', [False, True])
+def test_callback_exception_preserves_identity_and_checkpoint(table,
follow_up):
+ calls = []
+ failure = LookupError('bucket callback failed')
+
+ def select(bucket):
+ calls.append(bucket)
+ if len(calls) == 2:
+ raise failure
+ return True
+
+ scan = table.new_stream_scan().with_bucket_filter(select)
+ scan.restore(1 if follow_up else None)
+ with pytest.raises(LookupError) as caught:
+ scan.plan()
+ assert caught.value is failure
+ assert len(calls) == 2
+ assert scan.checkpoint() == (1 if follow_up else None)
+ assert scan.plan().snapshot_id() == (1 if follow_up else 2)
+
+
[email protected]('follow_up', [False, True])
+def test_projection_and_predicate_come_from_read_builder(table, follow_up):
+ builder =
(table.new_read_builder().with_projection(['value']).with_filter({
+ 'method': 'equal', 'field': 'id', 'literals': [2]}))
+ scan = builder.new_stream_scan()
+ scan.restore(1 if follow_up else None)
+ plan = scan.plan()
+ assert [row for batch in builder.new_read().read(plan.splits()) for row in
batch.to_pylist()] == [
+ {'value': 102 if follow_up else 202}]
+
+
+def test_consumer_is_only_persisted_by_explicit_acknowledgement(table):
+ first = table.new_stream_scan().with_consumer_id('job')
+ first.restore(1)
+ assert first.plan().snapshot_id() == 1
+ # Planning alone does not persist a consumer.
+ unacknowledged = table.new_stream_scan().with_consumer_id('job')
+ assert unacknowledged.plan().snapshot_id() == 2
+ assert not unacknowledged.plan()
+ first.notify_checkpoint_complete(first.checkpoint())
+ resumed = table.new_stream_scan().with_consumer_id('job')
+ plan = resumed.plan()
+ assert plan.snapshot_id() == 2
+ assert all(split.is_streaming() for split in plan.splits())
+ assert resumed.checkpoint() == 3
+
+
+def test_missing_checkpoint_waits_then_rejects_out_of_range(table):
+ scan = table.new_read_builder().with_limit(0).new_stream_scan()
+ scan.restore(12345)
+ for _ in range(15):
+ assert scan.plan() is None
+ assert scan.checkpoint() == 12345
+ with pytest.raises(ValueError, match='outside the available snapshots'):
+ scan.plan()
+ assert scan.checkpoint() == 12345
+
+
+def test_invalid_configuration_keeps_the_scan_usable(table):
+ scan = table.new_stream_scan()
+ for value in [0, -1]:
+ with pytest.raises(ValueError, match='snapshot id'):
+ scan.restore(value)
+ for index, count in [(0, 0), (2, 2)]:
+ with pytest.raises(ValueError, match='shard'):
+ scan.with_shard(index, count)
+ with pytest.raises(TypeError, match='callable'):
+ scan.with_bucket_filter(42)
+ for value in ['', '.', '..', 'x/../../snapshot/snapshot-1', r'x\y']:
+ with pytest.raises(ValueError, match='consumer id'):
+ scan.with_consumer_id(value)
+ assert scan.plan().snapshot_id() == 2
+
+
+def test_snapshot_reader_is_an_internal_component():
+ from pypaimon_rust import datafusion
+ assert not hasattr(datafusion, 'SnapshotReader')
+ assert not hasattr(datafusion.Table, 'new_snapshot_reader')
+ assert not hasattr(datafusion.ReadBuilder, 'new_snapshot_reader')
diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs
index 1d65128f..146c5e2b 100644
--- a/crates/paimon/src/lib.rs
+++ b/crates/paimon/src/lib.rs
@@ -62,3 +62,6 @@ pub use table::{
pub use table::{
HybridSearchBuilder, HybridSearchRanker, HybridSearchRoute,
HybridSearchRouteKind,
};
+
+#[cfg(test)]
+extern crate self as paimon;
diff --git a/crates/paimon/src/table/consumer_manager.rs
b/crates/paimon/src/table/consumer_manager.rs
index 35cb055b..20e1a084 100644
--- a/crates/paimon/src/table/consumer_manager.rs
+++ b/crates/paimon/src/table/consumer_manager.rs
@@ -22,7 +22,7 @@ use std::time::Duration;
use crate::io::FileIO;
use futures::{stream, StreamExt, TryStreamExt};
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
const CONSUMER_DIR: &str = "consumer";
const CONSUMER_PREFIX: &str = "consumer-";
@@ -53,6 +53,19 @@ impl ConsumerManager {
)
}
+ pub(super) fn validate_consumer_id(consumer_id: &str) -> crate::Result<()>
{
+ if consumer_id.is_empty()
+ || matches!(consumer_id, "." | "..")
+ || consumer_id.contains(['/', '\\'])
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "consumer id must be a non-empty name without path
separators".into(),
+ source: None,
+ });
+ }
+ Ok(())
+ }
+
/// Read one consumer's next snapshot id.
pub async fn get(&self, consumer_id: &str) -> crate::Result<Option<i64>> {
let ids = HashSet::from([consumer_id]);
@@ -64,6 +77,27 @@ impl ConsumerManager {
.map(|(_, next_snapshot)| next_snapshot))
}
+ /// Persist an explicitly acknowledged next snapshot, like Java
resetConsumer.
+ pub async fn reset(&self, consumer_id: &str, next_snapshot: i64) ->
crate::Result<()> {
+ Self::validate_consumer_id(consumer_id)?;
+ if next_snapshot < 1 {
+ return Err(crate::Error::DataInvalid {
+ message: "consumer next snapshot must be positive".into(),
+ source: None,
+ });
+ }
+ let directory = format!("{}/{}", self.table_path, CONSUMER_DIR);
+ self.file_io.mkdirs(&directory).await?;
+ let path = format!("{directory}/{CONSUMER_PREFIX}{consumer_id}");
+ let bytes = serde_json::to_vec(&Consumer { next_snapshot
}).map_err(|error| {
+ crate::Error::DataInvalid {
+ message: format!("consumer JSON invalid: {error}"),
+ source: Some(Box::new(error)),
+ }
+ })?;
+ self.file_io.new_output(&path)?.write(bytes.into()).await
+ }
+
async fn read_path(&self, consumer_id: &str, path: &str) ->
crate::Result<Option<i64>> {
let input = self.file_io.new_input(path)?;
for attempt in 0..READ_RETRIES {
@@ -143,7 +177,7 @@ impl ConsumerManager {
}
}
-#[derive(Deserialize)]
+#[derive(Deserialize, Serialize)]
struct Consumer {
#[serde(rename = "nextSnapshot")]
next_snapshot: i64,
@@ -156,6 +190,28 @@ mod tests {
use super::*;
use crate::io::FileIOBuilder;
+ #[tokio::test]
+ async fn reset_is_explicit_branch_scoped_and_rejects_path_traversal() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let manager = ConsumerManager::new(file_io.clone(),
"memory:/consumer-reset".into());
+ let branch = manager.with_branch("dev");
+ for invalid in ["", ".", "..", "x/../../snapshot/snapshot-1", "x\\y"] {
+ assert!(manager.reset(invalid, 2).await.is_err());
+ }
+ assert!(!file_io
+ .new_input("memory:/consumer-reset/snapshot/snapshot-1")
+ .unwrap()
+ .exists()
+ .await
+ .unwrap());
+ assert!(manager.reset("job", 0).await.is_err());
+ branch.reset("job", 2).await.unwrap();
+ assert_eq!(branch.get("job").await.unwrap(), Some(2));
+ assert_eq!(manager.get("job").await.unwrap(), None);
+ branch.reset("job", 3).await.unwrap();
+ assert_eq!(branch.get("job").await.unwrap(), Some(3));
+ }
+
#[tokio::test]
async fn retries_a_consumer_being_overwritten() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index d12e49df..4c32b659 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1188,18 +1188,31 @@ fn read_data_fields(
}
}
for expected in expected_fields {
- if expected.id() == crate::spec::SEQUENCE_NUMBER_FIELD_ID
- && expected.name() == crate::spec::SEQUENCE_NUMBER_FIELD_NAME
+ if is_physical_system_field(expected)
&& !read_fields.iter().any(|field| field.id() == expected.id())
{
- // Row tracking stores this physical system column outside the
- // logical table schema, including alongside selected MAP keys.
+ // Row tracking and KV files store these columns outside the
logical
+ // table schema. Keep them during MAP pruning/schema evolution too;
+ // an audit reader must receive the stored _VALUE_KIND.
read_fields.push(expected.clone());
}
}
Ok(read_fields)
}
+fn is_physical_system_field(field: &DataField) -> bool {
+ matches!(
+ (field.id(), field.name()),
+ (
+ crate::spec::SEQUENCE_NUMBER_FIELD_ID,
+ crate::spec::SEQUENCE_NUMBER_FIELD_NAME
+ ) | (
+ crate::spec::VALUE_KIND_FIELD_ID,
+ crate::spec::VALUE_KIND_FIELD_NAME
+ )
+ )
+}
+
fn prune_data_type(
read_type: &DataType,
data_type: &DataType,
@@ -1684,6 +1697,31 @@ mod row_tests {
assert_eq!(read[1], sequence);
}
+ #[test]
+ fn schema_pruning_keeps_value_kind_but_does_not_bind_an_alias() {
+ let value_kind = field(
+ crate::spec::VALUE_KIND_FIELD_ID,
+ crate::spec::VALUE_KIND_FIELD_NAME,
+ DataType::TinyInt(crate::spec::TinyIntType::with_nullable(false)),
+ );
+ let read = read_data_fields(&[], std::slice::from_ref(&value_kind),
false).unwrap();
+ assert_eq!(read, vec![value_kind]);
+ for alias in [
+ field(
+ 10,
+ crate::spec::VALUE_KIND_FIELD_NAME,
+ DataType::Int(IntType::new()),
+ ),
+ field(
+ crate::spec::VALUE_KIND_FIELD_ID,
+ "kind",
+ DataType::Int(IntType::new()),
+ ),
+ ] {
+ assert!(read_data_fields(&[], &[alias],
false).unwrap().is_empty());
+ }
+ }
+
#[test]
fn read_data_fields_preserves_variant_extraction_row_type() {
let configured = DataType::Row(RowType::new(vec
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index dfffa31d..ce971a17 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -298,6 +298,11 @@ impl<'a> ReadBuilder<'a> {
}
}
+ /// Create a stateful streaming scan using the configured filter and read
type.
+ pub fn new_stream_scan(&self) -> Result<super::StreamTableScan> {
+ super::StreamTableScan::new(self.new_scan())
+ }
+
/// Create a batch incremental scan over snapshot id range
/// `(start_exclusive, end_inclusive]`.
///
diff --git a/crates/paimon/src/table/snapshot_scan_tests.rs
b/crates/paimon/src/table/snapshot_scan_tests.rs
new file mode 100644
index 00000000..50910988
--- /dev/null
+++ b/crates/paimon/src/table/snapshot_scan_tests.rs
@@ -0,0 +1,566 @@
+// 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.
+
+#[path = "../../tests/common/mod.rs"]
+mod common;
+
+use super::SnapshotScanMode;
+use arrow_array::{Int32Array, RecordBatch};
+use common::incremental_helpers::{
+ make_batch, make_batch_with_kinds, memory_table, persist_table_schema,
setup_dirs, write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, IntType, Schema, TableSchema};
+use paimon::table::{IncrementalScanMode, Plan, Table, TableScan};
+use std::collections::HashMap;
+
+async fn table(name: &str, engine: &str, producer: &str, buckets: i32) ->
Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("changelog-producer", producer)
+ .option("source.split.target-size", "1 b")
+ .option("source.split.open-file-cost", "1 b");
+ match engine {
+ "append" => {
+ schema = schema.option("bucket", "-1");
+ }
+ "evolution" => {
+ schema = schema
+ .option("bucket", "-1")
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true");
+ }
+ _ => {
+ schema = schema
+ .primary_key(["id"])
+ .option("bucket", buckets.to_string())
+ .option("merge-engine", engine);
+ }
+ }
+ if engine == "aggregation" {
+ schema = schema.option("fields.value.aggregate-function", "sum");
+ }
+ let path = format!("memory:/snapshot_scan/{name}");
+ let (io, table) = memory_table(&path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ table
+}
+
+// Snapshot loading is test fixture setup; the production planner receives an
+// already selected Snapshot from the batch/stream scanning policy.
+async fn snapshot_plan(
+ scan: TableScan<'_>,
+ snapshot_id: Option<i64>,
+ mode: SnapshotScanMode,
+) -> paimon::Result<Plan> {
+ scan.validate_snapshot_read()?;
+ let manager = scan.snapshot_table().snapshot_manager();
+ let snapshot = match snapshot_id {
+ Some(id) => Some(manager.get_snapshot(id).await?),
+ None => manager.get_latest_snapshot().await?,
+ };
+ match snapshot {
+ Some(snapshot) => scan.plan_selected_snapshot(snapshot, mode).await,
+ None => Ok(Plan::new(Vec::new())),
+ }
+}
+
+async fn rows(table: &Table, plan: &Plan) -> Vec<(i32, i32)> {
+ let batches: Vec<RecordBatch> = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ rows.extend((0..batch.num_rows()).map(|row| (ids.value(row),
values.value(row))));
+ }
+ rows.sort_unstable();
+ rows
+}
+
+#[tokio::test]
+async fn
all_pins_snapshot_and_keeps_batch_merge_semantics_including_level_zero() {
+ for engine in [
+ "deduplicate",
+ "first-row",
+ "partial-update",
+ "aggregation",
+ "append",
+ "evolution",
+ ] {
+ let table = table(engine, engine, "none", 1).await;
+ write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await;
+ write_batch(&table, &make_batch(vec![1, 3], vec![11, 30])).await;
+ write_batch(&table, &make_batch(vec![4], vec![40])).await;
+ let plan = snapshot_plan(
+ table.new_read_builder().new_scan(),
+ Some(2),
+ SnapshotScanMode::All,
+ )
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert!(!plan.splits().is_empty());
+ assert!(plan.splits().iter().all(|split| !split.is_streaming()));
+ let expected = match engine {
+ "append" | "evolution" => vec![(1, 10), (1, 11), (2, 20), (3, 30)],
+ "first-row" => vec![(1, 10), (2, 20), (3, 30)],
+ "aggregation" => vec![(1, 21), (2, 20), (3, 30)],
+ _ => vec![(1, 11), (2, 20), (3, 30)],
+ };
+ assert_eq!(rows(&table, &plan).await, expected, "{engine}");
+ let current = snapshot_plan(
+ table.new_read_builder().new_scan(),
+ None,
+ SnapshotScanMode::All,
+ )
+ .await
+ .unwrap();
+ assert_eq!(current.snapshot_id(), Some(3));
+ assert!(rows(&table, ¤t).await.contains(&(4, 40)));
+ }
+}
+
+#[tokio::test]
+async fn explicit_snapshot_overrides_startup_options_without_mutating_table() {
+ let table = table("startup", "append", "none", 1).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![2], vec![20])).await;
+ let selected = table.copy_with_options(HashMap::from([
+ ("scan.snapshot-id".into(), "1".into()),
+ ("scan.mode".into(), "from-snapshot".into()),
+ ]));
+ let pinned = snapshot_plan(
+ selected.new_read_builder().new_scan(),
+ Some(2),
+ SnapshotScanMode::All,
+ )
+ .await
+ .unwrap();
+ assert_eq!(rows(&selected, &pinned).await, vec![(1, 10), (2, 20)]);
+ let batch = selected.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(batch.snapshot_id(), Some(1));
+ assert_eq!(rows(&selected, &batch).await, vec![(1, 10)]);
+}
+
+#[tokio::test]
+async fn changelog_reads_overwrite_while_batch_range_still_skips_it() {
+ let table = table("overwrite", "deduplicate", "input", 1).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![2], vec![20]))
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .overwrite(writer.prepare_commit().await.unwrap(), None)
+ .await
+ .unwrap();
+ writer.close().await;
+ // The local overwrite writer does not publish a changelog. Reuse a real
+ // physical changelog manifest to model a producer which supplies one;
+ // its rows deliberately differ from the overwrite's delta data.
+ let first = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ let overwrite = table.snapshot_manager().get_snapshot(2).await.unwrap();
+ let mut json = serde_json::to_value(overwrite).unwrap();
+ json["changelogManifestList"] =
first.changelog_manifest_list().unwrap().into();
+ table
+ .file_io()
+ .new_output(&table.snapshot_manager().snapshot_path(2))
+ .unwrap()
+ .write(serde_json::to_vec(&json).unwrap().into())
+ .await
+ .unwrap();
+ write_batch(&table, &make_batch(vec![3], vec![30])).await;
+ let scan = table.new_read_builder().new_scan();
+ let events = snapshot_plan(scan.clone(), Some(2),
SnapshotScanMode::Changelog)
+ .await
+ .unwrap();
+ assert_eq!(events.snapshot_id(), Some(2));
+ assert_eq!(rows(&table, &events).await, vec![(1, 10)]);
+ assert!(events.splits().iter().all(|split| split.is_streaming()));
+ assert!(events
+ .splits()
+ .iter()
+ .flat_map(|split| split.data_files())
+ .all(|file| file.file_name.starts_with("changelog-")));
+ let range = table
+ .new_read_builder()
+ .new_incremental_scan(IncrementalScanMode::Changelog, 1, 2)
+ .plan_combined()
+ .await
+ .unwrap();
+ assert!(range.splits().is_empty());
+ assert_eq!(range.snapshot_id(), Some(2));
+ let full = snapshot_plan(scan, Some(2), SnapshotScanMode::All)
+ .await
+ .unwrap();
+ assert_eq!(rows(&table, &full).await, vec![(2, 20)]);
+}
+
+#[tokio::test]
+async fn changelog_keeps_all_row_kinds_and_arrival_order() {
+ let table = table("events", "deduplicate", "input", 1).await;
+ write_batch(
+ &table,
+ &make_batch_with_kinds(vec![1, 1, 1, 1], vec![10, 10, 11, 11], vec![0,
1, 2, 3]),
+ )
+ .await;
+ let plan = snapshot_plan(
+ table.new_read_builder().new_scan(),
+ Some(1),
+ SnapshotScanMode::Changelog,
+ )
+ .await
+ .unwrap();
+ let builder = table.new_read_builder();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let kinds: Vec<&str> = batches
+ .iter()
+ .flat_map(|batch| {
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::StringArray>()
+ .unwrap()
+ .iter()
+ .map(Option::unwrap)
+ })
+ .collect();
+ assert_eq!(kinds, vec!["+I", "-U", "+U", "-D"]);
+}
+
+#[tokio::test]
+async fn bucket_filter_precedes_packing_and_limit_for_every_mode() {
+ let table = table("buckets", "deduplicate", "input", 4).await;
+ write_batch(&table, &make_batch((0..32).collect(),
(100..132).collect())).await;
+ for mode in [
+ SnapshotScanMode::All,
+ SnapshotScanMode::Delta,
+ SnapshotScanMode::Changelog,
+ ] {
+ let all = snapshot_plan(table.new_read_builder().new_scan(), None,
mode)
+ .await
+ .unwrap();
+ let selected_bucket = all.splits().last().unwrap().bucket();
+ let expected_ids: std::collections::HashSet<i32> = rows(
+ &table,
+ &Plan::new(
+ all.splits()
+ .iter()
+ .filter(|split| split.bucket() == selected_bucket)
+ .cloned()
+ .collect(),
+ ),
+ )
+ .await
+ .into_iter()
+ .map(|(id, _)| id)
+ .collect();
+ assert!(!expected_ids.is_empty());
+ let mut builder = table.new_read_builder();
+ builder.with_limit(1);
+ let selected = snapshot_plan(
+ builder
+ .new_scan()
+ .with_snapshot_bucket_filter(move |bucket| Ok(bucket ==
selected_bucket)),
+ None,
+ mode,
+ )
+ .await
+ .unwrap();
+ assert_eq!(selected.snapshot_id(), Some(1));
+ assert!(!selected.splits().is_empty());
+ assert!(selected
+ .splits()
+ .iter()
+ .all(|split| split.bucket() == selected_bucket));
+ assert!(rows(&table, &selected)
+ .await
+ .iter()
+ .all(|(id, _)| expected_ids.contains(id)));
+ let empty = snapshot_plan(
+ table
+ .new_read_builder()
+ .new_scan()
+ .with_snapshot_bucket_filter(|_| Ok(false)),
+ None,
+ mode,
+ )
+ .await
+ .unwrap();
+ assert_eq!(empty.snapshot_id(), Some(1));
+ assert!(empty.splits().is_empty());
+ }
+}
+
+#[tokio::test]
+async fn filter_errors_fail_instead_of_publishing_a_partial_plan() {
+ let table = table("callback", "deduplicate", "input", 4).await;
+ write_batch(&table, &make_batch((0..32).collect(),
(100..132).collect())).await;
+ for mode in [
+ SnapshotScanMode::All,
+ SnapshotScanMode::Delta,
+ SnapshotScanMode::Changelog,
+ ] {
+ let calls =
std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ let counter = calls.clone();
+ let error = snapshot_plan(
+ table
+ .new_read_builder()
+ .new_scan()
+ .with_snapshot_bucket_filter(move |_| {
+ if counter.fetch_add(1,
std::sync::atomic::Ordering::SeqCst) == 0 {
+ return Ok(true);
+ }
+ Err(paimon::Error::DataInvalid {
+ message: "bucket selection failed".into(),
+ source: None,
+ })
+ }),
+ None,
+ mode,
+ )
+ .await
+ .unwrap_err();
+ assert!(error.to_string().contains("bucket selection failed"));
+ assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
+ }
+}
+
+#[tokio::test]
+async fn replacing_bucket_filter_and_composing_shard_preserves_versions() {
+ let table = table("shard", "deduplicate", "input", 4).await;
+ write_batch(&table, &make_batch((0..32).collect(),
(100..132).collect())).await;
+ write_batch(&table, &make_batch((0..32).collect(),
(200..232).collect())).await;
+ let mut union = Vec::new();
+ for index in 0..2 {
+ let plan = snapshot_plan(
+ table
+ .new_read_builder()
+ .new_scan()
+ .with_snapshot_bucket_filter(|_| Ok(false))
+ .with_snapshot_bucket_filter(|bucket| Ok(bucket >= 0))
+ .with_shard(index, 2)
+ .unwrap(),
+ None,
+ SnapshotScanMode::All,
+ )
+ .await
+ .unwrap();
+ assert!(plan
+ .splits()
+ .iter()
+ .all(|split| split.bucket() as usize % 2 == index));
+ union.extend(rows(&table, &plan).await);
+ }
+ union.sort_unstable();
+ assert_eq!(union, (0..32).map(|id| (id, 200 + id)).collect::<Vec<_>>());
+}
+
+#[tokio::test]
+async fn read_configuration_and_empty_snapshot_metadata_survive_modes() {
+ for engine in ["append", "evolution", "deduplicate"] {
+ let table = table(&format!("projection_{engine}"), engine, "none",
1).await;
+ write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await;
+ let mut builder = table.new_read_builder();
+ builder.with_projection(&["value"]).unwrap();
+ builder.with_filter(
+ paimon::spec::PredicateBuilder::new(table.schema().fields())
+ .greater_than("id", paimon::spec::Datum::Int(1))
+ .unwrap(),
+ );
+ for mode in [SnapshotScanMode::All, SnapshotScanMode::Delta] {
+ let plan = snapshot_plan(builder.new_scan(), None,
mode).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>(), 1);
+ assert_eq!(batches[0].schema().field(0).name(), "value");
+ assert_eq!(
+ batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 20
+ );
+ }
+ let empty = snapshot_plan(builder.new_scan(), None,
SnapshotScanMode::Changelog)
+ .await
+ .unwrap();
+ assert!(empty.splits().is_empty());
+ assert_eq!(empty.snapshot_id(), Some(1));
+ }
+}
+
+#[tokio::test]
+async fn absent_and_missing_snapshots_are_distinct_even_with_zero_limit() {
+ let table = table("missing", "append", "none", 1).await;
+ let mut builder = table.new_read_builder();
+ builder.with_limit(0);
+ for mode in [
+ SnapshotScanMode::All,
+ SnapshotScanMode::Delta,
+ SnapshotScanMode::Changelog,
+ ] {
+ let scan = builder.new_scan();
+ let empty = snapshot_plan(scan.clone(), None, mode).await.unwrap();
+ assert!(empty.splits().is_empty());
+ assert_eq!(empty.snapshot_id(), None);
+ assert!(matches!(
+ snapshot_plan(scan, Some(1), mode).await,
+ Err(paimon::Error::SnapshotNotExist { snapshot_id: 1 })
+ ));
+ }
+ assert!(builder.new_scan().with_shard(0, 0).is_err());
+ assert!(builder.new_scan().with_shard(2, 2).is_err());
+}
+
+#[tokio::test]
+async fn all_dv_level_zero_versions_merge_before_value_predicates() {
+ for mor in ["true", "false"] {
+ let table = table(&format!("dv_{mor}"), "deduplicate", "none", 1)
+ .await
+ .copy_with_options(HashMap::from([
+ ("deletion-vectors.enabled".into(), "true".into()),
+ ("deletion-vectors.merge-on-read".into(), mor.into()),
+ ]));
+ write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await;
+ write_batch(&table, &make_batch(vec![1, 3], vec![11, 30])).await;
+ let plan = snapshot_plan(
+ table.new_read_builder().new_scan(),
+ None,
+ SnapshotScanMode::All,
+ )
+ .await
+ .unwrap();
+ assert_eq!(rows(&table, &plan).await, vec![(1, 11), (2, 20), (3, 30)]);
+ let mut builder = table.new_read_builder();
+ builder.with_filter(
+ paimon::spec::PredicateBuilder::new(table.schema().fields())
+ .equal("value", paimon::spec::Datum::Int(10))
+ .unwrap(),
+ );
+ let plan = snapshot_plan(builder.new_scan(), None,
SnapshotScanMode::All)
+ .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 snapshot_planning_never_bypasses_query_authorization() {
+ let table = table("auth", "append", "none", 1)
+ .await
+ .copy_with_options(HashMap::from([(
+ "query-auth.enabled".into(),
+ "true".into(),
+ )]));
+ for mode in [
+ SnapshotScanMode::All,
+ SnapshotScanMode::Delta,
+ SnapshotScanMode::Changelog,
+ ] {
+ for snapshot_id in [None, Some(1)] {
+ let scan = table
+ .new_read_builder()
+ .new_scan()
+ .with_snapshot_bucket_filter(|_| Ok(false));
+ let error = snapshot_plan(scan, snapshot_id,
mode).await.unwrap_err();
+ assert!(error.to_string().contains("query-auth.enabled"));
+ assert!(matches!(error, paimon::Error::Unsupported { .. }));
+ }
+ }
+}
+
+#[tokio::test]
+async fn postpone_pending_visibility_is_an_explicit_reader_selection() {
+ for producer in ["none", "input"] {
+ let table = table(&format!("pending_{producer}"), "deduplicate",
producer, -2).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ for mode in [
+ SnapshotScanMode::All,
+ SnapshotScanMode::Delta,
+ SnapshotScanMode::Changelog,
+ ] {
+ let plan = snapshot_plan(table.new_read_builder().new_scan(),
None, mode)
+ .await
+ .unwrap();
+ let expected = if mode == SnapshotScanMode::Changelog {
+ vec![]
+ } else {
+ vec![(1, 10)]
+ };
+ assert_eq!(rows(&table, &plan).await, expected, "{producer}
{mode:?}");
+ assert!(plan.splits().iter().all(|split| split.bucket() == -2));
+ let plan = snapshot_plan(
+ table
+ .new_read_builder()
+ .new_scan()
+ .with_snapshot_bucket_filter(|_| panic!("unassigned bucket
reached callback"))
+ .only_read_snapshot_real_buckets(),
+ None,
+ mode,
+ )
+ .await
+ .unwrap();
+ assert!(plan.splits().is_empty());
+ assert_eq!(plan.snapshot_id(), Some(1));
+ }
+ }
+}
diff --git a/crates/paimon/src/table/stream_table_scan.rs
b/crates/paimon/src/table/stream_table_scan.rs
new file mode 100644
index 00000000..fea63186
--- /dev/null
+++ b/crates/paimon/src/table/stream_table_scan.rs
@@ -0,0 +1,252 @@
+// 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.
+
+//! Stateful snapshot selection and progress for continuous table reads.
+
+use super::table_scan::SnapshotScanMode;
+use super::{Plan, TableScan};
+use crate::spec::{ChangelogProducer, CommitKind, CoreOptions, Snapshot};
+use crate::{Error, Result};
+
+/// Continuous table scan. `checkpoint` is the next snapshot to consume.
+///
+/// The first plan reads the latest full state. Follow-up plans read APPEND
+/// deltas when no changelog producer is configured, or physical changelogs
+/// otherwise. Empty follow-up plans and irrelevant commits are skipped, as in
+/// Java. `None` means there is no available next snapshot; clients can poll.
+/// Progress is persisted only by an explicit `notify_checkpoint_complete`.
+#[derive(Debug)]
+pub struct StreamTableScan {
+ scan: TableScan<'static>,
+ next_snapshot_id: Option<i64>,
+ watermark: Option<i64>,
+ consumer_id: Option<String>,
+ follow_up_mode: SnapshotScanMode,
+ consumer_restored: bool,
+ missing_snapshot_polls: usize,
+}
+
+impl StreamTableScan {
+ pub(super) fn new(scan: TableScan<'_>) -> Result<Self> {
+ let mut scan = scan.into_owned_paimon_scan()?;
+ let options =
CoreOptions::new(scan.snapshot_table().schema().options());
+ let producer = options.try_changelog_producer()?;
+ let follow_up_mode = if producer == ChangelogProducer::None {
+ SnapshotScanMode::Delta
+ } else {
+ SnapshotScanMode::Changelog
+ };
+ if options.bucket() == -2 && producer != ChangelogProducer::None {
+ scan = scan.only_read_snapshot_real_buckets();
+ }
+ let consumer_id = scan
+ .snapshot_table()
+ .schema()
+ .options()
+ .get("consumer-id")
+ .cloned();
+ if let Some(id) = &consumer_id {
+ super::ConsumerManager::validate_consumer_id(id)?;
+ }
+ Ok(Self {
+ scan,
+ next_snapshot_id: None,
+ watermark: None,
+ consumer_id,
+ follow_up_mode,
+ consumer_restored: false,
+ missing_snapshot_polls: 0,
+ })
+ }
+
+ /// Set the consumer whose acknowledged checkpoint will be restored.
+ pub fn with_consumer_id(&mut self, consumer_id: impl Into<String>) ->
Result<&mut Self> {
+ let consumer_id = consumer_id.into();
+ super::ConsumerManager::validate_consumer_id(&consumer_id)?;
+ self.consumer_id = Some(consumer_id);
+ Ok(self)
+ }
+
+ /// Select buckets before split packing. Replaces the previous filter.
+ pub fn with_bucket_filter(
+ &mut self,
+ filter: impl Fn(i32) -> Result<bool> + Send + Sync + 'static,
+ ) -> &mut Self {
+ self.scan = self.scan.clone().with_snapshot_bucket_filter(filter);
+ self
+ }
+
+ /// Distribute initial primary-key state by bucket so all key versions
+ /// merge together. Follow-up physical events use Java's shard rules.
+ pub fn with_shard(&mut self, index: usize, count: usize) -> Result<&mut
Self> {
+ self.scan = self.scan.clone().with_shard(index, count)?;
+ Ok(self)
+ }
+
+ pub fn checkpoint(&self) -> Option<i64> {
+ self.next_snapshot_id
+ }
+
+ pub fn watermark(&self) -> Option<i64> {
+ self.watermark
+ }
+
+ /// Restore the next snapshot to consume. `None` resets to the configured
starting position, including consumer progress.
+ pub fn restore(&mut self, next_snapshot_id: Option<i64>) -> Result<()> {
+ if next_snapshot_id.is_some_and(|id| id < 1) {
+ return Err(Error::DataInvalid {
+ message: "next snapshot id must be positive".into(),
+ source: None,
+ });
+ }
+ self.next_snapshot_id = next_snapshot_id;
+ self.consumer_restored = next_snapshot_id.is_some();
+ self.missing_snapshot_polls = 0;
+ Ok(())
+ }
+
+ 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()?;
+ if self.next_snapshot_id.is_none() && !self.consumer_restored {
+ if let Some(consumer_id) = &self.consumer_id {
+ let next = self
+ .scan
+ .snapshot_table()
+ .consumer_manager()
+ .get(consumer_id)
+ .await?;
+ if let Some(next) = next {
+ self.restore(Some(next))?;
+ }
+ }
+ self.consumer_restored = true;
+ }
+ let manager = self.scan.snapshot_table().snapshot_manager();
+ if self.next_snapshot_id.is_none() {
+ let Some(snapshot) = manager.get_latest_snapshot().await? else {
+ return Ok(None);
+ };
+ let next = snapshot.id() + 1;
+ let watermark = snapshot.watermark();
+ let plan = self.plan_initial_snapshot(snapshot).await?;
+ self.next_snapshot_id = Some(next);
+ self.watermark = watermark;
+ return Ok(Some(plan));
+ }
+ let mode = self.follow_up_mode;
+ loop {
+ let id = self.next_snapshot_id.expect("initial phase completed");
+ let snapshot = match manager.get_snapshot(id).await {
+ Ok(snapshot) => snapshot,
+ Err(Error::SnapshotNotExist { .. }) => {
+ // Like NextSnapshotFetcher, wait for publication but
detect
+ // expired checkpoints / recreated tables periodically.
+ self.missing_snapshot_polls += 1;
+ if self.missing_snapshot_polls.is_multiple_of(16) {
+ let earliest = manager.earliest_snapshot_id().await?;
+ let latest = manager.get_latest_snapshot_id().await?;
+ if earliest.is_some_and(|first| first > id)
+ || latest.is_some_and(|last| id > last + 1)
+ || (latest.is_none() && id > 1)
+ {
+ return Err(Error::DataInvalid {
+ message: format!("stream checkpoint {id} is
outside the available snapshots {earliest:?}..{latest:?}"),
+ source: None,
+ });
+ }
+ }
+ return Ok(None);
+ }
+ Err(error) => return Err(error),
+ };
+ self.missing_snapshot_polls = 0;
+ let should_scan = match mode {
+ SnapshotScanMode::Delta => snapshot.commit_kind() ==
&CommitKind::APPEND,
+ SnapshotScanMode::Changelog =>
snapshot.changelog_manifest_list().is_some(),
+ SnapshotScanMode::All => unreachable!("follow-up mode"),
+ };
+ if !should_scan {
+ self.next_snapshot_id = Some(id + 1);
+ continue;
+ }
+ 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?;
+ self.next_snapshot_id = Some(id + 1);
+ self.watermark = watermark;
+ if !plan.splits().is_empty() {
+ return Ok(Some(plan));
+ }
+ }
+ }
+
+ async fn plan_initial_snapshot(&self, snapshot: Snapshot) -> Result<Plan> {
+ let options = self.scan.snapshot_table().schema().options();
+ let mut scan = self.scan.clone();
+ let parse = |key, default| -> Result<i32> {
+ options
+ .get(key)
+ .map(|value| {
+ value.parse().map_err(|error| crate::Error::DataInvalid {
+ message: format!("invalid {key}: {value}"),
+ source: Some(Box::new(error)),
+ })
+ })
+ .unwrap_or(Ok(default))
+ };
+ match self
+ .scan
+ .snapshot_table()
+ .schema()
+ .core_options()
+ .try_changelog_producer()?
+ {
+ ChangelogProducer::Lookup => scan =
scan.with_snapshot_levels(1..=i32::MAX),
+ ChangelogProducer::FullCompaction => {
+ // Java numLevels defaults to
incrementSafely(compactionTrigger).
+ let levels = if options.contains_key("num-levels") {
+ parse("num-levels", 0)?
+ } else {
+ parse("num-sorted-run.compaction-trigger",
5)?.saturating_add(1)
+ };
+ let last = levels
+ .checked_sub(1)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "num-levels is out of range".into(),
+ source: None,
+ })?;
+ scan = scan.with_snapshot_levels(last..=last);
+ }
+ ChangelogProducer::None | ChangelogProducer::Input => {}
+ }
+ scan.plan_selected_snapshot(snapshot, SnapshotScanMode::All)
+ .await
+ }
+
+ /// Acknowledge a checkpoint after the caller has processed its plan.
+ pub async fn notify_checkpoint_complete(&self, next_snapshot: Option<i64>)
-> Result<()> {
+ if let (Some(consumer_id), Some(next)) = (&self.consumer_id,
next_snapshot) {
+ self.scan
+ .snapshot_table()
+ .consumer_manager()
+ .reset(consumer_id, next)
+ .await?;
+ }
+ Ok(())
+ }
+}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index e728c730..53f8bb63 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1097,7 +1097,121 @@ enum TableScanKind<'a> {
Format(FormatTableScan<'a>),
}
+/// File set to plan from one selected snapshot. Streaming progress is managed
+/// separately by StreamTableScan.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum SnapshotScanMode {
+ All,
+ Delta,
+ Changelog,
+}
+
+#[derive(Clone)]
+struct BucketFilter(std::sync::Arc<dyn Fn(i32) -> crate::Result<bool> + Send +
Sync>);
+
+impl std::fmt::Debug for BucketFilter {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result
{
+ formatter.write_str("BucketFilter")
+ }
+}
+
+impl BucketFilter {
+ fn test(&self, bucket: i32) -> crate::Result<bool> {
+ (self.0)(bucket)
+ }
+}
+
impl<'a> TableScan<'a> {
+ pub(super) fn into_owned_paimon_scan(self) ->
crate::Result<TableScan<'static>> {
+ let TableScanKind::Paimon(scan) = self.0 else {
+ return Err(crate::Error::Unsupported {
+ message: "StreamTableScan requires a Paimon table".into(),
+ });
+ };
+ Ok(TableScan(TableScanKind::Paimon(PaimonTableScan {
+ table:
ScanTable::Owned(std::sync::Arc::new((*scan.table).clone())),
+ partition_filter: scan.partition_filter,
+ data_predicates: scan.data_predicates,
+ bucket_predicate: scan.bucket_predicate,
+ limit: scan.limit,
+ row_ranges: scan.row_ranges,
+ split_selection: scan.split_selection,
+ row_range_optimization_disabled:
scan.row_range_optimization_disabled,
+ scan_all_files: scan.scan_all_files,
+ incremental_split_mode: scan.incremental_split_mode,
+ projected_read_field_ids: scan.projected_read_field_ids,
+ })))
+ }
+
+ pub(super) fn snapshot_table(&self) -> &Table {
+ let TableScanKind::Paimon(scan) = &self.0 else {
+ unreachable!("snapshot scans only contain Paimon tables")
+ };
+ &scan.table
+ }
+
+ pub(super) fn validate_snapshot_read(&self) -> crate::Result<()> {
+ let TableScanKind::Paimon(scan) = &self.0 else {
+ unreachable!("snapshot scans only contain Paimon tables")
+ };
+ scan.validate_read_options()
+ }
+
+ pub(super) fn with_snapshot_levels(mut self, levels:
std::ops::RangeInclusive<i32>) -> Self {
+ if let TableScanKind::Paimon(scan) = &mut self.0 {
+ scan.split_selection
+ .get_or_insert_with(Default::default)
+ .snapshot_levels = Some(levels);
+ }
+ self
+ }
+
+ pub(super) fn only_read_snapshot_real_buckets(mut self) -> Self {
+ if let TableScanKind::Paimon(scan) = &mut self.0 {
+ scan.split_selection
+ .get_or_insert_with(Default::default)
+ .only_real_buckets = true;
+ }
+ self
+ }
+
+ pub(super) fn with_snapshot_bucket_filter(
+ mut self,
+ filter: impl Fn(i32) -> crate::Result<bool> + Send + Sync + 'static,
+ ) -> Self {
+ if let TableScanKind::Paimon(scan) = &mut self.0 {
+ let selection =
scan.split_selection.get_or_insert_with(Default::default);
+ selection.bucket_filter =
Some(BucketFilter(std::sync::Arc::new(filter)));
+ }
+ self
+ }
+
+ pub(super) async fn plan_selected_snapshot(
+ &self,
+ snapshot: crate::spec::Snapshot,
+ mode: SnapshotScanMode,
+ ) -> crate::Result<Plan> {
+ self.validate_snapshot_read()?;
+ let TableScanKind::Paimon(scan) = &self.0 else {
+ unreachable!("snapshot scans only contain Paimon tables")
+ };
+ // Snapshot planning preserves the selected file set. Scan policies
+ // explicitly decide whether unassigned buckets should be hidden.
+ let mut scan = scan.clone();
+ scan.scan_all_files = true;
+ scan.split_selection
+ .get_or_insert_with(Default::default)
+ .snapshot_mode = Some(mode);
+ 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,
+ }
+ }
+
pub(crate) fn new(
table: &'a Table,
partition_filter: Option<PartitionFilter>,
@@ -1191,6 +1305,10 @@ impl<'a> TableScan<'a> {
scan.split_selection = Some(Box::new(ScanSplitSelection {
mode: Some(ScanSplitMode::ChunkShuffle(config)),
shard: scan.shard(),
+ bucket_filter: scan.bucket_filter().cloned(),
+ only_real_buckets: scan.only_real_buckets(),
+ snapshot_mode: scan.snapshot_mode(),
+ snapshot_levels: scan.snapshot_levels().cloned(),
}));
Ok(Self(TableScanKind::Paimon(scan)))
}
@@ -1230,6 +1348,10 @@ impl<'a> TableScan<'a> {
scan.split_selection =
Some(Box::new(ScanSplitSelection {
mode: None,
shard: Some((index, count)),
+ bucket_filter: None,
+ only_real_buckets: false,
+ snapshot_mode: None,
+ snapshot_levels: None,
}));
}
}
@@ -1288,6 +1410,10 @@ impl<'a> TableScan<'a> {
scan.split_selection = Some(Box::new(ScanSplitSelection {
mode: Some(ScanSplitMode::RowPosition(selection)),
shard: None,
+ bucket_filter: scan.bucket_filter().cloned(),
+ only_real_buckets: scan.only_real_buckets(),
+ snapshot_mode: scan.snapshot_mode(),
+ snapshot_levels: scan.snapshot_levels().cloned(),
}));
Ok(Self(TableScanKind::Paimon(scan)))
}
@@ -1409,15 +1535,38 @@ enum ScanSplitMode {
ChunkShuffle(ChunkShuffle),
}
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
struct ScanSplitSelection {
mode: Option<ScanSplitMode>,
shard: Option<(usize, usize)>,
+ bucket_filter: Option<BucketFilter>,
+ only_real_buckets: bool,
+ snapshot_mode: Option<SnapshotScanMode>,
+ snapshot_levels: Option<std::ops::RangeInclusive<i32>>,
+}
+
+// A stream scan owns its table so clients can retain the scan across polls.
+// Batch scans keep borrowing their table and retain their existing lifetime.
+#[derive(Debug, Clone)]
+enum ScanTable<'a> {
+ Borrowed(&'a Table),
+ Owned(std::sync::Arc<Table>),
+}
+
+impl std::ops::Deref for ScanTable<'_> {
+ type Target = Table;
+
+ fn deref(&self) -> &Table {
+ match self {
+ Self::Borrowed(table) => table,
+ Self::Owned(table) => table,
+ }
+ }
}
#[derive(Debug, Clone)]
struct PaimonTableScan<'a> {
- table: &'a Table,
+ table: ScanTable<'a>,
partition_filter: Option<PartitionFilter>,
data_predicates: Vec<Predicate>,
bucket_predicate: Option<Predicate>,
@@ -1441,6 +1590,30 @@ struct PaimonTableScan<'a> {
}
impl<'a> PaimonTableScan<'a> {
+ fn snapshot_levels(&self) -> Option<&std::ops::RangeInclusive<i32>> {
+ self.split_selection
+ .as_deref()
+ .and_then(|selection| selection.snapshot_levels.as_ref())
+ }
+
+ fn snapshot_mode(&self) -> Option<SnapshotScanMode> {
+ self.split_selection
+ .as_deref()
+ .and_then(|selection| selection.snapshot_mode)
+ }
+
+ fn only_real_buckets(&self) -> bool {
+ self.split_selection
+ .as_deref()
+ .is_some_and(|selection| selection.only_real_buckets)
+ }
+
+ fn bucket_filter(&self) -> Option<&BucketFilter> {
+ self.split_selection
+ .as_deref()
+ .and_then(|selection| selection.bucket_filter.as_ref())
+ }
+
fn row_position_selection(&self) -> Option<RowPositionSelection> {
match self
.split_selection
@@ -1482,7 +1655,7 @@ impl<'a> PaimonTableScan<'a> {
row_ranges: Option<Vec<RowRange>>,
) -> Self {
Self {
- table,
+ table: ScanTable::Borrowed(table),
partition_filter,
data_predicates,
bucket_predicate,
@@ -1567,7 +1740,7 @@ impl<'a> PaimonTableScan<'a> {
return Ok(plan.planned(grant));
}
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
- let plan = match
super::time_travel::resolve_snapshot(scan.table).await? {
+ let plan = match
super::time_travel::resolve_snapshot(&scan.table).await? {
Some(snapshot) => {
scan.plan_snapshot(snapshot,
data_evolution_read_field_ids.as_ref(), None)
.await?
@@ -1605,7 +1778,7 @@ impl<'a> PaimonTableScan<'a> {
return Ok((plan.planned(grant), trace));
}
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
- let plan = match
super::time_travel::resolve_snapshot(scan.table).await? {
+ let plan = match
super::time_travel::resolve_snapshot(&scan.table).await? {
Some(snapshot) => {
trace.snapshot_id = Some(snapshot.id());
let plan = scan
@@ -1697,7 +1870,7 @@ impl<'a> PaimonTableScan<'a> {
return Ok(Plan::new(Vec::new()));
};
super::IncrementalScan::new(
- self.table,
+ &self.table,
TableScan(TableScanKind::Paimon(self.clone())),
super::IncrementalScanMode::Auto,
start,
@@ -1788,16 +1961,21 @@ impl<'a> PaimonTableScan<'a> {
/// cannot be selected from packed splits: one split can contain files
owned
/// by several workers. Keep chunk-shuffle input complete until shuffling.
fn retain_shard_entries(&self, entries: &mut Vec<ManifestEntry>) ->
crate::Result<()> {
+ self.retain_bucket_entries(entries)?;
let Some((index, count)) = self.shard().filter(|_|
self.chunk_shuffle().is_none()) else {
return Ok(());
};
let options = self.table.schema().core_options();
let has_primary_keys = !self.table.schema().primary_keys().is_empty();
- // Match SplitGenerator.alwaysRawConvertible, independent of whether a
- // particular split is raw-convertible or the scan includes level zero.
+ let initial_state = self.snapshot_mode() ==
Some(SnapshotScanMode::All);
+ // Initial stream state merges overlapping key versions, including
+ // L0 files. Keep each bucket intact until that merge has completed.
+ // Follow-up physical events and ordinary raw scans retain Java's
+ // SplitGenerator.alwaysRawConvertible file-name distribution.
let by_file_name = if has_primary_keys {
- options.deletion_vectors_enabled()
- || options.merge_engine()? ==
crate::spec::MergeEngine::FirstRow
+ !initial_state
+ && (options.deletion_vectors_enabled()
+ || options.merge_engine()? ==
crate::spec::MergeEngine::FirstRow)
} else {
!options.data_evolution_enabled()
};
@@ -1814,12 +1992,43 @@ impl<'a> PaimonTableScan<'a> {
// also handles Integer.MIN_VALUE without overflow.
((i128::from(hash) % count as i128).unsigned_abs() as usize)
== index
} else {
- i128::from(entry.bucket()) % count as i128 == index as i128
+ let bucket = i128::from(entry.bucket());
+ // Initial state includes postpone buckets. Assign them to
+ // one worker as well, rather than dropping a negative
remainder.
+ let bucket = if initial_state { bucket.abs() } else { bucket };
+ bucket % count as i128 == index as i128
}
});
Ok(())
}
+ fn retain_bucket_entries(&self, entries: &mut Vec<ManifestEntry>) ->
crate::Result<()> {
+ let filter = self.bucket_filter();
+ let only_real = self.only_real_buckets();
+ let levels = self.snapshot_levels();
+ if filter.is_none() && !only_real && levels.is_none() {
+ return Ok(());
+ }
+ let mut selected = Vec::with_capacity(entries.len());
+ for entry in std::mem::take(entries) {
+ if levels.is_some_and(|levels|
!levels.contains(&entry.file().level)) {
+ continue;
+ }
+ if only_real && entry.bucket() < 0 {
+ continue;
+ }
+ if filter
+ .map(|filter| filter.test(entry.bucket()))
+ .transpose()?
+ .unwrap_or(true)
+ {
+ selected.push(entry);
+ }
+ }
+ *entries = selected;
+ Ok(())
+ }
+
fn projected_read_field_ids(&self) -> crate::Result<Option<HashSet<i32>>> {
Ok(self.projected_read_field_ids.clone())
}
@@ -2116,7 +2325,8 @@ impl<'a> PaimonTableScan<'a> {
fn requires_key_merge(&self) -> bool {
let options = self.table.schema().core_options();
- self.is_streaming()
+ self.snapshot_mode() == Some(SnapshotScanMode::All)
+ || self.is_streaming()
|| match options.merge_engine() {
Ok(crate::spec::MergeEngine::FirstRow) =>
!self.skip_level_zero(),
_ => {
@@ -2477,7 +2687,7 @@ impl<'a> PaimonTableScan<'a> {
for entry in entries {
if entry.file().schema_id == current_schema_id
|| data_file_matches_predicates_for_table(
- self.table,
+ &self.table,
entry.file(),
&stats_pruning_predicates,
&key_stats_predicates,
@@ -2614,7 +2824,7 @@ impl<'a> PaimonTableScan<'a> {
let mut groups = Vec::with_capacity(before);
for group in row_id_groups {
if data_evolution_group_matches_predicates_for_table(
- self.table,
+ &self.table,
&group,
&self.data_predicates,
&mut stats_schema_cache,
@@ -2823,7 +3033,7 @@ impl<'a> PaimonTableScan<'a> {
splits
};
let splits = if let Some(config) = self.chunk_shuffle() {
- chunk_shuffle_splits(self.table, splits, config,
self.shard()).await?
+ chunk_shuffle_splits(&self.table, splits, config,
self.shard()).await?
} else {
splits
};
@@ -6903,3 +7113,7 @@ mod tests {
assert_eq!(empty.snapshot_id(), Some(1));
}
}
+
+#[cfg(test)]
+#[path = "snapshot_scan_tests.rs"]
+mod snapshot_tests;
diff --git a/crates/paimon/tests/stream_table_scan_test.rs
b/crates/paimon/tests/stream_table_scan_test.rs
new file mode 100644
index 00000000..ab6a7575
--- /dev/null
+++ b/crates/paimon/tests/stream_table_scan_test.rs
@@ -0,0 +1,466 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use common::incremental_helpers::{
+ make_batch, memory_table, persist_table_schema, setup_dirs, write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, IntType, Schema, TableSchema};
+use paimon::table::{Plan, Table};
+use std::collections::HashMap;
+use std::sync::{
+ atomic::{AtomicBool, Ordering},
+ Arc,
+};
+
+async fn table(name: &str, producer: &str) -> Table {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("changelog-producer", producer)
+ .build()
+ .unwrap();
+ let (file_io, table) = memory_table(name, TableSchema::new(0, &schema));
+ setup_dirs(&file_io, name).await;
+ persist_table_schema(&file_io, name, table.schema()).await;
+ table
+}
+
+async fn values(table: &Table, plan: Plan) -> Vec<(i32, i32)> {
+ use arrow_array::Int32Array;
+ let batches: Vec<_> = table
+ .new_read_builder()
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut values = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let vals = batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ values.extend((0..batch.num_rows()).map(|row| (ids.value(row),
vals.value(row))));
+ }
+ values.sort_unstable();
+ values
+}
+
+async fn patch_snapshot(table: &Table, id: i64, key: &str, value:
serde_json::Value) {
+ let path = table.snapshot_manager().snapshot_path(id);
+ let bytes = table
+ .file_io()
+ .new_input(&path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let mut snapshot: serde_json::Value =
serde_json::from_slice(&bytes).unwrap();
+ snapshot[key] = value;
+ table
+ .file_io()
+ .new_output(&path)
+ .unwrap()
+ .write(serde_json::to_vec(&snapshot).unwrap().into())
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+async fn empty_poll_then_full_initial_and_follow_up() {
+ let table = table("memory:/stream-phases", "none").await;
+ let mut scan = table.new_stream_scan().unwrap();
+ assert!(scan.plan().await.unwrap().is_none());
+ assert_eq!(scan.checkpoint(), None);
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ patch_snapshot(&table, 1, "watermark", 42.into()).await;
+ let full = scan.plan().await.unwrap().unwrap();
+ assert!(!full.splits()[0].is_streaming());
+ assert_eq!(values(&table, full).await, vec![(1, 10)]);
+ assert_eq!(scan.checkpoint(), Some(2));
+ assert_eq!(scan.watermark(), Some(42));
+ assert!(scan.plan().await.unwrap().is_none());
+ write_batch(&table, &make_batch(vec![1], vec![11])).await;
+ let next = scan.plan().await.unwrap().unwrap();
+ assert!(next.splits()[0].is_streaming());
+ assert_eq!(values(&table, next).await, vec![(1, 11)]);
+ assert_eq!(scan.checkpoint(), Some(3));
+ scan.restore(Some(1)).unwrap();
+ assert_eq!(scan.plan().await.unwrap().unwrap().snapshot_id(), Some(1));
+}
+
+#[tokio::test]
+async fn stream_scan_owns_table_and_read_configuration() {
+ let table = table("memory:/stream-owned", "input").await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let mut scan = {
+ let copy = table.clone();
+ let builder = copy.new_read_builder();
+ builder.new_stream_scan().unwrap()
+ };
+ assert_eq!(scan.plan().await.unwrap().unwrap().snapshot_id(), Some(1));
+ scan.restore(Some(1)).unwrap();
+ assert_eq!(
+ values(&table, scan.plan().await.unwrap().unwrap()).await,
+ vec![(1, 10)]
+ );
+}
+
+#[tokio::test]
+async fn
initial_shards_merge_all_key_versions_and_follow_up_shards_keep_events() {
+ for (engine, dv, bucket, producer, expected) in [
+ ("first-row", false, "1", "none", 10),
+ ("first-row", false, "-2", "none", 10),
+ ("deduplicate", true, "1", "none", 13),
+ ("deduplicate", true, "-2", "none", 13),
+ ("deduplicate", true, "1", "input", 13),
+ ] {
+ for count in [2, 3] {
+ let table = table(
+
&format!("memory:/stream-shards-{engine}-{producer}-{bucket}-{count}"),
+ producer,
+ )
+ .await
+ .copy_with_options(HashMap::from([
+ ("merge-engine".into(), engine.into()),
+ ("deletion-vectors.enabled".into(), dv.to_string()),
+ ("bucket".into(), bucket.into()),
+ ]));
+ for version in 0..4 {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![1], vec![10 +
version]))
+ .await
+ .unwrap();
+ let mut messages = writer.prepare_commit().await.unwrap();
+ // Deterministic names distribute consecutive versions to
+ // different file-name shards, independent of writer UUIDs.
+ for message in &mut messages {
+ let bucket_path = format!(
+ "{}/{}",
+ table.location(),
+ paimon::spec::bucket_dir_name(message.bucket)
+ );
+ for (prefix, files) in [
+ ("data", &mut message.new_files),
+ ("changelog", &mut message.new_changelog_files),
+ ] {
+ for (index, file) in files.iter_mut().enumerate() {
+ let old_path = file.data_file_path(&bucket_path);
+ let bytes = table
+ .file_io()
+ .new_input(&old_path)
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ file.file_name =
format!("{prefix}-{version}-{index}.parquet");
+ table
+ .file_io()
+ .new_output(&file.data_file_path(&bucket_path))
+ .unwrap()
+ .write(bytes)
+ .await
+ .unwrap();
+ }
+ }
+ }
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ let full = table
+ .new_stream_scan()
+ .unwrap()
+ .plan()
+ .await
+ .unwrap()
+ .unwrap();
+ let expected_rows = values(&table, full).await;
+ assert_eq!(expected_rows, vec![(1, expected)]);
+ let mut initial_union = Vec::new();
+ let mut event_union = Vec::new();
+ for index in 0..count {
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.with_shard(index, count).unwrap();
+ let initial = scan.plan().await.unwrap().unwrap();
+ assert_eq!(initial.snapshot_id(), Some(4));
+ initial_union.extend(values(&table, initial).await);
+ scan.restore(Some(1)).unwrap();
+ let mut events = Vec::new();
+ while let Some(plan) = scan.plan().await.unwrap() {
+ assert!(plan.splits().iter().all(|split|
split.is_streaming()));
+ events.extend(values(&table, plan).await);
+ }
+ if count == 2 {
+ assert!(!events.is_empty(), "physical events should use
both shards");
+ }
+ assert_eq!(scan.checkpoint(), Some(5));
+ event_union.extend(events);
+ }
+ initial_union.sort_unstable();
+ assert_eq!(
+ initial_union, expected_rows,
+ "{engine}, {producer}, bucket={bucket}, count={count}"
+ );
+ event_union.sort_unstable();
+ assert_eq!(
+ event_union,
+ (10..14).map(|value| (1, value)).collect::<Vec<_>>()
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn delta_follow_up_skips_overwrite_and_compact_commits() {
+ let table = table("memory:/stream-delta-selection", "none").await;
+ for id in 1..=4 {
+ write_batch(&table, &make_batch(vec![id], vec![id * 10])).await;
+ }
+ patch_snapshot(&table, 2, "commitKind", "OVERWRITE".into()).await;
+ patch_snapshot(&table, 3, "commitKind", "COMPACT".into()).await;
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.restore(Some(2)).unwrap();
+ let plan = scan.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(4));
+ assert_eq!(values(&table, plan).await, vec![(4, 40)]);
+ assert_eq!(scan.checkpoint(), Some(5));
+}
+
+#[tokio::test]
+async fn changelog_follow_up_consumes_overwrite_physical_events() {
+ let table = table("memory:/stream-changelog-selection", "input").await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![2], vec![20])).await;
+ let first = table.snapshot_manager().get_snapshot(1).await.unwrap();
+ patch_snapshot(&table, 2, "commitKind", "OVERWRITE".into()).await;
+ patch_snapshot(
+ &table,
+ 2,
+ "changelogManifestList",
+ first.changelog_manifest_list().unwrap().into(),
+ )
+ .await;
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.restore(Some(2)).unwrap();
+ let plan = scan.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert_eq!(values(&table, plan).await, vec![(1, 10)]);
+}
+
+#[tokio::test]
+async fn empty_follow_up_is_skipped_and_initial_empty_plan_is_returned() {
+ let table = table("memory:/stream-empty-selection", "none").await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![2], vec![20])).await;
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.with_bucket_filter(|_| Ok(false));
+ let initial = scan.plan().await.unwrap().unwrap();
+ assert!(initial.splits().is_empty());
+ assert_eq!(initial.snapshot_id(), Some(2));
+ scan.restore(Some(1)).unwrap();
+ assert!(scan.plan().await.unwrap().is_none());
+ assert_eq!(scan.checkpoint(), Some(3));
+}
+
+#[tokio::test]
+async fn callback_failure_keeps_the_selected_snapshot_retryable() {
+ for initial in [false, true] {
+ let table = table(&format!("memory:/stream-retry-{initial}"),
"none").await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let fail = Arc::new(AtomicBool::new(true));
+ let flag = Arc::clone(&fail);
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.with_bucket_filter(move |_| {
+ if flag.swap(false, Ordering::SeqCst) {
+ return Err(paimon::Error::Unsupported {
+ message: "retry this frame".into(),
+ });
+ }
+ Ok(true)
+ });
+ let checkpoint = if initial { None } else { Some(1) };
+ scan.restore(checkpoint).unwrap();
+ assert!(scan.plan().await.is_err());
+ assert_eq!(scan.checkpoint(), checkpoint);
+ assert_eq!(scan.plan().await.unwrap().unwrap().snapshot_id(), Some(1));
+ }
+}
+
+#[tokio::test]
+async fn large_gap_preserves_every_changelog_frame() {
+ let table = table("memory:/stream-long-gap", "input").await;
+ for value in 1..=24 {
+ write_batch(&table, &make_batch(vec![1], vec![value])).await;
+ }
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.restore(Some(1)).unwrap();
+ for id in 1..=24 {
+ let plan = scan.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(id));
+ assert_eq!(values(&table, plan).await, vec![(1, id as i32)]);
+ }
+ assert!(scan.plan().await.unwrap().is_none());
+ assert_eq!(scan.checkpoint(), Some(25));
+}
+
+#[tokio::test]
+async fn consumer_acknowledgement_and_explicit_restore() {
+ let table = table("memory:/stream-consumer", "none").await;
+ for id in 1..=2 {
+ write_batch(&table, &make_batch(vec![id], vec![id * 10])).await;
+ }
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.with_consumer_id("job")
+ .unwrap()
+ .restore(Some(1))
+ .unwrap();
+ scan.plan().await.unwrap().unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), None);
+ scan.notify_checkpoint_complete(scan.checkpoint())
+ .await
+ .unwrap();
+ assert_eq!(table.consumer_manager().get("job").await.unwrap(), Some(2));
+ let mut resumed = table.new_stream_scan().unwrap();
+ resumed.with_consumer_id("job").unwrap();
+ let plan = resumed.plan().await.unwrap().unwrap();
+ assert_eq!(plan.snapshot_id(), Some(2));
+ assert!(plan.splits()[0].is_streaming());
+ resumed.restore(Some(1)).unwrap();
+ assert_eq!(
+ resumed.plan().await.unwrap().unwrap().snapshot_id(),
+ Some(1)
+ );
+}
+
+#[tokio::test]
+async fn missing_checkpoints_wait_without_skipping_a_hole() {
+ let table = table("memory:/stream-missing", "none").await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.restore(Some(100)).unwrap();
+ for _ in 0..15 {
+ assert!(scan.plan().await.unwrap().is_none());
+ }
+ assert!(scan.plan().await.is_err());
+ assert_eq!(scan.checkpoint(), Some(100));
+ scan.restore(Some(2)).unwrap();
+ for _ in 0..16 {
+ assert!(scan.plan().await.unwrap().is_none());
+ }
+ assert_eq!(scan.checkpoint(), Some(2));
+}
+
+#[tokio::test]
+async fn query_auth_is_checked_before_empty_polling() {
+ let table = table("memory:/stream-query-auth", "none").await;
+ let table = table.copy_with_options(HashMap::from([(
+ "query-auth.enabled".into(),
+ "true".into(),
+ )]));
+ let mut scan = table.new_stream_scan().unwrap();
+ assert!(scan.plan().await.is_err());
+ assert_eq!(scan.checkpoint(), None);
+}
+
+#[tokio::test]
+async fn producer_configuration_uses_typed_case_insensitive_parsing() {
+ for producer in ["none", "NONE", "input", "INPUT"] {
+ let table = table(&format!("memory:/stream-producer-{producer}"),
producer).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let mut scan = table.new_stream_scan().unwrap();
+ scan.restore(Some(1)).unwrap();
+ assert_eq!(
+ values(&table, scan.plan().await.unwrap().unwrap()).await,
+ vec![(1, 10)]
+ );
+ }
+ let table = table("memory:/stream-invalid-producer", "invalid").await;
+ assert!(table.new_stream_scan().is_err());
+}
+
+#[tokio::test]
+async fn initial_lookup_and_full_compaction_select_java_levels_only_once() {
+ use paimon::spec::{Manifest, ManifestEntry, ManifestList};
+ let table = table("memory:/stream-initial-levels", "input").await;
+ for (id, level) in [0, 1, 5].into_iter().enumerate() {
+ write_batch(&table, &make_batch(vec![id as i32], vec![id as i32 *
10])).await;
+ let snapshot = table
+ .snapshot_manager()
+ .get_snapshot(id as i64 + 1)
+ .await
+ .unwrap();
+ let list_path = format!(
+ "{}/manifest/{}",
+ table.location(),
+ snapshot.delta_manifest_list()
+ );
+ for meta in ManifestList::read(table.file_io(), &list_path)
+ .await
+ .unwrap()
+ {
+ let path = format!("{}/manifest/{}", table.location(),
meta.file_name());
+ let entries = Manifest::read(table.file_io(),
&path).await.unwrap();
+ let entries: Vec<ManifestEntry> = entries
+ .into_iter()
+ .map(|entry| {
+ let mut value = serde_json::to_value(entry).unwrap();
+ value["_FILE"]["_LEVEL"] = level.into();
+ serde_json::from_value(value).unwrap()
+ })
+ .collect();
+ Manifest::write(table.file_io(), &path, &entries)
+ .await
+ .unwrap();
+ }
+ }
+ for (producer, levels, expected) in [
+ ("lookup", None, vec![(1, 10), (2, 20)]),
+ ("full-compaction", None, vec![(2, 20)]),
+ ("full-compaction", Some("2"), vec![(1, 10)]),
+ ] {
+ let mut options = HashMap::from([("changelog-producer".into(),
producer.into())]);
+ if let Some(levels) = levels {
+ options.insert("num-levels".into(), levels.into());
+ }
+ let copy = table.copy_with_options(options);
+ let mut scan = copy.new_stream_scan().unwrap();
+ assert_eq!(
+ values(©, scan.plan().await.unwrap().unwrap()).await,
+ expected
+ );
+ // The initial level restriction must not hide L0 changelog events.
+ scan.restore(Some(1)).unwrap();
+ assert_eq!(
+ values(©, scan.plan().await.unwrap().unwrap()).await,
+ vec![(0, 0)]
+ );
+ }
+}
