ricky2129 opened a new pull request, #12808:
URL: https://github.com/apache/seatunnel/pull/12808
### Purpose of this pull request
Fixes #12806
`BufferReducedBatchStatementExecutor` keeps only the last change per key.
With `enable_upsert = false`, `InsertOrUpdateBatchStatementExecutor` picks
`INSERT` or `UPDATE` from that last change's row kind, so:
- `INSERT` then `UPDATE_AFTER` in one batch became an `UPDATE` of a row that
doesn't exist yet. The row is never inserted and no error is raised.
- `DELETE` then `INSERT` became an `INSERT` of a row still in the table,
which fails with a duplicate key.
This PR records, per key, whether the row existed before the batch, judged
from the key's first change. When the batch is flushed, the kept row is written
as `INSERT` if it did not exist before and `UPDATE_AFTER` if it did. Deletes
are unchanged. The state is cleared with the buffer, including in
`clearBatch()`.
With `enable_upsert = true` the upsert statement is the same for both kinds,
so behaviour there is unchanged.
### Does this PR introduce _any_ user-facing change?
Yes. With `enable_upsert = false`, rows that are inserted and changed within
one flush are now written. Before, they were silently missing from the target.
With `is_exactly_once = true` a flush spans a whole checkpoint, so this
affected every row created and changed within one checkpoint interval.
### How was this patch tested?
- Added
`BufferReducedBatchStatementExecutorTest#testReducedRowKeepsTheNetEffectOfTheBatch`,
covering insert→update, update→update, delete→insert, update→delete, and a
following batch. It fails on dev (`expected: <[+I[1, b], +U[2, y], +U[3, q]]>
but was: <[+U[1, b], +U[2, y], +I[3, q]]>`) and passes with this change. The
existing executor tests still pass.
- End to end, on a 2.3.12-based build: MySQL-CDC → JDBC with `enable_upsert
= false` and XA, 8 tables, about 2,600 mixed insert/update operations. Before:
50 of 960 rows missing with no restarts. After: 0 missing, also across three
runs with four `kill -9` restarts each.
### Check list
* [ ] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/contribution/new-license.md)
* [ ] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs
* [ ] If you are contributing the connector code, please check that the
following files are updated:
1. Update
[plugin-mapping.properties](https://github.com/apache/seatunnel/blob/dev/plugin-mapping.properties)
and add new connector information in it
2. Update the pom file of
[seatunnel-dist](https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml)
3. Add ci label in
[label-scope-conf](https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml)
4. Add e2e testcase in
[seatunnel-e2e](https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/)
5. Update connector
[plugin_config](https://github.com/apache/seatunnel/blob/dev/config/plugin_config)
--
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]
