diff --git a/api/v2/changefeed.go b/api/v2/changefeed.go index 0c9eb45b3f..4b167d12cb 100644 --- a/api/v2/changefeed.go +++ b/api/v2/changefeed.go @@ -1744,7 +1744,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 } @@ -1759,13 +1759,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 } diff --git a/api/v2/changefeed_test.go b/api/v2/changefeed_test.go index 38fd1155b9..a8594ffa1c 100644 --- a/api/v2/changefeed_test.go +++ b/api/v2/changefeed_test.go @@ -32,6 +32,10 @@ import ( "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" ) @@ -262,3 +266,45 @@ func TestVerifyRouteConflict(t *testing.T) { 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, + }, + }, + }) +} diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go index 5226d488e3..ab5f922435 100644 --- a/cmd/storage-consumer/consumer.go +++ b/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" "github.com/pingcap/ticdc/pkg/cloudstorage" commonType "github.com/pingcap/ticdc/pkg/common" @@ -68,6 +69,7 @@ type storageMetadata struct { type consumer struct { replicationCfg *config.ReplicaConfig codecCfg *common.Config + columnSelectors *columnselector.ColumnSelectors externalStorage storeapi.Storage fileExtension string sink sink.Sink @@ -132,6 +134,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) @@ -162,6 +168,7 @@ func newConsumer(ctx context.Context) (*consumer, error) { return &consumer{ replicationCfg: replicaConfig, codecCfg: codecConfig, + columnSelectors: columnSelectors, externalStorage: storage, fileExtension: extension, sink: sink, @@ -328,7 +335,15 @@ func (c *consumer) appendDMLEvents( var decoder common.Decoder switch c.codecCfg.Protocol { case config.ProtocolCsv: - decoder, err = csv.NewDecoder(ctx, c.codecCfg, schemaFile.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), + ) if err != nil { return errors.Trace(err) } diff --git a/downstreamadapter/sink/cloudstorage/buffer_manager.go b/downstreamadapter/sink/cloudstorage/buffer_manager.go index 6ad0451167..fa49c404e1 100644 --- a/downstreamadapter/sink/cloudstorage/buffer_manager.go +++ b/downstreamadapter/sink/cloudstorage/buffer_manager.go @@ -99,12 +99,14 @@ func (c *bufferManager) run(ctx context.Context) error { func (c *bufferManager) handleDMLTask(ctx context.Context, task *task) error { if len(task.encodedMsgs) == 0 { - task.callbacks.postEnqueue() + if task.postEnqueue != nil { + task.postEnqueue() + } return nil } for { - action, entry, err := c.spool.TryEnqueue(task.encodedMsgs, task.callbacks.postEnqueue) + action, entry, err := c.spool.TryEnqueue(task.encodedMsgs, task.postEnqueue) if err != nil { return err } diff --git a/downstreamadapter/sink/cloudstorage/buffer_manager_test.go b/downstreamadapter/sink/cloudstorage/buffer_manager_test.go index 48518eadd1..370ae3a986 100644 --- a/downstreamadapter/sink/cloudstorage/buffer_manager_test.go +++ b/downstreamadapter/sink/cloudstorage/buffer_manager_test.go @@ -171,7 +171,7 @@ func newBufferedTask(table string, dispatcherID commonType.DispatcherID, payload }, TableInfoVersion: 1, DispatcherID: dispatcherID, - }, event) + }, event, nil) msg := common.NewMsg(nil, []byte(payload)) msg.SetRowsCount(1) t.encodedMsgs = []*common.Message{msg} diff --git a/downstreamadapter/sink/cloudstorage/dml_writers.go b/downstreamadapter/sink/cloudstorage/dml_writers.go index 37f54fd258..6f4d8e9eb2 100644 --- a/downstreamadapter/sink/cloudstorage/dml_writers.go +++ b/downstreamadapter/sink/cloudstorage/dml_writers.go @@ -18,6 +18,7 @@ import ( "time" "github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" sinkmetrics "github.com/pingcap/ticdc/downstreamadapter/sink/metrics" "github.com/pingcap/ticdc/pkg/cloudstorage" commonType "github.com/pingcap/ticdc/pkg/common" @@ -43,8 +44,9 @@ type dmlWriters struct { encodeGroup *encoderGroup spool *spool.Spool - writers []*writer - closed atomic.Bool + columnSelector *columnselector.ColumnSelectors + writers []*writer + closed atomic.Bool } func newDMLWriters( @@ -54,6 +56,7 @@ func newDMLWriters( encoderConfig *common.Config, extension string, statistics *metrics.Statistics, + columnSelector *columnselector.ColumnSelectors, ) (*dmlWriters, error) { messageCh := chann.NewUnlimitedChannelDefault[*task]() encoderGroup := newEncoderGroup( @@ -76,12 +79,13 @@ func newDMLWriters( } return &dmlWriters{ - changefeedID: changefeedID, - statistics: statistics, - msgCh: messageCh, - encodeGroup: encoderGroup, - spool: spool, - writers: writers, + changefeedID: changefeedID, + statistics: statistics, + msgCh: messageCh, + encodeGroup: encoderGroup, + spool: spool, + columnSelector: columnSelector, + writers: writers, }, nil } @@ -168,7 +172,7 @@ func (d *dmlWriters) addDMLEvent(event *commonEvent.DMLEvent) { TableInfoVersion: event.TableInfoVersion, DispatcherID: event.GetDispatcherID(), } - d.msgCh.Push(newDMLTask(table, event)) + d.msgCh.Push(newDMLTask(table, event, d.columnSelector.GetForTableInfo(event.TableInfo))) } func (d *dmlWriters) flushDMLBeforeBlock(ctx context.Context, event commonEvent.BlockEvent) error { diff --git a/downstreamadapter/sink/cloudstorage/encoder_group.go b/downstreamadapter/sink/cloudstorage/encoder_group.go index 9b36deb502..023bd52467 100644 --- a/downstreamadapter/sink/cloudstorage/encoder_group.go +++ b/downstreamadapter/sink/cloudstorage/encoder_group.go @@ -108,13 +108,12 @@ func (eg *encoderGroup) runEncoder(ctx context.Context, index int) error { continue } - err = encoder.AppendTxnEvent(task.event) + err = encoder.AppendTxnEvent(task.rowEvents) if err != nil { return err } task.encodedMsgs = encoder.Build() - task.replacePostFlushCallbacks() - task.event = nil + task.rowEvents = nil future.Done() } } diff --git a/downstreamadapter/sink/cloudstorage/encoder_group_test.go b/downstreamadapter/sink/cloudstorage/encoder_group_test.go index ccc722df03..a0cf5d44c6 100644 --- a/downstreamadapter/sink/cloudstorage/encoder_group_test.go +++ b/downstreamadapter/sink/cloudstorage/encoder_group_test.go @@ -16,6 +16,7 @@ package cloudstorage import ( "context" "net/url" + "sync/atomic" "testing" "time" @@ -174,6 +175,15 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) { }) dispatcherID := commonType.NewDispatcherID() + var flushCount atomic.Int64 + var enqueueCount atomic.Int64 + event := newTestDMLEvent(dispatcherID, 100) + event.AddPostFlushFunc(func() { + flushCount.Add(1) + }) + event.AddPostEnqueueFunc(func() { + enqueueCount.Add(1) + }) taskValue := newDMLTask( cloudstorage.VersionedTableName{ TableNameWithPhysicTableID: commonType.TableName{ @@ -184,7 +194,8 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) { TableInfoVersion: 1, DispatcherID: dispatcherID, }, - newTestDMLEvent(dispatcherID, 100), + event, + nil, ) require.NoError(t, group.add(ctx, taskValue)) @@ -201,7 +212,15 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) { } task := future.task require.Equal(t, taskValue, task) - require.Nil(t, task.event) + require.Nil(t, task.rowEvents) + require.Len(t, task.encodedMsgs, 1) + require.NotNil(t, task.encodedMsgs[0].Callback) + task.encodedMsgs[0].Callback() + require.Equal(t, int64(1), flushCount.Load()) + require.Equal(t, int64(1), enqueueCount.Load()) + require.NotNil(t, task.postEnqueue) + task.postEnqueue() + require.Equal(t, int64(1), enqueueCount.Load()) done <- struct{}{} return nil } @@ -251,6 +270,8 @@ func newTestDMLEvent(dispatcherID commonType.DispatcherID, tableID int64) *commo PhysicalTableID: tableID, TableInfo: tableInfo, TableInfoVersion: 1, + Length: 1, + RowTypes: []commonType.RowType{commonType.RowTypeInsert}, Rows: chunk.MutRowFromValues(1, "hello world").ToRow().Chunk(), } } diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go index 79204a8901..297c5f1e06 100644 --- a/downstreamadapter/sink/cloudstorage/sink.go +++ b/downstreamadapter/sink/cloudstorage/sink.go @@ -21,6 +21,7 @@ import ( "time" "github.com/pingcap/log" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/pkg/cloudstorage" "github.com/pingcap/ticdc/pkg/common" @@ -86,6 +87,9 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url. if err != nil { return err } + if _, err = columnselector.New(sinkConfig); err != nil { + return err + } _, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt, math.MaxInt) if err != nil { return err @@ -123,6 +127,10 @@ func New( if err != nil { return nil, err } + columnSelectors, err := columnselector.New(sinkConfig) + if err != nil { + return nil, err + } storage, err := util.GetExternalStorageWithDefaultTimeout(ctx, sinkURI.String()) if err != nil { return nil, err @@ -134,7 +142,7 @@ func New( storage.Close() } }() - dmlWriters, err := newDMLWriters(changefeedID, storage, cfg, encoderConfig, ext, statistics) + dmlWriters, err := newDMLWriters(changefeedID, storage, cfg, encoderConfig, ext, statistics, columnSelectors) if err != nil { return nil, err } diff --git a/downstreamadapter/sink/cloudstorage/sink_test.go b/downstreamadapter/sink/cloudstorage/sink_test.go index 7ea81044d7..ea94d9913e 100644 --- a/downstreamadapter/sink/cloudstorage/sink_test.go +++ b/downstreamadapter/sink/cloudstorage/sink_test.go @@ -157,6 +157,76 @@ func TestBasicFunctionality(t *testing.T) { require.Equal(t, count.Load(), int64(3)) } +func TestCloudStorageSinkWithColumnSelector(t *testing.T) { + parentDir := t.TempDir() + uri := fmt.Sprintf("file:///%s?protocol=csv&flush-interval=3600s&file-size=1024", parentDir) + sinkURI, err := url.Parse(uri) + require.NoError(t, err) + + replicaConfig := config.GetDefaultReplicaConfig() + replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{ + {Matcher: []string{"test.table1"}, Columns: []string{"c1"}}, + } + err = replicaConfig.ValidateAndAdjust(sinkURI) + require.NoError(t, err) + replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String()) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + setPDClockForTest(t, pdutil.NewClock4Test()) + cloudStorageSink, err := newSinkForTest(ctx, replicaConfig, sinkURI, nil) + require.NoError(t, err) + + runDone := runSinkInBackground(t, ctx, cloudStorageSink) + defer cancelAndWaitSink(t, cancel, runDone) + + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + + helper.Tk().MustExec("use test") + job := helper.DDL2Job("create table table1(c1 int primary key, c2 varchar(255))") + require.NotNil(t, job) + helper.ApplyJob(job) + + dispatcherID := common.NewDispatcherID() + event := helper.DML2Event(job.SchemaName, job.TableName, `insert into table1 values (1, "filtered")`) + event.TableInfoVersion = job.BinlogInfo.FinishedTS + event.DispatcherID = dispatcherID + + var flushed atomic.Uint64 + event.AddPostFlushFunc(func() { + flushed.Add(1) + }) + + cloudStorageSink.AddDMLEvent(event) + err = cloudStorageSink.FlushDMLBeforeBlock(&commonEvent.DDLEvent{ + DispatcherID: dispatcherID, + FinishedTs: event.CommitTs + 1, + }) + require.NoError(t, err) + require.Equal(t, uint64(1), flushed.Load()) + + tableDir := path.Join(parentDir, job.SchemaName, job.TableName, fmt.Sprint(event.TableInfoVersion)) + var content []byte + require.Eventually(t, func() bool { + files, err := os.ReadDir(tableDir) + if err != nil { + return false + } + for _, file := range files { + if file.IsDir() || !strings.HasSuffix(file.Name(), ".csv") { + continue + } + content, err = os.ReadFile(path.Join(tableDir, file.Name())) + return err == nil + } + return false + }, testEventuallyTimeout, testEventuallyTick) + require.Contains(t, string(content), "1") + require.NotContains(t, string(content), "filtered") +} + func TestIgnoreCallsAfterRunError(t *testing.T) { uri := fmt.Sprintf("file:///%s?protocol=csv", t.TempDir()) sinkURI, err := url.Parse(uri) diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go index 3b9ff1a082..63ef6acd2b 100644 --- a/downstreamadapter/sink/cloudstorage/task.go +++ b/downstreamadapter/sink/cloudstorage/task.go @@ -15,8 +15,8 @@ package cloudstorage import ( "context" - "sync/atomic" + "github.com/pingcap/ticdc/downstreamadapter/sink/helper" "github.com/pingcap/ticdc/pkg/cloudstorage" commonType "github.com/pingcap/ticdc/pkg/common" commonEvent "github.com/pingcap/ticdc/pkg/common/event" @@ -38,10 +38,10 @@ type task struct { dispatcherID commonType.DispatcherID // DML-only fields. - event *commonEvent.DMLEvent // Original DML event to encode and flush. - callbacks *txnCallbacks // Lightweight txn callbacks detached from event. + postEnqueue func() // Transaction enqueue callback. tableInfo *commonType.TableInfo // Table info used after event is released. versionedTable cloudstorage.VersionedTableName // Versioned output identity for the DML event. + rowEvents []*commonEvent.RowEvent // Row events to encode and flush. encodedMsgs []*common.Message // Encoded result built from event. // Flush-only field. @@ -51,17 +51,20 @@ type task struct { func newDMLTask( version cloudstorage.VersionedTableName, event *commonEvent.DMLEvent, + selector commonEvent.Selector, ) *task { - // The dispatcher path registers progress callbacks before calling - // Sink.AddDMLEvent, so snapshot callbacks here and release the large event - // object after encoding. + postEnqueue, postFlush := event.DetachPostCallbacks() return &task{ kind: taskKindDML, - event: event, - callbacks: newTxnCallbacks(event), + postEnqueue: postEnqueue, tableInfo: event.TableInfo, versionedTable: version, - dispatcherID: event.GetDispatcherID(), + // Storage txn encoders attach only the last row callback to the built + // batch message, so the callback is triggered once per encoded txn + // message. Kafka uses row-level callbacks and counts all rows before + // PostFlush, which cannot be reused here for multi-row txns. + rowEvents: helper.NewRowEvents(event, selector, postFlush), + dispatcherID: event.GetDispatcherID(), } } @@ -80,66 +83,6 @@ func (t *task) isFlushTask() bool { return t != nil && t.kind == taskKindFlush } -func (t *task) replacePostFlushCallbacks() { - if len(t.encodedMsgs) == 0 { - return - } - - // Txn encoders put event.PostFlush into message.Callback. That method value - // keeps the original DMLEvent reachable through the encoded messages, so - // replace it with the lightweight callback copy before releasing task.event. - for _, msg := range t.encodedMsgs { - msg.Callback = nil - } - // One callback on the last message is enough because all messages in a task - // are enqueued and flushed as one spool entry. - t.encodedMsgs[len(t.encodedMsgs)-1].Callback = t.callbacks.postFlush -} - -// txnCallbacks is a lightweight copy of a DMLEvent's enqueue and flush -// callbacks. It lets cloud storage release the full DMLEvent after encoding -// while preserving the event callback semantics: each stage runs at most once. -type txnCallbacks struct { - flushed []func() - enqueued []func() - - flushedCalled atomic.Bool - enqueuedCalled atomic.Bool -} - -func newTxnCallbacks(event *commonEvent.DMLEvent) *txnCallbacks { - if event == nil { - return &txnCallbacks{} - } - return &txnCallbacks{ - flushed: append([]func(){}, event.PostTxnFlushed...), - enqueued: append([]func(){}, event.PostTxnEnqueued...), - } -} - -func (c *txnCallbacks) postFlush() { - if c == nil || !c.flushedCalled.CompareAndSwap(false, true) { - return - } - for _, f := range c.flushed { - if f != nil { - f() - } - } - c.postEnqueue() -} - -func (c *txnCallbacks) postEnqueue() { - if c == nil || !c.enqueuedCalled.CompareAndSwap(false, true) { - return - } - for _, f := range c.enqueued { - if f != nil { - f() - } - } -} - func (t *task) wait(ctx context.Context) error { if !t.isFlushTask() { return nil diff --git a/downstreamadapter/sink/cloudstorage/task_test.go b/downstreamadapter/sink/cloudstorage/task_test.go index 0750d199f2..ae9d010613 100644 --- a/downstreamadapter/sink/cloudstorage/task_test.go +++ b/downstreamadapter/sink/cloudstorage/task_test.go @@ -16,11 +16,9 @@ package cloudstorage import ( "context" "errors" - "sync/atomic" "testing" "time" - "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) @@ -50,42 +48,3 @@ func TestFlushMarkerWaitReturnsContextCause(t *testing.T) { err := marker.wait(ctx) require.ErrorIs(t, err, cause) } - -func TestReplacePostFlushCallbacksDetachesOriginalMessageCallbacks(t *testing.T) { - t.Parallel() - - var originalCallbackCount atomic.Int32 - var flushCallbackCount atomic.Int32 - var enqueueCallbackCount atomic.Int32 - task := &task{ - callbacks: &txnCallbacks{ - flushed: []func(){ - func() { - flushCallbackCount.Add(1) - }, - }, - enqueued: []func(){ - func() { - enqueueCallbackCount.Add(1) - }, - }, - }, - encodedMsgs: []*common.Message{ - {Callback: func() { originalCallbackCount.Add(1) }}, - {Callback: func() { originalCallbackCount.Add(1) }}, - {}, - }, - } - - task.replacePostFlushCallbacks() - - require.Nil(t, task.encodedMsgs[0].Callback) - require.Nil(t, task.encodedMsgs[1].Callback) - require.NotNil(t, task.encodedMsgs[2].Callback) - - task.encodedMsgs[2].Callback() - task.encodedMsgs[2].Callback() - require.Equal(t, int32(0), originalCallbackCount.Load()) - require.Equal(t, int32(1), flushCallbackCount.Load()) - require.Equal(t, int32(1), enqueueCallbackCount.Load()) -} diff --git a/downstreamadapter/sink/cloudstorage/writer_test.go b/downstreamadapter/sink/cloudstorage/writer_test.go index 5bbcf578af..7aaf02dd82 100644 --- a/downstreamadapter/sink/cloudstorage/writer_test.go +++ b/downstreamadapter/sink/cloudstorage/writer_test.go @@ -128,7 +128,7 @@ func TestWriterRun(t *testing.T) { TableInfo: tableInfo, Rows: chunk.MutRowFromValues(100, "hello world").ToRow().Chunk(), } - tableTask := newDMLTask(tableName, dmlEvent) + tableTask := newDMLTask(tableName, dmlEvent, nil) tableTask.encodedMsgs = []*common.Message{ { Value: []byte(fmt.Sprintf(`{"id":%d,"database":"test","table":"table1","pkNames":[],"isDdl":false,`+ @@ -201,6 +201,7 @@ func TestWriterFlushMarker(t *testing.T) { PhysicalTableID: 100, TableInfo: tableInfo, }, + nil, ) tableTask.encodedMsgs = []*common.Message{msg} require.NoError(t, d.enqueueTask(ctx, tableTask)) @@ -264,6 +265,7 @@ func TestWriterFlushMarkerOnlyFlushesTargetDispatcher(t *testing.T) { PhysicalTableID: 100, TableInfo: tableInfo, }, + nil, ) msgA := common.NewMsg(nil, []byte(`{"id":"a"}`)) msgA.SetRowsCount(1) @@ -292,6 +294,7 @@ func TestWriterFlushMarkerOnlyFlushesTargetDispatcher(t *testing.T) { }, }), }, + nil, ) msgB := common.NewMsg(nil, []byte(`{"id":"b"}`)) msgB.SetRowsCount(1) @@ -361,6 +364,7 @@ func TestWriterPostEnqueueAfterConsume(t *testing.T) { DispatcherID: dispatcherID, }, dmlEvent, + nil, ) tableTask.encodedMsgs = []*common.Message{ { @@ -384,7 +388,7 @@ func TestWriterPostEnqueueAfterConsume(t *testing.T) { require.ErrorIs(t, <-done, context.Canceled) } -func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) { +func TestWriterPostFlushDoesNotRunPausedPostEnqueue(t *testing.T) { t.Parallel() changefeedID := commonType.NewChangefeedID4Test("test", t.Name()) @@ -411,21 +415,13 @@ func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) { var secondFlushed atomic.Int64 var secondEnqueued atomic.Int64 - secondCallbacks := &txnCallbacks{ - flushed: []func(){ - func() { - secondFlushed.Add(1) - }, - }, - enqueued: []func(){ - func() { - secondEnqueued.Add(1) - }, - }, - } secondMsg := common.NewMsg(nil, []byte(strings.Repeat("b", 120))) - secondMsg.Callback = secondCallbacks.postFlush - secondEntry, err := spoolBuffer.Enqueue([]*common.Message{secondMsg}, secondCallbacks.postEnqueue) + secondMsg.Callback = func() { + secondFlushed.Add(1) + } + secondEntry, err := spoolBuffer.Enqueue([]*common.Message{secondMsg}, func() { + secondEnqueued.Add(1) + }) require.NoError(t, err) defer spoolBuffer.Release(secondEntry) require.Equal(t, int64(0), secondEnqueued.Load()) @@ -442,7 +438,7 @@ func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) { } require.Equal(t, int64(1), secondFlushed.Load()) - require.Equal(t, int64(1), secondEnqueued.Load()) + require.Equal(t, int64(0), secondEnqueued.Load()) spoolBuffer.Release(firstEntry) require.Equal(t, int64(1), secondEnqueued.Load()) @@ -513,6 +509,7 @@ func TestWriterStoresPendingMessagesInSpoolBeforeFlush(t *testing.T) { PhysicalTableID: 100, TableInfo: tableInfo, }, + nil, ) msg := common.NewMsg(nil, []byte(`{"id":1}`)) msg.SetRowsCount(1) @@ -678,6 +675,7 @@ func TestWriterIndexWriteError(t *testing.T) { PhysicalTableID: 100, TableInfo: tableInfo, }, + nil, ) msg := common.NewMsg(nil, []byte(`{"id":1}`)) msg.SetRowsCount(1) @@ -743,6 +741,7 @@ func TestWriterDataFileCloseError(t *testing.T) { PhysicalTableID: 100, TableInfo: tableInfo, }, + nil, ) var callbackCount atomic.Int64 diff --git a/downstreamadapter/sink/columnselector/column_selector.go b/downstreamadapter/sink/columnselector/column_selector.go index 0c619b42eb..df3342f6a9 100644 --- a/downstreamadapter/sink/columnselector/column_selector.go +++ b/downstreamadapter/sink/columnselector/column_selector.go @@ -14,6 +14,8 @@ package columnselector import ( + "slices" + "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter" "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter/partition" "github.com/pingcap/ticdc/pkg/common" @@ -99,9 +101,18 @@ func (c *ColumnSelectors) Get(schema, table string) *ColumnSelector { return &ColumnSelector{} } +func (c *ColumnSelectors) GetForTableInfo(tableInfo *common.TableInfo) *ColumnSelector { + if c == nil || tableInfo == nil { + return NewDefaultColumnSelector() + } + return c.Get(tableInfo.GetSchemaName(), tableInfo.GetTableName()) +} + // VerifyTables return the error if any given table cannot satisfy the column selector constraints. // 1. if the column is filter out, it must not be a part of handle key or the unique key. // 2. if the filtered out column is used in the column dispatcher, return error. +// +// The column dispatcher check is only applied when eventRouter is not nil. func (c *ColumnSelectors) VerifyTables( infos []*common.TableInfo, eventRouter *eventrouter.EventRouter, ) error { @@ -123,17 +134,17 @@ func (c *ColumnSelectors) VerifyTables( continue } - partitionDispatcher := eventRouter.GetPartitionGenerator(table.TableName.Schema, table.TableName.Table) - switch v := partitionDispatcher.(type) { - case *partition.ColumnsPartitionGenerator: - for _, col := range v.Columns { - if col == columnInfo.Name.O { + if eventRouter != nil { + partitionDispatcher := eventRouter.GetPartitionGenerator(table.TableName.Schema, table.TableName.Table) + switch v := partitionDispatcher.(type) { + case *partition.ColumnsPartitionGenerator: + if slices.Contains(v.Columns, columnInfo.Name.O) { return errors.ErrColumnSelectorFailed.GenWithStack( "the filtered out column is used in the column dispatcher, "+ "table: %v, column: %s", table.TableName, columnInfo.Name) } + default: } - default: } } @@ -151,16 +162,18 @@ func (c *ColumnSelectors) VerifyTables( func verifyIndices(table *common.TableInfo, retainedColumns map[string]struct{}) bool { primaryKeyColumns := table.GetPrimaryKeyColumnNames() - retained := true - for _, name := range primaryKeyColumns { - if _, ok := retainedColumns[name]; !ok { - retained = false - break + if len(primaryKeyColumns) > 0 { + retained := true + for _, name := range primaryKeyColumns { + if _, ok := retainedColumns[name]; !ok { + retained = false + break + } + } + // primary key columns are retained, return true. + if retained { + return true } - } - // primary key columns are retained, return true. - if retained { - return true } // at least one unique key columns are retained, return true. @@ -169,7 +182,7 @@ func verifyIndices(table *common.TableInfo, retainedColumns map[string]struct{}) continue } - retained = true + retained := true for _, col := range index.Columns { if _, ok := retainedColumns[col.Name.O]; !ok { retained = false diff --git a/downstreamadapter/sink/columnselector/column_selector_test.go b/downstreamadapter/sink/columnselector/column_selector_test.go index 255704f2ac..2a051b982d 100644 --- a/downstreamadapter/sink/columnselector/column_selector_test.go +++ b/downstreamadapter/sink/columnselector/column_selector_test.go @@ -16,9 +16,13 @@ package columnselector import ( "testing" + commonType "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/errors" "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" ) @@ -182,3 +186,56 @@ func TestColumnSelectorGetSelector(t *testing.T) { } } } + +func TestVerifyTablesRequiresFullUniqueKey(t *testing.T) { + t.Parallel() + + replicaConfig := config.GetDefaultReplicaConfig() + replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{ + { + Matcher: []string{"test.t"}, + Columns: []string{"a"}, + }, + } + selectors, err := New(replicaConfig.Sink) + require.NoError(t, err) + + tableInfo := commonType.WrapTableInfo("test", &model.TableInfo{ + Name: ast.NewCIStr("t"), + Columns: []*model.ColumnInfo{ + newColumnInfoForSelectorTest(1, "a", mysql.NotNullFlag), + newColumnInfoForSelectorTest(2, "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, + }, + }, + }) + + err = selectors.VerifyTables([]*commonType.TableInfo{tableInfo}, nil) + require.Error(t, err) + require.True(t, errors.ErrColumnSelectorFailed.Equal(err)) + + replicaConfig.Sink.ColumnSelectors[0].Columns = []string{"a", "b"} + selectors, err = New(replicaConfig.Sink) + require.NoError(t, err) + require.NoError(t, selectors.VerifyTables([]*commonType.TableInfo{tableInfo}, nil)) +} + +func newColumnInfoForSelectorTest(id int64, name string, flag uint) *model.ColumnInfo { + ft := types.NewFieldType(mysql.TypeLong) + ft.AddFlag(flag) + return &model.ColumnInfo{ + ID: id, + Name: ast.NewCIStr(name), + FieldType: *ft, + State: model.StatePublic, + } +} diff --git a/downstreamadapter/sink/helper/mq_row_event.go b/downstreamadapter/sink/helper/mq_row_event.go new file mode 100644 index 0000000000..bbd8de207c --- /dev/null +++ b/downstreamadapter/sink/helper/mq_row_event.go @@ -0,0 +1,99 @@ +// 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 helper + +import ( + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" + "github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter/partition" + commonEvent "github.com/pingcap/ticdc/pkg/common/event" +) + +func NewMQRowEvents( + event *commonEvent.DMLEvent, + topic string, + partitionNum int32, + partitionGenerator partition.Generator, + selector commonEvent.Selector, +) ([]*commonEvent.MQRowEvent, error) { + callback := NewPostFlushRowCallback(event, uint64(event.Len())) + events := make([]*commonEvent.MQRowEvent, 0, event.Len()) + if selector == nil { + selector = columnselector.NewDefaultColumnSelector() + } + + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + + index, key, err := partitionGenerator.GeneratePartitionIndexAndKey( + &row, partitionNum, event.TableInfo, event.CommitTs) + if err != nil { + return nil, err + } + + events = append(events, &commonEvent.MQRowEvent{ + Key: commonEvent.TopicPartitionKey{ + Topic: topic, + Partition: index, + PartitionKey: key, + TotalPartition: partitionNum, + }, + RowEvent: commonEvent.RowEvent{ + PhysicalTableID: event.PhysicalTableID, + TableInfo: event.TableInfo, + StartTs: event.StartTs, + CommitTs: event.CommitTs, + Event: row, + Callback: callback, + ColumnSelector: selector, + Checksum: row.Checksum, + }, + }) + } + return events, nil +} + +func NewRowEvents( + event *commonEvent.DMLEvent, + selector commonEvent.Selector, + callback func(), +) []*commonEvent.RowEvent { + if selector == nil { + selector = columnselector.NewDefaultColumnSelector() + } + + events := make([]*commonEvent.RowEvent, 0, event.Len()) + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + + events = append(events, &commonEvent.RowEvent{ + PhysicalTableID: event.PhysicalTableID, + TableInfo: event.TableInfo, + StartTs: event.StartTs, + CommitTs: event.CommitTs, + Event: row, + Callback: callback, + ColumnSelector: selector, + Checksum: row.Checksum, + }) + } + return events +} diff --git a/downstreamadapter/sink/helper/row_callback.go b/downstreamadapter/sink/helper/row_callback.go index e690ec16c0..9acbc6e671 100644 --- a/downstreamadapter/sink/helper/row_callback.go +++ b/downstreamadapter/sink/helper/row_callback.go @@ -18,9 +18,9 @@ import ( "go.uber.org/atomic" ) -// NewTxnPostFlushRowCallback returns a row-level callback that triggers txn-level +// NewPostFlushRowCallback returns a row-level callback that triggers txn-level // PostFlush exactly once when the callback has been invoked totalCount times. -func NewTxnPostFlushRowCallback(event *event.DMLEvent, totalCount uint64) func() { +func NewPostFlushRowCallback(event *event.DMLEvent, totalCount uint64) func() { var calledCount atomic.Uint64 return func() { if calledCount.Inc() == totalCount { diff --git a/downstreamadapter/sink/helper/row_callback_test.go b/downstreamadapter/sink/helper/row_callback_test.go index b4996ba4a4..ddcf076283 100644 --- a/downstreamadapter/sink/helper/row_callback_test.go +++ b/downstreamadapter/sink/helper/row_callback_test.go @@ -29,7 +29,7 @@ func TestTxnPostFlushRowCallback(t *testing.T) { flushCount++ }) - rowCallback := NewTxnPostFlushRowCallback(event, 3) + rowCallback := NewPostFlushRowCallback(event, 3) rowCallback() rowCallback() require.Equal(t, 0, flushCount) diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go index 9de1070e15..31cab1e900 100644 --- a/downstreamadapter/sink/kafka/sink.go +++ b/downstreamadapter/sink/kafka/sink.go @@ -317,41 +317,10 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { } partitionGenerator := s.comp.eventRouter.GetPartitionGenerator(schema, table) - selector := s.comp.columnSelector.Get(schema, table) - rowsCount := event.Len() - events := make([]*commonEvent.MQRowEvent, 0, rowsCount) - rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount)) - - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - - index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(&row, partitionNum, event.TableInfo, event.CommitTs) - if err != nil { - return errors.Trace(err) - } - - events = append(events, &commonEvent.MQRowEvent{ - Key: commonEvent.TopicPartitionKey{ - Topic: topic, - Partition: index, - PartitionKey: key, - TotalPartition: partitionNum, - }, - RowEvent: commonEvent.RowEvent{ - PhysicalTableID: event.PhysicalTableID, - TableInfo: event.TableInfo, - StartTs: event.StartTs, - CommitTs: event.CommitTs, - Event: row, - Callback: rowCallback, - ColumnSelector: selector, - Checksum: row.Checksum, - }, - }) + selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo) + events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector) + if err != nil { + return errors.Trace(err) } s.rowChan.Push(events...) } diff --git a/downstreamadapter/sink/pulsar/sink.go b/downstreamadapter/sink/pulsar/sink.go index 91c1d71729..9895541233 100644 --- a/downstreamadapter/sink/pulsar/sink.go +++ b/downstreamadapter/sink/pulsar/sink.go @@ -404,41 +404,10 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error { } partitionGenerator := s.comp.eventRouter.GetPartitionGenerator(schema, table) - selector := s.comp.columnSelector.Get(schema, table) - rowsCount := event.Len() - events := make([]*commonEvent.MQRowEvent, 0, rowsCount) - rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount)) - - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - - index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(&row, partitionNum, event.TableInfo, event.CommitTs) - if err != nil { - return errors.Trace(err) - } - - events = append(events, &commonEvent.MQRowEvent{ - Key: commonEvent.TopicPartitionKey{ - Topic: topic, - Partition: index, - PartitionKey: key, - TotalPartition: partitionNum, - }, - RowEvent: commonEvent.RowEvent{ - PhysicalTableID: event.PhysicalTableID, - TableInfo: event.TableInfo, - StartTs: event.StartTs, - CommitTs: event.CommitTs, - Event: row, - Callback: rowCallback, - ColumnSelector: selector, - Checksum: row.Checksum, - }, - }) + selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo) + events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector) + if err != nil { + return errors.Trace(err) } s.rowChan.Push(events...) } diff --git a/downstreamadapter/sink/redo/sink.go b/downstreamadapter/sink/redo/sink.go index 101a6e7311..d19fbf9c1e 100644 --- a/downstreamadapter/sink/redo/sink.go +++ b/downstreamadapter/sink/redo/sink.go @@ -168,7 +168,7 @@ func (s *Sink) WriteBlockEvent(event commonEvent.BlockEvent) error { func (s *Sink) AddDMLEvent(event *commonEvent.DMLEvent) { rowsCount := event.Len() events := make([]*commonEvent.RedoRowEvent, 0, rowsCount) - rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount)) + rowCallback := helper.NewPostFlushRowCallback(event, uint64(rowsCount)) var ( startTs = event.GetStartTs() diff --git a/pkg/common/event/dml_event.go b/pkg/common/event/dml_event.go index 9b6e8b7b6e..bb0f19dcf3 100644 --- a/pkg/common/event/dml_event.go +++ b/pkg/common/event/dml_event.go @@ -670,7 +670,7 @@ func (t *DMLEvent) PostFlush() { // This stage does not mean data is already written to downstream. The method is // idempotent and guarantees enqueue callbacks run at most once. func (t *DMLEvent) PostEnqueue() { - if !t.postEnqueueCalled.CAS(false, true) { + if !t.postEnqueueCalled.CompareAndSwap(false, true) { return } for _, f := range t.PostTxnEnqueued { @@ -678,6 +678,35 @@ func (t *DMLEvent) PostEnqueue() { } } +// DetachPostCallbacks returns callbacks with the same PostFlush/PostEnqueue +// semantics as this event, then removes the callback slices from the event. +// The returned closures intentionally do not capture the DMLEvent, so sinks can +// keep callbacks after encoding without retaining the event rows. +func (t *DMLEvent) DetachPostCallbacks() (postEnqueue func(), postFlush func()) { + postTxnEnqueued := append([]func(){}, t.PostTxnEnqueued...) + postTxnFlushed := append([]func(){}, t.PostTxnFlushed...) + t.PostTxnEnqueued = nil + t.PostTxnFlushed = nil + + var postEnqueueCalled atomic.Bool + postEnqueueCalled.Store(t.postEnqueueCalled.Load()) + postEnqueue = func() { + if !postEnqueueCalled.CompareAndSwap(false, true) { + return + } + for _, f := range postTxnEnqueued { + f() + } + } + postFlush = func() { + for _, f := range postTxnFlushed { + f() + } + postEnqueue() + } + return postEnqueue, postFlush +} + func (t *DMLEvent) GetSeq() uint64 { return t.Seq } diff --git a/pkg/common/event/dml_event_test.go b/pkg/common/event/dml_event_test.go index f662e118c6..7a04d9df42 100644 --- a/pkg/common/event/dml_event_test.go +++ b/pkg/common/event/dml_event_test.go @@ -463,6 +463,7 @@ func TestDMLEventPostCallbacks(t *testing.T) { t.Run("post flush triggers post enqueue once", verifyDMLEventPostFlushTriggersPostEnqueueOnce) t.Run("post flush order and fallback", verifyDMLEventPostFlushRunsFlushBeforePostEnqueueFallback) t.Run("post enqueue concurrent with post flush", verifyDMLEventPostEnqueueConcurrentWithPostFlush) + t.Run("detach callbacks", verifyDMLEventDetachPostCallbacks) } func verifyDMLEventPostFlushTriggersPostEnqueueOnce(t *testing.T) { @@ -529,3 +530,26 @@ func verifyDMLEventPostEnqueueConcurrentWithPostFlush(t *testing.T) { require.Equal(t, int64(1), enqueueCalled.Load()) } + +func verifyDMLEventDetachPostCallbacks(t *testing.T) { + t.Parallel() + + event := &DMLEvent{} + order := make([]string, 0, 3) + event.AddPostFlushFunc(func() { + order = append(order, "flush") + }) + event.AddPostEnqueueFunc(func() { + order = append(order, "enqueue") + }) + + postEnqueue, postFlush := event.DetachPostCallbacks() + require.Empty(t, event.PostTxnFlushed) + require.Empty(t, event.PostTxnEnqueued) + + postFlush() + postEnqueue() + event.PostFlush() + + require.Equal(t, []string{"flush", "enqueue"}, order) +} diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder.go b/pkg/sink/codec/canal/canal_json_txn_encoder.go index 3e10d98229..61f20ab173 100644 --- a/pkg/sink/codec/canal/canal_json_txn_encoder.go +++ b/pkg/sink/codec/canal/canal_json_txn_encoder.go @@ -17,7 +17,6 @@ import ( "bytes" "github.com/pingcap/log" - "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/sink/codec/common" @@ -33,34 +32,21 @@ type JSONTxnEventEncoder struct { valueBuf *bytes.Buffer batchSize int callback func() - - columnSelector commonEvent.Selector } // NewJSONTxnEventEncoder creates a new JSONTxnEventEncoder func NewJSONTxnEventEncoder(config *common.Config) common.TxnEventEncoder { return &JSONTxnEventEncoder{ - valueBuf: &bytes.Buffer{}, - terminator: []byte(config.Terminator), - columnSelector: columnselector.NewDefaultColumnSelector(), - config: config, + valueBuf: &bytes.Buffer{}, + terminator: []byte(config.Terminator), + config: config, } } // AppendTxnEvent appends a txn event to the encoder. -func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error { - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - value, err := newJSONMessageForDML(&commonEvent.RowEvent{ - TableInfo: event.TableInfo, - CommitTs: event.CommitTs, - Event: row, - ColumnSelector: j.columnSelector, - }, j.config, false, "") +func (j *JSONTxnEventEncoder) AppendTxnEvent(rowEvents []*commonEvent.RowEvent) error { + for _, rowEvent := range rowEvents { + value, err := newJSONMessageForDML(rowEvent, j.config, false, "") if err != nil { return err } @@ -69,14 +55,16 @@ func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error log.Warn("Single message is too large for canal-json", zap.Int("maxMessageBytes", j.config.MaxMessageBytes), zap.Int("length", length), - zap.Any("table", event.TableInfo.TableName)) - return errors.ErrMessageTooLarge.GenWithStackByArgs(event.TableInfo.GetTargetTableName(), length, j.config.MaxMessageBytes) + zap.Any("table", rowEvent.TableInfo.TableName)) + return errors.ErrMessageTooLarge.GenWithStackByArgs(rowEvent.TableInfo.GetTargetTableName(), length, j.config.MaxMessageBytes) } j.valueBuf.Write(value) j.valueBuf.Write(j.terminator) j.batchSize++ } - j.callback = event.PostFlush + if len(rowEvents) > 0 { + j.callback = rowEvents[len(rowEvents)-1].Callback + } return nil } diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go index c0c6943d26..5c18f8d533 100644 --- a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go +++ b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go @@ -16,12 +16,22 @@ package canal import ( "testing" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" + sinkhelper "github.com/pingcap/ticdc/downstreamadapter/sink/helper" commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) +func appendTxnEventForTest( + encoder common.TxnEventEncoder, + event *commonEvent.DMLEvent, + selector commonEvent.Selector, +) error { + return encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, selector, nil)) +} + func TestBuildCanalJSONTxnEventEncoder(t *testing.T) { t.Parallel() cfg := common.NewConfig(config.ProtocolCanalJSON) @@ -42,13 +52,13 @@ func TestCanalJSONTxnEventEncoderMaxMessageBytes(t *testing.T) { maxMessageBytes := 300 cfg := common.NewConfig(config.ProtocolCanalJSON).WithMaxMessageBytes(maxMessageBytes) encoder := NewJSONTxnEventEncoder(cfg) - err := encoder.AppendTxnEvent(testEvent) + err := appendTxnEventForTest(encoder, testEvent, nil) require.Nil(t, err) // the test message length is larger than max-message-bytes cfg = cfg.WithMaxMessageBytes(100) encoder = NewJSONTxnEventEncoder(cfg) - err = encoder.AppendTxnEvent(testEvent) + err = appendTxnEventForTest(encoder, testEvent, nil) require.NotNil(t, err) } @@ -56,7 +66,7 @@ func TestCanalJSONTxnEventEncoderUsesTargetNames(t *testing.T) { t.Parallel() encoder := NewJSONTxnEventEncoder(common.NewConfig(config.ProtocolCanalJSON)) - require.NoError(t, encoder.AppendTxnEvent(common.NewRoutedDMLEvent4Test())) + require.NoError(t, appendTxnEventForTest(encoder, common.NewRoutedDMLEvent4Test(), nil)) messages := encoder.Build() require.Len(t, messages, 1) @@ -76,23 +86,22 @@ func TestCanalJSONAppendTxnEventEncoderWithCallback(t *testing.T) { require.NotNil(t, encoder) event := helper.DML2Event("test", "t", `insert into test.t values("aa")`, `insert into test.t values("bb")`) - count := 0 - // Empty build makes sure that the callback build logic not broken. + // Empty build makes sure the build path handles an empty encoder. msgs := encoder.Build() require.Len(t, msgs, 0, "no message should be built and no panic") // Append the events. - event.AddPostFlushFunc(func() { + err := encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, nil, func() { count++ - }) - err := encoder.AppendTxnEvent(event) + })) require.Nil(t, err) require.Equal(t, 0, count, "nothing should be called") msgs = encoder.Build() require.Len(t, msgs, 1, "expected one message") + require.NotNil(t, msgs[0].Callback) msgs[0].Callback() require.Equal(t, 1, count, "expected one callback be called") // Assert the build reset all the internal states. @@ -100,3 +109,27 @@ func TestCanalJSONAppendTxnEventEncoderWithCallback(t *testing.T) { require.Equal(t, 0, encoder.(*JSONTxnEventEncoder).batchSize) require.Equal(t, 0, encoder.(*JSONTxnEventEncoder).valueBuf.Len()) } + +func TestCanalJSONTxnEventEncoderWithColumnSelector(t *testing.T) { + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + + helper.DDL2Event("create table test.t(col1 int primary key, col2 varchar(255))") + event := helper.DML2Event("test", "t", `insert into test.t values (1, "filtered")`) + + selectors, err := columnselector.New(&config.SinkConfig{ + ColumnSelectors: []*config.ColumnSelector{ + {Matcher: []string{"test.t"}, Columns: []string{"col1"}}, + }, + }) + require.NoError(t, err) + + encoder := NewJSONTxnEventEncoder(common.NewConfig(config.ProtocolCanalJSON)) + require.NoError(t, appendTxnEventForTest(encoder, event, selectors.GetForTableInfo(event.TableInfo))) + messages := encoder.Build() + require.Len(t, messages, 1) + value := string(messages[0].Value) + require.Contains(t, value, "col1") + require.NotContains(t, value, "col2") + require.NotContains(t, value, "filtered") +} diff --git a/pkg/sink/codec/common/encoder.go b/pkg/sink/codec/common/encoder.go index bcb9afd365..5a007434be 100644 --- a/pkg/sink/codec/common/encoder.go +++ b/pkg/sink/codec/common/encoder.go @@ -38,7 +38,7 @@ type EventEncoder interface { // TxnEventEncoder is an abstraction for events encoder type TxnEventEncoder interface { // AppendTxnEvent append a txn event into the buffer. - AppendTxnEvent(*commonEvent.DMLEvent) error + AppendTxnEvent([]*commonEvent.RowEvent) error // Build builds the batch and returns the bytes of key and value. // Should be called after `AppendTxnEvent` Build() []*Message diff --git a/pkg/sink/codec/csv/csv_decoder.go b/pkg/sink/codec/csv/csv_decoder.go index 5e5f6b80be..bc6d15ab57 100644 --- a/pkg/sink/codec/csv/csv_decoder.go +++ b/pkg/sink/codec/csv/csv_decoder.go @@ -31,6 +31,7 @@ import ( "github.com/pingcap/tidb/pkg/lightning/mydump" "github.com/pingcap/tidb/pkg/lightning/worker" "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/parser/ast" "github.com/pingcap/tidb/pkg/parser/charset" "github.com/pingcap/tidb/pkg/parser/mysql" "github.com/pingcap/tidb/pkg/types" @@ -54,6 +55,16 @@ func NewDecoder(ctx context.Context, codecConfig *common.Config, tableInfo *commonType.TableInfo, value []byte, +) (common.Decoder, error) { + return NewDecoderWithColumnSelector(ctx, codecConfig, tableInfo, value, nil) +} + +// NewDecoderWithColumnSelector creates a new BatchDecoder with a column selector. +func NewDecoderWithColumnSelector(ctx context.Context, + codecConfig *common.Config, + tableInfo *commonType.TableInfo, + value []byte, + selector commonEvent.Selector, ) (common.Decoder, error) { var backslashEscape bool @@ -78,6 +89,7 @@ func NewDecoder(ctx context.Context, if err != nil { return nil, err } + tableInfo = filterTableInfoByColumnSelector(tableInfo, selector) if codecConfig.CSVOutputFieldHeader { err := csvParser.ReadColumns() if err != nil { @@ -102,6 +114,79 @@ func NewDecoder(ctx context.Context, }, nil } +func filterTableInfoByColumnSelector( + tableInfo *commonType.TableInfo, + selector commonEvent.Selector, +) *commonType.TableInfo { + if tableInfo == nil || selector == nil { + return tableInfo + } + + columns := tableInfo.GetColumns() + selectedColumns := make([]*model.ColumnInfo, 0, len(columns)) + selectedColumnOffsets := make(map[int64]int, len(columns)) + for _, col := range columns { + if !shouldEncodeColumn(col, selector) { + continue + } + selectedCol := col.Clone() + selectedCol.Offset = len(selectedColumns) + selectedColumnOffsets[col.ID] = selectedCol.Offset + selectedColumns = append(selectedColumns, selectedCol) + } + if len(selectedColumns) == len(columns) { + return tableInfo + } + + tidbTableInfo := tableInfo.ToTiDBTableInfo().Clone() + tidbTableInfo.Columns = selectedColumns + tidbTableInfo.Indices = filterSelectedIndices(tableInfo.GetIndices(), columns, selectedColumnOffsets) + tidbTableInfo.PKIsHandle = tableInfo.PKIsHandle() && hasSelectedPKIsHandleColumn(selectedColumns) + + return commonType.NewTableInfo4Decoder(tableInfo.GetSchemaName(), tidbTableInfo) +} + +func filterSelectedIndices( + indices []*model.IndexInfo, + columns []*model.ColumnInfo, + selectedColumnOffsets map[int64]int, +) []*model.IndexInfo { + selectedIndices := make([]*model.IndexInfo, 0, len(indices)) + for _, index := range indices { + selectedIndex := index.Clone() + selectedIndex.Columns = make([]*model.IndexColumn, 0, len(index.Columns)) + allIndexColumnsSelected := true + for _, indexColumn := range index.Columns { + if indexColumn.Offset >= len(columns) { + allIndexColumnsSelected = false + break + } + selectedOffset, ok := selectedColumnOffsets[columns[indexColumn.Offset].ID] + if !ok { + allIndexColumnsSelected = false + break + } + selectedIndexColumn := indexColumn.Clone() + selectedIndexColumn.Offset = selectedOffset + selectedIndexColumn.Name = ast.NewCIStr(columns[indexColumn.Offset].Name.O) + selectedIndex.Columns = append(selectedIndex.Columns, selectedIndexColumn) + } + if allIndexColumnsSelected { + selectedIndices = append(selectedIndices, selectedIndex) + } + } + return selectedIndices +} + +func hasSelectedPKIsHandleColumn(columns []*model.ColumnInfo) bool { + for _, col := range columns { + if mysql.HasPriKeyFlag(col.GetFlag()) { + return true + } + } + return false +} + // AddKeyValue implements the Decoder interface. func (b *decoder) AddKeyValue(_, _ []byte) {} diff --git a/pkg/sink/codec/csv/csv_decoder_test.go b/pkg/sink/codec/csv/csv_decoder_test.go index d974f4ecd4..17ddfb1a33 100644 --- a/pkg/sink/codec/csv/csv_decoder_test.go +++ b/pkg/sink/codec/csv/csv_decoder_test.go @@ -16,7 +16,9 @@ import ( "context" "testing" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) @@ -56,3 +58,53 @@ func TestCSVBatchDecoder(t *testing.T) { _, hasNext := decoder.HasNext() require.False(t, hasNext) } + +func TestCSVBatchDecoderWithColumnSelector(t *testing.T) { + csvData := `"I","t","test",433305438660591626,1,"visible-value" +` + ctx := context.Background() + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + + createTableDDL := helper.DDL2Event( + "create table test.t(id int primary key, visible varchar(255), secret varchar(255))") + + selectors, err := columnselector.New(&config.SinkConfig{ + ColumnSelectors: []*config.ColumnSelector{ + {Matcher: []string{"test.t"}, Columns: []string{"id", "visible"}}, + }, + }) + require.NoError(t, err) + + codecConfig := &common.Config{ + Delimiter: ",", + Quote: "\"", + Terminator: "\n", + NullString: "\\N", + IncludeCommitTs: true, + } + decoder, err := NewDecoderWithColumnSelector( + ctx, + codecConfig, + createTableDDL.TableInfo, + []byte(csvData), + selectors.GetForTableInfo(createTableDDL.TableInfo), + ) + require.NoError(t, err) + + tp, hasNext := decoder.HasNext() + require.True(t, hasNext) + require.Equal(t, common.MessageTypeRow, tp) + + event := decoder.NextDMLMessage().ToDMLEvent() + require.Len(t, event.TableInfo.GetColumns(), 2) + require.Equal(t, "id", event.TableInfo.GetColumns()[0].Name.O) + require.Equal(t, "visible", event.TableInfo.GetColumns()[1].Name.O) + row, ok := event.GetNextRow() + require.True(t, ok) + require.Equal(t, int64(1), row.Row.GetInt64(0)) + require.Equal(t, "visible-value", string(row.Row.GetBytes(1))) + + _, hasNext = decoder.HasNext() + require.False(t, hasNext) +} diff --git a/pkg/sink/codec/csv/csv_encoder.go b/pkg/sink/codec/csv/csv_encoder.go index 0001fbd65c..4d14ebd3a8 100644 --- a/pkg/sink/codec/csv/csv_encoder.go +++ b/pkg/sink/codec/csv/csv_encoder.go @@ -39,28 +39,22 @@ func NewTxnEventEncoder(config *common.Config) common.TxnEventEncoder { } // AppendTxnEvent implements the TxnEventEncoder interface -func (b *batchEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error { +func (b *batchEncoder) AppendTxnEvent(rowEvents []*commonEvent.RowEvent) error { + if len(rowEvents) == 0 { + return nil + } if b.config.CSVOutputFieldHeader && b.batchSize == 0 { - b.setHeader(event.TableInfo) + b.setHeader(rowEvents[0].TableInfo, rowEvents[0].ColumnSelector) } - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - msg, err := rowChangedEvent2CSVMsg(b.config, &commonEvent.RowEvent{ - TableInfo: event.TableInfo, - CommitTs: event.CommitTs, - Event: row, - }) + for _, rowEvent := range rowEvents { + msg, err := rowChangedEvent2CSVMsg(b.config, rowEvent) if err != nil { return err } b.valueBuf.Write(msg.encode()) b.batchSize++ } - b.callback = event.PostFlush + b.callback = rowEvents[len(rowEvents)-1].Callback return nil } @@ -84,10 +78,13 @@ func (b *batchEncoder) Build() (messages []*common.Message) { return []*common.Message{ret} } -func (b *batchEncoder) setHeader(tableInfo *commonType.TableInfo) { +func (b *batchEncoder) setHeader(tableInfo *commonType.TableInfo, selector commonEvent.Selector) { buf := &bytes.Buffer{} colNames := make([]string, 0, len(tableInfo.GetColumns())) for _, col := range tableInfo.GetColumns() { + if !shouldEncodeColumn(col, selector) { + continue + } colNames = append(colNames, col.Name.O) } buf.Write(encodeHeader(b.config, colNames)) diff --git a/pkg/sink/codec/csv/csv_encoder_test.go b/pkg/sink/codec/csv/csv_encoder_test.go index bb51daaf6c..9482b9c868 100644 --- a/pkg/sink/codec/csv/csv_encoder_test.go +++ b/pkg/sink/codec/csv/csv_encoder_test.go @@ -17,11 +17,22 @@ import ( "strings" "testing" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" + sinkhelper "github.com/pingcap/ticdc/downstreamadapter/sink/helper" commonEvent "github.com/pingcap/ticdc/pkg/common/event" + "github.com/pingcap/ticdc/pkg/config" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/stretchr/testify/require" ) +func appendTxnEventForTest( + encoder common.TxnEventEncoder, + event *commonEvent.DMLEvent, + selector commonEvent.Selector, +) error { + return encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, selector, nil)) +} + func TestCSVBatchCodec(t *testing.T) { s := commonEvent.NewEventTestHelper(t) defer s.Close() @@ -38,7 +49,7 @@ func TestCSVBatchCodec(t *testing.T) { NullString: "\\N", IncludeCommitTs: true, }) - err := encoder.AppendTxnEvent(cs) + err := appendTxnEventForTest(encoder, cs, nil) require.Nil(t, err) messages := encoder.Build() if cs.Len() == 0 { @@ -50,7 +61,7 @@ func TestCSVBatchCodec(t *testing.T) { } } -func TestCSVAppendRowChangedEventWithCallback(t *testing.T) { +func TestCSVAppendTxnEventWithCallback(t *testing.T) { encoder := NewTxnEventEncoder(&common.Config{ Delimiter: ",", Quote: "\"", @@ -60,30 +71,28 @@ func TestCSVAppendRowChangedEventWithCallback(t *testing.T) { }) require.NotNil(t, encoder) - count := 0 - s := commonEvent.NewEventTestHelper(t) defer s.Close() s.DDL2Job("create table test.table1(col1 int primary key)") txn := s.DML2Event("test", "table1", "insert into test.table1 values (1)") - callback := func() { - count += 1 - } - txn.AddPostFlushFunc(callback) + count := 0 - // Empty build makes sure that the callback build logic not broken. + // Empty build makes sure the build path handles an empty encoder. msgs := encoder.Build() require.Len(t, msgs, 0, "no message should be built and no panic") // Append the event. - err := encoder.AppendTxnEvent(txn) + err := encoder.AppendTxnEvent(sinkhelper.NewRowEvents(txn, nil, func() { + count++ + })) require.Nil(t, err) require.Equal(t, 0, count, "nothing should be called") msgs = encoder.Build() require.Len(t, msgs, 1, "expected one message") + require.NotNil(t, msgs[0].Callback) msgs[0].Callback() - require.Equal(t, 1, count, "expected all callbacks to be called") + require.Equal(t, 1, count, "expected one callback be called") } func TestCSVBatchCodecWithHeader(t *testing.T) { @@ -103,7 +112,7 @@ func TestCSVBatchCodecWithHeader(t *testing.T) { CSVOutputFieldHeader: true, } encoder := NewTxnEventEncoder(cfg) - err := encoder.AppendTxnEvent(event) + err := appendTxnEventForTest(encoder, event, nil) require.Nil(t, err) messages := encoder.Build() require.Len(t, messages, 1) @@ -113,7 +122,7 @@ func TestCSVBatchCodecWithHeader(t *testing.T) { cfg.CSVOutputFieldHeader = false encoder = NewTxnEventEncoder(cfg) - err = encoder.AppendTxnEvent(event) + err = appendTxnEventForTest(encoder, event, nil) require.Nil(t, err) messages1 := encoder.Build() require.Len(t, messages1, 1) @@ -123,8 +132,88 @@ func TestCSVBatchCodecWithHeader(t *testing.T) { cfg.CSVOutputFieldHeader = true event.RowTypes = nil encoder = NewTxnEventEncoder(cfg) - err = encoder.AppendTxnEvent(event) + err = appendTxnEventForTest(encoder, event, nil) require.Nil(t, err) messages = encoder.Build() require.Len(t, messages, 0) } + +func TestCSVTxnEventEncoderWithColumnSelector(t *testing.T) { + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + + helper.DDL2Event("create table test.table1(col1 int primary key, col2 varchar(255))") + event := helper.DML2Event("test", "table1", `insert into test.table1 values (1, "filtered")`) + + selectors, err := columnselector.New(&config.SinkConfig{ + ColumnSelectors: []*config.ColumnSelector{ + {Matcher: []string{"test.table1"}, Columns: []string{"col1"}}, + }, + }) + require.NoError(t, err) + + cfg := &common.Config{ + Delimiter: ",", + Quote: "\"", + Terminator: "\n", + NullString: "\\N", + IncludeCommitTs: true, + CSVOutputFieldHeader: true, + } + encoder := NewTxnEventEncoder(cfg) + require.NoError(t, appendTxnEventForTest(encoder, event, selectors.GetForTableInfo(event.TableInfo))) + messages := encoder.Build() + require.Len(t, messages, 1) + require.Equal(t, "ticdc-meta$operation,ticdc-meta$table,ticdc-meta$schema,ticdc-meta$commit-ts,col1\n", string(messages[0].Key)) + require.NotContains(t, string(messages[0].Key), "col2") + require.NotContains(t, string(messages[0].Value), "filtered") +} + +func TestCSVTxnEventEncoderWithColumnSelectorForUpdateAndDelete(t *testing.T) { + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + + helper.DDL2Event("create table test.table1(id int primary key, visible varchar(255), secret varchar(255))") + updateEvent, _ := helper.DML2UpdateEvent( + "test", + "table1", + `insert into test.table1 values (1, "visible-before", "secret-before")`, + `update test.table1 set visible = "visible-after", secret = "secret-after" where id = 1`, + ) + deleteEvent := helper.DML2DeleteEvent( + "test", + "table1", + `insert into test.table1 values (2, "delete-visible", "delete-secret")`, + `delete from test.table1 where id = 2`, + ) + + selectors, err := columnselector.New(&config.SinkConfig{ + ColumnSelectors: []*config.ColumnSelector{ + {Matcher: []string{"test.table1"}, Columns: []string{"id", "visible"}}, + }, + }) + require.NoError(t, err) + + cfg := &common.Config{ + Delimiter: ",", + Quote: "\"", + Terminator: "\n", + NullString: "\\N", + OutputOldValue: true, + IncludeCommitTs: false, + } + selector := selectors.GetForTableInfo(updateEvent.TableInfo) + encoder := NewTxnEventEncoder(cfg) + require.NoError(t, appendTxnEventForTest(encoder, updateEvent, selector)) + require.NoError(t, appendTxnEventForTest(encoder, deleteEvent, selector)) + + messages := encoder.Build() + require.Len(t, messages, 1) + value := string(messages[0].Value) + require.Contains(t, value, "visible-before") + require.Contains(t, value, "visible-after") + require.Contains(t, value, "delete-visible") + require.NotContains(t, value, "secret-before") + require.NotContains(t, value, "secret-after") + require.NotContains(t, value, "delete-secret") +} diff --git a/pkg/sink/codec/csv/csv_message.go b/pkg/sink/codec/csv/csv_message.go index 9649bbfc3c..79b429d13f 100644 --- a/pkg/sink/codec/csv/csv_message.go +++ b/pkg/sink/codec/csv/csv_message.go @@ -332,6 +332,7 @@ func fromColValToCsvVal(csvConfig *common.Config, row *chunk.Row, idx int, colIn // rowChangedEvent2CSVMsg converts a RowChangedEvent to a csv record. func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMessage, error) { tableInfo := e.TableInfo + selector := e.ColumnSelector csvMsg := &csvMessage{ config: csvConfig, tableName: tableInfo.GetTargetTableName(), @@ -348,14 +349,14 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe var err error if e.IsDelete() { csvMsg.opType = operationDelete - csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo) + csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo, selector) if err != nil { return nil, err } } else if e.IsInsert() { // This is a insert operation. csvMsg.opType = operationInsert - csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo) + csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo, selector) if err != nil { return nil, err } @@ -368,12 +369,12 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe "the column length of preColumns %d doesn't equal to that of columns %d", e.GetPreRows().Len(), e.GetRows().Len()) } - csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo) + csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo, selector) if err != nil { return nil, err } } - csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo) + csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo, selector) if err != nil { return nil, err } @@ -381,13 +382,16 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe return csvMsg, nil } -func rowChangeColumns2CSVColumns(csvConfig *common.Config, row *chunk.Row, tableInfo *commonType.TableInfo) ([]any, error) { +func rowChangeColumns2CSVColumns( + csvConfig *common.Config, + row *chunk.Row, + tableInfo *commonType.TableInfo, + selector event.Selector, +) ([]any, error) { var csvColumns []any for i, col := range tableInfo.GetColumns() { - // column could be nil in a condition described in - // https://github.com/pingcap/ticdc/issues/6198#issuecomment-1191132951 - if col == nil || col.IsVirtualGenerated() { + if !shouldEncodeColumn(col, selector) { continue } @@ -402,6 +406,15 @@ func rowChangeColumns2CSVColumns(csvConfig *common.Config, row *chunk.Row, table return csvColumns, nil } +func shouldEncodeColumn(col *timodel.ColumnInfo, selector event.Selector) bool { + // column could be nil in a condition described in + // https://github.com/pingcap/ticdc/issues/6198#issuecomment-1191132951 + if col == nil || col.IsVirtualGenerated() { + return false + } + return selector == nil || selector.Select(col) +} + // The header should contain the name corresponding to the file record field, // and should have the same number as the record field. // | ticdc-meta$operation | ticdc-meta$table | ticdc-meta$schema | ticdc-meta$commit-ts | ticdc-meta$is-update | col1 | col2 | ... | diff --git a/tests/integration_tests/kafka_column_selector/conf/changefeed.toml b/tests/integration_tests/column_selector/conf/changefeed.toml similarity index 73% rename from tests/integration_tests/kafka_column_selector/conf/changefeed.toml rename to tests/integration_tests/column_selector/conf/changefeed.toml index addd3dd41a..fe5bfa1af5 100644 --- a/tests/integration_tests/kafka_column_selector/conf/changefeed.toml +++ b/tests/integration_tests/column_selector/conf/changefeed.toml @@ -1,7 +1,6 @@ [sink] column-selectors = [ {matcher = ['test.t1'], columns = ['a', 'b']}, - {matcher = ['test.*'], columns = ["*", "!b"]}, - + {matcher = ['test.*'], columns = ['*', '!b']}, {matcher = ['test1.t1'], columns = ['column*', '!column1']}, ] diff --git a/tests/integration_tests/column_selector/data/data.sql b/tests/integration_tests/column_selector/data/data.sql new file mode 100644 index 0000000000..def4ddbdf2 --- /dev/null +++ b/tests/integration_tests/column_selector/data/data.sql @@ -0,0 +1,45 @@ +drop database if exists `test`; +create database `test`; +use `test`; + +create table t1 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t1 values (1, 'keep_b_t1', 'filtered_c_t1'); +insert into t1 values (2, 'keep_b_t1_2', 'filtered_c_t1_2'); + +create table t2 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t2 values (1, 'filtered_b_t2', 'keep_c_t2'); +insert into t2 values (2, 'filtered_b_t2_2', 'keep_c_t2_2'); + +create table t3 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t3 values (1, 'filtered_b_t3', 'keep_c_t3'); +insert into t3 values (2, 'filtered_b_t3_2', 'keep_c_t3_2'); + +drop database if exists `test1`; +create database `test1`; +use `test1`; + +create table t1 ( + column0 int primary key, + column1 varchar(64), + column2 varchar(64) +); + +insert into t1 values (1, 'filtered_column1_test1', 'keep_column2_test1'); +insert into t1 values (2, 'filtered_column1_test1_2', 'keep_column2_test1_2'); + +create table finishmark(id int primary key); diff --git a/tests/integration_tests/kafka_column_selector/run.sh b/tests/integration_tests/column_selector/run.sh old mode 100755 new mode 100644 similarity index 54% rename from tests/integration_tests/kafka_column_selector/run.sh rename to tests/integration_tests/column_selector/run.sh index a7897c5432..91235725af --- a/tests/integration_tests/kafka_column_selector/run.sh +++ b/tests/integration_tests/column_selector/run.sh @@ -7,43 +7,79 @@ source $CUR/../_utils/test_prepare WORK_DIR=$OUT_DIR/$TEST_NAME CDC_BINARY=cdc.test SINK_TYPE=$1 +CHANGEFEED_ID="test" -function run() { - # test kafka sink only in this case - if [ "$SINK_TYPE" != "kafka" ]; then - return - fi - +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 +} - changefeed_id="test" - TOPIC_NAME="column-selector-test-$RANDOM" - SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&partition-num=1&enable-tidb-extension=true" - cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c ${changefeed_id} --config="$CUR/conf/changefeed.toml" - - cdc_kafka_consumer --upstream-uri $SINK_URI --downstream-uri="mysql://root@127.0.0.1:3306/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" --upstream-tidb-dsn="root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?" --config="$CUR/conf/changefeed.toml" --log-file $WORK_DIR/cdc_kafka_consumer.log 2>&1 & - - run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} - +function build_checksum_checker() { echo "Starting build checksum checker..." + local pwd=$(pwd) cd $CUR/../../utils/checksum_checker if [ ! -f ./checksum_checker ]; then GO111MODULE=on go build fi + cd $pwd +} +function run_checksum_checker() { + build_checksum_checker check_table_exists "test1.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} - ./checksum_checker --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" --databases "test,test1" --config="$CUR/conf/changefeed.toml" + $CUR/../../utils/checksum_checker/checksum_checker \ + --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" \ + --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" \ + --databases "test,test1" \ + --config="$CUR/conf/changefeed.toml" +} + +function run_kafka() { + prepare_cluster + + TOPIC_NAME="column-selector-test-$RANDOM" + SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&partition-num=1&enable-tidb-extension=true" + cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml" + + cdc_kafka_consumer --upstream-uri $SINK_URI --downstream-uri="mysql://root@127.0.0.1:3306/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" --upstream-tidb-dsn="root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?" --config="$CUR/conf/changefeed.toml" --log-file $WORK_DIR/cdc_kafka_consumer.log 2>&1 & + + run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} + run_checksum_checker cleanup_process $CDC_BINARY } +function run_storage() { + prepare_cluster + + SINK_URI="file://$WORK_DIR/storage_test?flush-interval=5s&protocol=canal-json&enable-tidb-extension=true" + cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml" + + run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} + run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml "" + run_checksum_checker +} + +function run() { + case "$SINK_TYPE" in + kafka) + run_kafka + ;; + storage) + run_storage + ;; + *) + return + ;; + esac +} + trap 'stop_test $WORK_DIR' EXIT run $* check_logs $WORK_DIR diff --git a/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml b/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml new file mode 100644 index 0000000000..836b106543 --- /dev/null +++ b/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml @@ -0,0 +1,15 @@ +[sink] +protocol = "csv" +terminator = "\n" +date-separator = "day" +column-selectors = [ + {matcher = ["test.t1"], columns = ["a", "b"]}, + {matcher = ["test.*"], columns = ["*", "!b"]}, + {matcher = ["test1.t1"], columns = ["column*", "!column1"]}, +] + +[sink.csv] +delimiter = "," +quote = '"' +null = '\N' +include-commit-ts = true diff --git a/tests/integration_tests/csv_storage_column_selector/data/data.sql b/tests/integration_tests/csv_storage_column_selector/data/data.sql new file mode 100644 index 0000000000..def4ddbdf2 --- /dev/null +++ b/tests/integration_tests/csv_storage_column_selector/data/data.sql @@ -0,0 +1,45 @@ +drop database if exists `test`; +create database `test`; +use `test`; + +create table t1 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t1 values (1, 'keep_b_t1', 'filtered_c_t1'); +insert into t1 values (2, 'keep_b_t1_2', 'filtered_c_t1_2'); + +create table t2 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t2 values (1, 'filtered_b_t2', 'keep_c_t2'); +insert into t2 values (2, 'filtered_b_t2_2', 'keep_c_t2_2'); + +create table t3 ( + a int primary key, + b varchar(64), + c varchar(64) +); + +insert into t3 values (1, 'filtered_b_t3', 'keep_c_t3'); +insert into t3 values (2, 'filtered_b_t3_2', 'keep_c_t3_2'); + +drop database if exists `test1`; +create database `test1`; +use `test1`; + +create table t1 ( + column0 int primary key, + column1 varchar(64), + column2 varchar(64) +); + +insert into t1 values (1, 'filtered_column1_test1', 'keep_column2_test1'); +insert into t1 values (2, 'filtered_column1_test1_2', 'keep_column2_test1_2'); + +create table finishmark(id int primary key); diff --git a/tests/integration_tests/csv_storage_column_selector/run.sh b/tests/integration_tests/csv_storage_column_selector/run.sh new file mode 100644 index 0000000000..7a0d028fb2 --- /dev/null +++ b/tests/integration_tests/csv_storage_column_selector/run.sh @@ -0,0 +1,59 @@ +#!/bin/bash + +set -eu + +CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +source $CUR/../_utils/test_prepare +WORK_DIR=$OUT_DIR/$TEST_NAME +CDC_BINARY=cdc.test +SINK_TYPE=$1 +CHANGEFEED_ID="test" + +function build_checksum_checker() { + echo "Starting build checksum checker..." + local pwd=$(pwd) + cd $CUR/../../utils/checksum_checker + if [ ! -f ./checksum_checker ]; then + GO111MODULE=on go build + fi + cd $pwd +} + +function run_checksum_checker() { + build_checksum_checker + check_table_exists "test1.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} + + $CUR/../../utils/checksum_checker/checksum_checker \ + --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" \ + --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" \ + --databases "test,test1" \ + --config="$CUR/conf/changefeed.toml" +} + +function run() { + if [ "$SINK_TYPE" != "storage" ]; then + return + fi + + rm -rf $WORK_DIR && 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 + + SINK_URI="file://$WORK_DIR/storage_test?flush-interval=5s&protocol=csv" + cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml" + + run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT} + run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml "" + run_checksum_checker + + cleanup_process $CDC_BINARY +} + +trap 'stop_test $WORK_DIR' EXIT +run $* +check_logs $WORK_DIR +echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>" diff --git a/tests/integration_tests/kafka_column_selector/data/data.sql b/tests/integration_tests/kafka_column_selector/data/data.sql deleted file mode 100644 index 79b4dcee0d..0000000000 --- a/tests/integration_tests/kafka_column_selector/data/data.sql +++ /dev/null @@ -1,49 +0,0 @@ -drop database if exists `test`; -create database `test`; -use `test`; - -create table t1 ( - a int primary key, - b int, - c int -); - -insert into t1 values (1, 2, 3); -insert into t1 values (2, 3, 4); -insert into t1 values (3, 4, 5); - -create table t2 ( - a int primary key, - b int, - c int -); - -insert into t2 values (1, 2, 3); -insert into t2 values (2, 3, 4); -insert into t2 values (3, 4, 5); - -create table t3 ( - a int primary key, - b int, - c int -); - -insert into t3 values (1, 2, 3); -insert into t3 values (2, 3, 4); -insert into t3 values (3, 4, 5); - -drop database if exists `test1`; -create database `test1`; -use `test1`; - -create table t1 ( - column0 int primary key, - column1 int, - column2 int -); - -insert into t1 values (1, 2, 3); -insert into t1 values (2, 3, 4); -insert into t1 values (3, 4, 5); - -create table finishmark(id int primary key); diff --git a/tests/integration_tests/run_heavy_it_in_ci.sh b/tests/integration_tests/run_heavy_it_in_ci.sh index a44f6fb87a..c8b098f09c 100755 --- a/tests/integration_tests/run_heavy_it_in_ci.sh +++ b/tests/integration_tests/run_heavy_it_in_ci.sh @@ -87,7 +87,7 @@ kafka_groups=( # G09 'kafka_simple_handle_key_only kafka_simple_handle_key_only_avro mq_sink_error_resume multi_source' # G10 - 'kafka_column_selector kafka_column_selector_avro ddl_with_random_move_table' + 'column_selector kafka_column_selector_avro ddl_with_random_move_table' # G11 'fail_over region_merge multi_changefeeds split_table_check' # G12 @@ -166,9 +166,9 @@ storage_groups=( # G13 'fail_over region_merge multi_changefeeds' # G14 - 'fail_over_ddl_mix' + 'column_selector fail_over_ddl_mix' # G15 - 'random_drop_message' + 'csv_storage_column_selector random_drop_message' ) # Source shared functions and check test coverage