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 5e5f2f4c77 [python] Support incremental-between with snapshot ids and
tags (#10857)
5e5f2f4c77 is described below
commit 5e5f2f4c7773af394a6d7832a26f049574dfd86e
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Oct 29 10:10:03 2026 +0800
[python] Support incremental-between with snapshot ids and tags (#10857)
---
.../pypaimon/common/options/core_options.py | 62 ++++++++-
.../pypaimon/execution/native/read/plan.py | 5 +
.../pypaimon/execution/python/read/table_scan.py | 61 ++++++++-
paimon-python/pypaimon/multimodal/temporal.py | 3 +-
.../pypaimon/multimodal/window_dataset.py | 3 +-
paimon-python/pypaimon/ray/join_common.py | 3 +-
paimon-python/pypaimon/read/file_scanner.py | 4 +-
paimon-python/pypaimon/read/scan_config.py | 17 ++-
paimon-python/pypaimon/table/file_store_table.py | 8 +-
.../pypaimon/tests/multimodal_temporal_test.py | 35 +++--
.../pypaimon/tests/native_plan_incremental_test.py | 151 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_plan_test.py | 13 ++
.../pypaimon/tests/table/file_store_table_test.py | 19 +++
.../pypaimon/tests/table_scan_mode_test.py | 29 ++++
.../pypaimon/tests/write/table_write_test.py | 22 +++
15 files changed, 391 insertions(+), 44 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 93f04872c0..91ff535c2b 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -87,6 +87,13 @@ class StartupMode(str, Enum):
INCREMENTAL = "incremental"
+class IncrementalBetweenScanMode(str, Enum):
+ AUTO = "auto"
+ DELTA = "delta"
+ CHANGELOG = "changelog"
+ DIFF = "diff"
+
+
class GlobalIndexColumnUpdateAction(str, Enum):
THROW_ERROR = "THROW_ERROR"
DROP_PARTITION_INDEX = "DROP_PARTITION_INDEX"
@@ -559,6 +566,20 @@ class CoreOptions:
.with_description("The fallback branch for scanning.")
)
+ INCREMENTAL_BETWEEN: ConfigOption[str] = (
+ ConfigOptions.key("incremental-between")
+ .string_type()
+ .no_default_value()
+ .with_description("Read changes between start snapshot (exclusive) and
end snapshot (inclusive).")
+ )
+
+ INCREMENTAL_BETWEEN_SCAN_MODE: ConfigOption[IncrementalBetweenScanMode] = (
+ ConfigOptions.key("incremental-between-scan-mode")
+ .enum_type(IncrementalBetweenScanMode)
+ .default_value(IncrementalBetweenScanMode.AUTO)
+ .with_description("Scan mode for incremental changes between
snapshots.")
+ )
+
INCREMENTAL_BETWEEN_TIMESTAMP: ConfigOption[str] = (
ConfigOptions.key("incremental-between-timestamp")
.string_type()
@@ -566,6 +587,26 @@ class CoreOptions:
.with_description("The timestamp range for incremental reading.")
)
+ INCREMENTAL_TO_AUTO_TAG: ConfigOption[str] = (
+ ConfigOptions.key("incremental-to-auto-tag")
+ .string_type()
+ .no_default_value()
+ .with_description("Read changes between an earlier auto-created tag
and the specified end tag (inclusive).")
+ )
+
+ INCREMENTAL_SCAN_KEYS = [
+ INCREMENTAL_BETWEEN.key(),
+ INCREMENTAL_BETWEEN_TIMESTAMP.key(),
+ INCREMENTAL_TO_AUTO_TAG.key(),
+ ]
+
+ INCREMENTAL_BETWEEN_TAG_TO_SNAPSHOT: ConfigOption[bool] = (
+ ConfigOptions.key("incremental-between-tag-to-snapshot")
+ .boolean_type()
+ .default_value(False)
+ .with_description("Read incremental changes between the snapshots
corresponding to the tags.")
+ )
+
SCAN_TAG_NAME: ConfigOption[str] = (
ConfigOptions.key("scan.tag-name")
.string_type()
@@ -1541,12 +1582,12 @@ class CoreOptions:
or self.options.contains(CoreOptions.SCAN_WATERMARK)
or self.options.contains(CoreOptions.SCAN_VERSION)):
return StartupMode.FROM_SNAPSHOT
- elif
self.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
- return StartupMode.INCREMENTAL
elif
self.options.contains(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS):
return StartupMode.FROM_FILE_CREATION_TIME
elif self.options.contains(CoreOptions.SCAN_CREATION_TIME_MILLIS):
return StartupMode.FROM_CREATION_TIMESTAMP
+ elif any(self.options.contains_key(key) for key in
CoreOptions.INCREMENTAL_SCAN_KEYS):
+ return StartupMode.INCREMENTAL
else:
return StartupMode.LATEST_FULL
elif mode == StartupMode.FULL:
@@ -1565,6 +1606,23 @@ class CoreOptions:
def incremental_between_timestamp(self, default=None):
return self.options.get(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
default)
+ def incremental_between(self):
+ value = self.options.get(CoreOptions.INCREMENTAL_BETWEEN)
+ # Java String.split discards trailing empty strings.
+ parts = value.rstrip(',').split(',')
+ if len(parts) != 2:
+ raise ValueError(
+ "The incremental-between must specific start(exclusive) and
end snapshot,"
+ " for example, 'incremental-between'='5,10' means changes
between snapshot 5 and snapshot 10. But is: "
+ + value)
+ return parts[0], parts[1]
+
+ def incremental_between_scan_mode(self):
+ return self.options.get(CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE)
+
+ def incremental_between_tag_to_snapshot(self):
+ return
self.options.get(CoreOptions.INCREMENTAL_BETWEEN_TAG_TO_SNAPSHOT)
+
def scan_tag_name(self, default=None):
return self.options.get(CoreOptions.SCAN_TAG_NAME, default)
diff --git a/paimon-python/pypaimon/execution/native/read/plan.py
b/paimon-python/pypaimon/execution/native/read/plan.py
index 9525a21a61..357de9d4f2 100644
--- a/paimon-python/pypaimon/execution/native/read/plan.py
+++ b/paimon-python/pypaimon/execution/native/read/plan.py
@@ -258,6 +258,11 @@ class NativeTableScan:
options = scan.table.options.options
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
unsupported_scan_keys = set(SCAN_KEYS) - _NATIVE_TIME_TRAVEL_OPTIONS
+ # Snapshot/tag ranges and explicit incremental modes use the Python
scanner.
+ unsupported_scan_keys.update((
+ CoreOptions.INCREMENTAL_BETWEEN.key(),
CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE.key(),
+ CoreOptions.INCREMENTAL_TO_AUTO_TAG.key(),
+ ))
return not any(options.contains_key(k) for k in unsupported_scan_keys)
def _native_global_index_result_supported(self, request:
Optional[ScanRequest] = None) -> bool:
diff --git a/paimon-python/pypaimon/execution/python/read/table_scan.py
b/paimon-python/pypaimon/execution/python/read/table_scan.py
index 22e1dacdd9..20e16ae530 100644
--- a/paimon-python/pypaimon/execution/python/read/table_scan.py
+++ b/paimon-python/pypaimon/execution/python/read/table_scan.py
@@ -17,7 +17,9 @@
"""Python file scanner construction for the public TableScan."""
-from pypaimon.common.options.core_options import CoreOptions, ChangelogProducer
+import re
+
+from pypaimon.common.options.core_options import CoreOptions,
ChangelogProducer, IncrementalBetweenScanMode, StartupMode
from pypaimon.catalog.catalog_exception import TableNoPermissionException
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.execution.python.read.auth_support import prune_scanner_by_auth
@@ -32,6 +34,8 @@ from pypaimon.utils.range import Range
def _resolve_incremental_snapshot_range(table):
"""Resolve Python scan boundaries only after selecting this backend."""
window = validate_scan_options(table)
+ if table.options.options.contains(CoreOptions.INCREMENTAL_BETWEEN):
+ return _resolve_incremental_ids(table)
if window is None:
return None
start_timestamp, end_timestamp = window
@@ -60,6 +64,50 @@ def _resolve_incremental_snapshot_range(table):
return start_id, end_id
+def _resolve_incremental_ids(table):
+ start, end = table.options.incremental_between()
+ tag_manager = table.tag_manager()
+ start_tag, end_tag = tag_manager.get(start), tag_manager.get(end)
+ if start_tag is not None and end_tag is not None:
+ start_id, end_id = start_tag.id, end_tag.id
+ if not table.options.incremental_between_tag_to_snapshot():
+ if end_id < start_id:
+ raise ValueError(
+ "Tag end %s with snapshot id %s should be >= tag start %s
with snapshot id %s"
+ % (end, end_id, start, start_id))
+ if start_id == end_id:
+ return None
+ raise NotImplementedError(
+ "Incremental diff between tags is not supported in pypaimon. "
+ "Set incremental-between-tag-to-snapshot=true to scan snapshot
deltas or changelogs.")
+ if table.options.incremental_between_scan_mode() ==
IncrementalBetweenScanMode.DIFF:
+ raise NotImplementedError("Unsupported incremental scan mode DIFF")
+ else:
+ try:
+ if any(re.fullmatch(r'[+-]?[0-9]+', value) is None for value in
(start, end)):
+ raise ValueError()
+ start_id, end_id = int(start), int(end)
+ if not all(-(1 << 63) <= value < (1 << 63) for value in (start_id,
end_id)):
+ raise ValueError()
+ except ValueError:
+ raise ValueError(
+ "Didn't find two tags for start '%s' and end '%s', and they
are not two snapshot Ids. "
+ "Please set two tags or two snapshot Ids." % (start, end))
+ if end_id < start_id:
+ raise ValueError("Ending snapshotId should >= starting snapshotId
%s. [%s]" % (end_id, start_id))
+ if table.snapshot_manager().get_latest_snapshot() is None or start_id
== end_id:
+ return None
+
+ snapshot_manager = table.snapshot_manager()
+ latest_id = snapshot_manager.get_latest_snapshot().id
+ earliest_id = snapshot_manager.try_get_earliest_snapshot(latest_id).id
+ if start_id < earliest_id - 1 or end_id > latest_id:
+ raise ValueError(
+ "The specified scan snapshotId range [%s, %s] is out of available
snapshotId range [%s, %s]."
+ % (start_id, end_id, earliest_id, latest_id))
+ return start_id, end_id
+
+
class PythonTableScan:
"""Plan with Python's file scanner and the public scan configuration."""
@@ -78,7 +126,7 @@ class PythonTableScan:
from pypaimon.snapshot.time_travel_util import TimeTravelUtil,
SCAN_KEYS
has_time_travel = any(options.contains_key(key) for key in SCAN_KEYS)
- has_incremental =
options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
+ has_incremental = scan.table.options.startup_mode() ==
StartupMode.INCREMENTAL
if has_incremental:
snapshot_range = _resolve_incremental_snapshot_range(scan.table)
@@ -89,9 +137,12 @@ class PythonTableScan:
partition_predicate=scan.partition_predicate,
)
start_id, end_id = snapshot_range
- # Java's AUTO mode uses physical changelogs when a producer is
- # configured, and APPEND deltas otherwise.
- changelog = scan.table.options.changelog_producer() !=
ChangelogProducer.NONE
+ mode = scan.table.options.incremental_between_scan_mode()
+ if mode == IncrementalBetweenScanMode.DIFF:
+ raise NotImplementedError("incremental-between-scan-mode
'diff' is not supported in pypaimon.")
+ changelog = (mode == IncrementalBetweenScanMode.CHANGELOG
+ or (mode == IncrementalBetweenScanMode.AUTO
+ and scan.table.options.changelog_producer() !=
ChangelogProducer.NONE))
def incremental_manifest():
snapshots_in_range = []
diff --git a/paimon-python/pypaimon/multimodal/temporal.py
b/paimon-python/pypaimon/multimodal/temporal.py
index 1affc09c5c..7b11127f6a 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -942,8 +942,7 @@ def _pin_scan_to_snapshot(query):
raise ValueError(
"Temporal alignment requires 'row-tracking.enabled' = 'true'.")
if (options.scan_mode() == StartupMode.INCREMENTAL
- or options.options.contains(
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)):
+ or any(options.options.contains_key(key) for key in
CoreOptions.INCREMENTAL_SCAN_KEYS)):
raise ValueError(
"Temporal alignment does not support incremental scans; "
"inputs must "
diff --git a/paimon-python/pypaimon/multimodal/window_dataset.py
b/paimon-python/pypaimon/multimodal/window_dataset.py
index f2240557e3..c697d6eac9 100644
--- a/paimon-python/pypaimon/multimodal/window_dataset.py
+++ b/paimon-python/pypaimon/multimodal/window_dataset.py
@@ -818,10 +818,9 @@ def _pin_table(table, snapshot_id):
tag_name = table.options.scan_tag_name()
if tag_name is not None:
_require_tag_snapshot(table, tag_name, snapshot_id)
- scan_keys = set(SCAN_KEYS)
+ scan_keys = set(SCAN_KEYS + CoreOptions.INCREMENTAL_SCAN_KEYS)
scan_keys.update(option.key() for option in (
CoreOptions.SCAN_MODE,
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
CoreOptions.SCAN_CREATION_TIME_MILLIS,
))
diff --git a/paimon-python/pypaimon/ray/join_common.py
b/paimon-python/pypaimon/ray/join_common.py
index 0892ea9c08..118d4aa5f9 100644
--- a/paimon-python/pypaimon/ray/join_common.py
+++ b/paimon-python/pypaimon/ray/join_common.py
@@ -84,9 +84,10 @@ def pin_latest_snapshot(table) -> Optional[int]:
for key in (CoreOptions.SCAN_MODE, CoreOptions.SCAN_SNAPSHOT_ID,
CoreOptions.SCAN_TAG_NAME, CoreOptions.SCAN_WATERMARK,
CoreOptions.SCAN_TIMESTAMP, CoreOptions.SCAN_TIMESTAMP_MILLIS,
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
CoreOptions.SCAN_CREATION_TIME_MILLIS):
opts.data.pop(key.key(), None)
+ for key in CoreOptions.INCREMENTAL_SCAN_KEYS:
+ opts.data.pop(key, None)
opts.set(CoreOptions.SCAN_SNAPSHOT_ID, snapshot.id)
return snapshot.id
diff --git a/paimon-python/pypaimon/read/file_scanner.py
b/paimon-python/pypaimon/read/file_scanner.py
index 2b958c8b41..d2f5f6458b 100644
--- a/paimon-python/pypaimon/read/file_scanner.py
+++ b/paimon-python/pypaimon/read/file_scanner.py
@@ -16,7 +16,7 @@
"""File-level configuration and dispatch for the public table scanner."""
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import StartupMode
from pypaimon.read.scan_request import scan_partition_predicate
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.utils.range import Range
@@ -60,7 +60,7 @@ class FileScanner:
self._global_index_result = request.global_index_result
self._row_ranges = (None if request.row_ranges is None else
[Range(start, end) for start, end in
request.row_ranges])
- self.is_streaming =
self.table.options.options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
+ self.is_streaming = self.table.options.startup_mode() ==
StartupMode.INCREMENTAL
self.skip_level0 = False if self.is_streaming else
self.table.options.batch_scan_skip_level0()
self.only_read_real_buckets = self.table.options.bucket() ==
BucketMode.POSTPONE_BUCKET.value
self.auth_partition_predicate = None
diff --git a/paimon-python/pypaimon/read/scan_config.py
b/paimon-python/pypaimon/read/scan_config.py
index e31106625c..e2c2cf1761 100644
--- a/paimon-python/pypaimon/read/scan_config.py
+++ b/paimon-python/pypaimon/read/scan_config.py
@@ -41,7 +41,7 @@ def _validate_scan_mode(table):
has_watermark = options.contains(CoreOptions.SCAN_WATERMARK)
has_timestamp_millis = options.contains(CoreOptions.SCAN_TIMESTAMP_MILLIS)
has_timestamp = options.contains(CoreOptions.SCAN_TIMESTAMP)
- has_incremental =
options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP)
+ incremental_keys = [key for key in CoreOptions.INCREMENTAL_SCAN_KEYS if
options.contains_key(key)]
has_file_creation_time =
options.contains(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS)
has_creation_time = options.contains(CoreOptions.SCAN_CREATION_TIME_MILLIS)
@@ -58,8 +58,7 @@ def _validate_scan_mode(table):
present_keys.append(CoreOptions.SCAN_TIMESTAMP_MILLIS.key())
if has_timestamp:
present_keys.append(CoreOptions.SCAN_TIMESTAMP.key())
- if has_incremental:
- present_keys.append(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key())
+ present_keys.extend(incremental_keys)
if has_file_creation_time:
present_keys.append(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())
if has_creation_time:
@@ -101,11 +100,11 @@ def _validate_scan_mode(table):
"scan.version, scan.snapshot-id, scan.tag-name, or
scan.watermark is set."
)
elif mode == StartupMode.INCREMENTAL:
- allowed = {CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key()}
- if not has_incremental:
+ allowed = set(CoreOptions.INCREMENTAL_SCAN_KEYS)
+ if len(incremental_keys) != 1:
raise ValueError(
- "scan.mode is 'incremental' but "
- "incremental-between-timestamp is not set."
+ "must set only one key in
[incremental-between,incremental-between-timestamp,incremental-to-auto-tag] "
+ "when you use incremental for scan.mode"
)
elif mode in (StartupMode.LATEST_FULL, StartupMode.LATEST):
allowed = set()
@@ -146,6 +145,10 @@ def validate_scan_options(table):
_validate_scan_mode(table)
options = table.options.options
+ if options.contains(CoreOptions.INCREMENTAL_TO_AUTO_TAG):
+ raise NotImplementedError("incremental-to-auto-tag is not supported in
pypaimon.")
+ if options.contains(CoreOptions.INCREMENTAL_BETWEEN):
+ table.options.incremental_between()
has_time_travel = any(options.contains_key(key) for key in SCAN_KEYS)
if not options.contains(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP):
return None
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 5bf2886dea..30111c91e9 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -493,8 +493,7 @@ class FileStoreTable(Table):
def _copy_with_snapshot(self, snapshot):
"""Keep one resolved read view, including tag metadata and empty
tables."""
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
- selectors = SCAN_KEYS + [
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(),
+ selectors = SCAN_KEYS + CoreOptions.INCREMENTAL_SCAN_KEYS + [
CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
]
@@ -529,9 +528,8 @@ class FileStoreTable(Table):
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
preserve_snapshot = hasattr(self, "_read_snapshot") and not any(
- key in options for key in SCAN_KEYS + [
- CoreOptions.SCAN_MODE.key(), CoreOptions.BRANCH.key(),
- CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key()])
+ key in options for key in SCAN_KEYS +
CoreOptions.INCREMENTAL_SCAN_KEYS + [
+ CoreOptions.SCAN_MODE.key(), CoreOptions.BRANCH.key()])
# Re-encode the branch into the identifier when the option changes, so
# current_branch() and any catalog-routed snapshot commit see the
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index b65f5f7309..b8b5f30867 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -2241,17 +2241,10 @@ class MultimodalTemporalTest(unittest.TestCase):
).to_list()
def test_alignment_rejects_incremental_scans(self):
- anchors = self.conn.create_table(
- "incremental_anchors",
- schema=pa.schema([
- pa.field("episode_id", pa.int32()),
- pa.field("event_time", pa.int64()),
- ]),
- options={
- "scan.mode": "incremental",
- "incremental-between-timestamp": "0,9999999999999",
- },
- )
+ anchors = self._table("incremental_anchors", {
+ "episode_id": pa.int32(),
+ "event_time": pa.int64(),
+ })
source = self._table("incremental_source", {
"episode_id": pa.int32(),
"event_time": pa.int64(),
@@ -2262,13 +2255,19 @@ class MultimodalTemporalTest(unittest.TestCase):
"episode_id": 1, "event_time": 100, "value": 7,
}])
- with self.assertRaisesRegex(
- ValueError, "Temporal alignment.*incremental"):
- pmm.join_asof(
- anchors.scan(), source.scan().select("value"),
- on="event_time", by="episode_id",
- direction="nearest", tolerance=0,
- ).to_list()
+ for mode in ("default", "incremental"):
+ for selector, value in (
+ ("incremental-between", "0,1"),
+ ("incremental-between-timestamp", "0,9999999999999"),
+ ("incremental-to-auto-tag", "end")):
+ with self.subTest(mode=mode, selector=selector):
+ query =
temporal.ScanQuery(anchors.raw_table.copy({"scan.mode": mode, selector: value}))
+ with self.assertRaisesRegex(ValueError, "Temporal
alignment.*incremental"):
+ pmm.join_asof(
+ query, source.scan().select("value"),
+ on="event_time", by="episode_id",
+ direction="nearest", tolerance=0,
+ ).to_list()
def test_alignment_supports_zero_output_columns(self):
anchors = self._table("zero_output_anchors", {
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 423fa8c899..62d90acf57 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -634,3 +634,154 @@ def test_timestamp_auto_reads_physical_changelog(native,
catalog):
assert all(split.snapshot_id == 3 for split in plan.splits())
assert all(file.file_name.startswith('changelog-') for split in
plan.splits() for file in split.files)
assert sorted((row['k'], row['v']) for row in rows) == [(1, 'new'), (2,
'insert')]
+
+
+def _read_incremental_options(table, options, configure_scanner=False):
+ builder = table.copy(options).new_read_builder()
+ scan = builder.new_scan()
+ if configure_scanner:
+ assert scan.file_scanner.is_streaming
+ assert not scan.file_scanner.skip_level0
+ plan = scan.plan()
+ return plan, builder.new_read().to_arrow(plan.splits()).to_pylist()
+
+
[email protected]('primary_key', [False, True])
[email protected]('window,expected', [
+ ('1,2', [2]), ('0,1', [1]), ('+1,02', [2]), ('1,2,', [2]),
+ ('1,3', [2, 3]), ('2,2', []), ('99,99', []),
+])
+def test_snapshot_id_window(catalog, primary_key, window, expected):
+ table = _table(catalog, 'snapshot_window', primary_key, {'bucket': '1'} if
primary_key else None)
+ for key in range(1, 4):
+ _write(table, key * 100, [{'k': key, 'v': str(key)}])
+ plan, rows = _read_incremental_options(table, {'incremental-between':
window}, configure_scanner=True)
+ assert sorted(row['k'] for row in rows) == expected
+ assert plan.snapshot_id == (max(expected) if expected else None)
+
+
[email protected]('window,message', [
+ ('3,1', 'Ending snapshotId should >= starting snapshotId 1. [3]'),
+ ('-1,2', 'The specified scan snapshotId range [-1, 2] is out of available
snapshotId range [1, 5].'),
+ ('1,6', 'The specified scan snapshotId range [1, 6] is out of available
snapshotId range [1, 5].'),
+ (',2', "Tag name shouldn't be blank."),
+])
+def test_snapshot_id_window_validation(history, window, message):
+ with pytest.raises(ValueError) as error:
+ _read_incremental_options(history, {'incremental-between': window})
+ assert str(error.value) == message
+
+
[email protected]('window', [
+ 'missing,end', '1,end', ' 1,2', '1,2.0', '1_0,20',
+ '1,9223372036854775808', '-9223372036854775809,2',
+])
+def test_snapshot_id_window_invalid_ids(catalog, window):
+ table = _table(catalog, 'invalid_ids')
+ with pytest.raises(ValueError) as error:
+ _read_incremental_options(table, {'incremental-between': window})
+ start, end = window.split(',')
+ assert str(error.value) == (
+ "Didn't find two tags for start '%s' and end '%s', and they are not
two snapshot Ids. "
+ "Please set two tags or two snapshot Ids." % (start, end))
+
+
+def test_snapshot_id_window_empty_table(catalog):
+ table = _table(catalog, 'empty_ids')
+ assert _read_incremental_options(table, {'incremental-between':
'0,99'})[1] == []
+ with pytest.raises(ValueError, match='Ending snapshotId'):
+ _read_incremental_options(table, {'incremental-between': '2,1'})
+
+
+def test_snapshot_id_window_expired_start(history):
+ manager = history.snapshot_manager()
+ history.file_io.delete_quietly(manager.get_snapshot_path(1))
+ assert _read_incremental_options(history, {'incremental-between':
'1,2'})[0].snapshot_id == 2
+ with pytest.raises(ValueError, match=r'available snapshotId range \[2,
5\]'):
+ _read_incremental_options(history, {'incremental-between': '0,2'})
+
+
+def test_snapshot_id_window_earliest_beyond_default_probe(history):
+ manager = history.snapshot_manager()
+ for snapshot_id in (1, 2):
+ history.file_io.delete_quietly(manager.get_snapshot_path(snapshot_id))
+ with
patch('pypaimon.snapshot.snapshot_manager.EARLIEST_SNAPSHOT_DEFAULT_RETRY_NUM',
1):
+ assert _read_incremental_options(history, {'incremental-between':
'2,3'})[0].snapshot_id == 3
+
+
[email protected]('names', [('tag1', 'tag2'), ('1', '2')])
+def test_tag_window_resolves_tags_before_snapshot_ids(history, names):
+ manager = history.snapshot_manager()
+ for name, snapshot_id in zip(names, (2, 3)):
+
history.tag_manager().create_tag(manager.get_snapshot_by_id(snapshot_id), name)
+ plan, rows = _read_incremental_options(history, {
+ 'incremental-between': ','.join(names),
'incremental-between-tag-to-snapshot': 'true'})
+ assert plan.snapshot_id == 3
+ assert sorted((row['k'], row['v']) for row in rows) == [(1, 'latest'), (4,
'fourth')]
+
+
+def test_tag_window_default_diff_and_ordering(history):
+ manager = history.snapshot_manager()
+ for name, snapshot_id in [('start', 1), ('end', 3)]:
+
history.tag_manager().create_tag(manager.get_snapshot_by_id(snapshot_id), name)
+ with pytest.raises(NotImplementedError, match='diff'):
+ _read_incremental_options(history, {'incremental-between':
'start,end'})
+ with pytest.raises(ValueError) as error:
+ _read_incremental_options(history, {'incremental-between':
'end,start'})
+ assert str(error.value) == 'Tag end start with snapshot id 1 should be >=
tag start end with snapshot id 3'
+ assert _read_incremental_options(history, {'incremental-between':
'start,start'})[1] == []
+ with pytest.raises(ValueError, match='Please set two tags or two snapshot
Ids'):
+ _read_incremental_options(history, {'incremental-between': 'start,3'})
+
+
+def test_tag_window_uses_branch_history(catalog):
+ table = _table(catalog, 'tag_branch')
+ _write(table, 100, [{'k': 1, 'v': 'base'}])
+ table.create_tag('base')
+ catalog.create_branch(table.identifier, 'test', tag_name='base')
+ branch = catalog.get_table(Identifier('default', 'tag_branch',
branch='test'))
+ for current, value in ((table, 'main'), (branch, 'branch')):
+ _write(current, 200, [{'k': 2, 'v': value}])
+ current.create_tag('end')
+ _, rows = _read_incremental_options(branch, {
+ 'incremental-between': 'base,end',
'incremental-between-tag-to-snapshot': 'true'})
+ assert rows == [{'k': 2, 'v': 'branch'}]
+
+
[email protected]('selector,window', [
+ ('incremental-between', '1,3'), ('incremental-between-timestamp',
'100,300'),
+])
[email protected]('producer', ['none', 'input'])
[email protected]('mode', ['auto', 'delta', 'changelog', 'diff',
'invalid'])
+def test_incremental_scan_mode(catalog, selector, window, producer, mode):
+ table = _table(catalog, 'scan_modes', True, {'bucket': '1',
'changelog-producer': producer})
+ for timestamp, value in ((100, '1'), (200, '2'), (300, '1')):
+ _write(table, timestamp, [{'k': 1, 'v': value}])
+ options = {selector: window, 'incremental-between-scan-mode': mode}
+ if mode in ('diff', 'invalid'):
+ with pytest.raises(NotImplementedError if mode == 'diff' else
ValueError, match=mode):
+ _read_incremental_options(table, options)
+ return
+ plan, rows = _read_incremental_options(table, options)
+ assert sorted(row['v'] for row in rows) == ([] if mode == 'changelog' and
producer == 'none' else ['1', '2'])
+ changelog = mode == 'changelog' or (mode == 'auto' and producer != 'none')
+ assert all(file.file_name.startswith('changelog-') == changelog
+ for split in plan.splits() for file in split.files)
+
+
+def test_incremental_scan_mode_skips_compact_and_overwrite(history):
+ manager = history.snapshot_manager()
+ path = manager.get_snapshot_path(3)
+ snapshot = json.loads(history.file_io.read_file_utf8(path))
+ snapshot['commitKind'] = 'COMPACT'
+ history.file_io.overwrite_file_utf8(path, json.dumps(snapshot))
+ plan, rows = _read_incremental_options(history, {'incremental-between':
'1,5'})
+ assert plan.snapshot_id == 5
+ assert sorted((row['k'], row['v']) for row in rows) == [(1,
'intermediate'), (3, 'third'), (5, 'fifth')]
+
+
[email protected]('mode', ['default', 'incremental'])
+def test_incremental_to_auto_tag_is_explicitly_unsupported(catalog, mode):
+ table = _table(catalog, 'auto_tag')
+ with pytest.raises(NotImplementedError, match='incremental-to-auto-tag'):
+ _read_incremental_options(table, {'scan.mode': mode,
'incremental-to-auto-tag': '2026-10-27'})
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index f8a35522bf..321fcfc7fa 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -1359,6 +1359,19 @@ class NativePlanTest(unittest.TestCase):
self.assertNotIn('incremental_range', native.call_args[1])
fs.scan.assert_not_called()
+ def test_incremental_selectors_use_python_planning(self):
+ for options in ({'incremental-between': '1,2'},
+ {'incremental-between-timestamp': '100,200',
'incremental-between-scan-mode': 'changelog'},
+ {'incremental-to-auto-tag': 'tag2'}):
+ with self.subTest(options=options):
+ fs = Mock(partition_key_predicate=None)
+ scan = _scan(True, fs)
+ scan.table.options.options = Options(options)
+ with patch('pypaimon.execution.native.read.plan.native_plan')
as native:
+ scan.plan()
+ native.assert_not_called()
+ fs.scan.assert_called_once()
+
def test_native_plan_forwards_explicit_incremental_mode(self):
table = _scan(True, Mock()).table
table.table_schema = Mock(fields=[], partition_keys=[])
diff --git a/paimon-python/pypaimon/tests/table/file_store_table_test.py
b/paimon-python/pypaimon/tests/table/file_store_table_test.py
index 0786984af8..1bac930648 100644
--- a/paimon-python/pypaimon/tests/table/file_store_table_test.py
+++ b/paimon-python/pypaimon/tests/table/file_store_table_test.py
@@ -104,6 +104,25 @@ class FileStoreTableTest(unittest.TestCase):
CoreOptions.SCAN_SNAPSHOT_ID.key()],
)
+ def test_snapshot_pin_clears_incremental_selectors(self):
+ for selector in ('incremental-between',
'incremental-between-timestamp', 'incremental-to-auto-tag'):
+ with self.subTest(selector=selector):
+ selected = self.table.copy({selector: '1,2'})
+ pinned = selected._copy_with_snapshot(None)
+ self.assertNotIn(selector, pinned.table_schema.options)
+ self.assertIsNone(pinned._read_snapshot)
+ self.assertEqual([],
pinned.new_read_builder().new_scan().plan().splits())
+ self.assertIn(selector, selected.table_schema.options)
+
+ def test_incremental_selector_copy_invalidates_pinned_snapshot(self):
+ pinned = self.table._copy_with_snapshot(None)
+ for selector in ('incremental-between',
'incremental-between-timestamp', 'incremental-to-auto-tag'):
+ for copy in (pinned.copy, pinned.copy_without_time_travel):
+ with self.subTest(selector=selector, copy=copy.__name__):
+ selected = copy({selector: '1,2'})
+ self.assertFalse(hasattr(selected, '_read_snapshot'))
+ self.assertIsNone(pinned.copy({'read.batch-size': '1'})._read_snapshot)
+
def test_consumer_manager(self):
"""Test that FileStoreTable has consumer_manager method."""
# Get consumer_manager
diff --git a/paimon-python/pypaimon/tests/table_scan_mode_test.py
b/paimon-python/pypaimon/tests/table_scan_mode_test.py
index c33fdf9beb..0b66785e2f 100644
--- a/paimon-python/pypaimon/tests/table_scan_mode_test.py
+++ b/paimon-python/pypaimon/tests/table_scan_mode_test.py
@@ -69,6 +69,35 @@ class TableScanModeTest(unittest.TestCase):
self.assertEqual(options.startup_mode(), StartupMode.FROM_SNAPSHOT)
_scan(options.options.to_map())._validate_scan_mode()
+ def test_incremental_selector_startup_mode(self):
+ for key in ('incremental-between', 'incremental-between-timestamp',
'incremental-to-auto-tag'):
+ with self.subTest(key=key):
+ scan = _scan({key: '1,2'})
+ self.assertEqual(scan.table.options.startup_mode(),
StartupMode.INCREMENTAL)
+ scan._validate_scan_mode()
+
+ def test_incremental_selector_requires_exactly_one_range(self):
+ for ranges in ({}, {'incremental-between': '1,2',
'incremental-between-timestamp': '100,200'},
+ {'incremental-between': '1,2',
'incremental-to-auto-tag': 'tag2'}):
+ with self.subTest(ranges=ranges):
+ scan = _scan(dict(ranges, **{'scan.mode': 'incremental'}))
+ with self.assertRaisesRegex(ValueError, 'must set only one
key'):
+ scan._validate_scan_mode()
+
+ def test_incremental_selector_conflicts_with_time_travel(self):
+ for key in ('incremental-between', 'incremental-to-auto-tag'):
+ for scan_options in ({'scan.snapshot-id': '1'}, {'scan.mode':
'latest-full'}):
+ with self.subTest(key=key, scan_options=scan_options):
+ with self.assertRaisesRegex(ValueError, key):
+ _scan(dict(scan_options, **{key:
'1,2'}))._validate_scan_mode()
+
+ def test_incremental_between_format(self):
+ for value in ('', '1', '1,', ',', '1,2,3'):
+ with self.subTest(value=value):
+ with self.assertRaisesRegex(ValueError, 'The
incremental-between must specific'):
+ CoreOptions.from_dict({'incremental-between':
value}).incremental_between()
+ self.assertEqual(CoreOptions.from_dict({'incremental-between':
'1,2,'}).incremental_between(), ('1', '2'))
+
def test_unsupported_scan_modes_raise_value_error(self):
scan = _scan({
CoreOptions.SCAN_MODE.key(): StartupMode.COMPACTED_FULL.value,
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 85609c222b..dabfa5898a 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -522,6 +522,28 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(True, snapshot_json.__contains__("writerVersion"))
self.assertEqual(False, snapshot_json.__contains__("nextRowId"))
+ @parameterized.expand([
+ ('incremental-between', '1,2'),
+ ('incremental-between-timestamp', '0,9999999999999'),
+ ('incremental-to-auto-tag', 'end'),
+ ])
+ def test_write_through_incremental_copy_restores_latest_sequence(self,
selector, value):
+ name = 'default.write_' + selector.replace('-', '_')
+ schema = pa.schema([pa.field('id', pa.int64(), nullable=False),
('value', pa.int64())])
+ self.catalog.create_table(name, Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'], options={'bucket': '1'}), False)
+ table = self.catalog.get_table(name)
+ for number in range(3):
+ self._commit_arrow(table, pa.Table.from_pydict({'id': [1],
'value': [number]}, schema=schema))
+
+ selected = table.copy({selector: value})
+ self._commit_arrow(selected, pa.Table.from_pydict({'id': [1], 'value':
[3]}, schema=schema))
+
+ self.assertEqual([{'id': 1, 'value': 3}], self._read_sorted(table,
'id').to_pylist())
+ splits = table.new_read_builder().new_scan().plan().splits()
+ self.assertEqual(3, max(file.max_sequence_number for split in splits
for file in split.files))
+ self.assertEqual(value, selected.table_schema.options[selector])
+
def test_write_row_append_only_partitioned_table(self):
from pypaimon.table.row.generic_row import GenericRow
