fix: Align SqlTable::supports_filters_pushdown with scan_to_sql by using Unparser (#637)

Author: sgrebnov

core/src/duckdb.rs

@@ -356,6 +356,33 @@ impl DuckDBTableProviderFactory {
 
         Ok(pool)
     }
+
+    /// Drop the cached pool entry for `key` if any, returning the previously
+    /// cached pool. Subsequent calls to `get_or_init_*` for the same key will
+    /// build a fresh pool.
+    ///
+    /// This is intended for callers that replace the underlying database file
+    /// out-of-band (for example, after restoring it from a snapshot in object
+    /// storage). Existing connections held by other clones of the previously
+    /// returned pool keep operating against the file descriptor they opened;
+    /// once they are dropped the OS releases the prior inode. New providers
+    /// built after invalidation will open the file fresh and observe the
+    /// replacement contents.
+    pub async fn invalidate_instance(&self, key: &DbInstanceKey) -> Option<DuckDbConnectionPool> {
+        self.instances.lock().await.remove(key)
+    }
+
+    /// Drop the cached pool entry for the file at `path` if any.
+    ///
+    /// Convenience wrapper over [`Self::invalidate_instance`] for the common
+    /// file-mode case.
+    pub async fn invalidate_file_instance(
+        &self,
+        path: impl Into<Arc<str>>,
+    ) -> Option<DuckDbConnectionPool> {
+        self.invalidate_instance(&DbInstanceKey::file(path.into()))
+            .await
+    }
 }
 
 type DynDuckDbConnectionPool = dyn DbConnectionPool<r2d2::PooledConnection<DuckdbConnectionManager>, DuckDBParameter>
@@ -779,6 +806,45 @@ pub(crate) mod tests {
     use std::collections::HashMap;
     use std::sync::Arc;
 
+    #[tokio::test]
+    async fn invalidate_instance_drops_cached_pool() {
+        let factory = DuckDBTableProviderFactory::new(duckdb::AccessMode::ReadWrite);
+        let options = HashMap::new();
+
+        // First call populates the cache.
+        let _pool1 = factory
+            .get_or_init_memory_instance(&options)
+            .await
+            .expect("first init");
+        assert_eq!(factory.instances.lock().await.len(), 1);
+
+        // Second call (without invalidation) returns the cached pool clone
+        // without growing the registry.
+        let _pool2 = factory
+            .get_or_init_memory_instance(&options)
+            .await
+            .expect("cached init");
+        assert_eq!(factory.instances.lock().await.len(), 1);
+
+        // Invalidate; entry is evicted and returned.
+        let evicted = factory.invalidate_instance(&DbInstanceKey::memory()).await;
+        assert!(evicted.is_some(), "invalidate returns evicted pool");
+        assert_eq!(factory.instances.lock().await.len(), 0);
+
+        // Re-invalidating a missing key is a no-op.
+        assert!(factory
+            .invalidate_instance(&DbInstanceKey::file("never-cached".into()))
+            .await
+            .is_none());
+
+        // Next get_or_init repopulates the cache.
+        let _pool3 = factory
+            .get_or_init_memory_instance(&options)
+            .await
+            .expect("reinit after invalidate");
+        assert_eq!(factory.instances.lock().await.len(), 1);
+    }
+
     #[tokio::test]
     async fn test_create_with_memory_limit() {
         let table_name = TableReference::bare("test_table");

core/src/sql/sql_provider_datafusion/expr.rs

@@ -267,20 +267,22 @@ fn handle_cast(cast: &Cast, engine: Option<Engine>, expr: &Expr) -> Result<Strin
     }
 }
 
-// Helper function to check if expression contains subquery
+// Helper function to check if expression contains subquery or outer reference
 use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion};
 
