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 0b67f36c3e [python] Enable Parquet page indexes when supported (#9976)
0b67f36c3e is described below
commit 0b67f36c3ed0d7cc139ce903e010f17263d4fc73
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Sep 20 13:31:01 2026 +0800
[python] Enable Parquet page indexes when supported (#9976)
---
docs/docs/pypaimon/writing.md | 10 +
.../pypaimon/common/options/core_options.py | 14 ++
.../pypaimon/table/format/format_table_write.py | 12 +-
.../tests/parquet_write_page_index_test.py | 250 +++++++++++++++++++++
.../pypaimon/write/map_shared_shredding_writer.py | 4 +-
.../pypaimon/write/table_update_by_row_id.py | 3 +-
paimon-python/pypaimon/write/writer/data_writer.py | 12 +-
.../write/writer/parquet_writer_options.py | 31 +++
.../pypaimon/write/writer/single_file_writer.py | 4 +-
.../pypaimon/write/writer/vector_writer.py | 8 +-
10 files changed, 338 insertions(+), 10 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index d42f5c4b06..e1d79ce544 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -83,6 +83,16 @@ write_builder = table.new_batch_write_builder().overwrite()
write_builder = table.new_batch_write_builder().overwrite({'dt': '2024-01-01'})
```
+### Parquet Page Indexes
+
+Python Parquet writers write page indexes by default on PyArrow >= 13; older
+versions omit them. Set `parquet.write-page-index.enabled=false` to disable
them,
+or `true` to require them. Explicit `true` on PyArrow < 13 fails before
creating files.
+
+This affects new files only, independently of read-side filtering. When
enabled,
+page-level statistics move from page headers into the index. Check reader
+compatibility when upgrading; PyArrow reads do not currently use these indexes.
+
### Manifest Merging
`manifest.merge.skip-on-write-only` defaults to `false` in both Python and
Java,
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 50e48ab3ca..bb312781bf 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -350,6 +350,17 @@ class CoreOptions:
)
# File format options
+ PARQUET_WRITE_PAGE_INDEX_ENABLED: ConfigOption[bool] = (
+ ConfigOptions.key("parquet.write-page-index.enabled")
+ .boolean_type()
+ .no_default_value()
+ .with_description(
+ "Whether Python Parquet writers write page indexes. Enabled by
default on PyArrow >= 13, "
+ "disabled on older versions. Explicit true requires PyArrow >= 13.
"
+ "Only affects newly written files; independent of read-side index
filtering."
+ )
+ )
+
FILE_FORMAT: ConfigOption[str] = (
ConfigOptions.key("file.format")
.string_type()
@@ -1325,6 +1336,9 @@ class CoreOptions:
def file_format(self, default=None):
return self.options.get(CoreOptions.FILE_FORMAT, default)
+ def parquet_write_page_index_enabled(self) -> Optional[bool]:
+ return self.options.get(CoreOptions.PARQUET_WRITE_PAGE_INDEX_ENABLED)
+
def file_compression(self, default=None):
return self.options.get(CoreOptions.FILE_COMPRESSION, default)
diff --git a/paimon-python/pypaimon/table/format/format_table_write.py
b/paimon-python/pypaimon/table/format/format_table_write.py
index 75b22d7186..ed33243730 100644
--- a/paimon-python/pypaimon/table/format/format_table_write.py
+++ b/paimon-python/pypaimon/table/format/format_table_write.py
@@ -31,6 +31,7 @@ from pypaimon.table.format.format_table import (
Format,
FormatTable,
)
+from pypaimon.write.writer.parquet_writer_options import
create_parquet_writer_options
def _partition_path(
@@ -100,9 +101,12 @@ class FormatTableWrite:
)
self._partition_only_value = opt.lower() == "true"
self._file_format = table.format()
- self._target_file_row_num = CoreOptions.from_dict(
- table.options()
- ).target_file_row_num()
+ options = CoreOptions.from_dict(table.options())
+ self._parquet_writer_options = (
+ create_parquet_writer_options(options)
+ if self._file_format == Format.PARQUET else {}
+ )
+ self._target_file_row_num = options.target_file_row_num()
max_target_file_row_num = (
CoreOptions.TARGET_FILE_ROW_NUM.default_value()
)
@@ -215,7 +219,7 @@ class FormatTableWrite:
if fmt == Format.PARQUET:
import pyarrow.parquet as pq
buf = io.BytesIO()
- pq.write_table(tbl, buf, compression="zstd")
+ pq.write_table(tbl, buf, compression="zstd",
**self._parquet_writer_options)
raw = buf.getvalue()
elif fmt == Format.CSV:
import pyarrow.csv as csv
diff --git a/paimon-python/pypaimon/tests/parquet_write_page_index_test.py
b/paimon-python/pypaimon/tests/parquet_write_page_index_test.py
new file mode 100644
index 0000000000..752637e9e2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/parquet_write_page_index_test.py
@@ -0,0 +1,250 @@
+# 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 unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.format.format_table import Format, FormatTable
+from pypaimon.write.table_update_by_row_id import _RowIdUpdateFileWriter
+
+
+OPTION = 'parquet.write-page-index.enabled'
+MODES = ['buffered', 'changelog', 'shared_shredding', 'row_id', 'format_table']
+HAS_PAGE_INDEX = int(pa.__version__.split('.')[0]) >= 13
+
+
+def _data():
+ return pa.Table.from_pydict({
+ 'id': [1, 2, 3],
+ 'items': [[1, None], [], None],
+ 'metrics': [[('a', 10), ('b', None)], [], None],
+ 'info': [{'score': 5}, None, {'score': None}],
+ }, schema=pa.schema([
+ pa.field('id', pa.int32(), nullable=False),
+ ('items', pa.list_(pa.int32())),
+ ('metrics', pa.map_(pa.string(), pa.int64())),
+ ('info', pa.struct([('score', pa.int32())])),
+ ]))
+
+
+def _table(tmp_path, mode, setting, file_format='parquet'):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ options = {'file.format': file_format}
+ if setting is not None:
+ options[OPTION] = setting
+ if mode == 'shared_shredding':
+ options['fields.metrics.map.storage-layout'] = 'shared-shredding'
+ options['fields.metrics.map.shared-shredding.max-columns'] = '2'
+ if mode == 'changelog':
+ options.update({'bucket': '1', 'changelog-producer': 'input'})
+ catalog.create_table('default.data', Schema.from_pyarrow_schema(
+ _data().schema, options=options,
+ primary_keys=['id'] if mode == 'changelog' else []), False)
+ table = catalog.get_table('default.data')
+ if mode == 'format_table':
+ table = FormatTable(table.file_io, table.identifier,
table.table_schema,
+ str(tmp_path / 'format'),
Format.parse(file_format))
+ return table
+
+
+def _write(table, mode, data):
+ if mode == 'row_id':
+ writer = _RowIdUpdateFileWriter(table, (), data.column_names)
+ writer.write_batches(iter(data.to_batches()))
+ return
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ commit = builder.new_commit()
+ try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('mode', MODES)
[email protected]('setting', [None, 'false', 'true'])
+def test_written_indexes_and_round_trip(tmp_path, mode, setting):
+ if setting == 'true' and not HAS_PAGE_INDEX:
+ pytest.skip('Writing page indexes requires PyArrow >= 13')
+ table = _table(tmp_path, mode, setting)
+ data = _data()
+ _write(table, mode, data)
+ paths = list(tmp_path.rglob('*.parquet'))
+ assert len(paths) == (2 if mode == 'changelog' else 1)
+ for path in paths:
+ metadata = pq.read_metadata(str(path))
+ if HAS_PAGE_INDEX:
+ for i in range(metadata.num_row_groups):
+ for j in range(metadata.num_columns):
+ column = metadata.row_group(i).column(j)
+ assert column.has_column_index == (setting != 'false')
+ assert column.has_offset_index == (setting != 'false')
+ if mode != 'shared_shredding':
+ actual = pq.read_table(str(path)).select(data.column_names)
+ assert actual.to_pydict() == data.to_pydict()
+ if mode != 'row_id':
+ reader = table.new_read_builder()
+ actual = reader.new_read().to_arrow(reader.new_scan().plan().splits())
+ assert actual.to_pydict() == data.to_pydict()
+ if mode not in ('row_id', 'format_table'):
+ reader = table.new_read_builder()
+
reader.with_filter(reader.new_predicate_builder().greater_or_equal('id', 2))
+ reader.with_projection(['id', 'items'])
+ actual = reader.new_read().to_arrow(reader.new_scan().plan().splits())
+ assert actual.to_pydict() == data.slice(1).select(['id',
'items']).to_pydict()
+
+
[email protected]('mode', MODES)
+def test_unsupported_arrow_rejected_before_output(tmp_path, mode):
+ table = _table(tmp_path, mode, 'true')
+ with patch.object(pa, '__version__', '12.0.1'), \
+ patch.object(table.file_io, 'new_output_stream',
+ wraps=table.file_io.new_output_stream) as
open_stream, \
+ patch.object(table.file_io, 'write_parquet',
+ wraps=table.file_io.write_parquet) as write_parquet:
+ with pytest.raises(ValueError,
match=r'parquet.write-page-index.enabled.*PyArrow >= 13'):
+ _write(table, mode, _data())
+ open_stream.assert_not_called()
+ write_parquet.assert_not_called()
+ assert not list(tmp_path.rglob('*.parquet'))
+
+
[email protected]('setting', [None, 'false'])
[email protected]('arrow_version', ['6.0.1', '12.0.1'])
+def test_legacy_arrow_omits_new_argument(tmp_path, setting, arrow_version):
+ table = _table(tmp_path, 'buffered', setting)
+ original = table.file_io.write_parquet
+ with patch.object(pa, '__version__', arrow_version), \
+ patch.object(table.file_io, 'write_parquet', wraps=original) as
write:
+ _write(table, 'buffered', _data())
+ assert write.called
+ assert 'write_page_index' not in write.call_args[1]
+
+
+def test_explicit_false_passed_to_supported_arrow(tmp_path):
+ if not HAS_PAGE_INDEX:
+ pytest.skip('Writing page indexes requires PyArrow >= 13')
+ table = _table(tmp_path, 'buffered', 'false')
+ with patch.object(table.file_io, 'write_parquet',
wraps=table.file_io.write_parquet) as write:
+ _write(table, 'buffered', _data())
+ assert write.call_args[1]['write_page_index'] is False
+
+
[email protected]('mode', ['buffered', 'format_table'])
+def test_other_formats_ignore_parquet_option(tmp_path, mode):
+ pytest.importorskip('pyarrow.orc')
+ table = _table(tmp_path, mode, 'true', file_format='orc')
+ with patch.object(pa, '__version__', '12.0.1'):
+ _write(table, mode, _data())
+ assert list(tmp_path.rglob('*.orc'))
+
+
+def test_parquet_changelog_with_orc_data(tmp_path):
+ pytest.importorskip('pyarrow.orc')
+ if not HAS_PAGE_INDEX:
+ pytest.skip('Writing page indexes requires PyArrow >= 13')
+ table = _table(tmp_path, 'changelog', 'true', file_format='orc')
+ table = table.copy({'changelog-file.format': 'parquet'})
+ _write(table, 'changelog', _data())
+ assert len(list(tmp_path.rglob('*.orc'))) == 1
+ paths = list(tmp_path.rglob('*.parquet'))
+ assert len(paths) == 1
+ column = pq.read_metadata(str(paths[0])).row_group(0).column(0)
+ assert column.has_column_index
+ assert column.has_offset_index
+
+
+def test_format_overwrite_checks_support_before_deleting(tmp_path):
+ table = _table(tmp_path, 'format_table', None)
+ _write(table, 'format_table', _data())
+ paths = {path: path.read_bytes() for path in tmp_path.rglob('*.parquet')}
+ table.options()[OPTION] = 'true'
+ from pypaimon.table.format.format_table_write import FormatTableWrite
+ with patch.object(pa, '__version__', '12.0.1'):
+ with pytest.raises(ValueError, match='PyArrow >= 13'):
+ FormatTableWrite(table, overwrite=True).write_arrow(_data())
+ assert {path: path.read_bytes() for path in tmp_path.rglob('*.parquet')}
== paths
+
+
+def _vector_table(tmp_path, file_format, setting):
+ data = pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'embedding': [[1.0, 2.0, 3.0], [4.0, 5.0, 6.0]],
+ }, schema=pa.schema([
+ ('id', pa.int64()),
+ ('embedding', pa.list_(pa.float32(), 3)),
+ ]))
+ options = {
+ 'file.format': file_format,
+ 'vector.file.format': 'parquet',
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ }
+ if setting is not None:
+ options[OPTION] = setting
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ catalog.create_table('default.vectors', Schema.from_pyarrow_schema(
+ data.schema, options=options), False)
+ return catalog.get_table('default.vectors'), data
+
+
[email protected]('file_format', ['parquet', 'orc'])
[email protected]('setting', [None, 'false', 'true'])
+def test_vector_parquet_indexes_and_round_trip(tmp_path, file_format, setting):
+ if setting == 'true' and not HAS_PAGE_INDEX:
+ pytest.skip('Writing page indexes requires PyArrow >= 13')
+ if file_format == 'orc':
+ pytest.importorskip('pyarrow.orc')
+ table, data = _vector_table(tmp_path, file_format, setting)
+ _write(table, 'buffered', data)
+ paths = list(tmp_path.rglob('*.parquet'))
+ assert len(list(tmp_path.rglob('*.vector.parquet'))) == 1
+ assert len(paths) == (2 if file_format == 'parquet' else 1)
+ for path in paths:
+ metadata = pq.read_metadata(str(path))
+ if HAS_PAGE_INDEX:
+ for i in range(metadata.num_row_groups):
+ for j in range(metadata.num_columns):
+ column = metadata.row_group(i).column(j)
+ assert column.has_column_index == (setting != 'false')
+ assert column.has_offset_index == (setting != 'false')
+ reader = table.new_read_builder()
+ actual = reader.new_read().to_arrow(reader.new_scan().plan().splits())
+ assert actual.to_pydict() == data.to_pydict()
+
+
+def test_vector_parquet_rejects_unsupported_arrow_before_output(tmp_path):
+ table, data = _vector_table(tmp_path, 'orc', 'true')
+ with patch.object(pa, '__version__', '12.0.1'), \
+ patch.object(table.file_io, 'write_parquet',
+ wraps=table.file_io.write_parquet) as write_parquet, \
+ patch.object(table.file_io, 'write_orc',
+ wraps=table.file_io.write_orc) as write_orc:
+ with pytest.raises(ValueError,
match=r'parquet.write-page-index.enabled.*PyArrow >= 13'):
+ _write(table, 'buffered', data)
+ write_parquet.assert_not_called()
+ write_orc.assert_not_called()
+ assert not list(tmp_path.rglob('*.parquet'))
+ assert not list(tmp_path.rglob('*.orc'))
diff --git a/paimon-python/pypaimon/write/map_shared_shredding_writer.py
b/paimon-python/pypaimon/write/map_shared_shredding_writer.py
index 07f21edf5c..539e9a0425 100644
--- a/paimon-python/pypaimon/write/map_shared_shredding_writer.py
+++ b/paimon-python/pypaimon/write/map_shared_shredding_writer.py
@@ -36,6 +36,7 @@ from pypaimon.schema.data_types import (
VectorType,
)
from pypaimon.table.bucket_mode import BucketMode
+from pypaimon.write.writer.parquet_writer_options import
create_parquet_writer_options
_FIELD_MAPPING = "__field_mapping"
@@ -88,6 +89,7 @@ class MapSharedShreddingWriter:
self._validate_format("file.format", file_format)
self._validate_format("changelog.file.format", changelog_format)
+ self._parquet_writer_options = create_parquet_writer_options(options)
self._validate_compression("file.compression",
options.file_compression())
if options.bucket() == BucketMode.POSTPONE_BUCKET.value:
raise ValueError(
@@ -158,7 +160,7 @@ class MapSharedShreddingWriter:
# Bound scratch space for slot mappings/indices as well as logical
values.
slots = sum(c.num_columns for c in converters.values())
batch_rows = max(1, min(1024, _CONVERSION_BYTES // max(1, slots * 16)))
- kwargs = {'compression': compression}
+ kwargs = dict(self._parquet_writer_options, compression=compression)
if compression.lower() == 'zstd':
kwargs['compression_level'] = zstd_level
try:
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index 9226f8d823..9e6275fd4a 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -172,7 +172,8 @@ class _RowIdUpdateFileWriter:
file_writer = SingleFileWriter(
writer.file_io, file_path, batch.schema,
writer.file_format,
writer.compression, writer.zstd_level,
- fields, writer._get_column_stats)
+ fields, writer._get_column_stats,
+ parquet_options=writer.parquet_writer_options)
file_writer.write(batch, row_group_size=batch.num_rows)
del batch
if file_writer is None:
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 79de18f606..ad336b0b39 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -31,6 +31,7 @@ from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.map_shared_shredding_writer import MapSharedShreddingWriter
from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
+from pypaimon.write.writer.parquet_writer_options import
create_parquet_writer_options
from pypaimon.write.writer.write_buffer import WriteBuffer
@@ -82,6 +83,13 @@ class DataWriter(ABC):
self.options.changelog_file_format()
or self.file_format
)
+ self.parquet_writer_options = (
+ create_parquet_writer_options(self.options)
+ if self.file_format == CoreOptions.FILE_FORMAT_PARQUET
+ or (changelog_producer == ChangelogProducer.INPUT
+ and self.changelog_file_format ==
CoreOptions.FILE_FORMAT_PARQUET)
+ else {}
+ )
self.write_cols = write_cols
self.blob_as_descriptor = self.options.blob_as_descriptor()
@@ -381,7 +389,9 @@ class DataWriter(ABC):
if self._map_shared_shredding.is_active():
return self._map_shared_shredding.write_parquet(
self.file_io, path, data, self.compression, self.zstd_level)
- self.file_io.write_parquet(path, data, compression=self.compression,
zstd_level=self.zstd_level)
+ self.file_io.write_parquet(
+ path, data, compression=self.compression,
zstd_level=self.zstd_level,
+ **self.parquet_writer_options)
return {}
def _create_data_file_meta(self, file_name, file_path, row_count,
diff --git a/paimon-python/pypaimon/write/writer/parquet_writer_options.py
b/paimon-python/pypaimon/write/writer/parquet_writer_options.py
new file mode 100644
index 0000000000..7665f580d4
--- /dev/null
+++ b/paimon-python/pypaimon/write/writer/parquet_writer_options.py
@@ -0,0 +1,31 @@
+# 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 pyarrow as pa
+
+from pypaimon.common.options.core_options import CoreOptions
+
+
+def create_parquet_writer_options(options: CoreOptions):
+ enabled = options.parquet_write_page_index_enabled()
+ if int(pa.__version__.split('.')[0]) >= 13:
+ return {'write_page_index': True if enabled is None else enabled}
+ if enabled:
+ raise ValueError(
+ "parquet.write-page-index.enabled requires PyArrow >= 13, got
{}".format(pa.__version__))
+ # Older PyArrow versions do not accept even write_page_index=False.
+ return {}
diff --git a/paimon-python/pypaimon/write/writer/single_file_writer.py
b/paimon-python/pypaimon/write/writer/single_file_writer.py
index 184d013822..5abe7e90b5 100644
--- a/paimon-python/pypaimon/write/writer/single_file_writer.py
+++ b/paimon-python/pypaimon/write/writer/single_file_writer.py
@@ -26,7 +26,7 @@ class SingleFileWriter:
"""Write one file incrementally; currently only Parquet is supported."""
def __init__(self, file_io, path, schema, file_format, compression,
zstd_level,
- stats_fields, stats_collector):
+ stats_fields, stats_collector, parquet_options=None):
if file_format != CoreOptions.FILE_FORMAT_PARQUET:
raise NotImplementedError(
'SingleFileWriter only supports Parquet, got
{}'.format(file_format))
@@ -41,7 +41,7 @@ class SingleFileWriter:
self.row_count = 0
self.column_stats = {}
- kwargs = {'compression': compression}
+ kwargs = dict(parquet_options or {}, compression=compression)
if compression.lower() == 'zstd':
kwargs['compression_level'] = zstd_level
try:
diff --git a/paimon-python/pypaimon/write/writer/vector_writer.py
b/paimon-python/pypaimon/write/writer/vector_writer.py
index a7cd6bfc58..50663af7b0 100644
--- a/paimon-python/pypaimon/write/writer/vector_writer.py
+++ b/paimon-python/pypaimon/write/writer/vector_writer.py
@@ -26,6 +26,7 @@ from pypaimon.manifest.schema.data_file_meta import
DataFileMeta
from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
+from pypaimon.write.writer.parquet_writer_options import
create_parquet_writer_options
class VectorWriter(AppendOnlyDataWriter):
@@ -44,6 +45,10 @@ class VectorWriter(AppendOnlyDataWriter):
self.vector_columns = vector_columns
self.vector_file_format = vector_file_format
self.file_format = vector_file_format
+ self.parquet_writer_options = (
+ create_parquet_writer_options(self.options)
+ if vector_file_format == CoreOptions.FILE_FORMAT_PARQUET else {}
+ )
self.target_file_size = options.vector_target_file_size()
# The parent coordinates rolling at logical group boundaries.
self.rolling_managed_by_parent = rolling_managed_by_parent
@@ -88,7 +93,8 @@ class VectorWriter(AppendOnlyDataWriter):
elif self.vector_file_format == CoreOptions.FILE_FORMAT_PARQUET:
self.file_io.write_parquet(file_path, data,
compression=self.compression,
- zstd_level=self.zstd_level)
+ zstd_level=self.zstd_level,
+ **self.parquet_writer_options)
elif self.vector_file_format == CoreOptions.FILE_FORMAT_VORTEX:
self.file_io.write_vortex(file_path, data)
else:
