morrySnow commented on code in PR #66530:
URL: https://github.com/apache/doris/pull/66530#discussion_r3859482034
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -414,20 +417,55 @@ public boolean supportsFileCache() {
*/
@Override
public List<ConnectorScanRange> planScan(ConnectorSession session,
ConnectorScanRequest request) {
- IcebergTableHandle handle = (IcebergTableHandle)
request.getTableHandle();
+ IcebergTableHandle icebergHandle = (IcebergTableHandle)
request.getTableHandle();
try {
- return planScanInternal(session, handle, request.getColumns(),
- request.getFilter(), request.isCountPushdown());
+ if (!isExternalScanTaskReuseEnabled(session)) {
+ return planScanInternal(session, icebergHandle,
request.getColumns(),
+ request.getFilter(), request.isCountPushdown());
+ }
+ if (icebergHandle.isSystemTable()) {
+ // System tables read connector metadata through their own
readers; never reuse them.
+ return planScanInternal(session, icebergHandle,
request.getColumns(),
+ request.getFilter(), request.isCountPushdown());
+ }
+ // Statement-scoped reuse: within one statement the identical scan
(same table, same
+ // snapshot/ref/schema pin, same filter, same COUNT pushdown)
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 from the key.
+ //
+ // The ranges are memoized in a map HUNG INSIDE the statement
scope rather than cached
+ // directly: planScanInternal re-enters the scope itself
(sharedTable and the v3
+ // rewritableDeleteSupply are scope-backed), and a loader of the
scope's
+ // ConcurrentHashMap must not touch that map (same-bin re-entry
throws
+ // IllegalStateException("Recursive update"); a mid-computation
resize silently drops the
+ // 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.
+ String memoKey = SCAN_REUSE_NAMESPACE + ":" +
session.getCatalogId() + ":" + session.getQueryId();
+ Map<IcebergScanReuseKey, List<ConnectorScanRange>> scanReuse =
+ session.getStatementScope().computeIfAbsent(memoKey, () ->
new ConcurrentHashMap<>());
+ IcebergScanReuseKey reuseKey = new
IcebergScanReuseKey(icebergHandle, request);
+ return scanReuse.computeIfAbsent(reuseKey,
+ k -> Collections.unmodifiableList(planScanInternal(session,
+ icebergHandle, request.getColumns(),
request.getFilter(),
+ request.isCountPushdown())));
} catch (RuntimeException e) {
// Normal data scans and native position_deletes run on File
Scanner V2. Keep the serialized JNI
// system-table route untouched because its deferred reads belong
to the V1 scanner contract.
- if (!handle.isSystemTable() || isPositionDeletesSysTable(handle)) {
- throw IcebergExceptionUtils.wrapMetadataReadFailure(handle, e);
+ if (!icebergHandle.isSystemTable() ||
isPositionDeletesSysTable(icebergHandle)) {
+ throw
IcebergExceptionUtils.wrapMetadataReadFailure(icebergHandle, e);
}
throw e;
}
}
+ private static boolean isExternalScanTaskReuseEnabled(ConnectorSession
session) {
+ return session != null && "true".equalsIgnoreCase(
+
session.getSessionProperties().get(ENABLE_EXTERNAL_SCAN_TASK_REUSE));
+ }
Review Comment:
这个是不是应当在session里面直接提供方法？而不是让不同的datasource自己去写？
--
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]
