Storage config owns data dirs + optional S3; NetCDF output to configured tmp

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 {

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<RawConfig> 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<RawConfig> 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,
         ] {

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?;
 
@@ -782,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).

beacon-core/tests/runtime_config.rs

@@ -14,17 +14,17 @@ fn config_with(default_table: &str, tag: &str) -> Arc<beacon_config::Config> {
     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,
     ] {

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 {

beacon-file-formats/beacon-arrow-netcdf/src/datafusion/mod.rs

@@ -335,6 +335,10 @@ impl FileFormat for NetcdfFormat {
         conf: FileSinkConfig,
         order_requirements: Option<LexRequirement>,
     ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+        // 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,

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<DefaultEncoder>`] 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<u64> {
         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<Self, DataFusionError> {
         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<TaskContext>,
     ) -> datafusion::error::Result<u64> {
-        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"));
     }

beacon-iceberg/src/catalog.rs

@@ -53,10 +53,9 @@ pub fn beacon_namespace() -> Vec<String> {
 pub async fn init_datasets_warehouse(
     datasets: std::sync::Arc<beacon_object_storage::DatasetsStore>,
     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 `<datasets_dir>/__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())
     };