allow table number reuse across components (#27509)

effectively revert #26245 now that we've aligned on the direction.

GitOrigin-RevId: e3f2953cc11d29793ca6e8ddf71ab8269a3a10b1

Author: ldanilek

crates/database/src/bootstrap_model/table.rs

@@ -247,23 +247,35 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
 
     // Checks both _tables and _virtual_tables to find a non-conflicting table
     // number
-    pub async fn next_user_table_number(&mut self) -> anyhow::Result<TableNumber> {
-        self.next_table_number(false).await
+    pub async fn next_user_table_number(
+        &mut self,
+        namespace: TableNamespace,
+    ) -> anyhow::Result<TableNumber> {
+        self.next_table_number(false, namespace).await
     }
 
-    pub async fn next_system_table_number(&mut self) -> anyhow::Result<TableNumber> {
-        self.next_table_number(true).await
+    pub async fn next_system_table_number(
+        &mut self,
+        namespace: TableNamespace,
+    ) -> anyhow::Result<TableNumber> {
+        self.next_table_number(true, namespace).await
     }
 
-    async fn next_table_number(&mut self, is_system: bool) -> anyhow::Result<TableNumber> {
+    async fn next_table_number(
+        &mut self,
+        is_system: bool,
+        namespace: TableNamespace,
+    ) -> anyhow::Result<TableNumber> {
         let occupied_table_numbers = {
             let mut occupied_table_numbers = BTreeSet::new();
             let tables_query = Query::full_table_scan(TABLES_TABLE.clone(), Order::Asc);
             let mut query_stream =
                 ResolvedQuery::new(self.tx, TableNamespace::Global, tables_query)?;
             while let Some(table_metadata) = query_stream.next(self.tx, None).await? {
                 let parsed_metadata: ParsedDocument<TableMetadata> = table_metadata.try_into()?;
-                occupied_table_numbers.insert(parsed_metadata.number);
+                if parsed_metadata.namespace == namespace {
+                    occupied_table_numbers.insert(parsed_metadata.number);
+                }
             }
             let virtual_tables_query =
                 Query::full_table_scan(VIRTUAL_TABLES_TABLE.clone(), Order::Asc);
@@ -481,7 +493,7 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
                 );
                 table_number
             } else {
-                self.next_user_table_number().await?
+                self.next_user_table_number(namespace).await?
             };
             let table_metadata =
                 TableMetadata::new_with_state(namespace, table.clone(), table_number, state);
@@ -733,18 +745,17 @@ mod tests {
     async fn test_next_table_number(rt: TestRuntime) -> anyhow::Result<()> {
         let mut tx = new_tx(rt).await?;
         let mut model = TableModel::new(&mut tx);
+        let namespace = TableNamespace::test_user();
 
-        let next_table_number: u32 = model.next_user_table_number().await?.into();
+        let next_table_number: u32 = model.next_user_table_number(namespace).await?.into();
         assert_eq!(
             (NUM_RESERVED_SYSTEM_TABLE_NUMBERS + 1) as usize,
             next_table_number as usize
         );
 
         let table_name = TableName::from_str("my_table")?;
-        model
-            .insert_table_metadata(TableNamespace::test_user(), &table_name)
-            .await?;
-        let new_next_table_number: u32 = model.next_user_table_number().await?.into();
+        model.insert_table_metadata(namespace, &table_name).await?;
+        let new_next_table_number: u32 = model.next_user_table_number(namespace).await?.into();
         assert_eq!(next_table_number + 1, new_next_table_number);
         Ok(())
     }

crates/database/src/table_registry.rs

@@ -107,7 +107,7 @@ impl TableRegistry {
                         if self.table_exists(metadata.namespace, &metadata.name) {
                             anyhow::bail!("Tried to create duplicate table {new_value}");
                         }
-                        self.validate_table_number(metadata.number)?;
+                        self.validate_table_number(metadata.namespace, metadata.number)?;
                     }
                     Some(TableUpdate {
                         namespace: metadata.namespace,
@@ -195,7 +195,7 @@ impl TableRegistry {
                     {
                         anyhow::bail!("Tried to create duplicate virtual table {new_value}");
                     }
-                    self.validate_table_number(metadata.number)?;
+                    self.validate_table_number(metadata.namespace, metadata.number)?;
                     virtual_table_creation =
                         Some((metadata.namespace, metadata.number, metadata.name));
                 },
@@ -211,19 +211,23 @@ impl TableRegistry {
         Ok(update)
     }
 
-    fn validate_table_number(&self, table_number: TableNumber) -> anyhow::Result<()> {
+    fn validate_table_number(
+        &self,
+        namespace: TableNamespace,
+        table_number: TableNumber,
+    ) -> anyhow::Result<()> {
         anyhow::ensure!(
             !self
                 .table_mapping
-                .iter()
-                .any(|(_, _, number, _)| number == table_number),
+                .namespace(namespace)
+                .table_number_exists()(table_number),
             "Cannot add a table with table number {table_number} since it already exists in the \
              table mapping"
         );
         anyhow::ensure!(
             !self
                 .virtual_table_mapping
-                .namespace(TableNamespace::TODO())
+                .namespace(namespace)
                 .number_exists(table_number),
             "Cannot add a table with table number {table_number} since it already exists in the \
              virtual table mapping"

crates/database/src/transaction.rs

@@ -24,7 +24,6 @@ use common::{
             TABLES_TABLE,
         },
     },
-    components::COMPONENTS_ENABLED,
     document::{
         CreationTime,
         DocumentUpdate,
@@ -642,21 +641,19 @@ impl<RT: Runtime> Transaction<RT> {
 
     async fn table_number_for_system_table(
         &mut self,
+        namespace: TableNamespace,
         table_name: &TableName,
         default_table_number: Option<TableNumber>,
     ) -> anyhow::Result<TableNumber> {
         Ok(if let Some(default_table_number) = default_table_number {
-            let existing_name = self
-                .table_mapping()
-                .iter()
-                .find(|(_, _, table_number, _)| *table_number == default_table_number);
-            let table_number = if let Some((_, _, _, existing_name)) = existing_name {
+            let existing_name =
+                self.table_mapping().namespace(namespace).number_to_name()(default_table_number)
+                    .ok();
+            let table_number = if let Some(existing_name) = existing_name {
                 // In tests, have a hard failure on conflicting default table numbers. In
                 // real system, have a looser fallback where we pick
                 // another table number.
-                // Also allow this when components are enabled, because system tables in
-                // different namespaces have different table numbers.
-                if !*COMPONENTS_ENABLED && cfg!(any(test, feature = "testing")) {
+                if cfg!(any(test, feature = "testing")) {
                     anyhow::bail!(
                         "{default_table_number} is used by both {table_name} and {existing_name}"
                     );
@@ -665,7 +662,9 @@ impl<RT: Runtime> Transaction<RT> {
                 // If the table_number requested is taken, just pick a higher table number.
                 // This might be true for older backends that have lower-numbered system
                 // tables.
-                TableModel::new(self).next_system_table_number().await?
+                TableModel::new(self)
+                    .next_system_table_number(namespace)
+                    .await?
             } else {
                 default_table_number
             };
@@ -680,7 +679,9 @@ impl<RT: Runtime> Transaction<RT> {
             );
             table_number
         } else {
-            TableModel::new(self).next_system_table_number().await?
+            TableModel::new(self)
+                .next_system_table_number(namespace)
+                .await?
         })
     }
 
@@ -701,7 +702,7 @@ impl<RT: Runtime> Transaction<RT> {
         let is_new = !TableModel::new(self).table_exists(namespace, table_name);
         if is_new {
             let table_number = self
-                .table_number_for_system_table(table_name, default_table_number)
+                .table_number_for_system_table(namespace, table_name, default_table_number)
                 .await?;
             let metadata = TableMetadata::new(namespace, table_name.clone(), table_number);
             let table_doc_id = SystemMetadataModel::new_global(self)
@@ -750,7 +751,7 @@ impl<RT: Runtime> Transaction<RT> {
             .name_exists(table_name);
         if is_new {
             let table_number = self
-                .table_number_for_system_table(table_name, default_table_number)
+                .table_number_for_system_table(namespace, table_name, default_table_number)
                 .await?;
             let metadata = VirtualTableMetadata::new(namespace, table_name.clone(), table_number);
             let table_doc_id = SystemMetadataModel::new_global(self)

crates/isolate/src/tests/user_error.rs

@@ -149,7 +149,9 @@ async fn test_nonexistent_table(rt: TestRuntime) -> anyhow::Result<()> {
         t.create_index("boatVotes.by_boat", "boat").await?;
         t.backfill_indexes().await?;
         let mut tx = t.database.begin(Identity::system()).await?;
-        let table_number = TableModel::new(&mut tx).next_user_table_number().await?;
+        let table_number = TableModel::new(&mut tx)
+            .next_user_table_number(TableNamespace::test_user())
+            .await?;
         let nonexistent_id = DeveloperDocumentId::new(table_number, InternalId::MIN);
         t.mutation(
             "userError:nonexistentTable",