github-actions[bot] commented on code in PR #69890:
URL: https://github.com/apache/doris/pull/69890#discussion_r4422773926
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandler.java:
##########
@@ -356,9 +367,15 @@ LogicalProject<?> buildApplyPlan(LogicalAggregate<?>
normalizedAgg,
new TinyIntLiteral((byte) -1), new
TinyIntLiteral((byte) 1));
List<NamedExpression> finalOutputs = new ArrayList<>();
+ Set<String> emittedColumnNames = new HashSet<>();
// Keep the normalized aggregate schema here. The normalize-added top
project computes row-id above this
// project, and the final sink project reorders columns by MV schema.
for (Slot target : normalizedAgg.getOutput()) {
+ if
(aggMeta.getInternalAggOutputNames().contains(target.getName())) {
Review Comment:
[P1] Preserve aggregate slot alignment when dropping internal outputs
For `SELECT MIN(x) AS m, SUM(y) AS s FROM t` with distinct x/y and K=1 (or
automatic K=0 on a DUP changelog source), the reduced normalized shape is:
```text
Project(..., count_x, count_y, min_topk, bound)
Aggregate(m, s, group_count, count_x, min_count_x, count_y)
```
This skip changes the apply output to
`[m,s,group_count,count_x,count_y,min_topk,bound,...]`, but
`IvmLinearDeltaHandler.rewriteProject` still pairs old/new child slots by
position. Old `count_y` therefore maps to the `min_topk` array and disappears
under its own name; `makeDeleteSignProject` then cannot find that persisted
COUNT column on the first nonempty incremental refresh. The owned-state
override only fixes the derived top-K/bound outputs. Preserve original
positions until rebinding or map surviving outputs by ExprId, and cover mixed
distinct-argument aggregates at K=0/1.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -557,17 +560,43 @@ public void alterMvProperties(AlterMTMV alterMTMV,
boolean isReplay) {
* backlog was skipped.
*/
private boolean rebuildsWholeMv(Set<TableNameInfo>
oldExcludedTriggerTables,
- Map<TableNameInfo, Integer> oldWindowLimits, Map<String, String>
oldSyncWindow) {
- // Judged once here rather than in each of the three: they answer "did
this property move in the
+ Map<TableNameInfo, Integer> oldWindowLimits, Map<String, String>
oldSyncWindow,
+ Long oldMinMaxTopK) {
+ // Judged once here rather than in each of the four: they answer "did
this property move in the
// direction that owes a rebuild", which is only a question an MV
maintaining an IVM baseline has.
if (!maintainsIvmBaseline()) {
return false;
}
- return unexcludesABaseTable(oldExcludedTriggerTables)
+ // The persisted MIN/MAX state was built for the old capacity. A
larger K needs values the old
+ // state already dropped; a smaller one would keep entries the new
apply never produces. Either
+ // direction makes the state unreadable, so any change rebuilds the
whole MV -- but only when
+ // the MV actually keeps that state: an aggregate MV without MIN/MAX,
or a detail MV, has
+ // nothing for the property to describe.
+ boolean minMaxTopKChanged = !Objects.equals(oldMinMaxTopK,
+
MTMVPropertyUtil.getIvmMinMaxTopK(this.mvProperties).orElse(null));
+ return (minMaxTopKChanged && keepsMinMaxTopKState())
Review Comment:
[P2] Publish a matching signature after the capacity rebuild
Create a MIN/MAX MV with K=2, establish its baseline, ALTER K to 4, run
COMPLETE, then insert a row and request strict INCREMENTAL. K appears in hidden
`min_topk`/bound expressions included by `IvmPlanSignatureGenerator`, so
normalization produces a different signature. This branch forces a
SCHEMA_CHANGE complete rebuild, but `MTMVTask` captures a replacement signature
only for a `PLAN_SIGNATURE_MISMATCH` fallback, and `addTaskResult` retains the
old signature otherwise. The next incremental refresh therefore fails
`PLAN_SIGNATURE_MISMATCH` even after the successful rebuild; repeating COMPLETE
does not repair it. Publish the new signature together with the rebuilt
baseline for this capacity change, and extend the ALTER test through a
subsequent strict incremental refresh.
##########
regression-test/plugins/plugin_planner.groovy:
##########
@@ -149,10 +153,39 @@ def maskIvmDryRunSequence = { row, meta ->
return row
}
-// Named ivm_dry_run_qt instead of order_qt_*: Suite.invokeMethod intercepts
every method whose
-// name starts with qt_ / order_qt_ and derives the tag from the method name,
so a metaClass
-// method could never be reached under those prefixes.
-Suite.metaClass.ivm_dry_run_qt = { String tag, String sql ->
- delegate.quickRunTest(tag, sql, true, maskIvmDryRunSequence)
+// One entry point for IVM output that carries a hidden sequence column:
REFRESH ... WITH DRY RUN
+// returns the delta rows, SELECT * on an IVM MV shows the hidden state
columns (for example the
+// MIN/MAX top-K struct). The sequence value differs between cloud and
shared-nothing deployments,
+// so the mask makes both modes share one .out; show_hidden_columns is
harmless to a REFRESH and is
+// what makes the hidden columns reach the SELECT * result.
+// Named ivm_hidden_qt instead of order_qt_*: Suite.invokeMethod intercepts
every method whose name
+// starts with qt_ / order_qt_ and derives the tag from the method name, so a
metaClass method could
+// never be reached under those prefixes.
+// Latest finished MV task row for one MV, waiting for a terminal status
first. Shared by the IVM
+// suites so each one does not carry its own copy; the row carries the columns
those suites assert
+// on (Status, RefreshMode, IvmFallbackReason, ErrorMsg).
+Suite.metaClass.latestIvmTask = { String mvName ->
+ def suite = delegate
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({
+ taskResult = suite.sql_return_maparray("""
+ SELECT Status, RefreshMode, IvmFallbackReason, ErrorMsg FROM
tasks('type'='mv')
+ WHERE MvDatabaseName = '${suite.context.dbName}' AND MvName =
'${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
Review Comment:
[P2] Require a fresh task ID before accepting terminal status
This helper drops the TaskId freshness condition removed from
`test_ivm_minmax_count` and `test_ivm_minmax_count_join`.
`MTMVTask.onSuccess/onFail` calls its superclass first, which removes the task
from `runningTasks` and journals the job; only the later `after()` publishes
the MV history. During that gap, `tasks()` returns the previous terminal
history row, and these status checks accept it immediately. For example, the
last-extreme FALLBACK assertion can read the preceding SUCCESS/INCREMENTAL task
instead of the new SUCCESS/COMPLETE task, causing intermittent failures or
stale success checks. The TVF does not take the worker's job lock. Pass the
pre-submission TaskId into the wait and require a new terminal task.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/CommonSubExpressionCollector.java:
##########
@@ -80,12 +80,14 @@ private int collectChildrenDepth(List<Expression> children,
Boolean inLambda) {
private int collectCommonExpressionByDepth(int depth, Expression expr,
boolean inLambda) {
Set<Expression> expressions = getExpressionsFromDepthMap(depth,
expressionsByDepth);
- // ArrayItemSlot and ArrayItemReference could not be common expressions
- // TODO: could not extract common expression when expression contains
same lambda expression
- // because ArrayItemSlot in Lambda are not same.
+ // An expression that contains a lambda argument cannot be evaluated
outside that lambda:
+ // hoisting it into an earlier projection leaves the argument without
its binding. The
+ // check must not depend on the inLambda flag -- a lambda nested in
another expression can
+ // be visited again from a context that does not carry the flag (for
example when the same
+ // subtree appears in two projections), and the extraction then breaks
the lambda scope.
if (expressions.contains(expr)
&& !(expr instanceof WhenClause)
- && !(inLambda && expr.containsType(ArrayItemSlot.class,
ArrayItemReference.class))) {
+ && !expr.containsType(ArrayItemSlot.class,
ArrayItemReference.class)) {
Review Comment:
[P2] Reconcile the existing closed-lambda CSE test with this guard
The unchanged `CommonSubExpressionTest.testLambdaExpression` constructs the
same bound `ArrayMap(Lambda(...))` under two aliases, calls this collector with
`false`, and asserts one common ArrayMap at depth 4. The complete ArrayMap
contains its bound `ArrayItemReference` and body `ArrayItemSlot`, so this
unconditional recursive check now excludes it along with every repeated body
expression. The aliases differ and leaves are never collected, leaving
`commonExprByDepth` empty where the existing test expects size 1. This
assertion conflict follows directly from the constructors and traversal; no
test was executed. Keep whole-function extraction scope-aware, or reconcile the
existing expectation with the intended conservative policy and add a case for
the unbound-body bug.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/agg/IvmAggFunctionProcessor.java:
##########
@@ -148,22 +204,43 @@ protected List<Expression>
targetArguments(AggregateFunction function) {
return ImmutableList.of(function.child(0));
}
- /** Aggregate function used to materialize one hidden state during
normalize. */
- protected AggregateFunction hiddenStateAggregate(IvmAggStateKey stateKey,
AggregateFunction function) {
- switch (stateKey) {
- case COUNT:
- return new Count(function.child(0));
- case SUM:
- return new Sum(function.child(0));
- case BITMAP_UNION:
- return new BitmapUnion(function.child(0));
- case MIN_COUNT:
- return new CountAtMin(function.child(0));
- case MAX_COUNT:
- return new CountAtMax(function.child(0));
- default:
- throw new IllegalArgumentException("Unsupported IVM aggregate
state: " + stateKey);
- }
+ /**
+ * Aggregate function used to materialize one hidden aggregate state
during normalize.
+ *
+ * <p>{@code minMaxTopK} is the MV's resolved MIN/MAX capacity; only the
MIN_TOPK/MAX_TOPK states
+ * read it, the other states ignore it.
+ */
+ protected AggregateFunction hiddenStateAggregate(IvmAggStateKey stateKey,
AggregateFunction function,
+ long minMaxTopK) {
+ return switch (stateKey) {
+ case COUNT -> new Count(function.child(0));
+ case SUM -> new Sum(function.child(0));
+ case BITMAP_UNION -> new BitmapUnion(function.child(0));
+ case MIN_TOPK -> new MinTopK(function.child(0), new
BigIntLiteral(minMaxTopK));
Review Comment:
[P2] Preserve supported ARRAY inputs when choosing the MIN/MAX state
For a changelog-enabled MOW table with `a ARRAY<INT>`, `SELECT MIN(a) FROM
t` selects default K=8 and now injects hidden `MIN_TOPK(a,8)`. `Min` and the
previous hidden `CountAtMin` explicitly allow ARRAY, and their BE
implementations support it through `SingleValueDataColumn`;
`TopKFunction.checkLegalityAfterRewrite` rejects ARRAY and the BE top-K factory
also omits it. Thus a previously supported MIN/MAX input can no longer be
created/refreshed with the default state. K=0/1 also introduces scalar-only
LEAST/GREATEST on array values. Keep an array-capable maintenance path or
extend the new helpers' type support, with ARRAY MIN/MAX coverage.
--
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]
