Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,7 @@ class JsonFunctionsValidateSuite extends FunctionsValidateSuite {
}
}

// TODO: fix on spark-4.0
testWithMaxSparkVersion("json_array_length", "3.5") {
test("json_array_length") {
runQueryAndCompare(
s"select *, json_array_length(string_field1) " +
s"from datatab limit 5")(checkGlutenPlan[ProjectExecTransformer])
Expand Down Expand Up @@ -349,8 +348,7 @@ class JsonFunctionsValidateSuite extends FunctionsValidateSuite {
}
}

// TODO: fix on spark-4.0
testWithMaxSparkVersion("json_object_keys", "3.5") {
test("json_object_keys") {
withTempPath {
path =>
Seq[String](
Expand Down Expand Up @@ -380,8 +378,7 @@ class JsonFunctionsValidateSuite extends FunctionsValidateSuite {
}
}

// TODO: fix on spark-4.0
testWithMaxSparkVersion("to_json function", "3.5") {
test("to_json function") {
withTable("t") {
spark.sql(
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.spark.{SPARK_REVISION, SPARK_VERSION_SHORT}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{StringTrimBoth, _}
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.expressions.objects.{Invoke, StaticInvoke, StructsToJsonInvoke}
import org.apache.spark.sql.catalyst.optimizer.NormalizeNaNAndZero
import org.apache.spark.sql.execution.ScalarSubquery
import org.apache.spark.sql.hive.HiveUDFTransformer
Expand Down Expand Up @@ -144,89 +144,106 @@ object ExpressionConverter extends SQLConfHelper with Logging {
DecimalArithmeticExpressionTransformer(substraitName, leftChild, rightChild, resultType, b)
}

// Mapping for Iceberg static invoke functions
private val icebergStaticInvokeMap = Map(
"BucketFunction" -> ExpressionNames.BUCKET,
"TruncateFunction" -> ExpressionNames.TRUNCATE,
"YearsFunction" -> ExpressionNames.YEARS,
"MonthsFunction" -> ExpressionNames.MONTHS,
"DaysFunction" -> ExpressionNames.DAYS,
"HoursFunction" -> ExpressionNames.HOURS
)

// Mapping for other static invoke functions
private val staticInvokeMap = Map(
"varcharTypeWriteSideCheck" -> ExpressionNames.VARCHAR_TYPE_WRITE_SIDE_CHECK,
"charTypeWriteSideCheck" -> ExpressionNames.CHAR_TYPE_WRITE_SIDE_CHECK,
"readSidePadding" -> ExpressionNames.READ_SIDE_PADDING,
"lengthOfJsonArray" -> ExpressionNames.JSON_ARRAY_LENGTH,
"jsonObjectKeys" -> ExpressionNames.JSON_OBJECT_KEYS
)

private def replaceStaticInvokeWithExpressionTransformer(
i: StaticInvoke,
attributeSeq: Seq[Attribute],
expressionsMap: Map[Class[_], String]): ExpressionTransformer = {

val objName = i.objectName
val funcName = i.functionName

def doTransform(child: Expression): ExpressionTransformer =
replaceWithExpressionTransformer0(child, attributeSeq, expressionsMap)

def validateAndTransform(
exprName: String,
childTransformers: => Seq[ExpressionTransformer]): ExpressionTransformer = {
if (!BackendsApiManager.getValidatorApiInstance.doExprValidate(exprName, i)) {
throw new GlutenNotSupportException(
s"Not supported to map current ${i.getClass} call on function: ${i.functionName}.")
s"Not supported to map current ${i.getClass} call on function: $funcName.")
}
GenericExpressionTransformer(exprName, childTransformers, i)
}

i.functionName match {
case "encode" | "decode" if i.objectName.endsWith("UrlCodec") =>
validateAndTransform(
"url_" + i.functionName,
Seq(replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap))
)
// Try to match Iceberg static invoke first
val icebergTransformer: Option[ExpressionTransformer] =
if (funcName == "invoke") {
icebergStaticInvokeMap.collectFirst {
case (func, name) if objName.startsWith("org.apache.iceberg.spark.functions." + func) =>
GenericExpressionTransformer(name, i.arguments.map(doTransform), i)
}
} else {
None
}

case "isLuhnNumber" =>
validateAndTransform(
ExpressionNames.LUHN_CHECK,
Seq(replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap))
)
icebergTransformer.getOrElse {
funcName match {
case "isLuhnNumber" =>
validateAndTransform(ExpressionNames.LUHN_CHECK, Seq(doTransform(i.arguments.head)))

case "encode" | "decode" if i.objectName.endsWith("Base64") =>
if (!BackendsApiManager.getValidatorApiInstance.doExprValidate(ExpressionNames.BASE64, i)) {
throw new GlutenNotSupportException(
s"Not supported to map current ${i.getClass} call on function: ${i.functionName}.")
}
BackendsApiManager.getSparkPlanExecApiInstance.genBase64StaticInvokeTransformer(
ExpressionNames.BASE64,
replaceWithExpressionTransformer0(i.arguments.head, attributeSeq, expressionsMap),
i
)
case fn @ ("encode" | "decode") if objName.endsWith("UrlCodec") =>
validateAndTransform("url_" + fn, Seq(doTransform(i.arguments.head)))

case fn
if i.objectName.endsWith("CharVarcharCodegenUtils") && Set(
"varcharTypeWriteSideCheck",
"charTypeWriteSideCheck",
"readSidePadding").contains(fn) =>
val exprName = fn match {
case "varcharTypeWriteSideCheck" => ExpressionNames.VARCHAR_TYPE_WRITE_SIDE_CHECK
case "charTypeWriteSideCheck" => ExpressionNames.CHAR_TYPE_WRITE_SIDE_CHECK
case "readSidePadding" => ExpressionNames.READ_SIDE_PADDING
}
validateAndTransform(
exprName,
i.arguments.map(replaceWithExpressionTransformer0(_, attributeSeq, expressionsMap))
)
case fn @ ("encode" | "decode")
if objName.endsWith("Base64") && BackendsApiManager.getValidatorApiInstance
.doExprValidate(ExpressionNames.BASE64, i) =>
BackendsApiManager.getSparkPlanExecApiInstance.genBase64StaticInvokeTransformer(
ExpressionNames.BASE64,
doTransform(i.arguments.head),
i
)

case _ =>
throw new GlutenNotSupportException(
s"Not supported to transform StaticInvoke with object: ${i.staticObject.getName}, " +
s"function: ${i.functionName}")
case fn if staticInvokeMap.contains(fn) =>
validateAndTransform(staticInvokeMap(fn), i.arguments.map(doTransform))

case _ =>
throw new GlutenNotSupportException(
s"Not supported to transform StaticInvoke with object: $objName, function: $funcName")
}
}
}

private def replaceIcebergStaticInvoke(
s: StaticInvoke,
private def replaceInvokeWithExpressionTransformer(
invoke: Invoke,
attributeSeq: Seq[Attribute],
expressionsMap: Map[Class[_], String]): ExpressionTransformer = {
val invokeMap = Map(
"BucketFunction" -> ExpressionNames.BUCKET,
"TruncateFunction" -> ExpressionNames.TRUNCATE,
"YearsFunction" -> ExpressionNames.YEARS,
"MonthsFunction" -> ExpressionNames.MONTHS,
"DaysFunction" -> ExpressionNames.DAYS,
"HoursFunction" -> ExpressionNames.HOURS
)
val objName = s.staticObject.getName
val transformer = invokeMap.find {
case (func, _) => objName.startsWith("org.apache.iceberg.spark.functions." + func)
}
if (transformer.isEmpty) {
throw new GlutenNotSupportException(s"Not supported staticInvoke call object: $objName")

// Pattern matching for different Invoke types
invoke match {
// StructsToJson evaluator
case StructsToJsonInvoke(options, child, timeZoneId) =>
val toJsonExpr = StructsToJson(options, child, timeZoneId)
val substraitExprName = getAndCheckSubstraitName(toJsonExpr, expressionsMap)
BackendsApiManager.getSparkPlanExecApiInstance.genToJsonTransformer(
substraitExprName,
replaceWithExpressionTransformer0(child, attributeSeq, expressionsMap),
toJsonExpr
)

// Unsupported invoke
case _ =>
throw new GlutenNotSupportException(
s"Not supported to transform Invoke with function: $invoke")
}
GenericExpressionTransformer(
transformer.get._2,
s.arguments.map(replaceWithExpressionTransformer0(_, attributeSeq, expressionsMap)),
s)
}

private def replaceWithExpressionTransformer0(
Expand All @@ -237,32 +254,59 @@ object ExpressionConverter extends SQLConfHelper with Logging {
s"replaceWithExpressionTransformer expr: $expr class: ${expr.getClass} " +
s"name: ${expr.prettyName}")

expr match {
case p: PythonUDF =>
return replacePythonUDFWithExpressionTransformer(p, attributeSeq, expressionsMap)
case s: ScalaUDF =>
return replaceScalaUDFWithExpressionTransformer(s, attributeSeq, expressionsMap)
case _ if HiveUDFTransformer.isHiveUDF(expr) =>
return BackendsApiManager.getSparkPlanExecApiInstance.genHiveUDFTransformer(
expr,
attributeSeq)
case i: StaticInvoke
if i.functionName == "invoke" && i.staticObject.getName.startsWith(
"org.apache.iceberg.spark.functions.") =>
return replaceIcebergStaticInvoke(i, attributeSeq, expressionsMap)
case i: StaticInvoke =>
return replaceStaticInvokeWithExpressionTransformer(i, attributeSeq, expressionsMap)
case _ =>
tryTransformWithoutExpressionMapping(expr, attributeSeq, expressionsMap).getOrElse {
val substraitExprName: String = getAndCheckSubstraitName(expr, expressionsMap)
val backendConverted = BackendsApiManager.getSparkPlanExecApiInstance
.extraExpressionConverter(substraitExprName, expr, attributeSeq)

backendConverted.getOrElse(
transformExpression(expr, attributeSeq, expressionsMap, substraitExprName))
}
}

val substraitExprName: String = getAndCheckSubstraitName(expr, expressionsMap)
val backendConverted = BackendsApiManager.getSparkPlanExecApiInstance.extraExpressionConverter(
substraitExprName,
expr,
attributeSeq)
if (backendConverted.isDefined) {
return backendConverted.get
/**
* Transform expressions that don't have direct expression class mapping in expressionsMap. This
* handles special cases like UDFs (Python, Scala, Hive), StaticInvoke, and Invoke expressions,
* where the transformation logic is based on runtime information rather than expression class
* type.
*
* @param expr
* The expression to transform
* @param attributeSeq
* The sequence of attributes for binding
* @param expressionsMap
* The expression class to substrait name mapping (not used for these cases)
* @return
* Some(ExpressionTransformer) if the expression matches one of these special cases, None
* otherwise
*/
private def tryTransformWithoutExpressionMapping(
expr: Expression,
attributeSeq: Seq[Attribute],
expressionsMap: Map[Class[_], String]): Option[ExpressionTransformer] = {
Option {
expr match {
case pythonUDF: PythonUDF =>
replacePythonUDFWithExpressionTransformer(pythonUDF, attributeSeq, expressionsMap)
case scalaUDF: ScalaUDF =>
replaceScalaUDFWithExpressionTransformer(scalaUDF, attributeSeq, expressionsMap)
case _ if HiveUDFTransformer.isHiveUDF(expr) =>
BackendsApiManager.getSparkPlanExecApiInstance.genHiveUDFTransformer(expr, attributeSeq)
case staticInvoke: StaticInvoke =>
replaceStaticInvokeWithExpressionTransformer(staticInvoke, attributeSeq, expressionsMap)
case invoke: Invoke =>
replaceInvokeWithExpressionTransformer(invoke, attributeSeq, expressionsMap)
case _ =>
null
}
}
}

private def transformExpression(
expr: Expression,
attributeSeq: Seq[Attribute],
expressionsMap: Map[Class[_], String],
substraitExprName: String): ExpressionTransformer = {
expr match {
case c: CreateArray =>
val children =
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.catalyst.expressions.objects

import org.apache.spark.sql.catalyst.expressions.Expression

/**
* Extractors for Invoke expressions to ensure compatibility across different Spark versions.
*
* For Spark 3.2, StructsToJson is not replaced with Invoke expressions, so this extractor returns
* None to maintain API compatibility with other versions.
*/
object StructsToJsonInvoke {
def unapply(expr: Expression): Option[(Map[String, String], Expression, Option[String])] = {
None
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.catalyst.expressions.objects

import org.apache.spark.sql.catalyst.expressions.Expression

/**
* Extractors for Invoke expressions to ensure compatibility across different Spark versions.
*
* For Spark 3.3, StructsToJson is not replaced with Invoke expressions, so this extractor returns
* None to maintain API compatibility with other versions.
*/
object StructsToJsonInvoke {
def unapply(expr: Expression): Option[(Map[String, String], Expression, Option[String])] = {
None
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.catalyst.expressions.objects

import org.apache.spark.sql.catalyst.expressions.Expression

/**
* Extractors for Invoke expressions to ensure compatibility across different Spark versions.
*
* For Spark 3.4, StructsToJson is not replaced with Invoke expressions, so this extractor returns
* None to maintain API compatibility with other versions.
*/
object StructsToJsonInvoke {
def unapply(expr: Expression): Option[(Map[String, String], Expression, Option[String])] = {
None
}
}
Loading