[BUG] Exception happened when getting a null row

[wbo4958]

This issue is for #3942

Exception

  Cause: java.lang.AssertionError: value at 6 is null
  at ai.rapids.cudf.HostColumnVectorCore.assertsForGet(HostColumnVectorCore.java:228)
  at ai.rapids.cudf.HostColumnVectorCore.getInt(HostColumnVectorCore.java:254)
  at com.nvidia.spark.rapids.RapidsHostColumnVectorCore.getInt(RapidsHostColumnVectorCore.java:126)
  at org.apache.spark.sql.vectorized.ColumnarArray.getInt(ColumnarArray.java:128)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificSafeProjection.MapObjects_0$(generated.java:57)
  at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificSafeProjection.apply(generated.java:28)
  at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$Deserializer.apply(ExpressionEncoder.scala:181)
  at com.nvidia.spark.rapids.shims.v2.GpuRowBasedScalaUDF.$anonfun$scalaConverter$2(GpuRowBasedScalaUDF.scala:70)
  at org.apache.spark.sql.rapids.GpuRowBasedScalaUDFBase.$anonfun$childAccessors$2(GpuScalaUDF.scala:141)

Root Cause,

Spark may have some bug on the CodeGen for MapObjects, please see https://github.com/apache/spark/blob/branch-3.2/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/objects/objects.scala#L1095-L1097

The generated code is like below

  private scala.collection.mutable.WrappedArray MapObjects_0(InternalRow i) {
    boolean isNull_1 = i.isNullAt(0);
    ArrayData value_1 = isNull_1 ?
        null : (i.getArray(0));
    scala.collection.mutable.WrappedArray value_0 = null;

    if (!isNull_1) {

      int dataLength_0 = value_1.numElements();

      scala.collection.mutable.Builder collectionBuilder_0 =
          scala.collection.mutable.WrappedArray$.MODULE$.newBuilder();
      collectionBuilder_0.sizeHint(dataLength_0);


      int loopIndex_0 = 0;

      while (loopIndex_0 < dataLength_0) {
        value_MapObject_lambda_variable_1 = (int) (value_1.getInt(loopIndex_0));
        isNull_MapObject_lambda_variable_1 = value_1.isNullAt(loopIndex_0);

        boolean isNull_2 = true;
        int value_2 = -1;

        if (!isNull_MapObject_lambda_variable_1) {


          isNull_2 = false; // resultCode could change nullability.

          value_2 = value_MapObject_lambda_variable_1 + 1;


        }
        if (isNull_2) {
          collectionBuilder_0.$plus$eq(null);
        } else {
          collectionBuilder_0.$plus$eq(value_2);
        }

        loopIndex_0 += 1;
      }

      value_0 = (scala.collection.mutable.WrappedArray) scala.collection.mutable.WrappedArray$.MODULE$.make(((scala.collection.IndexedSeq)collectionBuilder_0.result()).toArray(scala.reflect.ClassTag$.MODULE$.Object()));
    }
    globalIsNull_0 = isNull_1;
    return value_0;
  }

We can see, it first call getInt then call isNullAt instead of isNullAt and getInt

        value_MapObject_lambda_variable_1 = (int) (value_1.getInt(loopIndex_0));
        isNull_MapObject_lambda_variable_1 = value_1.isNullAt(loopIndex_0);

I checked RapidsHostColumnVectorCore and found it didn't check isNullAt and call cudf.getInt, where throw exception when the row is null.

I also checked the getInt of OnHeapColumnVector

  private int[] intData;
  public int getInt(int rowId) {
    if (dictionary == null) {
      return intData[rowId];
    } else {
      return dictionary.decodeToInt(dictionaryIds.getDictId(rowId));
    }
  }

it will return the default value. Same as OffHeapColumnVector.

Obviously, this is caused by Spark code. But we shouldn't throw exception which cause the spark job failed.