924060929 commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3801162804
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1366,6 +1377,261 @@ public List<org.apache.iceberg.FileScanTask>
getAndClearIcebergRewriteFileScanTa
return tasks;
}
+ public ExternalScanTaskCache getExternalScanTaskCache() {
+ return externalScanTaskCache;
+ }
+
+ /**
+ * Release scan tasks at the end of one execution without closing reusable
prepared-statement
+ * state. Delayed scan work retains only the invalidated generation and
cannot repopulate this
+ * StatementContext.
+ */
+ public void clearExternalScanTasks() {
+ externalScanTaskCache.invalidate();
+ }
+
+ /**
+ * One execution generation of statement-scoped external scan tasks.
+ *
+ * <p>Scan nodes capture this object when they are constructed. Resetting
a prepared statement
+ * swaps the generation before invalidating the old one, so a delayed
asynchronous scan from
+ * the previous execution cannot insert tasks into the next execution's
cache.
+ */
+ public static final class ExternalScanTaskCache {
+ /** Maximum number of connector tasks retained by one statement cache
generation. */
+ public static final long MAX_RETAINED_TASK_COUNT = 10_000;
+
+ /** Loads a value using the weight atomically reserved for this cache
entry. */
+ @FunctionalInterface
+ public interface WeightedLoader<T> {
+ List<T> load(long reservedWeight) throws Exception;
+ }
+
+ /** Independent cumulative budgets used by task-count and
serialized-byte retention. */
+ public enum WeightBudget {
+ TASK_COUNT,
+ ICEBERG_SERIALIZED_BYTES,
+ PAIMON_SERIALIZED_BYTES
+ }
+
+ private final Map<ExternalScanTaskCacheKey<?>,
CompletableFuture<List<?>>> tasks =
+ new ConcurrentHashMap<>();
+ private long retainedTaskCount;
+ private long retainedIcebergBytes;
+ private long retainedPaimonBytes;
+ private long reservedIcebergBytes;
+ private long reservedPaimonBytes;
+ private boolean invalidated;
+
+ /**
+ * Return the tasks for {@code key}, loading and publishing an
immutable result once per
+ * cache generation.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), tasks ->
Math.max(1, tasks.size()),
+ WeightBudget.TASK_COUNT, MAX_RETAINED_TASK_COUNT,
+ MAX_RETAINED_TASK_COUNT, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, retaining the loaded result only
when its weight fits
+ * within the cache generation's cumulative limit.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader,
+ ToLongFunction<List<T>> weigher, long maxRetainedWeight)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), weigher,
+ WeightBudget.TASK_COUNT, maxRetainedWeight,
maxRetainedWeight, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, optionally reserving an entry
allowance before running
+ * the loader. A reserving loader receives the remaining allowance so
it can stop producing
+ * a cache value early; other loaders are admitted atomically by their
actual weight.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, WeightedLoader<T> loader,
+ ToLongFunction<List<T>> weigher, WeightBudget weightBudget,
+ long maxEntryWeight, long maxRetainedWeight,
+ boolean reserveBeforeLoad) throws Exception {
+ CompletableFuture<List<?>> newLoad = new CompletableFuture<>();
+ CompletableFuture<List<?>> load;
+ boolean cacheable;
+ long reservedWeight = 0;
+ synchronized (this) {
Review Comment:
这个 synchronized(this) 保护的是 invalidated 标志 + tasks map +
权重记账这三者的原子性。并发来源：scan 的 off-thread pumps（streaming/分区批量模式的异步拆分线程）会并发访问同一个
statement cache（ConnectorStatementScopeImpl 的注释也说明 off-thread pumps 复用同一
scope）。锁内操作都很轻量（map putIfAbsent + 计数加减），loader 在锁外执行。
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1366,6 +1377,261 @@ public List<org.apache.iceberg.FileScanTask>
getAndClearIcebergRewriteFileScanTa
return tasks;
}
+ public ExternalScanTaskCache getExternalScanTaskCache() {
+ return externalScanTaskCache;
+ }
+
+ /**
+ * Release scan tasks at the end of one execution without closing reusable
prepared-statement
+ * state. Delayed scan work retains only the invalidated generation and
cannot repopulate this
+ * StatementContext.
+ */
+ public void clearExternalScanTasks() {
+ externalScanTaskCache.invalidate();
+ }
+
+ /**
+ * One execution generation of statement-scoped external scan tasks.
+ *
+ * <p>Scan nodes capture this object when they are constructed. Resetting
a prepared statement
+ * swaps the generation before invalidating the old one, so a delayed
asynchronous scan from
+ * the previous execution cannot insert tasks into the next execution's
cache.
+ */
+ public static final class ExternalScanTaskCache {
+ /** Maximum number of connector tasks retained by one statement cache
generation. */
+ public static final long MAX_RETAINED_TASK_COUNT = 10_000;
+
+ /** Loads a value using the weight atomically reserved for this cache
entry. */
+ @FunctionalInterface
+ public interface WeightedLoader<T> {
+ List<T> load(long reservedWeight) throws Exception;
+ }
+
+ /** Independent cumulative budgets used by task-count and
serialized-byte retention. */
+ public enum WeightBudget {
+ TASK_COUNT,
+ ICEBERG_SERIALIZED_BYTES,
+ PAIMON_SERIALIZED_BYTES
+ }
+
+ private final Map<ExternalScanTaskCacheKey<?>,
CompletableFuture<List<?>>> tasks =
+ new ConcurrentHashMap<>();
+ private long retainedTaskCount;
+ private long retainedIcebergBytes;
+ private long retainedPaimonBytes;
+ private long reservedIcebergBytes;
+ private long reservedPaimonBytes;
+ private boolean invalidated;
+
+ /**
+ * Return the tasks for {@code key}, loading and publishing an
immutable result once per
+ * cache generation.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), tasks ->
Math.max(1, tasks.size()),
+ WeightBudget.TASK_COUNT, MAX_RETAINED_TASK_COUNT,
+ MAX_RETAINED_TASK_COUNT, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, retaining the loaded result only
when its weight fits
+ * within the cache generation's cumulative limit.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader,
+ ToLongFunction<List<T>> weigher, long maxRetainedWeight)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), weigher,
+ WeightBudget.TASK_COUNT, maxRetainedWeight,
maxRetainedWeight, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, optionally reserving an entry
allowance before running
+ * the loader. A reserving loader receives the remaining allowance so
it can stop producing
+ * a cache value early; other loaders are admitted atomically by their
actual weight.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, WeightedLoader<T> loader,
+ ToLongFunction<List<T>> weigher, WeightBudget weightBudget,
+ long maxEntryWeight, long maxRetainedWeight,
+ boolean reserveBeforeLoad) throws Exception {
+ CompletableFuture<List<?>> newLoad = new CompletableFuture<>();
+ CompletableFuture<List<?>> load;
+ boolean cacheable;
Review Comment:
invalidated 标志用于 generation 失效后的隔离：invalidate() 置位后，延迟的异步 scan 线程即使再调用
getOrLoad 也不会插入新 entry（cacheable=false 直接走 loader），保证旧执行的延迟 worker 不能污染下一个
prepared 执行。
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -1366,6 +1377,261 @@ public List<org.apache.iceberg.FileScanTask>
getAndClearIcebergRewriteFileScanTa
return tasks;
}
+ public ExternalScanTaskCache getExternalScanTaskCache() {
+ return externalScanTaskCache;
+ }
+
+ /**
+ * Release scan tasks at the end of one execution without closing reusable
prepared-statement
+ * state. Delayed scan work retains only the invalidated generation and
cannot repopulate this
+ * StatementContext.
+ */
+ public void clearExternalScanTasks() {
+ externalScanTaskCache.invalidate();
+ }
+
+ /**
+ * One execution generation of statement-scoped external scan tasks.
+ *
+ * <p>Scan nodes capture this object when they are constructed. Resetting
a prepared statement
+ * swaps the generation before invalidating the old one, so a delayed
asynchronous scan from
+ * the previous execution cannot insert tasks into the next execution's
cache.
+ */
+ public static final class ExternalScanTaskCache {
+ /** Maximum number of connector tasks retained by one statement cache
generation. */
+ public static final long MAX_RETAINED_TASK_COUNT = 10_000;
+
+ /** Loads a value using the weight atomically reserved for this cache
entry. */
+ @FunctionalInterface
+ public interface WeightedLoader<T> {
+ List<T> load(long reservedWeight) throws Exception;
+ }
+
+ /** Independent cumulative budgets used by task-count and
serialized-byte retention. */
+ public enum WeightBudget {
+ TASK_COUNT,
+ ICEBERG_SERIALIZED_BYTES,
+ PAIMON_SERIALIZED_BYTES
+ }
+
+ private final Map<ExternalScanTaskCacheKey<?>,
CompletableFuture<List<?>>> tasks =
+ new ConcurrentHashMap<>();
+ private long retainedTaskCount;
+ private long retainedIcebergBytes;
+ private long retainedPaimonBytes;
+ private long reservedIcebergBytes;
+ private long reservedPaimonBytes;
+ private boolean invalidated;
+
+ /**
+ * Return the tasks for {@code key}, loading and publishing an
immutable result once per
+ * cache generation.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), tasks ->
Math.max(1, tasks.size()),
+ WeightBudget.TASK_COUNT, MAX_RETAINED_TASK_COUNT,
+ MAX_RETAINED_TASK_COUNT, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, retaining the loaded result only
when its weight fits
+ * within the cache generation's cumulative limit.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader,
+ ToLongFunction<List<T>> weigher, long maxRetainedWeight)
throws Exception {
+ return getOrLoad(key, ignored -> loader.call(), weigher,
+ WeightBudget.TASK_COUNT, maxRetainedWeight,
maxRetainedWeight, false);
+ }
+
+ /**
+ * Return the tasks for {@code key}, optionally reserving an entry
allowance before running
+ * the loader. A reserving loader receives the remaining allowance so
it can stop producing
+ * a cache value early; other loaders are admitted atomically by their
actual weight.
+ */
+ @SuppressWarnings("unchecked")
+ public <T> List<T> getOrLoad(
+ ExternalScanTaskCacheKey<T> key, WeightedLoader<T> loader,
+ ToLongFunction<List<T>> weigher, WeightBudget weightBudget,
+ long maxEntryWeight, long maxRetainedWeight,
+ boolean reserveBeforeLoad) throws Exception {
+ CompletableFuture<List<?>> newLoad = new CompletableFuture<>();
+ CompletableFuture<List<?>> load;
+ boolean cacheable;
+ long reservedWeight = 0;
+ synchronized (this) {
+ cacheable = !invalidated;
+ if (cacheable) {
+ load = tasks.putIfAbsent(key, newLoad);
+ if (load == null && reserveBeforeLoad) {
+ long retainedWeight = retainedWeight(weightBudget);
+ long alreadyReservedWeight =
reservedWeight(weightBudget);
+ long availableWeight = Math.max(
+ 0, maxRetainedWeight - retainedWeight -
alreadyReservedWeight);
+ reservedWeight = Math.min(maxEntryWeight,
availableWeight);
+ addReservedWeight(weightBudget, reservedWeight);
+ }
+ } else {
+ load = null;
+ }
+ }
+ if (!cacheable) {
+ return immutableCopy(loader.load(0));
+ }
+ if (load == null) {
+ List<T> loadedTasks;
+ long weight;
+ try {
+ loadedTasks = loader.load(reserveBeforeLoad ?
reservedWeight : maxEntryWeight);
+ weight = weigher.applyAsLong(loadedTasks);
+ } catch (Exception | Error throwable) {
+ synchronized (this) {
+ if (reserveBeforeLoad) {
+ releaseReservation(weightBudget, reservedWeight);
+ }
+ tasks.remove(key, newLoad);
+ }
+ newLoad.completeExceptionally(throwable);
+ throw throwable;
+ }
+ boolean mayRetain;
+ synchronized (this) {
+ long availableWeight = Math.max(
+ 0, maxRetainedWeight -
retainedWeight(weightBudget));
+ mayRetain = !invalidated && weight <= availableWeight
+ && (!reserveBeforeLoad || weight <=
reservedWeight);
+ }
+ List<T> retainedCopy = null;
+ boolean reservationReleased = false;
+ boolean retainedCommitted = false;
+ try {
+ if (mayRetain) {
+ retainedCopy = immutableCopy(loadedTasks);
+ }
+ synchronized (this) {
+ if (reserveBeforeLoad) {
+ releaseReservation(weightBudget, reservedWeight);
+ reservationReleased = true;
+ }
+ long availableWeight = Math.max(
+ 0, maxRetainedWeight -
retainedWeight(weightBudget));
+ retainedCommitted = mayRetain && !invalidated &&
weight <= availableWeight;
+ if (retainedCommitted) {
+ addRetainedWeight(weightBudget, weight);
+ } else {
+ tasks.remove(key, newLoad);
+ }
+ }
+ List<T> result = retainedCommitted ? retainedCopy :
loadedTasks;
+ newLoad.complete(result);
+ return result;
+ } catch (Exception | Error throwable) {
+ synchronized (this) {
+ if (retainedCommitted) {
+ addRetainedWeight(weightBudget, -weight);
+ }
+ if (reserveBeforeLoad && !reservationReleased) {
+ releaseReservation(weightBudget, reservedWeight);
+ }
+ tasks.remove(key, newLoad);
+ }
+ newLoad.completeExceptionally(throwable);
+ throw throwable;
+ }
+ }
+ try {
+ return (List<T>) load.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ throw (Error) cause;
+ }
+ }
+
+ private synchronized void invalidate() {
+ invalidated = true;
+ tasks.clear();
+ retainedTaskCount = 0;
+ retainedIcebergBytes = 0;
+ retainedPaimonBytes = 0;
+ reservedIcebergBytes = 0;
+ reservedPaimonBytes = 0;
+ }
+
+ private void releaseReservation(WeightBudget weightBudget, long
reservedWeight) {
+ addReservedWeight(weightBudget, -reservedWeight);
+ }
+
+ private long retainedWeight(WeightBudget weightBudget) {
+ switch (weightBudget) {
+ case TASK_COUNT:
+ return retainedTaskCount;
+ case ICEBERG_SERIALIZED_BYTES:
+ return retainedIcebergBytes;
+ case PAIMON_SERIALIZED_BYTES:
+ return retainedPaimonBytes;
+ default:
+ throw new IllegalStateException("Unknown external scan
task weight budget: " + weightBudget);
+ }
+ }
+
+ private long reservedWeight(WeightBudget weightBudget) {
+ switch (weightBudget) {
+ case TASK_COUNT:
+ return 0;
+ case ICEBERG_SERIALIZED_BYTES:
+ return reservedIcebergBytes;
+ case PAIMON_SERIALIZED_BYTES:
+ return reservedPaimonBytes;
+ default:
+ throw new IllegalStateException("Unknown external scan
task weight budget: " + weightBudget);
+ }
+ }
+
+ private void addRetainedWeight(WeightBudget weightBudget, long weight)
{
+ switch (weightBudget) {
+ case TASK_COUNT:
+ retainedTaskCount += weight;
+ break;
+ case ICEBERG_SERIALIZED_BYTES:
+ retainedIcebergBytes += weight;
+ break;
+ case PAIMON_SERIALIZED_BYTES:
+ retainedPaimonBytes += weight;
+ break;
+ default:
+ throw new IllegalStateException("Unknown external scan
task weight budget: " + weightBudget);
+ }
+ }
+
+ private void addReservedWeight(WeightBudget weightBudget, long weight)
{
+ switch (weightBudget) {
+ case TASK_COUNT:
+ break;
+ case ICEBERG_SERIALIZED_BYTES:
+ reservedIcebergBytes += weight;
+ break;
+ case PAIMON_SERIALIZED_BYTES:
+ reservedPaimonBytes += weight;
+ break;
+ default:
+ throw new IllegalStateException("Unknown external scan
task weight budget: " + weightBudget);
+ }
+ }
+
+ private static <T> List<T> immutableCopy(List<T> loadedTasks) {
+ return Collections.unmodifiableList(new ArrayList<>(loadedTasks));
+ }
Review Comment:
immutableCopy 是为了缓存值不被调用方修改：缓存的 List 会被同一语句内的多个 alias 共享，而 Hudi/Hive 的 task
对象是可变的，调用方（如 split 生成）可能改写列表结构。不可变包装 + 调用方自行拷贝（copyHudiSplit）保证了共享安全。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -887,9 +907,103 @@ private CloseableIterable<FileScanTask>
splitFiles(TableScan scan) {
} catch (Exception e) {
throw new RuntimeException("Failed to materialize file scan
tasks", e);
}
+ return fileScanTaskList;
+ }
- targetSplitSize = determineTargetFileSplitSize(fileScanTaskList);
- return
TableScanUtil.splitFiles(CloseableIterable.withNoopClose(fileScanTaskList),
targetSplitSize);
+ @VisibleForTesting
+ List<FileScanTask> getOrPlanFileScanTasks(TableScan scan,
Supplier<List<FileScanTask>> planner) {
+ try {
+ return
getOrPlanSerializedIcebergTasks(createFileScanTaskCacheKey(scan), planner::get);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to plan Iceberg file scan
tasks", e);
+ }
+ }
+
+ private IcebergScanTaskCacheKey<IcebergSerializedScanTask<FileScanTask>>
createFileScanTaskCacheKey(
+ TableScan scan) {
+ Snapshot snapshot = scan.snapshot();
+ return new IcebergScanTaskCacheKey<>(
+ source.getCatalog().getId(),
+ source.getTargetTable().getId(),
+ snapshot == null ? null : snapshot.snapshotId(),
+ scan.schema().schemaId(),
+ scan.filter(),
+ scan.isCaseSensitive(),
+ FileScanTask.class.getName());
+ }
+
+ @VisibleForTesting
+ List<PositionDeletesScanTask> getOrPlanPositionDeleteTasks(
+ BatchScan scan, Callable<List<PositionDeletesScanTask>> planner)
throws Exception {
+ Snapshot snapshot = scan.snapshot();
+
IcebergScanTaskCacheKey<IcebergSerializedScanTask<PositionDeletesScanTask>>
cacheKey
+ = new IcebergScanTaskCacheKey<>(
+ source.getCatalog().getId(),
+ source.getTargetTable().getId(),
+ snapshot == null ? null : snapshot.snapshotId(),
+ scan.schema().schemaId(),
+ scan.filter(),
+ scan.isCaseSensitive(),
+ PositionDeletesScanTask.class.getName());
+ return getOrPlanSerializedIcebergTasks(cacheKey, planner);
+ }
+
+ private <T> List<T> getOrPlanSerializedIcebergTasks(
+ IcebergScanTaskCacheKey<IcebergSerializedScanTask<T>> cacheKey,
+ Callable<List<T>> planner) throws Exception {
+ try {
+ List<IcebergSerializedScanTask<T>> serializedTasks =
getOrLoadExternalScanTasks(
+ cacheKey,
+ remainingBytes ->
serializeIcebergTasksWithinLimit(planner.call(), remainingBytes),
+ IcebergScanNode::serializedIcebergTaskBytes,
+
StatementContext.ExternalScanTaskCache.WeightBudget.ICEBERG_SERIALIZED_BYTES,
+ maxRetainedSerializedTaskBytes,
maxRetainedSerializedTaskBytes, true);
+ return serializedTasks.stream()
+ .map(IcebergSerializedScanTask::deserialize)
+ .collect(Collectors.toList());
+ } catch (IcebergTaskCacheLimitException e) {
+ List<T> plannedTasks = e.takePlannedTasks();
+ return plannedTasks == null ? planner.call() : plannedTasks;
+ }
+ }
+
+ private <T> List<IcebergSerializedScanTask<T>>
serializeIcebergTasksWithinLimit(
Review Comment:
是的，两者结构重复（逐 task 序列化 + 累计预算检查 + 超限异常回退）。抽公共工具到 fe-common 可作为 follow-up，本 PR
先保持现状以控制改动面。
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -320,8 +324,31 @@ private void
getFileSplitByPartitions(HiveExternalMetaCache cache, List<HivePart
}
} else {
boolean withCache = Config.max_external_file_cache_num > 0;
- fileCaches = cache.getFilesByPartitions(partitions, withCache,
partitions.size() > 1,
- directoryLister, hmsTable);
+ if (isBatchMode || !withCache) {
+ // Batch mode bounds FE memory by retaining only the
partitions currently in flight.
+ // Keeping every completed partition in the statement cache
would materialize the
+ // full scan again and defeat that bound. When the global file
cache is disabled,
+ // statement retention must not become an uncapped replacement
for that memory fence.
+ fileCaches = cache.getFilesByPartitions(partitions, withCache,
partitions.size() > 1,
+ directoryLister, hmsTable);
+ } else {
+ List<FileCacheValue> currentFileCaches =
cache.getFilesByPartitions(partitions, true,
+ partitions.size() > 1, directoryLister, hmsTable);
+ HiveFileScanTaskCacheKey cacheKey = new
HiveFileScanTaskCacheKey(
+ hmsTable.getCatalog().getId(), hmsTable.getId(),
partitions,
+
cache.getFileCacheInvalidationGeneration(hmsTable.getCatalog().getId()),
currentFileCaches);
+ try {
Review Comment:
不是重复缓存：这是两层。第一层是全局 file cache（HiveExternalMetaCache.fileEntry，跨 statement
共享、带 generation 版本），第二层是 statement 缓存（同一语句内重复 relation 复用已 plan 的
FileCacheValue，避免同一语句内对同一分区重复列目录+拆分）。statement 层的 key 包含全局层的
generation，所以全局层换代时 statement 层自动 miss。
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
