lpad/rpad do not use withNullShortCircuit, so ANSI errors are raised on rows Spark skips #7583

Describe the bug

CometStringLPad and CometStringRPad build their native expression with scalarFunctionExprToProto and do not wrap it in withNullShortCircuit:

• datafusion-comet/spark/src/main/scala/org/apache/comet/serde/strings.scala

  Lines 568 to 620 in 8f4ce5e

  	object CometStringRPad extends CometExpressionSerde[StringRPad] with CodegenDispatchFallback {
  	override def getUnsupportedReasons(): Seq[String] =
  	Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason)
  	override def getSupportLevel(expr: StringRPad): SupportLevel = {
  	if (expr.str.isInstanceOf[Literal]) {
  	return Unsupported(Some(PadReasons.literalStrReason))
  	}
  	if (!expr.pad.isInstanceOf[Literal]) {
  	return Unsupported(Some(PadReasons.nonLiteralPadReason))
  	}
  	Compatible()
  	}
  	override def convert(
  	expr: StringRPad,
  	inputs: Seq[Attribute],
  	binding: Boolean): Option[Expr] = {
  	scalarFunctionExprToProto(
  	"rpad",
  	exprToProtoInternal(expr.str, inputs, binding),
  	exprToProtoInternal(expr.len, inputs, binding),
  	exprToProtoInternal(expr.pad, inputs, binding))
  	}
  	}
  	object CometStringLPad extends CometExpressionSerde[StringLPad] with CodegenDispatchFallback {
  	override def getUnsupportedReasons(): Seq[String] =
  	Seq(PadReasons.literalStrReason, PadReasons.nonLiteralPadReason)
  	override def getSupportLevel(expr: StringLPad): SupportLevel = {
  	if (expr.str.isInstanceOf[Literal]) {
  	return Unsupported(Some(PadReasons.literalStrReason))
  	}
  	if (!expr.pad.isInstanceOf[Literal]) {
  	return Unsupported(Some(PadReasons.nonLiteralPadReason))
  	}
  	Compatible()
  	}
  	override def convert(
  	expr: StringLPad,
  	inputs: Seq[Attribute],
  	binding: Boolean): Option[Expr] = {
  	scalarFunctionExprToProto(
  	"lpad",
  	exprToProtoInternal(expr.str, inputs, binding),
  	exprToProtoInternal(expr.len, inputs, binding),
  	exprToProtoInternal(expr.pad, inputs, binding))
  	}

• withNullShortCircuit:

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

  Line 1408 in 8f4ce5e

  def withNullShortCircuit(expr: Expression, scalarFunc: Option[Expr]): Option[Expr] =

In Spark, StringLPad / StringRPad are TernaryExpressions with nullIntolerant = true. TernaryExpression.eval evaluates the len and pad children only when str is non-null. Comet evaluates all children for every row, so when a child can fail (for example an ANSI cast), Comet can raise an error on a row where str is NULL and Spark would have returned NULL without evaluating that child.

This came up in review of #7450: #7450 (comment)

Steps to reproduce

Found by reading the code; I have not run this repro yet.

SET spark.sql.ansi.enabled=true;
CREATE TABLE tbl(s STRING, t STRING) USING parquet;
INSERT INTO tbl VALUES (NULL, 'abc');
SELECT lpad(s, CAST(t AS INT)) FROM tbl;
SELECT rpad(s, CAST(t AS INT)) FROM tbl;

Expected behavior

Both queries return NULL, as Spark does, because s is NULL and Spark never evaluates CAST(t AS INT) for that row. Comet is expected to raise a cast error instead.

Additional context

Other serdes already use withNullShortCircuit for this (for example in arrays.scala, maps.scala and datetime.scala). The fix is probably to wrap the lpad/rpad proto in withNullShortCircuit and add a SQL file test for the case above.

Willingness to contribute

I can contribute a fix for this bug independently