From 93cefb2046c2f042fc2fad5e2ab910359ebcea68 Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Sun, 21 Jun 2026 01:17:27 +0200 Subject: [PATCH 1/2] Move data dirs into StorageConfig; make S3 config optional StorageConfig now owns the local storage layout (data_dir/datasets_dir/ tables_dir/tmp_dir), so ObjectStores::new(&StorageConfig) and the iceberg warehouse take a single config instead of loose path args re-derived at each call site. This removes the multi-source temp-dir coupling structurally. S3 is now Option: presence is the backend switch (None = local, Some = S3), replacing the separate data_lake bool. Within S3Config the bucket is a required String (object_store's AmazonS3Builder requires it regardless of addressing style and never infers it from the endpoint); Config::load rejects an empty bucket when S3 is enabled. endpoint/region stay optional. beacon-config keeps indexes/cache in DataDirsConfig; the storage dirs moved to config.storage. Updated runtime, iceberg, data-lake, and the runtime_config test accordingly. --- beacon-config/src/error.rs | 5 + beacon-config/src/lib.rs | 59 +++++---- beacon-core/src/runtime.rs | 10 +- beacon-core/tests/runtime_config.rs | 16 +-- beacon-data-lake/src/lib.rs | 2 +- beacon-iceberg/src/catalog.rs | 13 +- beacon-object-storage/src/config.rs | 83 ++++++------- beacon-object-storage/src/datasets_store.rs | 128 +++++++------------- beacon-object-storage/src/lib.rs | 20 ++- 9 files changed, 148 insertions(+), 188 deletions(-) diff --git a/beacon-config/src/error.rs b/beacon-config/src/error.rs index 3db3a4b0..c2ab9ecc 100644 --- a/beacon-config/src/error.rs +++ b/beacon-config/src/error.rs @@ -20,6 +20,11 @@ pub enum ConfigError { #[error("invalid BEACON_BASE_PATH: {0}")] InvalidBasePath(String), + /// Storage settings are internally inconsistent (e.g. S3 enabled without a + /// bucket name). + #[error("invalid storage configuration: {0}")] + InvalidStorage(String), + /// A required data directory could not be created. #[error("failed to create directory {}: {source}", .path.display())] CreateDir { diff --git a/beacon-config/src/lib.rs b/beacon-config/src/lib.rs index f93bc67f..fb2c18e1 100644 --- a/beacon-config/src/lib.rs +++ b/beacon-config/src/lib.rs @@ -114,10 +114,6 @@ pub struct ApiDocsConfig { /// `./data`). The directories are created when the config is loaded. #[derive(Debug, Clone)] pub struct DataDirsConfig { - pub root: PathBuf, - pub datasets: PathBuf, - pub tables: PathBuf, - pub tmp: PathBuf, pub indexes: PathBuf, pub cache: PathBuf, } @@ -311,17 +307,29 @@ impl From for Config { statement_ttl_secs: raw.flight_sql_statement_ttl_secs, prepared_statement_ttl_secs: raw.flight_sql_prepared_statement_ttl_secs, }, - storage: StorageConfig { - enable_fs_events: raw.enable_fs_events, - enable_s3_events: raw.enable_s3_events, - s3: S3Config { - bucket: raw.s3_bucket, - enable_virtual_hosting: raw.s3_enable_virtual_hosting, - data_lake: raw.s3_data_lake, - endpoint: raw.aws_endpoint, - region: raw.aws_region, - allow_http: raw.s3_allow_http, - }, + storage: { + let root = PathBuf::from(&raw.data_dir); + // S3 presence *is* the backend switch: `Some` => datasets on S3. + let s3 = if raw.s3_data_lake { + Some(S3Config { + bucket: raw.s3_bucket.unwrap_or_default(), + endpoint: raw.aws_endpoint, + region: raw.aws_region, + enable_virtual_hosting: raw.s3_enable_virtual_hosting, + allow_http: raw.s3_allow_http, + }) + } else { + None + }; + StorageConfig { + datasets_dir: root.join("datasets"), + tables_dir: root.join("tables"), + tmp_dir: root.join("tmp"), + data_dir: root, + enable_fs_events: raw.enable_fs_events, + enable_s3_events: raw.enable_s3_events, + s3, + } }, cors: CorsConfig { allowed_methods: raw.allowed_methods, @@ -356,12 +364,8 @@ impl From for Config { data: { let root = PathBuf::from(&raw.data_dir); DataDirsConfig { - datasets: root.join("datasets"), - tables: root.join("tables"), - tmp: root.join("tmp"), indexes: root.join("indexes"), cache: root.join("cache"), - root, } }, } @@ -405,12 +409,21 @@ impl Config { .into(); config.server.base_path = normalize_base_path(&config.server.base_path).map_err(ConfigError::InvalidBasePath)?; + // S3 always needs a bucket (object_store requires it; it is never inferred + // from the endpoint). + if let Some(s3) = &config.storage.s3 { + if s3.bucket.trim().is_empty() { + return Err(ConfigError::InvalidStorage( + "BEACON_S3_BUCKET is required when BEACON_S3_DATA_LAKE=true".to_string(), + )); + } + } // Create the configured data directories (idempotent). for dir in [ - &config.data.root, - &config.data.datasets, - &config.data.tables, - &config.data.tmp, + &config.storage.data_dir, + &config.storage.datasets_dir, + &config.storage.tables_dir, + &config.storage.tmp_dir, &config.data.indexes, &config.data.cache, ] { diff --git a/beacon-core/src/runtime.rs b/beacon-core/src/runtime.rs index f5fb9f0e..964f0701 100644 --- a/beacon-core/src/runtime.rs +++ b/beacon-core/src/runtime.rs @@ -54,13 +54,8 @@ impl Runtime { config.runtime.vm_memory_size * 1024 * 1024, )); - let object_stores = beacon_object_storage::ObjectStores::new( - &config.storage, - config.data.datasets.clone(), - config.data.tables.clone(), - config.data.tmp.clone(), - ) - .await?; + let object_stores = + beacon_object_storage::ObjectStores::new(&config.storage).await?; let session_ctx = Self::init_ctx(memory_pool, object_stores.datasets.clone(), config.clone())?; @@ -104,7 +99,6 @@ impl Runtime { beacon_iceberg::catalog::init_datasets_warehouse( object_stores.datasets.clone(), &config.storage, - &config.data.datasets, ) .await?; diff --git a/beacon-core/tests/runtime_config.rs b/beacon-core/tests/runtime_config.rs index 9c6f9e00..fd191fd3 100644 --- a/beacon-core/tests/runtime_config.rs +++ b/beacon-core/tests/runtime_config.rs @@ -14,17 +14,17 @@ fn config_with(default_table: &str, tag: &str) -> Arc { config.sql.default_table = default_table.to_string(); let root = std::env::temp_dir().join(format!("beacon-runtime-config-test-{tag}")); - config.data.datasets = root.join("datasets"); - config.data.tables = root.join("tables"); - config.data.tmp = root.join("tmp"); + config.storage.datasets_dir = root.join("datasets"); + config.storage.tables_dir = root.join("tables"); + config.storage.tmp_dir = root.join("tmp"); config.data.indexes = root.join("indexes"); config.data.cache = root.join("cache"); - config.data.root = root; + config.storage.data_dir = root; for dir in [ - &config.data.root, - &config.data.datasets, - &config.data.tables, - &config.data.tmp, + &config.storage.data_dir, + &config.storage.datasets_dir, + &config.storage.tables_dir, + &config.storage.tmp_dir, &config.data.indexes, &config.data.cache, ] { diff --git a/beacon-data-lake/src/lib.rs b/beacon-data-lake/src/lib.rs index 2f3d3f55..873b4a4d 100644 --- a/beacon-data-lake/src/lib.rs +++ b/beacon-data-lake/src/lib.rs @@ -165,7 +165,7 @@ impl DataLake { session_context, datasets_object_store_url.clone(), file_formats, - config.data.tmp.clone(), + config.storage.tmp_dir.clone(), )); Self { diff --git a/beacon-iceberg/src/catalog.rs b/beacon-iceberg/src/catalog.rs index fdb33a31..26ed6d30 100644 --- a/beacon-iceberg/src/catalog.rs +++ b/beacon-iceberg/src/catalog.rs @@ -53,10 +53,9 @@ pub fn beacon_namespace() -> Vec { pub async fn init_datasets_warehouse( datasets: std::sync::Arc, storage: &beacon_config::StorageConfig, - datasets_dir: &std::path::Path, ) -> anyhow::Result<()> { tracing::info!( - backend = if storage.s3.data_lake { "s3" } else { "local" }, + backend = if storage.s3.is_some() { "s3" } else { "local" }, "initializing Iceberg datasets warehouse" ); // Full warehouse prefix within the backing store, e.g. `__beacon__/iceberg`. @@ -65,15 +64,11 @@ pub async fn init_datasets_warehouse( // The file catalog needs an `ObjectStoreBuilder` (it cannot accept an // arbitrary `ObjectStore`), so mirror the datasets store's backend choice. // `catalog_path` is the warehouse root the catalog roots every table under. - let (object_store_builder, catalog_path) = if storage.s3.data_lake { - let bucket = storage.s3.bucket.as_deref().ok_or_else(|| { - anyhow::anyhow!("Iceberg warehouse on S3 requires a bucket name (set BEACON_S3_BUCKET)") - })?; + let (object_store_builder, catalog_path) = if let Some(s3) = &storage.s3 { // Mirror the datasets store's backend by consuming the same `S3Config` // values (credentials still come from the AWS env chain). Setting the // endpoint/region explicitly keeps the Iceberg warehouse on the same // backend as the datasets without re-reading the environment. - let s3 = &storage.s3; let mut builder = ObjectStoreBuilder::s3() .with_config("aws_allow_http", if s3.allow_http { "true" } else { "false" }) .and_then(|builder| { @@ -97,11 +92,11 @@ pub async fn init_datasets_warehouse( .with_config("aws_region", region) .map_err(|error| anyhow::anyhow!("Failed to configure Iceberg S3 store: {error}"))?; } - (builder, format!("s3://{bucket}/{warehouse_prefix}")) + (builder, format!("s3://{}/{warehouse_prefix}", s3.bucket)) } else { // Local: root the builder at the datasets directory so warehouse paths // resolve to `/__beacon__/iceberg/...`. - let builder = ObjectStoreBuilder::filesystem(datasets_dir.to_path_buf()); + let builder = ObjectStoreBuilder::filesystem(storage.datasets_dir.to_path_buf()); (builder, warehouse_prefix.clone()) }; diff --git a/beacon-object-storage/src/config.rs b/beacon-object-storage/src/config.rs index 5f9c1fda..c4be9178 100644 --- a/beacon-object-storage/src/config.rs +++ b/beacon-object-storage/src/config.rs @@ -1,23 +1,38 @@ //! Storage configuration types. //! -//! These plain data structs describe how the datasets store is backed (local -//! filesystem vs. S3). They live in this crate — the storage layer — so it has -//! no dependency on `beacon-config`; `beacon-config` re-exports them and fills -//! them from the environment. +//! These plain data structs describe how Beacon's object storage is laid out on +//! local disk and, optionally, how the datasets store is backed by S3. They live +//! in this crate — the storage layer — so it has no dependency on `beacon-config`; +//! `beacon-config` re-exports them and fills them from the environment. -use object_store::aws::AmazonS3Builder; +use std::path::PathBuf; -use crate::error::StorageError; +use object_store::aws::AmazonS3Builder; -/// How Beacon's object storage is configured. +/// How Beacon's object storage is configured: where data lives on local disk and +/// whether the datasets store is backed by S3 instead of the local filesystem. #[derive(Debug, Clone, Default)] pub struct StorageConfig { + /// Root data directory (parent of the stores below). + pub data_dir: PathBuf, + /// Local root of the datasets store. Also used for NetCDF URL translation of + /// local datasets. + pub datasets_dir: PathBuf, + /// Local root of the tables store. + pub tables_dir: PathBuf, + /// Local root of the temporary-files store. + pub tmp_dir: PathBuf, + /// Watch the local datasets directory for changes (local backend only). pub enable_fs_events: bool, + /// Reserved: wire S3 change notifications into the event listener. pub enable_s3_events: bool, - pub s3: S3Config, + /// S3 backing for the datasets store. `None` => local filesystem; `Some` => + /// the datasets store is backed by S3, configured from these settings. + pub s3: Option, } -/// S3-specific storage settings (used when `s3.data_lake` is set). +/// S3 settings for the datasets store, present only when datasets are backed by +/// S3 ([`StorageConfig::s3`] is `Some`). /// /// This is the single source of truth for the S3 backend: it drives both store /// construction ([`Self::amazon_s3_builder`]) and NetCDF URL translation, so the @@ -25,42 +40,32 @@ pub struct StorageConfig { /// flow through the standard AWS environment chain (`AmazonS3Builder::from_env`). #[derive(Debug, Clone)] pub struct S3Config { - pub bucket: Option, - pub enable_virtual_hosting: bool, - pub data_lake: bool, - /// S3-compatible endpoint, e.g. `http://minio:9000`. Captured explicitly so - /// store-building and NetCDF URL translation use the same value. `None` => - /// rely on the AWS endpoint resolution (real AWS). + /// Bucket name. Required: `object_store` needs it regardless of addressing + /// style, and it is never inferred from the endpoint. + pub bucket: String, + /// S3-compatible endpoint, e.g. `http://minio:9000`. `None` => the endpoint is + /// resolved from the region (real AWS). pub endpoint: Option, - /// Optional region; when `None`, `from_env` still reads `AWS_REGION`. + /// Region; when `None`, `from_env` still reads `AWS_REGION`. pub region: Option, - /// Allow plain HTTP (dev/MinIO). Defaults to `true` (current behavior). + /// Use virtual-hosted-style addressing (bucket in the host) instead of + /// path-style (`{endpoint}/{bucket}/{key}`). Also selects the NetCDF URL form. + pub enable_virtual_hosting: bool, + /// Allow plain HTTP (dev/MinIO). pub allow_http: bool, } -impl Default for S3Config { - fn default() -> Self { - Self { - bucket: None, - enable_virtual_hosting: false, - data_lake: false, - endpoint: None, - region: None, - allow_http: true, - } - } -} - impl S3Config { /// Build an [`AmazonS3Builder`] from this config — the single place the S3 /// backend is configured. Credentials are layered in by /// [`AmazonS3Builder::from_env`]; the explicit values here override the - /// corresponding environment variables so the configured endpoint/region - /// always win. - pub(crate) fn amazon_s3_builder(&self) -> Result { + /// corresponding environment variables so the configured endpoint/region/ + /// bucket always win. + pub(crate) fn amazon_s3_builder(&self) -> AmazonS3Builder { let mut builder = AmazonS3Builder::from_env() .with_allow_http(self.allow_http) - .with_virtual_hosted_style_request(self.enable_virtual_hosting); + .with_virtual_hosted_style_request(self.enable_virtual_hosting) + .with_bucket_name(&self.bucket); if let Some(endpoint) = &self.endpoint { builder = builder.with_endpoint(endpoint); @@ -69,14 +74,6 @@ impl S3Config { builder = builder.with_region(region); } - if !self.enable_virtual_hosting { - // Path-style requests need an explicit bucket name. - let bucket = self.bucket.as_ref().ok_or(StorageError::MissingConfig { - key: "BEACON_S3_BUCKET", - })?; - builder = builder.with_bucket_name(bucket); - } - - Ok(builder) + builder } } diff --git a/beacon-object-storage/src/datasets_store.rs b/beacon-object-storage/src/datasets_store.rs index c942191c..5fce5724 100644 --- a/beacon-object-storage/src/datasets_store.rs +++ b/beacon-object-storage/src/datasets_store.rs @@ -25,36 +25,23 @@ use crate::{ /// Build the datasets [`DatasetsStore`] from the given storage configuration. /// /// The backing object store is selected from `storage`: -/// - When `storage.s3.data_lake` is set, an S3-compatible store is configured -/// from the standard AWS environment variables (see [`AmazonS3Builder::from_env`]). -/// For path-style requests a bucket name (`BEACON_S3_BUCKET`) is required; for -/// virtual-hosted-style the bucket is encoded in the endpoint host. -/// - Otherwise a local filesystem store rooted at `datasets_dir` is used. +/// - When `storage.s3` is `Some`, an S3-compatible store is configured from those +/// settings (credentials still come from the standard AWS chain; see +/// [`AmazonS3Builder::from_env`]). +/// - Otherwise a local filesystem store rooted at `storage.datasets_dir` is used. /// -/// The resolved `storage`/`datasets_dir` are retained on the returned store so +/// The `storage` config is retained on the returned store so /// [`DatasetsStore::translate_netcdf_url_path`] uses the same backend selection /// without consulting any process-global config. -/// Build a [`DatasetsStore`] backed by the local filesystem rooted at -/// `datasets_dir`, using default storage settings (no S3, no filesystem events). -/// -/// Intended for tests and embedders that only need a plain local datasets store -/// without composing a full [`crate::ObjectStores`]. -pub async fn local_datasets_store(datasets_dir: PathBuf) -> StorageResult { - create_datasets_store(&StorageConfig::default(), datasets_dir).await -} - -pub(crate) async fn create_datasets_store( - storage: &StorageConfig, - datasets_dir: PathBuf, -) -> StorageResult { +pub(crate) async fn create_datasets_store(storage: &StorageConfig) -> StorageResult { let (inner, event_listener): ( Arc, Option>, - ) = if storage.s3.data_lake { + ) = if let Some(s3) = &storage.s3 { tracing::info!("Using S3 object store for datasets"); // `object_store::Error` converts into `StorageError::ObjectStore` via `?`. - let store = storage.s3.amazon_s3_builder()?.build()?; + let store = s3.amazon_s3_builder().build()?; // TODO: wire S3 change notifications into an `EventListener` so the // cache-backed fast path can be enabled for the S3 backend too. @@ -62,7 +49,7 @@ pub(crate) async fn create_datasets_store( } else { tracing::info!("Using local filesystem object store for datasets"); - let root = datasets_dir.clone(); + let root = storage.datasets_dir.clone(); let store = LocalFileSystem::new_with_prefix(&root)?.with_automatic_cleanup(true); let inner = Arc::new(store) as Arc; @@ -90,7 +77,20 @@ pub(crate) async fn create_datasets_store( Ok(DatasetsStore::new(inner, event_listener) .await - .with_storage(storage.clone(), datasets_dir)) + .with_storage(storage.clone())) +} + +/// Build a [`DatasetsStore`] backed by the local filesystem rooted at +/// `datasets_dir`, using default storage settings (no S3, no filesystem events). +/// +/// Intended for tests and embedders that only need a plain local datasets store +/// without composing a full [`crate::ObjectStores`]. +pub async fn local_datasets_store(datasets_dir: PathBuf) -> StorageResult { + let storage = StorageConfig { + datasets_dir, + ..StorageConfig::default() + }; + create_datasets_store(&storage).await } pub trait EventListener: Send + Sync { @@ -120,11 +120,9 @@ pub struct DatasetsStore { /// and fans events out to `subscribers`. Aborted when the store is dropped. poll_task: Option>, /// Storage configuration this store was built from, retained for NetCDF URL - /// translation. Defaults to a local configuration. + /// translation (backend selection + local datasets directory). Defaults to a + /// local configuration. storage: StorageConfig, - /// Root directory for the local datasets store (used by NetCDF URL - /// translation in local mode). - datasets_dir: PathBuf, } impl DatasetsStore { @@ -178,15 +176,13 @@ impl DatasetsStore { subscribers, poll_task, storage: StorageConfig::default(), - datasets_dir: PathBuf::new(), } } - /// Attach the storage config + datasets directory used to build this store so + /// Attach the storage config used to build this store so /// [`Self::translate_netcdf_url_path`] resolves URLs without a global config. - pub fn with_storage(mut self, storage: StorageConfig, datasets_dir: PathBuf) -> Self { + pub fn with_storage(mut self, storage: StorageConfig) -> Self { self.storage = storage; - self.datasets_dir = datasets_dir; self } @@ -322,25 +318,17 @@ impl DatasetsStore { /// /// This function intentionally never returns an `s3://...` URL. pub fn translate_netcdf_url_path(&self, object: &Path) -> StorageResult { - let storage = &self.storage; - if storage.s3.data_lake { - let endpoint = - storage - .s3 - .endpoint - .as_deref() - .ok_or(error::StorageError::MissingConfig { - key: "AWS_ENDPOINT", - })?; - let url = s3_object_url( - endpoint, - storage.s3.bucket.as_deref(), - storage.s3.enable_virtual_hosting, - object, - )?; + if let Some(s3) = &self.storage.s3 { + let endpoint = s3 + .endpoint + .as_deref() + .ok_or(error::StorageError::MissingConfig { + key: "AWS_ENDPOINT", + })?; + let url = s3_object_url(endpoint, &s3.bucket, s3.enable_virtual_hosting, object)?; Ok(append_mode_bytes(url)) } else { - local_object_path(&self.datasets_dir, object) + local_object_path(&self.storage.datasets_dir, object) } } } @@ -352,7 +340,7 @@ impl DatasetsStore { /// endpoint host, so no bucket segment is added). fn s3_object_url( endpoint: &str, - bucket: Option<&str>, + bucket: &str, is_virtual_hosted_style: bool, object: &Path, ) -> StorageResult { @@ -370,20 +358,16 @@ fn s3_object_url( let key = object.as_ref().trim_start_matches('/'); let url = if is_virtual_hosted_style { + // Bucket is encoded in the endpoint host, so no bucket segment is added. if key.is_empty() { endpoint.to_string() } else { format!("{endpoint}/{key}") } + } else if key.is_empty() { + format!("{endpoint}/{bucket}") } else { - let bucket = bucket.ok_or(error::StorageError::MissingConfig { - key: "BEACON_S3_BUCKET", - })?; - if key.is_empty() { - format!("{endpoint}/{bucket}") - } else { - format!("{endpoint}/{bucket}/{key}") - } + format!("{endpoint}/{bucket}/{key}") }; Ok(url) @@ -887,12 +871,7 @@ mod tests { #[test] fn s3_object_url_path_style_includes_bucket() { - let url = s3_object_url( - "https://example.test", - Some("my-bucket"), - false, - &Path::from("a/b.nc"), - ) + let url = s3_object_url("https://example.test", "my-bucket", false, &Path::from("a/b.nc")) .unwrap(); assert_eq!(url, "https://example.test/my-bucket/a/b.nc"); } @@ -903,7 +882,7 @@ mod tests { // trailing slash on the endpoint is trimmed. let url = s3_object_url( "https://my-bucket.example.test/", - None, + "my-bucket", true, &Path::from("a/b.nc"), ) @@ -913,13 +892,8 @@ mod tests { #[test] fn s3_object_url_requires_http_endpoint() { - let err = s3_object_url( - "s3://my-bucket", - Some("my-bucket"), - false, - &Path::from("a.nc"), - ) - .expect_err("non-http endpoint must be rejected"); + let err = s3_object_url("s3://my-bucket", "my-bucket", false, &Path::from("a.nc")) + .expect_err("non-http endpoint must be rejected"); assert!(matches!( err, error::StorageError::InvalidConfig { @@ -929,18 +903,6 @@ mod tests { )); } - #[test] - fn s3_object_url_path_style_requires_bucket() { - let err = s3_object_url("https://example.test", None, false, &Path::from("a.nc")) - .expect_err("path-style without bucket must be rejected"); - assert!(matches!( - err, - error::StorageError::MissingConfig { - key: "BEACON_S3_BUCKET" - } - )); - } - #[test] fn append_mode_bytes_is_idempotent() { assert_eq!( diff --git a/beacon-object-storage/src/lib.rs b/beacon-object-storage/src/lib.rs index 4d14ca63..34069548 100644 --- a/beacon-object-storage/src/lib.rs +++ b/beacon-object-storage/src/lib.rs @@ -10,7 +10,7 @@ //! prefix-scoped event subscription, and provides NetCDF URL translation. //! - **Tables** / **tmp**: local filesystem. -use std::{path::PathBuf, sync::Arc}; +use std::sync::Arc; use object_store::{ObjectStore, local::LocalFileSystem}; @@ -44,22 +44,16 @@ pub struct ObjectStores { impl ObjectStores { /// Build all three object stores, returning a structured error if any fails. /// - /// `datasets_dir`/`tables_dir`/`tmp_dir` are the resolved local roots (from - /// the runtime's configured data directories). - pub async fn new( - storage: &StorageConfig, - datasets_dir: PathBuf, - tables_dir: PathBuf, - tmp_dir: PathBuf, - ) -> StorageResult { - let datasets = - Arc::new(datasets_store::create_datasets_store(storage, datasets_dir).await?); + /// The local roots and S3 backend selection all come from `storage`. + pub async fn new(storage: &StorageConfig) -> StorageResult { + let datasets = Arc::new(datasets_store::create_datasets_store(storage).await?); let tables = Arc::new( - LocalFileSystem::new_with_prefix(tables_dir)?.with_automatic_cleanup(true), + LocalFileSystem::new_with_prefix(&storage.tables_dir)?.with_automatic_cleanup(true), ) as Arc; - let tmp = Arc::new(LocalFileSystem::new_with_prefix(tmp_dir)?) as Arc; + let tmp = + Arc::new(LocalFileSystem::new_with_prefix(&storage.tmp_dir)?) as Arc; tracing::info!("object stores initialized"); Ok(Self { From 0750b697136f6b3feac5d482bfc4d68041b997c0 Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Sun, 21 Jun 2026 11:23:01 +0200 Subject: [PATCH 2/2] Write NetCDF output to the configured tmp dir, not std::env::temp_dir() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The NetCDF sinks hardcoded std::env::temp_dir() for the output file, so `output: { format: netcdf }` ignored config.storage.tmp_dir and (after the tmp-store move) wrote to a different directory than the returned file handle — producing an empty download. Thread the tmp directory through: expose DatasetsStore::storage(), and in NetcdfFormat::create_writer_physical_plan read storage().tmp_dir (the factory already holds the datasets store, which now carries the full StorageConfig) and pass it to NetCDFSink/NetCDFNdSink, which write there instead of temp_dir(). Adds an end-to-end regression test asserting NetCDF output lands under the configured tmp dir and is non-empty. --- beacon-core/src/runtime.rs | 48 +++++++++++++++++++ .../beacon-arrow-netcdf/src/datafusion/mod.rs | 7 ++- .../src/datafusion/sink.rs | 34 ++++++++----- beacon-object-storage/src/datasets_store.rs | 7 +++ 4 files changed, 84 insertions(+), 12 deletions(-) diff --git a/beacon-core/src/runtime.rs b/beacon-core/src/runtime.rs index 964f0701..247c3110 100644 --- a/beacon-core/src/runtime.rs +++ b/beacon-core/src/runtime.rs @@ -776,6 +776,54 @@ mod client_query_tests { } } + /// NetCDF output goes through a custom `DataSink` that writes a real local + /// file (the netcdf-c writer cannot stream to an object store). This asserts + /// the file lands under the configured tmp store root — not the OS temp dir — + /// and is non-empty: the regression fixed by threading `StorageConfig` into + /// the NetCDF factory/sink. + #[tokio::test(flavor = "multi_thread")] + async fn query_with_netcdf_output_writes_under_configured_tmp() { + let config = std::sync::Arc::new(beacon_config::Config::load().unwrap()); + let tmp_dir = config.storage.tmp_dir.clone(); + let runtime = Runtime::new(config).await.expect("runtime should start"); + let suffix = uuid::Uuid::new_v4().simple(); + let table = format!("ncout_{suffix}"); + + run_sql(&runtime, &format!("CREATE TABLE {table} (a BIGINT)")).await; + run_sql(&runtime, &format!("INSERT INTO {table} VALUES (1), (2)")).await; + + let result = runtime + .run_query( + query(serde_json::json!({ + "from": table, + "select": ["a"], + "output": { "format": "netcdf" }, + })), + false, + ) + .await + .expect("netcdf query with output should run"); + + match result.query_output { + QueryOutput::File(file) => { + // tempfile resolves to an absolute path; canonicalize both + // sides so the comparison is independent of cwd-relative form. + let got = std::fs::canonicalize(file.path().parent().unwrap()) + .expect("canonicalize output parent"); + let want = std::fs::canonicalize(&tmp_dir).expect("canonicalize tmp dir"); + assert_eq!( + got, want, + "netcdf output should be written under the configured tmp dir" + ); + assert!( + file.size().expect("file size") > 0, + "netcdf output file should contain data" + ); + } + QueryOutput::Stream(_) => panic!("expected a file output"), + } + } + /// `validate_query_plan` is the single permission gate: non-super-users may run /// read-only SELECTs but not DDL/DML (standard nodes) nor any beacon extension /// operation (super-user-only). diff --git a/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/mod.rs b/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/mod.rs index 5578bb7b..7904df5d 100644 --- a/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/mod.rs +++ b/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/mod.rs @@ -335,6 +335,10 @@ impl FileFormat for NetcdfFormat { conf: FileSinkConfig, order_requirements: Option, ) -> datafusion::error::Result> { + // NetCDF needs a real local path (the netcdf-c writer cannot stream to an + // object store). Write into the configured tmp store root, threaded in via + // the datasets store's `StorageConfig`, rather than the OS temp dir. + let output_dir = self.datasets_object_store.storage().tmp_dir.clone(); match &self.options.write_dimensions { Some(dim_columns) if !dim_columns.is_empty() => { let unique_columns = dim_columns.clone(); @@ -362,6 +366,7 @@ impl FileFormat for NetcdfFormat { conf, unique_columns.len(), collection_handle, + output_dir.clone(), )?); Ok(Arc::new(DataSinkExec::new( @@ -371,7 +376,7 @@ impl FileFormat for NetcdfFormat { ))) } _ => { - let netcdf_sink = Arc::new(NetCDFSink::new(conf)); + let netcdf_sink = Arc::new(NetCDFSink::new(conf, output_dir)); Ok(Arc::new(DataSinkExec::new( input, netcdf_sink, diff --git a/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/sink.rs b/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/sink.rs index fe274946..b04697a0 100644 --- a/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/sink.rs +++ b/beacon-file-formats/beacon-arrow-netcdf/src/datafusion/sink.rs @@ -20,7 +20,7 @@ //! | `None` / empty | [`NetCDFSink`] | flat, unlimited `obs` | //! | `Some(["lat", "lon", …])` | [`NetCDFNdSink`]| gridded, named dims | -use std::{any::Any, env::temp_dir, fmt::Formatter, sync::Arc}; +use std::{any::Any, fmt::Formatter, path::PathBuf, sync::Arc}; use crate::{encoders::default::DefaultEncoder, writer::ArrowRecordBatchWriter}; use arrow::{ @@ -49,17 +49,23 @@ use ordered_float::OrderedFloat; /// [`ArrowRecordBatchWriter`] which serializes columns into /// NetCDF variables along a single unlimited `obs` dimension. /// -/// The output file is written to `std::env::temp_dir()` joined with the path -/// from [`FileSinkConfig::table_paths`]. +/// The output file is written to `output_dir` (the configured tmp store root) +/// joined with the path from [`FileSinkConfig::table_paths`]. #[derive(Debug, Clone)] pub struct NetCDFSink { sink_config: FileSinkConfig, + /// Directory the NetCDF file is written to — the configured tmp store root. + output_dir: PathBuf, } impl NetCDFSink { - /// Create a new flat sink bound to the given [`FileSinkConfig`]. - pub fn new(sink_config: FileSinkConfig) -> Self { - Self { sink_config } + /// Create a new flat sink bound to the given [`FileSinkConfig`], writing into + /// `output_dir`. + pub fn new(sink_config: FileSinkConfig, output_dir: PathBuf) -> Self { + Self { + sink_config, + output_dir, + } } } @@ -90,7 +96,7 @@ impl DataSink for NetCDFSink { ) -> datafusion::error::Result { let arrow_schema = self.sink_config.output_schema().clone(); let file_path = self.sink_config.table_paths[0].prefix(); - let full_path = temp_dir().join(file_path.as_ref()); + let full_path = self.output_dir.join(file_path.as_ref()); tracing::info!("Writing NetCDF to path: {:?}", full_path); let mut rows_written: u64 = 0; @@ -141,6 +147,8 @@ pub struct NetCDFNdSink { sink_config: FileSinkConfig, ndims: usize, unique_values: UniqueValuesHandleCollection, + /// Directory the NetCDF file is written to — the configured tmp store root. + output_dir: PathBuf, } impl NetCDFNdSink { @@ -152,6 +160,7 @@ impl NetCDFNdSink { sink_config: FileSinkConfig, ndims: usize, unique_values: UniqueValuesHandleCollection, + output_dir: PathBuf, ) -> Result { tracing::info!("Creating NetCDFNdSink with {} dimensions", ndims); for field in sink_config.output_schema().fields() { @@ -168,6 +177,7 @@ impl NetCDFNdSink { sink_config, ndims, unique_values, + output_dir, }) } } @@ -197,7 +207,9 @@ impl DataSink for NetCDFNdSink { data: SendableRecordBatchStream, _context: &Arc, ) -> datafusion::error::Result { - let output_path = temp_dir().join(self.sink_config.table_paths[0].prefix().as_ref()); + let output_path = self + .output_dir + .join(self.sink_config.table_paths[0].prefix().as_ref()); tracing::info!("Writing ND NetCDF to path: {:?}", output_path); let mut rows_written: u64 = 0; @@ -1439,7 +1451,7 @@ mod tests { let sink_config = test_sink_config(schema); let collection = UniqueValuesHandleCollection::new(); - let err = NetCDFNdSink::new(sink_config, 1, collection).unwrap_err(); + let err = NetCDFNdSink::new(sink_config, 1, collection, std::env::temp_dir()).unwrap_err(); assert!(err.to_string().contains("only supports primitive")); assert!(err.to_string().contains("name")); } @@ -1458,7 +1470,7 @@ mod tests { fn test_display_as_flat() { let schema = Arc::new(Schema::new(vec![Field::new("val", DataType::Int32, false)])); let sink_config = test_sink_config(schema); - let sink = NetCDFSink::new(sink_config); + let sink = NetCDFSink::new(sink_config, std::env::temp_dir()); let display = format!("{}", DisplayAsWrapper(&sink)); assert!(display.contains("NetCDFSink")); } @@ -1468,7 +1480,7 @@ mod tests { let schema = Arc::new(Schema::new(vec![Field::new("val", DataType::Int32, false)])); let sink_config = test_sink_config(schema); let collection = UniqueValuesHandleCollection::new(); - let sink = NetCDFNdSink::new(sink_config, 0, collection).unwrap(); + let sink = NetCDFNdSink::new(sink_config, 0, collection, std::env::temp_dir()).unwrap(); let display = format!("{}", DisplayAsWrapper(&sink)); assert!(display.contains("NetCDFNdSink")); } diff --git a/beacon-object-storage/src/datasets_store.rs b/beacon-object-storage/src/datasets_store.rs index 5fce5724..8d3b7777 100644 --- a/beacon-object-storage/src/datasets_store.rs +++ b/beacon-object-storage/src/datasets_store.rs @@ -186,6 +186,13 @@ impl DatasetsStore { self } + /// The storage configuration this store was built from (backend selection + + /// local directory layout). Writers that need the configured paths read it + /// here, e.g. the NetCDF output sink's temporary directory. + pub fn storage(&self) -> &StorageConfig { + &self.storage + } + /// Continuously drain events from `listener`, apply them to `cache`, and fan /// them out to matching `subscribers`. ///