Materialize workflow support for reference tables (vitessio#16787)

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

changelog/21.0/21.0.0/summary.md

@@ -1,21 +1,21 @@
-
 ## Summary
 
 ### Table of Contents
 
 - **[Major Changes](#major-changes)**
-  - **[Deprecations and Deletions](#deprecations-and-deletions)**
-    - [Deletion of deprecated metrics](#metric-deletion)
-    - [VTTablet Flags](#vttablet-flags)
-    - [Metrics](#deprecations-metrics)
-  - **[Traffic Mirroring](#traffic-mirroring)**
-  - **[New VTGate Shutdown Behavior](#new-vtgate-shutdown-behavior)**
-  - **[Tablet Throttler: Multi-Metric support](#tablet-throttler)**
-  - **[Allow Cross Cell Promotion in PRS](#allow-cross-cell)**
-  - **[Support for recursive CTEs](#recursive-cte)**
-  - **[VTGate Tablet Balancer](#tablet-balancer)**
-  - **[Query Timeout Override](#query-timeout)**
-  - **[Dynamic VReplication Configuration](#dynamic-vreplication-configuration)**
+    - **[Deprecations and Deletions](#deprecations-and-deletions)**
+        - [Deletion of deprecated metrics](#metric-deletion)
+        - [VTTablet Flags](#vttablet-flags)
+        - [Metrics](#deprecations-metrics)
+    - **[Traffic Mirroring](#traffic-mirroring)**
+    - **[New VTGate Shutdown Behavior](#new-vtgate-shutdown-behavior)**
+    - **[Tablet Throttler: Multi-Metric support](#tablet-throttler)**
+    - **[Allow Cross Cell Promotion in PRS](#allow-cross-cell)**
+    - **[Support for recursive CTEs](#recursive-cte)**
+    - **[VTGate Tablet Balancer](#tablet-balancer)**
+    - **[Query Timeout Override](#query-timeout)**
+    - **[Dynamic VReplication Configuration](#dynamic-vreplication-configuration)**
+    - **[Reference Table Materialization](#reference-table-materialization)**
 
 ## <a id="major-changes"/>Major Changes
 
@@ -25,7 +25,6 @@
 
 The following metrics that were deprecated in the previous release, have now been deleted.
 
-
 |                 Metric Name                  |
 |:--------------------------------------------:|
 |           `analysis.change.write`            |        
@@ -43,8 +42,8 @@ The following metrics that were deprecated in the previous release, have now bee
 
 #### <a id="vttablet-flags"/>VTTablet Flags
 
-- `queryserver-enable-settings-pool` flag, added in v15, has been on by default since v17. 
-It is now deprecated and will be removed in a future release.
+- `queryserver-enable-settings-pool` flag, added in v15, has been on by default since v17.
+  It is now deprecated and will be removed in a future release.
 
 #### <a id="deprecations-metrics"/>Metrics
 
@@ -59,10 +58,10 @@ The following metrics are now deprecated, if provided please use their replaceme
 | `vttablet` |   `QueryCacheHits`    |   `QueryEnginePlanCacheHits`    |
 | `vttablet` |  `QueryCacheMisses`   |  `QueryEnginePlanCacheMisses`   |
 
-
 ### <a id="traffic-mirroring"/>Traffic Mirroring
 
-Traffic mirroring is intended to help reduce some of the uncertainty inherent to `MoveTables SwitchTraffic`. When traffic mirroring is enabled, VTGate will mirror a percentage of traffic from one keyspace to another.
+Traffic mirroring is intended to help reduce some of the uncertainty inherent to `MoveTables SwitchTraffic`. When
+traffic mirroring is enabled, VTGate will mirror a percentage of traffic from one keyspace to another.
 
 Mirror rules may be enabled through `vtctldclient` with `MoveTables MirrorTraffic`. For example:
 
@@ -74,9 +73,11 @@ Mirror rules can be inspected with `GetMirrorRules`.
 
 ### <a id="new-vtgate-shutdown-behavior"/>New VTGate Shutdown Behavior
 
-We added a new option to affect the VTGate shutdown process in v21 by using a connection drain timeout rather than the older activity drain timeout.
+We added a new option to affect the VTGate shutdown process in v21 by using a connection drain timeout rather than the
+older activity drain timeout.
 The goal of this new behavior, connection draining option, is to disallow new connections when VTGate is shutting down,
-but continue allowing existing connections to finish their work until they manually disconnect or until the `--onterm_timeout` timeout is reached,
+but continue allowing existing connections to finish their work until they manually disconnect or until
+the `--onterm_timeout` timeout is reached,
 without getting a `Server shutdown in progress` error.
 
 This new behavior can be enabled by specifying the new `--mysql-server-drain-onterm` flag to VTGate.
@@ -85,11 +86,17 @@ See more information about this change by [reading its RFC](https://github.com/v
 
 ### <a id="tablet-throttler"/>Tablet Throttler: Multi-Metric support
 
-Up till `v20`, the tablet throttler would only monitor and use a single metric. That would be replication lag, by default, or could be the result of a custom query. `v21` introduces a major redesign where the throttler monitors and uses multiple metrics at the same time, including the above two.
+Up till `v20`, the tablet throttler would only monitor and use a single metric. That would be replication lag, by
+default, or could be the result of a custom query. `v21` introduces a major redesign where the throttler monitors and
+uses multiple metrics at the same time, including the above two.
 
-Backwards compatible with `v20`, the default behavior in `v21` is to monitor all metrics, but only use `lag` (if the cutsom query is undefined) or the `cutsom` metric (if the custom query is defined). A `v20` `PRIMARY` is compatible with a `v21` `REPLICA`, and a `v21` `PRIMARY` is compatible with a `v20` `REPLICA`.
+Backwards compatible with `v20`, the default behavior in `v21` is to monitor all metrics, but only use `lag` (if the
+cutsom query is undefined) or the `cutsom` metric (if the custom query is defined). A `v20` `PRIMARY` is compatible with
+a `v21` `REPLICA`, and a `v21` `PRIMARY` is compatible with a `v20` `REPLICA`.
 
-However, with `v21` it is possible to assign any combination of metrics (one or more) for a given app. The throttler would then accept or reject the app's requests based on the health of _all_ assigned metrics. `v21` comes with a preset list metrics, expected to be expanded:
+However, with `v21` it is possible to assign any combination of metrics (one or more) for a given app. The throttler
+would then accept or reject the app's requests based on the health of _all_ assigned metrics. `v21` comes with a preset
+list metrics, expected to be expanded:
 
 - `lag`: replication lag based on heartbeat injection.
 - `threads_running`: concurrent active threads on the MySQL server.
@@ -98,33 +105,46 @@ However, with `v21` it is possible to assign any combination of metrics (one or
 
 Each metric has a factory threshold which can be overridden by the `UpdateThrottlerConfig` command.
 
-The throttler also supports the catch-all `"all"` app name, and it is thus possible to assign metrics to _all_ apps. Explicit app to metric assignments will override the catch-all configuration.
+The throttler also supports the catch-all `"all"` app name, and it is thus possible to assign metrics to _all_ apps.
+Explicit app to metric assignments will override the catch-all configuration.
 
-Metrics are assigned a default _scope_, which could be `self` (isolated to the tablet) or `shard` (max, aka _worst_ value among shard tablets). It is further possible to require a different scope for each metric.
+Metrics are assigned a default _scope_, which could be `self` (isolated to the tablet) or `shard` (max, aka _worst_
+value among shard tablets). It is further possible to require a different scope for each metric.
 
 ### <a id="allow-cross-cell"/>Allow Cross Cell Promotion in PRS
-Up until now if the users wanted to promote a replica in a different cell than the current primary using `PlannedReparentShard`, they had to specify the new primary with the `--new-primary` flag.
 
-We have now added a new flag `--allow-cross-cell-promotion` that lets `PlannedReparentShard` choose a primary in a different cell even if no new primary is provided explicitly.
+Up until now if the users wanted to promote a replica in a different cell than the current primary
+using `PlannedReparentShard`, they had to specify the new primary with the `--new-primary` flag.
+
+We have now added a new flag `--allow-cross-cell-promotion` that lets `PlannedReparentShard` choose a primary in a
+different cell even if no new primary is provided explicitly.
 
 ### <a id="recursive-cte"/>Experimental support for recursive CTEs
-We have added experimental support for recursive CTEs in Vitess. We are marking it as experimental because it is not yet fully tested and may have some limitations. We are looking for feedback from the community to improve this feature.
+
+We have added experimental support for recursive CTEs in Vitess. We are marking it as experimental because it is not yet
+fully tested and may have some limitations. We are looking for feedback from the community to improve this feature.
 
 ### <a id="tablet-balancer"/>VTGate Tablet Balancer
-When a VTGate routes a query and has multiple available tablets for a given shard / tablet type (e.g. REPLICA), the current default behavior routes the query with local cell affinity and round robin policy. The VTGate Tablet Balancer provides an alternate mechanism that routes queries to maintain an even distribution of query load to each tablet, while preferentially routing to tablets in the same cell as the VTGate.
 
-The tablet balancer is enabled by a new flag `--enable-balancer` and configured by `--balancer-vtgate-cells` and `--balancer-keyspaces`.
+When a VTGate routes a query and has multiple available tablets for a given shard / tablet type (e.g. REPLICA), the
+current default behavior routes the query with local cell affinity and round robin policy. The VTGate Tablet Balancer
+provides an alternate mechanism that routes queries to maintain an even distribution of query load to each tablet, while
+preferentially routing to tablets in the same cell as the VTGate.
+
+The tablet balancer is enabled by a new flag `--enable-balancer` and configured by `--balancer-vtgate-cells`
+and `--balancer-keyspaces`.
 
 See [RFC for details](https://github.com/vitessio/vitess/issues/12241).
 
 ### <a id="query-timeout"/>Query Timeout Override
-VTGate sends an authoritative query timeout to VTTablet when the `QUERY_TIMEOUT_MS` comment directive, 
-`query_timeout` session system variable, or `query-timeout` flag is set. 
-The order of precedence is: `QUERY_TIMEOUT_MS` > `query_timeout` > `query-timeout`. 
-VTTablet overrides its default query timeout with the value received from VTGate. 
+
+VTGate sends an authoritative query timeout to VTTablet when the `QUERY_TIMEOUT_MS` comment directive,
+`query_timeout` session system variable, or `query-timeout` flag is set.
+The order of precedence is: `QUERY_TIMEOUT_MS` > `query_timeout` > `query-timeout`.
+VTTablet overrides its default query timeout with the value received from VTGate.
 All timeouts are specified in milliseconds.
 
-When a query is executed inside a transaction, this behavior does not apply; instead, 
+When a query is executed inside a transaction, this behavior does not apply; instead,
 the smaller of the transaction timeout or the query timeout from VTGate is used.
 
 A query can also be set to have no timeout by using the `QUERY_TIMEOUT_MS` comment directive with a value of `0`.
@@ -133,7 +153,14 @@ Example usage:
 `select /*vt+ QUERY_TIMEOUT_MS=30 */ col from tbl`
 
 ### <a id="dynamic-vreplication-configuration"/>Dynamic VReplication Configuration
-Currently many of the configuration options for VReplication Workflows are vttablet flags. This means that any change 
+
+Currently many of the configuration options for VReplication Workflows are vttablet flags. This means that any change
 requires restarts of vttablets. We now allow these to be overridden while creating a workflow or dynamically once
 the workflow is in progress. See https://github.com/vitessio/vitess/pull/16583 for details.
 
+### <a id="reference-table-materialization"/>Reference Table Materialization
+
+There is a new option in [`Materialize` workflows](https://vitess.io/docs/reference/vreplication/materialize/) to keep 
+a synced copy of [reference or lookup tables](https://vitess.io/docs/reference/vreplication/reference_tables/) 
+(countries, states, zip_codes, etc) from an unsharded keyspace, which holds the source of truth for the reference 
+table, to all shards in a sharded keyspace.

go/cmd/vtctldclient/command/vreplication/common/utils.go

@@ -68,6 +68,7 @@ var (
 		MySQLServerVersion           string
 		TruncateUILen                int
 		TruncateErrLen               int
+		ReferenceTables              []string
 		ConfigOverrides              []string
 	}{}
 )

go/cmd/vtctldclient/command/vreplication/materialize/create.go

@@ -109,6 +109,7 @@ func commandCreate(cmd *cobra.Command, args []string) error {
 		Cell:                      strings.Join(common.CreateOptions.Cells, ","),
 		TabletTypes:               topoproto.MakeStringTypeCSV(common.CreateOptions.TabletTypes),
 		TabletSelectionPreference: tsp,
+		ReferenceTables:           common.CreateOptions.ReferenceTables,
 		WorkflowOptions:           workflowOptions,
 	}
 

go/cmd/vtctldclient/command/vreplication/materialize/materialize.go

@@ -47,11 +47,11 @@ func registerCommands(root *cobra.Command) {
 	create.Flags().StringVar(&createOptions.SourceKeyspace, "source-keyspace", "", "Keyspace where the tables queried in the 'source_expression' values within table-settings live.")
 	create.MarkFlagRequired("source-keyspace")
 	create.Flags().Var(&createOptions.TableSettings, "table-settings", "A JSON array defining what tables to materialize using what select statements. See the --help output for more details.")
-	create.MarkFlagRequired("table-settings")
 	create.Flags().BoolVar(&common.CreateOptions.StopAfterCopy, "stop-after-copy", false, "Stop the workflow after it's finished copying the existing rows and before it starts replicating changes.")
 	create.Flags().StringVar(&common.CreateOptions.MySQLServerVersion, "mysql_server_version", fmt.Sprintf("%s-Vitess", config.DefaultMySQLVersion), "Configure the MySQL version to use for example for the parser.")
 	create.Flags().IntVar(&common.CreateOptions.TruncateUILen, "sql-max-length-ui", 512, "truncate queries in debug UIs to the given length (default 512)")
 	create.Flags().IntVar(&common.CreateOptions.TruncateErrLen, "sql-max-length-errors", 0, "truncate queries in error logs to the given length (default unlimited)")
+	create.Flags().StringSliceVarP(&common.CreateOptions.ReferenceTables, "reference-tables", "r", nil, "Used to specify the reference tables to materialize on every target shard.")
 	base.AddCommand(create)
 
 	// Generic workflow commands.

go/test/endtoend/vreplication/materialize_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package vreplication
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -235,3 +236,101 @@ func TestMaterializeVtctldClient(t *testing.T) {
 		testShardedMaterialize(t, true)
 	})
 }
+
+const (
+	refSchema = `
+  	create table ref1 (
+		id bigint not null,
+		val varbinary(10) not null,
+		primary key (id)
+	) engine=InnoDB default charset=utf8mb4 collate=utf8mb4_unicode_ci;
+	create table ref2 (
+		id bigint not null,
+		id2 bigint not null,
+		primary key (id)
+	) engine=InnoDB default charset=utf8mb4 collate=utf8mb4_unicode_ci;
+`
+	refSourceVSchema = `
+{
+  "tables": {
+    "ref1": {
+      "type": "reference"
+    },
+	"ref2": {
+      "type": "reference"
+	}
+  }
+}
+`
+	refTargetVSchema = `
+{
+  "tables": {
+	"ref1": {
+		  "type": "reference",
+		  "source": "ks1.ref1"
+	},
+	"ref2": {
+		  "type": "reference",
+		  "source": "ks1.ref2"
+	}
+  }
+}
+`
+	initRef1DataQuery = `insert into ks1.ref1(id, val) values (1, 'abc'), (2, 'def'), (3, 'ghi')`
+	initRef2DataQuery = `insert into ks1.ref2(id, id2) values (1, 1), (2, 2), (3, 3)`
+)
+
+// TestReferenceTableMaterialize tests materializing reference tables.
+func TestReferenceTableMaterialize(t *testing.T) {
+	vc = NewVitessCluster(t, nil)
+	require.NotNil(t, vc)
+	shards := []string{"-80", "80-"}
+	defer vc.TearDown()
+	defaultCell := vc.Cells[vc.CellNames[0]]
+	_, err := vc.AddKeyspace(t, []*Cell{defaultCell}, "ks1", "0", refSourceVSchema, refSchema, 0, 0, 100, nil)
+	require.NoError(t, err)
+	_, err = vc.AddKeyspace(t, []*Cell{defaultCell}, "ks2", strings.Join(shards, ","), refTargetVSchema, "", 0, 0, 200, nil)
+	require.NoError(t, err)
+	vtgateConn := getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
+	defer vtgateConn.Close()
+	verifyClusterHealth(t, vc)
+	_, err = vtgateConn.ExecuteFetch(initRef1DataQuery, 0, false)
+	require.NoError(t, err)
+	_, err = vtgateConn.ExecuteFetch(initRef2DataQuery, 0, false)
+	require.NoError(t, err)
+
+	err = vc.VtctldClient.ExecuteCommand("Materialize", "--target-keyspace", "ks2", "--workflow", "wf1", "create",
+		"--source-keyspace", "ks1", "--reference-tables", "ref1,ref2")
+	require.NoError(t, err, "Materialize")
+	for _, shard := range shards {
+		tab := vc.getPrimaryTablet(t, "ks2", shard)
+		catchup(t, tab, "wf1", "Materialize")
+	}
+
+	for _, shard := range shards {
+		waitForRowCount(t, vtgateConn, "ks2:"+shard, "ref1", 3)
+		waitForQueryResult(t, vtgateConn, "ks2:"+shard, "select id, val from ref1",
+			`[[INT64(1) VARBINARY("abc")] [INT64(2) VARBINARY("def")] [INT64(3) VARBINARY("ghi")]]`)
+		waitForRowCount(t, vtgateConn, "ks2:"+shard, "ref2", 3)
+		waitForQueryResult(t, vtgateConn, "ks2:"+shard, "select id, id2 from ref2",
+			`[[INT64(1) INT64(1)] [INT64(2) INT64(2)] [INT64(3) INT64(3)]]`)
+	}
+	vdiff(t, "ks2", "wf1", defaultCellName, false, true, nil)
+
+	queries := []string{
+		"update ks1.ref1 set val='xyz'",
+		"update ks1.ref2 set id2=3 where id=2",
+		"delete from ks1.ref1 where id=1",
+		"delete from ks1.ref2 where id=1",
+		"insert into ks1.ref1(id, val) values (4, 'jkl'), (5, 'mno')",
+		"insert into ks1.ref2(id, id2) values (4, 4), (5, 5)",
+	}
+	for _, query := range queries {
+		execVtgateQuery(t, vtgateConn, "ks1", query)
+	}
+	for _, shard := range shards {
+		waitForRowCount(t, vtgateConn, "ks2:"+shard, "ref1", 4)
+		waitForRowCount(t, vtgateConn, "ks2:"+shard, "ref2", 4)
+	}
+	vdiff(t, "ks2", "wf1", defaultCellName, false, true, nil)
+}