github-actions[bot] commented on code in PR #67549:
URL: https://github.com/apache/doris/pull/67549#discussion_r4321233303
##########
fe/fe-core/src/main/java/org/apache/doris/statistics/hbo/InputTableStatisticsInfo.java:
##########
@@ -17,25 +17,91 @@
package org.apache.doris.statistics.hbo;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapScan;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
/**
* Describes plan statistics which are derived from history based optimizer.
*/
public class InputTableStatisticsInfo {
- private final Optional<String> hash;
private final Optional<List<PlanStatistics>> inputTableStatistics;
- public InputTableStatisticsInfo(Optional<String> hash,
Optional<List<PlanStatistics>> inputTableStatistics) {
- this.hash = Objects.requireNonNull(hash, "hash is null");
+ public InputTableStatisticsInfo(Optional<List<PlanStatistics>>
inputTableStatistics) {
this.inputTableStatistics =
Objects.requireNonNull(inputTableStatistics,
"inputTableStatistics is null");
}
- public Optional<String> getHash() {
- return hash;
+ /**
+ * Fold the input table statistics of another node which shares the
fingerprint of this one: two
+ * runtime nodes can carry one fingerprint (the two scans of a self join,
for instance), and the
+ * provider stores one entry per fingerprint.
+ *
+ * <p>The two nodes read the same table population, so an entry of the
same scan (the same table, the
+ * same index and the same partitions) is one population however many
nodes read it: folding two of
+ * them keeps the larger measurement instead of adding the two up. Summing
them would describe a table
+ * which holds the rows of both scans at once, and the drift comparison -
which sums the rows of every
+ * recorded scan of a table - would then judge the next query of a single
scan stale by construction.
+ * Two different scans are both kept, because they describe two
populations.
+ */
+ public InputTableStatisticsInfo merge(InputTableStatisticsInfo other) {
+ if (inputTableStatistics.isEmpty()) {
+ return other;
+ }
+ if (other.inputTableStatistics.isEmpty()) {
+ return this;
+ }
+ Map<String, PlanStatistics> merged = new LinkedHashMap<>();
+ for (PlanStatistics statistics : inputTableStatistics.get()) {
+ merged.merge(identityOf(statistics), statistics,
InputTableStatisticsInfo::largerRows);
+ }
+ for (PlanStatistics statistics : other.inputTableStatistics.get()) {
+ merged.merge(identityOf(statistics), statistics,
InputTableStatisticsInfo::largerRows);
Review Comment:
[P2] Keep the predicate occurrence with its learned output
For `UnionAll(Project(a, Filter(b=1, Scan(t))), Project(a, Filter(b=2,
Scan(t))))`, both unaliased scans have the same S fingerprint. If both read N
mixed rows but return 100 and 900 rows, this dedup keeps only the first scan's
predicate/RelationId; `genPlanStatisticsMap` also keeps only its output. On the
identical next query, `getMatchedEntry` fetches predicates using that retained
historical RelationId for either filter's scan fallback. It therefore validates
the retained occurrence for both filters, and freshness now passes at N versus
N, so both receive the same output count. This is the new incorrect reuse after
replacing the previously reported 2N concatenation, without data growth or
runtime filters. Preserve separate output/input observations and match the
requested occurrence before applying one.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/stats/HboPlanInfoProvider.java:
##########
@@ -147,35 +315,62 @@ public void handle(Field field, String confVal) throws
Exception {
}
/**
- * Reference the above UpdateConfig comments.
+ * Reference the above UpdateConfig comments. Every per-query cache of
this provider is rebuilt
+ * (they are all sized and expired by the same two configs), and the
entries of a running query
+ * are carried over so that a runtime change cannot drop the plan info of
a query which is still
+ * in flight.
*/
public static synchronized void updateConfig() {
HboPlanStatisticsManager hboManger =
Env.getCurrentEnv().getHboPlanStatisticsManager();
if (hboManger == null) {
return;
}
- HboPlanInfoProvider planInfoProvider =
hboManger.getHboPlanInfoProvider();
- if (planInfoProvider == null) {
+ HboPlanInfoProvider provider = hboManger.getHboPlanInfoProvider();
+ if (provider == null) {
return;
}
+ int cacheNum = Config.hbo_plan_info_cache_num;
+ int pinnedCacheNum = Config.hbo_pinned_stats_cache_num;
+ long expireSeconds = Config.expire_hbo_plan_info_cache_in_fe_second;
- Cache<String, Map<Integer, PhysicalPlan>> idToPlanCache =
buildHboIdToPlanCache(
- Config.hbo_plan_info_cache_num,
- Config.expire_hbo_plan_info_cache_in_fe_second
- );
- Cache<String, Map<PhysicalPlan, Integer>> planToIdCache =
buildHboPlanToIdCache(
- Config.hbo_plan_info_cache_num,
- Config.expire_hbo_plan_info_cache_in_fe_second
- );
- Cache<String, Map<RelationId, Set<Expression>>> scanToFilterCache =
buildHboScanToFilterCache(
- Config.hbo_plan_info_cache_num,
- Config.expire_hbo_plan_info_cache_in_fe_second
- );
- idToPlanCache.putAll(planInfoProvider.idToPlanCache.asMap());
- planInfoProvider.idToPlanCache = idToPlanCache;
- planToIdCache.putAll(planInfoProvider.planToIdCache.asMap());
- planInfoProvider.planToIdCache = planToIdCache;
- scanToFilterCache.putAll(planInfoProvider.scanToFilterCache.asMap());
- planInfoProvider.scanToFilterCache = scanToFilterCache;
+ Cache<String, Map<Integer, PhysicalPlan>> idToPlanCache =
+ buildHboIdToPlanCache(cacheNum, expireSeconds);
+ idToPlanCache.putAll(provider.idToPlanCache.asMap());
+ provider.idToPlanCache = idToPlanCache;
+
+ Cache<String, Map<PhysicalPlan, Integer>> planToIdCache =
+ buildHboPlanToIdCache(cacheNum, expireSeconds);
+ planToIdCache.putAll(provider.planToIdCache.asMap());
+ provider.planToIdCache = planToIdCache;
+
+ Cache<String, Map<RelationId, Set<Expression>>> scanToFilterCache =
+ buildHboScanToFilterCache(cacheNum, expireSeconds);
+ scanToFilterCache.putAll(provider.scanToFilterCache.asMap());
+ provider.scanToFilterCache = scanToFilterCache;
+
+ Cache<String, Map<Integer, String>> nodeIdToFingerprintCache =
Review Comment:
[P3] Apply changed cache policies to the relation table cache
The new `relationTableIdCache` is built from `hbo_plan_info_cache_num` and
`expire_hbo_plan_info_cache_in_fe_second`, but this callback never rebuilds it.
Changing either mutable setting therefore updates the other per-query caches
while this one retains its startup policy. For example, starting with unbounded
size/expiry and later enabling finite limits leaves these per-query relation
maps unbounded and unexpired until restart. Rebuild and migrate this cache here
too, preserving in-flight writes.
##########
regression-test/suites/query_p0/hbo/hbo_join_expansion_inject_test.groovy:
##########
@@ -0,0 +1,246 @@
+// 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.
+
+suite("hbo_join_expansion_inject_test", "nonConcurrent") {
+ // HBO SET STATISTICS ... TYPE=JOIN_EXPANSION injects the measured fan-out
factor of a set of
+ // join equality conditions: it is a hbo statistics entry like any other,
only its value is a
+ // factor and its key is the condition fingerprint.
+ // Its purpose is join-order control: the injected join is estimated as
+ // clamp(expansion * max(leftEst, rightEst), 1, leftEst * rightEst), so a
join known to explode
+ // looks expensive and is scheduled as late as possible. The key carries
the join equality
+ // conditions, the join type and the residual (non equality) conditions,
because the factor
+ // replaces the output rows of the whole join; the shape of both sub trees
is excluded, so one
+ // entry is shared by every query with the same conditions and the same
join type. Semi / anti
+ // joins can never expand and deliberately ignore the entry.
+ sql "create database if not exists hbo_test;"
+ sql "use hbo_test;"
+ sql "drop table if exists hbo_je_t1;"
+ sql "drop table if exists hbo_je_t2;"
+ sql "drop table if exists hbo_je_t3;"
+ sql """create table hbo_je_t1(a int, b int) distributed by hash(a) buckets
1 properties("replication_num"="1");"""
+ sql """create table hbo_je_t2(a int, c int) distributed by hash(a) buckets
1 properties("replication_num"="1");"""
+ sql """create table hbo_je_t3(b int, d int) distributed by hash(b) buckets
1 properties("replication_num"="1");"""
+ sql """insert into hbo_je_t1 select number, number % 100 from
numbers("number"="1000");"""
+ sql """insert into hbo_je_t2 select number, number from
numbers("number"="2000");"""
+ sql """insert into hbo_je_t3 select number % 100, number from
numbers("number"="1000");"""
+ sql """analyze table hbo_je_t1 with sync;"""
+ sql """analyze table hbo_je_t2 with sync;"""
+ sql """analyze table hbo_je_t3 with sync;"""
+
+ sql "set enable_hbo_optimization=true;"
+ sql "set show_hbo_fingerprint=true;"
+ sql "set enable_sql_cache=false;"
+ sql "set enable_query_cache=false;"
+
+ def query = "select * from hbo_je_t1 join hbo_je_t2 on hbo_je_t1.a =
hbo_je_t2.a join hbo_je_t3 on hbo_je_t1.b = hbo_je_t3.b"
+ def physicalPlan = { (sql """ explain physical plan ${query}
""").flatten().join("\n") }
+ def annotation = { String sqlText -> (sql """ explain ${sqlText}
""").flatten().join("\n") }
+ // hash conditions of the physical hash joins, outermost first: the first
entry is the join that
+ // is computed last, which is exactly what the injected expansion is
supposed to influence
+ def joinOrder = { String text ->
+ (text =~
/PhysicalHashJoin\[\d+\][^\n]*hashCondition=\[([^\]]*)\]/).collect { it[1] }
+ }
+
+ def baseline = physicalPlan()
+ def baselineOrder = joinOrder(baseline)
+ log.info("baseline join order: ${baselineOrder}")
+ assertEquals(2, baselineOrder.size())
+ // baseline: the (t1.a = t2.a) join is the inner one (computed first), the
(t1.b = t3.b) join is
+ // the outermost one
+ assertTrue(baselineOrder[0].contains("b#"), baseline)
+ assertTrue(baselineOrder[1].contains("a#"), baseline)
+
+ // the condition fingerprint of the (t1.a = t2.a) join comes from the
explain annotation
+ def condCanonical =
"JE{INNER_JOIN;EqualTo(col(internal.hbo_test.hbo_je_t1.a),col(internal.hbo_test.hbo_je_t2.a))}"
+ // the expansion entry of the join is listed as its own line of the
annotation block
+ def matcher = (annotation(query) =~
+ /\] join type=join_expansion literal_mode=no_literal
fingerprint='([0-9a-f]+)'
struct='JE\{INNER_JOIN;EqualTo\(col\(internal\.hbo_test\.hbo_je_t1\.a\),col\(internal\.hbo_test\.hbo_je_t2\.a\)\)\}'/)
Review Comment:
[P2] Extract the complete expansion canonical from EXPLAIN
`HboJoinConditions.canonicalOf` now always appends `;bindings:[...]` before
the closing brace of an inner-join key. This matcher still requires the brace
immediately after the equality, so `matcher.find()` fails before the suite can
inject or check any expansion. The hardcoded `condCanonical` above is also
missing those bindings and would fail the fingerprint check even if only the
regex were relaxed. Capture the full emitted canonical together with its
fingerprint, as the semi-join case below already does. This test expectation is
separate from the existing memo-versus-physical lookup-key issue.
--
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]
