Skip to content

Commit

Permalink
tabledesc: improve index column ID and name validation
Browse files Browse the repository at this point in the history 
This commit adds missing index column ID and name validation, in
particular concerning stored- and key-suffix-columns.

Fixes #72771.

Release note: None
  • Loading branch information
Marius Posta committed Jan 14, 2022
1 parent 20eaf0b commit 218f887
Show file tree
Hide file tree
Showing 2 changed files with 130 additions and 2 deletions.
23 changes: 21 additions & 2 deletions pkg/sql/catalog/tabledesc/validate.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -834,19 +834,38 @@ func (desc *wrapper) validateTableIndexes(columnNames map[string]descpb.ColumnID

var validateIndexDup catalog.TableColSet
for i, name := range idx.IndexDesc().KeyColumnNames {
inIndexColID := idx.IndexDesc().KeyColumnIDs[i]
colID, ok := columnNames[name]
if !ok {
return errors.Newf("index %q contains unknown column %q", idx.GetName(), name)
}
if colID != idx.IndexDesc().KeyColumnIDs[i] {
if colID != inIndexColID {
return errors.Newf("index %q column %q should have ID %d, but found ID %d",
idx.GetName(), name, colID, idx.IndexDesc().KeyColumnIDs[i])
idx.GetName(), name, colID, inIndexColID)
}
if validateIndexDup.Contains(colID) {
return pgerror.Newf(pgcode.FeatureNotSupported, "index %q contains duplicate column %q", idx.GetName(), name)
}
validateIndexDup.Add(colID)
}
for i, colID := range idx.IndexDesc().StoreColumnIDs {
inIndexColName := idx.IndexDesc().StoreColumnNames[i]
col, exists := columnsByID[colID]
if !exists {
return errors.Newf("index %q contains stored column %q with unknown ID %d", idx.GetName(), inIndexColName, colID)
}
if col.GetName() != inIndexColName {
return errors.Newf("index %q stored column ID %d should have name %q, but found name %q",
idx.GetName(), colID, col.ColName(), inIndexColName)
}
}
for _, colID := range idx.IndexDesc().KeySuffixColumnIDs {
if _, exists := columnsByID[colID]; !exists {
return errors.Newf("index %q key suffix column ID %d is invalid",
idx.GetName(), colID)
}
}

if idx.IsSharded() {
if err := desc.ensureShardedIndexNotComputed(idx.IndexDesc()); err != nil {
return err
Expand Down
109 changes: 109 additions & 0 deletions pkg/sql/catalog/tabledesc/validate_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -854,6 +854,115 @@ func TestValidateTableDesc(t *testing.T) {
NextFamilyID: 1,
NextIndexID: 2,
}},
{`index "secondary" contains stored column "quux" with unknown ID 123`,
descpb.TableDescriptor{
ID: 2,
ParentID: 1,
Name: "foo",
FormatVersion: descpb.InterleavedFormatVersion,
Columns: []descpb.ColumnDescriptor{
{ID: 1, Name: "bar"},
{ID: 2, Name: "baz"},
},
Families: []descpb.ColumnFamilyDescriptor{
{ID: 0, Name: "primary", ColumnIDs: []descpb.ColumnID{1, 2}, ColumnNames: []string{"bar", "baz"}},
},
PrimaryIndex: descpb.IndexDescriptor{
ID: 1,
Name: "primary",
KeyColumnIDs: []descpb.ColumnID{1},
KeyColumnNames: []string{"bar"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
EncodingType: descpb.PrimaryIndexEncoding,
Version: descpb.PrimaryIndexWithStoredColumnsVersion,
},
Indexes: []descpb.IndexDescriptor{{
ID: 2,
Name: "secondary",
KeyColumnIDs: []descpb.ColumnID{1},
KeyColumnNames: []string{"bar"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
StoreColumnIDs: []descpb.ColumnID{123},
StoreColumnNames: []string{"quux"},
}},
NextColumnID: 3,
NextFamilyID: 1,
NextIndexID: 3,
}},
{`index "secondary" stored column ID 2 should have name "baz", but found name "quux"`,
descpb.TableDescriptor{
ID: 2,
ParentID: 1,
Name: "foo",
FormatVersion: descpb.InterleavedFormatVersion,
Columns: []descpb.ColumnDescriptor{
{ID: 1, Name: "bar"},
{ID: 2, Name: "baz"},
{ID: 3, Name: "quux"},
},
Families: []descpb.ColumnFamilyDescriptor{{
ID: 0,
Name: "primary",
ColumnIDs: []descpb.ColumnID{1, 2, 3},
ColumnNames: []string{"bar", "baz", "quux"},
}},
PrimaryIndex: descpb.IndexDescriptor{
ID: 1,
Name: "primary",
KeyColumnIDs: []descpb.ColumnID{1},
KeyColumnNames: []string{"bar"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
EncodingType: descpb.PrimaryIndexEncoding,
Version: descpb.PrimaryIndexWithStoredColumnsVersion,
},
Indexes: []descpb.IndexDescriptor{{
ID: 2,
Name: "secondary",
KeyColumnIDs: []descpb.ColumnID{2},
KeyColumnNames: []string{"baz"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
KeySuffixColumnIDs: []descpb.ColumnID{1},
StoreColumnIDs: []descpb.ColumnID{2},
StoreColumnNames: []string{"quux"},
}},
NextColumnID: 4,
NextFamilyID: 1,
NextIndexID: 3,
}},
{`index "secondary" key suffix column ID 123 is invalid`,
descpb.TableDescriptor{
ID: 2,
ParentID: 1,
Name: "foo",
FormatVersion: descpb.InterleavedFormatVersion,
Columns: []descpb.ColumnDescriptor{
{ID: 1, Name: "bar"},
{ID: 2, Name: "baz"},
},
Families: []descpb.ColumnFamilyDescriptor{
{ID: 0, Name: "primary", ColumnIDs: []descpb.ColumnID{1, 2}, ColumnNames: []string{"bar", "baz"}},
},
PrimaryIndex: descpb.IndexDescriptor{
ID: 1,
Name: "primary",
KeyColumnIDs: []descpb.ColumnID{1},
KeyColumnNames: []string{"bar"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
EncodingType: descpb.PrimaryIndexEncoding,
Version: descpb.PrimaryIndexWithStoredColumnsVersion,
},
Indexes: []descpb.IndexDescriptor{{
ID: 2,
Name: "secondary",
KeyColumnIDs: []descpb.ColumnID{2},
KeyColumnNames: []string{"baz"},
KeyColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
KeySuffixColumnIDs: []descpb.ColumnID{123},
}},
NextColumnID: 3,
NextFamilyID: 1,
NextIndexID: 3,
}},
{`index "primary" contains deprecated foreign key representation`,
descpb.TableDescriptor{
ID: 2,
Expand Down

0 comments on commit 218f887

Please sign in to comment.