924060929 commented on code in PR #69991:
URL: https://github.com/apache/doris/pull/69991#discussion_r4444601716
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -476,11 +496,12 @@ public void createScanRangeLocations() throws
UserException {
admissionResult = fileCacheAdmissionCheck();
}
- if (isBatchMode()) {
Review Comment:
Addressed in abcbb5cbf0c564be54c56e3925b8617e6ee2f3fa. Hudi and MaxCompute
now call the common batch-mode eligibility gate before their connector-specific
planning. Applicable scan-count admission rules and EXPLAIN VERBOSE use
complete split metadata. Ordinary EXPLAIN still describes the actual lazy
execution plan, preserving the Arrow Flight batch-path check. The related FE
suites cover the connector gates, exact admission/verbose metadata, and
ordinary EXPLAIN batching.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -2723,21 +2726,43 @@ public Map<String, Action> getForceEagerAggHintMap() {
@VariableMgr.VarAttr(name = ICEBERG_WRITE_TARGET_FILE_SIZE_BYTES,
needForward = true)
public long icebergWriteTargetFileSizeBytes = 0L;
+ @VariableMgr.VarAttr(name = SPLIT_BATCH_SIZE, checker =
"checkSplitBatchSize", needForward = true)
+ public int splitBatchSize = 8;
+
+ @VariableMgr.VarAttr(name = SPLIT_QUEUE_CAPACITY, checker =
"checkSplitQueueCapacity", needForward = true)
+ public int splitQueueCapacity = 1024;
+
+ @VariableMgr.VarAttr(name = SPLIT_PARTITION_BATCH_SIZE, checker =
"checkSplitPartitionBatchSize",
+ needForward = true)
+ public int splitPartitionBatchSize = 16;
+
+ public void checkSplitBatchSize(String value) throws Exception {
+ checkFieldValue(SPLIT_BATCH_SIZE, 1, value);
+ }
+
+ public void checkSplitQueueCapacity(String value) throws Exception {
+ checkFieldValue(SPLIT_QUEUE_CAPACITY, 1, value);
+ }
+
+ public void checkSplitPartitionBatchSize(String value) throws Exception {
+ checkFieldValue(SPLIT_PARTITION_BATCH_SIZE, 1, value);
+ }
+
@VariableMgr.VarAttr(
name = NUM_PARTITIONS_IN_BATCH_MODE,
fuzzy = true,
description = {"如果分区数量超过阈值，BE 将通过 batch 方式获取 scan ranges。作用于
Hive、Hudi、MaxCompute 表。",
"If the number of partitions exceeds the threshold, scan
ranges will be got through batch mode."},
needForward = true)
- public int numPartitionsInBatchMode = 1024;
+ public int numPartitionsInBatchMode = 1;
Review Comment:
Addressed in abcbb5cbf0c564be54c56e3925b8617e6ee2f3fa. MaxCompute now groups
partitions using the dedicated split_partition_batch_size (default 16),
independently of the threshold that enables batch mode. Each scan submits only
one continuation at a time, instead of submitting every group together.
Initial/continuation rejection and session failure publish the error and finish
the schedule. Tests cover 33 partitions grouped as 16/16/1, the single queued
continuation, both rejection paths, and cancellation.
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -1117,7 +1117,8 @@ public TFetchSplitBatchResult
fetchSplitBatch(TFetchSplitBatchRequest request) t
try {
List<TScanRangeLocations> locations =
splitSource.getNextBatch(request.getMaxNumSplits());
result.setSplits(locations);
- result.status = new TStatus(TStatusCode.OK);
+ result.status = new TStatus(locations.isEmpty() &&
!splitSource.isLastBatch()
+ ? TStatusCode.INCOMPLETE : TStatusCode.OK);
return result;
} catch (Exception e) {
Review Comment:
I checked both directions against the supported BE-first upgrade sequence
and the release-branch implementations. All BEs are upgraded before FEs, so an
old BE does not receive the new FE retry status in that sequence. With a new BE
and old FE, the old SplitSource still returns its existing INTERNAL_ERROR on an
empty-wait timeout; the old connector already treated any non-OK status as a
failure. That failure is therefore not introduced by INCOMPLETE. Under the
supported upgrade contract, no extra retry compatibility path is required for
an intermediate unreleased 4.2 build. The source-close extension in
abcbb5cbf0c564be54c56e3925b8617e6ee2f3fa is separately capability-gated by an
optional TSplitSource.supports_close field: a new BE sends no close request to
an old FE, even if the scan stops before its first fetch.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/SplitSource.java:
##########
@@ -67,40 +67,46 @@ public long getUniqueId() {
/**
* Get the next batch of file splits. If there's no more split, return
empty list.
*/
- public List<TScanRangeLocations> getNextBatch(int maxBatchSize) throws
UserException {
+ public boolean isLastBatch() {
+ return isLastBatch.get();
+ }
+
+ public synchronized List<TScanRangeLocations> getNextBatch(int
maxBatchSize) throws UserException {
+ Preconditions.checkArgument(maxBatchSize > 0);
if (isLastBatch.get()) {
return Collections.emptyList();
}
List<TScanRangeLocations> scanRanges =
Lists.newArrayListWithExpectedSize(maxBatchSize);
- long startTime = System.currentTimeMillis();
- while (scanRanges.size() < maxBatchSize && System.currentTimeMillis()
- startTime < maxWaitTime) {
- BlockingQueue<Collection<TScanRangeLocations>> splits =
splitAssignment.getAssignedSplits(backend);
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(maxWaitTime);
+ while (scanRanges.size() < maxBatchSize) {
Review Comment:
Addressed in abcbb5cbf0c564be54c56e3925b8617e6ee2f3fa. The session-variable
checker rejects fetch_splits_max_wait_time_ms <= 0. Planning also checks the
value before starting split generation, and SplitSource enforces the same
constructor precondition for programmatic/persisted values. The wait uses
elapsed-time arithmetic rather than adding a deadline that could overflow.
Tests cover zero/negative values and positive waits; a live empty batch still
waits before returning INCOMPLETE.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/SplitAssignment.java:
##########
@@ -113,15 +126,24 @@ private void appendBatch(Multimap<Backend, Split> batch)
throws UserException {
locations.add(splitToScanRange.getScanRange(backend,
locationProperties, split, pathPartitionKeys,
fileCacheAdmission));
}
- while (needMoreSplit()) {
- BlockingQueue<Collection<TScanRangeLocations>> queue =
- assignment.computeIfAbsent(backend, be -> new
LinkedBlockingQueue<>(10000));
- try {
- if (queue.offer(locations, 100, TimeUnit.MILLISECONDS)) {
- break;
+ BlockingQueue<TScanRangeLocations> queue =
assignment.computeIfAbsent(
+ backend.getId(), be -> new
LinkedBlockingQueue<>(queueCapacity));
+ for (TScanRangeLocations location : locations) {
+ while (needMoreSplit()) {
+ try {
Review Comment:
Addressed in abcbb5cbf0c564be54c56e3925b8617e6ee2f3fa. The scan operator
owner now releases its shared remote source after Base::close requests scanner
stop and releases ScannerContext's transfer lock; no RPC runs inside scanner
try_stop. FE marks only that backend consumer closed, clears its bounded queue,
skips subsequent publication to it, and clears an offer that races closure.
Other backends keep consuming; closing the last one stops the assignment.
Optional descriptor capability gates the new close request for old FEs. The
two-backend FE test forces the full-queue/close/offer race and verifies the
other backend still receives its split. BE tests cover retry, idempotent owner
close, old-FE capability absence, and an in-flight fetch returning after
closure, alongside the existing native V2 queue/cancellation suites.
--
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]
