github-actions[bot] commented on code in PR #67186:
URL: https://github.com/apache/doris/pull/67186#discussion_r3891887506
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveConnectorMetadata.java:
##########
@@ -1182,17 +1183,19 @@ public
Optional<FilterApplicationResult<ConnectorTableHandle>> applyFilter(
return Optional.empty();
}
- List<HmsPartitionInfo> prunedPartitions = matchedPartNames.isEmpty()
- ? Collections.emptyList()
- : hmsClient.getPartitions(hiveHandle.getDbName(),
- hiveHandle.getTableName(), matchedPartNames);
+ HmsPartitionBatchResult pruningResult = matchedPartNames.isEmpty()
+ ? null : hmsClient.getPartitionsWithStats(
Review Comment:
[P1] Keep list-derived scan lookups tolerant of vanished partitions. The
names here come from `listPartitionNames` (and can remain cached for the
24-hour default TTL), but this new exact-result call rejects an HMS response
that omits a partition dropped after that listing. The same list-derived exact
flow remains in ordinary/batch scan planning and write-plan discovery, so an
external drop can turn formerly omission-tolerant reads into repeatedly failing
queries until the name cache refreshes. Please use an omission-tolerant
result-with-stats path for identities derived from a prior listing, while
keeping caller-owned transaction identities exact, and cover cached
list-then-drop scans.
##########
fe/fe-connector/fe-connector-hms/src/main/java/org/apache/doris/connector/hms/ThriftHmsClient.java:
##########
@@ -909,6 +997,38 @@ public void close() throws Exception {
}
}
+ /** Reuses one temporary client across successful chunks when catalog
pooling is disabled. */
+ private final class UnpooledPartitionTransport implements
HmsPartitionTransport, AutoCloseable {
+ private PooledHmsClient current;
+
+ @Override
+ public List<HmsPartitionInfo> getPartitionsByNames(
+ String dbName, String tableName, List<String> partitionNames) {
+ if (closed) {
+ throw new HmsClientException("HMS client is closed");
+ }
+ if (current == null) {
+ current = createFreshClient();
+ }
+ try {
+ return executePartitionCall(current,
+ client -> loadPartitionsByNames(client, dbName,
tableName, partitionNames));
+ } catch (RuntimeException e) {
+ current.destroy();
Review Comment:
[P2] Preserve the remote failure when destroying the unpooled client. If
`current.destroy()` throws from `IMetaStoreClient.close()`, control never
reaches `current = null` or `throw e`; the executor sees the cleanup exception
instead of the explicit size-limit `RemoteCallException`, so it does not
halve/retry. The pooled try-with-resources path keeps the remote error primary
and suppresses cleanup failure. Please detach the failed client first and add
any destroy failure as suppressed to `e`, then test pool-size-zero fallback
with a throwing close.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRefreshContext.java:
##########
@@ -59,12 +76,103 @@ 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> missing = new LinkedHashSet<>(partitionNames);
+ missing.removeAll(cached.keySet());
+ if (missing.isEmpty()) {
+ return;
+ }
+ Map<String, MTMVSnapshotIf> loaded = table.getPartitionSnapshots(
+ missing, this, snapshotResolver.apply(table));
+ if (loaded == null || loaded.containsKey(null) ||
loaded.containsValue(null)
+ || !missing.equals(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);
+ }
+
public static MTMVRefreshContext buildContext(MTMV mtmv, Map<List<String>,
Set<String>> queryUsedPartitions)
throws AnalysisException {
- MTMVRefreshContext context = new MTMVRefreshContext(mtmv);
+ return buildContext(mtmv, queryUsedPartitions,
MvccUtil::getSnapshotFromContext);
+ }
+
+ 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 buildContext(mtmv, queryUsedPartitions,
+ table -> Optional.ofNullable(snapshotCopy.get(new
MvccTableInfo(table))));
+ }
+
+ private static MTMVRefreshContext buildContext(MTMV mtmv,
+ Map<List<String>, Set<String>> queryUsedPartitions,
+ Function<MTMVRelatedTableIf, Optional<MvccSnapshot>>
snapshotResolver) throws AnalysisException {
+ MTMVRefreshContext context = new MTMVRefreshContext(mtmv,
snapshotResolver);
Review Comment:
[P1] Build the partition mapping from the same pin as this context's bulk
loader. The new resolver makes `loadSnapshots` read S1, but
`calculatePartitionMappings` still reaches `MvccUtil.getSnapshotFromContext`
through its generators after the task's temporary `StatementContext` has been
restored. If a PCT partition is added after S1, the mapping can include that
S2-only name and the newly S1-pinned bulk load then rejects it as absent,
aborting refresh; before this change both mapping and scalar snapshot loading
used the same unpinned generation. Route mapping/align generation through this
resolver (or install the task pins before building it) and add an
add-between-capture-and-mapping 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]
