MoanasDaddyXu opened a new issue, #65705:
URL: https://github.com/apache/doris/issues/65705

   ### Tracking
   
   - Tracking issue: https://github.com/apache/doris/issues/65265
   ## Reference
   
   - Similar issue format: https://github.com/apache/doris/issues/65383
   - Regression log: 
`/mnt/disk1/jianxu/selectdb-cases/output/regression-test/log/doris-regression-test.20260716.152407.log`
   - Failed case: 
`/mnt/disk1/jianxu/selectdb-cases/regression-test/suites/stress/binlog/test_binlog_routine_load_merge_types.groovy`
   
   ## Search before asking
   
   I searched the existing Apache Doris issue format and used issue #65383 as 
the description template.
   
   ## Version
   
   The failure was reproduced on FE `172.20.49.4:9030`.
   
   The exact Doris build version was not printed in the regression log. The 
regression framework connected to:
   
   ```text
   jdbc:mysql://172.20.49.4:9030/regression_test_stress_binlog
   ```
   
   ## What's Wrong
   
   The regression case `test_binlog_routine_load_merge_types` failed after the 
base table DELETE phase.
   
   The case flow is:
   
   1. Create a source MOW unique-key table with row binlog enabled.
   2. Create a target MOW unique-key table.
   3. Create a `MIN_DELTA` table stream on the source table.
   4. Insert 100,000,000 rows into the source table through Routine Load and 
consume the table stream into the target table.
   5. Update the same 100,000,000 rows and consume the table stream into the 
target table.
   6. Execute `DELETE FROM routine_load_source WHERE id <= 20000000`.
   7. Consume DELETE events from the table stream into the target table by 
writing `__DORIS_DELETE_SIGN__ = 1`.
   
   The source table was correctly reduced to 80,000,000 rows, but the target 
table still had 100,000,000 rows.
   
   Regression log excerpt:
   
   ```text
   DELETE row count: source=80000000, target=100000000, expected=80000000
   org.opentest4j.AssertionFailedError: target row count mismatch after DELETE 
==> expected: <80000000> but was: <100000000>
   ```
   
   The table stream lag was consumed to zero after the DELETE phase, but the 
downstream target table did not receive the DELETE rows:
   
   ```text
   table stream lag after DELETE: [[routine_load_min_delta_stream, 
routine_load_source, 14005305344, 467714225933910017]]
   table stream lag after consuming DELETE: [[routine_load_min_delta_stream, 
routine_load_source, 0, 467714239939215361]]
   ```
   
   The case consumes the stream with:
   
   ```sql
   SET show_hidden_columns = true;
   
   INSERT INTO routine_load_target (id, version_no, payload, 
__DORIS_DELETE_SIGN__)
   SELECT id, version_no, payload,
          if(__DORIS_STREAM_CHANGE_TYPE_COL__ = 'DELETE', 1, 0)
   FROM routine_load_min_delta_stream
   WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN
         ('APPEND', 'INSERT', 'UPDATE_AFTER', 'DELETE');
   ```
   
   This works for INSERT and UPDATE events, but not for DELETE events.
   
   ## Expected Behavior
   
   When a `MIN_DELTA` table stream contains DELETE events, `INSERT INTO target 
SELECT ... FROM stream` should emit those DELETE rows to the target insert 
pipeline.
   
   For this case, writing the stream DELETE events into the target table with 
`__DORIS_DELETE_SIGN__ = 1` should reduce the target table from 100,000,000 
rows to 80,000,000 rows, matching the source table.
   
   ## How to Reproduce
   
   The following reduced SQL reproduces the product behavior without Kafka or 
