Function runTestInSchemaState

pkg/server/internal/testserverclient/server_client.go:2997–3094

(
	t *testing.T,
	conn *sql.Conn,
	cli *TestServerClient,
	state model.SchemaState,
	isOnJobUpdated bool,
	dropColumnSQL string,
	sqlWithErrs []sqlWithErr,
	expectQuery *expectQuery,
)

}

func runTestInSchemaState(
	t *testing.T,
	conn *sql.Conn,
	cli *TestServerClient,
	state model.SchemaState,
	isOnJobUpdated bool,
	dropColumnSQL string,
	sqlWithErrs []sqlWithErr,
	expectQuery *expectQuery,
) {
	ctx := context.Background()
	MustExec(ctx, t, conn, "use test_db_state")

	prevState := model.StateNone
	var checkErr error
	dbt := cli.getNewDB(t, func(config *mysql.Config) {
		config.MaxAllowedPacket = 1024
	})
	conn1, err := dbt.GetDB().Conn(ctx)
	require.NoError(t, err)
	defer func() {
		err := dbt.GetDB().Close()
		require.NoError(t, err)
	}()
	MustExec(ctx, t, conn1, "use test_db_state")

	for i, sqlWithErr := range sqlWithErrs {
		// Start the test txn.
		// Step 1: begin(when i = 0).
		if i == 0 || i == len(sqlWithErrs)-1 {
			sqlWithErr := sqlWithErrs[i]
			MustExec(ctx, t, conn1, sqlWithErr.sql)
		} else {
			// Step 2: prepare stmts.
			// SELECT a, c, d from stock where (a, b) IN ((?, ?),(?, ?)) FOR UPDATE
			// UPDATE stock SET c = ? WHERE a= ? AND b = 'a'
			// UPDATE stock SET c = ?, d = 'z' WHERE a= ? AND b = 'b'
			stmt, err := conn1.PrepareContext(ctx, sqlWithErr.sql)
			require.NoError(t, err)
			sqlWithErr.stmt = stmt
			sqlWithErrs[i] = sqlWithErr
		}
	}

	// Step 3: begin.
	sqlWithErr := sqlWithErrs[0]
	MustExec(ctx, t, conn1, sqlWithErr.sql)

	prevState = model.StateNone
	state = model.StateWriteOnly
	cbFunc1 := func(job *model.Job) {
		if jobStateOrLastSubJobState(job) == prevState || checkErr != nil {
			return
		}
		prevState = jobStateOrLastSubJobState(job)
		if prevState != state {
			return
		}