sink: add column selector for storage sink#5595
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
📝 WalkthroughWalkthroughThis PR adds column-selector support to cloud storage and messaging sinks, refactors DML processing around pre-built row events and detached callbacks, updates canal/CSV encoders, and expands unit and integration coverage. ChangesColumn selector and row-event contracts
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces column selector support to the cloud storage sink, enabling column filtering for CSV and Canal-JSON encoders, and refactors MQ row event generation. The review feedback highlights two critical layering violations where utility packages in pkg/ depend on downstreamadapter/, which should be resolved by allowing the column selector to be nil by default. Additionally, it is recommended to replace the variadic parameter in newDMLTask with a single optional selector to avoid Go API design anti-patterns.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
/test storage |
|
/test storage |
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
|
/test all |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
downstreamadapter/sink/helper/row_callback_test.go (1)
24-24: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTest name still references old
NewTxnPostFlushRowCallback.
TestTxnPostFlushRowCallbackwasn't renamed alongside theNewPostFlushRowCallbackrename at Line 32, leaving a stale name.✏️ Suggested rename
-func TestTxnPostFlushRowCallback(t *testing.T) { +func TestPostFlushRowCallback(t *testing.T) {Also applies to: 32-32
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/sink/helper/row_callback_test.go` at line 24, The test name is stale after the callback rename, so update the test identifier in TestTxnPostFlushRowCallback to match the new NewPostFlushRowCallback naming used in the helper code. Make sure any related references in row_callback_test.go that still mention the old Txn/PostFlush name are renamed consistently so the test suite and symbols align with the current API.downstreamadapter/sink/helper/mq_row_event.go (1)
52-81: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSolid refactor; consider adding GoDoc comments.
NewMQRowEvents/NewRowEventsare exported but lack doc comments describing behavior (rewind semantics, default selector fallback). Minor polish only.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/sink/helper/mq_row_event.go` around lines 52 - 81, The exported constructors NewRowEvents and NewMQRowEvents should have GoDoc comments. Add brief doc comments above these functions in mq_row_event.go that describe their behavior, including the default selector fallback when selector is nil and that NewRowEvents rewinds the DMLEvent after iterating through rows. Keep the comments concise and directly tied to the function names so they satisfy exported symbol documentation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/integration_tests/column_selector/run.sh`:
- Around line 12-20: Quote the variable expansions in prepare_cluster so shell
word-splitting and pathname expansion cannot alter the commands; this is
especially important for the rm -rf target. Update the calls in prepare_cluster
that use WORK_DIR, UP_PD_HOST_1, UP_PD_PORT_1, and CDC_BINARY so they are safely
passed as single arguments. Keep the fix localized to prepare_cluster in run.sh
and preserve the existing behavior.
---
Nitpick comments:
In `@downstreamadapter/sink/helper/mq_row_event.go`:
- Around line 52-81: The exported constructors NewRowEvents and NewMQRowEvents
should have GoDoc comments. Add brief doc comments above these functions in
mq_row_event.go that describe their behavior, including the default selector
fallback when selector is nil and that NewRowEvents rewinds the DMLEvent after
iterating through rows. Keep the comments concise and directly tied to the
function names so they satisfy exported symbol documentation.
In `@downstreamadapter/sink/helper/row_callback_test.go`:
- Line 24: The test name is stale after the callback rename, so update the test
identifier in TestTxnPostFlushRowCallback to match the new
NewPostFlushRowCallback naming used in the helper code. Make sure any related
references in row_callback_test.go that still mention the old Txn/PostFlush name
are renamed consistently so the test suite and symbols align with the current
API.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 1c2c26d7-fc5f-4a8e-b277-c8c59eb1925d
📒 Files selected for processing (28)
downstreamadapter/sink/cloudstorage/buffer_manager.godownstreamadapter/sink/cloudstorage/buffer_manager_test.godownstreamadapter/sink/cloudstorage/dml_writers.godownstreamadapter/sink/cloudstorage/encoder_group.godownstreamadapter/sink/cloudstorage/encoder_group_test.godownstreamadapter/sink/cloudstorage/sink.godownstreamadapter/sink/cloudstorage/sink_test.godownstreamadapter/sink/cloudstorage/task.godownstreamadapter/sink/cloudstorage/task_test.godownstreamadapter/sink/cloudstorage/writer_test.godownstreamadapter/sink/columnselector/column_selector.godownstreamadapter/sink/helper/mq_row_event.godownstreamadapter/sink/helper/row_callback.godownstreamadapter/sink/helper/row_callback_test.godownstreamadapter/sink/kafka/sink.godownstreamadapter/sink/pulsar/sink.godownstreamadapter/sink/redo/sink.gopkg/sink/codec/canal/canal_json_txn_encoder.gopkg/sink/codec/canal/canal_json_txn_encoder_test.gopkg/sink/codec/common/encoder.gopkg/sink/codec/csv/csv_encoder.gopkg/sink/codec/csv/csv_encoder_test.gopkg/sink/codec/csv/csv_message.gotests/integration_tests/column_selector/conf/changefeed.tomltests/integration_tests/column_selector/data/data.sqltests/integration_tests/column_selector/run.shtests/integration_tests/kafka_column_selector/data/data.sqltests/integration_tests/run_heavy_it_in_ci.sh
💤 Files with no reviewable changes (2)
- tests/integration_tests/kafka_column_selector/data/data.sql
- downstreamadapter/sink/cloudstorage/task_test.go
| function prepare_cluster() { | ||
| rm -rf $WORK_DIR && mkdir -p $WORK_DIR | ||
| start_tidb_cluster --workdir $WORK_DIR | ||
|
|
||
| # record tso before we create tables to skip the system table DDLs | ||
| # Record TSO before creating test tables to skip system table DDLs. | ||
| start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1) | ||
|
|
||
| run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Quote variable expansions in prepare_cluster, especially the rm -rf target.
rm -rf $WORK_DIR and the other $WORK_DIR/$UP_PD_HOST_1/$CDC_BINARY usages here are unquoted. Static analysis flags the rm -rf case specifically as risky if the variable is ever empty or contains glob/whitespace characters (word-splitting/pathname-expansion could target unintended paths).
🔒 Proposed fix
function prepare_cluster() {
- rm -rf $WORK_DIR && mkdir -p $WORK_DIR
- start_tidb_cluster --workdir $WORK_DIR
+ rm -rf -- "${WORK_DIR:?WORK_DIR is unset}" && mkdir -p "$WORK_DIR"
+ start_tidb_cluster --workdir "$WORK_DIR"
# Record TSO before creating test tables to skip system table DDLs.
- start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1)
+ start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1")
- run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+ run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY"
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| function prepare_cluster() { | |
| rm -rf $WORK_DIR && mkdir -p $WORK_DIR | |
| start_tidb_cluster --workdir $WORK_DIR | |
| # record tso before we create tables to skip the system table DDLs | |
| # Record TSO before creating test tables to skip system table DDLs. | |
| start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1) | |
| run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY | |
| } | |
| function prepare_cluster() { | |
| rm -rf -- "${WORK_DIR:?WORK_DIR is unset}" && mkdir -p "$WORK_DIR" | |
| start_tidb_cluster --workdir "$WORK_DIR" | |
| # Record TSO before creating test tables to skip system table DDLs. | |
| start_ts=$(run_cdc_cli_tso_query "$UP_PD_HOST_1" "$UP_PD_PORT_1") | |
| run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY" | |
| } |
🧰 Tools
🪛 ast-grep (0.44.1)
[error] 12-12: Recursive rm is given an unquoted variable expansion as a target. If the variable is empty, contains whitespace, or holds glob characters, word splitting and pathname expansion can make rm delete unintended paths (e.g. rm -rf $dir with dir="" or dir="/ tmp"). Always double-quote the expansion and guard against empty values: rm -rf -- "${dir:?dir is unset}".
Context: $WORK_DIR
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(rm-rf-unquoted-variable-bash)
🪛 Shellcheck (0.11.0)
[info] 13-13: Double quote to prevent globbing and word splitting.
(SC2086)
[info] 13-13: Double quote to prevent globbing and word splitting.
(SC2086)
[info] 14-14: Double quote to prevent globbing and word splitting.
(SC2086)
[info] 17-17: Double quote to prevent globbing and word splitting.
(SC2086)
[info] 17-17: Double quote to prevent globbing and word splitting.
(SC2086)
[info] 19-19: Double quote to prevent globbing and word splitting.
(SC2086)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/integration_tests/column_selector/run.sh` around lines 12 - 20, Quote
the variable expansions in prepare_cluster so shell word-splitting and pathname
expansion cannot alter the commands; this is especially important for the rm -rf
target. Update the calls in prepare_cluster that use WORK_DIR, UP_PD_HOST_1,
UP_PD_PORT_1, and CDC_BINARY so they are safely passed as single arguments. Keep
the fix localized to prepare_cluster in run.sh and preserve the existing
behavior.
Source: Linters/SAST tools
Signed-off-by: wk989898 <nhsmwk@gmail.com>
|
/retest |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: asddongmen The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
[LGTM Timeline notifier]Timeline:
|
|
/test all |
There was a problem hiding this comment.
🧹 Nitpick comments (1)
downstreamadapter/sink/columnselector/column_selector_test.go (1)
189-230: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a test for the PK-not-retained → unique-key-retained fallthrough case.
The
verifyIndiceschange (lines 165-176) introduces a key behavioral path: when a table has primary key columns but they are not fully retained by the selector, verification falls through to unique-key checks. If a unique key is fully retained, verification succeeds. This path is not currently tested — existing tests cover PK-retained (success), no-PK + UK-retained (success), and failure cases, but not the combination of PK-exists-not-retained + UK-retained.🧪 Suggested test
func TestVerifyTablesFallsThroughToUniqueKeyWhenPKNotRetained(t *testing.T) { t.Parallel() replicaConfig := config.GetDefaultReplicaConfig() replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{ { Matcher: []string{"test.t"}, Columns: []string{"a", "b"}, // UK columns retained, PK column "id" filtered out }, } selectors, err := New(replicaConfig.Sink) require.NoError(t, err) idFieldType := types.NewFieldType(mysql.TypeLong) idFieldType.AddFlag(mysql.PriKeyFlag | mysql.NotNullFlag) tableInfo := commonType.WrapTableInfo("test", &model.TableInfo{ Name: ast.NewCIStr("t"), Columns: []*model.ColumnInfo{ { ID: 1, Name: ast.NewCIStr("id"), FieldType: *idFieldType, State: model.StatePublic, }, newColumnInfoForSelectorTest(2, "a", mysql.NotNullFlag), newColumnInfoForSelectorTest(3, "b", mysql.NotNullFlag), }, Indices: []*model.IndexInfo{ { Name: ast.NewCIStr("uk_ab"), Columns: []*model.IndexColumn{ {Name: ast.NewCIStr("a"), Offset: 0}, {Name: ast.NewCIStr("b"), Offset: 1}, }, Unique: true, State: model.StatePublic, }, }, }) // PK "id" is filtered out, but unique key (a, b) is fully retained → should pass. require.NoError(t, selectors.VerifyTables([]*commonType.TableInfo{tableInfo}, nil)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@downstreamadapter/sink/columnselector/column_selector_test.go` around lines 189 - 230, Add a test alongside TestVerifyTablesRequiresFullUniqueKey covering a table with primary key column “id” excluded by the selector while unique-key columns “a” and “b” are retained. Build the table metadata with the primary key and unique index, then assert selectors.VerifyTables succeeds, using a test name such as TestVerifyTablesFallsThroughToUniqueKeyWhenPKNotRetained.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@downstreamadapter/sink/columnselector/column_selector_test.go`:
- Around line 189-230: Add a test alongside
TestVerifyTablesRequiresFullUniqueKey covering a table with primary key column
“id” excluded by the selector while unique-key columns “a” and “b” are retained.
Build the table metadata with the primary key and unique index, then assert
selectors.VerifyTables succeeds, using a test name such as
TestVerifyTablesFallsThroughToUniqueKeyWhenPKNotRetained.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: fc4060f4-4666-4955-a526-cfde7f0db6c5
📒 Files selected for processing (7)
api/v2/changefeed.goapi/v2/changefeed_test.godownstreamadapter/sink/cloudstorage/task.godownstreamadapter/sink/columnselector/column_selector.godownstreamadapter/sink/columnselector/column_selector_test.gopkg/common/event/dml_event.gopkg/common/event/dml_event_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- downstreamadapter/sink/cloudstorage/task.go
2 participants
What problem does this PR solve?
Issue Number: close #5592
What is changed and how it works?
The column selector logic was generalized from a Kafka-only path into a shared
column_selector flow that storage sink can also use.
Storage keeps PostEnqueue and PostFlush as separate lifecycle events. PostEnqueue remains
attached to the storage task and is triggered by the buffer/spool enqueue path. PostFlush
is attached to the encoded message callback and runs after the encoded storage message is
flushed.
About callback
For storage, the callback is intentionally different from Kafka.
Kafka produces row-level messages. It uses a row callback counter: each row callback
increments the count, and PostFlush runs only after all rows in the transaction have been
flushed.
Storage transaction encoders build a batch/transaction message. The encoder attaches only
the last row callback to the built batch message, so the callback is triggered once per
encoded storage message, not once per row. If storage used kafka's row-counting callback
for a multi-row transaction, the counter would never reach the row count and PostFlush
could hang forever.
That is why storage directly passes event.PostFlush as the row callback. The encoded
storage batch message calls it once when the message is flushed, which matches storage's
actual callback granularity.
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note
Summary by CodeRabbit
Summary by CodeRabbit