sink: add column selector for storage sink (#5595)

[ti-chi-bot]

This is an automated cherry-pick of #5595

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.

[ti-chi-bot]

@wk989898 This PR has conflicts, I have hold it.
Please resolve them or ask others to resolve them, then comment /unhold to remove the hold label.

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign charlescheung96 for approval. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

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

Details

Needs approval from an approver in each of these files:

• OWNERS

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: ## If you want to know how to resolve it, please read the guide in TiDB Dev Guide.

Details

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.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 0e827623-77d7-4dbc-956a-a5ad21856df3

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ 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

• X
• Mastodon
• Reddit
• LinkedIn

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