Skip to content

[SQL] GC of tables with keys and LATENESS should never use retain_values, but can use retain_keys when keys have LATENESS #6693

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
base: main
Choose a base branch
from
+689 −56
Open

[SQL] GC of tables with keys and LATENESS should never use retain_values, but can use retain_keys when keys have LATENESS #6693

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
14 changes: 4 additions & 10 deletions crates/dbsp/src/operator/dynamic/input.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -781,8 +781,6 @@ impl RootCircuit {
///
/// Retention conditions configured at logical time `t`
/// are applied starting from logical time `t+1`.
///
/// FIXME: see <https://github.com/feldera/feldera/issues/2669>
// TODO: Add a version that takes a custom hash function.
#[track_caller]
pub fn dyn_add_input_map<K, V, U>(
Expand Down Expand Up @@ -1970,10 +1968,8 @@ mod test {
Ok(handle)
}

// FIXME: the inputs to these tests are meant to exercise the logic that filters inputs based
// on lateness, but it does not currently work correctly (see https://github.com/feldera/feldera/issues/2669).
// We therefore don't use waterlines in tests and check for the standard upsert behavior
// without filtering.
// These tests check the standard upsert behavior without lateness-based filtering;
// filtering inputs based on lateness is implemented by `add_input_map_with_waterline`.
#[test]
fn map_test_st() {
let (mut circuit, mut input_handle) = Runtime::init_circuit(1, move |circuit| {
Expand Down Expand Up @@ -2033,10 +2029,8 @@ mod test {
dbsp.kill().unwrap();
}

// FIXME: the inputs to these tests are meant to exercise the logic that filters inputs based
// on lateness, but it does not currently work correctly (see https://github.com/feldera/feldera/issues/2669).
// We therefore don't use waterlines in tests and check for the standard upsert behavior
// without filtering.
// These tests check the standard upsert behavior without lateness-based filtering;
// filtering inputs based on lateness is implemented by `add_input_map_with_waterline`.
#[test]
fn map_test_mt1() {
map_test_mt(1, input_map_updates1, output_map_updates1);
Expand Down
2 changes: 0 additions & 2 deletions crates/dbsp/src/operator/input.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -606,8 +606,6 @@ impl RootCircuit {
///
/// Retention conditions configured at logical time `t`
/// are applied starting from logical time `t+1`.
///
/// FIXME: see <https://github.com/feldera/feldera/issues/2669>
// TODO: Add a version that takes a custom hash function.
#[track_caller]
pub fn add_input_map<K, V, U, PF>(
Expand Down
26 changes: 20 additions & 6 deletions ...main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainKeysOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,16 @@
public final class DBSPIntegrateTraceRetainKeysOperator
extends DBSPBinaryOperator implements IGCOperator
{
/** Selects the Rust function that the operator invokes.
* GC applied to an input table requires 'false'; everything else requires 'true'. */
public final boolean accumulate;

public DBSPIntegrateTraceRetainKeysOperator(
CalciteRelNode node, DBSPExpression expression,
OutputPort data, OutputPort control) {
super(node, "accumulate_integrate_trace_retain_keys", expression,
data.outputType(), data.isMultiset(), data, control);
OutputPort data, OutputPort control, boolean accumulate) {
super(node, accumulate ? "accumulate_integrate_trace_retain_keys" : "integrate_trace_retain_keys",
expression, data.outputType(), data.isMultiset(), data, control);
this.accumulate = accumulate;
}

@Override
Expand All @@ -43,6 +48,14 @@ public DBSPSimpleOperator asOperator() {
@Nullable
public static DBSPIntegrateTraceRetainKeysOperator create(
CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) {
return create(node, data, dataProjection, control, true);
}

/** Create an operator to retain keys and returns it. May return null if the keys contain no fields. */
@Nullable
public static DBSPIntegrateTraceRetainKeysOperator create(
CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control,
boolean accumulate) {
DBSPType controlType = control.outputType();
Utilities.enforce(controlType.is(DBSPTypeTupleBase.class), () -> "Control type is not a tuple: " + controlType);
DBSPTypeTupleBase controlTuple = controlType.to(DBSPTypeTupleBase.class);
Expand Down Expand Up @@ -83,7 +96,7 @@ public static DBSPIntegrateTraceRetainKeysOperator create(
compare = ExpressionCompiler.makeBinaryExpression(
node, compare.getType(), DBSPOpcode.OR, compare0, compare);
DBSPExpression closure = compare.closure(param, controlArg.asParameter());
return new DBSPIntegrateTraceRetainKeysOperator(node, closure, data, control);
return new DBSPIntegrateTraceRetainKeysOperator(node, closure, data, control, accumulate);
}

@Override
Expand All @@ -94,7 +107,7 @@ public DBSPSimpleOperator with(
Utilities.enforce(newInputs.size() == 2, () -> "Expected 2 inputs, got " + newInputs.size());
return new DBSPIntegrateTraceRetainKeysOperator(
this.getRelNode(), toClosure(function),
newInputs.get(0), newInputs.get(1)).copyAnnotations(this);
newInputs.get(0), newInputs.get(1), this.accumulate).copyAnnotations(this);
}
return this;
}
Expand All @@ -113,8 +126,9 @@ public void accept(CircuitVisitor visitor) {
@SuppressWarnings("unused")
public static DBSPIntegrateTraceRetainKeysOperator fromJson(JsonNode node, JsonDecoder decoder) {
DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder);
boolean accumulate = Utilities.getBooleanProperty(node, "accumulate");
return new DBSPIntegrateTraceRetainKeysOperator(CalciteEmptyRel.INSTANCE,
info.getFunction(), info.getInput(0), info.getInput(1))
info.getFunction(), info.getInput(0), info.getInput(1), accumulate)
.addAnnotations(info.annotations(), DBSPIntegrateTraceRetainKeysOperator.class);
}
}
22 changes: 6 additions & 16 deletions ...in/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainValuesOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,14 +27,11 @@
public final class DBSPIntegrateTraceRetainValuesOperator
extends DBSPBinaryOperator implements IGCOperator {

public final boolean accumulate;

public DBSPIntegrateTraceRetainValuesOperator(
CalciteRelNode node, DBSPExpression function,
OutputPort data, OutputPort control, boolean accumulate) {
super(node, accumulate ? "accumulate_integrate_trace_retain_values" : "integrate_trace_retain_values",
OutputPort data, OutputPort control) {
super(node, "accumulate_integrate_trace_retain_values",
function, data.outputType(), data.isMultiset(), data, control);
this.accumulate = accumulate;
}

@Override
Expand All @@ -43,7 +40,7 @@ public DBSPSimpleOperator asOperator() {
}

public static DBSPIntegrateTraceRetainValuesOperator create(
CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control, boolean accumulate) {
CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) {
DBSPType controlType = control.outputType();
Utilities.enforce(controlType.is(DBSPTypeTupleBase.class),
() -> "Control type is not a tuple: " + controlType);
Expand All @@ -66,14 +63,9 @@ public static DBSPIntegrateTraceRetainValuesOperator create(
compare = ExpressionCompiler.makeBinaryExpression(
node, compare.getType(), DBSPOpcode.OR, compare0, compare);
DBSPExpression closure = compare.closure(param, controlArg.asParameter());
return new DBSPIntegrateTraceRetainValuesOperator(node, closure, data, control, accumulate);
return new DBSPIntegrateTraceRetainValuesOperator(node, closure, data, control);
}

public static DBSPIntegrateTraceRetainValuesOperator create(
CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) {
return create(node, data, dataProjection, control, true);
}

@Override
public DBSPSimpleOperator with(
@Nullable DBSPExpression function, DBSPType outputType,
Expand All @@ -82,7 +74,7 @@ public DBSPSimpleOperator with(
Utilities.enforce(newInputs.size() == 2, () -> "Expected 2 inputs, got " + newInputs.size());
return new DBSPIntegrateTraceRetainValuesOperator(
this.getRelNode(), Objects.requireNonNull(function),
newInputs.get(0), newInputs.get(1), this.accumulate);
newInputs.get(0), newInputs.get(1));
}
return this;
}
Expand All @@ -101,10 +93,8 @@ public void accept(CircuitVisitor visitor) {
@SuppressWarnings("unused")
public static DBSPIntegrateTraceRetainValuesOperator fromJson(JsonNode node, JsonDecoder decoder) {
DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder);
boolean accumulate = Utilities.getBooleanProperty(node, "accumulate");

return new DBSPIntegrateTraceRetainValuesOperator(CalciteEmptyRel.INSTANCE,
info.getFunction(), info.getInput(0), info.getInput(1), accumulate)
info.getFunction(), info.getInput(0), info.getInput(1))
.addAnnotations(info.annotations(), DBSPIntegrateTraceRetainValuesOperator.class);
}
}
4 changes: 2 additions & 2 deletions .../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import org.dbsp.sqlCompiler.circuit.operator.DBSPUnaryOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPViewBaseOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPWindowOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainValuesOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainKeysOperator;
import org.dbsp.sqlCompiler.circuit.operator.DBSPWeightValidatorOperator;
import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ProgramIdentifier;
Expand Down Expand Up @@ -337,7 +337,7 @@ public VisitDecision preorder(DBSPWindowOperator operator) {
}

@Override
public VisitDecision preorder(DBSPIntegrateTraceRetainValuesOperator operator) {
public VisitDecision preorder(DBSPIntegrateTraceRetainKeysOperator operator) {
if (this.preorder(operator.to(DBSPBinaryOperator.class)).stop())
return VisitDecision.STOP;
this.property("accumulate");
Expand Down
4 changes: 2 additions & 2 deletions ...-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -722,7 +722,7 @@ public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) {
|| Linq.different(sources, operator.inputs)
|| function != operator.getFunction()) {
result = new DBSPIntegrateTraceRetainKeysOperator(operator.getRelNode(), function,
sources.get(0), sources.get(1))
sources.get(0), sources.get(1), operator.accumulate)
.copyAnnotations(operator);
}
this.map(operator, result);
Expand All @@ -738,7 +738,7 @@ public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) {
|| Linq.different(sources, operator.inputs)
|| function != operator.getFunction()) {
result = new DBSPIntegrateTraceRetainValuesOperator(operator.getRelNode(), function,
sources.get(0), sources.get(1), operator.accumulate)
sources.get(0), sources.get(1))
.copyAnnotations(operator);
}
this.map(operator, result);
Expand Down
21 changes: 21 additions & 0 deletions ...iler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/StrayGC.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,22 @@ public StrayGC(DBSPCompiler compiler, CircuitGraphs g) {
super(compiler, g);
}

/** Check that the retain operator invokes the runtime function variant that
* matches its data source: input tables require the non-accumulate variant,
* everything else the accumulate_ one. The wrong variant makes the
* operator a silent no-op. */
void checkAccumulate(DBSPBinaryOperator operator, boolean accumulate) {
boolean input = operator.left().operator.is(DBSPInputMapWithWaterlineOperator.class);
if (input && accumulate)
throw new InternalCompilerError(
"Operator " + operator + " garbage-collects an input table " +
"and must use the non-accumulate variant");
if (!input && !accumulate)
throw new InternalCompilerError(
"Operator " + operator + " garbage-collects a trace inside the " +
"accumulate framework and must use the accumulate_ variant");
}

void check(DBSPBinaryOperator operator) {
// At least one sibling on the left input must contain an integral
var left = operator.left();
Expand All @@ -45,16 +61,21 @@ void check(DBSPBinaryOperator operator) {

@Override
public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) {
// This operator always uses the accumulate_ variant
this.checkAccumulate(operator, true);
this.check(operator);
}

@Override
public void postorder(DBSPIntegrateTraceRetainNValuesOperator operator) {
// This operator always uses the accumulate_ variant
this.checkAccumulate(operator, true);
this.check(operator);
}

@Override
public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) {
this.checkAccumulate(operator, operator.accumulate);
DBSPOperator left = operator.left().operator;
if (left.is(DBSPAggregateLinearPostprocessRetainKeysOperator.class) ||
left.is(DBSPChainAggregateOperator.class) ||
Expand Down
61 changes: 50 additions & 11 deletions ...c/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1741,19 +1741,58 @@ DBSPOperator processLateness(
if (operator != expansion)
this.markBound(expansion.outputPort(), extend.outputPort());

if (INSERT_RETAIN_VALUES && replaceIndexedInput &&
if (INSERT_RETAIN_KEYS && replaceIndexedInput &&
multisetInput != null &&
// Do not GC materialized tables.
!multisetInput.getMetadata().materialized) {
// Do not GC materialized tables
IMaybeMonotoneType projection = me.getMonotoneType().to(MonotoneClosureType.class).getBodyType();
// The new input operator produces an indexed Zset, need to adjust the projection
projection = new PartiallyMonotoneTuple(
List.of(NonMonotoneType.nonMonotone(indexedOutputType.keyType), projection), true, false);
DBSPSimpleOperator retain = DBSPIntegrateTraceRetainValuesOperator.create(
operator.getRelNode(), newSource.getOutput(0),
// For inputs "accumulate" is 'false'.
projection, extend.outputPort(), false);
this.addOperator(retain);
// The trace of an input with a primary key resolves upserts and deletions,
// so it can be garbage-collected only by key, and only using LATENESS
// columns that belong to the key: a key pruned this way can never be
// legally referenced again, because any subsequent update or deletion
// carries the same key and is rejected as late.
// A LATENESS column outside the key permits no garbage collection at all:
// - a live row (timestamp at or above the waterline) must be retained
// to produce the retraction when its key is updated or deleted;
// - a frozen row (timestamp below the waterline) can never be retracted,
// but its presence is what lets the input operator reject writes to
// its key as late; if the row is dropped, the next insert on that key
// is accepted as a brand-new key without a retraction, so the output
// would depend on when compaction runs.
List<IMaybeMonotoneType> keyFieldsMonotone = new ArrayList<>();
List<Integer> keyWaterlineFields = new ArrayList<>();
int latenessIndex = 0;
for (IColumnMetadata column: multisetInput.to(IHasColumnsMetadata.class).getColumnsMetadata()) {
boolean hasLateness = column.getLateness() != null;
if (column.isPrimaryKey()) {
if (hasLateness) {
keyFieldsMonotone.add(new MonotoneType(column.getType()));
keyWaterlineFields.add(latenessIndex);
} else {
keyFieldsMonotone.add(NonMonotoneType.nonMonotone(column.getType()));
}
}
if (hasLateness)
latenessIndex++;
}
if (!keyWaterlineFields.isEmpty()) {
PartiallyMonotoneTuple keyProjection = new PartiallyMonotoneTuple(keyFieldsMonotone, false, false);
IMaybeMonotoneType projection = new PartiallyMonotoneTuple(
List.of(keyProjection, NonMonotoneType.nonMonotone(indexedOutputType.elementType)),
true, false);
// Project the waterline to the components that correspond to key columns
DBSPVariablePath w = timestamp.getType().ref().var();
DBSPExpression keyWaterline = new DBSPRawTupleExpression(
new DBSPTupleExpression(
Linq.map(keyWaterlineFields, f -> w.deepCopy().deref().field(f)),
false));
OutputPort control = this.createApply(extend.outputPort(), null, keyWaterline.closure(w));
DBSPSimpleOperator retain = DBSPIntegrateTraceRetainKeysOperator.create(
operator.getRelNode(), newSource.getOutput(0),
// For inputs "accumulate" is 'false'.
projection, control, false);
if (retain != null)
this.addOperator(retain);
}
}

return result;
Expand Down
2 changes: 1 addition & 1 deletion ...iler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/MergeGC.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,7 @@ DBSPIntegrateTraceRetainKeysOperator merge(List<DBSPIntegrateTraceRetainKeysOper
OutputPort apply = InsertLimiters.createApplyN(this.compiler, rights, min);
this.addOperator(apply.node());
return new DBSPIntegrateTraceRetainKeysOperator(
first.getRelNode(), first.getClosureFunction(), left, apply);
first.getRelNode(), first.getClosureFunction(), left, apply, first.accumulate);
}

public MergeRetain(DBSPCompiler compiler, FindMultipleRetainKeys fmk) {
Expand Down
6 changes: 4 additions & 2 deletions ...p-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/CatalogTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,9 @@ PARTITION BY customer_id, DATE_TRUNC(timestamp_column, MONTH)
int imww = 0;
int retain = 0;

// Check that InputMapWithWaterline is GC-ed
// Check that InputMapWithWaterline is NOT GC-ed: the LATENESS
// column is not part of the primary key, so every key can still be
// updated and the input trace must retain all rows.
@Override
public void postorder(DBSPIntegrateTraceRetainValuesOperator node) {
this.retain++;
Expand All @@ -254,7 +256,7 @@ public void postorder(DBSPInputMapWithWaterlineOperator operator) {
@Override
public void endVisit() {
Assert.assertEquals(1, this.imww);
Assert.assertEquals(1, this.retain);
Assert.assertEquals(0, this.retain);
}
});
}
Expand Down
4 changes: 1 addition & 3 deletions ...r/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegression2Tests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -744,16 +744,14 @@ public void issue2808() {
END
)
FROM T GROUP BY id, bid;
""");
""").compactAfterEachStep();
ccs.stepWeightOne("", """
id | bid | sum
----------------""");
ccs.blockForCompaction();
ccs.stepWeightOne("INSERT INTO T VALUES(1, 1, 0, 0, 0);", """
id | bid | sum
----------------
1 | 1 | 1""");
ccs.blockForCompaction();
}

@Test
Expand Down
Loading