diff --git a/Cargo.lock b/Cargo.lock
index b8ba0d5898..a316c7b8d6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -73,6 +73,12 @@ version = "1.0.55"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "159bb86af3a200e19a068f4224eae4c8bb2d0fa054c7e5d1cacd5cef95e684cd"
 
+[[package]]
+name = "arrayref"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4c527152e37cf757a3f78aae5a06fbeefdb07ccc535c980a3208ee3060dd544"
+
 [[package]]
 name = "arrayvec"
 version = "0.4.12"
@@ -82,6 +88,12 @@ dependencies = [
  "nodrop",
 ]
 
+[[package]]
+name = "arrayvec"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8da52d66c7071e2e3fa2a1e5c6d088fec47b593032b254f5e980de8ea54454d6"
+
 [[package]]
 name = "arrow"
 version = "9.1.0"
@@ -381,6 +393,29 @@ version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
+[[package]]
+name = "blake2"
+version = "0.10.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9cf849ee05b2ee5fba5e36f97ff8ec2533916700fc0758d40d92136a42f3388"
+dependencies = [
+ "digest 0.10.3",
+]
+
+[[package]]
+name = "blake3"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a08e53fc5a564bb15bfe6fae56bd71522205f1f91893f9c0116edad6496c183f"
+dependencies = [
+ "arrayref",
+ "arrayvec 0.7.2",
+ "cc",
+ "cfg-if",
+ "constant_time_eq",
+ "digest 0.10.3",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.7.3"
@@ -720,6 +755,12 @@ dependencies = [
  "workspace-hack",
 ]
 
+[[package]]
+name = "constant_time_eq"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "245097e9a4535ee1e3e3931fcfcd55a796a44c643e8596ff6566d68f09b87bbc"
+
 [[package]]
 name = "core-foundation"
 version = "0.9.3"
@@ -1040,7 +1081,7 @@ dependencies = [
 [[package]]
 name = "datafusion"
 version = "7.0.0"
-source = "git+https://github.com/apache/arrow-datafusion.git?rev=81e76edc330067332ac9a3972525ecbe92953267#81e76edc330067332ac9a3972525ecbe92953267"
+source = "git+https://github.com/apache/arrow-datafusion.git?rev=3261e2bcdfa13e460d7a11a64dd72f272c7d872b#3261e2bcdfa13e460d7a11a64dd72f272c7d872b"
 dependencies = [
  "ahash",
  "arrow",
@@ -1048,6 +1089,7 @@ dependencies = [
  "chrono",
  "datafusion-common",
  "datafusion-expr",
+ "datafusion-physical-expr",
  "futures",
  "hashbrown 0.12.0",
  "lazy_static",
@@ -1069,7 +1111,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common"
 version = "7.0.0"
-source = "git+https://github.com/apache/arrow-datafusion.git?rev=81e76edc330067332ac9a3972525ecbe92953267#81e76edc330067332ac9a3972525ecbe92953267"
+source = "git+https://github.com/apache/arrow-datafusion.git?rev=3261e2bcdfa13e460d7a11a64dd72f272c7d872b#3261e2bcdfa13e460d7a11a64dd72f272c7d872b"
 dependencies = [
  "arrow",
  "ordered-float 2.10.0",
@@ -1080,7 +1122,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr"
 version = "7.0.0"
-source = "git+https://github.com/apache/arrow-datafusion.git?rev=81e76edc330067332ac9a3972525ecbe92953267#81e76edc330067332ac9a3972525ecbe92953267"
+source = "git+https://github.com/apache/arrow-datafusion.git?rev=3261e2bcdfa13e460d7a11a64dd72f272c7d872b#3261e2bcdfa13e460d7a11a64dd72f272c7d872b"
 dependencies = [
  "ahash",
  "arrow",
@@ -1088,6 +1130,29 @@ dependencies = [
  "sqlparser",
 ]
 
+[[package]]
+name = "datafusion-physical-expr"
+version = "7.0.0"
+source = "git+https://github.com/apache/arrow-datafusion.git?rev=3261e2bcdfa13e460d7a11a64dd72f272c7d872b#3261e2bcdfa13e460d7a11a64dd72f272c7d872b"
+dependencies = [
+ "ahash",
+ "arrow",
+ "blake2",
+ "blake3",
+ "chrono",
+ "datafusion-common",
+ "datafusion-expr",
+ "hashbrown 0.12.0",
+ "lazy_static",
+ "md-5 0.10.1",
+ "ordered-float 2.10.0",
+ "paste",
+ "rand",
+ "regex",
+ "sha2 0.10.2",
+ "unicode-segmentation",
+]
+
 [[package]]
 name = "datafusion_util"
 version = "0.1.0"
@@ -2909,6 +2974,15 @@ dependencies = [
  "opaque-debug 0.3.0",
 ]
 
+[[package]]
+name = "md-5"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "658646b21e0b72f7866c7038ab086d3d5e1cd6271f060fd37defb241949d0582"
+dependencies = [
+ "digest 0.10.3",
+]
+
 [[package]]
 name = "md5"
 version = "0.7.0"
@@ -3246,7 +3320,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bafe4179722c2894288ee77a9f044f02811c86af699344c498b0840c698a2465"
 dependencies = [
- "arrayvec",
+ "arrayvec 0.4.12",
  "itoa 0.4.8",
 ]
 
@@ -4570,7 +4644,7 @@ dependencies = [
  "http",
  "hyper",
  "log",
- "md-5",
+ "md-5 0.9.1",
  "percent-encoding",
  "pin-project-lite",
  "rusoto_credential",
@@ -5236,7 +5310,7 @@ dependencies = [
  "itoa 1.0.1",
  "libc",
  "log",
- "md-5",
+ "md-5 0.9.1",
  "memchr",
  "once_cell",
  "paste",
@@ -6450,6 +6524,7 @@ dependencies = [
  "futures-sink",
  "futures-task",
  "futures-util",
+ "generic-array 0.14.5",
  "getrandom",
  "hashbrown 0.11.2",
  "hashbrown 0.12.0",
diff --git a/datafusion/Cargo.toml b/datafusion/Cargo.toml
index 42c89a2831..95bc520c2c 100644
--- a/datafusion/Cargo.toml
+++ b/datafusion/Cargo.toml
@@ -9,5 +9,5 @@ description = "Re-exports datafusion at a specific version"
 
 # Rename to workaround doctest bug
 # Turn off optional datafusion features (e.g. don't get support for crypo functions or avro)
-upstream = { git = "https://github.com/apache/arrow-datafusion.git", rev="81e76edc330067332ac9a3972525ecbe92953267", default-features = false, package = "datafusion" }
+upstream = { git = "https://github.com/apache/arrow-datafusion.git", rev="3261e2bcdfa13e460d7a11a64dd72f272c7d872b", default-features = false, package = "datafusion" }
 workspace-hack = { path = "../workspace-hack"}
diff --git a/parquet_file/src/storage.rs b/parquet_file/src/storage.rs
index 92be79aaf6..cb2a6c9f96 100644
--- a/parquet_file/src/storage.rs
+++ b/parquet_file/src/storage.rs
@@ -193,7 +193,7 @@ impl Storage {
     /// made simpler by using a plain `Iter` rather than a `Stream`.
     async fn record_batches_to_parquet_bytes(
         &self,
-        mut stream: impl Stream<Item = ArrowResult<RecordBatch>> + Send + Sync + Unpin,
+        mut stream: impl Stream<Item = ArrowResult<RecordBatch>> + Send + Unpin,
         schema: SchemaRef,
         metadata_bytes: &[u8],
     ) -> Result<Vec<u8>> {
diff --git a/predicate/src/rewrite.rs b/predicate/src/rewrite.rs
index 5fe8cd02da..f57ed53df2 100644
--- a/predicate/src/rewrite.rs
+++ b/predicate/src/rewrite.rs
@@ -76,6 +76,7 @@ fn is_comparison(op: Operator) -> bool {
     // added to the enum we will have to update this `match`
     match op {
         Operator::BitwiseAnd => false,
+        Operator::BitwiseOr => false,
         Operator::Eq => true,
         Operator::NotEq => true,
         Operator::Lt => true,
diff --git a/query/src/func/selectors.rs b/query/src/func/selectors.rs
index 6f811aa00e..f6dbc355ec 100644
--- a/query/src/func/selectors.rs
+++ b/query/src/func/selectors.rs
@@ -20,7 +20,7 @@ use arrow::{array::ArrayRef, datatypes::DataType};
 use datafusion::{
     error::{DataFusionError, Result as DataFusionResult},
     physical_plan::{
-        functions::{ReturnTypeFunction, Signature, Volatility},
+        functions::{Signature, Volatility},
         udaf::AggregateUDF,
         Accumulator,
     },
@@ -195,6 +195,8 @@ impl SelectorOutput {
     }
 }
 
+type ReturnTypeFunction = Arc<dyn Fn(&[DataType]) -> DataFusionResult<Arc<DataType>> + Send + Sync>;
+
 /// Factory function for creating the UDA function for DataFusion
 fn make_uda<SELECTOR>(name: &'static str, output: SelectorOutput) -> AggregateUDF
 where
diff --git a/workspace-hack/Cargo.toml b/workspace-hack/Cargo.toml
index 5d72404ebf..99263c3f05 100644
--- a/workspace-hack/Cargo.toml
+++ b/workspace-hack/Cargo.toml
@@ -27,6 +27,7 @@ futures-core = { version = "0.3", features = ["alloc", "std"] }
 futures-sink = { version = "0.3", features = ["alloc", "std"] }
 futures-task = { version = "0.3", default-features = false, features = ["alloc", "std"] }
 futures-util = { version = "0.3", features = ["alloc", "async-await", "async-await-macro", "channel", "futures-channel", "futures-io", "futures-macro", "futures-sink", "io", "memchr", "sink", "slab", "std"] }
+generic-array = { version = "0.14", default-features = false, features = ["more_lengths"] }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 hashbrown-a6292c17cd707f01 = { package = "hashbrown", version = "0.11", features = ["ahash", "inline-more", "raw"] }
 hashbrown-5ef9efb8ec2df382 = { package = "hashbrown", version = "0.12", features = ["ahash", "inline-more", "raw"] }
@@ -80,6 +81,7 @@ futures-core = { version = "0.3", features = ["alloc", "std"] }
 futures-sink = { version = "0.3", features = ["alloc", "std"] }
 futures-task = { version = "0.3", default-features = false, features = ["alloc", "std"] }
 futures-util = { version = "0.3", features = ["alloc", "async-await", "async-await-macro", "channel", "futures-channel", "futures-io", "futures-macro", "futures-sink", "io", "memchr", "sink", "slab", "std"] }
+generic-array = { version = "0.14", default-features = false, features = ["more_lengths"] }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 hashbrown-a6292c17cd707f01 = { package = "hashbrown", version = "0.11", features = ["ahash", "inline-more", "raw"] }
 indexmap = { version = "1", default-features = false, features = ["std"] }