Shift user item id allocation

Author: bkirwi

src/adapter/tests/testdata/sql

@@ -10,12 +10,12 @@
 add-table
 foo
 ----
-u1
+u2
 
 resolve
 SELECT 1 FROM foo
 ----
-SELECT 1 FROM [u1 AS materialize.public.foo]
+SELECT 1 FROM [u2 AS materialize.public.foo]
 
 resolve
 SELECT 1 FROM bar
@@ -25,7 +25,7 @@ error: unknown catalog item 'bar'
 resolve
 SELECT 1 FROM materialize.public.foo
 ----
-SELECT 1 FROM [u1 AS materialize.public.foo]
+SELECT 1 FROM [u2 AS materialize.public.foo]
 
 resolve
 WITH foo AS (SELECT 1)
@@ -43,13 +43,13 @@ resolve
 WITH bar AS (SELECT 1)
 SELECT 1 FROM foo CROSS JOIN bar
 ----
-WITH bar AS (SELECT 1) SELECT 1 FROM [u1 AS materialize.public.foo] CROSS JOIN bar
+WITH bar AS (SELECT 1) SELECT 1 FROM [u2 AS materialize.public.foo] CROSS JOIN bar
 
 resolve
 WITH bar AS (SELECT 1), baz AS (SELECT 2)
 SELECT 1 FROM foo CROSS JOIN bar CROSS JOIN baz
 ----
