yujun777 commented on code in PR #69890:
URL: https://github.com/apache/doris/pull/69890#discussion_r4510902348
##########
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:
Fixed in 1648691ba8aa840be909af805a865cbebf59143d. Normalize now collects
internal aggregate outputs separately and appends them after all persisted
aggregate outputs. Skipping COUNT_AT_MIN/MAX therefore trims only the suffix
and preserves positional alignment of the persisted slots. The focused FE
normalization/processor tests and all 124 IVM regression suites passed on the
updated branch.
##########
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:
Fixed in
https://github.com/apache/doris/commit/b153c38b1f7116d79671f11e0363d1b98cafe877.
Successful COMPLETE refreshes now capture a consistent layout signature across
batches and publish it with the rebuilt baseline through the existing IvmInfo
journal path. Partition refreshes do not publish a replacement signature;
append-only detail MVs remain excluded. Extended topk_1 through ALTER K=2 ->
K=4, COMPLETE, another insert, and strict INCREMENTAL; the task succeeds
without fallback and the hidden counts are correct. All 124 IVM suites passed,
including the snapshot and array/bitmap cases previously cited in the old guard
comment.
##########
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:
Fixed in 1449fef40a81ae4c8193c0d413a42ffe075212f0. Unsupported top-K
argument types resolve to the K=1 extreme-plus-count path, and apply compares
extremes with CASE and ordering comparisons instead of scalar-only
LEAST/GREATEST. test_ivm_agg_minmax_array covers the persisted capacity,
incremental inserts, and complete fallback after deleting the last minimum.
That suite passed again on b153c38b1f7.
##########
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:
Fixed in
https://github.com/apache/doris/commit/b153c38b1f7116d79671f11e0363d1b98cafe877.
refreshIvm records the latest TaskId immediately before submitting each
refresh, and latestIvmTask requires a different TaskId in terminal status. This
also covers the first helper call after an earlier refresh was waited for
through another helper, which a cache of previously returned IDs did not cover.
Updated all existing callers and made the shared cache concurrent. All 124 IVM
regression suites passed.
##########
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:
Addressed by aedebf639435b4de10ed9d25faa391c4788d1bfb and completed in
b153c38b1f7. Extraction is scope-aware: a whole ArrayMap whose lambda binds its
item references remains eligible, while expressions with free array-item
references are excluded. The existing testLambdaExpression expectation is
preserved. All 11 CommonSubExpressionTest tests passed, including the added
nested captured-input case.
--
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]
