XuQianJin-Stars commented on code in PR #3456: URL: https://github.com/apache/fluss/pull/3456#discussion_r3388401736
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitSerializer.java: ########## @@ -0,0 +1,51 @@ +/* + * 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.source; + +import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; +import org.apache.fluss.utils.InstantiationUtils; + +import java.io.IOException; + +/** Serializer for {@link HudiSplit}. */ +public class HudiSplitSerializer implements SimpleVersionedSerializer<HudiSplit> { + + private static final int CURRENT_VERSION = 1; + + @Override + public int getVersion() { + return CURRENT_VERSION; + } + + @Override + public byte[] serialize(HudiSplit hudiSplit) throws IOException { Review Comment: 1. `InstantiationUtils.deserializeObject` returns `Object` with **no type check** — a crafted byte stream can deserialize to arbitrary classes on the classpath. This violates the project security rule "Deserialization: always not trust, use safe load method." 2. Hudi internal types like `FileSlice` / `HoodieBaseFile` do not guarantee a stable `serialVersionUID` across versions, so this is also a forward-compatibility risk. **Minimum fix:** cast and validate the result, throwing `IOException` on type mismatch. **Preferred fix:** follow `PaimonSplitSerializer`'s field-by-field serialization — serialize `fileSlice` via Hudi's own Avro/JSON serializer (or its individual fields), and write `bucket` / `partition` explicitly. Don't bind our wire format to Hudi's internal Java serialization layout. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java: ########## @@ -0,0 +1,157 @@ +/* + * 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.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiTableInfo; +import org.apache.fluss.lake.source.Planner; +import org.apache.fluss.metadata.TablePath; + +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieFileGroupId; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.index.bucket.BucketIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** Planner for creating Hudi splits. */ +public class HudiSplitPlanner implements Planner<HudiSplit> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiSplitPlanner.class); + + private final Configuration hudiConfig; + private final TablePath tablePath; + private final long snapshotId; + + public HudiSplitPlanner(Configuration hudiConfig, TablePath tablePath, long snapshotId) { + this.hudiConfig = hudiConfig; + this.tablePath = tablePath; + this.snapshotId = snapshotId; + } + + @Override + public List<HudiSplit> plan() throws IOException { + String snapshotTime = String.valueOf(snapshotId); Review Comment: Hudi instant time is a string (canonical format is a 17-digit timestamp like `20260608010101000`). Round-tripping it through a `long` and then `String.valueOf` will drop any leading zeros. Today, Hudi writes don't normally produce leading-zero instants, but repaired or imported tables can. Note also that `LakeSource.PlannerContext#snapshotId()` returning `long` is itself a lossy carrier for Hudi. - Document the constraint that Hudi `snapshotId`s must not have leading zeros. - Or use `String.format("%017d", snapshotId)` to enforce the 17-digit instant format. - Add a "leading-zero instant" case to `HudiSplitPlannerTest`. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,338 @@ +/* + * 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.config.Configuration; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.table.catalog.CatalogOptions; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath; + +/** Resolved Hudi table metadata used by lake source planning. */ +public class HudiTableInfo implements AutoCloseable { + + private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse"; + private static final String DELIMITER = ","; + + private final TablePath tablePath; + private final Catalog hudiCatalog; + private final CatalogBaseTable hudiTable; + private final Map<String, String> tableOptions; + private final HoodieTableMetaClient metaClient; + private final HoodieEngineContext engineContext; + private final HoodieTimeline completedTimeline; + private final HoodieTableFileSystemView fileSystemView; + private final HoodieTableType tableType; + private final String basePath; + private final List<String> partitionFields; + private final boolean bucketAware; + + private HudiTableInfo( + TablePath tablePath, + Catalog hudiCatalog, + CatalogBaseTable hudiTable, + Map<String, String> tableOptions, + HoodieTableMetaClient metaClient, + HoodieEngineContext engineContext, + HoodieTimeline completedTimeline, + HoodieTableFileSystemView fileSystemView, + HoodieTableType tableType, + String basePath, + List<String> partitionFields, + boolean bucketAware) { + this.tablePath = tablePath; + this.hudiCatalog = hudiCatalog; + this.hudiTable = hudiTable; + this.tableOptions = tableOptions; + this.metaClient = metaClient; + this.engineContext = engineContext; + this.completedTimeline = completedTimeline; + this.fileSystemView = fileSystemView; + this.tableType = tableType; + this.basePath = basePath; + this.partitionFields = partitionFields; + this.bucketAware = bucketAware; + } + + public static HudiTableInfo create(TablePath tablePath, Configuration hudiConfig) + throws IOException { + Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig); + hudiCatalog.open(); + try { + CatalogBaseTable hudiTable = hudiCatalog.getTable(toHudiObjectPath(tablePath)); + Map<String, String> tableOptions = new HashMap<>(hudiTable.getOptions()); + tableOptions.putAll(hudiConfig.toMap()); + + String basePath = resolveBasePath(tablePath, tableOptions); + tableOptions.put(FlinkOptions.PATH.key(), basePath); + + HoodieTableMetaClient metaClient = createMetaClient(basePath, hudiConfig); + HoodieTimeline completedTimeline = Review Comment: `filterCompletedAndCompactionInstants()` returns completed commits **plus** REQUESTED / INFLIGHT compaction instants. So: - The local variable name `completedTimeline` is misleading. - More importantly, `HudiSplitPlanner#containsInstant(snapshotTime)` can match a **non-completed** compaction instant, and `getLatestMergedFileSlicesBeforeOrOn` at such an instant has undefined semantics. The PR description says "completed Hudi timeline" but the code does not match that. - Use `filterCompletedInstants()` if you only want completed commits. - Or rename the variable to `commitsAndCompactionTimeline` and explicitly handle the inflight-compaction case. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,338 @@ +/* + * 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.config.Configuration; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.table.catalog.CatalogOptions; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath; + +/** Resolved Hudi table metadata used by lake source planning. */ +public class HudiTableInfo implements AutoCloseable { + + private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse"; + private static final String DELIMITER = ","; + + private final TablePath tablePath; + private final Catalog hudiCatalog; + private final CatalogBaseTable hudiTable; + private final Map<String, String> tableOptions; + private final HoodieTableMetaClient metaClient; + private final HoodieEngineContext engineContext; + private final HoodieTimeline completedTimeline; + private final HoodieTableFileSystemView fileSystemView; + private final HoodieTableType tableType; + private final String basePath; + private final List<String> partitionFields; + private final boolean bucketAware; + + private HudiTableInfo( + TablePath tablePath, + Catalog hudiCatalog, + CatalogBaseTable hudiTable, + Map<String, String> tableOptions, + HoodieTableMetaClient metaClient, + HoodieEngineContext engineContext, + HoodieTimeline completedTimeline, + HoodieTableFileSystemView fileSystemView, + HoodieTableType tableType, + String basePath, + List<String> partitionFields, + boolean bucketAware) { + this.tablePath = tablePath; + this.hudiCatalog = hudiCatalog; + this.hudiTable = hudiTable; + this.tableOptions = tableOptions; + this.metaClient = metaClient; + this.engineContext = engineContext; + this.completedTimeline = completedTimeline; + this.fileSystemView = fileSystemView; + this.tableType = tableType; + this.basePath = basePath; + this.partitionFields = partitionFields; + this.bucketAware = bucketAware; + } + + public static HudiTableInfo create(TablePath tablePath, Configuration hudiConfig) + throws IOException { + Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig); + hudiCatalog.open(); + try { + CatalogBaseTable hudiTable = hudiCatalog.getTable(toHudiObjectPath(tablePath)); + Map<String, String> tableOptions = new HashMap<>(hudiTable.getOptions()); + tableOptions.putAll(hudiConfig.toMap()); + + String basePath = resolveBasePath(tablePath, tableOptions); + tableOptions.put(FlinkOptions.PATH.key(), basePath); + + HoodieTableMetaClient metaClient = createMetaClient(basePath, hudiConfig); + HoodieTimeline completedTimeline = + metaClient + .getCommitsAndCompactionTimeline() + .filterCompletedAndCompactionInstants(); + HoodieEngineContext engineContext = + new HoodieLocalEngineContext(metaClient.getStorageConf()); + HoodieTableFileSystemView fileSystemView = + HoodieTableFileSystemView.fileListingBasedFileSystemView( + engineContext, metaClient, completedTimeline); + HoodieTableType tableType = metaClient.getTableType(); + List<String> partitionFields = + splitCommaSeparated( + tableOptions.getOrDefault( + FLUSS_PARTITION_KEYS_OPTION, + tableOptions.get(FlinkOptions.PARTITION_PATH_FIELD.key()))); + boolean bucketAware = resolveBucketAware(tableOptions, tableType); + + return new HudiTableInfo( + tablePath, + hudiCatalog, + hudiTable, + Collections.unmodifiableMap(new HashMap<>(tableOptions)), + metaClient, + engineContext, + completedTimeline, + fileSystemView, + tableType, + basePath, + Collections.unmodifiableList(new ArrayList<>(partitionFields)), + bucketAware); + } catch (Exception e) { + closeCatalog(hudiCatalog); + if (e instanceof IOException) { + throw (IOException) e; + } + throw new IOException("Failed to resolve Hudi table info for " + tablePath + ".", e); + } + } + + private static HoodieTableMetaClient createMetaClient(String basePath, Configuration hudiConfig) + throws IOException { + try { + return HoodieTableMetaClient.builder() + .setBasePath(basePath) + .setConf(new HadoopStorageConfiguration(getHadoopConfiguration(hudiConfig))) + .build(); + } catch (TableNotFoundException e) { + throw new IOException("Hudi table not found at " + basePath + ".", e); + } + } + + public static org.apache.hadoop.conf.Configuration getHadoopConfiguration( + Configuration hudiConfig) { + org.apache.flink.configuration.Configuration flinkConfig = + org.apache.flink.configuration.Configuration.fromMap(hudiConfig.toMap()); + return HadoopConfigurations.getHadoopConf(flinkConfig); + } + + private static String resolveBasePath(TablePath tablePath, Map<String, String> tableOptions) { + String path = tableOptions.get(FlinkOptions.PATH.key()); + if (path != null && !path.trim().isEmpty()) { + return path; + } + String catalogPath = + tableOptions.getOrDefault( + CatalogOptions.CATALOG_PATH.key(), DEFAULT_HUDI_WAREHOUSE); + return ensureEndsWithSlash(catalogPath) + + tablePath.getDatabaseName() + + "/" + + tablePath.getTableName(); + } + + @VisibleForTesting + public static boolean resolveBucketAware( + Map<String, String> tableOptions, HoodieTableType tableType) { + String bucketAware = tableOptions.get(FLUSS_BUCKET_AWARE_OPTION); + if (bucketAware != null) { + return Boolean.parseBoolean(bucketAware.trim()); + } + String bucketKeys = tableOptions.get(FLUSS_BUCKET_KEYS_OPTION); + if (bucketKeys != null) { + return !bucketKeys.trim().isEmpty(); + } + return tableType == HoodieTableType.MERGE_ON_READ; + } + + @VisibleForTesting + public static List<String> extractPartitionValues( + String partitionPath, List<String> partitionFields) { + if (partitionFields.isEmpty() || partitionPath == null || partitionPath.isEmpty()) { + return Collections.emptyList(); + } + + String[] pathSegments = partitionPath.split("/"); Review Comment: The current tests cover "unknown field", "duplicate field" and "mixed style", but not the case where the number of segments equals `partitionFields.size()` yet the field names don't match (e.g. `dt=20260608/dt=20260609`). The implementation correctly throws via the `null` check, but please add a unit test for this edge case. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiConversions.java: ########## Review Comment: Trivial fix: `buket` → `bucket`. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplit.java: ########## @@ -0,0 +1,91 @@ +/* + * 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.source; + +import org.apache.fluss.lake.source.LakeSplit; + +import org.apache.hudi.common.model.FileSlice; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** A readable split of a Hudi table. */ +public class HudiSplit implements LakeSplit { + + private static final long serialVersionUID = 1L; + + private final FileSlice fileSlice; + private final int bucket; + private final List<String> partition; + + public HudiSplit(FileSlice fileSlice, int bucket, List<String> partition) { + this.fileSlice = Objects.requireNonNull(fileSlice, "fileSlice cannot be null"); + this.bucket = bucket; + this.partition = + Collections.unmodifiableList( + new ArrayList<>( + Objects.requireNonNull(partition, "partition cannot be null"))); + } + + @Override + public int bucket() { + return bucket; + } + + @Override + public List<String> partition() { + return partition; + } + + public FileSlice getFileSlice() { + return fileSlice; + } + + @Override + public boolean equals(Object o) { Review Comment: `FileSlice#equals/hashCode` only compare `HoodieFileGroupId + baseInstantTime` (log files are not considered). If Hudi changes that contract in a future release, the equality semantics of `HudiSplit` will silently shift. **Suggestion:** either document this dependency in the class javadoc, or implement equality explicitly using `fileGroupId + baseInstantTime + baseFile.path + logFiles.paths`. OK to defer to a follow-up PR if you prefer. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,338 @@ +/* + * 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.config.Configuration; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.table.catalog.CatalogOptions; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath; + +/** Resolved Hudi table metadata used by lake source planning. */ +public class HudiTableInfo implements AutoCloseable { + + private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse"; + private static final String DELIMITER = ","; + + private final TablePath tablePath; + private final Catalog hudiCatalog; + private final CatalogBaseTable hudiTable; + private final Map<String, String> tableOptions; + private final HoodieTableMetaClient metaClient; + private final HoodieEngineContext engineContext; + private final HoodieTimeline completedTimeline; + private final HoodieTableFileSystemView fileSystemView; + private final HoodieTableType tableType; + private final String basePath; + private final List<String> partitionFields; + private final boolean bucketAware; + + private HudiTableInfo( + TablePath tablePath, + Catalog hudiCatalog, + CatalogBaseTable hudiTable, + Map<String, String> tableOptions, + HoodieTableMetaClient metaClient, + HoodieEngineContext engineContext, + HoodieTimeline completedTimeline, + HoodieTableFileSystemView fileSystemView, + HoodieTableType tableType, + String basePath, + List<String> partitionFields, + boolean bucketAware) { + this.tablePath = tablePath; + this.hudiCatalog = hudiCatalog; + this.hudiTable = hudiTable; + this.tableOptions = tableOptions; + this.metaClient = metaClient; + this.engineContext = engineContext; + this.completedTimeline = completedTimeline; + this.fileSystemView = fileSystemView; + this.tableType = tableType; + this.basePath = basePath; + this.partitionFields = partitionFields; + this.bucketAware = bucketAware; + } + + public static HudiTableInfo create(TablePath tablePath, Configuration hudiConfig) + throws IOException { + Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig); + hudiCatalog.open(); + try { + CatalogBaseTable hudiTable = hudiCatalog.getTable(toHudiObjectPath(tablePath)); + Map<String, String> tableOptions = new HashMap<>(hudiTable.getOptions()); + tableOptions.putAll(hudiConfig.toMap()); + + String basePath = resolveBasePath(tablePath, tableOptions); + tableOptions.put(FlinkOptions.PATH.key(), basePath); + + HoodieTableMetaClient metaClient = createMetaClient(basePath, hudiConfig); + HoodieTimeline completedTimeline = + metaClient + .getCommitsAndCompactionTimeline() + .filterCompletedAndCompactionInstants(); + HoodieEngineContext engineContext = + new HoodieLocalEngineContext(metaClient.getStorageConf()); + HoodieTableFileSystemView fileSystemView = + HoodieTableFileSystemView.fileListingBasedFileSystemView( + engineContext, metaClient, completedTimeline); + HoodieTableType tableType = metaClient.getTableType(); + List<String> partitionFields = + splitCommaSeparated( + tableOptions.getOrDefault( + FLUSS_PARTITION_KEYS_OPTION, + tableOptions.get(FlinkOptions.PARTITION_PATH_FIELD.key()))); + boolean bucketAware = resolveBucketAware(tableOptions, tableType); + + return new HudiTableInfo( + tablePath, + hudiCatalog, + hudiTable, + Collections.unmodifiableMap(new HashMap<>(tableOptions)), + metaClient, + engineContext, + completedTimeline, + fileSystemView, + tableType, + basePath, + Collections.unmodifiableList(new ArrayList<>(partitionFields)), + bucketAware); + } catch (Exception e) { Review Comment: If `HoodieTableFileSystemView` has already been constructed and a subsequent step (e.g. `splitCommaSeparated(...)` / `resolveBucketAware(...)`) throws, the FSView is leaked. **Suggestion:** hold `fsView` in a local and `IOUtils.closeQuietly` it in the catch as well. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java: ########## @@ -0,0 +1,157 @@ +/* + * 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.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiTableInfo; +import org.apache.fluss.lake.source.Planner; +import org.apache.fluss.metadata.TablePath; + +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieFileGroupId; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.index.bucket.BucketIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** Planner for creating Hudi splits. */ +public class HudiSplitPlanner implements Planner<HudiSplit> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiSplitPlanner.class); + + private final Configuration hudiConfig; + private final TablePath tablePath; + private final long snapshotId; + + public HudiSplitPlanner(Configuration hudiConfig, TablePath tablePath, long snapshotId) { + this.hudiConfig = hudiConfig; + this.tablePath = tablePath; + this.snapshotId = snapshotId; + } + + @Override + public List<HudiSplit> plan() throws IOException { + String snapshotTime = String.valueOf(snapshotId); + try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath, hudiConfig)) { + if (!hudiTableInfo.getCompletedTimeline().containsInstant(snapshotTime)) { + throw new IOException( + String.format( + "Hudi instant time %s does not exist in table %s.", + snapshotTime, tablePath)); + } + + List<String> partitionPaths = + FSUtils.getAllPartitionPaths( + hudiTableInfo.getEngineContext(), hudiTableInfo.getMetaClient(), false); + if (partitionPaths.isEmpty()) { + partitionPaths = Collections.singletonList(""); + } + + List<HudiSplit> splits = new ArrayList<>(); + for (String partitionPath : partitionPaths) { + splits.addAll(planPartition(hudiTableInfo, snapshotTime, partitionPath)); + } + LOG.debug( + "Planned {} Hudi splits for table {} at instant {}.", + splits.size(), + tablePath, + snapshotTime); + return splits; + } catch (IOException e) { Review Comment: The first catch is redundant, and the second one converts every `RuntimeException` (e.g. `IllegalStateException`, `HoodieException`) into an `IOException`, hiding the original type and making operational diagnosis harder. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,338 @@ +/* + * 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.config.Configuration; +import org.apache.fluss.lake.hudi.utils.catalog.HudiCatalogUtils; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.engine.HoodieLocalEngineContext; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.configuration.HadoopConfigurations; +import org.apache.hudi.exception.TableNotFoundException; +import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration; +import org.apache.hudi.table.catalog.CatalogOptions; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_AWARE_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_BUCKET_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.FLUSS_PARTITION_KEYS_OPTION; +import static org.apache.fluss.lake.hudi.utils.HudiConversions.toHudiObjectPath; + +/** Resolved Hudi table metadata used by lake source planning. */ +public class HudiTableInfo implements AutoCloseable { + + private static final String DEFAULT_HUDI_WAREHOUSE = "/tmp/hudi_warehouse"; + private static final String DELIMITER = ","; + + private final TablePath tablePath; + private final Catalog hudiCatalog; + private final CatalogBaseTable hudiTable; + private final Map<String, String> tableOptions; + private final HoodieTableMetaClient metaClient; + private final HoodieEngineContext engineContext; + private final HoodieTimeline completedTimeline; + private final HoodieTableFileSystemView fileSystemView; + private final HoodieTableType tableType; + private final String basePath; + private final List<String> partitionFields; + private final boolean bucketAware; + + private HudiTableInfo( + TablePath tablePath, + Catalog hudiCatalog, + CatalogBaseTable hudiTable, + Map<String, String> tableOptions, + HoodieTableMetaClient metaClient, + HoodieEngineContext engineContext, + HoodieTimeline completedTimeline, + HoodieTableFileSystemView fileSystemView, + HoodieTableType tableType, + String basePath, + List<String> partitionFields, + boolean bucketAware) { + this.tablePath = tablePath; + this.hudiCatalog = hudiCatalog; + this.hudiTable = hudiTable; + this.tableOptions = tableOptions; + this.metaClient = metaClient; + this.engineContext = engineContext; + this.completedTimeline = completedTimeline; + this.fileSystemView = fileSystemView; + this.tableType = tableType; + this.basePath = basePath; + this.partitionFields = partitionFields; + this.bucketAware = bucketAware; + } + + public static HudiTableInfo create(TablePath tablePath, Configuration hudiConfig) + throws IOException { + Catalog hudiCatalog = HudiCatalogUtils.createHudiCatalog(hudiConfig); + hudiCatalog.open(); + try { + CatalogBaseTable hudiTable = hudiCatalog.getTable(toHudiObjectPath(tablePath)); + Map<String, String> tableOptions = new HashMap<>(hudiTable.getOptions()); + tableOptions.putAll(hudiConfig.toMap()); + + String basePath = resolveBasePath(tablePath, tableOptions); + tableOptions.put(FlinkOptions.PATH.key(), basePath); + + HoodieTableMetaClient metaClient = createMetaClient(basePath, hudiConfig); + HoodieTimeline completedTimeline = + metaClient + .getCommitsAndCompactionTimeline() + .filterCompletedAndCompactionInstants(); + HoodieEngineContext engineContext = + new HoodieLocalEngineContext(metaClient.getStorageConf()); + HoodieTableFileSystemView fileSystemView = + HoodieTableFileSystemView.fileListingBasedFileSystemView( + engineContext, metaClient, completedTimeline); + HoodieTableType tableType = metaClient.getTableType(); + List<String> partitionFields = + splitCommaSeparated( + tableOptions.getOrDefault( + FLUSS_PARTITION_KEYS_OPTION, + tableOptions.get(FlinkOptions.PARTITION_PATH_FIELD.key()))); + boolean bucketAware = resolveBucketAware(tableOptions, tableType); + + return new HudiTableInfo( + tablePath, + hudiCatalog, + hudiTable, + Collections.unmodifiableMap(new HashMap<>(tableOptions)), + metaClient, + engineContext, + completedTimeline, + fileSystemView, + tableType, + basePath, + Collections.unmodifiableList(new ArrayList<>(partitionFields)), + bucketAware); + } catch (Exception e) { + closeCatalog(hudiCatalog); + if (e instanceof IOException) { + throw (IOException) e; + } + throw new IOException("Failed to resolve Hudi table info for " + tablePath + ".", e); + } + } + + private static HoodieTableMetaClient createMetaClient(String basePath, Configuration hudiConfig) + throws IOException { + try { + return HoodieTableMetaClient.builder() + .setBasePath(basePath) + .setConf(new HadoopStorageConfiguration(getHadoopConfiguration(hudiConfig))) + .build(); + } catch (TableNotFoundException e) { + throw new IOException("Hudi table not found at " + basePath + ".", e); + } + } + + public static org.apache.hadoop.conf.Configuration getHadoopConfiguration( + Configuration hudiConfig) { + org.apache.flink.configuration.Configuration flinkConfig = + org.apache.flink.configuration.Configuration.fromMap(hudiConfig.toMap()); + return HadoopConfigurations.getHadoopConf(flinkConfig); + } + + private static String resolveBasePath(TablePath tablePath, Map<String, String> tableOptions) { + String path = tableOptions.get(FlinkOptions.PATH.key()); + if (path != null && !path.trim().isEmpty()) { + return path; + } + String catalogPath = + tableOptions.getOrDefault( + CatalogOptions.CATALOG_PATH.key(), DEFAULT_HUDI_WAREHOUSE); + return ensureEndsWithSlash(catalogPath) + + tablePath.getDatabaseName() + + "/" + + tablePath.getTableName(); + } + + @VisibleForTesting + public static boolean resolveBucketAware( + Map<String, String> tableOptions, HoodieTableType tableType) { + String bucketAware = tableOptions.get(FLUSS_BUCKET_AWARE_OPTION); + if (bucketAware != null) { + return Boolean.parseBoolean(bucketAware.trim()); + } + String bucketKeys = tableOptions.get(FLUSS_BUCKET_KEYS_OPTION); + if (bucketKeys != null) { + return !bucketKeys.trim().isEmpty(); + } + return tableType == HoodieTableType.MERGE_ON_READ; + } + + @VisibleForTesting + public static List<String> extractPartitionValues( + String partitionPath, List<String> partitionFields) { + if (partitionFields.isEmpty() || partitionPath == null || partitionPath.isEmpty()) { + return Collections.emptyList(); + } + + String[] pathSegments = partitionPath.split("/"); + if (pathSegments.length != partitionFields.size()) { + throw new IllegalArgumentException( + String.format( + "Hudi partition path '%s' does not match partition fields %s.", + partitionPath, partitionFields)); + } + + boolean hiveStyle = false; + boolean rawStyle = false; + Map<String, String> hivePartitionValues = new HashMap<>(); + List<String> rawPartitionValues = new ArrayList<>(pathSegments.length); + for (String pathSegment : pathSegments) { + int valueStart = pathSegment.indexOf('='); + if (valueStart > 0) { + hiveStyle = true; + String field = pathSegment.substring(0, valueStart); + if (!partitionFields.contains(field)) { + throw new IllegalArgumentException( + String.format( + "Hudi partition path '%s' contains unknown partition field '%s'.", + partitionPath, field)); + } + if (hivePartitionValues.put(field, pathSegment.substring(valueStart + 1)) != null) { + throw new IllegalArgumentException( + String.format( + "Hudi partition path '%s' contains duplicate partition field '%s'.", + partitionPath, field)); + } + } else { + rawStyle = true; + rawPartitionValues.add(pathSegment); + } + } + + if (hiveStyle && rawStyle) { + throw new IllegalArgumentException( + String.format( + "Hudi partition path '%s' mixes hive-style and raw partition segments.", + partitionPath)); + } + + if (!hiveStyle) { + return rawPartitionValues; + } + + List<String> partitionValues = new ArrayList<>(partitionFields.size()); + for (String partitionField : partitionFields) { + String partitionValue = hivePartitionValues.get(partitionField); + if (partitionValue == null) { + throw new IllegalArgumentException( + String.format( + "Hudi partition path '%s' does not contain partition field '%s'.", + partitionPath, partitionField)); + } + partitionValues.add(partitionValue); + } + return partitionValues; + } + + private static List<String> splitCommaSeparated(String value) { + if (value == null || value.trim().isEmpty()) { + return Collections.emptyList(); + } + String[] parts = value.split(DELIMITER); + List<String> values = new ArrayList<>(parts.length); + for (String part : parts) { + String trimmed = part.trim(); + if (!trimmed.isEmpty()) { + values.add(trimmed); + } + } + return values; + } + + private static String ensureEndsWithSlash(String path) { + return path.endsWith("/") ? path : path + "/"; + } + + private static void closeCatalog(Catalog hudiCatalog) { + if (hudiCatalog instanceof AutoCloseable) { + IOUtils.closeQuietly((AutoCloseable) hudiCatalog, "hudi catalog"); + } + } + + public TablePath getTablePath() { + return tablePath; + } + + public CatalogBaseTable getHudiTable() { + return hudiTable; + } + + public Map<String, String> getTableOptions() { + return tableOptions; + } + + public HoodieTableMetaClient getMetaClient() { + return metaClient; + } + + public HoodieEngineContext getEngineContext() { + return engineContext; + } + + public HoodieTimeline getCompletedTimeline() { + return completedTimeline; + } + + public HoodieTableFileSystemView getFileSystemView() { + return fileSystemView; + } + + public HoodieTableType getTableType() { + return tableType; + } + + public String getBasePath() { + return basePath; + } + + public boolean isBucketAware() { + return bucketAware; + } + + public List<String> partitionValues(String partitionPath) { + return extractPartitionValues(partitionPath, partitionFields); + } + + @Override + public void close() { + fileSystemView.close(); Review Comment: `close()` should be idempotent and must release **all** resources even if one of them throws. Currently: 1. If `fileSystemView.close()` throws, `closeCatalog(hudiCatalog)` is skipped → catalog leak. 2. Calling `close()` twice will throw on the second invocation. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/HudiLakeStorage.java: ########## Review Comment: `createLakeSource` is implemented in this PR, so calling the storage "a scaffold" is no longer accurate. Please rephrase along the lines of: > "Hudi lake tiering writer is not implemented yet (tracking <issue/PR link>)." so users can locate the tracking issue. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java: ########## @@ -0,0 +1,157 @@ +/* + * 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.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiTableInfo; +import org.apache.fluss.lake.source.Planner; +import org.apache.fluss.metadata.TablePath; + +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieFileGroupId; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.index.bucket.BucketIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** Planner for creating Hudi splits. */ +public class HudiSplitPlanner implements Planner<HudiSplit> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiSplitPlanner.class); + + private final Configuration hudiConfig; + private final TablePath tablePath; + private final long snapshotId; + + public HudiSplitPlanner(Configuration hudiConfig, TablePath tablePath, long snapshotId) { + this.hudiConfig = hudiConfig; + this.tablePath = tablePath; + this.snapshotId = snapshotId; + } + + @Override + public List<HudiSplit> plan() throws IOException { + String snapshotTime = String.valueOf(snapshotId); + try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath, hudiConfig)) { + if (!hudiTableInfo.getCompletedTimeline().containsInstant(snapshotTime)) { + throw new IOException( + String.format( + "Hudi instant time %s does not exist in table %s.", + snapshotTime, tablePath)); + } + + List<String> partitionPaths = + FSUtils.getAllPartitionPaths( + hudiTableInfo.getEngineContext(), hudiTableInfo.getMetaClient(), false); + if (partitionPaths.isEmpty()) { + partitionPaths = Collections.singletonList(""); + } + + List<HudiSplit> splits = new ArrayList<>(); + for (String partitionPath : partitionPaths) { + splits.addAll(planPartition(hudiTableInfo, snapshotTime, partitionPath)); + } + LOG.debug( Review Comment: Only logging on the success path makes "why is my query empty?" investigations harder. Please log at `info`/`warn` when `splits.isEmpty()` (especially after a successful `containsInstant` check), since that usually signals a filter or partition mismatch. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java: ########## @@ -0,0 +1,157 @@ +/* + * 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.source; + +import org.apache.fluss.config.Configuration; +import org.apache.fluss.lake.hudi.utils.HudiTableInfo; +import org.apache.fluss.lake.source.Planner; +import org.apache.fluss.metadata.TablePath; + +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.FileSlice; +import org.apache.hudi.common.model.HoodieBaseFile; +import org.apache.hudi.common.model.HoodieFileGroupId; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.view.HoodieTableFileSystemView; +import org.apache.hudi.index.bucket.BucketIdentifier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** Planner for creating Hudi splits. */ +public class HudiSplitPlanner implements Planner<HudiSplit> { + + private static final Logger LOG = LoggerFactory.getLogger(HudiSplitPlanner.class); + + private final Configuration hudiConfig; + private final TablePath tablePath; + private final long snapshotId; + + public HudiSplitPlanner(Configuration hudiConfig, TablePath tablePath, long snapshotId) { + this.hudiConfig = hudiConfig; + this.tablePath = tablePath; + this.snapshotId = snapshotId; + } + + @Override + public List<HudiSplit> plan() throws IOException { + String snapshotTime = String.valueOf(snapshotId); + try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath, hudiConfig)) { + if (!hudiTableInfo.getCompletedTimeline().containsInstant(snapshotTime)) { + throw new IOException( + String.format( + "Hudi instant time %s does not exist in table %s.", + snapshotTime, tablePath)); + } + + List<String> partitionPaths = + FSUtils.getAllPartitionPaths( + hudiTableInfo.getEngineContext(), hudiTableInfo.getMetaClient(), false); + if (partitionPaths.isEmpty()) { Review Comment: `FSUtils.getAllPartitionPaths(...)` returns an empty list for a **partitioned** table that has no data written yet. With this fallback we then ask `getLatestMergedFileSlicesBeforeOrOn("")`, which is meaningless on a partitioned layout. In addition, `HudiTableInfo#partitionValues("")` returns `emptyList()` for that case, so the resulting `HudiSplit.partition()` has a length different from `partitionFields.size()` and downstream consumers that rely on `partition().size() == partitionColumns.size()` will silently misalign. **Suggestion:** Only fall back to `""` when the table is **non-partitioned** (`partitionFields.isEmpty()`). For a partitioned table with no discovered partitions, return an empty split list and log at debug. -- 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]
