sunchao commented on code in PR #155:
URL:
https://github.com/apache/arrow-datafusion-comet/pull/155#discussion_r1511650731
##########
spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala:
##########
@@ -330,1243 +333,1263 @@ object QueryPlanSerde extends Logging with
ShimQueryPlanSerde {
}
}
- def exprToProto(expr: Expression, input: Seq[Attribute]): Option[Expr] = {
- val conf = SQLConf.get
- val newExpr =
- DecimalPrecision.promote(conf.decimalOperationsAllowPrecisionLoss, expr,
!conf.ansiEnabled)
- exprToProtoInternal(newExpr, input)
- }
-
- def exprToProtoInternal(expr: Expression, inputs: Seq[Attribute]):
Option[Expr] = {
- SQLConf.get
- expr match {
- case a @ Alias(_, _) =>
- exprToProtoInternal(a.child, inputs)
-
- case cast @ Cast(_: Literal, dataType, _, _) =>
- // This can happen after promoting decimal precisions
- val value = cast.eval()
- exprToProtoInternal(Literal(value, dataType), inputs)
-
- case Cast(child, dt, timeZoneId, _) =>
- val childExpr = exprToProtoInternal(child, inputs)
- val dataType = serializeDataType(dt)
-
- if (childExpr.isDefined && dataType.isDefined) {
- val castBuilder = ExprOuterClass.Cast.newBuilder()
- castBuilder.setChild(childExpr.get)
- castBuilder.setDatatype(dataType.get)
+ /**
+ * Convert a Spark expression to protobuf.
+ *
+ * @param expr
+ * The input expression
+ * @param inputs
+ * The input attributes
+ * @param binding
+ * Whether to bind the expression to the input attributes
+ * @return
+ * The protobuf representation of the expression, or None if the
expression is not supported
+ */
+ def exprToProto(
+ expr: Expression,
+ input: Seq[Attribute],
+ binding: Boolean = true): Option[Expr] = {
+
+ def exprToProtoInternal(expr: Expression, inputs: Seq[Attribute]):
Option[Expr] = {
+ SQLConf.get
+ expr match {
+ case a @ Alias(_, _) =>
+ exprToProtoInternal(a.child, inputs)
+
+ case cast @ Cast(_: Literal, dataType, _, _) =>
+ // This can happen after promoting decimal precisions
+ val value = cast.eval()
+ exprToProtoInternal(Literal(value, dataType), inputs)
+
+ case Cast(child, dt, timeZoneId, _) =>
+ val childExpr = exprToProtoInternal(child, inputs)
+ val dataType = serializeDataType(dt)
- val timeZone = timeZoneId.getOrElse("UTC")
- castBuilder.setTimezone(timeZone)
+ if (childExpr.isDefined && dataType.isDefined) {
+ val castBuilder = ExprOuterClass.Cast.newBuilder()
+ castBuilder.setChild(childExpr.get)
+ castBuilder.setDatatype(dataType.get)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setCast(castBuilder)
- .build())
- } else {
- None
- }
+ val timeZone = timeZoneId.getOrElse("UTC")
+ castBuilder.setTimezone(timeZone)
- case add @ Add(left, right, _) if supportedDataType(left.dataType) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val addBuilder = ExprOuterClass.Add.newBuilder()
- addBuilder.setLeft(leftExpr.get)
- addBuilder.setRight(rightExpr.get)
- addBuilder.setFailOnError(getFailOnError(add))
- serializeDataType(add.dataType).foreach { t =>
- addBuilder.setReturnType(t)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setCast(castBuilder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setAdd(addBuilder)
- .build())
- } else {
- None
- }
+ case add @ Add(left, right, _) if supportedDataType(left.dataType) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val addBuilder = ExprOuterClass.Add.newBuilder()
+ addBuilder.setLeft(leftExpr.get)
+ addBuilder.setRight(rightExpr.get)
+ addBuilder.setFailOnError(getFailOnError(add))
+ serializeDataType(add.dataType).foreach { t =>
+ addBuilder.setReturnType(t)
+ }
- case sub @ Subtract(left, right, _) if supportedDataType(left.dataType)
=>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Subtract.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
- builder.setFailOnError(getFailOnError(sub))
- serializeDataType(sub.dataType).foreach { t =>
- builder.setReturnType(t)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setAdd(addBuilder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setSubtract(builder)
- .build())
- } else {
- None
- }
+ case sub @ Subtract(left, right, _) if
supportedDataType(left.dataType) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Subtract.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+ builder.setFailOnError(getFailOnError(sub))
+ serializeDataType(sub.dataType).foreach { t =>
+ builder.setReturnType(t)
+ }
- case mul @ Multiply(left, right, _)
- if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Multiply.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
- builder.setFailOnError(getFailOnError(mul))
- serializeDataType(mul.dataType).foreach { t =>
- builder.setReturnType(t)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setSubtract(builder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setMultiply(builder)
- .build())
- } else {
- None
- }
+ case mul @ Multiply(left, right, _)
+ if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Multiply.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+ builder.setFailOnError(getFailOnError(mul))
+ serializeDataType(mul.dataType).foreach { t =>
+ builder.setReturnType(t)
+ }
- case div @ Divide(left, right, _)
- if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- // Datafusion now throws an exception for dividing by zero
- // See https://github.com/apache/arrow-datafusion/pull/6792
- // For now, use NullIf to swap zeros with nulls.
- val rightExpr = exprToProtoInternal(nullIfWhenPrimitive(right), inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Divide.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
- builder.setFailOnError(getFailOnError(div))
- serializeDataType(div.dataType).foreach { t =>
- builder.setReturnType(t)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setMultiply(builder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setDivide(builder)
- .build())
- } else {
- None
- }
+ case div @ Divide(left, right, _)
+ if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ // Datafusion now throws an exception for dividing by zero
+ // See https://github.com/apache/arrow-datafusion/pull/6792
+ // For now, use NullIf to swap zeros with nulls.
+ val rightExpr = exprToProtoInternal(nullIfWhenPrimitive(right),
inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Divide.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+ builder.setFailOnError(getFailOnError(div))
+ serializeDataType(div.dataType).foreach { t =>
+ builder.setReturnType(t)
+ }
- case rem @ Remainder(left, right, _)
- if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(nullIfWhenPrimitive(right), inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Remainder.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
- builder.setFailOnError(getFailOnError(rem))
- serializeDataType(rem.dataType).foreach { t =>
- builder.setReturnType(t)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setDivide(builder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setRemainder(builder)
- .build())
- } else {
- None
- }
-
- case EqualTo(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Equal.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
-
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setEq(builder)
- .build())
- } else {
- None
- }
-
- case Not(EqualTo(left, right)) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.NotEqual.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
-
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setNeq(builder)
- .build())
- } else {
- None
- }
-
- case EqualNullSafe(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.EqualNullSafe.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
-
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setEqNullSafe(builder)
- .build())
- } else {
- None
- }
-
- case Not(EqualNullSafe(left, right)) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ case rem @ Remainder(left, right, _)
+ if supportedDataType(left.dataType) &&
!decimalBeforeSpark34(left.dataType) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(nullIfWhenPrimitive(right),
inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Remainder.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+ builder.setFailOnError(getFailOnError(rem))
+ serializeDataType(rem.dataType).foreach { t =>
+ builder.setReturnType(t)
+ }
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.NotEqualNullSafe.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setRemainder(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setNeqNullSafe(builder)
- .build())
- } else {
- None
- }
+ case EqualTo(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case GreaterThan(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Equal.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.GreaterThan.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setEq(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setGt(builder)
- .build())
- } else {
- None
- }
+ case Not(EqualTo(left, right)) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case GreaterThanOrEqual(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.NotEqual.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.GreaterThanEqual.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setNeq(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setGtEq(builder)
- .build())
- } else {
- None
- }
+ case EqualNullSafe(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case LessThan(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.EqualNullSafe.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.LessThan.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setEqNullSafe(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setLt(builder)
- .build())
- } else {
- None
- }
+ case Not(EqualNullSafe(left, right)) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case LessThanOrEqual(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.NotEqualNullSafe.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.LessThanEqual.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setNeqNullSafe(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setLtEq(builder)
- .build())
- } else {
- None
- }
+ case GreaterThan(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case Literal(value, dataType) if supportedDataType(dataType) =>
- val exprBuilder = ExprOuterClass.Literal.newBuilder()
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.GreaterThan.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (value == null) {
- exprBuilder.setIsNull(true)
- } else {
- exprBuilder.setIsNull(false)
- dataType match {
- case _: BooleanType =>
exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
- case _: ByteType =>
exprBuilder.setByteVal(value.asInstanceOf[Byte])
- case _: ShortType =>
exprBuilder.setShortVal(value.asInstanceOf[Short])
- case _: IntegerType =>
exprBuilder.setIntVal(value.asInstanceOf[Int])
- case _: LongType =>
exprBuilder.setLongVal(value.asInstanceOf[Long])
- case _: FloatType =>
exprBuilder.setFloatVal(value.asInstanceOf[Float])
- case _: DoubleType =>
exprBuilder.setDoubleVal(value.asInstanceOf[Double])
- case _: StringType =>
- exprBuilder.setStringVal(value.asInstanceOf[UTF8String].toString)
- case _: TimestampType =>
exprBuilder.setLongVal(value.asInstanceOf[Long])
- case _: DecimalType =>
- // Pass decimal literal as bytes.
- val unscaled =
value.asInstanceOf[Decimal].toBigDecimal.underlying.unscaledValue
- exprBuilder.setDecimalVal(
- com.google.protobuf.ByteString.copyFrom(unscaled.toByteArray))
- case _: BinaryType =>
- val byteStr =
-
com.google.protobuf.ByteString.copyFrom(value.asInstanceOf[Array[Byte]])
- exprBuilder.setBytesVal(byteStr)
- case _: DateType => exprBuilder.setIntVal(value.asInstanceOf[Int])
- case dt =>
- logWarning(s"Unexpected date type '$dt' for literal value
'$value'")
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setGt(builder)
+ .build())
+ } else {
+ None
}
- }
- val dt = serializeDataType(dataType)
+ case GreaterThanOrEqual(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- if (dt.isDefined) {
- exprBuilder.setDatatype(dt.get)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.GreaterThanEqual.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setLiteral(exprBuilder)
- .build())
- } else {
- None
- }
-
- case Substring(str, Literal(pos, _), Literal(len, _)) =>
- val strExpr = exprToProtoInternal(str, inputs)
-
- if (strExpr.isDefined) {
- val builder = ExprOuterClass.Substring.newBuilder()
- builder.setChild(strExpr.get)
- builder.setStart(pos.asInstanceOf[Int])
- builder.setLen(len.asInstanceOf[Int])
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setGtEq(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setSubstring(builder)
- .build())
- } else {
- None
- }
+ case LessThan(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case Like(left, right, _) =>
- // TODO escapeChar
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.LessThan.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Like.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setLt(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setLike(builder)
- .build())
- } else {
- None
- }
+ case LessThanOrEqual(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- // TODO waiting for arrow-rs update
-// case RLike(left, right) =>
-// val leftExpr = exprToProtoInternal(left, inputs)
-// val rightExpr = exprToProtoInternal(right, inputs)
-//
-// if (leftExpr.isDefined && rightExpr.isDefined) {
-// val builder = ExprOuterClass.RLike.newBuilder()
-// builder.setLeft(leftExpr.get)
-// builder.setRight(rightExpr.get)
-//
-// Some(
-// ExprOuterClass.Expr
-// .newBuilder()
-// .setRlike(builder)
-// .build())
-// } else {
-// None
-// }
-
- case StartsWith(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.StartsWith.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.LessThanEqual.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setStartsWith(builder)
- .build())
- } else {
- None
- }
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setLtEq(builder)
+ .build())
+ } else {
+ None
+ }
- case EndsWith(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ case Literal(value, dataType) if supportedDataType(dataType) =>
+ val exprBuilder = ExprOuterClass.Literal.newBuilder()
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.EndsWith.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ if (value == null) {
+ exprBuilder.setIsNull(true)
+ } else {
+ exprBuilder.setIsNull(false)
+ dataType match {
+ case _: BooleanType =>
exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
+ case _: ByteType =>
exprBuilder.setByteVal(value.asInstanceOf[Byte])
+ case _: ShortType =>
exprBuilder.setShortVal(value.asInstanceOf[Short])
+ case _: IntegerType =>
exprBuilder.setIntVal(value.asInstanceOf[Int])
+ case _: LongType =>
exprBuilder.setLongVal(value.asInstanceOf[Long])
+ case _: FloatType =>
exprBuilder.setFloatVal(value.asInstanceOf[Float])
+ case _: DoubleType =>
exprBuilder.setDoubleVal(value.asInstanceOf[Double])
+ case _: StringType =>
+
exprBuilder.setStringVal(value.asInstanceOf[UTF8String].toString)
+ case _: TimestampType =>
exprBuilder.setLongVal(value.asInstanceOf[Long])
+ case _: DecimalType =>
+ // Pass decimal literal as bytes.
+ val unscaled =
value.asInstanceOf[Decimal].toBigDecimal.underlying.unscaledValue
+ exprBuilder.setDecimalVal(
+
com.google.protobuf.ByteString.copyFrom(unscaled.toByteArray))
+ case _: BinaryType =>
+ val byteStr =
+
com.google.protobuf.ByteString.copyFrom(value.asInstanceOf[Array[Byte]])
+ exprBuilder.setBytesVal(byteStr)
+ case _: DateType =>
exprBuilder.setIntVal(value.asInstanceOf[Int])
+ case dt =>
+ logWarning(s"Unexpected date type '$dt' for literal value
'$value'")
+ }
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setEndsWith(builder)
- .build())
- } else {
- None
- }
+ val dt = serializeDataType(dataType)
- case Contains(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
+ if (dt.isDefined) {
+ exprBuilder.setDatatype(dt.get)
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Contains.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setLiteral(exprBuilder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setContains(builder)
- .build())
- } else {
- None
- }
+ case Substring(str, Literal(pos, _), Literal(len, _)) =>
+ val strExpr = exprToProtoInternal(str, inputs)
- case StringSpace(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ if (strExpr.isDefined) {
+ val builder = ExprOuterClass.Substring.newBuilder()
+ builder.setChild(strExpr.get)
+ builder.setStart(pos.asInstanceOf[Int])
+ builder.setLen(len.asInstanceOf[Int])
- if (childExpr.isDefined) {
- val builder = ExprOuterClass.StringSpace.newBuilder()
- builder.setChild(childExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setSubstring(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setStringSpace(builder)
- .build())
- } else {
- None
- }
+ case Like(left, right, _) =>
+ // TODO escapeChar
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Like.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setLike(builder)
+ .build())
+ } else {
+ None
+ }
- case Hour(child, timeZoneId) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ // TODO waiting for arrow-rs update
+ // case RLike(left, right) =>
+ // val leftExpr = exprToProtoInternal(left, inputs)
+ // val rightExpr = exprToProtoInternal(right, inputs)
+ //
+ // if (leftExpr.isDefined && rightExpr.isDefined) {
+ // val builder = ExprOuterClass.RLike.newBuilder()
+ // builder.setLeft(leftExpr.get)
+ // builder.setRight(rightExpr.get)
+ //
+ // Some(
+ // ExprOuterClass.Expr
+ // .newBuilder()
+ // .setRlike(builder)
+ // .build())
+ // } else {
+ // None
+ // }
+
+ case StartsWith(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
+
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.StartsWith.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
+
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setStartsWith(builder)
+ .build())
+ } else {
+ None
+ }
- if (childExpr.isDefined) {
- val builder = ExprOuterClass.Hour.newBuilder()
- builder.setChild(childExpr.get)
+ case EndsWith(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- val timeZone = timeZoneId.getOrElse("UTC")
- builder.setTimezone(timeZone)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.EndsWith.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setHour(builder)
- .build())
- } else {
- None
- }
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setEndsWith(builder)
+ .build())
+ } else {
+ None
+ }
- case Minute(child, timeZoneId) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ case Contains(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- if (childExpr.isDefined) {
- val builder = ExprOuterClass.Minute.newBuilder()
- builder.setChild(childExpr.get)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Contains.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- val timeZone = timeZoneId.getOrElse("UTC")
- builder.setTimezone(timeZone)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setContains(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setMinute(builder)
- .build())
- } else {
- None
- }
+ case StringSpace(child) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case TruncDate(child, format) =>
- val childExpr = exprToProtoInternal(child, inputs)
- val formatExpr = exprToProtoInternal(format, inputs)
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.StringSpace.newBuilder()
+ builder.setChild(childExpr.get)
- if (childExpr.isDefined && formatExpr.isDefined) {
- val builder = ExprOuterClass.TruncDate.newBuilder()
- builder.setChild(childExpr.get)
- builder.setFormat(formatExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setStringSpace(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setTruncDate(builder)
- .build())
- } else {
- None
- }
+ case Hour(child, timeZoneId) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case TruncTimestamp(format, child, timeZoneId) =>
- val childExpr = exprToProtoInternal(child, inputs)
- val formatExpr = exprToProtoInternal(format, inputs)
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.Hour.newBuilder()
+ builder.setChild(childExpr.get)
- if (childExpr.isDefined && formatExpr.isDefined) {
- val builder = ExprOuterClass.TruncTimestamp.newBuilder()
- builder.setChild(childExpr.get)
- builder.setFormat(formatExpr.get)
+ val timeZone = timeZoneId.getOrElse("UTC")
+ builder.setTimezone(timeZone)
- val timeZone = timeZoneId.getOrElse("UTC")
- builder.setTimezone(timeZone)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setHour(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setTruncTimestamp(builder)
- .build())
- } else {
- None
- }
+ case Minute(child, timeZoneId) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case Second(child, timeZoneId) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.Minute.newBuilder()
+ builder.setChild(childExpr.get)
- if (childExpr.isDefined) {
- val builder = ExprOuterClass.Second.newBuilder()
- builder.setChild(childExpr.get)
+ val timeZone = timeZoneId.getOrElse("UTC")
+ builder.setTimezone(timeZone)
- val timeZone = timeZoneId.getOrElse("UTC")
- builder.setTimezone(timeZone)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setMinute(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setSecond(builder)
- .build())
- } else {
- None
- }
+ case TruncDate(child, format) =>
+ val childExpr = exprToProtoInternal(child, inputs)
+ val formatExpr = exprToProtoInternal(format, inputs)
- case Year(child) =>
- val periodType = exprToProtoInternal(Literal("year"), inputs)
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("datepart", Seq(periodType, childExpr): _*)
- .map(e => {
- Expr
- .newBuilder()
- .setCast(
- ExprOuterClass.Cast
- .newBuilder()
- .setChild(e)
- .setDatatype(serializeDataType(IntegerType).get)
- .build())
- .build()
- })
+ if (childExpr.isDefined && formatExpr.isDefined) {
+ val builder = ExprOuterClass.TruncDate.newBuilder()
+ builder.setChild(childExpr.get)
+ builder.setFormat(formatExpr.get)
- case IsNull(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setTruncDate(builder)
+ .build())
+ } else {
+ None
+ }
- if (childExpr.isDefined) {
- val castBuilder = ExprOuterClass.IsNull.newBuilder()
- castBuilder.setChild(childExpr.get)
+ case TruncTimestamp(format, child, timeZoneId) =>
+ val childExpr = exprToProtoInternal(child, inputs)
+ val formatExpr = exprToProtoInternal(format, inputs)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setIsNull(castBuilder)
- .build())
- } else {
- None
- }
+ if (childExpr.isDefined && formatExpr.isDefined) {
+ val builder = ExprOuterClass.TruncTimestamp.newBuilder()
+ builder.setChild(childExpr.get)
+ builder.setFormat(formatExpr.get)
- case IsNotNull(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ val timeZone = timeZoneId.getOrElse("UTC")
+ builder.setTimezone(timeZone)
- if (childExpr.isDefined) {
- val castBuilder = ExprOuterClass.IsNotNull.newBuilder()
- castBuilder.setChild(childExpr.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setTruncTimestamp(builder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setIsNotNull(castBuilder)
- .build())
- } else {
- None
- }
+ case Second(child, timeZoneId) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case SortOrder(child, direction, nullOrdering, _) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.Second.newBuilder()
+ builder.setChild(childExpr.get)
- if (childExpr.isDefined) {
- val sortOrderBuilder = ExprOuterClass.SortOrder.newBuilder()
- sortOrderBuilder.setChild(childExpr.get)
+ val timeZone = timeZoneId.getOrElse("UTC")
+ builder.setTimezone(timeZone)
- direction match {
- case Ascending => sortOrderBuilder.setDirectionValue(0)
- case Descending => sortOrderBuilder.setDirectionValue(1)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setSecond(builder)
+ .build())
+ } else {
+ None
}
- nullOrdering match {
- case NullsFirst => sortOrderBuilder.setNullOrderingValue(0)
- case NullsLast => sortOrderBuilder.setNullOrderingValue(1)
+ case Year(child) =>
+ val periodType = exprToProtoInternal(Literal("year"), inputs)
+ val childExpr = exprToProtoInternal(child, inputs)
+ scalarExprToProto("datepart", Seq(periodType, childExpr): _*)
+ .map(e => {
+ Expr
+ .newBuilder()
+ .setCast(
+ ExprOuterClass.Cast
+ .newBuilder()
+ .setChild(e)
+ .setDatatype(serializeDataType(IntegerType).get)
+ .build())
+ .build()
+ })
+
+ case IsNull(child) =>
+ val childExpr = exprToProtoInternal(child, inputs)
+
+ if (childExpr.isDefined) {
+ val castBuilder = ExprOuterClass.IsNull.newBuilder()
+ castBuilder.setChild(childExpr.get)
+
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setIsNull(castBuilder)
+ .build())
+ } else {
+ None
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setSortOrder(sortOrderBuilder)
- .build())
- } else {
- None
- }
-
- case And(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.And.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
-
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setAnd(builder)
- .build())
- } else {
- None
- }
-
- case Or(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
-
- if (leftExpr.isDefined && rightExpr.isDefined) {
- val builder = ExprOuterClass.Or.newBuilder()
- builder.setLeft(leftExpr.get)
- builder.setRight(rightExpr.get)
-
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setOr(builder)
- .build())
- } else {
- None
- }
-
- case UnaryExpression(child) if expr.prettyName == "promote_precision" =>
- // `UnaryExpression` includes `PromotePrecision` for Spark 3.2 & 3.3
- // `PromotePrecision` is just a wrapper, don't need to serialize it.
- exprToProtoInternal(child, inputs)
-
- case CheckOverflow(child, dt, nullOnOverflow) =>
- val childExpr = exprToProtoInternal(child, inputs)
+ case IsNotNull(child) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- if (childExpr.isDefined) {
- val builder = ExprOuterClass.CheckOverflow.newBuilder()
- builder.setChild(childExpr.get)
- builder.setFailOnError(!nullOnOverflow)
+ if (childExpr.isDefined) {
+ val castBuilder = ExprOuterClass.IsNotNull.newBuilder()
+ castBuilder.setChild(childExpr.get)
- // `dataType` must be decimal type
- val dataType = serializeDataType(dt)
- builder.setDatatype(dataType.get)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setIsNotNull(castBuilder)
+ .build())
+ } else {
+ None
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setCheckOverflow(builder)
- .build())
- } else {
- None
- }
+ case SortOrder(child, direction, nullOrdering, _) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case attr: AttributeReference =>
- val dataType = serializeDataType(attr.dataType)
+ if (childExpr.isDefined) {
+ val sortOrderBuilder = ExprOuterClass.SortOrder.newBuilder()
+ sortOrderBuilder.setChild(childExpr.get)
- if (dataType.isDefined) {
- val boundRef = BindReferences
- .bindReference(attr, inputs, allowFailures = false)
- .asInstanceOf[BoundReference]
- val boundExpr = ExprOuterClass.BoundReference
- .newBuilder()
- .setIndex(boundRef.ordinal)
- .setDatatype(dataType.get)
- .build()
+ direction match {
+ case Ascending => sortOrderBuilder.setDirectionValue(0)
+ case Descending => sortOrderBuilder.setDirectionValue(1)
+ }
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setBound(boundExpr)
- .build())
- } else {
- None
- }
+ nullOrdering match {
+ case NullsFirst => sortOrderBuilder.setNullOrderingValue(0)
+ case NullsLast => sortOrderBuilder.setNullOrderingValue(1)
+ }
- case Abs(child, _) =>
- exprToProtoInternal(child, inputs).map(childExpr => {
- val abs =
- ExprOuterClass.Abs
- .newBuilder()
- .setChild(childExpr)
- .build()
- Expr.newBuilder().setAbs(abs).build()
- })
-
- case Acos(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("acos", childExpr)
-
- case Asin(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("asin", childExpr)
-
- case Atan(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("atan", childExpr)
-
- case Atan2(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
- scalarExprToProto("atan2", leftExpr, rightExpr)
-
- case e @ Ceil(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- child.dataType match {
- case t: DecimalType if t.scale == 0 => // zero scale is no-op
- childExpr
- case t: DecimalType if t.scale < 0 => // Spark disallows negative
scale SPARK-30252
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setSortOrder(sortOrderBuilder)
+ .build())
+ } else {
None
- case _ =>
- scalarExprToProtoWithReturnType("ceil", e.dataType, childExpr)
- }
-
- case Cos(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("cos", childExpr)
+ }
- case Exp(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("exp", childExpr)
+ case And(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case e @ Floor(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- child.dataType match {
- case t: DecimalType if t.scale == 0 => // zero scale is no-op
- childExpr
- case t: DecimalType if t.scale < 0 => // Spark disallows negative
scale SPARK-30252
- None
- case _ =>
- scalarExprToProtoWithReturnType("floor", e.dataType, childExpr)
- }
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.And.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- case Log(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("ln", childExpr)
-
- case Log10(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("log10", childExpr)
-
- case Log2(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("log2", childExpr)
-
- case Pow(left, right) =>
- val leftExpr = exprToProtoInternal(left, inputs)
- val rightExpr = exprToProtoInternal(right, inputs)
- scalarExprToProto("pow", leftExpr, rightExpr)
-
- // round function for Spark 3.2 does not allow negative round target
scale. In addition,
- // it has different result precision/scale for decimals. Supporting only
3.3 and above.
- case r: Round if !isSpark32 =>
- // _scale s a constant, copied from Spark's RoundBase because it is a
protected val
- val scaleV: Any = r.scale.eval(EmptyRow)
- val _scale: Int = scaleV.asInstanceOf[Int]
-
- lazy val childExpr = exprToProtoInternal(r.child, inputs)
- r.child.dataType match {
- case t: DecimalType if t.scale < 0 => // Spark disallows negative
scale SPARK-30252
- None
- case _ if scaleV == null =>
- exprToProtoInternal(Literal(null), inputs)
- case _: ByteType | ShortType | IntegerType | LongType if _scale >= 0
=>
- childExpr // _scale(I.e. decimal place) >= 0 is a no-op for
integer types in Spark
- case _: FloatType | DoubleType =>
- // We cannot properly match with the Spark behavior for
floating-point numbers.
- // Spark uses BigDecimal for rounding float/double, and BigDecimal
fist converts a
- // double to string internally in order to create its own internal
representation.
- // The problem is BigDecimal uses java.lang.Double.toString() and
it has complicated
- // rounding algorithm. E.g. -5.81855622136895E8 is actually
- // -581855622.13689494132995605468750. Note the 5th fractional
digit is 4 instead of
- // 5. Java(Scala)'s toString() rounds it up to -581855622.136895.
This makes a
- // difference when rounding at 5th digit, I.e.
round(-5.81855622136895E8, 5) should be
- // -5.818556221369E8, instead of -5.8185562213689E8. There is also
an example that
- // toString() does NOT round up. 6.1317116247283497E18 is
6131711624728349696. It can
- // be rounded up to 6.13171162472835E18 that still represents the
same double number.
- // I.e. 6.13171162472835E18 == 6.1317116247283497E18. However,
toString() does not.
- // That results in round(6.1317116247283497E18, -5) ==
6.1317116247282995E18 instead
- // of 6.1317116247283999E18.
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setAnd(builder)
+ .build())
+ } else {
None
- case _ =>
- // `scale` must be Int64 type in DataFusion
- val scaleExpr = exprToProtoInternal(Literal(_scale.toLong,
LongType), inputs)
- scalarExprToProtoWithReturnType("round", r.dataType, childExpr,
scaleExpr)
- }
-
- case Signum(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("signum", childExpr)
-
- case Sin(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("sin", childExpr)
-
- case Sqrt(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("sqrt", childExpr)
-
- case Tan(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("tan", childExpr)
-
- case Ascii(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("ascii", childExpr)
-
- case BitLength(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("bit_length", childExpr)
-
- case If(predicate, trueValue, falseValue) =>
- val predicateExpr = exprToProtoInternal(predicate, inputs)
- val trueExpr = exprToProtoInternal(trueValue, inputs)
- val falseExpr = exprToProtoInternal(falseValue, inputs)
- if (predicateExpr.isDefined && trueExpr.isDefined &&
falseExpr.isDefined) {
- val builder = ExprOuterClass.IfExpr.newBuilder()
- builder.setIfExpr(predicateExpr.get)
- builder.setTrueExpr(trueExpr.get)
- builder.setFalseExpr(falseExpr.get)
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setIf(builder)
- .build())
- } else {
- None
- }
-
- case CaseWhen(branches, elseValue) =>
- val whenSeq = branches.map(elements =>
exprToProtoInternal(elements._1, inputs))
- val thenSeq = branches.map(elements =>
exprToProtoInternal(elements._2, inputs))
- assert(whenSeq.length == thenSeq.length)
- if (whenSeq.forall(_.isDefined) && thenSeq.forall(_.isDefined)) {
- val builder = ExprOuterClass.CaseWhen.newBuilder()
- builder.addAllWhen(whenSeq.map(_.get).asJava)
- builder.addAllThen(thenSeq.map(_.get).asJava)
- if (elseValue.isDefined) {
- val elseValueExpr = exprToProtoInternal(elseValue.get, inputs)
- if (elseValueExpr.isDefined) {
- builder.setElseExpr(elseValueExpr.get)
- } else {
- return None
- }
}
- Some(
- ExprOuterClass.Expr
- .newBuilder()
- .setCaseWhen(builder)
- .build())
- } else {
- None
- }
- case ConcatWs(children) =>
- val exprs = children.map(e => exprToProtoInternal(Cast(e, StringType),
inputs))
- scalarExprToProto("concat_ws", exprs: _*)
+ case Or(left, right) =>
+ val leftExpr = exprToProtoInternal(left, inputs)
+ val rightExpr = exprToProtoInternal(right, inputs)
- case Chr(child) =>
- val childExpr = exprToProtoInternal(child, inputs)
- scalarExprToProto("chr", childExpr)
+ if (leftExpr.isDefined && rightExpr.isDefined) {
+ val builder = ExprOuterClass.Or.newBuilder()
+ builder.setLeft(leftExpr.get)
+ builder.setRight(rightExpr.get)
- case InitCap(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("initcap", childExpr)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setOr(builder)
+ .build())
+ } else {
+ None
+ }
- case Length(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("length", childExpr)
+ case UnaryExpression(child) if expr.prettyName == "promote_precision"
=>
+ // `UnaryExpression` includes `PromotePrecision` for Spark 3.2 & 3.3
+ // `PromotePrecision` is just a wrapper, don't need to serialize it.
+ exprToProtoInternal(child, inputs)
- case Lower(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("lower", childExpr)
+ case CheckOverflow(child, dt, nullOnOverflow) =>
+ val childExpr = exprToProtoInternal(child, inputs)
- case Md5(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("md5", childExpr)
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.CheckOverflow.newBuilder()
+ builder.setChild(childExpr.get)
+ builder.setFailOnError(!nullOnOverflow)
- case OctetLength(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("octet_length", childExpr)
+ // `dataType` must be decimal type
+ val dataType = serializeDataType(dt)
+ builder.setDatatype(dataType.get)
- case Reverse(child) =>
- val childExpr = exprToProtoInternal(Cast(child, StringType), inputs)
- scalarExprToProto("reverse", childExpr)
+ Some(
+ ExprOuterClass.Expr
+ .newBuilder()
+ .setCheckOverflow(builder)
+ .build())
+ } else {
+ None
+ }
- case StringInstr(str, substr) =>
- val leftExpr = exprToProtoInternal(Cast(str, StringType), inputs)
- val rightExpr = exprToProtoInternal(Cast(substr, StringType), inputs)
- scalarExprToProto("strpos", leftExpr, rightExpr)
+ case attr: AttributeReference =>
+ val dataType = serializeDataType(attr.dataType)
- case StringRepeat(str, times) =>
- val leftExpr = exprToProtoInternal(Cast(str, StringType), inputs)
- val rightExpr = exprToProtoInternal(Cast(times, LongType), inputs)
- scalarExprToProto("repeat", leftExpr, rightExpr)
+ if (dataType.isDefined) {
Review Comment:
Hmm is the `binding` change accidentally removed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
