allow virtual table number = system table number (#28130)

turns out #28008 only covered one specific case: activating a hidden system table with the same table number as its virtual table. There are other cases we need to cover:

1. if default_table_number for a virtual table matches an existing system table, we should allow it to be used if the virtual and system tables match.
2. if the system table is created first, we should allow a new virtual table to be created later as long as they match

with these two cases covered, a local test that changes default table number of `_scheduled_jobs` to be the same as that of `_scheduled_functions` works!

don't be daunted by the size of this PR. it's basically doing this in two places:

```
self.virtual_system_mapping.is_virtual_table(table_name)
                && self
                    .virtual_system_mapping
                    .virtual_to_system_table(table_name)?
                    == existing_table
```

and then adds several tests

GitOrigin-RevId: 4f846add6b9f79effb53776b191acd4fa38c2ff9

crates/database/src/table_registry.rs

@@ -231,14 +231,27 @@ impl TableRegistry {
         table_number: TableNumber,
         table_name: &TableName,
     ) -> anyhow::Result<()> {
-        anyhow::ensure!(
-            !self
-                .table_mapping
-                .namespace(namespace)
-                .table_number_exists()(table_number),
-            "Cannot add a table with table number {table_number} since it already exists in the \
-             table mapping"
-        );
+        if let Some(existing_table) = self
+            .table_mapping
+            .namespace(namespace)
+            .name_by_number_if_exists(table_number)
+        {
+            if self.virtual_system_mapping.is_virtual_table(table_name)
+                && self
+                    .virtual_system_mapping
+                    .virtual_to_system_table(table_name)?
+                    == existing_table
+            {
+                // A virtual table can share a table number with its physical
+                // table.
+            } else {
+                anyhow::ensure!(
+                    existing_table == table_name,
+                    "Cannot add a table {table_name} with table number {table_number} since it \
+                     already exists in the table mapping as {existing_table}"
+                );
+            }
+        }
         if let Ok(existing_virtual_table) = self
             .virtual_table_mapping
             .namespace(namespace)
@@ -253,8 +266,8 @@ impl TableRegistry {
                 // table.
             } else {
                 anyhow::bail!(
-                    "Cannot add a table with table number {table_number} since it already exists \
-                     in the virtual table mapping"
+                    "Cannot add a table {table_name} with table number {table_number} since it \
+                     already exists in the virtual table mapping as {existing_virtual_table}"
                 );
             }
         }
@@ -397,3 +410,151 @@ impl<'a> Update<'a> {
         self.table_update
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{
+        collections::BTreeMap,
+        sync::Arc,
+    };
+
+    use common::{
+        bootstrap_model::{
+            index::{
+                database_index::IndexedFields,
+                TabletIndexMetadata,
+                INDEX_TABLE,
+            },
+            tables::{
+                TableMetadata,
+                TABLES_TABLE,
+            },
+        },
+        document::{
+            CreationTime,
+            ResolvedDocument,
+        },
+        testing::TestIdGenerator,
+        types::{
+            PersistenceVersion,
+            TabletIndexName,
+        },
+    };
+    use imbl::OrdMap;
+    use indexing::index_registry::IndexRegistry;
+    use value::{
+        TableNamespace,
+        VirtualTableMapping,
+    };
+
+    use crate::{
+        virtual_tables::NoopDocMapper,
+        TableRegistry,
+        VirtualSystemMapping,
+        VirtualTableMetadata,
+        VIRTUAL_TABLES_TABLE,
+    };
+
+    fn make_registries() -> anyhow::Result<(TableRegistry, IndexRegistry, TestIdGenerator)> {
+        let mut id_generator = TestIdGenerator::new();
+        let index_tablet = id_generator.system_table_id(&INDEX_TABLE).tablet_id;
+        let index_by_id = id_generator.system_generate(&INDEX_TABLE);
+        id_generator.system_table_id(&TABLES_TABLE);
+        id_generator.system_table_id(&VIRTUAL_TABLES_TABLE);
+        let by_id_index_metadata = TabletIndexMetadata::new_enabled(
+            TabletIndexName::by_id(index_tablet),
+            IndexedFields::by_id(),
+        );
+        let mut virtual_system_mapping = VirtualSystemMapping::default();
+        virtual_system_mapping.add_table(
+            &"_storage".parse()?,
+            &"_file_storage".parse()?,
+            BTreeMap::new(),
+            Arc::new(NoopDocMapper),
+        );
+        let table_registry = TableRegistry::bootstrap(
+            id_generator.clone(),
+            OrdMap::new(),
+            PersistenceVersion::V5,
+            VirtualTableMapping::new(),
+            virtual_system_mapping,
+        )?;
+        let index_documents = vec![ResolvedDocument::new(
+            index_by_id,
+            CreationTime::ONE,
+            by_id_index_metadata.try_into()?,
+        )?];
+        let index_registry = IndexRegistry::bootstrap(
+            &id_generator,
+            index_documents.iter(),
+            PersistenceVersion::V5,
+        )?;
+        Ok((table_registry, index_registry, id_generator))
+    }
+
+    #[test]
+    fn test_create_system_then_virtual_table_same_number() -> anyhow::Result<()> {
+        let (mut table_registry, index_registry, mut id_generator) = make_registries()?;
+
+        let table_number = 530.try_into()?;
+        let system_table_id = id_generator.system_generate(&TABLES_TABLE);
+        let virtual_table_id = id_generator.system_generate(&VIRTUAL_TABLES_TABLE);
+
+        let table_metadata = TableMetadata::new(
+            TableNamespace::test_user(),
+            "_file_storage".parse()?,
+            table_number,
+        );
+        table_registry.update(
+            &index_registry,
+            system_table_id,
+            None,
+            Some(&table_metadata.try_into()?),
+        )?;
+        let virtual_table_metadata = VirtualTableMetadata::new(
+            TableNamespace::test_user(),
+            "_storage".parse()?,
+            table_number,
+        );
+        table_registry.update(
+            &index_registry,
+            virtual_table_id,
+            None,
+            Some(&virtual_table_metadata.try_into()?),
+        )?;
+        Ok(())
+    }
+
+    #[test]
+    fn test_create_virtual_then_system_table_same_number() -> anyhow::Result<()> {
+        let (mut table_registry, index_registry, mut id_generator) = make_registries()?;
+
+        let table_number = 530.try_into()?;
+        let system_table_id = id_generator.system_generate(&TABLES_TABLE);
+        let virtual_table_id = id_generator.system_generate(&VIRTUAL_TABLES_TABLE);
+
+        let virtual_table_metadata = VirtualTableMetadata::new(
+            TableNamespace::test_user(),
+            "_storage".parse()?,
+            table_number,
+        );
+        table_registry.update(
+            &index_registry,
+            virtual_table_id,
+            None,
+            Some(&virtual_table_metadata.try_into()?),
+        )?;
+        let table_metadata = TableMetadata::new(
+            TableNamespace::test_user(),
+            "_file_storage".parse()?,
+            table_number,
+        );
+        table_registry.update(
+            &index_registry,
+            system_table_id,
+            None,
+            Some(&table_metadata.try_into()?),
+        )?;
+        Ok(())
+    }
+}

crates/database/src/tests/mod.rs

@@ -110,6 +110,7 @@ use crate::{
         DbFixtures,
         DbFixturesArgs,
     },
+    virtual_tables::NoopDocMapper,
     write_log::WriteSource,
     Database,
     DatabaseSnapshot,
@@ -1989,6 +1990,57 @@ async fn create_system_table_with_non_system_table_fails(rt: TestRuntime) -> any
     Ok(())
 }
 
