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 bb4a3e4f20 [python] Implement write_row in HdfsNativeFileIO (#10192)
bb4a3e4f20 is described below
commit bb4a3e4f20b3339e9edfc327e49f0b453ae8ae72
Author: jackylee <[email protected]>
AuthorDate: Sat Oct 3 17:10:19 2026 +0800
[python] Implement write_row in HdfsNativeFileIO (#10192)
---
.../pypaimon/filesystem/hdfs_native_file_io.py | 16 +++++
.../pypaimon/tests/hdfs_native_write_row_test.py | 74 ++++++++++++++++++++++
2 files changed, 90 insertions(+)
diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
index 78cad2140e..e06b683ef7 100644
--- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
+++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
@@ -617,6 +617,22 @@ class HdfsNativeFileIO(FileIO):
fastavro.writer(output_stream, avro_schema,
record_generator(), codec=codec, **kwargs)
+ def write_row(self, path: str, data: pyarrow.Table, fields=None,
+ zstd_level: int = 1, **kwargs):
+ try:
+ from pypaimon.write.writer.format_row_writer import FormatRowWriter
+
+ if fields is None:
+ fields = PyarrowFieldParser.to_paimon_schema(data.schema)
+
+ with self.new_output_stream(path) as output_stream:
+ writer = FormatRowWriter(output_stream, fields,
zstd_level=zstd_level)
+ writer.write_table(data)
+ writer.close()
+ except Exception as e:
+ self.delete_quietly(path)
+ raise RuntimeError(f"Failed to write row file {path}: {e}") from e
+
def write_blob(self, path: str, data: pyarrow.Table, **kwargs):
try:
if data.num_columns != 1:
diff --git a/paimon-python/pypaimon/tests/hdfs_native_write_row_test.py
b/paimon-python/pypaimon/tests/hdfs_native_write_row_test.py
new file mode 100644
index 0000000000..a46d4d9aa0
--- /dev/null
+++ b/paimon-python/pypaimon/tests/hdfs_native_write_row_test.py
@@ -0,0 +1,74 @@
+# 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 os
+import tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon.filesystem.hdfs_native_file_io import HdfsNativeFileIO
+from pypaimon.read.reader.format_row_reader import FormatRowReader
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
+class _LocalReadIO:
+ def get_file_size(self, path):
+ return os.path.getsize(path)
+
+ def new_input_stream(self, path):
+ return open(path, 'rb')
+
+
+class HdfsNativeWriteRowTest(unittest.TestCase):
+ """HdfsNativeFileIO is the default hdfs:// backend and must implement
+ write_row; before this it inherited FileIO.write_row's NotImplementedError,
+ so FILE_FORMAT_ROW / data-evolution sidecar writes on native HDFS
crashed."""
+
+ def test_write_row_roundtrips(self):
+ fields = [DataField(0, "id", AtomicType("INT")),
+ DataField(1, "name", AtomicType("STRING"))]
+ table = pa.table({"id": pa.array([1, 2, 3], pa.int32()),
+ "name": ["a", "b", "c"]})
+
+ # Bypass the HDFS __init__; write_row only needs new_output_stream
+ # (and delete_quietly on failure), which we point at the local FS.
+ fio = object.__new__(HdfsNativeFileIO)
+ fio.new_output_stream = lambda p: open(p, 'wb')
+ fio.delete_quietly = lambda p: None
+
+ with tempfile.TemporaryDirectory() as d:
+ path = os.path.join(d, "data.row")
+ fio.write_row(path, table, fields=fields)
+
+ reader = FormatRowReader(_LocalReadIO(), path,
+ [f.name for f in fields], fields, None)
+ batches = []
+ while True:
+ batch = reader.read_arrow_batch()
+ if batch is None:
+ break
+ batches.append(batch)
+ reader.close()
+ result = pa.Table.from_batches(batches)
+
+ self.assertEqual(result.column("id").to_pylist(), [1, 2, 3])
+ self.assertEqual(result.column("name").to_pylist(), ["a", "b", "c"])
+
+
+if __name__ == '__main__':
+ unittest.main()