Routine Load.
   
   ```sql
   DROP DATABASE IF EXISTS codex_stream_delete_probe3;
   CREATE DATABASE codex_stream_delete_probe3;
   USE codex_stream_delete_probe3;
   
   CREATE TABLE src (
       id BIGINT NOT NULL,
       version_no BIGINT NOT NULL,
       payload VARCHAR(32) NOT NULL
   ) ENGINE=OLAP
   UNIQUE KEY(id)
   DISTRIBUTED BY HASH(id) BUCKETS 1
   PROPERTIES (
       'replication_num' = '1',
       'enable_unique_key_merge_on_write' = 'true',
       'function_column.sequence_col' = 'version_no',
       'binlog.enable' = 'true',
       'binlog.format' = 'ROW',
       'binlog.need_historical_value' = 'true'
   );
   
   CREATE TABLE audit (
       id BIGINT,
       version_no BIGINT,
       payload VARCHAR(32),
       change_type VARCHAR(32)
   ) ENGINE=OLAP
   DUPLICATE KEY(id)
   DISTRIBUTED BY HASH(id) BUCKETS 1
   PROPERTIES (
       'replication_num' = '1'
   );
   
   CREATE STREAM s ON TABLE src
   PROPERTIES (
       'type' = 'min_delta',
       'show_initial_rows' = 'false'
   );
   
   INSERT INTO src VALUES (1, 1, 'a1'), (2, 1, 'a2'), (3, 1, 'a3');
   SYNC;
   
   INSERT INTO audit
   SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
   FROM s
   WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'INSERT', 
'UPDATE_AFTER', 'DELETE');
   
   SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER BY 
change_type;
   
   INSERT INTO src VALUES (1, 2, 'b1'), (2, 2, 'b2'), (3, 2, 'b3');
   SYNC;
   
   INSERT INTO audit
   SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
   FROM s
   WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'INSERT', 
'UPDATE_AFTER', 'DELETE');
   
   SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER BY 
change_type;
   
   DELETE FROM src WHERE id <= 2;
   SYNC;
   
   SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
   FROM s
   ORDER BY id;
   
   INSERT INTO audit
   SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
   FROM s
   WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'INSERT', 
'UPDATE_AFTER', 'DELETE');
   
   SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER BY 
change_type;
   ```
   
   Observed output:
   
   ```text
   change_type  count(*)
   APPEND       3
   
   change_type  count(*)
   APPEND       3
   UPDATE_AFTER 3
   
   id  version_no  payload  __DORIS_STREAM_CHANGE_TYPE_COL__
   1   2           b1       DELETE
   2   2           b2       DELETE
   
   OK (0 rows affected)
   
   change_type  count(*)
   APPEND       3
   UPDATE_AFTER 3
   ```
   
   The direct `SELECT FROM s` can see two DELETE events, but the following 
`INSERT INTO audit SELECT ... FROM s` inserts zero rows.
   
   ## Regression Test Result
   
   The failing regression command generated by the framework:
   
   ```text
   -cf 
/mnt/disk1/jianxu/selectdb-cases/regression-test/conf/regression-conf.groovy -d 
stress/binlog -s test_binlog_routine_load_merge_types --flowIds 
stress/binlog/test_binlog_routine_load_merge_types.groovy#test_binlog_routine_load_merge_types
   ```
   
   Failed suite:
   
   ```text
   
/mnt/disk1/jianxu/selectdb-cases/regression-test/suites/stress/binlog/test_binlog_routine_load_merge_types.groovy
   group=binlog,p0
   name=test_binlog_routine_load_merge_types
   elapsed=1450043ms
   ```
   
   ## Why this looks like a product issue instead of a case issue
   
   I also checked whether the target table delete-sign write itself was invalid.
   
   A reduced MOW table test showed that direct `INSERT INTO ... 
__DORIS_DELETE_SIGN__` works and can delete visible rows, even with the same 
sequence value:
   
   ```text
   c1
   1
   c2
   0
   c3
   0
   ```
   
   So the delete-sign write path is valid. The problem only appears when DELETE 
rows are read from a table stream through `INSERT SELECT`.
   
   This points to the table stream `INSERT SELECT` consumption path for DELETE 
events, not to the regression case assertion.
   
   ## Anything Else
   
   Temporary probe databases used during analysis were cleaned up:
   
   ```text
   codex_delete_sign_seq_probe
   codex_stream_delete_probe
   codex_stream_delete_probe2
   codex_stream_delete_probe3
   ```


-- 
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]

Reply via email to