[components] pass namespace into ResolvedQuery/DeveloperQuery (#26021)

some more threading of TableNamespace -- this time from ResolvedQuery all the way down to the table mapping, which involves passing it to everywhere we resolve `IndexName<TableName>` -> `TabletIndexName`.

Everything still uses TableNamespace::Global, it's just being passed explicitly to more places.

This refactor will allow different module definitions to be looked up in different namespaces.

GitOrigin-RevId: 8dd562c578c7d622984c8e1cb4ea8e5b0db50e2e

Author: ldanilek

crates/application/src/cron_jobs/mod.rs

@@ -78,7 +78,10 @@ use model::{
     modules::ModuleModel,
 };
 use usage_tracking::FunctionUsageTracker;
-use value::ResolvedDocumentId;
+use value::{
+    ResolvedDocumentId,
+    TableNamespace,
+};
 
 use crate::{
     application_function_runner::ApplicationFunctionRunner,
@@ -162,7 +165,8 @@ impl<RT: Runtime> CronJobExecutor<RT> {
                 range: vec![],
                 order: Order::Asc,
             });
-            let mut query_stream = ResolvedQuery::new(&mut tx, index_query)?;
+            let mut query_stream =
+                ResolvedQuery::new(&mut tx, TableNamespace::Global, index_query)?;
 
             let mut next_job_wait = None;
             while let Some(doc) = query_stream.next(&mut tx, None).await? {

crates/application/src/export_worker.rs

@@ -112,6 +112,7 @@ use usage_tracking::{
 use value::{
     export::ValueFormat,
     id_v6::DeveloperDocumentId,
+    TableNamespace,
     TableNumber,
     TabletId,
     VirtualTableMapping,
@@ -263,7 +264,7 @@ impl<RT: Runtime> ExportWorker<RT> {
             order: Order::Asc,
         };
         let query = common::query::Query::index_range(index_range);
-        let mut query_stream = ResolvedQuery::new(tx, query)?;
+        let mut query_stream = ResolvedQuery::new(tx, TableNamespace::Global, query)?;
         query_stream
             .expect_at_most_one(tx)
             .await?
@@ -287,7 +288,7 @@ impl<RT: Runtime> ExportWorker<RT> {
             order: Order::Desc,
         };
         let query = common::query::Query::index_range(index_range);
-        let mut query_stream = ResolvedQuery::new(tx, query)?;
+        let mut query_stream = ResolvedQuery::new(tx, TableNamespace::Global, query)?;
         query_stream.expect_at_most_one(tx).await
     }
 

crates/application/src/lib.rs

@@ -264,6 +264,7 @@ use value::{
     ConvexValue,
     Namespace,
     ResolvedDocumentId,
+    TableNamespace,
 };
 use vector::{
     PublicVectorSearchQueryResult,
@@ -2041,8 +2042,8 @@ impl<RT: Runtime> Application<RT> {
                 IndexMetadata::new_backfilling(*tx.begin_timestamp(), index_name, index_fields);
             let mut model = IndexModel::new(&mut tx);
             if let Some(existing_index_metadata) = model
-                .pending_index_metadata(&index_metadata.name)?
-                .or(model.enabled_index_metadata(&index_metadata.name)?)
+                .pending_index_metadata(TableNamespace::Global, &index_metadata.name)?
+                .or(model.enabled_index_metadata(TableNamespace::Global, &index_metadata.name)?)
             {
                 if !index_metadata
                     .config

crates/application/src/scheduled_jobs/mod.rs

@@ -85,7 +85,10 @@ use model::{
 use parking_lot::Mutex;
 use sync_types::Timestamp;
 use usage_tracking::FunctionUsageTracker;
-use value::ResolvedDocumentId;
+use value::{
+    ResolvedDocumentId,
+    TableNamespace,
+};
 
 use crate::{
     application_function_runner::ApplicationFunctionRunner,
@@ -326,7 +329,7 @@ impl<RT: Runtime> ScheduledJobExecutor<RT> {
             )],
             order: Order::Asc,
         });
-        let mut query_stream = ResolvedQuery::new(tx, index_query)?;
+        let mut query_stream = ResolvedQuery::new(tx, TableNamespace::Global, index_query)?;
         while let Some(doc) = query_stream.next(tx, None).await? {
             let job: ParsedDocument<ScheduledJob> = doc.try_into()?;
             let (job_id, job) = job.clone().into_id_and_value();
@@ -820,7 +823,8 @@ impl<RT: Runtime> ScheduledJobGarbageCollector<RT> {
                 order: Order::Asc,
             })
             .limit(*SCHEDULED_JOB_GARBAGE_COLLECTION_BATCH_SIZE);
-            let mut query_stream = ResolvedQuery::new(&mut tx, index_query)?;
+            let mut query_stream =
+                ResolvedQuery::new(&mut tx, TableNamespace::Global, index_query)?;
 
             let mut next_job_wait = None;
             let mut jobs_to_delete = vec![];

crates/application/src/snapshot_import.rs

@@ -2378,6 +2378,7 @@ mod tests {
         ConvexObject,
         FieldName,
         TableName,
+        TableNamespace,
     };
 
     use super::{
@@ -2992,7 +2993,7 @@ a
             let mut tx = app.begin(identity.clone()).await?;
             let mut index_model = IndexModel::new(&mut tx);
             let index = index_model
-                .enabled_index_metadata(&index_name)?
+                .enabled_index_metadata(TableNamespace::Global, &index_name)?
                 .context("index does not exist")?;
             assert_ne!(index.id(), index_id);
             assert!(index.config.is_enabled());
@@ -3072,7 +3073,7 @@ a
         let mut tx = app.begin(new_admin_id()).await?;
         let table_name = TableName::from_str(table_name)?;
         let query = common::query::Query::full_table_scan(table_name.clone(), Order::Asc);
-        let mut query_stream = ResolvedQuery::new(&mut tx, query)?;
+        let mut query_stream = ResolvedQuery::new(&mut tx, TableNamespace::Global, query)?;
 
         let mut docs: Vec<ResolvedDocument> = Vec::new();
         while let Some(doc) = query_stream.next(&mut tx, None).await? {

crates/application/src/tests/cron_jobs.rs

@@ -45,6 +45,7 @@ use model::{
 };
 use runtime::testing::TestRuntime;
 use serde_json::Value as JsonValue;
+use value::TableNamespace;
 
 use crate::{
     test_helpers::{
@@ -88,6 +89,7 @@ async fn create_cron_job(
 fn cron_log_query<RT: Runtime>(tx: &mut Transaction<RT>) -> anyhow::Result<DeveloperQuery<RT>> {
     DeveloperQuery::new(
         tx,
+        TableNamespace::Global,
         Query::index_range(IndexRange {
             index_name: CRON_JOB_LOGS_INDEX_BY_NAME_TS.clone(),
             range: vec![IndexRangeExpression::Eq(

crates/common/src/components/mod.rs

@@ -1,7 +1,10 @@
 use std::sync::LazyLock;
 
 use cmd_util::env::env_config;
-use value::InternalId;
+use value::{
+    InternalId,
+    TableNamespace,
+};
 
 mod component_definition_path;
 mod component_path;
@@ -91,3 +94,14 @@ impl ComponentDefinitionId {
         matches!(self, ComponentDefinitionId::Root)
     }
 }
+
+impl From<ComponentDefinitionId> for TableNamespace {
+    fn from(value: ComponentDefinitionId) -> Self {
+        // TODO(lee) convert these to the correct namespaces:
+        // RootComponentDefinition and ByComponentDefinition respectively.
+        match value {
+            ComponentDefinitionId::Root => TableNamespace::Global,
+            ComponentDefinitionId::Child(_id) => TableNamespace::Global,
+        }
+    }
+}

crates/database/src/bootstrap_model/components/mod.rs

@@ -107,6 +107,7 @@ impl<'a, RT: Runtime> BootstrapComponentsModel<'a, RT> {
         };
         let mut query = ResolvedQuery::new(
             self.tx,
+            TableNamespace::Global,
             Query::index_range(IndexRange {
                 index_name: COMPONENTS_BY_PARENT_INDEX.clone(),
                 range,

crates/database/src/bootstrap_model/index.rs

@@ -169,9 +169,13 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
     }
 
     #[cfg(any(test, feature = "testing"))]
-    pub async fn enable_index_for_testing(&mut self, index: &IndexName) -> anyhow::Result<()> {
+    pub async fn enable_index_for_testing(
+        &mut self,
+        namespace: TableNamespace,
+        index: &IndexName,
+    ) -> anyhow::Result<()> {
         let metadata = self
-            .pending_index_metadata(index)?
+            .pending_index_metadata(namespace, index)?
             .ok_or_else(|| anyhow::anyhow!("Failed to find pending index: {}", index))?;
         self.enable_index(&metadata.into_value()).await
     }
@@ -369,7 +373,7 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
         for new_index in indexes_in_schema {
             remaining_indexes.remove(&new_index.name);
 
-            match self.compare_new_and_existing_indexes(new_index)? {
+            match self.compare_new_and_existing_indexes(TableNamespace::Global, new_index)? {
                 IndexComparison::Added(index) => diff.added.push(index),
                 IndexComparison::Identical(index) => diff.identical.push(index),
                 IndexComparison::Replaced {
@@ -402,10 +406,11 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
 
     fn compare_new_and_existing_indexes(
         &mut self,
+        namespace: TableNamespace,
         new_index: DeveloperIndexMetadata,
     ) -> anyhow::Result<IndexComparison> {
-        let pending_index = self.pending_index_metadata(&new_index.name)?;
-        let enabled_index = self.enabled_index_metadata(&new_index.name)?;
+        let pending_index = self.pending_index_metadata(namespace, &new_index.name)?;
+        let enabled_index = self.enabled_index_metadata(namespace, &new_index.name)?;
 
         fn identical_or_replaced(
             existing_index: ParsedDocument<TabletIndexMetadata>,
@@ -543,9 +548,10 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
     /// pending_index_metadata.
     pub fn enabled_index_metadata(
         &mut self,
+        namespace: TableNamespace,
         index_name: &IndexName,
     ) -> anyhow::Result<Option<ParsedDocument<TabletIndexMetadata>>> {
-        self._index_metadata(index_name, |indexes, reads, index_name| {
+        self._index_metadata(namespace, index_name, |indexes, reads, index_name| {
             indexes.get_enabled(reads, &index_name)
         })
     }
@@ -558,9 +564,10 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
     /// or require_enabled_index_metadata instead.
     pub fn pending_index_metadata(
         &mut self,
+        namespace: TableNamespace,
         index_name: &IndexName,
     ) -> anyhow::Result<Option<ParsedDocument<TabletIndexMetadata>>> {
-        self._index_metadata(index_name, |indexes, reads, index_name| {
+        self._index_metadata(namespace, index_name, |indexes, reads, index_name| {
             indexes.get_pending(reads, &index_name)
         })
     }
@@ -575,6 +582,7 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
 
     fn _index_metadata<'b>(
         &'b mut self,
+        namespace: TableNamespace,
         index_name: &IndexName,
         getter: impl FnOnce(
             &'b mut TransactionIndex,
@@ -585,18 +593,19 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
         if !self
             .tx
             .table_mapping()
-            .namespace(TableNamespace::Global)
+            .namespace(namespace)
             .name_exists(index_name.table())
         {
             return Ok(None);
         }
-        let index_name = self.resolve_index_name(index_name)?;
+        let index_name = self.resolve_index_name(namespace, index_name)?;
         Ok(getter(&mut self.tx.index, &mut self.tx.reads, index_name)
             .map(|index| index.metadata.clone()))
     }
 
     pub fn stable_index_name(
         &mut self,
+        namespace: TableNamespace,
         index_name: &IndexName,
         table_filter: TableFilter,
     ) -> anyhow::Result<StableIndexName> {
@@ -612,24 +621,24 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
                 .clone();
             Ok(StableIndexName::Virtual(
                 index_name.clone(),
-                self.resolve_index_name(&physical_index_name)?,
+                self.resolve_index_name(namespace, &physical_index_name)?,
             ))
         } else if self
             .tx
             .table_mapping()
-            .namespace(TableNamespace::Global)
+            .namespace(namespace)
             .name_exists(index_name.table())
         {
             match table_filter {
                 TableFilter::IncludePrivateSystemTables => Ok(StableIndexName::Physical(
-                    self.resolve_index_name(index_name)?,
+                    self.resolve_index_name(namespace, index_name)?,
                 )),
                 TableFilter::ExcludePrivateSystemTables => {
                     if index_name.table().is_system() {
                         Ok(StableIndexName::Missing)
                     } else {
                         Ok(StableIndexName::Physical(
-                            self.resolve_index_name(index_name)?,
+                            self.resolve_index_name(namespace, index_name)?,
                         ))
                     }
                 },
@@ -639,14 +648,14 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
         }
     }
 
-    fn resolve_index_name(&mut self, index_name: &IndexName) -> anyhow::Result<TabletIndexName> {
-        let resolved = index_name.clone().map_table(
-            &self
-                .tx
-                .table_mapping()
-                .namespace(TableNamespace::Global)
-                .name_to_id(),
-        )?;
+    fn resolve_index_name(
+        &mut self,
+        namespace: TableNamespace,
+        index_name: &IndexName,
+    ) -> anyhow::Result<TabletIndexName> {
+        let resolved = index_name
+            .clone()
+            .map_table(&self.tx.table_mapping().namespace(namespace).name_to_id())?;
         Ok(resolved.into())
     }
 
@@ -699,7 +708,7 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
             range: vec![],
             order: Order::Asc,
         });
-        let mut query_stream = ResolvedQuery::new(self.tx, index_query)?;
+        let mut query_stream = ResolvedQuery::new(self.tx, TableNamespace::Global, index_query)?;
 
         let mut indexes = vec![];
         while let Some(document) = query_stream.next(self.tx, None).await? {

crates/database/src/bootstrap_model/index_workers/mod.rs

@@ -30,6 +30,7 @@ use value::{
     FieldPath,
     InternalId,
     TableName,
+    TableNamespace,
 };
 
 use crate::{
@@ -104,7 +105,7 @@ impl<'a, RT: Runtime> IndexWorkerMetadataModel<'a, RT> {
             range,
             order: Order::Asc,
         });
-        let mut query_stream = ResolvedQuery::new(self.tx, query)?;
+        let mut query_stream = ResolvedQuery::new(self.tx, TableNamespace::Global, query)?;
         let result = query_stream.next(self.tx, None).await?;
         result
             .map(ParsedDocument::<IndexWorkerMetadataRecord>::try_from)