sink(ticdc): Add DDL query into schema.json file (#7834) (#7848)

ref #6797, close #7847
pingcap · Dec 8, 2022 · 3693dc1 · 3693dc1
1 parent 075dab2
commit 3693dc1
Show file tree

Hide file tree

Showing 8 changed files with 333 additions and 110 deletions.
diff --git a/cdc/model/schema_storage.go b/cdc/model/schema_storage.go
@@ -66,9 +66,10 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 		TableInfo: info,
 		SchemaID:  schemaID,
 		TableName: TableName{
-			Schema:  schemaName,
-			Table:   info.Name.O,
-			TableID: info.ID,
+			Schema:      schemaName,
+			Table:       info.Name.O,
+			TableID:     info.ID,
+			IsPartition: info.GetPartitionInfo() != nil,
 		},
 		Version:          version,
 		columnsOffset:    make(map[int64]int, len(info.Columns)),

diff --git a/cdc/sinkv2/ddlsink/cloudstorage/cloud_storage_ddl_sink.go b/cdc/sinkv2/ddlsink/cloudstorage/cloud_storage_ddl_sink.go
@@ -68,24 +68,24 @@ func NewCloudStorageDDLSink(ctx context.Context, sinkURI *url.URL) (*ddlSink, er
 	return d, nil
 }
 
-func (d *ddlSink) generateSchemaPath(def cloudstorage.TableDetail) string {
-	return fmt.Sprintf("%s/%s/%d/schema.json", def.Schema, def.Table, def.Version)
+func generateSchemaPath(def cloudstorage.TableDefinition) string {
+	return fmt.Sprintf("%s/%s/%d/schema.json", def.Schema, def.Table, def.TableVersion)
 }
 
 func (d *ddlSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error {
-	var def cloudstorage.TableDetail
+	var def cloudstorage.TableDefinition
 
 	if ddl.TableInfo.TableInfo == nil {
 		return nil
 	}
 
-	def.FromTableInfo(ddl.TableInfo)
+	def.FromDDLEvent(ddl)
 	encodedDef, err := json.MarshalIndent(def, "", "    ")
 	if err != nil {
 		return errors.Trace(err)
 	}
 
-	path := d.generateSchemaPath(def)
+	path := generateSchemaPath(def)
 	err = d.statistics.RecordDDLExecution(func() error {
 		err1 := d.storage.WriteFile(ctx, path, encodedDef)
 		if err1 != nil {

diff --git a/cdc/sinkv2/ddlsink/cloudstorage/cloud_storage_ddl_sink_test.go b/cdc/sinkv2/ddlsink/cloudstorage/cloud_storage_ddl_sink_test.go
@@ -39,6 +39,8 @@ func TestWriteDDLEvent(t *testing.T) {
 	require.Nil(t, err)
 
 	ddlEvent := &model.DDLEvent{
+		Type:  timodel.ActionAddColumn,
+		Query: "alter table test.table1 add col2 varchar(64)",
 		TableInfo: &model.TableInfo{
 			Version: 100,
 			TableName: model.TableName{
@@ -70,7 +72,10 @@ func TestWriteDDLEvent(t *testing.T) {
 	require.JSONEq(t, `{
 		"Table": "table1",
 		"Schema": "test",
-		"Version": 100,
+		"Version": 1,
+		"TableVersion": 100,
+		"Query": "alter table test.table1 add col2 varchar(64)",
+		"Type": 5,
 		"TableColumns": [
 			{
 				"ColumnName": "col1",