github-actions[bot] commented on code in PR #66530:
URL: https://github.com/apache/doris/pull/66530#discussion_r3850860865
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1923,46 +1915,48 @@ public void startSplit(int numBackends) {
AtomicReference<UserException> batchException = new
AtomicReference<>(null);
AtomicInteger numFinishedPartitions = new AtomicInteger(0);
- CompletableFuture.runAsync(() -> {
- for (int begin = 0; begin < allPartitions.size(); begin +=
batchSize) {
- int end = Math.min(begin + batchSize, allPartitions.size());
- if (batchException.get() != null || splitAssignment.isStop()) {
- break;
- }
- List<String> batch = allPartitions.subList(begin, end);
- int curBatchSize = end - begin;
- try {
- CompletableFuture.runAsync(() -> {
- try {
- List<ConnectorScanRange> ranges =
onPluginClassLoader(scanProvider,
- () ->
scanProvider.planScanForPartitionBatch(
- connectorSession, batchRequest,
batch));
- List<Split> batchSplits = new
ArrayList<>(ranges.size());
- for (ConnectorScanRange range : ranges) {
- batchSplits.add(new PluginDrivenSplit(range));
- }
- if (splitAssignment.needMoreSplit()) {
- splitAssignment.addToQueue(batchSplits);
- }
- } catch (Exception e) {
- batchException.set(new
UserException(e.getMessage(), e));
- } finally {
- if (batchException.get() != null) {
-
splitAssignment.setException(batchException.get());
- }
- if (numFinishedPartitions.addAndGet(curBatchSize)
== allPartitions.size()) {
- splitAssignment.finishSchedule();
- }
+ for (int begin = 0; begin < allPartitions.size(); begin += batchSize) {
Review Comment:
[P2] Keep batch admission asynchronous until split consumers are exposed.
This loop now submits every batch on the `SplitAssignment.init()` caller, and
`init()` cannot return/register `SplitSource`s until the loop finishes. With
`num_partitions_in_batch_mode=1`, a 100k-partition Hive scan creates 100k
tasks: producers eventually fill the per-BE 10k assignment queues, the 64k
schedule-executor queue then fills, and its 10-second blocked policy rejects
the next submission because no consumer exists yet. The previous outer async
pump let `startSplit()` return after admission began, so consumers could drain
queues while later batches were scheduled. Please retain a tracked outer pump
or otherwise bound admission while allowing consumers to start.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -131,6 +135,33 @@ public boolean usesHiveParquetInt96TimeZone() {
@Override
public List<ConnectorScanRange> planScan(ConnectorSession session,
ConnectorScanRequest request) {
+ HiveTableHandle hiveHandle = (HiveTableHandle)
request.getTableHandle();
+ if (!isExternalScanTaskReuseEnabled(session)) {
+ return doPlanScan(session, request);
+ }
+ if (hiveHandle.isTransactional()) {
+ // ACID / INSERT_ONLY reads open a per-scan read transaction with
a write-id snapshot and
+ // a shared metastore lock; reusing the planned ranges would skip
that transaction.
+ return doPlanScan(session, request);
+ }
+ // Statement-scoped reuse: within one statement the identical scan
(same table, same
+ // partition set, same formats) plans once and every duplicated
relation shares the result.
+ // The scope is NONE for offline planning and tests, in which case the
loader runs on every
+ // call. Session variables are constant within a statement and
deliberately absent.
+ String memoKey = SCAN_REUSE_NAMESPACE + ":" + session.getCatalogId() +
":" + session.getQueryId();
+ Map<HiveScanReuseKey, List<ConnectorScanRange>> scanReuse =
session.getStatementScope().computeIfAbsent(
+ memoKey, () -> new ConcurrentHashMap<>());
+ HiveScanReuseKey reuseKey = new HiveScanReuseKey(hiveHandle);
+ return scanReuse.computeIfAbsent(reuseKey,
+ key -> Collections.unmodifiableList(doPlanScan(session,
request)));
+ }
+
+ private static boolean isExternalScanTaskReuseEnabled(ConnectorSession
session) {
+ return session != null && !"false".equalsIgnoreCase(
Review Comment:
[P2] Treat an absent reuse flag as disabled for mixed-version API-6
deployments. All four helpers currently enable reuse unless the value is
literally `false`. A newer 6.x connector can load on an older 6.x planning FE,
but that FE does not declare this new variable: it ignores the forwarded entry
and cannot put it in `ConnectorSession`, so even an explicit `SET
enable_external_scan_task_reuse=false` becomes absent and the newer plugin
turns reuse back on. Please require explicit `true` (the new FE already
serializes its default `true`) or bump/enforce the connector API major, and
cover the missing-property case.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -2006,13 +2000,21 @@ private void startStreamingSplit() {
}
pinRewriteFileScope();
final ConnectorTableHandle handle = currentHandle;
- final ConnectorScanPlanProvider scanProvider = resolveScanProvider();
Executor scheduleExecutor =
Env.getCurrentEnv().getExtMetaCacheMgr().getScheduleExecutor();
- CompletableFuture.runAsync(() -> {
+ splitAssignment.submitProducer(scheduleExecutor, () -> {
ConnectorSplitSource source = null;
+ Closeable sourceCloser = null;
try {
source = onPluginClassLoader(scanProvider,
() -> scanProvider.streamSplits(connectorSession,
handle, columns, remainingFilter, -1L));
+ ConnectorSplitSource registeredSource = source;
+ AtomicBoolean sourceClosed = new AtomicBoolean(false);
+ sourceCloser = () -> {
+ if (sourceClosed.compareAndSet(false, true)) {
+ registeredSource.close();
Review Comment:
[P2] Do not close this source concurrently with its pump.
`ConnectorSplitSource` explicitly says instances are not thread-safe and are
driven by one background task, but `SplitAssignment.stop()` invokes this
registered closer on the query-finalization thread before joining the producer,
which may be inside `hasNext()`/`next()`. Iceberg's implementation mutates and
closes the same unsynchronized iterator in those methods, and third-party API-6
sources are entitled to the same single-thread guarantee. Please keep `close()`
producer-owned and add a cancellation primitive whose contract permits
cross-thread wakeup, or version the SPI contract and make every source
concurrently close-safe; cover cancellation while `hasNext()` is active.
--
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]
