diff --git a/docs/source/contributor-guide/expression-audits/string_funcs.md b/docs/source/contributor-guide/expression-audits/string_funcs.md
index b362958a230..51154e3a095 100644
--- a/docs/source/contributor-guide/expression-audits/string_funcs.md
+++ b/docs/source/contributor-guide/expression-audits/string_funcs.md
@@ -111,7 +111,7 @@
 ## left
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `RuntimeReplaceable` with `replacement = Substring(str, Literal(1), len)`; accepts `StringType` or `BinaryType` plus `IntegerType`. Comet serde rewrites to a `Substring` proto with `start=1, len=lenValue`. `getSupportLevel` declares `Unsupported` for non-literal `len` so the dispatcher falls back uniformly.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `RuntimeReplaceable` with `replacement = Substring(str, Literal(1), len)`; accepts `StringType` or `BinaryType` plus `IntegerType`. Comet serde serialises `expr.replacement`, routing through `SparkSubstring` — so non-literal `len` is supported.
 - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened with `StringTypeWithCollation`; behaviour unchanged for `UTF8_BINARY`. Non-default collations not honoured by Comet (https://github.com/apache/datafusion-comet/issues/4496).
 
 ## len
@@ -174,7 +174,7 @@
 ## right
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `RuntimeReplaceable` with `replacement = If(IsNull(str), null, If(len <= 0, "", Substring(str, -len, len)))`; accepts `StringType` plus `IntegerType`. Comet serde rewrites positive `len` to a `Substring` proto with `start=-len, len=len`; for `len <= 0` it builds an `If(IsNull(str), null, "")` proto chain to preserve NULL propagation. `getSupportLevel` declares `Unsupported` for non-literal `len` so the dispatcher falls back uniformly.
+- Spark 3.5.8 (audited 2026-05-27): baseline. `RuntimeReplaceable` with `replacement = If(IsNull(str), null, If(len <= 0, "", Substring(str, -len, len)))`; accepts `StringType` plus `IntegerType`. Comet serde serialises `expr.replacement`, so NULL propagation for `len <= 0` and non-literal `len` are handled by the replacement tree itself.
 - Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened with collation; uses `UnaryMinus(len, failOnError = false)` to avoid integer-overflow exceptions on `len = Int.MinValue`. Semantics unchanged for `UTF8_BINARY`. Non-default collations not honoured by Comet (https://github.com/apache/datafusion-comet/issues/4496).
 
 ## rpad
@@ -217,8 +217,8 @@
 ## substring
 
 - Spark 3.4.3 (audited 2026-05-27): identical to 3.5.8.
-- Spark 3.5.8 (audited 2026-05-27): baseline. `TernaryExpression`; two-arg form defaults `len = Integer.MAX_VALUE`; supports `StringType` and `BinaryType`. Comet serializes to a dedicated `Substring` proto. `getSupportLevel` declares `Unsupported` when either `pos` or `len` is not a `Literal` so the dispatcher falls back uniformly.
-- Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened with `StringTypeWithCollation`; semantics unchanged for `UTF8_BINARY`. Native `SubstringExpr` implements Spark's negative-start clamping and is exercised against ASCII, multibyte UTF-8, emoji, decomposed and Telugu inputs. Non-default collations not honoured by Comet (https://github.com/apache/datafusion-comet/issues/4496).
+- Spark 3.5.8 (audited 2026-05-27): baseline. `TernaryExpression`; two-arg form defaults `len = Integer.MAX_VALUE`; supports `StringType` and `BinaryType`. Comet routes through the `datafusion-spark` `SparkSubstring` UDF (registered as `substring`), which accepts non-literal `pos`/`len` and Utf8/Utf8View/Binary/LargeBinary/BinaryView inputs.
+- Spark 4.0.1 (audited 2026-05-27): `inputTypes` widened with `StringTypeWithCollation`; semantics unchanged for `UTF8_BINARY`. `SparkSubstring` implements Spark's 1-indexed, negative-start, and negative-length clamping. Non-default collations not honoured by Comet (https://github.com/apache/datafusion-comet/issues/4496).
 
 ## substring_index
 
diff --git a/native/core/src/execution/expressions/strings.rs b/native/core/src/execution/expressions/strings.rs
index 4a7c44cc3e3..b8538084f93 100644
--- a/native/core/src/execution/expressions/strings.rs
+++ b/native/core/src/execution/expressions/strings.rs
@@ -17,7 +17,6 @@
 
 //! String expression builders
 
-use std::cmp::max;
 use std::sync::Arc;
 
 use arrow::datatypes::SchemaRef;
@@ -25,7 +24,7 @@ use datafusion::common::ScalarValue;
 use datafusion::physical_expr::expressions::{LikeExpr, Literal};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion_comet_proto::spark_expression::Expr;
-use datafusion_comet_spark_expr::{FromJson, RLike, SubstringExpr};
+use datafusion_comet_spark_expr::{FromJson, RLike};
 
 use crate::execution::{
     expressions::extract_expr,
@@ -34,31 +33,6 @@ use crate::execution::{
     serde::to_arrow_datatype,
 };
 
-/// Builder for Substring expressions
-pub struct SubstringBuilder;
-
-impl ExpressionBuilder for SubstringBuilder {
-    fn build(
-        &self,
-        spark_expr: &Expr,
-        input_schema: SchemaRef,
-        planner: &PhysicalPlanner,
-    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
-        let expr = extract_expr!(spark_expr, Substring);
-        let child = planner.create_expr(expr.child.as_ref().unwrap(), input_schema)?;
-        // Spark Substring's start is 1-based when start > 0
-        let start = expr.start - i32::from(expr.start > 0);
-        // substring negative len is treated as 0 in Spark
-        let len = max(expr.len, 0);
-
-        Ok(Arc::new(SubstringExpr::new(
-            child,
-            start as i64,
-            len as u64,
-        )))
-    }
-}
-
 /// Builder for Like expressions
 pub struct LikeBuilder;
 
diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index d2c1a5df023..68373735005 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -69,6 +69,7 @@ use datafusion_spark::function::string::char::CharFunc;
 use datafusion_spark::function::string::concat::SparkConcat;
 use datafusion_spark::function::string::luhn_check::SparkLuhnCheck;
 use datafusion_spark::function::string::space::SparkSpace;
+use datafusion_spark::function::string::substring::SparkSubstring;
 use datafusion_spark::function::url::try_url_decode::TryUrlDecode as SparkTryUrlDecode;
 use datafusion_spark::function::url::url_decode::UrlDecode as SparkUrlDecode;
 use datafusion_spark::function::url::url_encode::UrlEncode as SparkUrlEncode;
@@ -646,6 +647,7 @@ fn register_datafusion_spark_function(session_ctx: &SessionContext) {
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkRint::default()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkBitShift::right_unsigned()));
     session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSoundex::default()));