+#[convex_macro::test_runtime]
+async fn test_create_system_table_then_virtual_table(rt: TestRuntime) -> anyhow::Result<()> {
+    let mut virtual_system_mapping = VirtualSystemMapping::default();
+    virtual_system_mapping.add_table(
+        &"_storage".parse()?,
+        &"_file_storage".parse()?,
+        BTreeMap::new(),
+        Arc::new(NoopDocMapper),
+    );
+    let db = DbFixtures::new_with_args(
+        &rt,
+        DbFixturesArgs {
+            virtual_system_mapping,
+            ..Default::default()
+        },
+    )
+    .await?
+    .db;
+
+    let table_number = 530.try_into()?;
+    let mut tx = db.begin_system().await?;
+    tx.create_system_table(
+        TableNamespace::test_user(),
+        &"_file_storage".parse()?,
+        Some(table_number),
+    )
+    .await?;
+    tx.create_virtual_table(
+        TableNamespace::test_user(),
+        &"_storage".parse()?,
+        Some(table_number),
+    )
+    .await?;
+    db.commit(tx).await?;
+
+    let mut tx = db.begin_system().await?;
+    let physical_table_number = tx
+        .table_mapping()
+        .namespace(TableNamespace::test_user())
+        .id(&"_file_storage".parse()?)?
+        .table_number;
+    let virtual_table_number = tx
+        .virtual_table_mapping()
+        .namespace(TableNamespace::test_user())
+        .number(&"_storage".parse()?)?;
+    assert_eq!(physical_table_number, table_number);
+    assert_eq!(virtual_table_number, table_number);
+
+    Ok(())
+}
+
 #[convex_macro::test_runtime]
 async fn test_virtual_table_transaction(rt: TestRuntime) -> anyhow::Result<()> {
     let db = new_test_database(rt).await;

crates/database/src/transaction.rs

@@ -636,21 +636,30 @@ impl<RT: Runtime> Transaction<RT> {
                 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.
-                if cfg!(any(test, feature = "testing")) {
+                if self.virtual_system_mapping.is_virtual_table(table_name)
+                    && *self
+                        .virtual_system_mapping
+                        .virtual_to_system_table(table_name)?
+                        == existing_name
+                {
+                    // Table number is occupied by the system table for the virtual table we're
+                    // creating. Allow it.
+                    default_table_number
+                } else if cfg!(any(test, feature = "testing")) {
+                    // In tests, have a hard failure on conflicting default table numbers. In
+                    // real system, have a looser fallback where we pick
+                    // another table number.
                     anyhow::bail!(
                         "{default_table_number} is used by both {table_name} and {existing_name}"
                     );
+                } else {
+                    // 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(namespace)
+                        .await?
                 }
-
-                // 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(namespace)
-                    .await?
             } else {
                 default_table_number
             };

crates/database/src/virtual_tables/mod.rs

@@ -111,6 +111,43 @@ pub trait VirtualSystemDocMapper: Send + Sync {
     ) -> anyhow::Result<DeveloperDocument>;
 }
 
+#[cfg(any(test, feature = "testing"))]
+pub struct NoopDocMapper;
+
+#[cfg(any(test, feature = "testing"))]
+pub mod test_virtual_system_mapping {
+    use common::{
+        document::{
+            DeveloperDocument,
+            ResolvedDocument,
+        },
+        version::Version,
+    };
+    use value::{
+        TableMapping,
+        VirtualTableMapping,
+    };
+
+    use super::NoopDocMapper;
+    use crate::{
+        VirtualSystemDocMapper,
+        VirtualSystemMapping,
+    };
+
+    impl VirtualSystemDocMapper for NoopDocMapper {
+        fn system_to_virtual_doc(
+            &self,
+            _virtual_system_mapping: &VirtualSystemMapping,
+            doc: ResolvedDocument,
+            _table_mapping: &TableMapping,
+            _virtual_table_mapping: &VirtualTableMapping,
+            _version: Version,
+        ) -> anyhow::Result<DeveloperDocument> {
+            Ok(doc.to_developer())
+        }
+    }
+}
+
 #[derive(Clone, Default)]
 pub struct VirtualSystemMapping {
     virtual_to_system: OrdMap<TableName, TableName>,