You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/03/09 06:29:25 UTC
(paimon) branch master updated: [python] Fix time mapping by always using time32 (#7371)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 41bb2468c8 [python] Fix time mapping by always using time32 (#7371)
41bb2468c8 is described below
commit 41bb2468c818456e31cf367eafa1be73c20024c6
Author: XiaoHongbo <13...@qq.com>
AuthorDate: Mon Mar 9 14:29:20 2026 +0800
[python] Fix time mapping by always using time32 (#7371)
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 43 +++++++++++++---------
paimon-python/pypaimon/common/file_io.py | 29 +++++++++++++++
paimon-python/pypaimon/filesystem/local_file_io.py | 2 +
.../pypaimon/filesystem/pyarrow_file_io.py | 2 +
.../pypaimon/read/reader/format_pyarrow_reader.py | 25 +++++++++++++
paimon-python/pypaimon/schema/data_types.py | 19 ++--------
paimon-python/pypaimon/tests/data_types_test.py | 10 +++++
.../pypaimon/tests/e2e/java_py_read_write_test.py | 16 ++++++--
paimon-python/pypaimon/tests/reader_base_test.py | 8 ++--
.../pypaimon/tests/write/table_write_test.py | 32 ++++++++++++++++
10 files changed, 145 insertions(+), 41 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 ca5e2ef72c..c63456a2c2 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -154,6 +154,7 @@ public class JavaPyE2ETest {
.column("value", DataTypes.DOUBLE())
.column("ts", DataTypes.TIMESTAMP())
.column("ts_ltz", DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE())
+ .column("t", DataTypes.TIME())
.column(
"metadata",
DataTypes.ROW(
@@ -185,7 +186,7 @@ public class JavaPyE2ETest {
write.write(
createRow7Cols(
- 1, "Apple", "Fruit", 1.5, 1000000L, 2000000L, "store1", 1001L,
+ 1, "Apple", "Fruit", 1.5, 1000000L, 2000000L, 1000, "store1", 1001L,
"Beijing", "China"));
write.write(
createRow7Cols(
@@ -195,6 +196,7 @@ public class JavaPyE2ETest {
0.8,
1000001L,
2000001L,
+ 2000,
"store1",
1002L,
"Shanghai",
@@ -207,6 +209,7 @@ public class JavaPyE2ETest {
0.6,
1000002L,
2000002L,
+ 3000,
"store2",
1003L,
"Tokyo",
@@ -219,17 +222,18 @@ public class JavaPyE2ETest {
1.2,
1000003L,
2000003L,
+ 4000,
"store2",
1004L,
"Seoul",
"Korea"));
write.write(
createRow7Cols(
- 5, "Chicken", "Meat", 5.0, 1000004L, 2000004L, "store3", 1005L,
- "NewYork", "USA"));
+ 5, "Chicken", "Meat", 5.0, 1000004L, 2000004L, 5000, "store3",
+ 1005L, "NewYork", "USA"));
write.write(
createRow7Cols(
- 6, "Beef", "Meat", 8.0, 1000005L, 2000005L, "store3", 1006L,
+ 6, "Beef", "Meat", 8.0, 1000005L, 2000005L, 6000, "store3", 1006L,
"London", "UK"));
commit.commit(0, write.prepareCommit(true, 0));
@@ -242,12 +246,12 @@ public class JavaPyE2ETest {
getResult(read, splits, row -> rowToStringWithStruct(row, table.rowType()));
assertThat(res)
.containsExactlyInAnyOrder(
- "+I[1, Apple, Fruit, 1.5, 1970-01-01T00:16:40, 1970-01-01T00:33:20, (store1, 1001, (Beijing, China))]",
- "+I[2, Banana, Fruit, 0.8, 1970-01-01T00:16:40.001, 1970-01-01T00:33:20.001, (store1, 1002, (Shanghai, China))]",
- "+I[3, Carrot, Vegetable, 0.6, 1970-01-01T00:16:40.002, 1970-01-01T00:33:20.002, (store2, 1003, (Tokyo, Japan))]",
- "+I[4, Broccoli, Vegetable, 1.2, 1970-01-01T00:16:40.003, 1970-01-01T00:33:20.003, (store2, 1004, (Seoul, Korea))]",
- "+I[5, Chicken, Meat, 5.0, 1970-01-01T00:16:40.004, 1970-01-01T00:33:20.004, (store3, 1005, (NewYork, USA))]",
- "+I[6, Beef, Meat, 8.0, 1970-01-01T00:16:40.005, 1970-01-01T00:33:20.005, (store3, 1006, (London, UK))]");
+ "+I[1, Apple, Fruit, 1.5, 1970-01-01T00:16:40, 1970-01-01T00:33:20, 1000, (store1, 1001, (Beijing, China))]",
+ "+I[2, Banana, Fruit, 0.8, 1970-01-01T00:16:40.001, 1970-01-01T00:33:20.001, 2000, (store1, 1002, (Shanghai, China))]",
+ "+I[3, Carrot, Vegetable, 0.6, 1970-01-01T00:16:40.002, 1970-01-01T00:33:20.002, 3000, (store2, 1003, (Tokyo, Japan))]",
+ "+I[4, Broccoli, Vegetable, 1.2, 1970-01-01T00:16:40.003, 1970-01-01T00:33:20.003, 4000, (store2, 1004, (Seoul, Korea))]",
+ "+I[5, Chicken, Meat, 5.0, 1970-01-01T00:16:40.004, 1970-01-01T00:33:20.004, 5000, (store3, 1005, (NewYork, USA))]",
+ "+I[6, Beef, Meat, 8.0, 1970-01-01T00:16:40.005, 1970-01-01T00:33:20.005, 6000, (store3, 1006, (London, UK))]");
}
}
@@ -401,17 +405,18 @@ public class JavaPyE2ETest {
assertThat(table.rowType().getFieldTypes().get(4)).isEqualTo(DataTypes.TIMESTAMP());
assertThat(table.rowType().getFieldTypes().get(5))
.isEqualTo(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE());
- assertThat(table.rowType().getFieldTypes().get(6)).isInstanceOf(RowType.class);
- RowType metadataType = (RowType) table.rowType().getFieldTypes().get(6);
+ assertThat(table.rowType().getFieldTypes().get(6)).isEqualTo(DataTypes.TIME());
+ assertThat(table.rowType().getFieldTypes().get(7)).isInstanceOf(RowType.class);
+ RowType metadataType = (RowType) table.rowType().getFieldTypes().get(7);
assertThat(metadataType.getFieldTypes().get(2)).isInstanceOf(RowType.class);
assertThat(res)
.containsExactlyInAnyOrder(
- "+I[1, Apple, Fruit, 1.5, 1970-01-01T00:16:40, 1970-01-01T00:33:20, (store1, 1001, (Beijing, China))]",
- "+I[2, Banana, Fruit, 0.8, 1970-01-01T00:16:40.001, 1970-01-01T00:33:20.001, (store1, 1002, (Shanghai, China))]",
- "+I[3, Carrot, Vegetable, 0.6, 1970-01-01T00:16:40.002, 1970-01-01T00:33:20.002, (store2, 1003, (Tokyo, Japan))]",
- "+I[4, Broccoli, Vegetable, 1.2, 1970-01-01T00:16:40.003, 1970-01-01T00:33:20.003, (store2, 1004, (Seoul, Korea))]",
- "+I[5, Chicken, Meat, 5.0, 1970-01-01T00:16:40.004, 1970-01-01T00:33:20.004, (store3, 1005, (NewYork, USA))]",
- "+I[6, Beef, Meat, 8.0, 1970-01-01T00:16:40.005, 1970-01-01T00:33:20.005, (store3, 1006, (London, UK))]");
+ "+I[1, Apple, Fruit, 1.5, 1970-01-01T00:16:40, 1970-01-01T00:33:20, 1000, (store1, 1001, (Beijing, China))]",
+ "+I[2, Banana, Fruit, 0.8, 1970-01-01T00:16:40.001, 1970-01-01T00:33:20.001, 2000, (store1, 1002, (Shanghai, China))]",
+ "+I[3, Carrot, Vegetable, 0.6, 1970-01-01T00:16:40.002, 1970-01-01T00:33:20.002, 3000, (store2, 1003, (Tokyo, Japan))]",
+ "+I[4, Broccoli, Vegetable, 1.2, 1970-01-01T00:16:40.003, 1970-01-01T00:33:20.003, 4000, (store2, 1004, (Seoul, Korea))]",
+ "+I[5, Chicken, Meat, 5.0, 1970-01-01T00:16:40.004, 1970-01-01T00:33:20.004, 5000, (store3, 1005, (NewYork, USA))]",
+ "+I[6, Beef, Meat, 8.0, 1970-01-01T00:16:40.005, 1970-01-01T00:33:20.005, 6000, (store3, 1006, (London, UK))]");
PredicateBuilder predicateBuilder = new PredicateBuilder(table.rowType());
int[] ids = {1, 2, 3, 4, 5, 6};
@@ -704,6 +709,7 @@ public class JavaPyE2ETest {
double value,
long ts,
long tsLtz,
+ int timeMillis,
String metadataSource,
long metadataCreatedAt,
String city,
@@ -720,6 +726,7 @@ public class JavaPyE2ETest {
value,
org.apache.paimon.data.Timestamp.fromEpochMillis(ts),
org.apache.paimon.data.Timestamp.fromEpochMillis(tsLtz),
+ timeMillis,
metadataRow);
}
diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py
index c5e447d5b6..31539a9f00 100644
--- a/paimon-python/pypaimon/common/file_io.py
+++ b/paimon-python/pypaimon/common/file_io.py
@@ -206,6 +206,35 @@ class FileIO(ABC):
zstd_level: int = 1, **kwargs):
raise NotImplementedError("write_parquet must be implemented by FileIO subclasses")
+ @staticmethod
+ def _cast_time_columns_for_orc(data):
+ """Cast time32 columns to int32 before writing ORC.
+
+ PyArrow's ORC writer does not support time types.
+ """
+ has_time = any(pyarrow.types.is_time(f.type) for f in data.schema)
+ if not has_time:
+ return data
+ columns = []
+ for i, field in enumerate(data.schema):
+ col = data.column(i)
+ if pyarrow.types.is_time(field.type):
+ if not pyarrow.types.is_time32(field.type) \
+ or field.type != pyarrow.time32('ms'):
+ raise ValueError(
+ "Column '{}' has type {} which cannot be safely cast to int32 "
+ "for ORC writing. Use time32('ms') instead."
+ .format(field.name, field.type)
+ )
+ col = col.cast(pyarrow.int32())
+ columns.append(col)
+ orc_schema = pyarrow.schema([
+ pyarrow.field(f.name, pyarrow.int32(), f.nullable) if pyarrow.types.is_time(f.type)
+ else f
+ for f in data.schema
+ ])
+ return pyarrow.table(columns, schema=orc_schema)
+
def write_orc(self, path: str, data, compression: str = 'zstd',
zstd_level: int = 1, **kwargs):
raise NotImplementedError("write_orc must be implemented by FileIO subclasses")
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py b/paimon-python/pypaimon/filesystem/local_file_io.py
index 5726d601fe..cf9399309f 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -318,6 +318,8 @@ class LocalFileIO(FileIO):
if parent and not parent.exists():
parent.mkdir(parents=True, exist_ok=True)
+ data = self._cast_time_columns_for_orc(data)
+
with open(file_path, 'wb') as f:
if sys.version_info[:2] == (3, 6):
orc.write_table(data, f, **kwargs)
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 332898b59f..55f055ae9a 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -382,6 +382,8 @@ class PyArrowFileIO(FileIO):
import sys
import pyarrow.orc as orc
+ data = self._cast_time_columns_for_orc(data)
+
with self.new_output_stream(path) as output_stream:
# Check Python version - if 3.6, don't use compression parameter
if sys.version_info[:2] == (3, 6):
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index e9c9efd917..be80d63146 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -39,6 +39,7 @@ class FormatPyArrowReader(RecordBatchReader):
push_down_predicate: Any, batch_size: int = 1024):
file_path_for_pyarrow = file_io.to_filesystem_path(file_path)
self.dataset = ds.dataset(file_path_for_pyarrow, format=file_format, filesystem=file_io.filesystem)
+ self._file_format = file_format
self.read_fields = read_fields
self._read_field_names = [f.name for f in read_fields]
@@ -62,6 +63,9 @@ class FormatPyArrowReader(RecordBatchReader):
try:
batch = self.reader.read_next_batch()
+ if self._file_format == 'orc' and self._output_schema is not None:
+ batch = self._cast_orc_time_columns(batch)
+
if not self.missing_fields:
return batch
@@ -99,6 +103,27 @@ class FormatPyArrowReader(RecordBatchReader):
except StopIteration:
return None
+ def _cast_orc_time_columns(self, batch):
+ """Cast int32 TIME columns back to time32('ms') when reading ORC.
+ """
+ columns = []
+ fields = []
+ changed = False
+ for i, name in enumerate(batch.schema.names):
+ col = batch.column(i)
+ idx = self._output_schema.get_field_index(name)
+ if idx >= 0 and pa.types.is_int32(col.type) \
+ and pa.types.is_time(self._output_schema.field(idx).type):
+ col = col.cast(self._output_schema.field(idx).type)
+ fields.append(self._output_schema.field(idx))
+ changed = True
+ else:
+ fields.append(batch.schema.field(i))
+ columns.append(col)
+ if changed:
+ return pa.RecordBatch.from_arrays(columns, schema=pa.schema(fields))
+ return batch
+
def close(self):
if self.reader is not None:
self.reader = None
diff --git a/paimon-python/pypaimon/schema/data_types.py b/paimon-python/pypaimon/schema/data_types.py
index 6befaa4d40..0d95e7b95c 100755
--- a/paimon-python/pypaimon/schema/data_types.py
+++ b/paimon-python/pypaimon/schema/data_types.py
@@ -520,19 +520,7 @@ class PyarrowFieldParser:
elif type_name == 'DATE':
return pyarrow.date32()
if type_name.startswith('TIME'):
- if type_name == 'TIME':
- return pyarrow.time64('us') # default to 6
- match = re.fullmatch(r'TIME\((\d+)\)', type_name)
- if match:
- precision = int(match.group(1))
- if precision == 0:
- return pyarrow.time32('s')
- if 1 <= precision <= 3:
- return pyarrow.time32('ms')
- if 4 <= precision <= 6:
- return pyarrow.time64('us')
- if 7 <= precision <= 9:
- return pyarrow.time64('ns')
+ return pyarrow.time32('ms')
elif isinstance(data_type, ArrayType):
return pyarrow.list_(PyarrowFieldParser.from_paimon_type(data_type.element))
elif isinstance(data_type, MapType):
@@ -601,8 +589,7 @@ class PyarrowFieldParser:
elif types.is_date32(pa_type):
type_name = 'DATE'
elif types.is_time(pa_type):
- precision_mapping = {'s': 0, 'ms': 3, 'us': 6, 'ns': 9}
- type_name = f'TIME({precision_mapping[pa_type.unit]})'
+ type_name = 'TIME(0)'
elif types.is_list(pa_type) or types.is_large_list(pa_type):
pa_type: pyarrow.ListType
element_type = PyarrowFieldParser.to_paimon_type(pa_type.value_type, nullable)
@@ -677,6 +664,8 @@ class PyarrowFieldParser:
}
elif pyarrow.types.is_date(field_type):
return {"type": "int", "logicalType": "date"}
+ elif pyarrow.types.is_time(field_type):
+ return {"type": "int", "logicalType": "time-millis"}
elif pyarrow.types.is_timestamp(field_type):
unit = field_type.unit
if field_type.tz is None:
diff --git a/paimon-python/pypaimon/tests/data_types_test.py b/paimon-python/pypaimon/tests/data_types_test.py
index 6ace872997..de28235f38 100755
--- a/paimon-python/pypaimon/tests/data_types_test.py
+++ b/paimon-python/pypaimon/tests/data_types_test.py
@@ -134,3 +134,13 @@ class DataTypesTest(unittest.TestCase):
self.assertEqual(len(converted_nested_field.fields), 2)
self.assertEqual(converted_nested_field.fields[0].name, "inner_field1")
self.assertEqual(converted_nested_field.fields[1].name, "inner_field2")
+
+ def test_time_type(self):
+ pa_type = PyarrowFieldParser.from_paimon_type(AtomicType("TIME"))
+ self.assertEqual(pa_type, pa.time32('ms'))
+
+ pa_type_with_precision = PyarrowFieldParser.from_paimon_type(AtomicType("TIME(3)"))
+ self.assertEqual(pa_type_with_precision, pa.time32('ms'))
+
+ paimon_type = PyarrowFieldParser.to_paimon_type(pa.time32('ms'), nullable=True)
+ self.assertEqual(paimon_type.type, "TIME(0)")
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 98ae212a68..a5f9a8e3af 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
@@ -16,6 +16,7 @@
# limitations under the License.
################################################################################
+import datetime
import os
import sys
import unittest
@@ -59,7 +60,8 @@ class JavaPyReadWriteTest(unittest.TestCase):
('category', pa.string()),
('value', pa.float64()),
('ts', pa.timestamp('us')),
- ('ts_ltz', pa.timestamp('us', tz='UTC'))
+ ('ts_ltz', pa.timestamp('us', tz='UTC')),
+ ('t', pa.time32('ms'))
])
schema = Schema.from_pyarrow_schema(
@@ -78,7 +80,9 @@ class JavaPyReadWriteTest(unittest.TestCase):
'category': ['Fruit', 'Fruit', 'Vegetable', 'Vegetable', 'Meat', 'Meat'],
'value': [1.5, 0.8, 0.6, 1.2, 5.0, 8.0],
'ts': pd.to_datetime([1000000, 1000001, 1000002, 1000003, 1000004, 1000005], unit='ms'),
- 'ts_ltz': pd.to_datetime([2000000, 2000001, 2000002, 2000003, 2000004, 2000005], unit='ms', utc=True)
+ 'ts_ltz': pd.to_datetime([2000000, 2000001, 2000002, 2000003, 2000004, 2000005], unit='ms', utc=True),
+ 't': [datetime.time(0, 0, 1), datetime.time(0, 0, 2), datetime.time(0, 0, 3),
+ datetime.time(0, 0, 4), datetime.time(0, 0, 5), datetime.time(0, 0, 6)]
})
# Write initial data
write_builder = table.new_batch_write_builder()
@@ -129,6 +133,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
('value', pa.float64()),
('ts', pa.timestamp('us')),
('ts_ltz', pa.timestamp('us', tz='UTC')),
+ ('t', pa.time32('ms')),
('metadata', pa.struct([
pa.field('source', pa.string()),
pa.field('created_at', pa.int64()),
@@ -179,6 +184,8 @@ class JavaPyReadWriteTest(unittest.TestCase):
'value': [1.5, 0.8, 0.6, 1.2, 5.0, 8.0],
'ts': pd.to_datetime([1000000, 1000001, 1000002, 1000003, 1000004, 1000005], unit='ms'),
'ts_ltz': pd.to_datetime([2000000, 2000001, 2000002, 2000003, 2000004, 2000005], unit='ms', utc=True),
+ 't': [datetime.time(0, 0, 1), datetime.time(0, 0, 2), datetime.time(0, 0, 3),
+ datetime.time(0, 0, 4), datetime.time(0, 0, 5), datetime.time(0, 0, 6)],
'metadata': [
{'source': 'store1', 'created_at': 1001, 'location': {'city': 'Beijing', 'country': 'China'}},
{'source': 'store1', 'created_at': 1002, 'location': {'city': 'Shanghai', 'country': 'China'}},
@@ -232,9 +239,10 @@ class JavaPyReadWriteTest(unittest.TestCase):
if file_format != "lance":
self.assertEqual(table.fields[4].type.type, "TIMESTAMP(6)")
self.assertEqual(table.fields[5].type.type, "TIMESTAMP(6) WITH LOCAL TIME ZONE")
+ self.assertEqual(table.fields[6].type.type, "TIME(0)")
from pypaimon.schema.data_types import RowType
- self.assertIsInstance(table.fields[6].type, RowType)
- metadata_fields = table.fields[6].type.fields
+ self.assertIsInstance(table.fields[7].type, RowType)
+ metadata_fields = table.fields[7].type.fields
self.assertEqual(len(metadata_fields), 3)
self.assertEqual(metadata_fields[0].name, 'source')
self.assertEqual(metadata_fields[1].name, 'created_at')
diff --git a/paimon-python/pypaimon/tests/reader_base_test.py b/paimon-python/pypaimon/tests/reader_base_test.py
index 8a07c66239..6445111dfa 100644
--- a/paimon-python/pypaimon/tests/reader_base_test.py
+++ b/paimon-python/pypaimon/tests/reader_base_test.py
@@ -178,7 +178,7 @@ class ReaderBasicTest(unittest.TestCase):
('f10', pa.decimal128(10, 2)),
('f11', pa.timestamp('ms')),
('f12', pa.date32()),
- ('f13', pa.time64('us')),
+ ('f13', pa.time32('ms')),
])
stats_enabled = random.random() < 0.5
options = {'metadata.stats-mode': 'full'} if stats_enabled else {}
@@ -640,9 +640,9 @@ class ReaderBasicTest(unittest.TestCase):
DataField(10, "f10", AtomicType('BYTES'), 'desc'),
DataField(11, "f11", AtomicType('DATE'), 'desc'),
DataField(12, "f12", AtomicType('TIME(0)'), 'desc'),
- DataField(13, "f13", AtomicType('TIME(3)'), 'desc'),
- DataField(14, "f14", AtomicType('TIME(6)'), 'desc'),
- DataField(15, "f15", AtomicType('TIME(9)'), 'desc'),
+ DataField(13, "f13", AtomicType('TIME(0)'), 'desc'),
+ DataField(14, "f14", AtomicType('TIME(0)'), 'desc'),
+ DataField(15, "f15", AtomicType('TIME(0)'), 'desc'),
DataField(16, "f16", AtomicType('TIMESTAMP(0)'), 'desc'),
DataField(17, "f17", AtomicType('TIMESTAMP(3)'), 'desc'),
DataField(18, "f18", AtomicType('TIMESTAMP(6)'), 'desc'),
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py b/paimon-python/pypaimon/tests/write/table_write_test.py
index 34a892a725..22f6759610 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -16,6 +16,7 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
import glob
+import datetime
import os
import shutil
@@ -24,6 +25,7 @@ import unittest
from pypaimon import CatalogFactory, Schema
import pyarrow as pa
+from parameterized import parameterized
class TableWriteTest(unittest.TestCase):
@@ -384,3 +386,33 @@ class TableWriteTest(unittest.TestCase):
self.expected.sort_by(sort_keys),
actual.sort_by(sort_keys),
)
+
+ @parameterized.expand([('parquet',), ('orc',), ('avro',)])
+ def test_write_time_type(self, file_format):
+ time_schema = pa.schema([
+ ('id', pa.int32()),
+ ('t', pa.time32('ms'))
+ ])
+ expected = pa.Table.from_pydict({
+ 'id': [1, 2, 3],
+ 't': [datetime.time(0, 0, 1), datetime.time(0, 0, 2), datetime.time(0, 0, 3)]
+ }, schema=time_schema)
+
+ table_name = 'default.test_write_time_' + file_format
+ schema = Schema.from_pyarrow_schema(time_schema, options={'file.format': file_format})
+ self.catalog.create_table(table_name, schema, False)
+ table = self.catalog.get_table(table_name)
+
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(expected)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ read_builder = table.new_read_builder()
+ table_read = read_builder.new_read()
+ splits = read_builder.new_scan().plan().splits()
+ actual = table_read.to_arrow(splits)
+ self.assertEqual(expected, actual)