sql, *: allow table scoping under pg_temp_<session_id> physical schema

Previously, CRDB only supported the `public` physical schema. All table
entries in `system.namespace` assumed an implied `public` schema, so
name resolution for tables only required a databaseID and table name to
uniquely identify a table.

As highlighted in the temp tables RFC, temp tables will be scoped under
a session specific schema of the form `pg_temp_<session_id>`. This
motivated adding support for additional physical schemas.

This PR involves the following changes to `system.namespace`:

- A new `system.namespace` table is constructed for cluster versions
>= 20.1, which has an additional primary key column `publicSchemaID`.

- The older `system.namespace` table is marked deprecated. All
`system.namespace` read accesses default to the new `system.namespace`.
If a match isn't found, the deprecated `system.namespace` is checked.

- All `system.namespace` write accesses for key deletions remove entries
from both versions of the table. This ensures the fallback code doesn't
read an old key that was deleted.

- All `system.namespace` write accesses that involve creating new entries
are added to the `system.namespace` table according to the cluster version.

- Selecting from `system.namespace` in mixed version clusters actually
selects from `system.namespace_deprecated`, ensuring that the change is
invisible to users.

- The new `system.namespace` table is moved out of the SystemConfig
range. This means `system.namespace` is no longer gossiped for cluster
versions >= 20.1 .

- Every new database creation adds the `public` schema to
`system.namespace` by default.

The `search_path` is responsible for establishing the order in which
schemas are searched during name resolution. This PR involves the
following changes to the `search_path.go`:

- The search semantics are updated to match those described in the temp
tables RFC.

- The search path is now aware of the session specific temporary schema,
which can be used during name resolution.

- The distSQL api proto had to be updated to pass the temporary schema to
other nodes in addition to the list of paths.

Follow-up work:

- The `TableCollection` cache needs to be updated to have knowledge about
schemaIDs. Once this is done, there is a TODO in the code that allows the
`CachedPhysicalAccessor` to work correctly.

- Migration for clusters upgrading to 20.1. The new `system.namespace`
table needs to be populated from the deprecated table and a `public`
schema needs to be added for every database during migration.

Release note (sql change): CREATE TABLE pg_temp.abc(a int) now creates
a temporary table. See temp tables RFC (guide-level explanation) for
more details about the search path semantics.

docs/generated/settings/settings.html

@@ -134,6 +134,6 @@
 <tr><td><code>trace.debug.enable</code></td><td>boolean</td><td><code>false</code></td><td>if set, traces for recent requests can be seen in the /debug page</td></tr>
 <tr><td><code>trace.lightstep.token</code></td><td>string</td><td><code></code></td><td>if set, traces go to Lightstep using this token</td></tr>
 <tr><td><code>trace.zipkin.collector</code></td><td>string</td><td><code></code></td><td>if set, traces go to the given Zipkin instance (example: '127.0.0.1:9411'); ignored if trace.lightstep.token is set</td></tr>
-<tr><td><code>version</code></td><td>custom validation</td><td><code>19.2-3</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
+<tr><td><code>version</code></td><td>custom validation</td><td><code>19.2-4</code></td><td>set the active cluster version in the format '<major>.<minor>'</td></tr>
 </tbody>
 </table>

pkg/ccl/backupccl/backup.go

@@ -1383,7 +1383,9 @@ func (b *backupResumer) Resume(
 }
 
 // OnFailOrCancel is part of the jobs.Resumer interface.
-func (b *backupResumer) OnFailOrCancel(context.Context, *client.Txn) error { return nil }
+func (b *backupResumer) OnFailOrCancel(context.Context, *client.Txn) error {
+	return nil
+}
 
 // OnSuccess is part of the jobs.Resumer interface.
 func (b *backupResumer) OnSuccess(context.Context, *client.Txn) error { return nil }

pkg/ccl/backupccl/restore.go

@@ -11,6 +11,7 @@ package backupccl
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"math"
 	"runtime"
 	"sort"
