github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3877761197
##########
fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java:
##########
@@ -1350,6 +1354,34 @@ public synchronized void
addExternalTableGetFileScanTasksTime(long ms) {
addExternalCatalogMetaTimeInternal(ms);
}
+ /** Records one completed connector metadata request in this query's
profile. */
+ public synchronized void recordConnectorMetadataAccess(
+ String catalogName, ConnectorMetadataAccessEvent event) {
+ String key = catalogName + '\0' + event.getOperation() + '\0' +
event.getSource();
+ MetadataAccessProfileCounters counters =
connectorMetadataAccessCounters.get(key);
+ if (counters == null) {
+ counters = createMetadataAccessProfileCounters(catalogName, event);
+ connectorMetadataAccessCounters.put(key, counters);
+ }
+ counters.record(event);
+ addExternalCatalogMetaTimeInternal(event.getLogicalElapsedMillis());
Review Comment:
**[P2] Do not add nested wait spans to the legacy total.** The outer
`hms.get_partitions_by_names` timer starts before cache/load coordination, so
it already includes both in-flight and load-slot waits. Those waits are also
emitted as their own logical events, and this unconditional addition counts the
same interval again in `External Table Meta Time` (for example, a roughly 1s
pure waiter reports roughly 2s). Keep the detailed wait child counters, but
aggregate only top-level operations into the legacy total, with a
waiter-profile test that feeds both events.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java:
##########
@@ -105,6 +119,13 @@ public class ThriftHmsClient implements HmsClient {
private final AuthAction authAction;
private final MetaStoreClientProvider clientProvider;
private final HmsTypeMapping.Options typeMappingOptions;
+ private final HmsPartitionBatchLoader partitionBatchLoader;
+ private final ExecutorService clientCreationExecutor =
Executors.newCachedThreadPool(runnable -> {
Review Comment:
**[P1] Bound outstanding client creators.** `newCachedThreadPool` admits a
fresh creator for every request, while `discardLateClient()` can only interrupt
the old task. The new blocking-provider test explicitly models DNS/Kerberos
code that ignores interruption; after that caller aborts, a later request can
submit another creator and leave the first thread alive indefinitely. Repeated
deadlines/KILLs can therefore grow one daemon thread per attempt until the FE
exhausts threads. Use bounded creation admission and retain the in-progress
slot until a non-cooperative task actually exits, with repeated-abort coverage
asserting creator concurrency stays bounded.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java:
##########
@@ -55,15 +82,233 @@ public MTMVBaseVersions getBaseVersions() {
return baseVersions;
}
+ public Set<BaseTableInfo> getBaseTables() {
+ return baseTables;
+ }
+
+ public Set<MTMVRelatedTableIf> getPctTables() {
+ return pctTables;
+ }
+
+ public MTMVPartitionInfo.MTMVPartitionType getPartitionType() {
+ return partitionType;
+ }
+
+ public ConnectorMetadataAccessSource getMetadataAccessSource() {
+ return metadataAccessSource;
+ }
+
public Map<BaseTableInfo, MTMVSnapshotIf> getBaseTableSnapshotCache() {
return baseTableSnapshotCache;
}
- public static MTMVRefreshContext buildContext(MTMV mtmv, Map<List<String>,
Set<String>> queryUsedPartitions)
+ public TableIf getBaseTable(BaseTableInfo baseTableInfo) throws
AnalysisException {
+ AnalysisException previousFailure =
baseTableResolutionFailures.get(baseTableInfo);
+ if (previousFailure != null) {
+ throw previousFailure;
+ }
+ TableIf table = resolvedBaseTables.get(baseTableInfo);
+ if (table != null) {
+ return table;
+ }
+ try {
+ table = MTMVUtil.getTable(baseTableInfo);
+ resolvedBaseTables.put(baseTableInfo, table);
+ return table;
+ } catch (AnalysisException e) {
+ baseTableResolutionFailures.put(baseTableInfo, e);
+ throw e;
+ }
+ }
+
+ public Map<String, MTMVSnapshotIf>
getPartitionSnapshots(MTMVRelatedTableIf table,
+ Set<String> partitionNames, Optional<MvccSnapshot> snapshot)
throws AnalysisException {
+ Map<String, MTMVSnapshotIf> cached =
partitionSnapshotCache.computeIfAbsent(
+ table, ignored -> Maps.newHashMap());
+ Set<String> missing = Sets.difference(partitionNames,
cached.keySet()).copyInto(Sets.newLinkedHashSet());
+ if (!missing.isEmpty()) {
+ Map<String, MTMVSnapshotIf> loaded = table.getPartitionSnapshots(
+ new ArrayList<>(missing), this, snapshot);
+ if (!loaded.keySet().containsAll(missing)) {
+ Set<String> absent = Sets.difference(missing, loaded.keySet());
+ throw new AnalysisException("can not find partitions: " +
absent);
+ }
+ cached.putAll(loaded);
+ }
+ Map<String, MTMVSnapshotIf> result = new LinkedHashMap<>();
+ for (String partitionName : partitionNames) {
+ result.put(partitionName, cached.get(partitionName));
+ }
+ return Collections.unmodifiableMap(result);
+ }
+
+ public MTMVSnapshotIf getCachedPartitionSnapshot(MTMVRelatedTableIf table,
String partitionName) {
+ Map<String, MTMVSnapshotIf> cached = partitionSnapshotCache.get(table);
+ return cached == null ? null : cached.get(partitionName);
+ }
+
+ boolean hasPersistedPartitionSet(String mtmvPartitionName,
MTMVRelatedTableIf table,
+ Set<String> currentPartitions) {
+ BaseTableInfo tableInfo = pctTableInfos.computeIfAbsent(table,
BaseTableInfo::new);
+ return Objects.equals(currentPartitions,
+ mtmv.getRefreshSnapshot().getPctSnapshots(mtmvPartitionName,
tableInfo));
+ }
+
+ boolean persistedPartitionSetsMatch(String mtmvPartitionName) {
+ if (partitionType == MTMVPartitionInfo.MTMVPartitionType.SELF_MANAGE) {
+ return true;
+ }
+ Map<MTMVRelatedTableIf, Set<String>> mapping =
getByPartitionName(mtmvPartitionName);
+ if (mapping == null || mapping.isEmpty()) {
+ return false;
+ }
+ for (MTMVRelatedTableIf table : pctTables) {
+ if (table.needAutoRefresh() && !hasPersistedPartitionSet(
+ mtmvPartitionName, table, mapping.getOrDefault(table,
Collections.emptySet()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ boolean persistedPartitionSetsMatch(Set<String> mtmvPartitionNames) {
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ if (!persistedPartitionSetsMatch(mtmvPartitionName)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Preloads every external PCT partition after mapping capture and before
refresh calculations take locks. */
+ public void preloadPartitionSnapshots() throws AnalysisException {
+ preloadPartitionSnapshots(partitionMappings.keySet());
+ }
+
+ /** Preloads the external PCT partition union required by the selected
MTMV partitions. */
+ public void preloadPartitionSnapshots(Set<String> mtmvPartitionNames)
throws AnalysisException {
+ Map<MTMVRelatedTableIf, Set<String>> namesByTable = Maps.newHashMap();
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ Map<MTMVRelatedTableIf, Set<String>> mapping =
getByPartitionName(mtmvPartitionName);
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry :
mapping.entrySet()) {
+ if (entry.getKey().needAutoRefresh()
+ &&
entry.getKey().supportsPartitionSnapshotBatchLoading()
+ && hasPersistedPartitionSet(mtmvPartitionName,
entry.getKey(), entry.getValue())) {
+ namesByTable.computeIfAbsent(entry.getKey(), ignored ->
Sets.newLinkedHashSet())
+ .addAll(entry.getValue());
+ }
+ }
+ }
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry :
namesByTable.entrySet()) {
+ getPartitionSnapshots(entry.getKey(), entry.getValue(),
+ MvccUtil.getSnapshotFromContext(entry.getKey()));
+ }
+ }
+
+ /** Preloads every non-PCT table snapshot used by the context. */
+ public void preloadTableSnapshots() throws AnalysisException {
+ preloadTableSnapshots(baseTables, Collections.emptySet());
+ }
+
+ /** Preloads only the non-PCT table snapshots that the following freshness
comparison can observe. */
+ public void preloadTableSnapshots(Set<BaseTableInfo> tables,
Set<TableNameInfo> excludeTables)
+ throws AnalysisException {
+ for (BaseTableInfo baseTableInfo : tables) {
+ if (MTMVPartitionUtil.isTableExcluded(excludeTables,
baseTableInfo)) {
+ continue;
+ }
+ TableIf table = getBaseTable(baseTableInfo);
+ if (!(table instanceof MTMVRelatedTableIf)) {
+ continue;
+ }
+ MTMVRelatedTableIf relatedTable = (MTMVRelatedTableIf) table;
+ if (!relatedTable.needAutoRefresh()) {
+ continue;
+ }
+ if (partitionType !=
MTMVPartitionInfo.MTMVPartitionType.SELF_MANAGE
+ && pctTables.contains(relatedTable)) {
+ continue;
+ }
+ MTMVPartitionUtil.getTableSnapshotFromContext(relatedTable, this);
+ }
+ }
+
+ public void preloadSnapshots() throws AnalysisException {
+ preloadPartitionSnapshots();
+ preloadTableSnapshots();
+ }
+
+ public void preloadSnapshots(Set<String> mtmvPartitionNames) throws
AnalysisException {
+ preloadPartitionSnapshots(mtmvPartitionNames);
+ preloadTableSnapshots();
+ }
+
+ /** Preloads the exact partition/table snapshot set used by one freshness
comparison. */
+ public void preloadSnapshots(Set<String> mtmvPartitionNames,
Set<BaseTableInfo> tables,
+ Set<TableNameInfo> excludeTables) throws AnalysisException {
+ preloadPartitionSnapshots(mtmvPartitionNames);
+ preloadTableSnapshots(tables, excludeTables);
+ }
+
+ public void refreshLocalState() throws AnalysisException {
+ for (MTMVRelatedTableIf pctTable : pctTables) {
+ if (pctTable instanceof OlapTable) {
+ partitionItems.put(pctTable,
+
pctTable.getAndCopyPartitionItems(MvccUtil.getSnapshotFromContext(pctTable)));
+ }
+ }
+ partitionMappings = partitionItems.isEmpty()
+ ? mtmv.calculatePartitionMappings(queryUsedPartitions)
+ : mtmv.calculatePartitionMappings(queryUsedPartitions,
partitionItems);
+ baseVersions = MTMVPartitionUtil.getBaseVersions(mtmv,
partitionMappings, baseTables);
Review Comment:
**[P2] Keep cloud version RPCs outside FE table locks.**
`refreshLocalState()` reaches `Partition.getVisibleVersions()` and
`OlapTable.getVisibleVersionInBatch()`; in cloud mode their TTL-zero/expired
paths synchronously call meta service. `MTMVTask` and `PartitionsProcDir`
invoke this helper while holding all sorted table read locks, and rewrite
builds the same context after `StatementContext.lock()`, so a cold/disabled
version cache blocks DDL and writers for remote latency. This is earlier than
the connector freshness preload and still occurs on persisted-set short
circuits. Split locked structural capture from remote version loading and
revalidate afterward; add blocking cache-miss tests for task, display, and
rewrite.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java:
##########
@@ -55,15 +82,233 @@ public MTMVBaseVersions getBaseVersions() {
return baseVersions;
}
+ public Set<BaseTableInfo> getBaseTables() {
+ return baseTables;
+ }
+
+ public Set<MTMVRelatedTableIf> getPctTables() {
+ return pctTables;
+ }
+
+ public MTMVPartitionInfo.MTMVPartitionType getPartitionType() {
+ return partitionType;
+ }
+
+ public ConnectorMetadataAccessSource getMetadataAccessSource() {
+ return metadataAccessSource;
+ }
+
public Map<BaseTableInfo, MTMVSnapshotIf> getBaseTableSnapshotCache() {
return baseTableSnapshotCache;
}
- public static MTMVRefreshContext buildContext(MTMV mtmv, Map<List<String>,
Set<String>> queryUsedPartitions)
+ public TableIf getBaseTable(BaseTableInfo baseTableInfo) throws
AnalysisException {
+ AnalysisException previousFailure =
baseTableResolutionFailures.get(baseTableInfo);
+ if (previousFailure != null) {
+ throw previousFailure;
+ }
+ TableIf table = resolvedBaseTables.get(baseTableInfo);
+ if (table != null) {
+ return table;
+ }
+ try {
+ table = MTMVUtil.getTable(baseTableInfo);
+ resolvedBaseTables.put(baseTableInfo, table);
+ return table;
+ } catch (AnalysisException e) {
+ baseTableResolutionFailures.put(baseTableInfo, e);
+ throw e;
+ }
+ }
+
+ public Map<String, MTMVSnapshotIf>
getPartitionSnapshots(MTMVRelatedTableIf table,
+ Set<String> partitionNames, Optional<MvccSnapshot> snapshot)
throws AnalysisException {
+ Map<String, MTMVSnapshotIf> cached =
partitionSnapshotCache.computeIfAbsent(
+ table, ignored -> Maps.newHashMap());
+ Set<String> missing = Sets.difference(partitionNames,
cached.keySet()).copyInto(Sets.newLinkedHashSet());
+ if (!missing.isEmpty()) {
+ Map<String, MTMVSnapshotIf> loaded = table.getPartitionSnapshots(
+ new ArrayList<>(missing), this, snapshot);
+ if (!loaded.keySet().containsAll(missing)) {
+ Set<String> absent = Sets.difference(missing, loaded.keySet());
+ throw new AnalysisException("can not find partitions: " +
absent);
+ }
+ cached.putAll(loaded);
+ }
+ Map<String, MTMVSnapshotIf> result = new LinkedHashMap<>();
+ for (String partitionName : partitionNames) {
+ result.put(partitionName, cached.get(partitionName));
+ }
+ return Collections.unmodifiableMap(result);
+ }
+
+ public MTMVSnapshotIf getCachedPartitionSnapshot(MTMVRelatedTableIf table,
String partitionName) {
+ Map<String, MTMVSnapshotIf> cached = partitionSnapshotCache.get(table);
+ return cached == null ? null : cached.get(partitionName);
+ }
+
+ boolean hasPersistedPartitionSet(String mtmvPartitionName,
MTMVRelatedTableIf table,
+ Set<String> currentPartitions) {
+ BaseTableInfo tableInfo = pctTableInfos.computeIfAbsent(table,
BaseTableInfo::new);
+ return Objects.equals(currentPartitions,
+ mtmv.getRefreshSnapshot().getPctSnapshots(mtmvPartitionName,
tableInfo));
+ }
+
+ boolean persistedPartitionSetsMatch(String mtmvPartitionName) {
+ if (partitionType == MTMVPartitionInfo.MTMVPartitionType.SELF_MANAGE) {
+ return true;
+ }
+ Map<MTMVRelatedTableIf, Set<String>> mapping =
getByPartitionName(mtmvPartitionName);
+ if (mapping == null || mapping.isEmpty()) {
+ return false;
+ }
+ for (MTMVRelatedTableIf table : pctTables) {
+ if (table.needAutoRefresh() && !hasPersistedPartitionSet(
+ mtmvPartitionName, table, mapping.getOrDefault(table,
Collections.emptySet()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ boolean persistedPartitionSetsMatch(Set<String> mtmvPartitionNames) {
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ if (!persistedPartitionSetsMatch(mtmvPartitionName)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /** Preloads every external PCT partition after mapping capture and before
refresh calculations take locks. */
+ public void preloadPartitionSnapshots() throws AnalysisException {
+ preloadPartitionSnapshots(partitionMappings.keySet());
+ }
+
+ /** Preloads the external PCT partition union required by the selected
MTMV partitions. */
+ public void preloadPartitionSnapshots(Set<String> mtmvPartitionNames)
throws AnalysisException {
+ Map<MTMVRelatedTableIf, Set<String>> namesByTable = Maps.newHashMap();
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ Map<MTMVRelatedTableIf, Set<String>> mapping =
getByPartitionName(mtmvPartitionName);
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry :
mapping.entrySet()) {
+ if (entry.getKey().needAutoRefresh()
+ &&
entry.getKey().supportsPartitionSnapshotBatchLoading()
+ && hasPersistedPartitionSet(mtmvPartitionName,
entry.getKey(), entry.getValue())) {
Review Comment:
**[P1] Use an ungated union preload for snapshot persistence.** The
persisted-set predicate is correct for stale comparison, but `MTMVTask` also
calls this method after deciding to refresh so it can batch the snapshots that
will be persisted. On a first/incomplete baseline or a changed mapping, the old
set is empty/different, every affected mapping is rejected here, and
`generatePartitionSnapshots()` later loads each MV partition separately. A
one-to-one 160k Hive MV therefore becomes roughly 160k logical one-name HMS
requests instead of one union request. Split comparison-gated and persistence
preloads, and cover first, partially missing, and mapping-changed baselines
with request-count assertions.
--
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]
