MoveTables: adjust datetimes when importing from non-UTC sources into…

… UTC targets (#10102)

* Add e2e test case for source and target in different time zones

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Added source_time_zone option to MoveTables v2, pass it through to table plan builder and convert datetime columns if option is provided. e2e test demonstrates this works. VDiff needs to be modified

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Add time zone support in VDiff. Improve tests.

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Add vdiff unit test

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Add SourceTimeZone to Workflow Show

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Update workflow test for added attribute SourceTimeZone

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Support SourceTimeZone in reverse workflows, update e2e test

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Update unit tests

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Refactor vdiff timezone-related changes

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Address review comment nits

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Fix: lower col name also for time zone cols

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Use field type for type comparison instead of the string representation

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Check if current primaries have time zone loaded

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Address review comments

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

* Use same date for conversion and as date format

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

go/test/endtoend/vreplication/config.go

@@ -29,6 +29,7 @@ create table ` + "`Lead`(`Lead-id`" + ` binary(16), name varbinary(16), date1 da
 create table ` + "`Lead-1`(`Lead`" + ` binary(16), name varbinary(16), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key (` + "`Lead`" + `));
 create table _vt_PURGE_4f9194b43b2011eb8a0104ed332e05c2_20221210194431(id int, val varbinary(128), primary key(id));
 create table db_order_test (c_uuid varchar(64) not null default '', created_at datetime not null, dstuff varchar(128), dtstuff text, dbstuff blob, cstuff char(32), primary key (c_uuid,created_at)) CHARSET=utf8mb4;
+create table datze (id int, dt1 datetime not null default current_timestamp, dt2 datetime not null, ts1 timestamp default current_timestamp, primary key (id));
 `
 
 	// These should always be ignored in vreplication
@@ -60,7 +61,8 @@ create table db_order_test (c_uuid varchar(64) not null default '', created_at d
 	},
 	"Lead": {},
 	"Lead-1": {},
-	"db_order_test": {}
+	"db_order_test": {},
+	"datze": {}
   }
 }
 `
@@ -127,6 +129,14 @@ create table db_order_test (c_uuid varchar(64) not null default '', created_at d
           "name": "xxhash"
         }
       ]
+    },
+    "datze": {
+      "column_vindexes": [
+        {
+          "column": "id",
+          "name": "reverse_bits"
+        }
+      ]
     }
   }
 }

go/test/endtoend/vreplication/time_zone_test.go

