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 66e2530a73 [python] Honor sequence.field in primary-key writes (#10166)
66e2530a73 is described below
commit 66e2530a73b0c8185460c8ebb5b216fac2b47fed
Author: chaoyang <[email protected]>
AuthorDate: Fri Sep 25 21:29:50 2026 +0800
[python] Honor sequence.field in primary-key writes (#10166)
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 67 ++++++
paimon-python/dev/run_mixed_tests.sh | 4 +-
.../pypaimon/read/merge_engine_support.py | 25 ++-
.../pypaimon/read/reader/sort_merge_reader.py | 24 +-
paimon-python/pypaimon/read/table_read.py | 8 +
.../pypaimon/tests/e2e/java_py_read_write_test.py | 49 ++++
paimon-python/pypaimon/tests/native_read_test.py | 14 ++
paimon-python/pypaimon/tests/native_write_test.py | 42 +++-
.../pypaimon/tests/test_aggregation_e2e.py | 8 +-
.../pypaimon/tests/test_sequence_field_read.py | 29 ++-
.../pypaimon/tests/test_sequence_field_write.py | 249 +++++++++++++++++++++
.../pypaimon/tests/test_write_merge_buffer.py | 3 +
paimon-python/pypaimon/write/file_store_write.py | 5 +
paimon-python/pypaimon/write/native_write.py | 7 +
paimon-python/pypaimon/write/write_builder.py | 5 +
.../pypaimon/write/writer/key_value_data_writer.py | 39 +++-
16 files changed, 542 insertions(+), 36 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 8a59c43326..e48bf52b77 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -578,6 +578,73 @@ public class JavaPyE2ETest {
}
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testReadPythonFloatingSequence() throws Exception {
+ for (String type : Arrays.asList("float", "double")) {
+ for (String order : Arrays.asList("ascending", "descending")) {
+ for (String grouping : Arrays.asList("batch", "commits")) {
+ FileStoreTable table =
+ (FileStoreTable)
+ catalog.getTable(
+ identifier(
+ "floating_sequence_"
+ + type
+ + "_"
+ + order
+ + "_"
+ + grouping));
+ List<String> result =
+ getResult(
+ table.newRead(),
+ table.newScan().plan().splits(),
+ row -> {
+ String sequence =
+ row.isNullAt(1)
+ ? "null"
+ : "float".equals(type)
+ ?
Float.toString(row.getFloat(1))
+ :
Double.toString(row.getDouble(1));
+ return row.getInt(0)
+ + ":"
+ + row.getString(2)
+ + ":"
+ + sequence;
+ });
+ List<String> expected =
+ new ArrayList<>(
+ Arrays.asList(
+ "5:nan:NaN",
+ "6:nan:NaN",
+ "8:nan-last:NaN",
+ "10:null-last:null"));
+ if ("ascending".equals(order)) {
+ expected.addAll(
+ Arrays.asList(
+ "1:nan:NaN",
+ "2:nan:NaN",
+ "3:positive-zero:0.0",
+ "4:positive-zero:0.0",
+ "7:nan:NaN",
+ "9:finite:1.0"));
+ } else {
+ expected.addAll(
+ Arrays.asList(
+ "1:finite:1.0",
+ "2:finite:1.0",
+ "3:negative-zero:-0.0",
+ "4:negative-zero:-0.0",
+ "7:infinity:Infinity",
+ "9:negative-infinity:-Infinity"));
+ }
+ assertThat(result)
+ .as("%s sequence, %s, Python %s", type, order,
grouping)
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+ }
+ }
+
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testReadPythonDynamicBucketHashIndex() throws Exception {
diff --git a/paimon-python/dev/run_mixed_tests.sh
b/paimon-python/dev/run_mixed_tests.sh
index 319267b663..fc84202a0d 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -244,7 +244,7 @@ run_python_write_test() {
# Run the parameterized Python test method for writing data (pk table,
includes bucket num assertion)
echo "Running Python test for JavaPyReadWriteTest
(test_py_write_read_pk_table)..."
- if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k
"test_py_write_read_pk_table or test_py_write_dynamic_bucket_hash_index" -v;
then
+ if python -m pytest java_py_read_write_test.py::JavaPyReadWriteTest -k
"test_py_write_read_pk_table or test_py_write_dynamic_bucket_hash_index or
test_py_write_floating_sequence" -v; then
echo -e "${GREEN}✓ Python write test completed successfully${NC}"
return 0
else
@@ -263,7 +263,7 @@ run_java_read_test() {
echo "Running Maven test for JavaPyE2ETest.testReadPkTable (Java Read
Parquet/Orc/Avro)..."
echo "Note: Maven may download dependencies on first run, this may take a
while..."
local parquet_result=0
- if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadPythonDynamicBucketHashIndex
-pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
+ if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testReadPkTable+testReadPythonDynamicBucketHashIndex+testReadPythonFloatingSequence
-pl paimon-core -Drun.e2e.tests=true -Dpython.version="$PYTHON_VERSION"; then
echo -e "${GREEN}✓ Java read Parquet/Orc/Avro test completed
successfully${NC}"
else
echo -e "${RED}✗ Java read Parquet/Orc/Avro test failed${NC}"
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index a7d8337854..51025a0df5 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -156,20 +156,14 @@ def check_sequence_field_valid(table) -> None:
)
-def check_supported(table) -> None:
- """Raise ``NotImplementedError`` if the table's merge-engine
- configuration is outside what pypaimon's read path implements, or
- ``ValueError`` if it is an outright-invalid configuration that Java
- rejects at schema validation.
-
- Non-PK tables are always fine (no merge function involved).
- """
+def check_sequence_field_supported(table) -> None:
+ """Validate sequence configuration and types before reading or writing."""
if not table.is_primary_key_table:
return
# ``sequence.field`` validity is engine-independent in Java
# (SchemaValidation.validateSequenceField). pypaimon has no
# schema-creation validation, so enforce the same invariants here on
- # the read path, before per-engine dispatch.
+ # both paths, before per-engine dispatch.
check_sequence_field_valid(table)
# ``sequence.field`` validity (above) is Java-aligned and engine
# independent. Some field *types* are valid in Java but unimplemented in
@@ -186,6 +180,19 @@ def check_supported(table) -> None:
"handled by Java via RecordComparator) and VARIANT are not "
"supported. Open an issue to track support.".format(
", ".join(sorted(unsupported_seq))))
+
+
+def check_supported(table) -> None:
+ """Raise ``NotImplementedError`` if the table's merge-engine
+ configuration is outside what pypaimon's read path implements, or
+ ``ValueError`` if it is an outright-invalid configuration that Java
+ rejects at schema validation.
+
+ Non-PK tables are always fine (no merge function involved).
+ """
+ if not table.is_primary_key_table:
+ return
+ check_sequence_field_supported(table)
engine = table.options.merge_engine()
if engine == MergeEngine.DEDUPLICATE:
return
diff --git a/paimon-python/pypaimon/read/reader/sort_merge_reader.py
b/paimon-python/pypaimon/read/reader/sort_merge_reader.py
index b1965b2795..0a1479feff 100644
--- a/paimon-python/pypaimon/read/reader/sort_merge_reader.py
+++ b/paimon-python/pypaimon/read/reader/sort_merge_reader.py
@@ -16,6 +16,7 @@
# under the License.
import heapq
+import math
from typing import Any, Callable, List, Optional
from pypaimon.read.reader.deduplicate_merge_function import \
@@ -211,7 +212,8 @@ def is_comparable_seq_field(field: DataField) -> bool:
def _row_field_comparator(
fields: List[DataField],
indices: List[int],
- ascending: bool = True) -> Callable[[Any, Any], int]:
+ ascending: bool = True,
+ floating_sequence: bool = False) -> Callable[[Any, Any], int]:
"""Build a comparator over two rows on the given ``indices`` (positions
in ``fields`` / the row's ``get_field``), compared left-to-right.
@@ -224,8 +226,13 @@ def _row_field_comparator(
``nullIsLast=false`` (see ``CodeGeneratorImpl#getSortSpec``), where
descending order flips only the non-null value comparison and leaves
nulls sorting first.
+
+ ``floating_sequence`` enables Java's NaN and signed-zero ordering for
+ sequence fields, independently of primary-key equality.
"""
comparable_flags = [_base_type_name(fields[idx]) in _COMPARABLE_TYPE_NAMES
for idx in indices]
+ floating_flags = [floating_sequence and _base_type_name(fields[idx]) in
('FLOAT', 'DOUBLE')
+ for idx in indices]
sign = 1 if ascending else -1
def comparator(row1: InternalRow, row2: InternalRow) -> int:
@@ -249,6 +256,19 @@ def _row_field_comparator(
if not comparable_flags[pos]:
raise ValueError(f"Unsupported {fields[idx].type} comparison")
+ if floating_flags[pos]:
+ # Java Float/Double.compare: all NaNs tie above +inf, and
+ # -0.0 precedes +0.0. Primary-key equality is unchanged.
+ if math.isnan(val1):
+ if not math.isnan(val2):
+ return sign
+ continue
+ if math.isnan(val2):
+ return -sign
+ if val1 == val2 == 0.0:
+ val1 = math.copysign(1.0, val1)
+ val2 = math.copysign(1.0, val2)
+
if val1 < val2:
return -sign
elif val1 > val2:
@@ -315,4 +335,4 @@ def builtin_seq_comparator(
f"are not supported -- open an issue to track support.")
indices.append(idx)
- return _row_field_comparator(value_fields, indices, ascending)
+ return _row_field_comparator(value_fields, indices, ascending,
floating_sequence=True)
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index d8fbf2ea52..30b552047e 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -435,6 +435,14 @@ class TableRead:
return None
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 None
if not self._native_blob_view_supported():
return None
if (self._deferred_blob_limit_may_prune(splits)
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index d850206277..fa6ff03103 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -142,6 +142,55 @@ class JavaPyReadWriteTest(unittest.TestCase):
writer.close()
commit.close()
+ @parameterized.expand([
+ (type_name, order, grouping)
+ for type_name in ('float', 'double')
+ for order in ('ascending', 'descending')
+ for grouping in ('batch', 'commits')
+ ])
+ def test_py_write_floating_sequence(self, type_name, order, grouping):
+ table_name = 'default.floating_sequence_{}_{}_{}'.format(type_name,
order, grouping)
+ self.catalog.drop_table(table_name, True)
+ arrow_schema = pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ ('seq', pa.float32() if type_name == 'float' else pa.float64()),
+ ('val', pa.string()),
+ ])
+ self.catalog.create_table(table_name, Schema.from_pyarrow_schema(
+ arrow_schema, primary_keys=['id'], options={
+ 'bucket': '1', 'file.format': 'parquet',
+ 'sequence.field': 'seq', 'sequence.field.sort-order': order,
+ 'write.native.enabled': 'false', 'commit.native.enabled':
'false',
+ }), False)
+ table = self.catalog.get_table(table_name)
+ # Java verifies both the winning payload and the stored sequence value.
+ # The two versions are either folded together or merged by Java on
read.
+ versions = [
+ [(1, 1.0, 'finite'), (2, float('nan'), 'nan'),
+ (3, 0.0, 'positive-zero'), (4, -0.0, 'negative-zero'),
+ (5, None, 'null'), (6, float('nan'), 'nan'),
+ (7, float('inf'), 'infinity'), (8, float('nan'), 'nan-first'),
+ (9, float('-inf'), 'negative-infinity'), (10, None,
'null-first')],
+ [(1, float('nan'), 'nan'), (2, 1.0, 'finite'),
+ (3, -0.0, 'negative-zero'), (4, 0.0, 'positive-zero'),
+ (5, float('nan'), 'nan'), (6, None, 'null'),
+ (7, float('nan'), 'nan'), (8, -float('nan'), 'nan-last'),
+ (9, 1.0, 'finite'), (10, None, 'null-last')],
+ ]
+ groups = [versions[0] + versions[1]] if grouping == 'batch' else
versions
+ for rows in groups:
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pydict({
+ field.name: [row[i] for row in rows]
+ for i, field in enumerate(arrow_schema)
+ }, schema=arrow_schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
@parameterized.expand(get_file_format_params())
def test_py_write_read_append_table(self, file_format):
pa_schema = pa.schema([
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index e078b640f9..f57c9deff0 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -35,6 +35,7 @@ def _table_read(limit=None):
read = TableRead.__new__(TableRead)
read.table = Mock()
read.table.options.native_read_enabled.return_value = True
+ read.table.options.sequence_field.return_value = []
read.table.options.file_format.return_value = 'parquet'
read.table.options.data_file_path_directory.return_value = None
read.table.options.blob_as_descriptor.return_value = False
@@ -68,6 +69,19 @@ def _id_batch(values):
[pa.array(values, type=pa.int32())], names=['id'])
[email protected]('type_', ['FLOAT', 'DOUBLE'])
+def test_floating_sequence_falls_back_before_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()
+
+
def test_native_read_consumes_retained_rust_splits_and_enforces_limit():
read = _table_read(limit=2)
first, second = _Split(), _Split()
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 1b0f4972f3..73c555badc 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -30,7 +30,7 @@ from pypaimon.write.native_write import NativeTableWrite
requires_native = pytest.mark.native_plan
-def _table(tmp_path, primary_key=False, commit_native=True,
table_options=None):
+def _table(tmp_path, primary_key=False, commit_native=True,
table_options=None, id_type=None):
catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
catalog.create_database('default', True)
options = {'file.format': 'parquet', 'write.native.enabled': 'true',
@@ -39,7 +39,7 @@ def _table(tmp_path, primary_key=False, commit_native=True,
table_options=None):
options['bucket'] = '1'
options.update(table_options or {})
catalog.create_table('default.t', Schema.from_pyarrow_schema(
- pa.schema([('id', pa.int64()), ('pt', pa.string())]),
+ pa.schema([('id', id_type if id_type is not None else pa.int64()),
('pt', pa.string())]),
options=options, primary_keys=['id'] if primary_key else [],
partition_keys=[] if primary_key else ['pt']), False)
return catalog.get_table('default.t')
@@ -63,6 +63,44 @@ def test_native_write_is_opt_in():
assert CoreOptions(Options({'write.native.enabled':
'true'})).native_write_enabled()
[email protected]('streaming', [False, True])
[email protected]('sequence', ['missing', 'id,id', 'id,,pt'])
+def test_sequence_validation_precedes_native_selection(tmp_path, streaming,
sequence):
+ table = _table(tmp_path, primary_key=True,
table_options={'sequence.field': sequence})
+ builder = (table.new_stream_write_builder() if streaming
+ else table.new_batch_write_builder())
+ # A usable native backend must not bypass validation. This also runs when
+ # the optional Rust extension is absent.
+ with patch('pypaimon.write.native_write.create_native_write',
return_value=object()) as native:
+ with pytest.raises(ValueError):
+ builder.new_write()
+ native.assert_not_called()
+
+
[email protected]('type_,order,supported', [
+ (pa.int64(), 'ascending', True),
+ (pa.int64(), 'descending', False),
+ (pa.float32(), 'ascending', False),
+ (pa.float64(), 'ascending', False),
+])
+def test_native_sequence_write_capabilities(tmp_path, type_, order, supported):
+ from pypaimon.write.native_write import create_native_write
+
+ table = _table(tmp_path, primary_key=True, id_type=type_, table_options={
+ 'sequence.field': 'id', 'sequence.field.sort-order': order,
+ })
+ with patch('pypaimon.write.native_write.native_write_available',
return_value=True), \
+ patch('pypaimon.write.native_write.create_native_write_table') as
native:
+ writer = create_native_write(table, 'test')
+ if supported:
+ assert isinstance(writer, NativeTableWrite)
+ native.assert_called_once()
+ writer.close()
+ else:
+ assert writer is None
+ native.assert_not_called()
+
+
@requires_native
@pytest.mark.parametrize('primary_key', [False, True])
@pytest.mark.parametrize('commit_native', [False, True])
diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
index da22421475..c7bce9dba5 100644
--- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
+++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
@@ -232,8 +232,12 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
table = self._create_pk_table(
table_name, extra_options=extra_options
)
- # Writing is fine — the guard fires when a reader is built.
- self._write(table, [{'id': 1, 'total': 1, 'max_score': 1, 'label':
'a'}])
+ rows = [{'id': 1, 'total': 1, 'max_score': 1, 'label': 'a'}]
+ if error_type is ValueError:
+ with self.assertRaises(error_type):
+ self._write(table, rows)
+ else:
+ self._write(table, rows)
rb = table.new_read_builder()
with self.assertRaises(error_type) as cm:
rb.new_read()
diff --git a/paimon-python/pypaimon/tests/test_sequence_field_read.py
b/paimon-python/pypaimon/tests/test_sequence_field_read.py
index ea6df67f3e..ff7c2933b5 100644
--- a/paimon-python/pypaimon/tests/test_sequence_field_read.py
+++ b/paimon-python/pypaimon/tests/test_sequence_field_read.py
@@ -268,7 +268,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
table = self._create_pk_table(
'seq_fr', merge_engine='first-row',
extra_options={'sequence.field': 'ts'})
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'high'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'high'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
self.assertIn('FIRST_ROW', str(ctx.exception))
@@ -301,7 +302,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
"""
table = self._create_pk_table(
'seq_missing', extra_options={'sequence.field': 'nope'})
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
self.assertIn('nope', str(ctx.exception))
@@ -312,7 +314,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
"""
table = self._create_pk_table(
'seq_dup', extra_options={'sequence.field': 'ts,ts'})
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
self.assertIn('ts', str(ctx.exception))
@@ -326,7 +329,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
"""
table = self._create_pk_table(
'seq_empty_seg', extra_options={'sequence.field': 'ts,,ts2'})
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
# The empty field name is the one that can't be found in the schema.
@@ -340,7 +344,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
table = self._create_pk_table(
'seq_xpart', extra_options={'sequence.field': 'ts'},
partition_keys=['ts2'])
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
self.assertIn('cross partition', str(ctx.exception).lower())
@@ -356,7 +361,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
'seq_agg_on_seq', merge_engine='aggregation',
extra_options={'sequence.field': 'ts',
'fields.ts.aggregate-function': 'sum'})
- self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
+ with self.assertRaises(ValueError):
+ self._write(table, [{'id': 1, 'ts': 100, 'ts2': 0, 'val': 'x'}])
with self.assertRaises(ValueError) as ctx:
table.new_read_builder().new_read()
self.assertIn('fields.ts.aggregate-function', str(ctx.exception))
@@ -421,15 +427,8 @@ class SequenceFieldReadE2ETest(unittest.TestCase):
self.catalog.create_table('default.seq_complex', schema, False)
table = self.catalog.get_table('default.seq_complex')
wb = table.new_batch_write_builder()
- w = wb.new_write()
- c = wb.new_commit()
- try:
- w.write_arrow(pa.Table.from_pylist(
- [{'id': 1, 'seq': [1, 2], 'val': 'x'}], schema=pa_schema))
- c.commit(w.prepare_commit())
- finally:
- w.close()
- c.close()
+ with self.assertRaises(NotImplementedError):
+ wb.new_write()
with self.assertRaises(NotImplementedError) as ctx:
table.new_read_builder().new_read()
self.assertIn('seq', str(ctx.exception))
diff --git a/paimon-python/pypaimon/tests/test_sequence_field_write.py
b/paimon-python/pypaimon/tests/test_sequence_field_write.py
new file mode 100644
index 0000000000..58004b5401
--- /dev/null
+++ b/paimon-python/pypaimon/tests/test_sequence_field_write.py
@@ -0,0 +1,249 @@
+################################################################################
+# 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.
+################################################################################
+
+import datetime
+from decimal import Decimal
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.row.generic_row import GenericRow
+
+
+pytestmark = pytest.mark.python_write
+
+
[email protected]
+def make_table(tmp_path):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+
+ def create(options=None, sequence_type=None, partitioned=False):
+ arrow_schema = pa.schema([
+ pa.field('id', pa.int64(), nullable=False),
+ ('seq', sequence_type if sequence_type is not None else
pa.int64()),
+ ('seq2', pa.int64()), ('val', pa.string()), ('pt', pa.string()),
+ ])
+ opts = {'bucket': '1', 'sequence.field': 'seq'}
+ opts.update(options or {})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ arrow_schema, primary_keys=['pt', 'id'] if partitioned else ['id'],
+ partition_keys=['pt'] if partitioned else [], options=opts), False)
+ return catalog.get_table('default.t'), arrow_schema
+
+ return create
+
+
+def write_rows(table, schema, rows, grouping='batch', streaming=False):
+ builder = (table.new_stream_write_builder() if streaming
+ else table.new_batch_write_builder())
+ commit = builder.new_commit()
+ groups = [[row] for row in rows] if grouping == 'commits' else [rows]
+ try:
+ for identifier, group in enumerate(groups, 1):
+ writer = builder.new_write()
+ try:
+ batch = pa.Table.from_pylist(group, schema=schema)
+ if grouping == 'chunks':
+ for chunk in batch.to_batches(max_chunksize=1):
+ writer.write_arrow_batch(chunk)
+ elif grouping == 'rows':
+ for row in group:
+ writer.write_row(GenericRow(
+ [row.get(field.name) for field in table.fields],
table.fields))
+ else:
+ writer.write_arrow(batch)
+ if streaming:
+ commit.commit(writer.prepare_commit(identifier),
identifier)
+ else:
+ commit.commit(writer.prepare_commit())
+ commit.close()
+ commit = builder.new_commit()
+ finally:
+ writer.close()
+ finally:
+ commit.close()
+
+
+def read_rows(table, projection=None):
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder.with_projection(projection)
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+
+
[email protected]('engine', ['deduplicate', 'partial-update'])
[email protected]('grouping', ['batch', 'chunks', 'rows', 'commits'])
[email protected]('order', ['ascending', 'descending'])
[email protected]('streaming', [False, True])
+def test_sequence_order_is_independent_of_write_grouping(
+ make_table, engine, grouping, order, streaming):
+ table, schema = make_table({'merge-engine': engine,
'sequence.field.sort-order': order})
+ rows = [
+ {'id': 1, 'seq': 100, 'val': 'high'},
+ {'id': 2, 'seq': 100, 'val': 'tie-first'},
+ {'id': 3, 'seq': 100, 'val': 'non-null'},
+ {'id': 4, 'seq': None, 'val': 'null-first'},
+ {'id': 1, 'seq': 50, 'val': 'low'},
+ {'id': 2, 'seq': 100, 'val': 'tie-last'},
+ {'id': 3, 'seq': None, 'val': 'null'},
+ {'id': 4, 'seq': None, 'val': 'null-last'},
+ ]
+ write_rows(table, schema, rows, grouping, streaming)
+ assert read_rows(table, ['id', 'val']) == [
+ {'id': 1, 'val': 'high' if order == 'ascending' else 'low'},
+ {'id': 2, 'val': 'tie-last'},
+ {'id': 3, 'val': 'non-null'},
+ {'id': 4, 'val': 'null-last'},
+ ]
+
+
+def test_partial_update_fills_nulls_in_sequence_order(make_table):
+ table, schema = make_table({'merge-engine': 'partial-update'})
+ write_rows(table, schema, [
+ {'id': 1, 'seq': 100, 'seq2': 2, 'val': None},
+ {'id': 1, 'seq': 50, 'seq2': 1, 'val': 'filled'},
+ ], 'chunks')
+ assert read_rows(table, ['id', 'seq', 'seq2', 'val']) == [
+ {'id': 1, 'seq': 100, 'seq2': 2, 'val': 'filled'},
+ ]
+
+
+def test_sequence_order_across_buffer_flushes(make_table):
+ table, schema = make_table({'target-file-size': '1 b'})
+ write_rows(table, schema, [
+ {'id': 1, 'seq': 100, 'val': 'high'},
+ {'id': 1, 'seq': 50, 'val': 'low'},
+ ], 'chunks')
+ assert read_rows(table, ['id', 'val']) == [{'id': 1, 'val': 'high'}]
+
+
[email protected]('order,expected', [('ascending', 'high'),
('descending', 'low')])
+def test_compound_sequence_in_partitioned_table(make_table, order, expected):
+ table, schema = make_table({
+ 'sequence.field': 'seq,seq2', 'sequence.field.sort-order': order,
+ }, partitioned=True)
+ rows = []
+ for pt in ['a', 'b']:
+ rows.extend([
+ {'id': 1, 'pt': pt, 'seq': 10, 'seq2': 2, 'val': 'high'},
+ {'id': 1, 'pt': pt, 'seq': 9, 'seq2': 99, 'val': 'low'},
+ {'id': 1, 'pt': pt, 'seq': 10, 'seq2': 1, 'val': 'middle'},
+ {'id': 1, 'pt': pt, 'seq': 10, 'seq2': None, 'val': 'null'},
+ ])
+ write_rows(table, schema, rows, 'chunks')
+ assert sorted(read_rows(table, ['id', 'pt', 'val']), key=lambda row:
row['pt']) == [
+ {'id': 1, 'pt': pt, 'val': expected} for pt in ['a', 'b']
+ ]
+
+
[email protected]('type_,high,low', [
+ (pa.decimal128(10, 2), Decimal('100.50'), Decimal('50.25')),
+ (pa.timestamp('us'), datetime.datetime(2020, 1, 2),
datetime.datetime(2020, 1, 1)),
+ (pa.date32(), datetime.date(2020, 1, 2), datetime.date(2020, 1, 1)),
+ (pa.time32('ms'), datetime.time(12, 0), datetime.time(1, 0)),
+ (pa.float64(), 2.5, -1.0),
+ (pa.binary(), b'z', b'a'),
+])
+def test_typed_sequence_fields(make_table, type_, high, low):
+ table, schema = make_table(sequence_type=type_)
+ write_rows(table, schema, [
+ {'id': 1, 'seq': high, 'val': 'high'},
+ {'id': 1, 'seq': low, 'val': 'low'},
+ ])
+ assert read_rows(table)[0]['seq'] == high
+ assert read_rows(table, ['id', 'val']) == [{'id': 1, 'val': 'high'}]
+
+
[email protected]('type_', [pa.float32(), pa.float64()])
[email protected]('order', ['ascending', 'descending'])
[email protected]('grouping', ['batch', 'chunks', 'commits'])
[email protected]('streaming', [False, True])
+def test_floating_sequence_order(make_table, type_, order, grouping,
streaming):
+ table, schema = make_table({
+ 'sequence.field.sort-order': order,
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ }, sequence_type=type_)
+ # Pairs are in Java Float/Double.compare order, with null always first.
+ pairs = [
+ (1.0, float('nan')), (float('inf'), float('nan')),
+ (float('-inf'), 1.0), (1.0, float('inf')),
+ (-0.0, 0.0), (None, float('nan')), (None, 0.0),
+ ]
+ rows, expected = [], []
+ for low, high in pairs:
+ for reverse in (False, True):
+ key = len(expected)
+ pair = [{'id': key, 'seq': low, 'val': 'low'},
+ {'id': key, 'seq': high, 'val': 'high'}]
+ rows.extend(reversed(pair) if reverse else pair)
+ expected.append({'id': key, 'val': 'high' if low is None or order
== 'ascending'
+ else 'low'})
+ for values in ((float('nan'), -float('nan')), (None, None), (-0.0, -0.0)):
+ key = len(expected)
+ rows.extend({'id': key, 'seq': value, 'val': label}
+ for value, label in zip(values, ('first', 'last')))
+ expected.append({'id': key, 'val': 'last'})
+ write_rows(table, schema, rows, grouping, streaming)
+ assert read_rows(table, ['id', 'val']) == expected
+
+
[email protected]('order', ['ascending', 'descending'])
[email protected]('grouping', ['batch', 'commits'])
+def test_nan_ties_compare_next_sequence_field(make_table, order, grouping):
+ table, schema = make_table({
+ 'sequence.field': 'seq,seq2', 'sequence.field.sort-order': order,
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ }, sequence_type=pa.float64())
+ write_rows(table, schema, [
+ {'id': 1, 'seq': float('nan'), 'seq2': 2, 'val': 'high'},
+ {'id': 1, 'seq': -float('nan'), 'seq2': 1, 'val': 'low'},
+ ], grouping)
+ assert read_rows(table, ['id', 'val']) == [
+ {'id': 1, 'val': 'high' if order == 'ascending' else 'low'}]
+
+
[email protected]('streaming', [False, True])
[email protected]('direct', [False, True])
[email protected]('sequence,type_,error', [
+ ('missing', pa.int64(), ValueError),
+ ('seq,seq', pa.int64(), ValueError),
+ ('seq,,seq2', pa.int64(), ValueError),
+ ('seq', pa.list_(pa.int64()), NotImplementedError),
+])
+def test_dynamic_bucket_sequence_rejected_before_index_creation(
+ make_table, streaming, direct, sequence, type_, error):
+ from pathlib import Path
+ from pypaimon.write.table_write import BatchTableWrite, StreamTableWrite
+
+ table, _ = make_table({'bucket': '-1', 'sequence.field': sequence},
sequence_type=type_)
+ builder = (table.new_stream_write_builder() if streaming
+ else table.new_batch_write_builder())
+ writer_class = StreamTableWrite if streaming else BatchTableWrite
+ with patch.object(table, 'create_row_key_extractor') as extractor:
+ with pytest.raises(error):
+ if direct:
+ writer_class(table, 'test')
+ else:
+ builder.new_write()
+ extractor.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ assert list(Path(table.table_path).glob('index/*')) == []
+ assert list(Path(table.table_path).rglob('*.parquet')) == []
diff --git a/paimon-python/pypaimon/tests/test_write_merge_buffer.py
b/paimon-python/pypaimon/tests/test_write_merge_buffer.py
index b3ee177a39..42428e6388 100644
--- a/paimon-python/pypaimon/tests/test_write_merge_buffer.py
+++ b/paimon-python/pypaimon/tests/test_write_merge_buffer.py
@@ -31,6 +31,8 @@ from unittest.mock import Mock
import pyarrow as pa
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
from pypaimon.read.reader.deduplicate_merge_function import \
DeduplicateMergeFunction
from pypaimon.read.reader.partial_update_merge_function import \
@@ -83,6 +85,7 @@ class _Harness(KeyValueDataWriter):
"""
def __init__(self, merge_function, target_file_size: int = 10 ** 12):
+ self.options = CoreOptions(Options({}))
self.trimmed_primary_keys = ['id']
self._merge_function = merge_function
# Large enough that ``_check_and_roll_if_needed`` does not
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index a73b9df77d..596cd8594d 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -41,6 +41,11 @@ class FileStoreWrite:
def __init__(self, table, commit_user):
from pypaimon.table.file_store_table import FileStoreTable
+ from pypaimon.read.merge_engine_support import
check_sequence_field_supported
+
+ # TableWrite constructs this before the row-key extractor, whose
+ # dynamic bucket index must not retain hashes for rejected writes.
+ check_sequence_field_supported(table)
self.table: FileStoreTable = table
self.data_writers: Dict[Tuple, DataWriter] = {}
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index f96905f3e7..5185731614 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -42,6 +42,13 @@ def native_write_available() -> bool:
def create_native_write(table, commit_user, static_partition=None,
stream=False):
"""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 sequence_fields:
+ # The native writer currently sorts sequence fields ascending and
+ # does not implement Java's NaN/signed-zero ordering.
+ if (not table.options.sequence_field_sort_order_is_ascending()
+ or any(pa.types.is_floating(schema.field(name).type) for name
in sequence_fields)):
+ return None
partition_types = [schema.field(name).type for name in
table.partition_keys]
if (not native_write_available()
or table.options.data_evolution_enabled()
diff --git a/paimon-python/pypaimon/write/write_builder.py
b/paimon-python/pypaimon/write/write_builder.py
index 0586ff2d73..0f79308ab9 100644
--- a/paimon-python/pypaimon/write/write_builder.py
+++ b/paimon-python/pypaimon/write/write_builder.py
@@ -54,6 +54,11 @@ class WriteBuilder(ABC):
return str(uuid.uuid4())
def _native_write(self, static_partition=None, stream=False):
+ from pypaimon.read.merge_engine_support import
check_sequence_field_supported
+
+ # Keep invalid configurations outside the native fallback handler and
+ # enforce the same contract regardless of which writer is selected.
+ check_sequence_field_supported(self.table)
if not self.table.options.native_write_enabled():
return None
# data-file.path-directory relocates data files under a sub-directory
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index 1f1537f6d1..6992852bbc 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -17,6 +17,7 @@
from typing import List, Union
+import numpy as np
import pyarrow as pa
import pyarrow.compute as pc
@@ -157,7 +158,7 @@ class KeyValueDataWriter(DataWriter):
"""Fold same-PK runs in ``data`` using ``self._merge_function``.
``data`` is required to already be sorted by
- ``(primary_key, _SEQUENCE_NUMBER)``. ``_flush_all`` is the
+ ``(primary_key, sequence.field, _SEQUENCE_NUMBER)``. ``_flush_all`` is
the
only caller and runs ``_sort_by_primary_key`` immediately
before this method, so the precondition holds.
@@ -285,12 +286,42 @@ class KeyValueDataWriter(DataWriter):
# pc.sort_indices + .take work uniformly over RecordBatch and
# Table, so this serves both the per-batch entry path (legacy)
# and the buffer-wide sort path (used by ``_flush_all``).
- sort_keys = [(key, 'ascending') for key in self.trimmed_primary_keys]
+ sort_columns = [data.column(key) for key in self.trimmed_primary_keys]
+ sort_orders = ['ascending'] * len(sort_columns)
+ sequence_fields = self.options.sequence_field()
+ if sequence_fields:
+ sequence_order = ('ascending' if
self.options.sequence_field_sort_order_is_ascending()
+ else 'descending')
+ for field in sequence_fields:
+ column = data.column(field)
+ if pa.types.is_floating(column.type):
+ column = self._floating_sequence_sort_key(column)
+ sort_columns.append(column)
+ sort_orders.append(sequence_order)
if '_SEQUENCE_NUMBER' in data.schema.names:
- sort_keys.append(('_SEQUENCE_NUMBER', 'ascending'))
+ sort_columns.append(data.column('_SEQUENCE_NUMBER'))
+ sort_orders.append('ascending')
+ # Sort a separate key table so temporary keys cannot collide with user
+ # column names or change the stored values (including signed zero).
+ names = [str(i) for i in range(len(sort_columns))]
# Java MergeTree comparators order null keys first. Keep
Python-written files in the same
# order so their key ranges and sorted-run invariants are
interoperable with Java readers.
sorted_indices = pc.sort_indices(
- data, sort_keys=sort_keys, null_placement='at_start')
+ pa.table(sort_columns, names=names),
+ sort_keys=list(zip(names, sort_orders)), null_placement='at_start')
return data.take(sorted_indices)
+
+ @staticmethod
+ def _floating_sequence_sort_key(column):
+ """Unsigned keys in Java Float/Double.compare order, preserving
nulls."""
+ # The NumPy protocol supports Array and older ChunkedArray APIs alike.
+ values = np.asarray(column)
+ bits = values.view(np.dtype('uint{}'.format(column.type.bit_width)))
+ sign_bit = np.array(1 << (column.type.bit_width - 1), dtype=bits.dtype)
+ # Invert negative IEEE bits, flip the sign bit for nonnegative values.
+ # This orders -0.0 before +0.0, unlike Arrow's floating-point sort.
+ keys = np.where(bits & sign_bit, ~bits, bits ^ sign_bit)
+ # All NaN signs/payloads compare equal, above positive infinity.
+ keys[np.isnan(values)] = np.iinfo(bits.dtype).max
+ return pa.array(keys, mask=np.asarray(column.is_null()))
