kgyrtkirk commented on a change in pull request #1031:
URL: https://github.com/apache/hive/pull/1031#discussion_r434517408
##########
File path:
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveRewriteToDataSketchesRules.java
##########
@@ -368,4 +388,210 @@ void rewrite(AggregateCall aggCall) {
}
}
}
+
+ /**
+ * Generic support for rewriting Windowing expression into a different form
usually using joins.
+ */
+ private static abstract class WindowingToProjectAggregateJoinProject extends
RelOptRule {
+
+ protected final String sketchType;
+
+ public WindowingToProjectAggregateJoinProject(String sketchType) {
+ super(operand(HiveProject.class, any()), HiveRelFactories.HIVE_BUILDER,
null);
+ this.sketchType = sketchType;
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final Project project = call.rel(0);
+
+ VbuilderPAP vb = buildProcessor(call);
+ RelNode newProject = vb.processProject(project);
+
+ if (newProject == project) {
+ return;
+ } else {
+ call.transformTo(newProject);
+ }
+ }
+
+ protected abstract VbuilderPAP buildProcessor(RelOptRuleCall call);
+
+ protected static abstract class VbuilderPAP {
+ private final String sketchClass;
+ protected final RelBuilder relBuilder;
+ protected final RexBuilder rexBuilder;
+
+ protected VbuilderPAP(String sketchClass, RelBuilder relBuilder) {
+ this.sketchClass = sketchClass;
+ this.relBuilder = relBuilder;
+ rexBuilder = relBuilder.getRexBuilder();
+ }
+
+ final class ProcessShuttle extends RexShuttle {
+ public RexNode visitOver(RexOver over) {
+ return processCall(over);
+ }
+ };
+
+ protected final RelNode processProject(Project project) {
+ RelNode origInput = project.getInput();
+ relBuilder.push(origInput);
+ RexShuttle shuttle = new ProcessShuttle();
+ List<RexNode> newProjects = new ArrayList<RexNode>();
+ for (RexNode expr : project.getChildExps()) {
+ newProjects.add(expr.accept(shuttle));
+ }
+ if (relBuilder.peek() == origInput) {
+ relBuilder.clear();
+ return project;
+ }
+ relBuilder.project(newProjects);
+ return relBuilder.build();
+ }
+
+ private final RexNode processCall(RexNode expr) {
+ if (expr instanceof RexOver) {
+ RexOver over = (RexOver) expr;
+ if (isApplicable(over)) {
+ return rewrite(over);
+ }
+ }
+ return expr;
+ }
+
+ protected final SqlOperator getSqlOperator(String fnName) {
+ UDFDescriptor fn =
DataSketchesFunctions.INSTANCE.getSketchFunction(sketchClass, fnName);
+ if (!fn.getCalciteFunction().isPresent()) {
+ throw new RuntimeException(fn.toString() + " doesn't have a Calcite
function associated with it");
+ }
+ return fn.getCalciteFunction().get();
+ }
+
+ /**
+ * Do the rewrite for the given expression.
+ *
+ * When this method is invoked the {@link #relBuilder} will only contain
the current input.
+ * Expectation is to leave the new input there after the method finishes.
+ */
+ abstract RexNode rewrite(RexOver expr);
+
+ abstract boolean isApplicable(RexOver expr);
+
+ }
+ }
+
+ public static class CumeDistRewrite extends
WindowingToProjectAggregateJoinProject {
+
+ public CumeDistRewrite(String sketchType) {
+ super(sketchType);
+ }
+
+ @Override
+ protected VbuilderPAP buildProcessor(RelOptRuleCall call) {
+ return new VB(sketchType, call.builder());
+ }
+
+ private static class VB extends VbuilderPAP {
+
+ protected VB(String sketchClass, RelBuilder relBuilder) {
+ super(sketchClass, relBuilder);
+ }
+
+ @Override
+ boolean isApplicable(RexOver over) {
+ SqlAggFunction aggOp = over.getAggOperator();
+ RexWindow window = over.getWindow();
+ if (aggOp.getName().equalsIgnoreCase("cume_dist") &&
window.orderKeys.size() == 1
+ && window.getLowerBound().isUnbounded() &&
window.getUpperBound().isUnbounded()) {
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ RexNode rewrite(RexOver over) {
+ RexWindow w = over.getWindow();
+ RexFieldCollation orderKey = w.orderKeys.get(0);
+ // we don't really support nulls in aggregate/etc...they are actually
ignored
+ // so some hack will be needed for NULLs anyway..
+ ImmutableList<RexNode> partitionKeys = w.partitionKeys;
+
+ relBuilder.push(relBuilder.peek());
+ // the CDF function utilizes the '<' operator;
+ // negating the input will mirror the values on the x axis
+ // by using 1-CDF(-x) we could get a <= operator
+ RexNode key = orderKey.getKey();
+ key = rexBuilder.makeCall(SqlStdOperatorTable.UNARY_MINUS, key);
+ key = rexBuilder.makeCast(getFloatType(), key);
+
+ AggCall aggCall = ((HiveRelBuilder) relBuilder).aggregateCall(
+ (SqlAggFunction)
getSqlOperator(DataSketchesFunctions.DATA_TO_SKETCH),
+ /* distinct */ false,
+ /* approximate */ false,
+ /* ignoreNulls */ true,
+ null,
+ ImmutableList.of(),
+ null,
+ ImmutableList.of(key));
+
+ relBuilder.aggregate(relBuilder.groupKey(partitionKeys), aggCall);
+
+ List<RexNode> joinConditions;
+ joinConditions = Ord.zip(partitionKeys).stream().map(o -> {
+ RexNode f = relBuilder.field(2, 1, o.i);
+ return rexBuilder.makeCall(SqlStdOperatorTable.IS_NOT_DISTINCT_FROM,
o.e, f);
Review comment:
interestingly: CUME_DIST returns NULLs when the partitioning coulmn is
null - but the sketch based estimation (and postgres doesn't)
opened HIVE-23599
since this is not a simple '=' we could possibly loose some optimizations...
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscr...@hive.apache.org
For additional commands, e-mail: gitbox-h...@hive.apache.org
