Copilot commented on code in PR #3456: URL: https://github.com/apache/fluss/pull/3456#discussion_r3376895931
########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,286 @@ +/* + * 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); + + 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(); + } + + private static boolean resolveBucketAware(Map<String, String> tableOptions) { + String bucketAware = tableOptions.get(FLUSS_BUCKET_AWARE_OPTION); + if (bucketAware != null) { + return Boolean.parseBoolean(bucketAware); + } + String bucketKeys = tableOptions.get(FLUSS_BUCKET_KEYS_OPTION); + if (bucketKeys != null) { + return !bucketKeys.trim().isEmpty(); + } + return true; + } + + @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("/"); + List<String> partitionValues = new ArrayList<>(pathSegments.length); + for (String pathSegment : pathSegments) { + int valueStart = pathSegment.indexOf('='); + partitionValues.add( + valueStart < 0 ? pathSegment : pathSegment.substring(valueStart + 1)); + } + return partitionValues; + } Review Comment: `extractPartitionValues` currently ignores `partitionFields` beyond the empty-check and returns one value per path segment in encountered order. This can produce a partition value list whose size/order does not match the declared partition columns (see `LakeSplit#partition()` contract), e.g. when paths are hive-style (`k=v`) but reordered, or when the number of segments differs from the number of partition fields. This can break downstream partition-to-column mapping. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiSplitPlanner.java: ########## @@ -0,0 +1,134 @@ +/* + * 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) { + throw e; + } catch (Exception e) { + throw new IOException("Failed to plan Hudi splits for table " + tablePath + ".", e); + } + } + + private List<HudiSplit> planPartition( + HudiTableInfo hudiTableInfo, String snapshotTime, String partitionPath) { + HoodieTableFileSystemView fileSystemView = hudiTableInfo.getFileSystemView(); + if (hudiTableInfo.getTableType() == HoodieTableType.MERGE_ON_READ) { + return fileSystemView + .getLatestMergedFileSlicesBeforeOrOn(partitionPath, snapshotTime) + .map(fileSlice -> toHudiSplit(hudiTableInfo, partitionPath, fileSlice)) + .collect(Collectors.toList()); + } + + return fileSystemView + .getLatestBaseFilesBeforeOrOn(partitionPath, snapshotTime) + .map(baseFile -> toFileSlice(partitionPath, baseFile)) + .map(fileSlice -> toHudiSplit(hudiTableInfo, partitionPath, fileSlice)) + .collect(Collectors.toList()); + } + + private FileSlice toFileSlice(String partitionPath, HoodieBaseFile baseFile) { + return new FileSlice( + new HoodieFileGroupId(partitionPath, baseFile.getFileId()), + baseFile.getCommitTime(), + baseFile, + Collections.emptyList()); + } + + private HudiSplit toHudiSplit( + HudiTableInfo hudiTableInfo, String partitionPath, FileSlice fileSlice) { + return new HudiSplit( + fileSlice, + extractBucket(hudiTableInfo, fileSlice), + hudiTableInfo.partitionValues(partitionPath)); + } + + private int extractBucket(HudiTableInfo hudiTableInfo, FileSlice fileSlice) { + if (!hudiTableInfo.isBucketAware()) { + return -1; + } + String fileId = fileSlice.getFileGroupId().getFileId(); + if (fileId == null || fileId.isEmpty()) { + return -1; + } + return BucketIdentifier.bucketIdFromFileId(fileId); + } Review Comment: `extractBucket` calls `BucketIdentifier.bucketIdFromFileId(fileId)` directly. For tables/files where the fileId does not follow Hudi bucket-id encoding (e.g., legacy data or bucket-unaware Fluss tables when metadata is missing/misconfigured), this can throw and fail split planning entirely. Prefer to treat unparsable fileIds as bucket-unaware (`-1`) and continue planning. ########## fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/HudiTableInfo.java: ########## @@ -0,0 +1,286 @@ +/* + * 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); + + 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(); + } + + private static boolean resolveBucketAware(Map<String, String> tableOptions) { + String bucketAware = tableOptions.get(FLUSS_BUCKET_AWARE_OPTION); + if (bucketAware != null) { + return Boolean.parseBoolean(bucketAware); + } + String bucketKeys = tableOptions.get(FLUSS_BUCKET_KEYS_OPTION); + if (bucketKeys != null) { + return !bucketKeys.trim().isEmpty(); + } + return true; + } Review Comment: `resolveBucketAware` defaults to `true` when neither `fluss.bucket-aware` nor `fluss.bucket.keys` is present. For existing Hudi tables created before these properties were introduced, this will force bucket extraction and grouping even for Fluss bucket-unaware log tables (which should use bucket `-1`), potentially producing incorrect TableBucket IDs/state keys. A safer fallback is to treat missing metadata as bucket-unaware. -- 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]
