Skip to content

perf(index): keep tables and timeline reloads out of index lookup tasks #20095

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
yihua wants to merge 4 commits into
base: master
Choose a base branch
from
Draft

perf(index): keep tables and timeline reloads out of index lookup tasks #20095

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
581ffa6
feat(core): add an engine-agnostic broadcast to HoodieEngineContext
yihua Sep 26, 2026
cc02ac1
perf(index): keep tables and timeline reloads out of index lookup tasks
yihua Sep 27, 2026
e80431e
Read index lookup base files from the table's view so the timeline se…
yihua Sep 27, 2026
7898a7d
Sync the timeline server with a read before loading the index lookup …
yihua Sep 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 19 additions & 4 deletions ...lient/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieGlobalSimpleIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.apache.hudi.index;

import org.apache.hudi.common.data.HoodieBroadcast;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
Expand All @@ -27,11 +28,14 @@
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.HoodieKeyLocationFetchHandle;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.table.HoodieTable;

import java.util.List;
Expand Down Expand Up @@ -84,15 +88,26 @@ protected <R> HoodieData<HoodieRecord<R>> tagLocationInternal(
return taggedRecords;
}

private HoodiePairData<String, HoodieRecordGlobalLocation> fetchRecordGlobalLocations(
@VisibleForTesting
HoodiePairData<String, HoodieRecordGlobalLocation> fetchRecordGlobalLocations(
HoodieEngineContext context, HoodieTable hoodieTable,
List<Pair<String, HoodieBaseFile>> baseFiles) {
int parallelism = getParallelism(config.getGlobalSimpleIndexParallelism(), baseFiles.size());
return fetchRecordGlobalLocations(context, hoodieTable.getMetaClient(), baseFiles, parallelism, keyGeneratorOpt);
}

/**
* Static so that the tasks capture only the arguments, not the index or the table.
*/
private static HoodiePairData<String, HoodieRecordGlobalLocation> fetchRecordGlobalLocations(
HoodieEngineContext context, HoodieTableMetaClient metaClient, List<Pair<String, HoodieBaseFile>> baseFiles,
int parallelism, Option<BaseKeyGenerator> keyGeneratorOpt) {
HoodieBroadcast<StorageConfiguration<?>> storageConf = context.broadcast(metaClient.getStorageConf());
String basePath = metaClient.getBasePath().toString();
return context.parallelize(baseFiles, parallelism)
.flatMap(partitionPathBaseFile -> new HoodieKeyLocationFetchHandle(config, hoodieTable, partitionPathBaseFile, keyGeneratorOpt)
.globalLocations())
.mapToPair(e -> (Pair<String, HoodieRecordGlobalLocation>) e);
.flatMap(partitionPathBaseFile -> HoodieKeyLocationFetchHandle.globalLocations(
HoodieStorageUtils.getStorage(basePath, storageConf.value()), partitionPathBaseFile, keyGeneratorOpt))
.mapToPair(pair -> pair);
}

/**
Expand Down
76 changes: 60 additions & 16 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
import org.apache.hudi.common.table.read.HoodieFileGroupReader;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.util.ConfigUtils;
import org.apache.hudi.common.util.HoodieRecordUtils;
import org.apache.hudi.common.util.HoodieTimer;
Expand Down Expand Up @@ -109,6 +110,12 @@
@Slf4j
public class HoodieIndexUtils {

/**
* Characters of partition paths per load request. The list travels in the timeline server request
* line, where encoding can triple it, so this keeps the line under the server's 8 KB header limit.
*/
private static final int LOAD_PARTITIONS_MAX_PATHS_LENGTH = 2000;

/**
* Fetches Pair of partition path and {@link HoodieBaseFile}s for interested partitions.
*
Expand Down Expand Up @@ -182,6 +189,9 @@ public static List<FileSlice> getLatestFileSlicesForPartition(
/**
* Fetches Pair of partition path and {@link HoodieBaseFile}s for interested partitions.
*
* <p>With the metadata table, the driver reads the base files from the table's view after loading the
* partitions into it in batched metadata table lookups. Otherwise each partition is listed by its own task.
*
* @param partitions list of partitions of interest
* @param context instance of {@link HoodieEngineContext} to use
* @param hoodieTable instance of {@link HoodieTable} of interest
Expand All @@ -190,15 +200,51 @@ public static List<FileSlice> getLatestFileSlicesForPartition(
public static List<Pair<String, HoodieBaseFile>> getLatestBaseFilesForAllPartitions(final List<String> partitions,
final HoodieEngineContext context,
final HoodieTable hoodieTable) {
context.setJobStatus(HoodieIndexUtils.class.getSimpleName(), "Load latest base files from all partitions: " + hoodieTable.getConfig().getTableName());
return context.flatMap(partitions, partitionPath -> {
List<Pair<String, HoodieBaseFile>> filteredFiles =
getLatestBaseFilesForPartition(partitionPath, hoodieTable).stream()
.map(baseFile -> Pair.of(partitionPath, baseFile))
.collect(toList());

return filteredFiles.stream();
}, Math.max(partitions.size(), 1));
HoodieTableMetaClient metaClient = hoodieTable.getMetaClient();
if (!hoodieTable.getConfig().isMetadataTableEnabled() || !metaClient.getTableConfig().isMetadataTableAvailable()
|| metaClient.getMetaserverConfig().isMetaserverEnabled()) {
context.setJobStatus(HoodieIndexUtils.class.getSimpleName(), "Load latest base files from all partitions: " + hoodieTable.getConfig().getTableName());
return context.flatMap(partitions, partitionPath -> {
List<Pair<String, HoodieBaseFile>> filteredFiles =
getLatestBaseFilesForPartition(partitionPath, hoodieTable).stream()
.map(baseFile -> Pair.of(partitionPath, baseFile))
.collect(toList());

return filteredFiles.stream();
}, Math.max(partitions.size(), 1));
}
Option<HoodieInstant> latestCommitTime = metaClient.getCommitsTimeline().filterCompletedInstants().lastInstant();
if (partitions.isEmpty() || latestCommitTime.isEmpty()) {
return Collections.emptyList();
}
String maxCommitTime = latestCommitTime.get().requestedTime();
// The table's view, not a new one: with the embedded timeline server its reads sync the server's view
// to this write's timeline, which the write handles send later. A server view synced only by the
// handles would reload a timeline that a table service may have moved on, and reject them. The
// server does not sync on load requests, so the first read goes before the loads; a sync after them
// would drop the loaded partitions.
SyncableFileSystemView view = hoodieTable.getHoodieView();
List<Pair<String, HoodieBaseFile>> baseFiles = new ArrayList<>(getLatestBaseFiles(view, partitions.get(0), maxCommitTime));
List<String> remaining = partitions.subList(1, partitions.size());
int start = 0;
while (start < remaining.size()) {
int end = start;
int length = 0;
do {
length += remaining.get(end).length() + 3;
end++;
} while (end < remaining.size() && length + remaining.get(end).length() + 3 <= LOAD_PARTITIONS_MAX_PATHS_LENGTH);
view.loadPartitions(remaining.subList(start, end));
start = end;
}
remaining.forEach(partitionPath -> baseFiles.addAll(getLatestBaseFiles(view, partitionPath, maxCommitTime)));
return baseFiles;
}

private static List<Pair<String, HoodieBaseFile>> getLatestBaseFiles(SyncableFileSystemView view, String partitionPath, String maxCommitTime) {
return view.getLatestBaseFilesBeforeOrOn(partitionPath, maxCommitTime)
.map(baseFile -> Pair.of(partitionPath, baseFile))
.collect(toList());
}

/**
Expand Down Expand Up @@ -296,7 +342,8 @@ public static HoodieIndex createUserDefinedIndex(HoodieWriteConfig config) {
*
* @return {@link HoodieRecord}s that have the current location being set.
*/
private static <R> HoodieData<HoodieRecord<R>> getExistingRecords(
@VisibleForTesting
static <R> HoodieData<HoodieRecord<R>> getExistingRecords(
HoodieData<Pair<String, String>> partitionLocations, HoodieWriteConfig config, HoodieTable hoodieTable, ReaderContextFactory<R> readerContextFactory, HoodieSchema dataSchema) {
HoodieTableMetaClient metaClient = hoodieTable.getMetaClient();
final Option<String> instantTime = metaClient
Expand All @@ -307,16 +354,13 @@ private static <R> HoodieData<HoodieRecord<R>> getExistingRecords(
if (instantTime.isEmpty()) {
return hoodieTable.getContext().emptyHoodieData();
}
Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema());
return partitionLocations.flatMap(p -> {
Option<FileSlice> fileSliceOption = Option.fromJavaOptional(hoodieTable
.getHoodieView()
.getLatestMergedFileSlicesBeforeOrOn(p.getLeft(), instantTime.get())
.filter(fileSlice -> fileSlice.getFileId().equals(p.getRight()))
.findFirst());
Option<FileSlice> fileSliceOption = hoodieTable.getHoodieView()
.getLatestMergedFileSliceBeforeOrOn(p.getLeft(), instantTime.get(), p.getRight());
if (fileSliceOption.isEmpty()) {
return Collections.emptyIterator();
}
Option<InternalSchema> internalSchemaOption = SerDeHelper.fromJson(config.getInternalSchema());
FileSlice fileSlice = fileSliceOption.get();
HoodieReaderContext<R> readerContext = readerContextFactory.getContext();
HoodieFileGroupReader<R> fileGroupReader = HoodieFileGroupReader.<R>builder()
Expand Down
20 changes: 17 additions & 3 deletions hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieSimpleIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,19 +20,23 @@
package org.apache.hudi.index;

import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.data.HoodieBroadcast;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.HoodieStorageUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ImmutablePair;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.io.HoodieKeyLocationFetchHandle;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.table.HoodieTable;

import java.util.List;
Expand Down Expand Up @@ -142,11 +146,21 @@ protected HoodiePairData<HoodieKey, HoodieRecordLocation> fetchRecordLocations(
HoodieEngineContext context, HoodieTable hoodieTable,
List<Pair<String, HoodieBaseFile>> baseFiles) {
int parallelism = getParallelism(config.getSimpleIndexParallelism(), baseFiles.size());
return fetchRecordLocations(context, hoodieTable.getMetaClient(), baseFiles, parallelism, keyGeneratorOpt);
}

/**
* Static so that the tasks capture only the arguments, not the index or the table.
*/
private static HoodiePairData<HoodieKey, HoodieRecordLocation> fetchRecordLocations(
HoodieEngineContext context, HoodieTableMetaClient metaClient, List<Pair<String, HoodieBaseFile>> baseFiles,
int parallelism, Option<BaseKeyGenerator> keyGeneratorOpt) {
HoodieBroadcast<StorageConfiguration<?>> storageConf = context.broadcast(metaClient.getStorageConf());
String basePath = metaClient.getBasePath().toString();
return context.parallelize(baseFiles, parallelism)
.flatMap(partitionPathBaseFile -> new HoodieKeyLocationFetchHandle(config, hoodieTable, partitionPathBaseFile, keyGeneratorOpt)
.locations())
.mapToPair(e -> (Pair<HoodieKey, HoodieRecordLocation>) e);
.flatMap(partitionPathBaseFile -> HoodieKeyLocationFetchHandle.locations(
HoodieStorageUtils.getStorage(basePath, storageConf.value()), partitionPathBaseFile, keyGeneratorOpt))
.mapToPair(pair -> pair);
}

protected int getParallelism(int configuredParallelism, int numberOfBaseFiles) {
Expand Down
Loading
Loading