sundapeng commented on code in PR #1072:
URL: https://github.com/apache/paimon-rust/pull/1072#discussion_r4217998723
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -524,13 +600,21 @@ 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,
- })
+ // parquet-rs splits by bytes only once a row group holds rows, so
bound each write
+ // by the batch's in-memory row size, which is not smaller than its
encoded size.
+ let rows = batch.num_rows();
+ let step = self.max_row_group_bytes.map_or(rows, |limit| {
+ let row_bytes =
batch.get_array_memory_size().div_ceil(rows.max(1)).max(1);
+ (limit / row_bytes).max(1)
Review Comment:
Thanks, reproduced: the first row group was 8,259,966 bytes. Fixed in
43aab08. Each slice is now measured on its own rows (`batch_slice_bytes`: list
values and dictionaries outside the slice are not counted) and halved until it
holds at most a quarter of the limit. After every slice the writer checks the
encoded size of the open row group and closes it once it reaches the limit, so
the bound no longer depends on the batch average.
`test_parquet_block_size_bounds_row_groups_of_skewed_rows` is your case: one
65,536-row `(INT, VARCHAR)` batch whose first 1,000 values are 8 KiB, with
`file.block-size=1 mb`, for an append table and a primary-key table. It asserts
every row group stays within 1.5x the limit and reads all rows back after
commit.
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -445,7 +458,70 @@ 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 mut write_batch_size = positive(PAGE_SIZE_ROW_CHECK_MIN)?;
+ if let Some(rows) = positive(PAGE_ROW_COUNT_LIMIT)? {
+ props = props.set_data_page_row_count_limit(rows);
+ let batch =
write_batch_size.unwrap_or(parquet::file::properties::DEFAULT_WRITE_BATCH_SIZE);
+ write_batch_size = Some(batch.min(rows));
Review Comment:
Fixed in 43aab08: the check interval is now clamped against the effective
page row limit, which is parquet-rs' `DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT`
(20,000) when `parquet.page.row.count.limit` is not set.
`test_parquet_row_check_interval_keeps_default_page_row_limit` writes 131,072
rows through `TableWrite` with only `parquet.page.size.row.check.min=65536` and
checks the offset index; it showed pages of 65,536 rows before the change. The
unit test covers the same case, and the documentation now says "effective page
row limit".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
