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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
