leaves12138 commented on code in PR #10419:
URL: https://github.com/apache/paimon/pull/10419#discussion_r4214912067
##########
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:
Verified at 883b915 with a rebuilt native extension. Normalizing new Python
writes to the Java layout addresses the reproduced special-character partition
case: the 24 new source cases pass across append/PK/data-evolution tables,
batch/stream modes, and local/file-URI/configured-directory/external locations.
I am resolving this thread under the explicitly documented
Java-layout-only/no-migration contract. This does not imply that pre-existing
non-Java directories remain readable.
##########
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:
[P2] Check the source's partition types before entering native MERGE
The target is gated by `_native_partition_types_supported()` inside
`_native_row_id_table()`, but this source path calls
`create_native_write_table(source)` without the equivalent check. For a valid
Parquet append source partitioned on a FLOAT or DOUBLE column, the handle is
constructed successfully and core then fails while planning the source:
`NotImplementedError: ... Float/Double ... type is not supported as partition
key`. This happens even when ON and assignments only reference ordinary INT
columns, because core still needs to reconstruct source partition paths.
I reproduced this against the rebuilt extension at this exact head with an
unpartitioned DE target `(id INT, value INT)` and a Python-written source `(id
INT, value INT, part FLOAT/DOUBLE)` partitioned by `part`, using 1.5 and 2.25
as partition values. The same update-and-insert MERGE passes in both batch and
stream mode with target native writes disabled (4 controls), and fails in both
modes with them enabled (4 cases). These are newly written Java-layout files,
not the intentionally unsupported legacy directories.
Please apply the source partition capability gate before selecting the
native operation and return None for unsupported source layouts, so the
existing Python implementation is chosen before native execution. Add
FLOAT/DOUBLE batch/stream source regressions. Do not catch/retry the failure
after `prepare_commit()` has entered core.
--
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]
