morningman opened a new pull request, #69778:
URL: https://github.com/apache/doris/pull/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.
### Release note
Fix an IVM materialized view missing the rows of a base partition that was
added after the MV was created, or was empty then, when those rows were older
than the table's `changelog.ttl_seconds` at refresh time.
### Check List (For Author)
- Test: Regression test / Unit Test
- Regression test: `mtmv_p0/ivm/test_ivm_partition_without_offset` (new,
nonConcurrent): a non-partitioned 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, each refreshed after rows written to an added partition and to the
partition that was empty became older than the TTL, then again after more
changes with the TTL back to a day, and compared with the base table with MV
rewrite off. Local shared-nothing cluster: first on a build without this
change, where all three refreshes succeeded without those rows; with the change
the suite passes, and passes again against the generated `.out`. On the same
build the whole `mtmv_p0/ivm` directory (113 suites), `mtmv_ivm_restart_p0` and
`changelog_p0/test_changelog_before_column_name` pass. After the rebase onto
the latest master (two commits that touch neither this code nor IVM), the new
suite and the other `test_ivm_partition_*` suites pass again. Not r
un on a cloud cluster.
- Unit Test: `ExplainTableStreamPlanTest` (two new cases: a user stream
reads a partition without a consumption baseline from the changelog; the same
read in an IVM statement takes it from the base table and still reads the
partition with an offset from the changelog), run together with
`IvmIncrRefreshMTMVTest`, `IvmFullRefreshMTMVTest`,
`InsertIntoTableCommandTableStreamTest`, `UpdateMvByPartitionCommandTest` and
`CollectRelationChangelogTtlTest`: 84 tests pass (before the rebase).
- Behavior changed: Yes. An IVM refresh reads a partition with no stream
offset from the base table instead of from its changelog; for such a partition
`EXPLAIN REFRESH ... INCREMENTAL` shows a scan of the base table.
- Does this need documentation: No
--
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]
