morrySnow commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3800531493
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java:
##########
@@ -633,14 +635,73 @@ public static List<Column> parseSchema(RowType rowType,
List<String> primaryKeys
}
public static <T> String encodeObjectToString(T t) {
+ byte[] bytes = serializeObject(t);
+ return new String(BASE64_ENCODER.encode(bytes),
java.nio.charset.StandardCharsets.UTF_8);
+ }
+
+ public static <T> byte[] serializeObject(T object) {
+ try {
+ return InstantiationUtil.serializeObject(object);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static <T> Optional<byte[]> serializeObjectWithinLimit(T object,
long maxBytes) {
+ LimitedByteArrayOutputStream output = new
LimitedByteArrayOutputStream(maxBytes);
+ try (ObjectOutputStream objectOutput = new ObjectOutputStream(output))
{
+ objectOutput.writeObject(object);
+ objectOutput.flush();
+ return Optional.of(output.toByteArray());
+ } catch (SerializationSizeLimitException e) {
+ return Optional.empty();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static <T> T deserializeObject(byte[] bytes) {
try {
- byte[] bytes = InstantiationUtil.serializeObject(t);
- return new String(BASE64_ENCODER.encode(bytes),
java.nio.charset.StandardCharsets.UTF_8);
+ return InstantiationUtil.deserializeObject(bytes,
PaimonUtil.class.getClassLoader());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
+ private static final class LimitedByteArrayOutputStream extends
OutputStream {
Review Comment:
这种可以放到fe-foundation或者 fe-common中
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -910,6 +917,7 @@ protected void finalize() throws Throwable {
@Override
public void close() {
+ clearExternalScanTasks();
Review Comment:
这个为什么不放在 `releasePlannerResources` 里？
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java:
##########
@@ -82,6 +84,22 @@ private void runTask(AbstractTask task) {
task.runTask();
} catch (Exception e) {
log.warn("Execute task error, task id: {}", task.getTaskId(), e);
+ } finally {
Review Comment:
内存泄露的，用单独的PR修复
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -136,6 +142,46 @@ public FileQueryScanNode(PlanNodeId id, TupleDescriptor
desc, String planNodeNam
StatisticalType statisticalType, ScanContext scanContext, boolean
needCheckColumnPriv, SessionVariable sv) {
super(id, desc, planNodeName, statisticalType, scanContext,
needCheckColumnPriv);
this.sessionVariable = sv;
+ ConnectContext context = ConnectContext.get();
+ StatementContext statementContext = context == null ? null :
context.getStatementContext();
+ this.externalScanTaskCache = statementContext == null
+ ? null : statementContext.getExternalScanTaskCache();
+ }
+
+ protected <T> List<T> getOrLoadExternalScanTasks(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader) throws
Exception {
+ if (!sessionVariable.enableExternalScanTaskReuse ||
externalScanTaskCache == null) {
+ return loader.call();
+ }
+ return externalScanTaskCache.getOrLoad(key, loader);
+ }
+
+ protected <T> List<T> getOrLoadExternalScanTasks(
+ ExternalScanTaskCacheKey<T> key, Callable<List<T>> loader,
+ ToLongFunction<List<T>> weigher, long maxRetainedWeight) throws
Exception {
+ if (!sessionVariable.enableExternalScanTaskReuse ||
externalScanTaskCache == null) {
+ return loader.call();
+ }
+ return externalScanTaskCache.getOrLoad(key, loader, weigher,
maxRetainedWeight);
+ }
Review Comment:
没有用到的函数
##########
fe/fe-core/src/main/java/org/apache/doris/dictionary/DictionaryManager.java:
##########
@@ -453,129 +454,143 @@ public void dataLoad(ConnectContext ctx, Dictionary
dictionary, boolean adaptive
Thread.sleep(10);
}
- if (ctx == null) { // for run with scheduler, not by command.
+ boolean ownsContext = ctx == null;
Review Comment:
这个内存泄漏，单独提一个PR比较好呀
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java:
##########
@@ -980,6 +1017,7 @@ public String toString() {
@Data
public static class FileCacheValue {
private final List<HiveFileStatus> files = Lists.newArrayList();
+ private long cacheGeneration;
Review Comment:
增加注释，解释这个变量的意义
##########
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:
这个锁真的有必要吗？StatementContext什么时候会面对多线程？
##########
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;
Review Comment:
这个10000是怎么来的？需要改成可配置的吗？
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java:
##########
@@ -119,6 +121,8 @@ public class HiveExternalMetaCache extends
AbstractExternalMetaCache {
public static final String ERR_CACHE_INCONSISTENCY =
"ERR_CACHE_INCONSISTENCY: ";
private final ExecutorService fileListingExecutor;
+ private final Map<Long, AtomicLong> fileCacheInvalidationGenerations = new
ConcurrentHashMap<>();
+ private final Map<Long, AtomicLong> fileCacheValueGenerations = new
ConcurrentHashMap<>();
Review Comment:
增加注释，解释这俩是干什么用的，在什么情况下需要增加generation，在什么情况下不能增加generation
##########
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;
Review Comment:
感觉不用具体到每一种类型，抽象成 retrain 和 reserved 就可以了
##########
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:
看起来hive这段代码是没必要的，会导致重复cache？
##########
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:
为啥要引入这个变量？
##########
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:
为什么需要copy一下？
##########
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:
这个和paimonUtils中的代码其实是重复的
--
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]
