Skip to content

sink: add column selector for storage sink#5595

Merged
ti-chi-bot[bot] merged 17 commits into
pingcap:masterfrom
wk989898:column-selector
Jul 25, 2026
Merged

sink: add column selector for storage sink#5595
ti-chi-bot[bot] merged 17 commits into
pingcap:masterfrom
wk989898:column-selector

Conversation

@wk989898

@wk989898 wk989898 commented Jul 7, 2026

Copy link
Copy Markdown
Collaborator

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.

It does three main things:

  • Adds column selector support to storage sink by carrying the selector on RowEvent before txn encoding.
  • Refactors canal/csv/kafka/storage encoding paths to share selector and callback handling instead of using sink-specific setters.
  • Adds validation so storage changefeeds cannot filter out all primary/unique key columns, preventing unsafe output for consumers.

About callback

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.

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

  • Unit test
  • Integration test

Questions

Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?

Release note

Support column selector for storage sink

Summary by CodeRabbit

Summary

  • New Features
    • Added end-to-end column-selector support for storage and MQ pipelines (cloud storage, Kafka, Pulsar, and CSV/JSON), including per-table filtering that affects both encoded output and CSV headers/values.
    • Storage consumer now applies column selectors when decoding CSV DML files.
  • Bug Fixes
    • Improved DML task callback flow to ensure post-flush and post-enqueue hooks run reliably and in the expected order.
    • Updated encoders to consistently use row-level event data for txn message creation.
  • Tests
    • Expanded unit and integration coverage for column selection, CSV decoding/encoding, and callback lifecycle behavior.

wk989898 added 2 commits July 7, 2026 03:45
Signed-off-by: wk989898 <nhsmwk@gmail.com>
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@ti-chi-bot

ti-chi-bot Bot commented Jul 7, 2026

Copy link
Copy Markdown

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

@ti-chi-bot ti-chi-bot Bot added do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. release-note Denotes a PR that will be considered when it comes time to generate release notes. labels Jul 7, 2026
@coderabbitai

coderabbitai Bot commented Jul 7, 2026

Copy link
Copy Markdown
Contributor

Review Change Stack

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

  • @coderabbitai resume to resume automatic reviews.
  • @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

  • ▶️ Resume reviews
  • 🔍 Trigger review
📝 Walkthrough

Walkthrough

This PR adds column-selector support to storage and messaging sinks, refactors DML processing around pre-built row events and detached callbacks, updates JSON/CSV encoders and decoders, and expands unit and integration coverage.

Changes

Column selector and row-event pipeline

Layer / File(s) Summary
Selector and callback contracts
downstreamadapter/sink/columnselector/..., downstreamadapter/sink/helper/..., pkg/common/event/dml_event.go
Adds table-info selector lookup, shared row-event builders, detached callbacks, and the renamed callback factory.
Selector validation and sink wiring
api/v2/changefeed.go, downstreamadapter/sink/cloudstorage/..., downstreamadapter/sink/kafka/..., downstreamadapter/sink/pulsar/..., downstreamadapter/sink/redo/..., cmd/storage-consumer/...
Validates selectors, passes table-specific selectors through sinks and storage consumers, and centralizes MQ row-event construction.
Row-event encoding and filtering
pkg/sink/codec/common/..., pkg/sink/codec/canal/..., pkg/sink/codec/csv/...
Changes encoders to accept row-event slices and applies selector-aware JSON and CSV encoding and decoding.
Cloud storage task lifecycle
downstreamadapter/sink/cloudstorage/...
Stores precomputed row events and detached callbacks, then propagates callbacks through encoding and buffering.
Integration coverage
tests/integration_tests/...
Adds storage selector fixtures and execution paths, updates selector integration setup, and enables the tests in CI.

Estimated code review effort: 4 (Complex) | ~60 minutes

Sequence Diagram(s)

sequenceDiagram
  participant Changefeed
  participant CloudStorageSink
  participant DMLTask
  participant Encoder
  participant Spool
  Changefeed->>CloudStorageSink: validate column selectors
  CloudStorageSink->>DMLTask: create task with table selector
  DMLTask->>Encoder: append pre-built row events
  Encoder->>Spool: enqueue encoded messages
  Spool->>DMLTask: invoke post-enqueue callback
Loading

Possibly related PRs

Suggested labels: ok-to-test, contribution

Suggested reviewers: asddongmen, tenfyzhong

