yujun777 opened a new issue, #65909:
URL: https://github.com/apache/doris/issues/65909
## Summary
`MIN_DELTA` row-binlog streams return incorrect results when a predicate
references a non-key value column that changes in an update. The issue is
independent of IVM and can be reproduced by querying a stream directly.
For an update from `region = 'east'` to `region = 'north'`:
- an unfiltered `MIN_DELTA` stream returns both `UPDATE_BEFORE(east)` and
`UPDATE_AFTER(north)`;
- `WHERE region = 'east'` returns no rows, although it should return the
before row;
- `WHERE region = 'north'` returns only the after row, as expected after
filtering the reconstructed change rows.
## Environment
- Local non-cloud deployment: 1 FE + 1 BE, replication factor 1.
- FE: `doris-0.0.0-cbe6751fc3f`; BE: `doris-0.0.0-b9b4f89b7c8`.
- Source checkout used for investigation:
`a4dae0b64e9e7c6db5983b566656442f1ae2bc5d`.
## Minimal reproduction
```sql
DROP DATABASE IF EXISTS stream_predicate_pushdown_probe;
CREATE DATABASE stream_predicate_pushdown_probe;
USE stream_predicate_pushdown_probe;
CREATE TABLE events (
id BIGINT NOT NULL,
region VARCHAR(16) NOT NULL,
amount BIGINT NOT NULL
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES (
'replication_num' = '1',
'enable_unique_key_merge_on_write' = 'true',
'binlog.enable' = 'true',
'binlog.format' = 'ROW',
'binlog.need_historical_value' = 'true'
);
INSERT INTO events VALUES (1, 'east', 10);
-- Use independent streams so each query has the same initial offset.
CREATE STREAM s_all ON TABLE events
PROPERTIES ('type' = 'min_delta', 'show_initial_rows' = 'false');
CREATE STREAM s_east ON TABLE events
PROPERTIES ('type' = 'min_delta', 'show_initial_rows' = 'false');
CREATE STREAM s_north ON TABLE events
PROPERTIES ('type' = 'min_delta', 'show_initial_rows' = 'false');
UPDATE events SET region = 'north', amount = 20 WHERE id = 1;
SELECT id, region, amount, __DORIS_STREAM_CHANGE_TYPE_COL__ AS change_type
FROM s_all
ORDER BY change_type;
SELECT id, region, amount, __DORIS_STREAM_CHANGE_TYPE_COL__ AS change_type
FROM s_east
WHERE region = 'east'
ORDER BY change_type;
SELECT id, region, amount, __DORIS_STREAM_CHANGE_TYPE_COL__ AS change_type
FROM s_north
WHERE region = 'north'
ORDER BY change_type;
SELECT stream_name, consumption_status, lag, last_consumption_time
FROM information_schema.table_stream_consumption
WHERE db_name = 'stream_predicate_pushdown_probe'
ORDER BY stream_name;
```
## Observed result
The unfiltered stream correctly returns:
```text
1 | north | 20 | UPDATE_AFTER
1 | east | 10 | UPDATE_BEFORE
```
`s_east WHERE region = 'east'` returns zero rows. It should return:
```text
1 | east | 10 | UPDATE_BEFORE
```
`s_north WHERE region = 'north'` returns only:
```text
1 | north | 20 | UPDATE_AFTER
```
The three streams retain their initial consumption offsets and have positive
lag after the SELECT queries, so the behavior is not caused by stream
consumption.
## Expected result
The predicate must be evaluated against each reconstructed change row.
`UPDATE_BEFORE` should use its historical values and `UPDATE_AFTER` should use
its new values. Therefore `WHERE region = 'east'` must return the before row.
## Investigation
The row-binlog record stores the new value in the normal `region` column and
the historical value in `__BEFORE__region__`. Storage predicates are currently
pushed down before `BlockReader::_min_delta_next_block()` reconstructs
`UPDATE_BEFORE` rows from the `__BEFORE__*` mirrors. As a result, `region =
'east'` filters the raw after-image (`north`) before the before-image can be
produced.
Relevant paths:
- `be/src/exec/scan/olap_scanner.cpp`: `_init_tablet_reader_params()` copies
slot predicates into storage predicates.
- `be/src/storage/iterator/block_reader.cpp`: `_min_delta_next_block()`
later reconstructs before/after rows.
Suggested direction: for row-binlog `MIN_DELTA` and `DETAIL` scans, retain
TSO-range filtering but do not push down user value-column predicates, function
filters, or common expression filters before change-row reconstruction. They
should be evaluated by the upper scan/filter layer after reconstruction. A
later optimization could safely push a predicate only when it is evaluated
against both before and after images.
## Scope
This is a direct stream/binlog reproduction and does not require IVM,
complete refresh, or MV refresh. It causes IVM filtered incremental refreshes
to retain rows when a source row stops matching a filter.
Related issues: [#65265](https://github.com/apache/doris/issues/65265),
[#65418](https://github.com/apache/doris/issues/65418).
--
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]
