You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/12/14 09:26:41 UTC
(flink) 11/11: For debug
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch debug_q58
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 683eae44655b194587454bc850e71a9530eceda7
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Thu Dec 14 17:26:02 2023 +0800
For debug
---
.../table/api/config/OptimizerConfigOptions.java | 7 ++
.../program/FlinkRuntimeFilterProgram.java | 78 ++++++++++++++++++++--
.../planner/plan/metadata/FlinkRelMdRowCount.scala | 13 +++-
.../GlobalRuntimeFilterBuilderOperator.java | 7 ++
4 files changed, 99 insertions(+), 6 deletions(-)
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
index e38bcce654a..6ad3eb5b770 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java
@@ -193,6 +193,13 @@ public class OptimizerConfigOptions {
.defaultValue(false)
.withDescription("Whether consider benefit.");
+
+ @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH)
+ public static final ConfigOption<Boolean> TABLE_OPTIMIZER_RUNTIME_FILTER_IGNORE_BUILD_EXCHANGE_LIMIT =
+ key("table.optimizer.runtime-filter.ignore-build-exchange-limit")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Ignore build exchange limit.");
/**
* The data volume of build side needs to be under this value. If the data volume of build side
* is too large, the building overhead will be too large, which may lead to a negative impact on
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
index 8d20972be80..f9c8645b650 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
@@ -168,6 +168,31 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
return join;
}
+ System.out.println(
+ "Join is "
+ + join
+ + ", left side is "
+ + join.getLeft()
+ + ", right side is "
+ + join.getRight()
+ + ", left size is "
+ + getEstimatedDataSize(join.getLeft()).get() / (1024 * 1024)
+ + " MB"
+ + ", right size is "
+ + getEstimatedDataSize(join.getRight()).get() / (1024 * 1024)
+ + " MB"
+ + ", left row count is "
+ + getEstimatedRowCount(join.getLeft()).get()
+ + ", right row count is "
+ + getEstimatedRowCount(join.getRight()).get());
+
+ System.out.println(
+ "left can be probe "
+ + canBeProbeSide(join.getLeft())
+ + " , "
+ + "right can be probe "
+ + canBeProbeSide(join.getRight()));
+
boolean leftIsBuild;
if (canBeProbeSide(join.getLeft())) {
leftIsBuild = false;
@@ -177,6 +202,8 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
return join;
}
+ System.out.println("left is build " + leftIsBuild);
+
// check left join + left build
if (join.getJoinType() == JoinRelType.LEFT && !leftIsBuild) {
return join;
@@ -211,6 +238,15 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
(build, indices) ->
isSuitableDataSize(build, probeSide, indices, probeIndices));
+ if (!suitableBuildOpt.isPresent()
+ && ignoreBuildExchangeLimit(buildSide)
+ && isSuitableDataSize(buildSide, probeSide, buildIndices, probeIndices)) {
+ suitableBuildOpt = Optional.of(new BuildSideInfo(buildSide, buildIndices));
+ System.out.println("Using not-exchange build side " + suitableBuildOpt.get());
+ }
+
+ System.out.println("Can find build " + suitableBuildOpt.isPresent());
+
if (suitableBuildOpt.isPresent()) {
BuildSideInfo suitableBuildInfo = suitableBuildOpt.get();
RelNode newProbe =
@@ -385,6 +421,7 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
buildSideChecker);
} else {
+ System.out.println("findSuitableBuildSide more case " + rel.getClass());
// the above cases can cover all cases of TPC-DS test
// we may find more cases later
}
@@ -499,18 +536,29 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
return rel;
}
} else {
+ System.out.println(
+ "tryPushDownProbeSideAndInjectRuntimeFilter more case " + rel.getClass());
// the above cases can cover all cases of TPC-DS test
// we may find more cases later
}
if (considerBenefit(rel) && !filterHasBenefit) {
- System.out.println("considerBenefit " + considerBenefit(rel) + " filterHasBenefit " + filterHasBenefit);
+ System.out.println(
+ "considerBenefit "
+ + considerBenefit(rel)
+ + " filterHasBenefit "
+ + filterHasBenefit);
// If the probe side is a direct table source, or only simple Calc, no other operations,
// we will not inject runtime filter, because we believe the benefit to be small or even
// negative
return rel;
} else {
- System.out.println("considerBenefit " + considerBenefit(rel) + " filterHasBenefit " + filterHasBenefit + " has insert ");
+ System.out.println(
+ "considerBenefit "
+ + considerBenefit(rel)
+ + " filterHasBenefit "
+ + filterHasBenefit
+ + " has insert ");
return createNewProbeWithRuntimeFilter(
ignoreExchange(buildSideInfo.buildSide),
ignoreExchange(rel),
@@ -632,8 +680,17 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
return false;
}
- return computeFilterRatio(buildSide, probeSide, buildIndices, probeIndices)
- >= minFilterRatio;
+ double filterRatio = computeFilterRatio(buildSide, probeSide, buildIndices, probeIndices);
+ System.out.println(
+ "buildSize "
+ + buildSize.get()
+ + ", probeSize "
+ + probeSize.get() / (1024 * 1024)
+ + " MB"
+ + ", filterRatioooooo "
+ + filterRatio);
+
+ return filterRatio >= minFilterRatio;
}
private static double computeFilterRatio(
@@ -646,11 +703,17 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
Optional<Double> probeNdv = getEstimatedNdv(probeSide, ImmutableBitSet.of(probeIndices));
if (buildNdv.isPresent() && probeNdv.isPresent()) {
+ System.out.println("buildNdv is " + buildNdv.get() + ", probeNdv is " + probeNdv.get());
return Math.max(0, 1 - buildNdv.get() / probeNdv.get());
} else {
Optional<Double> buildRowCount = getEstimatedRowCount(buildSide);
Optional<Double> probeRowCount = getEstimatedRowCount(probeSide);
checkState(buildRowCount.isPresent() && probeRowCount.isPresent());
+ System.out.println(
+ "buildRowCountInRatio is "
+ + buildRowCount.get()
+ + ", probeRowCountInRatio is "
+ + probeRowCount.get());
return Math.max(0, 1 - buildRowCount.get() / probeRowCount.get());
}
}
@@ -697,6 +760,13 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
.get(OptimizerConfigOptions.TABLE_OPTIMIZER_RUNTIME_FILTER_CONSIDER_BENEFIT);
}
+ private static boolean ignoreBuildExchangeLimit(RelNode relNode) {
+ return unwrapTableConfig(relNode)
+ .get(
+ OptimizerConfigOptions
+ .TABLE_OPTIMIZER_RUNTIME_FILTER_IGNORE_BUILD_EXCHANGE_LIMIT);
+ }
+
private static long getMaxBuildDataSize(RelNode relNode) {
return unwrapTableConfig(relNode)
.get(OptimizerConfigOptions.TABLE_OPTIMIZER_RUNTIME_FILTER_MAX_BUILD_DATA_SIZE)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
index f5dec96af68..d50d79a6b07 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala
@@ -268,7 +268,9 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun
val joinInfo = JoinInfo.of(leftChild, rightChild, join.getCondition)
if (joinInfo.leftSet().nonEmpty) {
+ println("Start get join row count for " + join + ", leftRowCount is " + leftRowCount + ", rightRowCount is " + rightRowCount)
val innerJoinRowCount = getEquiInnerJoinRowCount(join, mq, leftRowCount, rightRowCount)
+ println("End get join row count for " + join + ", leftRowCount is " + leftRowCount + ", rightRowCount is " + rightRowCount)
require(innerJoinRowCount != null)
// Make sure outputRowCount won't be too small based on join type.
join.getJoinType match {
@@ -335,11 +337,15 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun
fmq.getSelectivity(joinWithOnlyEquiPred, nonEquiPred)
}
+ println("selectivityOfNonEquiPred is " + selectivityOfNonEquiPred)
+
if (leftNdv != null && rightNdv != null) {
// selectivity of equi part is 1 / Max(leftNdv, rightNdv)
val selectivityOfEquiPred = Math.min(1d, 1d / Math.max(leftNdv, rightNdv))
- return leftRowCount * rightRowCount * selectivityOfEquiPred *
+ val rowCount = leftRowCount * rightRowCount * selectivityOfEquiPred *
selectivityOfNonEquiPred
+ println("Has ndv, rowCount is " + rowCount)
+ return rowCount
}
val leftKeysAreUnique = fmq.areColumnsUnique(leftChild, leftKeySet)
@@ -357,6 +363,7 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun
} else {
leftRowCount * selectivityOfNonEquiPred
}
+ println("Has unique, rowCount is " + outputRowCount)
return outputRowCount
}
@@ -364,7 +371,9 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun
// rowCount = (leftRowCount + rightRowCount) * join condition selectivity
val crossJoin = copyJoinWithNewCondition(join, rexBuilder.makeLiteral(true))
val selectivity = fmq.getSelectivity(crossJoin, condition)
- (leftRowCount + rightRowCount) * selectivity
+ val rowCount = (leftRowCount + rightRowCount) * selectivity
+ println("has no ndv stats and no uniqueKeys stats, rowCount is " + rowCount + ", selectivity is " + selectivity)
+ rowCount
}
private def copyJoinWithNewCondition(join: Join, newCondition: RexNode): Join = {
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/runtimefilter/GlobalRuntimeFilterBuilderOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/runtimefilter/GlobalRuntimeFilterBuilderOperator.java
index f591df68f86..f39b2157028 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/runtimefilter/GlobalRuntimeFilterBuilderOperator.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/runtimefilter/GlobalRuntimeFilterBuilderOperator.java
@@ -28,6 +28,9 @@ import org.apache.flink.table.runtime.operators.runtimefilter.util.RuntimeFilter
import org.apache.flink.table.runtime.util.StreamRecordCollector;
import org.apache.flink.util.Collector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import static org.apache.flink.table.runtime.operators.runtimefilter.util.RuntimeFilterUtils.OVER_MAX_ROW_COUNT;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -46,6 +49,9 @@ public class GlobalRuntimeFilterBuilderOperator extends TableStreamOperator<RowD
private transient Collector<RowData> collector;
private transient int globalRowCount;
+ private static final Logger LOG =
+ LoggerFactory.getLogger(GlobalRuntimeFilterBuilderOperator.class);
+
public GlobalRuntimeFilterBuilderOperator(int maxRowCount) {
checkArgument(maxRowCount > 0);
this.maxRowCount = maxRowCount;
@@ -87,6 +93,7 @@ public class GlobalRuntimeFilterBuilderOperator extends TableStreamOperator<RowD
@Override
public void endInput() throws Exception {
+ LOG.warn("maxRowCount {}, globalRowCount {}", maxRowCount, globalRowCount);
collector.collect(
RuntimeFilterUtils.convertBloomFilterToRowData(
globalRowCount, serializedGlobalFilter));