fix(http): reject OR across different HTTP filter columns (spiceai#10625)

The HTTP connector's filter pushdown used the same PartitionAccumulator
for both AND and OR operators, so an OR across different virtual filter
columns (request_path, request_query, request_body, request_headers)
was silently rewritten as a cross product (AND), causing the connector
to issue a single combined HTTP request instead of two separate ones.
DataFusion's Inexact pushdown then re-applied the predicate and the
resulting row often passed, masking the bug while the upstream API
received a request the user never wrote.

Reject the cross-column OR case at extract_partitions time with a
clear error pointing the user to IN (...) or separate queries / UNION
ALL. OR within a single column continues to work (it's how DataFusion
sometimes lowers short IN lists).

Fixes spiceai#10623

Author: phillipleblanc

crates/data_components/src/http/provider.rs

@@ -2559,14 +2559,67 @@ impl HttpTableProvider {
     ) -> Result<()> {
         match expr.op {
             Operator::Eq => self.handle_equality_expr(expr, accumulator),
-            Operator::Or | Operator::And => {
+            Operator::And => {
+                self.extract_filter_values(expr.left.as_ref(), accumulator)?;
+                self.extract_filter_values(expr.right.as_ref(), accumulator)
+            }
+            Operator::Or => {
+                // OR within a single HTTP virtual filter column is treated as
+                // an IN list (alternative values). OR across different
+                // columns would be silently rewritten as a cross product
+                // (AND) by the partition accumulator, causing the connector
+                // to issue combined HTTP requests instead of separate ones.
+                // Reject the cross-column case explicitly. (Note: SQL
+                // `IN (...)` is sometimes pre-rewritten by DataFusion into a
+                // chain of OR-of-equality, which is why same-column OR must
+                // still be accepted.)
+                let mut columns = HashSet::new();
+                Self::collect_http_filter_columns(expr.left.as_ref(), &mut columns);
+                Self::collect_http_filter_columns(expr.right.as_ref(), &mut columns);
+                if columns.len() > 1 {
+                    let mut names: Vec<&str> = columns.into_iter().collect();
+                    names.sort_unstable();
+                    return Err(Error::FilterRejected {
+                        message: format!(
+                            "OR across different HTTP filter columns ({}) is not supported because the connector would otherwise issue combined HTTP requests instead of separate ones. Use IN (...) to enumerate values on a single column, or run separate queries (e.g. UNION ALL) for alternative requests.",
+                            names.join(", ")
+                        ),
+                    });
+                }
                 self.extract_filter_values(expr.left.as_ref(), accumulator)?;
                 self.extract_filter_values(expr.right.as_ref(), accumulator)
             }
             _ => Ok(()),
         }
     }
 
+    /// Walk an expression tree and collect the names of any HTTP virtual
+    /// filter columns (`request_path`, `request_query`, `request_body`,
+    /// `request_headers`) referenced anywhere inside it.
+    fn collect_http_filter_columns(expr: &Expr, columns: &mut HashSet<&'static str>) {
+        match expr {
+            Expr::BinaryExpr(BinaryExpr { left, right, .. }) => {
+                Self::collect_http_filter_columns(left.as_ref(), columns);
+                Self::collect_http_filter_columns(right.as_ref(), columns);
+            }
+            Expr::InList(in_list) => {
+                Self::collect_http_filter_columns(in_list.expr.as_ref(), columns);
+            }
+            Expr::Column(column) => {
+                if let Some(static_name) = match column.name.as_str() {
+                    "request_path" => Some("request_path"),
+                    "request_query" => Some("request_query"),
+                    "request_body" => Some("request_body"),
+                    "request_headers" => Some("request_headers"),
+                    _ => None,
+                } {
+                    columns.insert(static_name);
+                }
+            }
+            _ => {}
+        }
+    }
+
     fn handle_equality_expr(
         &self,
         expr: &BinaryExpr,
@@ -3228,6 +3281,82 @@ mod tests {
         assert!(partitions.contains(&(Some("/api/v2".to_string()), None, None, None)));
     }
 
+    #[test]
+    fn test_extract_partitions_with_or_across_columns_is_rejected() {
+        let provider = base_provider()
+            .with_allowed_paths(vec!["/a".to_string()])
+            .expect("allowed paths")
+            .enable_query_filters(64);
+        // request_path = '/a' OR request_query = 'b=1'
+        let filters = vec![Expr::BinaryExpr(BinaryExpr {
+            left: Box::new(Expr::BinaryExpr(BinaryExpr {
+                left: Box::new(Expr::Column(Column::from_name("request_path"))),
+                op: Operator::Eq,
+                right: Box::new(Expr::Literal(
+                    ScalarValue::Utf8(Some("/a".to_string())),
+                    None,
+                )),
+            })),
+            op: Operator::Or,
+            right: Box::new(Expr::BinaryExpr(BinaryExpr {
+                left: Box::new(Expr::Column(Column::from_name("request_query"))),
+                op: Operator::Eq,
+                right: Box::new(Expr::Literal(
+                    ScalarValue::Utf8(Some("b=1".to_string())),
+                    None,
+                )),
+            })),
+        })];
+
+        let err = provider
+            .extract_partitions(&filters)
+            .expect_err("OR across HTTP virtual columns must be rejected");
+        let message = err.to_string();
+        assert!(
+            message.contains("OR across different HTTP filter columns"),
+            "unexpected error message: {message}"
+        );
+        assert!(message.contains("request_path"));
+        assert!(message.contains("request_query"));
+    }
+
+    #[test]
+    fn test_extract_partitions_with_or_across_columns_nested_is_rejected() {
+        let provider = base_provider()
+            .with_allowed_paths(vec!["/a".to_string()])
+            .expect("allowed paths")
+            .enable_header_filters(DEFAULT_MAX_HEADERS_LENGTH, vec!["x-sandbox-id".to_string()])
+            .expect("enable header filters");
+        // request_path = '/a' OR request_headers IN ('{"x-sandbox-id":"a"}')
+        let filters = vec![Expr::BinaryExpr(BinaryExpr {
+            left: Box::new(Expr::BinaryExpr(BinaryExpr {
+                left: Box::new(Expr::Column(Column::from_name("request_path"))),
+                op: Operator::Eq,
+                right: Box::new(Expr::Literal(
+                    ScalarValue::Utf8(Some("/a".to_string())),
+                    None,
+                )),
+            })),
+            op: Operator::Or,
+            right: Box::new(Expr::InList(InList {
+                expr: Box::new(Expr::Column(Column::from_name("request_headers"))),
+                list: vec![Expr::Literal(
+                    ScalarValue::Utf8(Some(r#"{"x-sandbox-id":"a"}"#.to_string())),
+                    None,
+                )],
+                negated: false,
+            })),
+        })];
+
+        let err = provider
+            .extract_partitions(&filters)
+            .expect_err("OR across HTTP virtual columns must be rejected");
+        assert!(
+            err.to_string()
+                .contains("OR across different HTTP filter columns")
+        );
+    }
+
     #[test]
     fn test_extract_partitions_with_combined_filters() {
         let provider = base_provider()

docs/examples/http_refresh_sql_example.md

@@ -124,7 +124,7 @@ The HTTP connector's `refresh_sql` supports the following filter expressions on
 
 1. **Equality (`=`)**: `WHERE request_path = '/api/users'`
 2. **IN Lists**: `WHERE request_path IN ('/api/users', '/api/posts')`
-3. **OR expressions**: `WHERE request_path = '/api/users' OR request_path = '/api/posts'`
+3. **OR expressions** (single column only): `WHERE request_path = '/api/users' OR request_path = '/api/posts'`. OR across **different** filter columns is not supported — use separate queries (e.g. `UNION ALL`).
 4. **AND expressions**: `WHERE request_path = '/api/users' AND request_query = 'limit=10'`
 5. **POST requests**: `WHERE request_body = '{"key": "value"}'` (triggers POST with `http_post_content_type`)
 6. **Dynamic headers**: `WHERE request_headers IN ('{"x-sandbox-id":"sandbox-1"}', '{"x-sandbox-id":"sandbox-2"}')`
@@ -153,7 +153,7 @@ The HTTP connector's `refresh_sql` supports the following filter expressions on
 ## Performance Considerations
 
 - **Caching**: The HTTP connector respects `Cache-Control` headers and caches responses when `max-age` is set.
-- **Parallel Execution**: Multiple endpoints (from IN lists or OR expressions) are fetched in parallel.
+- **Parallel Execution**: Multiple endpoints (from IN lists or single-column OR expressions) are fetched in parallel.
 - **Filter Selectivity**: Use specific filters to minimize unnecessary HTTP requests.
 - **Partition Limits**: Use `max_request_partitions` to cap the number of HTTP requests created from cross-product filters.