+    session_ctx.register_udf(ScalarUDF::new_from_impl(SparkSubstring::default()));
 }
 
 /// Prepares arrow arrays for output.
diff --git a/native/core/src/execution/planner/expression_registry.rs b/native/core/src/execution/planner/expression_registry.rs
index 1cd44f3f81a..7fe7a477dd6 100644
--- a/native/core/src/execution/planner/expression_registry.rs
+++ b/native/core/src/execution/planner/expression_registry.rs
@@ -83,7 +83,6 @@ pub enum ExpressionType {
     CaseWhen,
     In,
     If,
-    Substring,
     Like,
     Rlike,
     CheckOverflow,
@@ -285,8 +284,6 @@ impl ExpressionRegistry {
     fn register_string_expressions(&mut self) {
         use crate::execution::expressions::strings::*;
 
-        self.builders
-            .insert(ExpressionType::Substring, Box::new(SubstringBuilder));
         self.builders
             .insert(ExpressionType::Like, Box::new(LikeBuilder));
         self.builders
@@ -359,7 +356,6 @@ impl ExpressionRegistry {
             Some(ExprStruct::CaseWhen(_)) => Ok(ExpressionType::CaseWhen),
             Some(ExprStruct::In(_)) => Ok(ExpressionType::In),
             Some(ExprStruct::If(_)) => Ok(ExpressionType::If),
-            Some(ExprStruct::Substring(_)) => Ok(ExpressionType::Substring),
             Some(ExprStruct::Like(_)) => Ok(ExpressionType::Like),
             Some(ExprStruct::Rlike(_)) => Ok(ExpressionType::Rlike),
             Some(ExprStruct::CheckOverflow(_)) => Ok(ExpressionType::CheckOverflow),
diff --git a/native/proto/src/proto/expr.proto b/native/proto/src/proto/expr.proto
index e4efae963b8..ee2a65b3d73 100644
--- a/native/proto/src/proto/expr.proto
+++ b/native/proto/src/proto/expr.proto
@@ -47,7 +47,6 @@ message Expr {
     BinaryExpr and = 17;
     BinaryExpr or = 18;
     SortOrder sort_order = 19;
-    Substring substring = 20;
     Hour hour = 22;
     Minute minute = 23;
     Second second = 24;
@@ -95,6 +94,9 @@ message Expr {
     Shuffle shuffle = 72;
   }
 
+  reserved 20;
+  reserved "substring";
+
   // Optional QueryContext for error reporting (contains SQL text and position)
   optional QueryContext query_context = 90;
 
@@ -355,12 +357,6 @@ message SortOrder {
   NullOrdering null_ordering = 3;
 }
 
-message Substring {
-  Expr child = 1;
-  int32 start = 2;
-  int32 len = 3;
-}
-
 message ToJson {
   Expr child = 1;
   string timezone = 2;
diff --git a/native/spark-expr/src/kernels/mod.rs b/native/spark-expr/src/kernels/mod.rs
index 0092f016c93..55ccfe28ad3 100644
--- a/native/spark-expr/src/kernels/mod.rs
+++ b/native/spark-expr/src/kernels/mod.rs
@@ -17,5 +17,4 @@
 
 //! Kernels
 
-pub mod strings;
 pub mod temporal;
diff --git a/native/spark-expr/src/kernels/strings.rs b/native/spark-expr/src/kernels/strings.rs
deleted file mode 100644
index 25fbf644660..00000000000
--- a/native/spark-expr/src/kernels/strings.rs
+++ /dev/null
@@ -1,62 +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.
-
-//! String kernels
-
-use std::sync::Arc;
-
-use arrow::{
-    array::*,
-    compute::kernels::substring::{substring as arrow_substring, substring_by_char},
-    datatypes::{DataType, Int32Type},
-};
-use datafusion::common::DataFusionError;
-
-pub fn substring(array: &dyn Array, start: i64, length: u64) -> Result<ArrayRef, DataFusionError> {
-    match array.data_type() {
-        DataType::LargeUtf8 => substring_by_char(
-            array
-                .as_any()
-                .downcast_ref::<LargeStringArray>()
-                .expect("A large string is expected"),
-            start,
-            Some(length),
-        )
-        .map_err(|e| e.into())
-        .map(|t| make_array(t.into_data())),
-        DataType::Utf8 => substring_by_char(
-            array
-                .as_any()
-                .downcast_ref::<StringArray>()
-                .expect("A string is expected"),
-            start,
-            Some(length),
-        )
-        .map_err(|e| e.into())
-        .map(|t| make_array(t.into_data())),
-        DataType::Binary | DataType::LargeBinary => {
-            arrow_substring(array, start, Some(length)).map_err(|e| e.into())
-        }
-        DataType::Dictionary(_, _) => {
-            let dict = as_dictionary_array::<Int32Type>(array);
-            let values = substring(dict.values(), start, length)?;
-            let result = DictionaryArray::try_new(dict.keys().clone(), values)?;
-            Ok(Arc::new(result))
-        }
-        dt => panic!("Unsupported input type for function 'substring': {dt:?}"),
-    }
-}
diff --git a/native/spark-expr/src/string_funcs/mod.rs b/native/spark-expr/src/string_funcs/mod.rs
index 6e706cc89b3..ca425fa0d13 100644
--- a/native/spark-expr/src/string_funcs/mod.rs
+++ b/native/spark-expr/src/string_funcs/mod.rs
@@ -22,7 +22,6 @@ mod regexp_extract;
 mod regexp_extract_all;
 mod regexp_extract_common;
 mod split;
-mod substring;
 
 pub use base64::spark_base64;
 pub use contains::SparkContains;
@@ -30,4 +29,3 @@ pub use get_json_object::spark_get_json_object;
 pub use regexp_extract::spark_regexp_extract;
 pub use regexp_extract_all::spark_regexp_extract_all;
 pub use split::{spark_split, spark_split_sql};
-pub use substring::SubstringExpr;
diff --git a/native/spark-expr/src/string_funcs/substring.rs b/native/spark-expr/src/string_funcs/substring.rs
deleted file mode 100644
index 3e3fdd4d59e..00000000000
--- a/native/spark-expr/src/string_funcs/substring.rs
+++ /dev/null
@@ -1,752 +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.
-
-#![allow(deprecated)]
-
-use crate::kernels::strings::substring;
-use arrow::array::{
-    as_dictionary_array, as_largestring_array, as_string_array, Array, ArrayRef, GenericStringArray,
-};
-use arrow::datatypes::{DataType, Int32Type, Schema};
-use arrow::record_batch::RecordBatch;
-use datafusion::logical_expr::ColumnarValue;
-use datafusion::physical_expr::PhysicalExpr;
-use std::{
-    fmt::{Display, Formatter},
-    hash::Hash,
-    sync::Arc,
-};
-
-#[derive(Debug, Eq)]
-pub struct SubstringExpr {
-    pub child: Arc<dyn PhysicalExpr>,
-    pub start: i64,
-    pub len: u64,
-}
-
-impl Hash for SubstringExpr {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.child.hash(state);
-        self.start.hash(state);
-        self.len.hash(state);
-    }
-}
-
-impl PartialEq for SubstringExpr {
-    fn eq(&self, other: &Self) -> bool {
-        self.child.eq(&other.child) && self.start.eq(&other.start) && self.len.eq(&other.len)
-    }
-}
-
-impl SubstringExpr {
-    pub fn new(child: Arc<dyn PhysicalExpr>, start: i64, len: u64) -> Self {
-        Self { child, start, len }
-    }
-}
-
-impl Display for SubstringExpr {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "Substring [start: {}, len: {}, child: {}]",
-            self.start, self.len, self.child
-        )
-    }
-}
-
-impl PhysicalExpr for SubstringExpr {
-    fn fmt_sql(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        Display::fmt(self, f)
-    }
-
-    fn data_type(&self, input_schema: &Schema) -> datafusion::common::Result<DataType> {
-        self.child.data_type(input_schema)
-    }
-
-    fn nullable(&self, _: &Schema) -> datafusion::common::Result<bool> {
-        Ok(true)
-    }
-
-    fn evaluate(&self, batch: &RecordBatch) -> datafusion::common::Result<ColumnarValue> {
-        let arg = self.child.evaluate(batch)?;
-        let is_scalar = matches!(arg, ColumnarValue::Scalar(_));
-        let array = arg.into_array(1)?;
-        // Spark and Arrow differ for negative start: Arrow clamps
-        // start to 0 then takes `len` chars, but Spark computes
-        // end = unclamped_start + len, then clamps both independently.
-        let result = if self.start < 0 {
-            spark_substring_negative_start(&array, self.start, self.len)?
-        } else {
-            substring(&array, self.start, self.len)?
-        };
-        if is_scalar {
-            let scalar = datafusion::common::ScalarValue::try_from_array(&result, 0)?;
-            Ok(ColumnarValue::Scalar(scalar))
-        } else {
-            Ok(ColumnarValue::Array(result))
-        }
-    }
-
-    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
-        vec![&self.child]
-    }
-
-    fn with_new_children(
-        self: Arc<Self>,
-        children: Vec<Arc<dyn PhysicalExpr>>,
-    ) -> datafusion::common::Result<Arc<dyn PhysicalExpr>> {
-        Ok(Arc::new(SubstringExpr::new(
-            Arc::clone(&children[0]),
-            self.start,
-            self.len,
-        )))
-    }
-}
-
-/// Implement Spark's substring semantics for negative start positions.
-/// Spark: start = numChars + pos, end = start + len, clamp both, empty if start >= end.
-/// Arrow: start = max(0, numChars + pos), take len chars — differs when start is clamped.
-fn spark_substring_negative_start(
-    array: &ArrayRef,
-    start: i64,
-    len: u64,
-) -> datafusion::common::Result<ArrayRef> {
-    use arrow::array::{DictionaryArray, GenericBinaryArray, OffsetSizeTrait};
-
-    fn substr_str<O: OffsetSizeTrait>(
-        str_array: &GenericStringArray<O>,
-        start: i64,
-        len: u64,
-    ) -> ArrayRef {
-        use arrow::array::GenericStringBuilder;
-        let mut builder = GenericStringBuilder::<O>::with_capacity(str_array.len(), 0);
-        for i in 0..str_array.len() {
-            // Always append; nulls are reattached in bulk below. This avoids
-            // per-row NullBufferBuilder maintenance.
-            let s = if str_array.is_null(i) {
-                ""
-            } else {
-                spark_substr_negative(str_array.value(i), start, len)
-            };
-            builder.append_value(s);
-        }
-        let (offsets, values, _) = builder.finish().into_parts();
-        Arc::new(GenericStringArray::<O>::new(
-            offsets,
-            values,
-            str_array.nulls().cloned(),
-        ))
-    }
-
-    fn substr_bin<O: OffsetSizeTrait>(
-        bin_array: &GenericBinaryArray<O>,
-        start: i64,
-        len: u64,
-    ) -> ArrayRef {
-        use arrow::array::GenericBinaryBuilder;
-        let mut builder = GenericBinaryBuilder::<O>::with_capacity(bin_array.len(), 0);
-        for i in 0..bin_array.len() {
-            let b: &[u8] = if bin_array.is_null(i) {
-                &[]
-            } else {
-                spark_binary_substr_negative(bin_array.value(i), start, len)
-            };
-            builder.append_value(b);
-        }
-        let (offsets, values, _) = builder.finish().into_parts();
-        Arc::new(GenericBinaryArray::<O>::new(
-            offsets,
-            values,
-            bin_array.nulls().cloned(),
-        ))
-    }
-
-    match array.data_type() {
-        DataType::Utf8 => Ok(substr_str::<i32>(as_string_array(array), start, len)),
-        DataType::LargeUtf8 => Ok(substr_str::<i64>(as_largestring_array(array), start, len)),
-        DataType::Binary => Ok(substr_bin::<i32>(
-            array.as_any().downcast_ref().unwrap(),
-            start,
-            len,
-        )),
-        DataType::LargeBinary => Ok(substr_bin::<i64>(
-            array.as_any().downcast_ref().unwrap(),
-            start,
-            len,
-        )),
-        DataType::Dictionary(_, _) => {
-            let dict = as_dictionary_array::<Int32Type>(array);
-            let values = spark_substring_negative_start(dict.values(), start, len)?;
-            let result = DictionaryArray::try_new(dict.keys().clone(), values)?;
-            Ok(Arc::new(result) as ArrayRef)
-        }
-        dt => Err(datafusion::common::DataFusionError::Internal(format!(
-            "Unsupported input type for substring with negative start: {dt:?}"
-        ))),
-    }
-}
-
-fn spark_substr_negative(s: &str, pos: i64, len: u64) -> &str {
-    let num_chars = s.chars().count() as i64;
-    let end = (num_chars + pos).saturating_add(len as i64).min(num_chars);
-    let start = (num_chars + pos).max(0);
-    if start >= end {
-        return "";
-    }
-
-    let mut it = s.char_indices();
-    let byte_start = it
-        .by_ref()
-        .nth(start as usize)
-        .map(|(b, _)| b)
-        .unwrap_or(s.len());
-    let span = (end - start - 1) as usize;
-    let byte_end = it.nth(span).map(|(b, _)| b).unwrap_or(s.len());
-
-    &s[byte_start..byte_end]
-}
-
-fn spark_binary_substr_negative(bytes: &[u8], pos: i64, len: u64) -> &[u8] {
-    let num_bytes = bytes.len() as i64;
-    let start = num_bytes + pos;
-    let end = start.saturating_add(len as i64).min(num_bytes);
-    let start = start.max(0);
-
-    if start >= end {
-        return &[];
-    }
-
-    &bytes[start as usize..end as usize]
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use arrow::array::{LargeStringArray, StringArray};
-    use arrow::datatypes::Field;
-    use datafusion::physical_expr::expressions::Column;
-
-    fn make_batch(values: Vec<Option<&str>>) -> RecordBatch {
-        let array = Arc::new(StringArray::from(values)) as ArrayRef;
-        let schema = Schema::new(vec![Field::new("s", DataType::Utf8, true)]);
-        RecordBatch::try_new(Arc::new(schema), vec![array]).unwrap()
-    }
-
-    fn evaluate_substring(values: Vec<Option<&str>>, start: i64, len: u64) -> Vec<Option<String>> {
-        let batch = make_batch(values);
-        let child = Arc::new(Column::new("s", 0)) as Arc<dyn PhysicalExpr>;
-        let expr = SubstringExpr::new(child, start, len);
-        let result = expr.evaluate(&batch).unwrap();
-        match result {
-            ColumnarValue::Array(arr) => {
-                let str_arr = as_string_array(&arr);
-                (0..str_arr.len())
-                    .map(|i| {
-                        if str_arr.is_null(i) {
-                            None
-                        } else {
-                            Some(str_arr.value(i).to_string())
-                        }
-                    })
-                    .collect()
-            }
-            _ => panic!("Expected Array result"),
-        }
-    }
-
-    // --- Unit tests for spark_substr_negative ---
-
-    #[test]
-    fn test_negative_basic() {
-        assert_eq!(spark_substr_negative("hello", -3, 3), "llo");
-    }
-
-    #[test]
-    fn test_negative_len_clips_at_end() {
-        assert_eq!(spark_substr_negative("hello", -3, 100), "llo");
-    }
-
-    #[test]
-    fn test_negative_len_shorter_than_available() {
-        assert_eq!(spark_substr_negative("hello", -3, 1), "l");
-    }
-
-    #[test]
-    fn test_negative_start_beyond_string() {
-        assert_eq!(spark_substr_negative("hello", -10, 3), "");
-    }
-
-    #[test]
-    fn test_negative_start_beyond_but_len_reaches_into_string() {
-        // pos=-7 on "hello"(5 chars): start = 5 + (-7) = -2, end = min(-2+8, 5) = 5,
-        // clamped start = 0, take 5 chars
-        assert_eq!(spark_substr_negative("hello", -7, 8), "hello");
-    }
-
-    #[test]
-    fn test_negative_start_equals_length() {
-        assert_eq!(spark_substr_negative("hello", -5, 5), "hello");
-    }
-
-    #[test]
-    fn test_negative_zero_len() {
-        assert_eq!(spark_substr_negative("hello", -3, 0), "");
-    }
-
-    #[test]
-    fn test_negative_empty_string() {
-        assert_eq!(spark_substr_negative("", -1, 1), "");
-    }
-
-    #[test]
-    fn test_negative_single_char() {
-        assert_eq!(spark_substr_negative("a", -1, 1), "a");
-    }
-
-    #[test]
-    fn test_negative_multibyte_utf8() {
-        assert_eq!(spark_substr_negative("こんにちは", -2, 2), "ちは");
-    }
-
-    #[test]
-    fn test_negative_emoji() {
-        assert_eq!(spark_substr_negative("🎉🎊🎈", -1, 1), "🎈");
-    }
-
-    #[test]
-    fn test_negative_mixed_ascii_multibyte() {
-        // "ab🎉cd" has 5 chars. pos=-3: start=5+(-3)=2, end=min(2+2,5)=4 → chars 2,3 = "🎉c"
-        assert_eq!(spark_substr_negative("ab🎉cd", -3, 2), "🎉c");
-    }
-
-    // --- End-to-end SubstringExpr tests (positive start) ---
-    // NOTE: SubstringExpr.start uses 0-based indexing. The serde layer
-    // converts Spark's 1-based positions before constructing SubstringExpr.
-
-    #[test]
-    fn test_basic_positive_start() {
-        // start=0 (0-based) → first character
-        let result =
-            evaluate_substring(vec![Some("hello world"), Some("abc"), Some(""), None], 0, 5);
-        assert_eq!(
-            result,
-            vec![
-                Some("hello".to_string()),
-                Some("abc".to_string()),
-                Some("".to_string()),
-                None,
-            ]
-        );
-    }
-
-    #[test]
-    fn test_positive_start_offset() {
-        // start=1 (0-based) → skip first character
-        let result = evaluate_substring(vec![Some("hello world"), Some("abc")], 1, 5);
-        assert_eq!(
-            result,
-            vec![Some("ello ".to_string()), Some("bc".to_string())]
-        );
-    }
-
-    #[test]
-    fn test_start_zero() {
-        let result = evaluate_substring(vec![Some("hello")], 0, 3);
-        assert_eq!(result, vec![Some("hel".to_string())]);
-    }
-
-    #[test]
-    fn test_start_beyond_string_length() {
-        let result = evaluate_substring(vec![Some("hello"), Some("ab")], 100, 5);
-        assert_eq!(result, vec![Some("".to_string()), Some("".to_string())]);
-    }
-
-    #[test]
-    fn test_len_zero() {
-        let result = evaluate_substring(vec![Some("hello")], 1, 0);
-        assert_eq!(result, vec![Some("".to_string())]);
-    }
-
-    #[test]
-    fn test_len_exceeds_string() {
-        // start=0 (0-based), len=100 on "hi" → "hi"
-        let result = evaluate_substring(vec![Some("hi")], 0, 100);
-        assert_eq!(result, vec![Some("hi".to_string())]);
-    }
-
-    #[test]
-    fn test_start_at_last_char() {
-        // "hello" has 5 chars, 0-based index 4 → 'o'
-        let result = evaluate_substring(vec![Some("hello")], 4, 10);
-        assert_eq!(result, vec![Some("o".to_string())]);
-    }
-
-    #[test]
-    fn test_very_large_start() {
-        let result = evaluate_substring(vec![Some("hello")], i64::from(i32::MAX), 5);
-        assert_eq!(result, vec![Some("".to_string())]);
-    }
-
-    #[test]
-    fn test_very_large_len() {
-        // start=0 (0-based) with u64::MAX length
-        let result = evaluate_substring(vec![Some("hello")], 0, u64::MAX);
-        assert_eq!(result, vec![Some("hello".to_string())]);
-    }
-
-    // --- End-to-end SubstringExpr tests (negative start) ---
-
-    #[test]
-    fn test_negative_start_end_to_end() {
-        let result = evaluate_substring(
-            vec![Some("hello world"), Some("abc"), Some(""), None],
-            -3,
-            3,
-        );
-        assert_eq!(
-            result,
-            vec![
-                Some("rld".to_string()),
-                Some("abc".to_string()),
-                Some("".to_string()),
-                None,
-            ]
-        );
-    }
-
-    #[test]
-    fn test_negative_start_with_clip() {
-        // -2 with len=1 on "hello": start=3, end=4 → "l"
-        let result = evaluate_substring(vec![Some("hello")], -2, 1);
-        assert_eq!(result, vec![Some("l".to_string())]);
-    }
-
-    #[test]
-    fn test_negative_start_beyond_string_end_to_end() {
-        let result = evaluate_substring(vec![Some("hello")], -10, 3);
-        assert_eq!(result, vec![Some("".to_string())]);
-    }
-
-    #[test]
-    fn test_negative_start_far_beyond_with_large_len() {
-        // -7 on "hello"(5): start=-2, end=min(-2+8,5)=5, clamped start=0 → "hello"
-        let result = evaluate_substring(vec![Some("hello")], -7, 8);
-        assert_eq!(result, vec![Some("hello".to_string())]);
-    }
-
-    #[test]
-    fn test_negative_start_equals_string_length() {
-        let result = evaluate_substring(vec![Some("hello")], -5, 5);
-        assert_eq!(result, vec![Some("hello".to_string())]);
-    }
-
-    // --- Multi-byte UTF-8 through SubstringExpr (0-based start) ---
-
-    #[test]
-    fn test_multibyte_positive_start() {
-        // start=0 (0-based), len=3 on "こんにちは世界" → "こんに"
-        let result = evaluate_substring(vec![Some("こんにちは世界")], 0, 3);
-        assert_eq!(result, vec![Some("こんに".to_string())]);
-    }
-
-    #[test]
-    fn test_multibyte_middle() {
-        // start=3 (0-based) on "こんにちは世界" → 'ち','は' → "ちは"
-        let result = evaluate_substring(vec![Some("こんにちは世界")], 3, 2);
-        assert_eq!(result, vec![Some("ちは".to_string())]);
-    }
-
-    #[test]
-    fn test_multibyte_negative_start() {
-        let result = evaluate_substring(vec![Some("こんにちは世界")], -2, 2);
-        assert_eq!(result, vec![Some("世界".to_string())]);
-    }
-
-    #[test]
-    fn test_emoji_substring() {
-        // start=1 (0-based) on "🎉🎊🎈🎁" → '🎊','🎈' → "🎊🎈"
-        let result = evaluate_substring(vec![Some("🎉🎊🎈🎁")], 1, 2);
-        assert_eq!(result, vec![Some("🎊🎈".to_string())]);
-    }
-
-    #[test]
-    fn test_mixed_ascii_emoji() {
-        // start=2 (0-based) on "ab🎉cd" → '🎉'
-        let result = evaluate_substring(vec![Some("ab🎉cd")], 2, 1);
-        assert_eq!(result, vec![Some("🎉".to_string())]);
-    }
-
-    // --- LargeUtf8 support ---
-
-    #[test]
-    fn test_large_utf8_negative_start() {
-        let array = Arc::new(LargeStringArray::from(vec![
-            Some("hello world"),
-            None,
-            Some("abc"),
-        ])) as ArrayRef;
-        let result = spark_substring_negative_start(&array, -3, 3).unwrap();
-        let str_arr = as_largestring_array(&result);
-        assert_eq!(str_arr.value(0), "rld");
-        assert!(str_arr.is_null(1));
-        assert_eq!(str_arr.value(2), "abc");
-    }
-
-    // --- Binary negative start ---
-
-    #[test]
-    fn test_binary_negative_basic() {
-        assert_eq!(
-            spark_binary_substr_negative(&[1, 2, 3, 4, 5], -2, 2),
-            &[4, 5]
-        );
-    }
-
-    #[test]
-    fn test_binary_negative_clips_at_end() {
-        assert_eq!(
-            spark_binary_substr_negative(&[1, 2, 3, 4, 5], -2, 100),
-            &[4, 5]
-        );
-    }
-
-    #[test]
-    fn test_binary_negative_beyond_length() {
-        let empty: &[u8] = &[];
-        assert_eq!(spark_binary_substr_negative(&[1, 2, 3], -10, 3), empty);
-    }
-
-    #[test]
-    fn test_binary_negative_start_array() {
-        use arrow::array::BinaryArray;
-        let array = Arc::new(BinaryArray::from(vec![
-            Some(vec![1, 2, 3, 4, 5].as_slice()),
-            Some(&[0xFF]),
-            Some(&[]),
-            None,
-        ])) as ArrayRef;
-        let result = spark_substring_negative_start(&array, -2, 2).unwrap();
-        let bin_arr = result.as_any().downcast_ref::<BinaryArray>().unwrap();
-        assert_eq!(bin_arr.value(0), &[4, 5]);
-        assert_eq!(bin_arr.value(1), &[0xFF]);
-        assert_eq!(bin_arr.value(2), &[] as &[u8]);
-        assert!(bin_arr.is_null(3));
-    }
-
-    // --- Unicode edge cases: decomposed vs precomposed and combining characters ---
-    // Spark substring operates on code points, not graphemes.
-    // "é" as e + \u{301} (combining acute) = 2 code points
-    // "é" as \u{e9} (precomposed) = 1 code point
-    // "తెలుగు" (Telugu) = 6 code points: త, ె, ల, ు, గ, ు
-
-    #[test]
-    fn test_negative_decomposed_e_acute() {
-        // "e\u{301}" has 2 code points; pos=-1 → just the combining accent
-        assert_eq!(spark_substr_negative("e\u{301}", -1, 1), "\u{301}");
-    }
-
-    #[test]
-    fn test_negative_precomposed_e_acute() {
-        // "\u{e9}" has 1 code point; pos=-1 → the whole character
-        assert_eq!(spark_substr_negative("\u{e9}", -1, 1), "\u{e9}");
-    }
-
-    #[test]
-    fn test_negative_telugu() {
-        // "తెలుగు" has 6 code points; pos=-2 → last 2 code points "గు"
-        assert_eq!(spark_substr_negative("తెలుగు", -2, 2), "గు");
-    }
-
-    #[test]
-    fn test_decomposed_e_acute_split() {
-        // "e\u{301}" = 2 code points; start=0, len=1 → just "e" (strips combining accent)
-        let result = evaluate_substring(vec![Some("e\u{301}")], 0, 1);
-        assert_eq!(result, vec![Some("e".to_string())]);
-    }
-
-    #[test]
-    fn test_decomposed_e_acute_accent_only() {
-        // start=1, len=1 → just the combining acute accent
-        let result = evaluate_substring(vec![Some("e\u{301}")], 1, 1);
-        assert_eq!(result, vec![Some("\u{301}".to_string())]);
-    }
-
-    #[test]
-    fn test_decomposed_e_acute_full() {
-        // start=0, len=2 → both code points "é" (decomposed)
-        let result = evaluate_substring(vec![Some("e\u{301}")], 0, 2);
-        assert_eq!(result, vec![Some("e\u{301}".to_string())]);
-    }
-
-    #[test]
-    fn test_precomposed_e_acute() {
-        // "\u{e9}" = 1 code point; start=0, len=1 → "é"
-        let result = evaluate_substring(vec![Some("\u{e9}")], 0, 1);
-        assert_eq!(result, vec![Some("\u{e9}".to_string())]);
-    }
-
-    #[test]
-    fn test_decomposed_vs_precomposed_different_len() {
-        // Same visual character but different code point counts
-        let decomposed = "e\u{301}";
-        let precomposed = "\u{e9}";
-        let result = evaluate_substring(vec![Some(decomposed), Some(precomposed)], 0, 1);
-        assert_eq!(
-            result,
-            vec![
-                Some("e".to_string()),      // only base 'e', accent stripped
-                Some("\u{e9}".to_string()), // full precomposed character
-            ]
-        );
-    }
-
-    #[test]
-    fn test_telugu_first_two_codepoints() {
-        // "తెలుగు" start=0, len=2 → "తె" (base + vowel sign)
-        let result = evaluate_substring(vec![Some("తెలుగు")], 0, 2);
-        assert_eq!(result, vec![Some("తె".to_string())]);
-    }
-
-    #[test]
-    fn test_telugu_middle() {
-        // "తెలుగు" start=2, len=2 → "లు"
-        let result = evaluate_substring(vec![Some("తెలుగు")], 2, 2);
-        assert_eq!(result, vec![Some("లు".to_string())]);
-    }
-
-    #[test]
-    fn test_telugu_negative_start() {
-        // "తెలుగు" has 6 code points; -3 with len=3 → last 3 code points "ుగు"
-        let result = evaluate_substring(vec![Some("తెలుగు")], -3, 3);
-        assert_eq!(result, vec![Some("ుగు".to_string())]);
-    }
-
-    #[test]
-    fn test_telugu_full() {
-        let result = evaluate_substring(vec![Some("తెలుగు")], 0, 100);
-        assert_eq!(result, vec![Some("తెలుగు".to_string())]);
-    }
-
-    // --- All-null input ---
-
-    #[test]
-    fn test_all_nulls() {
-        let result = evaluate_substring(vec![None, None, None], 1, 5);
-        assert_eq!(result, vec![None, None, None]);
-    }
-
-    // --- Empty strings ---
-
-    #[test]
-    fn test_all_empty_strings() {
-        let result = evaluate_substring(vec![Some(""), Some(""), Some("")], 1, 5);
-        assert_eq!(
-            result,
-            vec![
-                Some("".to_string()),
-                Some("".to_string()),
-                Some("".to_string()),
-            ]
-        );
-    }
-
-    // --- Scalar support ---
-
-    fn evaluate_scalar_substring(value: Option<&str>, start: i64, len: u64) -> ColumnarValue {
-        use datafusion::common::ScalarValue;
-        use datafusion::physical_expr::expressions::Literal;
-
-        let scalar = ScalarValue::Utf8(value.map(|s| s.to_string()));
-        let child = Arc::new(Literal::new(scalar)) as Arc<dyn PhysicalExpr>;
-        let expr = SubstringExpr::new(child, start, len);
-        let schema = Schema::new(vec![Field::new("dummy", DataType::Utf8, true)]);
-        let batch = RecordBatch::new_empty(Arc::new(schema));
-        expr.evaluate(&batch).unwrap()
-    }
-
-    #[test]
-    fn test_scalar_basic() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("hello world"), 0, 5) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "hello"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_negative_start() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("hello world"), -3, 3) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "rld"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_null() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(None, 0, 5) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(None)) => {}
-            other => panic!("Expected Scalar Utf8(None), got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_empty_string() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some(""), 0, 5) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, ""),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_multibyte() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("こんにちは"), 0, 3) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "こんに"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_negative_start_multibyte() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("こんにちは"), -2, 2) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "ちは"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_decomposed_e_acute() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("e\u{301}"), 0, 1) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "e"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-
-    #[test]
-    fn test_scalar_telugu() {
-        use datafusion::common::ScalarValue;
-        match evaluate_scalar_substring(Some("తెలుగు"), -2, 2) {
-            ColumnarValue::Scalar(ScalarValue::Utf8(Some(s))) => assert_eq!(s, "గు"),
-            other => panic!("Expected Scalar Utf8, got {:?}", other),
-        }
-    }
-}
diff --git a/spark/src/main/scala/org/apache/comet/serde/strings.scala b/spark/src/main/scala/org/apache/comet/serde/strings.scala
index 4918a7fd748..07b3c0c99fa 100644
--- a/spark/src/main/scala/org/apache/comet/serde/strings.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/strings.scala
@@ -19,12 +19,10 @@
 
 package org.apache.comet.serde
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, Cast, Concat, ConcatWs, Elt, Empty2Null, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, If, InitCap, IsNull, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Base64, BitLength, Cast, Concat, ConcatWs, Elt, Empty2Null, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, InitCap, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
-import org.apache.spark.unsafe.types.UTF8String
 
 import org.apache.comet.CometConf
