github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3821311163
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -904,11 +909,56 @@ public synchronized void releasePlannerResources() {
}
}
+ /**
+ * Returns one closeable resource per statement key and closes it when
this statement is closed.
+ * The supplier is invoked at most once for a key. This is intentionally
independent from planner locks,
+ * whose lifetime ends at the end of Nereids planning.
+ */
+ @SuppressWarnings("unchecked")
+ public synchronized <T extends Closeable> T getOrRegisterStatementResource(
+ Object resourceKey, java.util.function.Supplier<T> supplier) {
+ if (statementResourcesClosed) {
Review Comment:
[P1] Keep prepared-statement resources execution-scoped
COM_STMT_PREPARE goes through ConnectProcessor, whose finally closes the
same StatementContext that PrepareCommand stores for reuse. COM_STMT_EXECUTE
later reinstalls that context, so the first Iceberg lease or batch-Hudi owner
registration reaches this branch with statementResourcesClosed=true and fails
the query. Give each EXECUTE a fresh/resettable resource generation and close
that generation after execution; do not permanently seal the reusable prepared
context at PREPARE completion. Add protocol-lifecycle tests for prepared
Iceberg and batch Hudi queries.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalMetaCache.java:
##########
@@ -92,7 +96,8 @@ public IcebergExternalMetaCache(ExecutorService
refreshExecutor) {
super(ENGINE, refreshExecutor);
tableEntry = registerEntry(MetaCacheEntryDef.of(ENTRY_TABLE,
NameMapping.class, IcebergTableCacheValue.class,
this::loadTableCacheValue, defaultEntryCacheSpec(),
- MetaCacheEntryInvalidation.forNameMapping(nameMapping ->
nameMapping)));
+ false, MetaCacheEntryInvalidation.forNameMapping(nameMapping
-> nameMapping),
Review Comment:
[P1] Preserve freshness for hot Iceberg entries
Passing false removes this entry's previous refresh-after-write path. The
remaining 24-hour policy expires after access, so a continuously queried table
never reloads; unlike the Hudi wrapper, statementLease() performs no foreground
refresh, and IcebergTableCacheValue memoizes its latest-snapshot projection. An
external T2 commit after a hot T1 load can therefore remain invisible
indefinitely. Add an ownership-safe background replacement or foreground
refresh/reload, plus a hot-cache T1/T2 regression test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -558,53 +592,170 @@ private void initPrunedPartitions() throws UserException
{
public void startSplit(int numBackends) {
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
+ releaseFsViewOnce();
return;
}
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
+ acquireFsView();
ExecutorService scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
+ Executor producerExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
+ BatchFsViewOwner createdOwner = new BatchFsViewOwner(splitAssignment,
fsViewLease);
+ BatchFsViewOwner batchOwner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ batchOwner = statementContext.getOrRegisterStatementResource(
Review Comment:
[P1] Keep this owner alive through Arrow DoGet
Remote Arrow queries deliberately defer Coordinator.close() after
GetFlightInfo because the BE fetches lazy batch splits during the later DoGet
phase. ConnectProcessor still closes StatementContext immediately, so this
resource's close() stops SplitAssignment; subsequent getAssignedSplits() treats
the stream as finished and can omit partitions that DoGet has not consumed.
Transfer this owner to the deferred Arrow executor and release it from
finalizeArrowFlightQuery (with immediate cleanup on GetFlightInfo failure), and
cover slow multi-partition batch fetching.
--
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]
