perf: cache and validate small Parquet files to cut object-store round trips per query

.github/workflows/integration-test.yaml

@@ -67,6 +67,13 @@ jobs:
       PARSEABLE_TEST_IMAGE: parseable-ci:${{ github.sha }}
       COMPOSE_FILE: ${{ matrix.compose }}
       QUEST_MODE: ${{ inputs.full && 'load' || 'smoke' }}
+      # Quest is built from a fixed source commit instead of pulling the
+      # floating quay.io/parseablehq/quest:main image. Later Quest commits
+      # expect 202 from stream deletion, which this server does not return yet.
+      # Remove the pin when issue #151 is done.
+      QUEST_SOURCE_REF: f4a026eecde1badde8776ff366a743f9bbca2e67
+      QUEST_IMAGE: quest-ci:pinned
+      QUEST_PULL_POLICY: never
     steps:
       - uses: actions/checkout@v4
       - uses: actions/download-artifact@v4
@@ -84,6 +91,14 @@ jobs:
         run: |
           docker compose up --no-build --wait --wait-timeout 180 "$SERVER"
           docker compose exec -T "$SERVER" sh -ec 'curl --fail --silent --show-error --location http://localhost:8000/next/ | grep -q "Parseable · Observability"'
+      - name: Build Quest from the pinned source commit
+        run: |
+          src="$RUNNER_TEMP/quest-src"
+          git init --quiet "$src"
+          git -C "$src" fetch --quiet --depth 1 https://github.com/parseablehq/quest.git "$QUEST_SOURCE_REF"
+          git -C "$src" checkout --quiet FETCH_HEAD
+          test "$(git -C "$src" rev-parse HEAD)" = "$QUEST_SOURCE_REF"
+          docker build --tag "$QUEST_IMAGE" "$src"
       - name: Run Quest
         run: docker compose up --no-build --exit-code-from quest
       - name: Capture service logs

docker-compose-distributed-test-with-kafka.yaml

@@ -194,8 +194,8 @@ services:
 
   quest:
     platform: linux/amd64
-    image: quay.io/parseablehq/quest:main
-    pull_policy: always
+    image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main}
+    pull_policy: ${QUEST_PULL_POLICY:-always}
     command:
       [
         "load",

docker-compose-distributed-test.yaml

@@ -109,8 +109,8 @@ services:
 
   quest:
     platform: linux/amd64
-    image: quay.io/parseablehq/quest:main
-    pull_policy: always
+    image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main}
+    pull_policy: ${QUEST_PULL_POLICY:-always}
     environment:
       - QUEST_EDITION=oss
     command:

docker-compose-gcs-distributed-test.yaml

@@ -79,8 +79,8 @@ services:
 
   quest:
     platform: linux/amd64
-    image: quay.io/parseablehq/quest:main
-    pull_policy: always
+    image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main}
+    pull_policy: ${QUEST_PULL_POLICY:-always}
     command:
       [
         "load",

docker-compose-test-with-kafka.yaml

@@ -68,9 +68,9 @@ services:
         max_attempts: 3
 
   quest:
-    image: quay.io/parseablehq/quest:main
+    image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main}
     platform: linux/amd64
