-
Notifications
You must be signed in to change notification settings - Fork 555
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat(connector): support transaction for single-table CDC #11453
Conversation
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Codecov Report
@@ Coverage Diff @@
## main #11453 +/- ##
==========================================
- Coverage 70.42% 70.41% -0.02%
==========================================
Files 1351 1351
Lines 225283 225338 +55
==========================================
+ Hits 158655 158664 +9
- Misses 66628 66674 +46
Flags with carried forward coverage won't be shown. Click here to find out more.
... and 2 files with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The rest LGTM
split_offset_mapping: Some(split_offset_mapping), | ||
}; | ||
// If we are not in a transaction, we should yield the chunk now. | ||
if current_transaction.is_none() { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shall we limit the max size of a transaction? Similar with the limitation for DML transaction. Otherwise, it's possible that a very large transaction from upstream DB causes RW to crash
risingwave/src/stream/src/executor/dml.rs
Lines 57 to 64 in 0a9d2b4
/// If a transaction's data is less than `MAX_CHUNK_FOR_ATOMICITY` * `CHUNK_SIZE`, we can provide | |
/// atomicity. Otherwise, it is possible that part of transaction's data is sent to the downstream | |
/// without barrier boundaries. There are some cases that could cause non-atomicity for large | |
/// transaction. 1. The system crashes. | |
/// 2. Actor scale-in or migration. | |
/// 3. Dml's batch query error occurs at the middle of its execution. (e.g. Remove UDF function | |
/// server become unavailable). | |
const MAX_CHUNK_FOR_ATOMICITY: usize = 32; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Done. PTAL 🥰
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we need to unify these 2 numbers and document it somewhere in our docs. cc. @chenzl25
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
+1. We need to document it in DML and table CDC section.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@chenzl25 @fuyufjh @BugenZhao Please see this doc PR for the related doc updates. Please review and make edits as you see fit. Thanks.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
basically LGTM
so we cannot handle case that txn and scale occur at the same time? and how to prevent it?
#[allow(clippy::unused_async)] // false positive for `async_trait` | ||
async fn parse_one<'a>( | ||
&'a mut self, | ||
_key: Option<Vec<u8>>, | ||
_payload: Option<Vec<u8>>, | ||
_writer: SourceStreamChunkRowWriter<'a>, | ||
) -> Result<WriteGuard> { | ||
unreachable!("should call `parse_one_with_txn` instead") | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
just remove it?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's a required method of this trait, used for simplifying the implementation of other connectors that do not care about transactions.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Rest lgtm. I think we should set a maximum events limit for transactions to prevent OOM.
I don't quite get it. 👀 As we never yield a data chunk when currently inside a transaction, the offset persisted by the source executor for recovery and scaling will also respect the transaction. 🤔 |
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
Signed-off-by: Bugen Zhao i@bugenzhao.comI hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
Support for handling transaction boundaries in Debezium CDC messages. Users can specify
transactional = 'true'
in the table definition to enable this feature. Tested locally.The idea is relatively simple:
Checklist
./risedev check
(or alias,./risedev c
)Documentation
Click here for Documentation
Types of user-facing changes
Please keep the types that apply to your changes, and remove the others.
Release note
EXPERIMENTAL Specify
transactional = 'true'
in theWITH
clause of a CDC table definition to provide atomicity for transactions in the upstream for this single table. For the consideration of performance, transactions involving changes of more than 4096 rows cannot be guaranteed.