diff --git a/cpp/src/gandiva/CMakeLists.txt b/cpp/src/gandiva/CMakeLists.txt
index aabe4ec8bf70..f2d61a248a5e 100644
--- a/cpp/src/gandiva/CMakeLists.txt
+++ b/cpp/src/gandiva/CMakeLists.txt
@@ -65,6 +65,7 @@ set(SRC_FILES
     engine.cc
     date_utils.cc
     encrypt_utils.cc
+    expr_cse.cc
     expr_decomposer.cc
     expr_validator.cc
     expression.cc
@@ -259,6 +260,7 @@ add_gandiva_test(internals-test
                  annotator_test.cc
                  tree_expr_test.cc
                  encrypt_utils_test.cc
+                 expr_cse_test.cc
                  expr_decomposer_test.cc
                  exported_funcs_registry_test.cc
                  expression_registry_test.cc
diff --git a/cpp/src/gandiva/engine.cc b/cpp/src/gandiva/engine.cc
index 901421c86cb3..4cdc9238f678 100644
--- a/cpp/src/gandiva/engine.cc
+++ b/cpp/src/gandiva/engine.cc
@@ -543,6 +543,10 @@ Status Engine::FinalizeModule() {
   if (!cached_) {
     ARROW_RETURN_NOT_OK(RemoveUnusedFunctions());
 
+    if (conf_->dump_ir()) {
+      unoptimized_module_ir_ = DumpModuleIR(*module_);
+    }
+
     if (optimize_) {
       auto target_analysis = target_machine_->getTargetIRAnalysis();
 // misc passes to allow for inlining, vectorization, ..
@@ -615,4 +619,10 @@ const std::string& Engine::ir() {
   return module_ir_;
 }
 
+const std::string& Engine::unoptimized_ir() {
+  DCHECK(!unoptimized_module_ir_.empty())
+      << "dump_ir in Configuration must be set for dumping IR";
+  return unoptimized_module_ir_;
+}
+
 }  // namespace gandiva
diff --git a/cpp/src/gandiva/engine.h b/cpp/src/gandiva/engine.h
index 20165787cb66..6e8aefc39e0f 100644
--- a/cpp/src/gandiva/engine.h
+++ b/cpp/src/gandiva/engine.h
@@ -87,6 +87,11 @@ class GANDIVA_EXPORT Engine {
   /// Return the generated IR for the module.
   const std::string& ir();
 
+  /// Return the generated IR before the optimizer pipeline runs.
+  const std::string& unoptimized_ir();
+
+  bool has_unoptimized_ir() const { return !unoptimized_module_ir_.empty(); }
+
   /// Load the function IRs that can be accessed in the module.
   Status LoadFunctionIRs();
 
@@ -129,6 +134,7 @@ class GANDIVA_EXPORT Engine {
   bool cached_;
   bool functions_loaded_ = false;
   std::shared_ptr<FunctionRegistry> function_registry_;
+  std::string unoptimized_module_ir_;
   std::string module_ir_;
   // The lifetime of the TargetMachine is shared with LLJIT. This prevents unnecessary
   // duplication of this expensive object.
diff --git a/cpp/src/gandiva/expr_cse.cc b/cpp/src/gandiva/expr_cse.cc
new file mode 100644
index 000000000000..5867947191fd
--- /dev/null
+++ b/cpp/src/gandiva/expr_cse.cc
@@ -0,0 +1,251 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "gandiva/expr_cse.h"
+
+#include <cstddef>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "arrow/util/hash_util.h"
+#include "gandiva/condition.h"
+#include "gandiva/function_registry.h"
+#include "gandiva/function_signature.h"
+#include "gandiva/node.h"
+
+namespace gandiva {
+
+namespace {
+
+enum class NodeKind {
+  kField,
+  kLiteral,
+  kFunction,
+};
+
+struct NodeKey {
+  NodeKind kind;
+  std::string name;
+  std::string type;
+  std::vector<size_t> children;
+
+  bool operator==(const NodeKey& other) const {
+    return kind == other.kind && name == other.name && type == other.type &&
+           children == other.children;
+  }
+};
+
+struct NodeKeyHash {
+  size_t operator()(const NodeKey& key) const {
+    size_t hash = static_cast<size_t>(key.kind);
+    arrow::internal::hash_combine(hash, key.name);
+    arrow::internal::hash_combine(hash, key.type);
+    for (auto child : key.children) {
+      arrow::internal::hash_combine(hash, child);
+    }
+    return hash;
+  }
+};
+
+struct FoldedNode {
+  NodePtr node;
+  size_t id;
+  bool can_eliminate;
+};
+
+struct CanonicalNode {
+  NodePtr node;
+  size_t id;
+};
+
+class CommonSubexpressionFolder {
+ public:
+  explicit CommonSubexpressionFolder(const FunctionRegistry& registry)
+      : registry_(registry) {}
+
+  ExpressionPtr FoldExpression(const ExpressionPtr& expression) {
+    auto folded = Fold(expression->root());
+    if (folded.node == expression->root()) {
+      return expression;
+    }
+    return std::make_shared<Expression>(folded.node, expression->result());
+  }
+
+  ConditionPtr FoldCondition(const ConditionPtr& condition) {
+    auto folded = Fold(condition->root());
+    if (folded.node == condition->root()) {
+      return condition;
+    }
+    return std::make_shared<Condition>(folded.node);
+  }
+
+ private:
+  FoldedNode Fold(const NodePtr& node) {
+    if (node == nullptr) {
+      return Fresh(nullptr);
+    }
+
+    auto cached = folded_nodes_.find(node.get());
+    if (cached != folded_nodes_.end()) {
+      return cached->second;
+    }
+
+    auto folded = FoldUncached(node);
+    folded_nodes_.emplace(node.get(), folded);
+    return folded;
+  }
+
+  FoldedNode FoldUncached(const NodePtr& node) {
+    if (auto field_node = std::dynamic_pointer_cast<FieldNode>(node)) {
+      return Intern({NodeKind::kField, field_node->field()->ToString(), "", {}}, node);
+    }
+
+    if (auto literal_node = std::dynamic_pointer_cast<LiteralNode>(node)) {
+      return Intern({NodeKind::kLiteral, literal_node->ToString(), "", {}}, node);
+    }
+
+    if (auto function_node = std::dynamic_pointer_cast<FunctionNode>(node)) {
+      return FoldFunction(node, *function_node);
+    }
+
+    if (auto boolean_node = std::dynamic_pointer_cast<BooleanNode>(node)) {
+      return FoldBoolean(node, *boolean_node);
+    }
+
+    if (auto if_node = std::dynamic_pointer_cast<IfNode>(node)) {
+      return FoldIf(node, *if_node);
+    }
+
+    // InExpressionNode stores constants in unordered_sets, so keep it opaque in this
+    // conservative pass.
+    return Fresh(node);
+  }
+
+  FoldedNode FoldFunction(const NodePtr& original, const FunctionNode& function_node) {
+    NodeVector children;
+    children.reserve(function_node.children().size());
+
+    std::vector<size_t> child_ids;
+    child_ids.reserve(function_node.children().size());
+
+    bool children_unchanged = true;
+    bool children_can_eliminate = true;
+    for (const auto& child : function_node.children()) {
+      auto folded = Fold(child);
+      children_unchanged = children_unchanged && folded.node == child;
+      children_can_eliminate = children_can_eliminate && folded.can_eliminate;
+      children.push_back(folded.node);
+      child_ids.push_back(folded.id);
+    }
+
+    auto desc = function_node.descriptor();
+    auto return_type = desc->return_type() == NULLPTR ? std::string("untyped")
+                                                      : desc->return_type()->ToString();
+    NodeKey key{NodeKind::kFunction, desc->name(), std::move(return_type),
+                std::move(child_ids)};
+    auto folded_node =
+        children_unchanged
+            ? original
+            : std::make_shared<FunctionNode>(desc->name(), children, desc->return_type());
+    bool can_eliminate = children_can_eliminate && IsFunctionSafe(function_node);
+    return can_eliminate ? Intern(std::move(key), folded_node) : Fresh(folded_node);
+  }
+
+  FoldedNode FoldBoolean(const NodePtr& original, const BooleanNode& boolean_node) {
+    NodeVector folded_children;
+    folded_children.reserve(boolean_node.children().size());
+    bool children_unchanged = true;
+
+    for (const auto& child : boolean_node.children()) {
+      auto folded = Fold(child);
+      children_unchanged = children_unchanged && folded.node == child;
+      folded_children.push_back(std::move(folded.node));
+    }
+
+    auto folded_node =
+        children_unchanged
+            ? original
+            : std::make_shared<BooleanNode>(boolean_node.expr_type(), folded_children);
+    return Fresh(folded_node);
+  }
+
+  FoldedNode FoldIf(const NodePtr& original, const IfNode& if_node) {
+    auto condition = Fold(if_node.condition());
+    auto then_node = Fold(if_node.then_node());
+    auto else_node = Fold(if_node.else_node());
+
+    bool children_unchanged = condition.node == if_node.condition() &&
+                              then_node.node == if_node.then_node() &&
+                              else_node.node == if_node.else_node();
+    auto folded_node = children_unchanged ? original
+                                          : std::make_shared<IfNode>(
+                                                condition.node, then_node.node,
+                                                else_node.node, if_node.return_type());
+
+    return Fresh(folded_node);
+  }
+
+  bool IsFunctionSafe(const FunctionNode& node) const {
+    auto desc = node.descriptor();
+    FunctionSignature signature(desc->name(), desc->params(), desc->return_type());
+    const NativeFunction* native_function = registry_.LookupSignature(signature);
+    if (native_function == nullptr) {
+      return false;
+    }
+    return CanReuseNativeFunction(registry_, *native_function);
+  }
+
+  FoldedNode Intern(NodeKey key, const NodePtr& node) {
+    auto it = canonical_nodes_.find(key);
+    if (it != canonical_nodes_.end()) {
+      return {it->second.node, it->second.id, true};
+    }
+    auto id = next_id_++;
+    canonical_nodes_.emplace(std::move(key), CanonicalNode{node, id});
+    return {node, id, true};
+  }
+
+  FoldedNode Fresh(const NodePtr& node) { return {node, next_id_++, false}; }
+
+  const FunctionRegistry& registry_;
+  std::unordered_map<const Node*, FoldedNode> folded_nodes_;
+  std::unordered_map<NodeKey, CanonicalNode, NodeKeyHash> canonical_nodes_;
+  size_t next_id_ = 1;
+};
+
+}  // namespace
+
+ExpressionVector FoldCommonSubexpressions(const FunctionRegistry& registry,
+                                          const ExpressionVector& expressions) {
+  CommonSubexpressionFolder folder(registry);
+  ExpressionVector folded_expressions;
+  folded_expressions.reserve(expressions.size());
+  for (const auto& expression : expressions) {
+    folded_expressions.push_back(folder.FoldExpression(expression));
+  }
+  return folded_expressions;
+}
+
+ConditionPtr FoldCommonSubexpressions(const FunctionRegistry& registry,
+                                      const ConditionPtr& condition) {
+  CommonSubexpressionFolder folder(registry);
+  return folder.FoldCondition(condition);
+}
+
+}  // namespace gandiva
diff --git a/cpp/src/gandiva/expr_cse.h b/cpp/src/gandiva/expr_cse.h
new file mode 100644
index 000000000000..93ed1e9788e2
--- /dev/null
+++ b/cpp/src/gandiva/expr_cse.h
@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "gandiva/expression.h"
+#include "gandiva/function_registry.h"
+#include "gandiva/gandiva_aliases.h"
+#include "gandiva/native_function.h"
+
+namespace gandiva {
+
+class Condition;
+inline bool CanReuseNativeFunction(const FunctionRegistry& registry,
+                                   const NativeFunction& native_function) {
+  return registry.IsBuiltIn(native_function) &&
+         native_function.result_nullable_type() != kResultNullInternal &&
+         !native_function.NeedsContext() && !native_function.NeedsFunctionHolder() &&
+         !native_function.CanReturnErrors();
+}
+
+ExpressionVector FoldCommonSubexpressions(const FunctionRegistry& registry,
+                                          const ExpressionVector& expressions);
+
+ConditionPtr FoldCommonSubexpressions(const FunctionRegistry& registry,
+                                      const ConditionPtr& condition);
+
+}  // namespace gandiva
diff --git a/cpp/src/gandiva/expr_cse_test.cc b/cpp/src/gandiva/expr_cse_test.cc
new file mode 100644
index 000000000000..fe713c827ff8
--- /dev/null
+++ b/cpp/src/gandiva/expr_cse_test.cc
@@ -0,0 +1,247 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "gandiva/expr_cse.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/testing/gtest_util.h"
+#include "gandiva/function_registry.h"
+#include "gandiva/node.h"
+#include "gandiva/tree_expr_builder.h"
+
+namespace gandiva {
+namespace {
+
+ExpressionPtr MakeExpression(NodePtr root, const std::string& name = "out") {
+  auto return_type = root->return_type();
+  return TreeExprBuilder::MakeExpression(std::move(root),
+                                         arrow::field(name, return_type));
+}
+
+NodePtr MakeBinary(const std::string& name, const FieldPtr& left, const FieldPtr& right,
+                   const DataTypePtr& return_type) {
+  return TreeExprBuilder::MakeFunction(
+      name, {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+      return_type);
+}
+
+void ExpectDuplicateRootsNotFolded(const NodePtr& left, const NodePtr& right) {
+  auto registry = default_function_registry();
+  auto folded = FoldCommonSubexpressions(
+      *registry, {MakeExpression(left, "left"), MakeExpression(right, "right")});
+  ASSERT_EQ(2, folded.size());
+  EXPECT_NE(folded[0]->root(), folded[1]->root());
+}
+
+int32_t VolatileIdentity(int32_t value) { return value; }
+
+TEST(CommonSubexpressionFolderTest, FoldsRepeatedSafeSubtrees) {
+  auto left = arrow::field("left", arrow::int32());
+  auto right = arrow::field("right", arrow::int32());
+  auto root =
+      TreeExprBuilder::MakeFunction("multiply",
+                                    {MakeBinary("add", left, right, arrow::int32()),
+                                     MakeBinary("add", left, right, arrow::int32())},
+                                    arrow::int32());
+
+  auto folded =
+      FoldCommonSubexpressions(*default_function_registry(), {MakeExpression(root)});
+  auto folded_root = std::dynamic_pointer_cast<FunctionNode>(folded[0]->root());
+  ASSERT_NE(nullptr, folded_root);
+  ASSERT_EQ(2, folded_root->children().size());
+  EXPECT_EQ(folded_root->children()[0], folded_root->children()[1]);
+}
+
+TEST(CommonSubexpressionFolderTest, FoldsDeepSharedDagOncePerNode) {
+  auto value = arrow::field("value", arrow::int32());
+  NodePtr root = TreeExprBuilder::MakeField(value);
+  constexpr int kDepth = 24;
+  for (int depth = 0; depth < kDepth; ++depth) {
+    root = TreeExprBuilder::MakeFunction("add", {root, root}, arrow::int32());
+  }
+
+  auto folded =
+      FoldCommonSubexpressions(*default_function_registry(), {MakeExpression(root)});
+  auto current = folded[0]->root();
+  for (int depth = 0; depth < kDepth; ++depth) {
+    auto function = std::dynamic_pointer_cast<FunctionNode>(current);
+    ASSERT_NE(nullptr, function);
+    ASSERT_EQ(2, function->children().size());
+    EXPECT_EQ(function->children()[0], function->children()[1]);
+    current = function->children()[0];
+  }
+  EXPECT_NE(nullptr, std::dynamic_pointer_cast<FieldNode>(current));
+}
+
+TEST(CommonSubexpressionFolderTest, FoldsAcrossMultipleOutputExpressions) {
+  auto left = arrow::field("left", arrow::int32());
+  auto right = arrow::field("right", arrow::int32());
+  auto folded = FoldCommonSubexpressions(
+      *default_function_registry(),
+      {MakeExpression(MakeBinary("add", left, right, arrow::int32()), "out1"),
+       MakeExpression(MakeBinary("add", left, right, arrow::int32()), "out2")});
+
+  ASSERT_EQ(2, folded.size());
+  EXPECT_EQ(folded[0]->root(), folded[1]->root());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotApplyBooleanAlgebra) {
+  auto left = arrow::field("left", arrow::boolean());
+  auto right = arrow::field("right", arrow::boolean());
+  auto make_and = [&] {
+    return TreeExprBuilder::MakeAnd(
+        {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)});
+  };
+  auto root = TreeExprBuilder::MakeOr({make_and(), make_and()});
+
+  auto folded =
+      FoldCommonSubexpressions(*default_function_registry(), {MakeExpression(root)});
+  auto folded_or = std::dynamic_pointer_cast<BooleanNode>(folded[0]->root());
+  ASSERT_NE(nullptr, folded_or);
+  ASSERT_EQ(2, folded_or->children().size());
+  EXPECT_NE(folded_or->children()[0], folded_or->children()[1]);
+
+  auto first_and = std::dynamic_pointer_cast<BooleanNode>(folded_or->children()[0]);
+  auto second_and = std::dynamic_pointer_cast<BooleanNode>(folded_or->children()[1]);
+  ASSERT_NE(nullptr, first_and);
+  ASSERT_NE(nullptr, second_and);
+  EXPECT_EQ(first_and->children()[0], second_and->children()[0]);
+  EXPECT_EQ(first_and->children()[1], second_and->children()[1]);
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotApplyIfAlgebra) {
+  auto condition = arrow::field("condition", arrow::boolean());
+  auto value = arrow::field("value", arrow::int32());
+  auto root = TreeExprBuilder::MakeIf(TreeExprBuilder::MakeField(condition),
+                                      TreeExprBuilder::MakeField(value),
+                                      TreeExprBuilder::MakeField(value), arrow::int32());
+
+  auto folded =
+      FoldCommonSubexpressions(*default_function_registry(), {MakeExpression(root)});
+  auto folded_if = std::dynamic_pointer_cast<IfNode>(folded[0]->root());
+  ASSERT_NE(nullptr, folded_if);
+  EXPECT_EQ(folded_if->then_node(), folded_if->else_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldNullInternalFunctions) {
+  auto type = arrow::decimal128(38, 0);
+  auto value = arrow::field("value", type);
+  auto make_node = [&] {
+    return TreeExprBuilder::MakeFunction("castDECIMALNullOnOverflow",
+                                         {TreeExprBuilder::MakeField(value)}, type);
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldFunctionsNeedingContext) {
+  auto value = arrow::field("value", arrow::int32());
+  auto make_node = [&] {
+    return TreeExprBuilder::MakeFunction("chr", {TreeExprBuilder::MakeField(value)},
+                                         arrow::utf8());
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldFunctionsNeedingHolder) {
+  auto make_node = [] {
+    return TreeExprBuilder::MakeFunction("random", {}, arrow::float64());
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldFunctionsThatCanReturnErrors) {
+  auto left = arrow::field("left", arrow::int32());
+  auto right = arrow::field("right", arrow::int32());
+  ExpectDuplicateRootsNotFolded(MakeBinary("divide", left, right, arrow::int32()),
+                                MakeBinary("divide", left, right, arrow::int32()));
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldSafeParentWithUnsafeChild) {
+  auto left = arrow::field("left", arrow::int32());
+  auto right = arrow::field("right", arrow::int32());
+  auto make_node = [&] {
+    auto divide = MakeBinary("divide", left, right, arrow::int32());
+    return TreeExprBuilder::MakeFunction(
+        "add", {divide, TreeExprBuilder::MakeLiteral(int32_t{1})}, arrow::int32());
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldUnknownFunctions) {
+  auto value = arrow::field("value", arrow::int32());
+  auto make_node = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "unknown_function", {TreeExprBuilder::MakeField(value)}, arrow::int32());
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DoesNotFoldRegisteredFunctionsByDefault) {
+  auto registry = std::make_shared<FunctionRegistry>();
+  NativeFunction function("volatile_identity", {}, {arrow::int32()}, arrow::int32(),
+                          kResultNullIfNull, "volatile_identity_int32");
+  ASSERT_OK(registry->Register(std::move(function),
+                               reinterpret_cast<void*>(&VolatileIdentity)));
+
+  auto value = arrow::field("value", arrow::int32());
+  auto make_node = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "volatile_identity", {TreeExprBuilder::MakeField(value)}, arrow::int32());
+  };
+  auto folded = FoldCommonSubexpressions(
+      *registry,
+      {MakeExpression(make_node(), "left"), MakeExpression(make_node(), "right")});
+
+  ASSERT_EQ(2, folded.size());
+  EXPECT_NE(folded[0]->root(), folded[1]->root());
+}
+
+TEST(CommonSubexpressionFolderTest, KeepsInExpressionsOpaque) {
+  auto value = arrow::field("value", arrow::int32());
+  auto make_node = [&] {
+    return TreeExprBuilder::MakeInExpressionInt32(TreeExprBuilder::MakeField(value),
+                                                  std::unordered_set<int32_t>{1, 2, 3});
+  };
+  ExpectDuplicateRootsNotFolded(make_node(), make_node());
+}
+
+TEST(CommonSubexpressionFolderTest, DistinguishesStructuralDifferences) {
+  auto left = arrow::field("left", arrow::int32());
+  auto right = arrow::field("right", arrow::int32());
+  auto registry = default_function_registry();
+  auto folded = FoldCommonSubexpressions(
+      *registry,
+      {MakeExpression(MakeBinary("add", left, right, arrow::int32()), "ordered"),
+       MakeExpression(MakeBinary("add", right, left, arrow::int32()), "reversed"),
+       MakeExpression(TreeExprBuilder::MakeLiteral(int32_t{1}), "int32_literal"),
+       MakeExpression(TreeExprBuilder::MakeLiteral(int32_t{2}), "other_value"),
+       MakeExpression(TreeExprBuilder::MakeLiteral(int64_t{1}), "int64_literal")});
+
+  ASSERT_EQ(5, folded.size());
+  EXPECT_NE(folded[0]->root(), folded[1]->root());
+  EXPECT_NE(folded[2]->root(), folded[3]->root());
+  EXPECT_NE(folded[2]->root(), folded[4]->root());
+}
+
+}  // namespace
+}  // namespace gandiva
diff --git a/cpp/src/gandiva/expr_decomposer.cc b/cpp/src/gandiva/expr_decomposer.cc
index 921829db6a95..2a37cdde82c2 100644
--- a/cpp/src/gandiva/expr_decomposer.cc
+++ b/cpp/src/gandiva/expr_decomposer.cc
@@ -26,6 +26,7 @@
 #include "arrow/util/logging_internal.h"
 #include "gandiva/annotator.h"
 #include "gandiva/dex.h"
+#include "gandiva/expr_cse.h"
 #include "gandiva/function_holder_maker_registry.h"
 #include "gandiva/function_registry.h"
 #include "gandiva/function_signature.h"
@@ -61,6 +62,49 @@ const FunctionNode ExprDecomposer::TryOptimize(const FunctionNode& node) {
   }
 }
 
+Status ExprDecomposer::DecomposeNode(const Node& node, ValueValidityPairPtr* out) {
+  bool can_reuse = CanReuseDecomposition(node);
+  if (can_reuse) {
+    auto it = decomposed_cache_.find(&node);
+    if (it != decomposed_cache_.end()) {
+      *out = it->second;
+      return Status::OK();
+    }
+  }
+
+  ARROW_RETURN_NOT_OK(node.Accept(*this));
+  *out = std::move(result_);
+  if (can_reuse) {
+    decomposed_cache_.emplace(&node, *out);
+  }
+  return Status::OK();
+}
+
+bool ExprDecomposer::CanReuseDecomposition(const Node& node) {
+  auto cached = reusable_node_cache_.find(&node);
+  if (cached != reusable_node_cache_.end()) {
+    return cached->second;
+  }
+
+  bool reusable = false;
+  if (dynamic_cast<const FieldNode*>(&node) != nullptr ||
+      dynamic_cast<const LiteralNode*>(&node) != nullptr) {
+    reusable = true;
+  } else if (auto function_node = dynamic_cast<const FunctionNode*>(&node)) {
+    auto desc = function_node->descriptor();
+    FunctionSignature signature(desc->name(), desc->params(), desc->return_type());
+    const NativeFunction* native_function = registry_.LookupSignature(signature);
+    reusable =
+        native_function != nullptr && CanReuseNativeFunction(registry_, *native_function);
+    for (const auto& child : function_node->children()) {
+      reusable = reusable && CanReuseDecomposition(*child);
+    }
+  }
+
+  reusable_node_cache_.emplace(&node, reusable);
+  return reusable;
+}
+
 // Decompose a field node - wherever possible, merge the validity vectors of the
 // child nodes.
 Status ExprDecomposer::Visit(const FunctionNode& in_node) {
@@ -73,10 +117,9 @@ Status ExprDecomposer::Visit(const FunctionNode& in_node) {
   // decompose the children.
   std::vector<ValueValidityPairPtr> args;
   for (auto& child : node.children()) {
-    auto status = child->Accept(*this);
-    ARROW_RETURN_NOT_OK(status);
-
-    args.push_back(result());
+    ValueValidityPairPtr child_vv;
+    ARROW_RETURN_NOT_OK(DecomposeNode(*child, &child_vv));
+    args.push_back(child_vv);
   }
 
   // Make a function holder, if required.
@@ -93,7 +136,11 @@ Status ExprDecomposer::Visit(const FunctionNode& in_node) {
     // These functions are decomposable, merge the validity bits of the children.
 
     std::vector<DexPtr> merged_validity;
+    std::unordered_set<const ValueValidityPair*> merged_args;
     for (auto& decomposed : args) {
+      if (!merged_args.insert(decomposed.get()).second) {
+        continue;
+      }
       // Merge the validity_expressions of the children to build a combined validity
       // expression.
       merged_validity.insert(merged_validity.end(), decomposed->validity_exprs().begin(),
@@ -130,24 +177,24 @@ Status ExprDecomposer::Visit(const IfNode& node) {
   nested_if_else_ = false;
 
   PushConditionEntry(node);
-  auto status = node.condition()->Accept(*this);
+  ValueValidityPairPtr condition_vv;
+  auto status = DecomposeNode(*node.condition(), &condition_vv);
   ARROW_RETURN_NOT_OK(status);
-  auto condition_vv = result();
   PopConditionEntry(node);
 
   // Add a local bitmap to track the output validity.
   int local_bitmap_idx = PushThenEntry(node, svd_nested_if_else);
-  status = node.then_node()->Accept(*this);
+  ValueValidityPairPtr then_vv;
+  status = DecomposeNode(*node.then_node(), &then_vv);
   ARROW_RETURN_NOT_OK(status);
-  auto then_vv = result();
   PopThenEntry(node);
 
   PushElseEntry(node, local_bitmap_idx);
   nested_if_else_ = (dynamic_cast<IfNode*>(node.else_node().get()) != nullptr);
 
-  status = node.else_node()->Accept(*this);
+  ValueValidityPairPtr else_vv;
+  status = DecomposeNode(*node.else_node(), &else_vv);
   ARROW_RETURN_NOT_OK(status);
-  auto else_vv = result();
   bool is_terminal_else = PopElseEntry(node);
 
   auto validity_dex = std::make_shared<LocalBitMapValidityDex>(local_bitmap_idx);
@@ -164,10 +211,9 @@ Status ExprDecomposer::Visit(const BooleanNode& node) {
   // decompose the children.
   std::vector<ValueValidityPairPtr> args;
   for (auto& child : node.children()) {
-    auto status = child->Accept(*this);
-    ARROW_RETURN_NOT_OK(status);
-
-    args.push_back(result());
+    ValueValidityPairPtr child_vv;
+    ARROW_RETURN_NOT_OK(DecomposeNode(*child, &child_vv));
+    args.push_back(child_vv);
   }
 
   // Add a local bitmap to track the output validity.
@@ -190,9 +236,9 @@ Status ExprDecomposer::Visit(const BooleanNode& node) {
 Status ExprDecomposer::Visit(const InExpressionNode<gandiva::DecimalScalar128>& node) {
   /* decompose the children. */
   std::vector<ValueValidityPairPtr> args;
-  auto status = node.eval_expr()->Accept(*this);
-  ARROW_RETURN_NOT_OK(status);
-  args.push_back(result());
+  ValueValidityPairPtr eval_vv;
+  ARROW_RETURN_NOT_OK(DecomposeNode(*node.eval_expr(), &eval_vv));
+  args.push_back(eval_vv);
   /* In always outputs valid results, so no validity dex */
   auto value_dex = std::make_shared<InExprDex<gandiva::DecimalScalar128>>(
       args, node.values(), node.get_precision(), node.get_scale());
@@ -206,9 +252,9 @@ template <typename ctype>
 Status ExprDecomposer::VisitInGeneric(const InExpressionNode<ctype>& node) {
   /* decompose the children. */
   std::vector<ValueValidityPairPtr> args;
-  auto status = node.eval_expr()->Accept(*this);
-  ARROW_RETURN_NOT_OK(status);
-  args.push_back(result());
+  ValueValidityPairPtr eval_vv;
+  ARROW_RETURN_NOT_OK(DecomposeNode(*node.eval_expr(), &eval_vv));
+  args.push_back(eval_vv);
   /* In always outputs valid results, so no validity dex */
   auto value_dex = std::make_shared<InExprDex<ctype>>(args, node.values());
   int holder_idx = annotator_.AddHolderPointer(value_dex->in_holder().get());
diff --git a/cpp/src/gandiva/expr_decomposer.h b/cpp/src/gandiva/expr_decomposer.h
index 90a27744b362..e1d399709055 100644
--- a/cpp/src/gandiva/expr_decomposer.h
+++ b/cpp/src/gandiva/expr_decomposer.h
@@ -21,6 +21,7 @@
 #include <memory>
 #include <stack>
 #include <string>
+#include <unordered_map>
 #include <utility>
 
 #include "gandiva/arrow.h"
@@ -42,11 +43,7 @@ class GANDIVA_EXPORT ExprDecomposer : public NodeVisitor {
       : registry_(registry), annotator_(annotator), nested_if_else_(false) {}
 
   Status Decompose(const Node& root, ValueValidityPairPtr* out) {
-    auto status = root.Accept(*this);
-    if (status.ok()) {
-      *out = std::move(result_);
-    }
-    return status;
+    return DecomposeNode(root, out);
   }
 
  private:
@@ -75,6 +72,10 @@ class GANDIVA_EXPORT ExprDecomposer : public NodeVisitor {
   template <typename ctype>
   Status VisitInGeneric(const InExpressionNode<ctype>& node);
 
+  Status DecomposeNode(const Node& node, ValueValidityPairPtr* out);
+
+  bool CanReuseDecomposition(const Node& node);
+
   // Optimize a function node, if possible.
   const FunctionNode TryOptimize(const FunctionNode& node);
 
@@ -125,6 +126,8 @@ class GANDIVA_EXPORT ExprDecomposer : public NodeVisitor {
   Annotator& annotator_;
   std::stack<std::unique_ptr<IfStackEntry>> if_entries_stack_;
   ValueValidityPairPtr result_;
+  std::unordered_map<const Node*, ValueValidityPairPtr> decomposed_cache_;
+  std::unordered_map<const Node*, bool> reusable_node_cache_;
   bool nested_if_else_;
 };
 
diff --git a/cpp/src/gandiva/expr_decomposer_test.cc b/cpp/src/gandiva/expr_decomposer_test.cc
index 194c13bc82c8..8aa961861643 100644
--- a/cpp/src/gandiva/expr_decomposer_test.cc
+++ b/cpp/src/gandiva/expr_decomposer_test.cc
@@ -19,11 +19,13 @@
 
 #include <gtest/gtest.h>
 
+#include "arrow/testing/gtest_util.h"
 #include "gandiva/annotator.h"
 #include "gandiva/dex.h"
 #include "gandiva/function_registry.h"
 #include "gandiva/gandiva_aliases.h"
 #include "gandiva/node.h"
+#include "gandiva/tree_expr_builder.h"
 
 namespace gandiva {
 
@@ -34,6 +36,22 @@ class TestExprDecomposer : public ::testing::Test {
   std::shared_ptr<FunctionRegistry> registry_ = default_function_registry();
 };
 
+TEST_F(TestExprDecomposer, TestSharedDagValidityIsMergedOnce) {
+  auto value = arrow::field("value", arrow::int32());
+  NodePtr root = TreeExprBuilder::MakeField(value);
+  constexpr int kDepth = 24;
+  for (int depth = 0; depth < kDepth; ++depth) {
+    root = TreeExprBuilder::MakeFunction("add", {root, root}, arrow::int32());
+  }
+
+  Annotator annotator;
+  ExprDecomposer decomposer(*registry_, annotator);
+  ValueValidityPairPtr decomposed;
+  ASSERT_OK(decomposer.Decompose(*root, &decomposed));
+  ASSERT_NE(nullptr, decomposed);
+  EXPECT_EQ(1, decomposed->validity_exprs().size());
+}
+
 TEST_F(TestExprDecomposer, TestStackSimple) {
   Annotator annotator;
   ExprDecomposer decomposer(*registry_, annotator);
diff --git a/cpp/src/gandiva/filter.cc b/cpp/src/gandiva/filter.cc
index 8a270cfdc06f..ed49ec37acf1 100644
--- a/cpp/src/gandiva/filter.cc
+++ b/cpp/src/gandiva/filter.cc
@@ -23,6 +23,7 @@
 #include "gandiva/bitmap_accumulator.h"
 #include "gandiva/cache.h"
 #include "gandiva/condition.h"
+#include "gandiva/expr_cse.h"
 #include "gandiva/expr_validator.h"
 #include "gandiva/llvm_generator.h"
 #include "gandiva/selection_vector_impl.h"
@@ -45,10 +46,13 @@ Status Filter::Make(SchemaPtr schema, ConditionPtr condition,
   ARROW_RETURN_IF(configuration == nullptr,
                   Status::Invalid("Configuration cannot be null"));
 
+  auto folded_condition =
+      FoldCommonSubexpressions(*configuration->function_registry(), condition);
+
   std::shared_ptr<Cache<ExpressionCacheKey, std::shared_ptr<llvm::MemoryBuffer>>> cache =
       LLVMGenerator::GetCache();
 
-  Condition conditionToKey = *(condition.get());
+  Condition conditionToKey = *(folded_condition.get());
 
   ExpressionCacheKey cache_key(schema, configuration, conditionToKey);
 
@@ -73,13 +77,14 @@ Status Filter::Make(SchemaPtr schema, ConditionPtr condition,
     // Return if the expression is invalid since we will not be able to process further.
     ExprValidator expr_validator(llvm_gen->types(), schema,
                                  configuration->function_registry());
-    ARROW_RETURN_NOT_OK(expr_validator.Validate(condition));
+    ARROW_RETURN_NOT_OK(expr_validator.Validate(folded_condition));
   }
 
   // Set the object cache for LLVM
   ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(obj_cache));
 
-  ARROW_RETURN_NOT_OK(llvm_gen->Build({condition}, SelectionVector::Mode::MODE_NONE));
+  ARROW_RETURN_NOT_OK(
+      llvm_gen->Build({folded_condition}, SelectionVector::Mode::MODE_NONE));
 
   // Instantiate the filter with the completely built llvm generator
   *filter = std::make_shared<Filter>(std::move(llvm_gen), schema, configuration);
diff --git a/cpp/src/gandiva/function_registry.cc b/cpp/src/gandiva/function_registry.cc
index 7ef9178d8d8a..d62d746968e9 100644
--- a/cpp/src/gandiva/function_registry.cc
+++ b/cpp/src/gandiva/function_registry.cc
@@ -84,13 +84,20 @@ const NativeFunction* FunctionRegistry::LookupSignature(
   return got == pc_registry_map_.end() ? nullptr : got->second;
 }
 
-Status FunctionRegistry::Add(NativeFunction func) {
+bool FunctionRegistry::IsBuiltIn(const NativeFunction& function) const {
+  return built_in_functions_.find(&function) != built_in_functions_.end();
+}
+
+Status FunctionRegistry::Add(NativeFunction func, bool is_built_in) {
   if (pc_registry_.size() == kMaxFunctionSignatures) {
     return Status::CapacityError("Exceeded max function signatures limit of ",
                                  kMaxFunctionSignatures);
   }
   pc_registry_.emplace_back(std::move(func));
   const auto& last_func = pc_registry_.back();
+  if (is_built_in) {
+    built_in_functions_.insert(&last_func);
+  }
   for (const auto& func_signature : last_func.signatures()) {
     pc_registry_map_.emplace(&func_signature, &last_func);
   }
@@ -148,7 +155,7 @@ arrow::Result<std::shared_ptr<FunctionRegistry>> MakeDefaultFunctionRegistry() {
         GetHashFunctionRegistry(), GetMathOpsFunctionRegistry(),
         GetStringFunctionRegistry(), GetDateTimeArithmeticFunctionRegistry()}) {
     for (const auto& func_signature : funcs) {
-      ARROW_RETURN_NOT_OK(registry->Add(func_signature));
+      ARROW_RETURN_NOT_OK(registry->Add(func_signature, true));
     }
   }
   return registry;
diff --git a/cpp/src/gandiva/function_registry.h b/cpp/src/gandiva/function_registry.h
index 24b64fac5f3f..ce1d59b304fb 100644
--- a/cpp/src/gandiva/function_registry.h
+++ b/cpp/src/gandiva/function_registry.h
@@ -20,6 +20,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <unordered_set>
 #include <vector>
 
 #include "arrow/buffer.h"
@@ -48,6 +49,9 @@ class GANDIVA_EXPORT FunctionRegistry {
   /// Lookup a pre-compiled function by its signature.
   const NativeFunction* LookupSignature(const FunctionSignature& signature) const;
 
+  /// Return whether a function is part of Gandiva's built-in registry.
+  bool IsBuiltIn(const NativeFunction& function) const;
+
   /// \brief register a set of functions into the function registry from a given bitcode
   /// file
   arrow::Status Register(const std::vector<NativeFunction>& funcs,
@@ -85,11 +89,12 @@ class GANDIVA_EXPORT FunctionRegistry {
  private:
   std::vector<NativeFunction> pc_registry_;
   SignatureMap pc_registry_map_;
+  std::unordered_set<const NativeFunction*> built_in_functions_;
   std::vector<std::shared_ptr<arrow::Buffer>> bitcode_memory_buffers_;
   std::vector<std::pair<NativeFunction, void*>> c_functions_;
   FunctionHolderMakerRegistry holder_maker_registry_;
 
-  Status Add(NativeFunction func);
+  Status Add(NativeFunction func, bool is_built_in = false);
 };
 
 /// \brief get the default function registry
diff --git a/cpp/src/gandiva/function_registry_test.cc b/cpp/src/gandiva/function_registry_test.cc
index bbe72c0ee970..514bb3f4ecc3 100644
--- a/cpp/src/gandiva/function_registry_test.cc
+++ b/cpp/src/gandiva/function_registry_test.cc
@@ -20,6 +20,7 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 #include <algorithm>
+#include <cstdint>
 #include <string>
 #include <unordered_set>
 
@@ -27,6 +28,16 @@
 
 namespace gandiva {
 
+arrow::Result<std::shared_ptr<FunctionRegistry>> MakeDefaultFunctionRegistry();
+
+namespace {
+
+int32_t CustomIdentity(int32_t value) { return value; }
+
+int32_t CustomAdd(int32_t left, int32_t right) { return left + right; }
+
+}  // namespace
+
 class TestFunctionRegistry : public ::testing::Test {
  protected:
   std::shared_ptr<FunctionRegistry> registry_ = gandiva::default_function_registry();
@@ -122,4 +133,35 @@ TEST_F(TestFunctionRegistry, TestNoDuplicates) {
          "different precompiled functions:\n"
       << stream.str();
 }
+
+TEST_F(TestFunctionRegistry, TestBuiltInIdentitySurvivesCustomRegistrations) {
+  ASSERT_OK_AND_ASSIGN(auto registry, MakeDefaultFunctionRegistry());
+  FunctionSignature add_signature("add", {arrow::int32(), arrow::int32()},
+                                  arrow::int32());
+  const NativeFunction* built_in_add = registry->LookupSignature(add_signature);
+  ASSERT_NE(nullptr, built_in_add);
+  ASSERT_TRUE(registry->IsBuiltIn(*built_in_add));
+
+  for (int i = 0; i < 64; ++i) {
+    auto name = "custom_identity_" + std::to_string(i);
+    NativeFunction function(name, {}, {arrow::int32()}, arrow::int32(), kResultNullIfNull,
+                            name + "_int32");
+    ASSERT_OK(registry->Register(std::move(function),
+                                 reinterpret_cast<void*>(&CustomIdentity)));
+  }
+
+  EXPECT_EQ(built_in_add, registry->LookupSignature(add_signature));
+  EXPECT_TRUE(registry->IsBuiltIn(*built_in_add));
+  ASSERT_NE(registry->begin(), registry->end());
+  EXPECT_FALSE(registry->IsBuiltIn(*registry->back()));
+
+  NativeFunction shadow_add("add", {}, {arrow::int32(), arrow::int32()}, arrow::int32(),
+                            kResultNullIfNull, "custom_add_int32_int32");
+  ASSERT_OK(
+      registry->Register(std::move(shadow_add), reinterpret_cast<void*>(&CustomAdd)));
+
+  EXPECT_EQ(built_in_add, registry->LookupSignature(add_signature));
+  EXPECT_TRUE(registry->IsBuiltIn(*built_in_add));
+  EXPECT_FALSE(registry->IsBuiltIn(*registry->back()));
+}
 }  // namespace gandiva
diff --git a/cpp/src/gandiva/llvm_generator.cc b/cpp/src/gandiva/llvm_generator.cc
index a42f71a1f7f1..3268fd5e83b3 100644
--- a/cpp/src/gandiva/llvm_generator.cc
+++ b/cpp/src/gandiva/llvm_generator.cc
@@ -1241,15 +1241,27 @@ LValuePtr LLVMGenerator::Visitor::BuildIfElse(llvm::Value* condition,
 
 LValuePtr LLVMGenerator::Visitor::BuildValueAndValidity(const ValueValidityPair& pair) {
   // generate code for value
-  auto value_expr = pair.value_expr();
-  value_expr->Accept(*this);
-  auto value = result()->data();
-  auto length = result()->length();
+  auto value = BuildDex(pair.value_expr());
 
   // generate code for validity
   auto validity = BuildCombinedValidity(pair.validity_exprs());
 
-  return std::make_shared<LValue>(value, length, validity);
+  return std::make_shared<LValue>(value->data(), value->length(), validity);
+}
+
+LValuePtr LLVMGenerator::Visitor::BuildDex(const DexPtr& dex) {
+  auto* block = ir_builder()->GetInsertBlock();
+  auto cached = dex_cache_.find(dex.get());
+  if (cached != dex_cache_.end() && cached->second.block == block) {
+    return cached->second.value;
+  }
+
+  dex->Accept(*this);
+  auto value = result();
+  if (ir_builder()->GetInsertBlock() == block) {
+    dex_cache_.insert_or_assign(dex.get(), CachedDexValue{block, value});
+  }
+  return value;
 }
 
 LValuePtr LLVMGenerator::Visitor::BuildFunctionCall(const NativeFunction* func,
@@ -1335,9 +1347,7 @@ std::vector<llvm::Value*> LLVMGenerator::Visitor::BuildParams(
   // build the function params, along with the validities.
   for (auto& pair : args) {
     // build value.
-    DexPtr value_expr = pair->value_expr();
-    value_expr->Accept(*this);
-    LValue& result_ref = *result();
+    LValue& result_ref = *BuildDex(pair->value_expr());
 
     // append all the parameters corresponding to this LValue.
     result_ref.AppendFunctionParams(&params);
@@ -1359,8 +1369,8 @@ llvm::Value* LLVMGenerator::Visitor::BuildCombinedValidity(const DexVector& vali
 
   llvm::Value* isValid = types->true_constant();
   for (auto& dex : validities) {
-    dex->Accept(*this);
-    isValid = builder->CreateAnd(isValid, result()->data(), "validityBitAnd");
+    auto value = BuildDex(dex);
+    isValid = builder->CreateAnd(isValid, value->data(), "validityBitAnd");
   }
   ADD_VISITOR_TRACE("combined validity is %T", isValid);
   return isValid;
diff --git a/cpp/src/gandiva/llvm_generator.h b/cpp/src/gandiva/llvm_generator.h
index a60e2bf6b29e..e9099f56d924 100644
--- a/cpp/src/gandiva/llvm_generator.h
+++ b/cpp/src/gandiva/llvm_generator.h
@@ -22,6 +22,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "arrow/util/macros.h"
@@ -84,6 +85,8 @@ class GANDIVA_EXPORT LLVMGenerator {
   LLVMTypes* types() { return engine_->types(); }
   llvm::Module* module() { return engine_->module(); }
   const std::string& ir() { return engine_->ir(); }
+  const std::string& unoptimized_ir() { return engine_->unoptimized_ir(); }
+  bool has_unoptimized_ir() const { return engine_->has_unoptimized_ir(); }
 
  private:
   explicit LLVMGenerator(bool cached,
@@ -151,6 +154,9 @@ class GANDIVA_EXPORT LLVMGenerator {
     // Generate the code to build the validity and the value for the given pair.
     LValuePtr BuildValueAndValidity(const ValueValidityPair& pair);
 
+    // Generate or reuse code for a decomposed value in the current basic block.
+    LValuePtr BuildDex(const DexPtr& dex);
+
     // Generate code to build the params.
     std::vector<llvm::Value*> BuildParams(int holder_idx,
                                           const ValueValidityPairVector& args,
@@ -188,6 +194,11 @@ class GANDIVA_EXPORT LLVMGenerator {
     llvm::Value* arg_context_ptr_;
     llvm::Value* loop_var_;
     bool has_arena_allocs_;
+    struct CachedDexValue {
+      llvm::BasicBlock* block;
+      LValuePtr value;
+    };
+    std::unordered_map<const Dex*, CachedDexValue> dex_cache_;
   };
 
   // Generate the code for one expression for default mode, with the output of
diff --git a/cpp/src/gandiva/projector.cc b/cpp/src/gandiva/projector.cc
index ec0302146fff..27b94654d93a 100644
--- a/cpp/src/gandiva/projector.cc
+++ b/cpp/src/gandiva/projector.cc
@@ -24,6 +24,7 @@
 #include "arrow/util/logging.h"
 
 #include "gandiva/cache.h"
+#include "gandiva/expr_cse.h"
 #include "gandiva/expr_validator.h"
 #include "gandiva/llvm_generator.h"
 
@@ -61,11 +62,15 @@ Status Projector::Make(SchemaPtr schema, const ExpressionVector& exprs,
   ARROW_RETURN_IF(configuration == nullptr,
                   Status::Invalid("Configuration cannot be null"));
 
+  auto folded_exprs =
+      FoldCommonSubexpressions(*configuration->function_registry(), exprs);
+
   // see if equivalent projector was already built
   std::shared_ptr<Cache<ExpressionCacheKey, std::shared_ptr<llvm::MemoryBuffer>>> cache =
       LLVMGenerator::GetCache();
 
-  ExpressionCacheKey cache_key(schema, configuration, exprs, selection_vector_mode);
+  ExpressionCacheKey cache_key(schema, configuration, folded_exprs,
+                               selection_vector_mode);
 
   bool is_cached = false;
 
@@ -89,7 +94,7 @@ Status Projector::Make(SchemaPtr schema, const ExpressionVector& exprs,
   if (!is_cached) {
     ExprValidator expr_validator(llvm_gen->types(), schema,
                                  configuration->function_registry());
-    for (auto& expr : exprs) {
+    for (auto& expr : folded_exprs) {
       ARROW_RETURN_NOT_OK(expr_validator.Validate(expr));
     }
   }
@@ -97,12 +102,12 @@ Status Projector::Make(SchemaPtr schema, const ExpressionVector& exprs,
   // Set the object cache for LLVM
   ARROW_RETURN_NOT_OK(llvm_gen->SetLLVMObjectCache(obj_cache));
 
-  ARROW_RETURN_NOT_OK(llvm_gen->Build(exprs, selection_vector_mode));
+  ARROW_RETURN_NOT_OK(llvm_gen->Build(folded_exprs, selection_vector_mode));
 
   // save the output field types. Used for validation at Evaluate() time.
   std::vector<FieldPtr> output_fields;
-  output_fields.reserve(exprs.size());
-  for (auto& expr : exprs) {
+  output_fields.reserve(folded_exprs.size());
+  for (auto& expr : folded_exprs) {
     output_fields.push_back(expr->result());
   }
 
@@ -283,6 +288,13 @@ Status Projector::ValidateArrayDataCapacity(const arrow::ArrayData& array_data,
 
 const std::string& Projector::DumpIR() { return llvm_generator_->ir(); }
 
+Result<std::string> Projector::DumpUnoptimizedIR() {
+  ARROW_RETURN_IF(
+      !llvm_generator_->has_unoptimized_ir(),
+      Status::Invalid("Unoptimized IR was not captured when this projector was built"));
+  return llvm_generator_->unoptimized_ir();
+}
+
 void Projector::SetBuiltFromCache(bool flag) { built_from_cache_ = flag; }
 
 bool Projector::GetBuiltFromCache() { return built_from_cache_; }
diff --git a/cpp/src/gandiva/projector.h b/cpp/src/gandiva/projector.h
index f1ae7e4dc8cc..8352f3c662ca 100644
--- a/cpp/src/gandiva/projector.h
+++ b/cpp/src/gandiva/projector.h
@@ -120,6 +120,10 @@ class GANDIVA_EXPORT Projector {
 
   const std::string& DumpIR();
 
+  /// Return the generated IR before the optimizer pipeline runs.
+  /// Unavailable when IR dumping is disabled or the projector is built from cache.
+  Result<std::string> DumpUnoptimizedIR();
+
   void SetBuiltFromCache(bool flag);
 
   bool GetBuiltFromCache();
diff --git a/cpp/src/gandiva/tests/filter_test.cc b/cpp/src/gandiva/tests/filter_test.cc
index 749000aa0cf2..eef97f76fff6 100644
--- a/cpp/src/gandiva/tests/filter_test.cc
+++ b/cpp/src/gandiva/tests/filter_test.cc
@@ -87,6 +87,49 @@ TEST_F(TestFilter, TestFilterCache) {
   EXPECT_FALSE(should_be_new_filter->GetBuiltFromCache());
 }
 
+TEST_F(TestFilter, TestCommonSubexpressionSafety) {
+  auto field0 = field("filter_cse_f0", int32());
+  auto field1 = field("filter_cse_f1", int32());
+  auto schema = arrow::schema({field0, field1});
+
+  auto make_less_than = [&]() {
+    auto sum = TreeExprBuilder::MakeFunction(
+        "add", {TreeExprBuilder::MakeField(field0), TreeExprBuilder::MakeField(field1)},
+        arrow::int32());
+    return TreeExprBuilder::MakeFunction(
+        "less_than", {sum, TreeExprBuilder::MakeLiteral(static_cast<int32_t>(10))},
+        arrow::boolean());
+  };
+  auto repeated_condition = TreeExprBuilder::MakeCondition(
+      TreeExprBuilder::MakeAnd({make_less_than(), make_less_than(), make_less_than()}));
+  auto configuration = TestConfiguration();
+
+  std::shared_ptr<Filter> filter;
+  ASSERT_OK(Filter::Make(schema, repeated_condition, configuration, &filter));
+  ASSERT_FALSE(filter->GetBuiltFromCache());
+
+  auto input0 = MakeArrowArrayInt32({1, 2, 3, 4, 6}, {true, true, true, false, true});
+  auto input1 = MakeArrowArrayInt32({5, 9, 6, 17, 3}, {true, true, false, true, true});
+  auto batch = arrow::RecordBatch::Make(schema, 5, {input0, input1});
+  auto expected = MakeArrowArrayUint16({0, 4});
+
+  std::shared_ptr<SelectionVector> selection_vector;
+  ASSERT_OK(SelectionVector::MakeInt16(batch->num_rows(), pool_, &selection_vector));
+  ASSERT_OK(filter->Evaluate(*batch, selection_vector));
+  EXPECT_ARROW_ARRAY_EQUALS(expected, selection_vector->ToArray());
+
+  auto equivalent_condition = TreeExprBuilder::MakeCondition(make_less_than());
+  std::shared_ptr<Filter> cached_filter;
+  ASSERT_OK(Filter::Make(schema, equivalent_condition, configuration, &cached_filter));
+  ASSERT_FALSE(cached_filter->GetBuiltFromCache());
+
+  std::shared_ptr<SelectionVector> cached_selection_vector;
+  ASSERT_OK(
+      SelectionVector::MakeInt16(batch->num_rows(), pool_, &cached_selection_vector));
+  ASSERT_OK(cached_filter->Evaluate(*batch, cached_selection_vector));
+  EXPECT_ARROW_ARRAY_EQUALS(expected, cached_selection_vector->ToArray());
+}
+
 TEST_F(TestFilter, TestFilterCacheNullTreatment) {
   // schema for input fields
   auto field0 = field("f0", utf8());
diff --git a/cpp/src/gandiva/tests/micro_benchmarks.cc b/cpp/src/gandiva/tests/micro_benchmarks.cc
index 450e691323ca..54bbc96221b4 100644
--- a/cpp/src/gandiva/tests/micro_benchmarks.cc
+++ b/cpp/src/gandiva/tests/micro_benchmarks.cc
@@ -17,12 +17,20 @@
 
 #include <stdlib.h>
 
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
 #include "arrow/memory_pool.h"
 #include "arrow/status.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/type_fwd.h"
 #include "benchmark/benchmark.h"
 #include "gandiva/decimal_type_util.h"
+#include "gandiva/expr_cse.h"
+#include "gandiva/filter.h"
 #include "gandiva/projector.h"
 #include "gandiva/tests/test_util.h"
 #include "gandiva/tests/timed_evaluate.h"
@@ -35,6 +43,207 @@ using arrow::int32;
 using arrow::int64;
 using arrow::utf8;
 
+namespace {
+
+enum class CsePattern : int64_t {
+  kDeepUnique,
+  kRepeatedSafe,
+  kRepeatedUnsafe,
+  kSharedDag,
+};
+
+struct CseBenchmarkExpression {
+  SchemaPtr schema;
+  ExpressionPtr expression;
+  ConditionPtr condition;
+};
+
+NodePtr MakeBalancedTree(int64_t leaves, const std::function<NodePtr()>& make_leaf) {
+  if (leaves <= 1) {
+    return make_leaf();
+  }
+  auto left = MakeBalancedTree(leaves / 2, make_leaf);
+  auto right = MakeBalancedTree(leaves - leaves / 2, make_leaf);
+  return TreeExprBuilder::MakeFunction("add", {left, right}, arrow::float64());
+}
+
+CseBenchmarkExpression MakeCseBenchmarkExpression(CsePattern pattern, int64_t size,
+                                                  int64_t salt) {
+  auto left = field("cse_left", arrow::float64());
+  auto right = field("cse_right", arrow::float64());
+  auto schema = arrow::schema({left, right});
+
+  NodePtr root;
+  switch (pattern) {
+    case CsePattern::kDeepUnique: {
+      root = TreeExprBuilder::MakeField(left);
+      for (int64_t i = 0; i < size; ++i) {
+        root = TreeExprBuilder::MakeFunction(
+            "add", {root, TreeExprBuilder::MakeLiteral(static_cast<double>(i + 1))},
+            arrow::float64());
+      }
+      break;
+    }
+    case CsePattern::kRepeatedSafe: {
+      auto make_leaf = [&] {
+        auto make_sum = [&] {
+          return TreeExprBuilder::MakeFunction(
+              "add",
+              {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+              arrow::float64());
+        };
+        return TreeExprBuilder::MakeFunction("multiply", {make_sum(), make_sum()},
+                                             arrow::float64());
+      };
+      root = MakeBalancedTree(size, make_leaf);
+      break;
+    }
+    case CsePattern::kRepeatedUnsafe: {
+      auto make_leaf = [&] {
+        return TreeExprBuilder::MakeFunction(
+            "divide",
+            {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+            arrow::float64());
+      };
+      root = MakeBalancedTree(size, make_leaf);
+      break;
+    }
+    case CsePattern::kSharedDag: {
+      root = TreeExprBuilder::MakeFunction(
+          "add", {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+          arrow::float64());
+      for (int64_t i = 0; i < size; ++i) {
+        root = TreeExprBuilder::MakeFunction("add", {root, root}, arrow::float64());
+      }
+      break;
+    }
+  }
+
+  root = TreeExprBuilder::MakeFunction(
+      "add", {root, TreeExprBuilder::MakeLiteral(static_cast<double>(salt) + 0.125)},
+      arrow::float64());
+  auto condition_root = TreeExprBuilder::MakeFunction(
+      "greater_than", {root, TreeExprBuilder::MakeLiteral(0.0)}, arrow::boolean());
+  return {schema,
+          TreeExprBuilder::MakeExpression(
+              root, field("cse_result_" + std::to_string(salt), arrow::float64())),
+          TreeExprBuilder::MakeCondition(condition_root)};
+}
+
+void CseFoldOnly(benchmark::State& state) {
+  auto pattern = static_cast<CsePattern>(state.range(0));
+  auto size = state.range(1);
+  int64_t iteration = 0;
+  for (auto _ : state) {
+    state.PauseTiming();
+    auto input = MakeCseBenchmarkExpression(pattern, size, iteration++);
+    state.ResumeTiming();
+    auto folded = FoldCommonSubexpressions(*default_function_registry(),
+                                           ExpressionVector{input.expression});
+    benchmark::DoNotOptimize(folded);
+  }
+}
+
+void CseProjectorBuild(benchmark::State& state) {
+  auto pattern = static_cast<CsePattern>(state.range(0));
+  auto size = state.range(1);
+  int64_t iteration = 0;
+  int64_t total_ir_bytes = 0;
+  for (auto _ : state) {
+    state.PauseTiming();
+    auto input = MakeCseBenchmarkExpression(pattern, size, iteration++);
+    auto configuration = std::make_shared<Configuration>(
+        true, default_function_registry(), /*dump_ir=*/true);
+    state.ResumeTiming();
+
+    std::shared_ptr<Projector> projector;
+    ASSERT_OK(
+        Projector::Make(input.schema, {input.expression}, configuration, &projector));
+
+    state.PauseTiming();
+    ASSERT_OK_AND_ASSIGN(auto ir, projector->DumpUnoptimizedIR());
+    total_ir_bytes += static_cast<int64_t>(ir.size());
+    benchmark::DoNotOptimize(projector);
+    state.ResumeTiming();
+  }
+  state.counters["ir_bytes"] =
+      benchmark::Counter(total_ir_bytes, benchmark::Counter::kAvgIterations);
+}
+
+void CseFilterBuild(benchmark::State& state) {
+  auto pattern = static_cast<CsePattern>(state.range(0));
+  auto size = state.range(1);
+  int64_t iteration = 0;
+  for (auto _ : state) {
+    state.PauseTiming();
+    auto input = MakeCseBenchmarkExpression(pattern, size, iteration++);
+    auto configuration = std::make_shared<Configuration>(
+        true, default_function_registry(), /*dump_ir=*/false);
+    state.ResumeTiming();
+
+    std::shared_ptr<Filter> filter;
+    ASSERT_OK(Filter::Make(input.schema, input.condition, configuration, &filter));
+    benchmark::DoNotOptimize(filter);
+  }
+}
+
+void CseProjectorEvaluate(benchmark::State& state) {
+  auto pattern = static_cast<CsePattern>(state.range(0));
+  auto size = state.range(1);
+  auto input = MakeCseBenchmarkExpression(pattern, size, 0);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(
+      Projector::Make(input.schema, {input.expression}, TestConfiguration(), &projector));
+  auto left = MakeArrowArrayFloat64(std::vector<double>(1024, 12.0));
+  auto right = MakeArrowArrayFloat64(std::vector<double>(1024, 3.0));
+  auto batch = arrow::RecordBatch::Make(input.schema, 1024, {left, right});
+
+  for (auto _ : state) {
+    arrow::ArrayVector outputs;
+    ASSERT_OK(projector->Evaluate(*batch, arrow::default_memory_pool(), &outputs));
+    benchmark::DoNotOptimize(outputs);
+  }
+  state.SetItemsProcessed(state.iterations() * batch->num_rows());
+}
+
+void CseFilterEvaluate(benchmark::State& state) {
+  auto pattern = static_cast<CsePattern>(state.range(0));
+  auto size = state.range(1);
+  auto input = MakeCseBenchmarkExpression(pattern, size, 0);
+  std::shared_ptr<Filter> filter;
+  ASSERT_OK(Filter::Make(input.schema, input.condition, TestConfiguration(), &filter));
+  auto left = MakeArrowArrayFloat64(std::vector<double>(1024, 12.0));
+  auto right = MakeArrowArrayFloat64(std::vector<double>(1024, 3.0));
+  auto batch = arrow::RecordBatch::Make(input.schema, 1024, {left, right});
+  std::shared_ptr<SelectionVector> selection;
+  ASSERT_OK(SelectionVector::MakeInt32(batch->num_rows(), arrow::default_memory_pool(),
+                                       &selection));
+
+  for (auto _ : state) {
+    ASSERT_OK(filter->Evaluate(*batch, selection));
+    benchmark::DoNotOptimize(selection);
+  }
+  state.SetItemsProcessed(state.iterations() * batch->num_rows());
+}
+
+void CseBenchmarkArguments(benchmark::internal::Benchmark* benchmark) {
+  for (auto pattern : {CsePattern::kDeepUnique, CsePattern::kRepeatedSafe,
+                       CsePattern::kRepeatedUnsafe}) {
+    for (auto size : {10, 100, 1000}) {
+      benchmark->Args({static_cast<int64_t>(pattern), size});
+    }
+  }
+}
+
+void CseFoldBenchmarkArguments(benchmark::internal::Benchmark* benchmark) {
+  CseBenchmarkArguments(benchmark);
+  for (auto depth : {10, 20, 24}) {
+    benchmark->Args({static_cast<int64_t>(CsePattern::kSharedDag), depth});
+  }
+}
+
+}  // namespace
+
 static void TimedTestAdd3(benchmark::State& state) {
   // schema for input fields
   auto field0 = field("f0", int64());
@@ -491,6 +700,26 @@ static void DecimalAdd3Large(benchmark::State& state) {
 }
 
 BENCHMARK(TimedTestExprCompilation)->Unit(benchmark::kMicrosecond);
+BENCHMARK(CseFoldOnly)
+    ->Apply(CseFoldBenchmarkArguments)
+    ->ArgNames({"pattern", "size"})
+    ->Unit(benchmark::kMicrosecond);
+BENCHMARK(CseProjectorBuild)
+    ->Apply(CseBenchmarkArguments)
+    ->ArgNames({"pattern", "size"})
+    ->Unit(benchmark::kMillisecond);
+BENCHMARK(CseFilterBuild)
+    ->Apply(CseBenchmarkArguments)
+    ->ArgNames({"pattern", "size"})
+    ->Unit(benchmark::kMillisecond);
+BENCHMARK(CseProjectorEvaluate)
+    ->Apply(CseBenchmarkArguments)
+    ->ArgNames({"pattern", "size"})
+    ->Unit(benchmark::kMicrosecond);
+BENCHMARK(CseFilterEvaluate)
+    ->Apply(CseBenchmarkArguments)
+    ->ArgNames({"pattern", "size"})
+    ->Unit(benchmark::kMicrosecond);
 BENCHMARK(TimedTestAdd3)->Unit(benchmark::kMicrosecond);
 BENCHMARK(TimedTestBigNested)->Unit(benchmark::kMicrosecond);
 BENCHMARK(TimedTestExtractYear)->Unit(benchmark::kMicrosecond);
diff --git a/cpp/src/gandiva/tests/projector_test.cc b/cpp/src/gandiva/tests/projector_test.cc
index 268cb55a6422..8f369af68afe 100644
--- a/cpp/src/gandiva/tests/projector_test.cc
+++ b/cpp/src/gandiva/tests/projector_test.cc
@@ -24,6 +24,7 @@
 #include <gtest/gtest.h>
 
 #include <cmath>
+#include <string>
 
 #include "arrow/memory_pool.h"
 #include "gandiva/function_registry.h"
@@ -39,6 +40,45 @@ using arrow::float32;
 using arrow::int32;
 using arrow::int64;
 
+namespace {
+
+int CountOccurrences(const std::string& text, const std::string& needle) {
+  int count = 0;
+  std::string::size_type pos = 0;
+  while ((pos = text.find(needle, pos)) != std::string::npos) {
+    ++count;
+    pos += needle.size();
+  }
+  return count;
+}
+
+std::string ExtractFunctionIR(const std::string& ir, const std::string& function_name) {
+  const auto name_pos = ir.find("@" + function_name + "(");
+  if (name_pos == std::string::npos) {
+    return "";
+  }
+  const auto function_start = ir.rfind("\ndefine ", name_pos);
+  const auto start = function_start == std::string::npos ? 0 : function_start + 1;
+  const auto function_end = ir.find("\n}\n", name_pos);
+  if (function_end == std::string::npos) {
+    return ir.substr(start);
+  }
+  return ir.substr(start, function_end + 3 - start);
+}
+
+void ExpectProjectorOutput(const std::shared_ptr<Projector>& projector,
+                           const SchemaPtr& schema, const arrow::ArrayVector& inputs,
+                           const ArrayPtr& expected, arrow::MemoryPool* pool) {
+  ASSERT_FALSE(inputs.empty());
+  auto batch = arrow::RecordBatch::Make(schema, inputs[0]->length(), inputs);
+  arrow::ArrayVector outputs;
+  ASSERT_OK(projector->Evaluate(*batch, pool, &outputs));
+  ASSERT_EQ(1, outputs.size());
+  EXPECT_ARROW_ARRAY_EQUALS(expected, outputs[0]);
+}
+
+}  // namespace
+
 class TestProjector : public ::testing::Test {
  public:
   void SetUp() {
@@ -388,6 +428,506 @@ TEST_F(TestProjector, TestAllIntTypes) {
   TestArithmeticOpsForType<arrow::Int64Type, int64_t>(pool_);
 }
 
+TEST_F(TestProjector, TestCommonSubexpressionEliminationIR) {
+  auto field0 = arrow::field("cse_f0", arrow::int32());
+  auto field1 = arrow::field("cse_f1", arrow::int32());
+  auto schema = arrow::schema({field0, field1});
+
+  auto left_sum = TreeExprBuilder::MakeFunction(
+      "add", {TreeExprBuilder::MakeField(field0), TreeExprBuilder::MakeField(field1)},
+      arrow::int32());
+  auto right_sum = TreeExprBuilder::MakeFunction(
+      "add", {TreeExprBuilder::MakeField(field0), TreeExprBuilder::MakeField(field1)},
+      arrow::int32());
+  auto square =
+      TreeExprBuilder::MakeFunction("multiply", {left_sum, right_sum}, arrow::int32());
+  auto expr =
+      TreeExprBuilder::MakeExpression(square, arrow::field("cse_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+
+  EXPECT_EQ(1, CountOccurrences(unoptimized_expr_ir, "call i32 @add_int32_int32"));
+  EXPECT_EQ(1, CountOccurrences(unoptimized_expr_ir, "call i32 @multiply_int32_int32"));
+
+  auto input0 = MakeArrowArrayInt32({1, 2, 3, 4}, {true, true, false, true});
+  auto input1 = MakeArrowArrayInt32({10, -2, 3, 5}, {true, true, true, false});
+  auto expected = MakeArrowArrayInt32({121, 0, 0, 0}, {true, true, false, false});
+  ExpectProjectorOutput(projector, schema, {input0, input1}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestUnoptimizedIRUnavailableForCachedProjector) {
+  auto field0 = arrow::field("cached_ir_f0", arrow::int32());
+  auto field1 = arrow::field("cached_ir_f1", arrow::int32());
+  auto schema = arrow::schema({field0, field1});
+  auto expr = TreeExprBuilder::MakeExpression(
+      "add", {field0, field1}, arrow::field("cached_ir_out", arrow::int32()));
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+  ASSERT_FALSE(projector->GetBuiltFromCache());
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  ASSERT_FALSE(unoptimized_ir.empty());
+
+  std::shared_ptr<Projector> cached_projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &cached_projector));
+  ASSERT_TRUE(cached_projector->GetBuiltFromCache());
+  ASSERT_RAISES_WITH_MESSAGE(
+      Invalid, "Invalid: Unoptimized IR was not captured when this projector was built",
+      cached_projector->DumpUnoptimizedIR());
+}
+
+TEST_F(TestProjector, TestUnoptimizedIRAvailabilityUsesCapturedState) {
+  auto field0 = arrow::field("mutable_config_f0", arrow::int32());
+  auto schema = arrow::schema({field0});
+  auto expr =
+      TreeExprBuilder::MakeExpression(TreeExprBuilder::MakeField(field0),
+                                      arrow::field("mutable_config_out", arrow::int32()));
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/false);
+
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+  configuration->set_dump_ir(true);
+
+  ASSERT_RAISES_WITH_MESSAGE(
+      Invalid, "Invalid: Unoptimized IR was not captured when this projector was built",
+      projector->DumpUnoptimizedIR());
+}
+
+TEST_F(TestProjector, TestNestedCommonSubexpressionEliminationIR) {
+  auto field0 = arrow::field("nested_cse_f0", arrow::int32());
+  auto field1 = arrow::field("nested_cse_f1", arrow::int32());
+  auto schema = arrow::schema({field0, field1});
+
+  auto make_sum = [&]() {
+    return TreeExprBuilder::MakeFunction(
+        "add", {TreeExprBuilder::MakeField(field0), TreeExprBuilder::MakeField(field1)},
+        arrow::int32());
+  };
+  auto make_square = [&]() {
+    return TreeExprBuilder::MakeFunction("multiply", {make_sum(), make_sum()},
+                                         arrow::int32());
+  };
+  auto nested = TreeExprBuilder::MakeFunction("add", {make_square(), make_square()},
+                                              arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      nested, arrow::field("nested_cse_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+
+  EXPECT_EQ(2, CountOccurrences(unoptimized_expr_ir, "call i32 @add_int32_int32"));
+  EXPECT_EQ(1, CountOccurrences(unoptimized_expr_ir, "call i32 @multiply_int32_int32"));
+
+  auto input0 = MakeArrowArrayInt32({1, 2, 3, 4}, {true, true, false, true});
+  auto input1 = MakeArrowArrayInt32({10, -2, 3, 5}, {true, true, true, false});
+  auto expected = MakeArrowArrayInt32({242, 0, 0, 0}, {true, true, false, false});
+  ExpectProjectorOutput(projector, schema, {input0, input1}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestCommonSubexpressionEliminationForVarLenInput) {
+  auto value = arrow::field("varlen_cse_value", arrow::utf8());
+  auto schema = arrow::schema({value});
+
+  auto make_length = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "octet_length", {TreeExprBuilder::MakeField(value)}, arrow::int32());
+  };
+  auto root = TreeExprBuilder::MakeFunction("add", {make_length(), make_length()},
+                                            arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("varlen_cse_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(1, CountOccurrences(unoptimized_expr_ir, "call i32 @octet_length_utf8"));
+
+  auto values =
+      MakeArrowArrayUtf8({"", "abc", "路学", "ignored"}, {true, true, true, false});
+  auto expected = MakeArrowArrayInt32({0, 6, 12, 0}, {true, true, true, false});
+  ExpectProjectorOutput(projector, schema, {values}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestCommonSubexpressionEliminationForNullNeverFunction) {
+  auto value = arrow::field("null_never_cse_value", arrow::utf8());
+  auto schema = arrow::schema({value});
+
+  auto make_is_null = [&] {
+    return TreeExprBuilder::MakeFunction("isnull", {TreeExprBuilder::MakeField(value)},
+                                         arrow::boolean());
+  };
+  auto root = TreeExprBuilder::MakeFunction("equal", {make_is_null(), make_is_null()},
+                                            arrow::boolean());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("null_never_cse_out", arrow::boolean()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(1, CountOccurrences(unoptimized_expr_ir, "call i1 @isnull_utf8"));
+
+  auto values = MakeArrowArrayUtf8({"present", "ignored", "", "also present"},
+                                   {true, false, true, true});
+  auto expected = MakeArrowArrayBool({true, true, true, true});
+  ExpectProjectorOutput(projector, schema, {values}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestCommonSubexpressionEliminationAcrossMultipleOutputs) {
+  auto left = arrow::field("multi_output_cse_left", arrow::int32());
+  auto right = arrow::field("multi_output_cse_right", arrow::int32());
+  auto schema = arrow::schema({left, right});
+
+  auto make_sum = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "add", {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+        arrow::int32());
+  };
+  auto first = TreeExprBuilder::MakeExpression(
+      make_sum(), arrow::field("multi_output_cse_first", arrow::int32()));
+  auto second = TreeExprBuilder::MakeExpression(
+      make_sum(), arrow::field("multi_output_cse_second", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {first, second}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto first_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  const auto second_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_1_0");
+  ASSERT_FALSE(first_expr_ir.empty());
+  ASSERT_FALSE(second_expr_ir.empty());
+  EXPECT_EQ(1, CountOccurrences(first_expr_ir, "call i32 @add_int32_int32"));
+  EXPECT_EQ(1, CountOccurrences(second_expr_ir, "call i32 @add_int32_int32"));
+
+  auto left_values = MakeArrowArrayInt32({1, 2, 3}, {true, true, false});
+  auto right_values = MakeArrowArrayInt32({10, 20, 30});
+  auto expected = MakeArrowArrayInt32({11, 22, 0}, {true, true, false});
+  auto batch = arrow::RecordBatch::Make(schema, 3, {left_values, right_values});
+  arrow::ArrayVector outputs;
+  ASSERT_OK(projector->Evaluate(*batch, pool_, &outputs));
+  ASSERT_EQ(2, outputs.size());
+  EXPECT_ARROW_ARRAY_EQUALS(expected, outputs[0]);
+  EXPECT_ARROW_ARRAY_EQUALS(expected, outputs[1]);
+}
+
+TEST_F(TestProjector, TestCommonSubexpressionNotReusedAcrossIfBranches) {
+  auto condition_field = arrow::field("branch_cse_cond", arrow::boolean());
+  auto left = arrow::field("branch_cse_left", arrow::int32());
+  auto right = arrow::field("branch_cse_right", arrow::int32());
+  auto schema = arrow::schema({condition_field, left, right});
+
+  auto make_sum = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "add", {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+        arrow::int32());
+  };
+  auto else_node = TreeExprBuilder::MakeFunction(
+      "multiply", {make_sum(), TreeExprBuilder::MakeLiteral(int32_t{2})}, arrow::int32());
+  auto if_node = TreeExprBuilder::MakeIf(TreeExprBuilder::MakeField(condition_field),
+                                         make_sum(), else_node, arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      if_node, arrow::field("branch_cse_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(2, CountOccurrences(unoptimized_expr_ir, "call i32 @add_int32_int32"));
+
+  auto conditions = MakeArrowArrayBool({true, false, true, false});
+  auto left_values = MakeArrowArrayInt32({1, 2, 3, 4});
+  auto right_values = MakeArrowArrayInt32({10, 20, 30, 40});
+  auto expected = MakeArrowArrayInt32({11, 44, 33, 88});
+  ExpectProjectorOutput(projector, schema, {conditions, left_values, right_values},
+                        expected, pool_);
+}
+
+TEST_F(TestProjector, TestCommonSubexpressionRebuiltAfterIfMerge) {
+  auto condition = arrow::field("merge_cse_condition", arrow::boolean());
+  auto left = arrow::field("merge_cse_left", arrow::int32());
+  auto right = arrow::field("merge_cse_right", arrow::int32());
+  auto schema = arrow::schema({condition, left, right});
+
+  auto make_sum = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "add", {TreeExprBuilder::MakeField(left), TreeExprBuilder::MakeField(right)},
+        arrow::int32());
+  };
+  auto else_value = TreeExprBuilder::MakeFunction(
+      "add", {make_sum(), TreeExprBuilder::MakeLiteral(int32_t{1})}, arrow::int32());
+  auto conditional = TreeExprBuilder::MakeIf(TreeExprBuilder::MakeField(condition),
+                                             make_sum(), else_value, arrow::int32());
+  auto root =
+      TreeExprBuilder::MakeFunction("add", {conditional, make_sum()}, arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("merge_cse_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(5, CountOccurrences(unoptimized_expr_ir, "call i32 @add_int32_int32"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("res_value = phi"));
+
+  auto conditions =
+      MakeArrowArrayBool({true, false, false, true}, {true, true, false, true});
+  auto left_values = MakeArrowArrayInt32({1, 2, 3, 4}, {true, true, true, false});
+  auto right_values = MakeArrowArrayInt32({10, 20, 30, 40});
+  auto expected = MakeArrowArrayInt32({22, 45, 67, 0}, {true, true, true, false});
+  ExpectProjectorOutput(projector, schema, {conditions, left_values, right_values},
+                        expected, pool_);
+}
+
+TEST_F(TestProjector, TestUnsafeFunctionsAreNotCommoned) {
+  auto dividend = arrow::field("unsafe_cse_dividend", arrow::int32());
+  auto divisor = arrow::field("unsafe_cse_divisor", arrow::int32());
+  auto schema = arrow::schema({dividend, divisor});
+
+  auto make_divide = [&] {
+    return TreeExprBuilder::MakeFunction(
+        "divide",
+        {TreeExprBuilder::MakeField(dividend), TreeExprBuilder::MakeField(divisor)},
+        arrow::int32());
+  };
+  auto root = TreeExprBuilder::MakeFunction("add", {make_divide(), make_divide()},
+                                            arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("unsafe_cse_divide_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(2, CountOccurrences(unoptimized_expr_ir, "call i32 @divide_int32_int32"));
+
+  auto dividends = MakeArrowArrayInt32({4, 8});
+  auto divisors = MakeArrowArrayInt32({2, 0});
+  auto batch = arrow::RecordBatch::Make(schema, 2, {dividends, divisors});
+  arrow::ArrayVector outputs;
+  auto status = projector->Evaluate(*batch, pool_, &outputs);
+  EXPECT_EQ(StatusCode::ExecutionError, status.code());
+  EXPECT_NE(std::string::npos, status.message().find("divide by zero error"));
+}
+
+TEST_F(TestProjector, TestContextFunctionsAreNotCommoned) {
+  auto value = arrow::field("context_cse_value", arrow::int64());
+  auto schema = arrow::schema({value});
+
+  auto make_chr = [&] {
+    return TreeExprBuilder::MakeFunction("chr", {TreeExprBuilder::MakeField(value)},
+                                         arrow::utf8());
+  };
+  auto root =
+      TreeExprBuilder::MakeFunction("equal", {make_chr(), make_chr()}, arrow::boolean());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("context_cse_out", arrow::boolean()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(2, CountOccurrences(unoptimized_expr_ir, "@chr_int64("));
+
+  auto values = MakeArrowArrayInt64({65, 340, -5, 84}, {true, true, true, false});
+  auto expected =
+      MakeArrowArrayBool({true, true, true, false}, {true, true, true, false});
+  ExpectProjectorOutput(projector, schema, {values}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestHolderFunctionsAreNotCommoned) {
+  auto dummy = arrow::field("holder_cse_dummy", arrow::int32());
+  auto schema = arrow::schema({dummy});
+
+  auto make_random = [] {
+    return TreeExprBuilder::MakeFunction("random", {}, arrow::float64());
+  };
+  auto root = TreeExprBuilder::MakeFunction("subtract", {make_random(), make_random()},
+                                            arrow::float64());
+  auto expr = TreeExprBuilder::MakeExpression(
+      root, arrow::field("holder_cse_out", arrow::float64()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+  EXPECT_EQ(2, CountOccurrences(unoptimized_expr_ir, "call double @gdv_fn_random"));
+
+  constexpr int kNumRecords = 64;
+  auto dummy_values = MakeArrowArrayInt32(std::vector<int32_t>(kNumRecords, 0));
+  auto batch = arrow::RecordBatch::Make(schema, kNumRecords, {dummy_values});
+  arrow::ArrayVector outputs;
+  ASSERT_OK(projector->Evaluate(*batch, pool_, &outputs));
+  auto result = std::dynamic_pointer_cast<arrow::DoubleArray>(outputs[0]);
+  ASSERT_NE(nullptr, result);
+  ASSERT_EQ(0, result->null_count());
+  for (int64_t i = 0; i < result->length(); ++i) {
+    EXPECT_GT(result->Value(i), -1.0);
+    EXPECT_LT(result->Value(i), 1.0);
+  }
+}
+
+TEST_F(TestProjector, TestIfAlgebraicFoldIsNotApplied) {
+  auto condition_field = arrow::field("generated_if_cond", arrow::boolean());
+  auto value_field = arrow::field("generated_if_value", arrow::int32());
+  auto schema = arrow::schema({condition_field, value_field});
+
+  auto condition = TreeExprBuilder::MakeField(condition_field);
+  auto then_node = TreeExprBuilder::MakeField(value_field);
+  auto else_node = TreeExprBuilder::MakeField(value_field);
+  auto if_node = TreeExprBuilder::MakeIf(condition, then_node, else_node, arrow::int32());
+  auto expr = TreeExprBuilder::MakeExpression(
+      if_node, arrow::field("generated_if_out", arrow::int32()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("generated_if_value"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("generated_if_cond"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("then:"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("else:"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("validAndMatch"));
+  EXPECT_NE(std::string::npos, unoptimized_expr_ir.find("res_value = phi"));
+
+  auto conditions =
+      MakeArrowArrayBool({true, false, false, true}, {true, true, false, true});
+  auto values = MakeArrowArrayInt32({7, 8, 9, 10}, {true, false, true, true});
+  auto expected = MakeArrowArrayInt32({7, 0, 9, 10}, {true, false, true, true});
+  ExpectProjectorOutput(projector, schema, {conditions, values}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestBooleanAlgebraicFoldIsNotApplied) {
+  auto field0 = arrow::field("generated_bool_f0", arrow::boolean());
+  auto field1 = arrow::field("generated_bool_f1", arrow::boolean());
+  auto schema = arrow::schema({field0, field1});
+
+  auto make_and = [&]() {
+    return TreeExprBuilder::MakeAnd(
+        {TreeExprBuilder::MakeField(field0), TreeExprBuilder::MakeField(field1)});
+  };
+  auto bool_expr = TreeExprBuilder::MakeOr({make_and(), make_and()});
+  auto expr = TreeExprBuilder::MakeExpression(
+      bool_expr, arrow::field("generated_bool_out", arrow::boolean()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+
+  EXPECT_GT(CountOccurrences(unoptimized_expr_ir, "short_circuit"), 0);
+  EXPECT_GT(CountOccurrences(unoptimized_expr_ir, "non_short_circuit"), 0);
+  EXPECT_GT(CountOccurrences(unoptimized_expr_ir, "res_value = phi"), 0);
+
+  auto input0 = MakeArrowArrayBool({true, true, false, false, false},
+                                   {true, true, true, false, false});
+  auto input1 = MakeArrowArrayBool({true, false, true, true, false},
+                                   {true, false, false, true, true});
+  auto expected = MakeArrowArrayBool({true, false, false, false, false},
+                                     {true, false, true, false, true});
+  ExpectProjectorOutput(projector, schema, {input0, input1}, expected, pool_);
+}
+
+TEST_F(TestProjector, TestNestedBetweenDoesNotReuseAcrossBooleanControlFlow) {
+  auto value = arrow::field("nested_between_value", arrow::int32());
+  auto lower = arrow::field("nested_between_lower", arrow::int32());
+  auto upper = arrow::field("nested_between_upper", arrow::int32());
+  auto schema = arrow::schema({value, lower, upper});
+
+  auto make_between = [&]() {
+    auto ge_lower = TreeExprBuilder::MakeFunction(
+        "greater_than_or_equal_to",
+        {TreeExprBuilder::MakeField(value), TreeExprBuilder::MakeField(lower)},
+        arrow::boolean());
+    auto le_upper = TreeExprBuilder::MakeFunction(
+        "less_than_or_equal_to",
+        {TreeExprBuilder::MakeField(value), TreeExprBuilder::MakeField(upper)},
+        arrow::boolean());
+    return TreeExprBuilder::MakeAnd({ge_lower, le_upper});
+  };
+
+  auto nested_between = TreeExprBuilder::MakeAnd(
+      {make_between(), TreeExprBuilder::MakeOr({make_between(), make_between()})});
+  auto expr = TreeExprBuilder::MakeExpression(
+      nested_between, arrow::field("nested_between_out", arrow::boolean()));
+
+  auto configuration = std::make_shared<Configuration>(
+      true, gandiva::default_function_registry(), /*dump_ir=*/true);
+  std::shared_ptr<Projector> projector;
+  ASSERT_OK(Projector::Make(schema, {expr}, configuration, &projector));
+
+  ASSERT_OK_AND_ASSIGN(auto unoptimized_ir, projector->DumpUnoptimizedIR());
+  const auto unoptimized_expr_ir = ExtractFunctionIR(unoptimized_ir, "expr_0_0");
+  ASSERT_FALSE(unoptimized_expr_ir.empty());
+
+  EXPECT_EQ(3, CountOccurrences(unoptimized_expr_ir,
+                                "call i1 @greater_than_or_equal_to_int32_int32"));
+  EXPECT_EQ(3, CountOccurrences(unoptimized_expr_ir,
+                                "call i1 @less_than_or_equal_to_int32_int32"));
+
+  auto values = MakeArrowArrayInt32({5, 1, 10, 7, 4}, {true, true, true, true, false});
+  auto lowers = MakeArrowArrayInt32({1, 2, 10, 6, 0}, {true, true, true, false, true});
+  auto uppers = MakeArrowArrayInt32({10, 5, 10, 9, 10});
+  auto expected = MakeArrowArrayBool({true, false, true, false, false},
+                                     {true, true, true, false, false});
+  ExpectProjectorOutput(projector, schema, {values, lowers, uppers}, expected, pool_);
+}
+
 TEST_F(TestProjector, TestExtendedMath) {
 #ifdef __aarch64__
   GTEST_SKIP() << "Failed on aarch64 with 'JIT session error: Symbols not found: [ "