924060929 commented on code in PR #66530:
URL: https://github.com/apache/doris/pull/66530#discussion_r3773298801
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -725,4 +759,72 @@ private static final class PartitionScanInfo {
private ConnectorStorageContext storage() {
return context.getStorageContext();
}
+
+ /**
+ * Statement-scoped cache key for one Hive scan.
+ *
+ * <p>Includes every input that changes the planned split list: table
identity, the file formats
+ * (input format / serialization lib / JSON single-column gate), the
partition keys and the
+ * pruned partition set (each partition's location and values). ACID
tables are excluded
+ * upstream, and session variables are statement-constant, so both stay
out of the key.
+ */
+ private static final class HiveScanReuseKey {
+ private final String dbName;
+ private final String tableName;
+ private final String location;
+ private final String inputFormat;
+ private final String serializationLib;
+ private final boolean firstColumnIsString;
+ private final List<String> partitionKeyNames;
+ private final List<HmsPartitionInfo> prunedPartitions;
+
+ private HiveScanReuseKey(HiveTableHandle handle) {
+ // Catalog and query isolation are provided by the statement-scope
memo key. The table
+ // location identifies the data source of unpartitioned tables,
whose prunedPartitions
+ // is null.
+ this.dbName = handle.getDbName();
+ this.tableName = handle.getTableName();
+ this.location = handle.getLocation();
+ this.inputFormat = handle.getInputFormat();
+ this.serializationLib = handle.getSerializationLib();
+ this.firstColumnIsString = handle.isFirstColumnString();
+ this.partitionKeyNames = handle.getPartitionKeyNames() == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(new
ArrayList<>(handle.getPartitionKeyNames()));
+ this.prunedPartitions = handle.getPrunedPartitions() == null
Review Comment:
Addressed on the new head. `HiveScanReuseKey` now preserves `null` versus an
empty pruned-partition list, and the live-scope test plans zero-pruned first
followed by unpruned and verifies the latter still enumerates files instead of
reusing `[]`.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -2246,4 +2269,76 @@ private static String escapeJson(String s) {
private ConnectorStorageContext storage() {
return context.getStorageContext();
}
+
+ /**
+ * Statement-scoped cache key for one Paimon scan.
+ *
+ * <p>Includes every input that changes the planned split list: table
identity, the branch pin,
+ * the whole scan-options map (snapshot / tag / incremental / options
pins), the projected
+ * columns in order, the pushed filter and the COUNT pushdown flag. System
tables are excluded
+ * upstream, and session variables are statement-constant, so both stay
out of the key.
+ */
+ private static final class PaimonScanReuseKey {
+ private final String databaseName;
+ private final String tableName;
+ private final String branchName;
+ private final Map<String, String> scanOptions;
+ private final List<String> columnNames;
+ private final Optional<ConnectorExpression> filter;
+ private final boolean countPushdown;
+
+ private PaimonScanReuseKey(PaimonTableHandle handle,
ConnectorScanRequest request) {
+ // Catalog and query isolation are provided by the statement-scope
memo key. System
+ // tables are bypassed in planScan before this key is built, so
sysTableName is always
+ // null here; if the system-table bypass is ever relaxed, add it
back.
+ this.databaseName = handle.getDatabaseName();
+ this.tableName = handle.getTableName();
+ this.branchName = handle.getBranchName();
+ this.scanOptions = handle.getScanOptions() == null
+ ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new
HashMap<>(handle.getScanOptions()));
+ this.columnNames = request.getColumns().stream()
+ .map(PaimonScanReuseKey::toPaimonColumnName)
+ .collect(Collectors.toList());
+ this.filter = request.getFilter();
Review Comment:
Same resolution as Iceberg: all concrete `ConnectorExpression`
implementations already provide structural `equals`/`hashCode`, so
independently constructed equivalent Paimon filters select the same key. The
new head changes the real filesystem Paimon planning test to build two distinct
equivalent filter/request objects and verifies that the second `planScan`
reuses the first range list while authenticated table load plus manifest
enumeration still execute only once. No production key change is needed.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -131,6 +133,28 @@ public boolean usesHiveParquetInt96TimeZone() {
@Override
public List<ConnectorScanRange> planScan(ConnectorSession session,
ConnectorScanRequest request) {
+ HiveTableHandle hiveHandle = (HiveTableHandle)
request.getTableHandle();
+ if (session == null) {
+ 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 = "hive.scan-reuse:" + 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)));
Review Comment:
Confirmed against the current dispatch chain: a Hive scan that qualifies for
partition batch mode does not call synchronous `getSplits()`/`planScan()`
during execution. `FileQueryScanNode.createScanRangeLocations` takes the
`isBatchMode()` branch, `SplitAssignment.init()` calls `startSplit()`, and Hive
reaches `planScanForPartitionBatch()`, which deliberately bypasses the
statement memo. Thus the full eager plan described here is not created or
retained on the batch execution path. For non-batch scans, the statement-owned
memo is the intended scope for duplicate-alias reuse and closes through the
query-finish/StatementContext lifecycle. No reachable P1/P2 production change
is required.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -725,4 +759,72 @@ private static final class PartitionScanInfo {
private ConnectorStorageContext storage() {
return context.getStorageContext();
}
+
+ /**
+ * Statement-scoped cache key for one Hive scan.
+ *
+ * <p>Includes every input that changes the planned split list: table
identity, the file formats
+ * (input format / serialization lib / JSON single-column gate), the
partition keys and the
+ * pruned partition set (each partition's location and values). ACID
tables are excluded
+ * upstream, and session variables are statement-constant, so both stay
out of the key.
+ */
+ private static final class HiveScanReuseKey {
+ private final String dbName;
+ private final String tableName;
+ private final String location;
+ private final String inputFormat;
+ private final String serializationLib;
+ private final boolean firstColumnIsString;
+ private final List<String> partitionKeyNames;
+ private final List<HmsPartitionInfo> prunedPartitions;
+
+ private HiveScanReuseKey(HiveTableHandle handle) {
+ // Catalog and query isolation are provided by the statement-scope
memo key. The table
+ // location identifies the data source of unpartitioned tables,
whose prunedPartitions
+ // is null.
+ this.dbName = handle.getDbName();
+ this.tableName = handle.getTableName();
+ this.location = handle.getLocation();
+ this.inputFormat = handle.getInputFormat();
+ this.serializationLib = handle.getSerializationLib();
+ this.firstColumnIsString = handle.isFirstColumnString();
+ this.partitionKeyNames = handle.getPartitionKeyNames() == null
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(new
ArrayList<>(handle.getPartitionKeyNames()));
+ this.prunedPartitions = handle.getPrunedPartitions() == null
+ ? null
+ : Collections.unmodifiableList(new
ArrayList<>(handle.getPrunedPartitions()));
Review Comment:
Confirmed both notes against the current implementation. The defensive
partition-list snapshot and structural hash are intentional key-correctness
costs: the handle-owned list must not mutate after insertion, and comparing the
actual pruned identity necessarily reads that identity; large partition scans
that enter batch mode bypass this key. The format concern is not reachable on
this head: `convertPartitions` explicitly ignores `HmsPartitionInfo`
input-format/serde and sets `PartitionScanInfo.fileFormat` to null, so planning
uniformly falls back to the table-level input format/serde, both already
present in `HiveScanReuseKey`. No production change is needed.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -2922,4 +2953,70 @@ private Table resolveSysTable(ConnectorSession session,
IcebergTableHandle handl
private ConnectorStorageContext storage() {
return context.getStorageContext();
}
+
+ /**
+ * Statement-scoped cache key for one Iceberg scan.
+ *
+ * <p>Includes every input that changes the planned file list: table
identity, the typed
+ * time-travel pin (snapshot id / ref / schema id), the rewrite-file
scope, the pushed filter
+ * and the COUNT pushdown flag. System tables are excluded upstream, and
session variables are
+ * statement-constant, so both stay out of the key.
+ */
+ private static final class IcebergScanReuseKey {
+ private final String dbName;
+ private final String tableName;
+ private final long snapshotId;
+ private final String ref;
+ private final long schemaId;
+ private final List<String> rewriteFileScope;
+ private final Optional<ConnectorExpression> filter;
+ private final boolean countPushdown;
+
+ private IcebergScanReuseKey(IcebergTableHandle handle,
ConnectorScanRequest request) {
+ this.dbName = handle.getDbName();
+ this.tableName = handle.getTableName();
+ this.snapshotId = handle.getSnapshotId();
+ this.ref = handle.getRef();
+ this.schemaId = handle.getSchemaId();
+ // System tables are bypassed in planScan before this key is
built, so sysTableName is
+ // always null here; if the system-table bypass is ever relaxed,
add it back.
+ this.rewriteFileScope = handle.getRewriteFileScope() == null
+ ? Collections.emptyList()
+ : handle.getRewriteFileScope().stream().sorted()
+ .collect(Collectors.toList());
+ this.filter = request.getFilter();
Review Comment:
The production premise does not hold on this head: although the
`ConnectorExpression` interface does not redeclare `equals`/`hashCode`, every
concrete SPI expression node (`ConnectorAnd`, `ConnectorOr`, `ConnectorNot`,
comparison, column, literal, like, between, in, is-null, and function-call)
implements structural equality and hashing over its semantic fields. `Optional`
therefore delegates to structural expression equality. The new head also adds
an Iceberg production-path test that constructs two separate equivalent filter
trees and two separate requests, asserts the expression instances differ, and
verifies that `planScan` returns the same statement-cached range list. No
production canonicalization change is needed.
##########
fe/fe-core/src/main/java/org/apache/doris/job/executor/TaskProcessor.java:
##########
@@ -82,6 +84,22 @@ private void runTask(AbstractTask task) {
task.runTask();
} catch (Exception e) {
log.warn("Execute task error, task id: {}", task.getTaskId(), e);
+ } finally {
+ closeTaskContext();
+ }
+ }
+
+ private void closeTaskContext() {
+ ConnectContext connectContext = ConnectContext.get();
Review Comment:
Fixed on 5e9b9fe3de6. Cleanup now happens in MTMVTask.exec itself, so every
refresh chunk/retry closes the StatementContext it installed and removes the
worker thread-local ConnectContext, including failures before command.run().
The executor is reset at the start of each internal execution to avoid auditing
a stale executor from a previous retry. TaskProcessor cleanup remains as an
outer fallback. TaskProcessorTest now installs and closes two successive
statement contexts and verifies both closeables run and the thread-local
context is removed after each execution. Focused UT passed (4 tests),
Checkstyle passed, and the full FE Maven reactor built successfully.
--
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]
