JingsongLi commented on code in PR #1072:
URL: https://github.com/apache/paimon-rust/pull/1072#discussion_r4216558835
##########
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:
[P2] Clamp the check interval against the effective default page row limit
The clamp only runs when `parquet.page.row.count.limit` is explicitly set.
With only `parquet.page.size.row.check.min=65536`, the effective page row limit
remains 20,000, but `write_batch_size` becomes 65,536. A filesystem TableWrite
probe writing 131,072 INT32 rows with dictionary and compression disabled
produces two consecutive pages of 65,536 rows, as confirmed by the offset
index. Since parquet-rs checks the limit after each write batch, the larger
interval defeats the default row limit and increases page-index read
granularity. This also contradicts the documented promise that the interval
never exceeds the page row limit. Please clamp against the effective limit even
when it is omitted, using `DEFAULT_DATA_PAGE_ROW_COUNT_LIMIT`, and add a
behavior test that configures only the check interval above that default.
##########
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:
[P2] Account for row-size skew when bounding the first slice
The whole-batch average does not bound the size of a particular slice. I
reproduced on this head through filesystem `TableWrite`: one 65,536-row `(INT,
VARCHAR)` batch, with the first 1,000 strings each 8 KiB and the remainder each
1 byte, `file.block-size=1 mb`, dictionary disabled, and compression disabled.
The append path produces a first row group of 8,259,966 bytes; the primary-key
path also produces oversized groups, including one of 4,262,119 bytes. Every
individual row is well below the 1 MiB limit, and both tables commit and read
back correctly. The average permits a first slice containing too many large
values, and parquet-rs accepts that slice into an empty group before checking
the byte limit. This leaves point-read granularity dependent on value
distribution despite the new bounding logic. Please base the bound on the
actual rows in each slice, accounting for physical encoding, and check the
writer's encoded size at sufficiently small intervals. Add a skewed variable-
width TableWrite regression alongside the uniform-INT case.
--
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]
