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 062dc90b70 [python] Keep public APIs independent of execution backends
(#10779)
062dc90b70 is described below
commit 062dc90b70722dec990fb9a86e1cfe750c2afc8a
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 23 21:43:36 2026 +0800
[python] Keep public APIs independent of execution backends (#10779)
---
docs/docs/pypaimon/data-evolution.md | 6 +
paimon-python/dev/test_packaging.py | 4 +-
paimon-python/pypaimon/catalog/table_query_auth.py | 6 +-
paimon-python/pypaimon/common/transform_schema.py | 244 +++++++++++++++++++++
.../execution/native/write/native_update.py | 8 +-
.../execution/python/casting/data_type_casts.py | 53 -----
.../python/globalindex/create_global_index.py | 2 +-
.../python/globalindex/ray_index_build.py} | 59 +----
.../execution/python/read/predicate_json_parser.py | 79 +------
.../pypaimon/execution/python/read/table_read.py | 3 +-
.../pypaimon/execution/python/write/row_id_plan.py | 22 ++
.../execution/python/write/table_delete.py | 7 +
.../execution/python/write/table_update.py | 2 +-
.../pypaimon/filesystem/format_dispatch.py | 36 +++
.../pypaimon/filesystem/hdfs_native_file_io.py | 12 +-
paimon-python/pypaimon/filesystem/local_file_io.py | 12 +-
.../pypaimon/filesystem/pyarrow_file_io.py | 12 +-
paimon-python/pypaimon/globalindex/__init__.py | 27 ---
.../pypaimon/globalindex/index_dispatch.py | 26 +++
.../pypaimon/multimodal/lerobot/reader.py | 4 +-
paimon-python/pypaimon/multimodal/temporal.py | 29 +--
.../pypaimon/ray/data_evolution_merge_join.py | 32 +--
paimon-python/pypaimon/ray/merge_condition.py | 4 +-
paimon-python/pypaimon/ray/vector_index_build.py | 49 +----
paimon-python/pypaimon/read/read_dispatch.py | 32 +++
paimon-python/pypaimon/schema/data_type_casts.py | 53 ++++-
paimon-python/pypaimon/table/file_store_table.py | 26 +--
.../table/source/batch_vector_search_builder.py | 6 +-
.../table/source/full_text_search_builder.py | 12 +-
.../pypaimon/table/source/search_dispatch.py | 24 +-
.../pypaimon/table/source/vector_search_builder.py | 12 +-
.../tests/global_index_execution_boundary_test.py | 49 +++--
.../pypaimon/tests/lerobot_native_reader_test.py | 9 +-
.../tests/multimodal_temporal_boundary_test.py | 122 +++++++++++
.../tests/native_historical_update_test.py | 6 +-
.../pypaimon/tests/native_ray_row_id_plan_test.py | 55 +++++
paimon-python/pypaimon/tests/native_update_test.py | 9 +-
.../tests/public_execution_boundary_test.py | 113 ++++++++++
.../public_execution_factory_boundary_test.py | 209 ++++++++++++++++++
.../pypaimon/tests/schema_cast_support_test.py | 115 ++++++++++
.../tests/write/row_id_delete_input_test.py | 175 +++++++++++++++
.../pypaimon/utils/file_store_path_factory.py | 4 +-
paimon-python/pypaimon/write/backend_dispatch.py | 76 +++++++
paimon-python/pypaimon/write/row_id_conflict.py | 24 +-
paimon-python/pypaimon/write/row_id_plan.py | 5 +-
paimon-python/pypaimon/write/row_input.py | 26 ++-
paimon-python/pypaimon/write/table_delete.py | 8 +-
paimon-python/pypaimon/write/table_update.py | 11 +-
paimon-python/pypaimon/write/update_dispatch.py | 8 +-
49 files changed, 1459 insertions(+), 468 deletions(-)
diff --git a/docs/docs/pypaimon/data-evolution.md
b/docs/docs/pypaimon/data-evolution.md
index 83b26b5f09..faa94834d3 100644
--- a/docs/docs/pypaimon/data-evolution.md
+++ b/docs/docs/pypaimon/data-evolution.md
@@ -236,6 +236,12 @@ If you already have `_ROW_ID` values, use
`delete_by_row_id` to write deletion
vectors directly. This follow-up deletes the first row from the fresh table
above; use row IDs obtained from your target table in an existing dataset.
+The same method accepts an integer PyArrow `Array` or `ChunkedArray`, so a
+projected `_ROW_ID` column can be submitted without converting it to a Python
+list. NULL IDs and unsigned values outside the signed 64-bit range are
rejected;
+duplicate IDs delete the row once. Native execution retains Arrow buffers and
+chunk boundaries through the Rust binding.
+
```python
write_builder = table.new_batch_write_builder()
table_update = write_builder.new_update()
diff --git a/paimon-python/dev/test_packaging.py
b/paimon-python/dev/test_packaging.py
index 02fe0efdca..ddf4877ec2 100644
--- a/paimon-python/dev/test_packaging.py
+++ b/paimon-python/dev/test_packaging.py
@@ -121,6 +121,8 @@ class PackagingTest(unittest.TestCase):
self.assertFalse(any(name.endswith((".so", ".pyd", ".jar")) for name
in names))
self.assertIn("pypaimon/_version.py", names)
execution_files = (
+ "pypaimon/common/transform_schema.py",
+ "pypaimon/filesystem/format_dispatch.py",
"pypaimon/table/source/search_execution.py",
"pypaimon/ray/search_execution.py",
"pypaimon/execution/native/read/search_execution.py",
@@ -128,6 +130,7 @@ class PackagingTest(unittest.TestCase):
"pypaimon/execution/python/read/search/distributed_vector_search.py",
"pypaimon/execution/python/read/search/distributed_full_text_search.py",
"pypaimon/execution/python/map_blob_key_serializer.py",
+ "pypaimon/execution/python/globalindex/ray_index_build.py",
"pypaimon/execution/python/write/external_path_provider.py",
"pypaimon/execution/python/read/predicate_json_parser.py",
"pypaimon/execution/python/read/blob_view_lookup.py",
@@ -143,7 +146,6 @@ class PackagingTest(unittest.TestCase):
"pypaimon/utils/temporal_string.py",
"pypaimon/utils/row_to_string.py",
"pypaimon/execution/python/casting/__init__.py",
- "pypaimon/execution/python/casting/data_type_casts.py",
"pypaimon/execution/python/casting/schema_evolution.py",
"pypaimon/execution/python/casting/decimal_cast.py",
"pypaimon/execution/python/casting/temporal_cast.py",
diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py
b/paimon-python/pypaimon/catalog/table_query_auth.py
index 091c1f53a4..4a486cb6d0 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/paimon-python/pypaimon/catalog/table_query_auth.py
@@ -136,12 +136,12 @@ class TableQueryAuthResult:
def extract_row_filter(self) -> Optional[Callable[[pa.RecordBatch],
pa.Array]]:
if not self.filter:
return None
- from pypaimon.execution.python.read.predicate_json_parser import
parse_predicate_to_batch_filter
+ from pypaimon.read.read_dispatch import create_python_row_filter
if len(self.filter) == 1:
- return parse_predicate_to_batch_filter(self.filter[0])
+ return create_python_row_filter(self.filter[0])
# Java PredicateBuilder.and evaluates entries in their original order.
- return parse_predicate_to_batch_filter(json.dumps({
+ return create_python_row_filter(json.dumps({
'kind': 'COMPOUND', 'function': 'AND',
'children': [json.loads(rule) for rule in self.filter],
}))
diff --git a/paimon-python/pypaimon/common/transform_schema.py
b/paimon-python/pypaimon/common/transform_schema.py
new file mode 100644
index 0000000000..a8a2090da2
--- /dev/null
+++ b/paimon-python/pypaimon/common/transform_schema.py
@@ -0,0 +1,244 @@
+# 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.
+
+"""Infer authorization output schemas from rule metadata without reading
rows."""
+
+import re
+
+import pyarrow as pa
+
+from pypaimon.common.predicate_json_parser import _DATE_EXTRACT_NAMES
+
+
+_INT_MAX = (1 << 31) - 1
+_CHARACTER_TYPE = re.compile(
+ r"\s*(?:STRING(?=\s|\Z)|(?:CHAR|VARCHAR)\s*(?:\(\s*(\d+)\s*\)|(?=\s|\Z)))"
+ r"(?:\s*(?:NOT\s+)?NULL)?\s*\Z", re.IGNORECASE)
+_INTEGER_TYPE = re.compile(
+ r"\s*(?:TINYINT|SMALLINT|INT|INTEGER|BIGINT)(?=\s|\Z)"
+ r"(?:\s*(?:NOT\s+)?NULL)?\s*\Z", re.IGNORECASE)
+
+
+def effective_masked_schema(schema, masking):
+ if not masking:
+ return schema
+ fields = []
+ for field in schema:
+ transform = masking.get(field.name)
+ if transform is None:
+ fields.append(field)
+ else:
+ data_type = _transform_type(transform, schema, field.type)
+ fields.append(pa.field(field.name, data_type, nullable=True,
metadata=field.metadata))
+ return pa.schema(fields, metadata=schema.metadata)
+
+
+def _field_type(reference, schema):
+ name = reference.get('name') if isinstance(reference, dict) else None
+ if not isinstance(name, str) or name not in schema.names:
+ raise ValueError('Column masking refers to a field that is not
present: {!r}'.format(reference))
+ return schema.field(name).type
+
+
+def _string_type(value, schema, stored=False, coerce_literals=False):
+ if value is None or isinstance(value, str):
+ return pa.string()
+ if not isinstance(value, dict):
+ if coerce_literals:
+ return pa.string()
+ raise ValueError('Transform input must be a string or a field:
{!r}'.format(value))
+ if stored:
+ spelling = value.get('type')
+ match = _CHARACTER_TYPE.fullmatch(spelling) if isinstance(spelling,
str) else None
+ if match is None or (match.group(1) is not None and not 1 <=
int(match.group(1)) <= _INT_MAX):
+ raise ValueError('Transform field must be a string:
{!r}'.format(spelling))
+ data_type = _field_type(value, schema)
+ if not (pa.types.is_string(data_type) or
pa.types.is_large_string(data_type)):
+ raise ValueError('Transform field must be a string:
{}'.format(data_type))
+ return data_type
+
+
+def _integer_input(value, schema, stored=True):
+ if value is None or (isinstance(value, int) and not isinstance(value,
bool)):
+ return
+ if not isinstance(value, dict):
+ raise ValueError('Transform position must be an integer or a field:
{!r}'.format(value))
+ spelling = value.get('type')
+ if stored and (not isinstance(spelling, str) or
_INTEGER_TYPE.fullmatch(spelling) is None):
+ raise ValueError('Transform position field must be an integer:
{!r}'.format(spelling))
+ if not pa.types.is_integer(_field_type(value, schema)):
+ raise ValueError('Transform position field must be an integer')
+
+
+def _inputs(transform, minimum, maximum):
+ values = transform.get('inputs')
+ if not isinstance(values, list) or not minimum <= len(values) <= maximum:
+ raise ValueError('{} takes {} to {} inputs'.format(transform['name'],
minimum, maximum))
+ return values
+
+
+def _transform_type(transform, schema, null_type):
+ name = transform['name']
+ if name == 'NULL':
+ return null_type
+ if name == 'FIELD_REF':
+ return _field_type(transform['fieldRef'], schema)
+ if name == 'CAST':
+ _field_type(transform['fieldRef'], schema)
+ _paimon_type_to_arrow(transform['fieldRef']['type'])
+ target = transform['type']
+ width = re.fullmatch(r'\s*(CHAR|VARCHAR|BINARY|VARBINARY)(?=\s|\(|\Z)'
+ r'(?:\s*\(\s*([0-9]+)\s*\))?'
+ r'(?:\s*(?:NOT\s+)?NULL)?\s*', target,
flags=re.IGNORECASE)
+ if width is not None:
+ if not 1 <= int(width.group(2) or 1) <= _INT_MAX:
+ raise ValueError('Cast width must be between 1 and 2147483647')
+ return pa.string() if width.group(1).upper() in ('CHAR',
'VARCHAR') else pa.binary()
+ return _paimon_type_to_arrow(target)
+ if name in _DATE_EXTRACT_NAMES:
+ reference = transform['fieldRef']
+ stored = reference.get('type')
+ match =
re.fullmatch(r'\s*(DATE(?=\s|\Z)|TIMESTAMP(?=\s|\(|\Z)(?:\s*\(\s*([0-9]+)\s*\))?'
+ r'(?:\s*WITHOUT\s+TIME\s+ZONE(?=\s|\Z))?)'
+ r'(?:\s*(?:NOT\s+)?NULL)?\s*', stored,
flags=re.IGNORECASE) \
+ if isinstance(stored, str) else None
+ actual = _field_type(reference, schema)
+ if (match is None or (match.group(2) is not None and
int(match.group(2)) > 9)
+ or not (pa.types.is_date32(actual) or
(pa.types.is_timestamp(actual) and actual.tz is None))):
+ raise ValueError('Calendar transforms require DATE or TIMESTAMP
without time zone')
+ return pa.int32()
+ if name in ('UPPER', 'LOWER', 'LENGTH', 'BIT_LENGTH'):
+ values = _inputs(transform, 1, 1)
+ result = _string_type(values[0], schema, stored=name in ('LENGTH',
'BIT_LENGTH'),
+ coerce_literals=name in ('UPPER', 'LOWER'))
+ return pa.int32() if name in ('LENGTH', 'BIT_LENGTH') else result
+ if name in ('CONCAT', 'CONCAT_WS'):
+ values = _inputs(transform, 0 if name == 'CONCAT' else 1, _INT_MAX)
+ types = [_string_type(value, schema, coerce_literals=True) for value
in values]
+ # Python Arrow's CONCAT also supplies a string separator. A
large-string
+ # argument cannot bind that mixed-width kernel, even for an empty
batch.
+ if name == 'CONCAT' and any(pa.types.is_large_string(data_type) for
data_type in types):
+ raise TypeError('CONCAT inputs must use the same string width as
its separator')
+ return pa.string()
+ if name == 'SUBSTRING':
+ values = _inputs(transform, 2, 3)
+ result = _string_type(values[0], schema)
+ for value in values[1:]:
+ if isinstance(value, str):
+ continue
+ _integer_input(value, schema, stored=False)
+ return result
+ if name == 'TRIM':
+ values = _inputs(transform, 1, 2)
+ if transform.get('trimFlag') not in ('BOTH', 'LEADING', 'TRAILING'):
+ raise ValueError('Unknown trimFlag:
{}'.format(transform.get('trimFlag')))
+ types = [_string_type(value, schema, stored=True) for value in values]
+ return types[0]
+ if name in ('PAD', 'TRANSLATE', 'OVERLAY'):
+ values = _inputs(transform, 3, 4 if name == 'OVERLAY' else 3)
+ _string_type(values[0], schema, stored=True)
+ if name == 'PAD':
+ if transform.get('direction') not in ('LEFT', 'RIGHT'):
+ raise ValueError('PAD direction must be LEFT or RIGHT')
+ _integer_input(values[1], schema)
+ _string_type(values[2], schema, stored=True)
+ elif name == 'TRANSLATE':
+ for value in values[1:]:
+ _string_type(value, schema, stored=True)
+ else:
+ _string_type(values[1], schema, stored=True)
+ for value in values[2:]:
+ _integer_input(value, schema)
+ return pa.string()
+ raise ValueError('Unknown transform type: {}'.format(name))
+
+
+def _paimon_type_to_arrow(paimon_type: str) -> pa.DataType:
+ type_str = paimon_type.strip().upper()
+
+ ltz_match = re.match(
+ r"^TIMESTAMP\s*\((\d+)\)\s+WITH\s+LOCAL\s+TIME\s+ZONE", type_str)
+ if ltz_match:
+ precision = int(ltz_match.group(1))
+ return pa.timestamp(_timestamp_precision_to_unit(precision), tz="UTC")
+
+ m = re.match(r"^([A-Z_ ]+?)(?:\((.+)\))?(?:\s+NOT\s+NULL)?$", type_str)
+ if not m:
+ raise ValueError(f"Cannot parse Paimon type: '{paimon_type}'")
+ base_type = m.group(1).strip()
+ params = m.group(2)
+
+ simple_mapping = {
+ "INT": pa.int32(),
+ "BIGINT": pa.int64(),
+ "SMALLINT": pa.int16(),
+ "TINYINT": pa.int8(),
+ "FLOAT": pa.float32(),
+ "DOUBLE": pa.float64(),
+ "STRING": pa.string(),
+ "BOOLEAN": pa.bool_(),
+ "BYTES": pa.binary(),
+ "DATE": pa.date32(),
+ }
+ if base_type in simple_mapping:
+ return simple_mapping[base_type]
+
+ if base_type in ("VARCHAR", "CHAR"):
+ return pa.string()
+
+ if base_type in ("VARBINARY", "BINARY"):
+ return pa.binary()
+
+ if base_type == "TIMESTAMP":
+ precision = int(params) if params else 6
+ unit = _timestamp_precision_to_unit(precision)
+ return pa.timestamp(unit)
+
+ if base_type in ("TIMESTAMP WITH LOCAL TIME ZONE",
"TIMESTAMP_WITH_LOCAL_TIME_ZONE", "TIMESTAMP_LTZ"):
+ precision = int(params) if params else 6
+ unit = _timestamp_precision_to_unit(precision)
+ return pa.timestamp(unit, tz="UTC")
+
+ if base_type == "TIME":
+ precision = int(params) if params else 6
+ unit = _timestamp_precision_to_unit(precision)
+ return pa.time64(unit) if unit in ("us", "ns") else pa.time32(unit)
+
+ if base_type == "DECIMAL":
+ if params:
+ parts = [x.strip() for x in params.split(",")]
+ if len(parts) == 2:
+ return pa.decimal128(int(parts[0]), int(parts[1]))
+ raise ValueError(f"DECIMAL type requires (precision, scale):
'{paimon_type}'")
+
+ raise ValueError(
+ f"Unsupported Paimon type for PyArrow conversion: '{paimon_type}'. "
+ f"Supported: INT, BIGINT, SMALLINT, TINYINT, FLOAT, DOUBLE, STRING,
VARCHAR, CHAR, "
+ f"BOOLEAN, BYTES, VARBINARY, DATE, TIME(p), TIMESTAMP(p), "
+ f"TIMESTAMP WITH LOCAL TIME ZONE(p), DECIMAL(p,s)."
+ )
+
+
+def _timestamp_precision_to_unit(precision: int) -> str:
+ if precision == 0:
+ return "s"
+ elif precision <= 3:
+ return "ms"
+ elif precision <= 6:
+ return "us"
+ else:
+ return "ns"
diff --git a/paimon-python/pypaimon/execution/native/write/native_update.py
b/paimon-python/pypaimon/execution/native/write/native_update.py
index 7844079a1f..f0455bfa5e 100644
--- a/paimon-python/pypaimon/execution/native/write/native_update.py
+++ b/paimon-python/pypaimon/execution/native/write/native_update.py
@@ -33,7 +33,7 @@ from pypaimon.table.file_store_table import FileStoreTable
from pypaimon.execution.native.write.native_commit import (
create_native_write_table, from_native_commit_messages,
)
-from pypaimon.write.row_input import RowIdUpdateInputMixin
+from pypaimon.write.row_input import RowIdUpdateInputMixin, normalize_row_ids
from pypaimon.write.row_utils import _contains_blob_value, value_for_arrow
@@ -506,11 +506,7 @@ class NativeBatchTableUpdate:
return from_native_commit_messages(self.table, messages)
def delete_by_row_id(self, row_ids):
- ids = []
- for row_id in row_ids:
- if row_id is None:
- raise ValueError('_ROW_ID value must not be null.')
- ids.append(int(row_id))
+ ids = normalize_row_ids(row_ids)
try:
messages = self.writer.delete_by_row_id(ids)
except ValueError as error:
diff --git a/paimon-python/pypaimon/execution/python/casting/data_type_casts.py
b/paimon-python/pypaimon/execution/python/casting/data_type_casts.py
deleted file mode 100644
index ba21cf7db5..0000000000
--- a/paimon-python/pypaimon/execution/python/casting/data_type_casts.py
+++ /dev/null
@@ -1,53 +0,0 @@
-# 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.
-
-"""Probe executable schema-evolution casts using the Python read executor."""
-
-import pyarrow as pa
-
-from pypaimon.schema.data_types import PyarrowFieldParser
-
-
-# Caches the read-time cast probe per (source, target) pyarrow type so the
-# alter-time check stays cheap. Keyed by the pyarrow type strings.
-_EXECUTABLE_CAST_CACHE = {}
-
-
-def schema_evolution_cast_supported(source_type, target_type) -> bool:
- from pypaimon.execution.python.casting.schema_evolution import
cast_array_for_schema_evolution
-
- source_pa = PyarrowFieldParser.from_paimon_type(source_type)
- target_pa = PyarrowFieldParser.from_paimon_type(target_type)
- if source_pa == target_pa:
- return True
- cache_key = (str(source_pa), str(target_pa))
- cached = _EXECUTABLE_CAST_CACHE.get(cache_key)
- if cached is not None:
- return cached
- # Probe a one-row (null-valued) array rather than an empty one. Some Arrow
- # kernels validate target parameters only for a non-empty input. Using the
- # read executor also includes Java's per-value overflow-to-NULL rules, such
- # as integer-to-DECIMAL targets narrower than the source type's full range.
- try:
- cast_array_for_schema_evolution(
- pa.nulls(1, type=source_pa), target_pa, source_type, target_type)
- ok = True
- except (pa.lib.ArrowNotImplementedError, pa.lib.ArrowInvalid,
- pa.lib.ArrowTypeError, ValueError, NotImplementedError):
- ok = False
- _EXECUTABLE_CAST_CACHE[cache_key] = ok
- return ok
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/create_global_index.py
b/paimon-python/pypaimon/execution/python/globalindex/create_global_index.py
index 026b573c38..804b698ce3 100644
--- a/paimon-python/pypaimon/execution/python/globalindex/create_global_index.py
+++ b/paimon-python/pypaimon/execution/python/globalindex/create_global_index.py
@@ -118,7 +118,7 @@ class PythonGlobalIndexBuilder:
splits, unindexed_ranges, index_field, table_read, index_path,
DataEvolutionIndexSourceMeta(snapshot.id).serialize() if
snapshot else None)
if execution == "ray":
- from pypaimon.ray.vector_index_build import build_vector_index
+ from pypaimon.execution.python.globalindex.ray_index_build import
build_vector_index
return build_vector_index(
self, splits, unindexed_ranges, index_field, table_read,
index_path, concurrency, ray_remote_args)
diff --git a/paimon-python/pypaimon/ray/vector_index_build.py
b/paimon-python/pypaimon/execution/python/globalindex/ray_index_build.py
similarity index 53%
copy from paimon-python/pypaimon/ray/vector_index_build.py
copy to paimon-python/pypaimon/execution/python/globalindex/ray_index_build.py
index fa5f722578..982a648133 100644
--- a/paimon-python/pypaimon/ray/vector_index_build.py
+++ b/paimon-python/pypaimon/execution/python/globalindex/ray_index_build.py
@@ -15,24 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Ray shard construction with one driver-side global-index commit."""
-
-from pypaimon.globalindex.vindex_options import VINDEX_IDENTIFIERS
-from pypaimon.ray.search_execution import execution_options
-
-
-def validate_build_options(builder, concurrency, ray_remote_args):
- if builder._index_type not in VINDEX_IDENTIFIERS:
- raise ValueError("Ray index building supports only native vector
indexes.")
- concurrency, options = execution_options(concurrency, ray_remote_args)
- # Retrying a write task can race with a lost worker writing the same file.
- # A failed build can instead be rerun after its uncommitted outputs close.
- if options.get("max_retries", 0) != 0 or options.get("retry_exceptions",
False):
- raise ValueError("Ray index building requires max_retries=0 and
retry_exceptions=False.")
- options.update(max_retries=0, retry_exceptions=False)
- if builder._core_options.global_index_row_count_per_shard() <= 0:
- raise ValueError("Option 'global-index.row-count-per-shard' must be
greater than 0.")
- return concurrency, options
+"""Schedule the selected Python vector-index builder on Ray workers."""
def build_vector_index(builder, splits, unindexed_ranges, index_field,
table_read,
@@ -79,43 +62,3 @@ def _build_shard(context, shard, file_name):
index_split, index_range = shard
return builder._build_generic_shard(
index_split, index_range, index_field, table_read, index_path,
file_name=file_name)
-
-
-def build_native_vector_index(table, plan, concurrency, remote_args):
- """Schedule immutable Core tasks; all publication remains with the
caller."""
- import ray
-
- shards = plan.shards()
- if not shards:
- return []
- context = ray.put(table)
- remote = ray.remote(_build_native_shard).options(**remote_args)
- pending, results = {}, {}
- next_shard = 0
- try:
- while pending or next_shard < len(shards):
- while len(pending) < concurrency and next_shard < len(shards):
- ref = remote.remote(context, shards[next_shard])
- pending[ref] = next_shard
- next_shard += 1
- ready, _ = ray.wait(list(pending), num_returns=1)
- ref = ready[0]
- ordinal = pending[ref]
- results[ordinal] = ray.get(ref)
- del pending[ref]
- return [message for ordinal in range(len(shards)) for message in
results[ordinal]]
- except BaseException:
- # A worker may have completed even when result delivery failed. Wait
- # for dispatched tasks, but retain every output: its commit ownership
- # is unknown and another attempt may already hold the messages.
- for ref in pending:
- try:
- ray.get(ref)
- except BaseException:
- pass
- raise
-
-
-def _build_native_shard(table, shard):
- from pypaimon.execution.native.write.native_index_build import
build_native_index_shard
- return build_native_index_shard(table, shard)
diff --git
a/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
b/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
index cc7f6a6e04..f3bde3c2e9 100644
--- a/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
+++ b/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
@@ -25,6 +25,8 @@ from typing import Callable
import pyarrow as pa
import pyarrow.compute as pc
+from pypaimon.common.transform_schema import _paimon_type_to_arrow
+
# utf8_slice_codeunits needs an explicit integer stop on pyarrow 6
_MAX_STOP = 2 ** 31 - 1
@@ -870,80 +872,3 @@ def _convert_literal(literal, target_type: pa.DataType):
import decimal
return pa.scalar(decimal.Decimal(str(literal)), type=target_type)
return literal
-
-
-def _paimon_type_to_arrow(paimon_type: str) -> pa.DataType:
- type_str = paimon_type.strip().upper()
-
- ltz_match = re.match(
- r"^TIMESTAMP\s*\((\d+)\)\s+WITH\s+LOCAL\s+TIME\s+ZONE", type_str)
- if ltz_match:
- precision = int(ltz_match.group(1))
- return pa.timestamp(_timestamp_precision_to_unit(precision), tz="UTC")
-
- m = re.match(r"^([A-Z_ ]+?)(?:\((.+)\))?(?:\s+NOT\s+NULL)?$", type_str)
- if not m:
- raise ValueError(f"Cannot parse Paimon type: '{paimon_type}'")
- base_type = m.group(1).strip()
- params = m.group(2)
-
- simple_mapping = {
- "INT": pa.int32(),
- "BIGINT": pa.int64(),
- "SMALLINT": pa.int16(),
- "TINYINT": pa.int8(),
- "FLOAT": pa.float32(),
- "DOUBLE": pa.float64(),
- "STRING": pa.string(),
- "BOOLEAN": pa.bool_(),
- "BYTES": pa.binary(),
- "DATE": pa.date32(),
- }
- if base_type in simple_mapping:
- return simple_mapping[base_type]
-
- if base_type in ("VARCHAR", "CHAR"):
- return pa.string()
-
- if base_type in ("VARBINARY", "BINARY"):
- return pa.binary()
-
- if base_type == "TIMESTAMP":
- precision = int(params) if params else 6
- unit = _timestamp_precision_to_unit(precision)
- return pa.timestamp(unit)
-
- if base_type in ("TIMESTAMP WITH LOCAL TIME ZONE",
"TIMESTAMP_WITH_LOCAL_TIME_ZONE", "TIMESTAMP_LTZ"):
- precision = int(params) if params else 6
- unit = _timestamp_precision_to_unit(precision)
- return pa.timestamp(unit, tz="UTC")
-
- if base_type == "TIME":
- precision = int(params) if params else 6
- unit = _timestamp_precision_to_unit(precision)
- return pa.time64(unit) if unit in ("us", "ns") else pa.time32(unit)
-
- if base_type == "DECIMAL":
- if params:
- parts = [x.strip() for x in params.split(",")]
- if len(parts) == 2:
- return pa.decimal128(int(parts[0]), int(parts[1]))
- raise ValueError(f"DECIMAL type requires (precision, scale):
'{paimon_type}'")
-
- raise ValueError(
- f"Unsupported Paimon type for PyArrow conversion: '{paimon_type}'. "
- f"Supported: INT, BIGINT, SMALLINT, TINYINT, FLOAT, DOUBLE, STRING,
VARCHAR, CHAR, "
- f"BOOLEAN, BYTES, VARBINARY, DATE, TIME(p), TIMESTAMP(p), "
- f"TIMESTAMP WITH LOCAL TIME ZONE(p), DECIMAL(p,s)."
- )
-
-
-def _timestamp_precision_to_unit(precision: int) -> str:
- if precision == 0:
- return "s"
- elif precision <= 3:
- return "ms"
- elif precision <= 6:
- return "us"
- else:
- return "ns"
diff --git a/paimon-python/pypaimon/execution/python/read/table_read.py
b/paimon-python/pypaimon/execution/python/read/table_read.py
index 858527a735..0127249e44 100644
--- a/paimon-python/pypaimon/execution/python/read/table_read.py
+++ b/paimon-python/pypaimon/execution/python/read/table_read.py
@@ -440,7 +440,8 @@ class PythonTableRead:
defer_blob_resolution,
)
sr._blob_parallelism = blob_parallelism
- sr._parquet_row_group_cache = read._parquet_row_group_cache
+ from pypaimon.read.read_dispatch import python_row_group_cache
+ sr._parquet_row_group_cache =
python_row_group_cache(read._parquet_row_group_cache)
return sr
def _build_split_read(self, split: Split, read_type=None,
diff --git a/paimon-python/pypaimon/execution/python/write/row_id_plan.py
b/paimon-python/pypaimon/execution/python/write/row_id_plan.py
index c364187a94..403afaece3 100644
--- a/paimon-python/pypaimon/execution/python/write/row_id_plan.py
+++ b/paimon-python/pypaimon/execution/python/write/row_id_plan.py
@@ -74,6 +74,28 @@ def row_id_groups(plan):
return sorted(groups, key=lambda group: group[0])
+def validate_live_row_ids(table, info, first_row_id, row_ids):
+ """Reject deleted updates using only the routed snapshot's DV files.
+
+ The writer must still merge physical rows, including tombstones, to
preserve
+ row-id offsets. Filtering the writer's original data would shift those
offsets.
+ """
+ from pypaimon.execution.python.deletionvectors.deletion_vector import
DeletionVector
+
+ split, files = info.first_row_id_index[first_row_id]
+ if split.data_deletion_files is None:
+ return
+ row_ids = row_ids.to_pylist()
+ target_names = {file.file_name for file in files}
+ for file, deletion in zip(split.files, split.data_deletion_files):
+ if deletion is None or file.file_name not in target_names:
+ continue
+ vector = DeletionVector.read(table.file_io, deletion)
+ for row_id in row_ids:
+ if file.row_id_range().contains(row_id) and
vector.is_deleted(row_id - file.first_row_id):
+ raise ValueError("Cannot update deleted _ROW_ID
{}.".format(row_id))
+
+
class PythonRowIdPlan:
"""Route captured metadata with no file access or mutation state."""
diff --git a/paimon-python/pypaimon/execution/python/write/table_delete.py
b/paimon-python/pypaimon/execution/python/write/table_delete.py
index 1db2ad4276..40ba071e80 100644
--- a/paimon-python/pypaimon/execution/python/write/table_delete.py
+++ b/paimon-python/pypaimon/execution/python/write/table_delete.py
@@ -20,6 +20,8 @@ from bisect import bisect_right
from dataclasses import dataclass
from typing import Dict, List, Optional, Sequence, Set, Tuple
+import pyarrow as pa
+
from pypaimon.execution.python.deletionvectors.bitmap_deletion_vector import
BitmapDeletionVector
from pypaimon.execution.python.deletionvectors.deletion_vector import
DeletionVector
from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
@@ -114,6 +116,11 @@ class PythonTableDeleteByRowId:
@staticmethod
def _deduplicate_row_ids(row_ids: Sequence[int]) -> List[int]:
+ if isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
+ from pypaimon.write.row_input import normalize_row_ids
+ # Python's bitmap writer needs Python integers. Native retains
+ # the Arrow buffers through the SDK and never enters this path.
+ row_ids = normalize_row_ids(row_ids).to_pylist()
result = []
seen = set()
for row_id in row_ids:
diff --git a/paimon-python/pypaimon/execution/python/write/table_update.py
b/paimon-python/pypaimon/execution/python/write/table_update.py
index ce54a83348..b881983128 100644
--- a/paimon-python/pypaimon/execution/python/write/table_update.py
+++ b/paimon-python/pypaimon/execution/python/write/table_update.py
@@ -487,7 +487,7 @@ class PythonTableUpdate:
def _delete_rows_with_python(self, row_ids):
context = self._context
- return
TableDeleteByRowId._from_python(context.table).delete(list(row_ids))
+ return TableDeleteByRowId._from_python(context.table).delete(row_ids)
class _ShardBatchIterator:
diff --git a/paimon-python/pypaimon/filesystem/format_dispatch.py
b/paimon-python/pypaimon/filesystem/format_dispatch.py
new file mode 100644
index 0000000000..47111de1fd
--- /dev/null
+++ b/paimon-python/pypaimon/filesystem/format_dispatch.py
@@ -0,0 +1,36 @@
+# 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.
+
+"""Construct format helpers for explicitly selected Python FileIO methods.
+
+Native readers and writers use Core FileIO and never enter these adapters.
+"""
+
+
+def new_row_writer(output_stream, fields, zstd_level=1):
+ from pypaimon.execution.python.write.writer.format_row_writer import
FormatRowWriter
+ return FormatRowWriter(output_stream, fields, zstd_level=zstd_level)
+
+
+def new_blob_writer(output_stream):
+ from pypaimon.execution.python.write.blob_format_writer import
BlobFormatWriter
+ return BlobFormatWriter(output_stream)
+
+
+def vortex_location(file_io, path):
+ from pypaimon.execution.python.read.reader.vortex_utils import
to_vortex_specified
+ return to_vortex_specified(file_io, path)
diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
index 9bbd58dbc4..43db6ef266 100644
--- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
+++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
@@ -619,13 +619,13 @@ class HdfsNativeFileIO(FileIO):
def write_row(self, path: str, data: pyarrow.Table, fields=None,
zstd_level: int = 1, **kwargs):
try:
- from pypaimon.execution.python.write.writer.format_row_writer
import FormatRowWriter
+ from pypaimon.filesystem.format_dispatch import new_row_writer
if fields is None:
fields = PyarrowFieldParser.to_paimon_schema(data.schema)
with self.new_output_stream(path) as output_stream:
- writer = FormatRowWriter(output_stream, fields,
zstd_level=zstd_level)
+ writer = new_row_writer(output_stream, fields,
zstd_level=zstd_level)
writer.write_table(data)
writer.close()
except Exception as e:
@@ -633,7 +633,7 @@ class HdfsNativeFileIO(FileIO):
raise RuntimeError(f"Failed to write row file {path}: {e}") from e
def write_blob(self, path: str, data: pyarrow.Table, **kwargs):
- from pypaimon.execution.python.write.blob_format_writer import
BlobFormatWriter
+ from pypaimon.filesystem.format_dispatch import new_blob_writer
try:
if data.num_columns != 1:
@@ -651,7 +651,7 @@ class HdfsNativeFileIO(FileIO):
num_rows = data.num_rows
field_name = fields[0].name
with self.new_output_stream(path) as output_stream:
- writer = BlobFormatWriter(output_stream)
+ writer = new_blob_writer(output_stream)
for i in range(num_rows):
writer.write_value(records_dict[field_name][i],
fields, self.uri_reader_factory)
@@ -674,8 +674,8 @@ class HdfsNativeFileIO(FileIO):
import vortex
from vortex import store
- from pypaimon.execution.python.read.reader.vortex_utils import
(to_vortex_specified)
- file_path_for_vortex, store_kwargs = to_vortex_specified(self,
path)
+ from pypaimon.filesystem.format_dispatch import vortex_location
+ file_path_for_vortex, store_kwargs = vortex_location(self, path)
if store_kwargs:
vortex_store = store.from_url(file_path_for_vortex,
**store_kwargs)
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py
b/paimon-python/pypaimon/filesystem/local_file_io.py
index 7512ad0a3b..eb2c902950 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -427,8 +427,8 @@ class LocalFileIO(FileIO):
from vortex._lib.io import write as vortex_write
os.makedirs(os.path.dirname(path), exist_ok=True)
- from pypaimon.execution.python.read.reader.vortex_utils import
(to_vortex_specified)
- _, store_kwargs = to_vortex_specified(self, path)
+ from pypaimon.filesystem.format_dispatch import vortex_location
+ _, store_kwargs = vortex_location(self, path)
if store_kwargs:
from vortex import store
@@ -442,7 +442,7 @@ class LocalFileIO(FileIO):
def write_row(self, path: str, data: pyarrow.Table, fields=None,
zstd_level: int = 1, **kwargs):
try:
- from pypaimon.execution.python.write.writer.format_row_writer
import FormatRowWriter
+ from pypaimon.filesystem.format_dispatch import new_row_writer
from pypaimon.schema.data_types import PyarrowFieldParser
if fields is None:
@@ -454,7 +454,7 @@ class LocalFileIO(FileIO):
parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'wb') as output_stream:
- writer = FormatRowWriter(output_stream, fields,
zstd_level=zstd_level)
+ writer = new_row_writer(output_stream, fields,
zstd_level=zstd_level)
writer.write_table(data)
writer.close()
except Exception as e:
@@ -462,7 +462,7 @@ class LocalFileIO(FileIO):
raise RuntimeError(f"Failed to write row file {path}: {e}") from e
def write_blob(self, path: str, data: pyarrow.Table, **kwargs):
- from pypaimon.execution.python.write.blob_format_writer import
BlobFormatWriter
+ from pypaimon.filesystem.format_dispatch import new_blob_writer
try:
if data.num_columns != 1:
@@ -485,7 +485,7 @@ class LocalFileIO(FileIO):
parent.mkdir(parents=True, exist_ok=True)
with open(file_path, 'wb') as output_stream:
- writer = BlobFormatWriter(output_stream)
+ writer = new_blob_writer(output_stream)
for i in range(num_rows):
writer.write_value(records_dict[field_name][i], fields,
self.uri_reader_factory)
writer.close()
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 252d86841d..259281063b 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -1003,8 +1003,8 @@ class PyArrowFileIO(FileIO):
import vortex
from vortex import store
- from pypaimon.execution.python.read.reader.vortex_utils import
(to_vortex_specified)
- file_path_for_vortex, store_kwargs = to_vortex_specified(self,
path)
+ from pypaimon.filesystem.format_dispatch import vortex_location
+ file_path_for_vortex, store_kwargs = vortex_location(self, path)
if store_kwargs:
vortex_store = store.from_url(file_path_for_vortex,
**store_kwargs)
@@ -1018,13 +1018,13 @@ class PyArrowFileIO(FileIO):
def write_row(self, path: str, data: pyarrow.Table, fields=None,
zstd_level: int = 1, **kwargs):
try:
- from pypaimon.execution.python.write.writer.format_row_writer
import FormatRowWriter
+ from pypaimon.filesystem.format_dispatch import new_row_writer
if fields is None:
fields = PyarrowFieldParser.to_paimon_schema(data.schema)
with self.new_output_stream(path) as output_stream:
- writer = FormatRowWriter(output_stream, fields,
zstd_level=zstd_level)
+ writer = new_row_writer(output_stream, fields,
zstd_level=zstd_level)
writer.write_table(data)
writer.close()
except Exception as e:
@@ -1032,7 +1032,7 @@ class PyArrowFileIO(FileIO):
raise RuntimeError(f"Failed to write row file {path}: {e}") from e
def write_blob(self, path: str, data: pyarrow.Table, **kwargs):
- from pypaimon.execution.python.write.blob_format_writer import
BlobFormatWriter
+ from pypaimon.filesystem.format_dispatch import new_blob_writer
try:
if data.num_columns != 1:
@@ -1047,7 +1047,7 @@ class PyArrowFileIO(FileIO):
num_rows = data.num_rows
field_name = fields[0].name
with self.new_output_stream(path) as output_stream:
- writer = BlobFormatWriter(output_stream)
+ writer = new_blob_writer(output_stream)
for i in range(num_rows):
writer.write_value(records_dict[field_name][i], fields,
self.uri_reader_factory)
writer.close()
diff --git a/paimon-python/pypaimon/globalindex/__init__.py
b/paimon-python/pypaimon/globalindex/__init__.py
index 880b389e33..4840b9269b 100644
--- a/paimon-python/pypaimon/globalindex/__init__.py
+++ b/paimon-python/pypaimon/globalindex/__init__.py
@@ -30,15 +30,6 @@ if sys.version_info[:2] < (3, 7):
ScoreGetter,
)
from pypaimon.globalindex.global_index_meta import GlobalIndexMeta,
GlobalIndexIOMeta
- from pypaimon.execution.python.globalindex.global_index_evaluator import
GlobalIndexEvaluator
- from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
- DataEvolutionGlobalIndexScanner,
- )
- from pypaimon.execution.python.globalindex.key_serializer import
KeySerializer
- from pypaimon.execution.python.globalindex.memory_slice_input import
MemorySliceInput
- from pypaimon.execution.python.globalindex.offset_global_index_reader
import OffsetGlobalIndexReader
- from pypaimon.execution.python.globalindex.sorted_file_global_index_reader
import SortedFileGlobalIndexReader
- from pypaimon.execution.python.globalindex.sorted_file_meta_selector
import SortedFileMetaSelector
from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
from pypaimon.globalindex.create_global_index import (
GlobalIndexBuilder,
@@ -61,13 +52,6 @@ __all__ = [
'ScoreGetter',
'GlobalIndexMeta',
'GlobalIndexIOMeta',
- 'GlobalIndexEvaluator',
- 'DataEvolutionGlobalIndexScanner',
- 'KeySerializer',
- 'MemorySliceInput',
- 'OffsetGlobalIndexReader',
- 'SortedFileGlobalIndexReader',
- 'SortedFileMetaSelector',
'SortedIndexFileMeta',
'GlobalIndexBuilder',
'create_global_index',
@@ -87,17 +71,6 @@ _MODULE_BY_EXPORT = {
'ScoreGetter': 'pypaimon.globalindex.vector_search_result',
'GlobalIndexMeta': 'pypaimon.globalindex.global_index_meta',
'GlobalIndexIOMeta': 'pypaimon.globalindex.global_index_meta',
- 'GlobalIndexEvaluator':
'pypaimon.execution.python.globalindex.global_index_evaluator',
- 'DataEvolutionGlobalIndexScanner':
-
'pypaimon.execution.python.globalindex.data_evolution_global_index_scanner',
- 'KeySerializer': 'pypaimon.execution.python.globalindex.key_serializer',
- 'MemorySliceInput':
'pypaimon.execution.python.globalindex.memory_slice_input',
- 'OffsetGlobalIndexReader':
- 'pypaimon.execution.python.globalindex.offset_global_index_reader',
- 'SortedFileGlobalIndexReader':
-
'pypaimon.execution.python.globalindex.sorted_file_global_index_reader',
- 'SortedFileMetaSelector':
- 'pypaimon.execution.python.globalindex.sorted_file_meta_selector',
'SortedIndexFileMeta': 'pypaimon.globalindex.sorted_index_file_meta',
'GlobalIndexBuilder': 'pypaimon.globalindex.create_global_index',
'create_global_index': 'pypaimon.globalindex.create_global_index',
diff --git a/paimon-python/pypaimon/globalindex/index_dispatch.py
b/paimon-python/pypaimon/globalindex/index_dispatch.py
index 39c18df03b..821a238bea 100644
--- a/paimon-python/pypaimon/globalindex/index_dispatch.py
+++ b/paimon-python/pypaimon/globalindex/index_dispatch.py
@@ -104,3 +104,29 @@ def build_global_index_drop(request, partition_filter):
return messages
from pypaimon.execution.python.globalindex.drop_global_index import
build_global_index_drop as execute
return execute(request, partition_filter)
+
+
+def build_python_vector_index(builder, splits, unindexed_ranges, index_field,
table_read,
+ index_path, concurrency, remote_args):
+ from pypaimon.execution.python.globalindex.ray_index_build import
build_vector_index
+ return build_vector_index(
+ builder, splits, unindexed_ranges, index_field, table_read,
+ index_path, concurrency, remote_args)
+
+
+def build_python_vector_index_shard(context, shard, file_name):
+ from pypaimon.execution.python.globalindex.ray_index_build import
_build_shard
+ return _build_shard(context, shard, file_name)
+
+
+def build_native_vector_index_shard(table, shard):
+ from pypaimon.execution.native.write.native_index_build import
build_native_index_shard
+ return build_native_index_shard(table, shard)
+
+
+def create_python_scalar_index_scanner(table, *, predicate, snapshot):
+ """Open the scalar scanner after Python lookup execution was selected."""
+ from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ DataEvolutionGlobalIndexScanner,
+ )
+ return DataEvolutionGlobalIndexScanner.create(table, predicate=predicate,
snapshot=snapshot)
diff --git a/paimon-python/pypaimon/multimodal/lerobot/reader.py
b/paimon-python/pypaimon/multimodal/lerobot/reader.py
index f85269095d..ee636b2aeb 100644
--- a/paimon-python/pypaimon/multimodal/lerobot/reader.py
+++ b/paimon-python/pypaimon/multimodal/lerobot/reader.py
@@ -154,8 +154,8 @@ class _FrameLocator:
def _index_scanner(self, predicate):
if not self._scanner_initialized:
- from pypaimon.globalindex import DataEvolutionGlobalIndexScanner
- self._scanner = DataEvolutionGlobalIndexScanner.create(
+ from pypaimon.globalindex.index_dispatch import
create_python_scalar_index_scanner
+ self._scanner = create_python_scalar_index_scanner(
self._table,
predicate=predicate,
snapshot=self._snapshot,
diff --git a/paimon-python/pypaimon/multimodal/temporal.py
b/paimon-python/pypaimon/multimodal/temporal.py
index 3852ee46e9..ddd84e124d 100644
--- a/paimon-python/pypaimon/multimodal/temporal.py
+++ b/paimon-python/pypaimon/multimodal/temporal.py
@@ -29,16 +29,15 @@ import pyarrow.compute as pc
from pypaimon.catalog.table_query_auth import TableQueryAuthResult
from pypaimon.common.options.core_options import CoreOptions, StartupMode
from pypaimon.common.predicate_json_parser import
_collect_all_field_refs_from_transform
-from pypaimon.execution.python.read.predicate_json_parser import
_apply_predicate_transform
+from pypaimon.common.transform_schema import effective_masked_schema as
_effective_masked_schema
from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.multimodal.query import ScanQuery
from pypaimon.read.query_auth_split import QueryAuthSplit, resolve_auth_result
-from pypaimon.execution.python.read.reader.format_pyarrow_reader import
(_DecodedRowGroupCache)
+from pypaimon.read.read_dispatch import _ParquetRowGroupCacheRequest
from pypaimon.read.read_support import _ClosableArrowBatchReader
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.snapshot.time_travel_util import TimeTravelUtil
from pypaimon.table.special_fields import SpecialFields
-from pypaimon.execution.python.read.search.global_index_live_row_filter import
table_at_snapshot
from pypaimon.utils.range import Range
@@ -202,7 +201,7 @@ class TemporalAlignment:
anchor_metadata = _metadata_batches(
self._anchor, self._on, self._by, batch_size)
- row_group_cache = _DecodedRowGroupCache(
+ row_group_cache = _ParquetRowGroupCacheRequest(
_TEMPORAL_ROW_GROUP_CACHE_MAX_SIZE)
anchor_fetcher = _RowIdFetcher(self._anchor, row_group_cache)
source_fetchers = []
@@ -964,7 +963,7 @@ def _pin_scan_to_snapshot(query):
else None
)
if snapshot is not None and tag_name is None:
- table = table_at_snapshot(table, snapshot)
+ table = table._copy_with_snapshot(snapshot)
pinned = ScanQuery(table)
pinned._predicate = query._predicate
pinned._projection = query._projection
@@ -1459,26 +1458,6 @@ def _mask_dependencies(masking, targets, table_schema):
]
-def _effective_masked_schema(schema, masking):
- if not masking:
- return schema
- batch = pa.RecordBatch.from_arrays([
- pa.array([], type=field.type) for field in schema
- ], schema=schema)
- fields = []
- for field in schema:
- transform = masking.get(field.name)
- if transform is None:
- fields.append(field)
- continue
- masked = _apply_predicate_transform(
- transform, batch, null_type=field.type)
- fields.append(pa.field(
- field.name, masked.type, nullable=True,
- metadata=field.metadata))
- return pa.schema(fields, metadata=schema.metadata)
-
-
def _project_effective_schema(
schema, name_paths, effective_schema, masking):
if not masking:
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 5544353b96..543ab968ed 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -716,10 +716,9 @@ def distributed_update_apply(
)
update = captured_table.new_batch_write_builder().new_update()
worker = update.new_update_by_row_id(_precomputed_files_info=info)
- from pypaimon.write.backend_dispatch import
native_row_id_update_selected
- if not native_row_id_update_selected(worker):
- _validate_live_row_ids(
- captured_table, info, int(group[frid_col][0].as_py()),
for_update[row_id_name].to_pylist())
+ from pypaimon.write.backend_dispatch import validate_row_id_update_rows
+ validate_row_id_update_rows(
+ worker, info, int(group[frid_col][0].as_py()),
for_update[row_id_name])
msgs = worker.update_columns(for_update, list(captured_cols))
return pa.Table.from_pydict({
"msgs_blob": [pickle.dumps(msgs)],
@@ -900,7 +899,7 @@ def distributed_delete_apply(
"merging."
)
- row_ids = group.column(row_id_name).to_pylist()
+ row_ids = group.column(row_id_name)
worker = TableDeleteByRowId(
captured_table,
_precomputed_anchor_ranges=ray.get(precomputed_info_ref),
@@ -910,7 +909,7 @@ def distributed_delete_apply(
"msgs_blob": pa.array([pickle.dumps(msgs)], type=pa.binary()),
"n_deleted": pa.array([len(row_ids)], type=pa.int64()),
"row_ids_blob": pa.array(
- [pickle.dumps(row_ids if collect_row_ids else [])],
+ [pickle.dumps(row_ids.to_pylist() if collect_row_ids else [])],
type=pa.binary(),
),
})
@@ -1023,24 +1022,3 @@ def distributed_write_collect_msgs(
write_kwargs["concurrency"] = concurrency
insert_ds.write_datasink(sink, **write_kwargs)
return sink.collected
-
-
-def _validate_live_row_ids(table, info, first_row_id, row_ids):
- """Reject deleted updates using only the routed snapshot's DV files.
-
- The writer must still merge physical rows, including tombstones, to
preserve
- row-id offsets. Filtering the writer's original data would shift those
offsets.
- """
- from pypaimon.execution.python.deletionvectors.deletion_vector import
DeletionVector
-
- split, files = info.first_row_id_index[first_row_id]
- if split.data_deletion_files is None:
- return
- target_names = {file.file_name for file in files}
- for file, deletion in zip(split.files, split.data_deletion_files):
- if deletion is None or file.file_name not in target_names:
- continue
- vector = DeletionVector.read(table.file_io, deletion)
- for row_id in row_ids:
- if file.row_id_range().contains(row_id) and
vector.is_deleted(row_id - file.first_row_id):
- raise ValueError("Cannot update deleted _ROW_ID
{}.".format(row_id))
diff --git a/paimon-python/pypaimon/ray/merge_condition.py
b/paimon-python/pypaimon/ray/merge_condition.py
index a918488e92..7afcda4c7a 100644
--- a/paimon-python/pypaimon/ray/merge_condition.py
+++ b/paimon-python/pypaimon/ray/merge_condition.py
@@ -51,14 +51,14 @@ def remap_source_on_keys(
def filter_batch(
batch: pa.Table, condition: str, _pre_rewritten: bool = False,
) -> pa.Table:
- from pypaimon.execution.python.write.merge_condition import filter_batch
as execute
+ from pypaimon.write.backend_dispatch import filter_merge_condition_batch
as execute
return execute(batch, condition, _pre_rewritten)
def apply_condition(
batch: pa.Table, rewritten: str, empty_schema: pa.Schema,
) -> pa.Table:
- from pypaimon.execution.python.write.merge_condition import
apply_condition as execute
+ from pypaimon.write.backend_dispatch import apply_merge_condition as
execute
return execute(batch, rewritten, empty_schema)
diff --git a/paimon-python/pypaimon/ray/vector_index_build.py
b/paimon-python/pypaimon/ray/vector_index_build.py
index fa5f722578..b0afbe6704 100644
--- a/paimon-python/pypaimon/ray/vector_index_build.py
+++ b/paimon-python/pypaimon/ray/vector_index_build.py
@@ -37,48 +37,15 @@ def validate_build_options(builder, concurrency,
ray_remote_args):
def build_vector_index(builder, splits, unindexed_ranges, index_field,
table_read,
index_path, concurrency, remote_args):
- import ray
- from pypaimon.execution.python.globalindex.build_plan import
split_by_global_index_shard
- from pypaimon.execution.python.globalindex.index_file_utils import
new_global_index_file_name
-
- shards = split_by_global_index_shard(
- splits, builder._core_options.global_index_row_count_per_shard(),
unindexed_ranges)
- # Allocate names before dispatch, so failed/lost result delivery cannot
hide
- # files from cleanup. Workers never publish index manifests or snapshots.
- names = [new_global_index_file_name("vindex-" + builder._index_type) for _
in shards]
- context = ray.put((builder, index_field, table_read, index_path))
- remote = ray.remote(_build_shard).options(**remote_args)
- pending = {}
- results = {}
- next_shard = 0
- try:
- while pending or next_shard < len(shards):
- while len(pending) < concurrency and next_shard < len(shards):
- ref = remote.remote(context, shards[next_shard],
names[next_shard])
- pending[ref] = next_shard
- next_shard += 1
- ready, _ = ray.wait(list(pending), num_returns=1)
- ref = ready[0]
- results[pending.pop(ref)] = ray.get(ref)
- return [results[i] for i in range(len(shards)) if results[i] is not
None]
- except BaseException:
- # Do not delete a file while another worker may still be writing it.
- # Stop submissions and drain every dispatched task before cleanup.
- for ref in pending:
- try:
- ray.get(ref)
- except BaseException:
- pass
- for name in names:
- builder._table.file_io.delete_quietly(index_path.rstrip("/") + "/"
+ name)
- raise
+ from pypaimon.globalindex.index_dispatch import build_python_vector_index
+ return build_python_vector_index(
+ builder, splits, unindexed_ranges, index_field, table_read,
+ index_path, concurrency, remote_args)
def _build_shard(context, shard, file_name):
- builder, index_field, table_read, index_path = context
- index_split, index_range = shard
- return builder._build_generic_shard(
- index_split, index_range, index_field, table_read, index_path,
file_name=file_name)
+ from pypaimon.globalindex.index_dispatch import
build_python_vector_index_shard
+ return build_python_vector_index_shard(context, shard, file_name)
def build_native_vector_index(table, plan, concurrency, remote_args):
@@ -117,5 +84,5 @@ def build_native_vector_index(table, plan, concurrency,
remote_args):
def _build_native_shard(table, shard):
- from pypaimon.execution.native.write.native_index_build import
build_native_index_shard
- return build_native_index_shard(table, shard)
+ from pypaimon.globalindex.index_dispatch import
build_native_vector_index_shard
+ return build_native_vector_index_shard(table, shard)
diff --git a/paimon-python/pypaimon/read/read_dispatch.py
b/paimon-python/pypaimon/read/read_dispatch.py
index 31fd58ff9a..2e2867fedb 100644
--- a/paimon-python/pypaimon/read/read_dispatch.py
+++ b/paimon-python/pypaimon/read/read_dispatch.py
@@ -18,6 +18,7 @@
"""Preparation selection for reading and planning; execution is never
replayed."""
import logging
+import threading
from typing import Iterator, List, Optional, Tuple
import pyarrow
@@ -28,6 +29,37 @@ from pypaimon.read.split import Split
from pypaimon.read.scan_stats import ScanStats
+class _ParquetRowGroupCacheRequest:
+ """Share cache configuration until a Python reader needs decoded rows."""
+
+ def __init__(self, max_size):
+ self.max_size = max_size
+ self._cache = None
+ self._lock = threading.Lock()
+
+ def __getstate__(self):
+ # Decoded batches and locks belong to the process that reads them.
+ return {'max_size': self.max_size}
+
+ def __setstate__(self, state):
+ self.__init__(state['max_size'])
+
+
+def python_row_group_cache(request):
+ if not isinstance(request, _ParquetRowGroupCacheRequest):
+ return request
+ with request._lock:
+ if request._cache is None:
+ from pypaimon.execution.python.read.reader.format_pyarrow_reader
import _DecodedRowGroupCache
+ request._cache = _DecodedRowGroupCache(request.max_size)
+ return request._cache
+
+
+def create_python_row_filter(rule_json):
+ from pypaimon.execution.python.read.predicate_json_parser import
parse_predicate_to_batch_filter
+ return parse_predicate_to_batch_filter(rule_json)
+
+
def validate_read_backend(read):
"""Validate the explicitly selected Python backend without loading
Native."""
if not read.table.options.native_read_enabled():
diff --git a/paimon-python/pypaimon/schema/data_type_casts.py
b/paimon-python/pypaimon/schema/data_type_casts.py
index 83067563fb..69a9314438 100644
--- a/paimon-python/pypaimon/schema/data_type_casts.py
+++ b/paimon-python/pypaimon/schema/data_type_casts.py
@@ -27,8 +27,8 @@ applies the conversion leniently.
"""
from pypaimon.schema.data_types import (ArrayType, AtomicType, DataTypeParser,
- MapType, MultisetType, RowType,
- VectorType)
+ MapType, MultisetType,
PyarrowFieldParser,
+ RowType, VectorType)
# ---- Type roots
--------------------------------------------------------------
@@ -152,6 +152,38 @@ def _build_rules():
_IMPLICIT_RULES, _EXPLICIT_RULES = _build_rules()
+# Materialization capabilities are schema contracts, independent of which
+# executor is installed or selected. Keep the existing Arrow leaf conversions
+# and the Java conversions supplied by the readers here, without executing a
+# cast while validating a schema change. Logical admissibility is checked by
+# supports_cast separately; Arrow also has storage casts absent from that spec.
+_NUMERIC_CAST_TARGETS = CHARACTER_STRING | NUMERIC | {BOOLEAN}
+_TEXT_CAST_TARGETS = CHARACTER_STRING | BINARY_STRING | NUMERIC | {
+ BOOLEAN, DATE, BLOB,
+} | TIMESTAMP_FAMILY
+_BINARY_CAST_TARGETS = CHARACTER_STRING | BINARY_STRING | NUMERIC | {BOOLEAN,
BLOB}
+_TIMESTAMP_CAST_TARGETS = CHARACTER_STRING | TIMESTAMP_FAMILY | {BIGINT, DATE,
TIME}
+_EXECUTABLE_CAST_RULES = {
+ CHAR: _TEXT_CAST_TARGETS,
+ VARCHAR: _TEXT_CAST_TARGETS,
+ BINARY: _BINARY_CAST_TARGETS,
+ VARBINARY: _BINARY_CAST_TARGETS,
+ BLOB: _BINARY_CAST_TARGETS,
+ BOOLEAN: CHARACTER_STRING | INTEGER_NUMERIC | {BOOLEAN, FLOAT, DOUBLE},
+ DECIMAL: CHARACTER_STRING | NUMERIC,
+ TINYINT: _NUMERIC_CAST_TARGETS,
+ SMALLINT: _NUMERIC_CAST_TARGETS,
+ INTEGER: _NUMERIC_CAST_TARGETS | {DATE, TIME, TIMESTAMP},
+ BIGINT: _NUMERIC_CAST_TARGETS | TIMESTAMP_FAMILY,
+ FLOAT: _NUMERIC_CAST_TARGETS,
+ DOUBLE: _NUMERIC_CAST_TARGETS,
+ DATE: CHARACTER_STRING | TIMESTAMP_FAMILY | {DATE, INTEGER},
+ TIME: CHARACTER_STRING | {TIME, INTEGER},
+ TIMESTAMP: _TIMESTAMP_CAST_TARGETS,
+ TIMESTAMP_LTZ: _TIMESTAMP_CAST_TARGETS,
+ VARIANT: {VARIANT},
+}
+
def supports_cast(source_type, target_type, allow_explicit: bool = True) ->
bool:
"""Whether ``source_type`` can be cast to ``target_type`` for a column type
@@ -190,7 +222,7 @@ def _equals_ignore_nullable(source_type, target_type) ->
bool:
def can_execute_cast(source_type, target_type) -> bool:
- """Whether the Python read path can actually *materialize* a stored
+ """Whether the read path can actually *materialize* a stored
``source_type`` value as ``target_type`` when reading a file written before
the column type change.
@@ -222,7 +254,14 @@ def can_execute_cast(source_type, target_type) -> bool:
# Any other conversion touching a constructed type has no runtime cast.
if source_root in CONSTRUCTED or target_root in CONSTRUCTED:
return False
- # Leaf-to-leaf: probe the same executor used by old-file readers, including
- # Java conversions with no direct PyArrow kernel, such as INT -> TIMESTAMP.
- from pypaimon.execution.python.casting.data_type_casts import
schema_evolution_cast_supported
- return schema_evolution_cast_supported(source_type, target_type)
+ # Resolve Arrow type metadata to retain parameter validation and alternate
+ # LTZ spellings. No arrays, cast kernels or execution backends are needed.
+ source_pa = PyarrowFieldParser.from_paimon_type(source_type)
+ target_pa = PyarrowFieldParser.from_paimon_type(target_type)
+ if source_pa == target_pa:
+ return True
+ if source_root in TIMESTAMP_FAMILY:
+ source_root = TIMESTAMP_LTZ if source_pa.tz is not None else TIMESTAMP
+ if target_root in TIMESTAMP_FAMILY:
+ target_root = TIMESTAMP_LTZ if target_pa.tz is not None else TIMESTAMP
+ return target_root in _EXECUTABLE_CAST_RULES.get(source_root, set())
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 49324a28df..40e0dcb816 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -489,30 +489,8 @@ class FileStoreTable(Table):
def create_row_key_extractor(
self, ignore_existing: bool = False
) -> 'RowKeyExtractor':
- from pypaimon.execution.python.write.row_key_extractor import (
- DynamicBucketRowKeyExtractor, FixedBucketRowKeyExtractor,
- PostponeBucketRowKeyExtractor, UnawareBucketRowKeyExtractor)
-
- bucket_mode = self.bucket_mode()
- if bucket_mode == BucketMode.HASH_FIXED:
- return FixedBucketRowKeyExtractor(self.table_schema)
- elif bucket_mode == BucketMode.BUCKET_UNAWARE:
- return UnawareBucketRowKeyExtractor(self.table_schema)
- elif bucket_mode == BucketMode.POSTPONE_MODE:
- return PostponeBucketRowKeyExtractor(self.table_schema)
- elif bucket_mode == BucketMode.HASH_DYNAMIC:
- return DynamicBucketRowKeyExtractor(
- self.table_schema,
- table=self,
- ignore_existing=ignore_existing,
- )
- elif bucket_mode == BucketMode.CROSS_PARTITION:
- raise ValueError(
- "CROSS_PARTITION primary-key writes require a persistent "
- "global primary-key index, which PyPaimon does not yet support"
- )
- else:
- raise ValueError(f"Unsupported bucket mode: {bucket_mode}")
+ from pypaimon.write.backend_dispatch import create_row_key_extractor
+ return create_row_key_extractor(self, ignore_existing)
def copy(self, options: dict) -> 'FileStoreTable':
return self._copy(options, resolve_time_travel=True)
diff --git a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
index 91ca0e4e56..fecb09d4fa 100644
--- a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
+++ b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
@@ -127,8 +127,4 @@ class
BatchVectorSearchBuilderImpl(AbstractVectorSearchBuilderImpl,
execution = copy(self)
execution._options = dict(self._options)
execution._query_vectors = [copy(vector) for vector in
self._query_vectors]
- return create_vector_read(execution,
execution._new_python_batch_vector_read, batch=True)
-
- def _new_python_batch_vector_read(self):
- from pypaimon.execution.python.read.search.factories import
new_batch_vector_read
- return new_batch_vector_read(self)
+ return create_vector_read(execution, batch=True)
diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py
b/paimon-python/pypaimon/table/source/full_text_search_builder.py
index 0c3b5fa370..21bfbed2be 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -146,11 +146,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
from pypaimon.table.source.search_dispatch import create_full_text_scan
reject_search_under_query_auth(self._table)
self._primary_key_full_text_definition()
- return create_full_text_scan(self,
copy(self)._new_python_full_text_scan)
-
- def _new_python_full_text_scan(self):
- from pypaimon.execution.python.read.search.factories import
new_full_text_scan
- return new_full_text_scan(self)
+ return create_full_text_scan(self, copy(self))
def new_full_text_read(self) -> FullTextRead:
from pypaimon.table.source.search_dispatch import create_full_text_read
@@ -170,11 +166,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
"Primary-key full-text search only supports the FAST "
"full-text index search mode; FULL and DETAIL require "
"merge-aware logical-row fallback.")
- return create_full_text_read(self,
copy(self)._new_python_full_text_read)
-
- def _new_python_full_text_read(self):
- from pypaimon.execution.python.read.search.factories import
new_full_text_read
- return new_full_text_read(self)
+ return create_full_text_read(self, copy(self))
def _text_columns(self):
if self._query is None:
diff --git a/paimon-python/pypaimon/table/source/search_dispatch.py
b/paimon-python/pypaimon/table/source/search_dispatch.py
index 166d073b83..963d96ab18 100644
--- a/paimon-python/pypaimon/table/source/search_dispatch.py
+++ b/paimon-python/pypaimon/table/source/search_dispatch.py
@@ -77,7 +77,11 @@ def _native_requested(builder):
return callable(enabled) and enabled()
-def create_vector_scan(builder, python_factory, batch=False):
+def create_vector_scan(builder, execution, batch=False):
+ def python_factory():
+ from pypaimon.execution.python.read.search.factories import
new_vector_scan
+ return new_vector_scan(execution)
+
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_vector_scan
@@ -85,7 +89,11 @@ def create_vector_scan(builder, python_factory, batch=False):
return python_factory() if native is None else _SearchScan(native,
python_factory)
-def create_vector_read(builder, python_factory, batch=False):
+def create_vector_read(builder, batch=False):
+ def python_factory():
+ from pypaimon.execution.python.read.search.factories import
new_batch_vector_read, new_vector_read
+ return new_batch_vector_read(builder) if batch else
new_vector_read(builder)
+
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_vector_read
@@ -96,7 +104,11 @@ def create_vector_read(builder, python_factory,
batch=False):
return reader_type(native, python_factory)
-def create_full_text_scan(builder, python_factory):
+def create_full_text_scan(builder, execution):
+ def python_factory():
+ from pypaimon.execution.python.read.search.factories import
new_full_text_scan
+ return new_full_text_scan(execution)
+
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_full_text_scan
@@ -104,7 +116,11 @@ def create_full_text_scan(builder, python_factory):
return python_factory() if native is None else _SearchScan(native,
python_factory)
-def create_full_text_read(builder, python_factory):
+def create_full_text_read(builder, execution):
+ def python_factory():
+ from pypaimon.execution.python.read.search.factories import
new_full_text_read
+ return new_full_text_read(execution)
+
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_full_text_read
diff --git a/paimon-python/pypaimon/table/source/vector_search_builder.py
b/paimon-python/pypaimon/table/source/vector_search_builder.py
index 7b9a32d9af..f4dc16c201 100644
--- a/paimon-python/pypaimon/table/source/vector_search_builder.py
+++ b/paimon-python/pypaimon/table/source/vector_search_builder.py
@@ -223,11 +223,7 @@ class AbstractVectorSearchBuilderImpl:
from pypaimon.table.source.search_dispatch import create_vector_scan
execution = copy(self)
execution._options = dict(self._options)
- return create_vector_scan(self, execution._new_python_vector_scan,
self._native_batch_search)
-
- def _new_python_vector_scan(self):
- from pypaimon.execution.python.read.search.factories import
new_vector_scan
- return new_vector_scan(self)
+ return create_vector_scan(self, execution, self._native_batch_search)
def _is_primary_key_vector_search(self):
if self._vector_column is None:
@@ -265,8 +261,4 @@ class
VectorSearchBuilderImpl(AbstractVectorSearchBuilderImpl, VectorSearchBuild
execution = copy(self)
execution._options = dict(self._options)
execution._query_vector = copy(self._query_vector)
- return create_vector_read(execution, execution._new_python_vector_read)
-
- def _new_python_vector_read(self):
- from pypaimon.execution.python.read.search.factories import
new_vector_read
- return new_vector_read(self)
+ return create_vector_read(execution)
diff --git
a/paimon-python/pypaimon/tests/global_index_execution_boundary_test.py
b/paimon-python/pypaimon/tests/global_index_execution_boundary_test.py
index b23894e626..06b3ba44d2 100644
--- a/paimon-python/pypaimon/tests/global_index_execution_boundary_test.py
+++ b/paimon-python/pypaimon/tests/global_index_execution_boundary_test.py
@@ -17,9 +17,7 @@
"""Shared index contracts and Native builds do not load Python index
executors."""
-import importlib
from pathlib import Path
-import pickle
import subprocess
import sys
import textwrap
@@ -208,17 +206,36 @@ assert not any(name == 'ray' or
name.startswith('pypaimon.ray') for name in sys.
assert result.returncode == 0, result.stdout + result.stderr
[email protected]('public_module,new_module,name', [
- ('globalindex',
- 'execution.python.globalindex.global_index_evaluator',
'GlobalIndexEvaluator'),
- ('globalindex',
- 'execution.python.globalindex.data_evolution_global_index_scanner',
'DataEvolutionGlobalIndexScanner'),
- ('globalindex', 'execution.python.globalindex.key_serializer',
'KeySerializer'),
- ('globalindex', 'execution.python.globalindex.memory_slice_input',
'MemorySliceInput'),
-])
-def test_public_index_exports_resolve_to_canonical_class(public_module,
new_module, name):
- public_module, new_module = 'pypaimon.' + public_module, 'pypaimon.' +
new_module
- canonical = getattr(importlib.import_module(new_module), name)
- assert getattr(importlib.import_module(public_module), name) is canonical
- public_pickle = ('c%s\n%s\n.' % (public_module, name)).encode('ascii')
- assert pickle.loads(public_pickle) is canonical
[email protected](sys.version_info < (3, 7), reason='Python 3.6 retains
eager package exports')
+def test_public_index_exports_are_backend_independent_contracts():
+ script = r"""
+import importlib.abc
+import pickle
+import sys
+
+class RejectExecution(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname == 'pypaimon.execution' or
fullname.startswith('pypaimon.execution.'):
+ raise AssertionError('Public index export imported execution: ' +
fullname)
+
+sys.meta_path.insert(0, RejectExecution())
+import pypaimon.globalindex as indexes
+
+for name in indexes.__all__:
+ exported = getattr(indexes, name)
+ # Existing public import paths still resolve in transported references.
+ reference = ('cpypaimon.globalindex\n%s\n.' % name).encode('ascii')
+ assert pickle.loads(reference) is exported
+for name in ('GlobalIndexEvaluator', 'DataEvolutionGlobalIndexScanner',
+ 'KeySerializer', 'MemorySliceInput', 'OffsetGlobalIndexReader',
+ 'SortedFileGlobalIndexReader', 'SortedFileMetaSelector'):
+ assert name not in indexes.__all__ and not hasattr(indexes, name), name
+assert not any(name == 'pypaimon.execution' or
name.startswith('pypaimon.execution.')
+ for name in sys.modules)
+"""
+ result = subprocess.run(
+ [sys.executable, '-c', textwrap.dedent(script)],
+ cwd=str(Path(__file__).resolve().parents[2]),
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
+ )
+ assert result.returncode == 0, result.stdout + result.stderr
diff --git a/paimon-python/pypaimon/tests/lerobot_native_reader_test.py
b/paimon-python/pypaimon/tests/lerobot_native_reader_test.py
index c8f7a3ad17..7b7612bc75 100644
--- a/paimon-python/pypaimon/tests/lerobot_native_reader_test.py
+++ b/paimon-python/pypaimon/tests/lerobot_native_reader_test.py
@@ -140,7 +140,8 @@ def
test_missing_native_runtime_keeps_cached_python_lookup(frames):
@pytest.mark.skipif(not native_runtime_available(), reason="native reader
required")
def test_native_plan_failure_reuses_scanner_and_keeps_predicate(frames):
- from pypaimon.globalindex import DataEvolutionGlobalIndexScanner
+ from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ DataEvolutionGlobalIndexScanner)
frames.add(pa.table({"index": [40, 50], "value": [4, 5]}))
current = reader(frames, **{
"scan.native-plan.enabled": "true", "read.native.enabled": "false",
@@ -199,7 +200,8 @@ def test_auto_native_uses_private_table_copy(frames,
available):
def test_unsupported_native_plan_reuses_python_index_scanner(frames):
- from pypaimon.globalindex import DataEvolutionGlobalIndexScanner
+ from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ DataEvolutionGlobalIndexScanner)
frames.add(pa.table({"index": [40], "value": [4]}))
with patch("pypaimon.execution.native.runtime.native_runtime_available",
return_value=True):
current = reader(frames)
@@ -260,7 +262,8 @@ def
test_native_lookup_preserves_query_authorization(frames, denied):
@pytest.mark.parametrize("option", ["scan.native-plan.enabled",
"read.native.enabled"])
def test_pickled_native_locator_rebuilds_pinned_fallback_once(frames, option):
- from pypaimon.globalindex import DataEvolutionGlobalIndexScanner
+ from
pypaimon.execution.python.globalindex.data_evolution_global_index_scanner
import (
+ DataEvolutionGlobalIndexScanner)
frames.add(pa.table({"index": [40], "value": [4]}))
current = reader(frames, **{
"scan.native-plan.enabled": "false", "read.native.enabled": "false",
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
b/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
new file mode 100644
index 0000000000..4dc94cec86
--- /dev/null
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
@@ -0,0 +1,122 @@
+# 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.
+
+"""Temporal schema and cache preparation keep reader execution independent."""
+
+import pickle
+import subprocess
+import sys
+
+import pyarrow as pa
+import pytest
+
+
+def test_importing_multimodal_does_not_load_python_execution():
+ result = subprocess.run([sys.executable, '-B', '-c', '''
+import sys
+import pypaimon.multimodal
+loaded = [name for name in sys.modules if
name.startswith('pypaimon.execution.python.')]
+assert not loaded, loaded
+'''], capture_output=True, text=True)
+ assert result.returncode == 0, result.stderr
+
+
+def test_selected_native_batches_do_not_initialize_the_python_cache():
+ result = subprocess.run([sys.executable, '-B', '-c', '''
+import sys
+from types import SimpleNamespace
+import pyarrow as pa
+from pypaimon.read.read_dispatch import _ParquetRowGroupCacheRequest, to_arrow
+schema = pa.schema([('value', pa.int32())])
+batch = pa.record_batch([pa.array([7], type=pa.int32())], schema=schema)
+request = _ParquetRowGroupCacheRequest(1024)
+read = SimpleNamespace(
+ include_row_kind=False,
+ _parquet_row_group_cache=request,
+ _output_arrow_schema=lambda splits: schema,
+ _resolve_parallelism=lambda value, count: 1,
+ _resolve_blob_parallelism=lambda value: 1,
+ _try_native_batches=lambda *args, **kwargs: [batch],
+ _batches_to_arrow=lambda batches, output_schema:
pa.Table.from_batches(batches, schema=output_schema))
+assert to_arrow(read, [object()]).to_pydict() == {'value': [7]}
+assert request._cache is None
+loaded = [name for name in sys.modules if
name.startswith('pypaimon.execution.python.')]
+assert not loaded, loaded
+'''], capture_output=True, text=True)
+ assert result.returncode == 0, result.stderr
+
+
[email protected]('data_type,transform', [
+ (pa.int32(), {'name': 'NULL'}),
+ (pa.int32(), {'name': 'FIELD_REF', 'fieldRef': {'name': 'value'}}),
+ (pa.int32(), {'name': 'CAST', 'fieldRef': {'name': 'value', 'type':
'INT'}, 'type': 'STRING'}),
+ (pa.string(), {'name': 'UPPER', 'inputs': [{'name': 'value', 'type':
'STRING'}]}),
+ (pa.string(), {'name': 'UPPER', 'inputs': [42]}),
+ (pa.large_string(), {'name': 'LOWER', 'inputs': [{'name': 'value', 'type':
'STRING'}]}),
+ (pa.string(), {'name': 'CONCAT', 'inputs': [{'name': 'value'}, 'tail']}),
+ (pa.large_string(), {'name': 'CONCAT_WS', 'inputs': ['-', {'name':
'value'}]}),
+ (pa.large_string(), {'name': 'SUBSTRING', 'inputs': [{'name': 'value'}, 1,
2]}),
+ (pa.large_string(), {'name': 'SUBSTRING', 'inputs': [{'name': 'value'},
-1, 2]}),
+ (pa.large_string(), {'name': 'TRIM', 'trimFlag': 'BOTH', 'inputs':
[{'name': 'value', 'type': 'STRING'}]}),
+ (pa.large_string(), {'name': 'LENGTH', 'inputs': [{'name': 'value',
'type': 'STRING'}]}),
+ (pa.large_string(), {'name': 'BIT_LENGTH', 'inputs': [{'name': 'value',
'type': 'STRING'}]}),
+ (pa.large_string(), {'name': 'PAD', 'direction': 'LEFT', 'inputs':
[{'name': 'value', 'type': 'STRING'}, 4, 'x']}),
+ (pa.large_string(), {'name': 'TRANSLATE', 'inputs': [{'name': 'value',
'type': 'STRING'}, 'a', 'b']}),
+ (pa.large_string(), {'name': 'OVERLAY', 'inputs': [{'name': 'value',
'type': 'STRING'}, 'x', 1]}),
+ (pa.date32(), {'name': 'YEAR', 'fieldRef': {'name': 'value', 'type':
'DATE'}}),
+])
+def test_temporal_mask_schema_matches_actual_python_mask(data_type, transform):
+ from pypaimon.common.transform_schema import effective_masked_schema
+ from pypaimon.execution.python.read.predicate_json_parser import
_apply_predicate_transform
+
+ schema = pa.schema([pa.field('value', data_type, nullable=False,
metadata={b'field': b'value'})],
+ metadata={b'schema': b'value'})
+ batch = pa.RecordBatch.from_arrays([pa.array([], type=data_type)],
schema=schema)
+ actual = _apply_predicate_transform(transform, batch, null_type=data_type)
+ inferred = effective_masked_schema(schema, {'value': transform})
+ assert inferred.field('value').type == actual.type
+ assert inferred.field('value').nullable
+ assert inferred.field('value').metadata == schema.field('value').metadata
+ assert inferred.metadata == schema.metadata
+
+
[email protected]('transform', [
+ {'name': 'unknown'},
+ {'name': 'FIELD_REF', 'fieldRef': {'name': 'missing'}},
+ {'name': 'CAST', 'fieldRef': {'name': 'value', 'type': 'STRING'}, 'type':
'INVALID'},
+ {'name': 'SUBSTRING', 'inputs': ['value']},
+ {'name': 'TRIM', 'trimFlag': 'INVALID', 'inputs': ['value']},
+ {'name': 'PAD', 'direction': 'INVALID', 'inputs': ['value', 4, 'x']},
+ {'name': 'LENGTH', 'inputs': [{'name': 'value', 'type': 'INT'}]},
+])
+def test_temporal_mask_schema_rejects_invalid_rules(transform):
+ from pypaimon.common.transform_schema import effective_masked_schema
+
+ with pytest.raises((ValueError, KeyError, TypeError)):
+ effective_masked_schema(pa.schema([('value', pa.string())]), {'value':
transform})
+
+
+def
test_temporal_cache_is_created_only_for_python_and_recreated_after_transport():
+ from pypaimon.read.read_dispatch import _ParquetRowGroupCacheRequest,
python_row_group_cache
+
+ request = _ParquetRowGroupCacheRequest(1024)
+ assert request._cache is None
+ cached = python_row_group_cache(request)
+ assert python_row_group_cache(request) is cached
+ restored = pickle.loads(pickle.dumps(request))
+ assert restored._cache is None
+ assert python_row_group_cache(restored) is not cached
diff --git a/paimon-python/pypaimon/tests/native_historical_update_test.py
b/paimon-python/pypaimon/tests/native_historical_update_test.py
index 7bf9426591..020b235e8b 100644
--- a/paimon-python/pypaimon/tests/native_historical_update_test.py
+++ b/paimon-python/pypaimon/tests/native_historical_update_test.py
@@ -30,7 +30,8 @@ from pypaimon.schema.schema_change import SchemaChange
from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.execution.native.write.native_update import
NativeShardTableUpdator
-from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
+from pypaimon.write.backend_dispatch import native_row_id_update_selected
pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
@@ -91,7 +92,7 @@ def _update(table, operation, stream, native):
update = builder.new_update().with_update_type(['value'])
kwargs = dict(commit_identifier=37) if stream else {}
changed = pa.table({'_ROW_ID': [0], 'value': [11]})
- guard = patch.object(TableUpdateByRowId, 'update_columns',
+ guard = patch.object(PythonTableUpdateByRowId, 'update_columns',
side_effect=AssertionError('Python historical update
fallback')) if native else ExitStack()
with guard:
if operation == 'row_id':
@@ -100,6 +101,7 @@ def _update(table, operation, stream, native):
return update.update_by_arrow_batches_with_row_id(iter([changed]),
**kwargs)
if operation == 'incremental':
updater = update.new_update_by_row_id(**kwargs)
+ assert native_row_id_update_selected(updater) == native
return updater.update_columns(changed, ['value'])
if operation == 'predicate':
predicate =
table.new_read_builder().new_predicate_builder().equal('id', 1)
diff --git a/paimon-python/pypaimon/tests/native_ray_row_id_plan_test.py
b/paimon-python/pypaimon/tests/native_ray_row_id_plan_test.py
index 4072c17a6b..76543fa786 100644
--- a/paimon-python/pypaimon/tests/native_ray_row_id_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_ray_row_id_plan_test.py
@@ -186,6 +186,9 @@ def
test_ray_callbacks_route_read_update_and_delete_with_captured_core_plans(tmp
if native:
guards.enter_context(patch.object(PythonRowIdPlan, '_positions',
side_effect=AssertionError('Python row-ID routing')))
+ guards.enter_context(patch(
+
'pypaimon.execution.python.write.row_id_plan.validate_live_row_ids',
+ side_effect=AssertionError('Python deletion-vector
validation')))
source = _dataset(pa.table({'_ROW_ID': pa.array([3, 0, 3],
type=pa.int64())}))
read = distributed_read_by_row_id(source, table, ['id', 'value'],
num_partitions=2)
rows = pa.concat_tables(list(read.iter_batches())).sort_by('id')
@@ -214,3 +217,55 @@ def
test_ray_callbacks_route_read_update_and_delete_with_captured_core_plans(tmp
read = distributed_read_by_row_id(source, table, ['id', 'value'],
num_partitions=2)
rows = pa.concat_tables(list(read.iter_batches())).sort_by('id')
assert rows.to_pydict() == {'id': [1, 4], 'value': [8, 9], '_ROW_ID':
[0, 3]}
+
+
[email protected]('native', _ENGINES)
[email protected]('collect_row_ids', [False, True])
[email protected]('selected_ids', [[], [0, 3]])
+def test_ray_delete_transports_arrow_ids_and_collects_only_requested_actions(
+ tmp_path, native, collect_row_ids, selected_ids, monkeypatch):
+ from pypaimon.ray.data_evolution_merge_join import distributed_delete_apply
+
+ table = _table(tmp_path)
+ _append(table, [1, 2])
+ _append(table, [3, 4])
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+ scheduler = SimpleNamespace(
+ put=lambda value: pickle.loads(pickle.dumps(value)), get=lambda value:
value)
+ monkeypatch.setitem(sys.modules, 'ray', scheduler)
+ source = _dataset(pa.table({'_ROW_ID': pa.array(selected_ids,
type=pa.int64())}))
+ delete = TableDeleteByRowId.delete
+ received = []
+
+ def delete_arrow(worker, row_ids):
+ assert isinstance(row_ids, (pa.Array, pa.ChunkedArray))
+ assert row_ids.type == pa.int64()
+ received.append(row_ids)
+ return delete(worker, row_ids)
+
+ with patch.object(TableDeleteByRowId, 'delete', delete_arrow):
+ messages, count, actions = distributed_delete_apply(
+ source, table, num_partitions=2, collect_row_ids=collect_row_ids)
+ assert [ids.to_pylist() for ids in received] == ([selected_ids] if
selected_ids else [])
+ assert count == len(selected_ids)
+ assert actions == (selected_ids if collect_row_ids else [])
+ assert all(message.check_from_snapshot == snapshot_id for message in
messages)
+ table.new_batch_write_builder().new_commit().commit(messages)
+ read = table.new_read_builder()
+ rows =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert rows['id'].to_pylist() == ([2, 3] if selected_ids else [1, 2, 3, 4])
+
+
+def test_ray_delete_rejects_duplicate_ids_before_writing(tmp_path,
monkeypatch):
+ from pypaimon.ray.data_evolution_merge_join import distributed_delete_apply
+
+ table = _table(tmp_path)
+ _append(table, [1, 2])
+ scheduler = SimpleNamespace(
+ put=lambda value: pickle.loads(pickle.dumps(value)), get=lambda value:
value)
+ monkeypatch.setitem(sys.modules, 'ray', scheduler)
+ source = _dataset(pa.table({'_ROW_ID': pa.array([0, 0], type=pa.int64())}))
+ with patch.object(TableDeleteByRowId, 'delete',
side_effect=AssertionError('delete before validation')):
+ with pytest.raises(ValueError, match='multiple source rows'):
+ distributed_delete_apply(source, table, num_partitions=2)
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index 894b54426a..d42409c090 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -30,6 +30,7 @@ from pypaimon.execution.native.write.native_update import
create_native_delete,
from pypaimon.write.table_delete import TableDeleteByRowId
from pypaimon.read.table_read import TableRead
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+from pypaimon.write.backend_dispatch import native_row_id_update_selected
pytestmark = pytest.mark.native_plan
@@ -39,7 +40,6 @@ pytestmark = pytest.mark.native_plan
def
test_incremental_row_id_updater_uses_core_and_accumulates_columns(tmp_path,
stream):
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.table.row.generic_row import GenericRow
- from pypaimon.execution.native.write.native_update import
NativeTableUpdateByRowId
catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
catalog.create_database('default', True)
@@ -62,7 +62,8 @@ def
test_incremental_row_id_updater_uses_core_and_accumulates_columns(tmp_path,
with patch.object(PythonTableUpdateByRowId, '_load_existing_files_info',
side_effect=AssertionError('Python loaded the file
index')):
updater = update.new_update_by_row_id(commit_id) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater, TableUpdateByRowId)
+ assert native_row_id_update_selected(updater)
assert updater.commit_user == builder.commit_user
assert updater.commit_identifier == commit_id
assert updater.commit_messages == []
@@ -91,7 +92,6 @@ def
test_incremental_row_id_updater_uses_core_and_accumulates_columns(tmp_path,
@pytest.mark.parametrize('stream', [False, True])
def test_partitioned_upsert_and_incremental_update_use_core(tmp_path, stream):
- from pypaimon.execution.native.write.native_update import
NativeTableUpdateByRowId
from pypaimon.table.row.generic_row import GenericRow
catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
@@ -133,7 +133,8 @@ def
test_partitioned_upsert_and_incremental_update_use_core(tmp_path, stream):
current = read.new_read().to_arrow(read.new_scan().plan().splits())
selected = current.filter(pa.compute.equal(current['p'], 'a'))
updater = update.new_update_by_row_id(3) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater, TableUpdateByRowId)
+ assert native_row_id_update_selected(updater)
messages = updater.update_columns(selected.select(['p', '_ROW_ID']), ['p'])
commit = builder.new_commit()
commit.commit(messages, 3) if stream else commit.commit(messages)
diff --git a/paimon-python/pypaimon/tests/public_execution_boundary_test.py
b/paimon-python/pypaimon/tests/public_execution_boundary_test.py
new file mode 100644
index 0000000000..b6f9730fdb
--- /dev/null
+++ b/paimon-python/pypaimon/tests/public_execution_boundary_test.py
@@ -0,0 +1,113 @@
+# 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.
+
+"""Public imports and actual Native temporal reads keep Python execution
unloaded."""
+
+import subprocess
+import sys
+
+import pytest
+
+from pypaimon.execution.native.runtime import native_runtime_available
+
+
+def test_public_imports_and_shared_validation_do_not_load_either_executor():
+ script = r'''
+import importlib
+import importlib.abc
+import sys
+
+class RejectExecutors(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname.startswith(('pypaimon.execution.python.',
'pypaimon.execution.native.')):
+ raise AssertionError('Shared API loaded execution: ' + fullname)
+
+sys.meta_path.insert(0, RejectExecutors())
+for name in (
+ 'pypaimon.multimodal', 'pypaimon.api.rest_api',
'pypaimon.table.file_store_table',
+ 'pypaimon.catalog.table_query_auth', 'pypaimon.filesystem.local_file_io',
+ 'pypaimon.filesystem.pyarrow_file_io', 'pypaimon.ray.merge_condition',
+ 'pypaimon.ray.vector_index_build',
+):
+ importlib.import_module(name)
+import pypaimon.globalindex as indexes
+for name in indexes.__all__:
+ getattr(indexes, name)
+from pypaimon.schema.data_type_casts import supports_cast
+from pypaimon.schema.data_types import AtomicType
+assert supports_cast(AtomicType('INT'), AtomicType('BIGINT'))
+assert not any(name.startswith(('pypaimon.execution.python.',
'pypaimon.execution.native.'))
+ for name in sys.modules)
+'''
+ result = subprocess.run(
+ [sys.executable, '-B', '-c', script], stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, universal_newlines=True, timeout=30)
+ assert result.returncode == 0, result.stderr
+
+
[email protected](not native_runtime_available(), reason='pypaimon-rust not
installed')
+def test_native_temporal_reads_never_import_python_execution(tmp_path):
+ script = r'''
+# NATIVE_TEMPORAL_BOUNDARY: the test requires the current SDK in this process.
+import importlib.abc
+import sys
+
+class RejectPythonExecution(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname == 'pypaimon.execution.python' or
fullname.startswith('pypaimon.execution.python.'):
+ raise AssertionError('Native temporal loaded Python execution: ' +
fullname)
+
+sys.meta_path.insert(0, RejectPythonExecution())
+from pypaimon_rust.datafusion import SQLContext
+context = SQLContext()
+context.register_catalog('paimon', {'warehouse': sys.argv[1]})
+context.sql('CREATE SCHEMA paimon.temporal_boundary')
+for name, extra, values in (
+ ('anchors', '', '(1, 1), (1, 3)'),
+ ('samples', ', value INT', '(1, 0, 0), (1, 4, 40)'),
+):
+ definition = "CREATE TABLE paimon.temporal_boundary." + name + " (episode
INT, event_time BIGINT" + extra
+ context.sql(definition + """) WITH (
+ 'row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')""")
+ columns = '(episode, event_time)' if name == 'anchors' else '(episode,
event_time, value)'
+ context.sql('INSERT INTO paimon.temporal_boundary.' + name + ' ' + columns
+ ' VALUES ' + values)
+
+from pypaimon import CatalogFactory
+from pypaimon.multimodal import join_asof, interpolate, join_window
+from pypaimon.multimodal.query import ScanQuery
+catalog = CatalogFactory.create({'warehouse': sys.argv[1]})
+options = {'read.native.enabled': 'true', 'scan.native-plan.enabled': 'true'}
+anchors =
ScanQuery(catalog.get_table('temporal_boundary.anchors').copy(options))
+samples =
ScanQuery(catalog.get_table('temporal_boundary.samples').copy(options)).select('value')
+asof = join_asof(anchors, samples, on='event_time', by='episode')
+assert asof.to_arrow()['value'].to_pylist() == [0, 0]
+linear = interpolate(anchors, samples, on='event_time', by='episode')
+assert linear.to_arrow()['value'].to_pylist() == [10., 30.]
+window = join_window(anchors, samples, on='event_time', by='episode',
+ preceding=1, following=1, aggregations={'mean_value':
('value', 'mean')})
+assert window.to_arrow()['mean_value'].to_pylist() == [0., 40.]
+reader = asof.to_arrow_batch_reader(batch_size=1)
+try:
+ assert reader.read_next_batch().num_rows == 1
+finally:
+ reader.close()
+assert not any(name.startswith('pypaimon.execution.python') for name in
sys.modules)
+'''
+ result = subprocess.run(
+ [sys.executable, '-B', '-c', script, str(tmp_path)],
stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, universal_newlines=True, timeout=90)
+ assert result.returncode == 0, result.stderr
diff --git
a/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
b/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
new file mode 100644
index 0000000000..7869fedb20
--- /dev/null
+++ b/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
@@ -0,0 +1,209 @@
+# 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.
+
+"""Public factories retain captured inputs and enter their selected dispatch
boundary."""
+
+import subprocess
+import sys
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.source.batch_vector_search_builder import
BatchVectorSearchBuilderImpl
+from pypaimon.table.source.full_text_search_builder import
FullTextSearchBuilderImpl
+from pypaimon.table.source.vector_search_builder import VectorSearchBuilderImpl
+
+
+def _search_table(path, native):
+ catalog = CatalogFactory.create({'warehouse': str(path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.search', Schema.from_pyarrow_schema(
+ pa.schema([('embedding', pa.list_(pa.float32())), ('body',
pa.string())]),
+ options={'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'read.native.enabled': str(native).lower()}), False)
+ return catalog.get_table('default.search')
+
+
[email protected]('kind', ['vector', 'batch-vector', 'full-text'])
[email protected]('phase', ['scan', 'read'])
[email protected]('native', [False, True])
+def test_search_factories_keep_captured_configuration(tmp_path, kind, phase,
native):
+ from pypaimon.table.source.search_dispatch import python_search_reader
+
+ table = _search_table(tmp_path, native)
+ query = [1.0, 2.0]
+ if kind == 'vector':
+ builder =
(VectorSearchBuilderImpl(table).with_vector_column('embedding')
+
.with_query_vector(query).with_limit(2).with_option('ivf.nprobe', '1'))
+ elif kind == 'batch-vector':
+ builder =
(BatchVectorSearchBuilderImpl(table).with_vector_column('embedding')
+
.with_query_vectors([query]).with_limit(2).with_option('ivf.nprobe', '1'))
+ else:
+ builder = FullTextSearchBuilderImpl(table).with_query('body',
'term').with_limit(2)
+ family = 'full_text' if kind == 'full-text' else 'vector'
+ python_name = ('new_batch_vector_read' if kind == 'batch-vector' and phase
== 'read'
+ else 'new_' + family + '_' + phase)
+ native_name = 'native_' + family + '_' + phase
+ python_execution, native_execution = Mock(), Mock()
+ with patch('pypaimon.execution.python.read.search.factories.' +
python_name,
+ return_value=python_execution) as python_factory, \
+ patch('pypaimon.execution.native.read.native_search_plan.' +
native_name,
+ return_value=native_execution) as native_factory:
+ if phase == 'scan':
+ selected = (builder.new_full_text_scan() if kind == 'full-text'
+ else builder.new_vector_search_scan())
+ else:
+ selected = (builder.new_full_text_read() if kind == 'full-text'
+ else builder.new_batch_vector_search_read() if kind ==
'batch-vector'
+ else builder.new_vector_search_read())
+ if native:
+ python_factory.assert_not_called()
+ native_factory.assert_called_once()
+ else:
+ native_factory.assert_not_called()
+ builder.with_limit(9)
+ if kind == 'full-text':
+ builder.with_query('body', 'later')
+ else:
+ query[0] = 99.0
+ builder.with_option('ivf.nprobe', '9')
+ if phase == 'scan':
+ native_execution.available.return_value = False
+ assert selected.scan() is python_execution.scan.return_value
+ else:
+ assert python_search_reader(selected) is python_execution
+ python_factory.assert_called_once()
+ captured = python_factory.call_args[0][0]
+ assert captured is not builder and captured._limit == 2
+ if kind == 'full-text':
+ assert captured._query == 'term'
+ else:
+ assert captured._options == {'ivf.nprobe': '1'}
+ if phase == 'read':
+ assert (captured._query_vector if kind == 'vector' else
captured._query_vectors[0]) == [1.0, 2.0]
+
+
+def test_native_search_factories_do_not_import_python_execution(tmp_path):
+ source = r"""
+import importlib.abc
+import sys
+from unittest.mock import Mock, patch
+import pyarrow as pa
+
+class RejectPythonExecution(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname.startswith('pypaimon.execution.python'):
+ raise AssertionError('Native factory imported Python execution: '
+ fullname)
+
+sys.meta_path.insert(0, RejectPythonExecution())
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.source.vector_search_builder import VectorSearchBuilderImpl
+from pypaimon.table.source.batch_vector_search_builder import
BatchVectorSearchBuilderImpl
+from pypaimon.table.source.full_text_search_builder import
FullTextSearchBuilderImpl
+from pypaimon.execution.native.read import native_search_plan
+
+catalog = CatalogFactory.create({'warehouse': sys.argv[1]})
+catalog.create_database('default', True)
+catalog.create_table('default.search', Schema.from_pyarrow_schema(
+ pa.schema([('embedding', pa.list_(pa.float32())), ('body', pa.string())]),
+ options={'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'read.native.enabled': 'true'}), False)
+table = catalog.get_table('default.search')
+vector = (VectorSearchBuilderImpl(table).with_vector_column('embedding')
+ .with_query_vector([1., 2.]).with_limit(2))
+batch = (BatchVectorSearchBuilderImpl(table).with_vector_column('embedding')
+ .with_query_vectors([[1., 2.]]).with_limit(2))
+text = FullTextSearchBuilderImpl(table).with_query('body',
'term').with_limit(2)
+native = Mock()
+native.available.return_value = True
+expected = object()
+for name in ['scan', 'read_plan', 'read_batch_plan']:
+ getattr(native, name).return_value = expected
+constructors = {name: Mock(return_value=native) for name in [
+ 'native_vector_scan', 'native_vector_read', 'native_full_text_scan',
'native_full_text_read']}
+with patch.multiple(native_search_plan, **constructors):
+ operations = [
+ lambda: vector.new_vector_search_scan().scan(),
+ lambda: vector.new_vector_search_read().read_plan(object()),
+ lambda: batch.new_vector_search_scan().scan(),
+ lambda: batch.new_batch_vector_search_read().read_batch_plan(object()),
+ lambda: text.new_full_text_scan().scan(),
+ lambda: text.new_full_text_read().read_plan(object()),
+ ]
+ for operation in operations:
+ assert operation() is expected
+ failure = OSError('selected native executor failed')
+ for name in ['scan', 'read_plan', 'read_batch_plan']:
+ getattr(native, name).side_effect = failure
+ for operation in operations:
+ try:
+ operation()
+ except OSError as error:
+ assert error is failure
+ else:
+ raise AssertionError('Native failure was hidden')
+assert not any(name.startswith('pypaimon.execution.python') for name in
sys.modules)
+"""
+ result = subprocess.run(
+ [sys.executable, '-c', source, str(tmp_path)], stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, universal_newlines=True, timeout=45)
+ assert result.returncode == 0, result.stdout + result.stderr
+
+
[email protected]('bucket,primary_keys,expected', [
+ ('2', ['id', 'p'], 'FixedBucketRowKeyExtractor'),
+ ('-1', [], 'UnawareBucketRowKeyExtractor'),
+ ('-2', ['id', 'p'], 'PostponeBucketRowKeyExtractor'),
+ ('-1', ['id', 'p'], 'DynamicBucketRowKeyExtractor'),
+])
+def test_public_row_key_factory_preserves_per_row_contract(tmp_path, bucket,
primary_keys, expected):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ arrow_schema = pa.schema([('id', pa.int32()), ('p', pa.string())])
+ catalog.create_table('default.keys', Schema.from_pyarrow_schema(
+ arrow_schema, primary_keys=primary_keys, partition_keys=['p'],
options={
+ 'bucket': bucket, 'write.native.enabled': 'true',
'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false', 'commit.native.enabled':
'false'}), False)
+ table = catalog.get_table('default.keys')
+ rows = pa.record_batch([pa.array([1, 2], pa.int32()), pa.array(['a',
'a'])], schema=arrow_schema)
+ with patch('pypaimon.execution.native.runtime.native_runtime_available',
+ side_effect=AssertionError('Per-row helper rediscovered the
Native writer')):
+ extractor = table.create_row_key_extractor(ignore_existing=True)
+ partitions, buckets = extractor.extract_partition_bucket_batch(rows)
+ assert type(extractor).__name__ == expected
+ assert partitions == [('a',), ('a',)]
+ if bucket == '2':
+ assert all(0 <= value < 2 for value in buckets)
+ else:
+ assert buckets == [-2 if bucket == '-2' else 0] * 2
+
+
+def test_ray_condition_compatibility_helpers_keep_python_execution():
+ if sys.version_info < (3, 10):
+ pytest.skip('Python MERGE conditions require Python 3.10')
+ pytest.importorskip('datafusion')
+ from pypaimon.ray.merge_condition import apply_condition, filter_batch,
rewrite_condition
+
+ rows = pa.table({'s.id': pa.chunked_array([[1, 2], [3]]), 't.id': [2, 1,
2]})
+ with patch('pypaimon.execution.native.runtime.native_runtime_available',
+ side_effect=AssertionError('Python condition helper discovered
Native execution')):
+ assert filter_batch(rows, 's.id > t.id').column('s.id').to_pylist() ==
[2, 3]
+ schema = pa.schema([('s.id', pa.int64())])
+ empty = apply_condition(rows, rewrite_condition('s.id < 0'), schema)
+ assert empty.num_rows == 0 and empty.schema == schema
diff --git a/paimon-python/pypaimon/tests/schema_cast_support_test.py
b/paimon-python/pypaimon/tests/schema_cast_support_test.py
new file mode 100644
index 0000000000..c5fdfaca2b
--- /dev/null
+++ b/paimon-python/pypaimon/tests/schema_cast_support_test.py
@@ -0,0 +1,115 @@
+# 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.
+
+"""Schema cast validation describes capabilities without loading execution."""
+
+from pathlib import Path
+import subprocess
+import sys
+import textwrap
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.schema.data_type_casts import can_execute_cast
+from pypaimon.schema.data_types import AtomicType, PyarrowFieldParser
+
+
+_LEAF_TYPES = [
+ 'CHAR(3)', 'VARCHAR(5)', 'BOOLEAN', 'BINARY(3)', 'VARBINARY(5)',
+ 'DECIMAL(4,2)', 'DECIMAL(38,18)', 'TINYINT', 'SMALLINT', 'INT', 'BIGINT',
+ 'FLOAT', 'DOUBLE', 'DATE', 'TIME', 'TIME(3)',
+ 'TIMESTAMP(0)', 'TIMESTAMP(3)', 'TIMESTAMP(6)', 'TIMESTAMP(9)',
+ 'TIMESTAMP_LTZ(0)', 'TIMESTAMP_LTZ(3)', 'TIMESTAMP_LTZ(6)',
'TIMESTAMP_LTZ(9)',
+ 'TIMESTAMP(6) WITH LOCAL TIME ZONE', 'BLOB', 'VARIANT',
+]
+
+
[email protected]('source', _LEAF_TYPES)
+def
test_declared_cast_capabilities_match_existing_read_materialization(source):
+ from pypaimon.execution.python.casting.schema_evolution import
cast_array_for_schema_evolution
+
+ source_type = AtomicType(source)
+ values = pa.nulls(1, type=PyarrowFieldParser.from_paimon_type(source_type))
+ for target in _LEAF_TYPES:
+ target_type = AtomicType(target)
+ try:
+ cast_array_for_schema_evolution(
+ values, PyarrowFieldParser.from_paimon_type(target_type),
source_type, target_type)
+ executable = True
+ except (pa.ArrowNotImplementedError, pa.ArrowInvalid,
pa.ArrowTypeError,
+ ValueError, NotImplementedError):
+ executable = False
+ # TIME precision is not represented by its Arrow storage unit, so
+ # changing it is rejected independently of Arrow's identity kernel.
+ if (source, target) in (('TIME', 'TIME(3)'), ('TIME(3)', 'TIME')):
+ executable = False
+ assert can_execute_cast(source_type, target_type) == executable, '{}
-> {}'.format(source, target)
+
+
[email protected](sys.version_info < (3, 7), reason='Python 3.6 retains
eager package exports')
+def test_schema_type_changes_do_not_import_or_execute_either_backend():
+ script = r"""
+import importlib.abc
+import sys
+from unittest.mock import patch
+
+class RejectExecution(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname == 'pypaimon.execution' or
fullname.startswith('pypaimon.execution.'):
+ raise AssertionError('Schema validation imported execution: ' +
fullname)
+
+sys.meta_path.insert(0, RejectExecution())
+from pypaimon.schema.data_type_casts import can_execute_cast
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.schema.schema_manager import SchemaManager
+from pypaimon.schema.table_schema import TableSchema
+
+manager = SchemaManager(None, '/unused-schema-capabilities')
+with patch('pyarrow.nulls', side_effect=AssertionError('Schema validation
executed a cast probe')):
+ for source, target, supported in [
+ ('INT', 'BIGINT', True), ('INT', 'TIMESTAMP(3)', True),
+ ('BIGINT', 'TIMESTAMP_LTZ(6)', True), ('INT', 'DECIMAL(4,2)', True),
+ ('TIMESTAMP_LTZ(9)', 'TIMESTAMP(3)', True),
+ ('TIMESTAMP(3)', 'DECIMAL(10,2)', False), ('BOOLEAN', 'DECIMAL(4,2)',
False),
+ ('TIME', 'TIMESTAMP(6)', False), ('TIME(3)', 'TIME(1)', False),
+ ]:
+ source_type, target_type = AtomicType(source), AtomicType(target)
+ assert can_execute_cast(source_type, target_type) == supported,
(source, target)
+ schema = TableSchema(fields=[DataField(0, 'v', source_type)])
+ change = SchemaChange.update_column_type('v', target_type)
+ if supported:
+ changed = manager._generate_table_schema(schema, [change])
+ assert changed.fields[0].id == 0 and changed.fields[0].type ==
target_type
+ assert schema.fields[0].type == source_type
+ else:
+ try:
+ manager._generate_table_schema(schema, [change])
+ except ValueError as error:
+ assert 'no executable cast' in str(error)
+ else:
+ raise AssertionError('Unsupported schema cast was accepted: '
+ str((source, target)))
+assert not any(name == 'pypaimon.execution' or
name.startswith('pypaimon.execution.')
+ for name in sys.modules)
+"""
+ result = subprocess.run(
+ [sys.executable, '-c', textwrap.dedent(script)],
+ cwd=str(Path(__file__).resolve().parents[2]),
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True,
+ )
+ assert result.returncode == 0, result.stdout + result.stderr
diff --git a/paimon-python/pypaimon/tests/write/row_id_delete_input_test.py
b/paimon-python/pypaimon/tests/write/row_id_delete_input_test.py
new file mode 100644
index 0000000000..b07aa383b2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/row_id_delete_input_test.py
@@ -0,0 +1,175 @@
+# 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.
+
+"""Arrow row-ID deletes preserve the same snapshot and bitmap semantics."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.execution.native.runtime import native_runtime_available
+from pypaimon.execution.native.write.native_update import
NativeBatchTableUpdate
+from pypaimon.write.row_id_plan import capture_row_id_plan
+from pypaimon.write.row_input import normalize_row_ids
+from pypaimon.write.table_delete import TableDeleteByRowId
+
+pytestmark = [pytest.mark.python_plan, pytest.mark.python_read,
+ pytest.mark.python_write, pytest.mark.python_commit]
+_SCHEMA = pa.schema([('id', pa.int32()), ('v', pa.int32())])
+_ENGINES = [False, pytest.param(True, marks=pytest.mark.skipif(
+ not native_runtime_available(), reason='pypaimon-rust not installed'))]
+
+
[email protected]
+def table(tmp_path):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(_SCHEMA,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true', 'write.native.enabled': 'false',
+ 'commit.native.enabled': 'false', 'read.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false'}), False)
+ result = catalog.get_table('default.t')
+ _append(result, [0, 1, 2, 3])
+ return result
+
+
+def _append(table, ids):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': ids, 'v': ids}, schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _commit(table, messages):
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+
+def _ids(table):
+ read = table.new_read_builder().with_projection(['id'])
+ return
sorted(read.new_read().to_arrow(read.new_scan().plan().splits())['id'].to_pylist())
+
+
[email protected]('native', _ENGINES)
[email protected]('entry', ['direct', 'batch', 'stream'])
[email protected]('layout', ['sequence', 'slice', 'chunked'])
+def test_row_id_delete_inputs_are_idempotent(table, native, entry, layout):
+ worker_table = table.copy({'write.native.enabled': str(native).lower()})
+ row_ids = {
+ 'sequence': [1, 3, 1],
+ 'slice': pa.array([99, 1, 3, 1, 99], type=pa.uint32()).slice(1, 3),
+ 'chunked': pa.chunked_array([[], [1, 3], [], [1]], type=pa.int16()),
+ }[layout]
+ if entry == 'direct':
+ messages = TableDeleteByRowId(worker_table).delete(row_ids)
+ else:
+ builder = (worker_table.new_stream_write_builder() if entry == 'stream'
+ else worker_table.new_batch_write_builder())
+ update = builder.new_update()
+ messages = update.delete_by_row_id(row_ids, 7) if entry == 'stream'
else update.delete_by_row_id(row_ids)
+ assert messages and all(message.check_from_snapshot == 1 for message in
messages)
+ _commit(table, messages)
+ assert _ids(table) == [0, 2]
+ assert TableDeleteByRowId(worker_table).delete(row_ids) == []
+
+
[email protected]('native', _ENGINES)
[email protected]('row_ids,error', [
+ (pa.array([0.9]), TypeError),
+ (pa.array([True]), TypeError),
+ (pa.array([], type=pa.float64()), TypeError),
+ (pa.array([None]), TypeError),
+ (pa.array([0, None], type=pa.int64()), ValueError),
+ (pa.chunked_array([[0], [None]], type=pa.int16()), ValueError),
+ (pa.array([0, 1 << 63], type=pa.uint64()), pa.ArrowInvalid),
+ (pa.chunked_array([[0], [1 << 63]], type=pa.uint64()), pa.ArrowInvalid),
+])
+def test_invalid_arrow_ids_cannot_stage_a_partial_delete(table, native,
row_ids, error):
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ before = set(Path(table.table_path).rglob('*'))
+ update = table.new_batch_write_builder().new_update()
+ with pytest.raises(error):
+ update.delete_by_row_id(row_ids)
+ assert set(Path(table.table_path).rglob('*')) == before
+ assert _ids(table) == [0, 1, 2, 3]
+
+
[email protected]('native', _ENGINES)
+def test_captured_arrow_delete_preserves_other_anchors_and_new_appends(table,
native):
+ _append(table, [4, 5])
+ _commit(table, TableDeleteByRowId(table).delete([0]))
+ worker_table = table.copy({'write.native.enabled': str(native).lower()})
+ captured = capture_row_id_plan(worker_table)
+ captured = captured.for_split(next(split for start, _, split in
captured.groups if start == 4))
+ _append(table, [6])
+ with patch('pypaimon.read.table_scan.TableScan.plan_all',
side_effect=AssertionError('worker replanned')):
+ messages = TableDeleteByRowId(worker_table, captured).delete(
+ pa.chunked_array([[4], [], [4]], type=pa.int32()))
+ assert all(message.check_from_snapshot == 3 for message in messages)
+ assert sum(len(message.index_deletes) for message in messages) == 1
+ _commit(table, messages)
+ assert _ids(table) == [1, 2, 3, 5, 6]
+
+
[email protected]('row_ids', [pa.array([], type=pa.int32()),
pa.chunked_array([], type=pa.int64()), []])
+def test_empty_delete_keeps_native_execution_selected(table, row_ids):
+ from pypaimon.write import update_dispatch
+
+ class EmptyNativeDelete:
+ def delete_by_row_id(self, ids):
+ assert len(ids) == 0
+ return []
+
+ with patch.object(update_dispatch, 'prepare_native_write_operation',
return_value=EmptyNativeDelete()), \
+ patch.object(update_dispatch, 'create_python_update_operation',
+ side_effect=AssertionError('Python loaded')):
+ assert
table.new_batch_write_builder().new_update().delete_by_row_id(row_ids) == []
+
+
+def test_native_adapter_preserves_arrow_chunks_and_buffer_ownership(table):
+ original = pa.chunked_array([pa.array([99, 1, 3],
type=pa.int64()).slice(1),
+ pa.array([0], type=pa.int64())])
+ visited = []
+
+ class ArrowOnlyDelete:
+ def delete_by_row_id(self, ids):
+ assert isinstance(ids, pa.ChunkedArray)
+ assert ids.num_chunks == 2 and ids.chunk(0).offset == 1
+ assert [chunk.buffers()[1].address for chunk in ids.chunks] == [
+ chunk.buffers()[1].address for chunk in original.chunks]
+ visited.append(ids)
+ return []
+
+ assert NativeBatchTableUpdate(table,
ArrowOnlyDelete()).delete_by_row_id(original) == []
+ assert len(visited) == 1
+
+
+def test_sequence_callers_keep_integer_coercion():
+ assert normalize_row_ids([1, '2', 3.9, True]) == [1, 2, 3, 1]
+ with pytest.raises(ValueError, match='must not be null'):
+ normalize_row_ids([0, None])
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index e105329da0..7ba5342b7c 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -246,10 +246,10 @@ class FileStorePathFactory:
if not self.external_paths:
return None
- from pypaimon.execution.python.write.external_path_provider import
ExternalPathProvider
+ from pypaimon.write.backend_dispatch import
create_external_path_provider
relative_bucket_path = self.relative_bucket_path(partition, bucket,
canonical_partition)
- return ExternalPathProvider.create(
+ return create_external_path_provider(
self.external_path_strategy,
self.external_paths,
relative_bucket_path,
diff --git a/paimon-python/pypaimon/write/backend_dispatch.py
b/paimon-python/pypaimon/write/backend_dispatch.py
index 0181174f5a..ab773ac5af 100644
--- a/paimon-python/pypaimon/write/backend_dispatch.py
+++ b/paimon-python/pypaimon/write/backend_dispatch.py
@@ -459,6 +459,15 @@ def native_row_id_update_selected(worker):
return isinstance(execution, NativeTableUpdateByRowId)
+def validate_row_id_update_rows(worker, captured_plan, first_row_id, row_ids):
+ """Validate liveness in the selected executor before positional
rewriting."""
+ if native_row_id_update_selected(worker):
+ # Core validates the captured anchor's DV before staging any output.
+ return
+ from pypaimon.execution.python.write.row_id_plan import
validate_live_row_ids
+ validate_live_row_ids(worker.table, captured_plan, first_row_id, row_ids)
+
+
def python_commit(owner):
"""Create Python commit resources only after that backend is selected."""
if owner._python_commit is None:
@@ -554,3 +563,70 @@ def close_table_commit(owner):
callback.close()
except Exception:
pass
+
+
+def create_row_key_extractor(table, ignore_existing=False):
+ """Construct the public per-row extractor through the Python write
boundary."""
+ from pypaimon.table.bucket_mode import BucketMode
+ from pypaimon.execution.python.write.row_key_extractor import (
+ DynamicBucketRowKeyExtractor, FixedBucketRowKeyExtractor,
+ PostponeBucketRowKeyExtractor, UnawareBucketRowKeyExtractor,
+ )
+
+ bucket_mode = table.bucket_mode()
+ if bucket_mode == BucketMode.HASH_FIXED:
+ return FixedBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.BUCKET_UNAWARE:
+ return UnawareBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.POSTPONE_MODE:
+ return PostponeBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.HASH_DYNAMIC:
+ return DynamicBucketRowKeyExtractor(
+ table.table_schema, table=table, ignore_existing=ignore_existing)
+ elif bucket_mode == BucketMode.CROSS_PARTITION:
+ raise ValueError(
+ "CROSS_PARTITION primary-key writes require a persistent "
+ "global primary-key index, which PyPaimon does not yet support"
+ )
+ raise ValueError(f"Unsupported bucket mode: {bucket_mode}")
+
+
+def filter_merge_condition_batch(batch, condition, pre_rewritten=False):
+ """Execute the public Python condition helper without selecting a table
writer."""
+ from pypaimon.execution.python.write.merge_condition import filter_batch
+ return filter_batch(batch, condition, pre_rewritten)
+
+
+def apply_merge_condition(batch, rewritten, empty_schema):
+ from pypaimon.execution.python.write.merge_condition import apply_condition
+ return apply_condition(batch, rewritten, empty_schema)
+
+
+def create_external_path_provider(strategy, paths, relative_bucket_path,
weights):
+ from pypaimon.execution.python.write.external_path_provider import
ExternalPathProvider
+ return ExternalPathProvider.create(strategy, paths, relative_bucket_path,
weights)
+
+
+def execute_row_id_conflict_check(table, messages, latest_snapshot_id,
check_from_snapshot):
+ if table.options.native_commit_enabled():
+ from pypaimon.execution.native.write.native_commit import
prepare_native_row_id_conflict_check
+ # Unsupported configurations return None explicitly. Permission or
+ # transport failures must never restart the check in another backend.
+ native = prepare_native_row_id_conflict_check(table, messages)
+ if native is not None:
+ commit, native_messages = native
+ failure = None
+ try:
+ commit.check_row_id_conflicts(native_messages,
latest_snapshot_id, check_from_snapshot)
+ except BaseException as error:
+ failure = error
+ raise
+ finally:
+ try:
+ commit.close()
+ except BaseException:
+ if failure is None:
+ raise
+ return
+ from pypaimon.execution.python.write.commit.row_id_conflict import
check_row_id_conflicts as check
+ check(table, messages, latest_snapshot_id, check_from_snapshot)
diff --git a/paimon-python/pypaimon/write/row_id_conflict.py
b/paimon-python/pypaimon/write/row_id_conflict.py
index 61cda36f30..db4d8a264b 100644
--- a/paimon-python/pypaimon/write/row_id_conflict.py
+++ b/paimon-python/pypaimon/write/row_id_conflict.py
@@ -108,25 +108,5 @@ def check_row_id_conflicts(table, messages,
latest_snapshot_id, check_from_snaps
if (not baselines or invalid_baseline or min(baselines) !=
check_from_snapshot or missing_baseline):
raise ValueError('Row-ID conflict check baseline must match the commit
messages')
_authorize_update(table)
- if table.options.native_commit_enabled():
- from pypaimon.execution.native.write.native_commit import
prepare_native_row_id_conflict_check
- # Unsupported configurations return None explicitly. Permission or
- # transport failures must never restart the check in another backend.
- native = prepare_native_row_id_conflict_check(table, messages)
- if native is not None:
- commit, native_messages = native
- failure = None
- try:
- commit.check_row_id_conflicts(native_messages,
latest_snapshot_id, check_from_snapshot)
- except BaseException as error:
- failure = error
- raise
- finally:
- try:
- commit.close()
- except BaseException:
- if failure is None:
- raise
- return
- from pypaimon.execution.python.write.commit.row_id_conflict import
check_row_id_conflicts as check
- check(table, messages, latest_snapshot_id, check_from_snapshot)
+ from pypaimon.write.backend_dispatch import execute_row_id_conflict_check
+ execute_row_id_conflict_check(table, messages, latest_snapshot_id,
check_from_snapshot)
diff --git a/paimon-python/pypaimon/write/row_id_plan.py
b/paimon-python/pypaimon/write/row_id_plan.py
index 63de3f02d8..74fb5d08ad 100644
--- a/paimon-python/pypaimon/write/row_id_plan.py
+++ b/paimon-python/pypaimon/write/row_id_plan.py
@@ -103,9 +103,8 @@ def _row_id_array(row_ids):
"""Accept integer Arrow inputs without silently truncating floating IDs."""
if not isinstance(row_ids, (pa.Array, pa.ChunkedArray)) or not
pa.types.is_integer(row_ids.type):
raise TypeError('_ROW_ID input must be an integer Arrow array')
- if row_ids.null_count:
- raise ValueError('_ROW_ID is null; planner snapshot is stale or row
IDs come from a different table.')
- row_ids = row_ids.cast(pa.int64(), safe=True)
+ from pypaimon.write.row_input import normalize_row_ids
+ row_ids = normalize_row_ids(row_ids)
return row_ids.combine_chunks() if isinstance(row_ids, pa.ChunkedArray)
else row_ids
diff --git a/paimon-python/pypaimon/write/row_input.py
b/paimon-python/pypaimon/write/row_input.py
index 120abb3752..e0c4a702e8 100644
--- a/paimon-python/pypaimon/write/row_input.py
+++ b/paimon-python/pypaimon/write/row_input.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Sequence, Union
import pyarrow as pa
@@ -25,6 +25,30 @@ from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.row_utils import (inline_blob_value, require_columns,
row_to_named_values, value_for_arrow)
+RowIdInput = Union[Sequence[int], pa.Array, pa.ChunkedArray]
+
+
+def normalize_row_ids(row_ids: RowIdInput):
+ """Validate IDs without turning Arrow buffers into Python objects.
+
+ Keep chunk boundaries for Native transport. Only integer Arrow layouts
+ have a row-ID identity; floating, Boolean and untyped NULL arrays must not
+ be silently converted. Existing sequence callers retain integer coercion.
+ """
+ if isinstance(row_ids, (pa.Array, pa.ChunkedArray)):
+ if not pa.types.is_integer(row_ids.type):
+ raise TypeError('_ROW_ID input must be an integer Arrow array')
+ if row_ids.null_count:
+ raise ValueError('_ROW_ID value must not be null.')
+ return row_ids.cast(pa.int64(), safe=True)
+ values = []
+ for row_id in row_ids:
+ if row_id is None:
+ raise ValueError('_ROW_ID value must not be null.')
+ values.append(int(row_id))
+ return values
+
+
class RowIdUpdateInputMixin:
"""Normalize Python row inputs for either update execution backend."""
diff --git a/paimon-python/pypaimon/write/table_delete.py
b/paimon-python/pypaimon/write/table_delete.py
index c00bafaa81..544dbe0c52 100644
--- a/paimon-python/pypaimon/write/table_delete.py
+++ b/paimon-python/pypaimon/write/table_delete.py
@@ -15,9 +15,10 @@
# specific language governing permissions and limitations
# under the License.
-from typing import TYPE_CHECKING, List, Optional, Sequence
+from typing import TYPE_CHECKING, List, Optional
from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.row_input import RowIdInput, normalize_row_ids
if TYPE_CHECKING:
from pypaimon.execution.python.write.table_delete import _AnchorRangesInfo
@@ -40,8 +41,9 @@ class TableDeleteByRowId:
owner._execution = create_python_row_id_delete(table,
_precomputed_anchor_ranges)
return owner
- def delete(self, row_ids: Sequence[int]) -> List[CommitMessage]:
- return self._execution.delete(row_ids)
+ def delete(self, row_ids: RowIdInput) -> List[CommitMessage]:
+ """Delete sequence or integer Arrow IDs; duplicate IDs are
idempotent."""
+ return self._execution.delete(normalize_row_ids(row_ids))
def _snapshot_anchor_ranges(self):
# Ray's metadata coordinator selects _from_python explicitly. Native
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index a97863e989..aca41ec9e9 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -29,6 +29,7 @@ if TYPE_CHECKING:
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.row_input import RowIdInput
from pypaimon.write.commit_identifier import validate_commit_identifier
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -124,7 +125,7 @@ class TableUpdate:
from pypaimon.write.update_dispatch import TableUpdateOperations
return TableUpdateOperations(self)._delete_by_predicate(predicate,
commit_identifier)
- def _delete_by_row_id(self, row_ids: Sequence[int], commit_identifier:
int) -> List[CommitMessage]:
+ def _delete_by_row_id(self, row_ids: RowIdInput, commit_identifier: int)
-> List[CommitMessage]:
from pypaimon.write.update_dispatch import TableUpdateOperations
return TableUpdateOperations(self)._delete_by_row_id(row_ids,
commit_identifier)
@@ -194,12 +195,12 @@ class BatchTableUpdate(TableUpdate):
"""Delete rows matching ``predicate`` using deletion vectors."""
return self._delete_by_predicate(predicate, BATCH_COMMIT_IDENTIFIER)
- def delete_by_row_id(self, row_ids: Sequence[int]) -> List[CommitMessage]:
- """Delete rows by ``_ROW_ID`` using deletion vectors."""
+ def delete_by_row_id(self, row_ids: RowIdInput) -> List[CommitMessage]:
+ """Delete sequence or integer Arrow ``_ROW_ID`` values using deletion
vectors."""
return self._delete_by_row_id(row_ids, BATCH_COMMIT_IDENTIFIER)
def _delete_by_row_id(
- self, row_ids: Sequence[int], commit_identifier: int
+ self, row_ids: RowIdInput, commit_identifier: int
) -> List[CommitMessage]:
return super()._delete_by_row_id(row_ids, commit_identifier)
@@ -293,7 +294,7 @@ class StreamTableUpdate(TableUpdate):
def delete_by_row_id(
self,
- row_ids: Sequence[int],
+ row_ids: RowIdInput,
commit_identifier: int,
) -> List[CommitMessage]:
"""Delete rows by ``_ROW_ID`` using deletion vectors,
diff --git a/paimon-python/pypaimon/write/update_dispatch.py
b/paimon-python/pypaimon/write/update_dispatch.py
index 383192225c..40aa0a63d4 100644
--- a/paimon-python/pypaimon/write/update_dispatch.py
+++ b/paimon-python/pypaimon/write/update_dispatch.py
@@ -30,6 +30,7 @@ from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.row_utils import row_to_named_values
+from pypaimon.write.row_input import RowIdInput, normalize_row_ids
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
from pypaimon.write.table_upsert_by_key import TableUpsertByKey
from pypaimon.write.update_validation import _authorize_update
@@ -155,11 +156,10 @@ class TableUpdateOperations:
operation = create_python_update_operation(context,
commit_identifier, predicate)
return operation.delete()
- def _delete_by_row_id(self, row_ids: Sequence[int], commit_identifier:
int) -> List[CommitMessage]:
+ def _delete_by_row_id(self, row_ids: RowIdInput, commit_identifier: int)
-> List[CommitMessage]:
context = self._context
- operation = None
- if len(row_ids):
- operation = prepare_native_write_operation('delete',
context.table, context.commit_user)
+ row_ids = normalize_row_ids(row_ids)
+ operation = prepare_native_write_operation('delete', context.table,
context.commit_user)
if operation is None:
operation = create_python_update_operation(context,
commit_identifier)
return operation.delete_by_row_id(row_ids)
