Add optional Parquet page index writes

mkdocs/docs/configuration.md

@@ -84,6 +84,7 @@ Iceberg tables support table properties to configure table behavior.
 | `write.parquet.row-group-limit`          | Number of rows                     | 1048576                    | The upper bound of the number of entries within a single row group                                                                                   |
 | `write.parquet.page-size-bytes`          | Size in bytes                      | 1MB                        | Set a target threshold for the approximate encoded size of data pages within a column chunk                                                          |
 | `write.parquet.page-row-limit`           | Number of rows                     | 20000                      | Set a target threshold for the maximum number of rows within a column chunk                                                                          |
+| `write.parquet.page-index-enabled`        | Boolean                            | false                      | Write Parquet column and offset indexes for page-level filtering                                                                                      |
 | `write.parquet.dict-size-bytes`          | Size in bytes                      | 2MB                        | Set the dictionary page size limit per row group                                                                                                     |
 | `write.metadata.previous-versions-max`   | Integer                            | 100                        | The max number of previous version metadata files to keep before deleting after commit.                                                              |
 | `write.metadata.delete-after-commit.enabled` | Boolean                        | False                      | Whether to automatically delete old *tracked* metadata files after each table commit. It will retain a number of the most recent metadata files, which can be set using property `write.metadata.previous-versions-max`. |

pyiceberg/io/pyarrow.py

@@ -2957,6 +2957,11 @@ def _get_parquet_writer_kwargs(table_properties: Properties) -> dict[str, Any]:
             property_name=TableProperties.PARQUET_PAGE_ROW_LIMIT,
             default=TableProperties.PARQUET_PAGE_ROW_LIMIT_DEFAULT,
         ),
+        "write_page_index": property_as_bool(
+            properties=table_properties,
+            property_name=TableProperties.PARQUET_PAGE_INDEX_ENABLED,
+            default=TableProperties.PARQUET_PAGE_INDEX_ENABLED_DEFAULT,
+        ),
     }
 
 

pyiceberg/table/__init__.py

@@ -147,6 +147,9 @@ class TableProperties:
     PARQUET_PAGE_ROW_LIMIT = "write.parquet.page-row-limit"
     PARQUET_PAGE_ROW_LIMIT_DEFAULT = 20000
 
+    PARQUET_PAGE_INDEX_ENABLED = "write.parquet.page-index-enabled"
+    PARQUET_PAGE_INDEX_ENABLED_DEFAULT = False
+
     PARQUET_DICT_SIZE_BYTES = "write.parquet.dict-size-bytes"
     PARQUET_DICT_SIZE_BYTES_DEFAULT = 2 * 1024 * 1024  # 2 MB
 

tests/integration/test_writes/test_writes.py

@@ -669,6 +669,7 @@ def test_write_parquet_compression_properties(
     [
         ({"write.parquet.page-size-bytes": "42"}, {"data_page_size": 42}),
         ({"write.parquet.dict-size-bytes": "42"}, {"dictionary_pagesize_limit": 42}),
+        ({"write.parquet.page-index-enabled": "true"}, {"write_page_index": True}),
     ],
 )
 def test_write_parquet_other_properties(

tests/io/test_pyarrow.py

@@ -74,6 +74,7 @@
     _check_pyarrow_schema_compatible,
     _ConvertToArrowSchema,
     _determine_partitions,
+    _get_parquet_writer_kwargs,
     _primitive_to_physical,
     _read_deletes,
     _task_to_record_batches,
@@ -127,6 +128,11 @@
 )
 
 
+def test_parquet_page_index_writer_property() -> None:
+    assert _get_parquet_writer_kwargs({})["write_page_index"] is False
+    assert _get_parquet_writer_kwargs({"write.parquet.page-index-enabled": "true"})["write_page_index"] is True
+
+
 def test_pyarrow_infer_local_fs_from_path() -> None:
     """Test path with `file` scheme and no scheme both use LocalFileSystem"""
     assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)