Use TableSettings instead of strings for wider use cases

Signed-off-by: Noble Mittal <[email protected]>

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

@@ -54,11 +54,19 @@ var (
 )
 
 func commandAddReferenceTables(cmd *cobra.Command, args []string) error {
+	tableSettings := []*vtctldatapb.TableMaterializeSettings{}
+	for _, table := range addReferenceTablesOptions.Tables {
+		tableSettings = append(tableSettings, &vtctldatapb.TableMaterializeSettings{
+			TargetTable: table,
+		})
+	}
+
 	_, err := common.GetClient().MaterializeAddTables(common.GetCommandCtx(), &vtctldatapb.MaterializeAddTablesRequest{
-		Workflow: common.BaseOptions.Workflow,
-		Keyspace: common.BaseOptions.TargetKeyspace,
-		Tables:   addReferenceTablesOptions.Tables,
+		Workflow:      common.BaseOptions.Workflow,
+		Keyspace:      common.BaseOptions.TargetKeyspace,
+		TableSettings: tableSettings,
 	})
+
 	if err != nil {
 		return err
 	}

go/vt/vtctl/grpcvtctldserver/server.go

@@ -3104,7 +3104,7 @@ func (s *VtctldServer) MaterializeAddTables(ctx context.Context, req *vtctldatap
 
 	span.Annotate("workflow", req.Workflow)
 	span.Annotate("keyspace", req.Keyspace)
-	span.Annotate("tables", req.Tables)
+	span.Annotate("table_settings", req.TableSettings)
 
 	err = s.ws.MaterializeAddTables(ctx, req)
 	return resp, err

go/vt/vtctl/workflow/materializer.go

@@ -164,7 +164,7 @@ func (mz *materializer) createWorkflowStreams(req *tabletmanagerdatapb.CreateVRe
 		// Each tablet needs its own copy of the request as it will have a unique
 		// BinlogSource.
 		tabletReq := req.CloneVT()
-		tabletReq.BinlogSource, err = mz.generateBinlogSources(mz.ctx, target, sourceShards, streamKeyRangesEqual)
+		tabletReq.BinlogSource, err = mz.generateBinlogSources(target, sourceShards, streamKeyRangesEqual)
 		if err != nil {
 			return err
 		}
@@ -178,7 +178,7 @@ func (mz *materializer) getTenantClause() (*sqlparser.Expr, error) {
 	return getTenantClause(mz.ms.WorkflowOptions, mz.targetVSchema, mz.env.Parser())
 }
 
-func (mz *materializer) generateBinlogSources(ctx context.Context, targetShard *topo.ShardInfo, sourceShards []*topo.ShardInfo, keyRangesEqual bool) ([]*binlogdatapb.BinlogSource, error) {
+func (mz *materializer) generateBinlogSources(targetShard *topo.ShardInfo, sourceShards []*topo.ShardInfo, keyRangesEqual bool) ([]*binlogdatapb.BinlogSource, error) {
 	blses := make([]*binlogdatapb.BinlogSource, 0, len(mz.sourceShards))
 	for _, sourceShard := range sourceShards {
 		bls := &binlogdatapb.BinlogSource{
@@ -202,69 +202,76 @@ func (mz *materializer) generateBinlogSources(ctx context.Context, targetShard *
 		}
 
 		for _, ts := range mz.ms.TableSettings {
-			rule := &binlogdatapb.Rule{
-				Match: ts.TargetTable,
+			rule, err := mz.generateRule(ts, targetShard, tenantClause, keyRangesEqual)
+			if err != nil {
+				return nil, err
 			}
+			bls.Filter.Rules = append(bls.Filter.Rules, rule)
+		}
+		blses = append(blses, bls)
+	}
+	return blses, nil
+}
 
-			if ts.SourceExpression == "" {
-				bls.Filter.Rules = append(bls.Filter.Rules, rule)
-				continue
-			}
+func (mz *materializer) generateRule(ts *vtctldatapb.TableMaterializeSettings, targetShard *topo.ShardInfo, tenantClause *sqlparser.Expr, keyRangesEqual bool) (*binlogdatapb.Rule, error) {
+	rule := &binlogdatapb.Rule{
+		Match: ts.TargetTable,
+	}
 
-			// Validate non-empty query.
-			stmt, err := mz.env.Parser().Parse(ts.SourceExpression)
+	if ts.SourceExpression == "" {
+		return rule, nil
+	}
+
+	// Validate non-empty query.
+	stmt, err := mz.env.Parser().Parse(ts.SourceExpression)
+	if err != nil {
+		return nil, err
+	}
+	sel, ok := stmt.(*sqlparser.Select)
+	if !ok {
+		return nil, fmt.Errorf("unrecognized statement: %s", ts.SourceExpression)
+	}
+	if !keyRangesEqual && mz.targetVSchema.Keyspace.Sharded && mz.targetVSchema.Tables[ts.TargetTable].Type != vindexes.TypeReference {
+		cv, err := vindexes.FindBestColVindex(mz.targetVSchema.Tables[ts.TargetTable])
+		if err != nil {
+			return nil, err
+		}
+		mappedCols := make([]*sqlparser.ColName, 0, len(cv.Columns))
+		for _, col := range cv.Columns {
+			colName, err := matchColInSelect(col, sel)
 			if err != nil {
 				return nil, err
 			}
-			sel, ok := stmt.(*sqlparser.Select)
-			if !ok {
-				return nil, fmt.Errorf("unrecognized statement: %s", ts.SourceExpression)
+			mappedCols = append(mappedCols, colName)
+		}
+		subExprs := make([]sqlparser.Expr, 0, len(mappedCols)+2)
+		for _, mappedCol := range mappedCols {
+			subExprs = append(subExprs, mappedCol)
+		}
+		var vindexName string
+		if mz.workflowType == binlogdatapb.VReplicationWorkflowType_Migrate {
+			// For a Migrate, if the TargetKeyspace name is different from the SourceKeyspace name, we need to use the
+			// SourceKeyspace name to determine the vindex since the TargetKeyspace name is not known to the source.
+			// Note: it is expected that the source and target keyspaces have the same vindex name and data type.
+			keyspace := mz.ms.TargetKeyspace
+			if mz.ms.ExternalCluster != "" {
+				keyspace = mz.ms.SourceKeyspace
 			}
-			if !keyRangesEqual && mz.targetVSchema.Keyspace.Sharded && mz.targetVSchema.Tables[ts.TargetTable].Type != vindexes.TypeReference {
-				cv, err := vindexes.FindBestColVindex(mz.targetVSchema.Tables[ts.TargetTable])
-				if err != nil {
-					return nil, err
-				}
-				mappedCols := make([]*sqlparser.ColName, 0, len(cv.Columns))
-				for _, col := range cv.Columns {
-					colName, err := matchColInSelect(col, sel)
-					if err != nil {
-						return nil, err
-					}
-					mappedCols = append(mappedCols, colName)
-				}
-				subExprs := make([]sqlparser.Expr, 0, len(mappedCols)+2)
-				for _, mappedCol := range mappedCols {
-					subExprs = append(subExprs, mappedCol)
-				}
-				var vindexName string
-				if mz.workflowType == binlogdatapb.VReplicationWorkflowType_Migrate {
-					// For a Migrate, if the TargetKeyspace name is different from the SourceKeyspace name, we need to use the
-					// SourceKeyspace name to determine the vindex since the TargetKeyspace name is not known to the source.
-					// Note: it is expected that the source and target keyspaces have the same vindex name and data type.
-					keyspace := mz.ms.TargetKeyspace
-					if mz.ms.ExternalCluster != "" {
-						keyspace = mz.ms.SourceKeyspace
-					}
-					vindexName = fmt.Sprintf("%s.%s", keyspace, cv.Name)
-				} else {
-					vindexName = fmt.Sprintf("%s.%s", mz.ms.TargetKeyspace, cv.Name)
-				}
-
-				subExprs = append(subExprs, sqlparser.NewStrLiteral(vindexName))
-				subExprs = append(subExprs, sqlparser.NewStrLiteral(key.KeyRangeString(targetShard.KeyRange)))
-				inKeyRange := sqlparser.NewFuncExpr("in_keyrange", subExprs...)
-				addFilter(sel, inKeyRange)
-			}
-			if tenantClause != nil {
-				addFilter(sel, *tenantClause)
-			}
-			rule.Filter = sqlparser.String(sel)
-			bls.Filter.Rules = append(bls.Filter.Rules, rule)
+			vindexName = fmt.Sprintf("%s.%s", keyspace, cv.Name)
+		} else {
+			vindexName = fmt.Sprintf("%s.%s", mz.ms.TargetKeyspace, cv.Name)
 		}
-		blses = append(blses, bls)
+
+		subExprs = append(subExprs, sqlparser.NewStrLiteral(vindexName))
+		subExprs = append(subExprs, sqlparser.NewStrLiteral(key.KeyRangeString(targetShard.KeyRange)))
+		inKeyRange := sqlparser.NewFuncExpr("in_keyrange", subExprs...)
+		addFilter(sel, inKeyRange)
 	}
-	return blses, nil
+	if tenantClause != nil {
+		addFilter(sel, *tenantClause)
+	}
+	rule.Filter = sqlparser.String(sel)
+	return rule, nil
 }
 
 func (mz *materializer) deploySchema() error {