924060929 commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3801161576
##########
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（6ccc6d8eedd：DictionaryManager 的调度上下文现在走 production
cleanupScheduledContext 路径，关闭 StatementContext 并移除 worker 线程的
ConnectContext，DictionaryManagerTest 7/7 验证）。同意这个修复值得单独 PR 的价值，如果希望拆分，我可以把它从本
PR 拆出单独提交。
##########
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:
同意这个观察——Paimon 的 LimitedByteArrayOutputStream 和 Iceberg 的
serializeIcebergTaskWithinLimit 结构相同（逐 task 序列化 + 预算检查）。两者细节略有差异（Paimon 用
InstantiationUtil，Iceberg 用 SerializationUtil），抽到 fe-foundation/fe-common
的公共序列化工具可以作为 follow-up，避免在本 PR 继续膨胀。
##########
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（fbed4ac1092：TaskProcessor 外层 finally 关闭当前 StatementContext + 无条件移除
worker context；44a3365fd2a：MTMVTask 每个 refresh chunk 关闭自己的上下文）。如需拆分单独 PR 请告诉我。
##########
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:
clearExternalScanTasks() 放在 close() 里而不是 releasePlannerResources() 里，因为
releasePlannerResources 是 planner 阶段的资源（表锁等），可能在语句完成前被其它路径调用（例如 planner
失败重试时），此时 scan task 缓存仍需保留供重试复用；而 close() 是语句生命周期的终点，prepared statement 的每次
EXECUTE 也会走 resetConnectorStatementScope 单独清理。
##########
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:
MAX_RETAINED_TASK_COUNT=10000 是 statement 内重复 relation 场景的保守上限（Hive/Hudi 以
task 计数为单位的累计预算），基于单条语句内重复等价关系数量的经验值。改造成 session 变量是合理的 follow-up，当前先用常量避免
session 变量数量膨胀。
##########
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:
WeightBudget 分成三种（TASK_COUNT / ICEBERG_SERIALIZED_BYTES /
PAIMON_SERIALIZED_BYTES）是因为度量单位不同（个数 vs 字节）且需要跨 connector 隔离——一个 connector
的序列化字节预算不应被另一个 connector 的 entry 占用。抽象成单一的 retained/reserved 计数会失去这种隔离；如果抽象成
per-budget 的一对计数器则与现状等价。
--
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]
