Update DataFusion to version 49.0.2 (#11291)

### What

This PR is an update to DataFusion dependency to version 49.0.2.

With this PR, python users must likewise update their
`datafusion-python` versions to 49.0.2.

---------

Co-authored-by: Clement Rey <[email protected]>

Author: timsaucer

Cargo.lock

@@ -220,10 +220,10 @@ convert_case = "0.6"
 criterion = "0.5"
 cros-codecs = "0.0.6"
 crossbeam = "0.8"
-datafusion = { version = "47", default-features = false, features = [
+datafusion = { version = "49.0.2", default-features = false, features = [
   "nested_expressions",
 ] }
-datafusion-ffi = "47"
+datafusion-ffi = "49.0.2"
 directories = "6"
 document-features = "0.2.8"
 econtext = "0.2" # Prints error contexts on crashes

Cargo.toml

@@ -200,8 +200,8 @@ impl DataframeQueryTableProvider {
             }
             Expr::BinaryExpr(binary) => {
                 if binary.op == Operator::NotEq
-                    && let (Expr::Column(col), Expr::Literal(sv))
-                    | (Expr::Literal(sv), Expr::Column(col)) =
+                    && let (Expr::Column(col), Expr::Literal(sv, _))
+                    | (Expr::Literal(sv, _), Expr::Column(col)) =
                         (binary.left.as_ref(), binary.right.as_ref())
                     && sv.is_null()
                 {

crates/store/re_datafusion/src/dataframe_query_common.rs

@@ -227,13 +227,16 @@ impl PartitionStreamExec {
                     SortOptions::new(false, true),
                 ));
             }
-            vec![LexOrdering::new(physical_ordering)]
+            vec![
+                LexOrdering::new(physical_ordering)
+                    .expect("LexOrdering should return Some since input is not empty"),
+            ]
         } else {
             vec![]
         };
 
         let eq_properties =
-            EquivalenceProperties::new_with_orderings(Arc::clone(&projected_schema), &orderings);
+            EquivalenceProperties::new_with_orderings(Arc::clone(&projected_schema), orderings);
 
         let partition_in_output_schema = projection.map(|p| p.contains(&0)).unwrap_or(false);
 

crates/store/re_datafusion/src/dataframe_query_provider.rs

@@ -224,10 +224,13 @@ impl PartitionStreamExec {
             ));
         }
 
-        let orderings = vec![LexOrdering::new(physical_ordering)];
+        let orderings = vec![
+            LexOrdering::new(physical_ordering)
+                .expect("LexOrdering should return Some when non-empty vec is passed"),
+        ];
 
         let eq_properties =
-            EquivalenceProperties::new_with_orderings(Arc::clone(&projected_schema), &orderings);
+            EquivalenceProperties::new_with_orderings(Arc::clone(&projected_schema), orderings);
 
         let partition_in_output_schema = projection.map(|p| p.contains(&0)).unwrap_or(false);
 

crates/store/re_datafusion/src/dataframe_query_provider_wasm.rs

@@ -4,12 +4,13 @@ use std::sync::Arc;
 
 use arrow::array::{Array as _, ArrayRef, BooleanArray, ListArray, as_list_array};
 use arrow::datatypes::{DataType, Field};
+use datafusion::common::ExprSchema as _;
 use datafusion::common::{DFSchema, Result as DataFusionResult, exec_err};
 use datafusion::logical_expr::{
     ArrayFunctionArgument, ArrayFunctionSignature, ColumnarValue, ScalarFunctionArgs, ScalarUDF,
     ScalarUDFImpl, Signature, TypeSignature, Volatility,
 };
-use datafusion::prelude::{Column, Expr, array_to_string, col, lit, lower};
+use datafusion::prelude::{Column, Expr, array_to_string, col, contains, lit, lower};
 
 use super::{NonNullableBooleanFilter, NullableBooleanFilter};
 
@@ -297,7 +298,7 @@ impl FilterOperation {
                     }
                 };
 
-                Ok(contains_patch(lower(operand), lower(lit(query_string))))
+                Ok(contains(lower(operand), lower(lit(query_string))))
             }
 
             Self::NullableBoolean(boolean_filter) => {
@@ -480,7 +481,7 @@ impl ScalarUDFImpl for FilterOperationUdf {
         }
     }
 
