This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 487a411492 [python] Use Native StreamTableScan for streaming reads
(#10434)
487a411492 is described below
commit 487a4114923a7f7c29966ce8fa460f301ba75ce0
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 19:25:28 2026 +0800
[python] Use Native StreamTableScan for streaming reads (#10434)
---
paimon-python/conftest.py | 13 +
paimon-python/pypaimon/read/native_plan.py | 18 +-
paimon-python/pypaimon/read/stream_read_builder.py | 20 +-
.../pypaimon/read/streaming_table_scan.py | 181 +++++----
.../pypaimon/tests/native_pk_blob_limit_test.py | 12 +-
.../pypaimon/tests/native_plan_incremental_test.py | 16 +-
.../tests/native_stream_table_scan_test.py | 405 +++++++++++++++++++++
.../pypaimon/tests/stream_read_builder_test.py | 23 ++
.../pypaimon/tests/streaming_table_scan_test.py | 161 --------
9 files changed, 600 insertions(+), 249 deletions(-)
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index 0ca0ac11a0..85c1f635a0 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -113,6 +113,19 @@ def pytest_configure(config):
TableScan._try_native_plan = tracked_plan
+ from pypaimon.read.streaming_table_scan import StreamTableScan
+
+ original_stream_plan = StreamTableScan.plan
+
+ def tracked_stream_plan(self, *args, **kwargs):
+ global _native_plan_count
+ plan = original_stream_plan(self, *args, **kwargs)
+ if plan is not None and _force_native_for_test:
+ _native_plan_count += 1
+ return plan
+
+ StreamTableScan.plan = tracked_stream_plan
+
if _native_read_enabled():
from pypaimon.read.table_read import TableRead
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 60ece775c9..bd41fb6185 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -449,7 +449,23 @@ def native_plan(
scan = scan.with_chunk_shuffle(str(seed), chunk_size)
if shard is not None:
scan = scan.with_shard(*shard)
- rust_plan = scan.plan()
+ return _from_native_plan(table, scan.plan())
+
+
+def native_stream_scan(table, predicate=None, read_type=None,
bucket_filter=None, consumer_id=None):
+ """Create Rust's stateful StreamTableScan with the canonical reader
type."""
+ builder = _configure_native_read_builder(
+ _native_read_builder(table), predicate, None, None,
read_type=read_type)
+ scan = builder.new_stream_scan()
+ if bucket_filter is not None:
+ scan.with_bucket_filter(bucket_filter)
+ if consumer_id is not None:
+ scan.with_consumer_id(consumer_id)
+ return scan
+
+
+def _from_native_plan(table, rust_plan) -> Plan:
+ """Keep one split/metadata bridge for batch and streaming planning."""
rust_splits = rust_plan.splits()
pfields = _partition_fields(table)
# Trimmed primary keys decode per-file min/max keys (PK merge-on-read).
diff --git a/paimon-python/pypaimon/read/stream_read_builder.py
b/paimon-python/pypaimon/read/stream_read_builder.py
index 13ac1185e7..a3b48c000e 100644
--- a/paimon-python/pypaimon/read/stream_read_builder.py
+++ b/paimon-python/pypaimon/read/stream_read_builder.py
@@ -22,12 +22,15 @@ This module provides a builder for configuring streaming
reads from Paimon
tables, similar to ReadBuilder but for continuous streaming use cases.
"""
+import logging
+
from typing import Callable, Dict, List, Optional, Set, Union
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.read.native_plan import _raise_if_native_fork_safety_error
from pypaimon.read.read_builder import ReadBuilder
-from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan,
StreamTableScan
from pypaimon.read.table_read import TableRead
from pypaimon.schema.data_types import DataField
@@ -122,10 +125,21 @@ class StreamReadBuilder:
bucket_set: Set[int] = set(bucket_ids)
return self.with_bucket_filter(lambda bucket: bucket in bucket_set)
- def new_streaming_scan(self) -> AsyncStreamingTableScan:
- """Create a new AsyncStreamingTableScan with this builder's
settings."""
+ def new_streaming_scan(self) -> Union[AsyncStreamingTableScan,
StreamTableScan]:
+ """Create a stateful streaming scan using the configured backend."""
projection = self._projection_builder()
projection._validate_map_key_filter()
+ if (self.table.options.native_plan_enabled()
+ and self.table.catalog_environment.table_query_auth(
+ self.table.options, self.table.identifier) is None):
+ try:
+ return StreamTableScan(
+ self.table, predicate=self._predicate,
read_type=projection.read_type(),
+ poll_interval_ms=self._poll_interval_ms,
bucket_filter=self._bucket_filter,
+ consumer_id=self._consumer_id)
+ except Exception as error:
+ _raise_if_native_fork_safety_error(error)
+ logging.warning("Cannot create Native stream scan, using
Python: %s", error)
scan = AsyncStreamingTableScan(
table=self.table,
predicate=self._predicate,
diff --git a/paimon-python/pypaimon/read/streaming_table_scan.py
b/paimon-python/pypaimon/read/streaming_table_scan.py
index cc904e3961..bb666901f3 100644
--- a/paimon-python/pypaimon/read/streaming_table_scan.py
+++ b/paimon-python/pypaimon/read/streaming_table_scan.py
@@ -50,6 +50,7 @@ from pypaimon.read.scanner.incremental_diff_scanner import \
from pypaimon.read.scanner.primary_key_table_split_generator import \
PrimaryKeyTableSplitGenerator
from pypaimon.snapshot.snapshot import Snapshot
+from pypaimon.table.bucket_mode import BucketMode
class AsyncStreamingTableScan:
@@ -145,9 +146,10 @@ class AsyncStreamingTableScan:
if self.next_snapshot_id is None:
latest_snapshot = self._snapshot_manager.get_latest_snapshot()
if latest_snapshot:
+ plan = self._create_initial_plan(latest_snapshot)
self.next_snapshot_id = latest_snapshot.id + 1
self._stage_consumer()
- yield self._create_initial_plan(latest_snapshot)
+ yield plan
# Resumes here when caller calls __anext__() — after caller
processed the plan.
self._flush_pending_consumer()
@@ -211,14 +213,15 @@ class AsyncStreamingTableScan:
self.follow_up_scanner.should_scan,
lookahead_size=self._lookahead_size
)
- self._lookahead_skips += skipped_count
- self.next_snapshot_id = next_id
-
# Check if we found a scannable snapshot or skipped some
snapshot_processed = skipped_count > 0 or snapshot is not None
if snapshot is not None:
plan = self._create_follow_up_plan(snapshot)
+ # Advance only after constructing the selected frame. A failed
+ # callback/read must leave that snapshot available for retry.
+ self._lookahead_skips += skipped_count
+ self.next_snapshot_id = next_id
if plan is not None:
# Start prefetching next scannable snapshot before yielding
@@ -334,6 +337,12 @@ class AsyncStreamingTableScan:
return [e for e in entries if self._bucket_filter(e.bucket)]
return entries
+ def _only_read_real_buckets(self) -> bool:
+ # Java DataTableStreamScan leaves postpone pending data visible only
+ # when there is no changelog producer to emit it after assignment.
+ return (self.table.options.bucket() == BucketMode.POSTPONE_BUCKET.value
+ and self.table.options.changelog_producer() !=
ChangelogProducer.NONE)
+
def __create_initial_plan_raw(self, snapshot, auth_result=None):
def all_manifests():
return self._manifest_list_manager.read_all(snapshot), snapshot
@@ -344,90 +353,29 @@ class AsyncStreamingTableScan:
predicate=self.predicate,
limit=None
)
+ starting_scanner.only_read_real_buckets =
self._only_read_real_buckets()
if auth_result is not None:
from pypaimon.read.table_scan import prune_scanner_by_auth
prune_scanner_by_auth(self.table, starting_scanner, auth_result)
- return starting_scanner.scan()
+ plan = starting_scanner.scan()
+ if self._bucket_filter is not None:
+ plan = Plan([split for split in plan.splits() if
self._bucket_filter(split.bucket)],
+ snapshot_id=plan.snapshot_id)
+ return plan
def _create_initial_plan(self, snapshot: Snapshot) -> Plan:
"""Create a Plan for the initial full scan of the latest snapshot."""
auth_result = self.__auth_query()
- plan = None
- if auth_result is None:
- plan = self._try_native_plan(snapshot.id)
- if plan is None:
- plan = self.__create_initial_plan_raw(snapshot, auth_result)
+ plan = self.__create_initial_plan_raw(snapshot, auth_result)
return wrap_plan_with_auth(auth_result, plan)
def _create_delta_plan(self, snapshot: Snapshot) -> Plan:
"""Read new files from delta_manifest_list
(changelog-producer=none)."""
- plan = self._try_native_plan(
- snapshot.id, incremental_range=(snapshot.id - 1, snapshot.id))
- if plan is not None:
- return plan
manifest_files = self._manifest_list_manager.read_delta(snapshot)
return self._create_plan_from_manifests(manifest_files, snapshot.id)
- def _try_native_plan(self, expected_snapshot_id: int,
- incremental_range=None,
- incremental_mode='delta') -> Optional[Plan]:
- """Plan an initial, delta or changelog streaming frame with
pypaimon-rust.
-
- An arbitrary Python bucket predicate cannot be represented by the Rust
- planner, so sharded stream consumers retain the Python plan. Any native
- setup or planning error other than a fork-safety failure falls back
- before a plan is returned.
- """
- if (not self.table.options.native_plan_enabled()
- or self._bucket_filter is not None):
- return None
- # Rust's full-snapshot scan currently has batch visibility semantics.
- # For first-row and non-MOR deletion-vector tables it hides level-0
- # files, while an initial streaming scan must include those files.
- # Incremental scans already use Rust's streaming split mode and do not
- # need this fallback.
- if (incremental_range is None
- and self.table.options.batch_scan_skip_level0()):
- return None
- try:
- from pypaimon.read.native_plan import native_plan
- plan = native_plan(
- self.table,
- predicate=self.predicate,
- projection=(
- [field.name for field in self._read_type]
- if self._read_type is not None else None),
- incremental_range=incremental_range,
- incremental_mode=incremental_mode,
- )
- if plan.snapshot_id != expected_snapshot_id:
- logging.warning(
- "Native streaming plan resolved snapshot %s, expected %s; "
- "falling back to Python planning",
- plan.snapshot_id, expected_snapshot_id)
- return None
- return plan
- except Exception as error:
- _raise_if_native_fork_safety_error(error)
- logging.warning(
- "Native streaming plan failed, falling back to Python
planning: %s",
- error)
- return None
-
def _create_changelog_plan(self, snapshot: Snapshot) -> Plan:
- """Read from changelog_manifest_list
(changelog-producer=input/full-compaction/lookup)."""
- # Java's range-based incremental CHANGELOG scan skips OVERWRITE,
- # whereas ChangelogFollowUpScanner reads any follow-up snapshot with a
- # changelog manifest. Rust's incremental API implements the former, so
- # keep OVERWRITE on the per-snapshot Python planner.
- if snapshot.commit_kind != 'OVERWRITE':
- plan = self._try_native_plan(
- snapshot.id,
- incremental_range=(snapshot.id - 1, snapshot.id),
- incremental_mode='changelog',
- )
- if plan is not None:
- return plan
+ """Java ChangelogFollowUpScanner consumes any snapshot with a
changelog."""
manifest_files = self._manifest_list_manager.read_changelog(snapshot)
return self._create_plan_from_manifests(manifest_files, snapshot.id)
@@ -452,6 +400,8 @@ class AsyncStreamingTableScan:
)
# Apply shard/bucket filtering for parallel consumption
+ if self._only_read_real_buckets():
+ entries = [entry for entry in entries if entry.bucket >= 0]
entries = self._filter_entries_for_shard(entries) if entries else []
if not entries:
return Plan([], snapshot_id=snapshot_id)
@@ -511,3 +461,88 @@ class AsyncStreamingTableScan:
else:
plan = IncrementalDiffScanner(self.table).scan(start_snapshot,
end_snapshot)
return wrap_plan_with_auth(auth_result, plan)
+
+
+class StreamTableScan:
+ """Python iterator adapter for Rust's stateful StreamTableScan.
+
+ Rust owns snapshot selection, split planning, checkpoints and consumer
+ persistence. Python only polls, converts splits and acknowledges a yielded
+ plan when the caller resumes the iterator after processing it.
+ """
+
+ def __init__(self, table, predicate=None, read_type=None,
poll_interval_ms=1000,
+ bucket_filter=None, consumer_id=None):
+ from pypaimon.read.native_plan import native_stream_scan
+ self.table = table
+ self.poll_interval = poll_interval_ms / 1000.0
+ self._scan = native_stream_scan(
+ table, predicate=predicate, read_type=read_type,
+ bucket_filter=bucket_filter, consumer_id=consumer_id)
+ self._pending_consumer_snapshot = None
+
+ @property
+ def next_snapshot_id(self):
+ return self.checkpoint()
+
+ @next_snapshot_id.setter
+ def next_snapshot_id(self, next_snapshot_id):
+ self.restore(next_snapshot_id)
+
+ def checkpoint(self):
+ return self._scan.checkpoint()
+
+ def watermark(self):
+ return self._scan.watermark()
+
+ def restore(self, next_snapshot_id=None):
+ self._scan.restore(next_snapshot_id)
+ self._pending_consumer_snapshot = None
+
+ def notify_checkpoint_complete(self, next_snapshot):
+ self._scan.notify_checkpoint_complete(next_snapshot)
+
+ def plan(self):
+ from pypaimon.read.native_plan import _from_native_plan
+ # Decode before returning or acknowledging progress. A failed
conversion
+ # must leave the selected frame available for retry.
+ checkpoint = self.checkpoint()
+ native_plan = self._scan.plan()
+ if native_plan is None:
+ return None
+ try:
+ return _from_native_plan(self.table, native_plan)
+ except Exception:
+ self._scan.restore(checkpoint)
+ raise
+
+ def _flush_pending_consumer(self):
+ if self._pending_consumer_snapshot is not None:
+ self.notify_checkpoint_complete(self._pending_consumer_snapshot)
+ self._pending_consumer_snapshot = None
+
+ async def stream(self):
+ while True:
+ self._flush_pending_consumer()
+ checkpoint = self.checkpoint()
+ plan = self.plan()
+ if plan is None:
+ next_snapshot = self.checkpoint()
+ if next_snapshot is not None and next_snapshot != checkpoint:
+ # Skipped frames have no rows awaiting caller processing.
+ self._pending_consumer_snapshot = next_snapshot
+ self._flush_pending_consumer()
+ await asyncio.sleep(self.poll_interval)
+ continue
+ self._pending_consumer_snapshot = self.checkpoint()
+ yield plan
+
+ def stream_sync(self):
+ loop = asyncio.new_event_loop()
+ iterator = self.stream()
+ try:
+ while True:
+ yield loop.run_until_complete(iterator.__anext__())
+ finally:
+ loop.run_until_complete(iterator.aclose())
+ loop.close()
diff --git a/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
b/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
index c3a49b79d0..140faa3bf9 100644
--- a/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
+++ b/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
@@ -249,7 +249,11 @@ def
test_streaming_pk_blob_row_kinds_resolve_only_selected_payload(tmp_path, pay
dict(id=2, tag=1, payload=missing, output=missing)])
builder = table.new_stream_read_builder().with_projection(['id',
'payload'])
scan = builder.new_streaming_scan()
- splits =
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
+ if table.options.native_plan_enabled():
+ scan.restore(table.snapshot_manager().get_latest_snapshot().id)
+ splits = scan.plan().splits()
+ else:
+ splits =
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
assert splits and all(split.is_streaming for split in splits)
predicate = builder.new_predicate_builder().equal('payload', b'selected')
if payload_filter else None
reader = TableRead(table, predicate, builder.read_type(),
include_row_kind=True, limit=1)
@@ -268,7 +272,11 @@ def
test_pk_row_kind_limit_is_global_across_snapshot_and_delta_splits(tmp_path):
splits = table.new_read_builder().new_scan().plan().splits()
_write(table, [dict(id=2, tag=1, payload=missing, output=missing, pt='b')])
scan = table.new_stream_read_builder().new_streaming_scan()
- splits +=
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
+ if table.options.native_plan_enabled():
+ scan.restore(table.snapshot_manager().get_latest_snapshot().id)
+ splits += scan.plan().splits()
+ else:
+ splits +=
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
assert len(splits) == 2 and not splits[0].is_streaming and
splits[1].is_streaming
# Exercise the Rust quota directly: TableRead's outer truncation must not
# conceal extra rows or payload reads in the core row-kind path.
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 31c54ea997..a72c8ba3e2 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -406,12 +406,9 @@ def
test_incremental_reader_preserves_all_physical_row_kinds(catalog, native):
@pytest.mark.native_plan
[email protected](
- not native_method_available('ReadBuilder', 'with_nested_projection'),
- reason='pypaimon_rust nested native reader API required')
def test_stream_read_builder_combines_native_nested_projection_and_row_kind(
catalog):
- from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+ from pypaimon.read.streaming_table_scan import StreamTableScan
schema = pa.schema([
('k', pa.int64()),
@@ -445,11 +442,12 @@ def
test_stream_read_builder_combines_native_nested_projection_and_row_kind(
builder = (native_table.new_stream_read_builder()
.with_projection(['payload.score', 'k'])
.with_include_row_kind())
- # Use the same delta-plan primitive as the streaming loop without polling.
+ # Restore the public stream scan to consume this snapshot as a delta.
scan = builder.new_streaming_scan()
- assert isinstance(scan, AsyncStreamingTableScan)
+ assert isinstance(scan, StreamTableScan)
snapshot = native_table.snapshot_manager().get_latest_snapshot()
- plan = scan._create_delta_plan(snapshot)
+ scan.restore(snapshot.id)
+ plan = scan.plan()
with patch(
'pypaimon.read.table_read.TableRead._create_split_read',
side_effect=AssertionError('streaming nested native read fell
back')):
@@ -494,8 +492,8 @@ def
test_streaming_changelog_frames_use_native_plan_and_read(catalog):
if len(plans) == 2:
return plans
- with patch.object(
- scan, '_create_plan_from_manifests',
+ with patch(
+ 'pypaimon.read.streaming_table_scan.ManifestFileManager',
side_effect=AssertionError(
'streaming changelog native plan fell back to Python')):
plans = asyncio.run(first_two_frames())
diff --git a/paimon-python/pypaimon/tests/native_stream_table_scan_test.py
b/paimon-python/pypaimon/tests/native_stream_table_scan_test.py
new file mode 100644
index 0000000000..61b9d6f941
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_stream_table_scan_test.py
@@ -0,0 +1,405 @@
+# 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.
+
+"""Native StreamTableScan follows Java's initial and follow-up semantics."""
+
+import asyncio
+from contextlib import ExitStack
+import json
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+
+
+pytestmark = pytest.mark.native_plan
+SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+
+
+def _table(tmp_path, engine='deduplicate', producer='none', buckets=1,
extra=None, schema=SCHEMA):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ options = {
+ 'bucket': str(buckets), 'merge-engine': engine,
+ 'changelog-producer': producer, 'write.native.enabled': 'false',
+ 'source.split.target-size': '1 b', 'source.split.open-file-cost': '1
b',
+ }
+ options.update(extra or {})
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options=options), False)
+ return catalog.get_table('db.t')
+
+
+def _write(table, rows, overwrite=False, schema=SCHEMA):
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table.snapshot_manager().get_latest_snapshot()
+
+
+def _frame(table, snapshot, mode, native, bucket_filter=None, projection=None,
predicate=None):
+ if mode == 'delta':
+ table = table.copy({'changelog-producer': 'none'})
+ builder = table.copy({
+ 'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower(),
+ }).new_stream_read_builder().with_include_row_kind()
+ if bucket_filter is not None:
+ builder.with_bucket_filter(bucket_filter)
+ if projection is not None:
+ builder.with_projection(projection)
+ if predicate is not None:
+ builder.with_filter(predicate)
+ scan = builder.new_streaming_scan()
+ try:
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch(
+ 'pypaimon.read.streaming_table_scan.FileScanner',
+ side_effect=AssertionError('Python initial planning')))
+ stack.enter_context(patch(
+ 'pypaimon.read.streaming_table_scan.ManifestFileManager',
+ side_effect=AssertionError('Python incremental planning')))
+ if native:
+ if mode != 'initial':
+ scan.restore(snapshot.id)
+ plan = scan.plan()
+ else:
+ plan = getattr(scan, '_create_{}_plan'.format(mode))(snapshot)
+ if plan is None:
+ assert native and mode != 'initial'
+ assert scan.checkpoint() == snapshot.id + 1
+ return None, []
+ reader = builder.new_read()
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(
+ reader, '_create_split_read',
side_effect=AssertionError('Python read')))
+ rows = reader.to_arrow(plan.splits()).to_pylist()
+ assert plan.snapshot_id == snapshot.id
+ if native:
+ assert all(getattr(split, '_native_split', None) is not None for
split in plan.splits())
+ return plan, sorted(rows, key=lambda row: (row.get('id', 0),
repr(row)))
+ finally:
+ if getattr(scan, '_prefetch_executor', None) is not None:
+ scan._prefetch_executor.shutdown(wait=True)
+
+
[email protected]('engine', ['deduplicate', 'first-row'])
+def
test_initial_frame_includes_level_zero_and_merges_latest_snapshot(tmp_path,
engine):
+ table = _table(tmp_path, engine)
+ _write(table, [{'id': 1, 'value': 10}, {'id': 2, 'value': 20}])
+ selected = _write(table, [{'id': 1, 'value': 11}, {'id': 3, 'value': 30}])
+ expected = [
+ {'_row_kind': '+I', 'id': 1, 'value': 10 if engine == 'first-row' else
11},
+ {'_row_kind': '+I', 'id': 2, 'value': 20},
+ {'_row_kind': '+I', 'id': 3, 'value': 30},
+ ]
+ for native in (False, True):
+ plan, rows = _frame(table, selected, 'initial', native)
+ assert rows == expected
+ assert all(not split.is_streaming for split in plan.splits())
+
+
+def
test_overwrite_changelog_is_a_stream_frame_not_an_incremental_range(tmp_path):
+ table = _table(tmp_path, producer='input')
+ first = _write(table, [{'id': 1, 'value': 10}])
+ selected = _write(table, [{'id': 2, 'value': 20}], overwrite=True)
+ # The local writer does not publish an overwrite changelog. A fixture with
+ # a real physical changelog list exercises producers which do; its rows
+ # differ from the overwrite's delta data to catch reading the wrong list.
+ path = table.snapshot_manager().get_snapshot_path(selected.id)
+ with open(path) as reader:
+ metadata = json.load(reader)
+ metadata['changelogManifestList'] = first.changelog_manifest_list
+ with open(path, 'w') as writer:
+ json.dump(metadata, writer)
+ selected = table.snapshot_manager().get_snapshot_by_id(selected.id)
+ _write(table, [{'id': 3, 'value': 30}])
+ assert selected.commit_kind == 'OVERWRITE'
+ for native in (False, True):
+ plan, rows = _frame(table, selected, 'changelog', native)
+ assert rows == [{'_row_kind': '+I', 'id': 1, 'value': 10}]
+ assert all(split.is_streaming for split in plan.splits())
+ assert all(file.file_name.startswith('changelog-')
+ for split in plan.splits() for file in split.files)
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
[email protected]('selected', [[], [0], [0, 2], [0, 1, 2, 3]])
+def test_bucket_filter_is_applied_to_every_streaming_frame(tmp_path, mode,
selected):
+ table = _table(tmp_path, producer='input', buckets=4)
+ _write(table, [{'id': i, 'value': i + 100} for i in range(32)])
+ snapshot = _write(table, [{'id': i, 'value': i + 200} for i in range(32)])
+ predicate =
table.new_read_builder().new_predicate_builder().greater_than('id', 7)
+ callback = lambda bucket: bucket in selected
+ python_plan, expected = _frame(table, snapshot, mode, False, callback,
predicate=predicate)
+ native_plan, actual = _frame(table, snapshot, mode, True, callback,
predicate=predicate)
+ assert actual == expected
+ splits = python_plan.splits() + (native_plan.splits() if native_plan is
not None else [])
+ assert all(split.bucket in selected for split in splits)
+ if not selected:
+ assert not actual and (native_plan is None or not native_plan.splits())
+ elif len(selected) == 4:
+ assert len(actual) == 24
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
+def test_empty_selection_retains_snapshot_and_reader_schema(tmp_path, mode):
+ table = _table(tmp_path, producer='input', buckets=4)
+ snapshot = _write(table, [{'id': 1, 'value': 10}])
+ for native in (False, True):
+ plan, rows = _frame(table, snapshot, mode, native,
+ bucket_filter=lambda bucket: False,
projection=['value'])
+ assert (plan is None and mode != 'initial') or plan.snapshot_id ==
snapshot.id
+ assert rows == []
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
+def test_bucket_callback_failure_is_not_retried_or_hidden(tmp_path, mode):
+ table = _table(tmp_path, producer='input', buckets=4)
+ snapshot = _write(table, [{'id': i, 'value': i + 100} for i in range(32)])
+ error = LookupError('bucket callback failure')
+ calls = []
+
+ def callback(bucket):
+ calls.append(bucket)
+ if len(calls) == 2:
+ raise error
+ return True
+
+ with pytest.raises(LookupError) as raised:
+ _frame(table, snapshot, mode, True, bucket_filter=callback)
+ assert raised.value is error
+ assert len(calls) == 2
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
+def test_projected_predicate_operand_stays_internal(tmp_path, mode):
+ table = _table(tmp_path, producer='input')
+ snapshot = _write(table, [{'id': 1, 'value': 10}, {'id': 2, 'value': 20}])
+ predicate =
table.new_read_builder().new_predicate_builder().greater_than('id', 1)
+ for native in (False, True):
+ _, rows = _frame(table, snapshot, mode, native, projection=['value'],
predicate=predicate)
+ assert rows == [{'_row_kind': '+I', 'value': 20}]
+
+
[email protected]('merge_on_read', [False, True])
+def test_initial_deletion_vector_table_keeps_uncompacted_level_zero(tmp_path,
merge_on_read):
+ table = _table(tmp_path, extra={
+ 'deletion-vectors.enabled': 'true',
+ 'deletion-vectors.merge-on-read': str(merge_on_read).lower(),
+ })
+ snapshot = _write(table, [{'id': 1, 'value': 10}, {'id': 2, 'value': 20}])
+ snapshot = _write(table, [{'id': 1, 'value': 11}, {'id': 3, 'value': 30}])
+ for native in (False, True):
+ plan, rows = _frame(table, snapshot, 'initial', native)
+ assert rows == [{'_row_kind': '+I', 'id': 1, 'value': 11},
+ {'_row_kind': '+I', 'id': 2, 'value': 20},
+ {'_row_kind': '+I', 'id': 3, 'value': 30}]
+ assert all(file.level == 0 for split in plan.splits() for file in
split.files)
+ predicate =
table.new_read_builder().new_predicate_builder().equal('value', 10)
+ _, rows = _frame(table, snapshot, 'initial', native,
predicate=predicate)
+ assert rows == []
+
+
[email protected]('producer,mode', [
+ ('none', 'initial'), ('none', 'delta'), ('input', 'initial'), ('input',
'changelog')])
+def test_postpone_pending_visibility_matches_java_producer_rule(tmp_path,
producer, mode):
+ table = _table(tmp_path, producer=producer, buckets=-2)
+ snapshot = _write(table, [{'id': 1, 'value': 10}])
+ expected = ([] if producer != 'none' or mode == 'changelog' else
+ [{'_row_kind': '+I', 'id': 1, 'value': 10}])
+ for native in (False, True):
+ plan, rows = _frame(table, snapshot, mode, native)
+ assert rows == expected
+ if rows:
+ assert all(split.bucket == -2 for split in plan.splits())
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
+def
test_snapshot_plan_retains_structured_read_type_for_nested_and_map_projection(tmp_path,
mode):
+ schema = pa.schema([
+ ('id', pa.int32()),
+ ('payload', pa.struct([('x', pa.int32()), ('unused', pa.string())])),
+ ('attrs', pa.map_(pa.string(), pa.int32())),
+ ])
+ table = _table(tmp_path, producer='input', schema=schema)
+ snapshot = _write(table, [
+ {'id': 1, 'payload': {'x': 10, 'unused': 'drop'}, 'attrs':
[('selected', 100), ('unused', 200)]},
+ {'id': 2, 'payload': None, 'attrs': []},
+ ], schema=schema)
+ projection = {'leaf': 'payload.x', 'lookup': 'attrs["selected"]'}
+ for native in (False, True):
+ _, rows = _frame(table, snapshot, mode, native, projection=projection)
+ assert rows == [{'_row_kind': '+I', 'leaf': 10, 'lookup': 100},
+ {'_row_kind': '+I', 'leaf': None, 'lookup': None}]
+
+
[email protected]('mode', ['initial', 'delta', 'changelog'])
+def test_zero_column_projection_keeps_frame_row_count(tmp_path, mode):
+ table = _table(tmp_path, producer='input')
+ snapshot = _write(table, [{'id': 1, 'value': 10}, {'id': 2, 'value': 20}])
+ for native in (False, True):
+ _, rows = _frame(table, snapshot, mode, native, projection=[])
+ assert rows == [{'_row_kind': '+I'}, {'_row_kind': '+I'}]
+
+
[email protected]('native', [False, True])
[email protected]('initial', [False, True])
+def test_stream_callback_failure_does_not_advance_or_stage_consumer(tmp_path,
native, initial):
+ from pypaimon.consumer.consumer_manager import ConsumerManager
+ from pypaimon.read.streaming_table_scan import AsyncStreamingTableScan
+
+ table = _table(tmp_path)
+ _write(table, [{'id': 1, 'value': 10}])
+ table = table.copy({'scan.native-plan.enabled': str(native).lower(),
+ 'read.native.enabled': str(native).lower()})
+ failure = LookupError('retry selected snapshot')
+ calls = []
+
+ def callback(bucket):
+ calls.append(bucket)
+ if len(calls) == 1:
+ raise failure
+ return True
+
+ scan = (table.new_stream_read_builder().with_bucket_filter(callback)
+ .with_consumer_id('failed-frame').new_streaming_scan() if native
else
+ AsyncStreamingTableScan(table, bucket_filter=callback,
prefetch_enabled=False,
+ consumer_id='failed-frame'))
+ consumers = ConsumerManager(table.file_io, table.table_path)
+ if not initial:
+ scan.next_snapshot_id = 1
+
+ async def scenario():
+ failed_stream = scan.stream()
+ with pytest.raises(LookupError) as caught:
+ await failed_stream.__anext__()
+ assert caught.value is failure
+ assert scan.next_snapshot_id == (None if initial else 1)
+ assert scan._pending_consumer_snapshot is None
+ assert consumers.consumer('failed-frame') is None
+ _write(table, [{'id': 2, 'value': 20}])
+ retry = scan.stream()
+ try:
+ plan = await retry.__anext__()
+ expected_id = 2 if initial else 1
+ assert plan.snapshot_id == expected_id
+ builder = table.new_stream_read_builder().with_include_row_kind()
+ rows = builder.new_read().to_arrow(plan.splits()).to_pylist()
+ expected = [{'_row_kind': '+I', 'id': 1, 'value': 10}]
+ if initial:
+ expected.append({'_row_kind': '+I', 'id': 2, 'value': 20})
+ assert sorted(rows, key=lambda row: row['id']) == expected
+ assert scan.next_snapshot_id == expected_id + 1
+ assert scan._pending_consumer_snapshot == expected_id + 1
+ # Consumer progress waits for the caller to process the yielded
plan.
+ assert consumers.consumer('failed-frame') is None
+ finally:
+ await retry.aclose()
+ await failed_stream.aclose()
+
+ asyncio.run(scenario())
+
+
[email protected]('producer', ['none', 'input'])
+def test_empty_follow_up_acknowledges_skipped_progress(tmp_path, producer):
+ from pypaimon.consumer.consumer_manager import ConsumerManager
+ table = _table(tmp_path, producer=producer)
+ for value in range(3):
+ _write(table, [{'id': 1, 'value': value}])
+ table = table.copy({'scan.native-plan.enabled': 'true'})
+ scan = (table.new_stream_read_builder().with_bucket_filter(lambda _: False)
+ .with_consumer_id('filtered').new_streaming_scan())
+ scan.restore(1)
+
+ async def stop_polling(_):
+ raise asyncio.CancelledError()
+
+ async def scenario():
+ iterator = scan.stream()
+ try:
+ with patch('pypaimon.read.streaming_table_scan.asyncio.sleep',
new=stop_polling):
+ with pytest.raises(asyncio.CancelledError):
+ await iterator.__anext__()
+ finally:
+ await iterator.aclose()
+ asyncio.run(scenario())
+ assert scan.checkpoint() == 4
+ consumers = ConsumerManager(table.file_io, table.table_path)
+ assert consumers.consumer('filtered').next_snapshot == 4
+
+
+def test_large_gap_keeps_all_native_changelog_frames(tmp_path):
+ table = _table(tmp_path, producer='input')
+ for value in range(12):
+ _write(table, [{'id': 1, 'value': value}])
+ builder = table.copy({'scan.native-plan.enabled': 'true',
'read.native.enabled': 'true'})
+ builder = builder.new_stream_read_builder().with_include_row_kind()
+ scan = builder.new_streaming_scan()
+ scan.restore(1)
+ for value in range(12):
+ plan = scan.plan()
+ assert plan.snapshot_id == value + 1
+ assert builder.new_read().to_arrow(plan.splits()).to_pylist() == [
+ {'_row_kind': '+I', 'id': 1, 'value': value}]
+ assert scan.plan() is None
+ assert scan.checkpoint() == 13
+
+
+def test_failed_empty_poll_acknowledgement_is_retryable(tmp_path):
+ from pypaimon.consumer.consumer_manager import ConsumerManager
+ table = _table(tmp_path)
+ _write(table, [{'id': 1, 'value': 10}])
+ scan = (table.copy({'scan.native-plan.enabled':
'true'}).new_stream_read_builder()
+ .with_bucket_filter(lambda _:
False).with_consumer_id('retry-empty').new_streaming_scan())
+ scan.restore(1)
+ failure = IOError('retry consumer acknowledgement')
+
+ async def stop_polling(_):
+ raise asyncio.CancelledError()
+
+ async def scenario():
+ iterator = scan.stream()
+ try:
+ with patch.object(scan, 'notify_checkpoint_complete',
side_effect=failure):
+ with pytest.raises(IOError) as caught:
+ await iterator.__anext__()
+ assert caught.value is failure
+ assert scan._pending_consumer_snapshot == 2
+ finally:
+ await iterator.aclose()
+ iterator = scan.stream()
+ try:
+ with patch('pypaimon.read.streaming_table_scan.asyncio.sleep',
new=stop_polling):
+ with pytest.raises(asyncio.CancelledError):
+ await iterator.__anext__()
+ finally:
+ await iterator.aclose()
+ asyncio.run(scenario())
+ assert scan._pending_consumer_snapshot is None
+ assert ConsumerManager(table.file_io,
table.table_path).consumer('retry-empty').next_snapshot == 2
diff --git a/paimon-python/pypaimon/tests/stream_read_builder_test.py
b/paimon-python/pypaimon/tests/stream_read_builder_test.py
index a88bf4a69a..8e10db26a5 100644
--- a/paimon-python/pypaimon/tests/stream_read_builder_test.py
+++ b/paimon-python/pypaimon/tests/stream_read_builder_test.py
@@ -167,3 +167,26 @@ class TestConsumerIdPassthrough:
if __name__ == '__main__':
pytest.main([__file__, '-v'])
+
+
+class TestNativeStreamFactory:
+ def test_optional_native_initialization_falls_back(self, mock_scan_table):
+ mock_scan_table.options.native_plan_enabled.return_value = True
+ builder = StreamReadBuilder(mock_scan_table)
+ with patch('pypaimon.read.stream_read_builder.StreamTableScan',
+ side_effect=ImportError('optional native runtime
unavailable')):
+ assert isinstance(builder.new_streaming_scan(),
AsyncStreamingTableScan)
+
+ def test_native_planning_failure_is_not_retried_with_python(self,
mock_scan_table):
+ from pypaimon.read.streaming_table_scan import StreamTableScan
+ failure = LookupError('native planning failed')
+ native = MagicMock()
+ native.plan.side_effect = failure
+ with patch('pypaimon.read.native_plan.native_stream_scan',
return_value=native):
+ scan = StreamTableScan(mock_scan_table)
+ with patch('pypaimon.read.streaming_table_scan.FileScanner') as
python_planner:
+ with pytest.raises(LookupError) as caught:
+ scan.plan()
+ assert caught.value is failure
+ python_planner.assert_not_called()
+ native.plan.assert_called_once()
diff --git a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
index 41c1416a1c..2aea5287f1 100644
--- a/paimon-python/pypaimon/tests/streaming_table_scan_test.py
+++ b/paimon-python/pypaimon/tests/streaming_table_scan_test.py
@@ -18,10 +18,7 @@
"""Tests for AsyncStreamingTableScan."""
import asyncio
-import sys
import unittest
-from concurrent.futures import Future
-from types import ModuleType
from unittest.mock import Mock, patch
from pypaimon.common.options.core_options import ChangelogProducer
@@ -97,164 +94,6 @@ class AsyncStreamingTableScanTest(unittest.TestCase):
self.assertIsInstance(plan, Plan)
self.assertEqual(scan.next_snapshot_id, 6)
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_delta_scan_uses_native_plan_with_projection(
- self, native_plan, _manifest_files, _manifest_lists):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- native_plan.return_value = Plan([], snapshot_id=5)
- scan = AsyncStreamingTableScan(table, predicate=Mock())
- scan._read_type = [Mock(name='payload'), Mock(name='id')]
- scan._read_type[0].name = 'payload'
- scan._read_type[1].name = 'id'
-
- plan = scan._create_delta_plan(_create_mock_snapshot(5))
-
- self.assertIs(plan, native_plan.return_value)
- self.assertEqual(native_plan.call_args.kwargs['incremental_range'],
(4, 5))
- self.assertEqual(
- native_plan.call_args.kwargs['projection'], ['payload', 'id'])
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_changelog_scan_uses_explicit_native_mode(
- self, native_plan, _manifest_files, _manifest_lists):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- table.options.changelog_producer.return_value = ChangelogProducer.INPUT
- native_plan.return_value = Plan([], snapshot_id=5)
- scan = AsyncStreamingTableScan(table, predicate=Mock())
- scan._read_type = [Mock(name='payload')]
- scan._read_type[0].name = 'payload'
-
- plan = scan._create_changelog_plan(_create_mock_snapshot(5))
-
- self.assertIs(plan, native_plan.return_value)
- self.assertEqual(
- native_plan.call_args.kwargs['incremental_range'], (4, 5))
- self.assertEqual(
- native_plan.call_args.kwargs['incremental_mode'], 'changelog')
- self.assertEqual(
- native_plan.call_args.kwargs['projection'], ['payload'])
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_streaming_plans_propagate_native_fork_safety_error(
- self, native_plan, _manifest_files, manifest_lists):
- class ForkSafetyError(RuntimeError):
- pass
-
- module = ModuleType('pypaimon_rust')
- module.ForkSafetyError = ForkSafetyError
- native_plan.side_effect = ForkSafetyError('cannot reuse Jindo after
fork')
-
- for mode in ('initial', 'delta', 'changelog'):
- with self.subTest(mode=mode):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- table.options.batch_scan_skip_level0.return_value = False
- scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
- snapshot = _create_mock_snapshot(5)
- with patch.dict(sys.modules, {'pypaimon_rust': module}),
patch.object(
- scan, '_AsyncStreamingTableScan__auth_query',
return_value=None), patch.object(
- scan,
'_AsyncStreamingTableScan__create_initial_plan_raw') as initial, patch.object(
- scan, '_create_plan_from_manifests') as from_manifests:
- with self.assertRaises(ForkSafetyError):
- getattr(scan, '_create_{}_plan'.format(mode))(snapshot)
- initial.assert_not_called()
- from_manifests.assert_not_called()
- manifest_lists.return_value.read_all.assert_not_called()
- manifest_lists.return_value.read_delta.assert_not_called()
- manifest_lists.return_value.read_changelog.assert_not_called()
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan',
side_effect=RuntimeError('unsupported'))
- def test_delta_scan_falls_back_after_other_native_error(
- self, _native_plan, _manifest_files, manifest_lists):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- manifest_lists.return_value.read_delta.return_value = []
- scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
- snapshot = _create_mock_snapshot(5)
-
- self.assertEqual(scan._create_delta_plan(snapshot).snapshot_id, 5)
-
manifest_lists.return_value.read_delta.assert_called_once_with(snapshot)
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_prefetch_propagates_native_fork_safety_error(
- self, native_plan, _manifest_files, _manifest_lists):
- class ForkSafetyError(RuntimeError):
- pass
-
- module = ModuleType('pypaimon_rust')
- module.ForkSafetyError = ForkSafetyError
- error = ForkSafetyError('cannot reuse Jindo after fork')
- native_plan.side_effect = error
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- scan = AsyncStreamingTableScan(table, prefetch_enabled=False)
- scan._snapshot_manager.find_next_scannable.return_value = (
- _create_mock_snapshot(5), 6, 0)
-
- with patch.dict(sys.modules, {'pypaimon_rust': module}):
- with self.assertRaises(ForkSafetyError):
- scan._fetch_plan_with_lookahead(5)
-
- future = Future()
- future.set_exception(error)
- scan._prefetch_future = future
- scan.next_snapshot_id = 5
- with patch.object(scan, '_should_use_diff_catch_up',
return_value=False):
- with self.assertRaises(ForkSafetyError):
- asyncio.run(scan.stream().__anext__())
- scan._snapshot_manager.find_next_scannable.assert_called_once()
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_overwrite_changelog_keeps_per_snapshot_python_plan(
- self, native_plan, _manifest_files, manifest_lists):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- table.options.changelog_producer.return_value = ChangelogProducer.INPUT
- snapshot = _create_mock_snapshot(5, 'OVERWRITE')
- manifests = [Mock()]
- manifest_lists.return_value.read_changelog.return_value = manifests
- python_plan = Plan([], snapshot_id=5)
- scan = AsyncStreamingTableScan(table)
-
- with patch.object(
- scan, '_create_plan_from_manifests',
- return_value=python_plan) as create_plan:
- plan = scan._create_changelog_plan(snapshot)
-
- self.assertIs(plan, python_plan)
- native_plan.assert_not_called()
-
manifest_lists.return_value.read_changelog.assert_called_once_with(snapshot)
- create_plan.assert_called_once_with(manifests, 5)
-
- @patch('pypaimon.read.streaming_table_scan.ManifestListManager')
- @patch('pypaimon.read.streaming_table_scan.ManifestFileManager')
- @patch('pypaimon.read.native_plan.native_plan')
- def test_bucket_filtered_delta_scan_keeps_python_planner(
- self, native_plan, _manifest_files, manifest_lists):
- table, _ = _create_mock_table()
- table.options.native_plan_enabled.return_value = True
- manifest_lists.return_value.read_delta.return_value = []
- scan = AsyncStreamingTableScan(table, bucket_filter=lambda bucket:
True)
-
- plan = scan._create_delta_plan(_create_mock_snapshot(5))
-
- self.assertEqual(plan.snapshot_id, 5)
- native_plan.assert_not_called()
-
@patch('pypaimon.read.streaming_table_scan.ManifestListManager')
@patch('pypaimon.read.streaming_table_scan.FileScanner')
def test_stream_skips_non_append_commits(self, MockStartingScanner,
MockManifestListManager):
