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 4041b51a43 [python] Pass Mosaic options to writer (#8400)
4041b51a43 is described below
commit 4041b51a43cf0f350ef072354a9630d5d7cd8fb6
Author: QuakeWang <[email protected]>
AuthorDate: Wed Jul 1 13:29:04 2026 +0800
[python] Pass Mosaic options to writer (#8400)
Python Mosaic writes built data files with `mosaic.write_table(data,
stream)` and did not pass table options into Mosaic `WriterOptions`. As
a result, Python tables using `file.format=mosaic` ignored configured
`mosaic.num-buckets`, `mosaic.stats-columns`, `file.block-size`, and
`file.compression.zstd-level`.
This PR maps the relevant table options into Mosaic writer options and
passes them through Python data writers and FileIO implementations.
---
.../pypaimon/common/options/core_options.py | 30 +++++
.../pypaimon/filesystem/hdfs_native_file_io.py | 2 +-
paimon-python/pypaimon/filesystem/local_file_io.py | 2 +-
.../pypaimon/filesystem/pyarrow_file_io.py | 2 +-
.../pypaimon/tests/mosaic_writer_options_test.py | 132 +++++++++++++++++++++
.../pypaimon/write/writer/data_vector_writer.py | 2 +-
paimon-python/pypaimon/write/writer/data_writer.py | 8 +-
.../write/writer/dedicated_format_writer.py | 2 +-
.../pypaimon/write/writer/mosaic_writer_options.py | 44 +++++++
9 files changed, 218 insertions(+), 6 deletions(-)
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 2dc8049601..56b15ebfe6 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -263,6 +263,23 @@ class CoreOptions:
.with_description("Define the data block size.")
)
+ MOSAIC_STATS_COLUMNS: ConfigOption[str] = (
+ ConfigOptions.key("mosaic.stats-columns")
+ .string_type()
+ .default_value("")
+ .with_description(
+ "Comma-separated list of column names to collect statistics for. "
+ "Empty means no statistics collection."
+ )
+ )
+
+ MOSAIC_NUM_BUCKETS: ConfigOption[int] = (
+ ConfigOptions.key("mosaic.num-buckets")
+ .int_type()
+ .no_default_value()
+ .with_description("Number of column buckets for parallel IO.")
+ )
+
METADATA_STATS_MODE: ConfigOption[str] = (
ConfigOptions.key("metadata.stats-mode")
.string_type()
@@ -940,6 +957,19 @@ class CoreOptions:
def file_block_size(self, default=None):
return self.options.get(CoreOptions.FILE_BLOCK_SIZE, default)
+ def mosaic_stats_columns(self, default=None):
+ value = self.options.get(CoreOptions.MOSAIC_STATS_COLUMNS, default)
+ if value is None:
+ return []
+ if isinstance(value, str):
+ return [column.strip() for column in value.split(",") if
column.strip()]
+ if isinstance(value, (list, set, tuple)):
+ return [str(column).strip() for column in value if
str(column).strip()]
+ return []
+
+ def mosaic_num_buckets(self, default=None):
+ return self.options.get(CoreOptions.MOSAIC_NUM_BUCKETS, default)
+
def metadata_stats_enabled(self, default=None):
return self.options.get(CoreOptions.METADATA_STATS_MODE, default) ==
"full"
diff --git a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
index aa68e3b747..3c3d45ad86 100644
--- a/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
+++ b/paimon-python/pypaimon/filesystem/hdfs_native_file_io.py
@@ -671,7 +671,7 @@ class HdfsNativeFileIO(FileIO):
try:
import mosaic
with self.new_output_stream(path) as output_stream:
- mosaic.write_table(data, output_stream)
+ mosaic.write_table(data, output_stream,
options=kwargs.get("options"))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Mosaic file {path}: {e}")
from e
diff --git a/paimon-python/pypaimon/filesystem/local_file_io.py
b/paimon-python/pypaimon/filesystem/local_file_io.py
index d3f5f81f4f..c9848705aa 100644
--- a/paimon-python/pypaimon/filesystem/local_file_io.py
+++ b/paimon-python/pypaimon/filesystem/local_file_io.py
@@ -398,7 +398,7 @@ class LocalFileIO(FileIO):
import mosaic
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'wb') as f:
- mosaic.write_table(data, f)
+ mosaic.write_table(data, f, options=kwargs.get("options"))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Mosaic file {path}: {e}")
from e
diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
index 12d3e91b5c..c9a3a6437c 100644
--- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
+++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py
@@ -646,7 +646,7 @@ class PyArrowFileIO(FileIO):
try:
import mosaic
with self.new_output_stream(path) as output_stream:
- mosaic.write_table(data, output_stream)
+ mosaic.write_table(data, output_stream,
options=kwargs.get("options"))
except Exception as e:
self.delete_quietly(path)
raise RuntimeError(f"Failed to write Mosaic file {path}: {e}")
from e
diff --git a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
new file mode 100644
index 0000000000..0eac7c3849
--- /dev/null
+++ b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
@@ -0,0 +1,132 @@
+# 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 sys
+import types
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.filesystem.local_file_io import LocalFileIO
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
+from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
+
+
+class _FakeWriterOptions:
+ def __init__(self, **kwargs):
+ self.kwargs = kwargs
+
+
+class _PathFactory:
+ def __init__(self, bucket_path):
+ self._bucket_path = bucket_path
+
+ def create_external_path_provider(self, partition, bucket):
+ return None
+
+ def bucket_path(self, partition, bucket):
+ return self._bucket_path
+
+
+class _FileIO:
+ def __init__(self):
+ self.mosaic_options = None
+
+ def write_mosaic(self, path, data, **kwargs):
+ self.mosaic_options = kwargs.get("options")
+
+ def get_file_size(self, path):
+ return 1
+
+ def delete_quietly(self, path):
+ pass
+
+
+def test_create_mosaic_writer_options_maps_table_options(monkeypatch):
+ fake_mosaic = types.SimpleNamespace(WriterOptions=_FakeWriterOptions)
+ monkeypatch.setitem(sys.modules, "mosaic", fake_mosaic)
+ options = CoreOptions.from_dict({
+ "file.compression.zstd-level": "5",
+ "file.block-size": "64kb",
+ "mosaic.num-buckets": "8",
+ "mosaic.stats-columns": " id, name ,,",
+ })
+
+ writer_options = create_mosaic_writer_options(options)
+
+ assert writer_options.kwargs == {
+ "zstd_level": 5,
+ "num_buckets": 8,
+ "row_group_max_size": 64 * 1024,
+ "stats_columns": ["id", "name"],
+ }
+
+
+def
test_create_mosaic_writer_options_rejects_non_zstd_compression(monkeypatch):
+ fake_mosaic = types.SimpleNamespace(WriterOptions=_FakeWriterOptions)
+ monkeypatch.setitem(sys.modules, "mosaic", fake_mosaic)
+ options = CoreOptions.from_dict({"file.compression": "snappy"})
+
+ with pytest.raises(ValueError, match="Mosaic format only supports zstd
compression"):
+ create_mosaic_writer_options(options)
+
+
+def test_write_mosaic_local_file_io_passes_writer_options(monkeypatch,
tmp_path):
+ captured = {}
+ fake_mosaic = types.SimpleNamespace(
+ write_table=lambda data, output_stream, options=None: captured.update(
+ data=data, options=options))
+ monkeypatch.setitem(sys.modules, "mosaic", fake_mosaic)
+
+ data = pa.table({"id": pa.array([1], type=pa.int32())})
+ writer_options = object()
+ file_io = LocalFileIO({})
+
+ file_io.write_mosaic(str(tmp_path / "data.mosaic"), data,
options=writer_options)
+
+ assert captured["data"] is data
+ assert captured["options"] is writer_options
+
+
+def test_data_writer_passes_mosaic_writer_options(monkeypatch, tmp_path):
+ fake_mosaic = types.SimpleNamespace(WriterOptions=_FakeWriterOptions)
+ monkeypatch.setitem(sys.modules, "mosaic", fake_mosaic)
+ options = CoreOptions.from_dict({
+ "file.format": "mosaic",
+ "mosaic.num-buckets": "4",
+ "mosaic.stats-columns": "id",
+ })
+ file_io = _FileIO()
+ table = types.SimpleNamespace(
+ file_io=file_io,
+ options=options,
+ is_primary_key_table=True,
+ table_schema=types.SimpleNamespace(id=0),
+ trimmed_primary_keys=["id"],
+ trimmed_primary_keys_fields=[DataField(0, "id", AtomicType("INT"))],
+ fields=[DataField(0, "id", AtomicType("INT"))],
+ path_factory=lambda: _PathFactory(str(tmp_path)),
+ )
+ writer = AppendOnlyDataWriter(table, (), 0, 0, options)
+
+ writer._write_data_to_file(pa.table({"id": pa.array([1],
type=pa.int32())}))
+
+ assert isinstance(file_io.mosaic_options, _FakeWriterOptions)
+ assert file_io.mosaic_options.kwargs["num_buckets"] == 4
+ assert file_io.mosaic_options.kwargs["stats_columns"] == ["id"]
diff --git a/paimon-python/pypaimon/write/writer/data_vector_writer.py
b/paimon-python/pypaimon/write/writer/data_vector_writer.py
index f959a35b45..b474957870 100644
--- a/paimon-python/pypaimon/write/writer/data_vector_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_vector_writer.py
@@ -214,7 +214,7 @@ class DataVectorWriter(DataWriter):
elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
self.file_io.write_vortex(file_path, data)
elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data)
+ self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
self.file_io.write_row(file_path, data, zstd_level=self.zstd_level)
else:
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index c32bb6968b..221f0f0498 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -29,6 +29,7 @@ from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
class DataWriter(ABC):
@@ -60,6 +61,11 @@ class DataWriter(ABC):
self.file_format = self.options.file_format(default_format)
self.compression = self.options.file_compression()
self.zstd_level = self.options.file_compression_zstd_level()
+ self.mosaic_writer_options = (
+ create_mosaic_writer_options(self.options)
+ if self.file_format == CoreOptions.FILE_FORMAT_MOSAIC
+ else None
+ )
self.sequence_generator = SequenceGenerator(max_seq_number)
self.pending_data: Optional[pa.Table] = None
@@ -222,7 +228,7 @@ class DataWriter(ABC):
elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
self.file_io.write_vortex(file_path, data)
elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data)
+ self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
self.file_io.write_row(file_path, data,
zstd_level=self.zstd_level)
else:
diff --git a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
index 0ec7439359..4dc4fca2a5 100644
--- a/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
+++ b/paimon-python/pypaimon/write/writer/dedicated_format_writer.py
@@ -396,7 +396,7 @@ class DedicatedFormatWriter(DataWriter):
elif self.file_format == CoreOptions.FILE_FORMAT_VORTEX:
self.file_io.write_vortex(file_path, data)
elif self.file_format == CoreOptions.FILE_FORMAT_MOSAIC:
- self.file_io.write_mosaic(file_path, data)
+ self.file_io.write_mosaic(file_path, data,
options=self.mosaic_writer_options)
elif self.file_format == CoreOptions.FILE_FORMAT_ROW:
self.file_io.write_row(file_path, data, zstd_level=self.zstd_level)
else:
diff --git a/paimon-python/pypaimon/write/writer/mosaic_writer_options.py
b/paimon-python/pypaimon/write/writer/mosaic_writer_options.py
new file mode 100644
index 0000000000..974c0eb66f
--- /dev/null
+++ b/paimon-python/pypaimon/write/writer/mosaic_writer_options.py
@@ -0,0 +1,44 @@
+# 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.
+
+from pypaimon.common.options.core_options import CoreOptions
+
+
+def create_mosaic_writer_options(options: CoreOptions):
+ import mosaic
+
+ compression = options.file_compression()
+ if compression is not None and compression.lower() != "zstd":
+ raise ValueError(f"Mosaic format only supports zstd compression, but
got: {compression}")
+
+ kwargs = {
+ "zstd_level": options.file_compression_zstd_level(),
+ }
+
+ num_buckets = options.mosaic_num_buckets()
+ if num_buckets is not None:
+ kwargs["num_buckets"] = num_buckets
+
+ block_size = options.file_block_size()
+ if block_size is not None:
+ kwargs["row_group_max_size"] = block_size.get_bytes()
+
+ stats_columns = options.mosaic_stats_columns()
+ if stats_columns:
+ kwargs["stats_columns"] = stats_columns
+
+ return mosaic.WriterOptions(**kwargs)
