dramaticlly commented on code in PR #16094:
URL: https://github.com/apache/iceberg/pull/16094#discussion_r4474228829
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java:
##########
@@ -774,6 +781,148 @@ private Pair<File, Long> generateFile(
return Pair.of(file, size);
}
+ @Test
+ void globalCompressionCodecAppliesToAllColumns() throws Exception {
Review Comment:
nit: this one seems already covered by perColumnCompressionCodec below. One
global and one override
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -374,6 +376,46 @@ private void setColumnStatsConfig(
});
}
+ private void setColumnCompressionConfig(
+ Context context,
+ Map<String, String> colNameToParquetPathMap,
+ BiConsumer<String, CompressionCodecName> withCompressionCodec,
+ BiConsumer<String, Integer> withCompressionLevel) {
+
+ Map<String, String> columnCompressionCodec =
context.columnCompressionCodec();
+ Map<String, String> columnCompressionLevel =
context.columnCompressionLevel();
+
+ Sets.union(columnCompressionCodec.keySet(),
columnCompressionLevel.keySet())
+ .forEach(
+ colName -> {
+ String parquetColumnPath =
colNameToParquetPathMap.get(colName);
+ if (parquetColumnPath == null) {
+ LOG.warn("Skipping per-column compression config for missing
field: {}", colName);
+ return;
+ }
+
+ String codecStr = columnCompressionCodec.get(colName);
+ if (codecStr != null) {
+ withCompressionCodec.accept(parquetColumnPath,
Context.toCodec(codecStr));
+ }
+
+ String level = columnCompressionLevel.get(colName);
+ if (level != null) {
+ withCompressionLevel.accept(
+ parquetColumnPath, parseCompressionLevel(colName,
level));
+ }
+ });
+ }
+
+ private int parseCompressionLevel(String colName, String level) {
Review Comment:
sugg: this can be static as not rely on internal states of the class
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java:
##########
@@ -774,6 +781,148 @@ private Pair<File, Long> generateFile(
return Pair.of(file, size);
}
+ @Test
+ void globalCompressionCodecAppliesToAllColumns() throws Exception {
+ File file = writeIntAndStringFields(ImmutableMap.of(PARQUET_COMPRESSION,
"snappy"), 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.SNAPPY);
+ }
+
+ @Test
+ void perColumnCompressionCodec() throws Exception {
+ File file =
+ writeIntAndStringFields(
+ ImmutableMap.of(
+ PARQUET_COMPRESSION,
+ "zstd",
+ PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX + "int_field",
+ "snappy"),
+ 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.ZSTD);
+ }
+
+ @Test
+ void perColumnCompressionCodecForNonExistentColumnIsIgnored() throws
Exception {
+ File file =
+ writeIntAndStringFields(
+ ImmutableMap.of(
+ PARQUET_COMPRESSION,
+ "snappy",
+ PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX + "non_existent_field",
+ "zstd"),
+ 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.SNAPPY);
+ }
+
+ @Test
+ void perColumnCompressionLevelIsApplied() throws Exception {
+ File lowLevelFile = writeWithStringFieldZstdLevel(1);
+ File highLevelFile = writeWithStringFieldZstdLevel(19);
+
+ assertThat(readColumnCodecs(highLevelFile))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
Review Comment:
nit: do we need this snappy entry? seems unrelated
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java:
##########
@@ -774,6 +781,148 @@ private Pair<File, Long> generateFile(
return Pair.of(file, size);
}
+ @Test
+ void globalCompressionCodecAppliesToAllColumns() throws Exception {
+ File file = writeIntAndStringFields(ImmutableMap.of(PARQUET_COMPRESSION,
"snappy"), 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.SNAPPY);
+ }
+
+ @Test
+ void perColumnCompressionCodec() throws Exception {
+ File file =
+ writeIntAndStringFields(
+ ImmutableMap.of(
+ PARQUET_COMPRESSION,
+ "zstd",
+ PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX + "int_field",
+ "snappy"),
+ 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.ZSTD);
+ }
+
+ @Test
+ void perColumnCompressionCodecForNonExistentColumnIsIgnored() throws
Exception {
+ File file =
+ writeIntAndStringFields(
+ ImmutableMap.of(
+ PARQUET_COMPRESSION,
+ "snappy",
+ PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX + "non_existent_field",
+ "zstd"),
+ 5);
+
+ assertThat(readColumnCodecs(file))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.SNAPPY);
+ }
+
+ @Test
+ void perColumnCompressionLevelIsApplied() throws Exception {
+ File lowLevelFile = writeWithStringFieldZstdLevel(1);
+ File highLevelFile = writeWithStringFieldZstdLevel(19);
+
+ assertThat(readColumnCodecs(highLevelFile))
+ .containsEntry("int_field", CompressionCodecName.SNAPPY)
+ .containsEntry("string_field", CompressionCodecName.ZSTD);
+ assertThat(readColumnSizes(highLevelFile).get("string_field"))
+ .isLessThan(readColumnSizes(lowLevelFile).get("string_field"));
+ }
+
+ private File writeWithStringFieldZstdLevel(int level) throws IOException {
+ Random random = new Random(0);
+ return writeIntAndStringFields(
+ ImmutableMap.of(
+ PARQUET_COMPRESSION,
+ "snappy",
+ PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX + "string_field",
+ "zstd",
+ PARQUET_COLUMN_COMPRESSION_LEVEL_PREFIX + "string_field",
+ String.valueOf(level)),
+ 20_000,
+ i -> "value-" + random.nextInt(5_000) + "-" + random.nextInt(5_000));
+ }
+
+ @Test
+ void invalidCompressionLevelThrows() {
Review Comment:
maybe also add invalid codec in the same method for
`invalidCompressionInputThrows`?
##########
parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java:
##########
@@ -374,6 +376,46 @@ private void setColumnStatsConfig(
});
}
+ private void setColumnCompressionConfig(
+ Context context,
+ Map<String, String> colNameToParquetPathMap,
+ BiConsumer<String, CompressionCodecName> withCompressionCodec,
+ BiConsumer<String, Integer> withCompressionLevel) {
+
+ Map<String, String> columnCompressionCodec =
context.columnCompressionCodec();
+ Map<String, String> columnCompressionLevel =
context.columnCompressionLevel();
+
+ Sets.union(columnCompressionCodec.keySet(),
columnCompressionLevel.keySet())
Review Comment:
curious, do we expect the codec and compression level to always configured
together? Trying to see if it's even valid to set level without codec?
##########
core/src/main/java/org/apache/iceberg/TableProperties.java:
##########
@@ -201,6 +201,12 @@ private TableProperties() {}
public static final String PARQUET_COLUMN_STATS_ENABLED_PREFIX =
"write.parquet.stats-enabled.column.";
+ public static final String PARQUET_COLUMN_COMPRESSION_CODEC_PREFIX =
Review Comment:
I knew there's previous concern about the column name changes, for
rename/drop/reAddedBack, I am wondering if worth following the existing pattern
in
https://github.com/apache/iceberg/blob/050378663b3513499dc7900d8d368ec07336309d/core/src/main/java/org/apache/iceberg/SchemaUpdate.java#L518-L525
to apply automatically in schema update?
--
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]
