[AMORO-2204] Support Spark 3.5 for mixed format tables

amoro-ams/src/main/java/org/apache/amoro/server/table/internal/InternalTableConstants.java

@@ -19,6 +19,7 @@
 package org.apache.amoro.server.table.internal;
 
 import org.apache.amoro.mixed.InternalMixedIcebergCatalog;
+import org.apache.amoro.table.MixedTable;
 
 /** Constants defines for internal table */
 public class InternalTableConstants {
@@ -37,6 +38,5 @@ public class InternalTableConstants {
   public static final String OSS_PROTOCOL_PREFIX = "oss://";
 
   public static final String CHANGE_STORE_TABLE_NAME_SUFFIX =
-      InternalMixedIcebergCatalog.CHANGE_STORE_SEPARATOR
-          + InternalMixedIcebergCatalog.CHANGE_STORE_NAME;
+      InternalMixedIcebergCatalog.CHANGE_STORE_SEPARATOR + MixedTable.CHANGE_STORE_IDENTIFIER;
 }

amoro-ams/src/main/resources/mysql/upgrade-0.4.1-to-0.5.0.sql

@@ -59,7 +59,7 @@ CREATE TABLE `table_identifier`
     `table_name`      varchar(128) NOT NULL COMMENT 'Table name',
     PRIMARY KEY (`table_id`),
     UNIQUE KEY `table_name_index` (`catalog_name`,`db_name`,`table_name`)
-) ROW_FORMAT=DYNAMIC;
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Table identifier for AMS' ROW_FORMAT=DYNAMIC;
 INSERT INTO `table_identifier` (`catalog_name`, `db_name`, `table_name`) SELECT `catalog_name`, `db_name`, `table_name` FROM `table_metadata`;
 
 -- table_metadata

amoro-ams/src/test/java/org/apache/amoro/server/util/TestIcebergTableUtil.java

@@ -21,6 +21,7 @@
 import org.apache.amoro.io.IcebergDataTestHelpers;
 import org.apache.amoro.io.MixedDataTestHelpers;
 import org.apache.amoro.mixed.MixedTables;
