*: add tidb_row_checksum() as a builtin function (#43479)

ref #42747

executor/showtest/show_test.go

@@ -1524,7 +1524,7 @@ func TestShowBuiltin(t *testing.T) {
  res := tk.MustQuery("show builtins;")
  require.NotNil(t, res)
  rows := res.Rows()
- const builtinFuncNum = 287
+ const builtinFuncNum = 288
  require.Equal(t, builtinFuncNum, len(rows))
  require.Equal(t, rows[0][0].(string), "abs")
  require.Equal(t, rows[builtinFuncNum-1][0].(string), "yearweek")

expression/builtin.go

@@ -799,6 +799,7 @@ var funcs = map[string]functionClass{
  ast.UUIDToBin: &uuidToBinFunctionClass{baseFunctionClass{ast.UUIDToBin, 1, 2}},
  ast.BinToUUID: &binToUUIDFunctionClass{baseFunctionClass{ast.BinToUUID, 1, 2}},
  ast.TiDBShard: &tidbShardFunctionClass{baseFunctionClass{ast.TiDBShard, 1, 1}},
+ ast.TiDBRowChecksum: &tidbRowChecksumFunctionClass{baseFunctionClass{ast.TiDBRowChecksum, 0, 0}},
 
  ast.GetLock: &lockFunctionClass{baseFunctionClass{ast.GetLock, 2, 2}},
  ast.ReleaseLock: &releaseLockFunctionClass{baseFunctionClass{ast.ReleaseLock, 1, 1}},

expression/builtin_miscellaneous.go

@@ -1467,3 +1467,11 @@ func (b *builtinTidbShardSig) evalInt(row chunk.Row) (int64, bool, error) {
  hashed = hashed % tidbShardBucketCount
  return int64(hashed), false, nil
 }
+
+type tidbRowChecksumFunctionClass struct {
+ baseFunctionClass
+}
+
+func (c *tidbRowChecksumFunctionClass) getFunction(ctx sessionctx.Context, args []Expression) (builtinFunc, error) {
+ return nil, ErrNotSupportedYet.GenWithStack("FUNCTION tidb_row_checksum can only be used as a select field in a fast point plan")
+}

expression/integration_serial_test/BUILD.bazel

@@ -11,6 +11,7 @@ go_test(
  shard_count = 50,
  deps = [
  "//config",
+ "//expression",
  "//parser/mysql",
  "//parser/terror",
  "//planner/core",

expression/integration_serial_test/integration_serial_test.go

@@ -16,14 +16,17 @@ package integration_serial_test
 
 import (
  "context"
+ "encoding/binary"
  "fmt"
+ "hash/crc32"
  "math"
  "strings"
  "testing"
  "time"
 
  "github.com/pingcap/errors"
  "github.com/pingcap/failpoint"
+ "github.com/pingcap/tidb/expression"
  "github.com/pingcap/tidb/parser/mysql"
  "github.com/pingcap/tidb/parser/terror"
  plannercore "github.com/pingcap/tidb/planner/core"
@@ -4414,3 +4417,57 @@ func TestPartitionPruningRelaxOP(t *testing.T) {
  tk.MustQuery("SELECT COUNT(*) FROM t1 WHERE d < '2018-01-01'").Check(testkit.Rows("6"))
  tk.MustQuery("SELECT COUNT(*) FROM t1 WHERE d > '2018-01-01'").Check(testkit.Rows("12"))
 }
+
+func TestTiDBRowChecksumBuiltin(t *testing.T) {
+ store := testkit.CreateMockStore(t)
+
+ checksum := func(cols ...interface{}) uint32 {
+ buf := make([]byte, 0, 64)
+ for _, col := range cols {
+ switch x := col.(type) {
+ case int:
+ buf = binary.LittleEndian.AppendUint64(buf, uint64(x))
+ case string:
+ buf = binary.LittleEndian.AppendUint32(buf, uint32(len(x)))
+ buf = append(buf, []byte(x)...)
+ }
+ }
+ return crc32.ChecksumIEEE(buf)
+ }
+
+ tk := testkit.NewTestKit(t, store)
+ tk.MustExec("set global tidb_enable_row_level_checksum = 1")
+ tk.MustExec("use test")
+ tk.MustExec("create table t (id int primary key, c int)")
+
+ // row with 2 checksums
+ tk.MustExec("insert into t values (1, 10)")
+ tk.MustExec("alter table t change column c c varchar(10)")
+ checksum1 := fmt.Sprintf("%d,%d", checksum(1, 10), checksum(1, "10"))
+ // row with 1 checksum
+ tk.Session().GetSessionVars().EnableRowLevelChecksum = true
+ tk.MustExec("insert into t values (2, '20')")
+ checksum2 := fmt.Sprintf("%d", checksum(2, "20"))
+ // row without checksum
+ tk.Session().GetSessionVars().EnableRowLevelChecksum = false
+ tk.MustExec("insert into t values (3, '30')")
+ checksum3 := "<nil>"
+
+ // fast point-get
+ tk.MustQuery("select tidb_row_checksum() from t where id = 1").Check(testkit.Rows(checksum1))
+ tk.MustQuery("select tidb_row_checksum() from t where id = 2").Check(testkit.Rows(checksum2))
+ tk.MustQuery("select tidb_row_checksum() from t where id = 3").Check(testkit.Rows(checksum3))
+ // fast batch-point-get
+ tk.MustQuery("select tidb_row_checksum() from t where id in (1, 2, 3)").Check(testkit.Rows(checksum1, checksum2, checksum3))
+
+ // non-fast point-get
+ tk.MustGetDBError("select length(tidb_row_checksum()) from t where id = 1", expression.ErrNotSupportedYet)
+ tk.MustGetDBError("select c from t where id = 1 and tidb_row_checksum() is not null", expression.ErrNotSupportedYet)
+ // non-fast batch-point-get
+ tk.MustGetDBError("select length(tidb_row_checksum()) from t where id in (1, 2, 3)", expression.ErrNotSupportedYet)
+ tk.MustGetDBError("select c from t where id in (1, 2, 3) and tidb_row_checksum() is not null", expression.ErrNotSupportedYet)
+
+ // other plans
+ tk.MustGetDBError("select tidb_row_checksum() from t", expression.ErrNotSupportedYet)
+ tk.MustGetDBError("select tidb_row_checksum() from t where id > 0", expression.ErrNotSupportedYet)
+}

parser/ast/functions.go

@@ -298,6 +298,7 @@ const (
  BinToUUID = "bin_to_uuid"
  VitessHash = "vitess_hash"
  TiDBShard = "tidb_shard"
+ TiDBRowChecksum = "tidb_row_checksum"
  GetLock = "get_lock"
  ReleaseLock = "release_lock"
 

parser/model/model.go

@@ -387,6 +387,9 @@ const ExtraPidColID = -2
 // Must be after ExtraPidColID!
 const ExtraPhysTblID = -3
 
+// ExtraRowChecksumID is the column ID of column which holds the row checksum info.
+const ExtraRowChecksumID = -4
+
 const (
  // TableInfoVersion0 means the table info version is 0.
  // Upgrade from v2.1.1 or v2.1.2 to v2.1.3 and later, and then execute a "change/modify column" statement

planner/core/point_get_plan.go

@@ -1264,6 +1264,11 @@ func buildSchemaFromFields(
  }
  continue
  }
+ if name, column, ok := tryExtractRowChecksumColumn(field, len(columns)); ok {
+ names = append(names, name)
+ columns = append(columns, column)
+ continue
+ }
  colNameExpr, ok := field.Expr.(*ast.ColumnNameExpr)
  if !ok {
  return nil, nil
@@ -1305,6 +1310,38 @@ func buildSchemaFromFields(
  return schema, names
 }
 
+func tryExtractRowChecksumColumn(field *ast.SelectField, idx int) (*types.FieldName, *expression.Column, bool) {
+ f, ok := field.Expr.(*ast.FuncCallExpr)
+ if !ok || f.FnName.L != ast.TiDBRowChecksum || len(f.Args) != 0 {
+ return nil, nil, false
+ }
+ origName := f.FnName
+ origName.L += "()"
+ origName.O += "()"
+ asName := origName
+ if field.AsName.L != "" {
+ asName = field.AsName
+ }
+ cs, cl := types.DefaultCharsetForType(mysql.TypeString)
+ ftype := ptypes.NewFieldType(mysql.TypeString)
+ ftype.SetCharset(cs)
+ ftype.SetCollate(cl)
+ ftype.SetFlen(mysql.MaxBlobWidth)
+ ftype.SetDecimal(0)
+ name := &types.FieldName{
+ OrigColName: origName,
+ ColName: asName,
+ }
+ column := &expression.Column{
+ RetType: ftype,
+ ID: model.ExtraRowChecksumID,
+ UniqueID: model.ExtraRowChecksumID,
+ Index: idx,
+ OrigName: origName.L,
+ }
+ return name, column, true
+}
+
 // getSingleTableNameAndAlias return the ast node of queried table name and the alias string.
 // `tblName` is `nil` if there are multiple tables in the query.
 // `tblAlias` will be the real table name if there is no table alias in the query.

util/rowcodec/decoder.go

@@ -210,6 +210,14 @@ func (decoder *ChunkDecoder) DecodeToChunk(rowData []byte, handle kv.Handle, chk
  chk.AppendNull(colIdx)
  continue
  }
+ if col.ID == model.ExtraRowChecksumID {
+ if v := decoder.row.getChecksumInfo(); len(v) > 0 {
+ chk.AppendString(colIdx, v)
+ } else {
+ chk.AppendNull(colIdx)
+ }
+ continue
+ }
 
  idx, isNil, notFound := decoder.row.findColID(col.ID)
  if !notFound && !isNil {

util/rowcodec/row.go

@@ -16,6 +16,7 @@ package rowcodec
 
 import (
  "encoding/binary"
+ "strconv"
 )
 
 const (
@@ -100,6 +101,17 @@ func (r *row) setChecksums(checksums ...uint32) {
  }
 }
 
+func (r *row) getChecksumInfo() string {
+ var s string
+ if r.hasChecksum() {
+ s = strconv.FormatUint(uint64(r.checksum1), 10)
+ if r.hasExtraChecksum() {
+ s += "," + strconv.FormatUint(uint64(r.checksum2), 10)
+ }
+ }
+ return s
+}
+
 func (r *row) getData(i int) []byte {
  var start, end uint32
  if r.large() {