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

Consolidate data storage and caching into single abstraction to simplify architecture and eliminate the cached/uncached store distinction.

- Unified `DataStore` combines storage and cache in single type
- Integrated parquet metadata cache with memory-aware eviction
- Centralized physical table operations and file naming
- Single store interface across all services and modules

Signed-off-by: Lorenzo Delgado <lorenzo@edgeandnode.com>

Author: LNSD

Cargo.lock

@@ -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",

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/Cargo.toml

@@ -1,11 +1,12 @@
 use std::{net::SocketAddr, sync::Arc};
 
 use amp_config::Config as CommonConfig;
+use amp_data_store::DataStore;
 use amp_dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
 };
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use controller::config::Config;
 use monitoring::telemetry::metrics::Meter;
 
@@ -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/controller_cmd.rs

@@ -1,11 +1,12 @@
 use std::sync::Arc;
 
 use amp_config::{Addrs, Config as CommonConfig};
+use amp_data_store::DataStore;
 use amp_dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
 };
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use monitoring::telemetry::metrics::Meter;
 use server::config::Config as ServerConfig;
 
@@ -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/server_cmd.rs

@@ -1,11 +1,12 @@
 use std::{future::Future, pin::Pin, sync::Arc};
 
 use amp_config::Config as CommonConfig;
+use amp_data_store::DataStore;
 use amp_dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
 };
 use amp_object_store::ObjectStoreCreationError;
-use common::{BoxError, store::Store};
+use common::BoxError;
 use monitoring::telemetry::metrics::Meter;
 
 use crate::{controller_cmd, server_cmd, worker_cmd};
@@ -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/solo_cmd.rs

@@ -1,9 +1,9 @@
 use amp_config::Config;
+use amp_data_store::DataStore;
 use amp_dataset_store::{
     DatasetStore, manifests::DatasetManifestsStore, providers::ProviderConfigsStore,
 };
 use amp_object_store::ObjectStoreCreationError;
-use common::store::Store;
 use monitoring::telemetry::metrics::Meter;
 use worker::node_id::NodeId;
 
@@ -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/bin/ampd/src/worker_cmd.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/config/src/lib.rs

@@ -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