viirya commented on code in PR #6427: URL: https://github.com/apache/datafusion-comet/pull/6427#discussion_r4333323282
########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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, CometProjectExec, 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: SparkPlan): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + val node = SparkPlanGraph(info).nodes.head + // Older Spark versions omit the description from DOT output. + assert(node.makeDotNode(Map.empty).length < 2048) + Seq(plan.simpleString(100), plan.verboseStringWithOperatorId(), info.simpleString, node.desc) + } + + for (project <- Seq(false, true)) { + val operator = if (project) "projections" else "filters" + test(s"application-built Bloom $operator 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>" + + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = { + val predicate = getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head)) + if (project) input.select(predicate.as("contains")) else input.filter(predicate) + } + val expectedOperator = + if (project) classOf[CometProjectExec] else classOf[CometFilterExec] + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(expectedOperator)) + checkAnswer(query, if (project) Seq(Row(true), Row(false)) else Seq(Row(7L))) + val operators = plan.collect { + case op: CometFilterExec if !project => (op, Seq(op.condition)) + case op: CometProjectExec if project => (op, op.projectList) + } + operators.foreach { case (op, expressions) => + render(op).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } + assert(expressions.exists(_.exists { + case Literal(payload: Array[Byte], _) => payload.sameElements(bytes) + case _ => false + })) Review Comment: The display helper builds a new tree and never touches `op.condition` or `op.projectList`, and Catalyst expressions are immutable. So this checks that the planner kept the literal rather than anything this PR changes. Could it go? ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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, CometProjectExec, 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: SparkPlan): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + val node = SparkPlanGraph(info).nodes.head + // Older Spark versions omit the description from DOT output. + assert(node.makeDotNode(Map.empty).length < 2048) + Seq(plan.simpleString(100), plan.verboseStringWithOperatorId(), info.simpleString, node.desc) + } + + for (project <- Seq(false, true)) { + val operator = if (project) "projections" else "filters" + test(s"application-built Bloom $operator 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>" + + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = { + val predicate = getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head)) + if (project) input.select(predicate.as("contains")) else input.filter(predicate) + } + val expectedOperator = + if (project) classOf[CometProjectExec] else classOf[CometFilterExec] + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(expectedOperator)) + checkAnswer(query, if (project) Seq(Row(true), Row(false)) else Seq(Row(7L))) + val operators = plan.collect { + case op: CometFilterExec if !project => (op, Seq(op.condition)) + case op: CometProjectExec if project => (op, op.projectList) + } + operators.foreach { case (op, expressions) => + render(op).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } Review Comment: This renders only the collected operator, but the claim is about the EXPLAIN and SQL UI text users actually see. Could these assert on the whole executed plan instead? For example, `query.queryExecution.explainString(ExplainMode.fromString("formatted"))` and `SparkPlanInfo.fromSparkPlan(plan)` with its graph nodes. That tests the claim directly, drops the `collect` and the `project` guards in the match, and would catch the payload leaking through another node, such as an unaliased output name or a scan's `PartitionFilters`. ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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, CometProjectExec, 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 { Review Comment: Since the suite now covers projections too and the helper is `CometExpressionDisplay`, could it be renamed `CometExpressionDisplaySuite`? The entries in `pr_build_linux.yml` and `pr_build_macos.yml` would follow. ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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, CometProjectExec, 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: SparkPlan): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + val node = SparkPlanGraph(info).nodes.head + // Older Spark versions omit the description from DOT output. + assert(node.makeDotNode(Map.empty).length < 2048) + Seq(plan.simpleString(100), plan.verboseStringWithOperatorId(), info.simpleString, node.desc) + } + + for (project <- Seq(false, true)) { + val operator = if (project) "projections" else "filters" + test(s"application-built Bloom $operator 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>" + + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = { + val predicate = getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head)) + if (project) input.select(predicate.as("contains")) else input.filter(predicate) + } + val expectedOperator = + if (project) classOf[CometProjectExec] else classOf[CometFilterExec] + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(expectedOperator)) + checkAnswer(query, if (project) Seq(Row(true), Row(false)) else Seq(Row(7L))) + val operators = plan.collect { + case op: CometFilterExec if !project => (op, Seq(op.condition)) + case op: CometProjectExec if project => (op, op.projectList) + } + operators.foreach { case (op, expressions) => + render(op).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } + assert(expressions.exists(_.exists { + case Literal(payload: Array[Byte], _) => payload.sameElements(bytes) + case _ => false + })) + } + } + } + } + + 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) Review Comment: These compare constructor vals of an immutable plan. `plan.condition`, `plan.nativeOp`, `originalPlan.condition` and `payload.value` are all fixed at construction, so none of them can fail. The throwing `toString`/`eval` overrides above already prove the payload is neither formatted nor evaluated. Could these four lines go? ########## spark/src/test/scala/org/apache/spark/sql/execution/CometFilterDisplaySuite.scala: ########## @@ -0,0 +1,171 @@ +/* + * 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, CometProjectExec, 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: SparkPlan): Seq[String] = { + val info = SparkPlanInfo.fromSparkPlan(plan) + val node = SparkPlanGraph(info).nodes.head + // Older Spark versions omit the description from DOT output. + assert(node.makeDotNode(Map.empty).length < 2048) + Seq(plan.simpleString(100), plan.verboseStringWithOperatorId(), info.simpleString, node.desc) + } + + for (project <- Seq(false, true)) { + val operator = if (project) "projections" else "filters" + test(s"application-built Bloom $operator 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>" + + withParquetDataFrame(Seq(Tuple1(7L), Tuple1(8L))) { input => + def query = { + val predicate = getColumnFromExpression( + BloomFilterMightContain(Literal(bytes), input.queryExecution.analyzed.output.head)) + if (project) input.select(predicate.as("contains")) else input.filter(predicate) + } + val expectedOperator = + if (project) classOf[CometProjectExec] else classOf[CometFilterExec] + val (_, plan) = checkSparkAnswerAndOperator(query, Seq(expectedOperator)) + checkAnswer(query, if (project) Seq(Row(true), Row(false)) else Seq(Row(7L))) + val operators = plan.collect { + case op: CometFilterExec if !project => (op, Seq(op.condition)) + case op: CometProjectExec if project => (op, op.projectList) + } + operators.foreach { case (op, expressions) => + render(op).foreach { text => + assert(text.contains(summary)) + assert(text.length < 2048) + } + assert(expressions.exists(_.exists { + case Literal(payload: Array[Byte], _) => payload.sameElements(bytes) + case _ => false + })) + } + } + } + } + + 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 summarizes Bloom payloads and preserves 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) + render(plan).foreach { text => + assert(text.contains(s"<bloom: ${bytes.length} bytes>")) + assert(text.contains("0x0102")) + } + } + + test("empty and null Bloom literals remain distinguishable") { + Seq(Literal(Array.emptyByteArray), Literal.create(null, BinaryType)).foreach { payload => + val condition = BloomFilterMightContain(payload, Literal(7L)) + val expected = if (payload.value == null) "null" else "<bloom: 0 bytes>" + render(filter(condition)).foreach(text => + assert(text.contains(s"might_contain($expected,"))) + } + } + + test("rendering leaves an unexecuted Bloom scalar subquery unresolved") { Review Comment: I'm not sure this test adds much. `summarizeBloomLiterals` only matches `Literal`, so a `ScalarSubquery` operand is never rewritten or evaluated by it. The throwing-literal test already covers "no evaluation during rendering", and the closing `intercept[IllegalArgumentException](scalar.eval())` exercises Spark's `ScalarSubquery.eval` rather than this change, like the `eval()` assertions removed earlier. Could it go, together with the `SubqueryExec` metrics override it needs? -- 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]