-#[allow(dead_code)]
-pub(super) fn expr_contains_subquery(expr: &Expr) -> datafusion::error::Result<bool> {
-    let mut contains_subquery = false;
+pub(super) fn expr_contains_subquery_or_outer_ref(expr: &Expr) -> datafusion::error::Result<bool> {
+    let mut found = false;
     expr.apply(|expr| match expr {
-        Expr::ScalarSubquery(_) | Expr::InSubquery(_) | Expr::Exists(_) => {
-            contains_subquery = true;
+        Expr::ScalarSubquery(_)
+        | Expr::InSubquery(_)
+        | Expr::Exists(_)
+        | Expr::OuterReferenceColumn(_, _) => {
+            found = true;
             Ok(TreeNodeRecursion::Stop)
         }
         _ => Ok(TreeNodeRecursion::Continue),
     })?;
-    Ok(contains_subquery)
+    Ok(found)
 }
 
 #[cfg(test)]

core/src/sql/sql_provider_datafusion/mod.rs

@@ -293,11 +293,27 @@ impl<T, P> TableProvider for SqlTable<T, P> {
         &self,
         filters: &[&Expr],
     ) -> DataFusionResult<Vec<TableProviderFilterPushDown>> {
+        let dialect = self.dialect_arc();
+        let unparser = Unparser::new(dialect.as_ref());
+
         let filter_push_down: Vec<TableProviderFilterPushDown> = filters
             .iter()
-            .map(|f| match expr::to_sql_with_engine(f, self.engine) {
-                Ok(_) => TableProviderFilterPushDown::Exact,
-                Err(_) => TableProviderFilterPushDown::Unsupported,
+            .map(|f| {
+                // Expressions containing subqueries or outer references must not be
+                // pushed down. Subqueries are handled at the plan level and may
+                // reference tables in other databases not accessible from this
+                // connection. Outer references refer to columns from an enclosing
+                // query that the table provider cannot resolve.
+                if expr::expr_contains_subquery_or_outer_ref(f).unwrap_or(true) {
+                    return TableProviderFilterPushDown::Unsupported;
+                }
+                // Use the same Unparser that scan_to_sql() uses for actual SQL
+                // generation, so the capability check matches what can really
+                // be converted to SQL for the target dialect.
+                match unparser.expr_to_sql(f) {
+                    Ok(_) => TableProviderFilterPushDown::Exact,
+                    Err(_) => TableProviderFilterPushDown::Unsupported,
+                }
             })
             .collect();
 

core/src/sqlite.rs

@@ -246,6 +246,30 @@ impl SqliteTableProviderFactory {
 
         Ok(pool)
     }
+
+    /// Drop the cached pool entry for `key` if any, returning the previously
+    /// cached pool. Subsequent calls to [`Self::get_or_init_instance`] for
+    /// the same key will build a fresh pool.
+    ///
+    /// This is intended for callers that replace the underlying database file
+    /// out-of-band (for example, after restoring it from a snapshot in object
+    /// storage). See [`crate::duckdb::DuckDBTableProviderFactory::invalidate_instance`]
+    /// for the semantics around in-flight connections.
+    pub async fn invalidate_instance(&self, key: &DbInstanceKey) -> Option<SqliteConnectionPool> {
+        self.instances.lock().await.remove(key)
+    }
+
+    /// Drop the cached pool entry for the file at `db_path` if any.
+    ///
+    /// Convenience wrapper over [`Self::invalidate_instance`] for the common
+    /// file-mode case.
+    pub async fn invalidate_file_instance(
+        &self,
+        db_path: impl Into<Arc<str>>,
+    ) -> Option<SqliteConnectionPool> {
+        self.invalidate_instance(&DbInstanceKey::file(db_path.into()))
+            .await
+    }
 }
 
 impl Default for SqliteTableProviderFactory {
@@ -1511,6 +1535,52 @@ pub(crate) mod tests {
 
     use super::*;
 
+    #[tokio::test]
+    async fn invalidate_instance_drops_cached_pool() {
+        let factory = SqliteTableProviderFactory::new();
+        let path: Arc<str> = Arc::from("/tmp/spice-tp-invalidate-test.db");
+        // Best-effort cleanup of leftover files from prior runs.
+        let _ = std::fs::remove_file(path.as_ref());
+
+        let _pool = factory
+            .get_or_init_instance(
+                Arc::clone(&path),
+                Mode::File,
+                std::time::Duration::from_millis(5_000),
+            )
+            .await
+            .expect("init file pool");
+        assert_eq!(factory.instances.lock().await.len(), 1);
+
+        let evicted = factory
+            .invalidate_instance(&DbInstanceKey::file(Arc::clone(&path)))
+            .await;
+        assert!(evicted.is_some(), "invalidate returns evicted pool");
+        assert_eq!(factory.instances.lock().await.len(), 0);
+
+        // Convenience wrapper: re-init then invalidate via path.
+        let _pool2 = factory
+            .get_or_init_instance(
+                Arc::clone(&path),
+                Mode::File,
+                std::time::Duration::from_millis(5_000),
+            )
+            .await
+            .expect("reinit file pool");
+        assert_eq!(factory.instances.lock().await.len(), 1);
+        let evicted2 = factory.invalidate_file_instance(Arc::clone(&path)).await;
+        assert!(evicted2.is_some());
+        assert_eq!(factory.instances.lock().await.len(), 0);
+
+        // Missing key is a no-op.
+        assert!(factory
+            .invalidate_instance(&DbInstanceKey::file("never-cached".into()))
+            .await
+            .is_none());
+
+        let _ = std::fs::remove_file(path.as_ref());
+    }
+
     #[tokio::test]
     async fn test_sqlite_table_creation_with_indexes() {
         let schema = Arc::new(Schema::new(vec![