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) <[email protected]>
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: [email protected]
For additional commands, e-mail: [email protected]