-    pull_policy: always
+    pull_policy: ${QUEST_PULL_POLICY:-always}
     command: [
       "load",
       "http://parseable:8000",

docker-compose-test.yaml

@@ -64,9 +64,9 @@ services:
         max_attempts: 3
 
   quest:
-    image: quay.io/parseablehq/quest:main
+    image: ${QUEST_IMAGE:-quay.io/parseablehq/quest:main}
     platform: linux/amd64
-    pull_policy: always
+    pull_policy: ${QUEST_PULL_POLICY:-always}
     environment:
       - QUEST_EDITION=oss
     command: [

docs/log-severity-clustering.md

@@ -83,8 +83,8 @@ of the base commit and this change with the same toolchain and lockfile. The
 server was pinned to four CPUs with `s3-store` on a local MinIO. Each arm ingested
 the same 3.6 million JSON rows (about 1% `ERROR`, 22 fields), event-time
 partitioned, with default 262,144-row groups. Values are medians over three
-rotated blocks. Bytes are the server's logical requested ranges
-(`parseable_storage_read_bytes_total`), not physical GETs. The run used
+rotated blocks. Bytes are successful response body bytes returned by storage
+(`parseable_storage_read_bytes_total`), not physical GET counts. The run used
 `scripts/ingest-performance/clustering_bench.py`.
 
 | Measure | Baseline | Clustering off | Clustering on |

scripts/performance/request_cache_repro.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+"""Run against the isolated fixture/server described in validated-small-file-cache.md.
+Set CACHE_REPRO_ASSERT_ONCE=1 after the fix. Every shape runs twice, with a fresh
+HTTP request, and measures the second request. No third-party Python packages.
+"""
+import base64, collections, datetime, json, os, re, time, urllib.request, urllib.parse
+BASE = os.environ.get('CACHE_REPRO_URL', 'http://127.0.0.1:18398')
+LOG = os.environ.get('CACHE_REPRO_LOG', 'target/cache-repro/server.log')
+HEADERS = {'Authorization': 'Basic ' + base64.b64encode(b'admin:admin').decode(), 'X-P-Stream': 'repro'}
+def call(path, body=None):
+    headers = dict(HEADERS)
+    if body is not None:
+        headers['Content-Type'] = 'application/json'
+        body = json.dumps(body).encode()
+    try:
+        with urllib.request.urlopen(urllib.request.Request(BASE + path, data=body, headers=headers), timeout=180) as response:
+            return response.read().decode()
+    except urllib.error.HTTPError as error:
+        raise RuntimeError(error.read().decode()) from error
+def metrics():
+    return sum(float(line.rsplit(' ', 1)[1]) for line in call('/api/v1/metrics').splitlines() if line.startswith('parseable_storage_request_response_time_count{') and '"GET"' in line)
+def stamp(t):
+    return datetime.datetime.fromtimestamp(t, datetime.timezone.utc).isoformat()
+def query(sql, minutes):
+    now = time.time()
+    return '/api/v1/query', {'query': sql, 'startTime': stamp(now - minutes*60), 'endTime': stamp(now)}
+# Keep the recent staging/remote union in every reproduction.
+call('/v1/metrics', {'resourceMetrics': [{'resource': {'attributes': [{'key':'host','value':{'stringValue':'host-0'}}]}, 'scopeMetrics': [{'metrics': [{'name':'cpu','gauge':{'dataPoints':[{'timeUnixNano':str(time.time_ns()),'asDouble':0.5}]}}]}]}]})
+now = time.time()
+shapes = [
+    ('count_10m', *query('SELECT COUNT(*) FROM repro', 10)),
+    ('count_1h', *query('SELECT COUNT(*) FROM repro', 60)),
+    ('group_1h', *query("SELECT host, SUM(data_point_value) FROM repro WHERE metric_name = 'cpu' GROUP BY host", 60)),
+    ('promql_range_1h', '/prometheus/api/v1/query_range?' + urllib.parse.urlencode({'query':'cpu','start':now-3600,'end':now,'step':60}), None),
+    ('label_values_1h', '/prometheus/api/v1/label/host/values?' + urllib.parse.urlencode({'match[]':'cpu','start':now-3600,'end':now}), None),
+    ('count_24h', *query('SELECT COUNT(*) FROM repro',1440)),
+    ('promql_instant','/prometheus/api/v1/query?' + urllib.parse.urlencode({'query':'cpu','time':now}),None),
+]
+request_file = os.environ.get('CACHE_REPRO_REQUESTS')
+if request_file:
+    if os.path.exists(request_file):
+        with open(request_file) as saved:
+            shapes = json.load(saved)
+    else:
+        with open(request_file, 'w') as saved:
+            json.dump(shapes, saved)
+for name, path, body in shapes:
+    for run in range(2):
+        before = metrics()
+        offset = os.path.getsize(LOG)
+        result = call(path, body)
+        after = metrics()
+        with open(LOG) as trace:
+            trace.seek(offset)
+            lines = re.sub(r'\x1b\[[0-9;]*m', '', trace.read()).splitlines()
+            loads = collections.Counter(path for line in lines if 'CACHE_LOAD' in line for path in re.findall(r'\bpath=(\S+)', line))
+        if run:
+            print(json.dumps({'shape':name,'calls':int(after-before),'selected_files':len(loads),'validations':sum(loads.values()),'multiplicity':dict(collections.Counter(loads.values()))}), flush=True)
+        if os.environ.get('CACHE_REPRO_ASSERT_ONCE'):
+            if name in ('count_1h', 'group_1h', 'promql_range_1h', 'label_values_1h'):
+                assert len(loads) > 1, (name, 'fixture must select multiple remote files')
+            assert all(n == 1 for n in loads.values()), (name, loads)

src/cli.rs

@@ -496,6 +496,17 @@ pub struct Options {
     )]
     pub parquet_metadata_cache_size: usize,
 
+    #[arg(long, help = "Validate and cache small remote Parquet bodies", env = "P_SMALL_FILE_CACHE_ENABLED", default_value = "true", action = clap::ArgAction::Set)]
+    pub small_file_cache_enabled: bool,
+
+    #[arg(
+        long,
+        help = "Maximum resident small-file body bytes (zero disables caching)",
+        env = "P_SMALL_FILE_CACHE_SIZE",
+        default_value = "67108864"
+    )]
+    pub small_file_cache_size: usize,
+
     // reduced the max row group size from 1048576
     // smaller row groups help in faster query performance in multi threaded query
     #[arg(

