github-actions[bot] commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3900705618
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -494,22 +543,52 @@ private List<HudiSplit> planPartitionSplits(HivePartition
partition) throws IOEx
private void getPartitionsSplits(List<HivePartition> partitions,
List<Split> splits) {
Executor executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
- CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
+ ListingFsViewOwner createdOwner = new ListingFsViewOwner(fsViewLease);
+ ListingFsViewOwner owner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ owner =
statementContext.getOrRegisterStatementResource(listingFsViewResourceKey, () ->
createdOwner);
+ if (owner != createdOwner) {
+ throw new IllegalStateException("Hudi listing owner was
registered twice");
+ }
+ } catch (RuntimeException e) {
+ createdOwner.discardBeforeSubmission();
+ throw e;
+ }
+ }
+ // The owner now releases the exact fs-view generation after every
accepted task terminates.
+ if (!fsViewReleased.compareAndSet(false, true)) {
+ owner.discardBeforeSubmission();
+ throw new IllegalStateException("Hudi filesystem-view lease has
already been released");
+ }
AtomicReference<Throwable> throwable = new AtomicReference<>();
+ RuntimeException submissionFailure = null;
long startTime = System.currentTimeMillis();
- partitions.forEach(partition -> executor.execute(() -> {
+ for (HivePartition partition : partitions) {
Review Comment:
[P2] Stop submitting partitions after listing cancellation
Once statement cleanup sets this owner's `stopping` flag, every later
`owner.track(task)` cancels the task synchronously, but this loop still calls
`executor.execute(task)` and continues through all remaining partitions. Those
already-cancelled `FutureTask`s occupy the shared bounded file-listing queue
until workers drain them; if that queue is full, its rejection policy can also
keep the cancelled planner in `offer()` for up to ten seconds. With a raised
batch threshold or several cancelled scans this can delay or reject unrelated
listings. Break the submission loop when cancellation is visible (and handle
the track/execute race without enqueueing an already-cancelled task), then add
a close-during-submission regression.
--
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]
