diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java index e4951ef7229..83e4747bc80 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/DBSPCompiler.java @@ -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") diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java index ef4b0fcc156..edc3bc45747 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java @@ -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; } @@ -1149,7 +1160,8 @@ private void visitUnion(LogicalUnion union) { this.checkPermutation(node, unionInputs, "UNION"); // input type nullability may not match - List ports = Linq.map(inputs, o -> this.castOutput(node, o, outputType).outputPort()); + List 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)); @@ -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; @@ -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 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 inputs = new ArrayList<>(2); @@ -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 { diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java index 373c579f635..1f0b7813a69 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java @@ -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 )); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java index 3f8e50a3878..768ef268d7a 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java @@ -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())); @@ -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)); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/MergeSums.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/MergeSums.java index 575a562a617..84536a68789 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/MergeSums.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/MergeSums.java @@ -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 removeComplements(List ports) { + List results = new ArrayList<>(); + // Find non-negated operators + List 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 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; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PropagateConstants.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PropagateConstants.java new file mode 100644 index 00000000000..aaf1f72342d --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PropagateConstants.java @@ -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 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 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); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java new file mode 100644 index 00000000000..3fbdde9f9a7 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java @@ -0,0 +1,139 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.OutputPort; +import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateLinearPostprocessOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateLinearPostprocessRetainKeysOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateOperatorBase; +import org.dbsp.sqlCompiler.circuit.operator.DBSPChainAggregateOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPConstantOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPDifferentiateOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPMapIndexOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPNoopOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; +import org.dbsp.sqlCompiler.ir.expression.DBSPBinaryExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPOpcode; +import org.dbsp.sqlCompiler.ir.expression.DBSPRawTupleExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPVariablePath; +import org.dbsp.sqlCompiler.ir.expression.NoExpression; +import org.dbsp.sqlCompiler.ir.type.DBSPType; +import org.dbsp.sqlCompiler.ir.type.user.DBSPTypeIndexedZSet; +import org.dbsp.util.Linq; +import org.dbsp.util.Maybe; + +/** + * Move filters up in a plan, towards sources. + * - pull a filter above an aggregate if the filter only looks + * at that the aggregate groups on. + * - pull a filter above a map/mapindex if it doesn't become "too complex". + * - pull filters above noop, integrators, differentiators + * - combine two consecutive filters into one */ +public class PullFilterVisitor extends CircuitCloneWithGraphsVisitor { + public PullFilterVisitor(DBSPCompiler compiler, CircuitGraphs graphs) { + super(compiler, graphs); + } + + @Override + public void postorder(DBSPFilterOperator operator) { + OutputPort source = this.mapped(operator.input()); + int inputFanout = this.getGraph().getFanout(operator.input().node()); + if (inputFanout != 1) { + super.postorder(operator); + return; + } + if (source.node().is(DBSPAggregateLinearPostprocessOperator.class) + || source.node().is(DBSPAggregateLinearPostprocessRetainKeysOperator.class) + || source.node().is(DBSPAggregateOperatorBase.class) + || source.node().is(DBSPChainAggregateOperator.class)) { + DBSPClosureExpression filter = operator.getClosureFunction(); + DBSPTypeIndexedZSet aggInputType = source.node().inputs.get(0).getOutputIndexedZSetType(); + DBSPType elementType = operator.input().getOutputIndexedZSetType().elementType.ref(); + DBSPVariablePath var = aggInputType.getKVRefType().var(); + DBSPExpression newFilter = + filter.call(new DBSPRawTupleExpression(var.field(0), new NoExpression(elementType))) + .closure(var) + .reduce(this.compiler); + boolean hasNoExpression = FilterJoinVisitor.ContainsNoExpression.search(this.compiler, newFilter); + if (!hasNoExpression) { + DBSPSimpleOperator newFilterOperator = new DBSPFilterOperator(operator.getRelNode(), + newFilter, source.simpleNode().inputs.get(0)) + .copyAnnotations(operator); + this.addOperator(newFilterOperator); + DBSPSimpleOperator result = + source.simpleNode().withInputs(Linq.list(newFilterOperator.outputPort()), true) + .to(DBSPSimpleOperator.class); + this.map(operator, result); + return; + } + } else if (source.node().is(DBSPMapOperator.class) + || source.node().is(DBSPMapIndexOperator.class)) { + DBSPClosureExpression mapClosure = source.simpleNode().getClosureFunction(); + DBSPClosureExpression filterClosure = operator.getClosureFunction(); + if (filterClosure.shouldInlineComposition(this.compiler, mapClosure)) { + final DBSPClosureExpression newFilter; + if (source.node().is(DBSPMapOperator.class)) { + newFilter = filterClosure.applyAfter(this.compiler, mapClosure, Maybe.YES); + } else { + DBSPExpression argument = new DBSPRawTupleExpression( + mapClosure.body.field(0).borrow(), + mapClosure.body.field(1).borrow()); + DBSPExpression apply = filterClosure.call(argument).reduce(this.compiler()); + newFilter = apply.closure(mapClosure.parameters); + } + DBSPSimpleOperator newFilterOperator = new DBSPFilterOperator(operator.getRelNode(), + newFilter, source.simpleNode().inputs.get(0)) + .copyAnnotations(operator); + this.addOperator(newFilterOperator); + DBSPSimpleOperator result = + source.simpleNode().withInputs(Linq.list(newFilterOperator.outputPort()), true) + .to(DBSPSimpleOperator.class); + this.map(operator, result); + return; + } + } else if (source.node().is(DBSPNoopOperator.class) + || source.node().is(DBSPDifferentiateOperator.class) + || source.node().is(DBSPIntegrateOperator.class)) { + DBSPSimpleOperator newFilterOperator = + operator.withInputs(Linq.list(source.simpleNode().inputs.get(0)), true) + .copyAnnotations(operator) + .to(DBSPSimpleOperator.class); + this.addOperator(newFilterOperator); + DBSPSimpleOperator result = + source.simpleNode().withInputs(Linq.list(newFilterOperator.outputPort()), true) + .to(DBSPSimpleOperator.class); + this.map(operator, result); + return; + } else if (source.node().is(DBSPFilterOperator.class)) { + DBSPClosureExpression clo1 = source.simpleNode().getClosureFunction(); + DBSPClosureExpression clo2 = operator.getClosureFunction(); + DBSPVariablePath var = clo1.parameters[0].type.var(); + DBSPClosureExpression newFilter = + new DBSPBinaryExpression( + CalciteObject.EMPTY, clo1.getResultType(), + DBSPOpcode.AND, + clo1.call(var).reduce(this.compiler), + clo2.call(var).reduce(this.compiler)).closure(var); + DBSPSimpleOperator result = new DBSPFilterOperator( + operator.getRelNode().after(source.simpleNode().getRelNode()), + newFilter, source.simpleNode().inputs.get(0)) + .copyAnnotations(operator); + this.map(operator, result); + return; + } else if (source.node().is(DBSPConstantOperator.class)) { + DBSPClosureExpression filter = operator.getClosureFunction(); + DBSPConstantOperator c = source.node().to(DBSPConstantOperator.class); + DBSPConstantOperator filteredConstant = PropagateConstants.filterConstant(this.compiler, c, filter); + if (filteredConstant != null) { + this.map(operator, filteredConstant); + return; + } + } + super.postorder(operator); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java index 3f05ec70f8e..8ae584ae312 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java @@ -161,6 +161,7 @@ public IIndentStream toString(IIndentStream builder) { .append(")"); } + /** True if the composition this(before) can productively inline before */ public boolean shouldInlineComposition(DBSPCompiler compiler, DBSPClosureExpression before) { Projection projection = new Projection(compiler, true, true); projection.apply(this); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPIndexedZSetExpression.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPIndexedZSetExpression.java index 95d0f96e10a..90974085c8d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPIndexedZSetExpression.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPIndexedZSetExpression.java @@ -12,6 +12,7 @@ import org.dbsp.sqlCompiler.ir.type.DBSPType; import org.dbsp.sqlCompiler.ir.type.user.DBSPTypeIndexedZSet; import org.dbsp.util.IIndentStream; +import org.dbsp.util.Utilities; import javax.annotation.Nullable; @@ -39,15 +40,22 @@ public boolean isNull() { @Override public DBSPExpression deepCopy() { + Utilities.enforce(this.isEmpty()); return new DBSPIndexedZSetExpression(this.getNode(), this.type); } @Override public boolean equivalent(EquivalenceContext context, DBSPExpression other) { // This is accurate only for empty IndexedZSets, which is the only supported case so far + Utilities.enforce(this.isEmpty()); return this.getType().sameType(other.getType()); } + public DBSPIndexedZSetExpression negate() { + Utilities.enforce(this.isEmpty()); + return this; + } + @Override public void accept(InnerVisitor visitor) { VisitDecision decision = visitor.preorder(this); @@ -70,6 +78,7 @@ public boolean isEmpty() { @SuppressWarnings("SameReturnValue") public int size() { + Utilities.enforce(this.isEmpty()); return 0; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPZSetExpression.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPZSetExpression.java index 3159d2227d5..21461a6da78 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPZSetExpression.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPZSetExpression.java @@ -100,6 +100,22 @@ public static DBSPZSetExpression emptyWithElementType(DBSPType elementType) { return new DBSPZSetExpression(elementType); } + /** Return 'true' when this constant is not a multi-set. + * In general there is no easy way to do this, since we don't have a canonical representation of constants + * at compile-time, so this is only conservative. */ + public boolean isCertainlyDistinct() { + if (this.isEmpty()) + return true; + if (this.data.size() == 1) { + for (var weight : this.data.values()) { + if (weight <= 1) { + return true; + } + } + } + return false; + } + @SuppressWarnings("MethodDoesntCallSuperMethod") public DBSPZSetExpression clone() { return new DBSPZSetExpression(new HashMap<>(this.data), this.elementType); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/Utilities.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/Utilities.java index fbf2c209021..37b3908b7a5 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/Utilities.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/util/Utilities.java @@ -332,8 +332,7 @@ public static VE putNew(Map map, K key, VE value) { /** Print a ResultSet obtained from some database. */ @SuppressWarnings("unused") - public static void showResultSet(ResultSet result, PrintStream out) - throws SQLException { + public static void showResultSet(ResultSet result, PrintStream out) throws SQLException { int columnCount = result.getMetaData().getColumnCount(); while (result.next()) { for (int i = 1; i <= columnCount; i++) { diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/QATests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/QATests.java index 9ba3adc942f..c59e77638ab 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/QATests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/QATests.java @@ -96,7 +96,7 @@ public void qaTests() throws SQLException { for (File c : getQATests()) { // This program cannot be compiled because it contains a udf if (c.toString().matches(".*swiss.*-q1.*")) continue; - // if (!c.toString().contains("x.sql")) continue; + if (!c.toString().contains("x.sql")) continue; System.out.println("Compiling " + c); try { compileAndCheck(c); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegressionTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegressionTests.java index 7d38819cad8..6715e23c827 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegressionTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegressionTests.java @@ -8,6 +8,7 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceTableOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPStarJoinFilterMapOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPStarJoinIndexOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPUnaryOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPWindowOperator; import org.dbsp.sqlCompiler.compiler.CompilerOptions; @@ -2026,6 +2027,7 @@ CREATE VIEW V WITH ('emit_final' = 'y') ccs.step("INSERT INTO T VALUES(0, 0), (1, 2), (2, 2)", """ y | min | max | stddev | arg_max | weight -------------------------------------------"""); + ccs.blockForCompaction(); // Insert one tuple which produces no output yet; output for // data inserted so far is now emitted. ccs.step("INSERT INTO T VALUES(1, 5)", """ @@ -2036,7 +2038,7 @@ CREATE VIEW V WITH ('emit_final' = 'y') int joins = 0; @Override - public void postorder(DBSPStarJoinFilterMapOperator operator) { + public void postorder(DBSPStarJoinIndexOperator operator) { this.joins++; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java index cba768fe55b..324e745e60d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java @@ -1,5 +1,7 @@ package org.dbsp.sqlCompiler.compiler.sql.simple; +import org.dbsp.sqlCompiler.circuit.operator.DBSPFlatMapIndexOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceTableOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamJoinOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPWaterlineOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; @@ -63,9 +65,9 @@ public void issue5398() { CREATE VIEW W AS SELECT R[1], R[2], R[3][1] FROM V;"""); ccs.stepWeightOne(""" INSERT INTO T VALUES(0, 1, 2); INSERT INTO S VALUES(3, 4);""", """ - y | z | b - ----------- - 1 | 2 | 4"""); + y | z | b + ----------- + 1 | 2 | 4"""); } @Test @@ -144,8 +146,6 @@ CREATE TABLE purchase ( ) WITH ( 'append_only' = 'true' ); - - CREATE MATERIALIZED VIEW v1 WITH ('emit_final' = 'a_ts') AS @@ -811,4 +811,54 @@ public void testMapNullable() { {NULL: NULL} """); } + + @Test + public void issue6565() { + var ccs = this.getCCS(""" + CREATE TABLE employees(dept VARCHAR); + CREATE VIEW V AS SELECT dept, COUNT(*) AS n + FROM employees + GROUP BY dept + HAVING dept LIKE 'S%';"""); + ccs.visit(new CircuitVisitor(ccs.compiler) { + boolean filterFound = false; + + @Override + public void postorder(DBSPFlatMapIndexOperator node) { + // Source is input + this.filterFound = true; + Assert.assertTrue(node.input().node().is(DBSPSourceTableOperator.class)); + } + + @Override + public void endVisit() { + Assert.assertTrue(this.filterFound); + } + }); + } + + @Test + public void issue6565b() { + var ccs = this.getCCS(""" + CREATE TABLE employees(dept VARCHAR); + CREATE LOCAL VIEW V AS SELECT dept, COUNT(*) AS n + FROM employees + GROUP BY dept; + CREATE VIEW W AS SELECT * FROM V WHERE dept LIKE 'S%';"""); + ccs.visit(new CircuitVisitor(ccs.compiler) { + boolean filterFound = false; + + @Override + public void postorder(DBSPFlatMapIndexOperator node) { + // Source is input + this.filterFound = true; + Assert.assertTrue(node.input().node().is(DBSPSourceTableOperator.class)); + } + + @Override + public void endVisit() { + Assert.assertTrue(this.filterFound); + } + }); + } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/SetOpTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/SetOpTests.java index 68401aeb3d9..ef55f7647ff 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/SetOpTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/SetOpTests.java @@ -1,11 +1,40 @@ package org.dbsp.sqlCompiler.compiler.sql.simple; +import org.dbsp.sqlCompiler.circuit.operator.DBSPConstantOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSinkOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceTableOperator; import org.dbsp.sqlCompiler.compiler.sql.quidem.ScottBaseTests; +import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; +import org.junit.Assert; +import org.junit.Ignore; import org.junit.Test; /** End-to-end tests for SQL set operations: UNION, UNION ALL, INTERSECT, INTERSECT ALL, EXCEPT. * Some tests are from set-op.iq */ public class SetOpTests extends ScottBaseTests { + @Test @Ignore("TODO: This used to work when INTERSECT was implemented using JOIN, but it does not work" + + "using the new implementation using EXCEPT.") + public void simplifyIntersect() { + var ccs = this.getCCS(""" + CREATE TABLE tbl(x INT, y INT NOT NULL); + + CREATE MATERIALIZED VIEW v AS + SELECT x, y FROM tbl + INTERSECT + SELECT 5, NULL;"""); + ccs.visit(new CircuitVisitor(ccs.compiler) { + // This is simplified to an empty set + @Override + public void postorder(DBSPSimpleOperator operator) { + Assert.assertTrue( + operator.is(DBSPConstantOperator.class) + || operator.is(DBSPSourceTableOperator.class) + || operator.is(DBSPSinkOperator.class)); + } + }); + } + @Test public void calciteIntersect() { this.qst(""" diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json index 863aefbef72..03aa96b7ca3 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json @@ -382,7 +382,7 @@ {"start_line_number":19,"start_column":8,"end_line_number":19,"end_column":26}, {"start_line_number":20,"start_column":27,"end_line_number":20,"end_column":37} ], - "persistent_id": "1d55e90bf29e3f19e1986f4d01fa7ac67549ea1345bd04d223f43ff4694927f6" + "persistent_id": "5aa9e3ce872b5ae2f4047600a57ec8ac352119caed9f06a50ae9709d1c32ba60" }, "s4": { "operation": "flat_map_index", @@ -392,18 +392,18 @@ "calcite": { "seq": [ { - "final": 3 - },{ "final": 4 },{ "partial": 7 + },{ + "final": 3 }] }, "positions": [ {"start_line_number":19,"start_column":8,"end_line_number":19,"end_column":26}, {"start_line_number":20,"start_column":11,"end_line_number":20,"end_column":21} ], - "persistent_id": "532b61244e3f6ecac6e439489956f24fb4bdfedc3708b4dd7601e06c20c5ffc2" + "persistent_id": "1053dafeb120e48af7501505c4717953ea687294833b044a7b3a23fc5ca9e48b" }, "s5": { "operation": "delta0", @@ -430,7 +430,7 @@ {"start_line_number":16,"start_column":9,"end_line_number":16,"end_column":33}, {"start_line_number":19,"start_column":8,"end_line_number":19,"end_column":26} ], - "persistent_id": "41c15145ffd7f94a9c7335f8c82ca5f9fcd14d9d5c62c8aeea8616a7eb7433f4" + "persistent_id": "5a44d41ef4ed2601d42a7a00634ccb396c9c0230eab64a913761a6b73cb75c6d" }, "s7": { "operation": "sum", @@ -442,7 +442,7 @@ "final": 10 }, "positions": [], - "persistent_id": "f9e9f0eed2184cdd3cacd079404f8ff466e148cfafe969034ddacc0f747a8d17" + "persistent_id": "934c48e9f1e8daed99d74753d09f112e8fe1981a549d318476a3a4fbc18736f7" } }, "s8": { "operation": "inspect", @@ -457,7 +457,7 @@ {"start_line_number":4,"start_column":1,"end_line_number":21,"end_column":1}, {"start_line_number":4,"start_column":1,"end_line_number":21,"end_column":1} ], - "persistent_id": "c79ff4ab927d459ed213c8a0e70e93c7fdc0ed93011443b0ae9f1404fc33eca1" + "persistent_id": "e0ef8291b19f9950a61b410d5d368894412303197bc5d3465184f1fff0654253" }, "s9": { "operation": "constant", "inputs": [],