-    fn invoke_with_args(&self, args: ScalarFunctionArgs<'_>) -> DataFusionResult<ColumnarValue> {
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DataFusionResult<ColumnarValue> {
         let ColumnarValue::Array(input_array) = &args.args[0] else {
             return exec_err!("FilterOperation expected array inputs, not scalar values");
         };
@@ -512,14 +513,3 @@ impl ScalarUDFImpl for FilterOperationUdf {
         }
     }
 }
-
-// TODO(ab): this is a workaround for https://github.com/apache/datafusion/pull/16046. Next time we
-// update datafusion, this should break compilation. Remove this function and replace
-// `contains_patch` by `datafusion::prelude::contains` in the method above.
-fn contains_patch(arg1: Expr, arg2: Expr) -> Expr {
-    // make sure we break compilation when we update datafusion
-    #[cfg(debug_assertions)]
-    let _ = datafusion::prelude::contains();
-
-    datafusion::functions::string::contains().call(<[_]>::into_vec(Box::new([arg1, arg2])))
-}

crates/viewer/re_dataframe_ui/src/filters/filter.rs

@@ -29,7 +29,6 @@ all-features = true
 [advisories]
 version = 2
 ignore = [
-  "RUSTSEC-2024-0384", # Waiting for https://github.com/console-rs/indicatif/pull/666
   "RUSTSEC-2024-0436", # https://rustsec.org/advisories/RUSTSEC-2024-0436 - paste is unmaintained - https://github.com/dtolnay/paste
   "RUSTSEC-2024-0014", # https://rustsec.org/advisories/RUSTSEC-2024-0014 - generational-arena is unmaintained
 ]
@@ -54,6 +53,7 @@ deny = [
 skip = [
   { name = "base64" },              # Too popular
   { name = "block2" },              # Old version via rfd
+  { name = "bzip2" },               # Remove after https://github.com/apache/datafusion/pull/17509 closes
   { name = "cargo-platform" },      # Older version used by ply-rs. It's build-time only!
   { name = "cargo_metadata" },      # Older version used by ply-rs. It's small, and it's build-time only!
   { name = "core-foundation" },     # Currently, e.g. `webbrowser` and `winit` use different versions.
@@ -64,6 +64,7 @@ skip = [
   { name = "objc2-foundation" },    # `accesskit_macos` uses a different version than `arboard`
   { name = "objc2" },               # `accesskit_macos` uses a different version than `arboard`
   { name = "ordered-float" },       # Old version being used by parquet, but super small!
+  { name = "petgraph" },            # Remove after next release due to https://github.com/tokio-rs/prost/pull/1327
   { name = "pollster" },            # rfd is still on 0.3
   { name = "pulldown-cmark" },      # Build-dependency via `ply-rs` (!). TODO(emilk): use a better crate for .ply parsing
   { name = "redox_syscall" },       # Plenty of versions in the wild
@@ -97,8 +98,8 @@ allow = [
   "OFL-1.1",                        # https://spdx.org/licenses/OFL-1.1.html
   "Ubuntu-font-1.0",                # https://ubuntu.com/legal/font-licence
   "Unicode-3.0",                    # https://www.unicode.org/license.txt
-  "Unicode-DFS-2016",               # https://spdx.org/licenses/Unicode-DFS-2016.html
   "Zlib",                           # https://tldrlegal.com/license/zlib-libpng-license-(zlib)
+  "bzip2-1.0.6",                    # https://github.com/trifectatechfoundation/libbzip2-rs/blob/v0.2.2/COPYING
 ]
 exceptions = []
 
@@ -112,7 +113,6 @@ name = "ring"
 expression = "MIT AND ISC AND OpenSSL"
 license-files = [{ path = "LICENSE", hash = 0xbd0eed23 }]
 
-
 [sources]
 unknown-registry = "deny"
 unknown-git = "deny"

deny.toml

@@ -9,3 +9,9 @@ Use the new `timeout_sec` argument instead.
 For non-blocking, use `timeout_sec=0`.
 Mostly you can just call `.flush()` with no arguments.
 That will block until all writes either finishes or an error occurs (e.g. the gRPC connection is severed).
+
+## Python DataFusion interface: update to 49.0.0
+The DataFusion FFI that we rely on for user defined functions and
+table providers requires users to upgrade their `datafusion-python`
+version to 49.0.0. This only impacts customers who use the
+DataFusion tables provided through the `CatalogClient`.