refactor(source): respect max chunk size in source connector parser #19698

stdrc · 2024-12-06T06:52:46Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

In SourceStreamChunkBuilder, with the information of SourceCtrlOps, we can actually restrict the chunk size early during chunk building, instead emitting large chunks first and cutting them into small ones later in apply_rate_limit. This has two benefits:

Reduce the chunk splitting cost in apply_rate_limit used by FetchExecutor, FsSourceExecutor, SourceBackfillExecutor and SourceExecutor.
Source connectors and parsers can utilize their better understanding of the payload (e.g. transaction information), to avoid blindly cutting chunks at random positions.

Main changes happened in this PR:

Change SourceCtrlOpts::rate_limit: Option<u32> to split_txn: bool.

This is because we already pass in a limited chunk size (basically min(stream_chunk_size, rate_limit_burst)) via SourceCtrlOpts::chunk_size. We still need split_txn to control whether we need to cut chunks during (CDC) transactions, as concluded here.
Simplify SourceStreamChunkRowWriter, by directly store a reference to SourceStreamChunkBuilder inside it.
Rewrite SourceStreamChunkBuilder to allow creating chunk during writing to SourceStreamChunkRowWriter. The trick is to add a ready_chunks field.

This is because that it's possible to parse multiple rows from one SourceMessage, while the parser API doesn't use async stream to yield rows, instead, it directly write to SourceStreamChunkRowWriter.

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

stdrc · 2024-12-06T06:53:07Z

refactor(source): respect max chunk size in source connector parser #19698 : 2 dependent PRs (#19826 , #19863 ) 👈 (View in Graphite)
refactor(cdc source): merge cdc heartbeat chunk builder & data chunk builder #19671
refactor(connector): re-arrange the connector::parser module for better readibility #19697
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

tabVersion · 2024-12-12T03:35:57Z

any context about the refactor?

stdrc · 2024-12-12T06:42:59Z

any context about the refactor?

still WIP, will add description when it's ready

fuyufjh

Generally LGTM

fuyufjh · 2024-12-18T06:27:03Z

src/connector/src/parser/chunk_builder.rs

