924060929 commented on code in PR #66530:
URL: https://github.com/apache/doris/pull/66530#discussion_r3851353254
##########
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:
Fixed in f39b83c8773. Scan-task reuse now requires the session property to
be explicitly true for Hive, Hudi, Iceberg, and Paimon, so an older planning FE
that omits the property safely bypasses reuse. Added missing-property coverage
for Hive and Hudi.
##########
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:
Fixed in f39b83c8773. Partition batches are now planned sequentially by one
tracked asynchronous producer. This keeps startSplit non-blocking and preserves
assignment-queue backpressure without nested submissions to the same bounded
executor; initial executor rejection is routed through
SplitAssignment.setException.
##########
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:
Fixed in f39b83c8773. ConnectorSplitSource now has a thread-safe cancel
contract used by stop(), while final close remains producer-owned.
SplitAssignment runs cancellation, interrupts active producers, and joins them
before returning. Iceberg cancellation and lifecycle race tests were added.
--
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]
