From d914e1c81cb6879e214bcff41becb2b047086243 Mon Sep 17 00:00:00 2001 From: Noble Mittal Date: Sat, 16 Nov 2024 16:29:17 +0530 Subject: [PATCH] refac: Split prepareCreateLookup into individual funcs Signed-off-by: Noble Mittal --- go/vt/vtctl/workflow/lookup.go | 572 ++++++++++++---------- go/vt/vtctl/workflow/materializer_test.go | 104 ++-- go/vt/vtctl/workflow/server.go | 423 +--------------- 3 files changed, 404 insertions(+), 695 deletions(-) diff --git a/go/vt/vtctl/workflow/lookup.go b/go/vt/vtctl/workflow/lookup.go index 96842b07e3f..271984504b1 100644 --- a/go/vt/vtctl/workflow/lookup.go +++ b/go/vt/vtctl/workflow/lookup.go @@ -42,19 +42,7 @@ import ( // Lookup Vindex. func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace string, specs *vschemapb.Keyspace, continueAfterCopyWithOwner bool) ( ms *vtctldatapb.MaterializeSettings, sourceVSchema, targetVSchema *vschemapb.Keyspace, cancelFunc func() error, err error) { - // Important variables are pulled out here. var ( - vindexName string - vindex *vschemapb.Vindex - targetKeyspace string - targetTableName string - vindexFromCols []string - vindexToCol string - vindexIgnoreNulls bool - - sourceTableName string - // sourceTable is the supplied table info. - sourceTable *vschemapb.Table // sourceVSchemaTable is the table info present in the vschema. sourceVSchemaTable *vschemapb.Table // sourceVindexColumns are computed from the input sourceTable. @@ -66,179 +54,40 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s ) // Validate input vindex. - if specs == nil { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "no vindex provided") - } - if len(specs.Vindexes) != 1 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "only one vindex must be specified") - } - vindexName = maps.Keys(specs.Vindexes)[0] - vindex = maps.Values(specs.Vindexes)[0] - if !strings.Contains(vindex.Type, "lookup") { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex %s is not a lookup type", vindex.Type) - } - targetKeyspace, targetTableName, err = w.parser.ParseTable(vindex.Params["table"]) - if err != nil || targetKeyspace == "" { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex table name (%s) must be in the form .", vindex.Params["table"]) - } - vindexFromCols = strings.Split(vindex.Params["from"], ",") - for i, col := range vindexFromCols { - vindexFromCols[i] = strings.TrimSpace(col) - } - if strings.Contains(vindex.Type, "unique") { - if len(vindexFromCols) != 1 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unique vindex 'from' should have only one column") - } - } else { - if len(vindexFromCols) < 2 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "non-unique vindex 'from' should have more than one column") - } - } - vindexToCol = vindex.Params["to"] - // Make the vindex write_only. If one exists already in the vschema, - // it will need to match this vindex exactly, including the write_only setting. - vindex.Params["write_only"] = "true" - // See if we can create the vindex without errors. - if _, err := vindexes.CreateVindex(vindex.Type, vindexName, vindex.Params); err != nil { + vindex, vInfo, err := w.validateAndGetVindex(specs) + if err != nil { return nil, nil, nil, nil, err } - if ignoreNullsStr, ok := vindex.Params["ignore_nulls"]; ok { - // This mirrors the behavior of vindexes.boolFromMap(). - switch ignoreNullsStr { - case "true": - vindexIgnoreNulls = true - case "false": - vindexIgnoreNulls = false - default: - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ignore_nulls (%s) value must be 'true' or 'false'", - ignoreNullsStr) - } - } - // Validate input table. - if len(specs.Tables) < 1 || len(specs.Tables) > 2 { - return nil, nil, nil, nil, fmt.Errorf("one or two tables must be specified") - } - // Loop executes once or twice. - for tableName, table := range specs.Tables { - if len(table.ColumnVindexes) != 1 { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "exactly one ColumnVindex must be specified for the %s table", - tableName) - } - if tableName != targetTableName { // This is the source table. - sourceTableName = tableName - sourceTable = table - continue - } - // This is a primary vindex definition for the target table - // which allows you to override the vindex type used. - var vindexCols []string - if len(table.ColumnVindexes[0].Columns) != 0 { - vindexCols = table.ColumnVindexes[0].Columns - } else { - if table.ColumnVindexes[0].Column == "" { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", - tableName) - } - vindexCols = []string{table.ColumnVindexes[0].Column} - } - if !slices.Equal(vindexCols, vindexFromCols) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "columns in the lookup table %s primary vindex (%s) don't match the 'from' columns specified (%s)", - tableName, strings.Join(vindexCols, ","), strings.Join(vindexFromCols, ",")) - } + vInfo.sourceTable, vInfo.sourceTableName, err = getSourceTable(specs, vInfo.targetTableName, vInfo.fromCols) + if err != nil { + return nil, nil, nil, nil, err } - // Validate input table and vindex consistency. - if sourceTable == nil || len(sourceTable.ColumnVindexes) != 1 { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "No ColumnVindex found for the owner table (%s) in the %s keyspace", - sourceTable, keyspace) - } - if sourceTable.ColumnVindexes[0].Name != vindexName { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ColumnVindex name (%s) must match vindex name (%s)", - sourceTable.ColumnVindexes[0].Name, vindexName) - } - if vindex.Owner != "" && vindex.Owner != sourceTableName { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex owner (%s) must match table name (%s)", - vindex.Owner, sourceTableName) - } - if len(sourceTable.ColumnVindexes[0].Columns) != 0 { - sourceVindexColumns = sourceTable.ColumnVindexes[0].Columns - } else { - if sourceTable.ColumnVindexes[0].Column == "" { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", - sourceTableName) - } - sourceVindexColumns = []string{sourceTable.ColumnVindexes[0].Column} - } - if len(sourceVindexColumns) != len(vindexFromCols) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "length of table columns (%d) differs from length of vindex columns (%d)", - len(sourceVindexColumns), len(vindexFromCols)) + sourceVindexColumns, err = validateSourceTableAndGetVindexColumns(vInfo, vindex, keyspace) + if err != nil { + return nil, nil, nil, nil, err } - // Validate against source vschema. - sourceVSchema, err = w.ts.GetVSchema(ctx, keyspace) + sourceVSchema, targetVSchema, err = w.getTargetAndSourceVSchema(ctx, keyspace, vInfo.targetKeyspace) if err != nil { return nil, nil, nil, nil, err } - if sourceVSchema.Vindexes == nil { - sourceVSchema.Vindexes = make(map[string]*vschemapb.Vindex) - } - // If source and target keyspaces are the same, make vschemas point - // to the same object. - if keyspace == targetKeyspace { - targetVSchema = sourceVSchema - } else { - targetVSchema, err = w.ts.GetVSchema(ctx, targetKeyspace) - if err != nil { - return nil, nil, nil, nil, err - } - } - if targetVSchema.Vindexes == nil { - targetVSchema.Vindexes = make(map[string]*vschemapb.Vindex) - } - if targetVSchema.Tables == nil { - targetVSchema.Tables = make(map[string]*vschemapb.Table) - } - if existing, ok := sourceVSchema.Vindexes[vindexName]; ok { + + if existing, ok := sourceVSchema.Vindexes[vInfo.name]; ok { if !proto.Equal(existing, vindex) { // If the exact same vindex already exists then we can re-use it - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "a conflicting vindex named %s already exists in the %s keyspace", - vindexName, keyspace) + return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "a conflicting vindex named %s already exists in the %s keyspace", + vInfo.name, keyspace) } } - sourceVSchemaTable = sourceVSchema.Tables[sourceTableName] - if sourceVSchemaTable == nil && !schema.IsInternalOperationTableName(sourceTableName) { + + sourceVSchemaTable = sourceVSchema.Tables[vInfo.sourceTableName] + if sourceVSchemaTable == nil && !schema.IsInternalOperationTableName(vInfo.sourceTableName) { return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "table %s not found in the %s keyspace", sourceTableName, keyspace) + vterrors.Errorf(vtrpcpb.Code_INTERNAL, "table %s not found in the %s keyspace", vInfo.sourceTableName, keyspace) } - for _, colVindex := range sourceVSchemaTable.ColumnVindexes { - // For a conflict, the vindex name and column should match. - if colVindex.Name != vindexName { - continue - } - var colNames []string - if len(colVindex.Columns) == 0 { - colNames = []string{colVindex.Column} - } else { - colNames = colVindex.Columns - } - // If this is the exact same definition then we can use the existing one. If they - // are not the same then they are two distinct conflicting vindexes and we should - // not proceed. - if !slices.Equal(colNames, sourceVindexColumns) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting ColumnVindex on column(s) %s in table %s already exists in the %s keyspace", - strings.Join(colNames, ","), sourceTableName, keyspace) - } + if err := validateNonConflictingColumnVindex(sourceVSchemaTable, vInfo, sourceVindexColumns, keyspace); err != nil { + return nil, nil, nil, nil, err } // Validate against source schema. @@ -251,7 +100,8 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "source shard %s has no primary", onesource.ShardName()) } - req := &tabletmanagerdatapb.GetSchemaRequest{Tables: []string{sourceTableName}} + + req := &tabletmanagerdatapb.GetSchemaRequest{Tables: []string{vInfo.sourceTableName}} tableSchema, err := schematools.GetSchema(ctx, w.ts, w.tmc, onesource.PrimaryAlias, req) if err != nil { return nil, nil, nil, nil, err @@ -263,76 +113,13 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s } // Generate "create table" statement. - lines := strings.Split(tableSchema.TableDefinitions[0].Schema, "\n") - if len(lines) < 3 { - // Should never happen. - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "schema looks incorrect: %s, expecting at least four lines", - tableSchema.TableDefinitions[0].Schema) - } - var modified []string - modified = append(modified, strings.Replace(lines[0], sourceTableName, targetTableName, 1)) - for i := range sourceVindexColumns { - line, err := generateColDef(lines, sourceVindexColumns[i], vindexFromCols[i]) - if err != nil { - return nil, nil, nil, nil, err - } - modified = append(modified, line) - } - - if vindex.Params["data_type"] == "" || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { - modified = append(modified, fmt.Sprintf(" %s varbinary(128),", sqlescape.EscapeID(vindexToCol))) - } else { - modified = append(modified, fmt.Sprintf(" %s %s,", sqlescape.EscapeID(vindexToCol), sqlescape.EscapeID(vindex.Params["data_type"]))) - } - buf := sqlparser.NewTrackedBuffer(nil) - fmt.Fprintf(buf, " PRIMARY KEY (") - prefix := "" - for _, col := range vindexFromCols { - fmt.Fprintf(buf, "%s%s", prefix, sqlescape.EscapeID(col)) - prefix = ", " - } - fmt.Fprintf(buf, ")") - modified = append(modified, buf.String()) - modified = append(modified, ")") - createDDL = strings.Join(modified, "\n") - // Confirm that our DDL is valid before we create anything. - if _, err = w.parser.ParseStrictDDL(createDDL); err != nil { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "error: %v; invalid lookup table definition generated: %s", - err, createDDL) + createDDL, err = w.generateCreateDDLStatement(tableSchema, sourceVindexColumns, vInfo, vindex) + if err != nil { + return nil, nil, nil, nil, err } // Generate vreplication query. - buf = sqlparser.NewTrackedBuffer(nil) - buf.Myprintf("select ") - for i := range vindexFromCols { - buf.Myprintf("%s as %s, ", sqlparser.String(sqlparser.NewIdentifierCI(sourceVindexColumns[i])), sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - } - if strings.EqualFold(vindexToCol, "keyspace_id") || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { - buf.Myprintf("keyspace_id() as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } else { - buf.Myprintf("%s as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)), sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } - buf.Myprintf("from %s", sqlparser.String(sqlparser.NewIdentifierCS(sourceTableName))) - if vindexIgnoreNulls { - buf.Myprintf(" where ") - lastValIdx := len(vindexFromCols) - 1 - for i := range vindexFromCols { - buf.Myprintf("%s is not null", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - if i != lastValIdx { - buf.Myprintf(" and ") - } - } - } - if vindex.Owner != "" { - // Only backfill. - buf.Myprintf(" group by ") - for i := range vindexFromCols { - buf.Myprintf("%s, ", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - } - buf.Myprintf("%s", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } - materializeQuery = buf.String() + materializeQuery = generateMaterializeQuery(vInfo, vindex, sourceVindexColumns) // Save a copy of the original vschema if we modify it and need to provide // a cancelFunc. @@ -340,7 +127,7 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s targetChanged := false // Update targetVSchema. - targetTable := specs.Tables[targetTableName] + targetTable := specs.Tables[vInfo.targetTableName] if targetVSchema.Sharded { // Choose a primary vindex type for the lookup table based on the source // definition if one was not explicitly specified. @@ -369,11 +156,12 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s vterrors.Errorf(vtrpcpb.Code_INTERNAL, "column %s not found in target schema %s", sourceVindexColumns[0], tableSchema.TableDefinitions[0].Schema) } + if existing, ok := targetVSchema.Vindexes[targetVindexType]; ok { if !proto.Equal(existing, targetVindex) { return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting vindex named %v already exists in the %s keyspace", - targetVindexType, targetKeyspace) + targetVindexType, vInfo.targetKeyspace) } } else { targetVSchema.Vindexes[targetVindexType] = targetVindex @@ -382,28 +170,28 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s targetTable = &vschemapb.Table{ ColumnVindexes: []*vschemapb.ColumnVindex{{ - Column: vindexFromCols[0], + Column: vInfo.fromCols[0], Name: targetVindexType, }}, } } else { targetTable = &vschemapb.Table{} } - if existing, ok := targetVSchema.Tables[targetTableName]; ok { + if existing, ok := targetVSchema.Tables[vInfo.targetTableName]; ok { if !proto.Equal(existing, targetTable) { return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting table named %s already exists in the %s vschema", - targetTableName, targetKeyspace) + vInfo.targetTableName, vInfo.targetKeyspace) } } else { - targetVSchema.Tables[targetTableName] = targetTable + targetVSchema.Tables[vInfo.targetTableName] = targetTable targetChanged = true } if targetChanged { cancelFunc = func() error { // Restore the original target vschema. - return w.ts.SaveVSchema(ctx, targetKeyspace, ogTargetVSchema) + return w.ts.SaveVSchema(ctx, vInfo.targetKeyspace, ogTargetVSchema) } } @@ -411,22 +199,308 @@ func (w *workflow) prepareCreateLookup(ctx context.Context, workflow, keyspace s Workflow: workflow, MaterializationIntent: vtctldatapb.MaterializationIntent_CREATELOOKUPINDEX, SourceKeyspace: keyspace, - TargetKeyspace: targetKeyspace, + TargetKeyspace: vInfo.targetKeyspace, StopAfterCopy: vindex.Owner != "" && !continueAfterCopyWithOwner, TableSettings: []*vtctldatapb.TableMaterializeSettings{{ - TargetTable: targetTableName, + TargetTable: vInfo.targetTableName, SourceExpression: materializeQuery, CreateDdl: createDDL, }}, } // Update sourceVSchema - sourceVSchema.Vindexes[vindexName] = vindex - sourceVSchemaTable.ColumnVindexes = append(sourceVSchemaTable.ColumnVindexes, sourceTable.ColumnVindexes[0]) + sourceVSchema.Vindexes[vInfo.name] = vindex + sourceVSchemaTable.ColumnVindexes = append(sourceVSchemaTable.ColumnVindexes, vInfo.sourceTable.ColumnVindexes[0]) return ms, sourceVSchema, targetVSchema, cancelFunc, nil } +// vindexInfo holds the validated vindex configuration +type vindexInfo struct { + name string + targetKeyspace string + targetTableName string + fromCols []string + toCol string + ignoreNulls bool + + // sourceTable is the supplied table info. + sourceTable *vschemapb.Table + sourceTableName string +} + +// validateAndGetVindex validates and extracts vindex configuration +func (w *workflow) validateAndGetVindex(specs *vschemapb.Keyspace) (*vschemapb.Vindex, *vindexInfo, error) { + if specs == nil { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "no vindex provided") + } + if len(specs.Vindexes) != 1 { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "only one vindex must be specified") + } + + vindexName := maps.Keys(specs.Vindexes)[0] + vindex := maps.Values(specs.Vindexes)[0] + + if !strings.Contains(vindex.Type, "lookup") { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex %s is not a lookup type", vindex.Type) + } + + targetKeyspace, targetTableName, err := w.parser.ParseTable(vindex.Params["table"]) + if err != nil || targetKeyspace == "" { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, + "vindex table name (%s) must be in the form .
", vindex.Params["table"]) + } + + vindexFromCols := strings.Split(vindex.Params["from"], ",") + for i, col := range vindexFromCols { + vindexFromCols[i] = strings.TrimSpace(col) + } + + if strings.Contains(vindex.Type, "unique") { + if len(vindexFromCols) != 1 { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unique vindex 'from' should have only one column") + } + } else { + if len(vindexFromCols) < 2 { + return nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "non-unique vindex 'from' should have more than one column") + } + } + + vindexToCol := vindex.Params["to"] + // Make the vindex write_only. If one exists already in the vschema, + // it will need to match this vindex exactly, including the write_only setting. + vindex.Params["write_only"] = "true" + + // See if we can create the vindex without errors. + if _, err := vindexes.CreateVindex(vindex.Type, vindexName, vindex.Params); err != nil { + return nil, nil, err + } + + ignoreNulls := false + if ignoreNullsStr, ok := vindex.Params["ignore_nulls"]; ok { + // This mirrors the behavior of vindexes.boolFromMap(). + switch ignoreNullsStr { + case "true": + ignoreNulls = true + case "false": + ignoreNulls = false + default: + return nil, nil, + vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ignore_nulls (%s) value must be 'true' or 'false'", + ignoreNullsStr) + } + } + + // Validate input table. + if len(specs.Tables) < 1 || len(specs.Tables) > 2 { + return nil, nil, fmt.Errorf("one or two tables must be specified") + } + + return vindex, &vindexInfo{ + name: vindexName, + targetKeyspace: targetKeyspace, + targetTableName: targetTableName, + fromCols: vindexFromCols, + toCol: vindexToCol, + ignoreNulls: ignoreNulls, + }, nil +} + +func (w *workflow) getTargetAndSourceVSchema(ctx context.Context, sourceKeyspace string, targetKeyspace string) (sourceVSchema *vschemapb.Keyspace, targetVSchema *vschemapb.Keyspace, err error) { + sourceVSchema, err = w.ts.GetVSchema(ctx, sourceKeyspace) + if err != nil { + return nil, nil, err + } + if sourceVSchema.Vindexes == nil { + sourceVSchema.Vindexes = make(map[string]*vschemapb.Vindex) + } + // If source and target keyspaces are the same, make vschemas point + // to the same object. + if sourceKeyspace == targetKeyspace { + targetVSchema = sourceVSchema + } else { + targetVSchema, err = w.ts.GetVSchema(ctx, targetKeyspace) + if err != nil { + return nil, nil, err + } + } + if targetVSchema.Vindexes == nil { + targetVSchema.Vindexes = make(map[string]*vschemapb.Vindex) + } + if targetVSchema.Tables == nil { + targetVSchema.Tables = make(map[string]*vschemapb.Table) + } + + return sourceVSchema, targetVSchema, nil +} + +func getSourceTable(specs *vschemapb.Keyspace, targetTableName string, fromCols []string) (sourceTable *vschemapb.Table, sourceTableName string, err error) { + // Loop executes once or twice. + for tableName, table := range specs.Tables { + if len(table.ColumnVindexes) != 1 { + return nil, "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "exactly one ColumnVindex must be specified for the %s table", + tableName) + } + + if tableName != targetTableName { // This is the source table. + sourceTableName = tableName + sourceTable = table + continue + } + // This is a primary vindex definition for the target table + // which allows you to override the vindex type used. + var vindexCols []string + if len(table.ColumnVindexes[0].Columns) != 0 { + vindexCols = table.ColumnVindexes[0].Columns + } else { + if table.ColumnVindexes[0].Column == "" { + return nil, "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", + tableName) + } + vindexCols = []string{table.ColumnVindexes[0].Column} + } + if !slices.Equal(vindexCols, fromCols) { + return nil, "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "columns in the lookup table %s primary vindex (%s) don't match the 'from' columns specified (%s)", + tableName, strings.Join(vindexCols, ","), strings.Join(fromCols, ",")) + } + } + return sourceTable, sourceTableName, nil +} + +func (w *workflow) generateCreateDDLStatement(tableSchema *tabletmanagerdatapb.SchemaDefinition, sourceVindexColumns []string, vInfo *vindexInfo, vindex *vschemapb.Vindex) (string, error) { + lines := strings.Split(tableSchema.TableDefinitions[0].Schema, "\n") + if len(lines) < 3 { + // Should never happen. + return "", vterrors.Errorf(vtrpcpb.Code_INTERNAL, "schema looks incorrect: %s, expecting at least four lines", + tableSchema.TableDefinitions[0].Schema) + } + + var modified []string + modified = append(modified, strings.Replace(lines[0], vInfo.sourceTableName, vInfo.targetTableName, 1)) + for i := range sourceVindexColumns { + line, err := generateColDef(lines, sourceVindexColumns[i], vInfo.fromCols[i]) + if err != nil { + return "", err + } + modified = append(modified, line) + } + + if vindex.Params["data_type"] == "" || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { + modified = append(modified, fmt.Sprintf(" %s varbinary(128),", sqlescape.EscapeID(vInfo.toCol))) + } else { + modified = append(modified, fmt.Sprintf(" %s %s,", sqlescape.EscapeID(vInfo.toCol), sqlescape.EscapeID(vindex.Params["data_type"]))) + } + + buf := sqlparser.NewTrackedBuffer(nil) + fmt.Fprintf(buf, " PRIMARY KEY (") + prefix := "" + for _, col := range vInfo.fromCols { + fmt.Fprintf(buf, "%s%s", prefix, sqlescape.EscapeID(col)) + prefix = ", " + } + fmt.Fprintf(buf, ")") + + modified = append(modified, buf.String()) + modified = append(modified, ")") + createDDL := strings.Join(modified, "\n") + + // Confirm that our DDL is valid before we create anything. + if _, err := w.parser.ParseStrictDDL(createDDL); err != nil { + return "", vterrors.Errorf(vtrpcpb.Code_INTERNAL, "error: %v; invalid lookup table definition generated: %s", + err, createDDL) + } + + return createDDL, nil +} + +func generateMaterializeQuery(vInfo *vindexInfo, vindex *vschemapb.Vindex, sourceVindexColumns []string) string { + buf := sqlparser.NewTrackedBuffer(nil) + buf.Myprintf("select ") + for i := range vInfo.fromCols { + buf.Myprintf("%s as %s, ", sqlparser.String(sqlparser.NewIdentifierCI(sourceVindexColumns[i])), sqlparser.String(sqlparser.NewIdentifierCI(vInfo.fromCols[i]))) + } + if strings.EqualFold(vInfo.toCol, "keyspace_id") || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { + buf.Myprintf("keyspace_id() as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vInfo.toCol))) + } else { + buf.Myprintf("%s as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vInfo.toCol)), sqlparser.String(sqlparser.NewIdentifierCI(vInfo.toCol))) + } + buf.Myprintf("from %s", sqlparser.String(sqlparser.NewIdentifierCS(vInfo.sourceTableName))) + if vInfo.ignoreNulls { + buf.Myprintf(" where ") + lastValIdx := len(vInfo.fromCols) - 1 + for i := range vInfo.fromCols { + buf.Myprintf("%s is not null", sqlparser.String(sqlparser.NewIdentifierCI(vInfo.fromCols[i]))) + if i != lastValIdx { + buf.Myprintf(" and ") + } + } + } + if vindex.Owner != "" { + // Only backfill. + buf.Myprintf(" group by ") + for i := range vInfo.fromCols { + buf.Myprintf("%s, ", sqlparser.String(sqlparser.NewIdentifierCI(vInfo.fromCols[i]))) + } + buf.Myprintf("%s", sqlparser.String(sqlparser.NewIdentifierCI(vInfo.toCol))) + } + return buf.String() +} + +// validateSourceTableAndGetVindexColumns validates input table and vindex consistency, and returns sourceVindexColumns. +func validateSourceTableAndGetVindexColumns(vInfo *vindexInfo, vindex *vschemapb.Vindex, keyspace string) (sourceVindexColumns []string, err error) { + if vInfo.sourceTable == nil || len(vInfo.sourceTable.ColumnVindexes) != 1 { + return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "No ColumnVindex found for the owner table (%s) in the %s keyspace", + vInfo.sourceTable, keyspace) + } + if vInfo.sourceTable.ColumnVindexes[0].Name != vInfo.name { + return nil, + vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ColumnVindex name (%s) must match vindex name (%s)", + vInfo.sourceTable.ColumnVindexes[0].Name, vInfo.name) + } + if vindex.Owner != "" && vindex.Owner != vInfo.sourceTableName { + return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex owner (%s) must match table name (%s)", + vindex.Owner, vInfo.sourceTableName) + } + if len(vInfo.sourceTable.ColumnVindexes[0].Columns) != 0 { + sourceVindexColumns = vInfo.sourceTable.ColumnVindexes[0].Columns + } else { + if vInfo.sourceTable.ColumnVindexes[0].Column == "" { + return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", + vInfo.sourceTableName) + } + sourceVindexColumns = []string{vInfo.sourceTable.ColumnVindexes[0].Column} + } + if len(sourceVindexColumns) != len(vInfo.fromCols) { + return nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "length of table columns (%d) differs from length of vindex columns (%d)", + len(sourceVindexColumns), len(vInfo.fromCols)) + } + + return sourceVindexColumns, nil +} + +func validateNonConflictingColumnVindex(sourceVSchemaTable *vschemapb.Table, vInfo *vindexInfo, sourceVindexColumns []string, keyspace string) error { + for _, colVindex := range sourceVSchemaTable.ColumnVindexes { + // For a conflict, the vindex name and column should match. + if colVindex.Name != vInfo.name { + continue + } + var colNames []string + if len(colVindex.Columns) == 0 { + colNames = []string{colVindex.Column} + } else { + colNames = colVindex.Columns + } + // If this is the exact same definition then we can use the existing one. If they + // are not the same then they are two distinct conflicting vindexes and we should + // not proceed. + if !slices.Equal(colNames, sourceVindexColumns) { + return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting ColumnVindex on column(s) %s in table %s already exists in the %s keyspace", + strings.Join(colNames, ","), vInfo.sourceTableName, keyspace) + } + } + return nil +} + func generateColDef(lines []string, sourceVindexCol, vindexFromCol string) (string, error) { source := sqlescape.EscapeID(sourceVindexCol) target := sqlescape.EscapeID(vindexFromCol) diff --git a/go/vt/vtctl/workflow/materializer_test.go b/go/vt/vtctl/workflow/materializer_test.go index 746c5fe2bae..bb31fdde436 100644 --- a/go/vt/vtctl/workflow/materializer_test.go +++ b/go/vt/vtctl/workflow/materializer_test.go @@ -1515,7 +1515,13 @@ func TestCreateLookupVindexCreateDDL(t *testing.T) { setStartingVschema() }() } - outms, _, _, cancelFunc, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, tcase.specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + outms, _, _, cancelFunc, err := w.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, tcase.specs, false) if tcase.err != "" { require.Error(t, err) require.Contains(t, err.Error(), tcase.err, "prepareCreateLookup(%s) err: %v, does not contain %v", tcase.description, err, tcase.err) @@ -1763,7 +1769,13 @@ func TestCreateLookupVindexSourceVSchema(t *testing.T) { t.Fatal(err) } - _, got, _, _, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + _, got, _, _, err := w.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, specs, false) require.NoError(t, err) if !proto.Equal(got, tcase.out) { t.Errorf("%s: got:\n%v, want\n%v", tcase.description, got, tcase.out) @@ -1984,32 +1996,40 @@ func TestCreateLookupVindexTargetVSchema(t *testing.T) { err: "type SET is not recommended for a vindex", }} for _, tcase := range testcases { - env.tmc.schema[ms.SourceKeyspace+".t1"] = &tabletmanagerdatapb.SchemaDefinition{ - TableDefinitions: []*tabletmanagerdatapb.TableDefinition{{ - Fields: []*querypb.Field{{ - Name: "col2", - Type: tcase.sourceFieldType, + t.Run(tcase.description, func(t *testing.T) { + env.tmc.schema[ms.SourceKeyspace+".t1"] = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{{ + Fields: []*querypb.Field{{ + Name: "col2", + Type: tcase.sourceFieldType, + }}, + Schema: sourceSchema, }}, - Schema: sourceSchema, - }}, - } - specs.Vindexes["v"].Params["table"] = fmt.Sprintf("%s.%s", ms.TargetKeyspace, tcase.targetTable) - if err := env.topoServ.SaveVSchema(ctx, ms.TargetKeyspace, tcase.targetVSchema); err != nil { - t.Fatal(err) - } + } + specs.Vindexes["v"].Params["table"] = fmt.Sprintf("%s.%s", ms.TargetKeyspace, tcase.targetTable) + if err := env.topoServ.SaveVSchema(ctx, ms.TargetKeyspace, tcase.targetVSchema); err != nil { + t.Fatal(err) + } - _, _, got, cancelFunc, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, specs, false) - if tcase.err != "" { - if err == nil || !strings.Contains(err.Error(), tcase.err) { - t.Errorf("prepareCreateLookup(%s) err: %v, must contain %v", tcase.description, err, tcase.err) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), } - continue - } - require.NoError(t, err) - // withTable is a vschema that already contains the table and thus - // we don't make any vschema changes and there's nothing to cancel. - require.True(t, (cancelFunc != nil) == (tcase.targetVSchema != withTable)) - utils.MustMatch(t, tcase.out, got, tcase.description) + _, _, got, cancelFunc, err := w.prepareCreateLookup(ctx, "workflow", ms.SourceKeyspace, specs, false) + if tcase.err != "" { + if err == nil || !strings.Contains(err.Error(), tcase.err) { + t.Errorf("prepareCreateLookup(%s) err: %v, must contain %v", tcase.description, err, tcase.err) + } + return + } + require.NoError(t, err) + // withTable is a vschema that already contains the table and thus + // we don't make any vschema changes and there's nothing to cancel. + require.True(t, (cancelFunc != nil) == (tcase.targetVSchema != withTable)) + utils.MustMatch(t, tcase.out, got, tcase.description) + }) } } @@ -2119,7 +2139,13 @@ func TestCreateLookupVindexSameKeyspace(t *testing.T) { t.Fatal(err) } - _, got, _, _, err := env.ws.prepareCreateLookup(ctx, "keyspace", ms.TargetKeyspace, specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + _, got, _, _, err := w.prepareCreateLookup(ctx, "keyspace", ms.TargetKeyspace, specs, false) require.NoError(t, err) if !proto.Equal(got, want) { t.Errorf("same keyspace: got:\n%v, want\n%v", got, want) @@ -2245,7 +2271,13 @@ func TestCreateCustomizedVindex(t *testing.T) { t.Fatal(err) } - _, got, _, _, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + _, got, _, _, err := w.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) require.NoError(t, err) if !proto.Equal(got, want) { t.Errorf("customize create lookup error same: got:\n%v, want\n%v", got, want) @@ -2363,7 +2395,13 @@ func TestCreateLookupVindexIgnoreNulls(t *testing.T) { t.Fatal(err) } - ms, ks, _, _, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + ms, ks, _, _, err := w.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) require.NoError(t, err) if !proto.Equal(wantKs, ks) { t.Errorf("unexpected keyspace value: got:\n%v, want\n%v", ks, wantKs) @@ -2443,11 +2481,17 @@ func TestStopAfterCopyFlag(t *testing.T) { t.Fatal(err) } - ms1, _, _, _, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) + w := &workflow{ + ts: env.ws.ts, + tmc: env.ws.tmc, + logger: env.ws.Logger(), + parser: env.ws.SQLParser(), + } + ms1, _, _, _, err := w.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, false) require.NoError(t, err) require.Equal(t, ms1.StopAfterCopy, true) - ms2, _, _, _, err := env.ws.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, true) + ms2, _, _, _, err := w.prepareCreateLookup(ctx, "workflow", ms.TargetKeyspace, specs, true) require.NoError(t, err) require.Equal(t, ms2.StopAfterCopy, false) } diff --git a/go/vt/vtctl/workflow/server.go b/go/vt/vtctl/workflow/server.go index 1f1fc18b691..d62dfada001 100644 --- a/go/vt/vtctl/workflow/server.go +++ b/go/vt/vtctl/workflow/server.go @@ -29,18 +29,15 @@ import ( "time" "github.com/google/uuid" - "golang.org/x/exp/maps" "golang.org/x/sync/errgroup" "golang.org/x/sync/semaphore" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/encoding/prototext" - "google.golang.org/protobuf/proto" "vitess.io/vitess/go/constants/sidecar" "vitess.io/vitess/go/protoutil" "vitess.io/vitess/go/ptr" - "vitess.io/vitess/go/sqlescape" "vitess.io/vitess/go/sqltypes" "vitess.io/vitess/go/trace" "vitess.io/vitess/go/vt/concurrency" @@ -49,7 +46,6 @@ import ( "vitess.io/vitess/go/vt/log" "vitess.io/vitess/go/vt/logutil" "vitess.io/vitess/go/vt/mysqlctl/tmutils" - "vitess.io/vitess/go/vt/schema" "vitess.io/vitess/go/vt/sqlparser" "vitess.io/vitess/go/vt/topo" "vitess.io/vitess/go/vt/topo/topoproto" @@ -571,7 +567,13 @@ func (s *Server) LookupVindexCreate(ctx context.Context, req *vtctldatapb.Lookup span.Annotate("cells", req.Cells) span.Annotate("tablet_types", req.TabletTypes) - ms, sourceVSchema, targetVSchema, cancelFunc, err := s.prepareCreateLookup(ctx, req.Workflow, req.Keyspace, req.Vindex, req.ContinueAfterCopyWithOwner) + w := &workflow{ + ts: s.ts, + tmc: s.tmc, + logger: s.Logger(), + parser: s.SQLParser(), + } + ms, sourceVSchema, targetVSchema, cancelFunc, err := w.prepareCreateLookup(ctx, req.Workflow, req.Keyspace, req.Vindex, req.ContinueAfterCopyWithOwner) if err != nil { return nil, err } @@ -3416,417 +3418,6 @@ func fillStringTemplate(tmpl string, vars any) (string, error) { return data.String(), nil } -// prepareCreateLookup performs the preparatory steps for creating a -// Lookup Vindex. -func (s *Server) prepareCreateLookup(ctx context.Context, workflow, keyspace string, specs *vschemapb.Keyspace, continueAfterCopyWithOwner bool) ( - ms *vtctldatapb.MaterializeSettings, sourceVSchema, targetVSchema *vschemapb.Keyspace, cancelFunc func() error, err error) { - // Important variables are pulled out here. - var ( - vindexName string - vindex *vschemapb.Vindex - targetKeyspace string - targetTableName string - vindexFromCols []string - vindexToCol string - vindexIgnoreNulls bool - - sourceTableName string - // sourceTable is the supplied table info. - sourceTable *vschemapb.Table - // sourceVSchemaTable is the table info present in the vschema. - sourceVSchemaTable *vschemapb.Table - // sourceVindexColumns are computed from the input sourceTable. - sourceVindexColumns []string - - // Target table info. - createDDL string - materializeQuery string - ) - - // Validate input vindex. - if specs == nil { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "no vindex provided") - } - if len(specs.Vindexes) != 1 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "only one vindex must be specified") - } - vindexName = maps.Keys(specs.Vindexes)[0] - vindex = maps.Values(specs.Vindexes)[0] - if !strings.Contains(vindex.Type, "lookup") { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex %s is not a lookup type", vindex.Type) - } - targetKeyspace, targetTableName, err = s.env.Parser().ParseTable(vindex.Params["table"]) - if err != nil || targetKeyspace == "" { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex table name (%s) must be in the form .
", vindex.Params["table"]) - } - vindexFromCols = strings.Split(vindex.Params["from"], ",") - for i, col := range vindexFromCols { - vindexFromCols[i] = strings.TrimSpace(col) - } - if strings.Contains(vindex.Type, "unique") { - if len(vindexFromCols) != 1 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unique vindex 'from' should have only one column") - } - } else { - if len(vindexFromCols) < 2 { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "non-unique vindex 'from' should have more than one column") - } - } - vindexToCol = vindex.Params["to"] - // Make the vindex write_only. If one exists already in the vschema, - // it will need to match this vindex exactly, including the write_only setting. - vindex.Params["write_only"] = "true" - // See if we can create the vindex without errors. - if _, err := vindexes.CreateVindex(vindex.Type, vindexName, vindex.Params); err != nil { - return nil, nil, nil, nil, err - } - if ignoreNullsStr, ok := vindex.Params["ignore_nulls"]; ok { - // This mirrors the behavior of vindexes.boolFromMap(). - switch ignoreNullsStr { - case "true": - vindexIgnoreNulls = true - case "false": - vindexIgnoreNulls = false - default: - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ignore_nulls (%s) value must be 'true' or 'false'", - ignoreNullsStr) - } - } - - // Validate input table. - if len(specs.Tables) < 1 || len(specs.Tables) > 2 { - return nil, nil, nil, nil, fmt.Errorf("one or two tables must be specified") - } - // Loop executes once or twice. - for tableName, table := range specs.Tables { - if len(table.ColumnVindexes) != 1 { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "exactly one ColumnVindex must be specified for the %s table", - tableName) - } - if tableName != targetTableName { // This is the source table. - sourceTableName = tableName - sourceTable = table - continue - } - // This is a primary vindex definition for the target table - // which allows you to override the vindex type used. - var vindexCols []string - if len(table.ColumnVindexes[0].Columns) != 0 { - vindexCols = table.ColumnVindexes[0].Columns - } else { - if table.ColumnVindexes[0].Column == "" { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", - tableName) - } - vindexCols = []string{table.ColumnVindexes[0].Column} - } - if !slices.Equal(vindexCols, vindexFromCols) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "columns in the lookup table %s primary vindex (%s) don't match the 'from' columns specified (%s)", - tableName, strings.Join(vindexCols, ","), strings.Join(vindexFromCols, ",")) - } - } - - // Validate input table and vindex consistency. - if sourceTable == nil || len(sourceTable.ColumnVindexes) != 1 { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "No ColumnVindex found for the owner table (%s) in the %s keyspace", - sourceTable, keyspace) - } - if sourceTable.ColumnVindexes[0].Name != vindexName { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ColumnVindex name (%s) must match vindex name (%s)", - sourceTable.ColumnVindexes[0].Name, vindexName) - } - if vindex.Owner != "" && vindex.Owner != sourceTableName { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "vindex owner (%s) must match table name (%s)", - vindex.Owner, sourceTableName) - } - if len(sourceTable.ColumnVindexes[0].Columns) != 0 { - sourceVindexColumns = sourceTable.ColumnVindexes[0].Columns - } else { - if sourceTable.ColumnVindexes[0].Column == "" { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "at least one column must be specified in ColumnVindexes for the %s table", - sourceTableName) - } - sourceVindexColumns = []string{sourceTable.ColumnVindexes[0].Column} - } - if len(sourceVindexColumns) != len(vindexFromCols) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "length of table columns (%d) differs from length of vindex columns (%d)", - len(sourceVindexColumns), len(vindexFromCols)) - } - - // Validate against source vschema. - sourceVSchema, err = s.ts.GetVSchema(ctx, keyspace) - if err != nil { - return nil, nil, nil, nil, err - } - if sourceVSchema.Vindexes == nil { - sourceVSchema.Vindexes = make(map[string]*vschemapb.Vindex) - } - // If source and target keyspaces are the same, make vschemas point - // to the same object. - if keyspace == targetKeyspace { - targetVSchema = sourceVSchema - } else { - targetVSchema, err = s.ts.GetVSchema(ctx, targetKeyspace) - if err != nil { - return nil, nil, nil, nil, err - } - } - if targetVSchema.Vindexes == nil { - targetVSchema.Vindexes = make(map[string]*vschemapb.Vindex) - } - if targetVSchema.Tables == nil { - targetVSchema.Tables = make(map[string]*vschemapb.Table) - } - if existing, ok := sourceVSchema.Vindexes[vindexName]; ok { - if !proto.Equal(existing, vindex) { // If the exact same vindex already exists then we can re-use it - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "a conflicting vindex named %s already exists in the %s keyspace", - vindexName, keyspace) - } - } - sourceVSchemaTable = sourceVSchema.Tables[sourceTableName] - if sourceVSchemaTable == nil && !schema.IsInternalOperationTableName(sourceTableName) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "table %s not found in the %s keyspace", sourceTableName, keyspace) - } - for _, colVindex := range sourceVSchemaTable.ColumnVindexes { - // For a conflict, the vindex name and column should match. - if colVindex.Name != vindexName { - continue - } - var colNames []string - if len(colVindex.Columns) == 0 { - colNames = []string{colVindex.Column} - } else { - colNames = colVindex.Columns - } - // If this is the exact same definition then we can use the existing one. If they - // are not the same then they are two distinct conflicting vindexes and we should - // not proceed. - if !slices.Equal(colNames, sourceVindexColumns) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting ColumnVindex on column(s) %s in table %s already exists in the %s keyspace", - strings.Join(colNames, ","), sourceTableName, keyspace) - } - } - - // Validate against source schema. - sourceShards, err := s.ts.GetServingShards(ctx, keyspace) - if err != nil { - return nil, nil, nil, nil, err - } - onesource := sourceShards[0] - if onesource.PrimaryAlias == nil { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "source shard %s has no primary", onesource.ShardName()) - } - req := &tabletmanagerdatapb.GetSchemaRequest{Tables: []string{sourceTableName}} - tableSchema, err := schematools.GetSchema(ctx, s.ts, s.tmc, onesource.PrimaryAlias, req) - if err != nil { - return nil, nil, nil, nil, err - } - if len(tableSchema.TableDefinitions) != 1 { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected number of tables (%d) returned from %s schema", - len(tableSchema.TableDefinitions), keyspace) - } - - // Generate "create table" statement. - lines := strings.Split(tableSchema.TableDefinitions[0].Schema, "\n") - if len(lines) < 3 { - // Should never happen. - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "schema looks incorrect: %s, expecting at least four lines", - tableSchema.TableDefinitions[0].Schema) - } - var modified []string - modified = append(modified, strings.Replace(lines[0], sourceTableName, targetTableName, 1)) - for i := range sourceVindexColumns { - line, err := generateColDef(lines, sourceVindexColumns[i], vindexFromCols[i]) - if err != nil { - return nil, nil, nil, nil, err - } - modified = append(modified, line) - } - - if vindex.Params["data_type"] == "" || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { - modified = append(modified, fmt.Sprintf(" %s varbinary(128),", sqlescape.EscapeID(vindexToCol))) - } else { - modified = append(modified, fmt.Sprintf(" %s %s,", sqlescape.EscapeID(vindexToCol), sqlescape.EscapeID(vindex.Params["data_type"]))) - } - buf := sqlparser.NewTrackedBuffer(nil) - fmt.Fprintf(buf, " PRIMARY KEY (") - prefix := "" - for _, col := range vindexFromCols { - fmt.Fprintf(buf, "%s%s", prefix, sqlescape.EscapeID(col)) - prefix = ", " - } - fmt.Fprintf(buf, ")") - modified = append(modified, buf.String()) - modified = append(modified, ")") - createDDL = strings.Join(modified, "\n") - // Confirm that our DDL is valid before we create anything. - if _, err = s.env.Parser().ParseStrictDDL(createDDL); err != nil { - return nil, nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "error: %v; invalid lookup table definition generated: %s", - err, createDDL) - } - - // Generate vreplication query. - buf = sqlparser.NewTrackedBuffer(nil) - buf.Myprintf("select ") - for i := range vindexFromCols { - buf.Myprintf("%s as %s, ", sqlparser.String(sqlparser.NewIdentifierCI(sourceVindexColumns[i])), sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - } - if strings.EqualFold(vindexToCol, "keyspace_id") || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") { - buf.Myprintf("keyspace_id() as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } else { - buf.Myprintf("%s as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)), sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } - buf.Myprintf("from %s", sqlparser.String(sqlparser.NewIdentifierCS(sourceTableName))) - if vindexIgnoreNulls { - buf.Myprintf(" where ") - lastValIdx := len(vindexFromCols) - 1 - for i := range vindexFromCols { - buf.Myprintf("%s is not null", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - if i != lastValIdx { - buf.Myprintf(" and ") - } - } - } - if vindex.Owner != "" { - // Only backfill. - buf.Myprintf(" group by ") - for i := range vindexFromCols { - buf.Myprintf("%s, ", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i]))) - } - buf.Myprintf("%s", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol))) - } - materializeQuery = buf.String() - - // Save a copy of the original vschema if we modify it and need to provide - // a cancelFunc. - ogTargetVSchema := targetVSchema.CloneVT() - targetChanged := false - - // Update targetVSchema. - targetTable := specs.Tables[targetTableName] - if targetVSchema.Sharded { - // Choose a primary vindex type for the lookup table based on the source - // definition if one was not explicitly specified. - var targetVindexType string - var targetVindex *vschemapb.Vindex - for _, field := range tableSchema.TableDefinitions[0].Fields { - if sourceVindexColumns[0] == field.Name { - if targetTable != nil && len(targetTable.ColumnVindexes) > 0 { - targetVindexType = targetTable.ColumnVindexes[0].Name - } - if targetVindexType == "" { - targetVindexType, err = vindexes.ChooseVindexForType(field.Type) - if err != nil { - return nil, nil, nil, nil, err - } - } - targetVindex = &vschemapb.Vindex{ - Type: targetVindexType, - } - break - } - } - if targetVindex == nil { - // Unreachable. We validated column names when generating the DDL. - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INTERNAL, "column %s not found in target schema %s", - sourceVindexColumns[0], tableSchema.TableDefinitions[0].Schema) - } - if existing, ok := targetVSchema.Vindexes[targetVindexType]; ok { - if !proto.Equal(existing, targetVindex) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting vindex named %v already exists in the %s keyspace", - targetVindexType, targetKeyspace) - } - } else { - targetVSchema.Vindexes[targetVindexType] = targetVindex - targetChanged = true - } - - targetTable = &vschemapb.Table{ - ColumnVindexes: []*vschemapb.ColumnVindex{{ - Column: vindexFromCols[0], - Name: targetVindexType, - }}, - } - } else { - targetTable = &vschemapb.Table{} - } - if existing, ok := targetVSchema.Tables[targetTableName]; ok { - if !proto.Equal(existing, targetTable) { - return nil, nil, nil, nil, - vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "a conflicting table named %s already exists in the %s vschema", - targetTableName, targetKeyspace) - } - } else { - targetVSchema.Tables[targetTableName] = targetTable - targetChanged = true - } - - if targetChanged { - cancelFunc = func() error { - // Restore the original target vschema. - return s.ts.SaveVSchema(ctx, targetKeyspace, ogTargetVSchema) - } - } - - ms = &vtctldatapb.MaterializeSettings{ - Workflow: workflow, - MaterializationIntent: vtctldatapb.MaterializationIntent_CREATELOOKUPINDEX, - SourceKeyspace: keyspace, - TargetKeyspace: targetKeyspace, - StopAfterCopy: vindex.Owner != "" && !continueAfterCopyWithOwner, - TableSettings: []*vtctldatapb.TableMaterializeSettings{{ - TargetTable: targetTableName, - SourceExpression: materializeQuery, - CreateDdl: createDDL, - }}, - } - - // Update sourceVSchema - sourceVSchema.Vindexes[vindexName] = vindex - sourceVSchemaTable.ColumnVindexes = append(sourceVSchemaTable.ColumnVindexes, sourceTable.ColumnVindexes[0]) - - return ms, sourceVSchema, targetVSchema, cancelFunc, nil -} - -// func generateColDef(lines []string, sourceVindexCol, vindexFromCol string) (string, error) { -// source := sqlescape.EscapeID(sourceVindexCol) -// target := sqlescape.EscapeID(vindexFromCol) - -// for _, line := range lines[1:] { -// if strings.Contains(line, source) { -// line = strings.Replace(line, source, target, 1) -// line = strings.Replace(line, " AUTO_INCREMENT", "", 1) -// line = strings.Replace(line, " DEFAULT NULL", "", 1) -// // Ensure that the column definition ends with a comma as we will -// // be appending the TO column and PRIMARY KEY definitions. If the -// // souce column here was the last entity defined in the source -// // table's definition then it will not already have the comma. -// if !strings.HasSuffix(strings.TrimSpace(line), ",") { -// line += "," -// } -// return line, nil -// } -// } -// return "", fmt.Errorf("column %s not found in schema %v", sourceVindexCol, lines) -// } - func (s *Server) MigrateCreate(ctx context.Context, req *vtctldatapb.MigrateCreateRequest) (*vtctldatapb.WorkflowStatusResponse, error) { moveTablesCreateRequest := &vtctldatapb.MoveTablesCreateRequest{ Workflow: req.Workflow,