feat(amp-data-store): introduce crate with unified store interface

Cargo.toml

@@ -10,6 +10,7 @@ members = [
     "crates/bin/ampup",
     "crates/clients/flight",
     "crates/core/common",
+    "crates/core/data-store",
     "crates/core/dataset-store",
     "crates/core/datasets-common",
     "crates/core/datasets-derived",

crates/bin/ampd/Cargo.toml

@@ -14,6 +14,7 @@ console-subscriber = ["dep:console-subscriber"]
 snmalloc = ["dep:snmalloc-rs"]
 
 [dependencies]
+amp-data-store = { path = "../../core/data-store" }
 amp-object-store = { path = "../../core/object-store" }
 clap.workspace = true
 common = { path = "../../core/common" }

crates/bin/ampd/src/controller_cmd.rs

@@ -1,7 +1,8 @@
 use std::{net::SocketAddr, sync::Arc};
 
+use amp_data_store::DataStore;
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use config::Config as CommonConfig;
 use controller::config::Config;
 use dataset_store::{
@@ -16,8 +17,12 @@ pub async fn run(config: CommonConfig, meter: Option<Meter>, at: SocketAddr) ->
         .await
         .map_err(|err| Error::MetadataDbConnection(Box::new(err)))?;
 
-    let data_store = Store::new(metadata_db.clone(), config.data_store_url.clone())
-        .map_err(Error::DataStoreCreation)?;
+    let data_store = DataStore::new(
+        metadata_db.clone(),
+        config.data_store_url.clone(),
+        config.parquet.cache_size_mb,
+    )
+    .map_err(Error::DataStoreCreation)?;
 
     let dataset_store = {
         let provider_configs_store = ProviderConfigsStore::new(

crates/bin/ampd/src/server_cmd.rs

@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
+use amp_data_store::DataStore;
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use config::{Addrs, Config as CommonConfig};
 use dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
@@ -21,8 +22,12 @@ pub async fn run(
         .await
         .map_err(|err| Error::MetadataDbConnection(Box::new(err)))?;
 
-    let data_store = Store::new(metadata_db.clone(), config.data_store_url.clone())
-        .map_err(Error::DataStoreCreation)?;
+    let data_store = DataStore::new(
+        metadata_db.clone(),
+        config.data_store_url.clone(),
+        config.parquet.cache_size_mb,
+    )
+    .map_err(Error::DataStoreCreation)?;
 
     let dataset_store = {
         let provider_configs_store = ProviderConfigsStore::new(
@@ -146,6 +151,5 @@ pub fn config_from_common(config: &CommonConfig) -> ServerConfig {
         max_mem_mb: config.max_mem_mb,
         query_max_mem_mb: config.query_max_mem_mb,
         spill_location: config.spill_location.clone(),
-        parquet_cache_size_mb: config.parquet.cache_size_mb,
     }
 }

crates/bin/ampd/src/solo_cmd.rs

@@ -1,7 +1,8 @@
 use std::{future::Future, pin::Pin, sync::Arc};
 
+use amp_data_store::DataStore;
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use config::Config as CommonConfig;
 use dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
@@ -24,8 +25,12 @@ pub async fn run(
         .await
         .map_err(|err| Error::MetadataDbConnection(Box::new(err)))?;
 
-    let data_store = Store::new(metadata_db.clone(), config.data_store_url.clone())
-        .map_err(Error::DataStoreCreation)?;
+    let data_store = DataStore::new(
+        metadata_db.clone(),
+        config.data_store_url.clone(),
+        config.parquet.cache_size_mb,
+    )
+    .map_err(Error::DataStoreCreation)?;
 
     let dataset_store = {
         let provider_configs_store = ProviderConfigsStore::new(

crates/bin/ampd/src/worker_cmd.rs

@@ -1,5 +1,5 @@
+use amp_data_store::DataStore;
 use amp_object_store::ObjectStoreCreationError;
-use common::store::Store;
 use config::Config;
 use dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
@@ -13,8 +13,12 @@ pub async fn run(config: Config, meter: Option<Meter>, node_id: NodeId) -> Resul
         .await
         .map_err(|err| Error::MetadataDbConnection(Box::new(err)))?;
 
-    let data_store = Store::new(metadata_db.clone(), config.data_store_url.clone())
-        .map_err(Error::DataStoreCreation)?;
+    let data_store = DataStore::new(
+        metadata_db.clone(),
+        config.data_store_url.clone(),
+        config.parquet.cache_size_mb,
+    )
+    .map_err(Error::DataStoreCreation)?;
 
     let dataset_store = {
         let provider_configs_store = ProviderConfigsStore::new(

crates/config/src/lib.rs

@@ -314,7 +314,6 @@ impl Config {
             self.max_mem_mb,
             self.query_max_mem_mb,
             &self.spill_location,
-            self.parquet.cache_size_mb,
         )
     }
 

crates/core/common/Cargo.toml

@@ -6,7 +6,7 @@ license-file.workspace = true
 
 [dependencies]
 alloy.workspace = true
-amp-object-store = { path = "../object-store" }
+amp-data-store = { path = "../data-store" }
 async-stream.workspace = true
 async-trait.workspace = true
 axum.workspace = true
@@ -17,9 +17,6 @@ datafusion.workspace = true
 datafusion-datasource.workspace = true
 datafusion-tracing.workspace = true
 datasets-common = { path = "../datasets-common" }
-figment.workspace = true
-foyer = "0.21"
-fs-err.workspace = true
 futures.workspace = true
 governor.workspace = true
 indoc.workspace = true