From 343f130f4c510f8214463a684108eadc30d16eec Mon Sep 17 00:00:00 2001 From: Nuno Cruces Date: Wed, 21 Oct 2026 19:16:25 +0100 Subject: [PATCH 1/3] Fix #435. --- util/sql3util/combiner.go | 113 +++++++++++++++++++++++++++++++++ util/sql3util/combiner_test.go | 77 ++++++++++++++++++++++ 2 files changed, 190 insertions(+) create mode 100644 util/sql3util/combiner.go create mode 100644 util/sql3util/combiner_test.go diff --git a/util/sql3util/combiner.go b/util/sql3util/combiner.go new file mode 100644 index 00000000..8827a01f --- /dev/null +++ b/util/sql3util/combiner.go @@ -0,0 +1,113 @@ +package sql3util + +import ( + "errors" + "io" + + "github.com/ncruces/go-sqlite3" +) + +// WindowCombiner is the interface an aggregate function should implement +// to be used with [WindowFunction]. +type WindowCombiner[T sqlite3.AggregateFunction] interface { + sqlite3.AggregateFunction + + // Combine merges the state of other into the receiver. + // Implementations must not modify or retain other. + Combine(other T) +} + +// WindowFunction adapts a [WindowCombiner] constructor into an [sqlite3.AggregateConstructor] +// that implements [sqlite3.WindowFunction] using the two-stack sliding-window algorithm. +// If fn returns an [io.Closer], it will be called to free resources. +// +// https://orlp.net/blog/two-stack-sliding-window-aggregation/ +func WindowFunction[T WindowCombiner[T]](fn func() T) sqlite3.AggregateConstructor { + if fn == nil { + return nil + } + return func() sqlite3.AggregateFunction { + return &twoStack[T]{ + ctor: fn, + valuesAgg: fn(), + } + } +} + +type twoStack[T WindowCombiner[T]] struct { + ctor func() T + cumAggs []T + values []T + valuesAgg T +} + +func (s *twoStack[T]) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + unit := s.ctor() + unit.Step(ctx, arg...) + s.values = append(s.values, unit) + s.valuesAgg.Combine(unit) +} + +func (s *twoStack[T]) Inverse(ctx sqlite3.Context, arg ...sqlite3.Value) { + if len(s.cumAggs) > 0 { + last := len(s.cumAggs) - 1 + closeAgg(ctx, s.cumAggs[last]) + s.cumAggs[last] = *new(T) + s.cumAggs = s.cumAggs[:last] + return + } + + closeAgg(ctx, s.values[0]) + for i := len(s.values) - 1; i >= 1; i-- { + if i < len(s.values)-1 { + s.values[i].Combine(s.values[i+1]) + } + s.cumAggs = append(s.cumAggs, s.values[i]) + } + clear(s.values) + s.values = s.values[:0] + closeAgg(ctx, s.valuesAgg) + s.valuesAgg = s.ctor() +} + +func (s *twoStack[T]) Value(ctx sqlite3.Context) { + switch { + case len(s.cumAggs) == 0: + s.valuesAgg.Value(ctx) + case len(s.values) == 0: + s.cumAggs[len(s.cumAggs)-1].Value(ctx) + default: + tmp := s.ctor() + tmp.Combine(s.cumAggs[len(s.cumAggs)-1]) + tmp.Combine(s.valuesAgg) + tmp.Value(ctx) + closeAgg(ctx, tmp) + } +} + +func (s *twoStack[T]) Close() error { + var errs []error + if c, ok := any(s.valuesAgg).(io.Closer); ok { + errs = append(errs, c.Close()) + } + for _, v := range s.values { + if c, ok := any(v).(io.Closer); ok { + errs = append(errs, c.Close()) + } + } + for _, v := range s.cumAggs { + if c, ok := any(v).(io.Closer); ok { + errs = append(errs, c.Close()) + } + } + *s = twoStack[T]{} + return errors.Join(errs...) +} + +func closeAgg[T any](ctx sqlite3.Context, a T) { + if c, ok := any(a).(io.Closer); ok { + if err := c.Close(); err != nil { + ctx.ResultError(err) // notest + } + } +} diff --git a/util/sql3util/combiner_test.go b/util/sql3util/combiner_test.go new file mode 100644 index 00000000..0371a2de --- /dev/null +++ b/util/sql3util/combiner_test.go @@ -0,0 +1,77 @@ +package sql3util_test + +import ( + "fmt" + "log" + + "github.com/ncruces/go-sqlite3" + "github.com/ncruces/go-sqlite3/util/sql3util" +) + +func ExampleWindowFunction() { + db, err := sqlite3.Open(":memory:") + if err != nil { + log.Fatal(err) + } + defer db.Close() + + err = db.Exec(`CREATE TABLE flags (mask INTEGER)`) + if err != nil { + log.Fatal(err) + } + + err = db.Exec(`INSERT INTO flags VALUES (1), (2), (4), (4), (8)`) + if err != nil { + log.Fatal(err) + } + + err = db.CreateWindowFunction("bit_or", 1, sqlite3.DETERMINISTIC|sqlite3.INNOCUOUS, + sql3util.WindowFunction(func() *bitOr { return &bitOr{} })) + if err != nil { + log.Fatal(err) + } + + stmt, _, err := db.Prepare(`SELECT bit_or(mask) OVER (ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) FROM flags`) + if err != nil { + log.Fatal(err) + } + defer stmt.Close() + + for stmt.Step() { + fmt.Printf("%04b\n", stmt.ColumnInt64(0)) + } + if err := stmt.Err(); err != nil { + log.Fatal(err) + } + // Output: + // 0011 + // 0111 + // 0110 + // 1100 + // 1100 +} + +type bitOr struct { + val int64 + ok bool +} + +func (b *bitOr) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + if arg[0].NumericType() <= sqlite3.FLOAT { + b.val |= arg[0].Int64() + b.ok = true + } +} + +func (b *bitOr) Combine(other *bitOr) { + if other.ok { + b.val |= other.val + b.ok = true + } +} + +func (b *bitOr) Value(ctx sqlite3.Context) { + if b.ok { + ctx.ResultInt64(b.val) + } +} From cbdf78e016f04909429cd2d8a3ea0e095c0048c6 Mon Sep 17 00:00:00 2001 From: Nuno Cruces Date: Thu, 22 Oct 2026 16:51:59 +0100 Subject: [PATCH 2/3] Bitwise aggregates. --- context.go | 3 + ext/stats/bitwise.go | 97 ++++++++++++++++++++++++++++++ ext/stats/bitwise_test.go | 106 +++++++++++++++++++++++++++++++++ ext/stats/boolean.go | 10 ++-- ext/stats/stats | 13 ---- ext/stats/stats.go | 6 ++ ext/unicode/unicode.go | 2 +- func.go | 2 +- util/sql3util/combiner.go | 104 +++++++++++++++++++++++++------- util/sql3util/combiner_test.go | 70 ++++++++++++++++------ 10 files changed, 350 insertions(+), 63 deletions(-) create mode 100644 ext/stats/bitwise.go create mode 100644 ext/stats/bitwise_test.go delete mode 100644 ext/stats/stats diff --git a/context.go b/context.go index 4ed13985..2f9a052f 100644 --- a/context.go +++ b/context.go @@ -187,6 +187,9 @@ func (ctx Context) ResultValue(value Value) { // // https://sqlite.org/c3ref/result_blob.html func (ctx Context) ResultError(err error) { + if err == nil { + return + } if errors.Is(err, NOMEM) { ctx.c.wrp.Xsqlite3_result_error_nomem(int32(ctx.handle)) return diff --git a/ext/stats/bitwise.go b/ext/stats/bitwise.go new file mode 100644 index 00000000..33beff11 --- /dev/null +++ b/ext/stats/bitwise.go @@ -0,0 +1,97 @@ +package stats + +import ( + "github.com/ncruces/go-sqlite3" + "github.com/ncruces/go-sqlite3/util/sql3util" +) + +func newBitAnd() sqlite3.AggregateConstructor { + return sql3util.WindowFromCombiner(1e6, func() *bitAnd { + return &bitAnd{val: -1} + }) +} + +type bitAnd struct { + val int64 + ok bool +} + +func (b *bitAnd) Value(ctx sqlite3.Context) { + if b.ok { + ctx.ResultInt64(b.val) + } +} + +func (b *bitAnd) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + if arg[0].NumericType() <= sqlite3.FLOAT { + b.val &= arg[0].Int64() + b.ok = true + } +} + +func (b *bitAnd) Combine(other *bitAnd) { + if other.ok { + b.val &= other.val + b.ok = true + } +} + +func newBitOr() sqlite3.AggregateConstructor { + return sql3util.WindowFromCombiner(1e6, func() *bitOr { + return &bitOr{} + }) +} + +type bitOr struct { + val int64 + ok bool +} + +func (b *bitOr) Value(ctx sqlite3.Context) { + if b.ok { + ctx.ResultInt64(b.val) + } +} + +func (b *bitOr) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + if arg[0].NumericType() <= sqlite3.FLOAT { + b.val |= arg[0].Int64() + b.ok = true + } +} + +func (b *bitOr) Combine(other *bitOr) { + if other.ok { + b.val |= other.val + b.ok = true + } +} + +func newBitXor() sqlite3.AggregateFunction { + return &bitXor{} +} + +type bitXor struct { + val int64 + count int +} + +func (b *bitXor) Value(ctx sqlite3.Context) { + if b.count > 0 { + ctx.ResultInt64(b.val) + } +} + +func (b *bitXor) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + if arg[0].NumericType() <= sqlite3.FLOAT { + b.val ^= arg[0].Int64() + b.count++ + } +} + +func (b *bitXor) Inverse(ctx sqlite3.Context, arg ...sqlite3.Value) { + if arg[0].NumericType() <= sqlite3.FLOAT { + b.val ^= arg[0].Int64() + b.count-- + } +} diff --git a/ext/stats/bitwise_test.go b/ext/stats/bitwise_test.go new file mode 100644 index 00000000..7cd27dff --- /dev/null +++ b/ext/stats/bitwise_test.go @@ -0,0 +1,106 @@ +package stats_test + +import ( + "testing" + + "github.com/ncruces/go-sqlite3" + "github.com/ncruces/go-sqlite3/internal/testcfg" +) + +func TestRegister_bitwise(t *testing.T) { + t.Parallel() + + db, err := sqlite3.OpenContext(testcfg.Context(t), ":memory:") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + err = db.Exec(`CREATE TABLE data (x)`) + if err != nil { + t.Fatal(err) + } + + stmt, _, err := db.Prepare(`SELECT bit_and(x), bit_or(x), bit_xor(x) FROM data`) + if err != nil { + t.Fatal(err) + } + if !stmt.Step() { + t.Fatal(stmt.Err()) + } else { + for i := range 3 { + if got := stmt.ColumnType(i); got != sqlite3.NULL { + t.Errorf("got %v, want NULL", got) + } + } + } + stmt.Close() + + err = db.Exec(`INSERT INTO data (x) VALUES (3), (6.0), ('12'), (10), (NULL), ('abc')`) + if err != nil { + t.Fatal(err) + } + + stmt, _, err = db.Prepare(` + SELECT + bit_and(x), + bit_and(x | 1), + bit_or(x), + bit_xor(x) + FROM data`) + if err != nil { + t.Fatal(err) + } + if !stmt.Step() { + t.Fatal(stmt.Err()) + } else { + if got := stmt.ColumnInt64(0); got != 0 { + t.Errorf("got %v, want 0", got) + } + if got := stmt.ColumnInt64(1); got != 1 { + t.Errorf("got %v, want 1", got) + } + if got := stmt.ColumnInt64(2); got != 15 { + t.Errorf("got %v, want 15", got) + } + if got := stmt.ColumnInt64(3); got != 3 { + t.Errorf("got %v, want 3", got) + } + } + stmt.Close() + + stmt, _, err = db.Prepare(` + SELECT + bit_and(x) OVER (ROWS 1 PRECEDING), + bit_or(x) OVER (ROWS 1 PRECEDING), + bit_xor(x) OVER (ROWS 1 PRECEDING) + FROM data`) + if err != nil { + t.Fatal(err) + } + + wantAnd := [...]int64{3, 2, 4, 8, 10, 0} + wantOr := [...]int64{3, 7, 14, 14, 10, 0} + wantXor := [...]int64{3, 5, 10, 6, 10, 0} + for i := 0; stmt.Step(); i++ { + if got := stmt.ColumnInt64(0); got != wantAnd[i] { + t.Errorf("row %d: got %v, want %v", i, got, wantAnd[i]) + } + if got := stmt.ColumnInt64(1); got != wantOr[i] { + t.Errorf("row %d: got %v, want %v", i, got, wantOr[i]) + } + if got := stmt.ColumnInt64(2); got != wantXor[i] { + t.Errorf("row %d: got %v, want %v", i, got, wantXor[i]) + } + wantType := sqlite3.INTEGER + if i == 5 { + wantType = sqlite3.NULL + } + for c := range 3 { + if got := stmt.ColumnType(c); got != wantType { + t.Errorf("row %d col %d: got %v, want %v", i, c, got, wantType) + } + } + } + stmt.Close() +} diff --git a/ext/stats/boolean.go b/ext/stats/boolean.go index 7aa5aaf7..9ab622e8 100644 --- a/ext/stats/boolean.go +++ b/ext/stats/boolean.go @@ -26,21 +26,19 @@ func (b *boolean) Value(ctx sqlite3.Context) { } func (b *boolean) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { - a := arg[0] - if a.Bool() { + if arg[0].Bool() { b.count++ } - if a.Type() != sqlite3.NULL { + if arg[0].Type() != sqlite3.NULL { b.total++ } } func (b *boolean) Inverse(ctx sqlite3.Context, arg ...sqlite3.Value) { - a := arg[0] - if a.Bool() { + if arg[0].Bool() { b.count-- } - if a.Type() != sqlite3.NULL { + if arg[0].Type() != sqlite3.NULL { b.total-- } } diff --git a/ext/stats/stats b/ext/stats/stats deleted file mode 100644 index 00b7cc46..00000000 --- a/ext/stats/stats +++ /dev/null @@ -1,13 +0,0 @@ -goos: linux -goarch: amd64 -pkg: github.com/ncruces/go-sqlite3/ext/stats -cpu: Intel(R) Xeon(R) W-2135 CPU @ 3.70GHz - │ before │ after │ after-unsafe │ - │ sec/op │ sec/op vs base │ sec/op vs base │ -_average-12 128.5n ± 12% 183.6n ± 9% +42.88% (p=0.000 n=20) 139.2n ± 8% ~ (p=0.056 n=20) -_variance-12 323.6n ± 11% 248.7n ± 9% -23.16% (p=0.000 n=20) 194.4n ± 12% -39.93% (p=0.000 n=20) -_math/sqrt-12 169.3n ± 17% 286.6n ± 6% +69.21% (p=0.000 n=20) 185.5n ± 5% ~ (p=0.072 n=20) -_math/tan-12 207.8n ± 6% 302.7n ± 4% +45.69% (p=0.000 n=20) 207.9n ± 4% ~ (p=0.533 n=20) -_math/cot-12 465.1n ± 9% 365.0n ± 3% -21.51% (p=0.000 n=20) 260.8n ± 5% -43.93% (p=0.000 n=20) -_math/cbrt-12 434.8n ± 10% 344.4n ± 4% -20.79% (p=0.000 n=20) 262.9n ± 6% -39.55% (p=0.000 n=20) -geomean 258.1n 281.5n +9.06% 203.7n -21.08% diff --git a/ext/stats/stats.go b/ext/stats/stats.go index fa6c6b5a..af1542a5 100644 --- a/ext/stats/stats.go +++ b/ext/stats/stats.go @@ -29,6 +29,9 @@ // - mode: most frequent value // - every: boolean and // - some: boolean or +// - bit_and: bitwise and +// - bit_or: bitwise or +// - bit_xor: bitwise xor // // These join the [Built-in Aggregate Functions]: // - count: count rows/values @@ -91,6 +94,9 @@ func Register(db *sqlite3.Conn) error { db.CreateWindowFunction("percentile_disc", 2, order, newPercentile(percentile_disc)), db.CreateWindowFunction("every", 1, flags, newBoolean(every)), db.CreateWindowFunction("some", 1, flags, newBoolean(some)), + db.CreateWindowFunction("bit_and", 1, flags, newBitAnd()), + db.CreateWindowFunction("bit_or", 1, flags, newBitOr()), + db.CreateWindowFunction("bit_xor", 1, flags, newBitXor), db.CreateWindowFunction("mode", 1, order, newMode), db.CreateFunction("cbrt", 1, flags, cbrt), db.CreateFunction("cot", 1, flags, cot)) diff --git a/ext/unicode/unicode.go b/ext/unicode/unicode.go index 2571b54e..5bc34180 100644 --- a/ext/unicode/unicode.go +++ b/ext/unicode/unicode.go @@ -228,7 +228,7 @@ func like(ctx sqlite3.Context, arg ...sqlite3.Value) { b := arg[2].RawText() escape, size = utf8.DecodeRune(b) if size != len(b) { - ctx.ResultError(errutil.ErrorString("ESCAPE expression must be a single character")) + ctx.ResultError(errutil.ErrorString("unicode: ESCAPE expression must be a single character")) return } } diff --git a/func.go b/func.go index c96d22d1..2dfde038 100644 --- a/func.go +++ b/func.go @@ -226,7 +226,7 @@ func (e env) Xgo_inverse(pCtx, pAgg, nArg, pArg int32) { fn, ok := db.wrp.GetHandle(ptr_t(pAgg)).(WindowFunction) if !ok { Context{db, ptr_t(pCtx)}.ResultError( - errutil.ErrorString("may not be used as a window function")) + errutil.ErrorString("sqlite3: may not be used as a window function")) return // notest } fn.Inverse(Context{db, ptr_t(pCtx)}, *args...) diff --git a/util/sql3util/combiner.go b/util/sql3util/combiner.go index 8827a01f..c05a53d7 100644 --- a/util/sql3util/combiner.go +++ b/util/sql3util/combiner.go @@ -1,28 +1,46 @@ package sql3util import ( - "errors" + "cmp" "io" "github.com/ncruces/go-sqlite3" + "github.com/ncruces/go-sqlite3/internal/errutil" ) // WindowCombiner is the interface an aggregate function should implement -// to be used with [WindowFunction]. +// to be used with [WindowFromCombiner]. type WindowCombiner[T sqlite3.AggregateFunction] interface { sqlite3.AggregateFunction - // Combine merges the state of other into the receiver. + // Combine merges the state of other into the receiver, + // as if the rows aggregated by other had been added after + // the rows aggregated by the receiver, in window order. + // + // Combine must be associative, but need not be commutative: + // a.Combine(b) followed by a.Combine(c) must be equivalent to + // b.Combine(c) followed by a.Combine(b). + // + // Both receiver and other may be empty, with no rows aggregated: + // empty.Combine(x) and x.Combine(empty) must be equivalent to x. + // // Implementations must not modify or retain other. Combine(other T) } -// WindowFunction adapts a [WindowCombiner] constructor into an [sqlite3.AggregateConstructor] +// WindowFromCombiner adapts a [WindowCombiner] constructor into an [sqlite3.AggregateConstructor] // that implements [sqlite3.WindowFunction] using the two-stack sliding-window algorithm. -// If fn returns an [io.Closer], it will be called to free resources. // // https://orlp.net/blog/two-stack-sliding-window-aggregation/ -func WindowFunction[T WindowCombiner[T]](fn func() T) sqlite3.AggregateConstructor { +// +// If maxWindow > 0 and the window grows larger than maxWindow rows, +// the function degrades to a standard aggregate and may no longer be used as a sliding window. +// If T implements [io.Closer], it will be called to free resources. +// +// This is intended to help you implement aggregate window functions +// from primitives that are combinable (aka. mergeable) but not invertible. +// Some examples are: HyperLogLog, t-digest, KLL, CM and DD sketches, etc. +func WindowFromCombiner[T WindowCombiner[T]](maxWindow int, fn func() T) sqlite3.AggregateConstructor { if fn == nil { return nil } @@ -30,6 +48,7 @@ func WindowFunction[T WindowCombiner[T]](fn func() T) sqlite3.AggregateConstruct return &twoStack[T]{ ctor: fn, valuesAgg: fn(), + maxWindow: maxWindow, } } } @@ -39,34 +58,53 @@ type twoStack[T WindowCombiner[T]] struct { cumAggs []T values []T valuesAgg T + maxWindow int } func (s *twoStack[T]) Step(ctx sqlite3.Context, arg ...sqlite3.Value) { + if s.maxWindow > 0 && len(s.cumAggs)+len(s.values) >= s.maxWindow { + // Window exceeded maxWindow; degrade to a standard aggregate. + s.unwindow(ctx) + } + if s.ctor == nil { + // Degraded to a standard aggregate. + s.valuesAgg.Step(ctx, arg...) + return + } + unit := s.ctor() - unit.Step(ctx, arg...) s.values = append(s.values, unit) + unit.Step(ctx, arg...) s.valuesAgg.Combine(unit) } func (s *twoStack[T]) Inverse(ctx sqlite3.Context, arg ...sqlite3.Value) { + if s.ctor == nil { + // Degraded to a standard aggregate. + ctx.ResultError(errutil.ErrorString("combiner: window size exceeded maximum")) + return + } if len(s.cumAggs) > 0 { + // Pop the oldest cumulative aggregate. last := len(s.cumAggs) - 1 closeAgg(ctx, s.cumAggs[last]) - s.cumAggs[last] = *new(T) + clear(s.cumAggs[last:]) s.cumAggs = s.cumAggs[:last] return } - closeAgg(ctx, s.values[0]) + // Ignore the oldest value and accumulate the rest in reverse order. + for i := len(s.values) - 2; i >= 1; i-- { + s.values[i].Combine(s.values[i+1]) + } for i := len(s.values) - 1; i >= 1; i-- { - if i < len(s.values)-1 { - s.values[i].Combine(s.values[i+1]) - } s.cumAggs = append(s.cumAggs, s.values[i]) } + + // Cleanup. + closeAgg(ctx, s.valuesAgg, s.values[0]) clear(s.values) s.values = s.values[:0] - closeAgg(ctx, s.valuesAgg) s.valuesAgg = s.ctor() } @@ -77,37 +115,57 @@ func (s *twoStack[T]) Value(ctx sqlite3.Context) { case len(s.values) == 0: s.cumAggs[len(s.cumAggs)-1].Value(ctx) default: + // Combine the oldest cumulative aggregate with the values aggregate. tmp := s.ctor() + defer closeAgg(ctx, tmp) tmp.Combine(s.cumAggs[len(s.cumAggs)-1]) tmp.Combine(s.valuesAgg) tmp.Value(ctx) - closeAgg(ctx, tmp) } } func (s *twoStack[T]) Close() error { - var errs []error + var err error if c, ok := any(s.valuesAgg).(io.Closer); ok { - errs = append(errs, c.Close()) + err = cmp.Or(err, c.Close()) } for _, v := range s.values { if c, ok := any(v).(io.Closer); ok { - errs = append(errs, c.Close()) + err = cmp.Or(err, c.Close()) } } for _, v := range s.cumAggs { if c, ok := any(v).(io.Closer); ok { - errs = append(errs, c.Close()) + err = cmp.Or(err, c.Close()) } } *s = twoStack[T]{} - return errors.Join(errs...) + return err } -func closeAgg[T any](ctx sqlite3.Context, a T) { - if c, ok := any(a).(io.Closer); ok { - if err := c.Close(); err != nil { - ctx.ResultError(err) // notest +func (s *twoStack[T]) unwindow(ctx sqlite3.Context) { + if last := len(s.cumAggs) - 1; last >= 0 { + // Merge the oldest cumulative aggregate with valuesAgg. + agg := s.cumAggs[last] + agg.Combine(s.valuesAgg) + closeAgg(ctx, s.valuesAgg) + s.valuesAgg = agg + s.cumAggs = s.cumAggs[:last] + } + // Free stacks and clear ctor to mark as degraded. + closeAgg(ctx, s.values...) + closeAgg(ctx, s.cumAggs...) + *s = twoStack[T]{valuesAgg: s.valuesAgg} +} + +func closeAgg[T any](ctx sqlite3.Context, a ...T) { + var err error + for _, v := range a { + if c, ok := any(v).(io.Closer); ok { + if e := c.Close(); err == nil { + err = e + } } } + ctx.ResultError(err) } diff --git a/util/sql3util/combiner_test.go b/util/sql3util/combiner_test.go index 0371a2de..f8f93169 100644 --- a/util/sql3util/combiner_test.go +++ b/util/sql3util/combiner_test.go @@ -3,44 +3,46 @@ package sql3util_test import ( "fmt" "log" + "testing" "github.com/ncruces/go-sqlite3" + "github.com/ncruces/go-sqlite3/driver" + "github.com/ncruces/go-sqlite3/internal/testcfg" "github.com/ncruces/go-sqlite3/util/sql3util" ) -func ExampleWindowFunction() { - db, err := sqlite3.Open(":memory:") +func ExampleWindowFromCombiner() { + db, err := driver.Open(":memory:", func(c *sqlite3.Conn) error { + return c.CreateWindowFunction("bit_or", 1, sqlite3.DETERMINISTIC|sqlite3.INNOCUOUS, + sql3util.WindowFromCombiner(100, func() *bitOr { return &bitOr{} })) + }) if err != nil { log.Fatal(err) } defer db.Close() - err = db.Exec(`CREATE TABLE flags (mask INTEGER)`) + _, err = db.Exec(` + CREATE TABLE flags (mask INTEGER); + INSERT INTO flags VALUES (1), (2), (4), (4), (8); + `) if err != nil { log.Fatal(err) } - err = db.Exec(`INSERT INTO flags VALUES (1), (2), (4), (4), (8)`) + rows, err := db.Query(`SELECT bit_or(mask) OVER (ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) FROM flags`) if err != nil { log.Fatal(err) } + defer rows.Close() - err = db.CreateWindowFunction("bit_or", 1, sqlite3.DETERMINISTIC|sqlite3.INNOCUOUS, - sql3util.WindowFunction(func() *bitOr { return &bitOr{} })) - if err != nil { - log.Fatal(err) - } - - stmt, _, err := db.Prepare(`SELECT bit_or(mask) OVER (ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) FROM flags`) - if err != nil { - log.Fatal(err) - } - defer stmt.Close() - - for stmt.Step() { - fmt.Printf("%04b\n", stmt.ColumnInt64(0)) + for rows.Next() { + var mask int64 + if err := rows.Scan(&mask); err != nil { + log.Fatal(err) + } + fmt.Printf("%04b\n", mask) } - if err := stmt.Err(); err != nil { + if err := rows.Err(); err != nil { log.Fatal(err) } // Output: @@ -75,3 +77,33 @@ func (b *bitOr) Value(ctx sqlite3.Context) { ctx.ResultInt64(b.val) } } + +func TestWindowFromCombiner_overflow(t *testing.T) { + t.Parallel() + + ctx := testcfg.Context(t) + db, err := driver.Open(":memory:", func(c *sqlite3.Conn) error { + return c.CreateWindowFunction("bit_or", 1, sqlite3.DETERMINISTIC|sqlite3.INNOCUOUS, + sql3util.WindowFromCombiner(100, func() *bitOr { return &bitOr{} })) + }) + if err != nil { + t.Fatal(err) + } + defer db.Close() + + // Standard aggregate exceeding maxWindow degrades gracefully. + var got int64 + err = db.QueryRowContext(ctx, `SELECT bit_or(value) FROM generate_series(1, 113)`).Scan(&got) + if err != nil { + t.Fatal(err) + } + if got != 127 { + t.Errorf("got %b, want 127", got) + } + + // Sliding window exceeding maxWindow errors on subsequent Inverse. + _, err = db.ExecContext(ctx, `SELECT bit_or(value) OVER (ROWS 100 PRECEDING) FROM generate_series(1, 113)`) + if err == nil { + t.Error("want error") + } +} From 737bfc467578ec32caaeaca1996b3d3f97369f5b Mon Sep 17 00:00:00 2001 From: Nuno Cruces Date: Fri, 23 Oct 2026 11:03:44 +0100 Subject: [PATCH 3/3] Comments. --- util/sql3util/combiner.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/util/sql3util/combiner.go b/util/sql3util/combiner.go index c05a53d7..bf224881 100644 --- a/util/sql3util/combiner.go +++ b/util/sql3util/combiner.go @@ -18,28 +18,31 @@ type WindowCombiner[T sqlite3.AggregateFunction] interface { // the rows aggregated by the receiver, in window order. // // Combine must be associative, but need not be commutative: - // a.Combine(b) followed by a.Combine(c) must be equivalent to - // b.Combine(c) followed by a.Combine(b). + // a.Combine(b); a.Combine(c) must be equivalent to + // b.Combine(c); a.Combine(b). // // Both receiver and other may be empty, with no rows aggregated: - // empty.Combine(x) and x.Combine(empty) must be equivalent to x. + // x.Combine(empty) must be equivalent to x. // // Implementations must not modify or retain other. Combine(other T) } // WindowFromCombiner adapts a [WindowCombiner] constructor into an [sqlite3.AggregateConstructor] -// that implements [sqlite3.WindowFunction] using the two-stack sliding-window algorithm. -// +// that implements [sqlite3.WindowFunction] using the two-stack sliding-window algorithm: // https://orlp.net/blog/two-stack-sliding-window-aggregation/ // +// The memory efficiency of the algorithm depends on [WindowCombiner] +// efficiently representing small aggregates, +// particularly empty or single-row states. +// // If maxWindow > 0 and the window grows larger than maxWindow rows, // the function degrades to a standard aggregate and may no longer be used as a sliding window. // If T implements [io.Closer], it will be called to free resources. // -// This is intended to help you implement aggregate window functions -// from primitives that are combinable (aka. mergeable) but not invertible. -// Some examples are: HyperLogLog, t-digest, KLL, CM and DD sketches, etc. +// It helps implement aggregate window functions over primitives +// that are combinable (or mergeable) but not invertible. +// Typical examples are mergeable sketch algorithms like HyperLogLog. func WindowFromCombiner[T WindowCombiner[T]](maxWindow int, fn func() T) sqlite3.AggregateConstructor { if fn == nil { return nil