924060929 commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4448344644
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java:
##########
@@ -227,73 +230,150 @@ public List<Split> getSplits(int numBackends) throws
UserException {
return allFiles;
} catch (Throwable t) {
LOG.warn("get file split failed for table: {}",
hmsTable.getName(), t);
- throw new UserException(
- "get file split failed for table: " + hmsTable.getName() + ",
err: " + Util.getRootCauseMessage(t),
- t);
+ throw createSplitPlanningException(t);
}
}
+ private UserException createSplitPlanningException(Throwable cause) {
+ return new UserException("get file split failed for table: " +
hmsTable.getName()
+ + ", err: " + Util.getRootCauseMessage(cause), cause);
+ }
+
+ @Override
+ protected SplitAssignment createSplitAssignment(Map<String, String>
properties,
+ List<String> partitionKeys, boolean admission) {
+ return new SplitAssignment(backendPolicy, this,
this::splitToScanRange, properties,
+ partitionKeys, admission, sessionVariable.getSplitBatchSize(),
+ sessionVariable.getSplitQueueCapacity());
+ }
+
@Override
public void startSplit(int numBackends) {
+ ConnectContext context = ConnectContext.get();
+ StatementContext statement = context == null ? null :
context.getStatementContext();
+ if (statement != null) {
+ StmtExecutor executor = context.getExecutor();
+ statement.getOrRegisterStatementResource(this, () -> (Closeable)
() -> {
+ // A deferred Flight query still fetches splits after
statement handling ends.
+ // Its coordinator owns cleanup; abandoned planning and
EXPLAIN stop here.
+ if (executor == null || !executor.isDeferredForArrowFlight()) {
+ stop();
+ }
+ });
+ }
if (prunedPartitions.isEmpty()) {
splitAssignment.finishSchedule();
return;
}
HiveExternalMetaCache cache = Env.getCurrentEnv().getExtMetaCacheMgr()
.hive(hmsTable.getCatalog().getId());
- Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
- String bindBrokerName = hmsTable.getCatalog().bindBrokerName();
- AtomicInteger numFinishedPartitions = new AtomicInteger(0);
- long startTime = System.currentTimeMillis();
- CompletableFuture.runAsync(() -> {
- for (HivePartition partition : prunedPartitions) {
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- try {
- splittersOnFlight.acquire();
- CompletableFuture.runAsync(() -> {
+ Executor executor = getSplitTaskExecutor();
+ String broker = hmsTable.getCatalog().bindBrokerName();
+ // Without the file cache, the eager path listed partitions
concurrently. Keep that overlap.
+ boolean parallelListing = !fileCountBatch ||
Config.max_external_file_cache_num <= 0;
+ int producers = parallelListing
+ ? Math.min(prunedPartitions.size(),
Config.max_external_cache_loader_thread_pool_size) : 1;
+ AtomicInteger nextPartition = new AtomicInteger();
+ AtomicInteger remaining = new AtomicInteger(producers);
+ long started = System.currentTimeMillis();
+ for (int i = 0; i < producers; i++) {
+ new SplitProducer(splitAssignment, executor,
+ createSplitBatchSource(cache, nextPartition, broker,
numBackends), failure -> {
try {
- List<Split> allFiles = Lists.newArrayList();
- getFileSplitByPartitions(
- cache,
Collections.singletonList(partition), allFiles, bindBrokerName,
- numBackends, true);
- if (allFiles.size() > numSplitsPerPartition.get())
{
- numSplitsPerPartition.set(allFiles.size());
+ if (failure != null) {
+
splitAssignment.setException(createSplitPlanningException(failure));
}
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(allFiles);
- }
- } catch (Exception e) {
- batchException.set(new
UserException(e.getMessage(), e));
} finally {
- splittersOnFlight.release();
- if (batchException.get() != null) {
Review Comment:
Fixed in 663c74d0f2062295cca9b2dbde4ae6b8d25224f2. The streaming source now
counts splits for its current selected partition across producer batches and
files. It publishes the existing global maximum at a yield and before advancing
partitions, then resets only at that boundary. There is no per-range atomic
update or eager materialization. The ten-split floor and int saturation remain.
Tests reproduced 17 expected versus 10 actual for a one-range batch, and 51
versus 87 when unequal partitions share a large batch; both now pass. Counts
remain approximate while unseen files are still being lazily generated. All 147
related FE tests and full FE packaging/Checkstyle pass locally.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2472,6 +2472,15 @@ public class Config extends ConfigBase {
@ConfField(mutable = false, masterOnly = false)
public static int max_external_cache_loader_thread_pool_size = 64;
+ @ConfField(description = "Workers for fair split production")
+ public static int external_scan_split_threads = 64;
+
Review Comment:
Fixed in 663c74d0f2062295cca9b2dbde4ae6b8d25224f2. All three settings opt
into a pure positive-integer config validator during both ordinary and custom
startup-file loading. Zero/negative values are rejected with the config name
before that invalid field is updated. The existing empty-entry default rule and
runtime immutability are preserved; unrelated runtime callbacks with side
effects are still not dispatched during startup. Scheduler constructor/group
preconditions remain for direct programmatic callers. Config tests cover both
startup paths, 0/-1, valid positive/whitespace values, previous-value
preservation and runtime update rejection. All 147 related FE tests and full FE
packaging/Checkstyle pass locally.
--
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]
