Support hoodie catalog

Author: cuibo01

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieHiveCatalog.java

@@ -32,6 +32,8 @@
 import org.apache.hudi.util.StreamerUtil;
 
 import org.apache.avro.Schema;
+
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.sql.parser.hive.ddl.SqlAlterHiveDatabase;
 import org.apache.flink.sql.parser.hive.ddl.SqlAlterHiveDatabaseOwner;
@@ -47,8 +49,6 @@
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.CatalogView;
 import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.ResolvedCatalogTable;
-import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
 import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
@@ -352,7 +352,8 @@ private org.apache.avro.Schema getLatestTableSchema(String path) {
     return null;
   }
 
-  private Table getHiveTable(ObjectPath tablePath) throws TableNotExistException {
+  @VisibleForTesting
+  public Table getHiveTable(ObjectPath tablePath) throws TableNotExistException {
     try {
       Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
       return checkHoodieTable(hiveTable);
@@ -426,7 +427,7 @@ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
     }
 
     if (table instanceof CatalogView) {
-      throw new HoodieCatalogException("view not supported.");
+      throw new HoodieCatalogException("Hoodie catalog does not support to CREATE VIEW.");
     }
 
     try {
@@ -436,7 +437,7 @@ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
       //create hive table
       client.createTable(hiveTable);
       //init hoodie metaClient
-      initTableIfNotExists(tablePath, (ResolvedCatalogTable) table);
+      initTableIfNotExists(tablePath, (CatalogTable)table);
     } catch (AlreadyExistsException e) {
       if (!ignoreIfExists) {
         throw new TableAlreadyExistException(getName(), tablePath, e);
@@ -447,12 +448,10 @@ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
     }
   }
 
-  private void initTableIfNotExists(ObjectPath tablePath, ResolvedCatalogTable catalogTable) {
-    Map<String, String> options = catalogTable.getOptions();
-    Configuration flinkConf = Configuration.fromMap(catalogTable.getOptions());
+  private void initTableIfNotExists(ObjectPath tablePath, CatalogTable catalogTable) {
+    Configuration flinkConf = Configuration.fromMap(applyOptionsHook(catalogTable.getOptions()));
     flinkConf.addAllToProperties(hiveConf.getAllProperties());
-    ResolvedSchema resolvedSchema = catalogTable.getResolvedSchema();
-    final String avroSchema = AvroSchemaConverter.convertToSchema(resolvedSchema.toPhysicalRowDataType().getLogicalType()).toString();
+    final String avroSchema = AvroSchemaConverter.convertToSchema(catalogTable.getSchema().toPhysicalRowDataType().getLogicalType()).toString();
     flinkConf.setString(FlinkOptions.SOURCE_AVRO_SCHEMA, avroSchema);
 
     // stores two copies of options:
@@ -461,16 +460,23 @@ private void initTableIfNotExists(ObjectPath tablePath, ResolvedCatalogTable cat
     // because the HoodieTableMetaClient is a heavy impl, we try to avoid initializing it
     // when calling #getTable.
 
-    final String pkColumns = String.join(",", resolvedSchema.getPrimaryKey().get().getColumns());
-    flinkConf.setString(FlinkOptions.RECORD_KEY_FIELD, pkColumns);
-    options.put(TableOptionProperties.PK_CONSTRAINT_NAME, resolvedSchema.getPrimaryKey().get().getName());
-    options.put(TableOptionProperties.PK_COLUMNS, pkColumns);
+    if (catalogTable.getUnresolvedSchema().getPrimaryKey().isPresent()) {
+      final String pkColumns = String.join(",", catalogTable.getUnresolvedSchema().getPrimaryKey().get().getColumnNames());
+      String recordKey = flinkConf.get(FlinkOptions.RECORD_KEY_FIELD);
+      if (!Objects.equals(pkColumns, recordKey)) {
+        throw new HoodieCatalogException(String.format("%s and %s are the different", pkColumns, recordKey));
+      }
+    }
 
     if (catalogTable.isPartitioned()) {
       final String partitions = String.join(",", catalogTable.getPartitionKeys());
       flinkConf.setString(FlinkOptions.PARTITION_PATH_FIELD, partitions);
-      options.put(TableOptionProperties.PARTITION_COLUMNS, partitions);
     }
+
+    if (!flinkConf.getOptional(PATH).isPresent()) {
+      flinkConf.setString(PATH, inferTablePath(tablePath, catalogTable));
+    }
+
     flinkConf.setString(FlinkOptions.TABLE_NAME, tablePath.getObjectName());
     try {
       StreamerUtil.initTableIfNotExists(flinkConf);
@@ -485,14 +491,27 @@ private String inferTablePath(ObjectPath tablePath, CatalogBaseTable table) {
       try {
         Path dbLocation = new Path(client.getDatabase(tablePath.getDatabaseName()).getLocationUri());
         location = new Path(dbLocation, tablePath.getObjectName()).toString();
-        table.getOptions().put(PATH.key(), location);
       } catch (TException e) {
         throw new HoodieCatalogException(String.format("Failed to infer hoodie table path for table %s", tablePath), e);
       }
     }
     return location;
   }
 
+  private Map<String, String> applyOptionsHook(Map<String, String> options) {
+    Map<String, String> properties = new HashMap<>(options);
+    if (!options.containsKey(FlinkOptions.RECORD_KEY_FIELD)) {
+      properties.put(FlinkOptions.RECORD_KEY_FIELD.key(), FlinkOptions.RECORD_KEY_FIELD.defaultValue());
+    }
+    if (!options.containsKey(FlinkOptions.PRECOMBINE_FIELD)) {
+      properties.put(FlinkOptions.PRECOMBINE_FIELD.key(), FlinkOptions.PRECOMBINE_FIELD.defaultValue());
+    }
+    if (!options.containsKey(FlinkOptions.TABLE_TYPE)) {
+      properties.put(FlinkOptions.TABLE_TYPE.key(), FlinkOptions.TABLE_TYPE.defaultValue());
+    }
+    return properties;
+  }
+
   private Table instantiateHiveTable(ObjectPath tablePath, CatalogBaseTable table, String location, boolean useRealTimeInputFormat) throws IOException {
     // let Hive set default parameters for us, e.g. serialization.format
     Table hiveTable =
@@ -501,7 +520,7 @@ private Table instantiateHiveTable(ObjectPath tablePath, CatalogBaseTable table,
     hiveTable.setOwner(UserGroupInformation.getCurrentUser().getUserName());
     hiveTable.setCreateTime((int) (System.currentTimeMillis() / 1000));
 
-    Map<String, String> properties = new HashMap<>(table.getOptions());
+    Map<String, String> properties = applyOptionsHook(table.getOptions());
     properties.put("EXTERNAL", "TRUE");
     // Table comment
     if (table.getComment() != null) {
@@ -551,20 +570,23 @@ private Table instantiateHiveTable(ObjectPath tablePath, CatalogBaseTable table,
 
     hiveTable.setSd(sd);
 
-    properties.putAll(TableOptionProperties.translateFlinkTableProperties2Spark((ResolvedCatalogTable) table, hiveConf));
+    properties.putAll(TableOptionProperties.translateFlinkTableProperties2Spark((CatalogTable)table, hiveConf));
 
     //set pk
-    ResolvedSchema resolvedSchema = ((ResolvedCatalogTable) table).getResolvedSchema();
-    if (resolvedSchema.getPrimaryKey().isPresent()) {
-      String pkColumns = String.join(",", resolvedSchema.getPrimaryKey().get().getColumns());
+    if (table.getUnresolvedSchema().getPrimaryKey().isPresent()) {
+      String pkColumns = String.join(",", table.getUnresolvedSchema().getPrimaryKey().get().getColumnNames());
       String recordKey = properties.getOrDefault(FlinkOptions.RECORD_KEY_FIELD.key(), FlinkOptions.RECORD_KEY_FIELD.defaultValue());
       if (!Objects.equals(pkColumns, recordKey)) {
         throw new HoodieCatalogException(String.format("%s and %s are the different", pkColumns, recordKey));
       }
-      properties.put(PK_CONSTRAINT_NAME, resolvedSchema.getPrimaryKey().get().getName());
+      properties.put(PK_CONSTRAINT_NAME, table.getUnresolvedSchema().getPrimaryKey().get().getConstraintName());
       properties.put(PK_COLUMNS, pkColumns);
     }
 
+    if (!properties.containsKey(FlinkOptions.PATH.key())) {
+      properties.put(FlinkOptions.PATH.key(), location);
+    }
+
     hiveTable.setParameters(properties);
     return hiveTable;
   }
@@ -588,7 +610,7 @@ public List<String> listTables(String databaseName)
   @Override
   public List<String> listViews(String databaseName)
       throws DatabaseNotExistException, CatalogException {
-    throw new HoodieCatalogException("view not supported.");
+    throw new HoodieCatalogException("Hoodie catalog does not support to listViews");
   }
 
   @Override
@@ -684,7 +706,7 @@ public void alterTable(
       throw new HoodieCatalogException(String.format("The %s is not hoodie table", tablePath.getObjectName()));
     }
     if (newCatalogTable instanceof CatalogView) {
-      throw new HoodieCatalogException("view not supported.");
+      throw new HoodieCatalogException("Hoodie catalog does not support to ALTER VIEW");
     }
 
     try {

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/TableOptionProperties.java

@@ -30,6 +30,8 @@
 import org.apache.avro.Schema;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ResolvedCatalogTable;
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.table.types.DataType;
@@ -175,9 +177,8 @@ public static Map<String, String> getTableOptions(Map<String, String> options) {
     return copied;
   }
 
-  public static Map<String, String> translateFlinkTableProperties2Spark(ResolvedCatalogTable catalogTable, Configuration hadoopConf) {
-    ResolvedSchema resolvedSchema = catalogTable.getResolvedSchema();
-    Schema schema = AvroSchemaConverter.convertToSchema(resolvedSchema.toPhysicalRowDataType().getLogicalType());
+  public static Map<String, String> translateFlinkTableProperties2Spark(CatalogTable catalogTable, Configuration hadoopConf) {
+    Schema schema = AvroSchemaConverter.convertToSchema(catalogTable.getSchema().toPhysicalRowDataType().getLogicalType());
     MessageType messageType = TableSchemaResolver.convertAvroSchemaToParquetMessageType(schema, hadoopConf);
     String sparkVersion = catalogTable.getOptions().getOrDefault(SPARK_VERSION, DEFAULT_SPARK_VERSION);
     return Parquet2SparkSchemaUtils.getSparkTableProperties(catalogTable.getPartitionKeys(), sparkVersion, 4000, messageType);

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieCatalogTestUtils.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.catalog;
+
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+
+import static org.apache.hudi.table.catalog.HoodieCatalogFactoryOptions.HIVE_SITE_FILE;
+
+/** Test utils for Hoodie catalog. */
+public class HoodieCatalogTestUtils {
+    private static final String HIVE_WAREHOUSE_URI_FORMAT =
+            "jdbc:derby:;databaseName=%s;create=true";
+    private static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+    private static final String TEST_CATALOG_NAME = "test-catalog";
+
+    private static final byte[] SEPARATORS =
+            new byte[] {
+                (byte) 1, (byte) 2, (byte) 3, (byte) 4, (byte) 5, (byte) 6, (byte) 7, (byte) 8
+            };
+
+    /** Create a HiveCatalog with an embedded Hive Metastore. */
+    public static HoodieHiveCatalog createHiveCatalog() {
+        return createHiveCatalog(TEST_CATALOG_NAME);
+    }
+
+    public static HoodieHiveCatalog createHiveCatalog(String name) {
+        return new HoodieHiveCatalog(
+                name,
+                null,
+                createHiveConf(),
+                true);
+    }
+
+    public static HiveConf createHiveConf() {
+        ClassLoader classLoader = HoodieCatalogTestUtils.class.getClassLoader();
+
+        try {
+            TEMPORARY_FOLDER.create();
+            String warehouseDir = TEMPORARY_FOLDER.newFolder().getAbsolutePath() + "/metastore_db";
+            String warehouseUri = String.format(HIVE_WAREHOUSE_URI_FORMAT, warehouseDir);
+
+            HiveConf.setHiveSiteLocation(classLoader.getResource(HIVE_SITE_FILE));
+            HiveConf hiveConf = new HiveConf();
+            hiveConf.setVar(
+                    HiveConf.ConfVars.METASTOREWAREHOUSE,
+                    TEMPORARY_FOLDER.newFolder("hive_warehouse").getAbsolutePath());
+            hiveConf.setVar(HiveConf.ConfVars.METASTORECONNECTURLKEY, warehouseUri);
+            return hiveConf;
+        } catch (IOException e) {
+            throw new CatalogException("Failed to create test HiveConf to HiveCatalog.", e);
+        }
+    }
+}

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/HoodieHiveCatalogFactoryTest.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.catalog;
+
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.testutils.executor.TestExecutorResource;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.rules.TemporaryFolder;
+
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test cases for {@link HoodieHiveCatalogFactory}.
+ */
+public class HoodieHiveCatalogFactoryTest {
+  private static final URL CONF_DIR =
+      Thread.currentThread().getContextClassLoader().getResource("test-catalog-factory-conf");
+
+  @ClassRule
+  public static final TestExecutorResource<ExecutorService> EXECUTOR_RESOURCE =
+      new TestExecutorResource<>(() -> Executors.newFixedThreadPool(2));
+
+  @Rule
+  public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Test
+  public void testCreateHiveCatalog() {
+    final String catalogName = "mycatalog";
+
+    final HoodieHiveCatalog expectedCatalog = HoodieCatalogTestUtils.createHiveCatalog(catalogName);
+
+    final Map<String, String> options = new HashMap<>();
+    options.put(CommonCatalogOptions.CATALOG_TYPE.key(), HoodieCatalogFactoryOptions.IDENTIFIER);
+    options.put(HoodieCatalogFactoryOptions.HIVE_CONF_DIR.key(), CONF_DIR.getPath());
+
+    final Catalog actualCatalog =
+        FactoryUtil.createCatalog(
+            catalogName, options, null, Thread.currentThread().getContextClassLoader());
+
+    assertEquals(
+        ((HoodieHiveCatalog) actualCatalog)
+            .getHiveConf()
+            .getVar(HiveConf.ConfVars.METASTOREURIS), "dummy-hms");
+    checkEquals(expectedCatalog, (HoodieHiveCatalog) actualCatalog);
+  }
+
+  private static void checkEquals(HoodieHiveCatalog c1, HoodieHiveCatalog c2) {
+    // Only assert a few selected properties for now
+    assertEquals(c2.getName(), c1.getName());
+    assertEquals(c2.getDefaultDatabase(), c1.getDefaultDatabase());
+  }
+}