comphead commented on code in PR #6427: URL: https://github.com/apache/datafusion-comet/pull/6427#discussion_r4277235143
########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,183 @@ +/* + * 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.execution + +import java.io.ByteArrayOutputStream + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.comet.{CometFilterExec, SerializedPlan} +import org.apache.spark.sql.execution.{ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.execution.ui.SparkPlanGraph +import org.apache.spark.sql.types.{BinaryType, LongType} +import org.apache.spark.util.sketch.BloomFilter + +import org.apache.comet.CometSparkSessionExtensions.isSpark42Plus +import org.apache.comet.serde.OperatorOuterClass.Operator + +class CometFilterDisplaySuite extends CometTestBase { + private case class Input(override val output: Seq[Attribute]) extends LeafExecNode { + override protected def doExecute(): RDD[InternalRow] = + throw new AssertionError("Rendering must not execute the plan") + } + + private def filter(condition: Expression): CometFilterExec = { + val child = Input(condition.references.toSeq) + new CometFilterExec( + Operator.getDefaultInstance, + FilterExec(condition, child), + child.output, + condition, + child, + SerializedPlan(None)) { + // Plan-info capture should be testable without a SparkContext or native execution. + override lazy val metrics: Map[String, SQLMetric] = Map.empty + } + } + + private def render(plan: CometFilterExec): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + Seq( + plan.simpleString(100), + plan.verboseStringWithOperatorId(), + info.simpleString, + SparkPlanGraph(info).nodes.head.makeDotNode(Map.empty)) + } + + test("application-built Bloom filters have bounded explain and SQL UI displays") { + assume(!isSpark42Plus, "https://github.com/apache/datafusion-comet/issues/4968";) + val bloom = BloomFilter.create(10L, 8L * 1024 * 1024) + bloom.putLong(7L) + val buffer = new ByteArrayOutputStream() + bloom.writeTo(buffer) + val bytes = buffer.toByteArray + val summary = s"<bloom: ${bytes.length} bytes>" + + // This is an application-supplied literal. Spark's automatically injected runtime Bloom + // filters retain scalar subqueries; enabling that optimization does not create this case. + withSQLConf("spark.sql.optimizer.runtime.bloomFilter.enabled" -> "true") { + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = input.filter( + getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head))) + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(classOf[CometFilterExec])) + checkAnswer(query, Seq(Row(7L))) + val filters = plan.collect { case filter: CometFilterExec => filter } + assert(filters.nonEmpty) + filters.foreach { filter => + Seq( + filter.simpleString(100), + filter.verboseStringWithOperatorId(), + SparkPlanInfo.fromSparkPlan(filter).simpleString).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } + assert(filter.condition.find { Review Comment: This nested `find` with a `literal.value match` can be one `exists` with a pattern such as `case Literal(payload: Array[Byte], _) => payload.sameElements(bytes)`. The `assert(filters.nonEmpty)` at line 86 is also redundant, since `checkSparkAnswerAndOperator(query, Seq(classOf[CometFilterExec]))` already requires that operator in the plan. ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,183 @@ +/* + * 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.execution + +import java.io.ByteArrayOutputStream + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.comet.{CometFilterExec, SerializedPlan} +import org.apache.spark.sql.execution.{ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.execution.ui.SparkPlanGraph +import org.apache.spark.sql.types.{BinaryType, LongType} +import org.apache.spark.util.sketch.BloomFilter + +import org.apache.comet.CometSparkSessionExtensions.isSpark42Plus +import org.apache.comet.serde.OperatorOuterClass.Operator + +class CometFilterDisplaySuite extends CometTestBase { + private case class Input(override val output: Seq[Attribute]) extends LeafExecNode { + override protected def doExecute(): RDD[InternalRow] = + throw new AssertionError("Rendering must not execute the plan") + } + + private def filter(condition: Expression): CometFilterExec = { + val child = Input(condition.references.toSeq) + new CometFilterExec( + Operator.getDefaultInstance, + FilterExec(condition, child), + child.output, + condition, + child, + SerializedPlan(None)) { + // Plan-info capture should be testable without a SparkContext or native execution. + override lazy val metrics: Map[String, SQLMetric] = Map.empty + } + } + + private def render(plan: CometFilterExec): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + Seq( + plan.simpleString(100), + plan.verboseStringWithOperatorId(), + info.simpleString, + SparkPlanGraph(info).nodes.head.makeDotNode(Map.empty)) + } + + test("application-built Bloom filters have bounded explain and SQL UI displays") { + assume(!isSpark42Plus, "https://github.com/apache/datafusion-comet/issues/4968";) + val bloom = BloomFilter.create(10L, 8L * 1024 * 1024) + bloom.putLong(7L) + val buffer = new ByteArrayOutputStream() + bloom.writeTo(buffer) + val bytes = buffer.toByteArray + val summary = s"<bloom: ${bytes.length} bytes>" + + // This is an application-supplied literal. Spark's automatically injected runtime Bloom + // filters retain scalar subqueries; enabling that optimization does not create this case. + withSQLConf("spark.sql.optimizer.runtime.bloomFilter.enabled" -> "true") { Review Comment: `spark.sql.optimizer.runtime.bloomFilter.enabled` defaults to `true` in the Spark versions I checked (3.4.3, 3.5.9 and 4.1.3), and this is a single-table filter with no join, so the runtime filter injection cannot apply anyway. Could the `withSQLConf` wrapper and the comment explaining why it does not matter go? That would also remove a nesting level. ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,183 @@ +/* + * 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.execution + +import java.io.ByteArrayOutputStream + +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.{CometTestBase, Row} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.comet.{CometFilterExec, SerializedPlan} +import org.apache.spark.sql.execution.{ScalarSubquery => ExecScalarSubquery} +import org.apache.spark.sql.execution.metric.SQLMetric +import org.apache.spark.sql.execution.ui.SparkPlanGraph +import org.apache.spark.sql.types.{BinaryType, LongType} +import org.apache.spark.util.sketch.BloomFilter + +import org.apache.comet.CometSparkSessionExtensions.isSpark42Plus +import org.apache.comet.serde.OperatorOuterClass.Operator + +class CometFilterDisplaySuite extends CometTestBase { + private case class Input(override val output: Seq[Attribute]) extends LeafExecNode { + override protected def doExecute(): RDD[InternalRow] = + throw new AssertionError("Rendering must not execute the plan") + } + + private def filter(condition: Expression): CometFilterExec = { + val child = Input(condition.references.toSeq) + new CometFilterExec( + Operator.getDefaultInstance, + FilterExec(condition, child), + child.output, + condition, + child, + SerializedPlan(None)) { + // Plan-info capture should be testable without a SparkContext or native execution. + override lazy val metrics: Map[String, SQLMetric] = Map.empty + } + } + + private def render(plan: CometFilterExec): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + Seq( + plan.simpleString(100), + plan.verboseStringWithOperatorId(), + info.simpleString, + SparkPlanGraph(info).nodes.head.makeDotNode(Map.empty)) + } + + test("application-built Bloom filters have bounded explain and SQL UI displays") { + assume(!isSpark42Plus, "https://github.com/apache/datafusion-comet/issues/4968";) + val bloom = BloomFilter.create(10L, 8L * 1024 * 1024) + bloom.putLong(7L) + val buffer = new ByteArrayOutputStream() + bloom.writeTo(buffer) + val bytes = buffer.toByteArray + val summary = s"<bloom: ${bytes.length} bytes>" + + // This is an application-supplied literal. Spark's automatically injected runtime Bloom + // filters retain scalar subqueries; enabling that optimization does not create this case. + withSQLConf("spark.sql.optimizer.runtime.bloomFilter.enabled" -> "true") { + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = input.filter( + getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head))) + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(classOf[CometFilterExec])) + checkAnswer(query, Seq(Row(7L))) + val filters = plan.collect { case filter: CometFilterExec => filter } + assert(filters.nonEmpty) + filters.foreach { filter => + Seq( + filter.simpleString(100), + filter.verboseStringWithOperatorId(), + SparkPlanInfo.fromSparkPlan(filter).simpleString).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } + assert(filter.condition.find { + case predicate: BloomFilterMightContain => + predicate.bloomFilterExpression.find { + case literal: org.apache.spark.sql.catalyst.expressions.Literal => + literal.value match { + case payload: Array[Byte] => payload.sameElements(bytes) + case _ => false + } + case _ => false + }.nonEmpty + case _ => false + }.nonEmpty) + } + } + } + } + + test("UI and explain summarize large Bloom literals before formatting or evaluating them") { + val bytes = new Array[Byte](1024 * 1024) + val payload = new Literal(bytes, BinaryType) { + override def toString: String = + throw new AssertionError("Rendering must not hex-encode the Bloom payload") + override def eval(input: InternalRow): Any = + throw new AssertionError("Rendering must not evaluate the Bloom payload") + } + val key = AttributeReference("key", LongType)() + val condition = And(IsNotNull(key), BloomFilterMightContain(payload, key)) + val plan = filter(condition) + val nativeOp = plan.nativeOp + + render(plan).foreach { text => + assert(text.contains("<bloom: 1048576 bytes>")) + assert(text.contains("might_contain")) + assert(text.contains("key")) + assert(text.length < 1024) + } + assert(plan.condition eq condition) + assert(plan.nativeOp eq nativeOp) + assert(plan.originalPlan.asInstanceOf[FilterExec].condition eq condition) + assert(payload.value.asInstanceOf[Array[Byte]] eq bytes) + } + + test("nested Bloom rendering preserves predicate evaluation and ordinary binary literals") { + val bloom = BloomFilter.create(10L, 128L) + bloom.putLong(7L) + val buffer = new ByteArrayOutputStream() + bloom.writeTo(buffer) + val bytes = buffer.toByteArray + val contains = BloomFilterMightContain(Literal(bytes), Literal(7L)) + val binaryEquality = EqualTo(Literal(Array[Byte](1, 2)), Literal(Array[Byte](1, 2))) + val condition = And(contains, Or(Not(contains), binaryEquality)) + val plan = filter(condition) + assert(condition.eval() == true) Review Comment: The `eval()` assertions here, at line 154 and in the null case below, exercise `BloomFilterMightContain.eval`, which this PR does not touch. The `plan.condition eq condition` checks already show that the executable condition is the same object. The test name says it preserves predicate evaluation, which the `eval()` calls on Spark's own expression do not prove. Could they go and the test be renamed? The `eq` check is also repeated in the test above, so once is enough. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