src/metastore/metastores/object_store_metastore.rs

@@ -1552,7 +1552,6 @@ impl Metastore for ObjectStoreMetastore {
                 .map_err(MetastoreError::ObjectStorageError)
         } else {
             // not local-disk, object storage
-            let mut result_file_list = HashSet::new();
             let root = tenant_id
                 .as_ref()
                 .map(|tenant| object_store::path::Path::from_iter([tenant.clone()]));
@@ -1572,35 +1571,164 @@ impl Metastore for ObjectStoreMetastore {
                         && name != TOMBSTONE_ROOT_DIRECTORY
                 })
                 .collect::<Vec<_>>();
-            for stream in streams {
-                let stream_path = if let Some(root) = root.as_ref() {
-                    object_store::path::Path::from_iter([
-                        root.as_ref(),
-                        &stream,
-                        STREAM_ROOT_DIRECTORY,
-                    ])
-                } else {
-                    object_store::path::Path::from(format!("{}/{}", stream, STREAM_ROOT_DIRECTORY))
-                };
-                let resp = self.storage.list_with_delimiter(Some(stream_path)).await?;
-                if resp
-                    .objects
-                    .iter()
-                    .any(|name| name.location.filename().unwrap().ends_with("stream.json"))
-                {
-                    result_file_list.insert(stream);
+            let results = collect_dataset_probes(streams.into_iter().map(|stream| {
+                let root = root.clone();
+                async move {
+                    let stream_path = if let Some(root) = root.as_ref() {
+                        object_store::path::Path::from_iter([
+                            root.as_ref(),
+                            &stream,
+                            STREAM_ROOT_DIRECTORY,
+                        ])
+                    } else {
+                        object_store::path::Path::from(format!(
+                            "{}/{}",
+                            stream, STREAM_ROOT_DIRECTORY
+                        ))
+                    };
+                    let resp = self.storage.list_with_delimiter(Some(stream_path)).await?;
+                    if resp
+                        .objects
+                        .iter()
+                        .any(|name| name.location.filename().unwrap().ends_with("stream.json"))
+                    {
+                        Ok::<_, MetastoreError>(Some(stream))
+                    } else {
+                        Ok(None)
+                    }
                 }
-            }
-            Ok(result_file_list)
+            }))
+            .await?;
+            Ok(results.into_iter().flatten().collect())
         }
     }
 }
 
