[SPARK-54240] Translate get array item catalyst expression to connector expression

### What changes were proposed in this pull request?
- Support conversion of catalyst GetArrayItem expression to connector expression to allow data sources to implement pushdown of this expression

### Why are the changes needed?
- To allow data sources (built-in and third-party) to implement pushdown of get array item

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
No testing needed, since we did not implement pushdowns yet

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #52940 from urosstan-db/SPARK-54240-support-get-array-item-pushdown.

Lead-authored-by: Uros Stankovic <[email protected]>
Co-authored-by: Wenchen Fan <[email protected]>
Co-authored-by: Uros Stankovic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: urosstan-db

common/utils/src/main/resources/error/error-conditions.json

@@ -1693,6 +1693,12 @@
     ],
     "sqlState" : "42846"
   },
+  "EXPRESSION_TRANSLATION_TO_V2_IS_NOT_SUPPORTED" : {
+    "message" : [
+      "Expression <expr> cannot be translated to v2 expression."
+    ],
+    "sqlState" : "0A000"
+  },
   "EXPRESSION_TYPE_IS_NOT_ORDERABLE" : {
     "message" : [
       "Column expression <expr> cannot be sorted because its type <exprType> is not orderable."

sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GetArrayItem.java

@@ -0,0 +1,55 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.connector.expressions;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.internal.connector.ExpressionWithToString;
+
+/**
+ * Get array item expression.
+ *
+ * @since 4.1.0
+ */
+
+@Evolving
+public class GetArrayItem extends ExpressionWithToString {
+
+  private final Expression childArray;
+  private final Expression ordinal;
+  private final boolean failOnError;
+
+  /**
+   * Creates GetArrayItem expression.
+   * @param childArray Array that is source to get element from. Child of this expression.
+   * @param ordinal Ordinal of element. Zero-based indexing.
+   * @param failOnError Whether expression should throw exception for index out of bound or to
+   *                    return null.
+   */
+  public GetArrayItem(Expression childArray, Expression ordinal, boolean failOnError) {
+    this.childArray = childArray;
+    this.ordinal = ordinal;
+    this.failOnError = failOnError;
+  }
+
+  public Expression childArray() { return this.childArray; }
+  public Expression ordinal() { return this.ordinal; }
+  public boolean failOnError() { return this.failOnError; }
+
+  @Override
+  public Expression[] children() { return new Expression[]{ childArray, ordinal }; }
+}

sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java

@@ -29,6 +29,7 @@
 import org.apache.spark.sql.connector.expressions.Extract;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 import org.apache.spark.sql.connector.expressions.GeneralScalarExpression;
+import org.apache.spark.sql.connector.expressions.GetArrayItem;
 import org.apache.spark.sql.connector.expressions.Literal;
 import org.apache.spark.sql.connector.expressions.NullOrdering;
 import org.apache.spark.sql.connector.expressions.SortDirection;
@@ -84,6 +85,8 @@ public String build(Expression expr) {
     } else if (expr instanceof SortOrder sortOrder) {
       return visitSortOrder(
         build(sortOrder.expression()), sortOrder.direction(), sortOrder.nullOrdering());
+    } else if (expr instanceof GetArrayItem getArrayItem) {
+      return visitGetArrayItem(getArrayItem);
     } else if (expr instanceof GeneralScalarExpression e) {
       String name = e.name();
       return switch (name) {
@@ -348,6 +351,13 @@ protected String visitTrim(String direction, String[] inputs) {
     }
   }
 
+  protected String visitGetArrayItem(GetArrayItem getArrayItem) {
+    throw new SparkUnsupportedOperationException(
+      "EXPRESSION_TRANSLATION_TO_V2_IS_NOT_SUPPORTED",
+      Map.of("expr", getArrayItem.toString())
+    );
+  }
+
   protected String visitExtract(Extract extract) {
     return visitExtract(extract.field(), build(extract.source()));
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala

@@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression,
 import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke}
 import org.apache.spark.sql.catalyst.optimizer.ConstantFolding
 import org.apache.spark.sql.connector.catalog.functions.ScalarFunction
-import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression => V2Expression, Extract => V2Extract, FieldReference, GeneralScalarExpression, LiteralValue, NullOrdering, SortDirection, SortValue, UserDefinedScalarFunc}
+import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression => V2Expression, Extract => V2Extract, FieldReference, GeneralScalarExpression, GetArrayItem => V2GetArrayItem, LiteralValue, NullOrdering, SortDirection, SortValue, UserDefinedScalarFunc}
 import org.apache.spark.sql.connector.expressions.aggregate.{AggregateFunc, Avg, Count, CountStar, GeneralAggregateFunc, Max, Min, Sum, UserDefinedAggregateFunc}
 import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTrue, And => V2And, Not => V2Not, Or => V2Or, Predicate => V2Predicate}
 import org.apache.spark.sql.internal.SQLConf
@@ -326,6 +326,13 @@ class V2ExpressionBuilder(e: Expression, isPredicate: Boolean = false) extends L
     case _: Sha2 => generateExpressionWithName("SHA2", expr, isPredicate)
     case _: StringLPad => generateExpressionWithName("LPAD", expr, isPredicate)
     case _: StringRPad => generateExpressionWithName("RPAD", expr, isPredicate)
+    case GetArrayItem(child, ordinal, failOnError) =>
+      (generateExpression(child), generateExpression(ordinal)) match {
+        case (Some(v2ArrayChild), Some(v2Ordinal)) =>
+          Some(new V2GetArrayItem(v2ArrayChild, v2Ordinal, failOnError))
+        case _ =>
+          None
+      }
     // TODO supports other expressions
     case ApplyFunctionExpression(function, children) =>
       val childrenExpressions = children.flatMap(generateExpression(_))

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/ToStringSQLBuilder.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.internal.connector
 
+import org.apache.spark.sql.connector.expressions.GetArrayItem
 import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder
 
 /**
@@ -35,4 +36,8 @@ class ToStringSQLBuilder extends V2ExpressionSQLBuilder with Serializable {
     val distinct = if (isDistinct) "DISTINCT " else ""
     s"""$funcName($distinct${inputs.mkString(", ")})"""
   }
+
+  override protected def visitGetArrayItem(getArrayItem: GetArrayItem): String = {
+    s"${getArrayItem.childArray.toString}[${getArrayItem.ordinal.toString}]"
+  }
 }