+        if let Some(ref mut txn) = self.ongoing_txn {
+            txn.len += 1;
+
+            if txn.len >= MAX_TRANSACTION_SIZE


There was a warning log before. Shall we keep it?

The warning log is moved to finish_current_chunk, so that we won't miss any call site of finish_current_chunk that can cause an interrupt of transaction.

fuyufjh · 2024-12-18T06:33:42Z

src/connector/src/parser/chunk_builder.rs

-        let builder = std::mem::replace(self, Self::with_capacity(descs, next_cap));
-        builder.finish()
+    /// Consumes and returns the ready [`StreamChunk`]s.
+    pub fn consume_ready_chunks(&mut self) -> impl ExactSizeIterator<Item = StreamChunk> + '_ {


Rewrite SourceStreamChunkBuilder to allow creating chunk during writing to SourceStreamChunkRowWriter. The trick is to add a ready_chunks field.

This is because that it's possible to parse multiple rows from one SourceMessage, while the parser API doesn't use async stream to yield rows, instead, it directly write to SourceStreamChunkRowWriter.

I understand the motivation to return multiple chunks, but what's reason of separating finish() into finish_current_chunk() + consume_ready_chunks()? I mean, if you just want multiple chunks, changing finish() to return Vec<StreamChunk> can achieve that.

Because sometimes we need to call consume_ready_chunks separately even we (the user of SourceStreamChunkBuilder) don't call finish_current_chunk explicitly before it. For example 1) in the arm ParseResult::Rows of match parser.parse_one_with_txn(); 2) after chunk_builder.commit_transaction(id).

Signed-off-by: Richard Chien <[email protected]>

…iter Signed-off-by: Richard Chien <[email protected]>

… type Signed-off-by: Richard Chien <[email protected]>

Signed-off-by: Richard Chien <[email protected]>

BugenZhao

Good

BugenZhao · 2024-12-19T03:41:50Z

src/connector/src/parser/chunk_builder.rs

    builders: Vec<ArrayBuilderImpl>,
    op_builder: Vec<Op>,
    vis_builder: BitmapBuilder,
+    ongoing_txn: Option<Transaction>,
+    ready_chunks: SmallVec<[StreamChunk; 1]>,


SmallVec looks good.

Can you please elaborate in which (unusual) case we'll have multiple ready chunks stashed?

When chunk_size is limited to 1, and somehow we parsed 3 rows from one SourceMessage.

May update the doc comment to reflect it would build multiple chunks instead of only one.

Added a comment for the chunk builder in 307ebc2

stdrc mentioned this pull request Dec 6, 2024

refactor(cdc source): merge cdc heartbeat chunk builder & data chunk builder #19671

Merged

9 tasks

github-actions bot added the Invalid PR Title label Dec 6, 2024

stdrc mentioned this pull request Dec 6, 2024

refactor(connector): re-arrange the connector::parser module for better readibility #19697

Merged

9 tasks

stdrc changed the title ~~change SourceCtrlOpts::rate_limit to split_txn: bool~~ refactor(source): change SourceCtrlOpts::rate_limit to split_txn: bool Dec 6, 2024

github-actions bot added type/refactor and removed Invalid PR Title labels Dec 6, 2024

stdrc changed the title ~~refactor(source): change SourceCtrlOpts::rate_limit to split_txn: bool~~ refactor(source): Dec 6, 2024

stdrc changed the title ~~refactor(source):~~ refactor(source): respect max chunk size in Dec 6, 2024

stdrc changed the title ~~refactor(source): respect max chunk size in~~ refactor(source): respect max chunk size in source connector parser Dec 6, 2024

stdrc force-pushed the rc/merge-hb-and-data-chunk-for-cdc branch from a8d2c05 to 185defc Compare December 6, 2024 08:28

stdrc force-pushed the rc/respect-chunk-size-limit branch from d3af95f to 46e5995 Compare December 6, 2024 08:29

Base automatically changed from rc/merge-hb-and-data-chunk-for-cdc to main December 6, 2024 17:35

stdrc force-pushed the rc/respect-chunk-size-limit branch 2 times, most recently from 5f97579 to c7cdfb5 Compare December 11, 2024 08:56

stdrc force-pushed the rc/respect-chunk-size-limit branch 2 times, most recently from 3f2f0a0 to 1d9e42e Compare December 17, 2024 06:58

stdrc marked this pull request as ready for review December 17, 2024 07:07

stdrc requested a review from a team as a code owner December 17, 2024 07:07

stdrc requested review from MrCroxx, StrikeW, tabVersion, xxchan, kwannoel, fuyufjh and xiangjinwu December 17, 2024 07:07

graphite-app bot requested a review from a team December 17, 2024 07:31

stdrc mentioned this pull request Dec 17, 2024

refactor(source): remove chunk splitting logic in apply_rate_limit #19826

Open

8 tasks

BugenZhao self-requested a review December 18, 2024 02:45

stdrc force-pushed the rc/respect-chunk-size-limit branch from 9fc46c4 to 10e0ea4 Compare December 18, 2024 02:48

stdrc added the ci/main-cron/run-all label Dec 18, 2024

stdrc force-pushed the rc/respect-chunk-size-limit branch from eeeebb4 to 486f99c Compare December 18, 2024 06:09

fuyufjh approved these changes Dec 18, 2024

View reviewed changes

stdrc added 14 commits December 18, 2024 17:32

change SourceCtrlOpts::rate_limit to split_txn: bool

49b335f

Signed-off-by: Richard Chien <[email protected]>

minor

a6cc2a5

Signed-off-by: Richard Chien <[email protected]>

store reference of SourceStreamChunkBuilder in SourceStreamChunkRowWr…

7a7f58c

…iter Signed-off-by: Richard Chien <[email protected]>

add RecordType::ops method to get Ops corresponding to the record…

0cbf552

… type Signed-off-by: Richard Chien <[email protected]>

simplify SourceStreamChunkRowWriter

44603ed

Signed-off-by: Richard Chien <[email protected]>

rewrite SourceStreamChunkBuilder

0ab470c

Signed-off-by: Richard Chien <[email protected]>

remove old methods of SourceStreamChunkBuilder

a4074e9

Signed-off-by: Richard Chien <[email protected]>

fix clippy

79849fa

Signed-off-by: Richard Chien <[email protected]>

fix typo

666d415

Signed-off-by: Richard Chien <[email protected]>

remove ensure_max_chunk_size

f001393

Signed-off-by: Richard Chien <[email protected]>

fix clippy

6f1b54b

Signed-off-by: Richard Chien <[email protected]>

fix ut

0d98fa9

Signed-off-by: Richard Chien <[email protected]>

fix build

b32abd6

Signed-off-by: Richard Chien <[email protected]>

fix check

b1a892c

Signed-off-by: Richard Chien <[email protected]>

BugenZhao approved these changes Dec 19, 2024

View reviewed changes

stdrc added this pull request to the merge queue Dec 19, 2024

stdrc removed this pull request from the merge queue due to a manual request Dec 19, 2024

stdrc force-pushed the rc/respect-chunk-size-limit branch from d986091 to b1a892c Compare December 19, 2024 05:34

stdrc mentioned this pull request Dec 19, 2024

refactor(source): rename some source parser related types and methods #19863

Merged

8 tasks

stdrc enabled auto-merge December 19, 2024 05:53

stdrc added this pull request to the merge queue Dec 19, 2024

auto-merge was automatically disabled December 19, 2024 07:08
Pull Request is not mergeable

Merged via the queue into main with commit f3d8e0d Dec 19, 2024
31 of 33 checks passed

stdrc deleted the rc/respect-chunk-size-limit branch December 19, 2024 07:13

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(source): respect max chunk size in source connector parser #19698

refactor(source): respect max chunk size in source connector parser #19698

stdrc commented Dec 6, 2024 •

edited

Loading

stdrc commented Dec 6, 2024 •

edited

Loading

tabVersion commented Dec 12, 2024

stdrc commented Dec 12, 2024

fuyufjh left a comment

fuyufjh Dec 18, 2024

stdrc Dec 18, 2024

fuyufjh Dec 18, 2024

stdrc Dec 18, 2024 •

edited

Loading

BugenZhao left a comment

BugenZhao Dec 19, 2024

stdrc Dec 19, 2024

StrikeW Dec 19, 2024

stdrc Dec 19, 2024

refactor(source): respect max chunk size in source connector parser #19698

refactor(source): respect max chunk size in source connector parser #19698

Conversation

stdrc commented Dec 6, 2024 • edited Loading

What's changed and what's your intention?

Checklist

Documentation

Release note

stdrc commented Dec 6, 2024 • edited Loading

tabVersion commented Dec 12, 2024

stdrc commented Dec 12, 2024

fuyufjh left a comment

Choose a reason for hiding this comment

fuyufjh Dec 18, 2024

Choose a reason for hiding this comment

stdrc Dec 18, 2024

Choose a reason for hiding this comment

fuyufjh Dec 18, 2024

Choose a reason for hiding this comment

stdrc Dec 18, 2024 • edited Loading

Choose a reason for hiding this comment

BugenZhao left a comment

Choose a reason for hiding this comment

BugenZhao Dec 19, 2024

Choose a reason for hiding this comment

stdrc Dec 19, 2024

Choose a reason for hiding this comment

StrikeW Dec 19, 2024

Choose a reason for hiding this comment

stdrc Dec 19, 2024

Choose a reason for hiding this comment

stdrc commented Dec 6, 2024 •

edited

Loading

stdrc commented Dec 6, 2024 •

edited

Loading

stdrc Dec 18, 2024 •

edited

Loading