diff --git a/generators/esgen/bridgeutil.go b/generators/esgen/bridgeutil.go index a509ddf..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,8 +157,10 @@ 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;", + // 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 } 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 de466bf..0c2ae46 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,174 @@ 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, 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 := 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) + + 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() +} + +// 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 d +} diff --git a/vm/recurring_boundary_test.go b/vm/recurring_boundary_test.go new file mode 100644 index 0000000..237f770 --- /dev/null +++ b/vm/recurring_boundary_test.go @@ -0,0 +1,246 @@ +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. The +// 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() + + if n > 0 { + 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 + } + + 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 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 { + 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()) +} + +// 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) + } +}