From 38e1f5b73e7b5949810ccca6b760ae64cbcf6489 Mon Sep 17 00:00:00 2001 From: Junichi Furukawa Date: Tue, 28 Jul 2026 15:29:06 -0700 Subject: [PATCH 1/3] feat(vm): boundary time for recurring() so profiles get re-scheduled (LYT-391) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit findDateMathFn only produced a boundary fn for a "now±N" string literal inside a binary or BETWEEN node. recurring() is now-relative without containing any such literal, so it yielded no boundary fns at all -- DateConverter reported HasDateMath=false, callers never flagged the segment as needing recalculation, and nothing scheduled the profile to be re-evaluated on its anniversary. An anniversary audience then only picked up a profile when some unrelated event happened to re-evaluate it, which is the exact failure mode the feature request (lytics/lio#34919) was filed to escape. Adds a recurring case to findDateMathFn and RecurringBoundary, which returns the next UTC midnight at which the expression changes value: the start of the next recurrence day when it's currently false, or the start of tomorrow when it's matching today. Zero time when it never changes again (every-1-day past its anchor), so those don't get scheduled pointlessly. Mirrors the evaluators rather than idealizing them: the n-day path divides epoch seconds toward zero, and anchors with no counterpart in a period are skipped, not clamped -- Feb 29 recurs only in leap years, the 31st only in months that have one. Shapes the evaluators reject (unknown period, non-positive or fractional n, non-literal period or offset) return no boundary fn, so they aren't flagged for recalculation either. Tests cover exact boundaries per period, the DateConverter contract callers rely on, and a walk over 550 days per configuration asserting the boundary is the precise instant the predicate flips. Co-Authored-By: Claude Opus 5 (1M context) --- vm/datemath.go | 167 ++++++++++++++++++++++++++ vm/recurring_boundary_test.go | 219 ++++++++++++++++++++++++++++++++++ 2 files changed, 386 insertions(+) create mode 100644 vm/recurring_boundary_test.go diff --git a/vm/datemath.go b/vm/datemath.go index de466bf..296d1d0 100644 --- a/vm/datemath.go +++ b/vm/datemath.go @@ -201,6 +201,14 @@ func findDateMathFn(node expr.Node) BoundaryFns { fns = append(fns, findDateMathFn(arg)...) } case *expr.FuncNode: + // recurring() is now-relative without containing any datemath literal, so + // it needs its own boundary rather than a scan of its args. + if strings.EqualFold(n.Name, "recurring") { + if fn := findBoundaryForRecurring(n); fn != nil { + return BoundaryFns{fn} + } + return fns + } for _, arg := range n.Args { fns = append(fns, findDateMathFn(arg)...) } @@ -312,3 +320,162 @@ func findBoundaryForBetween(n *expr.TriNode) func(d *DateConverter, ctx expr.Eva d.bt = compareBoundaries(d.bt, d.at.Add(ct.Sub(lower))) } } + +// findBoundaryForRecurring builds the boundary fn for +// recurring(date_field, period[, offsetDays]). Unlike the datemath cases there is +// no "now±N" literal to invert: the expression is true for whole UTC days and +// flips at midnight, so the boundary is the next day on which its value changes. +// Returns nil for shapes the evaluators reject, so those don't get flagged as +// needing recalculation. +func findBoundaryForRecurring(n *expr.FuncNode) func(d *DateConverter, ctx expr.EvalContext, inc expr.Includer) { + if len(n.Args) < 2 || len(n.Args) > 3 { + return nil + } + if _, ok := n.Args[0].(*expr.IdentityNode); !ok { + return nil + } + + var period string + var nDays int + switch pn := n.Args[1].(type) { + case *expr.StringNode: + period = strings.ToLower(pn.Text) + switch period { + case "yearly", "monthly", "weekly": + default: + return nil + } + case *expr.NumberNode: + if !pn.IsInt || pn.Int64 <= 0 { + return nil + } + nDays = int(pn.Int64) + default: + return nil + } + + offsetDays := 0 + if len(n.Args) == 3 { + on, ok := n.Args[2].(*expr.NumberNode) + if !ok || !on.IsInt { + return nil + } + offsetDays = int(on.Int64) + } + + anchorNode := n.Args[0] + return func(d *DateConverter, ctx expr.EvalContext, inc expr.Includer) { + lhv, ok := EvalInc(inc, ctx, anchorNode) + if !ok { + return + } + anchor, ok := value.ValueToTime(lhv) + if !ok || anchor.IsZero() { + // No anchor date means the expression can't become true for this row. + return + } + if bt := RecurringBoundary(anchor, d.at, period, nDays, offsetDays); !bt.IsZero() { + d.bt = compareBoundaries(d.bt, bt) + } + } +} + +// RecurringBoundary returns the next UTC midnight at which recurring(anchor, +// period, offsetDays) changes value relative to `now`, or the zero time when it +// never changes again. Mirrors the evaluators: with n > 0 the recurrence is every +// n days from the anchor, otherwise period selects yearly/monthly/weekly. +func RecurringBoundary(anchor, now time.Time, period string, n, offsetDays int) time.Time { + if n > 0 { + return recurringBoundaryNDays(anchor, now, n, offsetDays) + } + return recurringBoundaryPeriod(anchor, now, period, offsetDays) +} + +// recurringBoundaryNDays works in epoch-days with truncating division, matching +// the every-n-days evaluators (both of which divide epoch seconds toward zero). +func recurringBoundaryNDays(anchor, now time.Time, n, offsetDays int) time.Time { + anchorDay := anchor.UTC().Unix() / secondsPerDay + nowDay := now.UTC().Unix() / secondsPerDay + // The evaluator subtracts the offset from the day difference, so the first + // matching day sits offsetDays after the anchor. + first := anchorDay + int64(offsetDays) + + if nowDay < first { + return dayStartFromEpochDay(first) + } + if n == 1 { + // Every day from `first` onward matches, so it never flips back. + return time.Time{} + } + if rem := (nowDay - first) % int64(n); rem != 0 { + return dayStartFromEpochDay(nowDay + int64(n) - rem) + } + // Matches today; goes false at the start of tomorrow. + return dayStartFromEpochDay(nowDay + 1) +} + +// recurringBoundaryPeriod handles yearly/monthly/weekly, which the evaluators +// compare on the calendar date of now-offsetDays. +func recurringBoundaryPeriod(anchor, now time.Time, period string, offsetDays int) time.Time { + anchorU := anchor.UTC() + today := dayStart(now.UTC()) + target := today.AddDate(0, 0, -offsetDays) + + next := nextPeriodRecurrence(anchorU, target, period) + if next.IsZero() { + return time.Time{} + } + if next.After(target) { + // Goes true at the start of that day, shifted back into "now" space. + return next.AddDate(0, 0, offsetDays) + } + // Matches today. yearly/monthly/weekly recurrences are never on consecutive + // days, so it goes false at the start of tomorrow. + return today.AddDate(0, 0, 1) +} + +// nextPeriodRecurrence returns the first UTC day on or after `from` whose +// calendar date is a recurrence of anchor. Anchors with no counterpart in a given +// period are skipped rather than clamped -- Feb 29 only recurs in leap years, and +// the 31st only in months that have one -- matching the evaluators. +func nextPeriodRecurrence(anchor, from time.Time, period string) time.Time { + switch period { + case "yearly": + // A Feb-29 anchor can skip up to 7 years across a non-leap century. + for y := from.Year(); y <= from.Year()+8; y++ { + c := time.Date(y, anchor.Month(), anchor.Day(), 0, 0, 0, 0, time.UTC) + if c.Month() != anchor.Month() || c.Day() != anchor.Day() { + continue + } + if !c.Before(from) { + return c + } + } + case "monthly": + first := time.Date(from.Year(), from.Month(), 1, 0, 0, 0, 0, time.UTC) + for i := 0; i < 14; i++ { + m := first.AddDate(0, i, 0) + c := time.Date(m.Year(), m.Month(), anchor.Day(), 0, 0, 0, 0, time.UTC) + if c.Month() != m.Month() { + continue + } + if !c.Before(from) { + return c + } + } + case "weekly": + delta := (int(anchor.Weekday()) - int(from.Weekday()) + 7) % 7 + return from.AddDate(0, 0, delta) + } + return time.Time{} +} + +const secondsPerDay = 86400 + +func dayStart(t time.Time) time.Time { + return time.Date(t.Year(), t.Month(), t.Day(), 0, 0, 0, 0, time.UTC) +} + +func dayStartFromEpochDay(day int64) time.Time { + return time.Unix(day*secondsPerDay, 0).UTC() +} diff --git a/vm/recurring_boundary_test.go b/vm/recurring_boundary_test.go new file mode 100644 index 0000000..18a2389 --- /dev/null +++ b/vm/recurring_boundary_test.go @@ -0,0 +1,219 @@ +package vm_test + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/lytics/qlbridge/datasource" + "github.com/lytics/qlbridge/rel" + "github.com/lytics/qlbridge/vm" +) + +func utcDay(y int, m time.Month, d int) time.Time { + return time.Date(y, m, d, 0, 0, 0, 0, time.UTC) +} + +// recurringMatches is an independent copy of the recurring() predicate the +// evaluators implement, used here as an oracle for the boundary calculation. +// Kept deliberately literal, including the truncating division on the n-day path. +func recurringMatches(anchor, now time.Time, period string, n, offsetDays int) bool { + anchorU, nowU := anchor.UTC(), now.UTC() + + if n > 0 { + diff := nowU.Unix()/86400 - anchorU.Unix()/86400 - int64(offsetDays) + return diff >= 0 && diff%int64(n) == 0 + } + + target := nowU.AddDate(0, 0, -offsetDays) + switch strings.ToLower(period) { + case "yearly": + return target.Month() == anchorU.Month() && target.Day() == anchorU.Day() + case "monthly": + return target.Day() == anchorU.Day() + case "weekly": + return target.Weekday() == anchorU.Weekday() + } + return false +} + +func TestRecurringBoundary(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + anchor time.Time + now time.Time + period string + n int + offset int + want time.Time + }{ + {"yearly upcoming", utcDay(1990, 6, 29), utcDay(2026, 3, 10).Add(9 * time.Hour), "yearly", 0, 0, utcDay(2026, 6, 29)}, + {"yearly today exits tomorrow", utcDay(1990, 6, 29), utcDay(2026, 6, 29).Add(9 * time.Hour), "yearly", 0, 0, utcDay(2026, 6, 30)}, + {"yearly just passed", utcDay(1990, 6, 29), utcDay(2026, 6, 30).Add(9 * time.Hour), "yearly", 0, 0, utcDay(2027, 6, 29)}, + // Feb 29 only recurs in leap years, so the wait can be four years. + {"leap anchor skips non-leap years", utcDay(2000, 2, 29), utcDay(2025, 3, 1), "yearly", 0, 0, utcDay(2028, 2, 29)}, + {"leap anchor today", utcDay(2000, 2, 29), utcDay(2024, 2, 29).Add(9 * time.Hour), "yearly", 0, 0, utcDay(2024, 3, 1)}, + + {"monthly upcoming", utcDay(1990, 6, 15), utcDay(2026, 3, 10), "monthly", 0, 0, utcDay(2026, 3, 15)}, + {"monthly today", utcDay(1990, 6, 15), utcDay(2026, 3, 15).Add(5 * time.Hour), "monthly", 0, 0, utcDay(2026, 3, 16)}, + {"monthly next month", utcDay(1990, 6, 15), utcDay(2026, 3, 20), "monthly", 0, 0, utcDay(2026, 4, 15)}, + // The 31st skips months that don't have one. + {"monthly 31st skips february", utcDay(1990, 1, 31), utcDay(2026, 2, 10), "monthly", 0, 0, utcDay(2026, 3, 31)}, + + // 2026-06-29 and 2026-07-06 are both Mondays. + {"weekly upcoming", utcDay(2026, 6, 29), utcDay(2026, 7, 1), "weekly", 0, 0, utcDay(2026, 7, 6)}, + {"weekly today", utcDay(2026, 6, 29), utcDay(2026, 7, 6).Add(9 * time.Hour), "weekly", 0, 0, utcDay(2026, 7, 7)}, + + {"every 30 days upcoming", utcDay(2026, 1, 1), utcDay(2026, 1, 15).Add(9 * time.Hour), "", 30, 0, utcDay(2026, 1, 31)}, + {"every 30 days today", utcDay(2026, 1, 1), utcDay(2026, 1, 31).Add(9 * time.Hour), "", 30, 0, utcDay(2026, 2, 1)}, + {"every 30 days before anchor", utcDay(2026, 1, 1), utcDay(2025, 12, 31).Add(9 * time.Hour), "", 30, 0, utcDay(2026, 1, 1)}, + + // Every day: true from the anchor onward, so it never flips back. + {"every 1 day after anchor never flips", utcDay(2026, 1, 1), utcDay(2026, 5, 1), "", 1, 0, time.Time{}}, + {"every 1 day before anchor", utcDay(2026, 1, 1), utcDay(2025, 12, 1), "", 1, 0, utcDay(2026, 1, 1)}, + + {"unknown period has no boundary", utcDay(1990, 6, 29), utcDay(2026, 3, 10), "fortnightly", 0, 0, time.Time{}}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got := vm.RecurringBoundary(tc.anchor, tc.now, tc.period, tc.n, tc.offset) + assert.Equal(t, tc.want.UTC(), got.UTC()) + }) + } +} + +// TestRecurringBoundaryIsExactFlipPoint walks a year and a half of "now" values +// per configuration and checks the returned boundary against the predicate +// itself: the value must be unchanged right up to the boundary and different at +// it. This is what makes the boundary safe to schedule a re-evaluation on. +func TestRecurringBoundaryIsExactFlipPoint(t *testing.T) { + t.Parallel() + + configs := []struct { + name string + anchor time.Time + period string + n int + offset int + }{ + {"yearly", utcDay(1990, 6, 29), "yearly", 0, 0}, + {"yearly leap anchor", utcDay(2000, 2, 29), "yearly", 0, 0}, + {"yearly half-birthday", utcDay(1990, 6, 29), "yearly", 0, 182}, + {"yearly negative offset", utcDay(1990, 6, 29), "yearly", 0, -30}, + {"monthly", utcDay(1990, 6, 15), "monthly", 0, 0}, + {"monthly 31st", utcDay(1990, 1, 31), "monthly", 0, 0}, + {"monthly with offset", utcDay(1990, 6, 15), "monthly", 0, 5}, + {"weekly", utcDay(2026, 6, 29), "weekly", 0, 0}, + {"weekly with offset", utcDay(2026, 6, 29), "weekly", 0, 3}, + {"every 30 days", utcDay(2026, 1, 1), "", 30, 0}, + {"every 7 days with offset", utcDay(2026, 1, 1), "", 7, 10}, + // Pre-1970 anchor: the n-day path divides toward zero, so the oracle has + // to share that quirk -- which it does, being a copy of the evaluator. + {"every 90 days pre-1970 anchor", utcDay(1965, 6, 29).Add(12 * time.Hour), "", 90, 0}, + } + + for _, cfg := range configs { + t.Run(cfg.name, func(t *testing.T) { + t.Parallel() + // Mid-morning, so a midnight boundary is a distinct instant from "now". + start := utcDay(2026, 1, 1).Add(9 * time.Hour) + + for i := 0; i < 550; i++ { + now := start.AddDate(0, 0, i) + bt := vm.RecurringBoundary(cfg.anchor, now, cfg.period, cfg.n, cfg.offset) + if bt.IsZero() { + continue + } + + require.True(t, bt.After(now), "boundary %v is not after now %v", bt, now) + require.Equal(t, bt, bt.Truncate(24*time.Hour).UTC(), "boundary %v is not a UTC midnight", bt) + + at := func(ts time.Time) bool { + return recurringMatches(cfg.anchor, ts, cfg.period, cfg.n, cfg.offset) + } + justBefore := bt.Add(-time.Nanosecond) + require.Equal(t, at(now), at(justBefore), + "value changed before the predicted boundary: now=%v boundary=%v", now, bt) + require.NotEqual(t, at(justBefore), at(bt), + "value did not change at the predicted boundary: now=%v boundary=%v", now, bt) + } + }) + } +} + +// TestRecurringBoundaryViaDateConverter covers the contract callers actually use: +// a recurring() filter must report HasDateMath so the segment gets flagged for +// re-calculation, and expose the boundary. +func TestRecurringBoundaryViaDateConverter(t *testing.T) { + t.Parallel() + + at := utcDay(2026, 3, 10).Add(9 * time.Hour) + evalCtx := datasource.NewContextMapTs(map[string]any{ + "birthday": utcDay(1990, 6, 29), + "last_event": at.Add(-12 * time.Hour), + }, true, at) + inc := &includectx{ContextReader: evalCtx} + + tests := []struct { + name string + filter string + hasDateMath bool + want time.Time + }{ + {"yearly", `FILTER recurring(birthday, "yearly")`, true, utcDay(2026, 6, 29)}, + {"yearly with offset", `FILTER recurring(birthday, "yearly", 10)`, true, utcDay(2026, 7, 9)}, + // 13038 days from the anchor to now, and 13038 mod 90 == 78, so the next + // multiple of 90 lands 12 days out. + {"every 90 days", `FILTER recurring(birthday, 90)`, true, utcDay(2026, 3, 22)}, + // Composed with datemath: the earliest boundary wins, and last_event goes + // stale in 12 hours -- long before June. + { + name: "earliest boundary wins", + filter: `FILTER AND ( recurring(birthday, "yearly"), last_event > "now-1d" )`, + hasDateMath: true, + want: at.Add(12 * time.Hour), + }, + // Shapes the evaluators reject must not be flagged for re-calculation. + {"unknown period", `FILTER recurring(birthday, "fortnightly")`, false, time.Time{}}, + {"zero day period", `FILTER recurring(birthday, 0)`, false, time.Time{}}, + {"non-literal period", `FILTER recurring(birthday, some_field)`, false, time.Time{}}, + {"too few args", `FILTER recurring(birthday)`, false, time.Time{}}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fs := rel.MustParseFilter(tc.filter) + + dc, err := vm.NewDateConverterWithAnchorTime(inc, inc, fs.Filter, at) + require.NoError(t, err) + assert.Equal(t, tc.hasDateMath, dc.HasDateMath) + assert.Equal(t, tc.want.UTC(), dc.Boundary().UTC()) + }) + } +} + +// TestRecurringBoundaryNoAnchor pins that a row with no usable date yields no +// boundary rather than an error, so one profile missing a birthday doesn't stop +// the rest of the filter's boundaries from being calculated. +func TestRecurringBoundaryNoAnchor(t *testing.T) { + t.Parallel() + + at := utcDay(2026, 3, 10).Add(9 * time.Hour) + evalCtx := datasource.NewContextMapTs(map[string]any{ + "last_event": at.Add(-12 * time.Hour), + }, true, at) + inc := &includectx{ContextReader: evalCtx} + + fs := rel.MustParseFilter(`FILTER AND ( recurring(birthday, "yearly"), last_event > "now-1d" )`) + dc, err := vm.NewDateConverterWithAnchorTime(inc, inc, fs.Filter, at) + require.NoError(t, err) + assert.Equal(t, at.Add(12*time.Hour).UTC(), dc.Boundary().UTC()) +} From 1d0152355476810ee63526cf63a2ecd9cc19b5b8 Mon Sep 17 00:00:00 2001 From: Junichi Furukawa Date: Tue, 28 Jul 2026 17:12:12 -0700 Subject: [PATCH 2/3] fix(recurring): floor epoch days instead of truncating toward zero (LYT-391) The every-n-days path divided epoch seconds with `/` on both sides -- Go in the boundary calculation, Java in the generated Painless. Both round toward zero, so an anchor before 1970 whose time-of-day was not midnight bucketed one day too high and its recurrences landed a day late. Pre-1970 birth dates are the headline input for recurring(), so this is the common case for the feature, not an edge. Switches the script to Math.floorDiv(..., 86400L) and the Go side to a matching floorDivInt64. The two must move together: flooring one side alone would make an Elasticsearch scan and in-process evaluation disagree for exactly these profiles. lio's RecurringMatch needs the same change, landing with its go.mod bump. Post-1970 values are unaffected -- truncation and flooring agree for non-negative seconds -- so no existing audience of recent dates changes. Co-Authored-By: Claude Opus 5 (1M context) --- generators/esgen/bridgeutil.go | 19 +++++++++++++-- generators/esgen/recurring_test.go | 8 ++++--- vm/datemath.go | 19 +++++++++++---- vm/recurring_boundary_test.go | 37 ++++++++++++++++++++++++++---- 4 files changed, 69 insertions(+), 14 deletions(-) diff --git a/generators/esgen/bridgeutil.go b/generators/esgen/bridgeutil.go index a509ddf..d3b5713 100644 --- a/generators/esgen/bridgeutil.go +++ b/generators/esgen/bridgeutil.go @@ -156,8 +156,11 @@ func makeRecurringQuery(lhs *gentypes.FieldType, period expr.Node, offsetDays in return nil, fmt.Errorf("'recurring' day period must be a positive integer, got %v", period) } n := num.Int64 - todayDay := now.UTC().Unix() / 86400 - src := fmt.Sprintf("if (%s) { long d = params.todayDay - (doc[%s].value.toInstant().getEpochSecond() / 86400) - params.offset; return d >= 0 && d %% params.n == 0; } return false;", + // Floor rather than truncate: Java's and Go's `/` both round toward zero, + // which puts a pre-1970 anchor with a non-midnight time one day late. Both + // sides of this (here and the in-process evaluator) must floor or neither. + todayDay := floorDivInt64(now.UTC().Unix(), secondsPerDay) + src := fmt.Sprintf("if (%s) { long d = params.todayDay - Math.floorDiv(doc[%s].value.toInstant().getEpochSecond(), 86400L) - params.offset; return d >= 0 && d %% params.n == 0; } return false;", exists, q) return Script(src, map[string]any{"todayDay": todayDay, "offset": offsetDays, "n": n}), nil } @@ -415,3 +418,15 @@ func makeTimeWindowQuery(lhs *gentypes.FieldType, threshold, window, ts int64) ( IgnoreUnmapped: true, }}, nil } + +const secondsPerDay = 86400 + +// floorDivInt64 mirrors Painless's Math.floorDiv: rounds toward negative +// infinity instead of toward zero, so pre-1970 epoch seconds map to the right day. +func floorDivInt64(a, b int64) int64 { + q := a / b + if a%b != 0 && (a < 0) != (b < 0) { + q-- + } + return q +} diff --git a/generators/esgen/recurring_test.go b/generators/esgen/recurring_test.go index eb6d11c..06ef4ff 100644 --- a/generators/esgen/recurring_test.go +++ b/generators/esgen/recurring_test.go @@ -55,9 +55,11 @@ func TestRecurring(t *testing.T) { wantParams: map[string]any{"month": 12, "day": 29}, // 2026-06-29 minus 182 days = 2025-12-29 }, { - name: "every 90 days", - filter: `FILTER recurring(signup, 90)`, - wantSrc: "if (doc['signup'].size() != 0) { long d = params.todayDay - (doc['signup'].value.toInstant().getEpochSecond() / 86400) - params.offset; return d >= 0 && d % params.n == 0; } return false;", + name: "every 90 days", + filter: `FILTER recurring(signup, 90)`, + // Math.floorDiv, not `/`: Java rounds toward zero, which put a pre-1970 + // anchor with a non-midnight time one day late. + wantSrc: "if (doc['signup'].size() != 0) { long d = params.todayDay - Math.floorDiv(doc['signup'].value.toInstant().getEpochSecond(), 86400L) - params.offset; return d >= 0 && d % params.n == 0; } return false;", wantParams: map[string]any{"todayDay": ts.UTC().Unix() / 86400, "offset": 0, "n": int64(90)}, }, { diff --git a/vm/datemath.go b/vm/datemath.go index 296d1d0..aa9ce24 100644 --- a/vm/datemath.go +++ b/vm/datemath.go @@ -391,11 +391,12 @@ func RecurringBoundary(anchor, now time.Time, period string, n, offsetDays int) return recurringBoundaryPeriod(anchor, now, period, offsetDays) } -// recurringBoundaryNDays works in epoch-days with truncating division, matching -// the every-n-days evaluators (both of which divide epoch seconds toward zero). +// recurringBoundaryNDays works in epoch-days, flooring toward negative infinity +// so a pre-1970 anchor buckets by calendar day. Matches the every-n-days +// evaluators, which floor the same way. func recurringBoundaryNDays(anchor, now time.Time, n, offsetDays int) time.Time { - anchorDay := anchor.UTC().Unix() / secondsPerDay - nowDay := now.UTC().Unix() / secondsPerDay + anchorDay := floorDivInt64(anchor.UTC().Unix(), secondsPerDay) + nowDay := floorDivInt64(now.UTC().Unix(), secondsPerDay) // The evaluator subtracts the offset from the day difference, so the first // matching day sits offsetDays after the anchor. first := anchorDay + int64(offsetDays) @@ -479,3 +480,13 @@ func dayStart(t time.Time) time.Time { func dayStartFromEpochDay(day int64) time.Time { return time.Unix(day*secondsPerDay, 0).UTC() } + +// floorDivInt64 mirrors Painless's Math.floorDiv: rounds toward negative +// infinity instead of toward zero. +func floorDivInt64(a, b int64) int64 { + q := a / b + if a%b != 0 && (a < 0) != (b < 0) { + q-- + } + return q +} diff --git a/vm/recurring_boundary_test.go b/vm/recurring_boundary_test.go index 18a2389..edafdf1 100644 --- a/vm/recurring_boundary_test.go +++ b/vm/recurring_boundary_test.go @@ -18,13 +18,21 @@ func utcDay(y int, m time.Month, d int) time.Time { } // recurringMatches is an independent copy of the recurring() predicate the -// evaluators implement, used here as an oracle for the boundary calculation. -// Kept deliberately literal, including the truncating division on the n-day path. +// evaluators implement, used here as an oracle for the boundary calculation. The +// n-day path floors toward negative infinity, matching Math.floorDiv in the +// generated Painless and the in-process evaluator. func recurringMatches(anchor, now time.Time, period string, n, offsetDays int) bool { anchorU, nowU := anchor.UTC(), now.UTC() if n > 0 { - diff := nowU.Unix()/86400 - anchorU.Unix()/86400 - int64(offsetDays) + floorDay := func(sec int64) int64 { + d := sec / 86400 + if sec%86400 != 0 && sec < 0 { + d-- + } + return d + } + diff := floorDay(nowU.Unix()) - floorDay(anchorU.Unix()) - int64(offsetDays) return diff >= 0 && diff%int64(n) == 0 } @@ -114,9 +122,10 @@ func TestRecurringBoundaryIsExactFlipPoint(t *testing.T) { {"weekly with offset", utcDay(2026, 6, 29), "weekly", 0, 3}, {"every 30 days", utcDay(2026, 1, 1), "", 30, 0}, {"every 7 days with offset", utcDay(2026, 1, 1), "", 7, 10}, - // Pre-1970 anchor: the n-day path divides toward zero, so the oracle has - // to share that quirk -- which it does, being a copy of the evaluator. + // Pre-1970 anchor with a non-midnight time: the case truncating division + // got wrong by a day. {"every 90 days pre-1970 anchor", utcDay(1965, 6, 29).Add(12 * time.Hour), "", 90, 0}, + {"every 30 days pre-1970 anchor", utcDay(1965, 6, 29).Add(23 * time.Hour), "", 30, 0}, } for _, cfg := range configs { @@ -217,3 +226,21 @@ func TestRecurringBoundaryNoAnchor(t *testing.T) { require.NoError(t, err) assert.Equal(t, at.Add(12*time.Hour).UTC(), dc.Boundary().UTC()) } + +// TestRecurringBoundaryFloorsPre1970 pins that the n-day path floors epoch days +// instead of truncating toward zero. Under truncation a pre-1970 anchor whose +// time-of-day wasn't midnight bucketed one day late, so its recurrences landed a +// day after those of the same calendar date at midnight. +func TestRecurringBoundaryFloorsPre1970(t *testing.T) { + t.Parallel() + + midnight := utcDay(1965, 6, 29) + now := utcDay(2026, 3, 10).Add(9 * time.Hour) + + for _, offset := range []time.Duration{time.Hour, 12 * time.Hour, 23 * time.Hour} { + assert.Equal(t, + vm.RecurringBoundary(midnight, now, "", 90, 0), + vm.RecurringBoundary(midnight.Add(offset), now, "", 90, 0), + "anchor at +%v should recur on the same days as the same date at midnight", offset) + } +} From 1d9c56673fd1065389addce91fd931c5cf320114 Mon Sep 17 00:00:00 2001 From: Junichi Furukawa Date: Wed, 29 Jul 2026 16:52:38 -0700 Subject: [PATCH 3/3] refactor: single definition of epoch-day flooring as vm.EpochDay (LYT-391) The floor-toward-negative-infinity day arithmetic had been copied into vm and esgen, and lio needs the same function for its in-process evaluators -- three copies of the rule that has to stay in lockstep with Math.floorDiv in the generated Painless. That is the drift that produced the pre-1970 bug in the first place. Exports it once as vm.EpochDay and points esgen at it (esgen already depends on vm, and vm does not depend on esgen, so no new cycle). lio calls the same function rather than keeping its own copy. The boundary test's oracle keeps its own hand-rolled flooring on purpose, with a comment saying so: an oracle that shares code with what it checks cannot catch a bug in that code. Co-Authored-By: Claude Opus 5 (1M context) --- generators/esgen/bridgeutil.go | 20 ++++---------------- vm/datemath.go | 19 ++++++++++--------- vm/recurring_boundary_test.go | 4 ++-- 3 files changed, 16 insertions(+), 27 deletions(-) diff --git a/generators/esgen/bridgeutil.go b/generators/esgen/bridgeutil.go index d3b5713..b688e74 100644 --- a/generators/esgen/bridgeutil.go +++ b/generators/esgen/bridgeutil.go @@ -11,6 +11,7 @@ import ( "github.com/lytics/qlbridge/generators/gentypes" "github.com/lytics/qlbridge/lex" "github.com/lytics/qlbridge/value" + "github.com/lytics/qlbridge/vm" ) type floatval interface { @@ -156,10 +157,9 @@ func makeRecurringQuery(lhs *gentypes.FieldType, period expr.Node, offsetDays in return nil, fmt.Errorf("'recurring' day period must be a positive integer, got %v", period) } n := num.Int64 - // Floor rather than truncate: Java's and Go's `/` both round toward zero, - // which puts a pre-1970 anchor with a non-midnight time one day late. Both - // sides of this (here and the in-process evaluator) must floor or neither. - todayDay := floorDivInt64(now.UTC().Unix(), secondsPerDay) + // Math.floorDiv, not `/`: Java rounds toward zero, which puts a pre-1970 + // anchor with a non-midnight time one day late. + todayDay := vm.EpochDay(now.UTC().Unix()) src := fmt.Sprintf("if (%s) { long d = params.todayDay - Math.floorDiv(doc[%s].value.toInstant().getEpochSecond(), 86400L) - params.offset; return d >= 0 && d %% params.n == 0; } return false;", exists, q) return Script(src, map[string]any{"todayDay": todayDay, "offset": offsetDays, "n": n}), nil @@ -418,15 +418,3 @@ func makeTimeWindowQuery(lhs *gentypes.FieldType, threshold, window, ts int64) ( IgnoreUnmapped: true, }}, nil } - -const secondsPerDay = 86400 - -// floorDivInt64 mirrors Painless's Math.floorDiv: rounds toward negative -// infinity instead of toward zero, so pre-1970 epoch seconds map to the right day. -func floorDivInt64(a, b int64) int64 { - q := a / b - if a%b != 0 && (a < 0) != (b < 0) { - q-- - } - return q -} diff --git a/vm/datemath.go b/vm/datemath.go index aa9ce24..0c2ae46 100644 --- a/vm/datemath.go +++ b/vm/datemath.go @@ -395,8 +395,8 @@ func RecurringBoundary(anchor, now time.Time, period string, n, offsetDays int) // so a pre-1970 anchor buckets by calendar day. Matches the every-n-days // evaluators, which floor the same way. func recurringBoundaryNDays(anchor, now time.Time, n, offsetDays int) time.Time { - anchorDay := floorDivInt64(anchor.UTC().Unix(), secondsPerDay) - nowDay := floorDivInt64(now.UTC().Unix(), secondsPerDay) + anchorDay := EpochDay(anchor.UTC().Unix()) + nowDay := EpochDay(now.UTC().Unix()) // The evaluator subtracts the offset from the day difference, so the first // matching day sits offsetDays after the anchor. first := anchorDay + int64(offsetDays) @@ -481,12 +481,13 @@ func dayStartFromEpochDay(day int64) time.Time { return time.Unix(day*secondsPerDay, 0).UTC() } -// floorDivInt64 mirrors Painless's Math.floorDiv: rounds toward negative -// infinity instead of toward zero. -func floorDivInt64(a, b int64) int64 { - q := a / b - if a%b != 0 && (a < 0) != (b < 0) { - q-- +// EpochDay returns the UTC calendar day for a unix-seconds timestamp, flooring +// toward negative infinity so pre-1970 dates don't bucket a day late. +// The generated Painless uses Math.floorDiv to stay in step with this. +func EpochDay(sec int64) int64 { + d := sec / secondsPerDay + if sec%secondsPerDay != 0 && sec < 0 { + d-- } - return q + return d } diff --git a/vm/recurring_boundary_test.go b/vm/recurring_boundary_test.go index edafdf1..237f770 100644 --- a/vm/recurring_boundary_test.go +++ b/vm/recurring_boundary_test.go @@ -19,8 +19,8 @@ func utcDay(y int, m time.Month, d int) time.Time { // recurringMatches is an independent copy of the recurring() predicate the // evaluators implement, used here as an oracle for the boundary calculation. The -// n-day path floors toward negative infinity, matching Math.floorDiv in the -// generated Painless and the in-process evaluator. +// day flooring is hand-rolled rather than calling vm.EpochDay on purpose: an +// oracle sharing code with what it checks can't catch a bug in that code. func recurringMatches(anchor, now time.Time, period string, n, offsetDays int) bool { anchorU, nowU := anchor.UTC(), now.UTC()