test: fix generate_column integration test case (pingcap#797)

Signed-off-by: dongmen <414110582@qq.com>

Author: asddongmen

.github/pull_request_template.md

@@ -0,0 +1,41 @@
+<!--
+Thank you for contributing to TiCDC! 
+Please read MD's [CONTRIBUTING](https://github.com/pingcap/ticdc/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
+-->
+
+### What problem does this PR solve?
+<!--
+Please create an issue first to describe the problem.
+
+There MUST be one line starting with "Issue Number:  " and 
+linking the relevant issues via the "close" or "ref".
+
+For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
+ -->
+
+Issue Number: close #xxx
+
+### What is changed and how it works?
+
+### Check List <!--REMOVE the items that are not applicable-->
+
+#### Tests <!-- At least one of them must be included. -->
+
+- Unit test
+- Integration test
+- Manual test (add detailed scripts or steps below)
+- No code
+
+#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->
+
+##### Will it cause performance regression or break compatibility?
+
+##### Do you need to update user documentation, design documentation or monitoring documentation?
+
+### Release note <!-- bugfixes or new features need a release note -->
+
+```release-note
+Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.
+
+If you don't think this PR needs a release note then fill it with `None`.
+```

.github/workflows/check_and_build.yaml

@@ -12,6 +12,7 @@ on:
       - 'OWNERS_ALIASES'
 
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review]
     branches:
       - master
       - "release-[0-9].[0-9]*"

.github/workflows/integration_test_mysql.yaml

@@ -12,6 +12,7 @@ on:
       - 'OWNERS_ALIASES'
 
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review]
     branches:
       - master
       - "release-[0-9].[0-9]*"
@@ -30,7 +31,7 @@ jobs:
   # To boost the test speed, we split every 10 test cases into a group.
   e2e_test_group_1:
     runs-on: ubuntu-latest 
-    name: E2E Test
+    name: E2E Test Group 1
     steps:
       - name: Check out code
         uses: actions/checkout@v2
@@ -82,6 +83,12 @@ jobs:
         run: |
           export TICDC_NEWARCH=true && make integration_test CASE=foreign_key
 
+      # The 7th case in this group
+      - name: Test generate_column
+        if: ${{ success() }}
+        run: |
+          export TICDC_NEWARCH=true && make integration_test CASE=generate_column
+
       - name: Upload test logs
         if: always()
         uses: ./.github/actions/upload-test-logs

.github/workflows/uint_test.yaml

@@ -12,6 +12,7 @@ on:
       - 'OWNERS_ALIASES'
 
   pull_request:
+    types: [opened, synchronize, reopened, ready_for_review]
     branches:
       - master
       - "release-[0-9].[0-9]*"

CONTRIBUTING.md

