apacheGH-48096 [Python][Parquet] Expose new WriterProperties::max_rows_per_page to Python bindings (apache#48101)

### Rationale for this change
See apache#48096, exposes `parquet.WriterProperties max_rows_per_page` argument to Python's API.

### What changes are included in this PR?
Added the argument

### Are these changes tested?
Yes, since the metadata doesn't have any info about the number of pages, a naive end-to-end test was used to ensure the implementation correctness.

### Are there any user-facing changes?
The ability to set the `max_rows_per_page` directly from PyArrow.

* GitHub Issue: apache#48096

Authored-by: Bogdan Romenskii <rmnsk@seznam.cz>
Signed-off-by: Raúl Cumplido <raulcumplido@gmail.com>

Author: rmnskb

python/pyarrow/_dataset_parquet.pyx

@@ -646,6 +646,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
             version=self._properties["version"],
             write_statistics=self._properties["write_statistics"],
             data_page_size=self._properties["data_page_size"],
+            max_rows_per_page=self._properties["max_rows_per_page"],
             compression_level=self._properties["compression_level"],
             use_byte_stream_split=(
                 self._properties["use_byte_stream_split"]
@@ -695,6 +696,7 @@ cdef class ParquetFileWriteOptions(FileWriteOptions):
             version="2.6",
             write_statistics=None,
             data_page_size=None,
+            max_rows_per_page=None,
             compression_level=None,
             use_byte_stream_split=False,
             column_encoding=None,

python/pyarrow/_parquet.pxd

@@ -44,6 +44,7 @@ cdef shared_ptr[WriterProperties] _create_writer_properties(
     version=*,
     write_statistics=*,
     data_page_size=*,
+    max_rows_per_page=*,
     compression_level=*,
     use_byte_stream_split=*,
     column_encoding=*,

python/pyarrow/_parquet.pyx

@@ -1984,6 +1984,7 @@ cdef shared_ptr[WriterProperties] _create_writer_properties(
         version=None,
         write_statistics=None,
         data_page_size=None,
+        max_rows_per_page=None,
         compression_level=None,
         use_byte_stream_split=False,
         column_encoding=None,
@@ -2129,6 +2130,9 @@ cdef shared_ptr[WriterProperties] _create_writer_properties(
     if data_page_size is not None:
         props.data_pagesize(data_page_size)
 
+    if max_rows_per_page is not None:
+        props.max_rows_per_page(max_rows_per_page)
+
     if write_batch_size is not None:
         props.write_batch_size(write_batch_size)
 
@@ -2300,6 +2304,7 @@ cdef class ParquetWriter(_Weakrefable):
                   use_deprecated_int96_timestamps=False,
                   coerce_timestamps=None,
                   data_page_size=None,
+                  max_rows_per_page=None,
                   allow_truncated_timestamps=False,
                   compression_level=None,
                   use_byte_stream_split=False,
@@ -2340,6 +2345,7 @@ cdef class ParquetWriter(_Weakrefable):
             version=version,
             write_statistics=write_statistics,
             data_page_size=data_page_size,
+            max_rows_per_page=max_rows_per_page,
             compression_level=compression_level,
             use_byte_stream_split=use_byte_stream_split,
             column_encoding=column_encoding,

python/pyarrow/includes/libparquet.pxd

@@ -492,6 +492,7 @@ cdef extern from "parquet/api/writer.h" namespace "parquet" nogil:
             Builder* enable_store_decimal_as_integer()
             Builder* disable_store_decimal_as_integer()
             Builder* data_pagesize(int64_t size)
+            Builder* max_rows_per_page(int64_t max_rows)
             Builder* encoding(ParquetEncoding encoding)
             Builder* encoding(const c_string& path,
                               ParquetEncoding encoding)

python/pyarrow/parquet/core.py

@@ -795,6 +795,10 @@ def _sanitize_table(table, new_schema, flavor):
     Set a target threshold for the approximate encoded size of data
     pages within a column chunk (in bytes). If None, use the default data page
     size of 1MByte.
+max_rows_per_page : int, default None
+    Maximum number of rows per page within a column chunk.
+    If None, use the default of 20000.
+    Smaller values reduce memory usage during reads but increase metadata overhead.
 flavor : {'spark'}, default None
     Sanitize schema or set other compatibility options to work with
     various target systems.
@@ -1042,6 +1046,7 @@ def __init__(self, where, schema, filesystem=None,
                  sorting_columns=None,
                  store_decimal_as_integer=False,
                  write_time_adjusted_to_utc=False,
+                 max_rows_per_page=None,
                  **options):
         if use_deprecated_int96_timestamps is None:
             # Use int96 timestamps for Spark
@@ -1096,6 +1101,7 @@ def __init__(self, where, schema, filesystem=None,
             sorting_columns=sorting_columns,
             store_decimal_as_integer=store_decimal_as_integer,
             write_time_adjusted_to_utc=write_time_adjusted_to_utc,
+            max_rows_per_page=max_rows_per_page,
             **options)
         self.is_open = True
 
@@ -1971,6 +1977,7 @@ def write_table(table, where, row_group_size=None, version='2.6',
                 sorting_columns=None,
                 store_decimal_as_integer=False,
                 write_time_adjusted_to_utc=False,
+                max_rows_per_page=None,
                 **kwargs):
     # Implementor's note: when adding keywords here / updating defaults, also
     # update it in write_to_dataset and _dataset_parquet.pyx ParquetFileWriteOptions
@@ -2003,6 +2010,7 @@ def write_table(table, where, row_group_size=None, version='2.6',
                 sorting_columns=sorting_columns,
                 store_decimal_as_integer=store_decimal_as_integer,
                 write_time_adjusted_to_utc=write_time_adjusted_to_utc,
+                max_rows_per_page=max_rows_per_page,
                 **kwargs) as writer:
             writer.write_table(table, row_group_size=row_group_size)
     except Exception:

python/pyarrow/tests/parquet/test_parquet_writer.py

@@ -487,3 +487,60 @@ def test_arrow_writer_props_time_adjusted_to_utc(
     result.validate(full=True)
 
     assert result.equals(table)
+
+
+@pytest.mark.parametrize(
+    "max_rows_per_page",
+    [1, 10, 100, 1_000, None],
+)
+def test_writer_props_max_rows_per_page(tempdir, max_rows_per_page):
+    # GH-48096
+    filename = tempdir / "max_rows_per_page.parquet"
+
+    table = pa.table({
+        "x": pa.array([1, 2, 3, 4, 5, 6, 7], type=pa.int8()),
+        "y": pa.array([11.0, 12.0, 13.0, 14.0, 15.0, 16.0, 17.0], type=pa.float16()),
+    })
+
+    schema = pa.schema([
+        ("x", pa.int8()),
+        ("y", pa.float16()),
+    ])
+
+    with pq.ParquetWriter(
+        where=filename,
+        schema=schema,
+        max_rows_per_page=max_rows_per_page,
+    ) as writer:
+        writer.write_table(table)
+
+    result = pq.read_table(filename, schema=schema)
+
+    result.validate(full=True)
+
+    assert result.equals(table)
+
+
+def test_writer_props_max_rows_per_page_file_size(tempdir):
+    # GH-48096
+    table = pa.table({
+        "x": pa.array(range(1_000_000))
+    })
+
+    local = fs.LocalFileSystem()
+    file_infos = []
+
+    for max_rows in (1_000, 10_000):
+        path = f"{tempdir}/max_rows_per_page_{max_rows}.parquet"
+
+        with pq.ParquetWriter(
+            where=path,
+            schema=table.schema,
+            max_rows_per_page=max_rows,
+        ) as writer:
+            writer.write_table(table)
+
+        file_infos.append(local.get_file_info(path))
+
+    # A smaller maximum rows parameter should produce a larger file
+    assert file_infos[0].size > file_infos[1].size