XuQianJin-Stars commented on code in PR #3395: URL: https://github.com/apache/fluss/pull/3395#discussion_r3330153412
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java: ########## @@ -0,0 +1,260 @@ +/* + * 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.fluss.lake.hudi; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiConversions; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.metadata.TableChange; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.types.AbstractDataType; +import org.apache.flink.table.types.DataType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Objects; + +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HUDI_CATALOG_DEFAULT_NAME; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.MODE_CONFIG; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** Implementation of {@link LakeCatalog} for Hudi. */ +public class HudiLakeCatalog implements LakeCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCatalog.class); + + public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new LinkedHashMap<>(); + + static { + SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); + SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); + SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP(6)); + } + + private final Catalog hudiCatalog; + private final String catalogMode; + + public HudiLakeCatalog(Configuration configuration) { + this.catalogMode = configuration.toMap().getOrDefault(MODE_CONFIG, HIVE_META_STORE_TYPE); + this.hudiCatalog = HudiCatalogUtils.createHudiCatalog(configuration); + this.hudiCatalog.open(); + } + + @VisibleForTesting + protected Catalog getHudiCatalog() { + return hudiCatalog; + } + + @Override + public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Context context) + throws org.apache.fluss.exception.TableAlreadyExistException { + LOG.info("create the lake table for : {} with props: {}", tablePath, tableDescriptor); + + ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath); + + boolean isPkTable = tableDescriptor.getSchema().getPrimaryKeyIndexes().length > 0; + + // Create Hudi catalog table + CatalogTable catalogTable = + HudiConversions.createHudiCatalogTable(tableDescriptor, isPkTable, catalogMode); + + // Create table in Hudi catalog + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException e) { + createDatabase(tablePath.getDatabaseName()); + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException t) { + // shouldn't happen in normal cases + throw new RuntimeException( + String.format( + "Fail to create table %s in Hudi, because " + + "Database %s still doesn't exist although create database " + + "successfully, please try again.", + tablePath, tablePath.getDatabaseName())); + } + } + } + + @Override + public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Context context) + throws org.apache.fluss.exception.TableNotExistException { + throw new UnsupportedOperationException( + "Alter table is not supported for Hudi at the moment"); + } + + private void createTable( + ObjectPath tablePath, CatalogBaseTable catalogTable, boolean isCreatingFlussTable) + throws DatabaseNotExistException { + try { + hudiCatalog.createTable(tablePath, catalogTable, false); + LOG.info("Table {} created successfully.", tablePath); + } catch (org.apache.flink.table.catalog.exceptions.TableAlreadyExistException e) { + // table already exists, check schema compatibility for idempotency + try { + CatalogBaseTable existingTable = hudiCatalog.getTable(tablePath); + if (!isHudiSchemaCompatible(existingTable, catalogTable)) { + throw new org.apache.fluss.exception.TableAlreadyExistException( + String.format( + "The table %s already exists in Hudi catalog, but the table schema is not compatible. " + + "Please first drop the table in Hudi catalog or use a new table name.", + tablePath), + e); + } + // if creating a new fluss table, we should ensure the lake table is empty + // TODO: add emptiness check for Hudi table once LakeTieringFactory is implemented + if (isCreatingFlussTable) { Review Comment: Compare with `PaimonLakeCatalog.createTable` (line 180) which directly calls `checkTableIsEmpty(...)`. Without the emptiness check, if a user attaches a new Fluss table to a Hudi table that **already contains data**, data corruption/mixing will occur. Even though TieringFactory is not yet ready, at the very least the semantics should be documented (in a comment or in the thrown exception), and consider throwing `TableAlreadyExistException` here as a safety fallback (safer), then loosening it once TieringFactory is ready. **At minimum, do not silently warn and return success.** ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/catalog/HudiCatalogUtils.java: ########## @@ -0,0 +1,106 @@ +/* + * 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.fluss.lake.hudi.utils.catalog; + +import org.apache.fluss.config.Configuration; + +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.hudi.exception.HoodieCatalogException; +import org.apache.hudi.table.catalog.CatalogOptions; +import org.apache.hudi.table.catalog.HoodieCatalog; +import org.apache.hudi.table.catalog.HoodieHiveCatalog; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Locale; +import java.util.Map; + +/** Implementation of {@link HudiCatalogUtils} for Hudi. */ +public class HudiCatalogUtils { + + private static final Logger LOG = LoggerFactory.getLogger(HudiCatalogUtils.class); + + public static final String MODE_CONFIG = "mode"; + public static final String CATALOG_NAME_CONFIG = "name"; + + public static final String HUDI_CATALOG_DEFAULT_NAME = "fluss-hudi-catalog"; + public static final String HIVE_META_STORE_TYPE = "hms"; + public static final String FILE_SYSTEM_TYPE = "dfs"; + + public static Catalog createHudiCatalog(Configuration configuration) { + Map<String, String> hudiProps = configuration.toMap(); Review Comment: `createHudiCatalog` uses the custom `MODE_CONFIG = "mode"` read in `HudiLakeCatalog`, while `buildHudiCatalog` internally uses `CatalogOptions.MODE.key()` (also `"mode"`) — same string by coincidence, but maintaining two sets of constants invites future drift. Standardize on `CatalogOptions.MODE.key()` as the authoritative key and drop the `MODE_CONFIG` constant to avoid long-term maintenance drift. Same for `CATALOG_NAME_CONFIG = "name"` — recommend using a public Hudi constant. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java: ########## @@ -0,0 +1,260 @@ +/* + * 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.fluss.lake.hudi; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiConversions; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.metadata.TableChange; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.types.AbstractDataType; +import org.apache.flink.table.types.DataType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Objects; + +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HUDI_CATALOG_DEFAULT_NAME; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.MODE_CONFIG; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** Implementation of {@link LakeCatalog} for Hudi. */ +public class HudiLakeCatalog implements LakeCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCatalog.class); + + public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new LinkedHashMap<>(); + + static { + SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); + SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); + SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP(6)); + } + + private final Catalog hudiCatalog; + private final String catalogMode; + + public HudiLakeCatalog(Configuration configuration) { + this.catalogMode = configuration.toMap().getOrDefault(MODE_CONFIG, HIVE_META_STORE_TYPE); + this.hudiCatalog = HudiCatalogUtils.createHudiCatalog(configuration); + this.hudiCatalog.open(); + } + + @VisibleForTesting + protected Catalog getHudiCatalog() { + return hudiCatalog; + } + + @Override + public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Context context) + throws org.apache.fluss.exception.TableAlreadyExistException { + LOG.info("create the lake table for : {} with props: {}", tablePath, tableDescriptor); + + ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath); + + boolean isPkTable = tableDescriptor.getSchema().getPrimaryKeyIndexes().length > 0; + + // Create Hudi catalog table + CatalogTable catalogTable = + HudiConversions.createHudiCatalogTable(tableDescriptor, isPkTable, catalogMode); + + // Create table in Hudi catalog + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException e) { Review Comment: The inner `createTable` throws `org.apache.fluss.exception.TableAlreadyExistException` (unchecked) on schema incompatibility, which is fine. The outer method signature declares `throws org.apache.fluss.exception.TableAlreadyExistException` — looks like it intends to comply with the `LakeCatalog.createTable` contract — but the exception **is not actually thrown as checked via the `throws` clause** because it's a `RuntimeException` subclass. OK, that's not really a problem. However: **The real bug is in the test `testCreateDuplicateTableWithIncompatibleSchema`**: based on the implementation of `isHudiSchemaCompatible(existingTable, catalogTable)` (with the `ResolvedCatalogBaseTable` vs `Schema.UnresolvedColumn` branches), the `catalogTable` passed into `hudiCatalog.createTable()` is a `CatalogTable.of(...)` (**unresolved**), while `hudiCatalog.getTable()` may return a `ResolvedCatalogBaseTable`. `extractColumns` will go through two different code paths (one uses `column.getDataType()` returning `DataType`, the other uses `UnresolvedColumn` returning `AbstractDataType<?>`). Comparison between `AbstractDataType` and `DataType` via `equals` is not necessarily symmetric — even with identical column names/types it may be judged as "incompatible", and conversely different schemas may fail comparison because the `AbstractDataType` is not fully resolved. **Suggestion**: First normalize both `existingTable` and `expectedTable` into the same representation (recommended: compare using `org.apache.flink.table.types.logical.LogicalType`, or convert via `getUnresolvedSchema()` to `Schema` then compare on column name + `AbstractDataType`). The dataType comparison should use the `LogicalType.equals` dimension (stripping the conversion class). Paimon's approach with `isPaimonSchemaCompatible` — directly comparing `name+type+nullable` on `Paimon Schema` fields — is more robust. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/FlussDataTypeToHudiDataType.java: ########## @@ -0,0 +1,180 @@ +/* + * 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.fluss.lake.hudi; + +import org.apache.fluss.types.ArrayType; +import org.apache.fluss.types.BigIntType; +import org.apache.fluss.types.BinaryType; +import org.apache.fluss.types.BooleanType; +import org.apache.fluss.types.BytesType; +import org.apache.fluss.types.CharType; +import org.apache.fluss.types.DataTypeVisitor; +import org.apache.fluss.types.DateType; +import org.apache.fluss.types.DecimalType; +import org.apache.fluss.types.DoubleType; +import org.apache.fluss.types.FloatType; +import org.apache.fluss.types.IntType; +import org.apache.fluss.types.LocalZonedTimestampType; +import org.apache.fluss.types.MapType; +import org.apache.fluss.types.RowType; +import org.apache.fluss.types.SmallIntType; +import org.apache.fluss.types.StringType; +import org.apache.fluss.types.TimeType; +import org.apache.fluss.types.TimestampType; +import org.apache.fluss.types.TinyIntType; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.types.DataType; + +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.FILE_SYSTEM_TYPE; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; + +/** Convert from Fluss's data type to Hudi's internal data type. */ +public class FlussDataTypeToHudiDataType implements DataTypeVisitor<DataType> { + + private String catalogMode; Review Comment: - The `catalogMode` field is not declared `final`, yet the class exposes a public constructor and provides two **singleton INSTANCEs**. If any caller obtains `DFS_INSTANCE` and accidentally invokes a future mutator, concurrency issues arise. Please change the field to `private final`, and either remove the public constructor or change it to `private` (external code uses the two INSTANCEs uniformly). - The class-level Javadoc lacks the `@ThreadSafe` annotation; per spec (AGENTS.md §5), multiple threads will share the same INSTANCE, so this should be explicitly marked. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java: ########## @@ -0,0 +1,271 @@ +/* + * 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.fluss.lake.hudi.utils; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.exception.InvalidConfigException; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.lake.hudi.FlussDataTypeToHudiDataType; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.index.HoodieIndex; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; + +/** Utils for conversion between Hudi and Fluss. */ +public class HudiConversions { + + private static final Logger LOG = LoggerFactory.getLogger(HudiConversions.class); + + // for fluss config + private static final String FLUSS_CONF_PREFIX = "fluss."; + // for hudi config + private static final String HUDI_CONF_PREFIX = "hudi."; + + private static final String DELIMITER = ","; + private static final String HUDI_TABLE_TYPE_KEY = "hoodie.datasource.write.table.type"; + + /** Hudi config options set by Fluss should not be set by users. */ + @VisibleForTesting public static final Set<String> HUDI_UNSETTABLE_OPTIONS = new HashSet<>(); + + static { + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.TABLE_TYPE.key()); + HUDI_UNSETTABLE_OPTIONS.add(HUDI_TABLE_TYPE_KEY); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.RECORD_KEY_FIELD.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.INDEX_TYPE.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.INDEX_KEY_FIELD.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.PARTITION_PATH_FIELD.key()); + } + + /** + * Converts a Fluss TablePath to a Hudi ObjectPath. + * + * @param tablePath the Fluss table path + * @return the corresponding Hudi ObjectPath + */ + public static ObjectPath toHudiObjectPath(TablePath tablePath) { + return new ObjectPath(tablePath.getDatabaseName(), tablePath.getTableName()); + } + + public static ResolvedSchema convertToFlinkResolvedSchema( + TableDescriptor tableDescriptor, boolean isPkTable, String catalogMode) { + // validate hudi options first + validateHudiOptions(tableDescriptor.getProperties(), isPkTable); + validateHudiOptions(tableDescriptor.getCustomProperties(), isPkTable); + + // choose the correct converter based on catalog mode + FlussDataTypeToHudiDataType converter = + HIVE_META_STORE_TYPE.equals(catalogMode) + ? FlussDataTypeToHudiDataType.HMS_INSTANCE + : FlussDataTypeToHudiDataType.DFS_INSTANCE; + + List<Column> columns = new ArrayList<>(); + + // Add regular columns + for (org.apache.fluss.metadata.Schema.Column column : + tableDescriptor.getSchema().getColumns()) { + String columnName = column.getName(); + if (SYSTEM_COLUMNS.containsKey(columnName)) { + throw new InvalidTableException( + "Column " + + columnName + + " conflicts with a system column name of hudi table, please rename the column."); + } + columns.add(Column.physical(columnName, column.getDataType().accept(converter))); + } + + // add system metadata columns to schema + for (Map.Entry<String, DataType> systemColumn : SYSTEM_COLUMNS.entrySet()) { + columns.add(Column.physical(systemColumn.getKey(), systemColumn.getValue())); + } + + UniqueConstraint constraint = null; + // Set primary key if this is a PK table + if (isPkTable && tableDescriptor.hasPrimaryKey()) { + constraint = + UniqueConstraint.primaryKey( + "primaryKey", extractPrimaryKeyColumns(tableDescriptor)); + } + + return new ResolvedSchema(columns, Collections.emptyList(), constraint); + } + + /** + * Builds Hudi table properties from Fluss TableDescriptor. + * + * @param tableDescriptor the Fluss table descriptor + * @param isPkTable whether this is a primary key table + * @return map of Hudi table properties + */ + public static Map<String, String> buildHudiTableProperties( + TableDescriptor tableDescriptor, boolean isPkTable) { + Map<String, String> hudiProperties = new HashMap<>(); + // Set connector type + hudiProperties.put(FactoryUtil.CONNECTOR.key(), "hudi"); + hudiProperties.put("storageType", "hudi"); + + // Set table type based on whether it's a PK table + if (isPkTable) { + hudiProperties.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name()); + hudiProperties.put( + FlinkOptions.RECORD_KEY_FIELD.key(), + String.join(DELIMITER, extractPrimaryKeyColumns(tableDescriptor))); + } else { + hudiProperties.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.COPY_ON_WRITE.name()); + // set primary key for Fluss Log Table. + String recordKeyField = getRecordKeyField(tableDescriptor); + if (recordKeyField == null || recordKeyField.isEmpty()) { + throw new IllegalArgumentException("Record key field should be set."); + } + hudiProperties.put(FlinkOptions.RECORD_KEY_FIELD.key(), recordKeyField); + hudiProperties.put( + FlinkOptions.INDEX_KEY_FIELD.key(), + recordKeyField); // use primary key as index key + } + + // buket keys column + hudiProperties.put(FlinkOptions.INDEX_TYPE.key(), HoodieIndex.IndexType.BUCKET.name()); + List<String> bucketKeys = tableDescriptor.getBucketKeys(); + int numBuckets = + tableDescriptor + .getTableDistribution() + .flatMap(TableDescriptor.TableDistribution::getBucketCount) + .orElseThrow( + () -> new IllegalArgumentException("Bucket count should be set.")); + + if (!bucketKeys.isEmpty()) { + hudiProperties.put( + FlinkOptions.INDEX_KEY_FIELD.key(), String.join(DELIMITER, bucketKeys)); + } + hudiProperties.put(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS.key(), String.valueOf(numBuckets)); + + // partition keys column + List<String> partitionKeys = tableDescriptor.getPartitionKeys(); + hudiProperties.put( + FlinkOptions.PARTITION_PATH_FIELD.key(), String.join(DELIMITER, partitionKeys)); + + // Convert Fluss properties to Hudi properties + tableDescriptor + .getProperties() + .forEach((k, v) -> setFlussPropertyToHudi(k, v, hudiProperties)); + tableDescriptor + .getCustomProperties() + .forEach((k, v) -> setFlussPropertyToHudi(k, v, hudiProperties)); + + return hudiProperties; + } + + /** + * Creates a CatalogTable for Hudi from Fluss TableDescriptor. + * + * @param tableDescriptor the Fluss table descriptor + * @param isPkTable whether this is a primary key table + * @return the created CatalogTable + */ + public static CatalogTable createHudiCatalogTable( + TableDescriptor tableDescriptor, boolean isPkTable, String catalogMode) { + ResolvedSchema resolvedSchema = + convertToFlinkResolvedSchema(tableDescriptor, isPkTable, catalogMode); + Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(); + List<String> partitionKeys = tableDescriptor.getPartitionKeys(); + Map<String, String> options = buildHudiTableProperties(tableDescriptor, isPkTable); + LOG.info("Hudi table properties: {}", options); + + String comment = tableDescriptor.getComment().orElse("Hudi table created from Fluss"); + return HIVE_META_STORE_TYPE.equals(catalogMode) + ? HudiCatalogUtils.createCatalogTable(schema, partitionKeys, options, comment) + : HudiCatalogUtils.createResolvedCatalogTable( + schema, partitionKeys, options, comment, resolvedSchema); + } + + private static void setFlussPropertyToHudi( + String key, String value, Map<String, String> hudiProperties) { + if (key.startsWith(HUDI_CONF_PREFIX)) { + hudiProperties.put(key.substring(HUDI_CONF_PREFIX.length()), value); + } else { + hudiProperties.put(FLUSS_CONF_PREFIX + key, value); + } + } + + private static String getRecordKeyField(TableDescriptor tableDescriptor) { + String recordKeyOption = HUDI_CONF_PREFIX + FlinkOptions.RECORD_KEY_FIELD.key(); + String recordKeyField = tableDescriptor.getCustomProperties().get(recordKeyOption); + if (recordKeyField == null) { + recordKeyField = tableDescriptor.getProperties().get(recordKeyOption); + } + return recordKeyField; + } + + private static void validateHudiOptions(Map<String, String> properties, boolean isPkTable) { Review Comment: Logic: non-PK tables allow users to set `RECORD_KEY_FIELD` (combined with `getRecordKeyField`); PK tables forbid setting it. **However, the test `testUnsettableOptionInCustomPropertiesThrowsException`** expects a PK table passing `hudi.hoodie.datasource.write.recordkey.field` to fail; this rule indeed works at runtime. But the test at line 152 `testCreateLogTable` sets the same key on a **non-PK table** and **expects success** — that looks OK. Readability issue: the two validation scenarios are easy to confuse — **suggest adding a Javadoc on `validateHudiOptions`** explaining that `RECORD_KEY_FIELD` is a legal usage for non-PK tables (i.e., "log table = use the user-specified primary key field as the index field"). More importantly, after the function lets `RECORD_KEY_FIELD` pass, it doesn't validate that "non-PK tables must provide `RECORD_KEY_FIELD`"; that's done in `buildHudiTableProperties` via `IllegalArgumentException("Record key field should be set.")`. **`IllegalArgumentException` does not belong to the Fluss `InvalidConfigException` family** (Section 4); the user sees a raw IAE. This should be changed to `InvalidConfigException` with helpful hints including `tablePath`/the specific key for easier troubleshooting. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java: ########## @@ -0,0 +1,271 @@ +/* + * 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.fluss.lake.hudi.utils; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.exception.InvalidConfigException; +import org.apache.fluss.exception.InvalidTableException; +import org.apache.fluss.lake.hudi.FlussDataTypeToHudiDataType; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; + +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.UniqueConstraint; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.types.DataType; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.index.HoodieIndex; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.fluss.lake.hudi.HudiLakeCatalog.SYSTEM_COLUMNS; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; + +/** Utils for conversion between Hudi and Fluss. */ +public class HudiConversions { + + private static final Logger LOG = LoggerFactory.getLogger(HudiConversions.class); + + // for fluss config + private static final String FLUSS_CONF_PREFIX = "fluss."; + // for hudi config + private static final String HUDI_CONF_PREFIX = "hudi."; + + private static final String DELIMITER = ","; + private static final String HUDI_TABLE_TYPE_KEY = "hoodie.datasource.write.table.type"; + + /** Hudi config options set by Fluss should not be set by users. */ + @VisibleForTesting public static final Set<String> HUDI_UNSETTABLE_OPTIONS = new HashSet<>(); + + static { + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.TABLE_TYPE.key()); + HUDI_UNSETTABLE_OPTIONS.add(HUDI_TABLE_TYPE_KEY); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.RECORD_KEY_FIELD.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.INDEX_TYPE.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.INDEX_KEY_FIELD.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.BUCKET_INDEX_NUM_BUCKETS.key()); + HUDI_UNSETTABLE_OPTIONS.add(FlinkOptions.PARTITION_PATH_FIELD.key()); + } + + /** + * Converts a Fluss TablePath to a Hudi ObjectPath. + * + * @param tablePath the Fluss table path + * @return the corresponding Hudi ObjectPath + */ + public static ObjectPath toHudiObjectPath(TablePath tablePath) { + return new ObjectPath(tablePath.getDatabaseName(), tablePath.getTableName()); + } + + public static ResolvedSchema convertToFlinkResolvedSchema( + TableDescriptor tableDescriptor, boolean isPkTable, String catalogMode) { + // validate hudi options first + validateHudiOptions(tableDescriptor.getProperties(), isPkTable); + validateHudiOptions(tableDescriptor.getCustomProperties(), isPkTable); + + // choose the correct converter based on catalog mode + FlussDataTypeToHudiDataType converter = + HIVE_META_STORE_TYPE.equals(catalogMode) + ? FlussDataTypeToHudiDataType.HMS_INSTANCE + : FlussDataTypeToHudiDataType.DFS_INSTANCE; + + List<Column> columns = new ArrayList<>(); + + // Add regular columns + for (org.apache.fluss.metadata.Schema.Column column : + tableDescriptor.getSchema().getColumns()) { Review Comment: `SYSTEM_COLUMNS` (defined at `HudiLakeCatalog.java` L66–72) adds `__bucket / __offset / __timestamp`, but Hudi itself reserves `_hoodie_commit_time` / `_hoodie_record_key` / `_hoodie_partition_path`, etc. In `HudiConversions.createHudiCatalogTable` (L105–119), only collisions against Fluss system columns are checked before writing fluss columns + system columns into the schema. Suggest adding a rejection rule for the `_hoodie_` prefix (or a "reserved column" set parallel to `HUDI_UNSETTABLE_OPTIONS`) with corresponding tests. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java: ########## @@ -0,0 +1,260 @@ +/* + * 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.fluss.lake.hudi; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiConversions; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.metadata.TableChange; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.types.AbstractDataType; +import org.apache.flink.table.types.DataType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Objects; + +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HUDI_CATALOG_DEFAULT_NAME; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.MODE_CONFIG; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** Implementation of {@link LakeCatalog} for Hudi. */ +public class HudiLakeCatalog implements LakeCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCatalog.class); + + public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new LinkedHashMap<>(); + + static { + SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); + SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); + SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP(6)); + } + + private final Catalog hudiCatalog; + private final String catalogMode; + + public HudiLakeCatalog(Configuration configuration) { + this.catalogMode = configuration.toMap().getOrDefault(MODE_CONFIG, HIVE_META_STORE_TYPE); + this.hudiCatalog = HudiCatalogUtils.createHudiCatalog(configuration); + this.hudiCatalog.open(); + } + + @VisibleForTesting + protected Catalog getHudiCatalog() { + return hudiCatalog; + } + + @Override + public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Context context) + throws org.apache.fluss.exception.TableAlreadyExistException { + LOG.info("create the lake table for : {} with props: {}", tablePath, tableDescriptor); + + ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath); + + boolean isPkTable = tableDescriptor.getSchema().getPrimaryKeyIndexes().length > 0; + + // Create Hudi catalog table + CatalogTable catalogTable = + HudiConversions.createHudiCatalogTable(tableDescriptor, isPkTable, catalogMode); + + // Create table in Hudi catalog + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException e) { + createDatabase(tablePath.getDatabaseName()); + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException t) { + // shouldn't happen in normal cases + throw new RuntimeException( Review Comment: This throws a raw `RuntimeException` without passing t as the cause, losing the stack trace. Please change to new `RuntimeException`(msg, t). Similarly at line 144, the inner throw `new RuntimeException`(...) (schema fallback) also drops the cause (`tableNotExistException`) — please add it. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java: ########## @@ -0,0 +1,260 @@ +/* + * 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.fluss.lake.hudi; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiConversions; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.lake.lakestorage.LakeCatalog; +import org.apache.fluss.metadata.TableChange; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.Column; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogBaseTable; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.types.AbstractDataType; +import org.apache.flink.table.types.DataType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Objects; + +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HIVE_META_STORE_TYPE; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.HUDI_CATALOG_DEFAULT_NAME; +import static org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils.MODE_CONFIG; +import static org.apache.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME; +import static org.apache.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME; + +/** Implementation of {@link LakeCatalog} for Hudi. */ +public class HudiLakeCatalog implements LakeCatalog { + + private static final Logger LOG = LoggerFactory.getLogger(HudiLakeCatalog.class); + + public static final LinkedHashMap<String, DataType> SYSTEM_COLUMNS = new LinkedHashMap<>(); + + static { + SYSTEM_COLUMNS.put(BUCKET_COLUMN_NAME, DataTypes.INT()); + SYSTEM_COLUMNS.put(OFFSET_COLUMN_NAME, DataTypes.BIGINT()); + SYSTEM_COLUMNS.put(TIMESTAMP_COLUMN_NAME, DataTypes.TIMESTAMP(6)); + } + + private final Catalog hudiCatalog; + private final String catalogMode; + + public HudiLakeCatalog(Configuration configuration) { + this.catalogMode = configuration.toMap().getOrDefault(MODE_CONFIG, HIVE_META_STORE_TYPE); + this.hudiCatalog = HudiCatalogUtils.createHudiCatalog(configuration); + this.hudiCatalog.open(); + } + + @VisibleForTesting + protected Catalog getHudiCatalog() { + return hudiCatalog; + } + + @Override + public void createTable(TablePath tablePath, TableDescriptor tableDescriptor, Context context) + throws org.apache.fluss.exception.TableAlreadyExistException { + LOG.info("create the lake table for : {} with props: {}", tablePath, tableDescriptor); + + ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath); + + boolean isPkTable = tableDescriptor.getSchema().getPrimaryKeyIndexes().length > 0; + + // Create Hudi catalog table + CatalogTable catalogTable = + HudiConversions.createHudiCatalogTable(tableDescriptor, isPkTable, catalogMode); + + // Create table in Hudi catalog + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException e) { + createDatabase(tablePath.getDatabaseName()); + try { + createTable(objectPath, catalogTable, context.isCreatingFlussTable()); + } catch (DatabaseNotExistException t) { + // shouldn't happen in normal cases + throw new RuntimeException( + String.format( + "Fail to create table %s in Hudi, because " + + "Database %s still doesn't exist although create database " + + "successfully, please try again.", + tablePath, tablePath.getDatabaseName())); + } + } + } + + @Override + public void alterTable(TablePath tablePath, List<TableChange> tableChanges, Context context) + throws org.apache.fluss.exception.TableNotExistException { + throw new UnsupportedOperationException( + "Alter table is not supported for Hudi at the moment"); + } + + private void createTable( + ObjectPath tablePath, CatalogBaseTable catalogTable, boolean isCreatingFlussTable) + throws DatabaseNotExistException { + try { + hudiCatalog.createTable(tablePath, catalogTable, false); + LOG.info("Table {} created successfully.", tablePath); + } catch (org.apache.flink.table.catalog.exceptions.TableAlreadyExistException e) { + // table already exists, check schema compatibility for idempotency + try { + CatalogBaseTable existingTable = hudiCatalog.getTable(tablePath); + if (!isHudiSchemaCompatible(existingTable, catalogTable)) { + throw new org.apache.fluss.exception.TableAlreadyExistException( + String.format( + "The table %s already exists in Hudi catalog, but the table schema is not compatible. " + + "Please first drop the table in Hudi catalog or use a new table name.", + tablePath), + e); + } + // if creating a new fluss table, we should ensure the lake table is empty + // TODO: add emptiness check for Hudi table once LakeTieringFactory is implemented + if (isCreatingFlussTable) { + LOG.warn( + "Table {} already exists in Hudi catalog with compatible schema. " + + "Skipping creation as the table may not be empty.", + tablePath); + } + } catch (TableNotExistException tableNotExistException) { + // shouldn't happen in normal cases + throw new RuntimeException( + String.format( + "Failed to create table %s in Hudi. The table already existed " + + "during the initial creation attempt, but subsequently " + + "could not be found when trying to get it. " + + "Please check whether the Hudi table was manually deleted, and try again.", + tablePath)); + } + } + } + + /** + * Checks whether the existing Hudi table schema is compatible with the expected schema. + * + * <p>Compatibility means the column names, types, and nullability match. This is used for + * crash-recovery idempotency: if the table already exists with a compatible schema, the + * creation is considered successful. + */ + @VisibleForTesting + boolean isHudiSchemaCompatible(CatalogBaseTable existingTable, CatalogBaseTable expectedTable) { + return extractColumns(existingTable).equals(extractColumns(expectedTable)); + } + + private static List<ColumnSignature> extractColumns(CatalogBaseTable table) { + if (table instanceof ResolvedCatalogBaseTable) { + ResolvedSchema resolvedSchema = + ((ResolvedCatalogBaseTable<?>) table).getResolvedSchema(); + List<ColumnSignature> columns = new ArrayList<>(); + for (Column column : resolvedSchema.getColumns()) { + columns.add(new ColumnSignature(column.getName(), column.getDataType())); + } + return columns; + } + + Schema schema = table.getUnresolvedSchema(); + List<ColumnSignature> columns = new ArrayList<>(); + for (Schema.UnresolvedColumn column : schema.getColumns()) { + columns.add(new ColumnSignature(column.getName(), getDataType(column))); + } + return columns; + } + + private static AbstractDataType<?> getDataType(Schema.UnresolvedColumn column) { Review Comment: `getDataType` does not handle `UnresolvedComputedColumn` and silently returns `null`, `Schema.UnresolvedComputedColumn` has no datatype; returning `null` causes `ColumnSignature.equals` to mistakenly consider different computed columns equal. Hudi tables shouldn't have computed columns in theory, but a safer approach is to throw `IllegalStateException("Unexpected column kind: " + column.getClass())`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
