Skip to content

[VL] Remove redundant string trimming before Velox casts #12965

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rui-mo wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[VL] Remove redundant string trimming before Velox casts #12965

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
43 changes: 0 additions & 43 deletions ...ends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxSparkPlanExecApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1286,49 +1286,6 @@ class VeloxSparkPlanExecApi extends SparkPlanExecApi with Logging {
VeloxGetStructFieldTransformer(substraitExprName, childTransformer, ordinal, original)
}

/**
* To align with spark in casting string type input to other types, add trim node for trimming
* space or whitespace. See spark's Cast.scala.
*/
override def genCastWithNewChild(c: Cast): Cast = {
// scalastyle:off nonascii
// Common whitespace to be trimmed, including: ' ', '\n', '\r', '\f', etc.
val trimWhitespaceStr = " \t\n\u000B\u000C\u000D\u001C\u001D\u001E\u001F"
// Space separator.
val trimSpaceSepStr = "\u1680\u2008\u2009\u200A\u205F\u3000" +
('\u2000' to '\u2006').toList.mkString
// Line separator.
val trimLineSepStr = "\u2028"
// Paragraph separator.
val trimParaSepStr = "\u2029"
// Needs to be trimmed for casting to float/double/decimal
val trimSpaceStr = ('\u0000' to '\u0020').toList.mkString
// ISOControl characters, refer java.lang.Character.isISOControl(int)
val isoControlStr = (('\u0000' to '\u001F') ++ ('\u007F' to '\u009F')).toList.mkString
// scalastyle:on nonascii
if (VeloxConfig.get.castFromVarcharAddTrimNode && c.child.dataType == StringType) {
val trimStr = c.dataType match {
case BinaryType | _: ArrayType | _: MapType | _: StructType | _: UserDefinedType[_] =>
None
case FloatType | DoubleType | _: DecimalType =>
Some(trimSpaceStr)
case _ =>
Some(
(trimWhitespaceStr + trimSpaceSepStr + trimLineSepStr
+ trimParaSepStr + isoControlStr).toSet.mkString
)
}
trimStr
.map {
trim =>
c.withNewChildren(Seq(StringTrim(c.child, Some(Literal(trim))))).asInstanceOf[Cast]
}
.getOrElse(c)
} else {
c
}
}

/** Define backend specfic expression mappings. */
override def extraExpressionMappings: Seq[Sig] = {
Seq(
Expand Down
11 changes: 0 additions & 11 deletions backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,6 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
ResizeRange(minSize, Int.MaxValue)
}

def castFromVarcharAddTrimNode: Boolean = getConf(CAST_FROM_VARCHAR_ADD_TRIM_NODE)

def enableVeloxFlushablePartialAggregation: Boolean =
getConf(VELOX_FLUSHABLE_PARTIAL_AGGREGATION_ENABLED)

Expand Down Expand Up @@ -696,15 +694,6 @@ object VeloxConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)

val CAST_FROM_VARCHAR_ADD_TRIM_NODE =
buildConf("spark.gluten.velox.castFromVarcharAddTrimNode")
.doc(
"If true, will add a trim node " +
"which has the same semantic as vanilla Spark to CAST-from-varchar." +
"Otherwise, do nothing.")
.booleanConf
.createWithDefault(false)

