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

api/v2/changefeed.go

@@ -1680,7 +1680,7 @@ func getVerifiedTables(
 		return nil, nil, nil, err
 	}
 
-	if err := verifyTable4MQ(replicaConfig, scheme, topic, protocol, tableInfos); err != nil {
+	if err := verifyTablesForSink(replicaConfig, scheme, topic, protocol, tableInfos); err != nil {
 		return nil, nil, nil, err
 	}
 
@@ -1691,13 +1691,21 @@ func getVerifiedTables(
 	return ineligibleTables, eligibleTables, allTables, nil
 }
 
-func verifyTable4MQ(
+func verifyTablesForSink(
 	replicaConfig *config.ReplicaConfig,
 	scheme string,
 	topic string,
 	protocol config.Protocol,
 	tableInfos []*common.TableInfo,
 ) error {
+	if config.IsStorageScheme(scheme) {
+		selectors, err := columnselector.New(replicaConfig.Sink)
+		if err != nil {
+			return err
+		}
+		return selectors.VerifyTables(tableInfos, nil)
+	}
+
 	if !config.IsMQScheme(scheme) {
 		return nil
 	}

api/v2/changefeed_test.go

@@ -0,0 +1,310 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/pingcap/kvproto/pkg/keyspacepb"
+	"github.com/pingcap/ticdc/maintainer"
+	"github.com/pingcap/ticdc/pkg/api"
+	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/errors"
+	"github.com/pingcap/ticdc/pkg/etcd"
+	"github.com/pingcap/ticdc/pkg/liveness"
+	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/pingcap/ticdc/pkg/server"
+	"github.com/pingcap/ticdc/pkg/util"
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/types"
+	"github.com/stretchr/testify/require"
+	pd "github.com/tikv/pd/client"
+)
+
+// TestValidateResumeChangefeedState covers the API-side guard that runs before
+// resume GC safepoint/barrier setup. Running states must fail fast, while states
+// that are actually stopped can proceed to the remaining resume validation.
+func TestValidateResumeChangefeedState(t *testing.T) {
+	for _, state := range []config.FeedState{config.StateStopped, config.StateFailed, config.StateFinished} {
+		require.NoError(t, validateResumeChangefeedState(state))
+	}
+
+	for _, state := range []config.FeedState{config.StateNormal, config.StateWarning, config.StatePending} {
+		err := validateResumeChangefeedState(state)
+		require.True(t, errors.ErrChangefeedUpdateRefused.Equal(err))
+		require.Contains(t, err.Error(), string(state))
+	}
+}
+
+// TestResumeChangefeedRejectsNormalBeforeGC covers the HTTP resume regression:
+// a normal changefeed must fail before the handler requests PD/etcd clients for
+// GC safepoint/barrier setup or calls the coordinator resume path.
+func TestResumeChangefeedRejectsNormalBeforeGC(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	co := &resumeNormalCoordinator{}
+	srv := &resumeNormalServer{coordinator: co}
+	h := &OpenAPIV2{server: srv}
+
+	w := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(w)
+	c.Request = httptest.NewRequest(http.MethodPost, "/api/v2/changefeeds/test/resume?keyspace=default", nil)
+	c.Params = gin.Params{{Key: api.APIOpVarChangefeedID, Value: "test"}}
+	c.Set("ctx-keyspace", &keyspacepb.KeyspaceMeta{
+		Id:    common.DefaultKeyspaceID,
+		State: keyspacepb.KeyspaceState_ENABLED,
+	})
+
+	h.ResumeChangefeed(c)
+
+	require.Len(t, c.Errors, 1)
+	require.True(t, errors.ErrChangefeedUpdateRefused.Equal(c.Errors.Last().Err))
+	require.False(t, srv.pdClientRequested)
+	require.False(t, srv.etcdClientRequested)
+	require.False(t, co.resumeCalled)
+}
+
+type resumeNormalServer struct {
+	coordinator         server.Coordinator
+	pdClientRequested   bool
+	etcdClientRequested bool
+}
+
+func (s *resumeNormalServer) Run(ctx context.Context) error { return nil }
+
+func (s *resumeNormalServer) Close() {}
+
+func (s *resumeNormalServer) SelfInfo() (*node.Info, error) { return nil, nil }
+
+func (s *resumeNormalServer) Liveness() liveness.Liveness { return liveness.CaptureAlive }
+
+func (s *resumeNormalServer) GetCoordinator() (server.Coordinator, error) {
+	return s.coordinator, nil
+}
+
+func (s *resumeNormalServer) IsCoordinator() bool { return true }
+
+func (s *resumeNormalServer) GetCoordinatorInfo(ctx context.Context) (*node.Info, error) {
+	return nil, nil
+}
+
+func (s *resumeNormalServer) GetPdClient() pd.Client {
+	s.pdClientRequested = true
+	return nil
+}
+
+func (s *resumeNormalServer) GetEtcdClient() etcd.CDCEtcdClient {
+	s.etcdClientRequested = true
+	return nil
+}
+
+func (s *resumeNormalServer) GetMaintainerManager() *maintainer.Manager { return nil }
+
+type resumeNormalCoordinator struct {
+	resumeCalled bool
+}
+
+func (c *resumeNormalCoordinator) Stop() {}
+
+func (c *resumeNormalCoordinator) Run(ctx context.Context) error { return nil }
+
+func (c *resumeNormalCoordinator) ListChangefeeds(ctx context.Context, keyspace string) ([]*config.ChangeFeedInfo, []*config.ChangeFeedStatus, error) {
+	return nil, nil, nil
+}
+
+func (c *resumeNormalCoordinator) GetChangefeed(ctx context.Context, changefeedDisplayName common.ChangeFeedDisplayName) (*config.ChangeFeedInfo, *config.ChangeFeedStatus, error) {
+	changefeedID := common.NewChangeFeedIDWithName(changefeedDisplayName.Name, changefeedDisplayName.Keyspace)
+	return &config.ChangeFeedInfo{
+			ChangefeedID: changefeedID,
+			State:        config.StateNormal,
+		}, &config.ChangeFeedStatus{
+			CheckpointTs: 123,
+		}, nil
+}
+
+func (c *resumeNormalCoordinator) GetPersistedChangefeedInfo(ctx context.Context, id common.ChangeFeedID) (*config.ChangeFeedInfo, error) {
+	return &config.ChangeFeedInfo{
+		ChangefeedID: id,
+		State:        config.StateNormal,
+	}, nil
+}
+
+func (c *resumeNormalCoordinator) CreateChangefeed(ctx context.Context, info *config.ChangeFeedInfo) error {
+	return nil
+}
+
+func (c *resumeNormalCoordinator) RemoveChangefeed(ctx context.Context, id common.ChangeFeedID) (uint64, error) {
+	return 0, nil
+}
+
+func (c *resumeNormalCoordinator) PauseChangefeed(ctx context.Context, id common.ChangeFeedID) error {
+	return nil
+}
+
+func (c *resumeNormalCoordinator) ResumeChangefeed(ctx context.Context, id common.ChangeFeedID, newCheckpointTs uint64, overwriteCheckpointTs bool) error {
+	c.resumeCalled = true
+	return nil
+}
+
+func (c *resumeNormalCoordinator) UpdateChangefeed(ctx context.Context, change *config.ChangeFeedInfo) error {
+	return nil
+}
+
+func (c *resumeNormalCoordinator) RequestResolvedTsFromLogCoordinator(ctx context.Context, changefeedDisplayName common.ChangeFeedDisplayName) {
+}
+
+func (c *resumeNormalCoordinator) DrainNode(ctx context.Context, target node.ID) (int, error) {
+	return 0, nil
+}
+
+func (c *resumeNormalCoordinator) Initialized() bool { return true }
+
+// TestMaskSinkURIForError verifies that error messages mask sensitive sink URI
+// fields. It checks both a valid URI with secret query parameters and an invalid
+// URI parse error that previously exposed raw credentials.
+func TestMaskSinkURIForError(t *testing.T) {
+	sinkURI := "kafka://127.0.0.1:9092/topic?protocol=canal-json" +
+		"&sasl-user=ticdc&sasl-password=verysecure&secret-access-key=rawsecret"
+
+	maskedURI := maskSinkURIForError(sinkURI)
+	require.NotContains(t, maskedURI, "verysecure")
+	require.NotContains(t, maskedURI, "rawsecret")
+	require.Contains(t, maskedURI, "sasl-password=xxxxx")
+	require.Contains(t, maskedURI, "secret-access-key=xxxxx")
+	require.Contains(t, maskedURI, "sasl-user=ticdc")
+
+	invalidURI := "mysql://root:verysecure@127.0.0.1/%zz"
+	require.Equal(t, "<invalid uri>", maskSinkURIForError(invalidURI))
+
+	err := genSinkURIInvalidError(invalidURI, mustParseURLError(t, invalidURI))
+	require.NotContains(t, err.Error(), "verysecure")
+	require.Contains(t, err.Error(), "<invalid uri>")
+	require.Contains(t, err.Error(), `parse "<invalid uri>"`)
+	require.Contains(t, err.Error(), "invalid URL escape")
+}
+
+func mustParseURLError(t *testing.T, rawURL string) error {
+	t.Helper()
+
+	_, err := url.Parse(rawURL)
+	require.Error(t, err)
+	return err
+}
+
+// TestVerifyRouteConflict covers route conflict detection for eligible and
+// ineligible source tables. It exercises the safe cases first, then verifies
+// that conflicts report both the shared target table and conflicting sources.
+func TestVerifyRouteConflict(t *testing.T) {
+	t.Parallel()
+
+	changefeedID := common.NewChangeFeedIDWithName("test-changefeed", common.DefaultKeyspaceName)
+	replicaCfg := config.GetDefaultReplicaConfig()
+	replicaCfg.Sink.DispatchRules = []*config.DispatchRule{
+		{Matcher: []string{"db1.*"}, TargetSchema: "archive", TargetTable: "{table}"},
+		{Matcher: []string{"db2.*"}, TargetSchema: "archive", TargetTable: "{table}"},
+	}
+
+	eligibleTables := []common.TableName{{Schema: "db1", Table: "orders"}}
+	ineligibleTables := []common.TableName{{Schema: "db2", Table: "orders"}}
+
+	replicaCfg.ForceReplicate = util.AddressOf(false)
+	replicaCfg.IgnoreIneligibleTable = util.AddressOf(true)
+	require.NoError(t, verifyRouteConflict(changefeedID, eligibleTables, ineligibleTables, replicaCfg))
+
+	replicaCfg.IgnoreIneligibleTable = util.AddressOf(false)
+	require.NoError(t, verifyRouteConflict(changefeedID, eligibleTables, ineligibleTables, replicaCfg))
+
+	err := verifyRouteConflict(
+		changefeedID,
+		[]common.TableName{{Schema: "db1", Table: "orders"}, {Schema: "db2", Table: "orders"}},
+		ineligibleTables,
+		replicaCfg,
+	)
+	require.Error(t, err)
+	require.True(t, errors.ErrTableRouteConflict.Equal(err))
+
+	replicaCfg.ForceReplicate = util.AddressOf(true)
+	err = verifyRouteConflict(changefeedID, eligibleTables, ineligibleTables, replicaCfg)
+	require.Error(t, err)
+	require.True(t, errors.ErrTableRouteConflict.Equal(err))
+	require.Contains(t, err.Error(), "target `archive`.`orders`")
+	require.Contains(t, err.Error(), "source `db1`.`orders`")
+	require.Contains(t, err.Error(), "source `db2`.`orders`")
+
+	replicaCfg.ForceReplicate = util.AddressOf(false)
+	replicaCfg.Sink.DispatchRules = []*config.DispatchRule{
+		{Matcher: []string{"db2.*"}, TargetSchema: "db1", TargetTable: "{table}"},
+	}
+	err = verifyRouteConflict(
+		changefeedID,
+		[]common.TableName{{Schema: "db1", Table: "orders"}, {Schema: "db2", Table: "orders"}},
+		nil,
+		replicaCfg,
+	)
+	require.Error(t, err)
+	require.True(t, errors.ErrTableRouteConflict.Equal(err))
+	require.Contains(t, err.Error(), "target `db1`.`orders`")
+	require.Contains(t, err.Error(), "source `db1`.`orders`")
+	require.Contains(t, err.Error(), "source `db2`.`orders`")
+}
+
+func TestVerifyTablesForSinkValidatesStorageColumnSelectors(t *testing.T) {
+	t.Parallel()
+
+	replicaCfg := config.GetDefaultReplicaConfig()
+	replicaCfg.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{Matcher: []string{"test.t"}, Columns: []string{"name"}},
+	}
+	tableInfos := []*common.TableInfo{newTableInfoWithPrimaryKeyForTest()}
+
+	err := verifyTablesForSink(replicaCfg, config.FileScheme, "", config.ProtocolCanalJSON, tableInfos)
+	require.Error(t, err)
+	require.True(t, errors.ErrColumnSelectorFailed.Equal(err))
+
+	replicaCfg.Sink.ColumnSelectors[0].Columns = []string{"id", "name"}
+	require.NoError(t, verifyTablesForSink(replicaCfg, config.FileScheme, "", config.ProtocolCanalJSON, tableInfos))
+}
+
+func newTableInfoWithPrimaryKeyForTest() *common.TableInfo {
+	idFieldType := types.NewFieldType(mysql.TypeLong)
+	idFieldType.AddFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
+
+	return common.WrapTableInfo("test", &timodel.TableInfo{
+		ID:         1,
+		Name:       ast.NewCIStr("t"),
+		PKIsHandle: true,
+		Columns: []*timodel.ColumnInfo{
+			{
+				ID:        1,
+				Name:      ast.NewCIStr("id"),
+				FieldType: *idFieldType,
+				State:     timodel.StatePublic,
+			},
+			{
+				ID:        2,
+				Name:      ast.NewCIStr("name"),
+				FieldType: *types.NewFieldType(mysql.TypeVarchar),
+				State:     timodel.StatePublic,
+			},
+		},
+	})
+}

cmd/storage-consumer/consumer.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cmd/util"
 	"github.com/pingcap/ticdc/downstreamadapter/sink"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/common/event"
@@ -60,6 +61,7 @@ type indexRange struct {
 type consumer struct {
 	replicationCfg  *config.ReplicaConfig
 	codecCfg        *common.Config
+	columnSelectors *columnselector.ColumnSelectors
 	externalStorage storeapi.Storage
 	fileExtension   string
 	sink            sink.Sink
@@ -122,6 +124,10 @@ func newConsumer(ctx context.Context) (*consumer, error) {
 	if err != nil {
 		return nil, err
 	}
+	columnSelectors, err := columnselector.New(replicaConfig.Sink)
+	if err != nil {
+		return nil, err
+	}
 
 	extension := helper.GetFileExtension(protocol)
 
@@ -152,6 +158,7 @@ func newConsumer(ctx context.Context) (*consumer, error) {
 	return &consumer{
 		replicationCfg:    replicaConfig,
 		codecCfg:          codecConfig,
+		columnSelectors:   columnSelectors,
 		externalStorage:   storage,
 		fileExtension:     extension,
 		sink:              sink,
@@ -302,7 +309,19 @@ func (c *consumer) appendDMLEvents(
 
 	switch c.codecCfg.Protocol {
 	case config.ProtocolCsv:
+<<<<<<< HEAD
 		decoder, err = csv.NewDecoder(ctx, c.codecCfg, tableInfo, content)
+=======
+		tableInfo := schemaFile.TableInfo()
+		// CSV rows contain selected values without column names, so decode with the same selector.
+		decoder, err = csv.NewDecoderWithColumnSelector(
+			ctx,
+			c.codecCfg,
+			tableInfo,
+			content,
+			c.columnSelectors.GetForTableInfo(tableInfo),
+		)
+>>>>>>> 07e944782 (sink: add column selector for storage sink (#5595))
 		if err != nil {
 			return errors.Trace(err)
 		}