github-actions[bot] commented on code in PR #69991:
URL: https://github.com/apache/doris/pull/69991#discussion_r4443306341
##########
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:
[P1] Reject or handle non-positive `fetch_splits_max_wait_time_ms` before
enabling the retry protocol. With a value of 0 (or negative),
`getNextBatch()`'s deadline is already expired, so every live call returns an
empty `INCOMPLETE` response immediately. The BE's new unbounded `INCOMPLETE`
loop then floods the FE with RPCs and burns CPU until split production
finishes; the previous implementation surfaced one timeout error instead. Add
validation or a bounded backoff/stop condition for this setting.
##########
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:
[P1] Release or mark a backend's split source when its scanner stops at a
local LIMIT. `appendBatch()` now blocks on each finite per-backend queue while
`needMoreSplit()` is globally true, but the legacy BE path only passes a local
`scanner_stop` and never tells FE that this backend will not consume again.
After that queue fills, the producer can remain in the offer loop and stop
publishing to other backends, whose fetches stay `INCOMPLETE` indefinitely. Add
per-source cancellation/queue closure or make publication skip stopped
consumers.
##########
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:
[P1] Preserve FE/BE rolling-upgrade compatibility for the new `INCOMPLETE`
response. This FE now returns `TStatusCode.INCOMPLETE` whenever a live
`SplitSource` is temporarily empty, but the previous BE connector treats every
non-OK status as an error and aborts the query. A new BE talking to an old FE
has the reverse problem because the old FE turns the same temporary empty
condition into `INTERNAL_ERROR`, which the new connector also returns. A
mixed-version FE/BE pair will therefore fail lazy external scans during the
normal wait window. Add a compatibility path or gate the status change until
both sides understand the retry contract.
##########
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:
[P1] Apply the new exact-planning guard to every batch-capable file scan.
`createScanRangeLocations()` dispatches to HudiScanNode.isBatchMode() and
MaxComputeScanNode.isBatchMode(), but those overrides never call
`isBatchModeEnabled()`. With the changed defaults of one partition/file, a Hudi
or MaxCompute query enters batch mode during EXPLAIN or when a
partition_num/tablet_num SQL block rule is active; FileQueryScanNode then
reports `numApproximateSplits()` (Hudi can still be zero while async listing
has not populated `numSplitsPerPartition`, and MaxCompute uses the partition
count) to StmtExecutor.checkBlockRulesByScan. This can show approximate EXPLAIN
details and admit or reject a query using the wrong split count. Make the
common guard apply to these overrides as well and add equivalent coverage for
each batch-capable subclass.
##########
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:
[P1] Keep MaxCompute's partition batching independent of the batch-mode
threshold. `MaxComputeScanNode.startSplit()` uses
`getNumPartitionsInBatchMode()` as `batchNumPartitions`, so this new default of
1 submits one `TableBatchReadSession` task per selected partition. Large tables
can exceed the fixed `scheduleExecutor` queue; rejected submissions leave
`numFinishedPartitions` short of the total, and the split source fails instead
of completing. Use a dedicated bounded batch size (or preserve a larger
MaxCompute default) and handle rejected tasks with terminal accounting.
--
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]
