QueryCacheNormalizer.java
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
// This file is copied from
// https://github.com/apache/impala/blob/branch-2.9.0/fe/src/main/java/org/apache/impala/PlanFragment.java
// and modified by Doris
package org.apache.doris.planner.normalize;
import org.apache.doris.analysis.DescriptorTable;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Tablet;
import org.apache.doris.common.Config;
import org.apache.doris.common.Pair;
import org.apache.doris.planner.AggregationNode;
import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.OlapScanNode;
import org.apache.doris.planner.PlanFragment;
import org.apache.doris.planner.PlanNode;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.thrift.TNormalizedPlanNode;
import org.apache.doris.thrift.TQueryCacheParam;
import org.apache.doris.thrift.TStringLiteral;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TCompactProtocol;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
/** QueryCacheNormalizer */
public class QueryCacheNormalizer implements Normalizer {
private final PlanFragment fragment;
private final DescriptorTable descriptorTable;
private final NormalizedIdGenerator normalizedPlanIds = new NormalizedIdGenerator();
private final NormalizedIdGenerator normalizedTupleIds = new NormalizedIdGenerator();
private final NormalizedIdGenerator normalizedSlotIds = new NormalizedIdGenerator();
// result
private final TQueryCacheParam queryCacheParam = new TQueryCacheParam();
public QueryCacheNormalizer(PlanFragment fragment, DescriptorTable descriptorTable) {
this.fragment = Objects.requireNonNull(fragment, "fragment can not be null");
this.descriptorTable = Objects.requireNonNull(descriptorTable, "descriptorTable can not be null");
}
public Optional<TQueryCacheParam> normalize(ConnectContext context) {
try {
Optional<CachePoint> cachePoint = computeCachePoint();
if (!cachePoint.isPresent()) {
return Optional.empty();
}
List<TNormalizedPlanNode> normalizedDigestPlans = normalizePlanTree(context, cachePoint.get());
byte[] digest = computeDigest(context, normalizedDigestPlans);
return setQueryCacheParam(cachePoint.get(), digest, context);
} catch (Throwable t) {
return Optional.empty();
}
}
@VisibleForTesting
public List<TNormalizedPlanNode> normalizePlans(ConnectContext context) {
Optional<CachePoint> cachePoint = computeCachePoint();
if (!cachePoint.isPresent()) {
return ImmutableList.of();
}
return normalizePlanTree(context, cachePoint.get());
}
private Optional<TQueryCacheParam> setQueryCacheParam(
CachePoint cachePoint, byte[] digest, ConnectContext context) {
SessionVariable sessionVariable = context.getSessionVariable();
queryCacheParam.setNodeId(cachePoint.cacheRoot.getId().asInt());
queryCacheParam.setDigest(digest);
queryCacheParam.setForceRefreshQueryCache(sessionVariable.isQueryCacheForceRefresh());
queryCacheParam.setEntryMaxBytes(sessionVariable.getQueryCacheEntryMaxBytes());
queryCacheParam.setEntryMaxRows(sessionVariable.getQueryCacheEntryMaxRows());
queryCacheParam.setAllowIncremental(computeAllowIncremental(cachePoint, sessionVariable));
queryCacheParam.setIsMergeOnWrite(computeIsMergeOnWrite(cachePoint));
queryCacheParam.setOutputSlotMapping(
cachePoint.cacheRoot.getOutputTupleIds()
.stream()
.flatMap(tupleId -> descriptorTable.getTupleDesc(tupleId).getSlots().stream())
.map(slot -> {
int slotId = slot.getId().asInt();
return Pair.of(slotId, normalizeSlotId(slotId));
})
.collect(Collectors.toMap(Pair::key, Pair::value))
);
return Optional.of(queryCacheParam);
}
private Optional<CachePoint> computeCachePoint() {
if (!fragment.getTargetRuntimeFilterIds().isEmpty()) {
return Optional.empty();
}
PlanNode planRoot = fragment.getPlanRoot();
return doComputeCachePoint(planRoot);
}
private Optional<CachePoint> doComputeCachePoint(PlanNode planRoot) {
if (planRoot instanceof AggregationNode) {
PlanNode child = planRoot.getChild(0);
if (child instanceof OlapScanNode) {
if (((AggregationNode) planRoot).isQueryCacheCandidate()) {
return Optional.of(new CachePoint(planRoot, planRoot));
}
} else if (child instanceof AggregationNode) {
Optional<CachePoint> childCachePoint = doComputeCachePoint(child);
if (childCachePoint.isPresent()) {
if (((AggregationNode) planRoot).isQueryCacheCandidate()) {
return Optional.of(new CachePoint(planRoot, planRoot));
}
return childCachePoint;
}
}
} else if (planRoot instanceof ExchangeNode) {
return Optional.empty();
} else if (planRoot.getChildren().size() == 1) {
return doComputeCachePoint(planRoot.getChildren().get(0));
}
return Optional.empty();
}
/**
* Decide whether BE may serve a stale cache entry by scanning only the delta
* rowsets since the cached version and emitting them together with the cached
* partial aggregation blocks (the upstream aggregation merges both).
*
* <p>This is only correct when all of the following hold:
* <ul>
* <li>The cache point aggregation does not finalize: its output is a partial
* state that an upstream aggregation always merges, so emitting the cached
* blocks and the delta blocks side by side yields the correct result. A
* finalized output has no downstream merge, and the two emissions would
* produce duplicated group keys.</li>
* <li>The cache point aggregates the raw detail rows directly (its child is
* the olap scan node): with another aggregation in between, that inner
* aggregation would see only the delta rows during an incremental run, and
* its finalized output over the delta is not a mergeable complement of its
* output over the cached snapshot.</li>
* <li>The scanned index is append-only, guaranteeing "cached snapshot +
* delta rowsets == new snapshot": either DUP_KEYS, or merge-on-write
* UNIQUE_KEYS, for which BE additionally verifies per tablet (through the
* delete bitmap of the delta window) that no pre-existing key was rewritten
* and falls back otherwise. Merge-on-read UNIQUE resolves duplicates by
* merging across rowsets at read time, so a delta-only scan cannot stand
* alone there; AGG tables merge rows inside the storage layer likewise.
* DELETE predicates are handled on BE: a delta containing delete predicates
* falls back to a full recompute.</li>
* </ul>
*/
private boolean computeAllowIncremental(CachePoint cachePoint, SessionVariable sessionVariable) {
if (!sessionVariable.getEnableQueryCacheIncremental()) {
return false;
}
// The cache point is always an aggregation node (see doComputeCachePoint).
if (((AggregationNode) cachePoint.cacheRoot).isNeedsFinalize()) {
return false;
}
// The cached partial state and the delta partial state merge correctly
// only when the cache point aggregates the raw detail rows directly.
// With a nested cache point (partial agg over a finalized/deduplicating
// agg over scan), the inner agg sees only the delta rows during an
// incremental run, so its finalized output over the delta is NOT a
// mergeable complement of its output over the cached snapshot (e.g.
// "group by cnt" buckets computed from partial counts are simply wrong).
if (!(cachePoint.cacheRoot.getChild(0) instanceof OlapScanNode)) {
return false;
}
OlapScanNode scanNode = (OlapScanNode) cachePoint.cacheRoot.getChild(0);
OlapTable olapTable = scanNode.getOlapTable();
long selectIndexId = scanNode.getSelectedIndexId() == -1
? olapTable.getBaseIndexId()
: scanNode.getSelectedIndexId();
// The scanned index must be append-only, guaranteeing "cached snapshot +
// delta rowsets == new snapshot". Note: judged on the selected index, not
// the base table -- a DUP table may serve the query from an aggregated
// materialized view, whose data is no longer append-only.
//
// DUP_KEYS is always append-only. A merge-on-write UNIQUE index is
// append-only as long as a load does not touch pre-existing keys, which
// covers the common "hourly append plus occasional backfill" pattern: BE
// verifies per tablet through the delete bitmap of the delta window and
// falls back to a full recompute for the rare load that rewrites history.
// A merge-on-read UNIQUE index resolves duplicates by merging across
// rowsets at read time, so a delta-only scan cannot stand alone there;
// AGG tables merge rows inside the storage layer likewise.
KeysType keysType = olapTable.getKeysTypeByIndexId(selectIndexId);
boolean mergeOnWrite =
keysType == KeysType.UNIQUE_KEYS && olapTable.getEnableUniqueKeyMergeOnWrite();
if (keysType != KeysType.DUP_KEYS && !mergeOnWrite) {
return false;
}
// Freshness tolerance and prefer-cached-rowset (both cloud-only in effect)
// trade exactness for speed and locality: the scan may read a warmed-up layout
// that stops below the queried version (freshness halts at the warmed boundary)
// or reaches beyond it (neither walk clips an edge spanning it). An incremental
// merge would defeat such a knob, because the delta capture always targets the
// exact queried version (it must, to keep the merged entry correct), forcing the
// un-warmed reads the query chose to skip. The suppression is mode-gated
// (Config.isCloudMode()) through a pure helper: these knobs are INERT on local
// storage, so suppressing incremental there would only forgo it for no
// correctness benefit (the local read is version-exact regardless). Correctness
// against version-inexact reads does NOT rest on this per-query gate anyway (an
// entry filled by such a read could still be reused by a knob-free query sharing
// the cache key): on cloud the BE suppresses their write-back, so no entry whose
// content mismatches its version stamp exists. Extracting the pure,
// mode-parameterized helper is what lets the unit test exercise BOTH modes
// directly, without planning under a flipped cloud flag (which casts
// SystemInfoService to CloudSystemInfoService and fails).
if (cloudKnobsSuppressIncremental(Config.isCloudMode(),
sessionVariable.getQueryFreshnessToleranceMs(),
sessionVariable.getEnablePreferCachedRowset(), mergeOnWrite)) {
return false;
}
return true;
}
// Whether the cloud warmed-read knobs suppress incremental merge for this query.
// Pure and mode-parameterized (see the call site for why): only cloud mode honors
// freshness-tolerance / prefer-cached-rowset, so in shared-nothing mode they are
// inert and never suppress. This FE decision is authoritative: the BE applies no
// further gate on allow_incremental.
@VisibleForTesting
public static boolean cloudKnobsSuppressIncremental(boolean cloudMode,
long queryFreshnessToleranceMs, boolean enablePreferCachedRowset,
boolean mergeOnWrite) {
if (!cloudMode) {
return false;
}
// Freshness tolerance may read a layout below the queried version, so an
// incremental merge (which targets the exact version) would defeat it: suppress
// for every table type.
if (queryFreshnessToleranceMs > 0) {
return true;
}
// Prefer-cached-rowset is honored by the storage layer only for non-MOW tables
// (CloudTablet::capture_consistent_versions_unlocked guards on
// !enable_unique_key_merge_on_write()), so a MOW query reads the exact queried
// version regardless of the knob and stays incrementally mergeable; only non-MOW
// is suppressed.
return enablePreferCachedRowset && !mergeOnWrite;
}
// Whether the selected index of the cache point's underlying scan is a merge-on-
// write UNIQUE table. BE consumes this in the cloud cache write-back gate: cloud
// ignores enable_prefer_cached_rowset for a MOW table, so a prefer-only MOW read is
// version-exact and its fill must not be suppressed. Reported independently of
// allow_incremental because the write-back gate also applies to a MISS (a MOW MISS
// under prefer must still be cached). Determined on the selected index, mirroring
// computeAllowIncremental.
//
// Resolves the scan through the (possibly multi-level) unary chain, not just the
// direct child: incremental merge requires a DIRECT-child scan (a nested agg is not
// mergeable, see computeAllowIncremental), but this write-back gate only needs the
// scan's TABLE TYPE, which is well-defined for a nested-agg cache point too. Reading
// it through the chain lets a nested-agg-over-MOW query populate the cache under
// prefer (its read is still version-exact) instead of being over-suppressed just
// because the scan is not the direct child. Returns false if the chain branches or
// bottoms out at a non-scan (no single underlying olap scan), leaving BE at its safe
// suppress default.
private boolean computeIsMergeOnWrite(CachePoint cachePoint) {
PlanNode node = cachePoint.cacheRoot.getChild(0);
while (node != null && !(node instanceof OlapScanNode) && node.getChildren().size() == 1) {
node = node.getChild(0);
}
if (!(node instanceof OlapScanNode)) {
return false;
}
OlapScanNode scanNode = (OlapScanNode) node;
OlapTable olapTable = scanNode.getOlapTable();
long selectIndexId = scanNode.getSelectedIndexId() == -1
? olapTable.getBaseIndexId()
: scanNode.getSelectedIndexId();
KeysType keysType = olapTable.getKeysTypeByIndexId(selectIndexId);
return keysType == KeysType.UNIQUE_KEYS && olapTable.getEnableUniqueKeyMergeOnWrite();
}
private List<TNormalizedPlanNode> normalizePlanTree(ConnectContext context, CachePoint cachePoint) {
List<TNormalizedPlanNode> normalizedPlans = new ArrayList<>();
doNormalizePlanTree(context, cachePoint.digestRoot, normalizedPlans);
return normalizedPlans;
}
private void doNormalizePlanTree(
ConnectContext context, PlanNode plan, List<TNormalizedPlanNode> normalizedPlans) {
for (PlanNode child : plan.getChildren()) {
doNormalizePlanTree(context, child, normalizedPlans);
}
normalizedPlans.add(plan.normalize(this));
}
public static byte[] computeDigest(
ConnectContext context, List<TNormalizedPlanNode> normalizedDigestPlans) throws Exception {
TSerializer serializer = new TSerializer(new TCompactProtocol.Factory());
MessageDigest digest = MessageDigest.getInstance("SHA-256");
for (TNormalizedPlanNode node : normalizedDigestPlans) {
digest.update(serializer.serialize(node));
}
StringBuffer variables = new StringBuffer();
context.getSessionVariable().readAffectQueryResultVariables((k, v) -> {
variables.append(k).append("=").append(v).append("|");
});
digest.update(serializer.serialize(new TStringLiteral(variables.toString())));
return digest.digest();
}
@Override
public int normalizeSlotId(int slotId) {
return normalizedSlotIds.normalize(slotId);
}
@Override
public void setSlotIdToNormalizeId(int slotId, int normalizedId) {
normalizedSlotIds.set(slotId, normalizedId);
}
@Override
public int normalizeTupleId(int tupleId) {
return normalizedTupleIds.normalize(tupleId);
}
@Override
public int normalizePlanId(int planId) {
return normalizedPlanIds.normalize(planId);
}
@Override
public DescriptorTable getDescriptorTable() {
return descriptorTable;
}
@Override
public void setNormalizedPartitionPredicates(OlapScanNode olapScanNode, NormalizedPartitionPredicates predicates) {
OlapTable olapTable = olapScanNode.getOlapTable();
long selectIndexId = olapScanNode.getSelectedIndexId() == -1
? olapTable.getBaseIndexId()
: olapScanNode.getSelectedIndexId();
Map<Long, String> tabletToRange = Maps.newLinkedHashMap();
for (Long partitionId : olapScanNode.getSelectedPartitionIds()) {
Set<Long> tabletIds = olapTable.getPartition(partitionId)
.getIndex(selectIndexId)
.getTablets()
.stream()
.map(Tablet::getId)
.collect(Collectors.toSet());
String filterRange = predicates.intersectPartitionRanges.get(partitionId);
for (Long tabletId : tabletIds) {
tabletToRange.put(tabletId, filterRange);
}
}
queryCacheParam.setTabletToRange(tabletToRange);
}
private static class CachePoint {
PlanNode digestRoot;
PlanNode cacheRoot;
public CachePoint(PlanNode digestRoot, PlanNode cacheRoot) {
this.digestRoot = digestRoot;
this.cacheRoot = cacheRoot;
}
}
private static class NormalizedIdGenerator {
private final AtomicInteger idGenerator = new AtomicInteger(0);
private final Map<Integer, Integer> originIdToNormalizedId = Maps.newLinkedHashMap();
public Integer normalize(Integer originId) {
return originIdToNormalizedId.computeIfAbsent(originId, id -> idGenerator.getAndIncrement());
}
public void set(int originId, Integer normalizedId) {
originIdToNormalizedId.put(originId, normalizedId);
}
}
}