Windows: reemit publishes the late rows alone, so the Postgres example drops

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## Unreleased
+
+### Fixed
+
+- `bluesky.postgres.windowed.yml` paired `late_rows: reemit` with an upsert
+  that replaces `posts`. `reemit` publishes `emit_sql` over the late rows
+  alone, so one late post replaced a closed minute's count with its own. The
+  example declares `drop`. The README, the config schema, the example
+  comments and `validate`'s warning said `reemit` publishes the bucket again
+  for a sink that upserts. They now say it publishes the late rows, for a
+  sink that adds them to the bucket it holds.
+
 ## v2026.09.14
 
 The first date-tagged release. Upgrade from v1.2.0. The pipeline file

README.md

@@ -863,12 +863,15 @@ bucket and every open bucket closes. Wall clock appears nowhere in the close.
 **Late rows.** A row for a bucket below the watermark arrived after that
 bucket was published. `late_rows` is required, because the two policies are
 different promises to the sink. `drop` deletes the row and counts it in
-`window_late_rows_total`, so a sink that appends sees each bucket once.
-`reemit` publishes the bucket again: a sink that upserts on the bucket's key
-replaces the value, and a sink that appends holds both rows, so its reader
-has to treat the later one as a correction. `sqlflow validate` warns when
-`reemit` is paired with the Iceberg or Kafka sink. A rising drop count means
-the grace is too short for the stream.
+`window_late_rows_total`, so the sink sees each bucket once, as it closed.
+`reemit` runs `emit_sql` over the late rows alone and publishes the result,
+because the bucket's other rows were deleted when it closed. The sink has to
+add that result to the bucket it holds. An upsert that replaces on the
+bucket's key overwrites the bucket's count with the late rows' count, so pair
+a replacing upsert with `drop`. An upsert that adds counts a republished close
+twice; see **Guarantees**. `sqlflow validate` warns when `reemit` is paired
+with the Iceberg or Kafka sink. A rising drop count means the grace is too
+short for the stream.
 
 The watermark is one value for the whole table, which makes it the fastest
 partition's clock. A topic whose partitions run at uneven rates has a slow

dev/config/examples/bluesky/bluesky.kafka.windowed.yml

@@ -22,8 +22,8 @@ tables:
       #
       # late_rows: drop discards a row for a bucket that already closed and
       # counts it in window_late_rows_total, so each bucket reaches the topic
-      # once. reemit would publish the bucket again for a consumer that
-      # merges.
+      # once. reemit would publish a second message for the bucket, counted
+      # over the late rows alone, for a consumer that adds it to the first.
       window:
         time_column: bucket
         size_seconds: 60

dev/config/examples/bluesky/bluesky.postgres.windowed.yml

@@ -48,15 +48,17 @@ tables:
       # land. After idle_close_seconds with nothing arriving, every open
       # bucket closes.
       #
-      # late_rows: reemit, because the sink upserts: a row for a bucket that
-      # already closed publishes the bucket again, and ON CONFLICT replaces
-      # the value.
+      # late_rows: drop, because the sink replaces. A row for a bucket that
+      # already closed is discarded and counted in window_late_rows_total.
+      # reemit would run emit_sql over the late rows alone, because the
+      # bucket's other rows were deleted when it closed, and ON CONFLICT
+      # would replace the bucket's count with theirs.
       window:
         time_column: bucket
         size_seconds: 60
         grace_seconds: 60
         idle_close_seconds: 60
-        late_rows: reemit
+        late_rows: drop
         poll_interval_seconds: 10
         emit_sql: |
           SELECT bucket, lang, sum(posts)::INTEGER AS posts

dev/config/examples/tumbling.window.yml

@@ -31,8 +31,8 @@ tables:
         grace_seconds: 0
         idle_close_seconds: 60
         # The sink is a topic, which appends. drop keeps each bucket to one
-        # message; reemit would publish a correction the consumer has to
-        # apply itself.
+        # message; reemit would publish a second message counted over the
+        # late rows alone, which the consumer has to add to the first.
         late_rows: drop
         # emit_sql shapes the closed rows. closed is every row of every bucket
         # that just closed. GROUP BY ALL groups by the output columns, which

internal/cli/testdata/config_example.golden

@@ -204,8 +204,9 @@ tables:
         # Absent means never: a stream that stops leaves its last bucket open.
         idle_close_seconds: <integer>
         # What happens to a row for a bucket that already closed. drop discards
-        # it and counts it. reemit publishes the bucket again, for a sink that
-        # merges on the bucket's key; a sink that appends holds both rows.
+        # it and counts it. reemit publishes emit_sql over the late rows alone,
+        # for a sink that adds them to the bucket it holds; a sink that replaces
+        # the bucket's value loses the rows published before.
         # Required: the two are different promises to the sink.
         late_rows: drop | reemit
         # How often the engine looks for closed buckets. Absent means 10.

internal/config/config.go

