From d074be0da32ff7a5c14f657c8f0bfc2168853a7c Mon Sep 17 00:00:00 2001
From: hemwang <hemwang@cisco.com>
Date: Wed, 20 Nov 2024 17:19:34 +0800
Subject: [PATCH] TestMetadataTableReadableMetrics get expected size from the
 underlying data

---
 .../TestMetadataTableReadableMetrics.java     | 37 ++++++++++++++-----
 gradle/libs.versions.toml                     |  2 +-
 2 files changed, 29 insertions(+), 10 deletions(-)

diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
index 9cf953342a18..f2fc6f4872bc 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestMetadataTableReadableMetrics.java
@@ -27,7 +27,9 @@
 import java.nio.ByteBuffer;
 import java.nio.file.Path;
 import java.util.Base64;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.table.api.TableEnvironment;
@@ -98,6 +100,8 @@ protected TableEnvironment getTableEnv() {
           optional(8, "fixedCol", Types.FixedType.ofLength(3)),
           optional(9, "binaryCol", Types.BinaryType.get()));
 
+  private Map<String, Long> columnNameSizeMap;
+
   private Table createPrimitiveTable() throws IOException {
     Table table =
         catalog.createTable(
@@ -133,6 +137,14 @@ private Table createPrimitiveTable() throws IOException {
 
     File testFile = File.createTempFile("junit", null, temp.toFile());
     DataFile dataFile = FileHelpers.writeDataFile(table, Files.localOutput(testFile), records);
+    Map<Integer, Long> columnSizes = dataFile.columnSizes();
+    columnNameSizeMap = new HashMap<>();
+    for (Map.Entry<Integer, Long> entry : columnSizes.entrySet()) {
+      int fieldId = entry.getKey();
+      long sizeInBytes = entry.getValue();
+      Types.NestedField field = PRIMITIVE_SCHEMA.findField(fieldId);
+      columnNameSizeMap.put(field.name(), sizeInBytes);
+    }
     table.newAppend().appendFile(dataFile).commit();
     return table;
   }
@@ -217,27 +229,34 @@ public void testPrimitiveColumns() throws Exception {
 
     Row binaryCol =
         Row.of(
-            52L,
+            columnNameSizeMap.get("binaryCol"),
             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(columnNameSizeMap.get("booleanCol"), 4L, 0L, null, false, true);
+    Row decimalCol =
+        Row.of(
+            columnNameSizeMap.get("decimalCol"),
+            4L,
+            1L,
+            null,
+            new BigDecimal("1.00"),
+            new BigDecimal("2.00"));
+    Row doubleCol = Row.of(columnNameSizeMap.get("doubleCol"), 4L, 0L, 1L, 1.0D, 2.0D);
     Row fixedCol =
         Row.of(
-            44L,
+            columnNameSizeMap.get("fixedCol"),
             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(columnNameSizeMap.get("floatCol"), 4L, 0L, 2L, 0f, 0f);
+    Row intCol = Row.of(columnNameSizeMap.get("intCol"), 4L, 0L, null, 1, 2);
+    Row longCol = Row.of(columnNameSizeMap.get("longCol"), 4L, 0L, null, 1L, 2L);
+    Row stringCol = Row.of(columnNameSizeMap.get("stringCol"), 4L, 0L, null, "1", "2");
 
     List<Row> expected =
         Lists.newArrayList(
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 96f62fb8163b..2e967150601c 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -75,7 +75,7 @@ netty-buffer = "4.1.115.Final"
 netty-buffer-compat = "4.1.115.Final"
 object-client-bundle = "3.3.2"
 orc = "1.9.4"
-parquet = "1.13.1"
+parquet = "1.14.3"
 roaringbitmap = "1.3.0"
 scala-collection-compat = "2.12.0"
 slf4j = "2.0.16"