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 f978e8bd5f [python] Run timestamp-window MERGE sources in native core
(#10428)
f978e8bd5f is described below
commit f978e8bd5fc776d44732031ef8f2f02ebdf75aa0
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 14:47:15 2026 +0800
[python] Run timestamp-window MERGE sources in native core (#10428)
---
paimon-python/README.md | 10 +-
paimon-python/pypaimon/read/table_scan.py | 243 +++++++++++----------
.../tests/native_merge_incremental_test.py | 192 ++++++++++++++++
.../pypaimon/tests/native_plan_incremental_test.py | 35 ++-
paimon-python/pypaimon/write/native_merge_into.py | 11 +-
5 files changed, 370 insertions(+), 121 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index dc2459e3c0..f4dd3c7fe8 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -184,8 +184,9 @@ except for a sole unconditional DELETE, following Paimon
Spark MERGE.
For table sources, core independently pins the selected source branch/snapshot
and reads only keys and referenced columns. Append and primary-key sources do
not require data evolution. Source scan options select the same point in time
-on Python and native paths. Incremental source scans, non-REST Blob views
-requiring catalog resolution, and packed-video inserts use the Python path
+on Python and native paths. Numeric `incremental-between-timestamp` source
+windows are resolved and read by core. Non-REST Blob views
+requiring catalog resolution and packed-video inserts use the Python path
selected before native execution starts. Native failures propagate without
Python retry or deleting files from earlier prepared actions.
@@ -435,7 +436,10 @@ counts can differ between shards. Limits are applied after
shard/slice selection
Timestamp incremental scans require `ReadBuilder.new_incremental_scan()` and
stream-aware splits exposing `Split.is_streaming()`. Python resolves
`(start_timestamp, end_timestamp]` to snapshot IDs; Rust packs the selected
APPEND
-deltas into one plan. Continuous streaming uses the same native path for
initial
+deltas into one plan. Equal timestamp bounds produce an empty result, as in
Java.
+AUTO selects physical changelog files when a changelog producer is configured;
+each split carries the selected ending snapshot ID. Numeric bounds use signed
+64-bit milliseconds and Java's integer syntax. Continuous streaming uses the
same native path for initial
and delta frames. When `changelog-producer` is enabled, follow-up frames
request
Rust's explicit `changelog` mode and read the physical changelog manifests.
OVERWRITE changelog frames retain per-snapshot Python planning because Java
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 44a09ee5ac..879072a12f 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -17,11 +17,12 @@
import json as _json
import logging
+import re
from typing import List, Optional, Tuple
from pypaimon.catalog.catalog_exception import TableNoPermissionException
from pypaimon.common.identifier import UNKNOWN_DATABASE
-from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.core_options import CoreOptions, ChangelogProducer
from pypaimon.common.predicate import Predicate
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.manifest.manifest_list_manager import ManifestListManager
@@ -183,6 +184,8 @@ class TableScan:
if self._incremental_snapshot_range is None:
return Plan([])
extra_options['incremental_range'] =
self._incremental_snapshot_range
+ if self.table.options.changelog_producer() !=
ChangelogProducer.NONE:
+ extra_options['incremental_mode'] = 'changelog'
chunk_shuffle = fs.chunk_shuffle
if chunk_shuffle is not None:
extra_options['chunk_shuffle'] = chunk_shuffle
@@ -317,26 +320,21 @@ class TableScan:
)
if has_incremental:
- ts =
options.get(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP).split(",")
- if len(ts) != 2:
- raise ValueError(
- "The incremental-between-timestamp must specific
start(exclusive) and end timestamp. But is: " +
- options.get(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP))
- start_timestamp = int(ts[0])
- end_timestamp = int(ts[1])
- if start_timestamp >= end_timestamp:
- raise ValueError(
- "Ending timestamp %s must be greater than starting
timestamp %s."
- % (end_timestamp, start_timestamp))
- earliest_snapshot = snapshot_manager.try_get_earliest_snapshot()
+ start_timestamp, end_timestamp =
_parse_incremental_timestamp_window(
+ options.get(CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP))
latest_snapshot = snapshot_manager.get_latest_snapshot()
+ earliest_snapshot = snapshot_manager.try_get_earliest_snapshot()
if latest_snapshot is not None else None
if earliest_snapshot is None or latest_snapshot is None:
return FileScanner(
self.table,
lambda: ([], None),
partition_predicate=self.partition_predicate,
)
- if (start_timestamp > latest_snapshot.time_millis
+ if start_timestamp > end_timestamp:
+ raise ValueError(
+ "Ending timestamp %s should be >= starting timestamp %s."
+ % (end_timestamp, start_timestamp))
+ if (start_timestamp == end_timestamp or start_timestamp >
latest_snapshot.time_millis
or end_timestamp < earliest_snapshot.time_millis):
return FileScanner(
self.table,
@@ -357,6 +355,9 @@ class TableScan:
latest_snapshot = snapshot_manager.get_latest_snapshot()
end_id = end_snapshot.id if end_snapshot else (latest_snapshot.id
if latest_snapshot else -1)
self._incremental_snapshot_range = (start_id, end_id)
+ # Java's AUTO mode uses physical changelogs when a producer is
+ # configured, and APPEND deltas otherwise.
+ changelog = self.table.options.changelog_producer() !=
ChangelogProducer.NONE
def incremental_manifest():
snapshots_in_range = []
@@ -364,13 +365,15 @@ class TableScan:
for snapshot_id in range(start_id + 1, end_id + 1):
snapshot = snapshot_manager.get_snapshot_by_id(snapshot_id)
end_snapshot = snapshot
- if snapshot.commit_kind == "APPEND":
+ if (snapshot.commit_kind != "OVERWRITE" and
snapshot.changelog_manifest_list
+ if changelog else snapshot.commit_kind ==
"APPEND"):
snapshots_in_range.append(snapshot)
manifests = []
for snapshot in snapshots_in_range:
- manifest_files = manifest_list_manager.read_delta(snapshot)
+ manifest_files =
(manifest_list_manager.read_changelog(snapshot) if changelog else
+
manifest_list_manager.read_delta(snapshot))
manifests.extend(manifest_files)
return manifests, end_snapshot
@@ -436,109 +439,125 @@ class TableScan:
return self
def _validate_scan_mode(self):
- """Validate scan.mode against companion options using a whitelist
approach.
+ _validate_scan_mode(self.table)
- Each StartupMode declares exactly which scan keys are allowed. Any
- scan key present but not in the whitelist for the resolved mode is
- rejected. This matches Java's SchemaValidation mutual-exclusion matrix.
- """
- from pypaimon.common.options.core_options import StartupMode
-
- core_options = self.table.options
- mode = core_options.startup_mode()
- options = core_options.options
-
- has_snapshot_id = options.contains(CoreOptions.SCAN_SNAPSHOT_ID)
- has_tag_name = options.contains(CoreOptions.SCAN_TAG_NAME)
- has_version = options.contains(CoreOptions.SCAN_VERSION)
- 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)
- has_file_creation_time =
options.contains(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS)
- has_creation_time =
options.contains(CoreOptions.SCAN_CREATION_TIME_MILLIS)
-
- present_keys = []
- if has_version:
- present_keys.append(CoreOptions.SCAN_VERSION.key())
- if has_snapshot_id:
- present_keys.append(CoreOptions.SCAN_SNAPSHOT_ID.key())
- if has_tag_name:
- present_keys.append(CoreOptions.SCAN_TAG_NAME.key())
- if has_watermark:
- present_keys.append(CoreOptions.SCAN_WATERMARK.key())
- if has_timestamp_millis:
- 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())
- if has_file_creation_time:
-
present_keys.append(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())
- if has_creation_time:
- present_keys.append(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())
-
- # scan.timestamp-millis and scan.timestamp are mutually exclusive
- if has_timestamp_millis and has_timestamp:
+
+def _validate_scan_mode(table):
+ """Validate scan.mode against companion options using a whitelist approach.
+
+ Each StartupMode declares exactly which scan keys are allowed. Any
+ scan key present but not in the whitelist for the resolved mode is
+ rejected. This matches Java's SchemaValidation mutual-exclusion matrix.
+ """
+ from pypaimon.common.options.core_options import StartupMode
+
+ core_options = table.options
+ mode = core_options.startup_mode()
+ options = core_options.options
+
+ has_snapshot_id = options.contains(CoreOptions.SCAN_SNAPSHOT_ID)
+ has_tag_name = options.contains(CoreOptions.SCAN_TAG_NAME)
+ has_version = options.contains(CoreOptions.SCAN_VERSION)
+ 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)
+ has_file_creation_time =
options.contains(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS)
+ has_creation_time = options.contains(CoreOptions.SCAN_CREATION_TIME_MILLIS)
+
+ present_keys = []
+ if has_version:
+ present_keys.append(CoreOptions.SCAN_VERSION.key())
+ if has_snapshot_id:
+ present_keys.append(CoreOptions.SCAN_SNAPSHOT_ID.key())
+ if has_tag_name:
+ present_keys.append(CoreOptions.SCAN_TAG_NAME.key())
+ if has_watermark:
+ present_keys.append(CoreOptions.SCAN_WATERMARK.key())
+ if has_timestamp_millis:
+ 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())
+ if has_file_creation_time:
+ present_keys.append(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())
+ if has_creation_time:
+ present_keys.append(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())
+
+ # scan.timestamp-millis and scan.timestamp are mutually exclusive
+ if has_timestamp_millis and has_timestamp:
+ raise ValueError(
+ "scan.timestamp-millis and scan.timestamp cannot both be set."
+ )
+
+ # Define allowed companion keys per mode
+ if mode == StartupMode.FROM_TIMESTAMP:
+ allowed = {
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ }
+ if not (has_timestamp_millis or has_timestamp):
raise ValueError(
- "scan.timestamp-millis and scan.timestamp cannot both be set."
+ "scan.mode is 'from-timestamp' but neither "
+ "scan.timestamp-millis nor scan.timestamp is set."
)
-
- # Define allowed companion keys per mode
- if mode == StartupMode.FROM_TIMESTAMP:
- allowed = {
- CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
- CoreOptions.SCAN_TIMESTAMP.key(),
- }
- if not (has_timestamp_millis or has_timestamp):
- raise ValueError(
- "scan.mode is 'from-timestamp' but neither "
- "scan.timestamp-millis nor scan.timestamp is set."
- )
- elif mode == StartupMode.FROM_SNAPSHOT_FULL:
- allowed = {CoreOptions.SCAN_SNAPSHOT_ID.key()}
- if not has_snapshot_id:
- raise ValueError(
- "scan.mode is 'from-snapshot-full' but scan.snapshot-id is
not set."
- )
- elif mode == StartupMode.FROM_SNAPSHOT:
- allowed = {
- CoreOptions.SCAN_SNAPSHOT_ID.key(),
- CoreOptions.SCAN_TAG_NAME.key(),
- CoreOptions.SCAN_WATERMARK.key(),
- CoreOptions.SCAN_VERSION.key(),
- }
- if not (has_snapshot_id or has_tag_name or has_watermark or
has_version):
- raise ValueError(
- "scan.mode is 'from-snapshot' but none of "
- "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:
- raise ValueError(
- "scan.mode is 'incremental' but "
- "incremental-between-timestamp is not set."
- )
- elif mode in (StartupMode.LATEST_FULL, StartupMode.LATEST):
- allowed = set()
- elif mode in (StartupMode.COMPACTED_FULL,
- StartupMode.FROM_CREATION_TIMESTAMP,
- StartupMode.FROM_FILE_CREATION_TIME):
+ elif mode == StartupMode.FROM_SNAPSHOT_FULL:
+ allowed = {CoreOptions.SCAN_SNAPSHOT_ID.key()}
+ if not has_snapshot_id:
raise ValueError(
- f"scan.mode '{mode.value}' is not yet supported in pypaimon."
+ "scan.mode is 'from-snapshot-full' but scan.snapshot-id is not
set."
)
- else:
- allowed = set()
-
- # Reject any scan key that's not in the whitelist for this mode
- disallowed = [k for k in present_keys if k not in allowed]
- if disallowed:
+ elif mode == StartupMode.FROM_SNAPSHOT:
+ allowed = {
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_VERSION.key(),
+ }
+ if not (has_snapshot_id or has_tag_name or has_watermark or
has_version):
+ raise ValueError(
+ "scan.mode is 'from-snapshot' but none of "
+ "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:
raise ValueError(
- f"scan.mode '{mode.value}' conflicts with: {disallowed}. "
- f"Only {sorted(allowed) if allowed else 'no scan keys'} "
- f"are allowed for this mode."
+ "scan.mode is 'incremental' but "
+ "incremental-between-timestamp is not set."
)
+ elif mode in (StartupMode.LATEST_FULL, StartupMode.LATEST):
+ allowed = set()
+ elif mode in (StartupMode.COMPACTED_FULL,
+ StartupMode.FROM_CREATION_TIMESTAMP,
+ StartupMode.FROM_FILE_CREATION_TIME):
+ raise ValueError(
+ f"scan.mode '{mode.value}' is not yet supported in pypaimon."
+ )
+ else:
+ allowed = set()
+
+ # Reject any scan key that's not in the whitelist for this mode
+ disallowed = [k for k in present_keys if k not in allowed]
+ if disallowed:
+ raise ValueError(
+ f"scan.mode '{mode.value}' conflicts with: {disallowed}. "
+ f"Only {sorted(allowed) if allowed else 'no scan keys'} "
+ f"are allowed for this mode."
+ )
+
+
+def _parse_incremental_timestamp_window(value):
+ """Numeric milliseconds use Java Long parsing, including its range."""
+ parts = value.split(',')
+ if len(parts) != 2 or any(re.fullmatch(r'[+-]?[0-9]+', part) is None for
part in parts):
+ raise ValueError('incremental-between-timestamp requires
start(exclusive),end(inclusive) '
+ 'in milliseconds, got: ' + value)
+ start, end = map(int, parts)
+ if not all(-(1 << 63) <= timestamp < (1 << 63) for timestamp in (start,
end)):
+ raise ValueError('incremental-between-timestamp milliseconds must be
signed 64-bit integers')
+ return start, end
def latest_auth_fields(table):
diff --git a/paimon-python/pypaimon/tests/native_merge_incremental_test.py
b/paimon-python/pypaimon/tests/native_merge_incremental_test.py
new file mode 100644
index 0000000000..8f73835463
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_merge_incremental_test.py
@@ -0,0 +1,192 @@
+# 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.
+
+"""Timestamp-window table sources must be planned and read by Rust core."""
+
+import json
+from contextlib import ExitStack
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
+from pypaimon.tests.native_merge_into_test import _SCHEMA, _table, _rows,
_write_rows
+
+pytestmark = pytest.mark.native_plan
+
+
+def _set_time(table, timestamp):
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ path = table.snapshot_manager().get_snapshot_path(snapshot.id)
+ with open(path) as reader:
+ data = json.load(reader)
+ data['timeMillis'] = timestamp
+ with open(path, 'w') as writer:
+ json.dump(data, writer)
+
+
+def _source(tmp_path, mode='append', options=None, schema=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ config = {'write.native.enabled': 'false', 'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false'}
+ if mode == 'pk':
+ config['bucket'] = '1'
+ if mode == 'evolution':
+ config.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ config.update(options or {})
+ catalog.create_table('db.source', Schema.from_pyarrow_schema(
+ schema or _SCHEMA, primary_keys=['id'] if mode == 'pk' else [],
options=config), False)
+ return catalog.get_table('db.source')
+
+
+def _merge(target, source, stream, native, value_column='value'):
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ kwargs = {'commit_identifier': 57} if stream else {}
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch.object(source, 'snapshot_manager',
+
side_effect=AssertionError('Python source snapshot resolution')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python
source materialization')))
+
stack.enter_context(patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python MERGE
matching')))
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col(value_column)})],
+ when_not_matched=[WhenNotMatched({
+ 'id': source_col('id'), 'value': source_col(value_column),
'name': source_col('name')})], **kwargs)
+ commit = builder.new_commit()
+ commit.commit(messages, 57) if stream else commit.commit(messages)
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('mode', ['append', 'pk', 'evolution'])
[email protected]('window,expected', [
+ ('100,200', [dict(id=1, value=11, name='old'), dict(id=2, value=22,
name='insert')]),
+ ('0,100', [dict(id=1, value=-1, name='old')]),
+ ('200,300', [dict(id=1, value=10, name='old'), dict(id=3, value=33,
name='later')]),
+ ('100,150', [dict(id=1, value=10, name='old')]),
+ ('100,100', [dict(id=1, value=10, name='old')]),
+ ('350,500', [dict(id=1, value=10, name='old')]),
+ ('0,50', [dict(id=1, value=10, name='old')]),
+])
+def test_timestamp_window_source_merge(tmp_path, native, stream, mode, window,
expected):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _source(tmp_path / 'source', mode)
+ for timestamp, values in [
+ (100, [dict(id=1, value=-1, name='base')]),
+ (200, [dict(id=1, value=11, name='new'), dict(id=2, value=22,
name='insert')]),
+ (300, [dict(id=3, value=33, name='later')])]:
+ _write_rows(source, _SCHEMA, values)
+ _set_time(source, timestamp)
+ source = source.copy({'scan.mode': 'incremental',
'incremental-between-timestamp': window})
+ _merge(target, source, stream, native)
+ assert _rows(target) == expected
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
+def test_incremental_source_uses_selected_branch(tmp_path, native, stream):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _source(tmp_path / 'source')
+ _write_rows(source, _SCHEMA, [dict(id=1, value=-1, name='base')])
+ _set_time(source, 100)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.create_tag(source.identifier, 'base')
+ catalog.create_branch(source.identifier, 'dev', tag_name='base')
+ branch = catalog.get_table('db.source$branch_dev')
+ _write_rows(branch, _SCHEMA, [dict(id=1, value=11, name='branch'),
dict(id=2, value=22, name='branch-insert')])
+ _set_time(branch, 200)
+ _write_rows(source, _SCHEMA, [dict(id=1, value=99, name='main'),
dict(id=3, value=33, name='main-insert')])
+ _set_time(source, 200)
+ _merge(target, branch.copy({'incremental-between-timestamp': '100,200'}),
stream, native)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='branch-insert')]
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('mode', ['append', 'pk'])
+def test_incremental_source_projects_evolved_field_ids(tmp_path, native,
stream, mode):
+ from pypaimon.schema.schema_change import SchemaChange
+ from pypaimon.schema.data_types import AtomicType
+
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ old_schema = pa.schema([_SCHEMA.field('id'), _SCHEMA.field('value')])
+ source = _source(tmp_path / 'source', mode, schema=old_schema)
+ _write_rows(source, old_schema, [dict(id=1, value=11)])
+ _set_time(source, 100)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.alter_table('db.source', [
+ SchemaChange.rename_column('value', 'incoming'),
SchemaChange.add_column('name', AtomicType('STRING'))], False)
+ source = catalog.get_table('db.source')
+ evolved_schema = pa.schema([_SCHEMA.field('id'), pa.field('incoming',
pa.int32()), _SCHEMA.field('name')])
+ _write_rows(source, evolved_schema, [dict(id=2, incoming=22,
name='insert')])
+ _set_time(source, 200)
+ _merge(target, source.copy({'incremental-between-timestamp': '0,200'}),
stream, native, 'incoming')
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='insert')]
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('window', ['200,100', '100', 'one,200', '100, 200',
'1_00,200',
+ '0,9223372036854775808'])
+def test_invalid_incremental_source_never_stages_target(tmp_path, native,
stream, window):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _source(tmp_path / 'source')
+ _write_rows(source, _SCHEMA, [dict(id=1, value=11, name='new')])
+ _set_time(source, 100)
+ before_snapshot = target.snapshot_manager().get_latest_snapshot().id
+ before_files = {path for path in (tmp_path / 'target').rglob('*') if
path.is_file()}
+ with pytest.raises(ValueError):
+ _merge(target, source.copy({'incremental-between-timestamp': window}),
stream, native)
+ assert target.snapshot_manager().get_latest_snapshot().id ==
before_snapshot
+ assert {path for path in (tmp_path / 'target').rglob('*') if
path.is_file()} == before_files
+ assert _rows(target) == [dict(id=1, value=10, name='old')]
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
+def test_incremental_source_auto_uses_changelog(tmp_path, native, stream):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _source(tmp_path / 'source', 'pk', options={'changelog-producer':
'input'})
+ for time, values in [(100, [dict(id=1, value=-1, name='base')]),
+ (200, [dict(id=1, value=11, name='new')]),
+ (300, [dict(id=2, value=22, name='insert')])]:
+ _write_rows(source, _SCHEMA, values)
+ _set_time(source, time)
+ _merge(target, source.copy({'incremental-between-timestamp': '100,300'}),
stream, native)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='insert')]
+
+
[email protected]('stream', [False, True])
+def test_missing_source_snapshot_does_not_stage_target(tmp_path, stream):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _source(tmp_path / 'source')
+ for time in (100, 200, 300):
+ _write_rows(source, _SCHEMA, [dict(id=time, value=time,
name='source')])
+ _set_time(source, time)
+ source.file_io.delete(source.snapshot_manager().get_snapshot_path(2),
False)
+ before_snapshot = target.snapshot_manager().get_latest_snapshot().id
+ before_files = {path for path in (tmp_path / 'target').rglob('*') if
path.is_file()}
+ with pytest.raises(ValueError, match='Snapshot 2 does not exist'):
+ _merge(target, source.copy({'incremental-between-timestamp':
'0,300'}), stream, True)
+ assert target.snapshot_manager().get_latest_snapshot().id ==
before_snapshot
+ assert {path for path in (tmp_path / 'target').rglob('*') if
path.is_file()} == before_files
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 1bffc16240..31c54ea997 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -286,7 +286,7 @@ def
test_incremental_ignores_deletion_vectors_from_window_end(native, catalog):
assert actual == [rows[4]]
[email protected]('window', ['100,100', '200,100', '100', 'one,200'])
[email protected]('window', ['100', 'one,200', '0,100,200'])
def test_invalid_timestamp_window_is_rejected_even_for_empty_tables(catalog,
window):
table = _table(catalog, 'invalid')
with pytest.raises(ValueError):
@@ -590,3 +590,36 @@ def
test_streaming_reader_honors_explicit_split_deletion_vector(catalog, native,
side_effect=AssertionError('explicit DV native read fell
back')))
result =
read_table.new_read_builder().new_read().to_arrow([dv_split]).to_pylist()
assert result == [{'k': 1, 'v': '1'}, {'k': 3, 'v': '3'}]
+
+
[email protected]('window', [(100, 100), (200, 100)])
+def test_empty_table_timestamp_windows_follow_java(catalog, native, window):
+ table = _table(catalog, 'empty_window')
+ plan, rows = _read(table, native, window)
+ assert rows == []
+ assert plan.snapshot_id is None
+
+
+def test_reverse_window_is_rejected_when_snapshots_exist(catalog):
+ table = _table(catalog, 'reverse_window')
+ _write(table, 100, [{'k': 1, 'v': 'base'}])
+ with pytest.raises(ValueError, match='Ending timestamp'):
+ table.copy({'incremental-between-timestamp':
'200,100'}).new_read_builder().new_scan()
+
+
+def test_equal_window_is_empty_for_populated_tables(native, history):
+ plan, rows = _read(history, native, (100, 100))
+ assert rows == []
+ assert plan.snapshot_id is None
+
+
+def test_timestamp_auto_reads_physical_changelog(native, catalog):
+ table = _table(catalog, 'batch_changelog', True, {'bucket': '1',
'changelog-producer': 'input'})
+ _write(table, 100, [{'k': 1, 'v': 'old'}])
+ _write(table, 200, [{'k': 1, 'v': 'new'}, {'k': 2, 'v': 'insert'}])
+ _write(table, 300, [{'k': 9, 'v': 'overwrite'}], overwrite=True)
+ plan, rows = _read(table, native, (100, 300))
+ assert plan.snapshot_id == 3
+ 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')]
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 7d535ce1dd..8652079f15 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -51,13 +51,14 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
or source.options.video_frame_fields() or
source.options.with_vector_format()
or (source.is_primary_key_table and not
source.trimmed_primary_keys)):
return None
- source.new_read_builder().new_scan()._validate_scan_mode()
+ from pypaimon.read.table_scan import _validate_scan_mode
+ _validate_scan_mode(source)
if not _native_blob_view_supported(source, source.field_names):
return None
- # This operation reads one full source snapshot. Incremental scans stay
- # on the Python path, whose scanner applies the requested delta range.
- if (source.options.scan_mode() not in ('default', 'from-snapshot',
'from-timestamp')
- or any(str(key).startswith('incremental-') for key in
source.table_schema.options)):
+ # Core resolves a full snapshot or PyPaimon's existing timestamp
window.
+ if (source.options.scan_mode() not in ('default', 'from-snapshot',
'from-timestamp', 'incremental')
+ or any(str(key).startswith('incremental-') and key !=
'incremental-between-timestamp'
+ for key in source.table_schema.options)):
return None
from pypaimon.schema.data_types import PyarrowFieldParser
source_schema =
PyarrowFieldParser.from_paimon_schema(source.table_schema.fields)
