chore: [FOLLOWUP] Drop support for Spark 3.3 (EOL) (apache#1534)

Author: kazuyukitanimura

common/pom.xml

@@ -181,7 +181,6 @@ under the License.
               <sources>
                 <source>src/main/${shims.majorVerSrc}</source>
                 <source>src/main/${shims.minorVerSrc}</source>
-                <source>src/main/${shims.pre35Src}</source>
               </sources>
             </configuration>
           </execution>

common/src/main/java/org/apache/comet/parquet/ConstantColumnReader.java

@@ -23,11 +23,10 @@
 
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns;
 import org.apache.spark.sql.types.*;
 import org.apache.spark.unsafe.types.UTF8String;
 
-import org.apache.comet.shims.ShimResolveDefaultColumns;
-
 /**
  * A column reader that always return constant vectors. Used for reading partition columns, for
  * instance.
@@ -41,7 +40,9 @@ public class ConstantColumnReader extends MetadataColumnReader {
 
   public ConstantColumnReader(StructField field, int batchSize, boolean useDecimal128) {
     this(field.dataType(), TypeUtil.convertToParquet(field), batchSize, useDecimal128);
-    this.value = ShimResolveDefaultColumns.getExistenceDefaultValue(field);
+    this.value =
+        ResolveDefaultColumns.getExistenceDefaultValues(new StructType(new StructField[] {field}))[
+            0];
     init(value);
   }
 

common/src/main/scala/org/apache/spark/sql/comet/CastOverflowException.scala

@@ -19,7 +19,17 @@
 
 package org.apache.spark.sql.comet
 
-import org.apache.spark.sql.comet.shims.ShimCastOverflowException
+import org.apache.spark.SparkArithmeticException
+import org.apache.spark.sql.errors.QueryExecutionErrors.toSQLConf
+import org.apache.spark.sql.internal.SQLConf
 
 class CastOverflowException(t: String, from: String, to: String)
-    extends ShimCastOverflowException(t, from, to) {}
+    extends SparkArithmeticException(
+      "CAST_OVERFLOW",
+      Map(
+        "value" -> t,
+        "sourceType" -> s""""$from"""",
+        "targetType" -> s""""$to"""",
+        "ansiConfig" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+      Array.empty,
+      "") {}

…comet/shims/ShimCometBatchScanExec.scala …/apache/comet/shims/ShimFileFormat.scalaspark/src/main/spark-3.x/org/apache/comet/shims/ShimCometBatchScanExec.scala renamed to common/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala spark/src/main/spark-3.x/org/apache/comet/shims/ShimCometBatchScanExec.scala renamed to common/src/main/spark-3.4/org/apache/comet/shims/ShimFileFormat.scala

@@ -19,15 +19,15 @@
 
 package org.apache.comet.shims
 
-import org.apache.spark.sql.catalyst.expressions.SortOrder
-import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+import org.apache.spark.sql.execution.datasources.{FileFormat, RowIndexUtil}
+import org.apache.spark.sql.types.StructType
 
-trait ShimCometBatchScanExec {
-  def wrapped: BatchScanExec
+object ShimFileFormat {
 
-  // Only for Spark 3.4+
-  def ordering: Option[Seq[SortOrder]] = wrapped.getClass.getDeclaredMethods
-    .filter(_.getName == "ordering")
-    .flatMap(_.invoke(wrapped).asInstanceOf[Option[Seq[SortOrder]]])
-    .headOption
+  // A name for a temporary column that holds row indexes computed by the file format reader
+  // until they can be placed in the _metadata struct.
+  val ROW_INDEX_TEMPORARY_COLUMN_NAME: String = FileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
+
+  def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int =
+    RowIndexUtil.findRowIndexColumnIndexInSchema(sparkSchema)
 }

common/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimCastOverflowException.scala

@@ -19,17 +19,15 @@
 
 package org.apache.comet.shims
 
-import org.apache.spark.sql.execution.TakeOrderedAndProjectExec
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.execution.datasources.parquet.ParquetRowIndexUtil
+import org.apache.spark.sql.types.StructType
 
-trait ShimCometTakeOrderedAndProjectExec {
+object ShimFileFormat {
+  // A name for a temporary column that holds row indexes computed by the file format reader
+  // until they can be placed in the _metadata struct.
+  val ROW_INDEX_TEMPORARY_COLUMN_NAME = ParquetFileFormat.ROW_INDEX_TEMPORARY_COLUMN_NAME
 
-  /**
-   * TODO: delete after dropping Spark 3.3 support
-   */
-  protected def getOffset(plan: TakeOrderedAndProjectExec): Option[Int] = {
-    plan.getClass.getDeclaredFields
-      .filter(_.getName == "offset")
-      .map { a => a.setAccessible(true); a.get(plan).asInstanceOf[Int] }
-      .headOption
-  }
+  def findRowIndexColumnIndexInSchema(sparkSchema: StructType): Int =
+    ParquetRowIndexUtil.findRowIndexColumnIndexInSchema(sparkSchema)
 }