You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fhan688 (via GitHub)" <gi...@apache.org> on 2026/05/28 10:39:20 UTC
[PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
fhan688 opened a new pull request, #3395:
URL: https://github.com/apache/fluss/pull/3395
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: https://github.com/apache/fluss/issues/3275
This PR introduces the Hudi `LakeCatalog` implementation, enabling Fluss to create tables in Hudi data lake storage. This aligns with the existing Paimon and Iceberg lake catalog support, completing the trio of supported lake formats for table creation.
### Brief change log
#### New modules & classes (`fluss-lake/fluss-lake-hudi`):
- `HudiLakeCatalog`: Implements `LakeCatalog` interface, supporting both HMS (Hive Metastore) and DFS (filesystem) catalog modes. Handles table creation with schema compatibility check for crash-recovery idempotency, automatic database creation, and system column (`__bucket, __offset, __timestamp`) appending.
- `FlussDataTypeToHudiDataType`: Implements `DataTypeVisitor` to convert Fluss data types to Flink types (Hudi's type system). Handles `LocalZonedTimestampType` specially: maps to `BIGINT` under HMS mode, `TIMESTAMP_WITH_LOCAL_TIME_ZONE` under DFS mode.
- `HudiConversions`: Core conversion utility. Converts Fluss `TablePath` → Hudi `ObjectPath`, `TableDescriptor` → `ResolvedSchema` / Hudi table properties. Validates `HUDI_UNSETTABLE_OPTIONS` (6 protected options that Fluss auto-manages), checks system column name conflicts, and handles property prefix rewriting (`hudi.xxx` → `xxx`, others → `fluss.xxx`).
- `HudiCatalogUtils`: Factory for creating Hudi `Catalog` instances (`HoodieHiveCatalog` for HMS, `HoodieCatalog for` DFS). Uses copied Configuration to avoid mutating the original.
#### Modifications to existing modules (`fluss-flink/fluss-flink-common`):
- `LakeFlinkCatalog`: Adds HUDI branch in `getLakeCatalog()` with a new `HudiCatalogFactory` inner class that uses reflection to instantiate Hudi catalog (mirroring the Iceberg pattern to avoid compile-time dependency on hudi-flink-bundle).
- `LakeTableFactory`: Adds HUDI branch in `getLakeTableFactory()` with `getHudiFactory()` that reflectively loads `HoodieTableFactory`.
- `HudiLakeStorage`: Replaces the `UnsupportedOperationException` in `createLakeCatalog()` with `new HudiLakeCatalog(hudiConfig)` to wire the SPI path.
Key design decisions:
| Aspect | Decision |
|--------|----------|
| Table type mapping | PK table → MERGE_ON_READ, Log table → COPY_ON_WRITE |
| Index strategy | BUCKET index type, aligned with Fluss's bucketing model |
| Dependency isolation | Hudi bundle loaded via reflection/plugin classloader (no compile-time dependency in fluss-flink-common) |
| Catalog mode | Supports hms (Hive Metastore) and dfs (filesystem) |
| Property rewriting | hudi. prefix stripped; non-hudi properties prefixed with fluss. |
| Idempotency | Schema-compatible duplicate creation is treated as success for crash recovery |
### Tests
`HudiLakeCatalogTest` (14 test cases):
- `testPropertyPrefixRewriting` — verifies hudi.xxx → xxx and non-hudi → fluss.xxx prefix rewriting
- `testCreatePrimaryKeyTable` — PK table (MOR) creation with system columns and primary key
- `testCreateLogTable` — Log table (COW) creation with record key from customProperties
- `testIsHudiSchemaCompatibleWithSameSchema` — compatible schemas return true
- `testIsHudiSchemaCompatibleWithDifferentColumnCount` — different column count returns false
- `testIsHudiSchemaCompatibleWithDifferentColumnName` — different column name returns false
- `testIsHudiSchemaCompatibleWithDifferentColumnType` — different column type returns false
- `testCreateDuplicateTableWithCompatibleSchema` — duplicate creation with compatible schema is idempotent
- t`estCreateDuplicateTableWithIncompatibleSchema` — duplicate creation with incompatible schema throws `TableAlreadyExistException`
- `testUnsettableOptionInPropertiesThrowsException` — protected option in `properties` throws `InvalidConfigException`
- `testUnsettableOptionInCustomPropertiesThrowsException` — protected option in customProperties throws `InvalidConfigException`
- `testNonProtectedHudiOptionPassesValidation` — non-protected option passes validation
- `testSystemColumnBucketConflictThrowsException` — `__bucket` conflict throws `InvalidTableException`
- `testSystemColumnOffsetConflictThrowsException` — `__offset` conflict throws `InvalidTableException`
- `testSystemColumnTimestampConflictThrowsException` — `__timestamp` conflict throws `InvalidTableException`
### API and Format
No API or storage format changes. This PR only adds new implementations for the existing `LakeCatalog` and `LakeStorage SPI` interfaces.
### Documentation
A new feature — Hudi lake catalog support for table creation. Will need documentation updates for the Hudi integration guide.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
XuQianJin-Stars commented on PR #3395:
URL: https://github.com/apache/fluss/pull/3395#issuecomment-4611637173
+1
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 closed pull request #3395: [lake/hudi] Introduce Hudi LakeCatalog to create table
URL: https://github.com/apache/fluss/pull/3395
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3395:
URL: https://github.com/apache/fluss/pull/3395#discussion_r3339450432
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java:
##########
@@ -86,6 +87,11 @@ public Catalog getLakeCatalog(
} else if (lakeFormat == ICEBERG) {
catalog = IcebergCatalogFactory.create(catalogName, catalogProperties);
this.lakeFormat = ICEBERG;
+ } else if (lakeFormat == HUDI) {
Review Comment:
Could you please remove these changes in `LakeFlinkCatalog` and `LakeTableFactory` in this pr. Let's focus Hudi own catalog instead of flink catalog in this pr.
We do add them back in another pr which requirs a IT to test we can use $lake to read hudi table
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java:
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.apache.flink.table.types.logical.LogicalType;
+import org.apache.hudi.table.catalog.CatalogOptions;
+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.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(CatalogOptions.MODE.key(), 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
Review Comment:
could you please add comment that why convert to flink catalog table and call flink catalog to create hudi table?
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/FlussDataTypeToHudiDataType.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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 javax.annotation.concurrent.ThreadSafe;
+
+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. */
Review Comment:
nit: not Hudi's internal data type. It's flink internal data type.
Also add comments why convert to flink internal data type intead of Hudi's internal data type.
##########
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java:
##########
@@ -0,0 +1,645 @@
+/*
+ * 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.config.Configuration;
+import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.exception.InvalidTableException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.lake.hudi.utils.HudiConversions;
+import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.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.Column;
+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.TableNotExistException;
+import org.apache.flink.table.types.DataType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit test for {@link HudiLakeCatalog}. */
+class HudiLakeCatalogTest {
+
+ @TempDir private File tempWarehouseDir;
+
+ private HudiLakeCatalog flussHudiLakeCatalog;
+
+ @BeforeEach
+ public void setUp() {
+ Configuration configuration = new Configuration();
+ configuration.setString("catalog.path", tempWarehouseDir.toURI().toString());
+ configuration.setString("mode", "dfs");
+ this.flussHudiLakeCatalog = new HudiLakeCatalog(configuration);
+ }
+
+ /** Verify property prefix rewriting. */
+ @Test
+ void testPropertyPrefixRewriting() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "test_table";
+
+ Schema flussSchema =
+ Schema.newBuilder().column("id", DataTypes.BIGINT()).primaryKey("id").build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(3)
+ .property("hudi.precombine.field", "id")
+ .property("table.datalake.freshness", "30s")
+ .build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ CatalogBaseTable table =
+ flussHudiLakeCatalog
+ .getHudiCatalog()
+ .getTable(HudiConversions.toHudiObjectPath(tablePath));
+
+ // Verify property prefix rewriting
+ assertThat(table.getOptions()).containsEntry("precombine.field", "id");
+ assertThat(table.getOptions()).containsEntry("fluss.table.datalake.freshness", "30s");
+ assertThat(table.getOptions())
+ .doesNotContainKeys("hudi.precombine.field", "table.datalake.freshness");
+ }
+
+ @Test
+ void testCreatePrimaryKeyTable() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "pk_table";
+
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .withComment("pk_table")
+ .primaryKey("id")
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(flussSchema).distributedBy(4, "id").build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath);
+ CatalogBaseTable table = flussHudiLakeCatalog.getHudiCatalog().getTable(objectPath);
+
+ assertThat(table).isNotNull();
+
+ List<String> primaryKeys = new ArrayList<>();
+ primaryKeys.add("id");
+ TableSchema expectHudiSchema =
Review Comment:
nit: 'org.apache.flink.table.api.TableSchema' is deprecated
##########
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java:
##########
@@ -0,0 +1,645 @@
+/*
+ * 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.config.Configuration;
+import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.exception.InvalidTableException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.lake.hudi.utils.HudiConversions;
+import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.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.Column;
+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.TableNotExistException;
+import org.apache.flink.table.types.DataType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit test for {@link HudiLakeCatalog}. */
+class HudiLakeCatalogTest {
+
+ @TempDir private File tempWarehouseDir;
+
+ private HudiLakeCatalog flussHudiLakeCatalog;
+
+ @BeforeEach
+ public void setUp() {
+ Configuration configuration = new Configuration();
+ configuration.setString("catalog.path", tempWarehouseDir.toURI().toString());
+ configuration.setString("mode", "dfs");
+ this.flussHudiLakeCatalog = new HudiLakeCatalog(configuration);
+ }
+
+ /** Verify property prefix rewriting. */
+ @Test
+ void testPropertyPrefixRewriting() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "test_table";
+
+ Schema flussSchema =
+ Schema.newBuilder().column("id", DataTypes.BIGINT()).primaryKey("id").build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(3)
+ .property("hudi.precombine.field", "id")
+ .property("table.datalake.freshness", "30s")
+ .build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ CatalogBaseTable table =
+ flussHudiLakeCatalog
+ .getHudiCatalog()
+ .getTable(HudiConversions.toHudiObjectPath(tablePath));
+
+ // Verify property prefix rewriting
+ assertThat(table.getOptions()).containsEntry("precombine.field", "id");
+ assertThat(table.getOptions()).containsEntry("fluss.table.datalake.freshness", "30s");
+ assertThat(table.getOptions())
+ .doesNotContainKeys("hudi.precombine.field", "table.datalake.freshness");
+ }
+
+ @Test
+ void testCreatePrimaryKeyTable() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "pk_table";
+
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .withComment("pk_table")
+ .primaryKey("id")
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(flussSchema).distributedBy(4, "id").build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath);
+ CatalogBaseTable table = flussHudiLakeCatalog.getHudiCatalog().getTable(objectPath);
+
+ assertThat(table).isNotNull();
+
+ List<String> primaryKeys = new ArrayList<>();
+ primaryKeys.add("id");
+ TableSchema expectHudiSchema =
+ TableSchema.builder()
+ .field("id", org.apache.flink.table.api.DataTypes.INT().notNull())
+ .field("name", org.apache.flink.table.api.DataTypes.STRING())
+ .field("__bucket", org.apache.flink.table.api.DataTypes.INT())
+ .field("__offset", org.apache.flink.table.api.DataTypes.BIGINT())
+ .field("__timestamp", org.apache.flink.table.api.DataTypes.TIMESTAMP(6))
+ .primaryKey("primaryKey", primaryKeys.toArray(new String[0]))
+ .build();
+
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectHudiSchema.toSchema());
+ }
+
+ @Test
+ void testCreateLogTable() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "log_table";
+
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .build();
+
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("hudi.hoodie.datasource.write.recordkey.field", "id");
+
+ TableDescriptor td =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(3) // no bucket key
+ .customProperties(customProperties)
+ .build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, td, context);
+
+ ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath);
+ CatalogBaseTable table = flussHudiLakeCatalog.getHudiCatalog().getTable(objectPath);
+
+ List<String> primaryKeys = new ArrayList<>();
+ primaryKeys.add("id");
+ TableSchema expectHudiSchema =
+ TableSchema.builder()
+ .field("id", org.apache.flink.table.api.DataTypes.BIGINT().notNull())
+ .field("name", org.apache.flink.table.api.DataTypes.STRING())
+ .field("__bucket", org.apache.flink.table.api.DataTypes.INT())
+ .field("__offset", org.apache.flink.table.api.DataTypes.BIGINT())
+ .field("__timestamp", org.apache.flink.table.api.DataTypes.TIMESTAMP(6))
+ .primaryKey("PK_id", primaryKeys.toArray(new String[0]))
+ .build();
+
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectHudiSchema.toSchema());
+ }
+
+ @Test
+ void testCreateLogTableWithoutRecordKeyThrowsException() {
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(flussSchema).distributedBy(3).build();
+
+ TablePath tablePath = TablePath.of("test_db", "log_table_without_record_key");
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+
+ assertThatThrownBy(
+ () -> flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context))
+ .isInstanceOf(InvalidConfigException.class)
+ .hasMessageContaining("hudi.hoodie.datasource.write.recordkey.field")
+ .hasMessageContaining(tablePath.toString());
+ }
+
+ // ------------------------------------------------------------------
+ // isHudiSchemaCompatible() tests
+ // ------------------------------------------------------------------
+
+ @Test
+ void testIsHudiSchemaCompatibleWithSameSchema() {
Review Comment:
nit: combine testIsHudiSchemaCompatible releated test into single one test method?
##########
fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/HudiLakeCatalogTest.java:
##########
@@ -0,0 +1,645 @@
+/*
+ * 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.config.Configuration;
+import org.apache.fluss.exception.InvalidConfigException;
+import org.apache.fluss.exception.InvalidTableException;
+import org.apache.fluss.exception.TableAlreadyExistException;
+import org.apache.fluss.lake.hudi.utils.HudiConversions;
+import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.types.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.Column;
+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.TableNotExistException;
+import org.apache.flink.table.types.DataType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Unit test for {@link HudiLakeCatalog}. */
+class HudiLakeCatalogTest {
+
+ @TempDir private File tempWarehouseDir;
+
+ private HudiLakeCatalog flussHudiLakeCatalog;
+
+ @BeforeEach
+ public void setUp() {
+ Configuration configuration = new Configuration();
+ configuration.setString("catalog.path", tempWarehouseDir.toURI().toString());
+ configuration.setString("mode", "dfs");
+ this.flussHudiLakeCatalog = new HudiLakeCatalog(configuration);
+ }
+
+ /** Verify property prefix rewriting. */
+ @Test
+ void testPropertyPrefixRewriting() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "test_table";
+
+ Schema flussSchema =
+ Schema.newBuilder().column("id", DataTypes.BIGINT()).primaryKey("id").build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(3)
+ .property("hudi.precombine.field", "id")
+ .property("table.datalake.freshness", "30s")
+ .build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ CatalogBaseTable table =
+ flussHudiLakeCatalog
+ .getHudiCatalog()
+ .getTable(HudiConversions.toHudiObjectPath(tablePath));
+
+ // Verify property prefix rewriting
+ assertThat(table.getOptions()).containsEntry("precombine.field", "id");
+ assertThat(table.getOptions()).containsEntry("fluss.table.datalake.freshness", "30s");
+ assertThat(table.getOptions())
+ .doesNotContainKeys("hudi.precombine.field", "table.datalake.freshness");
+ }
+
+ @Test
+ void testCreatePrimaryKeyTable() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "pk_table";
+
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .withComment("pk_table")
+ .primaryKey("id")
+ .build();
+
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder().schema(flussSchema).distributedBy(4, "id").build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, tableDescriptor, context);
+
+ ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath);
+ CatalogBaseTable table = flussHudiLakeCatalog.getHudiCatalog().getTable(objectPath);
+
+ assertThat(table).isNotNull();
+
+ List<String> primaryKeys = new ArrayList<>();
+ primaryKeys.add("id");
+ TableSchema expectHudiSchema =
+ TableSchema.builder()
+ .field("id", org.apache.flink.table.api.DataTypes.INT().notNull())
+ .field("name", org.apache.flink.table.api.DataTypes.STRING())
+ .field("__bucket", org.apache.flink.table.api.DataTypes.INT())
+ .field("__offset", org.apache.flink.table.api.DataTypes.BIGINT())
+ .field("__timestamp", org.apache.flink.table.api.DataTypes.TIMESTAMP(6))
+ .primaryKey("primaryKey", primaryKeys.toArray(new String[0]))
+ .build();
+
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectHudiSchema.toSchema());
+ }
+
+ @Test
+ void testCreateLogTable() throws TableNotExistException {
+ String database = "test_db";
+ String tableName = "log_table";
+
+ Schema flussSchema =
+ Schema.newBuilder()
+ .column("id", DataTypes.BIGINT())
+ .column("name", DataTypes.STRING())
+ .build();
+
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("hudi.hoodie.datasource.write.recordkey.field", "id");
+
+ TableDescriptor td =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(3) // no bucket key
+ .customProperties(customProperties)
+ .build();
+
+ TablePath tablePath = TablePath.of(database, tableName);
+ TestingLakeCatalogContext context = new TestingLakeCatalogContext();
+ flussHudiLakeCatalog.createTable(tablePath, td, context);
+
+ ObjectPath objectPath = HudiConversions.toHudiObjectPath(tablePath);
+ CatalogBaseTable table = flussHudiLakeCatalog.getHudiCatalog().getTable(objectPath);
+
+ List<String> primaryKeys = new ArrayList<>();
+ primaryKeys.add("id");
+ TableSchema expectHudiSchema =
Review Comment:
dito
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 closed pull request #3395: [lake/hudi] Introduce Hudi LakeCatalog to create table
URL: https://github.com/apache/fluss/pull/3395
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #3395:
URL: https://github.com/apache/fluss/pull/3395
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3395:
URL: https://github.com/apache/fluss/pull/3395#discussion_r3348039223
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeCatalog.java:
##########
@@ -0,0 +1,277 @@
+/*
+ * 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.apache.flink.table.types.logical.LogicalType;
+import org.apache.hudi.table.catalog.CatalogOptions;
+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.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(CatalogOptions.MODE.key(), 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
Review Comment:
Update: after disucss with author offline, unlike Iceberg and Paimon which provide their own catalog abstractions, Hudi does not have a standalone catalog API. We convert to a Flink catalog table and delegate to the Flink catalog to create the Hudi table.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3395:
URL: https://github.com/apache/fluss/pull/3395#discussion_r3332727928
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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_METADATA_COLUMN_PREFIX = "_hoodie_";
+ private static final String HUDI_TABLE_TYPE_KEY = "hoodie.datasource.write.table.type";
+ private static final String HUDI_RECORD_KEY_FIELD_OPTION =
+ HUDI_CONF_PREFIX + FlinkOptions.RECORD_KEY_FIELD.key();
+
+ /** 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(
+ TablePath tablePath,
+ 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(
+ String.format(
+ "Column %s in table %s conflicts with a system column name of Hudi table, "
+ + "please rename the column.",
+ columnName, tablePath));
+ }
+ if (columnName.startsWith(HUDI_METADATA_COLUMN_PREFIX)) {
+ throw new InvalidTableException(
+ String.format(
+ "Column %s in table %s conflicts with the reserved Hudi metadata column "
+ + "prefix '%s', please rename the column.",
+ columnName, tablePath, HUDI_METADATA_COLUMN_PREFIX));
+ }
+ 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 tablePath the path of the Fluss table
+ * @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(
+ TablePath tablePath, 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.trim().isEmpty()) {
+ throw new InvalidConfigException(
+ String.format(
+ "The Hudi record key field option %s should be set for log table %s. "
+ + "Please set it to the column used as the Hudi record key.",
+ HUDI_RECORD_KEY_FIELD_OPTION, tablePath));
+ }
+ 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 tablePath the path of the Fluss table
+ * @param tableDescriptor the Fluss table descriptor
+ * @param isPkTable whether this is a primary key table
+ * @return the created CatalogTable
+ */
+ public static CatalogTable createHudiCatalogTable(
+ TablePath tablePath,
+ TableDescriptor tableDescriptor,
+ boolean isPkTable,
+ String catalogMode) {
+ ResolvedSchema resolvedSchema =
+ convertToFlinkResolvedSchema(tablePath, tableDescriptor, isPkTable, catalogMode);
+ Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
+ List<String> partitionKeys = tableDescriptor.getPartitionKeys();
+ Map<String, String> options =
+ buildHudiTableProperties(tablePath, 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 recordKeyField =
+ tableDescriptor.getCustomProperties().get(HUDI_RECORD_KEY_FIELD_OPTION);
+ if (recordKeyField == null) {
+ recordKeyField = tableDescriptor.getProperties().get(HUDI_RECORD_KEY_FIELD_OPTION);
+ }
+ return recordKeyField;
+ }
Review Comment:
`getRecordKeyField()` only looks up the *prefixed* option key (`hudi.hoodie.datasource.write.recordkey.field`). However `validateHudiOptions()` explicitly allows `hoodie.datasource.write.recordkey.field` for log tables, and users may set the unprefixed (native Hudi) option. In that case table creation will still fail with "should be set" because the value is never found.
Consider accepting both prefixed and unprefixed keys when extracting the record key field so validation and extraction are consistent.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeTableFactory.java:
##########
@@ -101,4 +103,18 @@ private DynamicTableSourceFactory getIcebergFactory() {
e);
}
}
+
+ private DynamicTableSourceFactory getHudiFactory() {
+ try {
+ Class<?> hudiFactoryClass = Class.forName("org.apache.hudi.table.HoodieTableFactory");
+ return (DynamicTableSourceFactory)
+ hudiFactoryClass.getDeclaredConstructor().newInstance();
+ } catch (Exception e) {
Review Comment:
`getHudiFactory()` uses `Class.forName(...)` without specifying a classloader. In plugin deployments (as suggested by the error message), the Hudi bundle is typically loaded by Flink's plugin / context classloader, so resolving with the application classloader can fail.
Load `HoodieTableFactory` via the thread context classloader (or the Flink factory context classloader) to reliably find plugin-provided classes.
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/catalog/HudiCatalogUtils.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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);
+
+ private static final String CATALOG_NAME_KEY = "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();
+ // copy the configuration to avoid modifying the original
+ Configuration copiedConfig = new Configuration(configuration);
+ copiedConfig.setString(CatalogOptions.DEFAULT_DATABASE.key(), "tmp");
+ copiedConfig.setString(CatalogOptions.TABLE_EXTERNAL.key(), "true");
+ String catalogName = hudiProps.getOrDefault(CATALOG_NAME_KEY, HUDI_CATALOG_DEFAULT_NAME);
+ return buildHudiCatalog(
+ catalogName,
+ hudiProps,
+ org.apache.flink.configuration.Configuration.fromMap(copiedConfig.toMap()));
+ }
+
+ public static Catalog buildHudiCatalog(
+ String catalogName,
+ Map<String, String> hudiProps,
+ org.apache.flink.configuration.Configuration configuration) {
+ String catalogMode =
+ hudiProps.getOrDefault(CatalogOptions.MODE.key(), HIVE_META_STORE_TYPE);
+ LOG.info(
+ "create hudi catalog: {}, mode: {}, configuration: {}",
+ catalogName,
+ catalogMode,
+ configuration);
Review Comment:
`buildHudiCatalog()` logs the full Flink `Configuration` at INFO level. Catalog configurations frequently include sensitive values (e.g., metastore credentials, cloud storage secrets). Logging the entire configuration risks leaking secrets into logs.
Consider logging only the catalog name/mode (or logging options keys only at DEBUG) to avoid exposing sensitive configuration values.
##########
fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * 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_METADATA_COLUMN_PREFIX = "_hoodie_";
+ private static final String HUDI_TABLE_TYPE_KEY = "hoodie.datasource.write.table.type";
+ private static final String HUDI_RECORD_KEY_FIELD_OPTION =
+ HUDI_CONF_PREFIX + FlinkOptions.RECORD_KEY_FIELD.key();
+
+ /** 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(
+ TablePath tablePath,
+ 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(
+ String.format(
+ "Column %s in table %s conflicts with a system column name of Hudi table, "
+ + "please rename the column.",
+ columnName, tablePath));
+ }
+ if (columnName.startsWith(HUDI_METADATA_COLUMN_PREFIX)) {
+ throw new InvalidTableException(
+ String.format(
+ "Column %s in table %s conflicts with the reserved Hudi metadata column "
+ + "prefix '%s', please rename the column.",
+ columnName, tablePath, HUDI_METADATA_COLUMN_PREFIX));
+ }
+ 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 tablePath the path of the Fluss table
+ * @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(
+ TablePath tablePath, 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.trim().isEmpty()) {
+ throw new InvalidConfigException(
+ String.format(
+ "The Hudi record key field option %s should be set for log table %s. "
+ + "Please set it to the column used as the Hudi record key.",
+ HUDI_RECORD_KEY_FIELD_OPTION, tablePath));
+ }
+ 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 tablePath the path of the Fluss table
+ * @param tableDescriptor the Fluss table descriptor
+ * @param isPkTable whether this is a primary key table
+ * @return the created CatalogTable
+ */
+ public static CatalogTable createHudiCatalogTable(
+ TablePath tablePath,
+ TableDescriptor tableDescriptor,
+ boolean isPkTable,
+ String catalogMode) {
+ ResolvedSchema resolvedSchema =
+ convertToFlinkResolvedSchema(tablePath, tableDescriptor, isPkTable, catalogMode);
+ Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
+ List<String> partitionKeys = tableDescriptor.getPartitionKeys();
+ Map<String, String> options =
+ buildHudiTableProperties(tablePath, tableDescriptor, isPkTable);
+ LOG.info("Hudi table properties: {}", options);
Review Comment:
`createHudiCatalogTable()` logs the full resolved Hudi table `options` map at INFO. These options can include sensitive configuration (e.g., filesystem credentials, metastore URIs with embedded auth). Logging them at INFO risks leaking secrets.
Consider lowering this to DEBUG and/or redacting known sensitive keys before logging.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on PR #3395:
URL: https://github.com/apache/fluss/pull/3395#issuecomment-4564596587
all tests are passed, please help review, thanks! @XuQianJin-Stars
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/hudi] Introduce Hudi LakeCatalog to create table [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 closed pull request #3395: [lake/hudi] Introduce Hudi LakeCatalog to create table
URL: https://github.com/apache/fluss/pull/3395
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org