You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2026/10/13 15:11:02 UTC
(doris) branch master updated: [improvement](datasource) Drop the statement scan reuse memo when planning ends (#68901)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8027970f02d [improvement](datasource) Drop the statement scan reuse memo when planning ends (#68901)
8027970f02d is described below
commit 8027970f02da16d874d3aefdfd1548243dac2b99
Author: Mingyu Chen (Rayner) <mo...@gmail.com>
AuthorDate: Tue Oct 13 23:10:53 2026 +0800
[improvement](datasource) Drop the statement scan reuse memo when planning ends (#68901)
### What problem does this PR solve?
Issue Number: None
Related PR: #66473, #66530
Problem Summary:
**In short.** With `enable_external_scan_task_reuse` on (the default),
the Hive, Hudi, Iceberg and Paimon connectors keep the planned range
list of every scan in the statement scope until the statement ends, even
when no other scan of the statement can reuse it. Reuse can only happen
while the statement is being planned, so this PR drops the memo when
planning ends, in the planner's resource cleanup. branch-4.x (#66473)
bounds the same reuse with per-statement budgets instead; on master the
memo is dropped only when the statement ends.
**Background.** #66530 lets the identical scans of one statement (a self
join, a UNION of the same table, a CTE read twice) plan once: each of
the four scan providers memoizes the final `List<ConnectorScanRange>` of
a synchronous `planScan` in a map hung in the statement's
`ConnectorStatementScope`, keyed by everything the ranges depend on. The
scope lives until the statement's query-finish callback closes it.
Without the memo the ranges are garbage as soon as the scan node has
turned them into `TScanRangeLocations`, because the scan node keeps no
reference to them.
**The problem, and what it cost.** The memo is the only long-lived
reference to the ranges, and it keeps them for the whole execution:
- A statement with a single relation, which can never reuse anything,
still keeps all of its ranges until it ends.
- Paimon and Hudi have no batch path, so every scan of them keeps its
whole range list; a Paimon JNI range carries the Base64-serialized
`DataSplit`, which grows with the number of files. Hive keeps it for an
unpartitioned table or below the batch threshold, and Iceberg until a
scan switches to streaming.
- An Arrow Flight statement keeps it until the client has fetched the
whole result.
Reuse needs none of this. Scans are planned only inside
`NereidsPlanner.plan()` (translation finalizes each scan node, which
calls `planScan`); the coordinator, Arrow Flight's DoGet and
`handleQueryWithRetry` never plan a scan again, and the batch /
streaming paths that produce splits during execution do not use the
memo. A statement planned more than once (cloud replan, INSERT retry,
the probe of INSERT OVERWRITE) gets a new query id or a reset scope each
time, so no plan reuses the memo of another plan today either.
**How this PR fixes it.**
- `ConnectorStatementScope` (SPI) gains `computeIfAbsentForPlanning`,
for a value only planning reads, and `endPlanning`, which drops those
values. The defaults keep today's behavior (`computeIfAbsent` and a
no-op), and `NONE` still memoizes nothing. A value registered for
planning must hold nothing to close, since it is dropped without being
closed. The connector SPI has not shipped in a release, so
`connector.plugin.api.version` stays as it is; `ConnectorStatementScope`
is not part of the frozen plugin surface baselines.
- `ConnectorStatementScopeImpl` records the keys registered for
planning, and `endPlanning` removes them; the other values of the
statement (metadata, tables, transactions) stay until `closeAll`.
- `StatementContext.releasePlannerResources`, the cleanup
`NereidsPlanner.planWithLock` runs in its `finally` block once the plan
is translated and its scans are finalized (and `StatementContext.close`
runs as a fallback), ends connector planning after releasing the planner
resources and before rethrowing a failure to release one. It never
creates a scope for a statement no connector has touched.
- The Hive, Hudi, Iceberg and Paimon providers register their range memo
for planning. The Hudi and Paimon table-resolution maps, which
`getScanNodeProperties` also reads, stay for the statement. Fluss reads
its lake table through the Paimon provider, so it follows.
A scan planned during execution through the default
`planScanForPartitionBatch` (only MaxCompute uses it, and MaxCompute
keeps no memo) would register the memo again and keep it until the
statement ends, as today; the SPI javadoc says so.
**Results.**
| Statement | Before: the planned ranges are held until | After |
|---|---|---|
| SELECT, INSERT ... SELECT | the statement ends | planning ends |
| Arrow Flight query | the client has fetched the result | planning ends
|
| EXPLAIN | the statement ends | planning ends |
| Planning fails | the statement ends | planning fails |
Identical scans of one plan still share their ranges. On a local cluster
with a filesystem Paimon catalog, a self join, a UNION ALL of the same
table, a self join filtered on a partition and a repeated statement
return the expected rows, and EXPLAIN shows both scans of the self join
with the same splits.
---
.../doris/connector/hive/HiveScanPlanProvider.java | 7 +-
.../connector/hive/HiveScanBatchModeTest.java | 28 ++++++
.../doris/connector/hive/TestStatementScope.java | 15 +++
.../doris/connector/hudi/HudiScanPlanProvider.java | 7 +-
.../doris/connector/hudi/HudiScanReuseKeyTest.java | 40 ++++++++
.../connector/iceberg/IcebergScanPlanProvider.java | 7 +-
.../iceberg/IcebergScanPlanProviderTest.java | 26 ++++++
.../connector/iceberg/TestStatementScope.java | 15 +++
.../connector/paimon/PaimonScanPlanProvider.java | 7 +-
.../paimon/PaimonScanPlanProviderTest.java | 54 +++++++++++
.../connector/spi/ConnectorStatementScope.java | 21 +++++
.../connector/ConnectorStatementScopeImpl.java | 27 ++++++
.../org/apache/doris/nereids/StatementContext.java | 15 +++
.../connector/ConnectorStatementScopeTest.java | 103 +++++++++++++++++++++
.../NereidsPlannerConnectorPlanningTest.java | 63 +++++++++++++
15 files changed, 423 insertions(+), 12 deletions(-)
diff --git a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
index 3571696db49..28bcaef410d 100644
--- a/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
+++ b/fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java
@@ -154,10 +154,11 @@ public class HiveScanPlanProvider implements ConnectorScanPlanProvider {
// 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.
+ // call. Session variables are constant within a statement and deliberately absent. Every scan of a
+ // statement is planned within one plan, so the memo is dropped when planning ends.
String memoKey = SCAN_REUSE_NAMESPACE + ":" + session.getCatalogId() + ":" + session.getQueryId();
- Map<HiveScanReuseKey, List<ConnectorScanRange>> scanReuse = session.getStatementScope().computeIfAbsent(
- memoKey, () -> new ConcurrentHashMap<>());
+ Map<HiveScanReuseKey, List<ConnectorScanRange>> scanReuse =
+ session.getStatementScope().computeIfAbsentForPlanning(memoKey, () -> new ConcurrentHashMap<>());
HiveScanReuseKey reuseKey = new HiveScanReuseKey(hiveHandle);
AtomicReference<List<ConnectorScanRange>> uncached = new AtomicReference<>();
List<ConnectorScanRange> cached = scanReuse.computeIfAbsent(reuseKey, key -> {
diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
index 2b286d809db..aeb7bdf9beb 100644
--- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
+++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/HiveScanBatchModeTest.java
@@ -422,6 +422,34 @@ public class HiveScanBatchModeTest {
Assertions.assertEquals(1, lister.totalCalls, "the underlying Hive file listing must run once");
}
+ @Test
+ public void statementReuseLastsOnlyWhilePlanning() {
+ // Identical scans share their ranges only within one plan, so the memo is kept for planning only: once the
+ // engine ends planning no range is held, and a later plan of the statement plans the scan again.
+ // MUTATION: keeping the memo for the whole statement (computeIfAbsent) -> red.
+ HiveScanPlanProvider provider = provider(new FakeHmsClient(), new CountingLister());
+ HiveTableHandle handle = new HiveTableHandle.Builder("db", "t", HiveTableType.HIVE)
+ .inputFormat(PARQUET_INPUT_FORMAT)
+ .serializationLib(PARQUET_SERDE)
+ .partitionKeyNames(PART_KEYS)
+ .prunedPartitions(Collections.singletonList(part("year=2024/month=01")))
+ .build();
+ TestStatementScope scope = new TestStatementScope();
+ ConnectorSession session = new ScopeSession(7L, "same-statement", scope);
+ ConnectorScanRequest request = ConnectorScanRequest.builder(
+ handle, Collections.<ConnectorColumnHandle>emptyList()).build();
+ List<ConnectorScanRange> planned = provider.planScan(session, request);
+
+ scope.endPlanning();
+
+ Assertions.assertFalse(scope.contains(HiveScanPlanProvider.SCAN_REUSE_NAMESPACE + ":7:same-statement"),
+ "the planned ranges must not outlive planning");
+ List<ConnectorScanRange> replanned = provider.planScan(session, request);
+ Assertions.assertFalse(planned.isEmpty(), "the partition's file must plan a range");
+ Assertions.assertNotSame(planned, replanned, "a scan planned after planning ended plans again");
+ Assertions.assertEquals(planned.size(), replanned.size());
+ }
+
@Test
public void statementReuseRetriesAListingThatFailedOnce() {
FailOnceLister lister = new FailOnceLister();
diff --git a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/TestStatementScope.java b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/TestStatementScope.java
index bcec12a500d..9475490fa71 100644
--- a/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/TestStatementScope.java
+++ b/fe/fe-connector/fe-connector-hive/src/test/java/org/apache/doris/connector/hive/TestStatementScope.java
@@ -19,6 +19,7 @@ package org.apache.doris.connector.hive;
import org.apache.doris.connector.spi.ConnectorStatementScope;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
@@ -31,6 +32,7 @@ import java.util.function.Supplier;
final class TestStatementScope implements ConnectorStatementScope {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+ private final Set<String> planningKeys = ConcurrentHashMap.newKeySet();
@Override
@SuppressWarnings("unchecked")
@@ -38,6 +40,19 @@ final class TestStatementScope implements ConnectorStatementScope {
return (T) cache.computeIfAbsent(key, k -> loader.get());
}
+ @Override
+ public <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ T value = computeIfAbsent(key, loader);
+ planningKeys.add(key);
+ return value;
+ }
+
+ @Override
+ public void endPlanning() {
+ planningKeys.forEach(cache::remove);
+ planningKeys.clear();
+ }
+
/** Whether something was memoized under {@code key}, e.g. the statement reuse map. */
boolean contains(String key) {
return cache.containsKey(key);
diff --git a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
index 6345b5f7ae0..31cc0c9ac23 100644
--- a/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
+++ b/fe/fe-connector/fe-connector-hudi/src/main/java/org/apache/doris/connector/hudi/HudiScanPlanProvider.java
@@ -210,15 +210,16 @@ public class HudiScanPlanProvider implements ConnectorScanPlanProvider {
// with generation-A schema_id values) could be paired with alias B's generation-B
// history_schema_info dictionary, causing BE to map files through wrong field identities.
// 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.
+ // call. Session variables are constant within a statement and deliberately absent. Every scan of a
+ // statement is planned within one plan, so the memo is dropped when planning ends.
if (session == null || !session.isExternalScanTaskReuseEnabled()) {
return doPlanScan(session, request);
}
HudiStatementTable statementTable = resolveHudiTable(
session, (HudiTableHandle) request.getTableHandle());
String memoKey = SCAN_REUSE_NAMESPACE + ":" + session.getCatalogId() + ":" + session.getQueryId();
- Map<HudiScanReuseKey, List<ConnectorScanRange>> scanReuse = session.getStatementScope().computeIfAbsent(
- memoKey, () -> new ConcurrentHashMap<>());
+ Map<HudiScanReuseKey, List<ConnectorScanRange>> scanReuse =
+ session.getStatementScope().computeIfAbsentForPlanning(memoKey, () -> new ConcurrentHashMap<>());
HudiScanReuseKey reuseKey = hudiScanReuseKey(
(HudiTableHandle) request.getTableHandle(), statementTable.generation);
AtomicReference<List<ConnectorScanRange>> uncached = new AtomicReference<>();
diff --git a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiScanReuseKeyTest.java b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiScanReuseKeyTest.java
index e1b97ae758e..f7af3c4b7a7 100644
--- a/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiScanReuseKeyTest.java
+++ b/fe/fe-connector/fe-connector-hudi/src/test/java/org/apache/doris/connector/hudi/HudiScanReuseKeyTest.java
@@ -37,6 +37,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -201,6 +202,27 @@ class HudiScanReuseKeyTest {
Assertions.assertEquals(1, provider.planCalls, "the underlying Hudi planner must run once");
}
+ @Test
+ void statementReuseLastsOnlyWhilePlanning() {
+ // Identical scans share their ranges only within one plan, so the memo is kept for planning only: once the
+ // engine ends planning no range is held, and a later plan of the statement plans the scan again.
+ // MUTATION: keeping the memo for the whole statement (computeIfAbsent) -> red.
+ RecordingScanProvider provider = new RecordingScanProvider();
+ MemoScope scope = new MemoScope();
+ ConnectorSession session = new MemoSession(scope);
+ ConnectorScanRequest request = ConnectorScanRequest.builder(
+ handle().toBuilder().prunedPartitionPaths(null).build(),
+ Collections.<ConnectorColumnHandle>emptyList()).build();
+ List<ConnectorScanRange> planned = provider.planScan(session, request);
+
+ scope.endPlanning();
+
+ Assertions.assertFalse(scope.contains(HudiScanPlanProvider.SCAN_REUSE_NAMESPACE + ":7:same-statement"),
+ "the planned ranges must not outlive planning");
+ Assertions.assertNotSame(planned, provider.planScan(session, request));
+ Assertions.assertEquals(2, provider.planCalls, "a scan planned after planning ended plans again");
+ }
+
@Test
void statementReuseRetriesASchemaResolutionThatFailedOnce() {
RecordingScanProvider provider = new RecordingScanProvider();
@@ -348,12 +370,30 @@ class HudiScanReuseKeyTest {
private static final class MemoScope implements ConnectorStatementScope {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+ private final Set<String> planningKeys = ConcurrentHashMap.newKeySet();
@Override
@SuppressWarnings("unchecked")
public <T> T computeIfAbsent(String key, Supplier<T> loader) {
return (T) cache.computeIfAbsent(key, ignored -> loader.get());
}
+
+ @Override
+ public <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ T value = computeIfAbsent(key, loader);
+ planningKeys.add(key);
+ return value;
+ }
+
+ @Override
+ public void endPlanning() {
+ planningKeys.forEach(cache::remove);
+ planningKeys.clear();
+ }
+
+ boolean contains(String key) {
+ return cache.containsKey(key);
+ }
}
private static final class MemoSession implements ConnectorSession {
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index 05055535e72..8ad3d1cc2bf 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -463,10 +463,11 @@ public class IcebergScanPlanProvider implements ConnectorScanPlanProvider {
// outer entry). The scope loader only constructs the memo map; planning then runs on that
// separate map, so every scope call from planScanInternal is top-level again. The memo
// key is catalog-scoped, which also isolates same-named tables across a cross-catalog
- // statement.
+ // statement. Every scan of a statement is planned within one plan, so the memo is dropped
+ // when planning ends.
String memoKey = SCAN_REUSE_NAMESPACE + ":" + session.getCatalogId() + ":" + session.getQueryId();
- Map<IcebergScanReuseKey, List<ConnectorScanRange>> scanReuse =
- session.getStatementScope().computeIfAbsent(memoKey, () -> new ConcurrentHashMap<>());
+ Map<IcebergScanReuseKey, List<ConnectorScanRange>> scanReuse = session.getStatementScope()
+ .computeIfAbsentForPlanning(memoKey, () -> new ConcurrentHashMap<>());
IcebergScanReuseKey reuseKey = new IcebergScanReuseKey(icebergHandle, request);
return scanReuse.computeIfAbsent(reuseKey,
k -> Collections.unmodifiableList(planScanInternal(session,
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 3366a38ac88..e83a3049fde 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -589,6 +589,32 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals(1, first.size());
}
+ @Test
+ public void statementReuseLastsOnlyWhilePlanning() {
+ // Identical scans share their ranges only within one plan, so the memo is kept for planning only: once the
+ // engine ends planning no range is held, and a later plan of the statement plans the scan again.
+ // MUTATION: keeping the memo for the whole statement (computeIfAbsent) -> red.
+ Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(
+ dataFile(table.spec(), "s3://b/db/t1/f1.parquet", 1024, null, null)).commit();
+ IcebergScanPlanProvider provider = providerOver(table);
+ TestStatementScope scope = new TestStatementScope();
+ ConnectorSession session = new FakeScanSession("UTC",
+ Collections.singletonMap("enable_external_scan_task_reuse", "true"))
+ .withScope(scope);
+ ConnectorScanRequest request = ConnectorScanRequest.builder(
+ new IcebergTableHandle("db1", "t1"), Collections.emptyList()).build();
+ List<ConnectorScanRange> planned = provider.planScan(session, request);
+
+ scope.endPlanning();
+
+ Assertions.assertFalse(scope.contains(IcebergScanPlanProvider.SCAN_REUSE_NAMESPACE + ":0:q"),
+ "the planned ranges must not outlive planning");
+ List<ConnectorScanRange> replanned = provider.planScan(session, request);
+ Assertions.assertNotSame(planned, replanned, "a scan planned after planning ended plans again");
+ Assertions.assertEquals(1, replanned.size());
+ }
+
@Test
public void statementReuseMatchesStructurallyEqualFilters() {
Table table = createTable("t1", SCHEMA, PartitionSpec.unpartitioned());
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/TestStatementScope.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/TestStatementScope.java
index 5cab9c0d2ed..599b702e253 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/TestStatementScope.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/TestStatementScope.java
@@ -19,6 +19,7 @@ package org.apache.doris.connector.iceberg;
import org.apache.doris.connector.spi.ConnectorStatementScope;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
@@ -33,6 +34,7 @@ import java.util.function.Supplier;
final class TestStatementScope implements ConnectorStatementScope {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+ private final Set<String> planningKeys = ConcurrentHashMap.newKeySet();
private final AtomicBoolean closed = new AtomicBoolean();
@Override
@@ -41,6 +43,19 @@ final class TestStatementScope implements ConnectorStatementScope {
return (T) cache.computeIfAbsent(key, k -> loader.get());
}
+ @Override
+ public <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ T value = computeIfAbsent(key, loader);
+ planningKeys.add(key);
+ return value;
+ }
+
+ @Override
+ public void endPlanning() {
+ planningKeys.forEach(cache::remove);
+ planningKeys.clear();
+ }
+
/** Whether something was memoized under {@code key}, e.g. a connector's statement reuse map. */
boolean contains(String key) {
return cache.containsKey(key);
diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
index 2a2157e1bdc..2d06c4b826d 100644
--- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
+++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java
@@ -573,10 +573,11 @@ public class PaimonScanPlanProvider implements ConnectorScanPlanProvider {
// Statement-scoped reuse: within one statement the identical scan (same table, same
// branch/options pin, same generation, same projection, same filter, same limit, same COUNT
// pushdown) plans once and every duplicated relation shares the result. Session variables
- // are constant within a statement and deliberately absent from the key.
+ // are constant within a statement and deliberately absent from the key. Every scan of a
+ // statement is planned within one plan, so the memo is dropped when planning ends.
String memoKey = SCAN_REUSE_NAMESPACE + ":" + session.getCatalogId() + ":" + session.getQueryId();
- Map<PaimonScanReuseKey, List<ConnectorScanRange>> scanReuse = session.getStatementScope().computeIfAbsent(
- memoKey, () -> new ConcurrentHashMap<>());
+ Map<PaimonScanReuseKey, List<ConnectorScanRange>> scanReuse =
+ session.getStatementScope().computeIfAbsentForPlanning(memoKey, () -> new ConcurrentHashMap<>());
PaimonScanReuseKey reuseKey = new PaimonScanReuseKey(paimonHandle, request, generation);
return scanReuse.computeIfAbsent(reuseKey,
key -> Collections.unmodifiableList(planScanInternal(session,
diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
index 3bda0c5eb50..0ce6227673a 100644
--- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
+++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java
@@ -1024,6 +1024,46 @@ public class PaimonScanPlanProviderTest {
}
}
+ @Test
+ public void statementReuseLastsOnlyWhilePlanning(@TempDir Path warehouse) throws Exception {
+ // Identical scans share their ranges only within one plan, so the range memo is kept for planning only:
+ // once the engine ends planning no range is held, and a later plan of the statement plans the scan again.
+ // The table resolved for the statement stays, so that plan still reads the same table generation.
+ // MUTATION: keeping the range memo for the whole statement (computeIfAbsent) -> red.
+ try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(),
+ new org.apache.paimon.fs.Path(warehouse.toUri()))) {
+ catalog.createDatabase("db", false);
+ Identifier id = Identifier.create("db", "reuse_planning");
+ catalog.createTable(id, Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option("bucket", "-1")
+ .build(), false);
+ Table table = catalog.getTable(id);
+ commitRows(table, GenericRow.of(1));
+
+ RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps();
+ ops.table = table;
+ PaimonScanPlanProvider provider = new PaimonScanPlanProvider(
+ PaimonCatalogProperties.of(Collections.emptyMap()), ops);
+ TestStatementScope scope = new TestStatementScope();
+ ConnectorSession session = reuseSession(scope, 0L);
+ PaimonTableHandle handle = new PaimonTableHandle(
+ "db", "reuse_planning", Collections.emptyList(), Collections.emptyList());
+ List<ConnectorScanRange> planned = provider.planScan(session, scanOf(handle).build());
+
+ scope.endPlanning();
+
+ Assertions.assertFalse(scope.contains(PaimonScanPlanProvider.SCAN_REUSE_NAMESPACE + ":0:q"),
+ "the planned ranges must not outlive planning");
+ Assertions.assertTrue(scope.contains(PaimonScanPlanProvider.SCAN_REUSE_NAMESPACE + ":tables:0:q"),
+ "the table resolved for the statement stays");
+ List<ConnectorScanRange> replanned = provider.planScan(session, scanOf(handle).build());
+ Assertions.assertFalse(planned.isEmpty(), "one committed row must plan at least one split");
+ Assertions.assertNotSame(planned, replanned, "a scan planned after planning ended plans again");
+ Assertions.assertEquals(planned.size(), replanned.size());
+ }
+ }
+
@Test
public void statementReuseNeverMemoizesSystemTables(@TempDir Path warehouse) throws Exception {
// System tables resolve their snapshot on the BE and are never reused. MUTATION: dropping the
@@ -3227,6 +3267,7 @@ public class PaimonScanPlanProviderTest {
private static final class TestStatementScope implements ConnectorStatementScope {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+ private final Set<String> planningKeys = ConcurrentHashMap.newKeySet();
@Override
@SuppressWarnings("unchecked")
@@ -3234,6 +3275,19 @@ public class PaimonScanPlanProviderTest {
return (T) cache.computeIfAbsent(key, ignored -> loader.get());
}
+ @Override
+ public <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ T value = computeIfAbsent(key, loader);
+ planningKeys.add(key);
+ return value;
+ }
+
+ @Override
+ public void endPlanning() {
+ planningKeys.forEach(cache::remove);
+ planningKeys.clear();
+ }
+
/** Whether something was memoized under {@code key}, e.g. the statement reuse map. */
boolean contains(String key) {
return cache.containsKey(key);
diff --git a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatementScope.java b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatementScope.java
index f8598ba018d..5f767a9c7e8 100644
--- a/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatementScope.java
+++ b/fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorStatementScope.java
@@ -44,6 +44,27 @@ public interface ConnectorStatementScope {
*/
<T> T computeIfAbsent(String key, Supplier<T> loader);
+ /**
+ * {@link #computeIfAbsent} for a value only planning reads, such as a memo that lets the identical scans of
+ * one plan share their planned ranges. The engine drops such values once planning ends ({@link #endPlanning}),
+ * so nothing they reference is held while the statement executes. A call after that computes and registers
+ * the value again, which a scan planned at execution time (a partition batch planned through
+ * {@code ConnectorScanPlanProvider#planScanForPartitionBatch}'s default) would do: the value then stays until
+ * {@link #closeAll}. The value must hold nothing to close, because it is dropped without being closed. Under
+ * {@link #NONE} the loader runs on every call.
+ */
+ default <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ return computeIfAbsent(key, loader);
+ }
+
+ /**
+ * Drops every value registered through {@link #computeIfAbsentForPlanning}. The engine calls it when the
+ * planning of the statement ends, whether or not the plan succeeded; values registered through
+ * {@link #computeIfAbsent} stay until {@link #closeAll}. The default is a no-op.
+ */
+ default void endPlanning() {
+ }
+
/**
* Typed convenience over {@link #computeIfAbsent} for the ONE {@link ConnectorMetadata} a statement uses per
* {@code key}. The engine's metadata funnel builds {@code key} (as {@code "metadata:" + catalogId}, plus the
diff --git a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorStatementScopeImpl.java b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorStatementScopeImpl.java
index 06c3cd8a51d..07de8217d5d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorStatementScopeImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorStatementScopeImpl.java
@@ -20,9 +20,12 @@ package org.apache.doris.connector;
import org.apache.doris.connector.spi.ConnectorStatementScope;
import org.apache.doris.datasource.plugin.CatalogStatementTransaction;
+import com.google.common.base.Preconditions;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.util.Iterator;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
@@ -39,12 +42,18 @@ import java.util.function.Supplier;
* mid-computation resize. Connectors that need scope-backed state during scan planning nest a
* separate connector-private memo map inside the scope, avoiding re-entry and preserving the
* String-key SPI contract.</p>
+ *
+ * <p>A value registered through {@link #computeIfAbsentForPlanning} lives only until {@link #endPlanning}, which
+ * {@link org.apache.doris.nereids.StatementContext#releasePlannerResources} calls once planning is over, whether or
+ * not the plan succeeded.</p>
*/
public class ConnectorStatementScopeImpl implements ConnectorStatementScope {
private static final Logger LOG = LogManager.getLogger(ConnectorStatementScopeImpl.class);
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap<>();
+ // Keys of the values only planning reads; endPlanning drops them.
+ private final Set<String> planningKeys = ConcurrentHashMap.newKeySet();
private boolean closed;
@Override
@@ -53,6 +62,24 @@ public class ConnectorStatementScopeImpl implements ConnectorStatementScope {
return (T) cache.computeIfAbsent(key, k -> loader.get());
}
+ @Override
+ public <T> T computeIfAbsentForPlanning(String key, Supplier<T> loader) {
+ T value = computeIfAbsent(key, loader);
+ Preconditions.checkState(!(value instanceof AutoCloseable),
+ "planning value %s is dropped without being closed, so it must hold nothing to close", key);
+ planningKeys.add(key);
+ return value;
+ }
+
+ @Override
+ public void endPlanning() {
+ Iterator<String> keys = planningKeys.iterator();
+ while (keys.hasNext()) {
+ cache.remove(keys.next());
+ keys.remove();
+ }
+ }
+
/**
* Tears the statement's scoped values down at statement end, in two ordered passes. Pass 1 finalizes any
* {@link CatalogStatementTransaction} (rolling back a transaction the executor never committed — only a
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
index 686a62a51f4..2913af5f004 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java
@@ -851,6 +851,17 @@ public class StatementContext implements Closeable {
return this.connectorStatementScope;
}
+ /**
+ * Drops the values of this statement's connector scope that only planning reads (see
+ * {@link ConnectorStatementScope#endPlanning}), when a connector has opened the scope; unlike
+ * {@link #getOrCreateConnectorStatementScope}, it never creates one. {@link #releasePlannerResources} calls it.
+ */
+ private synchronized void endConnectorPlanning() {
+ if (this.connectorStatementScope != null) {
+ this.connectorStatementScope.endPlanning();
+ }
+ }
+
/**
* Closes (deterministically) then drops the connector scope so the next statement execution starts fresh.
* Prepared-statement EXECUTE reuses one StatementContext across executions (see
@@ -1159,6 +1170,10 @@ public class StatementContext implements Closeable {
}
}
}
+ // NereidsPlanner.planWithLock runs this once its plan is translated and the scans are finalized, so what
+ // connectors kept only for planning, such as the ranges identical scans share, can go, also when a resource
+ // failed to close.
+ endConnectorPlanning();
if (throwable != null) {
Throwables.throwIfInstanceOf(throwable, RuntimeException.class);
throw new IllegalStateException("Release resource failed", throwable);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorStatementScopeTest.java b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorStatementScopeTest.java
index 428ac2a2b80..cb7777dd108 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorStatementScopeTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/connector/ConnectorStatementScopeTest.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector;
+import org.apache.doris.catalog.TableIf;
import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
import org.apache.doris.connector.spi.ConnectorWriteOps;
@@ -40,7 +41,9 @@ import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
/**
* Tests for the per-statement {@link ConnectorStatementScope}: the {@link ConnectorStatementScope#NONE} no-op,
@@ -93,6 +96,106 @@ public class ConnectorStatementScopeTest {
Assertions.assertNotSame(firstA, b, "different keys are isolated (cross-catalog / cross-table)");
}
+ @Test
+ public void nonePlanningValuesAreNeverMemoized() {
+ // Under NONE a value only planning reads is loaded on every call too, and ending planning has nothing to do.
+ // MUTATION: NONE memoizing planning values -> one load -> red.
+ AtomicInteger loads = new AtomicInteger();
+ Supplier<Object> loader = () -> {
+ loads.incrementAndGet();
+ return new Object();
+ };
+ ConnectorStatementScope none = ConnectorStatementScope.NONE;
+
+ Assertions.assertNotSame(none.computeIfAbsentForPlanning("k", loader),
+ none.computeIfAbsentForPlanning("k", loader));
+ none.endPlanning();
+ Assertions.assertEquals(2, loads.get(), "NONE must run the loader every time (no memo)");
+ }
+
+ @Test
+ public void endPlanningDropsOnlyThePlanningValues() throws Exception {
+ // Identical scans share their planned ranges only within one plan, so a value only planning reads is dropped
+ // when planning ends, while the values the statement keeps (its tables, metadata, transactions) stay until
+ // closeAll. A later plan of the same statement registers the value again, and the end of that plan drops it.
+ // MUTATION: endPlanning dropping nothing, dropping every value, or forgetting a value registered again -> red.
+ ConnectorStatementScope scope = new ConnectorStatementScopeImpl();
+ AtomicInteger memoLoads = new AtomicInteger();
+ Supplier<Object> memoLoader = () -> {
+ memoLoads.incrementAndGet();
+ return new Object();
+ };
+ Object memo = scope.computeIfAbsentForPlanning("memo", memoLoader);
+ Assertions.assertSame(memo, scope.computeIfAbsentForPlanning("memo", memoLoader),
+ "within one plan the value is shared");
+ Object table = scope.computeIfAbsent("table", Object::new);
+ AtomicInteger closes = new AtomicInteger();
+ AutoCloseable metadata = () -> closes.incrementAndGet();
+ scope.computeIfAbsent("metadata", () -> metadata);
+
+ scope.endPlanning();
+
+ Assertions.assertSame(table, scope.computeIfAbsent("table", Object::new), "statement values outlive planning");
+ Object memoOfNextPlan = scope.computeIfAbsentForPlanning("memo", memoLoader);
+ Assertions.assertNotSame(memo, memoOfNextPlan, "the next plan of the statement loads the value again");
+ scope.endPlanning();
+ Assertions.assertNotSame(memoOfNextPlan, scope.computeIfAbsentForPlanning("memo", memoLoader));
+ Assertions.assertEquals(3, memoLoads.get());
+
+ scope.closeAll();
+ Assertions.assertEquals(1, closes.get(), "closeAll still closes the statement values");
+ }
+
+ @Test
+ public void planningValueMustHoldNothingToClose() {
+ // endPlanning drops a value without closing it, so a closeable one would leak. MUTATION: dropping the check
+ // -> no exception -> red.
+ ConnectorStatementScope scope = new ConnectorStatementScopeImpl();
+ AutoCloseable closeable = () -> { };
+
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> scope.computeIfAbsentForPlanning("closeable", () -> closeable));
+ }
+
+ @Test
+ public void releasingPlannerResourcesEndsThePlanningOfTheScope() {
+ // NereidsPlanner.planWithLock releases the planner resources when planning is over, which ends connector
+ // planning; a statement no connector has touched has no scope to end. MUTATION: releasePlannerResources not
+ // ending connector planning -> red.
+ StatementContext ctx = new StatementContext();
+ ctx.releasePlannerResources();
+ ConnectorStatementScope scope = ctx.getOrCreateConnectorStatementScope();
+ Object memo = scope.computeIfAbsentForPlanning("memo", Object::new);
+ Object table = scope.computeIfAbsent("table", Object::new);
+
+ ctx.releasePlannerResources();
+
+ Assertions.assertSame(scope, ctx.getOrCreateConnectorStatementScope(), "ending planning keeps the scope");
+ Assertions.assertSame(table, scope.computeIfAbsent("table", Object::new));
+ Assertions.assertNotSame(memo, scope.computeIfAbsentForPlanning("memo", Object::new));
+ }
+
+ @Test
+ public void releasingPlannerResourcesEndsPlanningWhenAResourceFailsToClose() {
+ // A table lock that fails to unlock still reaches the caller, after the planning values are gone.
+ // MUTATION: ending connector planning only after the release succeeded -> the memo survives -> red.
+ StatementContext ctx = new StatementContext();
+ ConnectorStatementScope scope = ctx.getOrCreateConnectorStatementScope();
+ Object memo = scope.computeIfAbsentForPlanning("memo", Object::new);
+ TableIf table = Mockito.mock(TableIf.class);
+ Mockito.when(table.needReadLockWhenPlan()).thenReturn(true);
+ Mockito.when(table.tryReadLock(Mockito.anyLong(), Mockito.any(TimeUnit.class))).thenReturn(true);
+ Mockito.when(table.getNameWithFullQualifiers()).thenReturn("internal.db.t");
+ IllegalStateException unlockFailure = new IllegalStateException("unlock failed");
+ Mockito.doThrow(unlockFailure).when(table).readUnlock();
+ ctx.getTables().put(Arrays.asList("internal", "db", "t"), table);
+ ctx.lock();
+
+ Assertions.assertSame(unlockFailure,
+ Assertions.assertThrows(IllegalStateException.class, ctx::releasePlannerResources));
+ Assertions.assertNotSame(memo, scope.computeIfAbsentForPlanning("memo", Object::new));
+ }
+
@Test
public void statementContextScopeIsStableThenResetsForReusedPreparedContext() {
// StatementContext lazily builds one scope and reuses it across the statement (getOrCreate...). A prepared
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/NereidsPlannerConnectorPlanningTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/NereidsPlannerConnectorPlanningTest.java
new file mode 100644
index 00000000000..b59dd394531
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/NereidsPlannerConnectorPlanningTest.java
@@ -0,0 +1,63 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids;
+
+import org.apache.doris.connector.spi.ConnectorStatementScope;
+import org.apache.doris.nereids.glue.LogicalPlanAdapter;
+import org.apache.doris.nereids.parser.NereidsParser;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** {@link NereidsPlanner#plan} ends the planning of the statement's connector scope. */
+public class NereidsPlannerConnectorPlanningTest extends TestWithFeService {
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabaseAndUse("connector_planning");
+ }
+
+ @Test
+ public void planEndsConnectorPlanningWhetherItSucceedsOrFails() {
+ // Identical scans share their planned ranges only within one plan, so what a connector keeps for planning
+ // must not outlive the plan, whether the plan is complete or planning fails; the values the statement keeps
+ // stay. MUTATION: planWithLock's cleanup (releasePlannerResources) not ending connector planning -> red for
+ // both statements.
+ assertPlanEndsConnectorPlanning("select 1", true);
+ assertPlanEndsConnectorPlanning("select * from no_such_table", false);
+ }
+
+ private void assertPlanEndsConnectorPlanning(String sql, boolean planned) {
+ StatementContext statementContext = new StatementContext();
+ ConnectorStatementScope scope = statementContext.getOrCreateConnectorStatementScope();
+ Object memo = scope.computeIfAbsentForPlanning("memo", Object::new);
+ Object table = scope.computeIfAbsent("table", Object::new);
+ NereidsPlanner planner = new NereidsPlanner(statementContext);
+ LogicalPlanAdapter statement = new LogicalPlanAdapter(new NereidsParser().parseSingle(sql), statementContext);
+
+ if (planned) {
+ planner.plan(statement);
+ } else {
+ Assertions.assertThrows(RuntimeException.class, () -> planner.plan(statement));
+ }
+
+ Assertions.assertNotSame(memo, scope.computeIfAbsentForPlanning("memo", Object::new), sql);
+ Assertions.assertSame(table, scope.computeIfAbsent("table", Object::new), sql);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org