This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 00e86fafa8dc test(flink): improve catalog coverage (#19393)
00e86fafa8dc is described below
commit 00e86fafa8dc75e1b915c3fcf670d96e44eb1976
Author: Danny Chan <[email protected]>
AuthorDate: Wed Jul 29 17:35:03 2026 +0800
test(flink): improve catalog coverage (#19393)
---
.../hudi/table/catalog/TestHiveSchemaUtils.java | 214 +++++++++++++++
.../hudi/table/catalog/TestHoodieCatalog.java | 76 +++++
.../hudi/table/catalog/TestHoodieCatalogUtil.java | 133 +++++++++
.../hudi/table/catalog/TestHoodieHiveCatalog.java | 305 +++++++++++++++++++++
4 files changed, 728 insertions(+)
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHiveSchemaUtils.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHiveSchemaUtils.java
new file mode 100644
index 000000000000..653bcd0b0dfa
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHiveSchemaUtils.java
@@ -0,0 +1,214 @@
+/*
+ * 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.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.configuration.FlinkOptions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.NullType;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for the Flink and Hive schema conversions used by the Hive catalog.
+ */
+class TestHiveSchemaUtils {
+
+ @ParameterizedTest
+ @MethodSource("flinkToHiveTypes")
+ void testFlinkToHiveTypeConversion(DataType flinkType, String
expectedHiveType) {
+ assertEquals(
+ expectedHiveType,
+
HiveSchemaUtils.toHiveTypeInfo(flinkType.getLogicalType()).getTypeName());
+ }
+
+ private static Stream<Arguments> flinkToHiveTypes() {
+ return Stream.of(
+ Arguments.of(DataTypes.CHAR(4), "string"),
+ Arguments.of(DataTypes.VARCHAR(12), "string"),
+ Arguments.of(DataTypes.BOOLEAN(), "boolean"),
+ Arguments.of(DataTypes.BYTES(), "binary"),
+ Arguments.of(DataTypes.DECIMAL(12, 3), "decimal(12,3)"),
+ Arguments.of(DataTypes.TINYINT(), "int"),
+ Arguments.of(DataTypes.SMALLINT(), "int"),
+ Arguments.of(DataTypes.INT(), "int"),
+ Arguments.of(DataTypes.BIGINT(), "bigint"),
+ Arguments.of(DataTypes.FLOAT(), "float"),
+ Arguments.of(DataTypes.DOUBLE(), "double"),
+ Arguments.of(DataTypes.DATE(), "date"),
+ Arguments.of(DataTypes.TIMESTAMP(6), "timestamp"),
+ Arguments.of(DataTypes.TIMESTAMP(9), "bigint"),
+ Arguments.of(DataTypes.ARRAY(DataTypes.INT()), "array<int>"),
+ Arguments.of(
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.BIGINT())),
+ "map<string,array<bigint>>"),
+ Arguments.of(
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("attributes",
DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()))),
+ "struct<id:int,attributes:map<string,string>>"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("hiveToFlinkTypes")
+ void testHiveToFlinkTypeConversion(String hiveType, DataType
expectedFlinkType) {
+ assertEquals(
+ expectedFlinkType,
+
HiveSchemaUtils.toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString(hiveType)));
+ }
+
+ private static Stream<Arguments> hiveToFlinkTypes() {
+ return Stream.of(
+ Arguments.of("char(4)", DataTypes.CHAR(4)),
+ Arguments.of("varchar(12)", DataTypes.VARCHAR(12)),
+ Arguments.of("string", DataTypes.STRING()),
+ Arguments.of("boolean", DataTypes.BOOLEAN()),
+ Arguments.of("tinyint", DataTypes.TINYINT()),
+ Arguments.of("smallint", DataTypes.SMALLINT()),
+ Arguments.of("int", DataTypes.INT()),
+ Arguments.of("bigint", DataTypes.BIGINT()),
+ Arguments.of("float", DataTypes.FLOAT()),
+ Arguments.of("double", DataTypes.DOUBLE()),
+ Arguments.of("date", DataTypes.DATE()),
+ Arguments.of("timestamp", DataTypes.TIMESTAMP(6)),
+ Arguments.of("binary", DataTypes.BYTES()),
+ Arguments.of("decimal(12,3)", DataTypes.DECIMAL(12, 3)),
+ Arguments.of("array<int>", DataTypes.ARRAY(DataTypes.INT())),
+ Arguments.of(
+ "map<string,array<bigint>>",
+ DataTypes.MAP(DataTypes.STRING(),
DataTypes.ARRAY(DataTypes.BIGINT()))),
+ Arguments.of(
+ "struct<id:int,attributes:map<string,string>>",
+ DataTypes.ROW(
+ DataTypes.FIELD("id", DataTypes.INT()),
+ DataTypes.FIELD("attributes",
DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())))));
+ }
+
+ @Test
+ void testUnsupportedTypeConversions() {
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
HiveSchemaUtils.toHiveTypeInfo(DataTypes.VARBINARY(4).getLogicalType()));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
HiveSchemaUtils.toHiveTypeInfo(DataTypes.TIME().getLogicalType()));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () ->
HiveSchemaUtils.toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString("void")));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> HiveSchemaUtils.toFlinkType(
+ TypeInfoUtils.getTypeInfoFromTypeString("uniontype<int,string>")));
+ assertThrows(NullPointerException.class, () ->
HiveSchemaUtils.toFlinkType(null));
+ assertThrows(NullPointerException.class, () ->
HiveSchemaUtils.toHiveTypeInfo(null));
+
+ assertEquals("void", HiveSchemaUtils.toHiveTypeInfo(new
NullType()).getTypeName());
+ assertEquals(
+ "int",
+ DataTypes.INT().getLogicalType().accept(new
TypeInfoLogicalTypeVisitor(DataTypes.INT())).getTypeName());
+ }
+
+ @Test
+ void testHiveTableSchemaRoundTrip() {
+ StorageDescriptor storageDescriptor = new StorageDescriptor();
+ storageDescriptor.setCols(Arrays.asList(
+ new FieldSchema(HoodieRecord.COMMIT_TIME_METADATA_FIELD, "string",
null),
+ new FieldSchema("id", "int", null),
+ new FieldSchema("payload", "struct<name:string,scores:array<int>>",
null)));
+
+ Table hiveTable = new Table();
+ hiveTable.setSd(storageDescriptor);
+ hiveTable.setPartitionKeys(Collections.singletonList(new
FieldSchema("part", "string", null)));
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(FlinkOptions.RECORD_KEY_FIELD.key(), "id");
+ parameters.put(TableOptionProperties.PK_CONSTRAINT_NAME, "pk_hms");
+ parameters.put(TableOptionProperties.METADATA_COLUMNS,
HoodieRecord.COMMIT_TIME_METADATA_FIELD);
+ hiveTable.setParameters(parameters);
+
+ Schema flinkSchema = HiveSchemaUtils.convertTableSchema(hiveTable);
+ assertEquals(
+ Arrays.asList("id", "payload", "part",
HoodieRecord.COMMIT_TIME_METADATA_FIELD),
+ flinkSchema.getColumns().stream()
+ .map(Schema.UnresolvedColumn::getName)
+ .collect(Collectors.toList()));
+ assertEquals(
+ Collections.singletonList("id"),
+ flinkSchema.getPrimaryKey().get().getColumnNames());
+ assertEquals("pk_hms",
flinkSchema.getPrimaryKey().get().getConstraintName());
+ assertTrue(flinkSchema.getColumns().get(0).toString().contains("NOT
NULL"));
+ assertTrue(
+ flinkSchema.getColumns().get(3) instanceof
Schema.UnresolvedMetadataColumn);
+
+ List<FieldSchema> roundTrip =
HiveSchemaUtils.toHiveFieldSchema(flinkSchema, true);
+ assertEquals(
+ HoodieRecord.HOODIE_META_COLUMNS.size() + 1 + 3,
+ roundTrip.size());
+ assertTrue(
+ HiveSchemaUtils.getFieldNames(roundTrip)
+ .contains(HoodieRecord.OPERATION_METADATA_FIELD));
+ assertEquals("int", fieldType(roundTrip, "id"));
+ assertEquals("struct<name:string,scores:array<int>>", fieldType(roundTrip,
"payload"));
+ }
+
+ @Test
+ void testSplitSchemaByPartitionKeys() {
+ List<FieldSchema> fields = Arrays.asList(
+ new FieldSchema("id", "int", null),
+ new FieldSchema("region", "string", null),
+ new FieldSchema("day", "date", null));
+
+ Pair<List<FieldSchema>, List<FieldSchema>> split =
+ HiveSchemaUtils.splitSchemaByPartitionKeys(fields,
Arrays.asList("day", "region"));
+
+ assertEquals(Collections.singletonList("id"),
HiveSchemaUtils.getFieldNames(split.getLeft()));
+ assertEquals(Arrays.asList("region", "day"),
HiveSchemaUtils.getFieldNames(split.getRight()));
+ assertFalse(split.getLeft().isEmpty());
+ }
+
+ private static String fieldType(List<FieldSchema> fields, String name) {
+ return fields.stream()
+ .filter(field -> field.getName().equals(name))
+ .findFirst()
+ .orElseThrow(AssertionError::new)
+ .getType();
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
index 2edc2f1255a5..8121fcc45df1 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalog.java
@@ -70,9 +70,12 @@ import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.hadoop.fs.FileSystem;
@@ -100,6 +103,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -585,6 +589,78 @@ public class TestHoodieCatalog extends
BaseTestHoodieCatalog {
assertFalse(catalog.partitionExists(tablePath, partitionSpec));
}
+ @Test
+ void testUnsupportedCatalogOperationsAndDefaults() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "missing");
+ ObjectPath functionPath = new ObjectPath(TEST_DEFAULT_DATABASE,
"function");
+ CatalogPartitionSpec partitionSpec =
+ new CatalogPartitionSpec(Collections.singletonMap("partition",
"20260728"));
+ CatalogDatabase database = new CatalogDatabaseImpl(Collections.emptyMap(),
null);
+
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterDatabase(TEST_DEFAULT_DATABASE, database, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.renameTable(tablePath, "renamed", false));
+ assertEquals(Collections.emptyList(),
catalog.listViews(TEST_DEFAULT_DATABASE));
+ assertEquals(Collections.emptyList(), catalog.listPartitions(tablePath));
+ assertEquals(Collections.emptyList(), catalog.listPartitions(tablePath,
partitionSpec));
+ assertEquals(
+ Collections.emptyList(),
+ catalog.listPartitionsByFilter(tablePath, Collections.emptyList()));
+ assertThrows(
+ PartitionNotExistException.class,
+ () -> catalog.getPartition(tablePath, partitionSpec));
+ assertFalse(catalog.partitionExists(tablePath, partitionSpec));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.createPartition(tablePath, partitionSpec, null, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterPartition(tablePath, partitionSpec, null, false));
+
+ assertEquals(Collections.emptyList(),
catalog.listFunctions(TEST_DEFAULT_DATABASE));
+ assertThrows(FunctionNotExistException.class, () ->
catalog.getFunction(functionPath));
+ assertFalse(catalog.functionExists(functionPath));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.createFunction(functionPath, null, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterFunction(functionPath, null, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.dropFunction(functionPath, false));
+
+ assertSame(CatalogTableStatistics.UNKNOWN,
catalog.getTableStatistics(tablePath));
+ assertSame(
+ CatalogColumnStatistics.UNKNOWN,
+ catalog.getTableColumnStatistics(tablePath));
+ assertSame(
+ CatalogTableStatistics.UNKNOWN,
+ catalog.getPartitionStatistics(tablePath, partitionSpec));
+ assertSame(
+ CatalogColumnStatistics.UNKNOWN,
+ catalog.getPartitionColumnStatistics(tablePath, partitionSpec));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterTableStatistics(
+ tablePath, CatalogTableStatistics.UNKNOWN, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterTableColumnStatistics(
+ tablePath, CatalogColumnStatistics.UNKNOWN, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterPartitionStatistics(
+ tablePath, partitionSpec, CatalogTableStatistics.UNKNOWN, false));
+ assertThrows(
+ UnsupportedOperationException.class,
+ () -> catalog.alterPartitionColumnStatistics(
+ tablePath, partitionSpec, CatalogColumnStatistics.UNKNOWN, false));
+ }
+
@Override
AbstractCatalog getCatalog() {
return catalog;
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalogUtil.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalogUtil.java
new file mode 100644
index 000000000000..9712dba61743
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalogUtil.java
@@ -0,0 +1,133 @@
+/*
+ * 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.hudi.configuration.FlinkOptions;
+import org.apache.hudi.utils.CatalogUtils;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link HoodieCatalogUtil}.
+ */
+class TestHoodieCatalogUtil {
+
+ @Test
+ void testPartitionKeyExtractionAndPathInference() {
+ Schema schema = Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("region", DataTypes.STRING())
+ .column("day", DataTypes.STRING())
+ .build();
+ Map<String, String> options = Collections.singletonMap(
+ FlinkOptions.PARTITION_PATH_FIELD.key(), "region,day");
+ CatalogTable optionPartitionedTable =
+ CatalogUtils.createCatalogTable(schema, Collections.emptyList(),
options, null);
+ CatalogTable declaredPartitionedTable =
+ CatalogUtils.createCatalogTable(schema,
Collections.singletonList("day"), options, null);
+
+ assertEquals(
+ Arrays.asList("region", "day"),
+ HoodieCatalogUtil.getPartitionKeys(optionPartitionedTable));
+ assertEquals(
+ Collections.singletonList("day"),
+ HoodieCatalogUtil.getPartitionKeys(declaredPartitionedTable));
+
+ Map<String, String> spec = new LinkedHashMap<>();
+ spec.put("region", "apac");
+ spec.put("day", "2026-07-28");
+ CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(spec);
+ assertEquals("region=apac/day=2026-07-28",
HoodieCatalogUtil.inferPartitionPath(true, partitionSpec));
+ assertEquals("apac/2026-07-28",
HoodieCatalogUtil.inferPartitionPath(false, partitionSpec));
+ }
+
+ @Test
+ void testOrderedPartitionValuesAndValidation() throws Exception {
+ HiveConf hiveConf = HoodieCatalogTestUtils.createHiveConf();
+ ObjectPath tablePath = new ObjectPath("default", "tbl");
+ Map<String, String> spec = new LinkedHashMap<>();
+ spec.put("day", null);
+ spec.put("region", "apac");
+
+ assertEquals(
+ Arrays.asList("apac",
hiveConf.getVar(HiveConf.ConfVars.DEFAULTPARTITIONNAME)),
+ HoodieCatalogUtil.getOrderedPartitionValues(
+ "catalog",
+ hiveConf,
+ new CatalogPartitionSpec(spec),
+ Arrays.asList("region", "day"),
+ tablePath));
+
+ assertThrows(
+ PartitionSpecInvalidException.class,
+ () -> HoodieCatalogUtil.getOrderedPartitionValues(
+ "catalog",
+ hiveConf,
+ new CatalogPartitionSpec(Collections.singletonMap("region",
"apac")),
+ Arrays.asList("region", "day"),
+ tablePath));
+
+ Map<String, String> wrongKey = new LinkedHashMap<>();
+ wrongKey.put("region", "apac");
+ wrongKey.put("month", "07");
+ assertThrows(
+ PartitionSpecInvalidException.class,
+ () -> HoodieCatalogUtil.getOrderedPartitionValues(
+ "catalog",
+ hiveConf,
+ new CatalogPartitionSpec(wrongKey),
+ Arrays.asList("region", "day"),
+ tablePath));
+ }
+
+ @Test
+ void testHiveConfLoadingFailureAndEmbeddedDetection() {
+ CatalogException exception = assertThrows(
+ CatalogException.class,
+ () -> HoodieCatalogUtil.createHiveConf(
+ "target/does-not-exist-" + System.nanoTime(),
+ new Configuration()));
+ assertTrue(exception.getMessage().contains("Failed to load
hive-site.xml"));
+
+ HiveConf embedded = new HiveConf();
+ embedded.setVar(HiveConf.ConfVars.METASTOREURIS, "");
+ assertTrue(HoodieCatalogUtil.isEmbeddedMetastore(embedded));
+
+ assertNotNull(HoodieCatalogUtil.createHiveConf(null, new Configuration()));
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
index 48ce7e26c203..91a598f00d99 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.catalog;
+import org.apache.hudi.adapter.HiveCatalogConstants.AlterHiveDatabaseOp;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
@@ -47,17 +48,24 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.AbstractCatalog;
import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
import org.apache.flink.table.catalog.CatalogPartitionSpec;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.junit.jupiter.api.AfterAll;
@@ -78,6 +86,12 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
+import static org.apache.hudi.adapter.HiveCatalogConstants.ALTER_DATABASE_OP;
+import static
org.apache.hudi.adapter.HiveCatalogConstants.DATABASE_LOCATION_URI;
+import static org.apache.hudi.adapter.HiveCatalogConstants.DATABASE_OWNER_NAME;
+import static org.apache.hudi.adapter.HiveCatalogConstants.DATABASE_OWNER_TYPE;
+import static org.apache.hudi.adapter.HiveCatalogConstants.ROLE_OWNER;
+import static org.apache.hudi.adapter.HiveCatalogConstants.USER_OWNER;
import static org.apache.hudi.configuration.FlinkOptions.ORDERING_FIELDS;
import static
org.apache.hudi.keygen.constant.KeyGeneratorOptions.RECORDKEY_FIELD_NAME;
import static
org.apache.hudi.table.catalog.HoodieCatalogTestUtils.createStorageConf;
@@ -89,6 +103,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -149,6 +164,296 @@ public class TestHoodieHiveCatalog extends
BaseTestHoodieCatalog {
}
}
+ @Test
+ void testDatabaseLifecycle() throws Exception {
+ String databaseName = "catalog_database_lifecycle";
+ hoodieCatalog.dropDatabase(databaseName, true, true);
+
+ try {
+ CatalogDatabase database =
+ new CatalogDatabaseImpl(new HashMap<>(), "catalog api database");
+ hoodieCatalog.createDatabase(databaseName, database, false);
+
+ assertTrue(hoodieCatalog.databaseExists(databaseName));
+ assertTrue(hoodieCatalog.listDatabases().contains(databaseName));
+ CatalogDatabase storedDatabase = hoodieCatalog.getDatabase(databaseName);
+ assertEquals("catalog api database", storedDatabase.getComment());
+ assertNotNull(storedDatabase.getProperties().get(DATABASE_LOCATION_URI));
+ assertThrows(
+ DatabaseAlreadyExistException.class,
+ () -> hoodieCatalog.createDatabase(databaseName, database, false));
+ hoodieCatalog.createDatabase(databaseName, database, true);
+
+ Map<String, String> changedProperties = new HashMap<>();
+ changedProperties.put("purpose", "coverage");
+ changedProperties.put("is_generic", "true");
+ hoodieCatalog.alterDatabase(
+ databaseName,
+ new CatalogDatabaseImpl(changedProperties, null),
+ false);
+ assertEquals(
+ "coverage",
+
hoodieCatalog.getDatabase(databaseName).getProperties().get("purpose"));
+ assertFalse(
+
hoodieCatalog.getDatabase(databaseName).getProperties().containsKey("is_generic"));
+
+ // Hive 2.x ObjectStore persists parameter and owner changes, but not
location changes.
+ Map<String, String> userOwner = new HashMap<>();
+ userOwner.put(ALTER_DATABASE_OP,
AlterHiveDatabaseOp.CHANGE_OWNER.name());
+ userOwner.put(DATABASE_OWNER_NAME, "catalog-user");
+ userOwner.put(DATABASE_OWNER_TYPE, USER_OWNER);
+ hoodieCatalog.alterDatabase(
+ databaseName,
+ new CatalogDatabaseImpl(userOwner, null),
+ false);
+ assertEquals(
+ PrincipalType.USER,
+ hoodieCatalog.getClient().getDatabase(databaseName).getOwnerType());
+ assertEquals(
+ "catalog-user",
+ hoodieCatalog.getClient().getDatabase(databaseName).getOwnerName());
+
+ Map<String, String> roleOwner = new HashMap<>();
+ roleOwner.put(ALTER_DATABASE_OP,
AlterHiveDatabaseOp.CHANGE_OWNER.name());
+ roleOwner.put(DATABASE_OWNER_NAME, "catalog-role");
+ roleOwner.put(DATABASE_OWNER_TYPE, ROLE_OWNER);
+ hoodieCatalog.alterDatabase(
+ databaseName,
+ new CatalogDatabaseImpl(roleOwner, null),
+ false);
+ assertEquals(
+ PrincipalType.ROLE,
+ hoodieCatalog.getClient().getDatabase(databaseName).getOwnerType());
+ assertEquals(
+ "catalog-role",
+ hoodieCatalog.getClient().getDatabase(databaseName).getOwnerName());
+
+ Map<String, String> invalidOwner = new HashMap<>();
+ invalidOwner.put(ALTER_DATABASE_OP,
AlterHiveDatabaseOp.CHANGE_OWNER.name());
+ invalidOwner.put(DATABASE_OWNER_NAME, "catalog-group");
+ invalidOwner.put(DATABASE_OWNER_TYPE, "GROUP");
+ assertThrows(
+ org.apache.flink.table.catalog.exceptions.CatalogException.class,
+ () -> hoodieCatalog.alterDatabase(
+ databaseName,
+ new CatalogDatabaseImpl(invalidOwner, null),
+ false));
+
+ hoodieCatalog.alterDatabase(
+ "missing_catalog_api_db",
+ new CatalogDatabaseImpl(Collections.emptyMap(), null),
+ true);
+ assertThrows(
+ DatabaseNotExistException.class,
+ () -> hoodieCatalog.alterDatabase(
+ "missing_catalog_api_db",
+ new CatalogDatabaseImpl(Collections.emptyMap(), null),
+ false));
+
+ hoodieCatalog.dropDatabase(databaseName, false, false);
+ assertFalse(hoodieCatalog.databaseExists(databaseName));
+ assertThrows(
+ DatabaseNotExistException.class,
+ () -> hoodieCatalog.dropDatabase(databaseName, false, false));
+ hoodieCatalog.dropDatabase(databaseName, true, false);
+ } finally {
+ hoodieCatalog.dropDatabase(databaseName, true, true);
+ }
+ }
+
+ @Test
+ void testTableLifecycleAgainstMetastore() throws Exception {
+ String databaseName = "catalog_table_lifecycle";
+ ObjectPath databaseTablePath = new ObjectPath(databaseName,
"catalog_api_table");
+ createCatalogDatabase(databaseName);
+
+ try {
+ CatalogTable catalogTable = createCatalogTable("stored in hms");
+ hoodieCatalog.createTable(databaseTablePath, catalogTable, false);
+ assertThrows(
+ TableAlreadyExistException.class,
+ () -> hoodieCatalog.createTable(databaseTablePath, catalogTable,
false));
+ hoodieCatalog.createTable(databaseTablePath, catalogTable, true);
+
+
assertTrue(hoodieCatalog.listTables(databaseName).contains(databaseTablePath.getObjectName()));
+ assertTrue(hoodieCatalog.tableExists(databaseTablePath));
+ Table hiveTable = hoodieCatalog.getHiveTable(databaseTablePath);
+ assertEquals("hudi", hiveTable.getParameters().get(CONNECTOR.key()));
+ assertEquals("stored in hms",
hiveTable.getParameters().get(TableOptionProperties.COMMENT));
+ assertEquals(
+
"uuid:int,name:string,age:int,infos:array<string>,ts_3:timestamp,ts_6:timestamp",
+ hiveTable.getSd().getCols().stream()
+ .filter(field -> !field.getName().startsWith("_hoodie_"))
+ .map(field -> field.getName() + ":" + field.getType())
+ .collect(Collectors.joining(",")));
+ assertEquals(
+ schema.getColumns().stream()
+ .map(Schema.UnresolvedColumn::getName)
+ .collect(Collectors.toList()),
+
hoodieCatalog.getTable(databaseTablePath).getUnresolvedSchema().getColumns().stream()
+ .map(Schema.UnresolvedColumn::getName)
+ .collect(Collectors.toList()));
+
+ assertThrows(
+ DatabaseNotEmptyException.class,
+ () -> hoodieCatalog.dropDatabase(databaseName, false, false));
+ hoodieCatalog.dropTable(databaseTablePath, false);
+ assertFalse(hoodieCatalog.tableExists(databaseTablePath));
+ assertThrows(
+ TableNotExistException.class,
+ () -> hoodieCatalog.dropTable(databaseTablePath, false));
+ } finally {
+ dropCatalogObjects(databaseName, databaseTablePath);
+ }
+ }
+
+ @Test
+ void testHiveSyncOptionsInjected() throws Exception {
+ String databaseName = "catalog_hive_sync_options";
+ ObjectPath databaseTablePath = new ObjectPath(databaseName,
"catalog_api_table");
+ createCatalogDatabase(databaseName);
+
+ try {
+ hoodieCatalog.createTable(databaseTablePath, createCatalogTable("hive
sync options"), false);
+
+ String metastoreUris = hoodieCatalog.getHiveConf().getVar(
+ org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS);
+ String remoteMetastoreUri = "thrift://localhost:9083";
+ try {
+ // Simulate a remote catalog so getTable must propagate its endpoint
to Hive sync.
+ hoodieCatalog.getHiveConf().setVar(
+ org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS,
+ remoteMetastoreUri);
+ Map<String, String> supplementedOptions =
+ hoodieCatalog.getTable(databaseTablePath).getOptions();
+ assertEquals("true",
supplementedOptions.get(FlinkOptions.HIVE_SYNC_ENABLED.key()));
+ assertEquals("hms",
supplementedOptions.get(FlinkOptions.HIVE_SYNC_MODE.key()));
+ assertEquals(databaseName,
supplementedOptions.get(FlinkOptions.HIVE_SYNC_DB.key()));
+ assertEquals(
+ databaseTablePath.getObjectName(),
+ supplementedOptions.get(FlinkOptions.HIVE_SYNC_TABLE.key()));
+ assertEquals(
+ remoteMetastoreUri,
+
supplementedOptions.get(FlinkOptions.HIVE_SYNC_METASTORE_URIS.key()));
+ } finally {
+ hoodieCatalog.getHiveConf().setVar(
+ org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS,
+ metastoreUris);
+ }
+ } finally {
+ dropCatalogObjects(databaseName, databaseTablePath);
+ }
+ }
+
+ private CatalogTable createCatalogTable(String comment) {
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put(CONNECTOR.key(), "hudi");
+ return CatalogUtils.createCatalogTable(schema, partitions, tableOptions,
comment);
+ }
+
+ private static void createCatalogDatabase(String databaseName) throws
Exception {
+ hoodieCatalog.dropDatabase(databaseName, true, true);
+ hoodieCatalog.createDatabase(
+ databaseName,
+ new CatalogDatabaseImpl(Collections.emptyMap(), "catalog api
database"),
+ false);
+ }
+
+ private static void dropCatalogObjects(
+ String databaseName, ObjectPath databaseTablePath) throws Exception {
+ if (hoodieCatalog.tableExists(databaseTablePath)) {
+ hoodieCatalog.dropTable(databaseTablePath, true);
+ }
+ hoodieCatalog.dropDatabase(databaseName, true, true);
+ }
+
+ @Test
+ void testUnsupportedCatalogOperationsAndDefaults() throws Exception {
+ CatalogPartitionSpec partitionSpec =
+ new CatalogPartitionSpec(Collections.singletonMap("par1", "20260728"));
+ ObjectPath functionPath = new ObjectPath(TEST_DEFAULT_DATABASE,
"function");
+ ObjectPath missingTablePath = new ObjectPath("missing_database",
"missing_table");
+
+ assertThrows(HoodieCatalogException.class, () ->
hoodieCatalog.listViews(TEST_DEFAULT_DATABASE));
+ assertEquals(
+ Collections.emptyList(),
+ hoodieCatalog.listTables(missingTablePath.getDatabaseName()));
+ assertFalse(hoodieCatalog.tableExists(missingTablePath));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.renameTable(missingTablePath, "renamed", false));
+ assertEquals(Collections.emptyList(),
hoodieCatalog.listPartitions(tablePath));
+ assertEquals(Collections.emptyList(),
hoodieCatalog.listPartitions(tablePath, partitionSpec));
+ assertEquals(
+ Collections.emptyList(),
+ hoodieCatalog.listPartitionsByFilter(tablePath,
Collections.emptyList()));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.getPartition(tablePath, partitionSpec));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.partitionExists(tablePath, partitionSpec));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.createPartition(tablePath, partitionSpec, null,
false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterPartition(tablePath, partitionSpec, null,
false));
+ assertThrows(
+ PartitionNotExistException.class,
+ () -> hoodieCatalog.dropPartition(missingTablePath, partitionSpec,
false));
+ hoodieCatalog.dropPartition(missingTablePath, partitionSpec, true);
+
+ Map<String, String> tableOptions =
Collections.singletonMap(CONNECTOR.key(), "hudi");
+ CatalogTable table =
+ CatalogUtils.createCatalogTable(schema, partitions, tableOptions,
"missing database");
+ assertThrows(
+ DatabaseNotExistException.class,
+ () -> hoodieCatalog.createTable(missingTablePath, table, false));
+
+ assertEquals(Collections.emptyList(),
hoodieCatalog.listFunctions(TEST_DEFAULT_DATABASE));
+ assertThrows(
+
org.apache.flink.table.catalog.exceptions.FunctionNotExistException.class,
+ () -> hoodieCatalog.getFunction(functionPath));
+ assertFalse(hoodieCatalog.functionExists(functionPath));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.createFunction(functionPath, null, false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterFunction(functionPath, null, false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.dropFunction(functionPath, false));
+
+ assertSame(CatalogTableStatistics.UNKNOWN,
hoodieCatalog.getTableStatistics(tablePath));
+ assertSame(
+ CatalogColumnStatistics.UNKNOWN,
+ hoodieCatalog.getTableColumnStatistics(tablePath));
+ assertSame(
+ CatalogTableStatistics.UNKNOWN,
+ hoodieCatalog.getPartitionStatistics(tablePath, partitionSpec));
+ assertSame(
+ CatalogColumnStatistics.UNKNOWN,
+ hoodieCatalog.getPartitionColumnStatistics(tablePath, partitionSpec));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterTableStatistics(
+ tablePath, CatalogTableStatistics.UNKNOWN, false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterTableColumnStatistics(
+ tablePath, CatalogColumnStatistics.UNKNOWN, false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterPartitionStatistics(
+ tablePath, partitionSpec, CatalogTableStatistics.UNKNOWN, false));
+ assertThrows(
+ HoodieCatalogException.class,
+ () -> hoodieCatalog.alterPartitionColumnStatistics(
+ tablePath, partitionSpec, CatalogColumnStatistics.UNKNOWN, false));
+ }
+
@ParameterizedTest
@EnumSource(value = HoodieTableType.class)
public void testCreateAndGetHoodieTable(HoodieTableType tableType) throws
Exception {
