steveloughran commented on code in PR #16327:
URL: https://github.com/apache/iceberg/pull/16327#discussion_r3310276013
##########
parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java:
##########
@@ -190,6 +204,30 @@ public List<Long> splitOffsets() {
}
private void checkSize() {
+ if (trackUncompressedSize) {
+ checkSizeUncompressed();
+ } else {
+ checkSizeCompressed();
+ }
+ }
+
+ private void checkSizeUncompressed() {
+ if (rowGroupUncompressedSize >= targetRowGroupSize) {
+ flushRowGroup(false);
+ } else if (recordCount >= nextCheckRecordCount) {
+ double avgRecordSize = ((double) rowGroupUncompressedSize) / recordCount;
+ if (rowGroupUncompressedSize > (targetRowGroupSize - 2 * avgRecordSize))
{
Review Comment:
this is very much a duplicate of the body of `checkSizeCompressed()`
replacing
` writeStore.getBufferedSize()` with `rowGroupUncompressedSize`.
I would normally have factored that out and reused in both places, passing
in the different size values as a parameter. The algorithm is fiddly enough to
justify it, especially with the type conversions between floats and longs;
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestParquetDataWriter.java:
##########
@@ -543,4 +546,69 @@ protected int resolveColumnIndex(Void engineSchema, String
columnName) {
variantSchema.asStruct(), variantRecords.get(i),
writtenRecords.get(i));
}
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"gzip", "snappy", "zstd", "uncompressed"})
Review Comment:
being ruthless here, is gzip needed?
I'm just thinking test time both locally and in GHA resource consumption.
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestParquetDataWriter.java:
##########
@@ -543,4 +546,69 @@ protected int resolveColumnIndex(Void engineSchema, String
columnName) {
variantSchema.asStruct(), variantRecords.get(i),
writtenRecords.get(i));
}
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"gzip", "snappy", "zstd", "uncompressed"})
+ public void testRowGroupSizeEnforcedWhenCompressionEnabled(String codec)
throws IOException {
+ // With uncompressed tracking, row groups split at the configured target
+ DataFile dataFile = writeCompressibleRecords(codec, true);
+
+ assertThat(dataFile.recordCount()).as("Record count should
match").isEqualTo(100);
+ assertThat(dataFile.splitOffsets().size())
+ .as("Row group count should reflect enforcement of the 8 MB target")
+ .isGreaterThanOrEqualTo(4);
+ }
+
+ @Test
+ public void testDefaultPathUsesCompressedSize() throws IOException {
+ // Without uncompressed tracking, compressed bytes never hit the target
+ DataFile dataFile = writeCompressibleRecords("gzip", false);
Review Comment:
use whatever codec is the fastest here
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