-WITH bar AS (SELECT 1), baz AS (SELECT 2) SELECT 1 FROM [u1 AS materialize.public.foo] CROSS JOIN bar CROSS JOIN baz
+WITH bar AS (SELECT 1), baz AS (SELECT 2) SELECT 1 FROM [u2 AS materialize.public.foo] CROSS JOIN bar CROSS JOIN baz
 
 resolve
 WITH outermost(x) AS (

src/catalog/src/durable.rs

@@ -355,9 +355,7 @@ pub trait DurableCatalogState: ReadOnlyDurableCatalogState {
         amount: u64,
         commit_ts: Timestamp,
     ) -> Result<Vec<(CatalogItemId, GlobalId)>, CatalogError> {
-        let ids = self
-            .allocate_id(&[USER_ITEM_ALLOC_KEY], amount, commit_ts)
-            .await?;
+        let ids = self.allocate_id(USER_ALLOC_KEYS, amount, commit_ts).await?;
         let ids = ids
             .iter()
             .map(|id| (CatalogItemId::User(*id), GlobalId::User(*id)))
@@ -372,9 +370,7 @@ pub trait DurableCatalogState: ReadOnlyDurableCatalogState {
         &mut self,
         commit_ts: Timestamp,
     ) -> Result<(CatalogItemId, GlobalId), CatalogError> {
-        let id = self
-            .allocate_id(&[USER_ITEM_ALLOC_KEY], 1, commit_ts)
-            .await?;
+        let id = self.allocate_id(USER_ALLOC_KEYS, 1, commit_ts).await?;
         let id = id.into_element();
         Ok((CatalogItemId::User(id), GlobalId::User(id)))
     }

src/catalog/src/durable/transaction.rs

@@ -65,8 +65,8 @@ use crate::durable::{
     DATABASE_ID_ALLOC_KEY, DefaultPrivilege, DurableCatalogError, DurableCatalogState,
     EXPRESSION_CACHE_SHARD_KEY, MOCK_AUTHENTICATION_NONCE_KEY, NetworkPolicy, OID_ALLOC_KEY,
     SCHEMA_ID_ALLOC_KEY, STORAGE_USAGE_ID_ALLOC_KEY, SYSTEM_ALLOC_KEYS, SYSTEM_ITEM_ALLOC_KEY,
-    Snapshot, SystemConfiguration, USER_ALLOC_KEYS, USER_ITEM_ALLOC_KEY,
-    USER_NETWORK_POLICY_ID_ALLOC_KEY, USER_ROLE_ID_ALLOC_KEY,
+    Snapshot, SystemConfiguration, USER_ALLOC_KEYS, USER_NETWORK_POLICY_ID_ALLOC_KEY,
+    USER_ROLE_ID_ALLOC_KEY,
 };
 use crate::memory::objects::{StateDiff, StateUpdate, StateUpdateKind};
 
@@ -889,7 +889,7 @@ impl<'a> Transaction<'a> {
         amount: u64,
     ) -> Result<Vec<(CatalogItemId, GlobalId)>, CatalogError> {
         Ok(self
-            .get_and_increment_id_by(&[USER_ITEM_ALLOC_KEY], amount)?
+            .get_and_increment_id_by(USER_ALLOC_KEYS, amount)?
             .into_iter()
             // TODO(alter_table): Use separate ID allocators.
             .map(|x| (CatalogItemId::User(x), GlobalId::User(x)))

src/environmentd/tests/server.rs

@@ -161,7 +161,7 @@ fn test_persistence() {
             .into_iter()
             .map(|row| row.get(0))
             .collect::<Vec<String>>(),
-        vec!["u1", "u2", "u3", "u4", "u5", "u6", "u7"]
+        vec!["u2", "u3", "u4", "u5", "u6", "u7", "u8"]
     );
 }
 

src/environmentd/tests/sql.rs

@@ -1244,13 +1244,13 @@ largest not in advance of upper:<TIMESTAMP>
                        timeline: Some(EpochMilliseconds)
               session wall time:<TIMESTAMP>
 
-source materialize.public.t1 (u1, storage):
+source materialize.public.t1 (u2, storage):
                   read frontier:[<TIMESTAMP>]
                  write frontier:[<TIMESTAMP>]
 
 binding constraints:
 lower:
-  (StorageInput([User(1)])): [<TIMESTAMP>]
+  (StorageInput([User(2)])): [<TIMESTAMP>]
   (IsolationLevel(StrictSerializable)): [<TIMESTAMP>]\n";
 
     let row = client

test/cluster/mzcompose.py

@@ -1461,8 +1461,8 @@ def workflow_test_system_table_indexes(c: Composition) -> None:
         FROM mz_views;
         CREATE DEFAULT INDEX ON v_mz_views;
 
-        > SELECT id FROM mz_indexes WHERE id like 'u%';
-        u2
+        > SELECT name FROM mz_indexes WHERE id like 'u%';
+        v_mz_views_primary_idx
     """
             )
         )
@@ -1476,8 +1476,8 @@ def workflow_test_system_table_indexes(c: Composition) -> None:
         c.testdrive(
             input=dedent(
                 """
-        > SELECT id FROM mz_indexes WHERE id like 'u%';
-        u2
+        > SELECT name FROM mz_indexes WHERE id like 'u%';
+        v_mz_views_primary_idx
     """
             )
         )
@@ -4794,7 +4794,7 @@ def workflow_test_adhoc_system_indexes(
         WHERE i.name = 'mz_test_idx1'
         """
     )
-    assert output[0] == ("u1", "mz_tables", "mz_catalog_server"), output
+    assert output[0][1:] == ("mz_tables", "mz_catalog_server"), output
     output = c.sql_query("EXPLAIN SELECT * FROM mz_tables WHERE char_length(name) = 9")
     assert "mz_test_idx1" in output[0][0], output
     output = c.sql_query("SELECT * FROM mz_tables WHERE char_length(name) = 9")
@@ -4823,7 +4823,7 @@ def workflow_test_adhoc_system_indexes(
         WHERE i.name = 'mz_test_idx2'
         """
     )
-    assert output[0] == ("u2", "mz_hydration_statuses", "mz_catalog_server"), output
+    assert output[0][1:] == ("mz_hydration_statuses", "mz_catalog_server"), output
     output = c.sql_query(
         "EXPLAIN SELECT * FROM mz_internal.mz_hydration_statuses WHERE hydrated"
     )
@@ -4848,8 +4848,8 @@ def workflow_test_adhoc_system_indexes(
         ORDER BY id
         """
     )
-    assert output[0] == ("u1", "mz_tables", "mz_catalog_server"), output
-    assert output[1] == ("u2", "mz_hydration_statuses", "mz_catalog_server"), output
+    assert output[0][1:] == ("mz_tables", "mz_catalog_server"), output
+    assert output[1][1:] == ("mz_hydration_statuses", "mz_catalog_server"), output
 
     # Make sure the new indexes can be dropped again.
     c.sql(

test/kafka-auth/mzcompose.py

@@ -240,7 +240,15 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     """
     )
     public_key = c.sql_query(
-        "select public_key_1 from mz_ssh_tunnel_connections where id = 'u1';"
+        """
+        SELECT
+            public_key_1
+        FROM
+            mz_connections JOIN
+            mz_ssh_tunnel_connections USING(id)
+        WHERE
+            mz_connections.name = 'ssh';
+        """
     )[0][0]
     c.exec(
         "ssh-bastion-host",
@@ -261,7 +269,15 @@ def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
     """
     )
     public_key = c.sql_query(
-        "select public_key_1 from mz_ssh_tunnel_connections where id = 'u2';"
+        """
+        SELECT
+            public_key_1
+        FROM
+            mz_connections JOIN
+            mz_ssh_tunnel_connections USING(id)
+        WHERE
+            mz_connections.name = 'ssh_backup';
+        """
     )[0][0]
     c.exec(
         "ssh-bastion-host",

test/kafka-auth/test-kafka-ssl.td

@@ -172,19 +172,19 @@ running
 > SELECT COUNT(*) FROM mz_ssh_tunnel_connections
 2
 
-> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies WHERE referenced_object_id = 'u1';
+> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies as mzo, mz_objects as mo WHERE mzo.referenced_object_id = mo.id and mo.name = 'ssh';
 1
 
 ! DROP CONNECTION testdrive_no_reset_connections.public.ssh;
 contains:still depended upon by connection "kafka_ssh"
 
-> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies WHERE referenced_object_id = 'u1';
+> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies as mzo, mz_objects as mo WHERE mzo.referenced_object_id = mo.id and mo.name = 'ssh';
 1
 
 > ALTER CONNECTION kafka_ssh SET (BROKER 'kafka:9093' USING SSH TUNNEL testdrive_no_reset_connections.public.ssh_backup);
 
 # We've removed all dependencies on testdrive_no_reset_connections.public.ssh, so it could be dropped
-> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies WHERE referenced_object_id = 'u1';
+> SELECT COUNT(*) FROM mz_internal.mz_object_dependencies as mzo, mz_objects as mo WHERE mzo.referenced_object_id = mo.id and mo.name = 'ssh';
 0
 
 # Break new SSH tunnel to show that we can fix it

test/persistence/mzcompose.py

@@ -194,7 +194,12 @@ def workflow_inspect_shard(c: Composition) -> None:
             """
         )
     )
-    json_dict = c.sql_query("INSPECT SHARD 'u1'", port=6877, user="mz_system")[0][0]
+    object_id = c.sql_query(
+        "SELECT id from mz_objects where name = 'foo'", port=6877, user="mz_system"
+    )[0][0]
+    json_dict = c.sql_query(
+        f"INSPECT SHARD '{object_id}'", port=6877, user="mz_system"
+    )[0][0]
     parts = [
         part
         for batch in json_dict["batches"]

test/sqllogictest/alter-table.slt

@@ -19,7 +19,7 @@ CREATE TABLE t2 (a int);
 
 # TODO(alter_table): Manually specifying a VERSION when referencing an item (maybe?) shouldn't be allowed.
 statement ok
-CREATE VIEW v1 AS SELECT * FROM [u1 AS "materialize"."public"."t2" VERSION 0];
+CREATE VIEW v1 AS SELECT * FROM [u2 AS "materialize"."public"."t2" VERSION 0];
 
 statement ok
 CREATE VIEW v2 AS SELECT * FROM t2;
@@ -57,7 +57,7 @@ statement ok
 ALTER TABLE t1 ADD COLUMN IF NOT EXISTS a int;
 
 query error db error: ERROR: invalid version "1000" for "t2"
-CREATE VIEW v1 AS SELECT * FROM [u1 AS "materialize"."public"."t2" VERSION 1000];
+CREATE VIEW v1 AS SELECT * FROM [u2 AS "materialize"."public"."t2" VERSION 1000];
 
 query TTTT
 SHOW COLUMNS FROM t1;
@@ -413,15 +413,15 @@ DROP TABLE t1 CASCADE;
 query TT
 SELECT id, name FROM mz_tables WHERE id LIKE 'u%';
 ----
-u1 t2
+u2 t2
 
 statement ok
 COMMENT ON COLUMN t2.a IS 'this column existed originally';
 
 query TTIT
-SELECT * FROM mz_internal.mz_comments WHERE id = 'u1';
+SELECT * FROM mz_internal.mz_comments WHERE id = 'u2';
 ----
-u1  table  1  this␠column␠existed␠originally
+u2  table  1  this␠column␠existed␠originally
 
 statement ok
 ALTER TABLE t2 ADD COLUMN c timestamp;
@@ -430,14 +430,14 @@ statement ok
 COMMENT ON COLUMN t2.c IS 'added later';
 
 query TTIT rowsort
-SELECT * FROM mz_internal.mz_comments WHERE id = 'u1';
+SELECT * FROM mz_internal.mz_comments WHERE id = 'u2';
 ----
-u1  table  3  added␠later
-u1  table  1  this␠column␠existed␠originally
+u2  table  3  added␠later
+u2  table  1  this␠column␠existed␠originally
 
 statement ok
 DROP TABLE t2 CASCADE;
 
 query TTIT
-SELECT * FROM mz_internal.mz_comments WHERE id = 'u1';
+SELECT * FROM mz_internal.mz_comments WHERE id = 'u2';
 ----