[HOTFIX][#3705] fix(spark-connector): write data to Iceberg table failed if drop and recreate Iceberg table (#3718)

Co-authored-by: FANNG <xiaojing@datastrato.com>
Fix: #3705

Author: jerryshao

docs/spark-connector/spark-catalog-iceberg.md

@@ -132,3 +132,8 @@ Gravitino spark connector will transform below property names which are defined
 | `warehouse`                     | `warehouse`                           | (none)        | No       | Catalog backend warehouse | 0.5.1         |
 
 Gravitino catalog property names with the prefix `spark.bypass.` are passed to Spark Iceberg connector. For example, using `spark.bypass.io-impl` to pass the `io-impl` to the Spark Iceberg connector.
+
+:::info
+Iceberg catalog property `cache-enabled` is setting to `false` internally and not allowed to change.
+:::
+

spark-connector/spark-common/src/main/java/com/datastrato/gravitino/spark/connector/iceberg/IcebergPropertiesConstants.java

@@ -75,5 +75,7 @@ public class IcebergPropertiesConstants {
   @VisibleForTesting
   public static final String ICEBERG_FORMAT_VERSION = IcebergTablePropertiesMetadata.FORMAT_VERSION;
 
+  static final String ICEBERG_CATALOG_CACHE_ENABLED = CatalogProperties.CACHE_ENABLED;
+
   private IcebergPropertiesConstants() {}
 }

spark-connector/spark-common/src/main/java/com/datastrato/gravitino/spark/connector/iceberg/IcebergPropertiesConverter.java

@@ -52,6 +52,9 @@ public Map<String, String> toSparkCatalogProperties(Map<String, String> properti
         throw new IllegalArgumentException(
             "Unsupported Iceberg Catalog backend: " + catalogBackend);
     }
+
+    all.put(IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED, "FALSE");
+
     return all;
   }
 

spark-connector/spark-common/src/main/java/com/datastrato/gravitino/spark/connector/iceberg/SparkIcebergTable.java

@@ -10,7 +10,6 @@
 import com.datastrato.gravitino.spark.connector.SparkTransformConverter;
 import com.datastrato.gravitino.spark.connector.SparkTypeConverter;
 import com.datastrato.gravitino.spark.connector.utils.GravitinoTableInfoHelper;
-import java.lang.reflect.Field;
 import java.util.Map;
 import org.apache.iceberg.spark.SparkCatalog;
 import org.apache.iceberg.spark.source.SparkTable;
@@ -37,7 +36,10 @@ public SparkIcebergTable(
       PropertiesConverter propertiesConverter,
       SparkTransformConverter sparkTransformConverter,
       SparkTypeConverter sparkTypeConverter) {
-    super(sparkTable.table(), !isCacheEnabled(sparkCatalog));
+    super(
+        sparkTable.table(), true
+        /** refreshEagerly */
+        );
     this.gravitinoTableInfoHelper =
         new GravitinoTableInfoHelper(
             true,
@@ -78,14 +80,4 @@ public Transform[] partitioning() {
   public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
     return ((SparkTable) sparkTable).newScanBuilder(options);
   }
-
-  private static boolean isCacheEnabled(SparkCatalog sparkCatalog) {
-    try {
-      Field cacheEnabled = sparkCatalog.getClass().getDeclaredField("cacheEnabled");
-      cacheEnabled.setAccessible(true);
-      return cacheEnabled.getBoolean(sparkCatalog);
-    } catch (NoSuchFieldException | IllegalAccessException e) {
-      throw new RuntimeException("Failed to get cacheEnabled field from SparkCatalog", e);
-    }
-  }
 }

spark-connector/spark-common/src/test/java/com/datastrato/gravitino/spark/connector/iceberg/TestIcebergPropertiesConverter.java

@@ -29,6 +29,8 @@ void testCatalogPropertiesWithHiveBackend() {
                 "value1"));
     Assertions.assertEquals(
         ImmutableMap.of(
+            IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED,
+            "FALSE",
             IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
             IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_HIVE,
             IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
@@ -57,6 +59,8 @@ void testCatalogPropertiesWithJdbcBackend() {
                 "value1"));
     Assertions.assertEquals(
         ImmutableMap.of(
+            IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED,
+            "FALSE",
             IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
             IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_JDBC,
             IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
@@ -85,6 +89,8 @@ void testCatalogPropertiesWithRestBackend() {
                 "value1"));
     Assertions.assertEquals(
         ImmutableMap.of(
+            IcebergPropertiesConstants.ICEBERG_CATALOG_CACHE_ENABLED,
+            "FALSE",
             IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE,
             IcebergPropertiesConstants.ICEBERG_CATALOG_BACKEND_REST,
             IcebergPropertiesConstants.ICEBERG_CATALOG_URI,

spark-connector/spark-common/src/test/java/com/datastrato/gravitino/spark/connector/integration/test/SparkCommonIT.java

@@ -748,6 +748,19 @@ void testTableOptions() {
     checkTableReadWrite(tableInfo);
   }
 
+  @Test
+  void testDropAndWriteTable() {
+    String tableName = "drop_then_create_write_table";
+
+    dropTableIfExists(tableName);
+    createSimpleTable(tableName);
+    checkTableReadWrite(getTableInfo(tableName));
+
+    dropTableIfExists(tableName);
+    createSimpleTable(tableName);
+    checkTableReadWrite(getTableInfo(tableName));
+  }
+
   @Test
   @EnabledIf("supportsDelete")
   void testDeleteOperation() {

spark-connector/spark-common/src/test/java/com/datastrato/gravitino/spark/connector/integration/test/iceberg/SparkIcebergCatalogIT.java

@@ -333,8 +333,6 @@ void testIcebergReservedProperties() throws NoSuchTableException {
             "CREATE TABLE %s (id INT NOT NULL COMMENT 'id comment', name STRING COMMENT '', age INT)",
             tableName));
 
-    SparkIcebergTable sparkIcebergTable = getSparkIcebergTableInstance(tableName);
-
     SparkTableInfo tableInfo = getTableInfo(tableName);
     Map<String, String> tableProperties = tableInfo.getTableProperties();
     Assertions.assertNotNull(tableProperties);
@@ -360,6 +358,7 @@ void testIcebergReservedProperties() throws NoSuchTableException {
     // create a new snapshot
     sql(String.format("INSERT INTO %s VALUES(1, '1', 1)", tableName));
 
+    SparkIcebergTable sparkIcebergTable = getSparkIcebergTableInstance(tableName);
     // set Identifier fields
     sparkIcebergTable.table().updateSchema().setIdentifierFields("id").commit();