Merge #57998

57998: sql: check use of virtual columns in indexes r=RaduBerinde a=RaduBerinde

This commit adds checks that disallow virtual columns being used as
primary key columns or stored columns in indexes.

Release note: None

Informs #57608.

Co-authored-by: Radu Berinde <[email protected]>

pkg/sql/catalog/tabledesc/structured.go

@@ -2307,6 +2307,20 @@ func (desc *Immutable) validateTableIndexes(columnNames map[string]descpb.Column
 		return ErrMissingPrimaryKey
 	}
 
+	var virtualCols catalog.TableColSet
+	for i := range desc.Columns {
+		if desc.Columns[i].Virtual {
+			virtualCols.Add(desc.Columns[i].ID)
+		}
+	}
+
+	// Verify that the primary index columns are not virtual.
+	for i, col := range desc.PrimaryIndex.ColumnIDs {
+		if virtualCols.Contains(col) {
+			return fmt.Errorf("primary index column %q cannot be virtual", desc.PrimaryIndex.ColumnNames[i])
+		}
+	}
+
 	indexNames := map[string]struct{}{}
 	indexIDs := map[descpb.IndexID]string{}
 	for _, index := range desc.AllNonDropIndexes() {
@@ -2349,6 +2363,12 @@ func (desc *Immutable) validateTableIndexes(columnNames map[string]descpb.Column
 			return fmt.Errorf("mismatched column IDs (%d) and directions (%d)",
 				len(index.ColumnIDs), len(index.ColumnDirections))
 		}
+		// In the old STORING encoding, stored columns are in ExtraColumnIDs;
+		// tolerate a longer list of column names.
+		if len(index.StoreColumnIDs) > len(index.StoreColumnNames) {
+			return fmt.Errorf("mismatched STORING column IDs (%d) and names (%d)",
+				len(index.StoreColumnIDs), len(index.StoreColumnNames))
+		}
 
 		if len(index.ColumnIDs) == 0 {
 			return fmt.Errorf("index %q must contain at least 1 column", index.Name)
@@ -2392,6 +2412,17 @@ func (desc *Immutable) validateTableIndexes(columnNames map[string]descpb.Column
 					index.Name, index.Predicate)
 			}
 		}
+		// Ensure that indexes do not STORE virtual columns.
+		for _, col := range index.ExtraColumnIDs {
+			if virtualCols.Contains(col) {
+				return fmt.Errorf("index %q cannot store virtual column %d", index.Name, col)
+			}
+		}
+		for i, col := range index.StoreColumnIDs {
+			if virtualCols.Contains(col) {
+				return fmt.Errorf("index %q cannot store virtual column %q", index.Name, index.StoreColumnNames[i])
+			}
+		}
 	}
 
 	return nil

pkg/sql/catalog/tabledesc/structured_test.go

@@ -670,6 +670,35 @@ func TestValidateTableDesc(t *testing.T) {
 				NextFamilyID: 1,
 				NextIndexID:  2,
 			}},
+		{`mismatched STORING column IDs (1) and names (0)`,
+			descpb.TableDescriptor{
+				ID:            2,
+				ParentID:      1,
+				Name:          "foo",
+				FormatVersion: descpb.FamilyFormatVersion,
+				Columns: []descpb.ColumnDescriptor{
+					{ID: 1, Name: "c1"},
+					{ID: 2, Name: "c2"},
+				},
+				Families: []descpb.ColumnFamilyDescriptor{
+					{
+						ID:          0,
+						Name:        "fam",
+						ColumnIDs:   []descpb.ColumnID{1, 2},
+						ColumnNames: []string{"c1", "c2"},
+					},
+				},
+				PrimaryIndex: descpb.IndexDescriptor{
+					ID: 1, Name: "primary",
+					ColumnIDs:        []descpb.ColumnID{1},
+					ColumnNames:      []string{"c1"},
+					ColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
+					StoreColumnIDs:   []descpb.ColumnID{2},
+				},
+				NextColumnID: 3,
+				NextFamilyID: 1,
+				NextIndexID:  2,
+			}},
 		{`at least one of LIST or RANGE partitioning must be used`,
 			// Verify that validatePartitioning is hooked up. The rest of these
 			// tests are in TestValidatePartitionion.
@@ -835,6 +864,63 @@ func TestValidateTableDesc(t *testing.T) {
 					},
 				},
 			}},
+		{`primary index column "v" cannot be virtual`,
+			descpb.TableDescriptor{
+				ID:            2,
+				ParentID:      1,
+				Name:          "foo",
+				FormatVersion: descpb.FamilyFormatVersion,
+				Columns: []descpb.ColumnDescriptor{
+					{ID: 1, Name: "bar"},
+					{ID: 2, Name: "v", ComputeExpr: &computedExpr, Virtual: true},
+				},
+				PrimaryIndex: descpb.IndexDescriptor{
+					ID:          1,
+					Name:        "primary",
+					Unique:      true,
+					ColumnIDs:   []descpb.ColumnID{1, 2},
+					ColumnNames: []string{"bar", "v"},
+				},
+				Families: []descpb.ColumnFamilyDescriptor{
+					{ID: 0, Name: "primary",
+						ColumnIDs:   []descpb.ColumnID{1},
+						ColumnNames: []string{"bar"},
+					},
+				},
+				NextColumnID: 3,
+				NextFamilyID: 1,
+			}},
+		{`index "sec" cannot store virtual column "v"`,
+			descpb.TableDescriptor{
+				ID:            2,
+				ParentID:      1,
+				Name:          "foo",
+				FormatVersion: descpb.FamilyFormatVersion,
+				Columns: []descpb.ColumnDescriptor{
+					{ID: 1, Name: "c1"},
+					{ID: 2, Name: "c2"},
+					{ID: 3, Name: "v", ComputeExpr: &computedExpr, Virtual: true},
+				},
+				Families: []descpb.ColumnFamilyDescriptor{
+					{ID: 0, Name: "primary", ColumnIDs: []descpb.ColumnID{1, 2}, ColumnNames: []string{"c1", "c2"}},
+				},
+				PrimaryIndex: descpb.IndexDescriptor{
+					ID: 1, Name: "pri", ColumnIDs: []descpb.ColumnID{1},
+					ColumnNames:      []string{"c1"},
+					ColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
+				},
+				Indexes: []descpb.IndexDescriptor{
+					{ID: 2, Name: "sec", ColumnIDs: []descpb.ColumnID{2},
+						ColumnNames:      []string{"c2"},
+						ColumnDirections: []descpb.IndexDescriptor_Direction{descpb.IndexDescriptor_ASC},
+						StoreColumnNames: []string{"v"},
+						StoreColumnIDs:   []descpb.ColumnID{3},
+					},
+				},
+				NextColumnID: 4,
+				NextFamilyID: 1,
+				NextIndexID:  3,
+			}},
 	}
 	for i, d := range testData {
 		t.Run(d.err, func(t *testing.T) {

pkg/sql/logictest/testdata/logic_test/virtual_columns

@@ -20,6 +20,22 @@ CREATE TABLE t (
   FAMILY (v)
 )
 
+statement error primary index column "v" cannot be virtual
+CREATE TABLE t (
+  a INT,
+  b INT,
+  v INT AS (a+b) VIRTUAL,
+  PRIMARY KEY (b,v)
+)
+
+statement error index "t_b_idx" cannot store virtual column "v"
+CREATE TABLE t (
+  a INT PRIMARY KEY,
+  b INT,
+  v INT AS (a+b) VIRTUAL,
+  INDEX (b) STORING (v)
+)
+
 statement ok
 CREATE TABLE t (
   a INT PRIMARY KEY,