Skip to content

[SQL] Pull filters above map, integrals, aggregates, filters #6600

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mihaibudiu wants to merge 1 commit into from
+509 −54
Open

[SQL] Pull filters above map, integrals, aggregates, filters #6600

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -885,7 +885,8 @@ public static String elapsedTimeInMs() {
@Nullable public DBSPCircuit getFinalCircuit(boolean temporary) {
compileStartTime = System.currentTimeMillis();
DBSPCircuit circuit = this.runAllCompilerStages();
this.postCompilationChecks();
if (circuit != null)
this.postCompilationChecks();
Logger.INSTANCE.belowLevel(this, 1)
.append("Compilation time ")
.appendSupplier(() -> elapsedTimeInMs() + "ms")
Expand Down
56 changes: 21 additions & 35 deletions ...-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1068,14 +1068,25 @@ void visitProject(LogicalProject project) {
this.assignOperator(project, op);
}

DBSPSimpleOperator castOutput(CalciteRelNode node, DBSPSimpleOperator operator, DBSPType outputElementType) {
DBSPType inputElementType = operator.getOutputZSetElementType();
/** Convert the type of a stream if needed, using a Map operator.
*
* @param source Output of this operator i sconverted.
* @param outputElementType Type to convert to.
* @param preservesDistinct If true and input is not a multiset, the output is not a multiset either.
* @return The operator inserting the cast.
*/
@SuppressWarnings("SameParameterValue")
DBSPSimpleOperator castOutput(CalciteRelNode node, DBSPSimpleOperator source,
DBSPType outputElementType, boolean preservesDistinct) {
DBSPType inputElementType = source.getOutputZSetElementType();
if (inputElementType.sameType(outputElementType))
return operator;
return source;
DBSPClosureExpression caster = inputElementType.caster(outputElementType, DBSPCastExpression.CastType.SqlUnsafe);
DBSPExpression function = caster.reduce(this.compiler);
DBSPSimpleOperator map = new DBSPMapOperator(
node, function, TypeCompiler.makeZSet(outputElementType), operator.outputPort());
node, function, TypeCompiler.makeZSet(outputElementType),
source.isMultiset || !preservesDistinct,
source.outputPort());
this.addOperator(map);
return map;
}
Expand Down Expand Up @@ -1149,7 +1160,8 @@ private void visitUnion(LogicalUnion union) {
this.checkPermutation(node, unionInputs, "UNION");

// input type nullability may not match
List<OutputPort> ports = Linq.map(inputs, o -> this.castOutput(node, o, outputType).outputPort());
List<OutputPort> ports = Linq.map(inputs,
o -> this.castOutput(node, o, outputType, true).outputPort());
if (union.all) {
DBSPSumOperator sum = new DBSPSumOperator(node.getFinal(), ports);
Utilities.enforce(sum.getOutputZSetElementType().sameType(outputType));
Expand Down Expand Up @@ -1185,10 +1197,10 @@ private void visitMinus(LogicalMinus minus) {

DBSPSimpleOperator filter =
this.filterNonNullFields(node, CalciteObject.EMPTY, filterIndexes, neg, false);
neg = this.castOutput(node, filter, outputType);
neg = this.castOutput(node, filter, outputType, true);
inputs.add(neg.outputPort());
} else {
opInput = this.castOutput(node, opInput, outputType);
opInput = this.castOutput(node, opInput, outputType, true);
inputs.add(opInput.outputPort());
}
first = false;
Expand Down Expand Up @@ -2404,37 +2416,11 @@ void visitLogicalValues(LogicalValues values) {
if (this.modifyTableTranslation != null) {
this.modifyTableTranslation.setResult(result);
} else {
// We currently don't have a reliable way to check whether there are duplicates
// in the Z-set, so we assume it is a multiset
DBSPSimpleOperator constant = new DBSPConstantOperator(node.getFinal(), result, true);
DBSPSimpleOperator constant = new DBSPConstantOperator(node.getFinal(), result, !result.isCertainlyDistinct());
this.assignOperator(values, constant);
}
}

/** Given an operator with output type ZSet[T], index is using
* (ZSet[S], Tup0). If a field is nullable in the input, but not in the result, insert a filter. */
DBSPMapIndexOperator indexEntireTuple(CalciteRelNode node, OutputPort port, DBSPTypeTuple resultType) {
DBSPTypeTuple inputRowType = port.getOutputZSetElementType().to(DBSPTypeTuple.class);
List<Integer> filterIndexes = new ArrayList<>();
for (int i = 0; i < resultType.size(); i++) {
if (inputRowType.getFieldType(i).mayBeNull && !resultType.getFieldType(i).mayBeNull)
filterIndexes.add(i);
}

DBSPSimpleOperator filter =
this.filterNonNullFields(node, CalciteObject.EMPTY, filterIndexes, port.simpleNode(), false);
DBSPVariablePath t = inputRowType.ref().var();
DBSPClosureExpression entireKey =
new DBSPRawTupleExpression(
DBSPTupleExpression.flatten(t.deref()).cast(
CalciteObject.EMPTY, resultType, DBSPCastExpression.CastType.SqlUnsafe),
new DBSPTupleExpression()).closure(t);
DBSPMapIndexOperator result = new DBSPMapIndexOperator(node, entireKey,
makeIndexedZSet(resultType, DBSPTypeTuple.EMPTY), filter.outputPort());
this.addOperator(result);
return result;
}

private DBSPSimpleOperator
except(IntermediateRel node, DBSPSimpleOperator left, DBSPSimpleOperator right, boolean all, boolean needsDistinct) {
List<OutputPort> inputs = new ArrayList<>(2);
Expand Down Expand Up @@ -2485,7 +2471,7 @@ void visitIntersect(LogicalIntersect intersect) {

DBSPSimpleOperator filter =
this.filterNonNullFields(node, CalciteObject.EMPTY, filterIndexes, opInput, false);
DBSPSimpleOperator next = this.castOutput(node, filter, outputType);
DBSPSimpleOperator next = this.castOutput(node, filter, outputType, true);
if (current == null)
current = next;
else {
Expand Down
3 changes: 2 additions & 1 deletion ...va/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -399,7 +399,8 @@ HepProgram getProgram(RelNode node, int level) {
// CoreRules.PROJECT_CORRELATE_TRANSPOSE,
CoreRules.PROJECT_WINDOW_TRANSPOSE,
CoreRules.PROJECT_SET_OP_TRANSPOSE,
CoreRules.FILTER_PROJECT_TRANSPOSE
CoreRules.FILTER_PROJECT_TRANSPOSE,
CoreRules.FILTER_AGGREGATE_TRANSPOSE
// Rule is unsound, replaced with UnusedFields done later.
//CoreRules.PROJECT_JOIN_TRANSPOSE
));
Expand Down
3 changes: 3 additions & 0 deletions ...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ void createOptimizer() {
this.add(new DeadCode(compiler, options.languageOptions.generateInputForEveryTable));
if (options.languageOptions.outputsAreSets)
this.add(new EnsureDistinctOutputs(compiler));
this.add(new PropagateConstants(compiler));
this.add(new PropagateEmptySources(compiler));
this.add(new MergeSums(compiler));
this.add(new Conditional(compiler, new CreateStarJoins(compiler), () -> !this.compiler.metadata.noStarJoins()));
Expand All @@ -100,6 +101,8 @@ void createOptimizer() {
this.add(new ExpandAggregateZero(compiler));
this.add(new Conditional(compiler, new RemoveStarJoins(compiler), this.compiler.metadata::noStarJoins));
this.add(new DeadCode(compiler, true));
this.add(new OptimizeWithGraph(compiler, g -> new PullFilterVisitor(compiler, g)));
this.add(new PropagateEmptySources(compiler));
this.add(new OptimizeDistinctVisitor(compiler));
// This is useful even without incrementalization if we have recursion
this.add(new OptimizeIncrementalVisitor(compiler));
Expand Down
49 changes: 48 additions & 1 deletion ...er/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/MergeSums.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,71 @@
package org.dbsp.sqlCompiler.compiler.visitors.outer;

import org.dbsp.sqlCompiler.circuit.operator.DBSPConstantOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPNegateOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPSumOperator;
import org.dbsp.sqlCompiler.circuit.OutputPort;
import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject;
import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPIndexedZSetExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPZSetExpression;
import org.dbsp.sqlCompiler.ir.type.user.DBSPTypeZSet;
import org.dbsp.util.Linq;

import java.util.ArrayList;
import java.util.List;

/** Replace Sum followed by Sum by a single Sum.
* Replace a sum with a single input by its input. */
* Replace a sum with a single input by its input.
* Replace a + (neg(a)) with nothing. */
public class MergeSums extends CircuitCloneVisitor {
public MergeSums(DBSPCompiler compiler) {
super(compiler, false);
}

List<OutputPort> removeComplements(List<OutputPort> ports) {
List<OutputPort> results = new ArrayList<>();
// Find non-negated operators
List<DBSPNegateOperator> negations = new ArrayList<>();
for (OutputPort port: ports) {
if (port.isSimpleNode()) {
DBSPSimpleOperator source = port.simpleNode();
if (source.is(DBSPNegateOperator.class)) {
negations.add(source.to(DBSPNegateOperator.class));
continue;
}
}
results.add(port);
}
for (DBSPNegateOperator neg: negations) {
OutputPort negated = neg.input();
if (results.contains(negated)) {
results.remove(negated);
} else {
results.add(neg.outputPort());
}
}
return results;
}

@Override
public void postorder(DBSPSumOperator operator) {
List<OutputPort> sources = Linq.map(operator.inputs, this::mapped);
sources = this.removeComplements(sources);

if (sources.isEmpty()) {
final DBSPExpression value;
if (operator.outputType().is(DBSPTypeZSet.class)) {
value = new DBSPZSetExpression(operator.getOutputZSetElementType());
} else {
value = new DBSPIndexedZSetExpression(CalciteObject.EMPTY, operator.getOutputIndexedZSetType());
}
DBSPConstantOperator constant = new DBSPConstantOperator(operator.getRelNode(), value, false);
this.map(operator, constant);
return;
}

if (sources.size() == 1) {
this.map(operator.outputPort(), sources.get(0), false);
return;
Expand Down
172 changes: 172 additions & 0 deletions ...mpiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PropagateConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,172 @@
package org.dbsp.sqlCompiler.compiler.visitors.outer;

import org.dbsp.sqlCompiler.circuit.OutputPort;
import org.dbsp.sqlCompiler.circuit.operator.DBSPConstantOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPDistinctOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPNegateOperator;
import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
import org.dbsp.sqlCompiler.compiler.visitors.inner.Simplify;
import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPIndexedZSetExpression;
import org.dbsp.sqlCompiler.ir.expression.DBSPZSetExpression;
import org.dbsp.sqlCompiler.ir.expression.literal.DBSPBoolLiteral;

import javax.annotation.Nullable;
import java.util.HashMap;
import java.util.Map;

/** Try to optimize operators applied to constant inputs */
public class PropagateConstants extends CircuitCloneVisitor {
public PropagateConstants(DBSPCompiler compiler) {
super(compiler, false);
}

boolean inputIsConstant(OutputPort port) {
return port.node().is(DBSPConstantOperator.class);
}

@Nullable
public static DBSPConstantOperator filterConstant(DBSPCompiler compiler,
DBSPConstantOperator constant, DBSPClosureExpression filter) {
DBSPExpression value = constant.getFunction();
Simplify simplify = new Simplify(compiler);
if (value.is(DBSPZSetExpression.class)) {
DBSPZSetExpression set = value.to(DBSPZSetExpression.class);
Map<DBSPExpression, Long> result = new HashMap<>();
boolean evaluated = true;
for (var entry : set.data.entrySet()) {
DBSPExpression filtered = filter.call(entry.getKey().borrow()).reduce(compiler);
DBSPExpression simplified = simplify.apply(filtered).to(DBSPExpression.class);
if (simplified.is(DBSPBoolLiteral.class)) {
DBSPBoolLiteral b = simplified.to(DBSPBoolLiteral.class);
if (b.value == null) {
// Should not happen, since filter expressions should not be nullable,
// but we are a bit paranoid.
evaluated = false;
break;
}
if (b.value) {
result.put(entry.getKey(), entry.getValue());
}
} else {
// Could not evaluate filter
evaluated = false;
break;
}
}
if (evaluated) {
DBSPZSetExpression zset = new DBSPZSetExpression(result, set.elementType);
boolean isDistinct = !constant.isMultiset || zset.isCertainlyDistinct();
return new DBSPConstantOperator(constant.getRelNode(), zset, !isDistinct);
}
}
return null;
}

@Nullable
public static DBSPConstantOperator mapConstant(
DBSPCompiler compiler, DBSPConstantOperator constant, DBSPClosureExpression map) {
DBSPExpression value = constant.getFunction();
Simplify simplify = new Simplify(compiler);
if (value.is(DBSPZSetExpression.class)) {
DBSPZSetExpression set = value.to(DBSPZSetExpression.class);
Map<DBSPExpression, Long> result = new HashMap<>();
boolean evaluated = true;
for (var entry : set.data.entrySet()) {
DBSPExpression filtered = map.call(entry.getKey().borrow()).reduce(compiler);
DBSPExpression simplified = simplify.apply(filtered).to(DBSPExpression.class);
if (simplified.isCompileTimeConstant()) {
result.put(simplified, entry.getValue());
} else {
// Could not evaluate filter
evaluated = false;
break;
}
}
if (evaluated) {
DBSPZSetExpression zset = new DBSPZSetExpression(result, map.getResultType());
boolean isDistinct = zset.isCertainlyDistinct();
return new DBSPConstantOperator(constant.getRelNode(), zset, !isDistinct);
}
}
return null;
}

@Override
public void postorder(DBSPFilterOperator operator) {
OutputPort source = this.mapped(operator.input());
if (!this.inputIsConstant(source)) {
super.postorder(operator);
return;
}
DBSPConstantOperator constant = source.simpleNode().to(DBSPConstantOperator.class);
DBSPConstantOperator filteredConstant = filterConstant(this.compiler, constant, operator.getClosureFunction());
if (filteredConstant != null) {
this.map(operator, filteredConstant);
return;
}
super.postorder(operator);
}

@Override
public void postorder(DBSPDistinctOperator operator) {
OutputPort source = this.mapped(operator.input());
if (source.isSimpleNode() && !source.simpleNode().isMultiset) {
// This includes constants
this.map(operator, source.simpleNode());
return;
}
super.postorder(operator);
}

@Override
public void postorder(DBSPNegateOperator operator) {
OutputPort source = this.mapped(operator.input());
if (!this.inputIsConstant(source)) {
super.postorder(operator);
return;
}
DBSPConstantOperator constant = source.simpleNode().to(DBSPConstantOperator.class);
DBSPZSetExpression value = constant.getFunction().as(DBSPZSetExpression.class);
if (value != null) {
value = value.negate();
boolean isMultiset = constant.isMultiset;
if (value.isCertainlyDistinct())
isMultiset = false;
DBSPConstantOperator result = new DBSPConstantOperator(operator.getRelNode(), value, isMultiset);
this.map(operator, result);
return;
}
DBSPIndexedZSetExpression ix = constant.getFunction().as(DBSPIndexedZSetExpression.class);
if (ix != null) {
ix = ix.negate();
DBSPConstantOperator result = new DBSPConstantOperator(operator.getRelNode(), ix, constant.isMultiset);
this.map(operator, result);
return;
}
super.postorder(operator);
}

@Override
public void postorder(DBSPMapOperator operator) {
OutputPort source = this.mapped(operator.input());
if (!this.inputIsConstant(source)) {
super.postorder(operator);
return;
}
DBSPConstantOperator constant = source.simpleNode().to(DBSPConstantOperator.class);
DBSPZSetExpression value = constant.getFunction().as(DBSPZSetExpression.class);
if (value != null) {
DBSPClosureExpression map = operator.getClosureFunction();
DBSPConstantOperator result = mapConstant(this.compiler, constant, map);
if (result != null) {
this.map(operator, result);
return;
}
}
super.postorder(operator);
}
}
Loading