+import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.server.utils.IcebergTableUtil;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
@@ -123,7 +124,12 @@ private static Table newIcebergTable(Catalog catalog) {
   }
 
   private static MixedTable newMixedTable(Catalog catalog, boolean withKey) {
-    MixedTables mixedTables = new MixedTables(TableMetaStore.EMPTY, Maps.newHashMap(), catalog);
+    MixedTables mixedTables =
+        new MixedTables(
+            TableMetaStore.EMPTY,
+            Maps.newHashMap(),
+            catalog,
+            CatalogMetaProperties.MIXED_FORMAT_TABLE_STORE_SEPARATOR_DEFAULT);
     return mixedTables.createTable(
         org.apache.amoro.table.TableIdentifier.of("cata", "db", "table"),
         schema,

amoro-common/src/main/java/org/apache/amoro/client/PoolConfig.java

@@ -88,7 +88,7 @@ public void setMaxReconnects(int maxReconnects) {
 
   public static PoolConfig<?> forUrl(String url) {
     PoolConfig<?> poolConfig = new PoolConfig<>();
-    URLEncodedUtils.parse(URI.create(url), Charset.defaultCharset())
+    URLEncodedUtils.parse(URI.create(url), String.valueOf(Charset.defaultCharset()))
         .forEach(
             pair -> {
               try {

amoro-common/src/main/java/org/apache/amoro/client/ThriftClientPool.java

@@ -40,7 +40,6 @@ public class ThriftClientPool<
   private static final Logger LOG = LoggerFactory.getLogger(ThriftClientPool.class);
   private static final int RECONNECT_INTERVAL = 2000;
   private static final int BORROW_ATTEMPTS = 5;
-
   private final ThriftClientFactory clientFactory;
   private final ThriftPingFactory pingFactory;
   private final GenericObjectPool<ThriftClient<T>> pool;

amoro-common/src/main/java/org/apache/amoro/properties/CatalogMetaProperties.java

@@ -25,10 +25,10 @@ public class CatalogMetaProperties {
   public static final String STORAGE_CONFIGS_KEY_HDFS_SITE = "hadoop.hdfs.site";
   public static final String STORAGE_CONFIGS_KEY_CORE_SITE = "hadoop.core.site";
   public static final String STORAGE_CONFIGS_KEY_HIVE_SITE = "hive.site";
+
   public static final String STORAGE_CONFIGS_KEY_REGION = "storage.s3.region";
   public static final String STORAGE_CONFIGS_KEY_S3_ENDPOINT = "storage.s3.endpoint";
   public static final String STORAGE_CONFIGS_KEY_OSS_ENDPOINT = "storage.oss.endpoint";
-
   public static final String STORAGE_CONFIGS_VALUE_TYPE_HDFS_LEGACY = "hdfs";
   public static final String STORAGE_CONFIGS_VALUE_TYPE_HADOOP = "Hadoop";
   public static final String STORAGE_CONFIGS_VALUE_TYPE_S3 = "S3";

amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/BasicMixedIcebergCatalog.java

@@ -65,6 +65,7 @@ public class BasicMixedIcebergCatalog implements MixedFormatCatalog {
   private AmsClient client;
   private MixedTables tables;
   private SupportsNamespaces asNamespaceCatalog;
+  private String separator;
 
   @Override
   public String name() {
@@ -93,6 +94,7 @@ public void initialize(String name, Map<String, String> properties, TableMetaSto
     }
     this.databaseFilterPattern = databaseFilterPattern;
     this.catalogProperties = properties;
+    this.separator = tableStoreSeparator();
     this.tables = newMixedTables(metaStore, properties, icebergCatalog());
     if (properties.containsKey(CatalogMetaProperties.AMS_URI)) {
       this.client = new PooledAmsClient(properties.get(CatalogMetaProperties.AMS_URI));
@@ -221,7 +223,13 @@ protected MixedTables newMixedTables(
       TableMetaStore metaStore,
       Map<String, String> catalogProperties,
       org.apache.iceberg.catalog.Catalog icebergCatalog) {
-    return new MixedTables(metaStore, catalogProperties, icebergCatalog);
+    return new MixedTables(metaStore, catalogProperties, icebergCatalog, separator);
+  }
+
+  protected String tableStoreSeparator() {
+    return catalogProperties.getOrDefault(
+        CatalogMetaProperties.MIXED_FORMAT_TABLE_STORE_SEPARATOR,
+        CatalogMetaProperties.MIXED_FORMAT_TABLE_STORE_SEPARATOR_DEFAULT);
   }
 
   private org.apache.iceberg.catalog.TableIdentifier toIcebergTableIdentifier(

amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/InternalMixedIcebergCatalog.java

@@ -21,6 +21,7 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.apache.amoro.table.MixedTable;
 import org.apache.amoro.table.PrimaryKeySpec;
 import org.apache.amoro.table.TableMetaStore;
 import org.apache.hadoop.conf.Configuration;
@@ -43,7 +44,6 @@
 public class InternalMixedIcebergCatalog extends BasicMixedIcebergCatalog {
 
   public static final String CHANGE_STORE_SEPARATOR = "@";
-  public static final String CHANGE_STORE_NAME = "change";
 
   public static final String HTTP_HEADER_LIST_TABLE_FILTER = "LIST-TABLE-FILTER";
 
@@ -81,29 +81,23 @@ public List<org.apache.amoro.table.TableIdentifier> listTables(String database)
   @Override
   protected MixedTables newMixedTables(
       TableMetaStore metaStore, Map<String, String> catalogProperties, Catalog icebergCatalog) {
-    return new InternalMixedTables(metaStore, catalogProperties, icebergCatalog);
+    return new InternalMixedTables(
+        metaStore, catalogProperties, icebergCatalog, tableStoreSeparator());
+  }
+
+  @Override
+  protected String tableStoreSeparator() {
+    return CHANGE_STORE_SEPARATOR;
   }
 
   static class InternalMixedTables extends MixedTables {
 
     public InternalMixedTables(
-        TableMetaStore tableMetaStore, Map<String, String> catalogProperties, Catalog catalog) {
-      super(tableMetaStore, catalogProperties, catalog);
-    }
-
-    /**
-     * For internal table, using {table-name}@change as change store identifier, this identifier
-     * cloud be recognized by AMS. Due to '@' is an invalid character of table name, the change
-     * store identifier will never be conflict with other table name.
-     *
-     * @param baseIdentifier base store table identifier.
-     * @return change store iceberg table identifier.
-     */
-    @Override
-    protected TableIdentifier generateChangeStoreIdentifier(TableIdentifier baseIdentifier) {
-      return TableIdentifier.of(
-          baseIdentifier.namespace(),
-          baseIdentifier.name() + CHANGE_STORE_SEPARATOR + CHANGE_STORE_NAME);
+        TableMetaStore tableMetaStore,
+        Map<String, String> catalogProperties,
+        Catalog catalog,
+        String separator) {
+      super(tableMetaStore, catalogProperties, catalog, separator);
     }
 
     /**
@@ -121,6 +115,13 @@ protected Table createChangeStore(
       return tableMetaStore.doAs(() -> icebergCatalog.loadTable(changeIdentifier));
     }
 
+    @Override
+    protected TableIdentifier generateChangeStoreIdentifier(TableIdentifier baseIdentifier) {
+      return TableIdentifier.of(
+          baseIdentifier.namespace(),
+          baseIdentifier.name() + CHANGE_STORE_SEPARATOR + MixedTable.CHANGE_STORE_IDENTIFIER);
+    }
+
     /**
      * The change store will be dropped automatically by AMS when dropping the base store, so we do
      * nothing here

amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/MixedTables.java

@@ -21,7 +21,6 @@
 import org.apache.amoro.TableFormat;
 import org.apache.amoro.io.AuthenticatedFileIO;
 import org.apache.amoro.io.AuthenticatedFileIOs;
-import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.table.BaseTable;
 import org.apache.amoro.table.BasicKeyedTable;
@@ -32,6 +31,7 @@
 import org.apache.amoro.table.TableMetaStore;
 import org.apache.amoro.table.UnkeyedTable;
 import org.apache.amoro.utils.MixedFormatCatalogUtil;
+import org.apache.amoro.utils.MixedTableUtil;
 import org.apache.amoro.utils.TablePropertyUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -47,16 +47,20 @@
 public class MixedTables {
   private static final Logger LOG = LoggerFactory.getLogger(MixedTables.class);
 
-  protected TableMetaStore tableMetaStore;
-  protected Catalog icebergCatalog;
-
-  protected Map<String, String> catalogProperties;
+  protected final TableMetaStore tableMetaStore;
+  protected final Catalog icebergCatalog;
+  protected final Map<String, String> catalogProperties;
+  protected final String separator;
 
   public MixedTables(
-      TableMetaStore tableMetaStore, Map<String, String> catalogProperties, Catalog catalog) {
+      TableMetaStore tableMetaStore,
+      Map<String, String> catalogProperties,
+      Catalog catalog,
+      String separator) {
     this.tableMetaStore = tableMetaStore;
     this.icebergCatalog = catalog;
     this.catalogProperties = catalogProperties;
+    this.separator = separator;
   }
 
   /**
@@ -86,12 +90,9 @@ public TableIdentifier parseChangeIdentifier(Table base) {
    * @return change store table identifier.
    */
   protected TableIdentifier generateChangeStoreIdentifier(TableIdentifier baseIdentifier) {
-    String separator =
-        catalogProperties.getOrDefault(
-            CatalogMetaProperties.MIXED_FORMAT_TABLE_STORE_SEPARATOR,
-            CatalogMetaProperties.MIXED_FORMAT_TABLE_STORE_SEPARATOR_DEFAULT);
     return TableIdentifier.of(
-        baseIdentifier.namespace(), baseIdentifier.name() + separator + "change" + separator);
+        baseIdentifier.namespace(),
+        MixedTableUtil.changeStoreName(baseIdentifier.name(), separator));
   }
 
   /**

amoro-format-iceberg/src/main/java/org/apache/amoro/table/MixedTable.java

@@ -31,6 +31,8 @@
 /** Represents an mixed-format table. */
 public interface MixedTable extends Serializable {
 
+  String CHANGE_STORE_IDENTIFIER = "change";
+
   /** Returns the {@link TableIdentifier} of this table */
   TableIdentifier id();
 

amoro-format-iceberg/src/main/java/org/apache/amoro/utils/MixedTableUtil.java

@@ -190,4 +190,27 @@ public static PartitionSpec getMixedTablePartitionSpecById(MixedTable mixedTable
       return spec;
     }
   }
+
+  /**
+   * Generate change store table name for mixed format tables.
+   *
+   * @param tableName mixed format table name.
+   * @param separator change store separator.
+   * @return change store table name.
+   */
+  public static String changeStoreName(String tableName, String separator) {
+    return tableName + separator + MixedTable.CHANGE_STORE_IDENTIFIER + separator;
+  }
+
+  /**
+   * Determine if it is a change store of a mixed format table by the table name.
+   *
+   * @param tableName table name.
+   * @param separator change store separator.
+   * @return if it is a change store.
+   */
+  public static boolean isChangeStore(String tableName, String separator) {
+    return tableName != null
+        && tableName.endsWith(separator + MixedTable.CHANGE_STORE_IDENTIFIER + separator);
+  }
 }

amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/test/java/org/apache/amoro/spark/test/SparkTestBase.java

@@ -220,10 +220,16 @@ protected void createHiveSource(
   }
 
   protected void createViewSource(Schema schema, List<Record> data) {
+    createViewSource(schema, data, Double.NaN);
+  }
+
+  protected void createViewSource(Schema schema, List<Record> data, double version) {
     Dataset<Row> ds =
         spark()
             .createDataFrame(
-                data.stream().map(TestTableUtil::recordToRow).collect(Collectors.toList()),
+                data.stream()
+                    .map(r -> TestTableUtil.recordToRow(r, version))
+                    .collect(Collectors.toList()),
                 SparkSchemaUtil.convert(schema));
 
     ds.createOrReplaceTempView(sourceTable);

amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/test/java/org/apache/amoro/spark/test/utils/TestTableUtil.java

@@ -72,17 +72,17 @@
 import java.util.stream.Collectors;
 
 public class TestTableUtil {
-
-  private static Object[] recordToObjects(Record record) {
+  private static Object[] recordToObjects(Record record, double sparkVersion) {
     Object[] values = new Object[record.size()];
     for (int i = 0; i < values.length; i++) {
       Object v = record.get(i);
       if (v instanceof LocalDateTime) {
-        Timestamp ts =
-            Timestamp.valueOf(((LocalDateTime) v).atZone(ZoneOffset.UTC).toLocalDateTime());
-        Timestamp tsUTC = Timestamp.valueOf((LocalDateTime) v);
-        values[i] = ts;
-        continue;
+        if (Double.isNaN(sparkVersion) || sparkVersion < 3.4) {
+          Timestamp ts =
+              Timestamp.valueOf(((LocalDateTime) v).atZone(ZoneOffset.UTC).toLocalDateTime());
+          values[i] = ts;
+          continue;
+        }
       } else if (v instanceof OffsetDateTime) {
         v = new Timestamp(((OffsetDateTime) v).toInstant().toEpochMilli());
       }
@@ -92,7 +92,11 @@ private static Object[] recordToObjects(Record record) {
   }
 
   public static Row recordToRow(Record record) {
-    Object[] values = recordToObjects(record);
+    return recordToRow(record, Double.NaN);
+  }
+
+  public static Row recordToRow(Record record, double sparkVersion) {
+    Object[] values = recordToObjects(record, sparkVersion);
     return RowFactory.create(values);
   }
 
@@ -103,6 +107,10 @@ public static InternalRow recordToInternalRow(Schema schema, Record record) {
   }
 
   public static Record rowToRecord(Row row, Types.StructType type) {
+    return rowToRecord(row, type, Double.NaN);
+  }
+
+  public static Record rowToRecord(Row row, Types.StructType type, double sparkVersion) {
     Record record = GenericRecord.create(type);
     for (int i = 0; i < type.fields().size(); i++) {
       Object v = row.get(i);
@@ -114,9 +122,14 @@ public static Record rowToRecord(Row row, Types.StructType type) {
         record.set(i, offsetDateTime);
         continue;
       } else if (field.type().equals(Types.TimestampType.withoutZone())) {
-        Preconditions.checkArgument(v instanceof Timestamp);
-        Object localDatetime = ((Timestamp) v).toLocalDateTime();
-        record.set(i, localDatetime);
+        if (Double.isNaN(sparkVersion) || sparkVersion < 3.4) {
+          Preconditions.checkArgument(v instanceof Timestamp);
+          Object localDatetime = ((Timestamp) v).toLocalDateTime();
+          record.set(i, localDatetime);
+        } else {
+          Preconditions.checkArgument(v instanceof LocalDateTime);
+          record.set(i, v);
+        }
         continue;
       }
       record.set(i, v);

amoro-format-mixed/amoro-mixed-spark/pom.xml

@@ -37,5 +37,7 @@
         <module>v3.2/amoro-mixed-spark-runtime-3.2</module>
         <module>v3.3/amoro-mixed-spark-3.3</module>
         <module>v3.3/amoro-mixed-spark-runtime-3.3</module>
+        <module>v3.5/amoro-mixed-spark-3.5</module>
+        <module>v3.5/amoro-mixed-spark-runtime-3.5</module>
     </modules>
 </project>