Make DuckDB database attachments logic more robust

Author: sgrebnov

src/sql/db_connection_pool/dbconnection/duckdbconn.rs

@@ -123,11 +123,29 @@ impl DuckDBAttachments {
     }
 
     /// Attaches the databases to the given connection and sets the search path for the newly attached databases.
+    /// If connection already contains attachments, it will skip the attachments override (including search_path).
     ///
     /// # Errors
     ///
     /// Returns an error if a specific attachment is missing, cannot be attached, search path cannot be set or the connection fails.
     pub fn attach(&self, conn: &Connection) -> Result<()> {
+        
+        // Check if attachments already exist, skip it in this case
+        let mut stmt = conn
+            .prepare("PRAGMA database_list;")
+            .context(DuckDBConnectionSnafu)?;
+        let mut rows = stmt.query([]).context(DuckDBConnectionSnafu)?;
+
+        while let Some(row) = rows.next()? {
+            let ds_name: String = row.get(1)?;
+            if ds_name.starts_with("attachment_") {
+                tracing::debug!(
+                    "Connection contains existing attachments, skipping attachments override",
+                );
+                return Ok(());
+            }
+        }
+
         for (i, db) in self.attachments.iter().enumerate() {
             // check the db file exists
             std::fs::metadata(db.as_ref()).context(UnableToAttachDatabaseSnafu {
@@ -154,7 +172,7 @@ impl DuckDBAttachments {
     pub fn detach(&self, conn: &Connection) -> Result<()> {
         for (i, _) in self.attachments.iter().enumerate() {
             conn.execute(
-                &format!("DETACH {}", Self::get_attachment_name(&self.random_id, i)),
+                &format!("DETACH DATABASE IF EXISTS {}", Self::get_attachment_name(&self.random_id, i)),
                 [],
             )
             .context(DuckDBConnectionSnafu)?;
@@ -320,7 +338,9 @@ impl SyncDbConnection<r2d2::PooledConnection<DuckdbConnectionManager>, DuckDBPar
     ) -> Result<SendableRecordBatchStream> {
         let (batch_tx, mut batch_rx) = tokio::sync::mpsc::channel::<RecordBatch>(4);
 
+        let conn = self.conn.try_clone()?;
         Self::attach(&self.conn, &self.attachments)?;
+ 
         let fetch_schema_sql =
             format!("WITH fetch_schema AS ({sql}) SELECT * FROM fetch_schema LIMIT 0");
         let mut stmt = self
@@ -334,21 +354,15 @@ impl SyncDbConnection<r2d2::PooledConnection<DuckdbConnectionManager>, DuckDBPar
             .boxed()
             .context(super::UnableToGetSchemaSnafu)?;
 
-        Self::detach(&self.conn, &self.attachments)?;
-
         let schema = result.get_schema();
 
         let params = params.iter().map(dyn_clone::clone).collect::<Vec<_>>();
 
-        let conn = self.conn.try_clone()?; // try_clone creates a new connection to the same database
-                                           // this creates a new connection session, requiring resetting the ATTACHments and search_path
         let sql = sql.to_string();
 
         let cloned_schema = schema.clone();
-        let attachments = self.attachments.clone();
 
         let join_handle = tokio::task::spawn_blocking(move || {
-            Self::attach(&conn, &attachments)?; // this attach could happen when we clone the connection, but we can't detach after the thread closes because the connection isn't thread safe
             let mut stmt = conn.prepare(&sql).context(DuckDBQuerySnafu)?;
             let params: &[&dyn ToSql] = &params
                 .iter()
@@ -360,8 +374,6 @@ impl SyncDbConnection<r2d2::PooledConnection<DuckdbConnectionManager>, DuckDBPar
             for i in result {
                 blocking_channel_send(&batch_tx, i)?;
             }
-
-            Self::detach(&conn, &attachments)?;
             Ok::<_, Box<dyn std::error::Error + Send + Sync>>(())
         });