feat: port from enterprise to core (#26991)

Author: philjb

Cargo.toml

@@ -231,6 +231,7 @@ inherits = "release"
 codegen-units = 16
 lto = false
 incremental = true
+debug = "line-tables-only"
 
 # This profile extends the `quick-release` profile with debuginfo turned on in order to
 # produce more human friendly symbols for profiling tools

influxdb3/src/commands/serve.rs

@@ -147,6 +147,13 @@ pub enum Error {
 
     #[error("Must set INFLUXDB3_NODE_IDENTIFIER_PREFIX to a valid string value")]
     NodeIdEnvVarMissing,
+
+    #[error(
+        "Python environment initialization failed: {0}\nPlease ensure Python and either pip or uv package manager is installed"
+    )]
+    PythonEnvironmentInitialization(
+        #[source] influxdb3_processing_engine::environment::PluginEnvironmentError,
+    ),
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -1104,7 +1111,8 @@ pub async fn command(config: Config, user_params: HashMap<String, String>) -> Re
         Arc::clone(&time_provider) as _,
         sys_events_store,
     )
-    .await;
+    .await
+    .map_err(Error::PythonEnvironmentInitialization)?;
 
     // Update query executor with processing engine reference
     query_executor.set_processing_engine(Arc::clone(&processing_engine));

influxdb3_authz/src/lib.rs

@@ -36,10 +36,13 @@ pub enum AccessRequest {
     Admin,
 }
 
-#[derive(Debug, Clone, Copy, thiserror::Error)]
+#[derive(Debug, Clone, thiserror::Error)]
 pub enum ResourceAuthorizationError {
     #[error("unauthorized to perform requested action with the token")]
     Unauthorized,
+
+    #[error("resource type not supported, {0}")]
+    ResourceNotSupported(String),
 }
 
 #[derive(Debug, thiserror::Error)]

influxdb3_cache/src/distinct_cache/mod.rs

@@ -3,7 +3,7 @@
 mod cache;
 pub use cache::{CacheError, CreateDistinctCacheArgs};
 mod provider;
-pub use provider::{DistinctCacheProvider, ProviderError};
+pub use provider::{DistinctCacheProvider, ProviderError, background_catalog_update};
 mod table_function;
 pub use table_function::DISTINCT_CACHE_UDTF_NAME;
 pub use table_function::DistinctCacheFunction;

influxdb3_cache/src/distinct_cache/provider.rs

@@ -271,7 +271,9 @@ impl DistinctCacheProvider {
     }
 }
 
