From de67c452ee85805de7d40ef1363d4c992888808f Mon Sep 17 00:00:00 2001 From: Juntao Zhang Date: Wed, 12 Aug 2026 22:55:42 +0800 Subject: [PATCH 1/2] [python] Support datetime and UUID encoding in GenericVariant and fix UUID byte order --- docs/docs/pypaimon/python-api.mdx | 5 +- docs/sidebars.js | 1 + .../java/org/apache/paimon/JavaPyE2ETest.java | 64 ++++- .../pypaimon/data/generic_variant.py | 25 +- .../tests/e2e/java_py_read_write_test.py | 46 +++- paimon-python/pypaimon/tests/variant_test.py | 238 ++++++++++++++++-- 6 files changed, 349 insertions(+), 30 deletions(-) diff --git a/docs/docs/pypaimon/python-api.mdx b/docs/docs/pypaimon/python-api.mdx index 1921a9764974..d22092cc7227 100644 --- a/docs/docs/pypaimon/python-api.mdx +++ b/docs/docs/pypaimon/python-api.mdx @@ -1119,8 +1119,9 @@ sub-columns for column-skipping via sub-field projection. Fields not listed in `variant.shreddingSchema` are stored in the overflow `value` bytes and remain fully accessible on the read path. -Supported Paimon type strings for shredded sub-fields: `BOOLEAN`, `INT`, `BIGINT`, `FLOAT`, `DOUBLE`, -`VARCHAR`, `DECIMAL(p,s)`, and nested `ROW` types for recursive object shredding. +Supported Paimon type strings for shredded sub-fields: `BOOLEAN`, `TINYINT`, `SMALLINT`, `INT`, +`BIGINT`, `FLOAT`, `DOUBLE`, `VARCHAR`, `BINARY`, `VARBINARY`, `DECIMAL(p,s)`, `ARRAY`, and nested +`ROW` types for recursive object shredding. diff --git a/docs/sidebars.js b/docs/sidebars.js index be7294db7b03..df6ece28ca73 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -114,6 +114,7 @@ const sidebars = { }, "items": [ "multimodal-table/data-evolution", + "multimodal-table/variant", "multimodal-table/blob", "multimodal-table/vector", { 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 b6801b4a7bca..b4d56386ae6b 100644 --- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java +++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java @@ -39,6 +39,8 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; import org.apache.paimon.data.variant.GenericVariant; +import org.apache.paimon.data.variant.GenericVariantBuilder; +import org.apache.paimon.data.variant.GenericVariantUtil.Type; import org.apache.paimon.deletionvectors.BitmapDeletionVector; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer; @@ -101,6 +103,10 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Paths; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1682,6 +1688,28 @@ public void testJavaWriteVariantTable() throws Exception { 3, BinaryString.fromString("Carol"), GenericVariant.fromJson("[1,2,3]"))); + + // Scalar DATE/TIMESTAMP/TIMESTAMP_NTZ/UUID values for cross-language compatibility. + GenericVariantBuilder b = new GenericVariantBuilder(false); + b.appendDate((int) LocalDate.of(2024, 1, 15).toEpochDay()); + GenericVariant v = b.result(); + write.write(GenericRow.of(4, BinaryString.fromString("Dave"), v)); + + b = new GenericVariantBuilder(false); + b.appendTimestampNtz( + toEpochMicros(LocalDateTime.of(2024, 1, 15, 12, 30, 45, 123456000))); + v = b.result(); + write.write(GenericRow.of(5, BinaryString.fromString("Eve"), v)); + + b = new GenericVariantBuilder(false); + b.appendTimestamp(toEpochMicros(Instant.parse("2024-01-15T12:30:45.123456Z"))); + v = b.result(); + write.write(GenericRow.of(6, BinaryString.fromString("Frank"), v)); + + b = new GenericVariantBuilder(false); + b.appendUuid(UUID.fromString("12345678-1234-5678-1234-567812345678")); + v = b.result(); + write.write(GenericRow.of(7, BinaryString.fromString("Grace"), v)); commit.commit(write.prepareCommit()); } @@ -1691,7 +1719,7 @@ public void testJavaWriteVariantTable() throws Exception { TableRead read = readTable.newRead(); List res = getResult(read, splits, row -> internalRowToString(row, readTable.rowType())); - assertThat(res).hasSize(3); + assertThat(res).hasSize(7); LOG.info("testJavaWriteVariantTable: wrote and read back {} VARIANT rows", res.size()); // Also write a shredded VARIANT table for Python to read (variant_shredded_test). @@ -1762,7 +1790,7 @@ public void testJavaReadVariantTable() throws Exception { TableRead read = table.newRead(); List res = getResult(read, splits, row -> internalRowToString(row, table.rowType())); - assertThat(res).hasSize(4); + assertThat(res).hasSize(7); // Verify the VARIANT column is present in the schema assertThat(table.rowType().getFieldNames()).contains("payload"); @@ -1781,8 +1809,29 @@ public void testJavaReadVariantTable() throws Exception { assertThat(row.isNullAt(2)).isTrue(); } else { assertThat(row.isNullAt(2)).isFalse(); - org.apache.paimon.data.variant.Variant v = row.getVariant(2); + GenericVariant v = (GenericVariant) row.getVariant(2); assertThat(v).isNotNull(); + if (id == 5) { + // DATE '2024-01-15' + assertThat(v.getType()).isEqualTo(Type.DATE); + assertThat(v.getLong()) + .isEqualTo(LocalDate.of(2024, 1, 15).toEpochDay()); + } else if (id == 6) { + // TIMESTAMP_NTZ '2024-01-15 12:30:45.123456' + assertThat(v.getType()).isEqualTo(Type.TIMESTAMP_NTZ); + long expectedMicros = + toEpochMicros( + LocalDateTime.of( + 2024, 1, 15, 12, 30, 45, 123456000)); + assertThat(v.getLong()).isEqualTo(expectedMicros); + } else if (id == 7) { + // UUID '12345678-1234-5678-1234-567812345678' + assertThat(v.getType()).isEqualTo(Type.UUID); + assertThat(v.getUuid()) + .isEqualTo( + UUID.fromString( + "12345678-1234-5678-1234-567812345678")); + } } }); } @@ -1830,6 +1879,15 @@ public void testJavaReadVariantTable() throws Exception { shreddedRes.size()); } + private static long toEpochMicros(LocalDateTime dateTime) { + return dateTime.toInstant(ZoneOffset.UTC).getEpochSecond() * 1_000_000L + + dateTime.getNano() / 1000L; + } + + private static long toEpochMicros(Instant instant) { + return instant.getEpochSecond() * 1_000_000L + instant.getNano() / 1000L; + } + /** Step 1: Write 5 base files for compact conflict test. */ @Test @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true") diff --git a/paimon-python/pypaimon/data/generic_variant.py b/paimon-python/pypaimon/data/generic_variant.py index 9d0e0a0b0c06..8d6dee4a7cbf 100644 --- a/paimon-python/pypaimon/data/generic_variant.py +++ b/paimon-python/pypaimon/data/generic_variant.py @@ -353,6 +353,13 @@ def append_binary(self, b): self._buf[self._pos:self._pos + len(b)] = b self._pos += len(b) + def append_uuid(self, u): + # UUID values are 16-byte big-endian: msb followed by lsb. + self._write_byte(_primitive_header(_UUID)) + self._ensure(16) + self._buf[self._pos:self._pos + 16] = u.bytes + self._pos += 16 + def append_date(self, days_since_epoch): self._write_byte(_primitive_header(_DATE)) self._write_le(days_since_epoch & 0xFFFFFFFF, 4) @@ -447,6 +454,18 @@ def build_python(self, obj): self._finish_writing_array(start, elem_offsets) elif isinstance(obj, bytes): self.append_binary(obj) + elif isinstance(obj, _uuid.UUID): + self.append_uuid(obj) + elif isinstance(obj, datetime.datetime): + if obj.tzinfo is not None: + micros = int((obj - _EPOCH_DT_UTC).total_seconds() * 1_000_000) + self.append_timestamp(micros) + else: + micros = int((obj - _EPOCH_DT_NTZ).total_seconds() * 1_000_000) + self.append_timestamp_ntz(micros) + elif isinstance(obj, datetime.date): + days = (obj - _EPOCH_DATE).days + self.append_date(days) else: raise TypeError(f'Unsupported Python type for variant encoding: {type(obj).__name__}') @@ -687,9 +706,9 @@ def _to_python_impl(self, value, metadata, pos): length = _read_unsigned(value, pos + 1, _U32_SIZE) return bytes(value[pos + 1 + _U32_SIZE:pos + 1 + _U32_SIZE + length]) if vtype == _Type.UUID: - # 16 bytes: two little-endian int64 (msb, lsb) → standard UUID - msb = _read_unsigned(value, pos + 1, 8) - lsb = _read_unsigned(value, pos + 9, 8) + # UUID values are 16-byte big-endian: msb followed by lsb. + msb = int.from_bytes(value[pos + 1:pos + 9], 'big', signed=False) + lsb = int.from_bytes(value[pos + 9:pos + 17], 'big', signed=False) return _uuid.UUID(int=(msb << 64) | lsb) if vtype == _Type.OBJECT: def _build_dict(size, id_size, offset_size, id_start, offset_start, data_start): 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 d14e1f7459fb..d011162b4999 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 @@ -20,6 +20,7 @@ import os import sys import unittest +import uuid from decimal import Decimal import pandas as pd @@ -1931,7 +1932,7 @@ def test_py_read_variant_table(self): splits = table_scan.plan().splits() result = table_read.to_arrow(splits) - self.assertEqual(result.num_rows, 3) + self.assertEqual(result.num_rows, 7) # VARIANT maps to struct payload_field = result.schema.field('payload') @@ -1973,6 +1974,31 @@ def test_py_read_variant_table(self): carol_data = GenericVariant.from_arrow_struct(payload_list[id_list.index(3)]).to_python() self.assertEqual(carol_data, [1, 2, 3]) + # Row 4: Dave, DATE '2024-01-15' + dave_data = GenericVariant.from_arrow_struct(payload_list[id_list.index(4)]).to_python() + self.assertEqual(dave_data, datetime.date(2024, 1, 15)) + + # Row 5: Eve, TIMESTAMP_NTZ '2024-01-15 12:30:45.123456' + eve_data = GenericVariant.from_arrow_struct(payload_list[id_list.index(5)]).to_python() + self.assertEqual( + eve_data, datetime.datetime(2024, 1, 15, 12, 30, 45, 123456) + ) + + # Row 6: Frank, TIMESTAMP '2024-01-15 12:30:45.123456 UTC' + frank_data = GenericVariant.from_arrow_struct(payload_list[id_list.index(6)]).to_python() + self.assertEqual( + frank_data, + datetime.datetime( + 2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc + ), + ) + + # Row 7: Grace, UUID '12345678-1234-5678-1234-567812345678' + grace_data = GenericVariant.from_arrow_struct(payload_list[id_list.index(7)]).to_python() + self.assertEqual( + grace_data, uuid.UUID('12345678-1234-5678-1234-567812345678') + ) + print("test_py_read_variant_table: verified {} VARIANT rows".format(result.num_rows)) # Also verify shredded VARIANT: Java wrote variant_shredded_test with @@ -2029,6 +2055,9 @@ def test_py_write_variant_table(self): id=2 payload=[10,20,30] id=3 payload="hello" id=4 payload=null + id=5 payload=DATE '2024-01-15' + id=6 payload=TIMESTAMP_NTZ '2024-01-15 12:30:45.123456' + id=7 payload=UUID '12345678-1234-5678-1234-567812345678' """ variant_type = pa.struct([ pa.field('value', pa.binary(), nullable=False), @@ -2046,15 +2075,24 @@ def test_py_write_variant_table(self): self.catalog.create_table(table_name, schema, False) table = self.catalog.get_table(table_name) + test_uuid = uuid.UUID('12345678-1234-5678-1234-567812345678') variant_col = GenericVariant.to_arrow_array([ GenericVariant.from_python({"name": "test", "value": 42}), GenericVariant.from_python([10, 20, 30]), GenericVariant.from_python("hello"), None, # SQL NULL at the column level, not a VARIANT containing JSON null + GenericVariant.from_python(datetime.date(2024, 1, 15)), + GenericVariant.from_python( + datetime.datetime(2024, 1, 15, 12, 30, 45, 123456) + ), + GenericVariant.from_python(test_uuid), ]) data = pa.table({ - 'id': pa.array([1, 2, 3, 4], type=pa.int32()), - 'name': pa.array(['row1', 'row2', 'row3', 'row4'], type=pa.string()), + 'id': pa.array([1, 2, 3, 4, 5, 6, 7], type=pa.int32()), + 'name': pa.array( + ['row1', 'row2', 'row3', 'row4', 'row5', 'row6', 'row7'], + type=pa.string() + ), 'payload': variant_col, }, schema=pa_schema) @@ -2065,7 +2103,7 @@ def test_py_write_variant_table(self): table_commit.commit(table_write.prepare_commit()) table_write.close() table_commit.close() - print("test_py_write_variant_table: wrote 4 VARIANT rows to {}".format(table_name)) + print("test_py_write_variant_table: wrote 7 VARIANT rows to {}".format(table_name)) # Also write a shredded VARIANT table (py_variant_shredded_test) for Java to read. # Python shreds the 'age' (BIGINT) and 'city' (VARCHAR) sub-fields of 'payload' diff --git a/paimon-python/pypaimon/tests/variant_test.py b/paimon-python/pypaimon/tests/variant_test.py index 6db6b8d8faa6..5c16f6f39a68 100644 --- a/paimon-python/pypaimon/tests/variant_test.py +++ b/paimon-python/pypaimon/tests/variant_test.py @@ -47,6 +47,9 @@ import struct as _struct import tempfile import unittest +import datetime +import uuid +from decimal import Decimal import pyarrow as pa import pyarrow.parquet as pq @@ -361,6 +364,38 @@ def test_repr_and_str(self): self.assertIn('hello', repr(gv)) self.assertIn('hello', str(gv)) + def test_from_python_date(self): + value = datetime.date(2024, 1, 15) + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) + + def test_from_python_timestamp_ntz(self): + value = datetime.datetime(2024, 1, 15, 12, 30, 45, 123456) + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) + + def test_from_python_timestamp_ltz(self): + value = datetime.datetime( + 2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc + ) + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) + + def test_from_python_nested_datetime(self): + obj = {'created_at': datetime.datetime(2024, 1, 15, 12, 0)} + gv = GenericVariant.from_python(obj) + self.assertEqual(gv.to_python(), obj) + + def test_from_python_uuid(self): + value = uuid.UUID('12345678-1234-5678-1234-567812345678') + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) + + def test_from_python_nested_uuid(self): + obj = {'id': uuid.UUID('12345678-1234-5678-1234-567812345678')} + gv = GenericVariant.from_python(obj) + self.assertEqual(gv.to_python(), obj) + class TestToArrowArray(unittest.TestCase): @@ -618,13 +653,50 @@ def _roundtrip(self, json_str: str, arrow_type: pa.DataType): gv = GenericVariant(value_bytes, b'\x01\x00') return gv.to_python() + def test_tinyint(self): + self.assertEqual(self._roundtrip('42', pa.int8()), 42) + + def test_smallint(self): + self.assertEqual(self._roundtrip('1000', pa.int16()), 1000) + def test_int(self): - self.assertEqual(self._roundtrip('42', pa.int64()), 42) + self.assertEqual(self._roundtrip('1000000', pa.int32()), 1000000) + + def test_bigint(self): + self.assertEqual(self._roundtrip('12345678901234', pa.int64()), 12345678901234) def test_float(self): - value_bytes = _encode_scalar_to_value_bytes(3.14, pa.float64()) + self.assertAlmostEqual(self._roundtrip('3.14159', pa.float32()), 3.14159, places=5) + + def test_double(self): + self.assertEqual(self._roundtrip('1.012345678901234567890123456789012345678', + pa.float64()), 1.012345678901234567890123456789012345678) + + def test_decimal(self): + value = Decimal('12345.6789') + value_bytes = _encode_scalar_to_value_bytes(value, pa.decimal128(10, 4)) + gv = GenericVariant(value_bytes, b'\x01\x00') + self.assertEqual(gv.to_python(), value) + + def test_date(self): + value = datetime.date(2024, 1, 15) + value_bytes = _encode_scalar_to_value_bytes(value, pa.date32()) + gv = GenericVariant(value_bytes, b'\x01\x00') + self.assertEqual(gv.to_python(), value) + + def test_timestamp_ntz(self): + value = datetime.datetime(2024, 1, 15, 12, 30, 45, 123456) + value_bytes = _encode_scalar_to_value_bytes(value, pa.timestamp('us')) + gv = GenericVariant(value_bytes, b'\x01\x00') + self.assertEqual(gv.to_python(), value) + + def test_timestamp_ltz(self): + value = datetime.datetime( + 2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc + ) + value_bytes = _encode_scalar_to_value_bytes(value, pa.timestamp('us', tz='UTC')) gv = GenericVariant(value_bytes, b'\x01\x00') - self.assertAlmostEqual(gv.to_python(), 3.14, places=5) + self.assertEqual(gv.to_python(), value) def test_bool_true(self): self.assertEqual(self._roundtrip('true', pa.bool_()), True) @@ -635,6 +707,11 @@ def test_bool_false(self): def test_string(self): self.assertEqual(self._roundtrip('"hello"', pa.string()), 'hello') + def test_binary(self): + value_bytes = _encode_scalar_to_value_bytes(b'\x00\x01\x02', pa.binary()) + gv = GenericVariant(value_bytes, b'\x01\x00') + self.assertEqual(gv.to_python(), b'\x00\x01\x02') + def test_null(self): value_bytes = _encode_scalar_to_value_bytes(None, pa.int64()) gv = GenericVariant(value_bytes, b'\x01\x00') @@ -1073,28 +1150,62 @@ def test_plain_variant_write_and_read(self): self.catalog.create_table('default.plain_variant', schema, False) table = self.catalog.get_table('default.plain_variant') + test_uuid = uuid.UUID('12345678-1234-5678-1234-567812345678') gvs = [ GenericVariant.from_python({'age': 30, 'city': 'Beijing'}), GenericVariant.from_python({'score': 99, 'active': True}), GenericVariant.from_python([1, 2, 3]), + GenericVariant.from_python({'dt': datetime.date(2024, 1, 15)}), + GenericVariant.from_python( + {'ts': datetime.datetime(2024, 1, 15, 12, 30, 45, 123456)} + ), + GenericVariant.from_python( + {'ts_ltz': datetime.datetime( + 2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc + )} + ), + GenericVariant.from_python({'id': test_uuid}), + GenericVariant.from_python(test_uuid), ] data = pa.table( - {'id': [1, 2, 3], 'payload': GenericVariant.to_arrow_array(gvs)}, + {'id': list(range(1, 9)), 'payload': GenericVariant.to_arrow_array(gvs)}, schema=self._pa_schema(), ) result = self._write_and_read(table, data) - self.assertEqual(result.num_rows, 3) + self.assertEqual(result.num_rows, 8) payload_col = result.column('payload') - gv0 = GenericVariant.from_arrow_struct(payload_col[0].as_py()) - self.assertEqual(gv0.to_python(), {'age': 30, 'city': 'Beijing'}) + self.assertEqual( + GenericVariant.from_arrow_struct(payload_col[0].as_py()).to_python(), + {'age': 30, 'city': 'Beijing'}, + ) + self.assertEqual( + GenericVariant.from_arrow_struct(payload_col[1].as_py()).to_python(), + {'score': 99, 'active': True}, + ) + self.assertEqual( + GenericVariant.from_arrow_struct(payload_col[2].as_py()).to_python(), + [1, 2, 3], + ) + + py3 = GenericVariant.from_arrow_struct(payload_col[3].as_py()).to_python() + self.assertEqual(py3['dt'], datetime.date(2024, 1, 15)) + + py4 = GenericVariant.from_arrow_struct(payload_col[4].as_py()).to_python() + self.assertEqual(py4['ts'], datetime.datetime(2024, 1, 15, 12, 30, 45, 123456)) - gv1 = GenericVariant.from_arrow_struct(payload_col[1].as_py()) - self.assertEqual(gv1.to_python(), {'score': 99, 'active': True}) + py5 = GenericVariant.from_arrow_struct(payload_col[5].as_py()).to_python() + self.assertEqual( + py5['ts_ltz'], + datetime.datetime(2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc), + ) - gv2 = GenericVariant.from_arrow_struct(payload_col[2].as_py()) - self.assertEqual(gv2.to_python(), [1, 2, 3]) + py6 = GenericVariant.from_arrow_struct(payload_col[6].as_py()).to_python() + self.assertEqual(py6['id'], test_uuid) + + py7 = GenericVariant.from_arrow_struct(payload_col[7].as_py()).to_python() + self.assertEqual(py7, test_uuid) def test_plain_variant_null_row(self): """SQL-NULL VARIANT rows are stored and retrieved as None.""" @@ -1116,7 +1227,46 @@ def test_plain_variant_null_row(self): def test_shredded_variant_write_and_read(self): """Shredded VARIANT: writer shreds automatically, reader assembles transparently.""" - shredding_json = _schema_json('payload', [('age', 'BIGINT'), ('city', 'VARCHAR')]) + # Build shredding schema manually because nested ARRAY/ROW require JSON objects, + # not just atomic type strings. + shredding_json = json.dumps({ + 'type': 'ROW', + 'fields': [ + { + 'id': 0, + 'name': 'payload', + 'type': { + 'type': 'ROW', + 'fields': [ + {'name': 'name', 'type': 'VARCHAR'}, + {'name': 'active', 'type': 'BOOLEAN'}, + {'name': 'age', 'type': 'TINYINT'}, + {'name': 'score', 'type': 'SMALLINT'}, + {'name': 'count', 'type': 'INT'}, + {'name': 'id', 'type': 'BIGINT'}, + {'name': 'ratio', 'type': 'DOUBLE'}, + {'name': 'amount', 'type': 'DECIMAL(10,2)'}, + {'name': 'fixed', 'type': 'BINARY'}, + {'name': 'raw', 'type': 'VARBINARY'}, + { + 'name': 'tags', + 'type': {'type': 'ARRAY', 'element': 'INT'}, + }, + { + 'name': 'address', + 'type': { + 'type': 'ROW', + 'fields': [ + {'name': 'city', 'type': 'VARCHAR'}, + {'name': 'zip', 'type': 'INT'}, + ], + }, + }, + ], + }, + } + ], + }) schema = Schema.from_pyarrow_schema( self._pa_schema(), options={'variant.shreddingSchema': shredding_json}, @@ -1125,8 +1275,40 @@ def test_shredded_variant_write_and_read(self): table = self.catalog.get_table('default.shredded_variant') gvs = [ - GenericVariant.from_python({'age': 28, 'city': 'Beijing'}), - GenericVariant.from_python({'age': 35, 'city': 'Shanghai'}), + GenericVariant.from_python( + { + 'name': 'Apache Paimon', + 'active': True, + 'age': 3, + 'score': 3000, + 'count': 400000, + 'id': 12345678901234, + 'ratio': 1.012345678901234567890123456789, + 'amount': Decimal('100.99'), + # BINARY stores raw bytes; fixed-length semantics are Parquet-level. + 'fixed': 'Apache Paimon'.encode('utf-8'), + # VARBINARY stores raw bytes without padding. + 'raw': b'\x01\x02\x03\x04\x05', + 'tags': [1, 2, 3], + 'address': {'city': 'Beijing', 'zip': 100000}, + } + ), + GenericVariant.from_python( + { + 'name': 'Pypaimon', + 'active': False, + 'age': 1, + 'score': 100, + 'count': 42, + 'id': 98765432109876, + 'ratio': 2.718281828459045, + 'amount': Decimal('42.50'), + 'fixed': b'\x00\x01\x02\x03', + 'raw': 'hello'.encode('utf-8'), + 'tags': [10, 20], + 'address': {'city': 'Shanghai', 'zip': 200000}, + } + ), ] data = pa.table( {'id': [1, 2], 'payload': GenericVariant.to_arrow_array(gvs)}, @@ -1138,12 +1320,32 @@ def test_shredded_variant_write_and_read(self): payload_col = result.column('payload') py0 = GenericVariant.from_arrow_struct(payload_col[0].as_py()).to_python() - self.assertEqual(py0['age'], 28) - self.assertEqual(py0['city'], 'Beijing') + self.assertEqual(py0['name'], 'Apache Paimon') + self.assertEqual(py0['active'], True) + self.assertEqual(py0['age'], 3) + self.assertEqual(py0['score'], 3000) + self.assertEqual(py0['count'], 400000) + self.assertEqual(py0['id'], 12345678901234) + self.assertAlmostEqual(py0['ratio'], 1.012345678901234567890123456789) + self.assertEqual(py0['amount'], Decimal('100.99')) + self.assertEqual(py0['fixed'], 'Apache Paimon'.encode('utf-8')) + self.assertEqual(py0['raw'], b'\x01\x02\x03\x04\x05') + self.assertEqual(py0['tags'], [1, 2, 3]) + self.assertEqual(py0['address'], {'city': 'Beijing', 'zip': 100000}) py1 = GenericVariant.from_arrow_struct(payload_col[1].as_py()).to_python() - self.assertEqual(py1['age'], 35) - self.assertEqual(py1['city'], 'Shanghai') + self.assertEqual(py1['name'], 'Pypaimon') + self.assertEqual(py1['active'], False) + self.assertEqual(py1['age'], 1) + self.assertEqual(py1['score'], 100) + self.assertEqual(py1['count'], 42) + self.assertEqual(py1['id'], 98765432109876) + self.assertAlmostEqual(py1['ratio'], 2.718281828459045) + self.assertEqual(py1['amount'], Decimal('42.50')) + self.assertEqual(py1['fixed'], b'\x00\x01\x02\x03') + self.assertEqual(py1['raw'], 'hello'.encode('utf-8')) + self.assertEqual(py1['tags'], [10, 20]) + self.assertEqual(py1['address'], {'city': 'Shanghai', 'zip': 200000}) def test_shredded_variant_overflow_preserved(self): """Fields outside the shredding schema survive in overflow bytes end-to-end.""" From 771db767b43eeb6b4877ebab077a0bebbab64ce0 Mon Sep 17 00:00:00 2001 From: Juntao Zhang Date: Fri, 14 Aug 2026 11:50:21 +0800 Subject: [PATCH 2/2] [python] use pure integer arithmetic to support datetime --- .../pypaimon/data/generic_variant.py | 12 ++++++-- paimon-python/pypaimon/tests/variant_test.py | 29 ++++++++++++++----- 2 files changed, 31 insertions(+), 10 deletions(-) diff --git a/paimon-python/pypaimon/data/generic_variant.py b/paimon-python/pypaimon/data/generic_variant.py index 8d6dee4a7cbf..f6b0dbc2b9d1 100644 --- a/paimon-python/pypaimon/data/generic_variant.py +++ b/paimon-python/pypaimon/data/generic_variant.py @@ -41,6 +41,7 @@ v.metadata() – raw metadata bytes """ +import calendar import datetime import decimal as _decimal import enum @@ -457,11 +458,10 @@ def build_python(self, obj): elif isinstance(obj, _uuid.UUID): self.append_uuid(obj) elif isinstance(obj, datetime.datetime): + micros = self._datetime_to_micros(obj) if obj.tzinfo is not None: - micros = int((obj - _EPOCH_DT_UTC).total_seconds() * 1_000_000) self.append_timestamp(micros) else: - micros = int((obj - _EPOCH_DT_NTZ).total_seconds() * 1_000_000) self.append_timestamp_ntz(micros) elif isinstance(obj, datetime.date): days = (obj - _EPOCH_DATE).days @@ -469,6 +469,14 @@ def build_python(self, obj): else: raise TypeError(f'Unsupported Python type for variant encoding: {type(obj).__name__}') + @staticmethod + def _datetime_to_micros(dt): + """Convert a datetime to microseconds since epoch using pure integer arithmetic. """ + if dt.tzinfo is not None: + dt = dt.astimezone(datetime.timezone.utc) + seconds = calendar.timegm(dt.timetuple()) + return seconds * 1_000_000 + dt.microsecond + def _try_decimal_or_double(self, d): try: sign, digits, exponent = d.as_tuple() diff --git a/paimon-python/pypaimon/tests/variant_test.py b/paimon-python/pypaimon/tests/variant_test.py index 5c16f6f39a68..fd5a6c9fef70 100644 --- a/paimon-python/pypaimon/tests/variant_test.py +++ b/paimon-python/pypaimon/tests/variant_test.py @@ -370,16 +370,29 @@ def test_from_python_date(self): self.assertEqual(gv.to_python(), value) def test_from_python_timestamp_ntz(self): - value = datetime.datetime(2024, 1, 15, 12, 30, 45, 123456) - gv = GenericVariant.from_python(value) - self.assertEqual(gv.to_python(), value) + cases = [ + datetime.datetime(2024, 1, 15, 12, 30, 45, 123456), + datetime.datetime(1600, 6, 15, 12, 30, 45, 123456), + datetime.datetime(1900, 1, 1, 0, 0, 0, 654321), + datetime.datetime(2500, 12, 31, 23, 59, 59, 111111), + datetime.datetime(5000, 3, 3, 3, 3, 3, 222222), + datetime.datetime(9998, 7, 8, 12, 34, 56, 654321), + ] + for value in cases: + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) def test_from_python_timestamp_ltz(self): - value = datetime.datetime( - 2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc - ) - gv = GenericVariant.from_python(value) - self.assertEqual(gv.to_python(), value) + cases = [ + datetime.datetime(2024, 1, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone.utc), + datetime.datetime( + 1600, 6, 15, 12, 30, 45, 123456, tzinfo=datetime.timezone(datetime.timedelta(hours=8))), + datetime.datetime( + 9998, 7, 8, 12, 34, 56, 654321, tzinfo=datetime.timezone(datetime.timedelta(hours=8))), + ] + for value in cases: + gv = GenericVariant.from_python(value) + self.assertEqual(gv.to_python(), value) def test_from_python_nested_datetime(self): obj = {'created_at': datetime.datetime(2024, 1, 15, 12, 0)}