@@ -0,0 +1,81 @@
+# How to contribute
+
+This document outlines some of the conventions on development workflow, commit
+message formatting, contact points and other resources to make it easier to get
+your contribution accepted.
+
+## Get started
+
+- Fork the repository on GitHub.
+- Read the README.md for build instructions.
+- Play with the project, submit bugs, submit patches!
+
+## Build TiCDC
+
+Developing TiCDC requires:
+
+* [Go 1.23+](https://go.dev/doc/code)
+* An internet connection to download the dependencies
+
+Simply run `make cdc` to build the program.
+
+```sh
+make cdc
+```
+
+### Run tests
+
+TODO
+
+### Update dependencies
+
+TiCDC uses [Go Modules](https://github.com/golang/go/wiki/Modules) to manage dependencies. To add or update a dependency: use the `go mod edit` command to change the dependency.
+
+## Contribution flow
+
+This is a rough outline of what a contributor's workflow looks like:
+
+- Create a topic branch from where you want to base your work. This is usually `master`.
+- Make commits of logical units and add test case if the change fixes a bug or adds new functionality.
+- Run tests and make sure all the tests are passed.
+- Make sure your commit messages are in the proper format (see below).
+- Push your changes to a topic branch in your fork of the repository.
+- Submit a pull request.
+- Your PR must receive LGTMs from two maintainers.
+
+Thanks for your contributions!
+
+### Code style
+
+The coding style suggested by the Golang community is used in TiCDC. See the [style doc](https://github.com/golang/go/wiki/CodeReviewComments) for details.
+
+Please follow this style to make TiCDC easy to review, maintain and develop.
+
+### Commit message format
+
+We follow a rough convention for commit messages that is designed to answer two
+questions: what changed and why. The subject line should feature the what and
+the body of the commit should describe the why.
+
+```shell
+maintainer: add comment for variable declaration
+
+Improve documentation.
+```
+
+The format can be described more formally as follows:
+
+```shell
+<subsystem>: <what changed>
+<BLANK LINE>
+<why this change was made>
+<BLANK LINE>
+<footer>(optional)
+```
+
+The first line is the subject and should be no longer than 70 characters, the second line is always blank, and other lines should be wrapped at 80 characters. This allows the message to be easier to read on GitHub as well as in various git tools.
+
+- If the change affects more than one subsystem, use a comma to separate them like ```maintainer,dispatcher:```.
+- If the change affects many subsystems, use ```*``` instead, like ```*:```.
+
+For the why part, if no specific reason for the change, you can use one of some generic reasons like "Improve documentation.", "Improve performance.", "Improve robustness.", "Improve test coverage."

pkg/common/table_info.go

@@ -74,6 +74,11 @@ const (
 	UnsignedFlag
 )
 
+func NewColumnFlagType(flag ColumnFlagType) *ColumnFlagType {
+	f := ColumnFlagType(flag)
+	return &f
+}
+
 // SetIsBinary sets BinaryFlag
 func (b *ColumnFlagType) SetIsBinary() {
 	(*Flag)(b).Add(Flag(BinaryFlag))

pkg/common/table_info_helper.go

@@ -454,7 +454,7 @@ func newColumnSchema(tableInfo *model.TableInfo, digest Digest) *columnSchema {
 			ID:            col.ID,
 			IsPKHandle:    pkIsHandle,
 			Ft:            col.FieldType.Clone(),
-			VirtualGenCol: col.IsGenerated(),
+			VirtualGenCol: !IsColCDCVisible(col),
 		}
 		colSchema.RowColFieldTps[col.ID] = colSchema.RowColInfos[i].Ft
 		colSchema.RowColFieldTpsSlice = append(colSchema.RowColFieldTpsSlice, colSchema.RowColInfos[i].Ft)
@@ -751,12 +751,13 @@ func (s *columnSchema) genPreSQLInsert(isReplace bool, needPlaceHolder bool) str
 		builder.WriteString("INSERT INTO %s")
 	}
 	builder.WriteString(" (")
-	builder.WriteString(s.getColumnList(false))
+	nonGeneratedColumnCount, columnList := s.getColumnList(false)
+	builder.WriteString(columnList)
 	builder.WriteString(") VALUES ")
 
 	if needPlaceHolder {
 		builder.WriteString("(")
-		builder.WriteString(placeHolder(len(s.Columns) - s.VirtualColumnCount))
+		builder.WriteString(placeHolder(nonGeneratedColumnCount))
 		builder.WriteString(")")
 	}
 	return builder.String()
@@ -766,7 +767,8 @@ func (s *columnSchema) genPreSQLUpdate() string {
 	var builder strings.Builder
 	builder.WriteString("UPDATE %s")
 	builder.WriteString(" SET ")
-	builder.WriteString(s.getColumnList(true))
+	_, columnList := s.getColumnList(true)
+	builder.WriteString(columnList)
 	return builder.String()
 }
 
@@ -784,12 +786,15 @@ func placeHolder(n int) string {
 	return builder.String()
 }
 
-func (s *columnSchema) getColumnList(isUpdate bool) string {
+// getColumnList returns non-generated columns number and column names
+func (s *columnSchema) getColumnList(isUpdate bool) (int, string) {
 	var b strings.Builder
+	nonGeneratedColumnCount := 0
 	for i, col := range s.Columns {
 		if col == nil || s.ColumnsFlag[col.ID].IsGeneratedColumn() {
 			continue
 		}
+		nonGeneratedColumnCount++
 		if i > 0 {
 			b.WriteString(",")
 		}
@@ -798,5 +803,5 @@ func (s *columnSchema) getColumnList(isUpdate bool) string {
 			b.WriteString(" = ?")
 		}
 	}
-	return b.String()
+	return nonGeneratedColumnCount, b.String()
 }

pkg/common/table_info_helper_test.go

@@ -0,0 +1,96 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/meta/model"
+	pmodel "github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestColumnSchema_GetColumnList(t *testing.T) {
+	tests := []struct {
+		name           string
+		columns        []*model.ColumnInfo
+		columnsFlag    map[int64]*ColumnFlagType
+		isUpdate       bool
+		wantCount      int
+		wantColumnList string
+	}{
+		{
+			name: "normal columns without update",
+			columns: []*model.ColumnInfo{
+				{Name: pmodel.CIStr{O: "id", L: "id"}, ID: 1},
+				{Name: pmodel.CIStr{O: "name", L: "name"}, ID: 2},
+				{Name: pmodel.CIStr{O: "age", L: "age"}, ID: 3},
+			},
+			columnsFlag: map[int64]*ColumnFlagType{
+				1: NewColumnFlagType(PrimaryKeyFlag),
+				2: NewColumnFlagType(UniqueKeyFlag),
+				3: NewColumnFlagType(NullableFlag),
+			},
+			isUpdate:       false,
+			wantCount:      3,
+			wantColumnList: "`id`,`name`,`age`",
+		},
+		{
+			name: "normal columns with update",
+			columns: []*model.ColumnInfo{
+				{Name: pmodel.CIStr{O: "id", L: "id"}, ID: 1},
+				{Name: pmodel.CIStr{O: "name", L: "name"}, ID: 2},
+				{Name: pmodel.CIStr{O: "age", L: "age"}, ID: 3},
+			},
+			columnsFlag: map[int64]*ColumnFlagType{
+				1: NewColumnFlagType(PrimaryKeyFlag),
+				2: NewColumnFlagType(UniqueKeyFlag),
+				3: NewColumnFlagType(NullableFlag),
+			},
+			isUpdate:       true,
+			wantCount:      3,
+			wantColumnList: "`id` = ?,`name` = ?,`age` = ?",
+		},
+		{
+			name: "with generated columns",
+			columns: []*model.ColumnInfo{
+				{Name: pmodel.CIStr{O: "id", L: "id"}, ID: 1},
+				{Name: pmodel.CIStr{O: "name", L: "name"}, ID: 2},
+				{Name: pmodel.CIStr{O: "full_name", L: "full_name"}, ID: 3}, // generated column
+			},
+			columnsFlag: map[int64]*ColumnFlagType{
+				1: NewColumnFlagType(PrimaryKeyFlag),
+				2: NewColumnFlagType(UniqueKeyFlag),
+				3: NewColumnFlagType(GeneratedColumnFlag),
+			},
+			isUpdate:       false,
+			wantCount:      2,
+			wantColumnList: "`id`,`name`",
+		},
+		{
+			name: "with nil column",
+			columns: []*model.ColumnInfo{
+				{Name: pmodel.CIStr{O: "id", L: "id"}, ID: 1},
+				nil,
+				{Name: pmodel.CIStr{O: "age", L: "age"}, ID: 3},
+			},
+			columnsFlag: map[int64]*ColumnFlagType{
+				1: NewColumnFlagType(PrimaryKeyFlag),
+				3: NewColumnFlagType(NullableFlag),
+			},
+			isUpdate:       false,
+			wantCount:      2,
+			wantColumnList: "`id`,`age`",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &columnSchema{
+				Columns:     tt.columns,
+				ColumnsFlag: tt.columnsFlag,
+			}
+			gotCount, gotColumnList := s.getColumnList(tt.isUpdate)
+			require.Equal(t, tt.wantCount, gotCount)
+			require.Equal(t, tt.wantColumnList, gotColumnList)
+		})
+	}
+}

pkg/sink/mysql/mysql_writer.go

@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/retry"
 	pmysql "github.com/pingcap/tiflow/pkg/sink/mysql"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 const (
@@ -748,6 +749,12 @@ func (w *MysqlWriter) prepareDMLs(events []*commonEvent.DMLEvent) (*preparedDMLs
 		}
 	}
 
+	// Pre-check log level to avoid dmls.String() being called unnecessarily
+	// This method is expensive, so we only log it when the log level is debug.
+	if log.GetLevel() == zapcore.DebugLevel {
+		log.Debug("prepareDMLs", zap.Any("dmls", dmls.String()), zap.Any("events", events))
+	}
+
 	return dmls, nil
 }
 

pkg/sink/mysql/sql_builder.go

@@ -14,6 +14,7 @@
 package mysql
 
 import (
+	"fmt"
 	"strings"
 	"sync"
 
@@ -33,6 +34,19 @@ type preparedDMLs struct {
 	startTs         []uint64
 }
 
+func (d *preparedDMLs) String() string {
+	return fmt.Sprintf("sqls: %v, values: %v, rowCount: %d, approximateSize: %d, startTs: %v", d.fmtSqls(), d.values, d.rowCount, d.approximateSize, d.startTs)
+}
+
+func (d *preparedDMLs) fmtSqls() string {
+	builder := strings.Builder{}
+	for _, sql := range d.sqls {
+		builder.WriteString(sql)
+		builder.WriteString(";")
+	}
+	return builder.String()
+}
+
 var dmlsPool = sync.Pool{
 	New: func() interface{} {
 		return &preparedDMLs{

tests/integration_tests/generate_column/data/virtual.sql

@@ -2,13 +2,6 @@ drop database if exists `generate_column`;
 create database `generate_column`;
 use `generate_column`;
 
-create table t (a int, b int as (a + 1) stored primary key);
-insert into t(a) values (1),(2), (3),(4),(5),(6),(7);
-update t set a = 10 where a = 1;
-update t set a = 11 where b = 3;
-delete from t where b=4;
-delete from t where a=4;
-
 create table t1 (a int, b int as (a + 1) virtual not null, c int not null, unique index idx1(b), unique index idx2(c));
 insert into t1 (a, c) values (1, 2),(2, 3), (3, 4),(4, 5),(5, 6),(6, 7),(7, 8);
 update t1 set a = 10 where a = 1;

utils/dynstream/parallel_dynamic_stream.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/pingcap/log"
 	. "github.com/pingcap/ticdc/pkg/apperror"
-	"go.uber.org/zap"
 )
 
 // Use a hasher to select target stream for the path.
@@ -93,14 +92,6 @@ func (s *parallelDynamicStream[A, P, T, D, H]) Push(path P, e T) {
 		timestamp: s.handler.GetTimestamp(e),
 		queueTime: time.Now(),
 	}
-	if s.memControl != nil {
-		log.Debug("fizz: add event to stream",
-			zap.Any("rawSize", s.handler.GetSize(e)),
-			zap.Any("extraSize", s.eventExtraSize),
-			zap.Any("totalSize", s.eventExtraSize+s.handler.GetSize(e)),
-			zap.Any("event", e),
-		)
-	}
 	pi.stream.in() <- ew
 }