-fn background_catalog_update(
+/// Background task that listens for catalog updates and maintains distinct caches.
+/// This should be started explicitly after creating the provider.
+pub fn background_catalog_update(
     provider: Arc<DistinctCacheProvider>,
     mut subscription: CatalogUpdateReceiver,
 ) -> tokio::task::JoinHandle<()> {

influxdb3_cache/src/last_cache/mod.rs

@@ -6,7 +6,7 @@ mod cache;
 pub use cache::CreateLastCacheArgs;
 mod metrics;
 mod provider;
-pub use provider::LastCacheProvider;
+pub use provider::{LastCacheProvider, background_catalog_update};
 mod table_function;
 use schema::InfluxColumnType;
 pub use table_function::{LAST_CACHE_UDTF_NAME, LastCacheFunction};

influxdb3_cache/src/last_cache/provider.rs

@@ -354,7 +354,9 @@ impl LastCacheProvider {
     }
 }
 
-fn background_catalog_update(
+/// Background task that listens for catalog updates and maintains last value caches.
+/// This should be started explicitly after creating the provider.
+pub fn background_catalog_update(
     provider: Arc<LastCacheProvider>,
     mut subscription: CatalogUpdateReceiver,
 ) -> tokio::task::JoinHandle<()> {

influxdb3_catalog/src/catalog.rs

@@ -219,7 +219,7 @@ impl<I: CatalogId, R: CatalogResource> Repository<I, R> {
     pub(crate) fn update(&mut self, id: I, resource: impl Into<Arc<R>>) -> Result<()> {
         let resource = resource.into();
         if !self.id_exists(&id) {
-            return Err(CatalogError::NotFound);
+            return Err(CatalogError::NotFound(format!("catalog id: {}", id)));
         }
         self.id_name_map.insert(id, resource.name());
         self.repo.insert(id, resource);
@@ -383,7 +383,7 @@ impl TokenRepository {
         let token_id = self
             .repo
             .name_to_id(&token_name)
-            .ok_or_else(|| CatalogError::NotFound)?;
+            .ok_or_else(|| CatalogError::NotFound(token_name))?;
         self.repo.remove(&token_id);
         self.hash_lookup_map.remove_by_left(&token_id);
         Ok(())

influxdb3_catalog/src/catalog/versions/v1/update.rs

@@ -211,7 +211,7 @@ impl Catalog {
         self.catalog_update_with_retry(|| {
             let time_ns = self.time_provider.now().timestamp_nanos();
             let Some(node) = self.node(node_id) else {
-                return Err(crate::CatalogError::NotFound);
+                return Err(crate::CatalogError::NotFound(node_id.to_string()));
             };
             if !node.is_running() {
                 return Err(crate::CatalogError::NodeAlreadyStopped {
@@ -259,16 +259,16 @@ impl Catalog {
 
     pub async fn soft_delete_database(
         &self,
-        name: &str,
+        db_name: &str,
         hard_delete_time: HardDeletionTime,
     ) -> Result<OrderedCatalogBatch> {
         self.catalog_update_with_retry(|| {
-            if name == INTERNAL_DB_NAME {
+            if db_name == INTERNAL_DB_NAME {
                 return Err(CatalogError::CannotDeleteInternalDatabase);
             };
 
-            let Some(db) = self.db_schema(name) else {
-                return Err(CatalogError::NotFound);
+            let Some(db) = self.db_schema(db_name) else {
+                return Err(CatalogError::NotFound(db_name.to_string()));
             };
 
             // If the request specifies the default hard-delete time, and the schema has an existing hard_delete_time,
@@ -338,10 +338,10 @@ impl Catalog {
     ) -> Result<OrderedCatalogBatch> {
         self.catalog_update_with_retry(|| {
             let Some(db) = self.db_schema(db_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(db_name.to_string()));
             };
             let Some(tbl_def) = db.table_definition(table_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(table_name.to_string()));
             };
 
             // If the request specifies the default hard-delete time, and the schema has an existing hard_delete_time,
@@ -402,10 +402,10 @@ impl Catalog {
         info!(?db_id, ?table_id, "Hard delete table.");
         self.catalog_update_with_retry(|| {
             let Some(db) = self.db_schema_by_id(db_id) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(format!("database id: {}", db_id)));
             };
             let Some(_table_def) = db.table_definition_by_id(table_id) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(format!("table id: {}", table_id)));
             };
 
             let deletion_time = self.time_provider.now().timestamp_nanos();
@@ -431,7 +431,7 @@ impl Catalog {
         info!(?db_id, "Hard delete database.");
         self.catalog_update_with_retry(|| {
             let Some(db) = self.db_schema_by_id(db_id) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(format!("database id: {}", db_id)));
             };
 
             // Prevent deletion of internal database
@@ -461,10 +461,10 @@ impl Catalog {
         info!(db_name, table_name, cache_name = ?cache_name, "create distinct cache");
         self.catalog_update_with_retry(|| {
             let Some(db) = self.db_schema(db_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(db_name.to_string()));
             };
             let Some(mut tbl) = db.table_definition(table_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(table_name.to_string()));
             };
             if columns.is_empty() {
                 return Err(CatalogError::invalid_configuration(
@@ -546,10 +546,10 @@ impl Catalog {
         info!(db_name, table_name, cache_name = ?cache_name, "create last cache");
         self.catalog_update_with_retry(|| {
             let Some(db) = self.db_schema(db_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(db_name.to_string()));
             };
             let Some(mut tbl) = db.table_definition(table_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(table_name.to_string()));
             };
 
             fn is_valid_last_cache_key_col(def: &ColumnDefinition) -> bool {
@@ -667,7 +667,7 @@ impl Catalog {
         info!(db_name, trigger_name, "create processing engine trigger");
         self.catalog_update_with_retry(|| {
             let Some(mut db) = self.db_schema(db_name) else {
-                return Err(CatalogError::NotFound);
+                return Err(CatalogError::NotFound(db_name.to_string()));
             };
             let trigger = TriggerSpecificationDefinition::from_string_rep(trigger_specification)?;
             if db.processing_engine_triggers.contains_name(trigger_name) {
@@ -709,7 +709,7 @@ impl Catalog {
             "create new retention policy"
         );
         let Some(db) = self.db_schema(db_name) else {
-            return Err(CatalogError::NotFound);
+            return Err(CatalogError::NotFound(db_name.to_string()));
         };
         self.catalog_update_with_retry(|| {
             Ok(CatalogBatch::database(
@@ -814,7 +814,7 @@ impl DatabaseCatalogTransaction {
         column_type: FieldDataType,
     ) -> Result<ColumnId> {
         let Some(table_def) = self.database_schema.table_definition(table_name) else {
-            return Err(CatalogError::NotFound);
+            return Err(CatalogError::NotFound(table_name.to_string()));
         };
         match table_def.column_definition(column_name) {
             Some(def) if def.data_type == column_type.into() => Ok(def.id),

influxdb3_catalog/src/catalog/versions/v2.rs

@@ -3959,13 +3959,13 @@ mod tests {
         let fake_db_id = DbId::from(999);
         let fake_table_id = TableId::from(123);
         let result = catalog.hard_delete_table(&fake_db_id, &fake_table_id).await;
-        assert!(matches!(result, Err(CatalogError::NotFound)));
+        assert!(matches!(result, Err(CatalogError::NotFound(_))));
 
         // Create database but try to delete non-existent table
         catalog.create_database("test").await.unwrap();
         let db_id = catalog.db_name_to_id("test").unwrap();
         let result = catalog.hard_delete_table(&db_id, &fake_table_id).await;
-        assert!(matches!(result, Err(CatalogError::NotFound)));
+        assert!(matches!(result, Err(CatalogError::NotFound(_))));
     }
 
     #[test_log::test(tokio::test)]
@@ -4149,7 +4149,7 @@ mod tests {
         // Try to delete non-existent database
         let fake_db_id = DbId::from(999);
         let result = catalog.hard_delete_database(&fake_db_id).await;
-        assert!(matches!(result, Err(CatalogError::NotFound)));
+        assert!(matches!(result, Err(CatalogError::NotFound(_))));
     }
 
     #[test_log::test(tokio::test)]