Test Parquet MR 1.14.0

Author: Fokko

.palantir/revapi.yml

@@ -1018,6 +1018,11 @@ acceptedBreaks:
       old: "method void org.apache.iceberg.PositionDeletesTable.PositionDeletesBatchScan::<init>(org.apache.iceberg.Table,\
         \ org.apache.iceberg.Schema, org.apache.iceberg.TableScanContext)"
       justification: "Removing deprecated code"
+  "1.5.0":
+    org.apache.iceberg:iceberg-parquet:
+    - code: "java.class.removed"
+      old: "class org.apache.iceberg.parquet.ParquetCodecFactory"
+      justification: "Workaround for a bug in Parquet-MR"
   apache-iceberg-0.14.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

build.gradle

@@ -128,6 +128,9 @@ allprojects {
   repositories {
     mavenCentral()
     mavenLocal()
+    maven {
+      url = uri("https://repository.apache.org/content/repositories/orgapacheparquet-1054")
+    }
   }
 }
 

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java

@@ -217,27 +217,27 @@ public void testPrimitiveColumns() throws Exception {
 
     Row binaryCol =
         Row.of(
-            52L,
+            55L,
             4L,
             2L,
             null,
             Base64.getDecoder().decode("1111"),
             Base64.getDecoder().decode("2222"));
-    Row booleanCol = Row.of(32L, 4L, 0L, null, false, true);
-    Row decimalCol = Row.of(85L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
-    Row doubleCol = Row.of(85L, 4L, 0L, 1L, 1.0D, 2.0D);
+    Row booleanCol = Row.of(36L, 4L, 0L, null, false, true);
+    Row decimalCol = Row.of(91L, 4L, 1L, null, new BigDecimal("1.00"), new BigDecimal("2.00"));
+    Row doubleCol = Row.of(91L, 4L, 0L, 1L, 1.0D, 2.0D);
     Row fixedCol =
         Row.of(
-            44L,
+            47L,
             4L,
             2L,
             null,
             Base64.getDecoder().decode("1111"),
             Base64.getDecoder().decode("2222"));
-    Row floatCol = Row.of(71L, 4L, 0L, 2L, 0f, 0f);
-    Row intCol = Row.of(71L, 4L, 0L, null, 1, 2);
-    Row longCol = Row.of(79L, 4L, 0L, null, 1L, 2L);
-    Row stringCol = Row.of(79L, 4L, 0L, null, "1", "2");
+    Row floatCol = Row.of(77L, 4L, 0L, 2L, 0f, 0f);
+    Row intCol = Row.of(77L, 4L, 0L, null, 1, 2);
+    Row longCol = Row.of(85L, 4L, 0L, null, 1L, 2L);
+    Row stringCol = Row.of(85L, 4L, 0L, null, "1", "2");
 
     List<Row> expected =
         Lists.newArrayList(

gradle/libs.versions.toml

@@ -70,7 +70,7 @@ netty-buffer = "4.1.109.Final"
 netty-buffer-compat = "4.1.109.Final"
 object-client-bundle = "3.3.2"
 orc = "1.9.3"
-parquet = "1.13.1"
+parquet = "1.14.0"
 pig = "0.17.0"
 roaringbitmap = "1.0.6"
 s3mock-junit5 = "2.11.0"

parquet/src/main/java/org/apache/iceberg/parquet/ParquetCodecFactory.java

@@ -86,8 +86,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
     this.targetRowGroupSize = rowGroupSize;
     this.props = properties;
     this.metadata = ImmutableMap.copyOf(metadata);
-    this.compressor =
-        new ParquetCodecFactory(conf, props.getPageSizeThreshold()).getCompressor(codec);
+    this.compressor = new CodecFactory(conf, props.getPageSizeThreshold()).getCompressor(codec);
     this.parquetSchema = ParquetSchemaUtil.convert(schema, "table");
     this.model = (ParquetValueWriter<T>) createWriterFunc.apply(parquetSchema);
     this.metricsConfig = metricsConfig;