Upgrade Iceberg to 1.9.2 and Avro to 1.12.0 (#784)

* upgrade to iceberg 1.10.1

* api changes

* use best available iceberg-hive-runtime

* upgrade avro version

* to run with java 17

* tests

* fix avro

* dont change notebook

* fix tests

* use iceberg 1.9.2

* unnecessary changes

* address ci issue

* fix version in LICENSE

* try revert

* add back avro schema change

* fix ci concurrency

* mvn spotless:apply

Author: kevinjqliu

pom.xml

@@ -63,7 +63,7 @@
         <project.build.outputTimestamp>2025-01-01T00:00:00Z</project.build.outputTimestamp>
         <maven.compiler.target>8</maven.compiler.target>
         <maven-compiler-plugin.version>3.13.0</maven-compiler-plugin.version>
-        <avro.version>1.11.4</avro.version>
+        <avro.version>1.12.0</avro.version>
         <log4j.version>2.22.0</log4j.version>
         <junit.version>5.11.4</junit.version>
         <junit.platform.runner.version>1.11.4</junit.platform.runner.version>
@@ -87,7 +87,8 @@
         <scala.binary.version>2.12</scala.binary.version>
         <spark.version>3.4.2</spark.version>
         <spark.version.prefix>3.4</spark.version.prefix>
-        <iceberg.version>1.4.2</iceberg.version>
+        <iceberg.version>1.9.2</iceberg.version>
+        <iceberg.hive.runtime.version>1.7.2</iceberg.hive.runtime.version>
         <delta.version>2.4.0</delta.version>
         <paimon.version>1.3.1</paimon.version>
         <jackson.version>2.18.2</jackson.version>
@@ -315,7 +316,7 @@
             <dependency>
                 <groupId>org.apache.iceberg</groupId>
                 <artifactId>iceberg-hive-runtime</artifactId>
-                <version>${iceberg.version}</version>
+                <version>${iceberg.hive.runtime.version}</version>
             </dependency>
 
             <!-- Delta -->

xtable-aws/src/main/resources/META-INF/LICENSE-bundled

@@ -217,7 +217,7 @@ com.google.guava:guava:32.1.3-jre
 com.lmax:disruptor:3.4.2
 io.delta:delta-core_2.12:2.4.0
 io.delta:delta-storage:2.4.0
-org.apache.avro:avro:1.11.4
+org.apache.avro:avro:1.12.0
 org.apache.hbase:hbase-client:2.4.9
 org.apache.hbase:hbase-common:2.4.9
 org.apache.hbase:hbase-server:2.4.9
@@ -226,8 +226,8 @@ org.apache.httpcomponents:fluent-hc:4.4.1
 org.apache.hudi:hudi-common:0.14.0
 org.apache.hudi:hudi-java-client:0.14.0
 org.apache.hudi:hudi-timeline-service:0.14.0
-org.apache.iceberg:iceberg-api:1.4.2
-org.apache.iceberg:iceberg-core:1.4.2
+org.apache.iceberg:iceberg-api:1.9.2
+org.apache.iceberg:iceberg-core:1.9.2
 org.apache.logging.log4j:log4j-1.2-api:2.22.0
 org.apache.logging.log4j:log4j-api:2.22.0
 org.javassist:javassist:3.25.0-GA

xtable-core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java

@@ -203,14 +203,18 @@ private static ColumnStat getColumnStatFromHudiStat(
     if (field.getSchema().getDataType() == InternalType.DECIMAL) {
       int scale =
           (int) field.getSchema().getMetadata().get(InternalSchema.MetadataKey.DECIMAL_SCALE);
-      minValue =
-          minValue instanceof ByteBuffer
-              ? convertBytesToBigDecimal((ByteBuffer) minValue, scale)
-              : ((BigDecimal) minValue).setScale(scale, RoundingMode.UNNECESSARY);
-      maxValue =
-          maxValue instanceof ByteBuffer
-              ? convertBytesToBigDecimal((ByteBuffer) maxValue, scale)
-              : ((BigDecimal) maxValue).setScale(scale, RoundingMode.UNNECESSARY);
+      if (minValue != null) {
+        minValue =
+            minValue instanceof ByteBuffer
+                ? convertBytesToBigDecimal((ByteBuffer) minValue, scale)
+                : ((BigDecimal) minValue).setScale(scale, RoundingMode.UNNECESSARY);
+      }
+      if (maxValue != null) {
+        maxValue =
+            maxValue instanceof ByteBuffer
+                ? convertBytesToBigDecimal((ByteBuffer) maxValue, scale)
+                : ((BigDecimal) maxValue).setScale(scale, RoundingMode.UNNECESSARY);
+      }
     }
     return getColumnStatFromValues(
         minValue,

xtable-core/src/test/java/org/apache/xtable/TestIcebergTable.java

@@ -371,7 +371,7 @@ private DataFile writeAndGetDataFile(List<Record> records, StructLike partitionK
     DataWriter<Record> dataWriter =
         Parquet.writeData(file)
             .schema(icebergTable.schema())
-            .createWriterFunc(GenericParquetWriter::buildWriter)
+            .createWriterFunc(GenericParquetWriter::create)
             .overwrite()
             .withSpec(icebergTable.spec())
             .withPartition(partitionKey)

xtable-core/src/test/java/org/apache/xtable/iceberg/ITIcebergConversionSource.java

@@ -41,6 +41,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -62,6 +64,7 @@
 import org.apache.xtable.model.storage.DataLayoutStrategy;
 import org.apache.xtable.model.storage.TableFormat;
 
+@Execution(ExecutionMode.SAME_THREAD)
 public class ITIcebergConversionSource {
   private static final Configuration hadoopConf = new Configuration();
 

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergConversionSource.java

@@ -37,6 +37,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 
 import org.apache.iceberg.*;
 import org.apache.iceberg.data.GenericRecord;
@@ -61,6 +63,7 @@
 import org.apache.xtable.model.storage.FileFormat;
 import org.apache.xtable.model.storage.InternalDataFile;
 
+@Execution(ExecutionMode.SAME_THREAD)
 class TestIcebergConversionSource {
 
   private IcebergTableManager tableManager;
@@ -432,7 +435,7 @@ private DataFile generateTestDataFile(int partition, Table table, String filePat
     DataWriter<GenericRecord> dataWriter =
         Parquet.writeData(table.io().newOutputFile(filePath))
             .schema(csSchema)
-            .createWriterFunc(GenericParquetWriter::buildWriter)
+            .createWriterFunc(GenericParquetWriter::create)
             .overwrite()
             .withSpec(table.spec())
             .withPartition(partitionInfo)

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergDataHelper.java

@@ -333,11 +333,7 @@ private Object generateRandomValueForType(
       case STRUCT:
         return generateInsertRecord(timeLowerBound, timeUpperBound, fieldType.asStructType());
       case UUID:
-        UUID uuid = UUID.randomUUID();
-        ByteBuffer byteBuffer = ByteBuffer.wrap(new byte[16]);
-        byteBuffer.putLong(uuid.getMostSignificantBits());
-        byteBuffer.putLong(uuid.getLeastSignificantBits());
-        return byteBuffer.array();
+        return UUID.randomUUID();
       case LIST:
         Types.ListType listType = (Types.ListType) fieldType;
         int listSize = RANDOM.nextInt(5) + 1;

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSchemaSync.java

@@ -181,7 +181,7 @@ public void testUpdateListElement() {
 
     verify(mockUpdateSchema).deleteColumn("array_field.element.element_string");
     verify(mockUpdateSchema)
-        .addColumn("array_field.element", "element_double", Types.DoubleType.get(), null);
+        .addColumn("array_field.element", "element_double", Types.DoubleType.get(), (String) null);
     verify(mockUpdateSchema).commit();
   }
 
@@ -194,7 +194,8 @@ public void testUpdateMapValue() {
 
     verify(mockUpdateSchema).deleteColumn("map_field.value.value_string");
     verify(mockUpdateSchema)
-        .addRequiredColumn("map_field.value", "value_double", Types.DoubleType.get(), null);
+        .addRequiredColumn(
+            "map_field.value", "value_double", Types.DoubleType.get(), (String) null);
     verify(mockUpdateSchema).commit();
   }
 

xtable-core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java

@@ -59,6 +59,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
@@ -106,6 +108,7 @@
  * Validates that the metadata for the table is properly created/updated. {@link
  * ITConversionController} validates that the table and its data can be properly read.
  */
+@Execution(ExecutionMode.SAME_THREAD)
 public class TestIcebergSync {
   private static final Random RANDOM = new Random();
   private static final Instant LAST_COMMIT_TIME = Instant.ofEpochSecond(1000);

xtable-core/src/test/resources/schemas/basic_schema.avsc

@@ -72,20 +72,23 @@
     },
     {
       "name": "bytes_field",
-      "type": "bytes",
-      "default": ""
+      "type": ["null", "bytes"],
+      "default": null
     },
     {
       "name": "decimal_field",
-      "type": {
-        "type": "fixed",
-        "name": "decimal_field_testing",
-        "size": 10,
-        "logicalType": "decimal",
-        "precision": 20,
-        "scale": 2
-      },
-      "default": "\u0000"
+      "type": [
+        "null",
+        {
+          "type": "fixed",
+          "name": "decimal_field_testing",
+          "size": 10,
+          "logicalType": "decimal",
+          "precision": 20,
+          "scale": 2
+        }
+      ],
+      "default": null
     },
     {
       "name": "nested_record",