github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3891380166
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java:
##########
@@ -251,6 +251,10 @@ public void run() throws JobException {
if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
return;
}
+ // Snapshot persistence happens after refresh partitions are split
into execution groups. Load the
+ // complete union here so the default one-partition group size
cannot turn a large Hive MTMV into
+ // one metadata request per MV partition;
generatePartitionSnapshots reuses this context cache.
+
context.preloadPartitionSnapshots(Sets.newHashSet(needRefreshPartitions));
Review Comment:
[P1] Preload with the same MVCC pins used by the refresh SQL.
`beforeMTMVRefresh` captures S1 only in this task's private `snapshots` map,
while the outer thread-local context still has no `StatementContext`; this call
therefore passes an empty pin and can materialize latest S2 for every group.
Each later `exec` explicitly installs S1 and scans it, but snapshot persistence
reuses the cached S2, so the MV can be marked fresh at S2 after reading only
S1. This is newly exposed for later groups by moving their loads before the
first pinned exec. Install/pass the task pins before this union preload and add
a two-group S1-to-S2 interleaving test.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -499,11 +509,18 @@ private List<PartitionScanInfo>
resolvePartitions(HiveTableHandle handle) {
if (partNames.isEmpty()) {
return Collections.emptyList();
}
- List<HmsPartitionInfo> hmsPartitions = hmsClient.getPartitions(
+ List<HmsPartitionInfo> hmsPartitions = loadPartitionsWithProfile(
handle.getDbName(), handle.getTableName(), partNames);
Review Comment:
[P2] Preserve the batch stats from partition predicate pushdown. A selective
scan resolves its nonempty subset earlier in
`HiveConnectorMetadata.applyFilter` via `hmsClient.getPartitions`, which
computes the new stats and then discards them when it stores only partition
objects on the handle. This pruned-handle branch consequently bypasses
`loadPartitionsWithProfile`, so the query profile reports no HMS
partition-batch request for precisely these scans. Carry the pruning result's
stats to this provider's one-time drain (or publish them once at the request
boundary), and cover the production applyFilter-to-planScan path.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveScanPlanProvider.java:
##########
@@ -725,4 +742,72 @@ private static final class PartitionScanInfo {
private ConnectorStorageContext storage() {
return context.getStorageContext();
}
+
+ /** Thread-safe aggregation because partition-batch scan planning runs on
the shared metadata executor. */
+ private static final class PartitionBatchProfile {
+ private String tableLabel;
+ private int logicalRequests;
+ private long requestedItems;
+ private long rpcAttempts;
+ private long rpcItems;
+ private int largestBatchSize;
+ private int smallestBatchSize;
+ private long fallbacks;
+ private long logicalElapsedNanos;
+ private long rpcElapsedNanos;
+ private long maxRpcElapsedNanos;
+
+ synchronized void record(String dbName, String tableName,
HmsPartitionBatchStats stats) {
+ tableLabel = dbName + "." + tableName;
+ logicalRequests++;
+ requestedItems += stats.getRequestedItems();
+ rpcAttempts += stats.getRpcAttempts();
+ rpcItems += stats.getRpcItems();
+ largestBatchSize = Math.max(largestBatchSize,
stats.getLargestBatchSize());
+ if (stats.getSmallestBatchSize() > 0) {
+ smallestBatchSize = smallestBatchSize == 0
+ ? stats.getSmallestBatchSize()
+ : Math.min(smallestBatchSize,
stats.getSmallestBatchSize());
+ }
+ fallbacks += stats.getFallbackCount();
+ logicalElapsedNanos += stats.getLogicalElapsedNanos();
+ rpcElapsedNanos += stats.getRpcElapsedNanos();
+ maxRpcElapsedNanos = Math.max(maxRpcElapsedNanos,
stats.getMaxRpcElapsedNanos());
+ }
+
+ synchronized List<ConnectorScanProfile> drain() {
+ if (logicalRequests == 0) {
+ return Collections.emptyList();
+ }
+ Map<String, String> metrics = new LinkedHashMap<>();
+ metrics.put("LogicalRequests", String.valueOf(logicalRequests));
+ metrics.put("RequestedItems", String.valueOf(requestedItems));
+ metrics.put("RpcAttempts", String.valueOf(rpcAttempts));
+ metrics.put("RpcItems", String.valueOf(rpcItems));
Review Comment:
[P2] Give each scan node a distinct profile child (or merge same-table
metrics). Every provider scanning the same `db.table` emits this identical
group/label, while `RuntimeProfile.addChild` removes and replaces an existing
child with the same name. A self-join or repeated scan therefore retains only
the last node's requested-item/RPC totals instead of reporting both. Please
carry a stable scan-instance identity or deliberately aggregate collisions,
with a two-provider same-table test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java:
##########
@@ -1951,6 +1954,14 @@ public void startSplit(int numBackends) {
splitAssignment.setException(batchException.get());
}
if (numFinishedPartitions.addAndGet(curBatchSize)
== allPartitions.size()) {
+ try {
+ List<ConnectorScanProfile> profiles =
onPluginClassLoader(scanProvider,
+ () ->
scanProvider.collectScanProfiles(connectorSession));
Review Comment:
[P2] Make profile-group creation atomic before publishing from batch
workers. Two large-table scan nodes in the same query can finish concurrently
here and both observe no `Connector Metadata Access` group in
`writeScanProfilesInto`; each creates a different group, and
`RuntimeProfile.addChild` then replaces the first same-named group with the
second, leaving one scan child unreachable even when the table labels differ.
Serialize/get-or-create through a RuntimeProfile-owned atomic API (or publish
on a serialized owner thread), and add a barrier test with two concurrent batch
finalizers.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -1431,6 +1431,26 @@ public OptionalLong
getPartitionFreshnessMillis(ConnectorSession session, Connec
return
OptionalLong.of(lastDdlMillis(partitions.get(0).getParameters()));
}
+ @Override
+ public Map<String, Long> getPartitionsFreshnessMillis(ConnectorSession
session, ConnectorTableHandle handle,
+ List<String> partitionNames) {
+ if (!(handle instanceof HiveTableHandle)) {
+ return siblingMetadata(session, handle)
+ .getPartitionsFreshnessMillis(session, handle,
partitionNames);
+ }
+ if (partitionNames.isEmpty()) {
+ return Collections.emptyMap();
+ }
+ HiveTableHandle hiveHandle = (HiveTableHandle) handle;
+ List<HmsPartitionInfo> partitions = hmsClient.getPartitions(
Review Comment:
[P2] Preserve the freshness API's vanished-partition result. A partition can
be present in the statement pin and then be dropped before this bulk fetch; the
SPI deliberately represents that race by omitting the name so fe-core raises
its checked `AnalysisException("can not find partition")`. The new exact
`HmsClient#getPartitions` path throws `HmsPartitionResultException` for the
missing object before this method can build such a map, so MTMV callers that
isolate `AnalysisException` instead receive an unchecked connector failure (the
scalar empty branch is now unreachable in production too). Keep strict
duplicate/unexpected/malformed validation, but translate missing-only results
for the freshness path into absent entries, and cover a production-stack
list/pin-then-drop race for both scalar and bulk probes.
--
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]