+// All-or-error collection: a failed probe cannot publish a partial inventory.
+async fn collect_dataset_probes<T, E>(
+    probes: impl IntoIterator<Item = impl std::future::Future<Output = Result<T, E>>>,
+) -> Result<Vec<T>, E> {
+    futures::stream::iter(probes)
+        .buffer_unordered(16)
+        .try_collect()
+        .await
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::storage::LocalFS;
 
+    #[tokio::test]
+    async fn dataset_probes_are_bounded_and_fail_atomically() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        for fail in [false, true] {
+            let active = AtomicUsize::new(0);
+            let peak = AtomicUsize::new(0);
+            let result = collect_dataset_probes((0..64).map(|i| {
+                let active = &active;
+                let peak = &peak;
+                async move {
+                    let n = active.fetch_add(1, Ordering::SeqCst) + 1;
+                    peak.fetch_max(n, Ordering::SeqCst);
+                    tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+                    active.fetch_sub(1, Ordering::SeqCst);
+                    if fail && i == 25 {
+                        Err("failed probe")
+                    } else {
+                        Ok(i)
+                    }
+                }
+            }))
+            .await;
+            assert_eq!(peak.load(Ordering::SeqCst), 16);
+            if fail {
+                assert_eq!(result, Err("failed probe"));
+            } else {
+                assert_eq!(result.unwrap().len(), 64);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn next_metadata_read_sees_external_writer_changes() {
+        for tenant in [None, Some("tenant".to_owned())] {
+            let dir = tempfile::tempdir_in(".").unwrap();
+            let storage = Arc::new(LocalFS::new(dir.path().to_owned()));
+            let reader = ObjectStoreMetastore {
+                storage: storage.clone(),
+            };
+            let stream_path = RelativePathBuf::from_iter([
+                tenant.as_deref().unwrap_or(""),
+                "metrics",
+                STREAM_ROOT_DIRECTORY,
+                STREAM_METADATA_FILE_NAME,
+            ]);
+            let manifest_path = RelativePathBuf::from_iter([
+                tenant.as_deref().unwrap_or(""),
+                "metrics",
+                "day",
+                "manifest.json",
+            ]);
+            for generation in ["first", "external-update"] {
+                // Write directly through storage, bypassing this reader's API.
+                let format = crate::storage::ObjectStoreFormat {
+                    created_at: generation.into(),
+                    ..Default::default()
+                };
+                storage
+                    .put_object(&stream_path, to_bytes(&format), &tenant)
+                    .await
+                    .unwrap();
+                let loaded: crate::storage::ObjectStoreFormat = serde_json::from_slice(
+                    &reader
+                        .get_stream_json("metrics", true, &tenant, false)
+                        .await
+                        .unwrap(),
+                )
+                .unwrap();
+                assert_eq!(loaded.created_at, generation);
+                let manifest = Manifest {
+                    version: generation.into(),
+                    ..Default::default()
+                };
+                storage
+                    .put_object(&manifest_path, encode_manifest(&manifest).unwrap(), &tenant)
+                    .await
+                    .unwrap();
+                let loaded = reader
+                    .get_manifest(
+                        "metrics",
+                        Utc::now(),
+                        Utc::now(),
+                        Some(manifest_path.to_string()),
+                        &tenant,
+                    )
+                    .await
+                    .unwrap()
+                    .unwrap();
+                assert_eq!(loaded.version, generation);
+            }
+            storage
+                .delete_object(&manifest_path, &tenant)
+                .await
+                .unwrap();
+            assert!(
+                reader
+                    .get_manifest(
+                        "metrics",
+                        Utc::now(),
+                        Utc::now(),
+                        Some(manifest_path.to_string()),
+                        &tenant
+                    )
+                    .await
+                    .unwrap()
+                    .is_none()
+            );
+        }
+    }
+
     #[tokio::test]
     async fn remote_schema_refresh_publishes_only_changes() {
         use crate::{event::commit_schema_to_metadata, metadata::LogStreamMetadata};

src/metrics/mod.rs

@@ -631,7 +631,7 @@ pub static STORAGE_READ_BYTES_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
     IntCounterVec::new(
         Opts::new(
             "storage_read_bytes_total",
-            "Logical bytes requested from an object store",
+            "Body bytes returned successfully by an object store",
         )
         .namespace(METRICS_NAMESPACE),
         &["provider"],

src/parseable/streams.rs

@@ -1956,7 +1956,7 @@ impl Stream {
     #[inline(always)]
     fn conversion_phase(&self, _phase: ConversionPhase<'_>) {}
 
-    fn parquet_writer_props(
+    pub(crate) fn parquet_writer_props(
         &self,
         merged_schema: &Schema,
         time_partition: Option<&String>,