expression: check timezone when encoding timestamp datum

expression/distsql_builtin.go

@@ -557,7 +557,7 @@ func convertTime(data []byte, ftPB *tipb.FieldType, tz *time.Location) (*Constan
 	if err != nil {
 		return nil, err
 	}
-	if ft.Tp == mysql.TypeTimestamp && !t.IsZero() {
+	if ft.Tp == mysql.TypeTimestamp && tz != time.UTC {
 		err = t.ConvertTimeZone(time.UTC, tz)
 		if err != nil {
 			return nil, err

expression/expr_to_pb.go

@@ -15,6 +15,7 @@ package expression
 
 import (
 	"context"
+	"time"
 
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/charset"
@@ -106,7 +107,7 @@ func (pc PbConverter) conOrCorColToPBExpr(expr Expression) *tipb.Expr {
 		logutil.Logger(context.Background()).Error("eval constant or correlated column", zap.String("expression", expr.ExplainInfo()), zap.Error(err))
 		return nil
 	}
-	tp, val, ok := pc.encodeDatum(d)
+	tp, val, ok := pc.encodeDatum(d, ft)
 	if !ok {
 		return nil
 	}
@@ -117,7 +118,7 @@ func (pc PbConverter) conOrCorColToPBExpr(expr Expression) *tipb.Expr {
 	return &tipb.Expr{Tp: tp, Val: val, FieldType: ToPBFieldType(ft)}
 }
 
-func (pc *PbConverter) encodeDatum(d types.Datum) (tipb.ExprType, []byte, bool) {
+func (pc *PbConverter) encodeDatum(d types.Datum, ft *types.FieldType) (tipb.ExprType, []byte, bool) {
 	var (
 		tp  tipb.ExprType
 		val []byte
@@ -158,6 +159,14 @@ func (pc *PbConverter) encodeDatum(d types.Datum) (tipb.ExprType, []byte, bool)
 		if pc.client.IsRequestTypeSupported(kv.ReqTypeDAG, int64(tipb.ExprType_MysqlTime)) {
 			tp = tipb.ExprType_MysqlTime
 			t := d.GetMysqlTime()
+			// To be compatible with `encode` and `PBToExpr`, convert timestamp to UTC timezone.
+			if ft.Tp == mysql.TypeTimestamp && pc.sc.TimeZone != time.UTC {
+				err := t.ConvertTimeZone(pc.sc.TimeZone, time.UTC)
+				if err != nil {
+					logutil.Logger(context.Background()).Error("encode mysql time", zap.Error(err))
+					return tp, nil, false
+				}
+			}
 			v, err := t.ToPackedUint()
 			if err != nil {
 				logutil.Logger(context.Background()).Error("encode mysql time", zap.Error(err))

expression/integration_test.go

@@ -4235,3 +4235,17 @@ where
     datediff(b.date8, date(from_unixtime(a.starttime))) >= 0`
 	tk.MustQuery(q)
 }
+
+func (s *testIntegrationSuite) TestTimestampDatumEncode(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec(`drop table if exists t;`)
+	tk.MustExec(`create table t (a bigint primary key, b timestamp)`)
+	tk.MustExec(`insert into t values (1, "2019-04-29 11:56:12")`)
+	tk.MustQuery(`explain select * from t where b = (select max(b) from t)`).Check(testkit.Rows(
+		"TableReader_43 10.00 root data:Selection_42",
+		"└─Selection_42 10.00 cop eq(test.t.b, 2019-04-29 11:56:12)",
+		"  └─TableScan_41 10000.00 cop table:t, range:[-inf,+inf], keep order:false, stats:pseudo",
+	))
+	tk.MustQuery(`select * from t where b = (select max(b) from t)`).Check(testkit.Rows(`1 2019-04-29 11:56:12`))
+}