c21 commented on code in PR #37083: URL: https://github.com/apache/spark/pull/37083#discussion_r918415377
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AdvancedStatsPlanVisitor.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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.plans.logical.statsEstimation + +import org.apache.spark.sql.catalyst.plans.logical._ + +/** + * A [[LogicalPlanVisitor]] that computes the statistics for the cost-based optimizer. + */ +object AdvancedStatsPlanVisitor extends LogicalPlanVisitor[Statistics] { + + /** Falls back to the estimation computed by [[BasicStatsPlanVisitor]]. */ + private def fallback(p: LogicalPlan): Statistics = BasicStatsPlanVisitor.visit(p) + + override def default(p: LogicalPlan): Statistics = fallback(p) + + override def visitAggregate(p: Aggregate): Statistics = { + AggregateEstimation.estimate(p).getOrElse(fallback(p)) + } + + override def visitDistinct(p: Distinct): Statistics = { + val child = p.child + visitAggregate(Aggregate(child.output, child.output, child)) + } + + override def visitExcept(p: Except): Statistics = fallback(p) + + override def visitExpand(p: Expand): Statistics = fallback(p) + + override def visitFilter(p: Filter): Statistics = { + FilterEstimation(p).estimate.getOrElse(fallback(p)) + } + + override def visitGenerate(p: Generate): Statistics = default(p) + + override def visitGlobalLimit(p: GlobalLimit): Statistics = fallback(p) + + override def visitOffset(p: Offset): Statistics = fallback(p) + + override def visitIntersect(p: Intersect): Statistics = fallback(p) + + override def visitJoin(p: Join): Statistics = fallback(p) Review Comment: Why we fallback here, but not use `JoinEstimation.estimate`? ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AdvancedStatsPlanVisitor.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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.plans.logical.statsEstimation + +import org.apache.spark.sql.catalyst.plans.logical._ + +/** + * A [[LogicalPlanVisitor]] that computes the statistics for the cost-based optimizer. + */ +object AdvancedStatsPlanVisitor extends LogicalPlanVisitor[Statistics] { + + /** Falls back to the estimation computed by [[BasicStatsPlanVisitor]]. */ + private def fallback(p: LogicalPlan): Statistics = BasicStatsPlanVisitor.visit(p) + + override def default(p: LogicalPlan): Statistics = fallback(p) + + override def visitAggregate(p: Aggregate): Statistics = { + AggregateEstimation.estimate(p).getOrElse(fallback(p)) + } + + override def visitDistinct(p: Distinct): Statistics = { + val child = p.child + visitAggregate(Aggregate(child.output, child.output, child)) + } + + override def visitExcept(p: Except): Statistics = fallback(p) + + override def visitExpand(p: Expand): Statistics = fallback(p) + + override def visitFilter(p: Filter): Statistics = { + FilterEstimation(p).estimate.getOrElse(fallback(p)) + } + + override def visitGenerate(p: Generate): Statistics = default(p) + + override def visitGlobalLimit(p: GlobalLimit): Statistics = fallback(p) + + override def visitOffset(p: Offset): Statistics = fallback(p) + + override def visitIntersect(p: Intersect): Statistics = fallback(p) + + override def visitJoin(p: Join): Statistics = fallback(p) + + override def visitLocalLimit(p: LocalLimit): Statistics = fallback(p) + + override def visitPivot(p: Pivot): Statistics = default(p) + + override def visitProject(p: Project): Statistics = { + ProjectEstimation.estimate(p).getOrElse(fallback(p)) + } + + override def visitRepartition(p: Repartition): Statistics = fallback(p) + + override def visitRepartitionByExpr(p: RepartitionByExpression): Statistics = fallback(p) + + override def visitRebalancePartitions(p: RebalancePartitions): Statistics = fallback(p) + + override def visitSample(p: Sample): Statistics = fallback(p) + + override def visitScriptTransform(p: ScriptTransformation): Statistics = default(p) + + override def visitUnion(p: Union): Statistics = { + UnionEstimation.estimate(p).getOrElse(fallback(p)) + } + + override def visitWindow(p: Window): Statistics = fallback(p) + + override def visitSort(p: Sort): Statistics = fallback(p) + + override def visitTail(p: Tail): Statistics = { Review Comment: nit: remove `{}` to be consistent as other places. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/BasicStatsPlanVisitor.scala: ########## @@ -17,16 +17,40 @@ package org.apache.spark.sql.catalyst.plans.logical.statsEstimation +import org.apache.spark.sql.catalyst.expressions.AttributeMap import org.apache.spark.sql.catalyst.plans.logical._ /** - * A [[LogicalPlanVisitor]] that computes the statistics for the cost-based optimizer. + * An [[LogicalPlanVisitor]] that computes a single dimension for plan stats: size in bytes. */ object BasicStatsPlanVisitor extends LogicalPlanVisitor[Statistics] { - /** Falls back to the estimation computed by [[SizeInBytesOnlyStatsPlanVisitor]]. */ - private def fallback(p: LogicalPlan): Statistics = SizeInBytesOnlyStatsPlanVisitor.visit(p) + /** + * A default, commonly used estimation for unary nodes. We assume the input row number is the + * same as the output row number, and compute sizes based on the column types. + */ + private def visitUnaryNode(p: UnaryNode): Statistics = { + // There should be some overhead in Row object, the size should not be zero when there is + // no columns, this help to prevent divide-by-zero error. + val childRowSize = EstimationUtils.getSizePerRow(p.child.output) + val outputRowSize = EstimationUtils.getSizePerRow(p.output) + // Assume there will be the same number of rows as child has. + var sizeInBytes = (p.child.stats.sizeInBytes * outputRowSize) / childRowSize + if (sizeInBytes == 0) { + // sizeInBytes can't be zero, or sizeInBytes of BinaryNode will also be zero + // (product of children). + sizeInBytes = 1 + } + + // v2 sources can bubble-up rowCount, so always propagate. + // Don't propagate attributeStats, since they are not estimated here. + Statistics(sizeInBytes = sizeInBytes, rowCount = p.child.stats.rowCount) Review Comment: I am confused here. In the top-level comment - `computes a single dimension for plan stats: size in bytes`. But why we populate `rowCount` as well here? ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AdvancedStatsPlanVisitor.scala: ########## @@ -0,0 +1,90 @@ +/* + * 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.plans.logical.statsEstimation + +import org.apache.spark.sql.catalyst.plans.logical._ + +/** + * A [[LogicalPlanVisitor]] that computes the statistics for the cost-based optimizer. Review Comment: Can you also mention here, what are the statistics we compute here? e.g. `sizeInBytes`, `rowCount` etc, similar to the top-level comment of `BasicStatsPlanVisitor`? ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/BasicStatsPlanVisitor.scala: ########## @@ -17,16 +17,40 @@ package org.apache.spark.sql.catalyst.plans.logical.statsEstimation +import org.apache.spark.sql.catalyst.expressions.AttributeMap import org.apache.spark.sql.catalyst.plans.logical._ /** - * A [[LogicalPlanVisitor]] that computes the statistics for the cost-based optimizer. + * An [[LogicalPlanVisitor]] that computes a single dimension for plan stats: size in bytes. Review Comment: nit: typo `An` -> `A`. -- 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: reviews-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
