Userwhite commented on code in PR #66889:
URL: https://github.com/apache/doris/pull/66889#discussion_r3841334909
##########
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:
fixed
##########
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:
fixed
##########
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:
fixed
--
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]
