[GLUTEN-12238][VL] Translate Velox invalid bitmap position errors to Spark's INVALID_BITMAP_POSITION exception

gluten-arrow/src/main/java/org/apache/gluten/vectorized/ColumnarBatchOutIterator.java

@@ -21,11 +21,14 @@
 import org.apache.gluten.iterator.ClosableIterator;
 import org.apache.gluten.runtime.Runtime;
 import org.apache.gluten.runtime.RuntimeAware;
+import org.apache.gluten.sql.shims.SparkShimLoader;
 
 import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException;
 import org.apache.spark.sql.vectorized.ColumnarBatch;
 
 import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class ColumnarBatchOutIterator extends ClosableIterator<ColumnarBatch>
     implements RuntimeAware {
@@ -132,6 +135,14 @@ public void requestBarrier() {
     nativeRequestBarrier(iterHandle);
   }
 
+  // Velox's bitmap_construct_agg uses a fixed 4096-byte bitmap, same as Spark's implementation.
+  private static final long BITMAP_NUM_BYTES = 4096;
+
+  // Velox's check failure message for an invalid bitmap_construct_agg position, e.g.
+  // "(-1 vs. 0) Bitmap position out of bounds". The first operand is the failing position.
+  private static final Pattern BITMAP_POSITION_OUT_OF_BOUNDS =
+      Pattern.compile("\\((-?\\d+) vs\\. -?\\d+\\)[^\\n]*Bitmap position out of bounds");
+
   /**
    * Translates a Velox type conversion error into a SchemaColumnConvertNotSupportedException.
    * Returns null if the message does not indicate a type conversion error.
@@ -143,6 +154,29 @@ private static RuntimeException translateToSchemaException(String msg) {
     return null;
   }
 
+  /**
+   * Translates a Velox invalid bitmap position error from bitmap_construct_agg into Spark's
+   * INVALID_BITMAP_POSITION exception. Returns null if the message does not indicate an invalid
+   * bitmap position, or if the running Spark version has no bitmap aggregate functions.
+   */
+  private static RuntimeException translateToInvalidBitmapPositionException(String msg) {
+    if (!msg.contains("Bitmap position out of bounds")) {
+      return null;
+    }
+    Matcher matcher = BITMAP_POSITION_OUT_OF_BOUNDS.matcher(msg);
+    if (!matcher.find()) {
+      return null;
+    }
+    final long bitPosition;
+    try {
+      bitPosition = Long.parseLong(matcher.group(1));
+    } catch (NumberFormatException ignored) {
+      return null;
+    }
+    return SparkShimLoader.getSparkShims()
+        .invalidBitmapPositionError(bitPosition, BITMAP_NUM_BYTES);
+  }
+
   @Override
   protected RuntimeException translateException(Exception e) {
     String msg = findFirstNonNullMessage(e);
@@ -152,6 +186,11 @@ protected RuntimeException translateException(Exception e) {
         schemaEx.initCause(e);
         return schemaEx;
       }
+      RuntimeException bitmapEx = translateToInvalidBitmapPositionException(msg);
+      if (bitmapEx != null) {
+        bitmapEx.initCause(e);
+        return bitmapEx;
+      }
     }
     return new GlutenException(e);
   }

gluten-ut/spark35/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -86,10 +86,6 @@ class VeloxTestSettings extends BackendTestSettings {
       "INCONSISTENT_BEHAVIOR_CROSS_VERSION: compatibility with Spark 2.4/3.2 in reading/writing dates")
     // Doesn't support unhex with failOnError=true.
     .exclude("CONVERSION_INVALID_INPUT: to_binary conversion function hex")
-    // bitmap_construct_agg offloaded to Velox throws GlutenException instead of
-    // SparkArrayIndexOutOfBoundsException.
-    .exclude("INVALID_BITMAP_POSITION: position out of bounds")
-    .exclude("INVALID_BITMAP_POSITION: negative position")
     // Different exceptions when reading Timestamp from ORC.
     .exclude("UNSUPPORTED_FEATURE - SPARK-36346: can't read Timestamp as TimestampNTZ")
   enableSuite[GlutenQueryParsingErrorsSuite]

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -288,10 +288,6 @@ class VeloxTestSettings extends BackendTestSettings {
       "INCONSISTENT_BEHAVIOR_CROSS_VERSION: compatibility with Spark 2.4/3.2 in reading/writing dates")
     // Doesn't support unhex with failOnError=true.
     .exclude("CONVERSION_INVALID_INPUT: to_binary conversion function hex")
-    // bitmap_construct_agg offloaded to Velox throws GlutenException instead of
-    // SparkArrayIndexOutOfBoundsException.
-    .exclude("INVALID_BITMAP_POSITION: position out of bounds")
-    .exclude("INVALID_BITMAP_POSITION: negative position")
     // Different exceptions when reading Timestamp from ORC.
     .exclude("UNSUPPORTED_FEATURE - SPARK-36346: can't read Timestamp as TimestampNTZ")
   enableSuite[GlutenQueryParsingErrorsSuite]

gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -297,10 +297,6 @@ class VeloxTestSettings extends BackendTestSettings {
       "INCONSISTENT_BEHAVIOR_CROSS_VERSION: compatibility with Spark 2.4/3.2 in reading/writing dates")
     // Doesn't support unhex with failOnError=true.
     .exclude("CONVERSION_INVALID_INPUT: to_binary conversion function hex")
-    // bitmap_construct_agg offloaded to Velox throws GlutenException instead of
-    // SparkArrayIndexOutOfBoundsException.
-    .exclude("INVALID_BITMAP_POSITION: position out of bounds")
-    .exclude("INVALID_BITMAP_POSITION: negative position")
     // Different exceptions when reading Timestamp from ORC.
     .exclude("UNSUPPORTED_FEATURE - SPARK-36346: can't read Timestamp as TimestampNTZ")
   enableSuite[GlutenQueryParsingErrorsSuite]

shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala

@@ -318,4 +318,11 @@ trait SparkShims {
 
   def getShuffleBlockFetcherIterator(params: ShuffleBlockFetcherIteratorParams)
       : GlutenShuffleBlockFetcherIteratorBase
+
+  /**
+   * Returns the Spark exception thrown for an invalid `bitmap_construct_agg` position
+   * (INVALID_BITMAP_POSITION), or null for Spark versions that don't have bitmap aggregate
+   * functions (< 3.5).
+   */
+  def invalidBitmapPositionError(bitPosition: Long, bitmapNumBytes: Long): RuntimeException = null
 }

shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala

@@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.util.{InternalRowComparableWrapper, Timesta
 import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
 import org.apache.spark.sql.connector.catalog.Table
 import org.apache.spark.sql.connector.read.{HasPartitionKey, InputPartition, Scan}
+import org.apache.spark.sql.errors.GlutenQueryExecutionErrors
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.datasources._
@@ -611,4 +612,9 @@ class Spark35Shims extends SparkShims {
       params.clock
     )
   }
+
+  override def invalidBitmapPositionError(
+      bitPosition: Long,
+      bitmapNumBytes: Long): RuntimeException =
+    GlutenQueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
 }

shims/spark35/src/main/scala/org/apache/spark/sql/errors/GlutenQueryExecutionErrors.scala

@@ -0,0 +1,23 @@
+/*
+ * 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.errors
+
+/** Bridge to the `private[sql]` [[QueryExecutionErrors]] for Gluten code outside `sql`. */
+object GlutenQueryExecutionErrors {
+  def invalidBitmapPositionError(bitPosition: Long, bitmapNumBytes: Long): RuntimeException =
+    QueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
+}

shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala

@@ -41,6 +41,7 @@ import org.apache.spark.sql.classic.ClassicConversions._
 import org.apache.spark.sql.connector.catalog.Table
 import org.apache.spark.sql.connector.read.{HasPartitionKey, InputPartition, Scan}
 import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.errors.GlutenQueryExecutionErrors
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.datasources._
@@ -676,4 +677,9 @@ class Spark40Shims extends SparkShims {
       params.clock
     )
   }
+
+  override def invalidBitmapPositionError(
+      bitPosition: Long,
+      bitmapNumBytes: Long): RuntimeException =
+    GlutenQueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
 }

shims/spark40/src/main/scala/org/apache/spark/sql/errors/GlutenQueryExecutionErrors.scala

@@ -0,0 +1,23 @@
+/*
+ * 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.errors
+
+/** Bridge to the `private[sql]` [[QueryExecutionErrors]] for Gluten code outside `sql`. */
+object GlutenQueryExecutionErrors {
+  def invalidBitmapPositionError(bitPosition: Long, bitmapNumBytes: Long): RuntimeException =
+    QueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
+}

shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala

@@ -40,6 +40,7 @@ import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec
 import org.apache.spark.sql.connector.catalog.Table
 import org.apache.spark.sql.connector.read.{HasPartitionKey, InputPartition, Scan}
 import org.apache.spark.sql.connector.read.streaming.SparkDataStream
+import org.apache.spark.sql.errors.GlutenQueryExecutionErrors
 import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec
 import org.apache.spark.sql.execution.datasources._
@@ -703,4 +704,9 @@ class Spark41Shims extends SparkShims {
       params.clock
     )
   }
+
+  override def invalidBitmapPositionError(
+      bitPosition: Long,
+      bitmapNumBytes: Long): RuntimeException =
+    GlutenQueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
 }

shims/spark41/src/main/scala/org/apache/spark/sql/errors/GlutenQueryExecutionErrors.scala

@@ -0,0 +1,23 @@
+/*
+ * 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.errors
+
+/** Bridge to the `private[sql]` [[QueryExecutionErrors]] for Gluten code outside `sql`. */
+object GlutenQueryExecutionErrors {
+  def invalidBitmapPositionError(bitPosition: Long, bitmapNumBytes: Long): RuntimeException =
+    QueryExecutionErrors.invalidBitmapPositionError(bitPosition, bitmapNumBytes)
+}