feldera · mihaibudiu · Aug 4, 2026 · Jul 22, 2026 · Jul 22, 2026 · Jul 23, 2026
diff --git a/docs.feldera.com/docs/connectors/index.mdx b/docs.feldera.com/docs/connectors/index.mdx
@@ -260,39 +260,13 @@ message that deletes the old value of a record and inserts the new one, so
 those two changes carry one timestamp; ranking the insertion first keeps the
 updated record, where ranking the deletion first would drop it.
 
-{/* Revise after implementing the append-only optimization for soft-delete tables.
-
-#### Bounding the state of the query
-
 As written, `live` needs unbounded state. A change carrying any timestamp can
 arrive at any moment and displace the record that is currently the latest one
 for its key, so Feldera has to keep every insertion and deletion the stream
-ever reported.
-
-An application that only needs values from a bounded time frame can filter the
-changes with a
-[temporal filter](/tutorials/time-series#now-and-temporal-filters) before
-ranking them, which leaves Feldera storing only the changes newer than the
-lower bound of the interval:
-
-```sql
-CREATE MATERIALIZED VIEW live_recent AS
-SELECT id, s, ts
-FROM (
-    SELECT *, ROW_NUMBER() OVER (
-        PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST
-    ) AS rn
-    FROM changes
-    WHERE ts >= NOW() - INTERVAL 7 DAYS
-)
-WHERE rn = 1 AND is_delete IS NOT TRUE;
-```
-
-A record whose latest change ages out of the window leaves `live_recent`, so
-this variant reports what is live among the keys the stream touched in the last
-seven days rather than what is live overall.
-
-*/}
+ever reported. An application that only needs values from a bounded time frame
+can filter the changes with a temporal filter before ranking them, which
+bounds the state of the query; see
+[Soft deletes with temporal filters](/sql/streaming#soft-deletes-with-temporal-filters).
 
 Notes and restrictions:
 

diff --git a/docs.feldera.com/docs/sql/streaming.md b/docs.feldera.com/docs/sql/streaming.md
@@ -54,4 +54,114 @@ The `emit_final` annotation on a view instructs Feldera to only output its final
 i.e., rows that are guaranteed to never get deleted or updated.
 
 See the [Time Series Analysis Guide](/tutorials/time-series#emitting-final-values-of-a-view-with-emit_final)
-for details.
+for details.
+
+## Soft deletes with temporal filters
+
+An input connector can be configured with the [`soft_delete`](/connectors#soft_delete)
+property to transform deletions into insertions; in this case the `is_delete` metadata
+attribute records the kind of change (insert/delete), essentially converting a table
+into a log.  Note that the table does *not* declare a `PRIMARY KEY` column, although
+the data may contain one.  Since the connector transforms every change into an
+insertion, the table only receives insertions, so it can be declared
+[`append_only`](#append_only-tables), which enables additional optimizations.
+
+The [Soft deletes](/connectors#soft-deletes) section shows how one can write a query
+to recover the current contents of the table from this log: group the changes
+on the columns forming the primary key, rank them by time, keep the latest one,
+and return it only when it is an insertion.  That query returns one row for each
+primary key, but it must remember every change in the log, so its state grows
+without bound.
+
+However, in some cases only a bounded window of the table is necessary for
+computing the desired results.  When a
+[temporal filter](/tutorials/time-series#now-and-temporal-filters) can be used
+to describe the window, the entire computation can be performed using finite
+state, by sequencing the computation as follows:
+
+```
+[connector with soft deletes] -> [temporal filter] -> [reconstruct table] -> [views]
+```
+
+The following program reconstructs only the recent contents
+of a change stream while never storing more than the last seven
+days of changes:
+
+```sql
+-- The 'soft_delete' connector property converts this
+-- table into a log of changes to the table.
+CREATE TABLE input_log (
+    id BIGINT, -- not declared as primary key
+    s VARCHAR,
+    ts TIMESTAMP,
+    -- Is the change a deletion?  Produced by the connector
+    is_delete BOOLEAN DEFAULT CAST(CONNECTOR_METADATA()['is_delete'] AS BOOLEAN)
+) WITH (
+    -- A soft-delete table only receives insertions
+    'append_only' = 'true',
+    'connectors' = '[{
+        "name": "changes",
+        "soft_delete": true,
+        "transport": {
+            "name": "kafka_input",
+            "config": {
+                "topic": "changes",
+                "start_from": "earliest",
+                "bootstrap.servers": "example.com:9092",
+                "include_timestamp": true
+            }
+        },
+        "format": {
+            "name": "json",
+            "config": { "update_format": "insert_delete" }
+        }
+    }]'
+);
+
+-- Contains only changes to 'input_log' from the last 7 days
+CREATE LOCAL VIEW recent AS
+SELECT * FROM input_log
+WHERE ts >= NOW() - INTERVAL 7 DAYS AND ts <= NOW();
+
+-- The contents of the 'input' table limited to the last 7 days
+CREATE LOCAL VIEW input AS
+SELECT id, s, ts
+FROM (
+    SELECT *, ROW_NUMBER() OVER (
+        PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST
+    ) AS rn
+    FROM recent
+)
+WHERE rn = 1 AND is_delete IS NOT TRUE;
+
+-- Rolling aggregates over the reconstructed table: for each record,
+-- the number of records with a timestamp in the preceding minute, hour, and day.
+CREATE VIEW input_stats AS
+SELECT
+    id, s, ts,
+    COUNT(*) OVER minute_window AS rows_last_minute,
+    COUNT(*) OVER hour_window AS rows_last_hour,
+    COUNT(*) OVER day_window AS rows_last_day
+FROM input
+WINDOW
+    minute_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 MINUTE PRECEDING AND CURRENT ROW),
+    hour_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW),
+    day_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 DAY PRECEDING AND CURRENT ROW);
+```
+
+The view `input_stats` consumes the reconstructed table using
+[rolling aggregates](/tutorials/time-series#rolling-aggregates) over three
+shorter intervals.  The aggregates see the reconstructed table rather than the
+log, so a deleted record stops contributing to the counts as soon as its
+deletion arrives.
+
+Two details of the query matter for correctness:
+
+* The `is_delete` term in the `ORDER BY` clause ranks an insertion ahead of a
+  deletion that carries the same timestamp, which keeps the new value of a
+  record that a CDC stream updates with a single delete-insert message pair.
+  See [Soft deletes](/connectors#soft-deletes) for details.
+
+* The polarity filter `is_delete IS NOT TRUE` must be outside the subquery
+  that ranks the changes. Filtering out the deletions before ranking would
+  "resurrect" the previous insertion of a deleted key.
diff --git a/...rc/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java b/...rc/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java
@@ -30,7 +30,7 @@
  * with records that missed the lateness. */
 public class DBSPInputMapWithWaterlineOperator
         extends DBSPOperator
-        implements IMultiOutput, IInputMapOperator, IInputOperator {
+        implements IMultiOutput, IInputMapOperator, IInputOperator, IStateful {
     // Fields that belong normally to SourceTableOperators (which we don't derive from)
     public final ProgramIdentifier tableName;
     public final DBSPTypeStruct originalRowType;

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java
@@ -27,9 +27,13 @@
 public final class DBSPWindowOperator extends DBSPBinaryOperator implements IContainsIntegrator, IIncremental {
     public final boolean lowerInclusive;
     public final boolean upperInclusive;
+    /** True if the window's lower bound is -infinity.
+     * This property is not used for code generation, but it is difficult to infer later.
+     * The validity relies on the fact that no optimizations change window bounds once created. */
+    public final boolean lowerUnbounded;
 
     public DBSPWindowOperator(CalciteRelNode node, boolean lowerInclusive, boolean upperInclusive,
-            OutputPort data, OutputPort control) {
+            boolean lowerUnbounded, OutputPort data, OutputPort control) {
         super(node, "window", null, data.outputType(), data.isMultiset(), data, control);
         // Check that the left input and output are indexed ZSets.
         DBSPTypeIndexedZSet indexedType = this.getOutputIndexedZSetType();
@@ -41,6 +45,7 @@ public DBSPWindowOperator(CalciteRelNode node, boolean lowerInclusive, boolean u
                         ", but have type " + control.outputType());
         this.lowerInclusive = lowerInclusive;
         this.upperInclusive = upperInclusive;
+        this.lowerUnbounded = lowerUnbounded;
     }
 
     @Override
@@ -52,6 +57,7 @@ public DBSPSimpleOperator with(
             if (force || this.inputsDiffer(newInputs))
                 return new DBSPWindowOperator(
                         this.getRelNode(), this.lowerInclusive, this.upperInclusive,
+                        this.lowerUnbounded,
                         newInputs.get(0), newInputs.get(1)).copyAnnotations(this);
         }
         return this;
@@ -71,8 +77,9 @@ public static DBSPWindowOperator fromJson(JsonNode node, JsonDecoder decoder) {
         DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder);
         boolean lowerInclusive = Utilities.getBooleanProperty(node, "lowerInclusive");
         boolean upperInclusive = Utilities.getBooleanProperty(node, "upperInclusive");
+        boolean lowerUnbounded = Utilities.getBooleanProperty(node, "lowerUnbounded");
         return new DBSPWindowOperator(CalciteEmptyRel.INSTANCE,
-                lowerInclusive, upperInclusive, info.getInput(0), info.getInput(1))
+                lowerInclusive, upperInclusive, lowerUnbounded, info.getInput(0), info.getInput(1))
                 .addAnnotations(info.annotations(), DBSPWindowOperator.class);
     }
 }
diff --git a/...r/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java b/...r/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java
@@ -16,7 +16,7 @@
 import java.util.List;
 
 /** Interface for a source operator that has primary keys */
-public interface IInputMapOperator extends IInputOperator {
+public interface IInputMapOperator extends IInputOperator, IStateful {
     TableMetadata getMetadata();
     List<Integer> getKeyFields();
     DBSPTypeIndexedZSet getOutputIndexedZSetType();

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java
@@ -333,6 +333,8 @@ public VisitDecision preorder(DBSPWindowOperator operator) {
         this.stream.append(operator.lowerInclusive);
         this.property("upperInclusive");
         this.stream.append(operator.upperInclusive);
+        this.property("lowerUnbounded");
+        this.stream.append(operator.lowerUnbounded);
         return VisitDecision.CONTINUE;
     }
 

diff --git a/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java b/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java
@@ -24,8 +24,8 @@
 import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange;
 import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision;
 import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor;
+import org.dbsp.sqlCompiler.compiler.visitors.outer.FindSourcePositions;
 import org.dbsp.sqlCompiler.compiler.visitors.outer.LowerCircuitVisitor;
-import org.dbsp.sqlCompiler.compiler.visitors.outer.ToJsonVisitor;
 import org.dbsp.sqlCompiler.ir.IDBSPInnerNode;
 import org.dbsp.sqlCompiler.ir.IDBSPOuterNode;
 import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
@@ -153,7 +153,7 @@ static String escapeString(String input) {
 
     String getPositions(IDBSPInnerNode node) {
         StringBuilder result = new StringBuilder();
-        ToJsonVisitor.FindSourcePositions finder = new ToJsonVisitor.FindSourcePositions(this.compiler, true);
+        FindSourcePositions finder = new FindSourcePositions(this.compiler, true);
         finder.apply(node);
         for (SourcePositionRange r : finder.getPositions()) {
             result.append(r.toShortString());

diff --git a/...r/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java b/...r/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java
@@ -124,6 +124,15 @@ else if (i == endLine)
         return result.toString();
     }
 
+    public String getFragments(SourcePositionRanges ranges) {
+        StringBuilder result = new StringBuilder();
+        for (SourcePositionRange r: ranges.positions) {
+            result.append(this.getFragment(r, true));
+            result.append("\n");
+        }
+        return result.toString();
+     }
+
     public String getFragment(SqlNode node) {
         return this.getFragment(
                 new SourcePositionRange(node.getParserPosition()), false);

diff --git a/...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java b/...SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java
@@ -8,7 +8,7 @@
 
 /** A set of source positions */
 public class SourcePositionRanges implements Iterable<SourcePositionRange> {
-    final List<SourcePositionRange> positions;
+    public final List<SourcePositionRange> positions;
 
     public SourcePositionRanges(Iterable<SourcePositionRange> positions) {
         List<SourcePositionRange> pos = Linq.list(positions);

diff --git a/...ler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java b/...ler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java
@@ -37,9 +37,6 @@ public String toString() {
         return "";
     }
 
-    /** Format the object in a way that can displayed in an error message */
-    public String getMessage() { return this.toString(); }
-
     public String toInternalString() {
         return this.toString();
     }

diff --git a/...er/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java b/...er/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java
@@ -70,9 +70,6 @@ public static String toSqlString(RelNode node) {
         }
     }
 
-    @Override
-    public String getMessage() { return ""; }
-
     public abstract IIndentStream asJson(IIndentStream stream, Map<RelNode, Integer> idRemap);
 
     public abstract CalciteRelNode remove(RelNode node);

diff --git a/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java b/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java
@@ -32,6 +32,15 @@ public RelAnd() {
         this.nodes = new HashSet<>();
     }
 
+    @Override
+    public List<SourcePositionRange> getSourcePositions() {
+        List<SourcePositionRange> result = new ArrayList<>();
+        for (LastRel lr: this.nodes) {
+            result.addAll(lr.getSourcePositions());
+        }
+        return result;
+    }
+
     @Override
     public IIndentStream asJson(IIndentStream stream, Map<RelNode, Integer> idRemap) {
         if (this.nodes.size() == 1) {

diff --git a/...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java
@@ -180,6 +180,7 @@ void createOptimizer() {
         this.add(new MerkleOuter(compiler, true));
         this.add(new MerkleOuter(compiler, false));
         this.add(new TagRegions(compiler));
+        this.add(new FindUnboundedState(compiler));
         this.add(new CircuitStatistics(compiler));
     }
 

diff --git a/...-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java b/...-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java
@@ -690,7 +690,7 @@ public void postorder(DBSPWindowOperator operator) {
         DBSPSimpleOperator result = operator;
         if (Linq.different(sources, operator.inputs))
             result = new DBSPWindowOperator(operator.getRelNode(), operator.lowerInclusive, operator.upperInclusive,
-                    sources.get(0), sources.get(1))
+                    operator.lowerUnbounded, sources.get(0), sources.get(1))
                     .copyAnnotations(operator);
         this.map(operator, result);
     }

diff --git a/...piler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java b/...piler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java
@@ -0,0 +1,58 @@
+package org.dbsp.sqlCompiler.compiler.visitors.outer;
+
+import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator;
+import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
+import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange;
+import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRanges;
+import org.dbsp.sqlCompiler.compiler.visitors.inner.InnerVisitor;
+import org.dbsp.sqlCompiler.ir.DBSPParameter;
+import org.dbsp.sqlCompiler.ir.IDBSPInnerNode;
+import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Visitor which extracts source position information from the various properties of an operator */
+public class FindSourcePositions extends InnerVisitor {
+    public final Set<SourcePositionRange> positions;
+    private final boolean reset;
+
+    public FindSourcePositions(DBSPCompiler compiler, boolean reset) {
+        super(compiler);
+        this.positions = new HashSet<>();
+        this.reset = reset;
+    }
+
+    @Override
+    public void postorder(DBSPExpression expression) {
+        SourcePositionRange positionRange = expression.getNode().getPositionRange();
+        if (positionRange.isValid())
+            this.positions.add(positionRange);
+    }
+
+    @Override
+    public void postorder(DBSPParameter parameter) {
+        SourcePositionRange positionRange = parameter.getNode().getPositionRange();
+        if (positionRange.isValid())
+            this.positions.add(positionRange);
+    }
+
+    @Override
+    public void startVisit(IDBSPInnerNode node) {
+        super.startVisit(node);
+        if (this.reset)
+            this.positions.clear();
+    }
+
+    public SourcePositionRanges getPositions() {
+        return new SourcePositionRanges(this.positions);
+    }
+
+    /** Find the source positions associated with the specified operator */
+    public static SourcePositionRanges getPositions(DBSPCompiler compiler, DBSPOperator operator) {
+        FindSourcePositions positions = new FindSourcePositions(compiler, true);
+        operator.accept(positions);
+        positions.positions.addAll(operator.getSourcePositions());
+        return positions.getPositions();
+    }
+}