From 58e5bd36f972b7618b1bb7e07a6941b34cb0b1ad Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Mon, 14 Sep 2026 06:53:02 -0400 Subject: [PATCH] Windows: reemit publishes the late rows alone, so the Postgres example drops bluesky.postgres.windowed.yml paired late_rows: reemit with ON CONFLICT (bucket, lang) DO UPDATE SET posts = EXCLUDED.posts. A close deletes the bucket's rows, so reemit runs emit_sql over the late rows alone, and the upsert replaced a published minute's count with theirs. Against the v2026.09.14 image, a webhook source and Postgres 18: five posts closed a minute, one late post arrived for it, and the row went from 5 to 1. The same run under drop kept 5, with window_late_rows_total at 1. The example now declares drop. The README, the config schema, the example comments, ParseLatePolicy's message and validate's reemit warning said reemit publishes the bucket again for a sink that upserts, which is the pairing that loses the count. They now say reemit publishes emit_sql over the late rows, for a sink that adds them to the bucket it holds. TestManagerWindow_ReemitPublishesTheLateRowsAlone pins the behaviour the docs now describe: a bucket closes with a sum of 5, a late row arrives, and the second flush carries 1. With the close's delete removed, so reemit republishes the whole bucket, the test fails with 6. If this is wrong, the docs steer a user with a replacing upsert away from reemit when reemit would have kept the count, or the example undercounts by every late post it drops. --- CHANGELOG.md | 12 +++++++++ README.md | 15 ++++++----- .../bluesky/bluesky.kafka.windowed.yml | 4 +-- .../bluesky/bluesky.postgres.windowed.yml | 10 ++++--- dev/config/examples/tumbling.window.yml | 4 +-- internal/cli/testdata/config_example.golden | 5 ++-- internal/config/config.go | 5 ++-- internal/managers/watermark.go | 9 ++++--- internal/managers/watermark_test.go | 26 +++++++++++++++++++ internal/validate/schemas/config.json | 2 +- internal/validate/window.go | 11 ++++---- 11 files changed, 76 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 69f780cb..63891523 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index b0a62e78..7e9230ef 100644 --- a/README.md +++ b/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 diff --git a/dev/config/examples/bluesky/bluesky.kafka.windowed.yml b/dev/config/examples/bluesky/bluesky.kafka.windowed.yml index 3a932458..852fce2d 100644 --- a/dev/config/examples/bluesky/bluesky.kafka.windowed.yml +++ b/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 diff --git a/dev/config/examples/bluesky/bluesky.postgres.windowed.yml b/dev/config/examples/bluesky/bluesky.postgres.windowed.yml index 811517c0..9f460b33 100644 --- a/dev/config/examples/bluesky/bluesky.postgres.windowed.yml +++ b/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 diff --git a/dev/config/examples/tumbling.window.yml b/dev/config/examples/tumbling.window.yml index 468b3d95..6fe9a847 100644 --- a/dev/config/examples/tumbling.window.yml +++ b/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 diff --git a/internal/cli/testdata/config_example.golden b/internal/cli/testdata/config_example.golden index 1c2780fc..2a927b28 100644 --- a/internal/cli/testdata/config_example.golden +++ b/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: # 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. diff --git a/internal/config/config.go b/internal/config/config.go index 6c322777..e11c4f79 100644 --- a/internal/config/config.go +++ b/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. diff --git a/internal/managers/watermark.go b/internal/managers/watermark.go index aebf4fd4..d833fe6d 100644 --- a/internal/managers/watermark.go +++ b/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) } diff --git a/internal/managers/watermark_test.go b/internal/managers/watermark_test.go index b713476a..13170f62 100644 --- a/internal/managers/watermark_test.go +++ b/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) { diff --git a/internal/validate/schemas/config.json b/internal/validate/schemas/config.json index 620b95f0..a99b3bd6 100644 --- a/internal/validate/schemas/config.json +++ b/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", diff --git a/internal/validate/window.go b/internal/validate/window.go index 89f5f4b9..bf46bdbb 100644 --- a/internal/validate/window.go +++ b/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")))) } }