This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bc49fd106e3 [fix](ivm) Read a base partition with no stream offset
from the base table in an IVM refresh (#69778)
bc49fd106e3 is described below
commit bc49fd106e36bcb54576a3c914dc3e23ad401b09
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 03:59:39 2026 +0000
[fix](ivm) Read a base partition with no stream offset from the base table
in an IVM refresh (#69778)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
**In short.** An IVM materialized view silently misses the rows of a
base partition its internal stream has no offset for (a partition added
after the MV was built, or one that was still empty then) when those
rows are older than the base table's `changelog.ttl_seconds` at refresh
time. The refresh succeeds, the rows are never read, and a base-table
query of the MV's shape that transparent rewrite answers from the MV
returns the same wrong result. This PR makes an IVM read take such a
partition from the base table, as an append of all its rows, instead of
from its changelog.
**Background.** An IVM MV reads each base table through an internal
table stream. The stream keeps an offset per partition and records one
only when it consumes the partition with data in it. It is created with
`show_initial_rows`, so a partition that has data at creation is a
historical partition, read once from the base table; a partition that is
empty then gets nothing; and every consumption, the reset of a COMPLETE
refresh included, records offsets only for the partitions it read with
data. Two kinds of partitions therefore have no offset and none of their
rows in the MV: a partition added after the stream was created, and a
partition that was empty when the stream was created and at every
consumption since (a range partition created ahead of its data is the
usual one).
**The problem, and what it cost.**
- `NormalizeOlapTableStreamScan.makeTableStreamScan` reads a partition
with no offset like any other incremental partition: from its changelog,
with no start. `applyChangelogTtl` turns the missing start into the
first TSO the changelog retains and does not reject it.
- Once the partition's rows are older than the TTL when the MV is
refreshed, the read skips them. The refresh succeeds and records the
partition's offset past them, so no later refresh reads them either: the
MV stays short of those rows for good.
- This holds for an MV over a unique-key table and for an append-only
detail MV over a duplicate-key table. An MV partitioned like its base
table rebuilds the MV partition of an added base partition, but its
partition that was empty when the MV was built is caught up
incrementally and loses its rows the same way.
**How this PR fixes it.**
- In an IVM statement, `makeTableStreamScan` moves the incremental
partitions that have no consumption baseline
(`OlapTableStreamWrapper.filterConsumedPartitionIds`: no offset, or the
sentinel offset of a partition empty at stream creation) to the branch
that reads historical partitions from the base table: every row as an
append, with the commit TSO as the sequence and the row LSN as the row
identity. For an MV that holds none of the partition's rows that is the
partition's whole change, whatever the TTL. The offset the refresh
records afterwards is the same as before.
- The base read and the end TSO the refresh commits come from the same
snapshot, as for a historical partition: in shared-nothing mode both are
taken under the table read lock that planning holds, and in cloud mode
the base read is pinned to the visible version of the stream read state.
- A user stream is not changed: reading the changelog it retains, from
the first retained TSO, is what it promises. The choice is made by the
statement being an IVM rewrite, not by the stream.
**Results** (local shared-nothing cluster; rows written to an added
partition and to the partition that was empty, older than a one-second
TTL when the MV is refreshed):
| MV | Before | After |
| --- | --- | --- |
| non-partitioned, over a unique-key table | SUCCESS, the rows of both
partitions missing for good | SUCCESS, the MV matches the base table |
| append-only detail MV, over a duplicate-key table | SUCCESS, the rows
of both partitions missing for good | SUCCESS, the MV matches the base
table |
| partitioned like its unique-key base table | SUCCESS, the added
partition rebuilt, the rows of the one that was empty missing | SUCCESS,
the MV matches the base table |
| the next refresh, after more changes, with the TTL back to a day |
incremental, the rows still missing | incremental, the new changes read
from the changelog |
**Classes, and how they call each other.**
- `IvmIncrRefreshMTMV` → `IvmDeltaRewriteState.createDeltaScan` builds a
`LogicalOlapTableStreamScan` over an `OlapTableStreamWrapper`.
- `NormalizeOlapTableStreamScan.makeTableStreamScan` →
`filterHistoryPartitionIds` / `filterIncrementalPartitionIds`, and in an
IVM statement `filterConsumedPartitionIds` → historical and unconsumed
partitions: `makeOlapScanOnBaseTable`; consumed partitions:
`makeIncrementalScanFromChangelog` → `applyChangelogTtl`.
- `StreamConsumptionInfoExtractor` records the new offsets from the
wrapper's output map, as before.
Not changed: the TTL check, the snapshot and reset reads (the snapshot
read already leaves the partitions without a baseline out and reports
that it did), user streams, and anything persisted.
---
.../rewrite/NormalizeOlapTableStreamScan.java | 19 +-
.../trees/plans/ExplainTableStreamPlanTest.java | 103 ++++++++--
.../ivm/test_ivm_partition_without_offset.out | 73 +++++++
.../ivm/test_ivm_partition_without_offset.groovy | 213 +++++++++++++++++++++
4 files changed, 395 insertions(+), 13 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
index 30ae919fc07..6b8092c9f97 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
@@ -377,7 +377,8 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
* <p>Selected partitions are split into:
* <ul>
* <li>historical partitions — never consumed history data; scanned from
the base table and all
- * rows are treated as APPEND (change type = "APPEND", seq = commit
tso, lsn = row lsn);</li>
+ * rows are treated as APPEND (change type = "APPEND", seq = commit
tso, lsn = row lsn). An IVM
+ * read adds the incremental partitions that have no consumption
baseline;</li>
* <li>incremental partitions — read row-level changes from changelog via
* {@link #makeIncrementalScanFromChangelog}.</li>
* </ul>
@@ -390,6 +391,22 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
List<Long> historicalPartitionIds =
streamWrapper.filterHistoryPartitionIds(scan.getSelectedPartitionIds());
List<Long> incrementalPartitionIds =
streamWrapper.filterIncrementalPartitionIds(scan.getSelectedPartitionIds());
+ if (cascadesContext.getStatementContext().isIvmMTMVRewrite()) {
+ // A partition with no consumption baseline -- added after the
stream was created, or still empty when
+ // the stream was created and whenever it was consumed -- has none
of its rows in the MV yet. Read from
+ // its changelog, it would be read from the first retained TSO,
and the rows older than the TTL would
+ // be left out while the refresh succeeds. An IVM read takes such
a partition from the base table, the
+ // way it takes a historical one. A user stream keeps reading the
changelog it retains.
+ Set<Long> consumedIds = ImmutableSet.copyOf(
+
streamWrapper.filterConsumedPartitionIds(incrementalPartitionIds));
+ historicalPartitionIds = ImmutableList.<Long>builder()
+ .addAll(historicalPartitionIds)
+ .addAll(incrementalPartitionIds.stream().filter(id ->
!consumedIds.contains(id)).iterator())
+ .build();
+ incrementalPartitionIds = incrementalPartitionIds.stream()
+ .filter(consumedIds::contains)
+ .collect(ImmutableList.toImmutableList());
+ }
Plan historyPlan = null;
Plan incrementalPlan = null;
List<Slot> originSlots = scan.getOutput();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
index 2b9e4f4ca92..1ff0b84099f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
@@ -23,10 +23,12 @@ import org.apache.doris.analysis.Expr;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
import org.apache.doris.catalog.OlapChangelogTableWrapper;
import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.OlapTableWrapper;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.Tablet;
@@ -38,6 +40,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.Pair;
import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.mtmv.ivm.IvmRewriteContext;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.exceptions.AnalysisException;
@@ -45,6 +48,7 @@ import
org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.properties.PhysicalProperties;
+import org.apache.doris.nereids.rules.rewrite.NormalizeOlapTableStreamScan;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.LessThan;
@@ -75,12 +79,15 @@ import org.apache.doris.thrift.TScanRangeLocations;
import org.apache.doris.tso.TSOTimestamp;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.ImmutableSet;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
/**
@@ -136,6 +143,21 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
createTable(createIncrementalStream);
bumpPartitionsAndReplicas(baseTable, 2002L);
+ // A stream with a partition that has no consumption baseline: p2 is
still empty when the stream is
+ // created, so it is recorded with the sentinel offset, and only gets
data afterwards. p1 has data
+ // before and after, so it has a real offset and data after it.
+ createTable("create table test_stream.tbl_stream_unconsumed (k1 int,
k2 int not null) unique key(k1)\n"
+ + "partition by range(k1) (partition p1 values less than
(\"100\"),"
+ + " partition p2 values less than (\"200\"))\n"
+ + "distributed by hash(k1) buckets 1
properties(\"replication_num\"=\"1\","
+ +
"\"enable_unique_key_merge_on_write\"=\"true\",\"changelog.enable\"=\"true\","
+ +
"\"changelog.ttl_seconds\"=\"9223372036854775807\",\"changelog.need_historical_value\"=\"true\")");
+ OlapTable unconsumedBaseTable = (OlapTable)
db.getTableOrMetaException("tbl_stream_unconsumed");
+ bumpPartitionAndReplicas(unconsumedBaseTable.getPartition("p1"),
1001L);
+ createTable("create stream if not exists test_stream.s_unconsumed on
table test_stream.tbl_stream_unconsumed"
+ + " properties('show_initial_rows' = 'false')");
+ bumpPartitionsAndReplicas(unconsumedBaseTable, 2002L);
+
String createDuplicateBaseTable = "create table
test_stream.tbl_dup_stream_base (\n"
+ " k1 int,\n"
+ " k2 int not null\n"
@@ -164,18 +186,22 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
private static void bumpPartitionsAndReplicas(OlapTable table, long
newVersion) {
for (Partition partition : table.getPartitions()) {
- // Use the same timestamp value as both visibleVersionTime and tso
to simulate a
- // transactional commit advancing partition.tso (the dedicated
commit-tso field).
- // Without populating tso explicitly, non-transactional setters
leave it as -1
- // which would make the stream consider the partition fully
consumed.
- long ts = System.currentTimeMillis();
- partition.setVisibleVersionAndTime(newVersion, ts, ts);
- partition.setNextVersion(newVersion + 1);
- for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
- for (Tablet tablet : index.getTablets()) {
- for (Replica replica : tablet.getReplicas()) {
- replica.updateVersion(newVersion);
- }
+ bumpPartitionAndReplicas(partition, newVersion);
+ }
+ }
+
+ private static void bumpPartitionAndReplicas(Partition partition, long
newVersion) {
+ // Use the same timestamp value as both visibleVersionTime and tso to
simulate a
+ // transactional commit advancing partition.tso (the dedicated
commit-tso field).
+ // Without populating tso explicitly, non-transactional setters leave
it as -1
+ // which would make the stream consider the partition fully consumed.
+ long ts = System.currentTimeMillis();
+ partition.setVisibleVersionAndTime(newVersion, ts, ts);
+ partition.setNextVersion(newVersion + 1);
+ for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
+ for (Tablet tablet : index.getTablets()) {
+ for (Replica replica : tablet.getReplicas()) {
+ replica.updateVersion(newVersion);
}
}
}
@@ -311,6 +337,59 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
}
}
+ @Test
+ public void testUserStreamReadsPartitionWithoutBaselineFromChangelog()
throws Exception {
+ // A user stream reads every partition with data after its offset from
the changelog, the one without a
+ // consumption baseline included: from the first retained TSO, which
is what the stream retains.
+ OlapTable baseTable = (OlapTable) ((Database)
Env.getCurrentInternalCatalog()
+
.getDbOrMetaException("test_stream")).getTableOrMetaException("tbl_stream_unconsumed");
+ Plan plan = normalizeStreamRead("select * from
test_stream.s_unconsumed",
+
TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis()),
Optional.empty());
+
+
Assertions.assertEquals(ImmutableSet.of(baseTable.getPartition("p1").getId(),
+ baseTable.getPartition("p2").getId()),
scannedPartitionIds(plan, true));
+ Assertions.assertEquals(ImmutableSet.of(), scannedPartitionIds(plan,
false));
+ }
+
+ @Test
+ public void testIvmReadsPartitionWithoutBaselineFromBaseTable() throws
Exception {
+ // The same read inside an IVM refresh takes the partition without a
consumption baseline from the base
+ // table: none of its rows are in the MV yet, and its changelog may no
longer reach back to its first row.
+ // The partition with an offset is still read from the changelog.
+ OlapTable baseTable = (OlapTable) ((Database)
Env.getCurrentInternalCatalog()
+
.getDbOrMetaException("test_stream")).getTableOrMetaException("tbl_stream_unconsumed");
+ Plan plan = normalizeStreamRead("select * from
test_stream.s_unconsumed",
+
TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis()),
+
Optional.of(IvmRewriteContext.incremental(Mockito.mock(MTMV.class))));
+
+
Assertions.assertEquals(ImmutableSet.of(baseTable.getPartition("p1").getId()),
+ scannedPartitionIds(plan, true));
+
Assertions.assertEquals(ImmutableSet.of(baseTable.getPartition("p2").getId()),
+ scannedPartitionIds(plan, false));
+ }
+
+ // The partitions the plan reads from the changelog, or from the base
table itself.
+ private static Set<Long> scannedPartitionIds(Plan plan, boolean changelog)
{
+ Set<LogicalOlapScan> scans = plan.collect(node -> node instanceof
LogicalOlapScan
+ && (changelog ? ((LogicalOlapScan) node).getTable() instanceof
OlapChangelogTableWrapper
+ : !(((LogicalOlapScan) node).getTable() instanceof
OlapTableWrapper)));
+ ImmutableSet.Builder<Long> partitionIds = ImmutableSet.builder();
+ scans.forEach(scan ->
partitionIds.addAll(scan.getSelectedPartitionIds()));
+ return partitionIds.build();
+ }
+
+ private Plan normalizeStreamRead(String sql, long referenceTso,
Optional<IvmRewriteContext> ivmRewriteContext) {
+ PlanChecker checker = PlanChecker.from(connectContext).parse(sql);
+ StatementContext statementContext =
checker.getCascadesContext().getStatementContext();
+ // Direct rewrite tests bypass the planner's statement reference
initialization.
+ statementContext.getOrRegisterChangelogReferenceTso(() ->
referenceTso);
+ checker.analyze();
+ // Installed after the analysis, whose IVM rules would rewrite the
query as an MV's: only how the stream
+ // read treats its offset inside an IVM statement is under test.
+ statementContext.setIvmRewriteContext(ivmRewriteContext);
+ return checker.applyTopDown(new
NormalizeOlapTableStreamScan()).getCascadesContext().getRewritePlan();
+ }
+
@Test
public void testIncrementalScanChangeTypeProjectionIsCaseExpr() throws
Exception {
ConnectContext ctx = createDefaultCtx();
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_without_offset.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_without_offset.out
new file mode 100644
index 00000000000..f8649546dcb
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_without_offset.out
@@ -0,0 +1,73 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !u_baseline_task --
+SUCCESS COMPLETE NONE
+
+-- !d_baseline_task --
+SUCCESS COMPLETE NONE
+
+-- !p_baseline_task --
+SUCCESS COMPLETE NONE
+
+-- !u_task --
+SUCCESS NONE NONE
+
+-- !u_mv --
+1 2026-01-10 10
+2 2026-02-10 20
+3 2026-03-10 30
+
+-- !d_task --
+SUCCESS NONE NONE
+
+-- !d_mv --
+1 2026-01-10 10
+2 2026-02-10 20
+3 2026-03-10 30
+
+-- !p_task --
+SUCCESS PARTIAL NONE
+
+-- !p_mv --
+1 2026-01-10 10
+2 2026-02-10 20
+3 2026-03-10 30
+
+-- !u_task_after_offsets --
+SUCCESS NONE NONE
+
+-- !u_mv_after_offsets --
+1 2026-01-10 10
+2 2026-02-10 21
+3 2026-03-10 30
+4 2026-03-11 40
+
+-- !d_task_after_offsets --
+SUCCESS NONE NONE
+
+-- !d_mv_after_offsets --
+1 2026-01-10 10
+2 2026-02-10 20
+3 2026-03-10 30
+4 2026-03-11 40
+
+-- !p_task_after_offsets --
+SUCCESS NONE NONE
+
+-- !p_mv_after_offsets --
+1 2026-01-10 10
+2 2026-02-10 21
+3 2026-03-10 30
+4 2026-03-11 40
+
+-- !base_u --
+1 2026-01-10 10
+2 2026-02-10 21
+3 2026-03-10 30
+4 2026-03-11 40
+
+-- !base_d --
+1 2026-01-10 10
+2 2026-02-10 20
+3 2026-03-10 30
+4 2026-03-11 40
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_without_offset.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_without_offset.groovy
new file mode 100644
index 00000000000..34ff7ed7a35
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_without_offset.groovy
@@ -0,0 +1,213 @@
+// 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
+
+/**
+ * An incremental refresh that reads a base partition the stream has no offset
for: one added after the MV was
+ * built, and one that was still empty then.
+ *
+ * <p>A stream records an offset for a partition when it consumes the
partition with data in it, so neither of
+ * the two has one, and none of their rows are in the MV. Read through the
changelog, such a partition is read
+ * from the first TSO the changelog retains, and the rows older than
changelog.ttl_seconds would be left out
+ * while the refresh succeeds. The refresh reads it from the base table
instead, as an append of all its rows.
+ *
+ * <p>Cases pinned here, for an MV over a unique-key table, an append-only
detail MV over a duplicate-key table
+ * and an MV partitioned like its unique-key base table:
+ * <ol>
+ * <li>the rows written to the two partitions are older than the TTL when
the MV is refreshed: a strict
+ * INCREMENTAL refresh succeeds and the MV has them;</li>
+ * <li>with the TTL back to a day, the two partitions have offsets now, and
the next refresh reads their
+ * changes from the changelog;</li>
+ * <li>the partitioned MV rebuilds the MV partition of the added base
partition, as before.</li>
+ * </ol>
+ *
+ * <p>nonConcurrent: the rows age past a one-second TTL by sleeping.
+ */
+suite("test_ivm_partition_without_offset", "nonConcurrent") {
+ def waitForNewTask = { String mvName, 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]
+ }
+
+ // How a refresh ended, in one row. Unset RefreshMode / IvmFallbackReason
come back as the literal
+ // two-character string "\N", which does not survive the .out round trip,
so fold them into a printable
+ // token; a reason is an enum name, which "\N" is not.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END,
+ CASE WHEN IvmFallbackReason REGEXP '^[A-Z_]+\$'
+ THEN IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_no_offset_u_mv"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_no_offset_d_mv"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_no_offset_u_pmv"""
+ sql """DROP TABLE IF EXISTS ivm_no_offset_u"""
+ sql """DROP TABLE IF EXISTS ivm_no_offset_d"""
+
+ sql """
+ CREATE TABLE ivm_no_offset_u (
+ k BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ v BIGINT
+ )
+ UNIQUE KEY(k, dt)
+ PARTITION BY RANGE(dt) (
+ PARTITION p202601 VALUES [('2026-01-01'), ('2026-02-01')),
+ PARTITION p202602 VALUES [('2026-02-01'), ('2026-03-01'))
+ )
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ sql """
+ CREATE TABLE ivm_no_offset_d (
+ k BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ v BIGINT
+ )
+ DUPLICATE KEY(k)
+ PARTITION BY RANGE(dt) (
+ PARTITION p202601 VALUES [('2026-01-01'), ('2026-02-01')),
+ PARTITION p202602 VALUES [('2026-02-01'), ('2026-03-01'))
+ )
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "changelog.enable" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ // p202602 stays empty until after the MVs are built.
+ sql """INSERT INTO ivm_no_offset_u VALUES (1, '2026-01-10', 10)"""
+ sql """INSERT INTO ivm_no_offset_d VALUES (1, '2026-01-10', 10)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ivm_no_offset_u_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT k, dt, v FROM ivm_no_offset_u
+ """
+ sql """
+ CREATE MATERIALIZED VIEW ivm_no_offset_d_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT k, dt, v FROM ivm_no_offset_d
+ """
+ sql """
+ CREATE MATERIALIZED VIEW ivm_no_offset_u_pmv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(k, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT k, dt, v FROM ivm_no_offset_u
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_mv INCREMENTAL"""
+ def uTask = waitForNewTask('ivm_no_offset_u_mv', null)
+ qt_u_baseline_task taskQuery(uTask.TaskId.toString())
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_d_mv INCREMENTAL"""
+ def dTask = waitForNewTask('ivm_no_offset_d_mv', null)
+ qt_d_baseline_task taskQuery(dTask.TaskId.toString())
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_pmv INCREMENTAL"""
+ def pTask = waitForNewTask('ivm_no_offset_u_pmv', null)
+ qt_p_baseline_task taskQuery(pTask.TaskId.toString())
+
+ // 1. Rows in the partition that was empty when the MVs were built and in
a partition added after that, both
+ // older than the TTL by the time the MVs are refreshed. Nothing else
changes, so no offset is read.
+ sql """ALTER TABLE ivm_no_offset_u ADD PARTITION p202603 VALUES
[('2026-03-01'), ('2026-04-01'))"""
+ sql """ALTER TABLE ivm_no_offset_d ADD PARTITION p202603 VALUES
[('2026-03-01'), ('2026-04-01'))"""
+ sql """INSERT INTO ivm_no_offset_u VALUES (2, '2026-02-10', 20), (3,
'2026-03-10', 30)"""
+ sql """INSERT INTO ivm_no_offset_d VALUES (2, '2026-02-10', 20), (3,
'2026-03-10', 30)"""
+ sql """ALTER TABLE ivm_no_offset_u SET ("changelog.ttl_seconds" = "1")"""
+ sql """ALTER TABLE ivm_no_offset_d SET ("changelog.ttl_seconds" = "1")"""
+ sleep(1500)
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_mv INCREMENTAL"""
+ uTask = waitForNewTask('ivm_no_offset_u_mv', uTask.TaskId.toString())
+ qt_u_task taskQuery(uTask.TaskId.toString())
+ order_qt_u_mv """SELECT k, dt, v FROM ivm_no_offset_u_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_d_mv INCREMENTAL"""
+ dTask = waitForNewTask('ivm_no_offset_d_mv', dTask.TaskId.toString())
+ qt_d_task taskQuery(dTask.TaskId.toString())
+ order_qt_d_mv """SELECT k, dt, v FROM ivm_no_offset_d_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_pmv INCREMENTAL"""
+ pTask = waitForNewTask('ivm_no_offset_u_pmv', pTask.TaskId.toString())
+ qt_p_task taskQuery(pTask.TaskId.toString())
+ order_qt_p_mv """SELECT k, dt, v FROM ivm_no_offset_u_pmv"""
+
+ // 2. With the TTL back to a day, the offsets the refreshes above recorded
are well inside the retention, and
+ // what changes after them is read from the changelog: an update and a new
row in the unique-key table, a
+ // new row in the duplicate-key one.
+ sql """ALTER TABLE ivm_no_offset_u SET ("changelog.ttl_seconds" =
"86400")"""
+ sql """ALTER TABLE ivm_no_offset_d SET ("changelog.ttl_seconds" =
"86400")"""
+ sql """INSERT INTO ivm_no_offset_u VALUES (2, '2026-02-10', 21), (4,
'2026-03-11', 40)"""
+ sql """INSERT INTO ivm_no_offset_d VALUES (4, '2026-03-11', 40)"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_mv INCREMENTAL"""
+ uTask = waitForNewTask('ivm_no_offset_u_mv', uTask.TaskId.toString())
+ qt_u_task_after_offsets taskQuery(uTask.TaskId.toString())
+ order_qt_u_mv_after_offsets """SELECT k, dt, v FROM ivm_no_offset_u_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_d_mv INCREMENTAL"""
+ dTask = waitForNewTask('ivm_no_offset_d_mv', dTask.TaskId.toString())
+ qt_d_task_after_offsets taskQuery(dTask.TaskId.toString())
+ order_qt_d_mv_after_offsets """SELECT k, dt, v FROM ivm_no_offset_d_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_no_offset_u_pmv INCREMENTAL"""
+ pTask = waitForNewTask('ivm_no_offset_u_pmv', pTask.TaskId.toString())
+ qt_p_task_after_offsets taskQuery(pTask.TaskId.toString())
+ order_qt_p_mv_after_offsets """SELECT k, dt, v FROM ivm_no_offset_u_pmv"""
+
+ // What the base tables answer for the MV queries. Rewrite is off: a query
of the MV's shape would otherwise
+ // be answered from an MV, and the comparison would compare the MV with
itself.
+ sql """SET enable_materialized_view_rewrite = false"""
+ order_qt_base_u """SELECT k, dt, v FROM ivm_no_offset_u"""
+ order_qt_base_d """SELECT k, dt, v FROM ivm_no_offset_d"""
+ sql """SET enable_materialized_view_rewrite = true"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
