yujun777 commented on code in PR #66889:
URL: https://github.com/apache/doris/pull/66889#discussion_r3809796334
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java:
##########
@@ -174,6 +175,11 @@ public List<Slot> computeOutput() {
// add stream exclusive virtual columns.
slots.add(SlotReference.fromColumn(
exprIdGenerator.getNextId(), table,
Column.STREAM_SEQ_VIRTUAL_COLUMN, qualified()));
+ // Only expose stream LSN when the base table stores row LSN, e.g.
dup table with binlog.
+ if (table instanceof OlapTable && ((OlapTable)
table).getKeysType() == KeysType.DUP_KEYS) {
Review Comment:
**Snapshot read mode does not expose the LSN slot for DUP tables.**
In `computeOutput()`, for `StreamReadMode.SNAPSHOT`: `readMode == RESET` is
false, so `table.getBaseSchema(false)` returns only visible columns (the hidden
`__DORIS_ROW_LSN_COL__` is filtered out), and since `isIncremental()` is false,
`STREAM_LSN_VIRTUAL_COLUMN` is not added either. Then in `makeSnapshotScan()`'s
DUP branch, `projectToOriginSlots(makeOlapScanOnBaseTable(...), originSlots)`
projects the base scan (which *does* contain `ROW_LSN_COL`) back onto the
snapshot `originSlots` (visible columns only), so the LSN is dropped.
Net effect: `SELECT __DORIS_ROW_LSN_COL__ / __DORIS_STREAM_LSN_COL__ FROM
stream@snapshot()` fails with unknown column, so `table` / `stream` /
`stream@reset` see the lsn slot but `stream@snapshot` does not. If the goal is
that a detail (DUP) table exposes the LSN slot in all three read modes (needed
for a stable row id in full refresh), this path is missing it.
Suggest one of: (a) for SNAPSHOT mode on DUP base tables, build `baseSchema`
from the full schema (like RESET) so `ROW_LSN_COL` flows through
`makeSnapshotScan`'s projection; or (b) also add `STREAM_LSN_VIRTUAL_COLUMN` in
snapshot mode and map it in the snapshot rebuild plan. Either way, add a
regression test querying the lsn slot on `stream@snapshot()`. The existing
`checkStreamVirtualColumnsHidden` only covers
`STREAM_CHANGE_TYPE_COL`/`STREAM_SEQUENCE_COL`, so this gap is not caught.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamBuildFactory.java:
##########
@@ -61,6 +63,13 @@ public BaseTableStream build() throws DdlException {
Column sequenceColumn = new Column(Column.STREAM_SEQ_COL, Type.BIGINT);
sequenceColumn.setIsVisible(false);
schema.add(sequenceColumn);
+ // Only expose stream LSN when the base table stores row LSN, e.g. dup
table with binlog.
+ if (params.baseTable instanceof OlapTable
Review Comment:
**LSN exposure is keyed on `keysType == DUP_KEYS` alone, without verifying
the base table actually stores a row LSN.**
This condition (and the identical one in
`LogicalOlapTableStreamScan.computeOutput()`) decides whether `STREAM_LSN_COL`
/ `STREAM_LSN_VIRTUAL_COLUMN` is exposed based only on the key type. If a DUP
table does not actually carry `__DORIS_ROW_LSN_COL__` (e.g. binlog enabled
without the hidden columns, or an old schema), the stream schema will still
advertise `STREAM_LSN_COL` while the history path's
`Preconditions.checkArgument(lsnSlot != null)` will fail. Since
`binlogConfig.isRowFormat()` already gates column injection at `CREATE TABLE`
(`createRowBinlogHiddenColumnsIfNecessary`), consider keying this on the actual
presence of the `ROW_LSN_COL` column in the base table schema (or on
`needRowBinlog()`/`isRowFormat()`) so the stream schema and the normalize rule
stay consistent.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java:
##########
@@ -197,17 +206,13 @@ private Plan
makeIncrementalScanFromBinlog(CascadesContext cascadesContext, Logi
if (isIncremental) {
// replace stream virtual column with alias slot reference
for (Slot slot : originSlots) {
- if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
+ if (isStreamVirtualSlot(slot,
Column.STREAM_CHANGE_TYPE_VIRTUAL_COLUMN)) {
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), buildChangeTypeExpr(opSlot),
Column.STREAM_CHANGE_TYPE_COL));
- } else if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference) slot).getOriginalColumn().get()
- .equals(Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
+ } else if (isStreamVirtualSlot(slot,
Column.STREAM_SEQ_VIRTUAL_COLUMN)) {
project.add(new
Alias(StatementScopeIdGenerator.newExprId(), seqSlot, Column.STREAM_SEQ_COL));
+ } else if (isStreamVirtualSlot(slot,
Column.STREAM_LSN_VIRTUAL_COLUMN)) {
+ project.add(new
Alias(StatementScopeIdGenerator.newExprId(), lsnSlot, Column.STREAM_LSN_COL));
Review Comment:
**`lsnSlot` is used without a null check in the incremental path, unlike the
history path.**
The history path guards with `Preconditions.checkArgument(lsnSlot != null,
"Row lsn column not found in base table output")` for DUP tables (line ~387),
but here `lsnSlot` (found by matching `BINLOG_LSN_COL` in the binlog output) is
passed directly into an `Alias`. Today the row-binlog schema always contains
`BINLOG_LSN_COL`, so this happens to be safe, but it will NPE rather than fail
cleanly if the binlog output ever lacks the LSN column (e.g. a DUP table whose
binlog was enabled through a path that does not add the LSN column). For
consistency with the history path, add the same precondition (or
`Preconditions.checkState`) before using `lsnSlot`.
--
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]