val DECIMAL_TO_FLOAT_HIGH_PRECISION_CAST_ENABLED =
buildConf("spark.gluten.velox.decimalToFloatHighPrecisionCastEnabled")
.doc(
Expand Down
1 change: 0 additions & 1 deletion docs/velox-configuration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,6 @@ nav_order: 16
| spark.gluten.sql.rewrite.castArrayToString | 🔄 Dynamic | true | When true, rewrite `cast(array as String)` to `concat('[', array_join(array, ', ', null), ']')` to allow offloading to Velox. |
| spark.gluten.velox.broadcast.build.targetBytesPerThread | ⚓ Static | 32MB | It is used to calculate the number of hash table build threads. Based on our testing across various thresholds (1MB to 128MB), we recommend a value of 32MB or 64MB, as these consistently provided the most significant performance gains. |
| spark.gluten.velox.broadcastBuild.mergeBatches | 🔄 Dynamic | false | If enabled, all columnar batches in a broadcast build relation will be serialized into a single buffer to reduce the number of addInput calls in HashBuild operator. This can significantly improve BHJ performance when the broadcast table has many small batches, but may increase driver-side peak memory and is not suitable for very large broadcasts. |
| spark.gluten.velox.castFromVarcharAddTrimNode | 🔄 Dynamic | false | If true, will add a trim node which has the same semantic as vanilla Spark to CAST-from-varchar.Otherwise, do nothing. |
| spark.gluten.velox.decimalToFloatHighPrecisionCastEnabled | 🔄 Dynamic | false | If true, enables high-precision casts from DECIMAL to REAL/DOUBLE in Velox, which match vanilla Spark for values that cannot be represented exactly by floating-point arithmetic. Disabled by default because it is slower than the default conversion; enable it if precision matters more than throughput. |
| spark.gluten.velox.s3MaxConcurrentUploadNum | ⚓ Static | 4 | The maximum number of in-flight S3 part uploads per file. |
| spark.gluten.velox.s3UploadPartAsync | ⚓ Static | false | If true, S3 multipart upload parts are uploaded asynchronously. |
Expand Down
2 changes: 1 addition & 1 deletion ep/build-velox/src/get-velox.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ RUN_SETUP_SCRIPT=ON
ENABLE_ENHANCED_FEATURES=OFF

# Developer use only for testing Velox PR.
UPSTREAM_VELOX_PR_ID=""
UPSTREAM_VELOX_PR_ID="18821"
Comment thread
rui-mo marked this conversation as resolved.

OS=`uname -s`

Expand Down
2 changes: 0 additions & 2 deletions gluten-substrait/src/main/scala/org/apache/gluten/backendsapi/SparkPlanExecApi.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -530,8 +530,6 @@ trait SparkPlanExecApi {
startDate: ExpressionTransformer,
original: DateDiff): ExpressionTransformer

Comment thread
rui-mo marked this conversation as resolved.
def genCastWithNewChild(c: Cast): Cast = c

def genHashExpressionTransformer(
substraitExprName: String,
exprs: Seq[ExpressionTransformer],
Expand Down
7 changes: 2 additions & 5 deletions gluten-substrait/src/main/scala/org/apache/gluten/expression/ExpressionConverter.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,13 +484,10 @@ object ExpressionConverter extends SQLConfHelper with Logging {
}
}
}
// Add trim node, as necessary.
val newCast =
BackendsApiManager.getSparkPlanExecApiInstance.genCastWithNewChild(c)
CastTransformer(
substraitExprName,
replaceWithExpressionTransformer0(newCast.child, attributeSeq, expressionsMap),
newCast)
replaceWithExpressionTransformer0(c.child, attributeSeq, expressionsMap),
c)
Comment thread
rui-mo marked this conversation as resolved.
case s: String2TrimExpression =>
val (srcStr, trimStr) = s match {
case StringTrim(srcStr, trimStr) => (srcStr, trimStr)
Expand Down
31 changes: 13 additions & 18 deletions gluten-ut/spark34/src/test/scala/org/apache/spark/sql/GlutenDataFrameSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.spark.sql

import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.execution.{ProjectExecTransformer, WholeStageTransformer}

import org.apache.spark.SparkException
Expand All @@ -29,7 +30,6 @@ import org.apache.spark.sql.test.SQLTestData.TestData2
import org.apache.spark.sql.types.StringType

import java.io.ByteArrayOutputStream
import java.nio.charset.StandardCharsets

import scala.util.Random

Expand Down Expand Up @@ -325,8 +325,13 @@ class GlutenDataFrameSuite extends DataFrameSuite with GlutenSQLTestsTrait {
}

testGluten("Allow leading/trailing whitespace in string before casting") {
def checkResult(df: DataFrame, expectedResult: Seq[Row]): Unit = {
checkAnswer(df, expectedResult)
def checkResult(sql: String): Unit = {
var expected: Seq[Row] = null
withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
expected = spark.sql(sql).collect()
}
val df = spark.sql(sql)
checkAnswer(df, expected)
assert(find(df.queryExecution.executedPlan)(_.isInstanceOf[ProjectExecTransformer]).isDefined)
}

Expand All @@ -335,31 +340,21 @@ class GlutenDataFrameSuite extends DataFrameSuite with GlutenSQLTestsTrait {
.toDF("col1")
.createOrReplaceTempView("t1")
// scalastyle:on nonascii
val expectedIntResult = Row(123) :: Row(123) ::
Row(123) :: Row(123) :: Row(123) :: Row(123) :: Row(123) :: Nil
var df = spark.sql("select cast(col1 as int) from t1")
checkResult(df, expectedIntResult)
df = spark.sql("select cast(col1 as long) from t1")
checkResult(df, expectedIntResult)
checkResult("select cast(col1 as int) from t1")
checkResult("select cast(col1 as long) from t1")

Seq(" 123.5", "123.5 ", " 123.5 ", "123.5\n\n\n", "123.5\r\r\r", "123.5\f\f\f", "123.5\u000C")
.toDF("col1")
.createOrReplaceTempView("t1")
val expectedFloatResult = Row(123.5) :: Row(123.5) ::
Row(123.5) :: Row(123.5) :: Row(123.5) :: Row(123.5) :: Row(123.5) :: Nil
df = spark.sql("select cast(col1 as float) from t1")
checkResult(df, expectedFloatResult)
df = spark.sql("select cast(col1 as double) from t1")
checkResult(df, expectedFloatResult)
checkResult("select cast(col1 as float) from t1")
checkResult("select cast(col1 as double) from t1")

// scalastyle:off nonascii
val rawData =
Seq(" abc", "abc ", " abc ", "\u2000abc\n\n\n", "abc\r\r\r", "abc\f\f\f", "abc\u000C")
// scalastyle:on nonascii
rawData.toDF("col1").createOrReplaceTempView("t1")
val expectedBinaryResult = rawData.map(d => Row(d.getBytes(StandardCharsets.UTF_8))).seq
df = spark.sql("select cast(col1 as binary) from t1")
checkResult(df, expectedBinaryResult)
checkResult("select cast(col1 as binary) from t1")
}

testGluten("SPARK-27439: Explain result should match collected result after view change") {
Expand Down
85 changes: 38 additions & 47 deletions gluten-ut/spark35/src/test/scala/org/apache/spark/sql/GlutenDataFrameSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.spark.sql

import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.execution.{ProjectExecTransformer, WholeStageTransformer}

import org.apache.spark.SparkException
Expand All @@ -29,7 +30,6 @@ import org.apache.spark.sql.test.SQLTestData.TestData2
import org.apache.spark.sql.types.StringType

import java.io.ByteArrayOutputStream
import java.nio.charset.StandardCharsets

import scala.util.Random

Expand Down Expand Up @@ -325,54 +325,45 @@ class GlutenDataFrameSuite extends DataFrameSuite with GlutenSQLTestsTrait {
}

testGluten("Allow leading/trailing whitespace in string before casting") {
withSQLConf(
"spark.gluten.velox.castFromVarcharAddTrimNode" -> "true",
"spark.gluten.bolt.castFromVarcharAddTrimNode" -> "true") {
def checkResult(df: DataFrame, expectedResult: Seq[Row]): Unit = {
checkAnswer(df, expectedResult)
assert(
find(df.queryExecution.executedPlan)(_.isInstanceOf[ProjectExecTransformer]).isDefined)
def checkResult(sql: String): Unit = {
var expected: Seq[Row] = null
withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
expected = spark.sql(sql).collect()
}

// scalastyle:off nonascii
Seq(
" 123",
"123 ",
" 123 ",
"\u2000123\n\n\n",
"123\r\r\r",
"123\f\f\f",
"123\u000C",
"123\u0000")
.toDF("col1")
.createOrReplaceTempView("t1")
// scalastyle:on nonascii
val expectedIntResult = Row(123) :: Row(123) ::
Row(123) :: Row(123) :: Row(123) :: Row(123) :: Row(123) :: Row(123) :: Nil
var df = spark.sql("select cast(col1 as int) from t1")
checkResult(df, expectedIntResult)
df = spark.sql("select cast(col1 as long) from t1")
checkResult(df, expectedIntResult)

Seq(" 123.5", "123.5 ", " 123.5 ", "123.5\n\n\n", "123.5\r\r\r", "123.5\f\f\f", "123.5\u000C")
.toDF("col1")
.createOrReplaceTempView("t1")
val expectedFloatResult = Row(123.5) :: Row(123.5) ::
Row(123.5) :: Row(123.5) :: Row(123.5) :: Row(123.5) :: Row(123.5) :: Nil
df = spark.sql("select cast(col1 as float) from t1")
checkResult(df, expectedFloatResult)
df = spark.sql("select cast(col1 as double) from t1")
checkResult(df, expectedFloatResult)

// scalastyle:off nonascii
val rawData =
Seq(" abc", "abc ", " abc ", "\u2000abc\n\n\n", "abc\r\r\r", "abc\f\f\f", "abc\u000C")
// scalastyle:on nonascii
rawData.toDF("col1").createOrReplaceTempView("t1")
val expectedBinaryResult = rawData.map(d => Row(d.getBytes(StandardCharsets.UTF_8))).seq
df = spark.sql("select cast(col1 as binary) from t1")
checkResult(df, expectedBinaryResult)
val df = spark.sql(sql)
checkAnswer(df, expected)
assert(
find(df.queryExecution.executedPlan)(_.isInstanceOf[ProjectExecTransformer]).isDefined)
Comment thread
rui-mo marked this conversation as resolved.
}

// scalastyle:off nonascii
Seq(
" 123",
"123 ",
" 123 ",
"\u2000123\n\n\n",
"123\r\r\r",
"123\f\f\f",
"123\u000C",
"123\u0000")
.toDF("col1")
.createOrReplaceTempView("t1")
// scalastyle:on nonascii
checkResult("select cast(col1 as int) from t1")
checkResult("select cast(col1 as long) from t1")

Seq(" 123.5", "123.5 ", " 123.5 ", "123.5\n\n\n", "123.5\r\r\r", "123.5\f\f\f", "123.5\u000C")
.toDF("col1")
.createOrReplaceTempView("t1")
checkResult("select cast(col1 as float) from t1")
checkResult("select cast(col1 as double) from t1")

// scalastyle:off nonascii
val rawData =
Seq(" abc", "abc ", " abc ", "\u2000abc\n\n\n", "abc\r\r\r", "abc\f\f\f", "abc\u000C")
// scalastyle:on nonascii
rawData.toDF("col1").createOrReplaceTempView("t1")
checkResult("select cast(col1 as binary) from t1")
}

testGluten("SPARK-27439: Explain result should match collected result after view change") {
Expand Down
Loading
Loading