Enhance catalog management, caching, and error handling in pChronicle

Cargo.toml

@@ -60,6 +60,7 @@ lance-core = "11.0.0"
 lance-file = "11.0.0"
 lance-datafusion = "11.0.0"
 lance-index = { version = "11.0.0", features = ["tokenizer-jieba"] }
+lance-io = "11.0.0"
 lance-linalg = "11.0.0"
 lance-table = "11.0.0"
 libc = "0.2"
@@ -75,6 +76,7 @@ opendal = { version = "0.57.0", default-features = true, features = [
     "services-s3",
     "services-tos",
 ] }
+object_store = "0.13.2"
 persisting-agentctl = { path = "crates/persisting-agentctl" }
 persisting-events = { path = "crates/persisting-events" }
 persisting-gateway = { path = "crates/persisting-gateway" }

crates/persisting-pchronicle-cli/src/gateway_partition.rs

@@ -1,7 +1,7 @@
 //! Safe, bounded physical partitioning below one logical Gateway Dataset.
 
 use std::collections::HashMap;
-use std::sync::Mutex;
+use std::sync::{Mutex, MutexGuard};
 
 use anyhow::{Context, Result};
 use chrono::{DateTime, Datelike, Timelike, Utc};
@@ -10,6 +10,12 @@ const MAX_TEMPLATE_BYTES: usize = 256;
 const MAX_TEMPLATE_SEGMENTS: usize = 16;
 const MAX_USER_SEGMENT_BYTES: usize = 80;
 
+fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
+    mutex
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 enum Segment {
     Literal(String),
@@ -149,7 +155,7 @@ impl GatewayPartitionRouter {
         let Some(split) = &self.split else {
             return self.dataset_uri.clone();
         };
-        let mut routes = self.routes.lock().unwrap();
+        let mut routes = lock_recover(&self.routes);
         routes
             .entry(route_key.to_string())
             .or_insert_with(|| {

crates/persisting-pchronicle-cli/src/lib.rs

@@ -121,7 +121,9 @@ impl Cli {
 /// invocations return None and retain their existing execution path.
 pub fn run_catalog_worker_before_runtime(cli: &Cli) -> Option<Result<()>> {
     match &cli.command {
-        Command::Serve(args) if args.catalog_query_worker => Some(server::catalog_worker::run()),
+        Command::Serve(args) if args.catalog_query_worker => {
+            Some(server::catalog_worker::run(cli.log_level))
+        }
         _ => None,
     }
 }
@@ -172,6 +174,18 @@ pub enum LogLevel {
     Debug,
 }
 
+impl LogLevel {
+    /// The `--log-level` value that parses back to this variant.
+    pub(crate) fn as_arg(self) -> &'static str {
+        match self {
+            Self::Error => "error",
+            Self::Warn => "warn",
+            Self::Info => "info",
+            Self::Debug => "debug",
+        }
+    }
+}
+
 struct DiagnosticWriter<'a> {
     level: LogLevel,
     inner: &'a mut dyn Write,
@@ -3589,51 +3603,63 @@ pub(crate) async fn find_expression_predicate_for_dataset(
     for predicate in text_predicates {
         let mut matches = Vec::<String>::new();
         let mut searched = false;
+        let mut jobs = Vec::new();
         for dataset in snapshot.datasets() {
             if dataset_filter.is_some_and(|filter| dataset.mount.name != filter) {
                 continue;
             }
             for source in &dataset.sources {
                 if source.status != CatalogSourceStatus::Ready
                     || source.kind != CatalogSourceKind::Store
-                    || source_filter.is_some_and(|filter| filter != source.file)
+                    || source_filter.is_some_and(|filter| {
+                        filter != source.file && !source.file.starts_with(&format!("{filter}/"))
+                    })
                 {
                     continue;
                 }
-                let Some(paths) =
+                if let Some(paths) =
                     snapshot.storyline_table_paths(&dataset.mount.name, &source.file)?
-                else {
-                    continue;
-                };
-                match search_storyline_step_matches_fts_in_columns(
-                    &paths,
-                    &predicate.query,
-                    predicate.field.columns(),
-                )
-                .await
                 {
-                    Ok(step_matches) => {
-                        available = true;
-                        searched = true;
-                        let source_predicate = predicate
-                            .field
-                            .source_predicate()
-                            .map(|value| format!(" AND ({value})"))
-                            .unwrap_or_default();
-                        matches.extend(step_matches.into_iter().map(|(document_id, step_id)| {
-                            format!(
-                                "(_file_ = {} AND document_id = {} AND step_id = {}{})",
-                                sql_string(&source.file),
-                                sql_string(&document_id),
-                                step_id,
-                                source_predicate,
-                            )
-                        }));
-                    }
-                    Err(error) => errors.push(format!(
-                        "FTS unavailable for {} / {}: {error:#}",
-                        dataset.mount.name, source.file
-                    )),
+                    jobs.push((dataset.mount.name.clone(), source.file.clone(), paths));
+                }
+            }
+        }
+        let query = predicate.query.clone();
+        let columns = predicate.field.columns();
+        let results = stream::iter(jobs)
+            .map(|(dataset, file, paths)| {
+                let query = query.clone();
+                async move {
+                    let result =
+                        search_storyline_step_matches_fts_in_columns(&paths, &query, columns).await;
+                    (dataset, file, result)
+                }
+            })
+            .buffer_unordered(4)
+            .collect::<Vec<_>>()
+            .await;
+        for (dataset, file, result) in results {
+            match result {
+                Ok(step_matches) => {
+                    available = true;
+                    searched = true;
+                    let source_predicate = predicate
+                        .field
+                        .source_predicate()
+                        .map(|value| format!(" AND ({value})"))
+                        .unwrap_or_default();
+                    matches.extend(step_matches.into_iter().map(|(document_id, step_id)| {
+                        format!(
+                            "(_file_ = {} AND document_id = {} AND step_id = {}{})",
+                            sql_string(&file),
+                            sql_string(&document_id),
+                            step_id,
+                            source_predicate,
+                        )
+                    }));
+                }
+                Err(error) => {
+                    errors.push(format!("FTS unavailable for {dataset} / {file}: {error:#}"))
                 }
             }
         }