@@ -369,12 +370,22 @@ func allocateTableRewrites(
 	if err := p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *client.Txn) error {
 		// Check that any DBs being restored do _not_ exist.
 		for name := range restoreDBNames {
-			dKey := sqlbase.NewDatabaseKey(name)
-			existingDatabaseID, err := txn.Get(ctx, dKey.Key())
+			dKey := sqlbase.NewDatabaseKey(name).Key()
+			existingDatabaseID, err := txn.Get(ctx, dKey)
 			if err != nil {
 				return err
 			}
-			if existingDatabaseID.Value != nil {
+			if existingDatabaseID.Exists() {
+				return errors.Errorf("database %q already exists", name)
+			}
+			// Check the older system.namespace as well.
+			// TODO(whomever): This can be removed in 20.2
+			dKey = sqlbase.NewDeprecatedDatabaseKey(name).Key()
+			existingDatabaseID, err = txn.Get(ctx, dKey)
+			if err != nil {
+				return err
+			}
+			if existingDatabaseID.Exists() {
 				return errors.Errorf("database %q already exists", name)
 			}
 		}
@@ -397,14 +408,23 @@ func allocateTableRewrites(
 			} else {
 				var parentID sqlbase.ID
 				{
-					dKey := sqlbase.NewDatabaseKey(targetDB)
-					existingDatabaseID, err := txn.Get(ctx, dKey.Key())
+					dKey := sqlbase.NewDatabaseKey(targetDB).Key()
+					existingDatabaseID, err := txn.Get(ctx, dKey)
 					if err != nil {
 						return err
 					}
-					if existingDatabaseID.Value == nil {
-						return errors.Errorf("a database named %q needs to exist to restore table %q",
-							targetDB, table.Name)
+					if !existingDatabaseID.Exists() {
+						// Try the older system.namespace before throwing an error.
+						// TODO(whomever): This can be removed in 20.2
+						dKey = sqlbase.NewDatabaseKey(targetDB).Key()
+						existingDatabaseID, err = txn.Get(ctx, dKey)
+						if err != nil {
+							return err
+						}
+						if !existingDatabaseID.Exists() {
+							return errors.Errorf("a database named %q needs to exist to restore table %q",
+								targetDB, table.Name)
+						}
 					}
 
 					newParentID, err := existingDatabaseID.Value.GetInt()
@@ -488,14 +508,25 @@ func allocateTableRewrites(
 func CheckTableExists(
 	ctx context.Context, txn *client.Txn, parentID sqlbase.ID, name string,
 ) error {
-	tKey := sqlbase.NewTableKey(parentID, name)
+	tKey := sqlbase.NewPublicTableKey(parentID, name)
 	res, err := txn.Get(ctx, tKey.Key())
 	if err != nil {
 		return err
 	}
 	if res.Exists() {
 		return sqlbase.NewRelationAlreadyExistsError(name)
 	}
+	// Check the deprecated system.namespace for mixed version clusters as well.
+	// TODO(whomever): This can be removed in 20.2
+	dtKey := sqlbase.NewDeprecatedTableKey(parentID, name)
+	res, err = txn.Get(ctx, dtKey.Key())
+	if err != nil {
+		return err
+	}
+	if res.Exists() {
+		fmt.Printf("!!! res.value is %v\n", res.Value)
+		return sqlbase.NewRelationAlreadyExistsError(name)
+	}
 	return nil
 }
 
@@ -1057,7 +1088,15 @@ func WriteTableDescs(
 			if err := sql.WriteNewDescToBatch(ctx, false /* kvTrace */, settings, b, desc.ID, desc); err != nil {
 				return err
 			}
-			b.CPut(sqlbase.NewDatabaseKey(desc.Name).Key(), desc.ID, nil)
+			dKey := sqlbase.NewDatabaseKey(desc.Name).Key()
+			// Depending on which cluster version we are restoring to, we decide which
+			// namespace table to write the descriptor into. This may cause wrong
+			// behavior if the cluster version is bumped DURING a restore.
+			// TODO(whomever): This if condition can be removed in 20.2
+			if !cluster.Version.IsActive(ctx, settings, cluster.VersionNamespaceTableWithSchemas) {
+				dKey = sqlbase.NewDeprecatedDatabaseKey(desc.Name).Key()
+			}
+			b.CPut(dKey, desc.ID, nil)
 		}
 		for i := range tables {
 			if wrote, ok := wroteDBs[tables[i].ParentID]; ok {
@@ -1079,7 +1118,15 @@ func WriteTableDescs(
 			if err := sql.WriteNewDescToBatch(ctx, false /* kvTrace */, settings, b, tables[i].ID, tables[i]); err != nil {
 				return err
 			}
-			b.CPut(sqlbase.NewTableKey(tables[i].ParentID, tables[i].Name).Key(), tables[i].ID, nil)
+			var tkey sqlbase.DescriptorKey = sqlbase.NewPublicTableKey(tables[i].ParentID, tables[i].Name)
+			// Depending on which cluster version we are restoring to, we decide which
+			// namespace table to write the descriptor into. This may cause wrong
+			// behavior if the cluster version is bumped DURING a restore.
+			// TODO(whomever): This if condition can be removed in 20.2
+			if !cluster.Version.IsActive(ctx, settings, cluster.VersionNamespaceTableWithSchemas) {
+				tkey = sqlbase.NewDeprecatedTableKey(tables[i].ParentID, tables[i].Name)
+			}
+			b.CPut(tkey.Key(), tables[i].ID, nil)
 		}
 		for _, kv := range extra {
 			b.InitPut(kv.Key, &kv.Value, false)
@@ -1790,8 +1837,17 @@ func (r *restoreResumer) OnFailOrCancel(ctx context.Context, txn *client.Txn) er
 		tableDesc.State = sqlbase.TableDescriptor_DROP
 		var existingIDVal roachpb.Value
 		existingIDVal.SetInt(int64(tableDesc.ID))
-		b.CPut(
-			sqlbase.NewTableKey(tableDesc.ParentID, tableDesc.Name).Key(),
+		// The (parentID, name) mapping could be in either the new system.namespace
+		// or the deprecated version. Thus we try to remove the mapping from both.
+		b.CPutAllowingIfNotExists(
+			sqlbase.NewPublicTableKey(tableDesc.ParentID, tableDesc.Name).Key(),
+			nil,
+			&existingIDVal,
+		)
+		// TODO(whomever): This can be removed in 20.2, and the above
+		//  CPutAllowingIfNotExists should change to CPut.
+		b.CPutAllowingIfNotExists(
+			sqlbase.NewDeprecatedTableKey(tableDesc.ParentID, tableDesc.Name).Key(),
 			nil,
 			&existingIDVal,
 		)

pkg/ccl/backupccl/targets_test.go

@@ -104,7 +104,7 @@ func TestDescriptorsMatchingTargets(t *testing.T) {
 		// {"", `TABLE system."FOO"`, []string{"system"}},
 		// {"system", `TABLE "FOO"`, []string{"system"}},
 	}
-	searchPath := sessiondata.MakeSearchPath([]string{"public", "pg_catalog"})
+	searchPath := sessiondata.MakeSearchPath([]string{"public", "pg_catalog"}, sessiondata.DefaultTemporarySchemaName)
 	for i, test := range tests {
 		t.Run(fmt.Sprintf("%d/%s/%s", i, test.sessionDatabase, test.pattern), func(t *testing.T) {
 			sql := fmt.Sprintf(`GRANT ALL ON %s TO ignored`, test.pattern)

pkg/ccl/importccl/import_stmt.go

@@ -924,8 +924,16 @@ func (r *importResumer) OnFailOrCancel(ctx context.Context, txn *client.Txn) err
 			tableDesc.DropTime = 1
 			var existingIDVal roachpb.Value
 			existingIDVal.SetInt(int64(tableDesc.ID))
-			tKey := sqlbase.NewTableKey(tableDesc.ParentID, tableDesc.Name)
-			b.CPut(tKey.Key(), nil, &existingIDVal)
+			var copyExistingValue roachpb.Value
+			copyExistingValue.SetInt(int64(tableDesc.ID))
+			// The (parentID, name) mapping could be in either the new system.namespace
+			// or the deprecated version. Thus we try to remove the mapping from both.
+			tKey := sqlbase.NewPublicTableKey(tableDesc.ParentID, tableDesc.Name)
+			b.CPutAllowingIfNotExists(tKey.Key(), nil, &existingIDVal)
+			// TODO(whomever): This can be removed in 20.2, and the above
+			//  CPutAllowingIfNotExists should change to CPut.
+			dtKey := sqlbase.NewDeprecatedTableKey(tableDesc.ParentID, tableDesc.Name)
+			b.CPutAllowingIfNotExists(dtKey.Key(), nil, &existingIDVal)
 		} else {
 			// IMPORT did not create this table, so we should not drop it.
 			tableDesc.State = sqlbase.TableDescriptor_PUBLIC

pkg/ccl/partitionccl/partition_test.go

@@ -1095,10 +1095,14 @@ func verifyScansOnNode(
 		}
 		traceLines = append(traceLines, traceLine.String)
 		if strings.Contains(traceLine.String, "read completed") {
-			if strings.Contains(traceLine.String, "SystemCon") {
+			if strings.Contains(traceLine.String, "SystemCon") || strings.Contains(traceLine.String, "NamespaceTab") {
 				// Ignore trace lines for the system config range (abbreviated as
 				// "SystemCon" in pretty printing of the range descriptor). A read might
 				// be performed to the system config range to update the table lease.
+				//
+				// Also ignore trace lines for the system.namespace table, which is a
+				// system table that resides outside the system config range. (abbreviated
+				// as "NamespaceTab" in pretty printing of the range descriptor).
 				continue
 			}
 			if !strings.Contains(traceLine.String, node) {

pkg/cli/cli_test.go

@@ -2215,6 +2215,8 @@ writing ` + os.DevNull + `
   debug/nodes/1/ranges/22.json
   debug/nodes/1/ranges/23.json
   debug/nodes/1/ranges/24.json
+  debug/nodes/1/ranges/25.json
+  debug/nodes/1/ranges/26.json
   debug/schema/[email protected]
   debug/schema/[email protected]
   debug/schema/[email protected]
@@ -2225,6 +2227,7 @@ writing ` + os.DevNull + `
   debug/schema/system/lease.json
   debug/schema/system/locations.json
   debug/schema/system/namespace.json
+  debug/schema/system/namespace_deprecated.json
   debug/schema/system/rangelog.json
   debug/schema/system/replication_constraint_stats.json
   debug/schema/system/replication_critical_localities.json

pkg/cli/sql_util_test.go

@@ -150,6 +150,7 @@ SET
 
 	expectedRows := [][]string{
 		{`parentID`, `INT8`, `false`, `NULL`, ``, `{primary}`, `false`},
+		{`parentSchemaID`, `INT8`, `false`, `NULL`, ``, `{primary}`, `false`},
 		{`name`, `STRING`, `false`, `NULL`, ``, `{primary}`, `false`},
 		{`id`, `INT8`, `true`, `NULL`, ``, `{}`, `false`},
 	}
@@ -163,12 +164,13 @@ SET
 	}
 
 	expected = `
-  column_name | data_type | is_nullable | column_default | generation_expression |  indices  | is_hidden  
-+-------------+-----------+-------------+----------------+-----------------------+-----------+-----------+
-  parentID    | INT8      |    false    | NULL           |                       | {primary} |   false    
-  name        | STRING    |    false    | NULL           |                       | {primary} |   false    
-  id          | INT8      |    true     | NULL           |                       | {}        |   false    
-(3 rows)
+   column_name   | data_type | is_nullable | column_default | generation_expression |  indices  | is_hidden  
++----------------+-----------+-------------+----------------+-----------------------+-----------+-----------+
+  parentID       | INT8      |    false    | NULL           |                       | {primary} |   false    
+  parentSchemaID | INT8      |    false    | NULL           |                       | {primary} |   false    
+  name           | STRING    |    false    | NULL           |                       | {primary} |   false    
+  id             | INT8      |    true     | NULL           |                       | {}        |   false    
+(4 rows)
 `
 
 	if a, e := b.String(), expected[1:]; a != e {
@@ -183,9 +185,9 @@ SET
 	}
 
 	expected = `
-  parentID |    name    | id  
-+----------+------------+----+
-         1 | descriptor |  3  
+  parentID | parentSchemaID |    name    | id  
++----------+----------------+------------+----+
+         1 |             29 | descriptor |  3  
 (1 row)
 `
 	if a, e := b.String(), expected[1:]; a != e {

pkg/config/system_test.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/config"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/util/encoding"
@@ -163,7 +164,7 @@ func TestGetLargestID(t *testing.T) {
 			ms := sqlbase.MakeMetadataSchema(config.DefaultZoneConfigRef(), config.DefaultSystemZoneConfigRef())
 			descIDs := ms.DescriptorIDs()
 			maxDescID := descIDs[len(descIDs)-1]
-			kvs, _ /* splits */ := ms.GetInitialValues()
+			kvs, _ /* splits */ := ms.GetInitialValues(cluster.TestingClusterVersion)
 			return testCase{kvs, uint32(maxDescID), 0, ""}
 		}(),
 
@@ -256,7 +257,7 @@ func TestComputeSplitKeySystemRanges(t *testing.T) {
 	}
 
 	cfg := config.NewSystemConfig(config.DefaultZoneConfigRef())
-	kvs, _ /* splits */ := sqlbase.MakeMetadataSchema(cfg.DefaultZoneConfig, config.DefaultSystemZoneConfigRef()).GetInitialValues()
+	kvs, _ /* splits */ := sqlbase.MakeMetadataSchema(cfg.DefaultZoneConfig, config.DefaultSystemZoneConfigRef()).GetInitialValues(cluster.TestingClusterVersion)
 	cfg.SystemConfigEntries = config.SystemConfigEntries{
 		Values: kvs,
 	}
@@ -288,12 +289,14 @@ func TestComputeSplitKeyTableIDs(t *testing.T) {
 
 	schema := sqlbase.MakeMetadataSchema(config.DefaultZoneConfigRef(), config.DefaultSystemZoneConfigRef())
 	// Real system tables only.
-	baseSql, _ /* splits */ := schema.GetInitialValues()
+	baseSql, _ /* splits */ := schema.GetInitialValues(cluster.TestingClusterVersion)
 	// Real system tables plus some user stuff.
-	kvs, _ /* splits */ := schema.GetInitialValues()
+	kvs, _ /* splits */ := schema.GetInitialValues(cluster.TestingClusterVersion)
 	userSQL := append(kvs, descriptor(start), descriptor(start+1), descriptor(start+5))
 	// Real system tables and partitioned user tables.
-	subzoneSQL := append(userSQL,
+	var subzoneSQL = make([]roachpb.KeyValue, len(userSQL))
+	copy(subzoneSQL, userSQL)
+	subzoneSQL = append(subzoneSQL,
 		zoneConfig(start+1, subzone("a", ""), subzone("c", "e")),
 		zoneConfig(start+5, subzone("b", ""), subzone("c", "d"), subzone("d", "")))
 
@@ -404,13 +407,13 @@ func TestGetZoneConfigForKey(t *testing.T) {
 		{roachpb.RKey(keys.SystemConfigSplitKey), keys.SystemDatabaseID},
 
 		// Gossiped system tables should refer to the SystemDatabaseID.
-		{tkey(keys.NamespaceTableID), keys.SystemDatabaseID},
 		{tkey(keys.ZonesTableID), keys.SystemDatabaseID},
 
 		// Non-gossiped system tables should refer to themselves.
 		{tkey(keys.LeaseTableID), keys.LeaseTableID},
 		{tkey(keys.JobsTableID), keys.JobsTableID},
 		{tkey(keys.LocationsTableID), keys.LocationsTableID},
+		{tkey(keys.NamespaceTableID), keys.NamespaceTableID},
 
 		// Pseudo-tables should refer to the SystemDatabaseID.
 		{tkey(keys.MetaRangesID), keys.SystemDatabaseID},
@@ -430,7 +433,7 @@ func TestGetZoneConfigForKey(t *testing.T) {
 	}()
 	cfg := config.NewSystemConfig(config.DefaultZoneConfigRef())
 
-	kvs, _ /* splits */ := sqlbase.MakeMetadataSchema(cfg.DefaultZoneConfig, config.DefaultSystemZoneConfigRef()).GetInitialValues()
+	kvs, _ /* splits */ := sqlbase.MakeMetadataSchema(cfg.DefaultZoneConfig, config.DefaultSystemZoneConfigRef()).GetInitialValues(cluster.TestingClusterVersion)
 	cfg.SystemConfigEntries = config.SystemConfigEntries{
 		Values: kvs,
 	}

pkg/keys/constants.go

@@ -265,6 +265,12 @@ var (
 	MaxKey = roachpb.KeyMax
 	// MinKey is a minimum key value which sorts before all other keys.
 	MinKey = roachpb.KeyMin
+
+	// NamespaceTableMin is the start key of system.namespace, which is a system
+	// table that does not reside in the same range as other system tables.
+	NamespaceTableMin = roachpb.Key(MakeTablePrefix(NamespaceTableID))
+	// NamespaceTableMax is the end key of system.namespace.
+	NamespaceTableMax = roachpb.Key(MakeTablePrefix(NamespaceTableID + 1))
 )
 
 // Various IDs used by the structured data layer.
@@ -295,12 +301,14 @@ const (
 	// SystemDatabaseID and following are the database/table IDs for objects
 	// in the system span.
 	// NOTE: IDs must be <= MaxSystemConfigDescID.
-	SystemDatabaseID  = 1
-	NamespaceTableID  = 2
-	DescriptorTableID = 3
-	UsersTableID      = 4
-	ZonesTableID      = 5
-	SettingsTableID   = 6
+	SystemDatabaseID = 1
+	// DeprecatedNamespaceTableID was the tableID for the system.namespace table
+	// for pre-20.1 clusters.
+	DeprecatedNamespaceTableID = 2
+	DescriptorTableID          = 3
+	UsersTableID               = 4
+	ZonesTableID               = 5
+	SettingsTableID            = 6
 
 	// IDs for the important columns and indexes in the zones table live here to
 	// avoid introducing a dependency on sql/sqlbase throughout the codebase.
@@ -330,6 +338,10 @@ const (
 	ReplicationCriticalLocalitiesTableID = 26
 	ReplicationStatsTableID              = 27
 	ReportsMetaTableID                   = 28
+	PublicSchemaID                       = 29
+	// New NamespaceTableID for cluster version >= 20.1
+	// Ensures that NamespaceTable does not get gossiped again
+	NamespaceTableID = 30
 
 	// CommentType is type for system.comments
 	DatabaseCommentType = 0
@@ -342,4 +354,4 @@ const (
 // there's no table descriptor). They're grouped here because the cluster
 // bootstrap process needs to create splits for them; splits for the tables
 // happen separately.
-var PseudoTableIDs = []uint32{MetaRangesID, SystemRangesID, TimeseriesRangesID, LivenessRangesID}
+var PseudoTableIDs = []uint32{MetaRangesID, SystemRangesID, TimeseriesRangesID, LivenessRangesID, PublicSchemaID}

pkg/keys/printer.go

@@ -126,6 +126,9 @@ var (
 				PSFunc: parseUnsupported,
 			},
 		}},
+		{Name: "/NamespaceTable", start: NamespaceTableMin, end: NamespaceTableMax, Entries: []DictEntry{
+			{Name: "", prefix: nil, ppFunc: decodeKeyPrint, PSFunc: parseUnsupported},
+		}},
 		{Name: "/Table", start: TableDataMin, end: TableDataMax, Entries: []DictEntry{
 			{Name: "", prefix: nil, ppFunc: decodeKeyPrint, PSFunc: tableKeyParse},
 		}},