planner: take logical schema producer into logical operator's hash ge…

…neration (#57323)

ref #51664

pkg/meta/model/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//pkg/parser/mysql",
         "//pkg/parser/terror",
         "//pkg/parser/types",
+        "//pkg/planner/cascades/base",
         "//pkg/util/intest",
         "@com_github_pingcap_errors//:errors",
         "@com_github_tikv_pd_client//http",

pkg/meta/model/table.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/duration"
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/planner/cascades/base"
 )
 
 // ExtraHandleID is the column ID of column which we need to append to schema to occupy the handle's position
@@ -196,6 +197,27 @@ type TableInfo struct {
 	DBID int64 `json:"-"`
 }
 
+// Hash64 implement HashEquals interface.
+func (t *TableInfo) Hash64(h base.Hasher) {
+	h.HashInt64(t.ID)
+}
+
+// Equals implements HashEquals interface.
+func (t *TableInfo) Equals(other any) bool {
+	// any(nil) can still be converted as (*TableInfo)(nil)
+	t2, ok := other.(*TableInfo)
+	if !ok {
+		return false
+	}
+	if t == nil {
+		return t2 == nil
+	}
+	if t2 == nil {
+		return false
+	}
+	return t.ID == t2.ID
+}
+
 // SepAutoInc decides whether _rowid and auto_increment id use separate allocator.
 func (t *TableInfo) SepAutoInc() bool {
 	return t.Version >= TableInfoVersion5 && t.AutoIDCache == 1

pkg/planner/core/generator/hash64_equals/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "github.com/pingcap/tidb/pkg/planner/core/generator/hash64_equals",
     visibility = ["//visibility:private"],
     deps = [
+        "//pkg/parser/types",
         "//pkg/planner/cascades/base",
         "//pkg/planner/core/operator/logicalop",
     ],

pkg/planner/core/generator/hash64_equals/hash64_equals_generator.go

@@ -23,6 +23,7 @@ import (
 	"reflect"
 	"strings"
 
+	"github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tidb/pkg/planner/cascades/base"
 	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 )
@@ -36,7 +37,7 @@ import (
 // If a field is not tagged, then it will be skipped.
 func GenHash64Equals4LogicalOps() ([]byte, error) {
 	var structures = []any{logicalop.LogicalJoin{}, logicalop.LogicalAggregation{}, logicalop.LogicalApply{},
-		logicalop.LogicalExpand{}, logicalop.LogicalLimit{}, logicalop.LogicalMaxOneRow{}}
+		logicalop.LogicalExpand{}, logicalop.LogicalLimit{}, logicalop.LogicalMaxOneRow{}, logicalop.DataSource{}}
 	c := new(cc)
 	c.write(codeGenHash64EqualsPrefix)
 	for _, s := range structures {
@@ -49,7 +50,14 @@ func GenHash64Equals4LogicalOps() ([]byte, error) {
 	return c.format()
 }
 
+// IHashEquals is the interface for hash64 and equals inside parser pkg.
+type IHashEquals interface {
+	Hash64(h types.IHasher)
+	Equals(other any) bool
+}
+
 var hashEqualsType = reflect.TypeOf((*base.HashEquals)(nil)).Elem()
+var iHashEqualsType = reflect.TypeOf((*IHashEquals)(nil)).Elem()
 
 func genHash64EqualsForLogicalOps(x any) ([]byte, error) {
 	c := new(cc)
@@ -78,9 +86,10 @@ func genHash64EqualsForLogicalOps(x any) ([]byte, error) {
 	// for Equals function.
 	c.write("// Equals implements the Hash64Equals interface, only receive *%v pointer.", vType.Name())
 	c.write("func (op *%v) Equals(other any) bool {", vType.Name())
-	c.write("if other == nil { return false }")
 	c.write("op2, ok := other.(*%v)", vType.Name())
 	c.write("if !ok { return false }")
+	c.write("if op == nil { return op2 == nil }")
+	c.write("if op2 == nil { return false }")
 	hasValidField := false
 	for i := 0; i < vType.NumField(); i++ {
 		f := vType.Field(i)
@@ -114,6 +123,8 @@ func logicalOpName2PlanCodecString(name string) string {
 		return "plancodec.TypeLimit"
 	case "LogicalMaxOneRow":
 		return "plancodec.TypeMaxOneRow"
+	case "DataSource":
+		return "plancodec.TypeDataSource"
 	default:
 		return ""
 	}
@@ -149,7 +160,8 @@ func (c *cc) EqualsElement(fType reflect.Type, lhs, rhs string, i string) {
 		reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Float32, reflect.Float64:
 		c.write("if %v != %v {return false}", lhs, rhs)
 	default:
-		if fType.Implements(hashEqualsType) || reflect.PtrTo(fType).Implements(hashEqualsType) {
+		if fType.Implements(hashEqualsType) || fType.Implements(iHashEqualsType) ||
+			reflect.PtrTo(fType).Implements(hashEqualsType) || reflect.PtrTo(fType).Implements(iHashEqualsType) {
 			if fType.Kind() == reflect.Struct {
 				rhs = "&" + rhs
 			}
@@ -183,7 +195,8 @@ func (c *cc) Hash64Element(fType reflect.Type, callName string) {
 	case reflect.Float32, reflect.Float64:
 		c.write("h.HashFloat64(float64(%v))", callName)
 	default:
-		if fType.Implements(hashEqualsType) || reflect.PtrTo(fType).Implements(hashEqualsType) {
+		if fType.Implements(hashEqualsType) || fType.Implements(iHashEqualsType) ||
+			reflect.PtrTo(fType).Implements(hashEqualsType) || reflect.PtrTo(fType).Implements(iHashEqualsType) {
 			c.write("%v.Hash64(h)", callName)
 		} else {
 			panic("doesn't support element type" + fType.Kind().String())

pkg/planner/core/operator/logicalop/logical_aggregation.go

@@ -38,7 +38,7 @@ import (
 
 // LogicalAggregation represents an aggregate plan.
 type LogicalAggregation struct {
-	LogicalSchemaProducer
+	LogicalSchemaProducer `hash64-equals:"true"`
 
 	AggFuncs     []*aggregation.AggFuncDesc `hash64-equals:"true"`
 	GroupByItems []expression.Expression    `hash64-equals:"true"`