Poem

A rabbit selects each field with care,
Carries callbacks through the air.
Rows hop neatly, encoded bright,
Storage keeps the chosen right.
Clean columns appear! 🐇

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name Status Explanation Resolution
Docstring Coverage ⚠️ Warning Docstring coverage is 17.65% which is insufficient. The required threshold is 80.00%. Write docstrings for the functions missing them to satisfy the coverage threshold.
✅ Passed checks (4 passed)
Check name Status Explanation
Title check ✅ Passed The title is concise and accurately describes the main change: adding column selector support for the storage sink.
Description check ✅ Passed The description includes the issue number, change summary, tests, questions, and release note, with only minor template details omitted.
Linked Issues check ✅ Passed The changes implement the requested storage-sink column selector support from issue #5592.
Out of Scope Changes check ✅ Passed The changed files are all related to column-selector support and its required test coverage; no unrelated code stands out.
✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create PR with unit tests

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.

❤️ Share

Comment @coderabbitai help to get the list of available commands.

@ti-chi-bot ti-chi-bot Bot added the size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. label Jul 7, 2026

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment thread pkg/sink/codec/csv/csv_encoder.go Outdated
Comment thread pkg/sink/codec/canal/canal_json_txn_encoder.go Outdated
Comment thread downstreamadapter/sink/cloudstorage/task.go
@wk989898

wk989898 commented Jul 7, 2026

Copy link
Copy Markdown
Collaborator Author

/test storage

wk989898 added 3 commits July 7, 2026 07:41
Signed-off-by: wk989898 <nhsmwk@gmail.com>
.
Signed-off-by: wk989898 <nhsmwk@gmail.com>
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@wk989898

wk989898 commented Jul 7, 2026

Copy link
Copy Markdown
Collaborator Author

/test storage

Signed-off-by: wk989898 <nhsmwk@gmail.com>
@wk989898
wk989898 marked this pull request as ready for review July 7, 2026 09:27
@ti-chi-bot ti-chi-bot Bot removed the do-not-merge/work-in-progress Indicates that a PR should not merge because it is a work in progress. label Jul 7, 2026
@wk989898

wk989898 commented Jul 7, 2026

Copy link
Copy Markdown
Collaborator Author

/test all

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 1

🧹 Nitpick comments (2)
downstreamadapter/sink/helper/row_callback_test.go (1)

24-24: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value

Test name still references old NewTxnPostFlushRowCallback.

TestTxnPostFlushRowCallback wasn't renamed alongside the NewPostFlushRowCallback rename 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 value

Solid refactor; consider adding GoDoc comments.

NewMQRowEvents/NewRowEvents are 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

📥 Commits

Reviewing files that changed from the base of the PR and between 2258270 and 5102f18.

📒 Files selected for processing (28)
  • downstreamadapter/sink/cloudstorage/buffer_manager.go
  • downstreamadapter/sink/cloudstorage/buffer_manager_test.go
  • downstreamadapter/sink/cloudstorage/dml_writers.go
  • downstreamadapter/sink/cloudstorage/encoder_group.go
  • downstreamadapter/sink/cloudstorage/encoder_group_test.go
  • downstreamadapter/sink/cloudstorage/sink.go
  • downstreamadapter/sink/cloudstorage/sink_test.go
  • downstreamadapter/sink/cloudstorage/task.go
  • downstreamadapter/sink/cloudstorage/task_test.go
  • downstreamadapter/sink/cloudstorage/writer_test.go
  • downstreamadapter/sink/columnselector/column_selector.go
  • downstreamadapter/sink/helper/mq_row_event.go
  • downstreamadapter/sink/helper/row_callback.go
  • downstreamadapter/sink/helper/row_callback_test.go
  • downstreamadapter/sink/kafka/sink.go
  • downstreamadapter/sink/pulsar/sink.go
  • downstreamadapter/sink/redo/sink.go
  • pkg/sink/codec/canal/canal_json_txn_encoder.go
  • pkg/sink/codec/canal/canal_json_txn_encoder_test.go
  • pkg/sink/codec/common/encoder.go
  • pkg/sink/codec/csv/csv_encoder.go
  • pkg/sink/codec/csv/csv_encoder_test.go
  • pkg/sink/codec/csv/csv_message.go
  • tests/integration_tests/column_selector/conf/changefeed.toml
  • tests/integration_tests/column_selector/data/data.sql
  • tests/integration_tests/column_selector/run.sh
  • tests/integration_tests/kafka_column_selector/data/data.sql
  • tests/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