@@ -0,0 +1,200 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package vreplication
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"vitess.io/vitess/go/test/endtoend/cluster"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/vt/log"
+)
+
+// TestMoveTablesTZ tests the conversion of datetime based on the source timezone passed to the MoveTables workflow
+func TestMoveTablesTZ(t *testing.T) {
+	allCellNames = "zone1"
+	defaultCellName := "zone1"
+	workflow := "tz"
+	sourceKs := "product"
+	targetKs := "customer"
+	ksWorkflow := fmt.Sprintf("%s.%s", targetKs, workflow)
+	ksReverseWorkflow := fmt.Sprintf("%s.%s_reverse", sourceKs, workflow)
+
+	vc = NewVitessCluster(t, "TestCellAliasVreplicationWorkflow", []string{"zone1"}, mainClusterConfig)
+	require.NotNil(t, vc)
+	defaultCell = vc.Cells[defaultCellName]
+	cells := []*Cell{defaultCell}
+
+	defer vc.TearDown(t)
+
+	cell1 := vc.Cells["zone1"]
+	vc.AddKeyspace(t, []*Cell{cell1}, sourceKs, "0", initialProductVSchema, initialProductSchema, 0, 0, 100, sourceKsOpts)
+
+	vtgate = cell1.Vtgates[0]
+	require.NotNil(t, vtgate)
+	vtgate.WaitForStatusOfTabletInShard(fmt.Sprintf("%s.%s.primary", "product", "0"), 1)
+
+	vtgateConn = getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
+	defer vtgateConn.Close()
+	verifyClusterHealth(t, vc)
+
+	productTab := vc.Cells[defaultCell.Name].Keyspaces[sourceKs].Shards["0"].Tablets["zone1-100"].Vttablet
+	timeZoneSQLBytes, _ := os.ReadFile("tz.sql")
+	timeZoneSQL := string(timeZoneSQLBytes)
+
+	// it seems to take some time for the mysql server to load time zone info after the tables in mysql db have been populated
+	loadTimeZoneInfo := func(tab *cluster.VttabletProcess, sql, timezone string) {
+		_, err := tab.QueryTabletWithDB(timeZoneSQL, "mysql")
+		require.NoError(t, err)
+		timer := time.NewTimer(1 * time.Minute)
+		for {
+			select {
+			case <-timer.C:
+				require.Fail(t, "could not load time zone info")
+			default:
+			}
+			_, err = tab.QueryTablet(fmt.Sprintf("SET GLOBAL time_zone = '%s';", timezone), "", false)
+			if err == nil {
+				timer.Stop()
+				return
+			}
+			time.Sleep(100 * time.Millisecond)
+		}
+	}
+	loadTimeZoneInfo(productTab, timeZoneSQL, "US/Pacific")
+
+	insertInitialData(t)
+
+	if _, err := vc.AddKeyspace(t, cells, targetKs, "0", customerVSchema, customerSchema, defaultReplicas, defaultRdonly, 200, targetKsOpts); err != nil {
+		t.Fatal(err)
+	}
+	if err := vtgate.WaitForStatusOfTabletInShard(fmt.Sprintf("%s.%s.primary", "customer", "0"), 1); err != nil {
+		t.Fatal(err)
+	}
+
+	defaultCell := vc.Cells["zone1"]
+	custKs := vc.Cells[defaultCell.Name].Keyspaces[targetKs]
+	customerTab := custKs.Shards["0"].Tablets["zone1-200"].Vttablet
+
+	loadTimeZoneInfo(customerTab, timeZoneSQL, "UTC")
+
+	tables := "datze"
+
+	ksErrorWorkflow := fmt.Sprintf("%s.%s", targetKs, "tzerr")
+	output, err := vc.VtctlClient.ExecuteCommandWithOutput("MoveTables", "--", "--source", sourceKs, "--tables",
+		tables, "--source_time_zone", "US/Pacifik", "Create", ksErrorWorkflow)
+	require.Error(t, err, output)
+	require.Contains(t, output, "time zone is invalid")
+
+	output, err = vc.VtctlClient.ExecuteCommandWithOutput("MoveTables", "--", "--source", sourceKs, "--tables",
+		tables, "--source_time_zone", "US/Pacific", "Create", ksWorkflow)
+	require.NoError(t, err, output)
+
+	catchup(t, customerTab, workflow, "MoveTables")
+
+	// inserts to test date conversions in replication (vplayer) mode
+	_, err = vtgateConn.ExecuteFetch("insert into datze(id, dt2) values (11, '2022-01-01 10:20:30')", 1, false) // standard time
+	require.NoError(t, err)
+	_, err = vtgateConn.ExecuteFetch("insert into datze(id, dt2) values (12, '2022-04-01 5:06:07')", 1, false) // dst
+	require.NoError(t, err)
+
+	vdiff(t, ksWorkflow, "")
+
+	query := "select * from datze"
+	qrSourceUSPacific, err := productTab.QueryTablet(query, sourceKs, true)
+	require.NoError(t, err)
+	require.NotNil(t, qrSourceUSPacific)
+
+	qrTargetUTC, err := customerTab.QueryTablet(query, targetKs, true)
+	require.NoError(t, err)
+	require.NotNil(t, qrTargetUTC)
+
+	require.Equal(t, len(qrSourceUSPacific.Rows), len(qrTargetUTC.Rows))
+
+	pacificLocation, err := time.LoadLocation("US/Pacific")
+	require.NoError(t, err)
+
+	// for reference the columns in the test are as follows:
+	//  * dt1 datetime default current_timestamp, constant for all rows
+	//  * dt2 datetime, different values. First row is in standard time, rest with daylight savings including times around the time zone switch
+	//  * ts1 timestamp default current_timestamp, constant for all rows
+	for i, row := range qrSourceUSPacific.Named().Rows {
+		// source and UTC results must differ since source is in US/Pacific
+		require.NotEqual(t, row.AsString("dt1", ""), qrTargetUTC.Named().Rows[i].AsString("dt1", ""))
+		require.NotEqual(t, row.AsString("dt2", ""), qrTargetUTC.Named().Rows[i].AsString("dt2", ""))
+		require.NotEqual(t, row.AsString("ts1", ""), qrTargetUTC.Named().Rows[i].AsString("ts1", ""))
+
+		dtLayout := "2006-01-02 15:04:05"
+		// now compare times b/w source and target (actual). VDiff has already compared, but we want to validate that vdiff is right too!
+		dt2a, err := time.Parse(dtLayout, qrTargetUTC.Named().Rows[i].AsString("dt2", ""))
+		require.NoError(t, err)
+		targetUTCTUnix := dt2a.Unix()
+
+		dt2b, err := time.Parse(dtLayout, qrSourceUSPacific.Named().Rows[i].AsString("dt2", ""))
+		require.NoError(t, err)
+		sourceUSPacific := dt2b.Unix()
+
+		dtt := dt2b.In(pacificLocation)
+		zone, _ := dtt.Zone()
+		var hoursBehind int64
+		if zone == "PDT" { // daylight savings is on
+			hoursBehind = 7
+		} else {
+			hoursBehind = 8
+		}
+		// extra logging, so that we can spot any issues in CI test runs
+		log.Infof("times are %s, %s, hours behind %d", dt2a, dt2b, hoursBehind)
+		require.Equal(t, hoursBehind*3600, targetUTCTUnix-sourceUSPacific)
+	}
+
+	// user should be either running this query or have set their location in their driver to map from the time in Vitess/UTC to local
+	query = "select id, convert_tz(dt1, 'UTC', 'US/Pacific') dt1, convert_tz(dt2, 'UTC', 'US/Pacific') dt2, convert_tz(ts1, 'UTC', 'US/Pacific') ts1 from datze"
+	qrTargetUSPacific, err := customerTab.QueryTablet(query, "customer", true)
+	require.NoError(t, err)
+	require.NotNil(t, qrTargetUSPacific)
+	require.Equal(t, len(qrSourceUSPacific.Rows), len(qrTargetUSPacific.Rows))
+
+	for i, row := range qrSourceUSPacific.Named().Rows {
+		// source and target results must match since source is in US/Pacific and we are converting target columns explicitly to US/Pacific
+		require.Equal(t, row.AsString("dt1", ""), qrTargetUSPacific.Named().Rows[i].AsString("dt1", ""))
+		require.Equal(t, row.AsString("dt2", ""), qrTargetUSPacific.Named().Rows[i].AsString("dt2", ""))
+		require.Equal(t, row.AsString("ts1", ""), qrTargetUSPacific.Named().Rows[i].AsString("ts1", ""))
+	}
+
+	output, err = vc.VtctlClient.ExecuteCommandWithOutput("MoveTables", "--", "SwitchTraffic", ksWorkflow)
+	require.NoError(t, err, output)
+
+	qr, err := productTab.QueryTablet(fmt.Sprintf("select * from _vt.vreplication where workflow='%s_reverse'", workflow), "", false)
+	if err != nil {
+		return
+	}
+	for _, row := range qr.Named().Rows {
+		bls := row["source"].ToString()
+		require.Contains(t, bls, "source_time_zone:\"UTC\"")
+		require.Contains(t, bls, "target_time_zone:\"US/Pacific\"")
+	}
+
+	// inserts to test date conversions in reverse replication
+	execVtgateQuery(t, vtgateConn, "customer", "insert into datze(id, dt2) values (13, '2022-01-01 18:20:30')")
+	execVtgateQuery(t, vtgateConn, "customer", "insert into datze(id, dt2) values (14, '2022-04-01 12:06:07')")
+	vdiff(t, ksReverseWorkflow, "")
+}