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 5d02e09dbd [Python] Enable native primary-key BLOB limits and floating
sequence fields (#10410)
5d02e09dbd is described below
commit 5d02e09dbd3f140520fd7c33f33765118f5b098d
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Oct 7 11:11:05 2026 +0800
[Python] Enable native primary-key BLOB limits and floating sequence fields
(#10410)
---
paimon-python/pypaimon/read/table_read.py | 42 ++-
.../pypaimon/tests/native_pk_blob_limit_test.py | 283 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_read_test.py | 76 ++++--
.../pypaimon/tests/native_sequence_test.py | 1 +
paimon-python/pypaimon/tests/native_write_test.py | 4 +-
paimon-python/pypaimon/write/native_write.py | 4 -
6 files changed, 352 insertions(+), 58 deletions(-)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 551d0300be..ee8f8e9f8d 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -43,7 +43,7 @@ from pypaimon.read.split_read import (DataEvolutionSplitRead,
MergeFileSplitRead, RawFileSplitRead,
SplitRead, deferred_blob_field_names)
from pypaimon.schema.data_types import (
- DataField, MapType, PyarrowFieldParser, is_map_blob_type)
+ DataField, MapType, PyarrowFieldParser, is_blob_file_field,
is_map_blob_type)
from pypaimon.table.row.offset_row import OffsetRow
from pypaimon.utils.arrow_utils import zero_column_batch
@@ -254,13 +254,12 @@ class TableRead:
table reads do not guarantee row order. Python fallback reads remain
serial.
"""
- # Cleanup ownership follows the uncapped concurrency decision. LIMIT
- # may reduce the actual native worker count to one, but the returned
- # PyArrow reader still does not close its suspended batch iterator.
+ # PyArrow does not close a suspended Python batch iterator. Keep its
+ # cleanup even when LIMIT or BLOB pruning serializes native reads.
effective = self._resolve_parallelism(parallelism, len(splits))
reader, batch_iterator = self._new_arrow_batch_reader(
splits, blob_parallelism, parallelism)
- if self._should_run_parallel(splits, effective):
+ if self._requires_batch_iterator_cleanup(splits, effective):
return _ClosableArrowBatchReader(reader, batch_iterator)
return reader
@@ -280,7 +279,7 @@ class TableRead:
effective = self._resolve_parallelism(parallelism, len(splits))
reader, batch_iterator = self._new_arrow_batch_reader(
splits, blob_parallelism, parallelism)
- if self._should_run_parallel(splits, effective):
+ if self._requires_batch_iterator_cleanup(splits, effective):
return _ClosableArrowBatchReader(reader, batch_iterator)
if (_RECORD_BATCH_READER_FROM_STREAM is not None
and hasattr(reader, "close")):
@@ -377,9 +376,9 @@ class TableRead:
``>= 1``. Note that with ``>= 2`` (or auto)
and a ``limit`` set, the returned rows are an arbitrary
subset of the requested size, since which splits fill the row
- quota first is non-deterministic. Data-evolution reads with
- deferred BLOB resolution falls back to the Python reader when
- a limit may discard rows, so discarded payloads are not read.
+ quota first is non-deterministic. Reads with BLOB payloads run
+ serially when a limit may discard rows, so discarded payloads
+ are not fetched by speculative split workers.
blob_parallelism: maximum concurrent blob range reads within each
split reader. ``None`` or ``1`` (default) reads blobs serially;
``>= 2`` enables concurrent ranged reads. On the
@@ -498,22 +497,9 @@ class TableRead:
"the table file format is not supported")
if not splits:
return []
- sequence_fields = self.table.options.sequence_field()
- if self.table.is_primary_key_table and sequence_fields:
- sequence_schema = PyarrowFieldParser.from_paimon_schema(
- [self.table.field_dict[name] for name in sequence_fields])
- # Native merge cannot extract floating sequence values: they
- # would silently become missing sequence values.
- if any(pyarrow.types.is_floating(field.type) for field in
sequence_schema):
- return self._native_fallback(
- "floating sequence fields are not supported")
if not self._native_blob_view_supported():
return self._native_fallback(
"the table uses an unsupported BLOB view")
- if (self._deferred_blob_limit_may_prune(splits)
- and not self.table.options.data_evolution_enabled()):
- return self._native_fallback(
- "deferred BLOB LIMIT pruning requires the Python reader")
# Query authorization has additional filtering, masking and projection
# semantics which are already implemented by the Python reader.
if any(isinstance(split, QueryAuthSplit) for split in splits):
@@ -982,9 +968,13 @@ class TableRead:
def _deferred_blob_limit_may_prune(self, splits: List[Split]) -> bool:
return (self.limit is not None
and (self._deferred_blob_fields
- or self._native_inline_blob_fields())
+ or self._native_resolved_blob_fields())
and not self._limit_covers_all_splits(splits))
+ def _requires_batch_iterator_cleanup(self, splits: List[Split], effective:
int) -> bool:
+ return (self._should_run_parallel(splits, effective)
+ or self._deferred_blob_limit_may_prune(splits))
+
def _native_blob_view_supported(self) -> bool:
"""Use native view resolution only with the REST catalog
environment."""
read_names = {field.name for field in self._scan_read_type}
@@ -999,11 +989,13 @@ class TableRead:
from pypaimon.read.native_plan import _catalog_metastore
return _catalog_metastore(loader) == 'rest'
- def _native_inline_blob_fields(self) -> set:
- """Return configured BLOB fields that native reads resolve eagerly."""
+ def _native_resolved_blob_fields(self) -> set:
+ """BLOB fields Rust resolves before emitting a batch."""
options = self.table.options
if options.blob_as_descriptor():
return set()
+ if self.table.is_primary_key_table:
+ return {field.name for field in self._scan_read_type if
is_blob_file_field(field)}
inline_fields = (
options.blob_descriptor_fields()
| options.blob_view_fields()
diff --git a/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
b/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
new file mode 100644
index 0000000000..c3a49b79d0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_pk_blob_limit_test.py
@@ -0,0 +1,283 @@
+# 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.
+
+"""PK BLOB filtering and LIMIT must leave unselected payloads unopened."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory
+from pypaimon.read.table_read import TableRead
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.row.blob import BlobDescriptor
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, engine='deduplicate', batch_size=8192, partitioned=False):
+ from pypaimon_rust.datafusion import SQLContext
+
+ options = {'bucket': '1', 'merge-engine': engine, 'read.batch-size':
str(batch_size),
+ 'blob-descriptor-field': 'payload,output',
'read.native.enabled': 'true',
+ 'write.native.enabled': 'true', 'scan.native-plan.enabled':
'true'}
+ if engine == 'aggregation':
+ options['fields.default-aggregate-function'] = 'last_value'
+ ctx = SQLContext()
+ ctx.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ ctx.sql('CREATE SCHEMA paimon.pkblob')
+ columns = 'id INT, tag INT, payload BLOB, output BLOB'
+ key, partition = 'PRIMARY KEY (id)', ''
+ if partitioned:
+ columns += ', pt STRING'
+ key, partition = 'PRIMARY KEY (pt, id)', ' PARTITIONED BY (pt)'
+ properties = ', '.join("'%s' = '%s'" % item for item in options.items())
+ ctx.sql('CREATE TABLE paimon.pkblob.t (' + columns + ', ' + key + ')' +
partition
+ + ' WITH (' + properties + ')')
+ table = CatalogFactory.create({'warehouse':
str(tmp_path)}).get_table('pkblob.t')
+ if engine == 'first-row':
+ # Java pk clustering permits first-row DVs. Loaded options let the
+ # reader consume level-0 files directly, without requiring compaction.
+ table = table.copy({'pk-clustering-override': 'true',
'deletion-vectors.enabled': 'true',
+ 'deletion-vectors.merge-on-read': 'true'})
+ return table
+
+
+def _descriptor(tmp_path, name, value=None):
+ file = tmp_path / name
+ if value is not None:
+ file.write_bytes(value)
+ return BlobDescriptor(file.as_uri(), 0, len(value) if value is not None
else 8).serialize()
+
+
+def _write(table, rows):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.Table.from_pylist(
+ rows, schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ commit.commit(writer.prepare_commit())
+ assert writer._python_writer is None
+ finally:
+ writer.close()
+ commit.close()
+
+
+def _read(table, projection, predicate=None, limit=None, parallelism=1,
streaming=False, splits=None):
+ builder = table.new_read_builder().with_projection(projection)
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if limit is not None:
+ builder.with_limit(limit)
+ splits = builder.new_scan().plan().splits() if splits is None else splits
+ reader = builder.new_read()
+ with patch.object(reader, '_create_split_read',
side_effect=AssertionError('Python read fallback')):
+ if streaming:
+ with reader.to_arrow_batch_reader(splits, parallelism=parallelism,
blob_parallelism=2) as batches:
+ return pa.Table.from_batches(list(batches),
schema=batches.schema).to_pylist()
+ return reader.to_arrow(splits, parallelism=parallelism,
blob_parallelism=2).to_pylist()
+
+
[email protected]('engine', ['deduplicate', 'first-row',
'partial-update', 'aggregation'])
[email protected]('batch_size', [1, 8192])
[email protected]('payload_filter', [False, True])
+def test_pk_blob_limit_stops_before_later_invalid_uri(tmp_path, engine,
batch_size, payload_filter):
+ table = _table(tmp_path, engine, batch_size)
+ payload = _descriptor(tmp_path, 'payload', b'selected')
+ output = _descriptor(tmp_path, 'output', b'returned')
+ missing = _descriptor(tmp_path, 'missing')
+ _write(table, [dict(id=1, tag=1, payload=payload, output=output),
+ dict(id=2, tag=1, payload=missing, output=missing)])
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.equal('payload', b'selected') if payload_filter else None
+ assert _read(table, ['id', 'payload', 'output'], predicate, 1) == [
+ dict(id=1, payload=b'selected', output=b'returned')]
+
+
[email protected]('limit', [None, 1])
[email protected]('streaming', [False, True])
+def test_pk_output_blob_is_resolved_only_for_matching_rows(tmp_path, limit,
streaming):
+ table = _table(tmp_path)
+ rejected = _descriptor(tmp_path, 'rejected', b'no')
+ selected = _descriptor(tmp_path, 'selected', b'yes')
+ output = _descriptor(tmp_path, 'output', b'returned')
+ missing = _descriptor(tmp_path, 'missing')
+ _write(table, [dict(id=1, tag=1, payload=rejected, output=missing),
+ dict(id=2, tag=1, payload=selected, output=output)])
+ pb = table.new_read_builder().new_predicate_builder()
+ assert _read(table, ['output', 'id'], pb.equal('payload', b'yes'), limit,
+ streaming=streaming) == [dict(output=b'returned', id=2)]
+
+
+def test_pk_ordinary_conjunct_filters_before_payload_access(tmp_path):
+ table = _table(tmp_path)
+ payload = _descriptor(tmp_path, 'payload', b'yes')
+ missing = _descriptor(tmp_path, 'missing')
+ _write(table, [dict(id=1, tag=0, payload=missing, output=missing),
+ dict(id=2, tag=1, payload=payload, output=payload)])
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.and_predicates([pb.equal('tag', 1), pb.equal('payload',
b'yes')])
+ assert _read(table, ['id'], predicate, 1) == [dict(id=2)]
+
+
[email protected]('batch_size', [1, 8192])
+def test_pk_blob_limit_shrinks_candidate_batches_after_rejected_rows(tmp_path,
batch_size):
+ table = _table(tmp_path, batch_size=batch_size)
+ missing = _descriptor(tmp_path, 'must-not-open-after-second-match')
+ rejected = _descriptor(tmp_path, 'rejected', b'no')
+ selected = _descriptor(tmp_path, 'selected', b'yes')
+ _write(table, [dict(id=1, tag=1, payload=rejected, output=missing),
+ dict(id=2, tag=1, payload=selected, output=selected),
+ dict(id=3, tag=1, payload=selected, output=selected),
+ dict(id=4, tag=1, payload=missing, output=missing)])
+ pb = table.new_read_builder().new_predicate_builder()
+ assert _read(table, ['id', 'output'], pb.equal('payload', b'yes'), 2) == [
+ dict(id=2, output=b'yes'), dict(id=3, output=b'yes')]
+
+
[email protected]('limit', [None, 1, 2])
+def test_pk_blob_or_short_circuits_payload_access(tmp_path, limit):
+ table = _table(tmp_path)
+ missing = _descriptor(tmp_path, 'unneeded-branch')
+ selected = _descriptor(tmp_path, 'selected', b'yes')
+ rejected = _descriptor(tmp_path, 'rejected', b'no')
+ _write(table, [dict(id=1, tag=1, payload=missing, output=missing),
+ dict(id=2, tag=0, payload=selected, output=missing),
+ dict(id=3, tag=0, payload=rejected, output=missing)])
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.or_predicates([pb.equal('tag', 1), pb.equal('payload',
b'yes')])
+ assert _read(table, ['id'], predicate, limit) == [dict(id=1),
dict(id=2)][:limit]
+
+
+def test_pk_blob_nested_boolean_branches_only_resolve_pending_rows(tmp_path):
+ table = _table(tmp_path)
+ missing = _descriptor(tmp_path, 'unneeded-branch')
+ selected = _descriptor(tmp_path, 'selected', b'yes')
+ _write(table, [dict(id=1, tag=1, payload=missing, output=missing),
+ dict(id=2, tag=0, payload=selected, output=missing)])
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.or_predicates([
+ pb.and_predicates([pb.equal('tag', 1), pb.equal('id', 1)]),
+ pb.and_predicates([pb.equal('payload', b'yes'), pb.equal('payload',
b'yes')])])
+ assert _read(table, ['id'], predicate) == [dict(id=1), dict(id=2)]
+
+
+def test_pk_blob_predicate_payload_is_reused_for_output(tmp_path):
+ table = _table(tmp_path)
+ inner_descriptor = _descriptor(tmp_path, 'must-not-resolve-returned-bytes')
+ payload = _descriptor(tmp_path, 'outer', inner_descriptor)
+ _write(table, [dict(id=1, tag=1, payload=payload, output=payload)])
+ pb = table.new_read_builder().new_predicate_builder()
+ assert _read(table, ['payload'], pb.equal('payload', inner_descriptor)) ==
[
+ dict(payload=inner_descriptor)]
+
+
[email protected]('limit', [None, 1])
[email protected]('not_null', [False, True])
+def test_pk_blob_null_predicates_do_not_open_payload(tmp_path, limit,
not_null):
+ table = _table(tmp_path)
+ missing = _descriptor(tmp_path, 'must-not-read-for-null-check')
+ _write(table, [dict(id=1, tag=1, payload=missing, output=missing),
+ dict(id=2, tag=1, payload=None, output=None)])
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.is_not_null('payload') if not_null else
pb.is_null('payload')
+ assert _read(table, ['id'], predicate, limit) == [dict(id=1 if not_null
else 2)]
+
+
[email protected]('engine', ['deduplicate', 'first-row',
'partial-update', 'aggregation'])
+def test_pk_blob_limit_and_payload_filter_run_after_merge(tmp_path, engine):
+ table = _table(tmp_path, engine)
+ good = _descriptor(tmp_path, 'good', b'kept')
+ missing = _descriptor(tmp_path, 'discarded-version')
+ older, newer = (good, missing) if engine == 'first-row' else (missing,
good)
+ for reference in (older, newer):
+ _write(table, [dict(id=1, tag=1, payload=reference, output=reference)])
+ pb = table.new_read_builder().new_predicate_builder()
+ assert _read(table, ['id', 'output'], pb.equal('output', b'kept'), 1) == [
+ dict(id=1, output=b'kept')]
+
+
[email protected]('parallelism', [None, 1, 4])
[email protected]('streaming', [False, True])
+def test_pk_blob_limit_is_shared_across_partitions(tmp_path, parallelism,
streaming):
+ table = _table(tmp_path, partitioned=True)
+ payload = _descriptor(tmp_path, 'payload', b'yes')
+ missing = _descriptor(tmp_path, 'missing')
+ _write(table, [dict(id=1, tag=1, payload=payload, output=payload, pt='a'),
+ dict(id=2, tag=1, payload=missing, output=missing, pt='b')])
+ splits = sorted(table.new_read_builder().new_scan().plan().splits(),
+ key=lambda split: split.files[0].file_path)
+ assert len(splits) == 2
+ assert _read(table, ['id', 'payload'], limit=1, parallelism=parallelism,
+ streaming=streaming, splits=splits) == [dict(id=1,
payload=b'yes')]
+
+
+def test_pk_blob_zero_limit_and_descriptor_output_require_no_payload(tmp_path):
+ table = _table(tmp_path)
+ missing = _descriptor(tmp_path, 'missing')
+ _write(table, [dict(id=1, tag=1, payload=missing, output=missing)])
+ splits = table.new_read_builder().new_scan().plan().splits()
+ assert _read(table, ['payload'], limit=0, splits=splits) == []
+ assert _read(table.copy({'blob-as-descriptor': 'true'}), ['payload'],
limit=1) == [dict(payload=missing)]
+ with pytest.raises(ValueError, match='missing'):
+ _read(table, ['payload'], limit=1)
+
+
[email protected]('payload_filter', [False, True])
+def test_streaming_pk_blob_row_kinds_resolve_only_selected_payload(tmp_path,
payload_filter):
+ table = _table(tmp_path)
+ payload = _descriptor(tmp_path, 'payload', b'selected')
+ missing = _descriptor(tmp_path, 'unselected')
+ _write(table, [dict(id=1, tag=1, payload=payload, output=payload),
+ dict(id=2, tag=1, payload=missing, output=missing)])
+ builder = table.new_stream_read_builder().with_projection(['id',
'payload'])
+ scan = builder.new_streaming_scan()
+ splits =
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
+ assert splits and all(split.is_streaming for split in splits)
+ predicate = builder.new_predicate_builder().equal('payload', b'selected')
if payload_filter else None
+ reader = TableRead(table, predicate, builder.read_type(),
include_row_kind=True, limit=1)
+ with patch.object(reader, '_create_split_read',
side_effect=AssertionError('Python read fallback')):
+ assert reader.to_arrow(splits).to_pylist() == [
+ {'_row_kind': '+I', 'id': 1, 'payload': b'selected'}]
+
+
+def
test_pk_row_kind_limit_is_global_across_snapshot_and_delta_splits(tmp_path):
+ from pypaimon.read.native_plan import native_read, native_split_from_python
+
+ table = _table(tmp_path, partitioned=True)
+ payload = _descriptor(tmp_path, 'payload', b'selected')
+ missing = _descriptor(tmp_path, 'unselected-delta')
+ _write(table, [dict(id=1, tag=1, payload=payload, output=payload, pt='a')])
+ splits = table.new_read_builder().new_scan().plan().splits()
+ _write(table, [dict(id=2, tag=1, payload=missing, output=missing, pt='b')])
+ scan = table.new_stream_read_builder().new_streaming_scan()
+ splits +=
scan._create_delta_plan(table.snapshot_manager().get_latest_snapshot()).splits()
+ assert len(splits) == 2 and not splits[0].is_streaming and
splits[1].is_streaming
+ # Exercise the Rust quota directly: TableRead's outer truncation must not
+ # conceal extra rows or payload reads in the core row-kind path.
+ fields = [field for field in table.fields if field.name in ('id',
'payload')]
+ rust_splits = [native_split_from_python(split) for split in splits]
+ reader = native_read(table, rust_splits, read_type=fields,
include_row_kind=True, limit=1)
+ try:
+ rows = [row for batch in reader for row in batch.to_pylist()]
+ finally:
+ reader.close()
+ assert len(rows) == 1
+ assert rows[0]['id'] == 1 and rows[0]['payload'] == b'selected'
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index b50c9c15e4..46a568515f 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -38,6 +38,7 @@ class _Split:
def _table_read(limit=None):
read = TableRead.__new__(TableRead)
read.table = Mock()
+ read.table.is_primary_key_table = False
read.table.options.native_read_enabled.return_value = True
read.table.options.sequence_field.return_value = []
read.table.options.file_format.return_value = 'parquet'
@@ -160,16 +161,17 @@ def
test_variant_fields_never_silently_falls_back_to_python():
@pytest.mark.parametrize('type_', ['FLOAT', 'DOUBLE'])
-def test_floating_sequence_falls_back_before_native_read(type_):
+def test_floating_sequence_uses_native_read(type_):
read = _table_read()
read.table.is_primary_key_table = True
read.table.options.sequence_field.return_value = ['seq']
read.table.field_dict = {'seq': DataField(1, 'seq', AtomicType(type_))}
split = _Split()
split._native_split = object()
- with patch('pypaimon.read.native_plan.native_read') as native:
- assert read._try_native_batches([split], pa.schema([('id',
pa.int32())])) is None
- native.assert_not_called()
+ batch = _id_batch([1])
+ with patch('pypaimon.read.native_plan.native_read', return_value=[batch])
as native:
+ assert list(read._try_native_batches([split], pa.schema([('id',
pa.int32())]))) == [batch]
+ native.assert_called_once()
def test_native_read_consumes_retained_rust_splits_and_enforces_limit():
@@ -678,6 +680,37 @@ def
test_native_read_limit_close_reaches_capped_native_reader():
assert native_reader.closed
[email protected]('managed_reader', [False, True])
+def
test_pruning_pk_blob_limit_close_reaches_serial_native_reader(managed_reader):
+ read = _blob_table_read(limit=2)
+ read.table.is_primary_key_table = True
+ splits = [_Split(), _Split()]
+ for index, split in enumerate(splits):
+ split._native_split = index
+ split.merged_row_count = Mock(return_value=None)
+
+ class CloseTrackingReader:
+ closed = False
+
+ def __iter__(self):
+ return self
+
+ def __next__(self):
+ return pa.record_batch([pa.array([b'payload'],
type=pa.large_binary())], names=['payload'])
+
+ def close(self):
+ self.closed = True
+
+ native_reader = CloseTrackingReader()
+ method = read._to_managed_arrow_batch_reader if managed_reader else
read.to_arrow_batch_reader
+ with patch('pypaimon.read.native_plan.native_read',
return_value=native_reader):
+ reader = method(splits, parallelism=4)
+ assert reader.read_next_batch().column('payload').to_pylist() ==
[b'payload']
+ assert not native_reader.closed
+ reader.close()
+ assert native_reader.closed
+
+
def test_filtered_native_read_limit_keeps_split_parallelism():
read = _table_read(limit=1)
read.predicate = Mock()
@@ -965,35 +998,24 @@ def
test_native_read_preserves_parallel_large_binary_blob_schema():
assert result.schema == pa.schema([('payload', pa.large_binary())])
-def test_native_read_defers_to_python_for_pruning_blob_limit():
- read = _blob_table_read(limit=1)
- read._deferred_blob_fields = {'payload'}
- split = _Split('payload.blob')
- split._native_split = object()
- split.merged_row_count = Mock(return_value=2)
-
- with patch(
- 'pypaimon.read.native_plan.native_read') as native:
- assert read._try_native_batches(
- [split], pa.schema([('payload', pa.large_binary())]),
- blob_parallelism=1) is None
-
- native.assert_not_called()
-
-
-def test_native_read_defers_to_python_for_pruning_descriptor_blob_limit():
[email protected]('descriptor', [False, True])
+def test_native_primary_key_blob_limit_passes_quota_to_rust(descriptor):
read = _blob_table_read(limit=1)
- read.table.options.blob_descriptor_fields.return_value = {'payload'}
+ read.table.is_primary_key_table = True
+ if descriptor:
+ read.table.options.blob_descriptor_fields.return_value = {'payload'}
split = _Split('payload.parquet')
split._native_split = object()
split.merged_row_count = Mock(return_value=2)
- with patch('pypaimon.read.native_plan.native_read') as native:
- assert read._try_native_batches(
+ batch = pa.record_batch([pa.array([b'first'], type=pa.large_binary())],
names=['payload'])
+ with patch('pypaimon.read.native_plan.native_read', return_value=[batch])
as native:
+ assert list(read._try_native_batches(
[split], pa.schema([('payload', pa.large_binary())]),
- blob_parallelism=1) is None
-
- native.assert_not_called()
+ blob_parallelism=1)) == [batch]
+ native.assert_called_once()
+ assert native.call_args.kwargs['limit'] == 1
+ assert not read._should_run_parallel([split, split], 2)
@pytest.mark.parametrize('descriptor', [False, True])
diff --git a/paimon-python/pypaimon/tests/native_sequence_test.py
b/paimon-python/pypaimon/tests/native_sequence_test.py
index 20058bcc12..aa29c8f22b 100644
--- a/paimon-python/pypaimon/tests/native_sequence_test.py
+++ b/paimon-python/pypaimon/tests/native_sequence_test.py
@@ -98,6 +98,7 @@ def _read(table, native, predicate=None, projection=None):
@pytest.mark.parametrize('engine', ['deduplicate', 'partial-update',
'aggregation'])
@pytest.mark.parametrize('sequence_type,high,low', [
(pa.int64(), 100, 50), (pa.string(), 'z', 'a'), (pa.binary(), b'z', b'a'),
+ (pa.float32(), 100.5, 50.25), (pa.float64(), 100.5, 50.25),
(pa.decimal128(10, 2), Decimal('100.50'), Decimal('50.25')),
(pa.timestamp('us'), datetime.datetime(2020, 1, 2),
datetime.datetime(2020, 1, 1)),
(pa.time32('ms'), datetime.time(12, 0), datetime.time(1, 0)),
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 8403d06c4d..85145cfa52 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -80,8 +80,8 @@ def
test_sequence_validation_precedes_native_selection(tmp_path, streaming, sequ
@pytest.mark.parametrize('type_,order,supported', [
(pa.int64(), 'ascending', True),
(pa.int64(), 'descending', True),
- (pa.float32(), 'ascending', False),
- (pa.float64(), 'ascending', False),
+ (pa.float32(), 'ascending', True),
+ (pa.float64(), 'ascending', True),
])
def test_native_sequence_write_capabilities(tmp_path, type_, order, supported):
from pypaimon.write.native_write import create_native_write
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index be04e4f9bd..43cf8e40c2 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -51,10 +51,6 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
*, fixed_bucket=False, bucket_plan=None):
"""Return a native writer if the table can use the filesystem write
path."""
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
- sequence_fields = table.options.sequence_field()
- if (table.is_primary_key_table and any(
- pa.types.is_floating(schema.field(name).type) for name in
sequence_fields)):
- return None
if (not native_write_available()
or table.bucket_mode() not in (BucketMode.HASH_FIXED,
BucketMode.HASH_DYNAMIC,
BucketMode.BUCKET_UNAWARE,
