Skip to content
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

[VL] Make bloom_filter_agg fall back when might_contain is not transformable #3917

Merged
merged 5 commits into from
Dec 6, 2023
Merged

[VL] Make bloom_filter_agg fall back when might_contain is not transformable #3917

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
119062f
[VL] fallback bloom_filter_agg when might_contain is fallbacked
zhli1142015 Dec 4, 2023
f04d6d1
address comments
zhli1142015 Dec 5, 2023
a0fba76
update doc
zhli1142015 Dec 5, 2023
03f4570
address comment
zhli1142015 Dec 5, 2023
4e62718
minor fix
zhli1142015 Dec 5, 2023
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
7 changes: 7 additions & 0 deletions gluten-core/src/main/scala/io/glutenproject/extension/columnar/TransformHintRule.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -714,6 +714,13 @@ case class AddTransformHintRule() extends Rule[SparkPlan] {
s"${e.getMessage}, original sparkplan is " +
s"${plan.getClass}(${plan.children.toList.map(_.getClass)})")
}

if (TransformHints.isAlreadyTagged(plan) && TransformHints.isNotTransformable(plan)) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe, we can move this part into a dedicated rule after AddTransformHintRule (see link).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, moved to a separate rule.

// Velox BloomFilter's implementation is different from Spark's.
// So if might_contain falls back, we need fall back related bloom filter agg.
SparkShimLoader.getSparkShims.handleBloomFilterFallback(plan)(
p => TransformHints.tagNotTransformable(p, "related BloomFilterMightContain is fallbacked"))
}
}

implicit class EncodeTransformableTagImplicits(validationResult: ValidationResult) {
Expand Down
2 changes: 0 additions & 2 deletions gluten-ut/spark33/src/test/scala/io/glutenproject/utils/velox/VeloxTestSettings.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("string split function with positive limit")
.exclude("string split function with negative limit")
enableSuite[GlutenBloomFilterAggregateQuerySuite]
// fallback might_contain, the input argument binary is not same with vanilla spark
.exclude("Test NULL inputs for might_contain")
enableSuite[GlutenDataSourceV2DataFrameSessionCatalogSuite]
enableSuite[GlutenDataSourceV2DataFrameSuite]
enableSuite[GlutenDataSourceV2FunctionSuite]
Expand Down
2 changes: 0 additions & 2 deletions gluten-ut/spark34/src/test/scala/io/glutenproject/utils/velox/VeloxTestSettings.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@ class VeloxTestSettings extends BackendTestSettings {
.exclude("string split function with positive limit")
.exclude("string split function with negative limit")
enableSuite[GlutenBloomFilterAggregateQuerySuite]
// fallback might_contain, the input argument binary is not same with vanilla spark
.exclude("Test NULL inputs for might_contain")
enableSuite[GlutenDataSourceV2DataFrameSessionCatalogSuite]
enableSuite[GlutenDataSourceV2DataFrameSuite]
enableSuite[GlutenDataSourceV2FunctionSuite]
Expand Down
4 changes: 3 additions & 1 deletion shims/common/src/main/scala/io/glutenproject/sql/shims/SparkShims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
import org.apache.spark.sql.catalyst.plans.physical.Distribution
import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.{FileSourceScanExec, SparkPlan}
import org.apache.spark.sql.execution.datasources.{FilePartition, FileScanRDD, PartitionDirectory, PartitionedFile, PartitioningAwareFileIndex}
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.execution.datasources.v2.text.TextScan
Expand Down Expand Up @@ -81,4 +81,6 @@ trait SparkShims {
start: Long,
length: Long,
@transient locations: Array[String] = Array.empty): PartitionedFile

def handleBloomFilterFallback(plan: SparkPlan)(fun: SparkPlan => Unit): Unit
}
4 changes: 3 additions & 1 deletion shims/spark32/src/main/scala/io/glutenproject/sql/shims/spark32/Spark32Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.Expression
import org.apache.spark.sql.catalyst.plans.physical.{Distribution, HashClusteredDistribution}
import org.apache.spark.sql.connector.catalog.Table
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.{FileSourceScanExec, PartitionedFileUtil}
import org.apache.spark.sql.execution.{FileSourceScanExec, PartitionedFileUtil, SparkPlan}
import org.apache.spark.sql.execution.datasources.{BucketingUtils, FilePartition, FileScanRDD, PartitionDirectory, PartitionedFile, PartitioningAwareFileIndex}
import org.apache.spark.sql.execution.datasources.FileFormatWriter.Empty2Null
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
Expand Down Expand Up @@ -101,4 +101,6 @@ class Spark32Shims extends SparkShims {
length: Long,
@transient locations: Array[String] = Array.empty): PartitionedFile =
PartitionedFile(partitionValues, filePath, start, length, locations)

override def handleBloomFilterFallback(plan: SparkPlan)(fun: SparkPlan => Unit): Unit = {}
}
27 changes: 27 additions & 0 deletions shims/spark33/src/main/scala/io/glutenproject/sql/shims/spark33/Spark33Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,33 @@ class Spark33Shims extends SparkShims {
@transient locations: Array[String] = Array.empty): PartitionedFile =
PartitionedFile(partitionValues, filePath, start, length, locations)

