github-actions[bot] commented on code in PR #68170: URL: https://github.com/apache/doris/pull/68170#discussion_r4061166482
########## regression-test/suites/mtmv_p0/ivm/test_ivm_chained_stream_scope.groovy: ########## @@ -0,0 +1,162 @@ +// 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. + +import org.awaitility.Awaitility + +import static java.util.concurrent.TimeUnit.SECONDS + +/** + * Which streams decide that a refresh needs the complete attempt. + * + * <p>An IVM MV is created with a stream for every base table in its relation, and for a chained MV the + * relation carries more than the plan reads: mv2 reads mv1, while the table behind mv1 is in mv2's + * relation only through the closure. Both rewrites that read streams -- the incremental delta rewriter + * and the full refresh -- take the streams of the plan's scans, so mv2's stream on that table is never + * read, and its absence must not turn mv2's incremental refresh into a rebuild of the whole MV. + * + * <p>Pinned here: + * <ol> + * <li>the unused stream exists, so the setup is the one the case is about;</li> + * <li>an incremental refresh of the upstream MV succeeds with it dropped, which is what makes the + * refresh below able to stay incremental;</li> + * <li>the downstream MV still refreshes incrementally rather than falling back to a complete + * refresh for a stream nothing reads.</li> + * </ol> + */ +suite("test_ivm_chained_stream_scope") { + def baseTable = "ivm_chained_scope_base" + def upstreamMv = "ivm_chained_scope_up" + def downstreamMv = "ivm_chained_scope_down" + + def waitForNewTask = { String mvName, String previousTaskId -> + def taskResult + Awaitility.await().atMost(300, SECONDS).pollInterval(2, SECONDS).until({ + taskResult = sql_return_maparray(""" + SELECT TaskId, Status + FROM tasks('type'='mv') + WHERE MvDatabaseName = '${context.dbName}' + AND MvName = '${mvName}' + ORDER BY CreateTime DESC, TaskId DESC LIMIT 1 + """) + return !taskResult.isEmpty() + && taskResult[0].TaskId.toString() != previousTaskId + && taskResult[0].Status.toString() != 'PENDING' + && taskResult[0].Status.toString() != 'RUNNING' + }) + return taskResult[0].TaskId.toString() + } + + def taskRecord = { String taskId -> + // Unset RefreshMode comes back as the literal two-character string "\N", which does not survive + // the .out round trip, so fold the unset value into a printable token. + return sql_return_maparray(""" + SELECT Status, + CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL', 'NOT_REFRESH') + THEN RefreshMode ELSE 'NONE' END AS RefreshMode + FROM tasks('type'='mv') + WHERE TaskId = '${taskId}' + """)[0] + } + + def mvId = { String mvName -> + def rows = sql_return_maparray(""" + SELECT Id FROM mv_infos('database'='${context.dbName}') WHERE Name = '${mvName}' + """) + return rows[0].Id.toString() + } + + // Streams are named after the MV that owns them, so the ones of the downstream MV on the upstream's + // base table can be told apart from the ones the upstream MV has on it. + def streamNamesOf = { String mvName, String baseTableName -> + def rows = sql_return_maparray(""" + SELECT STREAM_NAME FROM information_schema.table_streams + WHERE DB_NAME = '${context.dbName}' AND BASE_TABLE_NAME = '${baseTableName}' + """) + def prefix = "__doris_ivm_stream_${mvId(mvName)}_" + return rows.collect { it.STREAM_NAME.toString() }.findAll { it.startsWith(prefix) } + } + + sql """DROP MATERIALIZED VIEW IF EXISTS ${downstreamMv}""" + sql """DROP MATERIALIZED VIEW IF EXISTS ${upstreamMv}""" + sql """DROP TABLE IF EXISTS ${baseTable}""" + + sql """ + CREATE TABLE ${baseTable} ( + k1 INT NOT NULL, + v1 INT + ) + UNIQUE KEY(k1) + DISTRIBUTED BY HASH(k1) BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "enable_unique_key_merge_on_write" = "true", + "binlog.enable" = "true", + "binlog.format" = "ROW", + "binlog.need_historical_value" = "true" + ) + """ + sql """INSERT INTO ${baseTable} VALUES (1, 10), (2, 20)""" + + // The upstream MV carries row binlog, which is what the downstream MV reads its changes from. + sql """ + CREATE MATERIALIZED VIEW ${upstreamMv} + BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ( + "replication_num" = "1", + "binlog.enable" = "true", + "binlog.format" = "ROW", + "binlog.need_historical_value" = "true" + ) + AS SELECT * FROM ${baseTable} + """ + sql """REFRESH MATERIALIZED VIEW ${upstreamMv} COMPLETE""" + def upstreamTaskId = waitForNewTask(upstreamMv, null) + + sql """ + CREATE MATERIALIZED VIEW ${downstreamMv} + BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL + DISTRIBUTED BY RANDOM BUCKETS 1 + PROPERTIES ("replication_num" = "1") + AS SELECT k1, v1 FROM ${upstreamMv} + """ + sql """REFRESH MATERIALIZED VIEW ${downstreamMv} COMPLETE""" + def downstreamTaskId = waitForNewTask(downstreamMv, null) + + // The downstream MV's plan scans the upstream MV only, but its relation also carries the upstream's + // base table, and the MV was created with a stream for it. + def unusedStreams = streamNamesOf(downstreamMv, baseTable) + assertEquals(1, unusedStreams.size()) + sql """DROP STREAM ${context.dbName}.${unusedStreams[0]}""" Review Comment: **[P1] Use FORCE so this regression reaches its assertions in cloud mode** `DropStreamCommand.validate` rejects every non-force drop when `Config.isCloudMode()` is true, and this common IVM suite has no cloud skip, so it aborts here before testing the chained-stream behavior. The other new stream-unusable regression already uses the portable form. ```suggestion sql """DROP STREAM ${context.dbName}.${unusedStreams[0]} FORCE""" ``` ########## fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java: ########## @@ -445,6 +445,17 @@ private List<RefreshAttemptType> buildAttempts(RefreshRequest request, boolean c if (shouldUseCompleteForInitialIvmRefresh(containsOneRowRelation)) { return Lists.newArrayList(RefreshAttemptType.COMPLETE); } + // A base table the plan scans without a usable stream makes every other attempt fail: the + // incremental rewrite reads the stream of every table it scans, and a partition refresh reads + // those of the PCT tables it realigns as well. Only the COMPLETE attempt reconciles streams, so + // it is the one that has to run. Deciding it here keeps the attempt that cannot succeed -- and + // the baseline barrier a partition refresh writes before it starts -- out of the way altogether. + if (request.allowFallback && mtmv.isIvm() && hasUnusableIvmStream()) { Review Comment: **[P1] Treat a replacement stream as a broken baseline** This preflight validates only the deterministic name, owner, base-table ID, and state, so a stream that was dropped and recreated under the same name before the refresh is considered usable even though its offsets no longer match the MV baseline. On a partial refresh, non-PCT tables are read in SNAPSHOT mode; the recreated `show_initial_rows` stream has no consumed offsets, so `filterConsumedPartitionIds` can turn such a joined dimension into an empty scan and the task can publish wrong MV rows. SNAPSHOT emits no stream-offset identity update, and stream DDL does not invalidate the owning MV or bump its schema version, so the success path can clear the guard as well. Please persist/revalidate the stream generation associated with the MV baseline, or make replacement invalidate the dependent MV's COMPLETE baseline/version. ########## fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java: ########## @@ -855,6 +889,55 @@ private void executePartitionBasedRefresh(MTMVRefreshContext context, RefreshMod mtmv.getDatabase().getFullName(), mtmv.getName(), getTaskId()); } + /** + * Whether a base table the refresh reads has no stream that can be read, which no attempt other + * than COMPLETE can work around. + * + * <p>A base table that cannot be resolved is skipped rather than judged: it says nothing about the + * streams, and the refresh fails on it for its own reasons -- the attempt that runs reports that, + * this one only decides which attempt that should be. + */ + private boolean hasUnusableIvmStream() { + Database mvDb = (Database) mtmv.getDatabase(); + if (mvDb == null) { + // Nothing to look the streams up in, so there is nothing to decide here. + return false; + } + Set<TableNameInfo> excluded = mtmv.getExcludedTriggerTables(); + // The tables in the plan, not the relation's closure: a chained MV is created with a stream for + // every base table behind the MVs it reads, but no rewrite ever looks those up -- the incremental + // rewriter and the full refresh take the streams of the plan's scans -- so judging them would + // rebuild an MV whose refresh had nothing wrong with it. + for (BaseTableInfo baseTableInfo : relation.getBaseTablesOneLevelAndFromView()) { Review Comment: **[P2] Scope the preflight to the selected PCT mappings** For a partitioned multi-PCT IVM such as `t1 UNION ALL t2` with disjoint ranges, a change confined to a `t1`-backed MV partition produces a RESET map containing only `t1`. `IvmFullRefreshMTMV` keeps `t2` as an ordinary scan in that case and never calls `getIvmStream` for it, but this loop still checks `t2`; if only that unselected stream is missing, `PARTITIONS FALLBACK` is sent to COMPLETE and rebuilds every MV partition unnecessarily. This is distinct from the earlier closure-only thread because `t2` is a real plan leaf but is not stream-read for the selected partition scope. Please derive the PARTITIONS preflight from the planned partitions: all streamed non-PCT scans plus only PCT tables present in their RESET mappings. -- 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]
