leaves12138 commented on code in PR #10419:
URL: https://github.com/apache/paimon/pull/10419#discussion_r4214374814
##########
paimon-python/pypaimon/write/native_merge_into.py:
##########
@@ -36,12 +36,35 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
self_merge = _is_self_merge(table, source, target_keys, source_keys)
if not_matched and table.options.video_frame_fields():
return None
- if _is_table_like(source) and not self_merge:
- return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
- source, matched, not_matched, context = _prepare(table, source,
list(matched), list(not_matched), on)
+ source_schema = None
+ native_source = None
+ if _is_table_like(source) and not self_merge:
+ # The source can be an append or PK table; only the target requires DE.
+ from pypaimon.read.native_plan import _native_blob_view_supported
+ from pypaimon.table.file_store_table import FileStoreTable
+ if (type(source) is not FileStoreTable
+ or source.options.file_format() != 'parquet'
+ 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()
+ 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)):
+ return None
+ native_source = create_native_write_table(source)
Review Comment:
[P1] Keep Python-written legacy partition sources off the native path
The target goes through `_native_row_id_table()`, which checks
`_native_update_paths_supported()`, but the source is passed directly to
`create_native_write_table()`. Valid Python-written sources can still store a
string partition such as `part='a/b'` under `part=a/b/bucket-0`. The ordinary
Python planner/reader repairs and prefers that existing path, whereas
`MergeInput::read()` now plans the source entirely in Rust and opens
`part=a%2Fb/bucket-0`.
I reproduced this with an unpartitioned DE target and a Python-written
Parquet append source: `_normalize_source(source)` and Python MERGE both
succeed; `_native_update_paths_supported(source)` returns false, but
`create_native_merge_into()` still selects the native operation. Running the
actual new Rust `MergeSource::Table` against that source fails with
`IoUnexpected` / `NotFound` at the encoded path. Copying the same Parquet file
to the canonical directory makes the same Rust MERGE pass. Since native
failures deliberately do not retry in Python, enabling native target writes
regresses previously valid table-source MERGEs.
Please apply equivalent source-path compatibility gating before selecting
the native operation, or pass repaired source splits into core. Add
batch/stream regressions for a Python-written special-character partition and a
canonical-path control, preserving the existing preference for a legacy path
when both paths exist.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
