sql/parser: update CREATE VIRTUAL CLUSTER to support a direct tenant ID

Previously, the CREATE VIRTUAL CLUSTER (or CREATE TENANT) syntax only accepted
a tenant name for creating a virtual cluster, with the system automatically
generating the tenant ID. However, this approach doesn't work with CockroachDB
Cloud's PCR setup, as we need to create a replication stream using an existing
tenant ID. To address this, this commit modifies the CREATE VIRTUAL CLUSTER
syntax to allow for a direct tenant ID input, enabling the creation of a
tenant with a specific ID. While we have crdb_internal.create_tenant(ID) for
this purpose, there's no equivalent built-in for creating a replication stream,
so we will have to resort to using the CREATE VIRTUAL CLUSTER ... FROM
REPLICATION syntax.

No release note as this is meant to be used internally only. This change can
be removed once we transition away from using tenant IDs in CockroachDB Cloud.

Epic: none

Release note: None

pkg/ccl/crosscluster/physical/stream_ingestion_job_test.go

@@ -56,10 +56,10 @@ func TestTenantStreamingCreationErrors(t *testing.T) {
 
 	telemetry.GetFeatureCounts(telemetry.Raw, telemetry.ResetCounts)
 	t.Run("destination cannot be system tenant", func(t *testing.T) {
-		sysSQL.ExpectErr(t, `pq: the destination tenant "system" \(0\) cannot be the system tenant`,
-			"CREATE TENANT system FROM REPLICATION OF source ON $1", srcPgURL.String())
+		sysSQL.ExpectErr(t, `pq: the destination tenant "" \(1\) cannot be the system tenant`,
+			"CREATE TENANT [1] FROM REPLICATION OF source ON $1", srcPgURL.String())
 	})
-	t.Run("cannot set expiration window on creat tenant from replication", func(t *testing.T) {
+	t.Run("cannot set expiration window on create tenant from replication", func(t *testing.T) {
 		sysSQL.ExpectErr(t, `pq: cannot specify EXPIRATION WINDOW option while starting a physical replication stream`,
 			"CREATE TENANT system FROM REPLICATION OF source ON $1 WITH EXPIRATION WINDOW='42s'", srcPgURL.String())
 	})
@@ -77,7 +77,6 @@ func TestTenantStreamingCreationErrors(t *testing.T) {
 	})
 	counts := telemetry.GetFeatureCounts(telemetry.Raw, telemetry.ResetCounts)
 	require.Equal(t, int32(0), counts["physical_replication.started"])
-
 }
 
 func TestTenantStreamingFailback(t *testing.T) {

pkg/ccl/sqlproxyccl/tenant/directory_cache_test.go

@@ -596,14 +596,19 @@ func TestRefreshThrottling(t *testing.T) {
 func createTenant(tc serverutils.TestClusterInterface, id roachpb.TenantID) error {
 	srv := tc.Server(0)
 	conn := srv.InternalExecutor().(*sql.InternalExecutor)
-	if _, err := conn.Exec(
-		context.Background(),
-		"testserver-create-tenant",
-		nil, /* txn */
-		"SELECT crdb_internal.create_tenant($1)",
-		id.ToUint64(),
-	); err != nil {
-		return err
+	for _, stmt := range []string{
+		`CREATE VIRTUAL CLUSTER [$1]`,
+		`ALTER VIRTUAL CLUSTER [$1] START SERVICE EXTERNAL`,
+	} {
+		if _, err := conn.Exec(
+			context.Background(),
+			"testserver-create-tenant",
+			nil, /* txn */
+			stmt,
+			id.ToUint64(),
+		); err != nil {
+			return err
+		}
 	}
 	return nil
 }

pkg/sql/parser/sql.y

@@ -4749,42 +4749,42 @@ logical_replication_options:
 // %Help: CREATE VIRTUAL CLUSTER - create a new virtual cluster
 // %Category: Experimental
 // %Text:
-// CREATE VIRTUAL CLUSTER [ IF NOT EXISTS ] name [ <replication> ]
+// CREATE VIRTUAL CLUSTER [ IF NOT EXISTS ] <virtual_cluster_spec> [ <replication> ]
 //
 // Replication option:
-//    FROM REPLICATION OF <virtual_cluster_spec> ON <location> [ WITH OPTIONS ... ]
+//    FROM REPLICATION OF name ON <location> [ WITH OPTIONS ... ]
 create_virtual_cluster_stmt:
-  CREATE virtual_cluster d_expr
+  CREATE virtual_cluster virtual_cluster_spec
   {
     /* SKIP DOC */
     $$.val = &tree.CreateTenant{
-      TenantSpec: &tree.TenantSpec{IsName: true, Expr: $3.expr()},
+      TenantSpec: $3.tenantSpec(),
     }
   }
-| CREATE virtual_cluster IF NOT EXISTS d_expr
+| CREATE virtual_cluster IF NOT EXISTS virtual_cluster_spec
   {
     /* SKIP DOC */
     $$.val = &tree.CreateTenant{
       IfNotExists: true,
-      TenantSpec: &tree.TenantSpec{IsName: true, Expr: $6.expr()},
+      TenantSpec: $6.tenantSpec(),
     }
   }
-| CREATE virtual_cluster d_expr FROM REPLICATION OF d_expr ON d_expr opt_with_replication_options
+| CREATE virtual_cluster virtual_cluster_spec FROM REPLICATION OF d_expr ON d_expr opt_with_replication_options
   {
     /* SKIP DOC */
     $$.val = &tree.CreateTenantFromReplication{
-      TenantSpec: &tree.TenantSpec{IsName: true, Expr: $3.expr()},
+      TenantSpec: $3.tenantSpec(),
       ReplicationSourceTenantName: &tree.TenantSpec{IsName: true, Expr: $7.expr()},
       ReplicationSourceAddress: $9.expr(),
       Options: *$10.tenantReplicationOptions(),
     }
   }
-| CREATE virtual_cluster IF NOT EXISTS d_expr FROM REPLICATION OF d_expr ON d_expr opt_with_replication_options
+| CREATE virtual_cluster IF NOT EXISTS virtual_cluster_spec FROM REPLICATION OF d_expr ON d_expr opt_with_replication_options
   {
     /* SKIP DOC */
     $$.val = &tree.CreateTenantFromReplication{
       IfNotExists: true,
-      TenantSpec: &tree.TenantSpec{IsName: true, Expr: $6.expr()},
+      TenantSpec: $6.tenantSpec(),
       ReplicationSourceTenantName: &tree.TenantSpec{IsName: true, Expr: $10.expr()},
       ReplicationSourceAddress: $12.expr(),
       Options: *$13.tenantReplicationOptions(),

pkg/sql/parser/testdata/create_virtual_cluster

@@ -6,6 +6,14 @@ CREATE VIRTUAL CLUSTER (bar) -- fully parenthesized
 CREATE VIRTUAL CLUSTER bar -- literals removed
 CREATE VIRTUAL CLUSTER _ -- identifiers removed
 
+parse
+CREATE VIRTUAL CLUSTER [123::INT]
+----
+CREATE VIRTUAL CLUSTER [123::INT8] -- normalized!
+CREATE VIRTUAL CLUSTER [((123)::INT8)] -- fully parenthesized
+CREATE VIRTUAL CLUSTER [_::INT8] -- literals removed
+CREATE VIRTUAL CLUSTER [123::INT8] -- identifiers removed
+
 parse
 CREATE TENANT bar
 ----
@@ -30,6 +38,14 @@ CREATE VIRTUAL CLUSTER IF NOT EXISTS (bar) -- fully parenthesized
 CREATE VIRTUAL CLUSTER IF NOT EXISTS bar -- literals removed
 CREATE VIRTUAL CLUSTER IF NOT EXISTS _ -- identifiers removed
 
+parse
+CREATE VIRTUAL CLUSTER IF NOT EXISTS [123::INT]
+----
+CREATE VIRTUAL CLUSTER IF NOT EXISTS [123::INT8] -- normalized!
+CREATE VIRTUAL CLUSTER IF NOT EXISTS [((123)::INT8)] -- fully parenthesized
+CREATE VIRTUAL CLUSTER IF NOT EXISTS [_::INT8] -- literals removed
+CREATE VIRTUAL CLUSTER IF NOT EXISTS [123::INT8] -- identifiers removed
+
 parse
 CREATE VIRTUAL CLUSTER destination FROM REPLICATION OF source ON 'pgurl'
 ----
@@ -38,6 +54,14 @@ CREATE VIRTUAL CLUSTER (destination) FROM REPLICATION OF (source) ON ('pgurl') -
 CREATE VIRTUAL CLUSTER destination FROM REPLICATION OF source ON '_' -- literals removed
 CREATE VIRTUAL CLUSTER _ FROM REPLICATION OF _ ON 'pgurl' -- identifiers removed
 
+parse
+CREATE VIRTUAL CLUSTER [123::INT] FROM REPLICATION OF source ON 'pgurl'
+----
+CREATE VIRTUAL CLUSTER [123::INT] FROM REPLICATION OF source ON 'pgurl'
+CREATE VIRTUAL CLUSTER ([123::INT]) FROM REPLICATION OF (source) ON ('pgurl') -- fully parenthesized
+CREATE VIRTUAL CLUSTER [_::INT] FROM REPLICATION OF source ON '_' -- literals removed
+CREATE VIRTUAL CLUSTER [123::INT] FROM REPLICATION OF _ ON 'pgurl' -- identifiers removed
+
 parse
 CREATE VIRTUAL CLUSTER IF NOT EXISTS destination FROM REPLICATION OF source ON 'pgurl'
 ----

pkg/sql/sem/tree/walk.go

@@ -1069,9 +1069,22 @@ func (n *AlterTenantReplication) walkStmt(v Visitor) Statement {
 	return ret
 }
 
+// copyNode makes a copy of this Statement without recursing in any child Statements.
+func (n *CreateTenant) copyNode() *CreateTenant {
+	stmtCopy := *n
+	return &stmtCopy
+}
+
 // walkStmt is part of the walkableStmt interface.
 func (n *CreateTenant) walkStmt(v Visitor) Statement {
 	ret := n
+	ts, changed := walkTenantSpec(v, n.TenantSpec)
+	if changed {
+		if ret == n {
+			ret = n.copyNode()
+		}
+		ret.TenantSpec = ts
+	}
 	return ret
 }
 
@@ -1084,6 +1097,13 @@ func (n *CreateTenantFromReplication) copyNode() *CreateTenantFromReplication {
 // walkStmt is part of the walkableStmt interface.
 func (n *CreateTenantFromReplication) walkStmt(v Visitor) Statement {
 	ret := n
+	ts, changed := walkTenantSpec(v, n.TenantSpec)
+	if changed {
+		if ret == n {
+			ret = n.copyNode()
+		}
+		ret.TenantSpec = ts
+	}
 	e, changed := WalkExpr(v, n.ReplicationSourceTenantName.Expr)
 	if changed {
 		if ret == n {