diff --git a/datafusion-cli/Cargo.lock b/datafusion-cli/Cargo.lock
index deda497d9dd34..8e2a2c353e2d5 100644
--- a/datafusion-cli/Cargo.lock
+++ b/datafusion-cli/Cargo.lock
@@ -378,13 +378,13 @@ dependencies = [
 
 [[package]]
 name = "async-trait"
-version = "0.1.77"
+version = "0.1.78"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c980ee35e870bd1a4d2c8294d4c04d0499e67bca1e4b5cefcc693c2fa00caea9"
+checksum = "461abc97219de0eaaf81fe3ef974a540158f3d079c2ab200f891f1a2ef201e85"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -776,9 +776,9 @@ dependencies = [
 
 [[package]]
 name = "brotli"
-version = "3.4.0"
+version = "3.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "516074a47ef4bce09577a3b379392300159ce5b1ba2e501ff1c819950066100f"
+checksum = "d640d25bc63c50fb1f0b545ffd80207d2e10a4c965530809b40ba3386825c391"
 dependencies = [
  "alloc-no-stdlib",
  "alloc-stdlib",
@@ -1080,7 +1080,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ad291aa74992b9b7a7e88c38acbbf6ad7e107f1d90ee8775b7bc1fc3394f485c"
 dependencies = [
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -1270,6 +1270,7 @@ dependencies = [
  "arrow",
  "arrow-array",
  "arrow-buffer",
+ "arrow-ord",
  "arrow-schema",
  "datafusion-common",
  "datafusion-execution",
@@ -1639,7 +1640,7 @@ checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -1713,9 +1714,9 @@ checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"
 
 [[package]]
 name = "h2"
-version = "0.3.24"
+version = "0.3.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb2c4422095b67ee78da96fbb51a4cc413b3b25883c7717ff7ca1ab31022c9c9"
+checksum = "4fbd2820c5e49886948654ab546d0688ff24530286bdcf8fca3cefb16d4618eb"
 dependencies = [
  "bytes",
  "fnv",
@@ -2560,7 +2561,7 @@ checksum = "2f38a4412a78282e09a2cf38d195ea5420d15ba0602cb375210efbc877243965"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3101,7 +3102,7 @@ checksum = "7eb0b34b42edc17f6b7cac84a52a1c5f0e1bb2227e997ca9011ea3dd34e8610b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3236,7 +3237,7 @@ checksum = "01b2e185515564f15375f593fb966b5718bc624ba77fe49fa4616ad619690554"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3282,7 +3283,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3295,7 +3296,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustversion",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3317,9 +3318,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.52"
+version = "2.0.53"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b699d15b36d1f02c3e7c69f8ffef53de37aefae075d8488d4ba1a7788d574a07"
+checksum = "7383cd0e49fff4b6b90ca5670bfd3e9d6a733b3f90c686605aa7eec8c4996032"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3403,7 +3404,7 @@ checksum = "c61f3ba182994efc43764a46c018c347bc492c79f024e705f46567b418f6d4f7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3498,7 +3499,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3595,7 +3596,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3640,7 +3641,7 @@ checksum = "f03ca4cb38206e2bef0700092660bb74d696f808514dae47fa1467cbfe26e96e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
@@ -3794,7 +3795,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
  "wasm-bindgen-shared",
 ]
 
@@ -3828,7 +3829,7 @@ checksum = "e94f17b526d0a461a191c78ea52bbce64071ed5c04c9ffe424dcb38f74171bb7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -4086,7 +4087,7 @@ checksum = "9ce1b18ccd8e73a9321186f97e46f9f04b778851177567b1975109d26a08d2a6"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.52",
+ "syn 2.0.53",
 ]
 
 [[package]]
diff --git a/datafusion/core/benches/array_expression.rs b/datafusion/core/benches/array_expression.rs
index 95bc93e0e353a..c980329620aa6 100644
--- a/datafusion/core/benches/array_expression.rs
+++ b/datafusion/core/benches/array_expression.rs
@@ -22,48 +22,32 @@ extern crate datafusion;
 
 mod data_utils;
 use crate::criterion::Criterion;
-use arrow_array::cast::AsArray;
-use arrow_array::types::Int64Type;
-use arrow_array::{ArrayRef, Int64Array, ListArray};
-use datafusion_physical_expr::array_expressions;
-use std::sync::Arc;
+use datafusion::functions_array::expr_fn::{array_replace_all, make_array};
+use datafusion_expr::lit;
 
 fn criterion_benchmark(c: &mut Criterion) {
     // Construct large arrays for benchmarking
 
     let array_len = 100000000;
 
-    let array = (0..array_len).map(|_| Some(2_i64)).collect::<Vec<_>>();
-    let list_array = ListArray::from_iter_primitive::<Int64Type, _, _>(vec![
-        Some(array.clone()),
-        Some(array.clone()),
-        Some(array),
-    ]);
-    let from_array = Int64Array::from_value(2, 3);
-    let to_array = Int64Array::from_value(-2, 3);
+    let array = (0..array_len).map(|_| lit(2_i64)).collect::<Vec<_>>();
+    let list_array = make_array(vec![make_array(array); 3]);
+    let from_array = make_array(vec![lit(2_i64); 3]);
+    let to_array = make_array(vec![lit(-2_i64); 3]);
 
-    let args = vec![
-        Arc::new(list_array) as ArrayRef,
-        Arc::new(from_array) as ArrayRef,
-        Arc::new(to_array) as ArrayRef,
-    ];
-
-    let array = (0..array_len).map(|_| Some(-2_i64)).collect::<Vec<_>>();
-    let expected_array = ListArray::from_iter_primitive::<Int64Type, _, _>(vec![
-        Some(array.clone()),
-        Some(array.clone()),
-        Some(array),
-    ]);
+    let expected_array = list_array.clone();
 
     // Benchmark array functions
 
     c.bench_function("array_replace", |b| {
         b.iter(|| {
             assert_eq!(
-                array_expressions::array_replace_all(args.as_slice())
-                    .unwrap()
-                    .as_list::<i32>(),
-                criterion::black_box(&expected_array)
+                array_replace_all(
+                    list_array.clone(),
+                    from_array.clone(),
+                    to_array.clone()
+                ),
+                *criterion::black_box(&expected_array)
             )
         })
     });
diff --git a/datafusion/expr/src/built_in_function.rs b/datafusion/expr/src/built_in_function.rs
index fe3397b1af525..79cd6a24ce39d 100644
--- a/datafusion/expr/src/built_in_function.rs
+++ b/datafusion/expr/src/built_in_function.rs
@@ -102,14 +102,6 @@ pub enum BuiltinScalarFunction {
     /// cot
     Cot,
 
-    // array functions
-    /// array_replace
-    ArrayReplace,
-    /// array_replace_n
-    ArrayReplaceN,
-    /// array_replace_all
-    ArrayReplaceAll,
-
     // string functions
     /// ascii
     Ascii,
@@ -262,9 +254,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Cbrt => Volatility::Immutable,
             BuiltinScalarFunction::Cot => Volatility::Immutable,
             BuiltinScalarFunction::Trunc => Volatility::Immutable,
-            BuiltinScalarFunction::ArrayReplace => Volatility::Immutable,
-            BuiltinScalarFunction::ArrayReplaceN => Volatility::Immutable,
-            BuiltinScalarFunction::ArrayReplaceAll => Volatility::Immutable,
             BuiltinScalarFunction::Ascii => Volatility::Immutable,
             BuiltinScalarFunction::BitLength => Volatility::Immutable,
             BuiltinScalarFunction::Btrim => Volatility::Immutable,
@@ -322,9 +311,6 @@ impl BuiltinScalarFunction {
         // the return type of the built in function.
         // Some built-in functions' return type depends on the incoming type.
         match self {
-            BuiltinScalarFunction::ArrayReplace => Ok(input_expr_types[0].clone()),
-            BuiltinScalarFunction::ArrayReplaceN => Ok(input_expr_types[0].clone()),
-            BuiltinScalarFunction::ArrayReplaceAll => Ok(input_expr_types[0].clone()),
             BuiltinScalarFunction::Ascii => Ok(Int32),
             BuiltinScalarFunction::BitLength => {
                 utf8_to_int_type(&input_expr_types[0], "bit_length")
@@ -477,11 +463,6 @@ impl BuiltinScalarFunction {
 
         // for now, the list is small, as we do not have many built-in functions.
         match self {
-            BuiltinScalarFunction::ArrayReplace => Signature::any(3, self.volatility()),
-            BuiltinScalarFunction::ArrayReplaceN => Signature::any(4, self.volatility()),
-            BuiltinScalarFunction::ArrayReplaceAll => {
-                Signature::any(3, self.volatility())
-            }
             BuiltinScalarFunction::Concat
             | BuiltinScalarFunction::ConcatWithSeparator => {
                 Signature::variadic(vec![Utf8], self.volatility())
@@ -779,15 +760,6 @@ impl BuiltinScalarFunction {
             BuiltinScalarFunction::Levenshtein => &["levenshtein"],
             BuiltinScalarFunction::SubstrIndex => &["substr_index", "substring_index"],
             BuiltinScalarFunction::FindInSet => &["find_in_set"],
-
-            // hashing functions
-            BuiltinScalarFunction::ArrayReplace => &["array_replace", "list_replace"],
-            BuiltinScalarFunction::ArrayReplaceN => {
-                &["array_replace_n", "list_replace_n"]
-            }
-            BuiltinScalarFunction::ArrayReplaceAll => {
-                &["array_replace_all", "list_replace_all"]
-            }
             BuiltinScalarFunction::OverLay => &["overlay"],
         }
     }
diff --git a/datafusion/expr/src/expr_fn.rs b/datafusion/expr/src/expr_fn.rs
index c5ad2a9b3ce49..b76164a1c83c6 100644
--- a/datafusion/expr/src/expr_fn.rs
+++ b/datafusion/expr/src/expr_fn.rs
@@ -584,25 +584,6 @@ scalar_expr!(
 scalar_expr!(Uuid, uuid, , "returns uuid v4 as a string value");
 scalar_expr!(Log, log, base x, "logarithm of a `x` for a particular `base`");
 
-scalar_expr!(
-    ArrayReplace,
-    array_replace,
-    array from to,
-    "replaces the first occurrence of the specified element with another specified element."
-);
-scalar_expr!(
-    ArrayReplaceN,
-    array_replace_n,
-    array from to max,
-    "replaces the first `max` occurrences of the specified element with another specified element."
-);
-scalar_expr!(
-    ArrayReplaceAll,
-    array_replace_all,
-    array from to,
-    "replaces all occurrences of the specified element with another specified element."
-);
-
 // string functions
 scalar_expr!(Ascii, ascii, chr, "ASCII code value of the character");
 scalar_expr!(
@@ -1145,11 +1126,6 @@ mod test {
         test_scalar_expr!(Translate, translate, string, from, to);
         test_scalar_expr!(Trim, trim, string);
         test_scalar_expr!(Upper, upper, string);
-
-        test_scalar_expr!(ArrayReplace, array_replace, array, from, to);
-        test_scalar_expr!(ArrayReplaceN, array_replace_n, array, from, to, max);
-        test_scalar_expr!(ArrayReplaceAll, array_replace_all, array, from, to);
-
         test_nary_scalar_expr!(OverLay, overlay, string, characters, position, len);
         test_nary_scalar_expr!(OverLay, overlay, string, characters, position);
         test_scalar_expr!(Levenshtein, levenshtein, string1, string2);
diff --git a/datafusion/functions-array/Cargo.toml b/datafusion/functions-array/Cargo.toml
index 99239ffb3bdc3..80c0e5e187687 100644
--- a/datafusion/functions-array/Cargo.toml
+++ b/datafusion/functions-array/Cargo.toml
@@ -40,6 +40,7 @@ path = "src/lib.rs"
 arrow = { workspace = true }
 arrow-array = { workspace = true }
 arrow-buffer = { workspace = true }
+arrow-ord = { workspace = true }
 arrow-schema = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-execution = { workspace = true }
diff --git a/datafusion/functions-array/src/core.rs b/datafusion/functions-array/src/core.rs
index 4c84b7018c99b..fdd127cc3f32b 100644
--- a/datafusion/functions-array/src/core.rs
+++ b/datafusion/functions-array/src/core.rs
@@ -96,7 +96,7 @@ impl ScalarUDFImpl for MakeArray {
         }
     }
 
-    fn invoke(&self, args: &[ColumnarValue]) -> datafusion_common::Result<ColumnarValue> {
+    fn invoke(&self, args: &[ColumnarValue]) -> Result<ColumnarValue> {
         make_scalar_function(make_array_inner)(args)
     }
 
diff --git a/datafusion/functions-array/src/lib.rs b/datafusion/functions-array/src/lib.rs
index 2c19dfad62223..fb16acdef2bd4 100644
--- a/datafusion/functions-array/src/lib.rs
+++ b/datafusion/functions-array/src/lib.rs
@@ -36,6 +36,7 @@ mod extract;
 mod kernels;
 mod position;
 mod remove;
+mod replace;
 mod rewrite;
 mod set_ops;
 mod udf;
@@ -66,6 +67,9 @@ pub mod expr_fn {
     pub use super::remove::array_remove;
     pub use super::remove::array_remove_all;
     pub use super::remove::array_remove_n;
+    pub use super::replace::array_replace;
+    pub use super::replace::array_replace_all;
+    pub use super::replace::array_replace_n;
     pub use super::set_ops::array_distinct;
     pub use super::set_ops::array_intersect;
     pub use super::set_ops::array_union;
@@ -120,8 +124,11 @@ pub fn register_all(registry: &mut dyn FunctionRegistry) -> Result<()> {
         position::array_position_udf(),
         position::array_positions_udf(),
         remove::array_remove_udf(),
-        remove::array_remove_n_udf(),
         remove::array_remove_all_udf(),
+        remove::array_remove_n_udf(),
+        replace::array_replace_n_udf(),
+        replace::array_replace_all_udf(),
+        replace::array_replace_udf(),
     ];
     functions.into_iter().try_for_each(|udf| {
         let existing_udf = registry.register_udf(udf)?;
diff --git a/datafusion/functions-array/src/position.rs b/datafusion/functions-array/src/position.rs
index 4988e0ded1061..627cf3cb0cf0b 100644
--- a/datafusion/functions-array/src/position.rs
+++ b/datafusion/functions-array/src/position.rs
@@ -27,8 +27,7 @@ use std::sync::Arc;
 
 use arrow_array::types::UInt64Type;
 use arrow_array::{
-    Array, ArrayRef, BooleanArray, GenericListArray, ListArray, OffsetSizeTrait, Scalar,
-    UInt32Array, UInt64Array,
+    Array, ArrayRef, GenericListArray, ListArray, OffsetSizeTrait, UInt64Array,
 };
 use datafusion_common::cast::{
     as_generic_list_array, as_int64_array, as_large_list_array, as_list_array,
@@ -36,6 +35,8 @@ use datafusion_common::cast::{
 use datafusion_common::{exec_err, internal_err};
 use itertools::Itertools;
 
+use crate::utils::compare_element_to_list;
+
 make_udf_function!(
     ArrayPosition,
     array_position,
@@ -173,107 +174,6 @@ fn generic_position<OffsetSize: OffsetSizeTrait>(
     Ok(Arc::new(UInt64Array::from(data)))
 }
 
-/// Computes a BooleanArray indicating equality or inequality between elements in a list array and a specified element array.
-///
-/// # Arguments
-///
-/// * `list_array_row` - A reference to a trait object implementing the Arrow `Array` trait. It represents the list array for which the equality or inequality will be compared.
-///
-/// * `element_array` - A reference to a trait object implementing the Arrow `Array` trait. It represents the array with which each element in the `list_array_row` will be compared.
-///
-/// * `row_index` - The index of the row in the `element_array` and `list_array` to use for the comparison.
-///
-/// * `eq` - A boolean flag. If `true`, the function computes equality; if `false`, it computes inequality.
-///
-/// # Returns
-///
-/// Returns a `Result<BooleanArray>` representing the comparison results. The result may contain an error if there are issues with the computation.
-///
-/// # Example
-///
-/// ```text
-/// compare_element_to_list(
-///     [1, 2, 3], [1, 2, 3], 0, true => [true, false, false]
-///     [1, 2, 3, 3, 2, 1], [1, 2, 3], 1, true => [false, true, false, false, true, false]
-///
-///     [[1, 2, 3], [2, 3, 4], [3, 4, 5]], [[1, 2, 3], [2, 3, 4], [3, 4, 5]], 0, true => [true, false, false]
-///     [[1, 2, 3], [2, 3, 4], [2, 3, 4]], [[1, 2, 3], [2, 3, 4], [3, 4, 5]], 1, false => [true, false, false]
-/// )
-/// ```
-fn compare_element_to_list(
-    list_array_row: &dyn Array,
-    element_array: &dyn Array,
-    row_index: usize,
-    eq: bool,
-) -> datafusion_common::Result<BooleanArray> {
-    if list_array_row.data_type() != element_array.data_type() {
-        return exec_err!(
-            "compare_element_to_list received incompatible types: '{:?}' and '{:?}'.",
-            list_array_row.data_type(),
-            element_array.data_type()
-        );
-    }
-
-    let indices = UInt32Array::from(vec![row_index as u32]);
-    let element_array_row = arrow::compute::take(element_array, &indices, None)?;
-
-    // Compute all positions in list_row_array (that is itself an
-    // array) that are equal to `from_array_row`
-    let res = match element_array_row.data_type() {
-        // arrow_ord::cmp::eq does not support ListArray, so we need to compare it by loop
-        DataType::List(_) => {
-            // compare each element of the from array
-            let element_array_row_inner = as_list_array(&element_array_row)?.value(0);
-            let list_array_row_inner = as_list_array(list_array_row)?;
-
-            list_array_row_inner
-                .iter()
-                // compare element by element the current row of list_array
-                .map(|row| {
-                    row.map(|row| {
-                        if eq {
-                            row.eq(&element_array_row_inner)
-                        } else {
-                            row.ne(&element_array_row_inner)
-                        }
-                    })
-                })
-                .collect::<BooleanArray>()
-        }
-        DataType::LargeList(_) => {
-            // compare each element of the from array
-            let element_array_row_inner =
-                as_large_list_array(&element_array_row)?.value(0);
-            let list_array_row_inner = as_large_list_array(list_array_row)?;
-
-            list_array_row_inner
-                .iter()
-                // compare element by element the current row of list_array
-                .map(|row| {
-                    row.map(|row| {
-                        if eq {
-                            row.eq(&element_array_row_inner)
-                        } else {
-                            row.ne(&element_array_row_inner)
-                        }
-                    })
-                })
-                .collect::<BooleanArray>()
-        }
-        _ => {
-            let element_arr = Scalar::new(element_array_row);
-            // use not_distinct so we can compare NULL
-            if eq {
-                arrow::compute::kernels::cmp::not_distinct(&list_array_row, &element_arr)?
-            } else {
-                arrow::compute::kernels::cmp::distinct(&list_array_row, &element_arr)?
-            }
-        }
-    };
-
-    Ok(res)
-}
-
 make_udf_function!(
     ArrayPositions,
     array_positions,
diff --git a/datafusion/functions-array/src/replace.rs b/datafusion/functions-array/src/replace.rs
new file mode 100644
index 0000000000000..8ff65d3154318
--- /dev/null
+++ b/datafusion/functions-array/src/replace.rs
@@ -0,0 +1,362 @@
+// 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.
+
+//! [`ScalarUDFImpl`] definitions for array functions.
+
+use arrow::array::{
+    Array, ArrayRef, AsArray, Capacities, MutableArrayData, OffsetSizeTrait,
+};
+use arrow::datatypes::DataType;
+
+use arrow_array::GenericListArray;
+use arrow_buffer::{BooleanBufferBuilder, NullBuffer, OffsetBuffer};
+use arrow_schema::Field;
+use datafusion_common::cast::as_int64_array;
+use datafusion_common::{exec_err, Result};
+use datafusion_expr::expr::ScalarFunction;
+use datafusion_expr::Expr;
+use datafusion_expr::{ColumnarValue, ScalarUDFImpl, Signature, Volatility};
+
+use crate::utils::compare_element_to_list;
+use crate::utils::make_scalar_function;
+
+use std::any::Any;
+use std::sync::Arc;
+
+// Create static instances of ScalarUDFs for each function
+make_udf_function!(ArrayReplace,
+    array_replace,
+    array from to,
+    "replaces the first occurrence of the specified element with another specified element.",
+    array_replace_udf
+);
+make_udf_function!(ArrayReplaceN,
+    array_replace_n,
+    array from to max,
+    "replaces the first `max` occurrences of the specified element with another specified element.",
+    array_replace_n_udf
+);
+make_udf_function!(ArrayReplaceAll,
+    array_replace_all,
+    array from to,
+    "replaces all occurrences of the specified element with another specified element.",
+    array_replace_all_udf
+);
+
+#[derive(Debug)]
+pub(super) struct ArrayReplace {
+    signature: Signature,
+    aliases: Vec<String>,
+}
+
+impl ArrayReplace {
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::any(3, Volatility::Immutable),
+            aliases: vec![String::from("array_replace"), String::from("list_replace")],
+        }
+    }
+}
+
+impl ScalarUDFImpl for ArrayReplace {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn name(&self) -> &str {
+        "array_replace"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, args: &[DataType]) -> datafusion_common::Result<DataType> {
+        Ok(args[0].clone())
+    }
+
+    fn invoke(&self, args: &[ColumnarValue]) -> datafusion_common::Result<ColumnarValue> {
+        make_scalar_function(array_replace_inner)(args)
+    }
+
+    fn aliases(&self) -> &[String] {
+        &self.aliases
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct ArrayReplaceN {
+    signature: Signature,
+    aliases: Vec<String>,
+}
+
+impl ArrayReplaceN {
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::any(4, Volatility::Immutable),
+            aliases: vec![
+                String::from("array_replace_n"),
+                String::from("list_replace_n"),
+            ],
+        }
+    }
+}
+
+impl ScalarUDFImpl for ArrayReplaceN {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn name(&self) -> &str {
+        "array_replace_n"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, args: &[DataType]) -> datafusion_common::Result<DataType> {
+        Ok(args[0].clone())
+    }
+
+    fn invoke(&self, args: &[ColumnarValue]) -> datafusion_common::Result<ColumnarValue> {
+        make_scalar_function(array_replace_n_inner)(args)
+    }
+
+    fn aliases(&self) -> &[String] {
+        &self.aliases
+    }
+}
+
+#[derive(Debug)]
+pub(super) struct ArrayReplaceAll {
+    signature: Signature,
+    aliases: Vec<String>,
+}
+
+impl ArrayReplaceAll {
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::any(3, Volatility::Immutable),
+            aliases: vec![
+                String::from("array_replace_all"),
+                String::from("list_replace_all"),
+            ],
+        }
+    }
+}
+
+impl ScalarUDFImpl for ArrayReplaceAll {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+    fn name(&self) -> &str {
+        "array_replace_all"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, args: &[DataType]) -> datafusion_common::Result<DataType> {
+        Ok(args[0].clone())
+    }
+
+    fn invoke(&self, args: &[ColumnarValue]) -> datafusion_common::Result<ColumnarValue> {
+        make_scalar_function(array_replace_all_inner)(args)
+    }
+
+    fn aliases(&self) -> &[String] {
+        &self.aliases
+    }
+}
+
+/// For each element of `list_array[i]`, replaces up to `arr_n[i]`  occurences
+/// of `from_array[i]`, `to_array[i]`.
+///
+/// The type of each **element** in `list_array` must be the same as the type of
+/// `from_array` and `to_array`. This function also handles nested arrays
+/// (\[`ListArray`\] of \[`ListArray`\]s)
+///
+/// For example, when called to replace a list array (where each element is a
+/// list of int32s, the second and third argument are int32 arrays, and the
+/// fourth argument is the number of occurrences to replace
+///
+/// ```text
+/// general_replace(
+///   [1, 2, 3, 2], 2, 10, 1    ==> [1, 10, 3, 2]   (only the first 2 is replaced)
+///   [4, 5, 6, 5], 5, 20, 2    ==> [4, 20, 6, 20]  (both 5s are replaced)
+/// )
+/// ```
+fn general_replace<O: OffsetSizeTrait>(
+    list_array: &GenericListArray<O>,
+    from_array: &ArrayRef,
+    to_array: &ArrayRef,
+    arr_n: Vec<i64>,
+) -> Result<ArrayRef> {
+    // Build up the offsets for the final output array
+    let mut offsets: Vec<O> = vec![O::usize_as(0)];
+    let values = list_array.values();
+    let original_data = values.to_data();
+    let to_data = to_array.to_data();
+    let capacity = Capacities::Array(original_data.len());
+
+    // First array is the original array, second array is the element to replace with.
+    let mut mutable = MutableArrayData::with_capacities(
+        vec![&original_data, &to_data],
+        false,
+        capacity,
+    );
+
+    let mut valid = BooleanBufferBuilder::new(list_array.len());
+
+    for (row_index, offset_window) in list_array.offsets().windows(2).enumerate() {
+        if list_array.is_null(row_index) {
+            offsets.push(offsets[row_index]);
+            valid.append(false);
+            continue;
+        }
+
+        let start = offset_window[0];
+        let end = offset_window[1];
+
+        let list_array_row = list_array.value(row_index);
+
+        // Compute all positions in list_row_array (that is itself an
+        // array) that are equal to `from_array_row`
+        let eq_array =
+            compare_element_to_list(&list_array_row, &from_array, row_index, true)?;
+
+        let original_idx = O::usize_as(0);
+        let replace_idx = O::usize_as(1);
+        let n = arr_n[row_index];
+        let mut counter = 0;
+
+        // All elements are false, no need to replace, just copy original data
+        if eq_array.false_count() == eq_array.len() {
+            mutable.extend(
+                original_idx.to_usize().unwrap(),
+                start.to_usize().unwrap(),
+                end.to_usize().unwrap(),
+            );
+            offsets.push(offsets[row_index] + (end - start));
+            valid.append(true);
+            continue;
+        }
+
+        for (i, to_replace) in eq_array.iter().enumerate() {
+            let i = O::usize_as(i);
+            if let Some(true) = to_replace {
+                mutable.extend(replace_idx.to_usize().unwrap(), row_index, row_index + 1);
+                counter += 1;
+                if counter == n {
+                    // copy original data for any matches past n
+                    mutable.extend(
+                        original_idx.to_usize().unwrap(),
+                        (start + i).to_usize().unwrap() + 1,
+                        end.to_usize().unwrap(),
+                    );
+                    break;
+                }
+            } else {
+                // copy original data for false / null matches
+                mutable.extend(
+                    original_idx.to_usize().unwrap(),
+                    (start + i).to_usize().unwrap(),
+                    (start + i).to_usize().unwrap() + 1,
+                );
+            }
+        }
+
+        offsets.push(offsets[row_index] + (end - start));
+        valid.append(true);
+    }
+
+    let data = mutable.freeze();
+
+    Ok(Arc::new(GenericListArray::<O>::try_new(
+        Arc::new(Field::new("item", list_array.value_type(), true)),
+        OffsetBuffer::<O>::new(offsets.into()),
+        arrow_array::make_array(data),
+        Some(NullBuffer::new(valid.finish())),
+    )?))
+}
+
+pub(crate) fn array_replace_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
+    if args.len() != 3 {
+        return exec_err!("array_replace expects three arguments");
+    }
+
+    // replace at most one occurence for each element
+    let arr_n = vec![1; args[0].len()];
+    let array = &args[0];
+    match array.data_type() {
+        DataType::List(_) => {
+            let list_array = array.as_list::<i32>();
+            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
+        }
+        DataType::LargeList(_) => {
+            let list_array = array.as_list::<i64>();
+            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
+        }
+        array_type => exec_err!("array_replace does not support type '{array_type:?}'."),
+    }
+}
+
+pub(crate) fn array_replace_n_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
+    if args.len() != 4 {
+        return exec_err!("array_replace_n expects four arguments");
+    }
+
+    // replace the specified number of occurences
+    let arr_n = as_int64_array(&args[3])?.values().to_vec();
+    let array = &args[0];
+    match array.data_type() {
+        DataType::List(_) => {
+            let list_array = array.as_list::<i32>();
+            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
+        }
+        DataType::LargeList(_) => {
+            let list_array = array.as_list::<i64>();
+            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
+        }
+        array_type => {
+            exec_err!("array_replace_n does not support type '{array_type:?}'.")
+        }
+    }
+}
+
+pub(crate) fn array_replace_all_inner(args: &[ArrayRef]) -> Result<ArrayRef> {
+    if args.len() != 3 {
+        return exec_err!("array_replace_all expects three arguments");
+    }
+
+    // replace all occurrences (up to "i64::MAX")
+    let arr_n = vec![i64::MAX; args[0].len()];
+    let array = &args[0];
+    match array.data_type() {
+        DataType::List(_) => {
+            let list_array = array.as_list::<i32>();
+            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
+        }
+        DataType::LargeList(_) => {
+            let list_array = array.as_list::<i64>();
+            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
+        }
+        array_type => {
+            exec_err!("array_replace_all does not support type '{array_type:?}'.")
+        }
+    }
+}
diff --git a/datafusion/functions-array/src/utils.rs b/datafusion/functions-array/src/utils.rs
index ad613163c6afd..9589cb05fe9ba 100644
--- a/datafusion/functions-array/src/utils.rs
+++ b/datafusion/functions-array/src/utils.rs
@@ -20,6 +20,7 @@
 use std::sync::Arc;
 
 use arrow::{array::ArrayRef, datatypes::DataType};
+
 use arrow_array::{
     Array, BooleanArray, GenericListArray, OffsetSizeTrait, Scalar, UInt32Array,
 };
@@ -27,6 +28,7 @@ use arrow_buffer::OffsetBuffer;
 use arrow_schema::Field;
 use datafusion_common::cast::{as_large_list_array, as_list_array};
 use datafusion_common::{exec_err, plan_err, Result, ScalarValue};
+
 use datafusion_expr::{ColumnarValue, ScalarFunctionImplementation};
 
 pub(crate) fn check_datatypes(name: &str, args: &[&ArrayRef]) -> Result<()> {
@@ -202,9 +204,9 @@ pub(crate) fn compare_element_to_list(
             let element_arr = Scalar::new(element_array_row);
             // use not_distinct so we can compare NULL
             if eq {
-                arrow::compute::kernels::cmp::not_distinct(&list_array_row, &element_arr)?
+                arrow_ord::cmp::not_distinct(&list_array_row, &element_arr)?
             } else {
-                arrow::compute::kernels::cmp::distinct(&list_array_row, &element_arr)?
+                arrow_ord::cmp::distinct(&list_array_row, &element_arr)?
             }
         }
     };
diff --git a/datafusion/physical-expr/src/array_expressions.rs b/datafusion/physical-expr/src/array_expressions.rs
deleted file mode 100644
index c3c0f4c822820..0000000000000
--- a/datafusion/physical-expr/src/array_expressions.rs
+++ /dev/null
@@ -1,423 +0,0 @@
-// 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.
-
-//! Array expressions
-
-use std::sync::Arc;
-
-use arrow::array::*;
-use arrow::buffer::OffsetBuffer;
-use arrow::datatypes::{DataType, Field};
-use arrow_buffer::NullBuffer;
-
-use datafusion_common::cast::{as_int64_array, as_large_list_array, as_list_array};
-use datafusion_common::utils::array_into_list_array;
-use datafusion_common::{exec_err, plan_err, Result};
-
-/// Computes a BooleanArray indicating equality or inequality between elements in a list array and a specified element array.
-///
-/// # Arguments
-///
-/// * `list_array_row` - A reference to a trait object implementing the Arrow `Array` trait. It represents the list array for which the equality or inequality will be compared.
-///
-/// * `element_array` - A reference to a trait object implementing the Arrow `Array` trait. It represents the array with which each element in the `list_array_row` will be compared.
-///
-/// * `row_index` - The index of the row in the `element_array` and `list_array` to use for the comparison.
-///
-/// * `eq` - A boolean flag. If `true`, the function computes equality; if `false`, it computes inequality.
-///
-/// # Returns
-///
-/// Returns a `Result<BooleanArray>` representing the comparison results. The result may contain an error if there are issues with the computation.
-///
-/// # Example
-///
-/// ```text
-/// compare_element_to_list(
-///     [1, 2, 3], [1, 2, 3], 0, true => [true, false, false]
-///     [1, 2, 3, 3, 2, 1], [1, 2, 3], 1, true => [false, true, false, false, true, false]
-///
-///     [[1, 2, 3], [2, 3, 4], [3, 4, 5]], [[1, 2, 3], [2, 3, 4], [3, 4, 5]], 0, true => [true, false, false]
-///     [[1, 2, 3], [2, 3, 4], [2, 3, 4]], [[1, 2, 3], [2, 3, 4], [3, 4, 5]], 1, false => [true, false, false]
-/// )
-/// ```
-fn compare_element_to_list(
-    list_array_row: &dyn Array,
-    element_array: &dyn Array,
-    row_index: usize,
-    eq: bool,
-) -> Result<BooleanArray> {
-    if list_array_row.data_type() != element_array.data_type() {
-        return exec_err!(
-            "compare_element_to_list received incompatible types: '{:?}' and '{:?}'.",
-            list_array_row.data_type(),
-            element_array.data_type()
-        );
-    }
-
-    let indices = UInt32Array::from(vec![row_index as u32]);
-    let element_array_row = arrow::compute::take(element_array, &indices, None)?;
-
-    // Compute all positions in list_row_array (that is itself an
-    // array) that are equal to `from_array_row`
-    let res = match element_array_row.data_type() {
-        // arrow_ord::cmp::eq does not support ListArray, so we need to compare it by loop
-        DataType::List(_) => {
-            // compare each element of the from array
-            let element_array_row_inner = as_list_array(&element_array_row)?.value(0);
-            let list_array_row_inner = as_list_array(list_array_row)?;
-
-            list_array_row_inner
-                .iter()
-                // compare element by element the current row of list_array
-                .map(|row| {
-                    row.map(|row| {
-                        if eq {
-                            row.eq(&element_array_row_inner)
-                        } else {
-                            row.ne(&element_array_row_inner)
-                        }
-                    })
-                })
-                .collect::<BooleanArray>()
-        }
-        DataType::LargeList(_) => {
-            // compare each element of the from array
-            let element_array_row_inner =
-                as_large_list_array(&element_array_row)?.value(0);
-            let list_array_row_inner = as_large_list_array(list_array_row)?;
-
-            list_array_row_inner
-                .iter()
-                // compare element by element the current row of list_array
-                .map(|row| {
-                    row.map(|row| {
-                        if eq {
-                            row.eq(&element_array_row_inner)
-                        } else {
-                            row.ne(&element_array_row_inner)
-                        }
-                    })
-                })
-                .collect::<BooleanArray>()
-        }
-        _ => {
-            let element_arr = Scalar::new(element_array_row);
-            // use not_distinct so we can compare NULL
-            if eq {
-                arrow_ord::cmp::not_distinct(&list_array_row, &element_arr)?
-            } else {
-                arrow_ord::cmp::distinct(&list_array_row, &element_arr)?
-            }
-        }
-    };
-
-    Ok(res)
-}
-
-/// Convert one or more [`ArrayRef`] of the same type into a
-/// `ListArray` or 'LargeListArray' depending on the offset size.
-///
-/// # Example (non nested)
-///
-/// Calling `array(col1, col2)` where col1 and col2 are non nested
-/// would return a single new `ListArray`, where each row was a list
-/// of 2 elements:
-///
-/// ```text
-/// ┌─────────┐   ┌─────────┐           ┌──────────────┐
-/// │ ┌─────┐ │   │ ┌─────┐ │           │ ┌──────────┐ │
-/// │ │  A  │ │   │ │  X  │ │           │ │  [A, X]  │ │
-/// │ ├─────┤ │   │ ├─────┤ │           │ ├──────────┤ │
-/// │ │NULL │ │   │ │  Y  │ │──────────▶│ │[NULL, Y] │ │
-/// │ ├─────┤ │   │ ├─────┤ │           │ ├──────────┤ │
-/// │ │  C  │ │   │ │  Z  │ │           │ │  [C, Z]  │ │
-/// │ └─────┘ │   │ └─────┘ │           │ └──────────┘ │
-/// └─────────┘   └─────────┘           └──────────────┘
-///   col1           col2                    output
-/// ```
-///
-/// # Example (nested)
-///
-/// Calling `array(col1, col2)` where col1 and col2 are lists
-/// would return a single new `ListArray`, where each row was a list
-/// of the corresponding elements of col1 and col2.
-///
-/// ``` text
-/// ┌──────────────┐   ┌──────────────┐        ┌─────────────────────────────┐
-/// │ ┌──────────┐ │   │ ┌──────────┐ │        │ ┌────────────────────────┐  │
-/// │ │  [A, X]  │ │   │ │    []    │ │        │ │    [[A, X], []]        │  │
-/// │ ├──────────┤ │   │ ├──────────┤ │        │ ├────────────────────────┤  │
-/// │ │[NULL, Y] │ │   │ │[Q, R, S] │ │───────▶│ │ [[NULL, Y], [Q, R, S]] │  │
-/// │ ├──────────┤ │   │ ├──────────┤ │        │ ├────────────────────────│  │
-/// │ │  [C, Z]  │ │   │ │   NULL   │ │        │ │    [[C, Z], NULL]      │  │
-/// │ └──────────┘ │   │ └──────────┘ │        │ └────────────────────────┘  │
-/// └──────────────┘   └──────────────┘        └─────────────────────────────┘
-///      col1               col2                         output
-/// ```
-fn array_array<O: OffsetSizeTrait>(
-    args: &[ArrayRef],
-    data_type: DataType,
-) -> Result<ArrayRef> {
-    // do not accept 0 arguments.
-    if args.is_empty() {
-        return plan_err!("Array requires at least one argument");
-    }
-
-    let mut data = vec![];
-    let mut total_len = 0;
-    for arg in args {
-        let arg_data = if arg.as_any().is::<NullArray>() {
-            ArrayData::new_empty(&data_type)
-        } else {
-            arg.to_data()
-        };
-        total_len += arg_data.len();
-        data.push(arg_data);
-    }
-
-    let mut offsets: Vec<O> = Vec::with_capacity(total_len);
-    offsets.push(O::usize_as(0));
-
-    let capacity = Capacities::Array(total_len);
-    let data_ref = data.iter().collect::<Vec<_>>();
-    let mut mutable = MutableArrayData::with_capacities(data_ref, true, capacity);
-
-    let num_rows = args[0].len();
-    for row_idx in 0..num_rows {
-        for (arr_idx, arg) in args.iter().enumerate() {
-            if !arg.as_any().is::<NullArray>()
-                && !arg.is_null(row_idx)
-                && arg.is_valid(row_idx)
-            {
-                mutable.extend(arr_idx, row_idx, row_idx + 1);
-            } else {
-                mutable.extend_nulls(1);
-            }
-        }
-        offsets.push(O::usize_as(mutable.len()));
-    }
-    let data = mutable.freeze();
-
-    Ok(Arc::new(GenericListArray::<O>::try_new(
-        Arc::new(Field::new("item", data_type, true)),
-        OffsetBuffer::new(offsets.into()),
-        arrow_array::make_array(data),
-        None,
-    )?))
-}
-
-/// `make_array` SQL function
-pub fn make_array(arrays: &[ArrayRef]) -> Result<ArrayRef> {
-    let mut data_type = DataType::Null;
-    for arg in arrays {
-        let arg_data_type = arg.data_type();
-        if !arg_data_type.equals_datatype(&DataType::Null) {
-            data_type = arg_data_type.clone();
-            break;
-        }
-    }
-
-    match data_type {
-        // Either an empty array or all nulls:
-        DataType::Null => {
-            let array =
-                new_null_array(&DataType::Null, arrays.iter().map(|a| a.len()).sum());
-            Ok(Arc::new(array_into_list_array(array)))
-        }
-        DataType::LargeList(..) => array_array::<i64>(arrays, data_type),
-        _ => array_array::<i32>(arrays, data_type),
-    }
-}
-
-/// For each element of `list_array[i]`, replaces up to `arr_n[i]`  occurences
-/// of `from_array[i]`, `to_array[i]`.
-///
-/// The type of each **element** in `list_array` must be the same as the type of
-/// `from_array` and `to_array`. This function also handles nested arrays
-/// ([`ListArray`] of [`ListArray`]s)
-///
-/// For example, when called to replace a list array (where each element is a
-/// list of int32s, the second and third argument are int32 arrays, and the
-/// fourth argument is the number of occurrences to replace
-///
-/// ```text
-/// general_replace(
-///   [1, 2, 3, 2], 2, 10, 1    ==> [1, 10, 3, 2]   (only the first 2 is replaced)
-///   [4, 5, 6, 5], 5, 20, 2    ==> [4, 20, 6, 20]  (both 5s are replaced)
-/// )
-/// ```
-fn general_replace<O: OffsetSizeTrait>(
-    list_array: &GenericListArray<O>,
-    from_array: &ArrayRef,
-    to_array: &ArrayRef,
-    arr_n: Vec<i64>,
-) -> Result<ArrayRef> {
-    // Build up the offsets for the final output array
-    let mut offsets: Vec<O> = vec![O::usize_as(0)];
-    let values = list_array.values();
-    let original_data = values.to_data();
-    let to_data = to_array.to_data();
-    let capacity = Capacities::Array(original_data.len());
-
-    // First array is the original array, second array is the element to replace with.
-    let mut mutable = MutableArrayData::with_capacities(
-        vec![&original_data, &to_data],
-        false,
-        capacity,
-    );
-
-    let mut valid = BooleanBufferBuilder::new(list_array.len());
-
-    for (row_index, offset_window) in list_array.offsets().windows(2).enumerate() {
-        if list_array.is_null(row_index) {
-            offsets.push(offsets[row_index]);
-            valid.append(false);
-            continue;
-        }
-
-        let start = offset_window[0];
-        let end = offset_window[1];
-
-        let list_array_row = list_array.value(row_index);
-
-        // Compute all positions in list_row_array (that is itself an
-        // array) that are equal to `from_array_row`
-        let eq_array =
-            compare_element_to_list(&list_array_row, &from_array, row_index, true)?;
-
-        let original_idx = O::usize_as(0);
-        let replace_idx = O::usize_as(1);
-        let n = arr_n[row_index];
-        let mut counter = 0;
-
-        // All elements are false, no need to replace, just copy original data
-        if eq_array.false_count() == eq_array.len() {
-            mutable.extend(
-                original_idx.to_usize().unwrap(),
-                start.to_usize().unwrap(),
-                end.to_usize().unwrap(),
-            );
-            offsets.push(offsets[row_index] + (end - start));
-            valid.append(true);
-            continue;
-        }
-
-        for (i, to_replace) in eq_array.iter().enumerate() {
-            let i = O::usize_as(i);
-            if let Some(true) = to_replace {
-                mutable.extend(replace_idx.to_usize().unwrap(), row_index, row_index + 1);
-                counter += 1;
-                if counter == n {
-                    // copy original data for any matches past n
-                    mutable.extend(
-                        original_idx.to_usize().unwrap(),
-                        (start + i).to_usize().unwrap() + 1,
-                        end.to_usize().unwrap(),
-                    );
-                    break;
-                }
-            } else {
-                // copy original data for false / null matches
-                mutable.extend(
-                    original_idx.to_usize().unwrap(),
-                    (start + i).to_usize().unwrap(),
-                    (start + i).to_usize().unwrap() + 1,
-                );
-            }
-        }
-
-        offsets.push(offsets[row_index] + (end - start));
-        valid.append(true);
-    }
-
-    let data = mutable.freeze();
-
-    Ok(Arc::new(GenericListArray::<O>::try_new(
-        Arc::new(Field::new("item", list_array.value_type(), true)),
-        OffsetBuffer::<O>::new(offsets.into()),
-        arrow_array::make_array(data),
-        Some(NullBuffer::new(valid.finish())),
-    )?))
-}
-
-pub fn array_replace(args: &[ArrayRef]) -> Result<ArrayRef> {
-    if args.len() != 3 {
-        return exec_err!("array_replace expects three arguments");
-    }
-
-    // replace at most one occurence for each element
-    let arr_n = vec![1; args[0].len()];
-    let array = &args[0];
-    match array.data_type() {
-        DataType::List(_) => {
-            let list_array = array.as_list::<i32>();
-            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
-        }
-        DataType::LargeList(_) => {
-            let list_array = array.as_list::<i64>();
-            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
-        }
-        array_type => exec_err!("array_replace does not support type '{array_type:?}'."),
-    }
-}
-
-pub fn array_replace_n(args: &[ArrayRef]) -> Result<ArrayRef> {
-    if args.len() != 4 {
-        return exec_err!("array_replace_n expects four arguments");
-    }
-
-    // replace the specified number of occurences
-    let arr_n = as_int64_array(&args[3])?.values().to_vec();
-    let array = &args[0];
-    match array.data_type() {
-        DataType::List(_) => {
-            let list_array = array.as_list::<i32>();
-            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
-        }
-        DataType::LargeList(_) => {
-            let list_array = array.as_list::<i64>();
-            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
-        }
-        array_type => {
-            exec_err!("array_replace_n does not support type '{array_type:?}'.")
-        }
-    }
-}
-
-pub fn array_replace_all(args: &[ArrayRef]) -> Result<ArrayRef> {
-    if args.len() != 3 {
-        return exec_err!("array_replace_all expects three arguments");
-    }
-
-    // replace all occurrences (up to "i64::MAX")
-    let arr_n = vec![i64::MAX; args[0].len()];
-    let array = &args[0];
-    match array.data_type() {
-        DataType::List(_) => {
-            let list_array = array.as_list::<i32>();
-            general_replace::<i32>(list_array, &args[1], &args[2], arr_n)
-        }
-        DataType::LargeList(_) => {
-            let list_array = array.as_list::<i64>();
-            general_replace::<i64>(list_array, &args[1], &args[2], arr_n)
-        }
-        array_type => {
-            exec_err!("array_replace_all does not support type '{array_type:?}'.")
-        }
-    }
-}
diff --git a/datafusion/physical-expr/src/functions.rs b/datafusion/physical-expr/src/functions.rs
index 994c17309ec07..c6c185e002f0b 100644
--- a/datafusion/physical-expr/src/functions.rs
+++ b/datafusion/physical-expr/src/functions.rs
@@ -32,8 +32,8 @@
 
 use crate::sort_properties::SortProperties;
 use crate::{
-    array_expressions, conditional_expressions, math_expressions, string_expressions,
-    PhysicalExpr, ScalarFunctionExpr,
+    conditional_expressions, math_expressions, string_expressions, PhysicalExpr,
+    ScalarFunctionExpr,
 };
 use arrow::{
     array::ArrayRef,
@@ -253,18 +253,6 @@ pub fn create_physical_fun(
         BuiltinScalarFunction::Cot => {
             Arc::new(|args| make_scalar_function_inner(math_expressions::cot)(args))
         }
-
-        // array functions
-        BuiltinScalarFunction::ArrayReplace => Arc::new(|args| {
-            make_scalar_function_inner(array_expressions::array_replace)(args)
-        }),
-        BuiltinScalarFunction::ArrayReplaceN => Arc::new(|args| {
-            make_scalar_function_inner(array_expressions::array_replace_n)(args)
-        }),
-        BuiltinScalarFunction::ArrayReplaceAll => Arc::new(|args| {
-            make_scalar_function_inner(array_expressions::array_replace_all)(args)
-        }),
-
         // string functions
         BuiltinScalarFunction::Ascii => Arc::new(|args| match args[0].data_type() {
             DataType::Utf8 => {
diff --git a/datafusion/physical-expr/src/lib.rs b/datafusion/physical-expr/src/lib.rs
index e8b80ee4e1e67..1791a6ed60b24 100644
--- a/datafusion/physical-expr/src/lib.rs
+++ b/datafusion/physical-expr/src/lib.rs
@@ -17,7 +17,6 @@
 
 pub mod aggregate;
 pub mod analysis;
-pub mod array_expressions;
 pub mod binary_map;
 pub mod conditional_expressions;
 pub mod equivalence;
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto
index 597094758584a..6879f70cd05c2 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -637,7 +637,7 @@ enum ScalarFunction {
   // 93 was ArrayPositions
   // 94 was ArrayPrepend
   // 95 was ArrayRemove
-  ArrayReplace = 96;
+  // 96 was ArrayReplace
   // 97 was ArrayToString
   // 98 was Cardinality
   // 99 was ArrayElement
@@ -647,9 +647,9 @@ enum ScalarFunction {
   // 105 was ArrayHasAny
   // 106 was ArrayHasAll
   // 107 was ArrayRemoveN
-  ArrayReplaceN = 108;
+  // 108 was ArrayReplaceN
   // 109 was ArrayRemoveAll
-  ArrayReplaceAll = 110;
+  // 110 was ArrayReplaceAll
   Nanvl = 111;
   // 112 was Flatten
   // 113 was IsNan
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs
index cb9633338e8f4..75c135fd01b45 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -22951,10 +22951,7 @@ impl serde::Serialize for ScalarFunction {
             Self::Factorial => "Factorial",
             Self::Lcm => "Lcm",
             Self::Gcd => "Gcd",
-            Self::ArrayReplace => "ArrayReplace",
             Self::Cot => "Cot",
-            Self::ArrayReplaceN => "ArrayReplaceN",
-            Self::ArrayReplaceAll => "ArrayReplaceAll",
             Self::Nanvl => "Nanvl",
             Self::Iszero => "Iszero",
             Self::OverLay => "OverLay",
@@ -23032,10 +23029,7 @@ impl<'de> serde::Deserialize<'de> for ScalarFunction {
             "Factorial",
             "Lcm",
             "Gcd",
-            "ArrayReplace",
             "Cot",
-            "ArrayReplaceN",
-            "ArrayReplaceAll",
             "Nanvl",
             "Iszero",
             "OverLay",
@@ -23142,10 +23136,7 @@ impl<'de> serde::Deserialize<'de> for ScalarFunction {
                     "Factorial" => Ok(ScalarFunction::Factorial),
                     "Lcm" => Ok(ScalarFunction::Lcm),
                     "Gcd" => Ok(ScalarFunction::Gcd),
-                    "ArrayReplace" => Ok(ScalarFunction::ArrayReplace),
                     "Cot" => Ok(ScalarFunction::Cot),
-                    "ArrayReplaceN" => Ok(ScalarFunction::ArrayReplaceN),
-                    "ArrayReplaceAll" => Ok(ScalarFunction::ArrayReplaceAll),
                     "Nanvl" => Ok(ScalarFunction::Nanvl),
                     "Iszero" => Ok(ScalarFunction::Iszero),
                     "OverLay" => Ok(ScalarFunction::OverLay),
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs
index f5ef6c1f74f05..c9cc4a9b073b8 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -2910,7 +2910,7 @@ pub enum ScalarFunction {
     /// 93 was ArrayPositions
     /// 94 was ArrayPrepend
     /// 95 was ArrayRemove
-    ArrayReplace = 96,
+    /// 96 was ArrayReplace
     /// 97 was ArrayToString
     /// 98 was Cardinality
     /// 99 was ArrayElement
@@ -2920,9 +2920,9 @@ pub enum ScalarFunction {
     /// 105 was ArrayHasAny
     /// 106 was ArrayHasAll
     /// 107 was ArrayRemoveN
-    ArrayReplaceN = 108,
+    /// 108 was ArrayReplaceN
     /// 109 was ArrayRemoveAll
-    ArrayReplaceAll = 110,
+    /// 110 was ArrayReplaceAll
     Nanvl = 111,
     /// 112 was Flatten
     /// 113 was IsNan
@@ -3019,10 +3019,7 @@ impl ScalarFunction {
             ScalarFunction::Factorial => "Factorial",
             ScalarFunction::Lcm => "Lcm",
             ScalarFunction::Gcd => "Gcd",
-            ScalarFunction::ArrayReplace => "ArrayReplace",
             ScalarFunction::Cot => "Cot",
-            ScalarFunction::ArrayReplaceN => "ArrayReplaceN",
-            ScalarFunction::ArrayReplaceAll => "ArrayReplaceAll",
             ScalarFunction::Nanvl => "Nanvl",
             ScalarFunction::Iszero => "Iszero",
             ScalarFunction::OverLay => "OverLay",
@@ -3094,10 +3091,7 @@ impl ScalarFunction {
             "Factorial" => Some(Self::Factorial),
             "Lcm" => Some(Self::Lcm),
             "Gcd" => Some(Self::Gcd),
-            "ArrayReplace" => Some(Self::ArrayReplace),
             "Cot" => Some(Self::Cot),
-            "ArrayReplaceN" => Some(Self::ArrayReplaceN),
-            "ArrayReplaceAll" => Some(Self::ArrayReplaceAll),
             "Nanvl" => Some(Self::Nanvl),
             "Iszero" => Some(Self::Iszero),
             "OverLay" => Some(Self::OverLay),
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs
index 3822b74bc18cb..06aab16edd57c 100644
--- a/datafusion/proto/src/logical_plan/from_proto.rs
+++ b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -48,9 +48,9 @@ use datafusion_expr::expr::Unnest;
 use datafusion_expr::expr::{Alias, Placeholder};
 use datafusion_expr::window_frame::{check_window_frame, regularize_window_order_by};
 use datafusion_expr::{
-    acosh, array_replace, array_replace_all, array_replace_n, ascii, asinh, atan, atan2,
-    atanh, bit_length, btrim, cbrt, ceil, character_length, chr, coalesce, concat_expr,
-    concat_ws_expr, cos, cosh, cot, degrees, ends_with, exp,
+    acosh, ascii, asinh, atan, atan2, atanh, bit_length, btrim, cbrt, ceil,
+    character_length, chr, coalesce, concat_expr, concat_ws_expr, cos, cosh, cot,
+    degrees, ends_with, exp,
     expr::{self, InList, Sort, WindowFunction},
     factorial, find_in_set, floor, gcd, initcap, iszero, lcm, left, levenshtein, ln, log,
     log10, log2,
@@ -466,9 +466,6 @@ impl From<&protobuf::ScalarFunction> for BuiltinScalarFunction {
             ScalarFunction::Trim => Self::Trim,
             ScalarFunction::Ltrim => Self::Ltrim,
             ScalarFunction::Rtrim => Self::Rtrim,
-            ScalarFunction::ArrayReplace => Self::ArrayReplace,
-            ScalarFunction::ArrayReplaceN => Self::ArrayReplaceN,
-            ScalarFunction::ArrayReplaceAll => Self::ArrayReplaceAll,
             ScalarFunction::Log2 => Self::Log2,
             ScalarFunction::Signum => Self::Signum,
             ScalarFunction::Ascii => Self::Ascii,
@@ -1362,22 +1359,6 @@ pub fn parse_expr(
                 ScalarFunction::Acosh => {
                     Ok(acosh(parse_expr(&args[0], registry, codec)?))
                 }
-                ScalarFunction::ArrayReplace => Ok(array_replace(
-                    parse_expr(&args[0], registry, codec)?,
-                    parse_expr(&args[1], registry, codec)?,
-                    parse_expr(&args[2], registry, codec)?,
-                )),
-                ScalarFunction::ArrayReplaceN => Ok(array_replace_n(
-                    parse_expr(&args[0], registry, codec)?,
-                    parse_expr(&args[1], registry, codec)?,
-                    parse_expr(&args[2], registry, codec)?,
-                    parse_expr(&args[3], registry, codec)?,
-                )),
-                ScalarFunction::ArrayReplaceAll => Ok(array_replace_all(
-                    parse_expr(&args[0], registry, codec)?,
-                    parse_expr(&args[1], registry, codec)?,
-                    parse_expr(&args[2], registry, codec)?,
-                )),
                 ScalarFunction::Sqrt => Ok(sqrt(parse_expr(&args[0], registry, codec)?)),
                 ScalarFunction::Cbrt => Ok(cbrt(parse_expr(&args[0], registry, codec)?)),
                 ScalarFunction::Sin => Ok(sin(parse_expr(&args[0], registry, codec)?)),
diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs
index 7a17d2a2b4051..478f7c7795525 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -1453,9 +1453,6 @@ impl TryFrom<&BuiltinScalarFunction> for protobuf::ScalarFunction {
             BuiltinScalarFunction::Trim => Self::Trim,
             BuiltinScalarFunction::Ltrim => Self::Ltrim,
             BuiltinScalarFunction::Rtrim => Self::Rtrim,
-            BuiltinScalarFunction::ArrayReplace => Self::ArrayReplace,
-            BuiltinScalarFunction::ArrayReplaceN => Self::ArrayReplaceN,
-            BuiltinScalarFunction::ArrayReplaceAll => Self::ArrayReplaceAll,
             BuiltinScalarFunction::Log2 => Self::Log2,
             BuiltinScalarFunction::Signum => Self::Signum,
             BuiltinScalarFunction::Ascii => Self::Ascii,
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index 479f80fbdddfd..93de560dbee56 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -605,6 +605,14 @@ async fn roundtrip_expr_api() -> Result<()> {
             make_array(vec![lit(3), lit(3), lit(2), lit(3), lit(1)]),
             lit(3),
         ),
+        array_replace(make_array(vec![lit(1), lit(2), lit(3)]), lit(2), lit(4)),
+        array_replace_n(
+            make_array(vec![lit(1), lit(2), lit(3)]),
+            lit(2),
+            lit(4),
+            lit(1),
+        ),
+        array_replace_all(make_array(vec![lit(1), lit(2), lit(3)]), lit(2), lit(4)),
     ];
 
     // ensure expressions created with the expr api can be round tripped