-import org.apache.comet.CometSparkSessionExtensions.withFallbackReason
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.QueryPlanSerde.{createBinaryExpr, exprToProtoInternal, optExprWithFallbackReason, scalarFunctionExprToProto, scalarFunctionExprToProtoWithReturnType}
 import org.apache.comet.shims.CometTypeShim
@@ -185,36 +183,7 @@ object CometStringReplace
   }
 }
 
-object CometSubstring extends CometExpressionSerde[Substring] {
-
-  override def getSupportLevel(expr: Substring): SupportLevel = (expr.pos, expr.len) match {
-    case (_: Literal, _: Literal) => Compatible()
-    case _ => Unsupported(Some("Substring pos and len must be literals"))
-  }
-
-  override def convert(
-      expr: Substring,
-      inputs: Seq[Attribute],
-      binding: Boolean): Option[Expr] = {
-    (expr.pos, expr.len) match {
-      case (Literal(pos, _), Literal(len, _)) =>
-        exprToProtoInternal(expr.str, inputs, binding) match {
-          case Some(strExpr) =>
-            val builder = ExprOuterClass.Substring.newBuilder()
-            builder.setChild(strExpr)
-            builder.setStart(pos.asInstanceOf[Int])
-            builder.setLen(len.asInstanceOf[Int])
-            Some(ExprOuterClass.Expr.newBuilder().setSubstring(builder).build())
-          case None =>
-            withFallbackReason(expr, expr.str)
-            None
-        }
-      case _ =>
-        // Unreachable: getSupportLevel gates non-literal pos/len.
-        None
-    }
-  }
-}
+object CometSubstring extends CometScalarFunction[Substring]("substring")
 
 object CometSubstringIndex extends CometExpressionSerde[SubstringIndex] {
 
@@ -234,88 +203,30 @@ object CometSubstringIndex extends CometExpressionSerde[SubstringIndex] {
 
 object CometLeft extends CometExpressionSerde[Left] {
 
-  override def getUnsupportedReasons(): Seq[String] = Seq(
-    "Only supports `BinaryType` and `StringType` input",
-    "The length argument must be a literal value")
-
-  override def convert(expr: Left, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
-    expr.len match {
-      case Literal(lenValue, _) =>
-        exprToProtoInternal(expr.str, inputs, binding) match {
-          case Some(strExpr) =>
-            val builder = ExprOuterClass.Substring.newBuilder()
-            builder.setChild(strExpr)
-            builder.setStart(1)
-            builder.setLen(lenValue.asInstanceOf[Int])
-            Some(ExprOuterClass.Expr.newBuilder().setSubstring(builder).build())
-          case None =>
-            withFallbackReason(expr, expr.str)
-            None
-        }
-      case _ =>
-        // Unreachable: getSupportLevel gates a non-literal length.
-        None
-    }
-  }
+  override def convert(expr: Left, inputs: Seq[Attribute], binding: Boolean): Option[Expr] =
+    // Left is RuntimeReplaceable; its `replacement` is `Substring(str, Literal(1), len)`,
+    // which routes through CometSubstring -> DataFusion's SparkSubstring UDF and handles
+    // non-literal `len`, len <= 0, len > length(str), NULL propagation, and BinaryType input.
+    exprToProtoInternal(expr.replacement, inputs, binding)
 
-  override def getSupportLevel(expr: Left): SupportLevel = {
-    expr.str.dataType match {
-      case _: BinaryType | _: StringType =>
-        expr.len match {
-          case _: Literal => Compatible()
-          case _ => Unsupported(Some("LEFT len must be a literal"))
-        }
-      case _ => Unsupported(Some(s"LEFT does not support ${expr.str.dataType}"))
-    }
+  override def getSupportLevel(expr: Left): SupportLevel = expr.str.dataType match {
+    case _: BinaryType | _: StringType => Compatible()
+    case dt => Unsupported(Some(s"LEFT does not support $dt"))
   }
 }
 
 object CometRight extends CometExpressionSerde[Right] {
 
-  override def convert(expr: Right, inputs: Seq[Attribute], binding: Boolean): Option[Expr] = {
-    expr.len match {
-      case Literal(lenValue, _) =>
-        val lenInt = lenValue.asInstanceOf[Int]
-        if (lenInt <= 0) {
-          // Match Spark's behavior: If(IsNull(str), NULL, "")
-          // This ensures NULL propagation: RIGHT(NULL, 0) -> NULL, RIGHT("hello", 0) -> ""
-          val isNullExpr = IsNull(expr.str)
-          val nullLiteral = Literal.create(null, StringType)
-          val emptyStringLiteral = Literal(UTF8String.EMPTY_UTF8, StringType)
-          val ifExpr = If(isNullExpr, nullLiteral, emptyStringLiteral)
-
-          // Serialize the If expression using existing infrastructure
-          exprToProtoInternal(ifExpr, inputs, binding)
-        } else {
-          exprToProtoInternal(expr.str, inputs, binding) match {
-            case Some(strExpr) =>
-              val builder = ExprOuterClass.Substring.newBuilder()
-              builder.setChild(strExpr)
-              builder.setStart(-lenInt)
-              builder.setLen(lenInt)
-              Some(ExprOuterClass.Expr.newBuilder().setSubstring(builder).build())
-            case None =>
-              withFallbackReason(expr, expr.str)
-              None
-          }
-        }
-      case _ =>
-        // Unreachable: getSupportLevel gates a non-literal length.
-        None
-    }
-  }
+  override def convert(expr: Right, inputs: Seq[Attribute], binding: Boolean): Option[Expr] =
+    // Right is RuntimeReplaceable; its `replacement` is
+    //   If(IsNull(str), NULL, If(len <= 0, "", Substring(str, -len, len)))
+    // Serializing that tree preserves Spark's NULL-propagation for len <= 0 (RIGHT(NULL, 0)
+    // must return NULL, not "") and routes the substring path through SparkSubstring.
+    exprToProtoInternal(expr.replacement, inputs, binding)
 
-  override def getUnsupportedReasons(): Seq[String] = Seq("Only supports `StringType` input")
-
-  override def getSupportLevel(expr: Right): SupportLevel = {
-    expr.str.dataType match {
-      case _: StringType =>
-        expr.len match {
-          case _: Literal => Compatible()
-          case _ => Unsupported(Some("RIGHT len must be a literal"))
-        }
-      case _ => Unsupported(Some(s"RIGHT does not support ${expr.str.dataType}"))
-    }
+  override def getSupportLevel(expr: Right): SupportLevel = expr.str.dataType match {
+    case _: StringType => Compatible()
+    case dt => Unsupported(Some(s"RIGHT does not support $dt"))
   }
 }
 
diff --git a/spark/src/test/resources/sql-tests/expressions/string/left.sql b/spark/src/test/resources/sql-tests/expressions/string/left.sql
index 7c05ecac35c..c46e67e1a77 100644
--- a/spark/src/test/resources/sql-tests/expressions/string/left.sql
+++ b/spark/src/test/resources/sql-tests/expressions/string/left.sql
@@ -15,13 +15,19 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
+-- Note: Left is a RuntimeReplaceable expression whose replacement is
+-- Substring(str, Literal(1), len). Comet serialises expr.replacement, so
+-- non-literal len goes through the SparkSubstring UDF natively.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
 statement
 CREATE TABLE test_str_left(s string, n int) USING parquet
 
 statement
 INSERT INTO test_str_left VALUES ('hello', 3), ('hello', 0), ('hello', -1), ('hello', 10), ('', 3), (NULL, 3), ('hello', NULL)
 
-query expect_fallback(Substring pos and len must be literals)
+query
 SELECT left(s, n) FROM test_str_left
 
 -- column + literal
@@ -40,13 +46,44 @@ query
 SELECT left(s, 10) FROM test_str_left
 
 -- literal + column
-query expect_fallback(Substring pos and len must be literals)
+query
 SELECT left('hello', n) FROM test_str_left
 
 -- literal + literal
 query
 SELECT left('hello', 3), left('hello', 0), left('hello', -1), left('', 3), left(NULL, 3)
 
+-- integer boundaries
+query
+SELECT left('hello', 2147483647), left('hello', -2147483648)
+
+-- null propagation across combinations
+query
+SELECT left(CAST(NULL AS STRING), 0), left(CAST(NULL AS STRING), -1), left(CAST(NULL AS STRING), 2)
+
+-- non-literal len across integer widths
+statement
+CREATE TABLE test_str_left_int_widths(s string, nt tinyint, ns smallint, ni int, nb bigint) USING parquet
+
+statement
+INSERT INTO test_str_left_int_widths VALUES ('hello', 2, 3, 4, 5), ('hello', -1, 0, 10, NULL), (NULL, 2, 2, 2, 2)
+
+query
+SELECT left(s, nt), left(s, ns), left(s, ni), left(s, nb) FROM test_str_left_int_widths
+
+-- BinaryType input (Spark's Left accepts StringType and BinaryType)
+statement
+CREATE TABLE test_str_left_bin(b binary, n int) USING parquet
+
+statement
+INSERT INTO test_str_left_bin VALUES (CAST('hello' AS BINARY), 3), (CAST('' AS BINARY), 3), (NULL, 3), (CAST('hello' AS BINARY), 0), (CAST('hello' AS BINARY), -1), (X'00010203', 2)
+
+query
+SELECT left(b, n) FROM test_str_left_bin
+
+query
+SELECT left(b, 3) FROM test_str_left_bin
+
 -- unicode
 statement
 CREATE TABLE test_str_left_unicode(s string) USING parquet
@@ -62,3 +99,7 @@ SELECT s, left(s, 4) FROM test_str_left_unicode
 
 query
 SELECT s, left(s, 0) FROM test_str_left_unicode
+
+-- equivalence with substring
+query
+SELECT s, left(s, 3), substring(s, 1, 3) FROM test_str_left_unicode
diff --git a/spark/src/test/resources/sql-tests/expressions/string/right.sql b/spark/src/test/resources/sql-tests/expressions/string/right.sql
index 0af2d562f63..d4026910fbb 100644
--- a/spark/src/test/resources/sql-tests/expressions/string/right.sql
+++ b/spark/src/test/resources/sql-tests/expressions/string/right.sql
@@ -15,45 +15,52 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Note: Right is a RuntimeReplaceable expression. Spark replaces it with
--- If(IsNull(str), null, If(len <= 0, "", Substring(str, -len, len)))
--- before Comet sees it. CometRight handles the serde, but the optimizer
--- may replace it first. We use spark_answer_only to verify correctness.
+-- Note: Right is a RuntimeReplaceable expression whose replacement is
+-- If(IsNull(str), null, If(len <= 0, "", Substring(str, -len, len))).
+-- Comet serialises expr.replacement, so NULL propagation for len <= 0
+-- and non-literal len are handled by the replacement tree.
+
+-- ConfigMatrix: parquet.enable.dictionary=false,true
+
 statement
 CREATE TABLE test_str_right(s string, n int) USING parquet
 
 statement
 INSERT INTO test_str_right VALUES ('hello', 3), ('hello', 0), ('hello', -1), ('hello', 10), ('', 3), (NULL, 3), ('hello', NULL)
 
--- both columns: len must be literal, falls back
-query spark_answer_only
+query
 SELECT right(s, n) FROM test_str_right
 
 -- column + literal: basic
-query spark_answer_only
+query
 SELECT right(s, 3) FROM test_str_right
 
 -- column + literal: edge cases
-query spark_answer_only
+query
 SELECT right(s, 0) FROM test_str_right
 
-query spark_answer_only
+query
 SELECT right(s, -1) FROM test_str_right
 
-query spark_answer_only
+query
 -- n exceeds length of 'hello' (5 chars)
 SELECT right(s, 10) FROM test_str_right
 
--- literal + column: falls back
-query spark_answer_only
+-- literal + column
+query
 SELECT right('hello', n) FROM test_str_right
 
 -- literal + literal
-query spark_answer_only
+query
 SELECT right('hello', 3), right('hello', 0), right('hello', -1), right('', 3), right(NULL, 3)
 
+-- integer boundaries: Int.MinValue exercises Spark 4.x UnaryMinus(len, failOnError=false),
+-- which overflows to Int.MinValue and only reaches Substring when the len<=0 guard is honoured.
+query
+SELECT right('hello', 2147483647), right('hello', -2147483648)
+
 -- null propagation with len <= 0 (critical: NULL str with non-positive len must return NULL, not empty string)
-query spark_answer_only
+query
 SELECT right(CAST(NULL AS STRING), 0), right(CAST(NULL AS STRING), -1), right(CAST(NULL AS STRING), 2)
 
 -- mixed null and non-null values with len <= 0
@@ -63,19 +70,29 @@ CREATE TABLE test_str_right_nulls(s string) USING parquet
 statement
 INSERT INTO test_str_right_nulls VALUES ('hello'), (NULL), (''), ('world')
 
-query spark_answer_only
+query
 SELECT s, right(s, 0) FROM test_str_right_nulls
 
-query spark_answer_only
+query
 SELECT s, right(s, -1) FROM test_str_right_nulls
 
-query spark_answer_only
+query
 SELECT s, right(s, 2) FROM test_str_right_nulls
 
 -- equivalence with substring
-query spark_answer_only
+query
 SELECT s, right(s, 3), substring(s, -3, 3) FROM test_str_right_nulls
 
+-- non-literal len across integer widths
+statement
+CREATE TABLE test_str_right_int_widths(s string, nt tinyint, ns smallint, ni int, nb bigint) USING parquet
+
+statement
+INSERT INTO test_str_right_int_widths VALUES ('hello', 2, 3, 4, 5), ('hello', -1, 0, 10, NULL), (NULL, 2, 2, 2, 2)
+
+query
+SELECT right(s, nt), right(s, ns), right(s, ni), right(s, nb) FROM test_str_right_int_widths
+
 -- unicode
 statement
 CREATE TABLE test_str_right_unicode(s string) USING parquet
@@ -83,11 +100,11 @@ CREATE TABLE test_str_right_unicode(s string) USING parquet
 statement
 INSERT INTO test_str_right_unicode VALUES ('café'), ('hello世界'), ('😀emoji'), ('తెలుగు'), (NULL)
 
-query spark_answer_only
+query
 SELECT s, right(s, 2) FROM test_str_right_unicode
 
-query spark_answer_only
+query
 SELECT s, right(s, 4) FROM test_str_right_unicode
 
-query spark_answer_only
+query
 SELECT s, right(s, 0) FROM test_str_right_unicode