Cleanup: remove state_dir config (#36)

zfarrell · web-flow · commit 3809c6daa652 · 2025-12-14T09:10:03.000-08:00
remove state_dir
diff --git a/.dockerignore b/.dockerignore
@@ -5,7 +5,6 @@ target/
 .gitignore
 *.db
 cache/
-state/
 venv/
 __pycache__/
 *.pyc
diff --git a/.gitignore b/.gitignore
@@ -15,5 +15,4 @@
 # Project-specific
 catalog.db*
 cache/
-state/
 
diff --git a/config-docker.toml b/config-docker.toml
@@ -10,4 +10,3 @@ type = "filesystem"
 
 [paths]
 cache_dir = "/app/cache"
-state_dir = "/app/state"
diff --git a/config-local.toml b/config-local.toml
@@ -25,4 +25,3 @@ type = "filesystem"
 
 [paths]
 #cache_dir = "~/.hotdata/rivetdb/cache"
-#state_dir = "~/.hotdata/rivetdb/state"
diff --git a/src/catalog/backend.rs b/src/catalog/backend.rs
@@ -183,7 +183,7 @@ where
 
     pub async fn list_tables(&self, connection_id: Option<i32>) -> Result<Vec<TableInfo>> {
         let mut sql = String::from(
-            "SELECT id, connection_id, schema_name, table_name, parquet_path, state_path, \
+            "SELECT id, connection_id, schema_name, table_name, parquet_path, \
              CAST(last_sync AS TEXT) as last_sync, arrow_schema_json \
              FROM tables",
         );
@@ -210,7 +210,7 @@ where
         table_name: &str,
     ) -> Result<Option<TableInfo>> {
         let sql = format!(
-            "SELECT id, connection_id, schema_name, table_name, parquet_path, state_path, \
+            "SELECT id, connection_id, schema_name, table_name, parquet_path, \
              CAST(last_sync AS TEXT) as last_sync, arrow_schema_json \
              FROM tables WHERE connection_id = {} AND schema_name = {} AND table_name = {}",
             DB::bind_param(1),
@@ -227,23 +227,16 @@ where
             .map_err(Into::into)
     }
 
-    pub async fn update_table_sync(
-        &self,
-        table_id: i32,
-        parquet_path: &str,
-        state_path: &str,
-    ) -> Result<()> {
+    pub async fn update_table_sync(&self, table_id: i32, parquet_path: &str) -> Result<()> {
         let sql = format!(
-            "UPDATE tables SET parquet_path = {}, state_path = {}, last_sync = CURRENT_TIMESTAMP \
+            "UPDATE tables SET parquet_path = {}, last_sync = CURRENT_TIMESTAMP \
              WHERE id = {}",
             DB::bind_param(1),
             DB::bind_param(2),
-            DB::bind_param(3),
         );
 
         query(&sql)
             .bind(parquet_path)
-            .bind(state_path)
             .bind(table_id)
             .execute(&self.pool)
             .await?;
@@ -263,7 +256,7 @@ where
             .ok_or_else(|| anyhow!("Table '{}.{}' not found", schema_name, table_name))?;
 
         let sql = format!(
-            "UPDATE tables SET parquet_path = NULL, state_path = NULL, last_sync = NULL WHERE id = {}",
+            "UPDATE tables SET parquet_path = NULL, last_sync = NULL WHERE id = {}",
             DB::bind_param(1)
         );
 
@@ -279,7 +272,7 @@ where
             .ok_or_else(|| anyhow!("Connection '{}' not found", name))?;
 
         let sql = format!(
-            "UPDATE tables SET parquet_path = NULL, state_path = NULL, last_sync = NULL \
+            "UPDATE tables SET parquet_path = NULL, last_sync = NULL \
              WHERE connection_id = {}",
             DB::bind_param(1)
         );
diff --git a/src/catalog/manager.rs b/src/catalog/manager.rs
@@ -19,7 +19,6 @@ pub struct TableInfo {
     pub schema_name: String,
     pub table_name: String,
     pub parquet_path: Option<String>,
-    pub state_path: Option<String>,
     pub last_sync: Option<String>,
     pub arrow_schema_json: Option<String>,
 }
@@ -54,12 +53,7 @@ pub trait CatalogManager: Debug + Send + Sync {
         schema_name: &str,
         table_name: &str,
     ) -> Result<Option<TableInfo>>;
-    async fn update_table_sync(
-        &self,
-        table_id: i32,
-        parquet_path: &str,
-        state_path: &str,
-    ) -> Result<()>;
+    async fn update_table_sync(&self, table_id: i32, parquet_path: &str) -> Result<()>;
 
     /// Clear table cache metadata (set paths to NULL) without deleting files.
     async fn clear_table_cache_metadata(
diff --git a/src/catalog/postgres_manager.rs b/src/catalog/postgres_manager.rs
@@ -42,7 +42,6 @@ impl PostgresCatalogManager {
                 schema_name TEXT NOT NULL,
                 table_name TEXT NOT NULL,
                 parquet_path TEXT,
-                state_path TEXT,
                 last_sync TIMESTAMP,
                 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                 arrow_schema_json TEXT,
@@ -146,15 +145,8 @@ impl CatalogManager for PostgresCatalogManager {
             .await
     }
 
-    async fn update_table_sync(
-        &self,
-        table_id: i32,
-        parquet_path: &str,
-        state_path: &str,
-    ) -> Result<()> {
-        self.backend
-            .update_table_sync(table_id, parquet_path, state_path)
-            .await
+    async fn update_table_sync(&self, table_id: i32, parquet_path: &str) -> Result<()> {
+        self.backend.update_table_sync(table_id, parquet_path).await
     }
 
     async fn clear_table_cache_metadata(
diff --git a/src/catalog/sqlite_manager.rs b/src/catalog/sqlite_manager.rs
@@ -56,7 +56,6 @@ impl SqliteCatalogManager {
                 schema_name TEXT NOT NULL,
                 table_name TEXT NOT NULL,
                 parquet_path TEXT,
-                state_path TEXT,
                 last_sync TIMESTAMP,
                 arrow_schema_json TEXT,
                 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
@@ -124,15 +123,8 @@ impl CatalogManager for SqliteCatalogManager {
             .await
     }
 
-    async fn update_table_sync(
-        &self,
-        table_id: i32,
-        parquet_path: &str,
-        state_path: &str,
-    ) -> Result<()> {
-        self.backend
-            .update_table_sync(table_id, parquet_path, state_path)
-            .await
+    async fn update_table_sync(&self, table_id: i32, parquet_path: &str) -> Result<()> {
+        self.backend.update_table_sync(table_id, parquet_path).await
     }
 
     async fn clear_table_cache_metadata(
diff --git a/src/config/mod.rs b/src/config/mod.rs
@@ -49,7 +49,6 @@ pub struct StorageConfig {
 #[derive(Debug, Clone, Deserialize, Serialize, Default)]
 pub struct PathsConfig {
     pub cache_dir: Option<String>,
-    pub state_dir: Option<String>,
 }
 
 impl AppConfig {
diff --git a/src/datafusion/lazy_table_provider.rs b/src/datafusion/lazy_table_provider.rs
@@ -147,10 +147,7 @@ impl LazyTableProvider {
             .get_table(self.connection_id, &self.schema_name, &self.table_name)
             .await
         {
-            let _ = self
-                .catalog
-                .update_table_sync(info.id, &parquet_url, "")
-                .await;
+            let _ = self.catalog.update_table_sync(info.id, &parquet_url).await;
         }
 
         Ok(parquet_url)
diff --git a/src/engine.rs b/src/engine.rs
@@ -28,19 +28,17 @@ pub struct RivetEngine {
 impl RivetEngine {
     /// Create a new engine instance and register all existing connections.
     pub async fn new(catalog_path: &str) -> Result<Self> {
-        Self::new_with_paths(catalog_path, "cache", "state", false).await
+        Self::new_with_paths(catalog_path, "cache", false).await
     }
 
     /// Create a new engine instance with custom paths for catalog, cache, and state.
     pub async fn new_with_paths(
         catalog_path: &str,
         cache_base: &str,
-        state_base: &str,
         readonly: bool,
     ) -> Result<Self> {
         // Create filesystem storage manager
-        let storage: Arc<dyn StorageManager> =
-            Arc::new(FilesystemStorage::new(cache_base, state_base));
+        let storage: Arc<dyn StorageManager> = Arc::new(FilesystemStorage::new(cache_base));
         Self::new_with_storage(catalog_path, storage, readonly).await
     }
 
@@ -74,19 +72,10 @@ impl RivetEngine {
     /// Delete cache and state directories for a connection.
     async fn delete_connection_files(&self, connection_id: i32) -> Result<()> {
         let cache_prefix = self.storage.cache_prefix(connection_id);
-        let state_prefix = self.storage.state_prefix(connection_id);
 
-        let (cache_res, state_res) = tokio::join!(
-            self.storage.delete_prefix(&cache_prefix),
-            self.storage.delete_prefix(&state_prefix),
-        );
-
-        if let Err(e) = cache_res {
+        if let Err(e) = self.storage.delete_prefix(&cache_prefix).await {
             warn!("Failed to delete cache prefix {}: {}", cache_prefix, e);
         }
-        if let Err(e) = state_res {
-            warn!("Failed to delete state prefix {}: {}", state_prefix, e);
-        }
 
         Ok(())
     }
@@ -100,13 +89,6 @@ impl RivetEngine {
             }
         }
 
-        // Delete state path if it exists
-        if let Some(state_path) = &table_info.state_path {
-            if let Err(e) = self.storage.delete(state_path).await {
-                warn!("Failed to delete state file {}: {}", state_path, e);
-            }
-        }
-
         Ok(())
     }
 
@@ -456,11 +438,6 @@ impl RivetEngine {
                 .parent()
                 .ok_or_else(|| anyhow::anyhow!("Invalid cache directory path"))?
                 .to_path_buf()
-        } else if let Some(state_dir) = &config.paths.state_dir {
-            PathBuf::from(state_dir)
-                .parent()
-                .ok_or_else(|| anyhow::anyhow!("Invalid state directory path"))?
-                .to_path_buf()
         } else {
             // Default to ~/.hotdata/rivetdb
             let home = std::env::var("HOME")
@@ -476,17 +453,10 @@ impl RivetEngine {
             .as_ref()
             .map(PathBuf::from)
             .unwrap_or_else(|| metadata_dir.join("cache"));
-        let state_dir_path = config
-            .paths
-            .state_dir
-            .as_ref()
-            .map(PathBuf::from)
-            .unwrap_or_else(|| metadata_dir.join("state"));
 
         // Ensure directories exist before using them
         std::fs::create_dir_all(&metadata_dir)?;
         std::fs::create_dir_all(&cache_dir_path)?;
-        std::fs::create_dir_all(&state_dir_path)?;
 
         // Create catalog manager based on config
         let catalog: Arc<dyn CatalogManager> = match config.catalog.catalog_type.as_str() {
@@ -540,10 +510,7 @@ impl RivetEngine {
                 let cache_dir_str = cache_dir_path
                     .to_str()
                     .ok_or_else(|| anyhow::anyhow!("Invalid cache directory path"))?;
-                let state_dir_str = state_dir_path
-                    .to_str()
-                    .ok_or_else(|| anyhow::anyhow!("Invalid state directory path"))?;
-                Arc::new(FilesystemStorage::new(cache_dir_str, state_dir_str))
+                Arc::new(FilesystemStorage::new(cache_dir_str))
             }
             "s3" => {
                 let bucket = config
@@ -610,7 +577,6 @@ mod tests {
 
         let storage = Arc::new(FilesystemStorage::new(
             temp_dir.path().join("cache").to_str().unwrap(),
-            temp_dir.path().join("state").to_str().unwrap(),
         ));
 
         // Build engine using builder pattern
@@ -659,7 +625,6 @@ mod tests {
 
         let temp_dir = TempDir::new().unwrap();
         let cache_dir = temp_dir.path().join("cache");
-        let state_dir = temp_dir.path().join("state");
 
         let config = AppConfig {
             server: ServerConfig {
@@ -682,7 +647,6 @@ mod tests {
             },
             paths: PathsConfig {
                 cache_dir: Some(cache_dir.to_str().unwrap().to_string()),
-                state_dir: Some(state_dir.to_str().unwrap().to_string()),
             },
         };
 
diff --git a/src/storage/filesystem.rs b/src/storage/filesystem.rs
@@ -10,14 +10,12 @@ use super::StorageManager;
 #[derive(Debug)]
 pub struct FilesystemStorage {
     cache_base: PathBuf,
-    state_base: PathBuf,
 }
 
 impl FilesystemStorage {
-    pub fn new(cache_base: &str, state_base: &str) -> Self {
+    pub fn new(cache_base: &str) -> Self {
         Self {
             cache_base: PathBuf::from(cache_base),
-            state_base: PathBuf::from(state_base),
         }
     }
 }
@@ -35,29 +33,13 @@ impl StorageManager for FilesystemStorage {
         format!("file://{}", path.display())
     }
 
-    fn state_url(&self, connection_id: i32, schema: &str, table: &str) -> String {
-        let path = self
-            .state_base
-            .join(connection_id.to_string())
-            .join(schema)
-            .join(format!("{}.json", table));
-        format!("file://{}", path.display())
-    }
-
     fn cache_prefix(&self, connection_id: i32) -> String {
         self.cache_base
             .join(connection_id.to_string())
             .to_string_lossy()
             .to_string()
     }
 
-    fn state_prefix(&self, connection_id: i32) -> String {
-        self.state_base
-            .join(connection_id.to_string())
-            .to_string_lossy()
-            .to_string()
-    }
-
     async fn read(&self, url: &str) -> Result<Vec<u8>> {
         let path = url
             .strip_prefix("file://")
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
@@ -12,7 +12,7 @@ pub mod s3;
 pub use filesystem::FilesystemStorage;
 pub use s3::S3Storage;
 
-/// S3 credentials for passing to sync scripts (e.g., DLT).
+/// S3 credentials for passing to sync scripts
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct S3Credentials {
     pub aws_access_key_id: String,
@@ -24,9 +24,7 @@ pub struct S3Credentials {
 pub trait StorageManager: Debug + Send + Sync {
     // Path construction
     fn cache_url(&self, connection_id: i32, schema: &str, table: &str) -> String;
-    fn state_url(&self, connection_id: i32, schema: &str, table: &str) -> String;
     fn cache_prefix(&self, connection_id: i32) -> String;
-    fn state_prefix(&self, connection_id: i32) -> String;
 
     // File operations
     async fn read(&self, url: &str) -> Result<Vec<u8>>;
@@ -38,7 +36,7 @@ pub trait StorageManager: Debug + Send + Sync {
     // DataFusion integration
     fn register_with_datafusion(&self, ctx: &SessionContext) -> Result<()>;
 
-    /// Get S3 credentials for DLT sync script.
+    /// Get S3 credentials
     /// Returns None for non-S3 storage backends.
     fn get_s3_credentials(&self) -> Option<S3Credentials> {
         None
diff --git a/src/storage/s3.rs b/src/storage/s3.rs
@@ -86,21 +86,10 @@ impl StorageManager for S3Storage {
         )
     }
 
-    fn state_url(&self, connection_id: i32, schema: &str, table: &str) -> String {
-        format!(
-            "s3://{}/state/{}/{}/{}.json",
-            self.bucket, connection_id, schema, table
-        )
-    }
-
     fn cache_prefix(&self, connection_id: i32) -> String {
         format!("s3://{}/cache/{}", self.bucket, connection_id)
     }
 
-    fn state_prefix(&self, connection_id: i32) -> String {
-        format!("s3://{}/state/{}", self.bucket, connection_id)
-    }
-
     async fn read(&self, url: &str) -> Result<Vec<u8>> {
         let path = self.url_to_path(url)?;
         let result = self.store.get(&path).await?;
diff --git a/tests/catalog_manager_suite.rs b/tests/catalog_manager_suite.rs
diff --git a/tests/engine_sync_tests.rs b/tests/engine_sync_tests.rs
diff --git a/tests/http_server_tests.rs b/tests/http_server_tests.rs
diff --git a/tests/integration_tests.rs b/tests/integration_tests.rs
diff --git a/tests/s3_storage_tests.rs b/tests/s3_storage_tests.rs
diff --git a/tests/storage_tests.rs b/tests/storage_tests.rs

Original file line number	Diff line number	Diff line change
`@@ -10,4 +10,3 @@ type = "filesystem"`
`10`	`10`
`11`	`11`	`[paths]`
`12`	`12`	`cache_dir = "/app/cache"`
`13`		`-state_dir = "/app/state"`
Original file line number	Diff line number	Diff line change
`@@ -25,4 +25,3 @@ type = "filesystem"`
`25`	`25`
`26`	`26`	`[paths]`
`27`	`27`	`#cache_dir = "~/.hotdata/rivetdb/cache"`
`28`		`-#state_dir = "~/.hotdata/rivetdb/state"`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,6 @@ pub struct StorageConfig {`
`49`	`49`	`#[derive(Debug, Clone, Deserialize, Serialize, Default)]`
`50`	`50`	`pub struct PathsConfig {`
`51`	`51`	`pub cache_dir: Option<String>,`
`52`		`- pub state_dir: Option<String>,`
`53`	`52`	`}`
`54`	`53`
`55`	`54`	`impl AppConfig {`
Original file line number	Diff line number	Diff line change
`@@ -147,10 +147,7 @@ impl LazyTableProvider {`
`147`	`147`	`.get_table(self.connection_id, &self.schema_name, &self.table_name)`
`148`	`148`	`.await`
`149`	`149`	`{`
`150`		`- let _ = self`
`151`		`- .catalog`
`152`		`- .update_table_sync(info.id, &parquet_url, "")`
`153`		`- .await;`
	`150`	`+ let _ = self.catalog.update_table_sync(info.id, &parquet_url).await;`
`154`	`151`	`}`
`155`	`152`
`156`	`153`	`Ok(parquet_url)`