| /* |
| * 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.flink.table.plan.util |
| |
| import org.apache.flink.table.calcite.FlinkRelBuilder.NamedWindowProperty |
| import org.apache.flink.table.plan.nodes.calcite.{Expand, LogicalRank, LogicalWindowAggregate, Rank} |
| import org.apache.flink.table.plan.nodes.logical.{FlinkLogicalRank, FlinkLogicalWindowAggregate} |
| import org.apache.flink.table.plan.nodes.physical.batch._ |
| import org.apache.flink.table.plan.nodes.physical.stream.StreamExecRank |
| import org.apache.flink.table.plan.util.FlinkRelOptUtil.{checkAndGetFullGroupSet, checkAndSplitAggCalls} |
| |
| import org.apache.calcite.avatica.util.TimeUnitRange._ |
| import org.apache.calcite.plan.RelOptUtil |
| import org.apache.calcite.rel.core._ |
| import org.apache.calcite.rel.metadata.{RelMdUtil, RelMetadataQuery} |
| import org.apache.calcite.rel.{RelNode, SingleRel} |
| import org.apache.calcite.rex._ |
| import org.apache.calcite.sql.SqlKind |
| import org.apache.calcite.sql.`type`.SqlTypeName._ |
| import org.apache.calcite.util.{ImmutableBitSet, NumberUtil} |
| import com.google.common.collect.ImmutableList |
| |
| import java.lang.Double |
| import java.math.BigDecimal |
| import java.util |
| |
| import scala.collection.JavaConversions._ |
| import scala.collection.mutable |
| |
| /** |
| * FlinkRelMdUtil provides utility methods used by the metadata provider methods. |
| */ |
| object FlinkRelMdUtil { |
| |
| /** |
| * Creates a RexNode that stores a selectivity value corresponding to the |
| * selectivity of a semi-join/anti-join. This can be added to a filter to simulate the |
| * effect of the semi-join/anti-join during costing, but should never appear in a real |
| * plan since it has no physical implementation. |
| * |
| * @param mq instance of metadata query |
| * @param rel the semiJoin or antiJoin of interest |
| * @return constructed rexNode |
| */ |
| def makeSemiJoinSelectivityRexNode( |
| mq: RelMetadataQuery, |
| rel: SemiJoin): RexNode = { |
| val joinInfo = rel.analyzeCondition() |
| val rexBuilder = rel.getCluster.getRexBuilder |
| makeSemiJoinSelectivityRexNode(mq, joinInfo, rel.getLeft, rel.getRight, rel.isAnti, rexBuilder) |
| } |
| |
| private def makeSemiJoinSelectivityRexNode( |
| mq: RelMetadataQuery, |
| joinInfo: JoinInfo, |
| left: RelNode, |
| right: RelNode, |
| isAnti: Boolean, |
| rexBuilder: RexBuilder): RexNode = { |
| |
| val equiSelectivity: Double = if (!joinInfo.leftKeys.isEmpty) { |
| RelMdUtil.computeSemiJoinSelectivity(mq, left, right, joinInfo.leftKeys, joinInfo.rightKeys) |
| } else { |
| 1D |
| } |
| |
| val nonEquiSelectivity = RelMdUtil.guessSelectivity(joinInfo.getRemaining(rexBuilder)) |
| val semiJoinSelectivity = equiSelectivity * nonEquiSelectivity |
| |
| val selectivity = if (isAnti) { |
| val antiJoinSelectivity = 1.0 - semiJoinSelectivity |
| if (antiJoinSelectivity == 0.0) { |
| // we don't expect that anti-join's selectivity is 0.0, so choose a default value 0.1 |
| 0.1 |
| } else { |
| antiJoinSelectivity |
| } |
| } else { |
| semiJoinSelectivity |
| } |
| |
| rexBuilder.makeCall( |
| RelMdUtil.ARTIFICIAL_SELECTIVITY_FUNC, |
| rexBuilder.makeApproxLiteral(new BigDecimal(selectivity))) |
| } |
| |
| /** |
| * Creates a RexNode that stores a selectivity value corresponding to the |
| * selectivity of a NamedProperties predicate. |
| * |
| * @param winAgg window aggregate node |
| * @param predicate a RexNode |
| * @return constructed rexNode including non-NamedProperties predicates and |
| * a predicate that stores NamedProperties predicate's selectivity |
| */ |
| def makeNamePropertiesSelectivityRexNode( |
| winAgg: LogicalWindowAggregate, |
| predicate: RexNode): RexNode = { |
| val fullGroupSet = checkAndGetFullGroupSet(winAgg) |
| makeNamePropertiesSelectivityRexNode(winAgg, fullGroupSet, winAgg.getNamedProperties, predicate) |
| } |
| |
| /** |
| * Creates a RexNode that stores a selectivity value corresponding to the |
| * selectivity of a NamedProperties predicate. |
| * |
| * @param winAgg window aggregate node |
| * @param predicate a RexNode |
| * @return constructed rexNode including non-NamedProperties predicates and |
| * a predicate that stores NamedProperties predicate's selectivity |
| */ |
| def makeNamePropertiesSelectivityRexNode( |
| winAgg: FlinkLogicalWindowAggregate, |
| predicate: RexNode): RexNode = { |
| val fullGroupSet = checkAndGetFullGroupSet(winAgg) |
| makeNamePropertiesSelectivityRexNode(winAgg, fullGroupSet, winAgg.getNamedProperties, predicate) |
| } |
| |
| /** |
| * Creates a RexNode that stores a selectivity value corresponding to the |
| * selectivity of a NamedProperties predicate. |
| * |
| * @param globalWinAgg global window aggregate node |
| * @param predicate a RexNode |
| * @return constructed rexNode including non-NamedProperties predicates and |
| * a predicate that stores NamedProperties predicate's selectivity |
| */ |
| def makeNamePropertiesSelectivityRexNode( |
| globalWinAgg: BatchExecWindowAggregateBase, |
| predicate: RexNode): RexNode = { |
| require(globalWinAgg.isFinal, "local window agg does not contain NamedProperties!") |
| val fullGrouping = globalWinAgg.getGrouping ++ globalWinAgg.getAuxGrouping |
| makeNamePropertiesSelectivityRexNode( |
| globalWinAgg, fullGrouping, globalWinAgg.getNamedProperties, predicate) |
| } |
| |
| /** |
| * Creates a RexNode that stores a selectivity value corresponding to the |
| * selectivity of a NamedProperties predicate. |
| * |
| * @param winAgg window aggregate node |
| * @param fullGrouping full groupSets |
| * @param namedProperties NamedWindowProperty list |
| * @param predicate a RexNode |
| * @return constructed rexNode including non-NamedProperties predicates and |
| * a predicate that stores NamedProperties predicate's selectivity |
| */ |
| def makeNamePropertiesSelectivityRexNode( |
| winAgg: SingleRel, |
| fullGrouping: Array[Int], |
| namedProperties: Seq[NamedWindowProperty], |
| predicate: RexNode): RexNode = { |
| if (predicate == null || predicate.isAlwaysTrue || namedProperties.isEmpty) { |
| return predicate |
| } |
| val rexBuilder = winAgg.getCluster.getRexBuilder |
| val namePropertiesStartIdx = winAgg.getRowType.getFieldCount - namedProperties.size |
| // split non-nameProperties predicates and nameProperties predicates |
| val pushable = new util.ArrayList[RexNode] |
| val notPushable = new util.ArrayList[RexNode] |
| RelOptUtil.splitFilters( |
| ImmutableBitSet.range(0, namePropertiesStartIdx), |
| predicate, |
| pushable, |
| notPushable) |
| if (notPushable.nonEmpty) { |
| val pred = RexUtil.composeConjunction(rexBuilder, notPushable, true) |
| val selectivity = RelMdUtil.guessSelectivity(pred) |
| val fun = rexBuilder.makeCall( |
| RelMdUtil.ARTIFICIAL_SELECTIVITY_FUNC, |
| rexBuilder.makeApproxLiteral(new BigDecimal(selectivity))) |
| pushable.add(fun) |
| } |
| RexUtil.composeConjunction(rexBuilder, pushable, true) |
| } |
| |
| /** |
| * Computes the cardinality of a particular expression from the projection |
| * list. |
| * |
| * @param mq metadata query instance |
| * @param calc calc RelNode |
| * @param expr projection expression |
| * @return cardinality |
| */ |
| def cardOfCalcExpr(mq: RelMetadataQuery, calc: Calc, expr: RexNode): Double = { |
| expr.accept(new CardOfCalcExpr(mq, calc)) |
| } |
| |
| /** |
| * Visitor that walks over a scalar expression and computes the |
| * cardinality of its result. |
| * The code is borrowed from RelMdUtil |
| * |
| * @param mq metadata query instance |
| * @param calc calc relnode |
| */ |
| private class CardOfCalcExpr( |
| mq: RelMetadataQuery, |
| calc: Calc) |
| extends RexVisitorImpl[Double](true) { |
| private val program = calc.getProgram |
| |
| private val condition = if (program.getCondition != null) { |
| program.expandLocalRef(program.getCondition) |
| } else { |
| null |
| } |
| |
| override def visitInputRef(inputRef: RexInputRef): Double = { |
| val col = ImmutableBitSet.of(inputRef.getIndex) |
| val distinctRowCount = mq.getDistinctRowCount(calc.getInput, col, condition) |
| if (distinctRowCount == null) { |
| null |
| } else { |
| RelMdUtil.numDistinctVals(distinctRowCount, mq.getAverageRowSize(calc)) |
| } |
| } |
| |
| override def visitLiteral(literal: RexLiteral): Double = { |
| RelMdUtil.numDistinctVals(1D, mq.getAverageRowSize(calc)) |
| } |
| |
| override def visitCall(call: RexCall): Double = { |
| val rowCount = mq.getRowCount(calc) |
| val distinctRowCount: Double = if (call.isA(SqlKind.MINUS_PREFIX)) { |
| cardOfCalcExpr(mq, calc, call.getOperands.get(0)) |
| } else if (call.isA(ImmutableList.of(SqlKind.PLUS, SqlKind.MINUS))) { |
| val card0 = cardOfCalcExpr(mq, calc, call.getOperands.get(0)) |
| if (card0 == null) { |
| null |
| } else { |
| val card1 = cardOfCalcExpr(mq, calc, call.getOperands.get(1)) |
| if (card1 == null) { |
| null |
| } else { |
| Math.max(card0, card1) |
| } |
| } |
| } else if (call.isA(ImmutableList.of(SqlKind.TIMES, SqlKind.DIVIDE))) { |
| NumberUtil.multiply( |
| cardOfCalcExpr(mq, calc, call.getOperands.get(0)), |
| cardOfCalcExpr(mq, calc, call.getOperands.get(1))) |
| } else if (call.isA(SqlKind.EXTRACT)) { |
| val extractUnit = call.getOperands.get(0) |
| val timeOperand = call.getOperands.get(1) |
| extractUnit match { |
| // go https://www.postgresql.org/docs/9.1/static/ |
| // functions-datetime.html#FUNCTIONS-DATETIME-EXTRACT to get the definitions of timeunits |
| case unit: RexLiteral => |
| val unitValue = unit.getValue |
| val timeOperandType = timeOperand.getType.getSqlTypeName |
| // assume min time is 1970-01-01 00:00:00, max time is 2100-12-31 21:59:59 |
| unitValue match { |
| case YEAR => 130D // [1970, 2100] |
| case MONTH => 12D |
| case DAY => 31D |
| case HOUR => 24D |
| case MINUTE => 60D |
| case SECOND => timeOperandType match { |
| case TIMESTAMP | TIME => 60 * 1000D // [0.000, 59.999] |
| case _ => 60D // [0, 59] |
| } |
| case QUARTER => 4D |
| case WEEK => 53D // [1, 53] |
| case MILLISECOND => timeOperandType match { |
| case TIMESTAMP | TIME => 60 * 1000D // [0.000, 59.999] |
| case _ => 60D // [0, 59] |
| } |
| case MICROSECOND => timeOperandType match { |
| case TIMESTAMP | TIME => 60 * 1000D * 1000D // [0.000, 59.999] |
| case _ => 60D // [0, 59] |
| } |
| case DOW => 7D // [0, 6] |
| case DOY => 366D // [1, 366] |
| case EPOCH => timeOperandType match { |
| // the number of seconds since 1970-01-01 00:00:00 UTC |
| case TIMESTAMP | TIME => 130 * 24 * 60 * 60 * 1000D |
| case _ => 130 * 24 * 60 * 60D |
| } |
| case DECADE => 13D // The year field divided by 10 |
| case CENTURY => 2D |
| case MILLENNIUM => 2D |
| case _ => cardOfCalcExpr(mq, calc, timeOperand) |
| } |
| case _ => cardOfCalcExpr(mq, calc, timeOperand) |
| } |
| } else if (call.getOperands.size() == 1) { |
| cardOfCalcExpr(mq, calc, call.getOperands.get(0)) |
| } else { |
| if (rowCount != null) rowCount / 10 else null |
| } |
| if (distinctRowCount == null) { |
| null |
| } else { |
| RelMdUtil.numDistinctVals(distinctRowCount, rowCount) |
| } |
| } |
| |
| } |
| |
| /** |
| * Takes a bitmap representing a set of input references and extracts the |
| * ones that reference the group by columns in an aggregate. |
| * |
| * |
| * @param groupKey the original bitmap |
| * @param aggRel the aggregate |
| */ |
| def setAggChildKeys( |
| groupKey: ImmutableBitSet, |
| aggRel: Aggregate): (ImmutableBitSet, Array[AggregateCall]) = { |
| val childKeyBuilder = ImmutableBitSet.builder |
| val aggCalls = new mutable.ArrayBuffer[AggregateCall]() |
| val groupSet = aggRel.getGroupSet.toArray |
| val (auxGroupSet, otherAggCalls) = checkAndSplitAggCalls(aggRel) |
| val fullGroupSet = groupSet ++ auxGroupSet |
| // does not need to take keys in aggregate call into consideration if groupKey contains all |
| // groupSet element in aggregate |
| val containsAllAggGroupKeys = fullGroupSet.indices.forall(groupKey.get) |
| groupKey.foreach( |
| bit => |
| if (bit < fullGroupSet.length) { |
| childKeyBuilder.set(fullGroupSet(bit)) |
| } else if (!containsAllAggGroupKeys) { |
| // getIndicatorCount return 0 if auxGroupSet is not empty |
| val agg = otherAggCalls.get(bit - (fullGroupSet.length + aggRel.getIndicatorCount)) |
| aggCalls += agg |
| } |
| ) |
| (childKeyBuilder.build(), aggCalls.toArray) |
| } |
| |
| /** |
| * Takes a bitmap representing a set of input references and extracts the |
| * ones that reference the group by columns in an aggregate. |
| * |
| * @param groupKey the original bitmap |
| * @param aggRel the aggregate |
| */ |
| def setAggChildKeys( |
| groupKey: ImmutableBitSet, |
| aggRel: BatchExecGroupAggregateBase): (ImmutableBitSet, Array[AggregateCall]) = { |
| require(!aggRel.isFinal || !aggRel.isMerge, "Cannot handle global agg which has local agg!") |
| setChildKeysOfAgg(groupKey, aggRel) |
| } |
| |
| /** |
| * Takes a bitmap representing a set of input references and extracts the |
| * ones that reference the group by columns in an aggregate. |
| * |
| * @param groupKey the original bitmap |
| * @param aggRel the aggregate |
| */ |
| def setAggChildKeys( |
| groupKey: ImmutableBitSet, |
| aggRel: BatchExecWindowAggregateBase): (ImmutableBitSet, Array[AggregateCall]) = { |
| require(!aggRel.isFinal || !aggRel.isMerge, "Cannot handle global agg which has local agg!") |
| setChildKeysOfAgg(groupKey, aggRel) |
| } |
| |
| private def setChildKeysOfAgg( |
| groupKey: ImmutableBitSet, |
| aggRel: SingleRel): (ImmutableBitSet, Array[AggregateCall]) = { |
| val (aggCalls, fullGroupSet) = aggRel match { |
| case agg: BatchExecLocalSortWindowAggregate => |
| // grouping + assignTs + auxGrouping |
| (agg.getAggCallList, |
| agg.getGrouping ++ Array(agg.inputTimestampIndex) ++ agg.getAuxGrouping) |
| case agg: BatchExecLocalHashWindowAggregate => |
| // grouping + assignTs + auxGrouping |
| (agg.getAggCallList, |
| agg.getGrouping ++ Array(agg.inputTimestampIndex) ++ agg.getAuxGrouping) |
| case agg: BatchExecWindowAggregateBase => |
| (agg.getAggCallList, agg.getGrouping ++ agg.getAuxGrouping) |
| case agg: BatchExecGroupAggregateBase => |
| (agg.getAggCallList, agg.getGrouping ++ agg.getAuxGrouping) |
| case _ => throw new IllegalArgumentException(s"Unknown relnode type ${aggRel.getRelTypeName}") |
| } |
| // does not need to take keys in aggregate call into consideration if groupKey contains all |
| // groupSet element in aggregate |
| val containsAllAggGroupKeys = fullGroupSet.indices.forall(groupKey.get) |
| val childKeyBuilder = ImmutableBitSet.builder |
| val aggs = new mutable.ArrayBuffer[AggregateCall]() |
| groupKey.foreach( |
| bit => |
| if (bit < fullGroupSet.length) { |
| childKeyBuilder.set(fullGroupSet(bit)) |
| } else if (!containsAllAggGroupKeys) { |
| val agg = aggCalls.get(bit - fullGroupSet.length) |
| aggs += agg |
| } |
| ) |
| (childKeyBuilder.build(), aggs.toArray) |
| } |
| |
| /** |
| * Takes a bitmap representing a set of local window aggregate references. |
| * |
| * global win-agg output type: groupSet + auxGroupSet + aggCall + namedProperties |
| * local win-agg output type: groupSet + assignTs + auxGroupSet + aggCalls |
| * |
| * Skips `assignTs` when mapping `groupKey` to `childKey`. |
| * |
| * @param groupKey the original bitmap |
| * @param globalWinAgg the global window aggregate |
| */ |
| def setChildKeysOfWinAgg( |
| groupKey: ImmutableBitSet, |
| globalWinAgg: BatchExecWindowAggregateBase): ImmutableBitSet = { |
| require(globalWinAgg.isMerge, "Cannot handle global agg which does not have local agg!") |
| val childKeyBuilder = ImmutableBitSet.builder |
| groupKey.toArray.foreach { key => |
| if (key < globalWinAgg.getGrouping.length) { |
| childKeyBuilder.set(key) |
| } else { |
| // skips `assignTs` |
| childKeyBuilder.set(key + 1) |
| } |
| } |
| childKeyBuilder.build() |
| } |
| |
| /** |
| * Split groupKeys on Agregate/ BatchExecGroupAggregateBase/ BatchExecWindowAggregateBase |
| * into keys on aggregate's groupKey and aggregate's aggregateCalls. |
| * |
| * @param agg the aggregate |
| * @param groupKey the original bitmap |
| */ |
| private[flink] def splitGroupKeysOnAggregate( |
| agg: SingleRel, |
| groupKey: ImmutableBitSet): (ImmutableBitSet, Array[AggregateCall]) = { |
| |
| def removeAuxKey( |
| groupKey: ImmutableBitSet, |
| groupSet: Array[Int], |
| auxGroupSet: Array[Int]): ImmutableBitSet = { |
| if (groupKey.contains(ImmutableBitSet.of(groupSet: _*))) { |
| // remove auxGroupSet from groupKey if groupKey contain both full-groupSet |
| // and (partial-)auxGroupSet |
| groupKey.except(ImmutableBitSet.of(auxGroupSet: _*)) |
| } else { |
| groupKey |
| } |
| } |
| |
| agg match { |
| case rel: Aggregate => |
| val (auxGroupSet, _) = FlinkRelOptUtil.checkAndSplitAggCalls(rel) |
| val (childKeys, aggCalls) = setAggChildKeys(groupKey, rel) |
| val childKeyExcludeAuxKey = removeAuxKey(childKeys, rel.getGroupSet.toArray, auxGroupSet) |
| (childKeyExcludeAuxKey, aggCalls) |
| case rel: BatchExecGroupAggregateBase => |
| // set the bits as they correspond to the child input |
| val (childKeys, aggCalls) = setAggChildKeys(groupKey, rel) |
| val childKeyExcludeAuxKey = removeAuxKey(childKeys, rel.getGrouping, rel.getAuxGrouping) |
| (childKeyExcludeAuxKey, aggCalls) |
| case rel: BatchExecWindowAggregateBase => |
| val (childKeys, aggCalls) = setAggChildKeys(groupKey, rel) |
| val childKeyExcludeAuxKey = removeAuxKey(childKeys, rel.getGrouping, rel.getAuxGrouping) |
| (childKeyExcludeAuxKey, aggCalls) |
| case _ => throw new IllegalArgumentException( |
| s"Unknown aggregate type: ${agg.getRelTypeName}.") |
| } |
| } |
| |
| /** |
| * Shifts every [[RexInputRef]] in an expression higher than length of full grouping |
| * (for skips `assignTs`). |
| * |
| * global win-agg output type: groupSet + auxGroupSet + aggCall + namedProperties |
| * local win-agg output type: groupSet + assignTs + auxGroupSet + aggCalls |
| * |
| * @param predicate a RexNode |
| * @param globalWinAgg the global window aggregate |
| */ |
| def setChildPredicateOfWinAgg( |
| predicate: RexNode, |
| globalWinAgg: BatchExecWindowAggregateBase): RexNode = { |
| require(globalWinAgg.isMerge, "Cannot handle global agg which does not have local agg!") |
| if (predicate == null) { |
| return null |
| } |
| // grouping + assignTs + auxGrouping |
| val fullGrouping = globalWinAgg.getGrouping ++ globalWinAgg.getAuxGrouping |
| // skips `assignTs` |
| RexUtil.shift(predicate, fullGrouping.length, 1) |
| } |
| |
| /** |
| * Split a predicate on Aggregate into two parts, the first one is pushable part, |
| * the second one is rest part. |
| * |
| * @param agg Aggregate which to analyze |
| * @param predicate Predicate which to analyze |
| * @return a tuple, first element is pushable part, second element is rest part. |
| * Note, pushable condition will be converted based on the input field position. |
| */ |
| def splitPredicateOnAggregate( |
| agg: Aggregate, |
| predicate: RexNode): (Option[RexNode], Option[RexNode]) = { |
| val fullGroupSet = checkAndGetFullGroupSet(agg) |
| splitPredicateOnAgg(fullGroupSet, agg, predicate) |
| } |
| |
| /** |
| * Split a predicate on BatchExecGroupAggregateBase into two parts, |
| * the first one is pushable part, the second one is rest part. |
| * |
| * @param agg Aggregate which to analyze |
| * @param predicate Predicate which to analyze |
| * @return a tuple, first element is pushable part, second element is rest part. |
| * Note, pushable condition will be converted based on the input field position. |
| */ |
| def splitPredicateOnAggregate( |
| agg: BatchExecGroupAggregateBase, |
| predicate: RexNode): (Option[RexNode], Option[RexNode]) = { |
| splitPredicateOnAgg(agg.getGrouping ++ agg.getAuxGrouping, agg, predicate) |
| } |
| |
| /** |
| * Split a predicate on WindowAggregateBatchExecBase into two parts, |
| * the first one is pushable part, the second one is rest part. |
| * |
| * @param agg Aggregate which to analyze |
| * @param predicate Predicate which to analyze |
| * @return a tuple, first element is pushable part, second element is rest part. |
| * Note, pushable condition will be converted based on the input field position. |
| */ |
| def splitPredicateOnAggregate( |
| agg: BatchExecWindowAggregateBase, |
| predicate: RexNode): (Option[RexNode], Option[RexNode]) = { |
| splitPredicateOnAgg(agg.getGrouping ++ agg.getAuxGrouping, agg, predicate) |
| } |
| |
| private def splitPredicateOnAgg( |
| grouping: Array[Int], |
| agg: SingleRel, |
| predicate: RexNode): (Option[RexNode], Option[RexNode]) = { |
| val notPushable = new util.ArrayList[RexNode] |
| val pushable = new util.ArrayList[RexNode] |
| val numOfGroupKey = grouping.length |
| RelOptUtil.splitFilters( |
| ImmutableBitSet.range(0, numOfGroupKey), |
| predicate, |
| pushable, |
| notPushable) |
| val rexBuilder = agg.getCluster.getRexBuilder |
| val childPred = if (pushable.isEmpty) { |
| None |
| } else { |
| // Converts a list of expressions that are based on the output fields of a |
| // Aggregate to equivalent expressions on the Aggregate's input fields. |
| val aggOutputFields = agg.getRowType.getFieldList |
| val aggInputFields = agg.getInput.getRowType.getFieldList |
| val adjustments = new Array[Int](aggOutputFields.size) |
| grouping.zipWithIndex foreach { |
| case (bit, index) => adjustments(index) = bit - index |
| } |
| val pushableConditions = pushable map { |
| pushCondition => |
| pushCondition.accept( |
| new RelOptUtil.RexInputConverter( |
| rexBuilder, |
| aggOutputFields, |
| aggInputFields, |
| adjustments)) |
| } |
| Option(RexUtil.composeConjunction(rexBuilder, pushableConditions, true)) |
| } |
| val restPred = if (notPushable.isEmpty) { |
| None |
| } else { |
| Option(RexUtil.composeConjunction(rexBuilder, notPushable, true)) |
| } |
| (childPred, restPred) |
| } |
| |
| def getRankFunColumnIndex(rank: Rank): Int = { |
| rank match { |
| case r: LogicalRank => getRankFunColumnIndex(rank, outputRankFunColumn = true) |
| case r: FlinkLogicalRank => getRankFunColumnIndex(rank, r.outputRankFunColumn) |
| case r: BatchExecRank => getRankFunColumnIndex(rank, r.outputRankFunColumn) |
| case r: StreamExecRank => getRankFunColumnIndex(rank, r.outputRankFunColumn) |
| } |
| } |
| |
| private def getRankFunColumnIndex(rank: Rank, outputRankFunColumn: Boolean): Int = { |
| if (outputRankFunColumn) { |
| require(rank.getRowType.getFieldCount == rank.getInput.getRowType.getFieldCount + 1) |
| rank.getRowType.getFieldCount - 1 |
| } else { |
| require(rank.getRowType.getFieldCount == rank.getInput.getRowType.getFieldCount) |
| -1 |
| } |
| } |
| |
| def splitPredicateOnRank( |
| rank: Rank, |
| predicate: RexNode): (Option[RexNode], Option[RexNode]) = { |
| val rankFunColumnIndex = getRankFunColumnIndex(rank) |
| if (predicate == null || predicate.isAlwaysTrue || rankFunColumnIndex < 0) { |
| return (Some(predicate), None) |
| } |
| |
| val rankNodes = new util.ArrayList[RexNode] |
| val nonRankNodes = new util.ArrayList[RexNode] |
| RelOptUtil.splitFilters( |
| ImmutableBitSet.range(0, rankFunColumnIndex), |
| predicate, |
| nonRankNodes, |
| rankNodes) |
| val rexBuilder = rank.getCluster.getRexBuilder |
| val nonRankPred = if (nonRankNodes.isEmpty) { |
| None |
| } else { |
| Option(RexUtil.composeConjunction(rexBuilder, nonRankNodes, true)) |
| } |
| val rankPred = if (rankNodes.isEmpty) { |
| None |
| } else { |
| Option(RexUtil.composeConjunction(rexBuilder, rankNodes, true)) |
| } |
| (nonRankPred, rankPred) |
| } |
| |
| def getRankRangeNdv(rankRange: RankRange): Double = rankRange match { |
| case r: ConstantRankRange => (r.rankEnd - r.rankStart + 1).toDouble |
| case _ => 100D // default value now |
| } |
| |
| /** Splits a column set between left and right sets. */ |
| def splitColumnsIntoLeftAndRight( |
| leftCount: Int, |
| columns: ImmutableBitSet): (ImmutableBitSet, ImmutableBitSet) = { |
| val leftBuilder = ImmutableBitSet.builder |
| val rightBuilder = ImmutableBitSet.builder |
| columns.foreach { |
| bit => if (bit < leftCount) leftBuilder.set(bit) else rightBuilder.set(bit - leftCount) |
| } |
| (leftBuilder.build, rightBuilder.build) |
| } |
| |
| |
| /** |
| * Estimates ratio outputRowCount/ inputRowCount of agg when ndv of groupKeys is unavailable. |
| * |
| * the value of `1.0 - math.exp(-0.1 * groupCount)` increases with groupCount |
| * from 0.095 until close to 1.0. when groupCount is 1, the formula result is 0.095, |
| * when groupCount is 2, the formula result is 0.18, |
| * when groupCount is 3, the formula result is 0.25. |
| * ... |
| * |
| * @param groupingLength grouping keys length of aggregate |
| * @return the ratio outputRowCount/ inputRowCount of agg when ndv of groupKeys is unavailable. |
| */ |
| def getAggregationRatioIfNdvUnavailable(groupingLength: Int): Double = |
| 1.0 - math.exp(-0.1 * groupingLength) |
| |
| /** |
| * Estimates outputRowCount of local aggregate. |
| * |
| * output rowcount of local agg is (1 - pow((1 - 1/x) , n/m)) * m * x, based on two assumption: |
| * 1. even distribution of all distinct data |
| * 2. even distribution of all data in each concurrent local agg worker |
| * |
| * @param parallelism number of concurrent worker of local aggregate |
| * @param inputRowCount rowcount of input node of aggregate. |
| * @param globalAggRowCount rowcount of output of global aggregate. |
| * @return outputRowCount of local aggregate. |
| */ |
| def getRowCountOfLocalAgg( |
| parallelism: Int, |
| inputRowCount: Double, |
| globalAggRowCount: Double): Double = |
| Math.min((1 - math.pow(1 - 1.0 / parallelism, inputRowCount / globalAggRowCount)) |
| * globalAggRowCount * parallelism, inputRowCount) |
| |
| /** |
| * Estimates new distinctRowCount of currentNode after it applies a condition. |
| * The estimation based on one assumption: |
| * even distribution of all distinct data |
| * |
| * @param rowCount rowcount of node. |
| * @param distinctRowCount distinct rowcount of node. |
| * @param selectivity selectivity of condition expression. |
| * @return new distinctRowCount |
| */ |
| def adaptNdvBasedOnSelectivity( |
| rowCount: Double, |
| distinctRowCount: Double, |
| selectivity: Double): Double = { |
| val ndv = Math.min(distinctRowCount, rowCount) |
| (1 - Math.pow(1 - selectivity, rowCount / ndv)) * ndv |
| } |
| |
| /** |
| * Returns [[RexInputRef]] index set of projects corresponding to the given column index. |
| * The index will be set as -1 if the given column in project is not a [[RexInputRef]]. |
| */ |
| def getInputRefIndices(index: Int, expand: Expand): util.Set[Int] = { |
| val inputRefs = new util.HashSet[Int]() |
| for (project <- expand.projects) { |
| project.get(index) match { |
| case inputRef: RexInputRef => inputRefs.add(inputRef.getIndex) |
| case _ => inputRefs.add(-1) |
| } |
| } |
| inputRefs |
| } |
| |
| } |
