feat: Upgrade iceberg-java to 1.9.2 and support adding required columns (#145)

Author: subkanthi

.github/workflows/verify.yaml

@@ -55,6 +55,6 @@ jobs:
         run: >
           ./mvnw -pl ice-rest-catalog -am install -DskipTests=true -Pno-check &&
           ./mvnw -pl ice-rest-catalog failsafe:integration-test failsafe:verify
-          -Dit.test=DockerScenarioBasedIT,DockerLocalFileIOClickHouseIT
+          -Dit.test=DockerScenarioBasedIT,DockerLocalFileIOClickHouseIT,DockerLocalFileIOClickHouseAllTypesIT
           -Ddocker.image=altinity/ice-rest-catalog:debug-with-ice-latest-master-amd64
           -Dclickhouse.image=altinity/clickhouse-server:25.8.16.20002.altinityantalya

ice-rest-catalog/pom.xml

@@ -244,6 +244,18 @@
           <groupId>com.fasterxml.jackson.core</groupId>
           <artifactId>jackson-core</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.core</groupId>
+          <artifactId>jackson-databind</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.core</groupId>
+          <artifactId>jackson-annotations</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.datatype</groupId>
+          <artifactId>jackson-datatype-jsr310</artifactId>
+        </exclusion>
         <exclusion>
           <groupId>com.google.errorprone</groupId>
           <artifactId>error_prone_annotations</artifactId>
@@ -577,6 +589,7 @@
           <excludes>
             <exclude>**/DockerScenarioBasedIT.java</exclude>
             <exclude>**/DockerLocalFileIOClickHouseIT.java</exclude>
+            <exclude>**/DockerLocalFileIOClickHouseAllTypesIT.java</exclude>
           </excludes>
         </configuration>
       </plugin>

ice-rest-catalog/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import org.apache.iceberg.BaseMetastoreOperations.CommitStatus;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
@@ -294,20 +295,39 @@ public long newSnapshotId() {
    * were attempting to set. This is used as a last resort when we are dealing with exceptions that
    * may indicate the commit has failed but are not proof that this is the case. Past locations must
    * also be searched on the chance that a second committer was able to successfully commit on top
-   * of our commit.
+   * of our commit. When the {@code newMetadataLocation} is not found, the method returns {@link
+   * CommitStatus#UNKNOWN}.
    *
    * @param newMetadataLocation the path of the new commit file
    * @param config metadata to use for configuration
-   * @return Commit Status of Success, Failure or Unknown
+   * @return Commit Status of Success, Unknown
    */
   protected CommitStatus checkCommitStatus(String newMetadataLocation, TableMetadata config) {
-    return CommitStatus.valueOf(
-        checkCommitStatus(
-                tableName(),
-                newMetadataLocation,
-                config.properties(),
-                () -> checkCurrentMetadataLocation(newMetadataLocation))
-            .name());
+    return checkCommitStatus(
+        tableName(),
+        newMetadataLocation,
+        config.properties(),
+        () -> checkCurrentMetadataLocation(newMetadataLocation));
+  }
+
+  /**
+   * Attempt to load the table and see if any current or past metadata location matches the one we
+   * were attempting to set. This is used as a last resort when we are dealing with exceptions that
+   * may indicate the commit has failed but are not proof that this is the case. Past locations must
+   * also be searched on the chance that a second committer was able to successfully commit on top
+   * of our commit. When the {@code newMetadataLocation} is not found, the method returns {@link
+   * CommitStatus#FAILURE}.
+   *
+   * @param newMetadataLocation the path of the new commit file
+   * @param config metadata to use for configuration
+   * @return Commit Status of Success, Failure or Unknown
+   */
+  protected CommitStatus checkCommitStatusStrict(String newMetadataLocation, TableMetadata config) {
+    return checkCommitStatusStrict(
+        tableName(),
+        newMetadataLocation,
+        config.properties(),
+        () -> checkCurrentMetadataLocation(newMetadataLocation));
   }
 
   /**