@@ -140,8 +140,9 @@ type Window struct {
 	// Absent means never: a stream that stops leaves its last bucket open.
 	IdleCloseSeconds int `yaml:"idle_close_seconds,omitempty" jsonschema:"minimum=0"`
 	// What happens to a row for a bucket that already closed. drop discards
-	// it and counts it. reemit publishes the bucket again, for a sink that
-	// merges on the bucket's key; a sink that appends holds both rows.
+	// it and counts it. reemit publishes emit_sql over the late rows alone,
+	// for a sink that adds them to the bucket it holds; a sink that replaces
+	// the bucket's value loses the rows published before.
 	// Required: the two are different promises to the sink.
 	LateRows string `yaml:"late_rows" jsonschema:"enum=drop,enum=reemit"`
 	// How often the engine looks for closed buckets. Absent means 10.

internal/managers/watermark.go

@@ -28,9 +28,12 @@ const defaultPollInterval = 10 * time.Second
 type LatePolicy string
 
 const (
-	// LateReemit publishes the bucket again. For a sink that merges.
+	// LateReemit publishes emit_sql over the late rows alone: the bucket's
+	// other rows were deleted when it closed. For a sink that adds them to
+	// the bucket it holds.
 	LateReemit LatePolicy = "reemit"
-	// LateDrop discards the row and counts it. For a sink that appends.
+	// LateDrop discards the row and counts it. For a sink that appends or
+	// replaces.
 	LateDrop LatePolicy = "drop"
 )
 
@@ -43,7 +46,7 @@ func ParseLatePolicy(s string) (LatePolicy, error) {
 	case LateDrop:
 		return LateDrop, nil
 	case "":
-		return "", errs.New(errs.CodeConfigInvalid, "late_rows is required: drop, or reemit for a sink that upserts")
+		return "", errs.New(errs.CodeConfigInvalid, "late_rows is required: drop, or reemit for a sink that adds late rows to the bucket it holds")
 	default:
 		return "", errs.New(errs.CodeConfigInvalid, "late_rows must be drop or reemit, not %q", s)
 	}

internal/managers/watermark_test.go

@@ -197,6 +197,32 @@ func TestManagerWindow_LateRowsFollowThePolicy(t *testing.T) {
 	}
 }
 
+// reemit runs emit_sql over the late rows alone. The bucket's earlier rows
+// were deleted when it closed, so a sum over the bucket after a late row is
+// the late rows' sum, not the bucket's total. A sink that replaces the
+// bucket's value with it loses everything published before.
+func TestManagerWindow_ReemitPublishesTheLateRowsAlone(t *testing.T) {
+	coverage.Covers(t, "manager.window")
+	ctx := context.Background()
+	d := newTestDB(t, "")
+	createWindowTable(t, d.pipeline)
+	now := live(d, t)
+	sink := &recordingSink{}
+	decl := testDecl()
+	decl.Late = LateReemit
+	decl.EmitSQL = "SELECT sum(count)::INT AS total, bucket, city FROM closed GROUP BY ALL"
+	w := newTestWatermark(t, d, decl, sink, now)
+
+	insertBucket(t, d.pipeline, 0, "NYC", 5)
+	insertBucket(t, d.pipeline, 2, "NYC", 1)
+	assert.NoError(t, w.Poll(ctx))
+
+	insertBucket(t, d.pipeline, 0, "NYC", 1)
+	assert.NoError(t, w.Poll(ctx))
+
+	assert.DeepEqual(t, sink.published(), [][]string{{"5"}, {"1"}})
+}
+
 // A close reads committed rows only. Rows an open transaction on another
 // connection has written are not in the bucket the sink receives.
 func TestManagerWindow_UncommittedRowsAreNotPublished(t *testing.T) {

internal/validate/schemas/config.json

@@ -734,7 +734,7 @@
                       "drop",
                       "reemit"
                     ],
-                    "description": "What happens to a row for a bucket that already closed. drop discards\nit and counts it. reemit publishes the bucket again, for a sink that\nmerges on the bucket's key; a sink that appends holds both rows.\nRequired: the two are different promises to the sink."
+                    "description": "What happens to a row for a bucket that already closed. drop discards\nit and counts it. reemit publishes emit_sql over the late rows alone,\nfor a sink that adds them to the bucket it holds; a sink that replaces\nthe bucket's value loses the rows published before.\nRequired: the two are different promises to the sink."
                   },
                   "poll_interval_seconds": {
                     "type": "integer",

internal/validate/window.go

@@ -87,14 +87,15 @@ func checkWindows(rendered []byte, rep *Report) {
 						"Drop the index and sum the appended rows in emit_sql, or set "+
 						"pipeline.state.path", i), position(node)))
 			}
-			// reemit publishes a bucket the sink already holds. A sink that
-			// upserts replaces it; a sink that appends keeps both rows, and
-			// its reader cannot tell which is current.
+			// reemit publishes emit_sql over the late rows alone, for a bucket
+			// the sink already holds. A sink that appends keeps both rows, and
+			// its reader has to add them rather than keep the newest.
 			if w.LateRows == "reemit" && appendsOnly(w.Sink.Type) {
 				rep.Add(diagnostic(errs.CodeConfigInvalid, SeverityWarning, fmt.Sprintf(
 					"tables.sql[%d] window: late_rows is reemit and the %s sink appends, so a "+
-						"late row publishes a second row for a bucket the sink already holds. "+
-						"Use drop, or a sink that upserts on the bucket's key",
+						"late row publishes a second row for a bucket the sink already holds, "+
+						"computed over the late rows alone. Its reader has to add the two. "+
+						"Use drop unless it does",
 					i, w.Sink.Type), position(mappingKey(node, "late_rows"))))
 			}
 		}