This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 107b3a60 [core] Align Parquet writer row group and page sizing options
with Java (#1072)
107b3a60 is described below
commit 107b3a60bfeb315119f542d141e3bc7fb85fe444
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Oct 8 20:58:32 2026 +0800
[core] Align Parquet writer row group and page sizing options with Java
(#1072)
---
crates/paimon/src/arrow/format/parquet.rs | 319 +++++++++++++++++++++++++++++-
crates/paimon/src/table/table_write.rs | 207 +++++++++++++++++++
docs/src/getting-started.md | 16 ++
3 files changed, 534 insertions(+), 8 deletions(-)
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 48132c27..cbc434a4 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -288,6 +288,7 @@ pub(crate) struct ParquetFormatWriter {
write_fields: Option<Vec<DataField>>,
stats_modes: Option<Vec<MetadataStatsMode>>,
stats_dense_store: bool,
+ max_row_group_bytes: Option<usize>,
}
pub(crate) struct ParquetWriterFactory {
@@ -370,6 +371,7 @@ impl ParquetFormatWriter {
let core_options = CoreOptions::new(format_options);
let page_index_enabled =
core_options.parquet_write_page_index_enabled()?;
let props = parquet_writer_properties(codec, page_index_enabled,
format_options)?;
+ let max_row_group_bytes = props.max_row_group_bytes();
let async_write = output.async_writer().await?;
let input_schema = schema;
let schema = timestamp_millis_schema(&input_schema);
@@ -384,8 +386,19 @@ impl ParquetFormatWriter {
write_fields: write_fields.map(|fields| fields.to_vec()),
stats_modes,
stats_dense_store: core_options.metadata_stats_dense_store(),
+ max_row_group_bytes,
})
}
+
+ async fn write_physical(&mut self, batch: &RecordBatch) ->
crate::Result<()> {
+ self.inner
+ .write(batch)
+ .await
+ .map_err(|e| crate::Error::DataInvalid {
+ message: format!("Failed to write parquet batch: {e}"),
+ source: None,
+ })
+ }
}
fn create_parquet_arrow_writer(
@@ -445,7 +458,134 @@ fn parquet_writer_properties(
);
}
}
- Ok(props.build())
+ Ok(parquet_sizing_properties(props, options)?.build())
+}
+
+/// Java's parquet writer sizing keys; `file.block-size` overrides
`parquet.block.size` as in
+/// Java's `ParquetFileFormat`.
+fn parquet_sizing_properties(
+ mut props: parquet::file::properties::WriterPropertiesBuilder,
+ options: &HashMap<String, String>,
+) -> crate::Result<parquet::file::properties::WriterPropertiesBuilder> {
+ const BLOCK_SIZE: &str = "parquet.block.size";
+ const PAGE_SIZE: &str = "parquet.page.size";
+ const DICTIONARY_PAGE_SIZE: &str = "parquet.dictionary.page.size";
+ const PAGE_ROW_COUNT_LIMIT: &str = "parquet.page.row.count.limit";
+ const PAGE_SIZE_ROW_CHECK_MIN: &str = "parquet.page.size.row.check.min";
+ let positive = |key: &str| -> crate::Result<Option<usize>> {
+ options
+ .get(key)
+ .map(|value| {
+ value
+ .trim()
+ .parse::<usize>()
+ .ok()
+ .filter(|parsed| *parsed > 0)
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!("Option '{key}' must be a positive
integer, got: {value}"),
+ })
+ })
+ .transpose()
+ };
+ let file_block_size = CoreOptions::new(options)
+ .file_block_size()?
+ .map(|bytes| {
+ usize::try_from(bytes)
+ .ok()
+ .filter(|bytes| *bytes > 0)
+ .ok_or_else(|| crate::Error::ConfigInvalid {
+ message: format!("Option 'file.block-size' must be
positive, got: {bytes}"),
+ })
+ })
+ .transpose()?;
+ let block_size = match file_block_size {
+ Some(bytes) => Some(bytes),
+ None => positive(BLOCK_SIZE)?,
+ };
+ if let Some(bytes) = block_size {
+ props = props.set_max_row_group_bytes(Some(bytes));
+ }
+ if let Some(bytes) = positive(PAGE_SIZE)? {
+ props = props.set_data_page_size_limit(bytes);
+ }
+ if let Some(bytes) = positive(DICTIONARY_PAGE_SIZE)? {
+ props = props.set_dictionary_page_size_limit(bytes);
+ }
+ // parquet-rs checks page limits between write batches, so the batch
bounds the check interval.
+ let row_check = positive(PAGE_SIZE_ROW_CHECK_MIN)?;
+ let page_rows = positive(PAGE_ROW_COUNT_LIMIT)?;
+ if let Some(rows) = page_rows {
+ props = props.set_data_page_row_count_limit(rows);
+ }
+ if row_check.is_some() || page_rows.is_some() {
+ let limit =
+
page_rows.unwrap_or(parquet::file::properties::DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT);
+ let batch =
row_check.unwrap_or(parquet::file::properties::DEFAULT_WRITE_BATCH_SIZE);
+ props = props.set_write_batch_size(batch.min(limit));
+ }
+ Ok(props)
+}
+
+/// Bytes the rows of a (possibly sliced) batch occupy; unlike
`get_array_memory_size`, list
+/// values and dictionaries outside the slice are not counted.
+fn batch_slice_bytes(batch: &RecordBatch) -> usize {
+ batch
+ .columns()
+ .iter()
+ .map(|column| array_slice_bytes(column.as_ref()))
+ .sum()
+}
+
+fn array_slice_bytes(array: &dyn arrow_array::Array) -> usize {
+ use arrow_array::cast::AsArray;
+ use arrow_array::{Array, ArrayRef};
+ use arrow_schema::DataType as ArrowDataType;
+ let nulls = array.nulls().map_or(0, |nulls| nulls.len().div_ceil(8));
+ let list_bytes = |values: &ArrayRef, start: usize, end: usize,
offset_width: usize| {
+ offset_width * (array.len() + 1)
+ + nulls
+ + array_slice_bytes(values.slice(start, end - start).as_ref())
+ };
+ match array.data_type() {
+ ArrowDataType::List(_) => {
+ let list = array.as_list::<i32>();
+ let offsets = list.value_offsets();
+ let (start, end) = (offsets[0] as usize, offsets[list.len()] as
usize);
+ list_bytes(list.values(), start, end, 4)
+ }
+ ArrowDataType::LargeList(_) => {
+ let list = array.as_list::<i64>();
+ let offsets = list.value_offsets();
+ let (start, end) = (offsets[0] as usize, offsets[list.len()] as
usize);
+ list_bytes(list.values(), start, end, 8)
+ }
+ ArrowDataType::Map(_, _) => {
+ let map = array.as_map();
+ let offsets = map.value_offsets();
+ let (start, end) = (offsets[0] as usize, offsets[map.len()] as
usize);
+ let entries: ArrayRef = Arc::new(map.entries().clone());
+ list_bytes(&entries, start, end, 4)
+ }
+ ArrowDataType::Struct(_) => {
+ nulls
+ + array
+ .as_struct()
+ .columns()
+ .iter()
+ .map(|column| array_slice_bytes(column.as_ref()))
+ .sum::<usize>()
+ }
+ ArrowDataType::Dictionary(_, _) => {
+ let dictionary = array.as_any_dictionary();
+ let values = dictionary.values();
+ let value_bytes = array_slice_bytes(values.as_ref());
+ array_slice_bytes(dictionary.keys()) + value_bytes * array.len() /
values.len().max(1)
+ }
+ _ => array
+ .to_data()
+ .get_slice_memory_size()
+ .unwrap_or_else(|_| array.get_array_memory_size()),
+ }
}
/// Every `file.compression` value a parquet write accepts.
@@ -524,13 +664,29 @@ impl FormatFileWriter for ParquetFormatWriter {
physical_batch = cast_record_batch_to_schema(batch, &self.schema)?;
&physical_batch
};
- self.inner
- .write(batch)
- .await
- .map_err(|e| crate::Error::DataInvalid {
- message: format!("Failed to write parquet batch: {e}"),
- source: None,
- })
+ let Some(limit) = self.max_row_group_bytes else {
+ return self.write_physical(batch).await;
+ };
+ // parquet-rs splits by bytes only once a row group holds rows, so
write slices of at
+ // most a quarter of the limit, measured on their own rows, and close
the group here.
+ let target = (limit / 4).max(1);
+ let rows = batch.num_rows();
+ let guess = (target /
batch_slice_bytes(batch).div_ceil(rows.max(1)).max(1)).max(1);
+ let mut offset = 0;
+ while offset < rows {
+ let mut len = guess.min(rows - offset);
+ let mut slice = batch.slice(offset, len);
+ while len > 1 && batch_slice_bytes(&slice) > target {
+ len /= 2;
+ slice = batch.slice(offset, len);
+ }
+ self.write_physical(&slice).await?;
+ offset += len;
+ if self.inner.in_progress_size() >= limit {
+ self.flush().await?;
+ }
+ }
+ Ok(())
}
fn num_bytes(&self) -> usize {
@@ -4176,6 +4332,7 @@ mod tests {
write_fields: None,
stats_modes: None,
stats_dense_store: false,
+ max_row_group_bytes: None,
}),
Some(&resources),
);
@@ -4339,6 +4496,152 @@ mod tests {
}
}
+ #[test]
+ fn test_parquet_writer_sizing_options() {
+ let defaults =
+ parquet_writer_properties(Compression::UNCOMPRESSED, true,
&HashMap::new()).unwrap();
+ assert_eq!(defaults.max_row_group_bytes(), None);
+ assert_eq!(
+ defaults.write_batch_size(),
+ parquet::file::properties::DEFAULT_WRITE_BATCH_SIZE
+ );
+ let mut options = HashMap::from([
+ ("parquet.block.size".to_owned(), "1048576".to_owned()),
+ ("parquet.page.size".to_owned(), "65536".to_owned()),
+ (
+ "parquet.dictionary.page.size".to_owned(),
+ "32768".to_owned(),
+ ),
+ ("parquet.page.row.count.limit".to_owned(), "500".to_owned()),
+ ]);
+ let props = parquet_writer_properties(Compression::UNCOMPRESSED, true,
&options).unwrap();
+ assert_eq!(props.max_row_group_bytes(), Some(1_048_576));
+ assert_eq!(props.data_page_size_limit(), 65_536);
+ assert_eq!(props.dictionary_page_size_limit(), 32_768);
+ assert_eq!(props.data_page_row_count_limit(), 500);
+ assert_eq!(props.write_batch_size(), 500);
+
+ options.insert("file.block-size".to_owned(), "2 mb".to_owned());
+ options.insert("parquet.block.size".to_owned(), "128mb".to_owned());
+ options.insert(
+ "parquet.page.size.row.check.min".to_owned(),
+ "100".to_owned(),
+ );
+ let props = parquet_writer_properties(Compression::UNCOMPRESSED, true,
&options).unwrap();
+ assert_eq!(props.max_row_group_bytes(), Some(2 * 1024 * 1024));
+ assert_eq!(props.write_batch_size(), 100);
+
+ // Without an explicit page row limit, the default one still bounds
the check interval.
+ let options = HashMap::from([(
+ "parquet.page.size.row.check.min".to_owned(),
+ "65536".to_owned(),
+ )]);
+ let props = parquet_writer_properties(Compression::UNCOMPRESSED, true,
&options).unwrap();
+ assert_eq!(
+ props.data_page_row_count_limit(),
+ parquet::file::properties::DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT
+ );
+ assert_eq!(
+ props.write_batch_size(),
+ parquet::file::properties::DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT
+ );
+ }
+
+ #[tokio::test]
+ async fn test_parquet_writer_invalid_sizing_options_create_no_file() {
+ for (key, value) in [
+ ("parquet.block.size", "0"),
+ ("parquet.block.size", "128mb"),
+ ("parquet.page.size", "-1"),
+ ("parquet.dictionary.page.size", "x"),
+ ("parquet.page.row.count.limit", "0"),
+ ("parquet.page.size.row.check.min", ""),
+ ("file.block-size", "0 b"),
+ ("file.block-size", "invalid"),
+ ] {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/invalid_sizing.parquet";
+ let options = HashMap::from([(key.to_owned(), value.to_owned())]);
+ let result = ParquetFormatWriter::new(
+ &file_io.new_output(path).unwrap(),
+ writer_arrow_schema(),
+ "zstd",
+ 1,
+ None,
+ &options,
+ )
+ .await;
+ assert!(
+ matches!(result, Err(crate::Error::ConfigInvalid { .. })),
+ "{key}={value}"
+ );
+ assert!(!file_io.exists(path).await.unwrap());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_parquet_writer_sizing_options_shape_the_file() {
+ use parquet::arrow::arrow_reader::{ArrowReaderOptions,
ParquetRecordBatchReaderBuilder};
+
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/sizing_options.parquet";
+ let options = HashMap::from([
+ ("parquet.enable.dictionary".to_owned(), "false".to_owned()),
+ ("parquet.block.size".to_owned(), "16384".to_owned()),
+ ("parquet.page.row.count.limit".to_owned(), "100".to_owned()),
+ ]);
+ let schema = writer_arrow_schema();
+ let mut writer = ParquetFormatWriter::new(
+ &file_io.new_output(path).unwrap(),
+ schema.clone(),
+ "none",
+ 1,
+ None,
+ &options,
+ )
+ .await
+ .unwrap();
+ for chunk in 0..40 {
+ let ids = (chunk * 100..(chunk + 1) * 100).collect::<Vec<i32>>();
+ let values = ids.iter().map(|id| id.wrapping_mul(7919)).collect();
+ writer
+ .write(&writer_test_batch(&schema, ids, values))
+ .await
+ .unwrap();
+ }
+ Box::new(writer).close().await.unwrap();
+
+ let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
+ let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(
+ bytes,
+
ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required),
+ )
+ .unwrap();
+ let metadata = reader.metadata().clone();
+ assert!(metadata.num_row_groups() > 1);
+ let offsets = metadata.offset_index().expect("page index");
+ for (row_group, columns) in metadata.row_groups().iter().zip(offsets) {
+ assert!(row_group.compressed_size() <= 2 * 16384);
+ for column in columns {
+ let firsts = column
+ .page_locations()
+ .iter()
+ .map(|page| page.first_row_index)
+ .chain(std::iter::once(row_group.num_rows()))
+ .collect::<Vec<_>>();
+ // parquet-rs checks the limit between write batches, so a
split batch can add one batch.
+ assert!(firsts.windows(2).all(|pair| pair[1] - pair[0] < 200));
+ assert!(firsts.len() > (row_group.num_rows() / 100) as usize);
+ }
+ }
+ let rows = reader
+ .build()
+ .unwrap()
+ .map(|batch| batch.unwrap().num_rows())
+ .sum::<usize>();
+ assert_eq!(rows, 4000);
+ }
+
#[tokio::test]
async fn test_parquet_writer_write_and_close() {
let file_io = FileIOBuilder::new("memory").build().unwrap();
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 4e8032be..59ed4a4c 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -2301,6 +2301,213 @@ pub(in crate::table) mod tests {
assert_eq!(snapshot.total_record_count(), Some(3));
}
+ fn sizing_table(
+ file_io: &FileIO,
+ table_path: &str,
+ primary_key: bool,
+ options: &[(&str, &str)],
+ ) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .option("file.format", "parquet")
+ .option("file.compression", "none")
+ .option("parquet.enable.dictionary", "false");
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ if primary_key {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ }
+ Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test_table"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema.build().unwrap()),
+ None,
+ )
+ }
+
+ fn sizing_batch(names: Vec<String>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, false),
+ ]));
+ let ids = (0..names.len() as i32).collect::<Vec<_>>();
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ ],
+ )
+ .unwrap()
+ }
+
+ /// Writes one batch, returns the footers (with page indexes) of the new
files, then
+ /// commits and checks that every row reads back.
+ async fn write_sizing_batch(
+ file_io: &FileIO,
+ table: &Table,
+ batch: &RecordBatch,
+ ) -> Vec<(String, parquet::file::metadata::ParquetMetaData)> {
+ use parquet::arrow::arrow_reader::{ArrowReaderOptions,
ParquetRecordBatchReaderBuilder};
+ use parquet::file::metadata::PageIndexPolicy;
+
+ let mut table_write = TableWrite::new(table,
"test-user".to_string()).unwrap();
+ table_write.write_arrow_batch(batch).await.unwrap();
+ let messages = table_write.prepare_commit().await.unwrap();
+ let mut footers = Vec::new();
+ for message in &messages {
+ for file in &message.new_files {
+ let path = format!(
+ "{}/{}/{}",
+ table.location(),
+ bucket_dir_name(message.bucket),
+ file.file_name
+ );
+ let bytes =
file_io.new_input(&path).unwrap().read().await.unwrap();
+ let metadata =
ParquetRecordBatchReaderBuilder::try_new_with_options(
+ bytes,
+
ArrowReaderOptions::new().with_page_index_policy(PageIndexPolicy::Required),
+ )
+ .unwrap()
+ .metadata()
+ .as_ref()
+ .clone();
+ footers.push((path, metadata));
+ }
+ }
+ TableCommit::new(table.clone(), "test-user".to_string())
+ .commit(messages)
+ .await
+ .unwrap();
+ let read_builder = table.new_read_builder();
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = futures::TryStreamExt::try_collect(
+ read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap(),
+ )
+ .await
+ .unwrap();
+ let mut read_ids = batches
+ .iter()
+ .flat_map(|batch| {
+ let ids = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ ids.values().to_vec()
+ })
+ .collect::<Vec<_>>();
+ read_ids.sort_unstable();
+ assert_eq!(read_ids, (0..batch.num_rows() as i32).collect::<Vec<_>>());
+ footers
+ }
+
+ #[tokio::test]
+ async fn test_parquet_block_size_bounds_row_groups_of_one_large_batch() {
+ for primary_key in [false, true] {
+ let file_io = test_file_io();
+ let table_path =
format!("memory:/test_parquet_block_size_{primary_key}");
+ setup_dirs(&file_io, &table_path).await;
+ // parquet.block.size is shadowed by file.block-size as in Java,
so it is never parsed.
+ let table = sizing_table(
+ &file_io,
+ &table_path,
+ primary_key,
+ &[
+ ("file.block-size", "16 kb"),
+ ("parquet.block.size", "128mb"),
+ ],
+ );
+ let names = (0..65_536).map(|id| format!("{:08}", id *
7919)).collect();
+ for (path, metadata) in write_sizing_batch(&file_io, &table,
&sizing_batch(names)).await
+ {
+ assert!(metadata.num_row_groups() > 1, "{path}");
+ for row_group in metadata.row_groups() {
+ assert!(row_group.compressed_size() <= 2 * 16 * 1024,
"{path}");
+ }
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_parquet_block_size_bounds_row_groups_of_skewed_rows() {
+ const LIMIT: i64 = 1024 * 1024;
+ for primary_key in [false, true] {
+ let file_io = test_file_io();
+ let table_path =
format!("memory:/test_parquet_block_size_skewed_{primary_key}");
+ setup_dirs(&file_io, &table_path).await;
+ let table = sizing_table(
+ &file_io,
+ &table_path,
+ primary_key,
+ &[("file.block-size", "1 mb")],
+ );
+ // The first rows hold nearly all bytes, so the batch average says
little about them.
+ let names = (0..65_536)
+ .map(|id| {
+ if id < 1_000 {
+ "x".repeat(8 * 1024)
+ } else {
+ "y".to_string()
+ }
+ })
+ .collect();
+ for (path, metadata) in write_sizing_batch(&file_io, &table,
&sizing_batch(names)).await
+ {
+ assert!(metadata.num_row_groups() > 1, "{path}");
+ for row_group in metadata.row_groups() {
+ assert!(
+ row_group.compressed_size() <= LIMIT * 3 / 2,
+ "{path}: {}",
+ row_group.compressed_size()
+ );
+ }
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_parquet_row_check_interval_keeps_default_page_row_limit() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_parquet_row_check_interval";
+ setup_dirs(&file_io, table_path).await;
+ let table = sizing_table(
+ &file_io,
+ table_path,
+ false,
+ &[("parquet.page.size.row.check.min", "65536")],
+ );
+ let names = (0..131_072).map(|id| (id % 10).to_string()).collect();
+ let default_limit =
parquet::file::properties::DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT as i64;
+ for (path, metadata) in write_sizing_batch(&file_io, &table,
&sizing_batch(names)).await {
+ let offsets = metadata.offset_index().expect("page index");
+ for (row_group, columns) in
metadata.row_groups().iter().zip(offsets) {
+ for column in columns {
+ let firsts = column
+ .page_locations()
+ .iter()
+ .map(|page| page.first_row_index)
+ .chain(std::iter::once(row_group.num_rows()))
+ .collect::<Vec<_>>();
+ assert!(
+ firsts
+ .windows(2)
+ .all(|pair| pair[1] - pair[0] <= default_limit),
+ "{path}: {firsts:?}"
+ );
+ }
+ }
+ }
+ }
+
#[tokio::test]
async fn test_append_write_populates_value_stats() {
let file_io = test_file_io();
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 8a417b2b..85e6ecc5 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -132,6 +132,22 @@ Dictionary encoding is enabled by default. Disable it for
all columns with
Paths use dot-separated physical Parquet leaves, not logical MAP keys.
These options affect new writes only.
+## Parquet Row Groups and Pages
+
+The Parquet writer accepts Java's sizing keys as plain integers:
+`parquet.block.size` (row group bytes), `parquet.page.size`,
+`parquet.dictionary.page.size`, `parquet.page.row.count.limit` and
+`parquet.page.size.row.check.min`. As in Java, `file.block-size` (for example
+`32 mb`) overrides `parquet.block.size`. Unset keys keep the parquet-rs
+defaults: no row group byte limit with at most 1,048,576 rows, 1 MiB data and
+dictionary pages, and 20,000 rows per page. parquet-rs checks page limits
between
+write batches of `parquet.page.size.row.check.min` rows (1,024 by default, and
+never more than the effective page row limit), so a page can exceed its row
limit
+by up to one batch. With a row group byte limit, the writer passes rows in
slices
+of at most a quarter of the limit and closes the row group once its encoded
size
+reaches the limit, so large or skewed batches stay close to it. These options
+affect new writes only.
+
## Parquet Page Pruning
Parquet Page Index pruning is enabled by default. Set