Comment on lines +12 to +20
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
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 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.

Suggested change
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>
@wk989898

wk989898 commented Jul 9, 2026

Copy link
Copy Markdown
Collaborator Author

/retest

@asddongmen asddongmen left a comment

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two comments.

Comment thread downstreamadapter/sink/cloudstorage/task.go Outdated
Comment thread downstreamadapter/sink/cloudstorage/sink.go
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@ti-chi-bot ti-chi-bot Bot added needs-1-more-lgtm Indicates a PR needs 1 more LGTM. approved labels Jul 10, 2026
@wk989898

Copy link
Copy Markdown
Collaborator Author

/test all

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧹 Nitpick comments (1)
downstreamadapter/sink/columnselector/column_selector_test.go (1)

189-230: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Consider adding a test for the PK-not-retained → unique-key-retained fallthrough case.

The verifyIndices change (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

📥 Commits

Reviewing files that changed from the base of the PR and between 5102f18 and b547286.

📒 Files selected for processing (7)
  • api/v2/changefeed.go
  • api/v2/changefeed_test.go
  • downstreamadapter/sink/cloudstorage/task.go
  • downstreamadapter/sink/columnselector/column_selector.go
  • downstreamadapter/sink/columnselector/column_selector_test.go
  • pkg/common/event/dml_event.go
  • pkg/common/event/dml_event_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
  • downstreamadapter/sink/cloudstorage/task.go

wk989898 added 2 commits July 21, 2026 08:50
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@wk989898

Copy link
Copy Markdown
Collaborator Author

/test all

wk989898 added 3 commits July 22, 2026 03:45
Signed-off-by: wk989898 <nhsmwk@gmail.com>
Signed-off-by: wk989898 <nhsmwk@gmail.com>
.
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@ti-chi-bot

ti-chi-bot Bot commented Jul 22, 2026

Copy link
Copy Markdown

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: 3AceShowHand, asddongmen

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details Needs approval from an approver in each of these files:
  • OWNERS [3AceShowHand,asddongmen]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

@ti-chi-bot ti-chi-bot Bot added lgtm and removed needs-1-more-lgtm Indicates a PR needs 1 more LGTM. labels Jul 22, 2026
@ti-chi-bot

ti-chi-bot Bot commented Jul 22, 2026

Copy link
Copy Markdown

[LGTM Timeline notifier]

Timeline:

  • 2026-07-10 07:20:41.162445962 +0000 UTC m=+352627.198541008: ☑️ agreed by asddongmen.
  • 2026-07-22 07:26:18.856753128 +0000 UTC m=+1389764.892848254: ☑️ agreed by 3AceShowHand.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 1

🧹 Nitpick comments (1)
downstreamadapter/sink/helper/mq_row_event.go (1)

35-46: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win

Make event.Rewind() unconditional.

If partition generation fails at Line 44, the function returns while the DMLEvent cursor remains advanced. A retry or subsequent encoder using the same event can then skip already-consumed rows. Register defer event.Rewind() on entry and remove the success-only rewind; add a focused error-path test.

Proposed fix
 ) ([]*commonEvent.MQRowEvent, error) {
+	defer event.Rewind()
 	callback := NewPostFlushRowCallback(event, uint64(event.Len()))
...
 		row, ok := event.GetNextRow()
 		if !ok {
-			event.Rewind()
 			break
 		}
🤖 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 35 - 46, Make
event rewinding unconditional in the function containing the row-processing
loop: register defer event.Rewind() on entry so it runs on successful completion
and when GeneratePartitionIndexAndKey returns an error, then remove the loop’s
success-only event.Rewind() call. Add a focused test covering
partition-generation failure and verifying the DMLEvent cursor is rewound.
🤖 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/csv_storage_column_selector/run.sh`:
- Line 38: Update the cleanup command in the test script around WORK_DIR to
safely handle spaces and glob characters: quote the WORK_DIR expansion for both
recursive removal and directory creation, and guard against an empty or unset
WORK_DIR before running cleanup. Preserve the existing workspace setup behavior
for valid paths.

---

Nitpick comments:
In `@downstreamadapter/sink/helper/mq_row_event.go`:
- Around line 35-46: Make event rewinding unconditional in the function
containing the row-processing loop: register defer event.Rewind() on entry so it
runs on successful completion and when GeneratePartitionIndexAndKey returns an
error, then remove the loop’s success-only event.Rewind() call. Add a focused
test covering partition-generation failure and verifying the DMLEvent cursor is
rewound.
🪄 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 Plus

Run ID: 87fcd489-2122-4ccf-9a9c-b48930f04d5e

📥 Commits

Reviewing files that changed from the base of the PR and between 6128efa and 55d3e71.

📒 Files selected for processing (13)
  • cmd/storage-consumer/consumer.go
  • downstreamadapter/sink/cloudstorage/encoder_group_test.go
  • downstreamadapter/sink/cloudstorage/sink.go
  • downstreamadapter/sink/helper/mq_row_event.go
  • downstreamadapter/sink/kafka/sink.go
  • pkg/sink/codec/canal/canal_json_txn_encoder.go
  • pkg/sink/codec/csv/csv_decoder.go
  • pkg/sink/codec/csv/csv_decoder_test.go
  • pkg/sink/codec/csv/csv_encoder_test.go
  • tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml
  • tests/integration_tests/csv_storage_column_selector/data/data.sql
  • tests/integration_tests/csv_storage_column_selector/run.sh
  • tests/integration_tests/run_light_it_in_ci.sh
💤 Files with no reviewable changes (1)
  • pkg/sink/codec/canal/canal_json_txn_encoder.go
🚧 Files skipped from review as they are similar to previous changes (4)
  • downstreamadapter/sink/cloudstorage/encoder_group_test.go
  • downstreamadapter/sink/kafka/sink.go
  • downstreamadapter/sink/cloudstorage/sink.go
  • pkg/sink/codec/csv/csv_encoder_test.go

return
fi

rm -rf $WORK_DIR && mkdir -p $WORK_DIR

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win

Quote and guard the recursive cleanup path.

Line 38 can split or glob-expand WORK_DIR, potentially deleting paths outside the intended test workspace.

Proposed fix
-	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	rm -rf -- "${WORK_DIR:?WORK_DIR must be set}" && mkdir -p -- "$WORK_DIR"
📝 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.

Suggested change
rm -rf $WORK_DIR && mkdir -p $WORK_DIR
rm -rf -- "${WORK_DIR:?WORK_DIR must be set}" && mkdir -p -- "$WORK_DIR"
🧰 Tools
🪛 Shellcheck (0.11.0)

[info] 38-38: Double quote to prevent globbing and word splitting.

(SC2086)


[info] 38-38: 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/csv_storage_column_selector/run.sh` at line 38,
Update the cleanup command in the test script around WORK_DIR to safely handle
spaces and glob characters: quote the WORK_DIR expansion for both recursive
removal and directory creation, and guard against an empty or unset WORK_DIR
before running cleanup. Preserve the existing workspace setup behavior for valid
paths.

Source: Linters/SAST tools

wk989898 added 2 commits July 24, 2026 09:30
Signed-off-by: wk989898 <nhsmwk@gmail.com>
@wk989898

Copy link
Copy Markdown
Collaborator Author

/retest

Signed-off-by: wk989898 <nhsmwk@gmail.com>
@wk989898

Copy link
Copy Markdown
Collaborator Author

/retest

@ti-chi-bot
ti-chi-bot Bot merged commit 07e9447 into pingcap:master Jul 25, 2026
25 checks passed
@wk989898 wk989898 added the needs-cherry-pick-release-8.5 Should cherry pick this PR to release-8.5 branch. label Jul 25, 2026
@wk989898

Copy link
Copy Markdown
Collaborator Author

/cherry-pick release-nextgen-202603

@ti-chi-bot

Copy link
Copy Markdown
Member

In response to a cherrypick label: new pull request created to branch release-8.5: #5742.
But this PR has conflicts, please resolve them!

@ti-chi-bot

Copy link
Copy Markdown
Member

@wk989898: new pull request created to branch release-nextgen-202603: #5743.
But this PR has conflicts, please resolve them!

Details

In response to this:

/cherry-pick release-nextgen-202603

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the ti-community-infra/tichi repository.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

approved lgtm needs-cherry-pick-release-8.5 Should cherry pick this PR to release-8.5 branch. release-note Denotes a PR that will be considered when it comes time to generate release notes. size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

support column selector in storage sink

4 participants