Skip to content

Commit ebde310

Browse files
austenLacymattlord
austenLacy
and
mattlord
authored
[BACK-PORT] Properly support ignore_nulls in CreateLookupVindex (vitessio#13913) (#122)
* Properly support ignore_nulls in CreateLookupVindex (vitessio#13913) Signed-off-by: Matt Lord <mattalord@gmail.com> * remove new vreplication tests that dont work with v15 Signed-off-by: Austen Lacy <austen.lacy@shopify.com> --------- Signed-off-by: Matt Lord <mattalord@gmail.com> Signed-off-by: Austen Lacy <austen.lacy@shopify.com> Co-authored-by: Matt Lord <mattalord@gmail.com> Co-authored-by: Austen Lacy <austen.lacy@shopify.com>
1 parent 0869e39 commit ebde310

File tree

4 files changed

+208
-13
lines changed

4 files changed

+208
-13
lines changed

‎go/test/endtoend/vreplication/config_test.go

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ var (
3535
initialProductSchema = `
3636
create table product(pid int, description varbinary(128), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key(pid)) CHARSET=utf8mb4;
3737
create table customer(cid int, name varchar(128) collate utf8mb4_bin, meta json default null, typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),
38-
ts timestamp not null default current_timestamp, bits bit(2) default b'11', date1 datetime not null default '0000-00-00 00:00:00',
38+
ts timestamp not null default current_timestamp, bits bit(2) default b'11', date1 datetime not null default '0000-00-00 00:00:00',
3939
date2 datetime not null default '2021-00-01 00:00:00', dec80 decimal(8,0), primary key(cid,typ)) CHARSET=utf8mb4;
4040
create table customer_seq(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
4141
create table merchant(mname varchar(128), category varchar(128), primary key(mname)) CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
@@ -84,6 +84,33 @@ create table datze (id int, dt1 datetime not null default current_timestamp, dt2
8484
}
8585
}
8686
`
87+
88+
createLookupVindexVSchema = `
89+
{
90+
"sharded": true,
91+
"vindexes": {
92+
"customer_name_keyspace_id": {
93+
"type": "consistent_lookup",
94+
"params": {
95+
"table": "product.customer_name_keyspace_id",
96+
"from": "name,cid",
97+
"to": "keyspace_id",
98+
"ignore_nulls": "true"
99+
},
100+
"owner": "customer"
101+
}
102+
},
103+
"tables": {
104+
"customer": {
105+
"column_vindexes": [{
106+
"columns": ["name", "cid"],
107+
"name": "customer_name_keyspace_id"
108+
}]
109+
}
110+
}
111+
}
112+
`
113+
87114
customerSchema = ""
88115
customerVSchema = `
89116
{

‎go/vt/vttablet/tabletserver/vstreamer/planbuilder.go

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,8 @@ const (
8181
GreaterThanEqual
8282
// NotEqual is used to filter a comparable column if != specific value
8383
NotEqual
84+
// IsNotNull is used to filter a column if it is NULL
85+
IsNotNull
8486
)
8587

8688
// Filter contains opcodes for filtering.
@@ -226,6 +228,10 @@ func (plan *Plan) filter(values, result []sqltypes.Value, charsets []collations.
226228
if !key.KeyRangeContains(filter.KeyRange, ksid) {
227229
return false, nil
228230
}
231+
case IsNotNull:
232+
if values[filter.ColNum].IsNull() {
233+
return false, nil
234+
}
229235
default:
230236
match, err := compare(filter.Opcode, values[filter.ColNum], filter.Value, charsets[filter.ColNum])
231237
if err != nil {
@@ -552,6 +558,25 @@ func (plan *Plan) analyzeWhere(vschema *localVSchema, where *sqlparser.Where) er
552558
if err := plan.analyzeInKeyRange(vschema, expr.Exprs); err != nil {
553559
return err
554560
}
561+
case *sqlparser.IsExpr: // Needed for CreateLookupVindex with ignore_nulls
562+
if expr.Right != sqlparser.IsNotNullOp {
563+
return fmt.Errorf("unsupported constraint: %v", sqlparser.String(expr))
564+
}
565+
qualifiedName, ok := expr.Left.(*sqlparser.ColName)
566+
if !ok {
567+
return fmt.Errorf("unexpected: %v", sqlparser.String(expr))
568+
}
569+
if !qualifiedName.Qualifier.IsEmpty() {
570+
return fmt.Errorf("unsupported qualifier for column: %v", sqlparser.String(qualifiedName))
571+
}
572+
colnum, err := findColumn(plan.Table, qualifiedName.Name)
573+
if err != nil {
574+
return err
575+
}
576+
plan.Filters = append(plan.Filters, Filter{
577+
Opcode: IsNotNull,
578+
ColNum: colnum,
579+
})
555580
default:
556581
return fmt.Errorf("unsupported constraint: %v", sqlparser.String(expr))
557582
}

‎go/vt/wrangler/materializer.go

Lines changed: 36 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@ import (
5454
tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
5555
vschemapb "vitess.io/vitess/go/vt/proto/vschema"
5656
vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
57+
vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
5758
)
5859

5960
type materializer struct {
@@ -440,12 +441,13 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
440441
// Important variables are pulled out here.
441442
var (
442443
// lookup vindex info
443-
vindexName string
444-
vindex *vschemapb.Vindex
445-
targetKeyspace string
446-
targetTableName string
447-
vindexFromCols []string
448-
vindexToCol string
444+
vindexName string
445+
vindex *vschemapb.Vindex
446+
targetKeyspace string
447+
targetTableName string
448+
vindexFromCols []string
449+
vindexToCol string
450+
vindexIgnoreNulls bool
449451

450452
// source table info
451453
sourceTableName string
@@ -496,6 +498,18 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
496498
if _, err := vindexes.CreateVindex(vindex.Type, vindexName, vindex.Params); err != nil {
497499
return nil, nil, nil, err
498500
}
501+
if ignoreNullsStr, ok := vindex.Params["ignore_nulls"]; ok {
502+
// This mirrors the behavior of vindexes.boolFromMap().
503+
switch ignoreNullsStr {
504+
case "true":
505+
vindexIgnoreNulls = true
506+
case "false":
507+
vindexIgnoreNulls = false
508+
default:
509+
return nil, nil, nil, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "ignore_nulls value must be 'true' or 'false': '%s'",
510+
ignoreNullsStr)
511+
}
512+
}
499513

500514
// Validate input table
501515
if len(specs.Tables) != 1 {
@@ -632,21 +646,31 @@ func (wr *Wrangler) prepareCreateLookup(ctx context.Context, keyspace string, sp
632646
buf = sqlparser.NewTrackedBuffer(nil)
633647
buf.Myprintf("select ")
634648
for i := range vindexFromCols {
635-
buf.Myprintf("%v as %v, ", sqlparser.NewIdentifierCI(sourceVindexColumns[i]), sqlparser.NewIdentifierCI(vindexFromCols[i]))
649+
buf.Myprintf("%s as %s, ", sqlparser.String(sqlparser.NewIdentifierCI(sourceVindexColumns[i])), sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i])))
636650
}
637651
if strings.EqualFold(vindexToCol, "keyspace_id") || strings.EqualFold(vindex.Type, "consistent_lookup_unique") || strings.EqualFold(vindex.Type, "consistent_lookup") {
638-
buf.Myprintf("keyspace_id() as %v ", sqlparser.NewIdentifierCI(vindexToCol))
652+
buf.Myprintf("keyspace_id() as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)))
639653
} else {
640-
buf.Myprintf("%v as %v ", sqlparser.NewIdentifierCI(vindexToCol), sqlparser.NewIdentifierCI(vindexToCol))
654+
buf.Myprintf("%s as %s ", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)), sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)))
655+
}
656+
buf.Myprintf("from %s", sqlparser.String(sqlparser.NewIdentifierCS(sourceTableName)))
657+
if vindexIgnoreNulls {
658+
buf.Myprintf(" where ")
659+
lastValIdx := len(vindexFromCols) - 1
660+
for i := range vindexFromCols {
661+
buf.Myprintf("%s is not null", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i])))
662+
if i != lastValIdx {
663+
buf.Myprintf(" and ")
664+
}
665+
}
641666
}
642-
buf.Myprintf("from %v", sqlparser.NewIdentifierCS(sourceTableName))
643667
if vindex.Owner != "" {
644668
// Only backfill
645669
buf.Myprintf(" group by ")
646670
for i := range vindexFromCols {
647-
buf.Myprintf("%v, ", sqlparser.NewIdentifierCI(vindexFromCols[i]))
671+
buf.Myprintf("%s, ", sqlparser.String(sqlparser.NewIdentifierCI(vindexFromCols[i])))
648672
}
649-
buf.Myprintf("%v", sqlparser.NewIdentifierCI(vindexToCol))
673+
buf.Myprintf("%s", sqlparser.String(sqlparser.NewIdentifierCI(vindexToCol)))
650674
}
651675
materializeQuery = buf.String()
652676

‎go/vt/wrangler/materializer_test.go

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1274,6 +1274,125 @@ func TestCreateCustomizedVindex(t *testing.T) {
12741274
}
12751275
}
12761276

1277+
func TestCreateLookupVindexIgnoreNulls(t *testing.T) {
1278+
ms := &vtctldatapb.MaterializeSettings{
1279+
SourceKeyspace: "ks",
1280+
TargetKeyspace: "ks",
1281+
}
1282+
1283+
env := newTestMaterializerEnv(t, ms, []string{"0"}, []string{"0"})
1284+
defer env.close()
1285+
1286+
specs := &vschemapb.Keyspace{
1287+
Vindexes: map[string]*vschemapb.Vindex{
1288+
"v": {
1289+
Type: "consistent_lookup",
1290+
Params: map[string]string{
1291+
"table": "ks.lkp",
1292+
"from": "col2,col1",
1293+
"to": "keyspace_id",
1294+
"ignore_nulls": "true",
1295+
},
1296+
Owner: "t1",
1297+
},
1298+
},
1299+
Tables: map[string]*vschemapb.Table{
1300+
"t1": {
1301+
ColumnVindexes: []*vschemapb.ColumnVindex{{
1302+
Name: "v",
1303+
Columns: []string{"col2", "col1"},
1304+
}},
1305+
},
1306+
},
1307+
}
1308+
// Dummy sourceSchema
1309+
sourceSchema := "CREATE TABLE `t1` (\n" +
1310+
" `col1` int(11) NOT NULL AUTO_INCREMENT,\n" +
1311+
" `col2` int(11) DEFAULT NULL,\n" +
1312+
" PRIMARY KEY (`id`)\n" +
1313+
") ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=latin1"
1314+
1315+
vschema := &vschemapb.Keyspace{
1316+
Sharded: true,
1317+
Vindexes: map[string]*vschemapb.Vindex{
1318+
"hash": {
1319+
Type: "hash",
1320+
},
1321+
},
1322+
Tables: map[string]*vschemapb.Table{
1323+
"t1": {
1324+
ColumnVindexes: []*vschemapb.ColumnVindex{{
1325+
Name: "hash",
1326+
Column: "col1",
1327+
}},
1328+
},
1329+
},
1330+
}
1331+
1332+
wantKs := &vschemapb.Keyspace{
1333+
Sharded: true,
1334+
Vindexes: map[string]*vschemapb.Vindex{
1335+
"hash": {
1336+
Type: "hash",
1337+
},
1338+
"v": {
1339+
Type: "consistent_lookup",
1340+
Params: map[string]string{
1341+
"table": "ks.lkp",
1342+
"from": "col2,col1",
1343+
"to": "keyspace_id",
1344+
"write_only": "true",
1345+
"ignore_nulls": "true",
1346+
},
1347+
Owner: "t1",
1348+
},
1349+
},
1350+
Tables: map[string]*vschemapb.Table{
1351+
"t1": {
1352+
ColumnVindexes: []*vschemapb.ColumnVindex{{
1353+
Name: "hash",
1354+
Column: "col1",
1355+
}, {
1356+
Name: "v",
1357+
Columns: []string{"col2", "col1"},
1358+
}},
1359+
},
1360+
"lkp": {
1361+
ColumnVindexes: []*vschemapb.ColumnVindex{{
1362+
Column: "col2",
1363+
Name: "hash",
1364+
}},
1365+
},
1366+
},
1367+
}
1368+
wantQuery := "select col2 as col2, col1 as col1, keyspace_id() as keyspace_id from t1 where col2 is not null and col1 is not null group by col2, col1, keyspace_id"
1369+
1370+
env.tmc.schema[ms.SourceKeyspace+".t1"] = &tabletmanagerdatapb.SchemaDefinition{
1371+
TableDefinitions: []*tabletmanagerdatapb.TableDefinition{{
1372+
Fields: []*querypb.Field{{
1373+
Name: "col1",
1374+
Type: querypb.Type_INT64,
1375+
}, {
1376+
Name: "col2",
1377+
Type: querypb.Type_INT64,
1378+
}},
1379+
Schema: sourceSchema,
1380+
}},
1381+
}
1382+
if err := env.topoServ.SaveVSchema(context.Background(), ms.SourceKeyspace, vschema); err != nil {
1383+
t.Fatal(err)
1384+
}
1385+
1386+
ms, ks, _, err := env.wr.prepareCreateLookup(context.Background(), ms.SourceKeyspace, specs, false)
1387+
require.NoError(t, err)
1388+
if !proto.Equal(wantKs, ks) {
1389+
t.Errorf("unexpected keyspace value: got:\n%v, want\n%v", ks, wantKs)
1390+
}
1391+
require.NotNil(t, ms)
1392+
require.GreaterOrEqual(t, len(ms.TableSettings), 1)
1393+
require.Equal(t, wantQuery, ms.TableSettings[0].SourceExpression, "unexpected query")
1394+
}
1395+
12771396
func TestStopAfterCopyFlag(t *testing.T) {
12781397
ms := &vtctldatapb.MaterializeSettings{
12791398
SourceKeyspace: "ks",

0 commit comments

Comments
 (0)