feat: Support left expression (#3206)

Author: Shekharrajak

docs/source/user-guide/latest/configs.md

@@ -269,6 +269,7 @@ These settings can be used to determine which parts of the plan are accelerated
 | `spark.comet.expression.JsonToStructs.enabled` | Enable Comet acceleration for `JsonToStructs` | true |
 | `spark.comet.expression.KnownFloatingPointNormalized.enabled` | Enable Comet acceleration for `KnownFloatingPointNormalized` | true |
 | `spark.comet.expression.LastDay.enabled` | Enable Comet acceleration for `LastDay` | true |
+| `spark.comet.expression.Left.enabled` | Enable Comet acceleration for `Left` | true |
 | `spark.comet.expression.Length.enabled` | Enable Comet acceleration for `Length` | true |
 | `spark.comet.expression.LessThan.enabled` | Enable Comet acceleration for `LessThan` | true |
 | `spark.comet.expression.LessThanOrEqual.enabled` | Enable Comet acceleration for `LessThanOrEqual` | true |

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -170,6 +170,7 @@ object QueryPlanSerde extends Logging with CometExprShim {
     classOf[StringTrimBoth] -> CometScalarFunction("btrim"),
     classOf[StringTrimLeft] -> CometScalarFunction("ltrim"),
     classOf[StringTrimRight] -> CometScalarFunction("rtrim"),
+    classOf[Left] -> CometLeft,
     classOf[Substring] -> CometSubstring,
     classOf[Upper] -> CometUpper)
 

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -21,7 +21,7 @@ package org.apache.comet.serde
 
 import java.util.Locale
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, Expression, InitCap, Length, Like, Literal, Lower, RegExpReplace, RLike, StringLPad, StringRepeat, StringRPad, Substring, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Concat, Expression, InitCap, Left, Length, Like, Literal, Lower, RegExpReplace, RLike, StringLPad, StringRepeat, StringRPad, Substring, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
 
 import org.apache.comet.CometConf
@@ -113,6 +113,36 @@ object CometSubstring extends CometExpressionSerde[Substring] {
   }
 }
 
+object CometLeft extends CometExpressionSerde[Left] {
+
+  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 =>
+            withInfo(expr, expr.str)
+            None
+        }
+      case _ =>
+        withInfo(expr, "LEFT len must be a literal")
+        None
+    }
+  }
+
+  override def getSupportLevel(expr: Left): SupportLevel = {
+    expr.str.dataType match {
+      case _: BinaryType | _: StringType => Compatible()
+      case _ => Unsupported(Some(s"LEFT does not support ${expr.str.dataType}"))
+    }
+  }
+}
+
 object CometConcat extends CometScalarFunction[Concat]("concat") {
   val unsupportedReason = "CONCAT supports only string input parameters"
 

spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala

@@ -506,6 +506,52 @@ class CometExpressionSuite extends CometTestBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("LEFT function") {
+    withParquetTable((0 until 10).map(i => (s"test$i", i)), "tbl") {
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 2) FROM tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 4) FROM tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 0) FROM tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, -1) FROM tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 100) FROM tbl")
+      checkSparkAnswerAndOperator("SELECT LEFT(CAST(NULL AS STRING), 2) FROM tbl LIMIT 1")
+    }
+  }
+
+  test("LEFT function with unicode") {
+    val data = Seq("café", "hello世界", "😀emoji", "తెలుగు")
+    withParquetTable(data.zipWithIndex, "unicode_tbl") {
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 2) FROM unicode_tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 3) FROM unicode_tbl")
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 0) FROM unicode_tbl")
+    }
+  }
+
+  test("LEFT function equivalence with SUBSTRING") {
+    withParquetTable((0 until 20).map(i => Tuple1(s"test$i")), "equiv_tbl") {
+      val df = spark.sql("""
+        SELECT _1,
+          LEFT(_1, 3) as left_result,
+          SUBSTRING(_1, 1, 3) as substring_result
+        FROM equiv_tbl
+      """)
+      checkAnswer(
+        df.filter(
+          "left_result != substring_result OR " +
+            "(left_result IS NULL AND substring_result IS NOT NULL) OR " +
+            "(left_result IS NOT NULL AND substring_result IS NULL)"),
+        Seq.empty)
+    }
+  }
+
+  test("LEFT function with dictionary") {
+    val data = (0 until 1000)
+      .map(_ % 5)
+      .map(i => s"value$i")
+    withParquetTable(data.zipWithIndex, "dict_tbl") {
+      checkSparkAnswerAndOperator("SELECT _1, LEFT(_1, 3) FROM dict_tbl")
+    }
+  }
+
   test("hour, minute, second") {
     Seq(true, false).foreach { dictionaryEnabled =>
       withTempDir { dir =>