diff --git a/paimon-python/pypaimon/read/split_serializer.py b/paimon-python/pypaimon/read/split_serializer.py index 4b50c86c289b..ff5397c7c2af 100644 --- a/paimon-python/pypaimon/read/split_serializer.py +++ b/paimon-python/pypaimon/read/split_serializer.py @@ -18,7 +18,7 @@ """Deserialize the cross-language ``SplitSerializer`` v1 binary into a pypaimon :class:`DataSplit`. -Mirror of the Java ``DataSplit#serialize`` (VERSION 8) frame wrapped in the +Mirror of the Java ``DataSplit#serialize`` (versions 8 and 9) frame wrapped in the ``SplitSerializer`` v1 header, as produced by ``pypaimon_rust``'s ``Split.serialize()``. Extracts the fields the reader needs, plus per-file min/max keys for PK merge-on-read; key/value stats (planning-only) stay empty. @@ -46,16 +46,14 @@ _INDEXED_SPLIT_MAGIC = -938472394838495695 _INDEXED_SPLIT_VERSION = 1 _SPLIT_MAGIC = -2394839472490812314 -_SPLIT_VERSION = 8 - -_DFM_ARITY = 20 +_DFM_ARITY_BY_VERSION = {8: 20, 9: 21} def _f(idx, name, dtype): return DataField(idx, name, dtype) -# DataFileMeta 20-field layout (order/types mirror DataFileMetaSerializer#toRow). +# DataFileMeta layout (order/types mirror DataFileMetaSerializer#toRow). # Fields 3/4 (min/max key) are decoded for PK tables; 5/6 (stats) stay unread. _DFM_FIELDS: List[DataField] = [ _f(0, '_FILE_NAME', AtomicType('STRING')), @@ -78,10 +76,11 @@ def _f(idx, name, dtype): _f(17, '_EXTERNAL_PATH', AtomicType('STRING')), _f(18, '_FIRST_ROW_ID', AtomicType('BIGINT')), _f(19, '_WRITE_COLS', AtomicType('BYTES')), + # Added in v9 for compaction/index refresh; unused by the Python reader. + _f(20, '_WRITE_COLS_SEQUENCES', AtomicType('BYTES')), ] -# Arity is fixed by DataSplit VERSION 8; keep the field list and arity in lockstep. -assert len(_DFM_FIELDS) == _DFM_ARITY +assert len(_DFM_FIELDS) == max(_DFM_ARITY_BY_VERSION.values()) def _decode_str_array(b: Optional[bytes]) -> Optional[List[str]]: @@ -214,9 +213,9 @@ def _read_datasplit_body(r: _Reader, partition_fields: List[DataField], if r.i64() != _SPLIT_MAGIC: raise ValueError("bad DataSplit magic") version = r.i32() - if version != _SPLIT_VERSION: + if version not in _DFM_ARITY_BY_VERSION: raise ValueError( - "unsupported DataSplit version %d (expected %d)" % (version, _SPLIT_VERSION)) + "unsupported DataSplit version %d (expected 8 or 9)" % version) snapshot_id = r.i64() # scanned snapshot; row-id conflict detection needs it partition = GenericRowDeserializer.from_bytes(r.take(r.i32()), partition_fields) bucket = r.i32() @@ -228,7 +227,8 @@ def _read_datasplit_body(r: _Reader, partition_fields: List[DataField], if r.u8() != 0: # beforeDeletionFiles must be null raise ValueError("cannot deserialize a split with before deletion files") file_count = r.i32() - files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, key_fields) + arity = _DFM_ARITY_BY_VERSION[version] + files = [_datafilemeta_from_row(r.take(r.i32()), bucket_path, arity, key_fields) for _ in range(file_count)] data_deletion_files = _read_deletion_list(r) r.u8() # isStreaming @@ -250,9 +250,9 @@ def _decode_key(b: Optional[bytes], key_fields: Optional[List[DataField]]) -> Ge return GenericRowDeserializer.from_bytes(b, key_fields) -def _datafilemeta_from_row(row_bytes: bytes, bucket_path: str, +def _datafilemeta_from_row(row_bytes: bytes, bucket_path: str, arity: int, key_fields: Optional[List[DataField]] = None) -> DataFileMeta: - row = BinaryRow(struct.pack('>i', _DFM_ARITY) + row_bytes, _DFM_FIELDS) + row = BinaryRow(struct.pack('>i', arity) + row_bytes, _DFM_FIELDS[:arity]) g = row.get_field file_name = g(0) external_path = g(17) diff --git a/paimon-python/pypaimon/tests/split_serializer_test.py b/paimon-python/pypaimon/tests/split_serializer_test.py index d168c6ff4bd5..0c35e1064073 100644 --- a/paimon-python/pypaimon/tests/split_serializer_test.py +++ b/paimon-python/pypaimon/tests/split_serializer_test.py @@ -16,6 +16,7 @@ # under the License. import base64 +import struct import unittest from pypaimon.globalindex.indexed_split import IndexedSplit @@ -56,7 +57,7 @@ ) # Same split serialized as an IndexedSplit (type id 3): identical DataSplit body -# with a trailing row-ranges/scores section that the reader skips. +# with a trailing row-ranges/scores section that must survive decoding. _GOLDEN_INDEXED_SPLIT_V1 = base64.b64decode( "U1BMSVRfVjEAAAABAAAAA/L54FRCJC4xAAAAAd7D0jAsGexmAAAACAAAAAAAAAAqAAAAHAAAAAIA" "AAAAAAAAAOoHAAAAAAAABwAAAAAAAAAAAAADABRkdD0yMDI2MDcwNi9idWNrZXQtMwEAAAAIAAAA" @@ -78,6 +79,26 @@ ) +# Java DataSplitCompatibleTest's v9 golden, also used by paimon-rust's +# serialize_matches_datasplit_v9 test. Includes the new non-null per-column +# sequence numbers [15, 100, 150, 200] after write_cols. +_GOLDEN_DATA_SPLIT_V9 = base64.b64decode( + "3sPSMCwZ7GYAAAAJAAAAAAAAABIAAAAUAAAAAQAAAAAAAAAAYWFhYWEAAIUAAAAUAAdteSBwYXRo" + "AQAAACAAAAAAAAAAAAEAAAJoAAAAAAAAAABteV9maWxlhwAAEAAAAAAAAAQAAAAAAAAUAAAAsAAA" + "ABQAAADIAAAAYAAAAOAAAACAAAAAQAEAAA8AAAAAAAAAyAAAAAAAAAAFAAAAAAAAAAMAAAAAAAAA" + "GAAAAMABAABgTEpMfwEAAAsAAAAAAAAAAQIEAAAAAIMBAAAAAAAAACAAAADYAQAAGQAAAPgBAAAM" + "AAAAAAAAACgAAAAYAgAAKAAAAEACAAAAAAABAAAAAAAAAABtaW5fa2V5hwAAAAAAAAABAAAAAAAA" + "AABtYXhfa2V5hwAAAAAAAAAAAAAAABQAAAAgAAAAFAAAADgAAAAQAAAAUAAAAAAAAAEAAAAAAAAA" + "AG1pbl9rZXmHAAAAAAAAAAEAAAAAAAAAAG1heF9rZXmHAAAAAAEAAAAAAAAAAAAAAAAAAAAAAAAA" + "AAAAACQAAAAgAAAAJAAAAEgAAAAQAAAAcAAAAAAAAAEAAAAAAAAAAAkAAAAQAAAAbWluX3ZhbHVl" + "AAAAAAAAAAAAAAAAAAABAAAAAAAAAAAJAAAAEAAAAG1heF92YWx1ZQAAAAAAAAAAAAAAAQAAAAAA" + "AAAAAAAAAAAAAAIAAAAAAAAAZXh0cmExAIZleHRyYTIAhgMAAAAAAAAAZmllbGQxAIZmaWVsZDIA" + "hmZpZWxkMwCGaGRmczovLy9wYXRoL3RvL3dhcmVob3VzZQAAAAAAAAAEAAAAAAAAAGEAAAAAAACB" + "YgAAAAAAAIFjAAAAAAAAgWYAAAAAAACBBAAAAAAAAAAPAAAAAAAAAGQAAAAAAAAAlgAAAAAAAADI" + "AAAAAAAAAAEAAAABAQANZGVsZXRpb25fZmlsZQAAAAAAAABkAAAAAAAAABYAAAAAAAAAIQAA" +) + + class SplitSerializerTest(unittest.TestCase): def _partition_fields(self): @@ -139,6 +160,51 @@ def test_decode_modified_utf8_supplementary_char(self): _decode_modified_utf8(bytes([0xED, 0xA0, 0xBD, 0xED, 0xB8, 0x80])), '\U0001F600') + def test_deserialize_v9_data_and_indexed_splits(self): + fields = [DataField(0, 's', AtomicType('STRING'))] + for indexed in (False, True): + with self.subTest(indexed=indexed): + if indexed: + data = (_GOLDEN_INDEXED_SPLIT_V1[:28] + _GOLDEN_DATA_SPLIT_V9 + + struct.pack('>iqqBif', 1, 13, 14, 1, 1, 0.75)) + else: + data = _GOLDEN_DATA_SPLIT_V1[:16] + _GOLDEN_DATA_SPLIT_V9 + split = deserialize_split_v1(data, fields, fields) + self.assertEqual(split.snapshot_id, 18) + self.assertEqual(list(split.partition.values), ['aaaaa']) + self.assertEqual(split.bucket, 20) + self.assertFalse(split.raw_convertible) + self.assertEqual(len(split.files), 1) + file = split.files[0] + self.assertEqual(file.file_name, 'my_file') + self.assertEqual(file.file_path, 'hdfs:///path/to/warehouse') + self.assertEqual((file.file_size, file.row_count), (1024 * 1024, 1024)) + self.assertEqual(list(file.min_key.values), ['min_key']) + self.assertEqual(list(file.max_key.values), ['max_key']) + self.assertEqual((file.min_sequence_number, file.max_sequence_number), (15, 200)) + self.assertEqual((file.schema_id, file.level), (5, 3)) + self.assertEqual(file.extra_files, ['extra1', 'extra2']) + self.assertEqual(file.embedded_index, bytes([1, 2, 4])) + self.assertEqual(file.value_stats_cols, ['field1', 'field2', 'field3']) + self.assertEqual(file.first_row_id, 12) + self.assertEqual(file.write_cols, ['a', 'b', 'c', 'f']) + self.assertEqual(len(split.data_deletion_files), 1) + dv = split.data_deletion_files[0] + self.assertEqual( + (dv.dv_index_path, dv.offset, dv.length, dv.cardinality), + ('deletion_file', 100, 22, 33)) + if indexed: + self.assertEqual([(r.from_, r.to) for r in split.row_ranges()], [(13, 14)]) + self.assertEqual(split.scores(), [0.75]) + + def test_unsupported_data_split_version_raises(self): + for version in (7, 10): + with self.subTest(version=version): + data = (_GOLDEN_DATA_SPLIT_V1[:24] + struct.pack('>i', version) + + _GOLDEN_DATA_SPLIT_V1[28:]) + with self.assertRaisesRegex(ValueError, 'unsupported DataSplit version'): + deserialize_split_v1(data, self._partition_fields()) + def test_decode_str_array_inline_and_pointer(self): # Covers both element encodings: inline (<=7 bytes) and var pointer (>7). self.assertEqual(_decode_str_array(_BINARY_ARRAY_STR), ['id', 'longcolumn12'])