[components] fix TODOs related to source packages (#26994)

enable looking up source packages and udf config within components

GitOrigin-RevId: acdb963dbdd7963de0f2f9d26bc359f16d1a3cc0

Author: ldanilek

crates/application/src/application_function_runner/mod.rs

@@ -187,7 +187,6 @@ use usage_tracking::{
 use value::{
     heap_size::HeapSize,
     id_v6::DeveloperDocumentId,
-    TableNamespace,
 };
 use vector::{
     PublicVectorSearchQueryResult,
@@ -1019,10 +1018,11 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
             FunctionOutcome::Mutation(o) => o,
             _ => anyhow::bail!("Received non-mutation outcome for mutation"),
         };
+        let (_, component) = BootstrapComponentsModel::new(&mut tx)
+            .component_path_to_ids(path.component.clone())
+            .await?;
 
-        let table_mapping = tx
-            .table_mapping()
-            .namespace(TableNamespace::by_component_TODO());
+        let table_mapping = tx.table_mapping().namespace(component.into());
         let virtual_table_mapping = tx.virtual_table_mapping().clone();
 
         let outcome = ValidatedUdfOutcome::new(
@@ -1162,6 +1162,9 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
             UdfType::Action,
         )
         .await?;
+        let (_, component) = BootstrapComponentsModel::new(&mut tx)
+            .component_path_to_ids(path.component.clone())
+            .await?;
 
         // Fetch the returns_validator now to be used at a later ts.
         let (path_and_args, returns_validator) = match validate_result {
@@ -1189,9 +1192,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
 
         // We should use table mappings from the same transaction as the output
         // validator was retrieved.
-        let table_mapping = tx
-            .table_mapping()
-            .namespace(TableNamespace::by_component_TODO());
+        let table_mapping = tx.table_mapping().namespace(component.into());
         let virtual_table_mapping = tx.virtual_table_mapping().clone();
         let udf_server_version = path_and_args.npm_version().clone();
         // We should not be missing the module given we validated the path above
@@ -1273,11 +1274,10 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
                 }
 
                 let source_package_id = module.source_package_id;
-                let source_package =
-                    SourcePackageModel::new(&mut tx, TableNamespace::by_component_TODO())
-                        .get(source_package_id)
-                        .await?
-                        .into_value();
+                let source_package = SourcePackageModel::new(&mut tx, component.into())
+                    .get(source_package_id)
+                    .await?
+                    .into_value();
                 let mut environment_variables =
                     EnvironmentVariablesModel::new(&mut tx).get_all().await?;
                 // Insert special environment variables if not already provided by user

crates/application/src/cache/mod.rs

@@ -41,6 +41,7 @@ use common::{
     RequestId,
 };
 use database::{
+    BootstrapComponentsModel,
     Database,
     Token,
 };
@@ -80,7 +81,6 @@ use usage_tracking::FunctionUsageTracker;
 use value::{
     heap_size::HeapSize,
     ConvexValue,
-    TableNamespace,
 };
 
 use crate::{
@@ -471,9 +471,10 @@ impl<RT: Runtime> CacheManager<RT> {
                         };
                         if let Ok(ref json_packed_value) = &query_outcome.result {
                             let output: ConvexValue = json_packed_value.unpack();
-                            let table_mapping = tx
-                                .table_mapping()
-                                .namespace(TableNamespace::by_component_TODO());
+                            let (_, component) = BootstrapComponentsModel::new(&mut tx)
+                                .component_path_to_ids(path.component.clone())
+                                .await?;
+                            let table_mapping = tx.table_mapping().namespace(component.into());
                             let virtual_table_mapping = tx.virtual_table_mapping().clone();
                             let returns_validation_error = returns_validator.check_output(
                                 &output,

crates/application/src/tests/source_package.rs

@@ -69,7 +69,7 @@ pub async fn assemble_package<RT: Runtime>(
     modifications: BTreeMap<CanonicalizedModulePath, Option<ModuleConfig>>,
 ) -> anyhow::Result<Vec<ModuleConfig>> {
     let existing_modules = ModuleModel::new(tx)
-        .get_application_modules(ComponentId::TODO(), module_loader)
+        .get_application_modules(ComponentId::test_user(), module_loader)
         .await?;
     let mut modules = BTreeMap::new();
     for (path, module) in existing_modules {

crates/common/src/shapes/dashboard.rs

@@ -180,7 +180,7 @@ mod tests {
             ShapeEnumJson::Id { table_name } => {
                 let name: TableName = table_name.parse()?;
                 ReducedShape::Id(all_tables_name_to_number(
-                    TableNamespace::by_component_TODO(),
+                    TableNamespace::test_user(),
                     mapping,
                     virtual_mapping,
                 )(name)?)

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

@@ -127,10 +127,7 @@ impl<'a, RT: Runtime> SchemaModel<'a, RT> {
     }
 
     pub async fn enforce(&mut self, document: &ResolvedDocument) -> anyhow::Result<()> {
-        let schema_table_mapping = self
-            .tx
-            .table_mapping()
-            .namespace(TableNamespace::by_component_TODO());
+        let schema_table_mapping = self.tx.table_mapping().namespace(self.namespace);
         self.enforce_with_table_mapping(document, &schema_table_mapping)
             .await
     }
@@ -249,9 +246,9 @@ impl<'a, RT: Runtime> SchemaModel<'a, RT> {
     ) -> anyhow::Result<(ResolvedDocumentId, SchemaState)> {
         let mut table_model = TableModel::new(self.tx);
         for name in schema.tables.keys() {
-            if !table_model.table_exists(TableNamespace::by_component_TODO(), name) {
+            if !table_model.table_exists(self.namespace, name) {
                 table_model
-                    .insert_table_metadata(TableNamespace::by_component_TODO(), name)
+                    .insert_table_metadata(self.namespace, name)
                     .await?;
             }
         }

crates/database/src/query/search_query.rs

@@ -97,13 +97,18 @@ impl SearchQuery {
             .into_iter()
             .filter(|(_, index_key)| self.cursor_interval.contains(index_key))
             .collect();
+        let namespace = match self.stable_index_name.tablet_index_name() {
+            Some(index_name) => tx.table_mapping().tablet_namespace(*index_name.table())?,
+            None => TableNamespace::Global,
+        };
         let table_number = tx
             .table_mapping()
-            .namespace(TableNamespace::by_component_TODO())
+            .namespace(namespace)
             .id(&self.query.table)?
             .table_number;
         Ok(SearchResultIterator::new(
             revisions_in_range,
+            namespace,
             table_number,
             self.version.clone(),
         ))
@@ -178,6 +183,7 @@ impl QueryStream for SearchQuery {
 
 #[derive(Clone)]
 struct SearchResultIterator {
+    namespace: TableNamespace,
     table_number: TableNumber,
     candidates: Vec<(CandidateRevision, IndexKeyBytes)>,
     next_index: usize,
@@ -188,10 +194,12 @@ struct SearchResultIterator {
 impl SearchResultIterator {
     fn new(
         candidates: Vec<(CandidateRevision, IndexKeyBytes)>,
+        namespace: TableNamespace,
         table_number: TableNumber,
         version: Option<Version>,
     ) -> Self {
         Self {
+            namespace,
             table_number,
             candidates,
             next_index: 0,
@@ -232,13 +240,12 @@ impl SearchResultIterator {
         self.next_index += 1;
 
         let id = self.table_number.id(candidate.id);
-        let (document, existing_doc_ts) =
-            UserFacingModel::new(tx, TableNamespace::by_component_TODO())
-                .get_with_ts(id, self.version.clone())
-                .await?
-                .ok_or_else(|| {
-                    anyhow::anyhow!("Unable to load search result {id}@{:?}", candidate.ts)
-                })?;
+        let (document, existing_doc_ts) = UserFacingModel::new(tx, self.namespace)
+            .get_with_ts(id, self.version.clone())
+            .await?
+            .ok_or_else(|| {
+                anyhow::anyhow!("Unable to load search result {id}@{:?}", candidate.ts)
+            })?;
 
         self.bytes_read += document.size();
 

crates/isolate/src/environment/action/phase.rs

@@ -4,10 +4,8 @@ use std::{
     sync::Arc,
 };
 
-use anyhow::Context;
 use common::{
     components::{
-        ComponentDefinitionId,
         ComponentId,
         ComponentPath,
         Reference,
@@ -52,7 +50,6 @@ use sync_types::{
     CanonicalizedModulePath,
     ModulePath,
 };
-use value::TableNamespace;
 
 use crate::{
     concurrency_limiter::ConcurrencyPermit,
@@ -137,10 +134,18 @@ impl<RT: Runtime> ActionPhase<RT> {
             anyhow::bail!("ActionPhase initialized twice");
         };
 
+        let component_id = with_release_permit(timeout, permit_slot, async {
+            let (_, component) = BootstrapComponentsModel::new(&mut tx)
+                .component_path_to_ids(self.component.clone())
+                .await?;
+            anyhow::Ok(component)
+        })
+        .await?;
+
         let udf_config = with_release_permit(
             timeout,
             permit_slot,
-            UdfConfigModel::new(&mut tx, TableNamespace::TODO()).get(),
+            UdfConfigModel::new(&mut tx, component_id.into()).get(),
         )
         .await?;
 
@@ -156,18 +161,6 @@ impl<RT: Runtime> ActionPhase<RT> {
                     .get_all_metadata(ComponentId::Root)
                     .await?
             } else {
-                let metadata = BootstrapComponentsModel::new(&mut tx)
-                    .resolve_path(self.component.clone())
-                    .await?
-                    .context("Failed to find component")?;
-                let (component_id, _) = if self.component.is_root() {
-                    (ComponentId::Root, ComponentDefinitionId::Root)
-                } else {
-                    (
-                        ComponentId::Child(metadata.id().internal_id()),
-                        ComponentDefinitionId::Child(metadata.definition_id),
-                    )
-                };
                 let module_metadata = ModuleModel::new(&mut tx)
                     .get_all_metadata(component_id)
                     .await?;

crates/isolate/src/environment/helpers/validation.rs

@@ -3,6 +3,7 @@ use common::{
     components::{
         CanonicalizedComponentFunctionPath,
         ComponentFunctionPath,
+        ComponentId,
         ComponentPath,
     },
     errors::JsError,
@@ -24,6 +25,7 @@ use common::{
 };
 use database::{
     unauthorized_error,
+    BootstrapComponentsModel,
     Transaction,
 };
 use errors::ErrorMetadata;
@@ -56,7 +58,6 @@ use value::{
     ConvexArray,
     ConvexValue,
     NamespacedTableMapping,
-    TableNamespace,
     VirtualTableMapping,
 };
 
@@ -146,11 +147,10 @@ fn missing_or_internal_error(path: &CanonicalizedComponentFunctionPath) -> anyho
 
 async fn udf_version<RT: Runtime>(
     path: &CanonicalizedComponentFunctionPath,
+    component: ComponentId,
     tx: &mut Transaction<RT>,
 ) -> anyhow::Result<Result<Version, JsError>> {
-    let udf_config = UdfConfigModel::new(tx, TableNamespace::TODO())
-        .get()
-        .await?;
+    let udf_config = UdfConfigModel::new(tx, component.into()).get().await?;
 
     let udf_version = match udf_config {
         Some(udf_config) if udf_config.server_version > DEPRECATION_THRESHOLD.npm.unsupported => {
@@ -259,6 +259,10 @@ impl ValidatedPathAndArgs {
             return Ok(result);
         }
 
+        let (_, component) = BootstrapComponentsModel::new(tx)
+            .component_path_to_ids(path.component.clone())
+            .await?;
+
         let mut backend_state_model = BackendStateModel::new(tx);
         let backend_state = backend_state_model.get_backend_state().await?;
         match backend_state {
@@ -273,7 +277,7 @@ impl ValidatedPathAndArgs {
             },
         }
 
-        let udf_version = match udf_version(&path, tx).await? {
+        let udf_version = match udf_version(&path, component, tx).await? {
             Ok(udf_version) => udf_version,
             Err(e) => return Ok(Err(e)),
         };
@@ -289,6 +293,7 @@ impl ValidatedPathAndArgs {
             allowed_visibility,
             tx,
             path,
+            component,
             args,
             expected_udf_type,
             analyzed_function,
@@ -338,7 +343,11 @@ impl ValidatedPathAndArgs {
             },
         }
 
-        let udf_version = match udf_version(&path, tx).await? {
+        let (_, component) = BootstrapComponentsModel::new(tx)
+            .component_path_to_ids(path.component.clone())
+            .await?;
+
+        let udf_version = match udf_version(&path, component, tx).await? {
             Ok(udf_version) => udf_version,
             Err(e) => return Ok(Err(e)),
         };
@@ -362,6 +371,7 @@ impl ValidatedPathAndArgs {
             allowed_visibility,
             tx,
             path,
+            component,
             args,
             expected_udf_type,
             analyzed_function,
@@ -378,6 +388,7 @@ impl ValidatedPathAndArgs {
         allowed_visibility: AllowedVisibility,
         tx: &mut Transaction<RT>,
         path: CanonicalizedComponentFunctionPath,
+        component: ComponentId,
         args: ConvexArray,
         expected_udf_type: UdfType,
         analyzed_function: AnalyzedFunction,
@@ -420,9 +431,7 @@ impl ValidatedPathAndArgs {
             Err(err) => return Ok(Err(err)),
         }
 
-        let table_mapping = &tx
-            .table_mapping()
-            .namespace(TableNamespace::by_component_TODO());
+        let table_mapping = &tx.table_mapping().namespace(component.into());
         let virtual_table_mapping = &tx.virtual_table_mapping().clone();
 
         // If the UDF has an args validator, check that these args match.
@@ -565,7 +574,10 @@ impl ValidatedHttpPath {
                 .fail_while_paused_or_disabled()
                 .await?;
         }
-        let udf_version = match udf_version(&path, tx).await? {
+        let (_, component) = BootstrapComponentsModel::new(tx)
+            .component_path_to_ids(path.component.clone())
+            .await?;
+        let udf_version = match udf_version(&path, component, tx).await? {
             Ok(udf_version) => udf_version,
             Err(e) => return Ok(Err(e)),
         };