GeezerVison is showing github.com — [SQL] Dataflow analysis to compute for each stream whether it's a (multi)set by mihaibudiu · Pull Request #7725 · feldera/feldera
[SQL] Dataflow analysis to compute for each stream whether it's a (multi)set
#7725
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.
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
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.java
(139)
.json
(2)
All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Original file line number
Diff line number
Diff line change
Expand Up
@@ -118,10 +118,6 @@ public DBSPType getOutputRowRefType() {
return this.outputType();
}
public boolean isMultiset() {
return this.node().isMultiset(this.outputNumber);
}
public DBSPType outputType() {
return this.node().outputType(this.outputNumber);
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -28,7 +28,7 @@ public DBSPAggregateLinearPostprocessOperator(
DBSPTypeIndexedZSet outputType,
DBSPExpression function,
DBSPClosureExpression postProcess, OutputPort input) {
super(node, "aggregate_linear_postprocess", function, outputType, false, input);
super(node, "aggregate_linear_postprocess", function, outputType, input);
this.postProcess = postProcess;
Utilities.enforce(outputType.elementType.sameType(postProcess.getResultType()));
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -32,7 +32,7 @@ public DBSPAggregateLinearPostprocessRetainKeysOperator(
DBSPClosureExpression retainKeysFunction,
OutputPort left, OutputPort right) {
super(node, "aggregate_linear_postprocess_retain_keys",
function, outputType, false, left, right);
function, outputType, left, right);
this.postProcess = postProcess;
this.retainKeysFunction = retainKeysFunction;
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -47,7 +47,7 @@ public DBSPAggregateOperator(
DBSPTypeIndexedZSet outputType,
@Nullable DBSPAggregator function,
@Nullable DBSPAggregateList aggregate, OutputPort input) {
super(node, "aggregate", outputType, function, aggregate, false, input);
super(node, "aggregate", outputType, function, aggregate, input);
Utilities.enforce(input.getOutputIndexedZSetType().keyType.sameType(outputType.keyType));
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -25,9 +25,8 @@ protected DBSPAggregateOperatorBase(CalciteRelNode node, String operation,
DBSPTypeIndexedZSet outputType,
@Nullable DBSPAggregator function,
@Nullable DBSPAggregateList aggregateList,
boolean multiset,
OutputPort source) {
super(node, operation, function, outputType, multiset, source);
super(node, operation, function, outputType, source);
this.aggregateList = aggregateList;
// There are really two different representations of an aggregate operator,
// which reuse the same classes: a high-level one, which contains an Aggregate,
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -51,8 +51,7 @@ public class DBSPAggregateZeroOperator extends DBSPUnaryOperator implements ILin
*/
public DBSPAggregateZeroOperator(
CalciteRelNode node, DBSPExpression zero, OutputPort source) {
super(node, "aggregate_zero", zero, TypeCompiler.makeZSet(zero.getType()),
false, source);
super(node, "aggregate_zero", zero, TypeCompiler.makeZSet(zero.getType()), source);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -18,7 +18,7 @@
* in the right stream. */
public final class DBSPAntiJoinOperator extends DBSPBinaryOperator implements IJoin, IIncremental {
public DBSPAntiJoinOperator(CalciteRelNode node, OutputPort left, OutputPort right) {
super(node, "antijoin", null, left.outputType(), left.isMultiset(), left , right);
super(node, "antijoin", null, left.outputType(), left, right);
// Inputs must be indexed
left.getOutputIndexedZSetType();
right.getOutputIndexedZSetType();
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -23,7 +23,7 @@
public final class DBSPApply2Operator extends DBSPBinaryOperator implements ILinear {
public DBSPApply2Operator(CalciteRelNode node, DBSPClosureExpression function,
OutputPort left, OutputPort right) {
super(node, "apply2", function, function.getResultType(), false, left, right);
super(node, "apply2", function, function.getResultType(), left, right);
Utilities.enforce(function.parameters.length == 2,
() -> "Expected 2 parameters for function " + function);
DBSPType param0Type = function.parameters[0].getType().deref();
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -24,7 +24,7 @@
public final class DBSPApplyNOperator extends DBSPSimpleOperator implements ILinear {
public DBSPApplyNOperator(CalciteRelNode node, DBSPClosureExpression function,
List<OutputPort> inputs) {
super(node, "apply_n", function, function.getResultType(), false );
super(node, "apply_n", function, function.getResultType());
Utilities.enforce(!inputs.isEmpty());
for (var input: inputs)
this.addInput(input);
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -36,7 +36,7 @@ public static void noZsets(DBSPType type) {
public DBSPApplyOperator(CalciteRelNode node, DBSPClosureExpression function,
DBSPType outputType, OutputPort input, @Nullable String comment) {
super(node, "apply", function, outputType, false, input, comment);
super(node, "apply", function, outputType, input, comment);
Utilities.enforce(function.parameters.length == 1,
() -> "Expected 1 parameter for function " + function);
DBSPType paramType = function.parameters[0].getType().deref();
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -40,7 +40,6 @@ public final class DBSPAsofJoinOperator extends DBSPJoinBaseOperator implements
* @param leftTimestampIndex Index of left timestamp in left tuple.
* @param rightTimestampIndex Index of right timestamp in right tuple
* @param comparator Function that compares two timestamps.
* @param isMultiset True if output is a multiset
* @param isLeft True if this is a left join
* @param left Left input
* @param right Right input
Expand All
@@ -50,9 +49,9 @@ public DBSPAsofJoinOperator(CalciteRelNode node, DBSPTypeZSet outputType,
int leftTimestampIndex,
int rightTimestampIndex,
DBSPExpression comparator,
boolean isMultiset, boolean isLeft,
boolean isLeft,
OutputPort left, OutputPort right) {
super(node, "asof_join_abstract", function, outputType, isMultiset, left, right, false);
super(node, "asof_join_abstract", function, outputType, left, right, false);
this.isLeft = isLeft;
this.comparator = comparator;
this.leftTimestampIndex = leftTimestampIndex;
Expand Down
Expand Up
@@ -101,7 +100,7 @@ public DBSPSimpleOperator with(
return new DBSPAsofJoinOperator(this.getRelNode(),
outputType.to(DBSPTypeZSet.class), Objects.requireNonNull(function),
this.leftTimestampIndex, this.rightTimestampIndex,
this.comparator, this.isMultiset, this. isLeft, newInputs.get(0), newInputs.get(1))
this.comparator, this.isLeft, newInputs.get(0), newInputs.get(1))
.copyAnnotations(this);
}
return this;
Expand Down
Expand Up
@@ -130,7 +129,7 @@ public static DBSPAsofJoinOperator fromJson(JsonNode node, JsonDecoder decoder)
return new DBSPAsofJoinOperator(
CalciteEmptyRel.INSTANCE, info.getZsetType(), info.getFunction(),
leftTimestampIndex, rightTimestampIndex, comparator,
info.isMultiset(), isLeft, info.getInput(0), info.getInput(1))
isLeft, info.getInput(0), info.getInput(1))
.addAnnotations(info.annotations(), DBSPAsofJoinOperator.class);
}
}
Original file line number
Diff line number
Diff line change
Expand Up
@@ -19,7 +19,7 @@
* different: it guarantees that successors will not see partial results within a transaction sub-step */
public final class DBSPAtomicSumOperator extends DBSPSimpleOperator implements ILinear {
public DBSPAtomicSumOperator(CalciteRelNode node, List<OutputPort> inputs) {
super(node, "atomic_sum", null, inputs.get(0).outputType(), true );
super(node, "atomic_sum", null, inputs.get(0).outputType());
for (OutputPort op: inputs) {
this.addInput(op);
if (!op.outputType().sameType(this.outputType)) {
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -24,7 +24,7 @@ public final class DBSPBinaryDistinctOperator extends DBSPBinaryOperator
public final boolean positive;
public DBSPBinaryDistinctOperator(CalciteRelNode node, OutputPort integral, OutputPort delta, boolean positive) {
super(node, "distinct_component", null, delta.outputType(), false, integral, delta);
super(node, "distinct_component", null, delta.outputType(), integral, delta);
this.positive = positive;
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -11,8 +11,8 @@
public abstract class DBSPBinaryOperator extends DBSPSimpleOperator {
protected DBSPBinaryOperator(CalciteRelNode node, String operation,
@Nullable DBSPExpression function, DBSPType outputType,
boolean isMultiset, OutputPort left, OutputPort right) {
super(node, operation, function, outputType, isMultiset );
OutputPort left, OutputPort right) {
super(node, operation, function, outputType);
this.addInput(left);
this.addInput(right);
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -25,7 +25,7 @@ public class DBSPChainAggregateOperator extends DBSPUnaryOperator implements ILi
public DBSPChainAggregateOperator(CalciteRelNode node, DBSPClosureExpression init,
DBSPClosureExpression function, DBSPType outputType, OutputPort source) {
super(node, "chain_aggregate", function, outputType, false, source);
super(node, "chain_aggregate", function, outputType, source);
this.init = init;
Utilities.enforce(init.parameters.length == 2);
Utilities.enforce(function.parameters.length == 3);
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -43,8 +43,8 @@
public class DBSPChainOperator extends DBSPUnaryOperator implements ILinear {
public final ComputationChain chain;
public DBSPChainOperator(CalciteRelNode node, ComputationChain chain, boolean isMultiset, OutputPort source) {
super(node, "chain" + chain.summary(), null, chain.getOutputType(), isMultiset, source);
public DBSPChainOperator(CalciteRelNode node, ComputationChain chain, OutputPort source) {
super(node, "chain" + chain.summary(), null, chain.getOutputType(), source);
this.chain = chain;
Utilities.enforce(this.chain.size() > 1);
}
Expand Down
Expand Up
@@ -338,7 +338,7 @@ public DBSPSimpleOperator with(
List<OutputPort> newInputs, boolean force) {
if (this.mustReplace(force, function, newInputs, outputType)) {
Utilities.enforce(function == null);
return new DBSPChainOperator(this.getRelNode(), this.chain, this.isMultiset, newInputs.get(0))
return new DBSPChainOperator(this.getRelNode(), this.chain, newInputs.get(0))
.copyAnnotations(this);
}
return this;
Expand Down
Expand Up
@@ -379,7 +379,7 @@ public static DBSPChainOperator fromJson(JsonNode node, JsonDecoder decoder) {
CommonInfo info = DBSPSimpleOperator.commonInfoFromJson(node, decoder);
ComputationChain chain = ComputationChain.fromJson(node, decoder);
return new DBSPChainOperator(CalciteEmptyRel.INSTANCE,
chain, info.isMultiset(), info. getInput(0))
chain, info.getInput(0))
.addAnnotations(info.annotations(), DBSPChainOperator.class);
}
}
Original file line number
Diff line number
Diff line change
Expand Up
@@ -38,7 +38,6 @@ public final class DBSPConcreteAsofJoinOperator extends DBSPJoinBaseOperator imp
* @param leftTimestamp Function that extracts a "timestamp" from the left input
* @param rightTimestamp Function that extracts a "timestamp" from the right input
* @param comparator Function that compares two timestamps. Not yet used - hardwired.
* @param isMultiset True if output is a multiset
* @param isLeft True if this is a left join
* @param left Left input
* @param right Right input
Expand All
@@ -48,9 +47,9 @@ public DBSPConcreteAsofJoinOperator(CalciteRelNode node, DBSPTypeZSet outputType
DBSPClosureExpression leftTimestamp,
DBSPClosureExpression rightTimestamp,
DBSPExpression comparator,
boolean isMultiset, boolean isLeft,
boolean isLeft,
OutputPort left, OutputPort right) {
super(node, "asof_join", function, outputType, isMultiset, left, right, false);
super(node, "asof_join", function, outputType, left, right, false);
this.isLeft = isLeft;
this.comparator = comparator;
this.leftTimestamp = leftTimestamp;
Expand Down
Expand Up
@@ -107,7 +106,7 @@ public DBSPSimpleOperator with(
return new DBSPConcreteAsofJoinOperator(
this.getRelNode(), this.getOutputZSetType(),
Objects.requireNonNull(function), this.leftTimestamp, this.rightTimestamp,
this.comparator, this.isMultiset, this. isLeft,
this.comparator, this.isLeft,
newInputs.get(0), newInputs.get(1))
.copyAnnotations(this);
}
Expand Down
Expand Up
@@ -137,7 +136,7 @@ public static DBSPConcreteAsofJoinOperator fromJson(JsonNode node, JsonDecoder d
return new DBSPConcreteAsofJoinOperator(
CalciteEmptyRel.INSTANCE, info.getZsetType(), info.getFunction(),
leftTimestamp, rightTimestamp, comparator,
info.isMultiset(), isLeft, info.getInput(0), info.getInput(1))
isLeft, info.getInput(0), info.getInput(1))
.addAnnotations(info.annotations(), DBSPConcreteAsofJoinOperator.class);
}
}
Original file line number
Diff line number
Diff line change
Expand Up
@@ -39,10 +39,10 @@
import org.dbsp.sqlCompiler.ir.type.user.StreamKind;
public final class DBSPConstantOperator extends DBSPSimpleOperator {
public DBSPConstantOperator(CalciteRelNode node, DBSPExpression value, boolean isMultiset ) {
public DBSPConstantOperator(CalciteRelNode node, DBSPExpression value) {
// Notice that we use the 'this.function' field to represent
// the constant value. Constants are not ClosureExpressions.
super(node, "constant", value, value.getType(), isMultiset );
super(node, "constant", value, value.getType());
}
@Override
Expand All
@@ -59,7 +59,7 @@ public DBSPSimpleOperator with(
@Nullable DBSPExpression function, DBSPType outputType,
List<OutputPort> newInputs, boolean force) {
if (this.mustReplace(force, function, newInputs, outputType)) {
return new DBSPConstantOperator(this.getRelNode(), Objects.requireNonNull(function), this.isMultiset )
return new DBSPConstantOperator(this.getRelNode(), Objects.requireNonNull(function))
.copyAnnotations(this);
}
return this;
Expand All
@@ -69,7 +69,7 @@ public DBSPSimpleOperator with(
public static DBSPConstantOperator fromJson(JsonNode node, JsonDecoder decoder) {
CommonInfo info = DBSPSimpleOperator.commonInfoFromJson(node, decoder);
return new DBSPConstantOperator(
CalciteEmptyRel.INSTANCE, info.getFunction(), info.isMultiset() )
CalciteEmptyRel.INSTANCE, info.getFunction())
.addAnnotations(info.annotations(), DBSPConstantOperator.class);
}
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -34,12 +34,12 @@ static DBSPExpression function(CalciteObject node, DBSPType inputType) {
public DBSPDeindexOperator(CalciteRelNode node, CalciteObject functionNode, OutputPort input) {
super(node, "map", function(functionNode, input.outputType()),
outputType(input.getOutputIndexedZSetType()), true, input);
outputType(input.getOutputIndexedZSetType()), input);
}
/** Rebuilds a decoded operator with its serialized function. */
DBSPDeindexOperator(CalciteRelNode node, DBSPClosureExpression function, DBSPType outputType, OutputPort input) {
super(node, "map", function, outputType, true, input);
super(node, "map", function, outputType, input);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -20,7 +20,7 @@
public final class DBSPDelayOperator extends DBSPUnaryOperator implements IStateful, ILinear {
public DBSPDelayOperator(CalciteRelNode node, @Nullable DBSPExpression initial, OutputPort source) {
super(node, initial == null ? "transaction_delay" : "transaction_delay_with_initial_value",
initial, source.outputType(), source.isMultiset(), source );
initial, source.outputType(), source);
if (initial != null && !initial.getType().sameType(source.outputType()))
throw new InternalCompilerError("Delay input has type " + source.outputType() +
" but initial value has type " + initial.getType());
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -20,7 +20,7 @@
@NonCoreIR
public final class DBSPDelayedIntegralOperator extends DBSPUnaryOperator implements IContainsIntegrator {
public DBSPDelayedIntegralOperator(CalciteRelNode node, OutputPort source) {
super(node, "accumulate_delay_trace", null, source.outputType(), source.isMultiset(), source );
super(node, "accumulate_delay_trace", null, source.outputType(), source);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -17,7 +17,7 @@
/** Represents a delta operator (called delta0 in DBSP) */
public class DBSPDeltaOperator extends DBSPUnaryOperator implements ILinear {
public DBSPDeltaOperator(CalciteRelNode node, OutputPort source) {
super(node, "delta0", null, source.outputType(), source.isMultiset(), source );
super(node, "delta0", null, source.outputType(), source);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -39,7 +39,7 @@
public final class DBSPDifferentiateOperator extends DBSPUnaryOperator implements IStateful, ILinear {
public DBSPDifferentiateOperator(CalciteRelNode node, OutputPort source) {
super(node, "differentiate", null, source.outputType(), source.isMultiset(), source );
super(node, "differentiate", null, source.outputType(), source);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -39,7 +39,7 @@
public final class DBSPDistinctOperator extends DBSPUnaryOperator
implements IHasInputIntegrator, IIncremental {
public DBSPDistinctOperator(CalciteRelNode node, OutputPort input) {
super(node, "distinct", null, input.outputType(), false, input);
super(node, "distinct", null, input.outputType(), input);
}
@Override
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -40,7 +40,7 @@
public final class DBSPFilterOperator extends DBSPUnaryOperator implements ILinear {
public DBSPFilterOperator(CalciteRelNode node, DBSPExpression condition, OutputPort input) {
super(node, "filter", condition, input.outputType(), input.isMultiset(), input );
super(node, "filter", condition, input.outputType(), input);
this.checkResultType(condition, new DBSPTypeBool(CalciteEmptyRel.INSTANCE, false));
checkArgumentFunctionType(condition, input);
}
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.