override def handleBloomFilterFallback(plan: SparkPlan)(fun: SparkPlan => Unit): Unit = {
def tagNotTransformableRecursive(p: SparkPlan): Unit = {
p match {
case agg: org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec
if agg.aggregateExpressions.exists(
expr => expr.aggregateFunction.isInstanceOf[BloomFilterAggregate]) =>
fun(agg)
tagNotTransformableRecursive(agg.child)
case a: org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec =>
tagNotTransformableRecursive(a.executedPlan)
case _ =>
p.children.map(tagNotTransformableRecursive)
}
}

plan.transformAllExpressions {
case mc @ BloomFilterMightContain(sub: org.apache.spark.sql.execution.ScalarSubquery, _) =>
tagNotTransformableRecursive(sub.plan)
mc
case mc @ BloomFilterMightContain(
g @ GetStructField(sub: org.apache.spark.sql.execution.ScalarSubquery, _, _),
_) =>
tagNotTransformableRecursive(sub.plan)
mc
}
}

private def invalidBucketFile(path: String): Throwable = {
new SparkException(
errorClass = "INVALID_BUCKET_FILE",
Expand Down
27 changes: 27 additions & 0 deletions shims/spark34/src/main/scala/io/glutenproject/sql/shims/spark34/Spark34Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,33 @@ class Spark34Shims extends SparkShims {
@transient locations: Array[String] = Array.empty): PartitionedFile =
PartitionedFile(partitionValues, SparkPath.fromPathString(filePath), start, length, locations)

override def handleBloomFilterFallback(plan: SparkPlan)(fun: SparkPlan => Unit): Unit = {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method looks repeated. If so, maybe, we can just move this method into a common place. But just move the check for BloomFilterAggregate into shims, as I note it doesn't exist in spark3.2.

Spark 3.3/3.4:

def hasBloomFilterAggregate(agg): Boolean = {
  agg.aggregateExpressions.exists( expr.aggregateFunction.isInstanceOf[BloomFilterAggregate])
}

Spark 3.2:

def hasBloomFilterAggregate(agg): Boolean = {
   false
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated, thanks,

def tagNotTransformableRecursive(p: SparkPlan): Unit = {
p match {
case agg: org.apache.spark.sql.execution.aggregate.ObjectHashAggregateExec
if agg.aggregateExpressions.exists(
expr => expr.aggregateFunction.isInstanceOf[BloomFilterAggregate]) =>
fun(agg)
tagNotTransformableRecursive(agg.child)
case a: org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec =>
tagNotTransformableRecursive(a.executedPlan)
case _ =>
p.children.map(tagNotTransformableRecursive)
}
}

plan.transformAllExpressions {
case mc @ BloomFilterMightContain(sub: org.apache.spark.sql.execution.ScalarSubquery, _) =>
tagNotTransformableRecursive(sub.plan)
mc
case mc @ BloomFilterMightContain(
g @ GetStructField(sub: org.apache.spark.sql.execution.ScalarSubquery, _, _),
_) =>
tagNotTransformableRecursive(sub.plan)
mc
}
}

private def invalidBucketFile(path: String): Throwable = {
new SparkException(
errorClass = "INVALID_BUCKET_FILE",
Expand Down
Loading