github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3921408053
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java:
##########
@@ -59,12 +77,127 @@ public Map<BaseTableInfo, MTMVSnapshotIf>
getBaseTableSnapshotCache() {
return baseTableSnapshotCache;
}
+ /** Loads the union of mapped base partitions once per related table. */
+ public PreparedPartitionSnapshots preparePartitionSnapshots(Set<String>
mtmvPartitionNames)
+ throws AnalysisException {
+ return preparePartitionSnapshots(mtmvPartitionNames, false);
+ }
+
+ /** Loads only mappings whose persisted partition-name set still matches
and needs version comparison. */
+ public PreparedPartitionSnapshots
prepareComparablePartitionSnapshots(Set<String> mtmvPartitionNames)
+ throws AnalysisException {
+ return preparePartitionSnapshots(mtmvPartitionNames, true);
+ }
+
+ private PreparedPartitionSnapshots preparePartitionSnapshots(
+ Set<String> mtmvPartitionNames, boolean comparableOnly)
+ throws AnalysisException {
+ Map<MTMVRelatedTableIf, Set<String>> namesByTable = new
LinkedHashMap<>();
+ Map<MTMVRelatedTableIf, BaseTableInfo> tableInfos = comparableOnly
+ ? new LinkedHashMap<>() : Collections.emptyMap();
+ for (String mtmvPartitionName : mtmvPartitionNames) {
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry
+ : getByPartitionName(mtmvPartitionName).entrySet()) {
+ if (!entry.getKey().needAutoRefresh()) {
+ continue;
+ }
+ if (comparableOnly) {
+ BaseTableInfo tableInfo =
tableInfos.computeIfAbsent(entry.getKey(), BaseTableInfo::new);
+ if (!Objects.equals(entry.getValue(),
mtmv.getRefreshSnapshot()
+ .getPctSnapshots(mtmvPartitionName, tableInfo))) {
+ continue;
+ }
+ }
+ namesByTable.computeIfAbsent(entry.getKey(), ignored -> new
LinkedHashSet<>())
+ .addAll(entry.getValue());
+ }
+ }
+ for (Map.Entry<MTMVRelatedTableIf, Set<String>> entry :
namesByTable.entrySet()) {
+ loadSnapshots(entry.getKey(), entry.getValue());
+ }
+ return new PreparedPartitionSnapshots(this);
+ }
+
+ private void loadSnapshots(MTMVRelatedTableIf table, Set<String>
partitionNames) throws AnalysisException {
+ Map<String, MTMVSnapshotIf> cached =
partitionSnapshotCache.computeIfAbsent(
+ table, ignored -> new LinkedHashMap<>());
+ Set<String> knownMissing =
missingPartitionSnapshotCache.computeIfAbsent(
+ table, ignored -> new LinkedHashSet<>());
+ Set<String> missing = new LinkedHashSet<>(partitionNames);
+ missing.removeAll(cached.keySet());
+ missing.removeAll(knownMissing);
+ if (missing.isEmpty()) {
+ return;
+ }
+ Map<String, MTMVSnapshotIf> loaded = table.getPartitionSnapshots(
+ missing, this, resolveSnapshot(table));
+ if (loaded == null || loaded.containsKey(null) ||
loaded.containsValue(null)
+ || !missing.containsAll(loaded.keySet())) {
+ throw new AnalysisException("Invalid partition snapshot result for
table " + table.getName()
+ + ": requestedCount=" + missing.size() + ", returnedCount="
+ + (loaded == null ? "null" : loaded.size()));
+ }
+ cached.putAll(loaded);
+ missing.removeAll(loaded.keySet());
+ knownMissing.addAll(missing);
+ }
+
+ void recordPartitionSnapshotFailure(
+ MTMVRelatedTableIf table, String partitionName, AnalysisException
failure) {
+ partitionSnapshotFailureCache.computeIfAbsent(table, ignored -> new
LinkedHashMap<>())
+ .put(partitionName, failure);
+ }
+
public static MTMVRefreshContext buildContext(MTMV mtmv, Map<List<String>,
Set<String>> queryUsedPartitions)
throws AnalysisException {
- MTMVRefreshContext context = new MTMVRefreshContext(mtmv);
- context.partitionMappings =
mtmv.calculatePartitionMappings(queryUsedPartitions);
+ return buildContextInternal(mtmv, queryUsedPartitions, null);
+ }
+
+ public static MTMVRefreshContext buildContext(MTMV mtmv, Map<List<String>,
Set<String>> queryUsedPartitions,
+ Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots) throws
AnalysisException {
+ Map<MvccTableInfo, MvccSnapshot> snapshotCopy = new
LinkedHashMap<>(pinnedSnapshots);
+ return buildContextInternal(mtmv, queryUsedPartitions, snapshotCopy);
+ }
+
+ private static MTMVRefreshContext buildContextInternal(MTMV mtmv,
+ Map<List<String>, Set<String>> queryUsedPartitions,
+ Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots) throws
AnalysisException {
+ MTMVRefreshContext context = new MTMVRefreshContext(mtmv,
pinnedSnapshots);
+ context.partitionMappings =
mtmv.calculatePartitionMappings(queryUsedPartitions, pinnedSnapshots);
context.baseVersions = MTMVPartitionUtil.getBaseVersions(mtmv,
context.partitionMappings);
return context;
}
+ private Optional<MvccSnapshot> resolveSnapshot(MTMVRelatedTableIf table) {
Review Comment:
[P1] Apply the task pin to non-PCT table snapshots too. This resolver now
keeps mapping and PCT partition snapshots at S1, but
`MTMVPartitionUtil.getTableSnapshotFromContext` still calls
`MvccUtil.getSnapshotFromContext` directly. During `MTMVTask.run` the outer
context has no `StatementContext`, so an Iceberg/Paimon non-PCT base can
materialize latest S2 here; each later refresh SQL explicitly scans the task
pin S1, while `generatePartitionSnapshots` persists the cached S2. The MV can
then be considered fresh at S2 and used for rewrite although S2 was never
materialized. Reuse this context resolver for table-level snapshots and add a
non-PCT S1-to-S2 task interleaving test.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/CachingHmsClient.java:
##########
@@ -203,76 +210,292 @@ public List<String> listPartitionNamesFresh(String
dbName, String tableName, int
@Override
public List<HmsPartitionInfo> getPartitions(String dbName, String
tableName, List<String> partNames) {
- if (partNames == null || partNames.isEmpty()) {
- return Collections.emptyList();
- }
- // Per-partition assembly (Trino CachingHiveMetastore / legacy
HiveExternalMetaCache shape): serve each
- // requested partition from its own entry and fetch only the misses in
ONE delegate round-trip, so
- // overlapping requests share partition objects and the capacity
bounds partition OBJECTS, not
- // request-lists. Correctness is independent of name-parse fidelity: a
LOOKUP is keyed by the requested
- // name parsed to values, but a STORE is always keyed by the
partition's OWN values, so a name whose
- // parse diverges (a rare escaped value) simply misses and is
re-fetched — never a wrong or dropped
- // partition. Callers consume the result as a SET (they never rely on
order or 1:1 name↔result
- // correspondence — the delegate get_partitions_by_names never
guaranteed either).
- List<HmsPartitionInfo> result = new ArrayList<>(partNames.size());
- List<String> missNames = null;
- for (String name : partNames) {
- HmsPartitionInfo hit =
- partitionsCache.getIfPresent(new PartitionKey(dbName,
tableName, toPartitionValues(name)));
+ return getPartitionsWithStats(dbName, tableName,
partNames).getPartitions();
+ }
+
+ @Override
+ public HmsPartitionBatchResult getPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames) {
+ return getPartitionsWithStats(dbName, tableName, partNames, false);
+ }
+
+ @Override
+ public List<HmsPartitionInfo> getExistingPartitions(
+ String dbName, String tableName, List<String> partNames) {
+ return getExistingPartitionsWithStats(dbName, tableName,
partNames).getPartitions();
+ }
+
+ @Override
+ public HmsPartitionBatchResult getExistingPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames) {
+ return getPartitionsWithStats(dbName, tableName, partNames, true);
+ }
+
+ private HmsPartitionBatchResult getPartitionsWithStats(
+ String dbName, String tableName, List<String> partNames, boolean
allowMissing) {
+ long logicalStartNanos = System.nanoTime();
+ if (partNames.isEmpty()) {
+ HmsPartitionBatchStats stats = HmsPartitionBatchStats.builder()
+ .logicalElapsedNanos(System.nanoTime() - logicalStartNanos)
+ .build();
+ return new HmsPartitionBatchResult(Collections.emptyList(), stats);
+ }
+ HmsPartitionRequest request = new HmsPartitionRequest(dbName,
tableName, partNames);
+ // Keep the existing cache policy: aggregate every miss into one
logical delegate request and publish
+ // only after that request succeeds. Reassemble from partition
identities afterwards because a mixed
+ // hit/miss request must preserve the caller's exact order even when
HMS returns a different order.
+ List<List<String>> requestedValues = new ArrayList<>(partNames.size());
+ Map<List<String>, HmsPartitionInfo> resultByIdentity = new HashMap<>();
+ List<HmsPartitionIdentity.ParsedPartitionName> misses = new
ArrayList<>();
+ for (HmsPartitionIdentity.ParsedPartitionName partition :
request.getPartitions()) {
+ List<String> values = partition.getValues();
+ requestedValues.add(values);
+ HmsPartitionInfo hit = partitionsCache.getIfPresent(new
PartitionKey(dbName, tableName, values));
if (hit != null) {
- result.add(hit);
+ resultByIdentity.put(values, hit);
} else {
- if (missNames == null) {
- missNames = new ArrayList<>();
+ misses.add(partition);
+ }
+ }
+ PartitionStatsAccumulator physicalStats = new
PartitionStatsAccumulator();
+ if (!misses.isEmpty()) {
+ try {
+ loadMissingPartitions(dbName, tableName, allowMissing, misses,
+ resultByIdentity, physicalStats);
+ } catch (HmsClientException e) {
+ HmsPartitionBatchStats failedStats =
e.getPartitionBatchStats();
+ if (failedStats != null) {
+ physicalStats.add(failedStats);
+ e.withPartitionBatchStats(physicalStats.build(
+ partNames.size(), System.nanoTime() -
logicalStartNanos));
}
- missNames.add(name);
+ throw e;
+ }
+ }
+ List<HmsPartitionInfo> result = new ArrayList<>(partNames.size());
+ for (int i = 0; i < requestedValues.size(); i++) {
+ List<String> values = requestedValues.get(i);
+ if (allowMissing && !resultByIdentity.containsKey(values)) {
+ continue;
+ }
+ HmsPartitionInfo partition = resultByIdentity.get(values);
+ if (partition == null) {
+ throw HmsPartitionResultException.builder(partNames.size(),
resultByIdentity.size())
+ .missing(request.getPartitions().get(i).getName())
+ .build();
}
+ result.add(partition);
}
- if (missNames != null) {
- // Capture the invalidation generation BEFORE the delegate RPC so
a REFRESH (flush) that races this
- // in-flight cold-cache fetch does not get silently undone by
re-caching the pre-refresh partitions.
- // The pre-D2 code went through partitionsCache.get(key, loader)
-> getWithManualLoad, which had this
- // guard; the per-partition put must restore it
(getTable/listPartitionNames/getTableColumnStatistics
- // still use the guarded get path). The delegate results still
populate the RESULT list directly,
- // preserving the misparse->never-drop safety (only the CACHE put
is generation-guarded).
- try (MetaCache.BulkLoad<PartitionKey, HmsPartitionInfo> load =
- partitionsCache.beginBulkLoad(ScopePath.table(dbName,
tableName))) {
- for (HmsPartitionInfo info : delegate.getPartitions(dbName,
tableName, missNames)) {
- load.publish(new PartitionKey(dbName, tableName,
info.getValues()), info);
- result.add(info);
+ HmsPartitionBatchStats stats = physicalStats.build(
+ partNames.size(), System.nanoTime() - logicalStartNanos);
+ return new HmsPartitionBatchResult(result, stats);
+ }
+
+ private void loadMissingPartitions(String dbName, String tableName,
boolean allowMissing,
+ List<HmsPartitionIdentity.ParsedPartitionName> initialMisses,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ PartitionStatsAccumulator physicalStats) {
+ List<HmsPartitionIdentity.ParsedPartitionName> pending = initialMisses;
+ while (!pending.isEmpty()) {
+ // Elect one owner per missing identity. One caller can own a
batch and wait on identities owned by
+ // another caller, so partially overlapping requests still issue
one transport load per identity.
+ PartitionLoadBatch ownedBatch = new
PartitionLoadBatch(allowMissing);
+ List<PartitionLoadRegistration> owned = new ArrayList<>();
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting =
new IdentityHashMap<>();
+ try {
+ for (HmsPartitionIdentity.ParsedPartitionName partition :
pending) {
+ registerPartitionLoad(dbName, tableName, partition,
ownedBatch,
+ resultByIdentity, owned, waiting);
}
+ afterPartitionLoadRegistrationForTest();
+ if (!owned.isEmpty()) {
+ loadOwnedPartitions(dbName, tableName, allowMissing,
registrationsToPartitions(owned),
+ resultByIdentity, physicalStats, ownedBatch);
+ }
+ ownedBatch.complete(null);
+ } catch (RuntimeException | Error failure) {
+ ownedBatch.complete(failure);
+ releaseWaitingBatches(waiting.keySet());
+ throw failure;
+ } finally {
+ releaseOwnedPartitionLoads(ownedBatch);
}
+ List<HmsPartitionIdentity.ParsedPartitionName> retries = new
ArrayList<>();
+ consumeWaitingBatches(waiting, resultByIdentity, retries,
allowMissing);
+ pending = retries;
}
- return result;
}
- /**
- * Splits a Hive partition name ("c1=a/c2=b") into its ordered values
("a", "b"), unescaping each via
- * Hive's {@code FileUtils} (already a hms-module dependency — {@code
HmsEventParser} uses it). Semantics
- * match the write path's {@code HiveWriteUtils.toPartitionValues}, so
scan and write correlate partitions
- * identically. Only used to build the per-partition LOOKUP key: a parse
that diverges from the stored
- * partition's own values just misses and re-fetches (never a
wrong/dropped partition), so this is a
- * hit-rate optimization, not a correctness dependency.
- */
- private static List<String> toPartitionValues(String partitionName) {
- List<String> values = new ArrayList<>();
- int start = 0;
+ void afterPartitionLoadRegistrationForTest() {
+ }
+
+ void beforePartitionLoadElectionForTest() {
+ }
+
+ void afterPartitionLoadOwnershipForTest() {
+ }
+
+ int inFlightPartitionLoadCountForTest() {
+ return inFlightPartitionLoads.size();
+ }
+
+ private void registerPartitionLoad(String dbName, String tableName,
+ HmsPartitionIdentity.ParsedPartitionName partition,
PartitionLoadBatch ownedBatch,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<PartitionLoadRegistration> owned,
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting) {
+ PartitionKey key = new PartitionKey(dbName, tableName,
partition.getValues());
+ HmsPartitionInfo hit = partitionsCache.getIfPresent(key);
+ if (hit != null) {
+ resultByIdentity.put(partition.getValues(), hit);
+ return;
+ }
+ beforePartitionLoadElectionForTest();
while (true) {
- while (start < partitionName.length() &&
partitionName.charAt(start) != '=') {
- start++;
+ PartitionLoadBatch existing =
inFlightPartitionLoads.putIfAbsent(key, ownedBatch);
+ if (existing == null) {
+ ownedBatch.claimedKeys.add(key);
+ afterPartitionLoadOwnershipForTest();
+ hit = partitionsCache.getIfPresent(key);
+ if (hit == null) {
+ owned.add(new PartitionLoadRegistration(partition, key));
+ } else {
+ resultByIdentity.put(partition.getValues(), hit);
+ ownedBatch.claimedKeys.remove(key);
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ }
+ return;
}
- start++;
- int end = start;
- while (end < partitionName.length() && partitionName.charAt(end)
!= '/') {
- end++;
+ List<PartitionLoadRegistration> registrations =
waiting.get(existing);
+ if (registrations != null) {
+ registrations.add(new PartitionLoadRegistration(partition,
key));
+ return;
}
- if (start > partitionName.length()) {
- break;
+ if (existing.tryRegisterWaiter()) {
+ waiting.computeIfAbsent(existing, ignored -> new ArrayList<>())
+ .add(new PartitionLoadRegistration(partition, key));
+ return;
+ }
+ inFlightPartitionLoads.remove(key, existing);
+ }
+ }
+
+ private void loadOwnedPartitions(String dbName, String tableName, boolean
allowMissing,
+ List<HmsPartitionIdentity.ParsedPartitionName> owned,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ PartitionStatsAccumulator physicalStats, PartitionLoadBatch
ownedBatch) {
+ List<String> names = new ArrayList<>(owned.size());
+ for (HmsPartitionIdentity.ParsedPartitionName partition : owned) {
+ names.add(partition.getName());
+ }
+ MetaCache.BulkLoad<PartitionKey, HmsPartitionInfo> load =
+ partitionsCache.beginBulkLoad(ScopePath.table(dbName,
tableName));
+ ownedBatch.setLoad(load);
+ HmsPartitionBatchResult loadedResult = allowMissing
+ ? delegate.getExistingPartitionsWithStats(dbName, tableName,
names)
+ : delegate.getPartitionsWithStats(dbName, tableName, names);
+ physicalStats.add(loadedResult.getStats());
+ List<HmsPartitionInfo> loaded = loadedResult.getPartitions();
+ for (HmsPartitionInfo info : loaded) {
+ PartitionKey key = new PartitionKey(dbName, tableName,
info.getValues());
+ load.publish(key, info);
+ resultByIdentity.put(info.getValues(), info);
+ ownedBatch.resolvedPartitions.put(key, info);
+ }
+ }
+
+ private static List<HmsPartitionIdentity.ParsedPartitionName>
registrationsToPartitions(
+ List<PartitionLoadRegistration> registrations) {
+ List<HmsPartitionIdentity.ParsedPartitionName> partitions = new
ArrayList<>(registrations.size());
+ for (PartitionLoadRegistration registration : registrations) {
+ partitions.add(registration.partition);
+ }
+ return partitions;
+ }
+
+ private void releaseOwnedPartitionLoads(PartitionLoadBatch ownedBatch) {
+ for (PartitionKey key : ownedBatch.claimedKeys) {
+ inFlightPartitionLoads.remove(key, ownedBatch);
+ }
+ ownedBatch.releaseOwner();
+ }
+
+ private static void releaseWaitingBatches(Set<PartitionLoadBatch> batches)
{
+ for (PartitionLoadBatch batch : batches) {
+ batch.releaseWaiter();
+ }
+ }
+
+ private void consumeWaitingBatches(
+ Map<PartitionLoadBatch, List<PartitionLoadRegistration>> waiting,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<HmsPartitionIdentity.ParsedPartitionName> retries, boolean
allowMissing) {
+ List<Map.Entry<PartitionLoadBatch, List<PartitionLoadRegistration>>>
entries =
+ new ArrayList<>(waiting.entrySet());
+ for (int i = 0; i < entries.size(); i++) {
+ try {
+ Map.Entry<PartitionLoadBatch, List<PartitionLoadRegistration>>
entry = entries.get(i);
+ consumeWaitingBatch(entry.getKey(), entry.getValue(),
resultByIdentity, retries, allowMissing);
+ } catch (RuntimeException | Error failure) {
+ for (int remaining = i + 1; remaining < entries.size();
remaining++) {
+ entries.get(remaining).getKey().releaseWaiter();
+ }
+ throw failure;
}
-
values.add(FileUtils.unescapePathName(partitionName.substring(start, end)));
- start = end + 1;
}
- return values;
+ }
+
+ private void consumeWaitingBatch(PartitionLoadBatch batch,
+ List<PartitionLoadRegistration> registrations,
+ Map<List<String>, HmsPartitionInfo> resultByIdentity,
+ List<HmsPartitionIdentity.ParsedPartitionName> retries, boolean
allowMissing) {
+ try {
+ Throwable failure = batch.await();
+ if (failure != null) {
+ if (batch.allowMissing != allowMissing
Review Comment:
[P2] Share request-independent terminal failures with partial waiters. This
compatibility check runs before the failure is classified, so a waiter covering
a subset (or using the other missing-result contract) retries even
pool/client-creation, authentication, connection failures, and `Error`. For one
owner of N cold keys plus N singleton waiters, a global owner failure releases
every waiter into a separate delegate attempt under the same outage, defeating
single-flight and potentially saturating the metadata pool. Only
result-integrity failures can depend on identities outside a waiter's request;
propagate request-independent failures and `Error` to all overlapping waiters,
and add a broad-owner/singleton-waiters outage test.
--
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]
