This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new 9fb8a8fb6 Introduce reusable `Prepared` QL binding (#1213)
9fb8a8fb6 is described below
commit 9fb8a8fb63336916690adcbcbd4600a67ed26814
Author: mrproliu <[email protected]>
AuthorDate: Mon Jul 13 08:49:33 2026 +0800
Introduce reusable `Prepared` QL binding (#1213)
* Introduce reusable `Prepared` QL binding
---
CHANGES.md | 1 +
banyand/liaison/grpc/bydbql.go | 119 +++++++++-
banyand/liaison/grpc/bydbql_cache.go | 142 +++++++++++
banyand/liaison/grpc/bydbql_cache_test.go | 199 ++++++++++++++++
banyand/liaison/grpc/bydbql_test.go | 50 +++-
banyand/liaison/grpc/metrics.go | 14 ++
banyand/liaison/grpc/server.go | 24 +-
banyand/liaison/grpc/topk.go | 132 +++++++++++
banyand/liaison/grpc/topk_test.go | 124 ++++++++++
docs/operation/configuration.md | 27 +++
pkg/bydbql/binder.go | 157 +++++++++----
pkg/bydbql/binder_equivalence_test.go | 79 ++++++-
pkg/bydbql/binder_test.go | 2 +-
pkg/bydbql/grammar.go | 34 ++-
pkg/bydbql/prepared.go | 289 +++++++++++++++++++++++
pkg/bydbql/prepared_internal_test.go | 335 ++++++++++++++++++++++++++
pkg/bydbql/transformer.go | 377 +++++++++++++++++++-----------
17 files changed, 1897 insertions(+), 208 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 41467ba54..1ae9f7a1b 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -78,6 +78,7 @@ Release Notes.
- Add bounded plugin telemetry (meter + logger) to the trace-pipeline sampler
SDK. A sampler plugin opts in by implementing `sdk.HostAware`; the engine calls
`UseHost(h)` once per group before the first `Decide`, delivering a scoped
`sdk.Host` the plugin may cache and use in `Decide`. The host enforces all
resource limits: metric names are forced under the
`banyandb_trace_pipeline_plugin_` prefix; `{group, plugin_name}` labels are
always present; per-plugin cardinality is capped at 100 s [...]
- Add a Claude/Codex plugin packaging the BanyanDB MCP server with a
parse-only `validate_bydbql` tool (backed by the `mcp/tools/bydbql-parse` Go
validator) and a BydbQL skill for read-only natural-language-to-BydbQL
generation over STREAM/MEASURE/TRACE/PROPERTY resources.
- Introduce positional parameter binding (`?` placeholders) into BydbQL to
eliminate QL injection.
+- Add reusable BydbQL binding: Prepare a query once, then Bind it many times
without re-parsing or mutating the template. The liaison caches prepared
statements on the gRPC query path (LRU bounded by entry count and bytes, on by
default; `--bydbql-prepared-cache-size`/`--bydbql-prepared-cache-max-bytes`,
`bydbql_prepared_cache_*` metrics) so repeated templates skip parsing. To
pinpoint un-cacheable and slow queries without high-cardinality labels, the
query access log tags each entry by [...]
### Bug Fixes
diff --git a/banyand/liaison/grpc/bydbql.go b/banyand/liaison/grpc/bydbql.go
index e22420736..97e714ac9 100644
--- a/banyand/liaison/grpc/bydbql.go
+++ b/banyand/liaison/grpc/bydbql.go
@@ -35,6 +35,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/accesslog"
"github.com/apache/skywalking-banyandb/pkg/bydbql"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/run"
)
type bydbQLService struct {
@@ -44,10 +45,15 @@ type bydbQLService struct {
metrics *metrics
repo metadata.Repo
transformer *bydbql.Transformer
+ cache *preparedCache
streamSvc *streamService
measureSvc *measureService
traceSvc *traceService
propertyServer *propertyServer
+ // dumper tracks and periodically logs the top cache-miss and slow
queries; it is
+ // nil (its observe methods no-op) unless the periodic top-K log is
enabled.
+ dumper *topKDumper
+ slowThreshold time.Duration
}
func (b *bydbQLService) setLogger(log *logger.Logger) {
@@ -65,6 +71,10 @@ func (b *bydbQLService) activeQueryAccessLog(root string,
sampled bool) (err err
func (b *bydbQLService) Query(ctx context.Context, req *bydbqlv1.QueryRequest)
(resp *bydbqlv1.QueryResponse, err error) {
start := time.Now()
b.metrics.totalStarted.Inc(1, "", "bydbql", "query")
+ // cacheResult tags the access-log entry with the prepared-statement
cache
+ // outcome so operators can find un-cached queries: entries logged under
+ // "bydbql-miss" / "bydbql-bypass" are the ones that did not hit the
cache.
+ var cacheResult string
defer func() {
duration := time.Since(start)
b.metrics.totalFinished.Inc(1, "", "bydbql", "query")
@@ -73,23 +83,36 @@ func (b *bydbQLService) Query(ctx context.Context, req
*bydbqlv1.QueryRequest) (
}
b.metrics.totalLatency.Inc(duration.Seconds(), "", "bydbql",
"query")
+ if b.slowThreshold > 0 && duration > b.slowThreshold {
+ b.metrics.bydbqlSlowQueryTotal.Inc(1)
+ b.dumper.observeSlow(req.Query, duration)
+ }
+
if b.queryAccessLog != nil {
- if errAccessLog :=
b.queryAccessLog.WriteQuery("bydbql", start, duration, req, err); errAccessLog
!= nil {
+ service := "bydbql"
+ if cacheResult != "" {
+ service = "bydbql-" + cacheResult
+ }
+ if errAccessLog := b.queryAccessLog.WriteQuery(service,
start, duration, req, err); errAccessLog != nil {
b.l.Error().Err(errAccessLog).Msg("bydbql
access log error")
}
}
}()
- // parse query, bind parameters, and transform to native request
+ // prepare (parse once, cached), bind parameters, and transform to
native request
parseStart := time.Now()
- query, err := bydbql.ParseQuery(req.Query)
+ stmt, cacheResult, err := b.cache.getOrPrepare(req.Query)
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "failed to
parse query: %v", err)
}
- if err = bydbql.BindParams(query, req.Params); err != nil {
+ if cacheResult == "miss" {
+ b.dumper.observeMiss(req.Query)
+ }
+ bound, err := stmt.Bind(req.Params)
+ if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "failed to
bind parameters: %v", err)
}
- result, err := b.transformer.Transform(ctx, query)
+ result, err := b.transformer.TransformBound(ctx, bound)
if err != nil {
return nil, status.Errorf(codes.Internal, "failed to transform
to native request: %v", err)
}
@@ -143,6 +166,92 @@ func (b *bydbQLService) Query(ctx context.Context, req
*bydbqlv1.QueryRequest) (
return resp, nil
}
+// topKDumper tracks the top cache-miss and slow queries and, on a supervised
+// goroutine, periodically logs the cumulative top-K. All methods are
nil-safe, so the
+// call sites need no guards when the top-K log is disabled (the dumper is
nil).
+type topKDumper struct {
+ miss *topK
+ slow *topK
+ l *logger.Logger
+ cancel context.CancelFunc
+}
+
+// newTopKDumper starts the trackers and the dump goroutine; a non-positive
interval
+// disables the feature and returns nil.
+func newTopKDumper(interval time.Duration, l *logger.Logger) *topKDumper {
+ if interval <= 0 {
+ return nil
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ d := &topKDumper{miss: newTopK(bydbqlTopKSize), slow:
newTopK(bydbqlTopKSize), l: l, cancel: cancel}
+ run.Go(ctx, "liaison.grpc.bydbql.topk-dump", l, func(ctx
context.Context) {
+ ticker := time.NewTicker(interval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ d.dump()
+ }
+ }
+ })
+ return d
+}
+
+func (d *topKDumper) observeMiss(query string) {
+ if d != nil {
+ d.miss.observe(query, 0)
+ }
+}
+
+func (d *topKDumper) observeSlow(query string, dur time.Duration) {
+ if d != nil {
+ d.slow.observe(query, dur)
+ }
+}
+
+func (d *topKDumper) close() {
+ if d != nil && d.cancel != nil {
+ d.cancel()
+ }
+}
+
+// minReparseMisses is the smallest cumulative miss count worth logging. Every
+// parameterized template misses exactly once on its cold-start lookup, so
count==1 is
+// benign; only count>=2 means the template was evicted and re-parsed
(thrashing).
+const minReparseMisses = 2
+
+func (d *topKDumper) dump() {
+ d.logTopK(d.miss.snapshot(), minReparseMisses, "top bydbql cache-miss
queries", func(s topKSlot) string {
+ return fmt.Sprintf("%q count=%d", s.key, s.count)
+ })
+ d.logTopK(d.slow.snapshotByLatency(), 1, "top bydbql slow queries",
func(s topKSlot) string {
+ return fmt.Sprintf("%q count=%d max_latency=%s", s.key,
s.count, s.maxDur)
+ })
+}
+
+// logTopK logs the entries with at least minCount occurrences, formatted by
line.
+func (d *topKDumper) logTopK(entries []topKSlot, minCount uint64, msg string,
line func(topKSlot) string) {
+ lines := formatTopK(entries, minCount, line)
+ if len(lines) == 0 {
+ return
+ }
+ d.l.Info().Strs("top", lines).Msg(msg)
+}
+
+// formatTopK renders the entries with at least minCount occurrences into log
lines.
+func formatTopK(entries []topKSlot, minCount uint64, line func(topKSlot)
string) []string {
+ lines := make([]string, 0, len(entries))
+ for _, s := range entries {
+ if s.count < minCount {
+ continue
+ }
+ lines = append(lines, line(s))
+ }
+ return lines
+}
+
func (b *bydbQLService) Close() error {
if b.queryAccessLog != nil {
if err := b.queryAccessLog.Close(); err != nil {
diff --git a/banyand/liaison/grpc/bydbql_cache.go
b/banyand/liaison/grpc/bydbql_cache.go
new file mode 100644
index 000000000..ba03a96f7
--- /dev/null
+++ b/banyand/liaison/grpc/bydbql_cache.go
@@ -0,0 +1,142 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package grpc
+
+import (
+ "sync/atomic"
+
+ lru "github.com/hashicorp/golang-lru"
+
+ "github.com/apache/skywalking-banyandb/pkg/bydbql"
+)
+
+// cacheValue is a cached prepared statement plus its accounted byte cost
(query
+// text plus the estimated statement footprint), so evictions can adjust
curBytes.
+type cacheValue struct {
+ ps *bydbql.PreparedStatement
+ cost int
+}
+
+// preparedCache is a concurrency-safe LRU over parsed BydbQL prepared
statements,
+// keyed by the exact query text. A cached statement is immutable and safe to
Bind
+// concurrently, and the parse result depends only on the query text, so
entries
+// never need invalidation. The count bound is enforced by the underlying LRU;
the
+// byte bound (accounted key text plus estimated statement size) is enforced
on top
+// via curBytes and RemoveOldest. All shared state is either behind the LRU's
own
+// lock or atomic, so the cache needs no mutex of its own.
+type preparedCache struct {
+ metrics *metrics
+ lru *lru.Cache
+ maxBytes int
+ curBytes atomic.Int64
+ hits atomic.Uint64
+ misses atomic.Uint64
+}
+
+// newPreparedCache builds a cache holding up to size entries and maxBytes of
+// accounted size. A size <= 0 disables caching (lru is nil): getOrPrepare then
+// parses every request and bypasses the cache without recording hits or
misses.
+// A maxBytes <= 0 drops the byte bound, leaving only the count.
+func newPreparedCache(size, maxBytes int, m *metrics) *preparedCache {
+ c := &preparedCache{maxBytes: maxBytes, metrics: m}
+ if size > 0 {
+ // onEvict keeps curBytes in step with both the LRU's own
count-based
+ // evictions and our byte-based RemoveOldest calls; it runs
without the
+ // cache lock, so the atomic add is safe.
+ if l, err := lru.NewWithEvict(size, func(_, value interface{}) {
+ c.curBytes.Add(-int64(value.(*cacheValue).cost))
+ }); err == nil {
+ c.lru = l
+ }
+ }
+ return c
+}
+
+// getOrPrepare returns the prepared statement for query, parsing and caching
it on
+// a miss. It records the cache metrics internally; the returned result ("hit",
+// "miss", "bypass", or "" for a disabled cache) lets the caller tag the query
+// access log so un-cached queries can be found. Callers handle only the parse
error.
+func (c *preparedCache) getOrPrepare(query string) (*bydbql.PreparedStatement,
string, error) {
+ if c.lru != nil {
+ if v, ok := c.lru.Get(query); ok {
+ c.hits.Add(1)
+ c.emit("hit")
+ return v.(*cacheValue).ps, "hit", nil
+ }
+ }
+ ps, err := bydbql.Prepare(query)
+ if err != nil {
+ // A parse failure is not a cache event: it is never cached (so
it cannot
+ // pin memory) and is already counted by the RPC-level error
metric.
+ // Excluding it keeps hit_ratio from being depressed by
malformed queries.
+ return nil, "", err
+ }
+ if c.lru == nil {
+ return ps, "", nil // caching disabled entirely; nothing to
record
+ }
+ // Only parameterized queries are reusable templates worth caching. A
literal
+ // query bakes its values into the text, so every distinct value is a
unique key
+ // that would never hit and would only evict useful templates. It
bypasses the
+ // cache; a bypass is counted separately from hit/miss so the hit ratio
still
+ // reflects only cacheable queries.
+ if ps.NumPlaceholders() == 0 {
+ c.emit("bypass")
+ return ps, "bypass", nil
+ }
+ c.misses.Add(1)
+ c.store(query, ps)
+ c.emit("miss")
+ return ps, "miss", nil
+}
+
+// store caches ps under the byte bound, skipping a statement that alone
exceeds
+// maxBytes. ContainsOrAdd inserts atomically only when the key is absent, so
+// concurrent misses on the same query count its bytes exactly once. Call only
for
+// an enabled cache and a parameterized query.
+func (c *preparedCache) store(query string, ps *bydbql.PreparedStatement) {
+ cost := len(query) + ps.EstimatedSize()
+ if c.maxBytes > 0 && cost > c.maxBytes {
+ return
+ }
+ if found, _ := c.lru.ContainsOrAdd(query, &cacheValue{ps: ps, cost:
cost}); found {
+ return
+ }
+ c.curBytes.Add(int64(cost))
+ for c.maxBytes > 0 && c.curBytes.Load() > int64(c.maxBytes) &&
c.lru.Len() > 1 {
+ c.lru.RemoveOldest()
+ }
+}
+
+// emit publishes the cache counter (labeled by result: hit/miss/bypass) and
the
+// gauges from the atomic counters.
+func (c *preparedCache) emit(result string) {
+ if c.metrics == nil {
+ return
+ }
+ hits, misses := c.hits.Load(), c.misses.Load()
+ c.metrics.bydbqlPreparedCacheTotal.Inc(1, result)
+ if total := hits + misses; total > 0 {
+ c.metrics.bydbqlPreparedCacheHitRatio.Set(float64(hits) /
float64(total))
+ }
+ entries := 0
+ if c.lru != nil {
+ entries = c.lru.Len()
+ }
+ c.metrics.bydbqlPreparedCacheCount.Set(float64(entries))
+ c.metrics.bydbqlPreparedCacheBytes.Set(float64(c.curBytes.Load()))
+}
diff --git a/banyand/liaison/grpc/bydbql_cache_test.go
b/banyand/liaison/grpc/bydbql_cache_test.go
new file mode 100644
index 000000000..3c2dcf368
--- /dev/null
+++ b/banyand/liaison/grpc/bydbql_cache_test.go
@@ -0,0 +1,199 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package grpc
+
+import (
+ "fmt"
+ "sync"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// bydbqlQuery is a distinct, parameterized (cacheable) query per i.
+func bydbqlQuery(i int) string {
+ return fmt.Sprintf("SELECT * FROM STREAM sw IN default WHERE service_id
= 'v%d' AND duration > ?", i)
+}
+
+// bydbqlLiteralQuery is a distinct query per i with no placeholder (not
cacheable).
+func bydbqlLiteralQuery(i int) string {
+ return fmt.Sprintf("SELECT * FROM STREAM sw IN default WHERE service_id
= 'v%d'", i)
+}
+
+// probeEntryCost returns the accounted byte cost of one cached entry (query
text
+// plus estimated statement size), so byte-bound tests stay valid as the
estimate
+// or query template changes.
+func probeEntryCost(t *testing.T) int {
+ t.Helper()
+ c := newPreparedCache(10, 0, nil)
+ _, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ require.Positive(t, c.curBytes.Load())
+ return int(c.curBytes.Load())
+}
+
+func TestPreparedCacheHitReturnsSameInstance(t *testing.T) {
+ c := newPreparedCache(16, 0, nil)
+ first, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ require.NotNil(t, first)
+ second, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ assert.Same(t, first, second, "a cache hit must return the same
prepared statement")
+ assert.Equal(t, uint64(1), c.hits.Load())
+ assert.Equal(t, uint64(1), c.misses.Load())
+}
+
+func TestPreparedCacheEvictsLeastRecentlyUsedByCount(t *testing.T) {
+ c := newPreparedCache(2, 0, nil)
+ get := func(i int) {
+ _, _, err := c.getOrPrepare(bydbqlQuery(i))
+ require.NoError(t, err)
+ }
+ get(1)
+ get(2)
+ get(1) // touch q1 so q2 becomes the least-recently-used entry
+ get(3) // adding q3 must evict q2, not q1
+ assert.Equal(t, 2, c.lru.Len())
+
+ missBefore := c.misses.Load()
+ get(1)
+ assert.Equal(t, missBefore, c.misses.Load(), "recently used q1 should
still be cached")
+ get(2)
+ assert.Equal(t, missBefore+1, c.misses.Load(), "least-recently-used q2
should have been evicted")
+}
+
+func TestPreparedCacheEvictsByBytes(t *testing.T) {
+ // A cap of 2.5x one entry's cost holds at most two entries. Deriving
it from a
+ // probe keeps the test valid regardless of the estimated statement
size.
+ perEntry := probeEntryCost(t)
+ maxBytes := perEntry*2 + perEntry/2
+ c := newPreparedCache(100, maxBytes, nil)
+ for i := 0; i < 5; i++ {
+ _, _, err := c.getOrPrepare(bydbqlQuery(i))
+ require.NoError(t, err)
+ }
+ assert.LessOrEqual(t, int(c.curBytes.Load()), maxBytes, "byte cap must
bound total accounted size")
+ assert.Equal(t, 2, c.lru.Len(), "a 2.5x-cost byte cap must hold exactly
two entries")
+ assert.Equal(t, perEntry*c.lru.Len(), int(c.curBytes.Load()), "curBytes
must track the live entries")
+}
+
+func TestPreparedCacheSkipsOversizedQuery(t *testing.T) {
+ c := newPreparedCache(16, 8, nil) // every real query exceeds 8 bytes
+ stmt, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ require.NotNil(t, stmt, "an oversized query must still be prepared,
just not cached")
+ assert.Equal(t, 0, c.lru.Len())
+ assert.Zero(t, c.curBytes.Load())
+}
+
+func TestPreparedCacheDisabled(t *testing.T) {
+ c := newPreparedCache(0, 0, nil)
+ first, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ require.NotNil(t, first)
+ second, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ assert.NotSame(t, first, second, "a disabled cache must re-prepare
every request")
+ assert.Nil(t, c.lru, "a disabled cache holds no LRU")
+ // A disabled cache bypasses without recording cache hits or misses.
+ assert.Equal(t, uint64(0), c.hits.Load())
+ assert.Equal(t, uint64(0), c.misses.Load())
+}
+
+func TestPreparedCacheSkipsLiteralQuery(t *testing.T) {
+ c := newPreparedCache(16, 1<<20, nil)
+ first, _, err := c.getOrPrepare(bydbqlLiteralQuery(0))
+ require.NoError(t, err)
+ require.NotNil(t, first)
+ second, _, err := c.getOrPrepare(bydbqlLiteralQuery(0))
+ require.NoError(t, err)
+ // A placeholder-free query is not a reusable template: never cached,
re-parsed
+ // each time, and not counted so it cannot depress hit_ratio.
+ assert.NotSame(t, first, second, "a literal query must be re-prepared,
not cached")
+ assert.Equal(t, 0, c.lru.Len(), "a literal query must never be stored")
+ assert.Equal(t, uint64(0), c.hits.Load())
+ assert.Equal(t, uint64(0), c.misses.Load())
+}
+
+func TestPreparedCacheBypassNotCounted(t *testing.T) {
+ // A literal query is counted as a bypass (emit("bypass")) but must not
move the
+ // hit/miss counters, so the hit ratio stays meaningful.
+ c := newPreparedCache(16, 1<<20, newBypassMetrics())
+ for i := 0; i < 5; i++ {
+ _, _, err := c.getOrPrepare(bydbqlLiteralQuery(i))
+ require.NoError(t, err)
+ }
+ assert.Equal(t, 0, c.lru.Len(), "literal queries are never cached")
+ assert.Equal(t, uint64(0), c.hits.Load())
+ assert.Equal(t, uint64(0), c.misses.Load(), "a bypass must not count as
a hit or miss")
+}
+
+func TestPreparedCacheParseErrorNotCached(t *testing.T) {
+ c := newPreparedCache(16, 0, nil)
+ _, _, err := c.getOrPrepare("NOT A QUERY")
+ require.Error(t, err)
+ assert.Equal(t, 0, c.lru.Len(), "a parse failure must not be cached")
+ // A parse failure is not a cache event, so it must not move hit/miss
counters.
+ assert.Equal(t, uint64(0), c.hits.Load())
+ assert.Equal(t, uint64(0), c.misses.Load())
+}
+
+func TestPreparedCacheEmitsMetrics(t *testing.T) {
+ // A non-nil metrics wiring exercises emit() end to end (counter label,
hit-ratio
+ // math, and the count/bytes gauges) on both the miss and hit paths.
+ c := newPreparedCache(4, 1<<20, newBypassMetrics())
+ _, _, err := c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ _, _, err = c.getOrPrepare(bydbqlQuery(0))
+ require.NoError(t, err)
+ assert.Equal(t, uint64(1), c.hits.Load())
+ assert.Equal(t, uint64(1), c.misses.Load())
+}
+
+func TestPreparedCacheConcurrentReuse(t *testing.T) {
+ // size 8 with 64 distinct keys and a byte cap of ~4 entries forces
both count-
+ // and byte-driven eviction to run concurrently, so -race covers the
eviction
+ // path that mutates curBytes.
+ const (
+ workers = 32
+ iters = 50
+ keys = 64
+ size = 8
+ )
+ maxBytes := probeEntryCost(t) * 4
+ c := newPreparedCache(size, maxBytes, nil)
+ var wg sync.WaitGroup
+ for i := 0; i < workers; i++ {
+ wg.Add(1)
+ go func(n int) {
+ defer wg.Done()
+ for j := 0; j < iters; j++ {
+ stmt, _, err :=
c.getOrPrepare(bydbqlQuery((n*iters + j) % keys))
+ assert.NoError(t, err)
+ assert.NotNil(t, stmt)
+ }
+ }(i)
+ }
+ wg.Wait()
+ assert.LessOrEqual(t, c.lru.Len(), size)
+ assert.LessOrEqual(t, int(c.curBytes.Load()), maxBytes, "byte cap must
hold under concurrent eviction")
+ assert.GreaterOrEqual(t, c.curBytes.Load(), int64(0), "curBytes must
never drift negative")
+ assert.Equal(t, uint64(workers*iters), c.hits.Load()+c.misses.Load())
+}
diff --git a/banyand/liaison/grpc/bydbql_test.go
b/banyand/liaison/grpc/bydbql_test.go
index 2d91de7f2..cac5bb4ea 100644
--- a/banyand/liaison/grpc/bydbql_test.go
+++ b/banyand/liaison/grpc/bydbql_test.go
@@ -20,6 +20,7 @@ package grpc
import (
"context"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -28,10 +29,12 @@ import (
bydbqlv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/bydbql/v1"
modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
)
func newTestBydbQLService() *bydbQLService {
- return &bydbQLService{metrics: newBypassMetrics()}
+ m := newBypassMetrics()
+ return &bydbQLService{metrics: m, cache: newPreparedCache(16, 1<<20, m)}
}
func bydbqlStrParam(v string) *modelv1.TagValue {
@@ -88,3 +91,48 @@ func
TestBydbQLQuery_ParamTypeMismatch_ReturnsInvalidArgument(t *testing.T) {
assert.Equal(t, codes.InvalidArgument, st.Code())
assert.Contains(t, st.Message(), "failed to bind parameters")
}
+
+// newTestDumper builds a topKDumper without starting the dump goroutine.
+func newTestDumper(l *logger.Logger) *topKDumper {
+ return &topKDumper{miss: newTopK(bydbqlTopKSize), slow:
newTopK(bydbqlTopKSize), l: l}
+}
+
+func TestBydbQLQuery_TracksCacheMiss(t *testing.T) {
+ svc := newTestBydbQLService()
+ svc.dumper = newTestDumper(nil)
+ // A cacheable query misses on first sight; the miss is observed before
Bind runs
+ // (Bind then fails on missing params, but the miss was already
recorded).
+ _, _ = svc.Query(context.Background(), &bydbqlv1.QueryRequest{
+ Query: "SELECT * FROM STREAM sw IN default WHERE service_id =
?",
+ })
+ assert.NotEmpty(t, svc.dumper.miss.snapshot(), "a cacheable miss must
be tracked")
+}
+
+func TestBydbQLQuery_TracksSlowQuery(t *testing.T) {
+ svc := newTestBydbQLService()
+ svc.slowThreshold = time.Nanosecond // any query exceeds it
+ svc.dumper = newTestDumper(nil)
+ _, _ = svc.Query(context.Background(), &bydbqlv1.QueryRequest{
+ Query: "SELECT * FROM STREAM sw IN default WHERE service_id =
?",
+ })
+ assert.NotEmpty(t, svc.dumper.slow.snapshot(), "a query over the
threshold must be tracked")
+}
+
+func TestBydbQLDumpTopK(t *testing.T) {
+ d := newTestDumper(logger.GetLogger("test-bydbql"))
+ d.miss.observe("q-miss", 0)
+ d.miss.observe("q-miss", 0) // count>=2 so it survives the cold-start
filter
+ d.slow.observe("q-slow", time.Millisecond)
+ d.dump() // must not panic; the cumulative trackers keep their entries
+ assert.NotEmpty(t, d.miss.snapshot())
+ assert.NotEmpty(t, d.slow.snapshot())
+}
+
+func TestFormatTopKFiltersColdStartMisses(t *testing.T) {
+ entries := []topKSlot{
+ {key: "thrashing", count: 5},
+ {key: "cold-start", count: 1},
+ }
+ lines := formatTopK(entries, minReparseMisses, func(s topKSlot) string
{ return s.key })
+ assert.Equal(t, []string{"thrashing"}, lines, "count==1 cold-start
misses are filtered out")
+}
diff --git a/banyand/liaison/grpc/metrics.go b/banyand/liaison/grpc/metrics.go
index e97e35ce7..710b3e39e 100644
--- a/banyand/liaison/grpc/metrics.go
+++ b/banyand/liaison/grpc/metrics.go
@@ -63,6 +63,15 @@ type metrics struct {
schemaBarrierLaggards meter.Counter
schemaStatusNotApplied meter.Counter
schemaStatusExpired meter.Counter
+
+ // BydbQL prepared-statement cache. bydbqlPreparedCacheTotal counts
lookups by
+ // `result` ("hit"/"miss"); the gauges track the live hit ratio and the
cache's
+ // current entry count and approximate byte size.
+ bydbqlPreparedCacheTotal meter.Counter
+ bydbqlPreparedCacheHitRatio meter.Gauge
+ bydbqlPreparedCacheCount meter.Gauge
+ bydbqlPreparedCacheBytes meter.Gauge
+ bydbqlSlowQueryTotal meter.Counter
}
func newMetrics(factory observability.Factory) *metrics {
@@ -93,6 +102,11 @@ func newMetrics(factory observability.Factory) *metrics {
schemaBarrierLaggards:
factory.NewCounter("schema_barrier_laggard_nodes_total", "barrier", "role",
"node"),
schemaStatusNotApplied:
factory.NewCounter("schema_status_schema_not_applied_total", "rpc", "group",
"reason"),
schemaStatusExpired:
factory.NewCounter("schema_status_expired_schema_total", "rpc", "group"),
+ bydbqlPreparedCacheTotal:
factory.NewCounter("bydbql_prepared_cache_total", "result"),
+ bydbqlPreparedCacheHitRatio:
factory.NewGauge("bydbql_prepared_cache_hit_ratio"),
+ bydbqlPreparedCacheCount:
factory.NewGauge("bydbql_prepared_cache_count"),
+ bydbqlPreparedCacheBytes:
factory.NewGauge("bydbql_prepared_cache_bytes"),
+ bydbqlSlowQueryTotal:
factory.NewCounter("bydbql_slow_query_total"),
}
}
diff --git a/banyand/liaison/grpc/server.go b/banyand/liaison/grpc/server.go
index fe731c87e..3c5a347b8 100644
--- a/banyand/liaison/grpc/server.go
+++ b/banyand/liaison/grpc/server.go
@@ -137,6 +137,10 @@ type server struct {
queryAccessLogRecorders []queryAccessLogRecorder
maxRecvMsgSize run.Bytes
grpcBufferMemoryRatio float64
+ bydbqlSlowThreshold time.Duration
+ bydbqlTopKLogInterval time.Duration
+ bydbqlCacheSize int
+ bydbqlCacheMaxBytes int
port uint32
tls bool
enableIngestionAccessLog bool
@@ -279,7 +283,7 @@ func NewServer(_ context.Context, tir1Client, tir2Client,
broadcaster queue.Clie
)
}
s.accessLogRecorders = []accessLogRecorder{streamSVC, measureSVC,
traceSVC, s.propertyServer}
- s.queryAccessLogRecorders = []queryAccessLogRecorder{streamSVC,
measureSVC, traceSVC, s.propertyServer}
+ s.queryAccessLogRecorders = []queryAccessLogRecorder{streamSVC,
measureSVC, traceSVC, s.propertyServer, bydbQLSVC}
return s
}
@@ -364,6 +368,11 @@ func (s *server) PreRun(ctx context.Context) error {
s.measureSVC.metrics = metrics
s.traceSVC.metrics = metrics
s.bydbQLSVC.metrics = metrics
+ s.bydbQLSVC.cache = newPreparedCache(s.bydbqlCacheSize,
s.bydbqlCacheMaxBytes, metrics)
+ s.bydbQLSVC.slowThreshold = s.bydbqlSlowThreshold
+ // The dump goroutine lives for the server's lifetime and is stopped by
Close(),
+ // so it is rooted at a background context rather than PreRun's setup
context.
+ s.bydbQLSVC.dumper = newTopKDumper(s.bydbqlTopKLogInterval,
s.bydbQLSVC.l) //nolint:contextcheck
s.propertyServer.metrics = metrics
if s.barrierSVC != nil {
s.barrierSVC.metrics = metrics
@@ -449,6 +458,14 @@ func (s *server) FlagSet() *run.FlagSet {
fs.DurationVar(&s.traceSVC.maxWaitDuration,
"trace-metadata-cache-wait-duration", 0,
"the maximum duration to wait for metadata cache to load (for
testing purposes)")
fs.IntVar(&s.propertyServer.repairQueueCount,
"property-repair-queue-count", 128, "the number of queues for property repair")
+ fs.IntVar(&s.bydbqlCacheSize, "bydbql-prepared-cache-size", 2000,
+ "max number of prepared BydbQL statements cached on the query
path; 0 disables the cache")
+ fs.IntVar(&s.bydbqlCacheMaxBytes, "bydbql-prepared-cache-max-bytes",
10*1024*1024,
+ "max total estimated size (bytes) of the cached BydbQL prepared
statements; 0 removes the byte bound")
+ fs.DurationVar(&s.bydbqlSlowThreshold, "bydbql-slow-query-threshold",
time.Second,
+ "end-to-end latency above which a BydbQL query is counted as
slow; 0 disables slow-query tracking")
+ fs.DurationVar(&s.bydbqlTopKLogInterval, "bydbql-topk-log-interval",
5*time.Minute,
+ "how often to log the top BydbQL cache-miss and slow queries; 0
disables the top-K log")
s.grpcBufferMemoryRatio = 0.1
fs.Float64Var(&s.grpcBufferMemoryRatio, "grpc-buffer-memory-ratio", 0.1,
"ratio of memory limit to use for gRPC buffer size calculation
(0.0 < ratio <= 1.0)")
@@ -717,6 +734,11 @@ func (s *server) GracefulStop() {
stopped := make(chan struct{})
go func() {
s.ser.GracefulStop()
+ // The top-K dumper is independent of the access log, so stop
its goroutine
+ // unconditionally (nil-safe when the feature is disabled).
+ if s.bydbQLSVC != nil {
+ s.bydbQLSVC.dumper.close()
+ }
if s.enableIngestionAccessLog {
for _, alr := range s.accessLogRecorders {
_ = alr.Close()
diff --git a/banyand/liaison/grpc/topk.go b/banyand/liaison/grpc/topk.go
new file mode 100644
index 000000000..f6ef17773
--- /dev/null
+++ b/banyand/liaison/grpc/topk.go
@@ -0,0 +1,132 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package grpc
+
+import (
+ "math"
+ "sort"
+ "sync"
+ "time"
+)
+
+// bydbqlTopKSize is the number of hot entries each top-K tracker keeps. It is
sized
+// well above the expected number of distinct parameterized query templates so
the
+// Space-Saving eviction (and its count over-estimate) effectively never
triggers for
+// a normal workload, keeping the reported counts exact while still bounding
memory.
+const bydbqlTopKSize = 128
+
+// topKSlot is one tracked query and its accumulated statistics.
+type topKSlot struct {
+ key string
+ count uint64
+ maxDur time.Duration
+}
+
+// topK is a bounded approximate heavy-hitters tracker (Space-Saving): it
keeps at
+// most k entries and, when full, evicts the least-frequent one, letting the
new key
+// inherit that entry's count + 1 so a fresh key gets a fair chance instead of
being
+// evicted again immediately. With k modest (128) the min scan and the
snapshot sort are
+// cheap, and observing an existing key needs no reordering at all.
+type topK struct {
+ slots map[string]*topKSlot
+ k int
+ mu sync.Mutex
+}
+
+func newTopK(k int) *topK {
+ if k < 1 {
+ k = 1
+ }
+ return &topK{slots: make(map[string]*topKSlot, k), k: k}
+}
+
+// observe records one occurrence of key; dur is the query latency (0 when
latency is
+// not tracked, e.g. the cache-miss queue). maxDur keeps the largest latency
seen.
+func (t *topK) observe(key string, dur time.Duration) {
+ t.mu.Lock()
+ defer t.mu.Unlock()
+ if s, ok := t.slots[key]; ok {
+ s.count++
+ if dur > s.maxDur {
+ s.maxDur = dur
+ }
+ return
+ }
+ if len(t.slots) < t.k {
+ t.slots[key] = &topKSlot{key: key, count: 1, maxDur: dur}
+ return
+ }
+ // Full: evict the least-frequent entry and let the new key inherit its
count.
+ minKey := ""
+ minCount := uint64(math.MaxUint64)
+ for k, s := range t.slots {
+ if s.count < minCount {
+ minCount, minKey = s.count, k
+ }
+ }
+ delete(t.slots, minKey)
+ t.slots[key] = &topKSlot{key: key, count: minCount + 1, maxDur: dur}
+}
+
+// snapshot returns the tracked entries ranked by frequency: (count desc,
maxDur desc,
+// key asc). The tracker is cumulative, so each dump reflects the hottest
queries since
+// process start. The full tie-break makes the order deterministic across
dumps.
+func (t *topK) snapshot() []topKSlot {
+ out := t.copyOut()
+ sort.Slice(out, func(i, j int) bool { return lessByCount(out[i],
out[j]) })
+ return out
+}
+
+// snapshotByLatency returns the tracked entries ranked by peak latency:
(maxDur desc,
+// count desc, key asc), so a rarely-but-catastrophically slow query is not
buried under
+// frequently-mildly-slow ones.
+func (t *topK) snapshotByLatency() []topKSlot {
+ out := t.copyOut()
+ sort.Slice(out, func(i, j int) bool { return lessByLatency(out[i],
out[j]) })
+ return out
+}
+
+func (t *topK) copyOut() []topKSlot {
+ t.mu.Lock()
+ out := make([]topKSlot, 0, len(t.slots))
+ for _, s := range t.slots {
+ out = append(out, *s)
+ }
+ t.mu.Unlock()
+ return out
+}
+
+func lessByCount(a, b topKSlot) bool {
+ if a.count != b.count {
+ return a.count > b.count
+ }
+ if a.maxDur != b.maxDur {
+ return a.maxDur > b.maxDur
+ }
+ return a.key < b.key
+}
+
+func lessByLatency(a, b topKSlot) bool {
+ if a.maxDur != b.maxDur {
+ return a.maxDur > b.maxDur
+ }
+ if a.count != b.count {
+ return a.count > b.count
+ }
+ return a.key < b.key
+}
diff --git a/banyand/liaison/grpc/topk_test.go
b/banyand/liaison/grpc/topk_test.go
new file mode 100644
index 000000000..db864ee03
--- /dev/null
+++ b/banyand/liaison/grpc/topk_test.go
@@ -0,0 +1,124 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package grpc
+
+import (
+ "fmt"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func topKByKey(slots []topKSlot) map[string]topKSlot {
+ m := make(map[string]topKSlot, len(slots))
+ for _, s := range slots {
+ m[s.key] = s
+ }
+ return m
+}
+
+func TestTopKCountsAndMaxDur(t *testing.T) {
+ tk := newTopK(4)
+ tk.observe("a", 5*time.Millisecond)
+ tk.observe("a", 8*time.Millisecond) // largest, observed in the middle
+ tk.observe("a", 3*time.Millisecond) // smaller and last: maxDur must
stay 8ms, not 3ms
+ tk.observe("b", time.Millisecond)
+
+ byKey := topKByKey(tk.snapshot())
+ assert.Equal(t, uint64(3), byKey["a"].count)
+ assert.Equal(t, 8*time.Millisecond, byKey["a"].maxDur, "maxDur keeps
the largest, not the last")
+ assert.Equal(t, uint64(1), byKey["b"].count)
+}
+
+func TestTopKEvictsMinAndInheritsCount(t *testing.T) {
+ tk := newTopK(2)
+ tk.observe("a", 0)
+ tk.observe("a", 0)
+ tk.observe("a", 0) // a.count = 3
+ tk.observe("b", 0) // b.count = 1; now full {a:3, b:1}
+
+ // A new key evicts the least-frequent (b) and inherits its count + 1,
so it is
+ // not immediately evicted again (Space-Saving) — this is what lets new
keys in.
+ tk.observe("c", 0)
+
+ byKey := topKByKey(tk.snapshot())
+ assert.Len(t, byKey, 2)
+ assert.Equal(t, uint64(3), byKey["a"].count, "the frequent key
survives")
+ assert.Equal(t, uint64(2), byKey["c"].count, "the new key inherits
evicted count + 1")
+ _, hasB := byKey["b"]
+ assert.False(t, hasB, "the least-frequent key was evicted")
+}
+
+func TestTopKSnapshotSortedAndCumulative(t *testing.T) {
+ tk := newTopK(4)
+ tk.observe("hi", 0)
+ tk.observe("hi", 0)
+ tk.observe("lo", 0)
+
+ snap := tk.snapshot()
+ assert.Equal(t, "hi", snap[0].key, "snapshot is sorted by count
descending")
+ assert.Equal(t, "lo", snap[1].key)
+ // The tracker is cumulative: a snapshot does not clear it.
+ assert.Len(t, tk.snapshot(), 2)
+}
+
+func TestTopKSnapshotByLatency(t *testing.T) {
+ tk := newTopK(bydbqlTopKSize)
+ tk.observe("frequent", time.Millisecond) // high count, low latency
+ tk.observe("frequent", time.Millisecond)
+ tk.observe("frequent", time.Millisecond)
+ tk.observe("rare-but-slow", time.Second) // count 1, catastrophic
latency
+
+ assert.Equal(t, "frequent", tk.snapshot()[0].key, "snapshot ranks by
count")
+ assert.Equal(t, "rare-but-slow", tk.snapshotByLatency()[0].key,
+ "snapshotByLatency surfaces the catastrophic outlier first")
+}
+
+func TestTopKDeterministicTieBreak(t *testing.T) {
+ // Equal counts and durations must order by key, not by map iteration.
+ a := newTopK(bydbqlTopKSize)
+ b := newTopK(bydbqlTopKSize)
+ for _, k := range []string{"c", "a", "b"} {
+ a.observe(k, 0)
+ }
+ for _, k := range []string{"b", "c", "a"} {
+ b.observe(k, 0)
+ }
+ assert.Equal(t, a.snapshot(), b.snapshot(), "same entries yield the
same order regardless of insertion")
+ snap := a.snapshot()
+ assert.Equal(t, []string{"a", "b", "c"}, []string{snap[0].key,
snap[1].key, snap[2].key})
+}
+
+func TestTopKConcurrentObserve(t *testing.T) {
+ tk := newTopK(bydbqlTopKSize)
+ const workers, iters = 16, 500
+ var wg sync.WaitGroup
+ for w := 0; w < workers; w++ {
+ wg.Add(1)
+ go func(n int) {
+ defer wg.Done()
+ for i := 0; i < iters; i++ {
+ tk.observe(fmt.Sprintf("q%d", (n*iters+i)%50),
time.Duration(i)*time.Microsecond)
+ }
+ }(w)
+ }
+ wg.Wait()
+ assert.LessOrEqual(t, len(tk.snapshot()), bydbqlTopKSize)
+}
diff --git a/docs/operation/configuration.md b/docs/operation/configuration.md
index 599b34c07..59bd5f636 100644
--- a/docs/operation/configuration.md
+++ b/docs/operation/configuration.md
@@ -89,6 +89,33 @@ The following flags are used to configure the timeout of
data sending from liais
- `--measure-write-timeout duration`: Measure write timeout (default: 1m).
- `--trace-write-timeout duration`: Trace write timeout (default: 1m).
+The following flags tune the BydbQL prepared-statement cache on the query
path. The cache stores parsed statements keyed by query text so repeated,
parameterized (`?`) queries skip re-parsing; literal queries without
placeholders are never cached. It is bounded by both an entry count and the
estimated in-memory size of the cached statements, evicting least-recently-used
entries when either bound is exceeded. Effectiveness is observable via the
`bydbql_prepared_cache_*` metrics (a `hit`/` [...]
+
+- `--bydbql-prepared-cache-size int`: Max number of prepared BydbQL statements
cached on the query path; `0` disables the cache (default: 2000).
+- `--bydbql-prepared-cache-max-bytes int`: Max total estimated size (in bytes)
of the cached prepared statements; `0` removes the byte bound (default:
10485760, i.e. 10MiB).
+
+These flags surface the queries behind cache misses and slow responses without
exposing high-cardinality query text as metric labels: Prometheus gets only two
counters (`bydbql_prepared_cache_total{result="miss"}` and
`bydbql_slow_query_total`), while the specific hot queries are logged.
+
+- `--bydbql-slow-query-threshold duration`: End-to-end latency above which a
BydbQL query is counted as slow (increments `bydbql_slow_query_total`) and
tracked in the slow-query top-K; `0` disables slow-query tracking (default:
`1s`).
+- `--bydbql-topk-log-interval duration`: How often to log the hottest
cache-miss and slow queries. Counts are cumulative since process start. The
cache-miss list only shows templates re-parsed at least twice (`count>=2`):
every template misses once on its cold-start lookup, so a `count==1` entry is
benign and is filtered out — only repeatedly evicted-and-re-parsed (thrashing)
templates are surfaced. The slow-query list is ranked by peak latency
(`max_latency`) so a rarely-but-catastrophi [...]
+
+#### Diagnosing an ineffective BydbQL cache
+
+A cache hit costs about 25 ns and 0 allocations, so the cache is effectively
free when it works. When it does not, every request re-parses the query — for a
wide trace/log query that is roughly **220 µs and ~147 KB / ~2,100 allocations
per request** (about four orders of magnitude more CPU and heap churn than a
hit). Watch these signals; several going abnormal together points at an
ineffective cache:
+
+- **`bydbql_prepared_cache_hit_ratio` drops toward 0** and the **`miss` rate
of `bydbql_prepared_cache_total{result="miss"}` climbs** — the primary
indicator.
+- **BydbQL query latency rises** (the liaison gRPC `query` latency series)
because the parse cost is added back to every request.
+- **Process CPU and Go GC pressure rise** — each miss allocates a fresh parse
tree, so the allocation rate, GC frequency, GC pause, and GC CPU fraction all
increase.
+- **`bydbql_prepared_cache_count` / `bydbql_prepared_cache_bytes` sit pinned
at the cap while the miss rate stays high** — the set of distinct query
templates exceeds the cache capacity and entries are constantly evicted.
+- **The `bypass` rate of `bydbql_prepared_cache_total{result="bypass"}` is
high** — many queries are literal (contain no `?`) and are never cached.
Bypasses are excluded from the hit ratio.
+
+To find the exact queries behind a high miss/bypass rate, enable the query
access log (`--enable-query-access-log`). Each BydbQL entry is tagged with its
cache outcome via the log's service field: `bydbql-hit`, `bydbql-miss`
(cacheable but not in the cache), and `bydbql-bypass` (literal, never cached).
Filtering the `bydbql-query-*` log for `bydbql-miss` / `bydbql-bypass` yields
exactly the queries that did not hit the cache, together with their full text.
+
+Remedies:
+
+- **High miss rate with the cache pinned at capacity** → the working set of
distinct query templates is larger than the cache. Raise
`--bydbql-prepared-cache-size` (and `--bydbql-prepared-cache-max-bytes` if the
byte bound binds first) until the hit ratio recovers.
+- **High bypass rate (near-empty cache)** → queries are not parameterized.
Literal queries (no `?`) are never cached, and inlining values (e.g. `id = 'a'
OR id = 'b' …`) explodes the template count. Use the `bydbql-bypass` access-log
entries to find them, then parameterize the varying values with `?`, and
express value sets as `IN (?)` bound to a `str_array`/`int_array` so a query is
one reusable template regardless of the number of values.
+
### TLS
If you want to enable TLS for the communication between the client and
liaison/standalone, you can use the following flags:
diff --git a/pkg/bydbql/binder.go b/pkg/bydbql/binder.go
index 4b63c29fe..3d57620cd 100644
--- a/pkg/bydbql/binder.go
+++ b/pkg/bydbql/binder.go
@@ -34,8 +34,8 @@ import (
//
// Binding mutates the grammar in place, so a grammar binds exactly once:
rebinding is
// rejected and a failed bind leaves the grammar partially bound and unusable
— parse the
-// query again in both cases. Parse-once/bind-many reuse requires an immutable
binding
-// design and is tracked as a follow-up.
+// query again in both cases. For parse-once/bind-many reuse, use Prepare and
+// (*PreparedStatement).Bind, which never mutate the parsed template.
func BindParams(g *Grammar, params []*modelv1.TagValue) error {
if g.paramsBound {
return fmt.Errorf("grammar is already bound; parse the query
again to bind new parameters")
@@ -159,15 +159,11 @@ func (b *binder) collectIntSlot(value *int, param *bool,
maxValue int64) {
return
}
b.slots = append(b.slots, func(p *modelv1.TagValue) error {
- intVal, ok := p.Value.(*modelv1.TagValue_Int)
- if !ok {
- return fmt.Errorf("this position only accepts int
parameters, got %T", p.Value)
- }
- bound := intVal.Int.GetValue()
- if err := validateCountValue("int parameter", bound, maxValue);
err != nil {
+ bound, err := resolveCountParam(p, maxValue)
+ if err != nil {
return err
}
- *value = int(bound)
+ *value = bound
*param = false
return nil
})
@@ -287,31 +283,17 @@ func (b *binder) collectValueList(values []*GrammarValue,
assign func([]*Grammar
}
func (b *binder) bindListValue(v *GrammarValue, p *modelv1.TagValue) error {
- switch val := p.Value.(type) {
+ switch p.Value.(type) {
case *modelv1.TagValue_Str, *modelv1.TagValue_Int,
*modelv1.TagValue_Null:
return bindScalarValue(v, p)
- case *modelv1.TagValue_StrArray:
- if len(val.StrArray.GetValue()) == 0 {
- return fmt.Errorf("array parameter must not be empty")
- }
- expanded := make([]*GrammarValue, 0,
len(val.StrArray.GetValue()))
- for _, s := range val.StrArray.GetValue() {
- expanded = append(expanded, &GrammarValue{String: &s})
- }
- b.recordExpansion(v, expanded)
- case *modelv1.TagValue_IntArray:
- if len(val.IntArray.GetValue()) == 0 {
- return fmt.Errorf("array parameter must not be empty")
- }
- expanded := make([]*GrammarValue, 0,
len(val.IntArray.GetValue()))
- for _, i := range val.IntArray.GetValue() {
- expanded = append(expanded, &GrammarValue{Integer: &i})
+ default:
+ expanded, err := resolveArrayElements(p)
+ if err != nil {
+ return err
}
b.recordExpansion(v, expanded)
- default:
- return fmt.Errorf("value list only accepts str, int, null,
str_array, or int_array parameters, got %T", p.Value)
+ return nil
}
- return nil
}
func (b *binder) recordExpansion(v *GrammarValue, expanded []*GrammarValue) {
@@ -336,40 +318,115 @@ func (b *binder) expandLists() {
}
func bindTimeValue(v *GrammarTimeValue, p *modelv1.TagValue) error {
+ strVal, err := resolveTimeParam(p)
+ if err != nil {
+ return err
+ }
+ v.String = &strVal
+ v.Param = false
+ return nil
+}
+
+func bindScalarValue(v *GrammarValue, p *modelv1.TagValue) error {
+ resolved, err := resolveScalarParam(p)
+ if err != nil {
+ return err
+ }
+ v.String, v.Integer, v.Null = resolved.String, resolved.Integer,
resolved.Null
+ v.Param = false
+ return nil
+}
+
+// resolveScalarParam produces a fresh literal value node for a str/int/null
+// parameter. It is the single source of scalar type acceptance, shared by the
+// in-place bind path and the reusable Bind path.
+func resolveScalarParam(p *modelv1.TagValue) (*GrammarValue, error) {
switch val := p.Value.(type) {
case *modelv1.TagValue_Str:
strVal := val.Str.GetValue()
- v.String = &strVal
+ return &GrammarValue{String: &strVal}, nil
+ case *modelv1.TagValue_Int:
+ intVal := val.Int.GetValue()
+ return &GrammarValue{Integer: &intVal}, nil
+ case *modelv1.TagValue_Null:
+ return &GrammarValue{Null: true}, nil
+ default:
+ return nil, fmt.Errorf("this position only accepts str, int, or
null parameters, got %T", p.Value)
+ }
+}
+
+// resolveTimeParam produces the time string for a str/timestamp parameter. int
+// is rejected: the transformer cannot parse a bare integer as a timestamp, so
+// accepting it would only defer a certain failure.
+func resolveTimeParam(p *modelv1.TagValue) (string, error) {
+ switch val := p.Value.(type) {
+ case *modelv1.TagValue_Str:
+ return val.Str.GetValue(), nil
case *modelv1.TagValue_Timestamp:
// A nil inner timestamp would silently decode as the Unix
epoch and an
// out-of-range one would format as a nonsense year; reject
both here.
if err := val.Timestamp.CheckValid(); err != nil {
- return fmt.Errorf("invalid timestamp parameter: %w",
err)
+ return "", fmt.Errorf("invalid timestamp parameter:
%w", err)
}
- strVal := val.Timestamp.AsTime().Format(time.RFC3339Nano)
- v.String = &strVal
+ return val.Timestamp.AsTime().Format(time.RFC3339Nano), nil
default:
- // int is rejected as well: the transformer cannot parse a bare
integer
- // as a timestamp, so accepting it would only defer a certain
failure.
- return fmt.Errorf("time clause only accepts str or timestamp
parameters, got %T", p.Value)
+ return "", fmt.Errorf("time clause only accepts str or
timestamp parameters, got %T", p.Value)
}
- v.Param = false
- return nil
}
-func bindScalarValue(v *GrammarValue, p *modelv1.TagValue) error {
+// resolveArrayElements produces the literal nodes a str_array/int_array
+// parameter expands to, rejecting empty arrays.
+func resolveArrayElements(p *modelv1.TagValue) ([]*GrammarValue, error) {
switch val := p.Value.(type) {
- case *modelv1.TagValue_Str:
- strVal := val.Str.GetValue()
- v.String = &strVal
- case *modelv1.TagValue_Int:
- intVal := val.Int.GetValue()
- v.Integer = &intVal
- case *modelv1.TagValue_Null:
- v.Null = true
+ case *modelv1.TagValue_StrArray:
+ if len(val.StrArray.GetValue()) == 0 {
+ return nil, fmt.Errorf("array parameter must not be
empty")
+ }
+ expanded := make([]*GrammarValue, 0,
len(val.StrArray.GetValue()))
+ for _, s := range val.StrArray.GetValue() {
+ expanded = append(expanded, &GrammarValue{String: &s})
+ }
+ return expanded, nil
+ case *modelv1.TagValue_IntArray:
+ if len(val.IntArray.GetValue()) == 0 {
+ return nil, fmt.Errorf("array parameter must not be
empty")
+ }
+ expanded := make([]*GrammarValue, 0,
len(val.IntArray.GetValue()))
+ for _, i := range val.IntArray.GetValue() {
+ expanded = append(expanded, &GrammarValue{Integer: &i})
+ }
+ return expanded, nil
default:
- return fmt.Errorf("this position only accepts str, int, or null
parameters, got %T", p.Value)
+ return nil, fmt.Errorf("value list only accepts str, int, null,
str_array, or int_array parameters, got %T", p.Value)
}
- v.Param = false
- return nil
+}
+
+// resolveListParam produces the value node(s) an IN/MATCH/HAVING placeholder
+// resolves to: one for a scalar, N for an array expanded in place. It mirrors
+// bindListValue's scalar-vs-array split so both paths accept the same types.
+func resolveListParam(p *modelv1.TagValue) ([]*GrammarValue, error) {
+ switch p.Value.(type) {
+ case *modelv1.TagValue_Str, *modelv1.TagValue_Int,
*modelv1.TagValue_Null:
+ v, err := resolveScalarParam(p)
+ if err != nil {
+ return nil, err
+ }
+ return []*GrammarValue{v}, nil
+ default:
+ return resolveArrayElements(p)
+ }
+}
+
+// resolveCountParam produces the int for a LIMIT/OFFSET/TOP N parameter within
+// that position's upper bound.
+func resolveCountParam(p *modelv1.TagValue, maxValue int64) (int, error) {
+ intVal, ok := p.Value.(*modelv1.TagValue_Int)
+ if !ok {
+ return 0, fmt.Errorf("this position only accepts int
parameters, got %T", p.Value)
+ }
+ bound := intVal.Int.GetValue()
+ if err := validateCountValue("int parameter", bound, maxValue); err !=
nil {
+ return 0, err
+ }
+ return int(bound), nil
}
diff --git a/pkg/bydbql/binder_equivalence_test.go
b/pkg/bydbql/binder_equivalence_test.go
index 3551ca0a4..0a9abf41e 100644
--- a/pkg/bydbql/binder_equivalence_test.go
+++ b/pkg/bydbql/binder_equivalence_test.go
@@ -19,6 +19,8 @@ package bydbql_test
import (
"context"
+ "fmt"
+ "sync"
"time"
"github.com/alecthomas/participle/v2/lexer"
@@ -329,6 +331,20 @@ func equivalenceCases() []equivalenceCase {
params: []*modelv1.TagValue{intParam(5),
strParam("-1h"), strParam("svc")},
ignoreTimeRange: true,
},
+ // PROPERTY query: the id-extraction path resolves placeholders
separately
+ // from the criteria path, so it needs its own coverage.
+ {
+ name: "str param in PROPERTY id =",
+ parameterized: "SELECT env FROM PROPERTY sw_prop IN
default WHERE id = ?",
+ literal: "SELECT env FROM PROPERTY sw_prop IN
default WHERE id = 'k1'",
+ params: []*modelv1.TagValue{strParam("k1")},
+ },
+ {
+ name: "str_array param expands in PROPERTY id
IN with a criteria param",
+ parameterized: "SELECT env FROM PROPERTY sw_prop IN
default WHERE id IN (?) AND env = ?",
+ literal: "SELECT env FROM PROPERTY sw_prop IN
default WHERE id IN ('k1', 'k2') AND env = 'prod'",
+ params: []*modelv1.TagValue{strArrayParam("k1",
"k2"), strParam("prod")},
+ },
}
}
@@ -353,10 +369,16 @@ var _ = Describe("BindParams equivalence with literal
queries", func() {
}},
Fields: []*databasev1.FieldSpec{{Name: "value",
FieldType: databasev1.FieldType_FIELD_TYPE_INT}},
}, nil).AnyTimes()
+ propertyRegistry := schema.NewMockProperty(ctrl)
+ propertyRegistry.EXPECT().GetProperty(gomock.Any(),
gomock.Any()).Return(&databasev1.Property{
+ Metadata: &commonv1.Metadata{Name: "sw_prop", Group:
"default"},
+ Tags: []*databasev1.TagSpec{{Name: "env", Type:
databasev1.TagType_TAG_TYPE_STRING}},
+ }, nil).AnyTimes()
mockRepo := metadata.NewMockRepo(ctrl)
mockRepo.EXPECT().StreamRegistry().AnyTimes().Return(streamRegistry)
mockRepo.EXPECT().TopNAggregationRegistry().AnyTimes().Return(topNRegistry)
mockRepo.EXPECT().MeasureRegistry().AnyTimes().Return(measureRegistry)
+
mockRepo.EXPECT().PropertyRegistry().AnyTimes().Return(propertyRegistry)
transformer = NewTransformer(mockRepo)
})
@@ -406,19 +428,31 @@ var _ = Describe("BindParams equivalence with literal
queries", func() {
astDiff := cmp.Diff(literalGrammar, boundGrammar,
cmpopts.IgnoreTypes(lexer.Position{}), cmpopts.IgnoreUnexported(Grammar{}))
Expect(astDiff).To(BeEmpty(), "AST mismatch:\n%s",
astDiff)
- // Both must transform to the same native query request
+ // Both the in-place bind and the reusable Prepare/Bind
path must
+ // transform to the same native request as the literal
query.
ctx := context.Background()
literalResult, literalErr := transformer.Transform(ctx,
literalGrammar)
boundResult, boundErr := transformer.Transform(ctx,
boundGrammar)
+
+ prepared, prepErr := Prepare(testCase.parameterized)
+ Expect(prepErr).To(BeNil())
+ bq, bindErr := prepared.Bind(testCase.params)
+ Expect(bindErr).To(BeNil())
+ preparedResult, preparedErr :=
transformer.TransformBound(ctx, bq)
+
if literalErr != nil {
Expect(testCase.expectError).To(BeTrue(),
"unexpected transform error: %v", literalErr)
Expect(boundErr).To(HaveOccurred())
Expect(boundErr.Error()).To(Equal(literalErr.Error()))
+ Expect(preparedErr).To(HaveOccurred())
+
Expect(preparedErr.Error()).To(Equal(literalErr.Error()))
return
}
Expect(testCase.expectError).To(BeFalse(), "expected a
transform error but both sides succeeded")
Expect(boundErr).To(BeNil())
+ Expect(preparedErr).To(BeNil())
Expect(boundResult.Type).To(Equal(literalResult.Type))
+
Expect(preparedResult.Type).To(Equal(literalResult.Type))
// SELECT * projections are built from a map, so their
tag order is
// nondeterministic per Transform call; sort them
before comparing.
opts := []cmp.Option{
@@ -430,8 +464,47 @@ var _ = Describe("BindParams equivalence with literal
queries", func() {
protocmp.IgnoreFields(&streamv1.QueryRequest{}, "time_range"),
protocmp.IgnoreFields(&measurev1.TopNRequest{}, "time_range"))
}
- requestDiff := cmp.Diff(literalResult.QueryRequest,
boundResult.QueryRequest, opts...)
- Expect(requestDiff).To(BeEmpty(), "query request
mismatch:\n%s", requestDiff)
+ Expect(cmp.Diff(literalResult.QueryRequest,
boundResult.QueryRequest, opts...)).
+ To(BeEmpty(), "in-place bind request mismatch")
+ Expect(cmp.Diff(literalResult.QueryRequest,
preparedResult.QueryRequest, opts...)).
+ To(BeEmpty(), "prepared bind request mismatch")
})
}
+
+ It("reuses one prepared statement concurrently without cross-talk",
func() {
+ prepared, err := Prepare("SELECT * FROM STREAM sw IN default
WHERE service_id = ?")
+ Expect(err).To(BeNil())
+ const workers = 32
+ var wg sync.WaitGroup
+ errs := make(chan error, workers)
+ for i := 0; i < workers; i++ {
+ wg.Add(1)
+ go func(n int) {
+ defer wg.Done()
+ want := fmt.Sprintf("svc-%d", n)
+ bq, bindErr :=
prepared.Bind([]*modelv1.TagValue{strParam(want)})
+ if bindErr != nil {
+ errs <- bindErr
+ return
+ }
+ result, transformErr :=
transformer.TransformBound(context.Background(), bq)
+ if transformErr != nil {
+ errs <- transformErr
+ return
+ }
+ got :=
result.QueryRequest.(*streamv1.QueryRequest).Criteria.GetCondition().GetValue().GetStr().GetValue()
+ if got != want {
+ errs <- fmt.Errorf("worker %d: got %q,
want %q", n, got, want)
+ }
+ }(i)
+ }
+ wg.Wait()
+ close(errs)
+ for e := range errs {
+ Expect(e).NotTo(HaveOccurred())
+ }
+ // Under -race, correct per-worker results also prove the
shared template
+ // is never mutated during concurrent binds.
TestBindTemplateStaysImmutable
+ // asserts the template fields directly.
+ })
})
diff --git a/pkg/bydbql/binder_test.go b/pkg/bydbql/binder_test.go
index b3f4a7646..e6845d4ca 100644
--- a/pkg/bydbql/binder_test.go
+++ b/pkg/bydbql/binder_test.go
@@ -377,7 +377,7 @@ var _ = Describe("BindParams", func() {
}
walk(reflect.TypeOf(Grammar{}))
Expect(found).To(Equal(expected),
- "grammar @Param positions changed: wire any new
position into binder.collect, the acceptance matrix, and this list")
+ "grammar @Param positions changed: wire any new
position into binder.collect, preparer.walkGrammar, the acceptance matrix, and
this list")
})
})
diff --git a/pkg/bydbql/grammar.go b/pkg/bydbql/grammar.go
index 7062461a5..e0022a139 100644
--- a/pkg/bydbql/grammar.go
+++ b/pkg/bydbql/grammar.go
@@ -58,10 +58,11 @@ type GrammarSelectStatement struct {
// GrammarTopNStatement represents a SHOW TOP N statement.
type GrammarTopNStatement struct {
Pos lexer.Position
- Show string `parser:"@'SHOW'"`
- Top string `parser:"@'TOP'"`
- N int `parser:"( @Int"`
- NParam bool `parser:"| @Param )"`
+ Show string `parser:"@'SHOW'"`
+ Top string `parser:"@'TOP'"`
+ N int `parser:"( @Int"`
+ NParam bool `parser:"| @Param )"`
+ NParamIndex int
From *GrammarFromClause `parser:"@@"`
Time *GrammarTimeClause `parser:"@@?"`
Where *GrammarTopNWhereClause `parser:"@@?"`
@@ -80,8 +81,9 @@ type GrammarProjection struct {
// GrammarTopNProjection represents TOP N projection.
type GrammarTopNProjection struct {
- N int `parser:"( @Int"`
- NParam bool `parser:"| @Param )"`
+ N int `parser:"( @Int"`
+ NParam bool `parser:"| @Param )"`
+ NParamIndex int
OrderField *GrammarIdentifierPath `parser:"@@"`
Direction *string `parser:"@('ASC'|'DESC')?"`
OtherColumns []*GrammarColumn `parser:" ( ',' @@ ( ',' @@ )* )?"`
@@ -152,6 +154,9 @@ type GrammarTimeValue struct {
String *string `parser:" @String"`
Integer *int64 `parser:"| @Int"`
Param bool `parser:"| @Param"`
+ // ParamIndex is the placeholder's positional index, assigned by
Prepare and
+ // meaningful only when Param is true. It keys the per-request bound
overlay.
+ ParamIndex int
}
// GrammarSelectWhereClause represents WHERE clause.
@@ -260,6 +265,9 @@ type GrammarValue struct {
Integer *int64 `parser:"| @Int"`
Null bool `parser:"| @'NULL'"`
Param bool `parser:"| @Param"`
+ // ParamIndex is the placeholder's positional index, assigned by
Prepare and
+ // meaningful only when Param is true. It keys the per-request bound
overlay.
+ ParamIndex int
}
// GrammarIdentifierPart Can be either an Ident or a Keyword (keywords are
allowed in paths, but not as standalone identifiers).
@@ -327,16 +335,18 @@ type GrammarOrderByWithIdent struct {
// GrammarLimitClause represents LIMIT clause.
type GrammarLimitClause struct {
- Limit string `parser:"@'LIMIT'"`
- Value int `parser:"( @Int"`
- Param bool `parser:"| @Param )"`
+ Limit string `parser:"@'LIMIT'"`
+ Value int `parser:"( @Int"`
+ Param bool `parser:"| @Param )"`
+ ParamIndex int
}
// GrammarOffsetClause represents OFFSET clause.
type GrammarOffsetClause struct {
- Offset string `parser:"@'OFFSET'"`
- Value int `parser:"( @Int"`
- Param bool `parser:"| @Param )"`
+ Offset string `parser:"@'OFFSET'"`
+ Value int `parser:"( @Int"`
+ Param bool `parser:"| @Param )"`
+ ParamIndex int
}
// GrammarWithTraceClause represents WITH QUERY_TRACE clause.
diff --git a/pkg/bydbql/prepared.go b/pkg/bydbql/prepared.go
new file mode 100644
index 000000000..d917648a4
--- /dev/null
+++ b/pkg/bydbql/prepared.go
@@ -0,0 +1,289 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package bydbql
+
+import (
+ "fmt"
+ "math"
+ "reflect"
+
+ modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+)
+
+// placeholderKind classifies a `?` placeholder by the position it occupies,
+// which determines the parameter types Bind accepts and how Transform reads
it.
+type placeholderKind uint8
+
+const (
+ // phScalar is a WHERE/HAVING comparison value: str, int, or null.
+ phScalar placeholderKind = iota
+ // phList is an element of an IN/MATCH/HAVING value list: scalar or an
array
+ // that expands in place.
+ phList
+ // phTime is a TIME value: str or timestamp.
+ phTime
+ // phCount is a LIMIT/OFFSET/TOP N count: an int within maxCount.
+ phCount
+)
+
+// placeholderSpec describes one placeholder, recorded once by Prepare and read
+// per request by Bind (to validate the parameter) and Transform (to read it).
+type placeholderSpec struct {
+ maxCount int64 // upper bound for phCount positions
+ kind placeholderKind
+}
+
+// PreparedStatement is an immutable, reusable parsed query. It is safe to
cache
+// and to Bind concurrently: Prepare numbers the placeholders once, and Bind
+// never mutates the template — the bound values live in a per-request overlay.
+type PreparedStatement struct {
+ template *Grammar
+ specs []placeholderSpec
+}
+
+// Prepare parses a query and numbers its `?` placeholders so it can be bound
+// repeatedly with different parameters without re-parsing.
+func Prepare(query string) (*PreparedStatement, error) {
+ g, err := ParseQuery(query)
+ if err != nil {
+ return nil, err
+ }
+ p := &preparer{}
+ p.walkGrammar(g)
+ return &PreparedStatement{template: g, specs: p.specs}, nil
+}
+
+// NumPlaceholders reports how many `?` placeholders the statement carries. A
zero
+// count means a literal query with no reusable parameters.
+func (ps *PreparedStatement) NumPlaceholders() int {
+ return len(ps.specs)
+}
+
+// EstimatedSize returns an approximate in-memory byte footprint of the
prepared
+// statement — its parsed grammar template and placeholder specs. It is a
+// heuristic for cache byte-accounting (a reflection walk of the object graph),
+// not an exact measurement, and is meant to be called off the hot path.
+func (ps *PreparedStatement) EstimatedSize() int {
+ return deepSize(reflect.ValueOf(ps))
+}
+
+// deepSize sums the bytes reachable from v: the inline size of each pointed-to
+// value plus the out-of-line bytes of strings and slice backing arrays. Inline
+// struct/array fields are already covered by their container's size, so only
+// their referenced memory is added. It is shaped for the grammar, an acyclic
tree
+// of concrete structs, pointers, slices, strings, and scalars.
+func deepSize(v reflect.Value) int {
+ switch v.Kind() {
+ case reflect.Pointer:
+ if v.IsNil() {
+ return 0
+ }
+ elem := v.Elem()
+ return int(elem.Type().Size()) + deepSize(elem)
+ case reflect.Struct:
+ total := 0
+ for i := 0; i < v.NumField(); i++ {
+ total += deepSize(v.Field(i))
+ }
+ return total
+ case reflect.Slice:
+ if v.IsNil() {
+ return 0
+ }
+ total := v.Cap() * int(v.Type().Elem().Size())
+ for i := 0; i < v.Len(); i++ {
+ total += deepSize(v.Index(i))
+ }
+ return total
+ case reflect.String:
+ return v.Len()
+ default:
+ return 0
+ }
+}
+
+// resolvedParam holds one placeholder's bound value in the per-request
overlay.
+// Which field is populated follows the placeholder's kind: a scalar/list
+// position fills values (one node, or several for an expanded array), a TIME
+// position fills timeStr, and a count position fills count.
+type resolvedParam struct {
+ timeStr string
+ values []*GrammarValue
+ count int
+}
+
+// BoundQuery pairs an immutable prepared statement with the values bound for
one
+// request. Its values are the per-request overlay indexed by placeholder
+// position, allocated by Bind and never shared or mutated after. Pass it to
+// Transformer.TransformBound.
+type BoundQuery struct {
+ stmt *PreparedStatement
+ values []resolvedParam
+}
+
+// Bind resolves params against the statement's placeholders and returns a
+// per-request BoundQuery, leaving the template untouched, so a statement may
be
+// bound repeatedly and concurrently. Parameter positions in errors are
1-based,
+// matching BindParams.
+func (ps *PreparedStatement) Bind(params []*modelv1.TagValue) (*BoundQuery,
error) {
+ if len(params) != len(ps.specs) {
+ return nil, fmt.Errorf("parameter count mismatch: query
contains %d placeholder(s) but %d parameter(s) provided", len(ps.specs),
len(params))
+ }
+ values := make([]resolvedParam, len(ps.specs))
+ for i, spec := range ps.specs {
+ p := params[i]
+ if p == nil || p.GetValue() == nil {
+ return nil, fmt.Errorf("parameter #%d has no value",
i+1)
+ }
+ var err error
+ switch spec.kind {
+ case phScalar:
+ var v *GrammarValue
+ v, err = resolveScalarParam(p)
+ if err == nil {
+ values[i].values = []*GrammarValue{v}
+ }
+ case phList:
+ values[i].values, err = resolveListParam(p)
+ case phTime:
+ values[i].timeStr, err = resolveTimeParam(p)
+ case phCount:
+ values[i].count, err = resolveCountParam(p,
spec.maxCount)
+ }
+ if err != nil {
+ return nil, fmt.Errorf("failed to bind parameter #%d:
%w", i+1, err)
+ }
+ }
+ return &BoundQuery{stmt: ps, values: values}, nil
+}
+
+// preparer walks a parsed grammar in textual order, assigning each placeholder
+// its positional index and recording its spec. The traversal mirrors
+// binder.collect; the reflection tripwire test asserts both cover every
+// placeholder-bearing grammar field.
+type preparer struct {
+ specs []placeholderSpec
+}
+
+func (p *preparer) add(kind placeholderKind, maxCount int64) int {
+ idx := len(p.specs)
+ p.specs = append(p.specs, placeholderSpec{kind: kind, maxCount:
maxCount})
+ return idx
+}
+
+func (p *preparer) walkGrammar(g *Grammar) {
+ if g.Select != nil {
+ if g.Select.Projection != nil && g.Select.Projection.TopN !=
nil {
+ if topN := g.Select.Projection.TopN; topN.NParam {
+ topN.NParamIndex = p.add(phCount, math.MaxInt32)
+ }
+ }
+ p.walkTime(g.Select.Time)
+ if g.Select.Where != nil {
+ p.walkOrExpr(g.Select.Where.Expr)
+ }
+ if g.Select.Limit != nil && g.Select.Limit.Param {
+ g.Select.Limit.ParamIndex = p.add(phCount,
math.MaxUint32)
+ }
+ if g.Select.Offset != nil && g.Select.Offset.Param {
+ g.Select.Offset.ParamIndex = p.add(phCount,
math.MaxUint32)
+ }
+ }
+ if g.TopN != nil {
+ if g.TopN.NParam {
+ g.TopN.NParamIndex = p.add(phCount, math.MaxInt32)
+ }
+ p.walkTime(g.TopN.Time)
+ if g.TopN.Where != nil {
+ p.walkAndExpr(g.TopN.Where.Expr)
+ }
+ }
+}
+
+func (p *preparer) walkTime(clause *GrammarTimeClause) {
+ if clause == nil {
+ return
+ }
+ p.walkTimeValue(clause.Value)
+ if clause.Between != nil {
+ p.walkTimeValue(clause.Between.Begin)
+ p.walkTimeValue(clause.Between.End)
+ }
+}
+
+func (p *preparer) walkTimeValue(v *GrammarTimeValue) {
+ if v != nil && v.Param {
+ v.ParamIndex = p.add(phTime, 0)
+ }
+}
+
+func (p *preparer) walkOrExpr(expr *GrammarOrExpr) {
+ if expr == nil {
+ return
+ }
+ p.walkAndExpr(expr.Left)
+ for _, right := range expr.Right {
+ p.walkAndExpr(right.Right)
+ }
+}
+
+func (p *preparer) walkAndExpr(expr *GrammarAndExpr) {
+ if expr == nil {
+ return
+ }
+ p.walkPredicate(expr.Left)
+ for _, right := range expr.Right {
+ p.walkPredicate(right.Right)
+ }
+}
+
+func (p *preparer) walkPredicate(pred *GrammarPredicate) {
+ if pred == nil {
+ return
+ }
+ switch {
+ case pred.Paren != nil:
+ p.walkOrExpr(pred.Paren)
+ case pred.Binary != nil && pred.Binary.Tail != nil:
+ if compare := pred.Binary.Tail.Compare; compare != nil &&
compare.Value != nil && compare.Value.Param {
+ compare.Value.ParamIndex = p.add(phScalar, 0)
+ }
+ if match := pred.Binary.Tail.Match; match != nil &&
match.Values != nil {
+ p.walkValueList(match.Values.Single, match.Values.Array)
+ }
+ case pred.In != nil:
+ p.walkValueList(nil, pred.In.Values)
+ case pred.Having != nil && pred.Having.Values != nil:
+ p.walkValueList(pred.Having.Values.Single,
pred.Having.Values.Array)
+ }
+}
+
+// walkValueList numbers the placeholders in a MATCH/HAVING single-or-array or
an
+// IN list. A container is parsed as either a single value or an array; both
are
+// walked the same way here.
+func (p *preparer) walkValueList(single *GrammarValue, array []*GrammarValue) {
+ values := array
+ if single != nil {
+ values = []*GrammarValue{single}
+ }
+ for _, v := range values {
+ if v != nil && v.Param {
+ v.ParamIndex = p.add(phList, 0)
+ }
+ }
+}
diff --git a/pkg/bydbql/prepared_internal_test.go
b/pkg/bydbql/prepared_internal_test.go
new file mode 100644
index 000000000..0a82bd35c
--- /dev/null
+++ b/pkg/bydbql/prepared_internal_test.go
@@ -0,0 +1,335 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package bydbql
+
+import (
+ "context"
+ "math"
+ "strings"
+ "testing"
+
+ modelv1
"github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+)
+
+func tvStr(v string) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str:
&modelv1.Str{Value: v}}}
+}
+
+func tvInt(v int64) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int:
&modelv1.Int{Value: v}}}
+}
+
+func tvStrArray(vs ...string) *modelv1.TagValue {
+ return &modelv1.TagValue{Value: &modelv1.TagValue_StrArray{StrArray:
&modelv1.StrArray{Value: vs}}}
+}
+
+func TestPrepareNumbersPlaceholdersInTextOrder(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default TIME > ? WHERE
service_id = ? AND codes IN (?) LIMIT ? OFFSET ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if got := len(ps.specs); got != 5 {
+ t.Fatalf("placeholder count = %d, want 5", got)
+ }
+ wantSpecs := []placeholderSpec{
+ {kind: phTime},
+ {kind: phScalar},
+ {kind: phList},
+ {kind: phCount, maxCount: math.MaxUint32},
+ {kind: phCount, maxCount: math.MaxUint32},
+ }
+ for i, want := range wantSpecs {
+ if ps.specs[i] != want {
+ t.Errorf("specs[%d] = %+v, want %+v", i, ps.specs[i],
want)
+ }
+ }
+ sel := ps.template.Select
+ if idx := sel.Time.Value.ParamIndex; idx != 0 {
+ t.Errorf("TIME ParamIndex = %d, want 0", idx)
+ }
+ if idx :=
sel.Where.Expr.Left.Left.Binary.Tail.Compare.Value.ParamIndex; idx != 1 {
+ t.Errorf("scalar ParamIndex = %d, want 1", idx)
+ }
+ if idx := sel.Where.Expr.Left.Right[0].Right.In.Values[0].ParamIndex;
idx != 2 {
+ t.Errorf("IN element ParamIndex = %d, want 2", idx)
+ }
+ if idx := sel.Limit.ParamIndex; idx != 3 {
+ t.Errorf("LIMIT ParamIndex = %d, want 3", idx)
+ }
+ if idx := sel.Offset.ParamIndex; idx != 4 {
+ t.Errorf("OFFSET ParamIndex = %d, want 4", idx)
+ }
+}
+
+func TestPrepareNumbersShowTopCount(t *testing.T) {
+ ps, err := Prepare("SHOW TOP ? FROM MEASURE m IN default TIME > ? WHERE
service = ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ wantSpecs := []placeholderSpec{
+ {kind: phCount, maxCount: math.MaxInt32},
+ {kind: phTime},
+ {kind: phScalar},
+ }
+ if len(ps.specs) != len(wantSpecs) {
+ t.Fatalf("specs len = %d, want %d", len(ps.specs),
len(wantSpecs))
+ }
+ for i, want := range wantSpecs {
+ if ps.specs[i] != want {
+ t.Errorf("specs[%d] = %+v, want %+v", i, ps.specs[i],
want)
+ }
+ }
+ if idx := ps.template.TopN.NParamIndex; idx != 0 {
+ t.Errorf("SHOW TOP NParamIndex = %d, want 0", idx)
+ }
+}
+
+func TestPrepareLiteralQueryHasNoPlaceholders(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default WHERE service_id
= 'webapp'")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if got := len(ps.specs); got != 0 {
+ t.Fatalf("placeholder count = %d, want 0", got)
+ }
+}
+
+func TestPrepareCoversSamePlaceholdersAsBinder(t *testing.T) {
+ // Prepare and binder.collect are parallel traversals; a query
exercising
+ // every placeholder position must yield the same count from both, or
one
+ // walk has drifted from the other.
+ query := "SELECT TOP ? f FROM MEASURE m IN default TIME BETWEEN ? AND ?
" +
+ "WHERE a = ? AND b IN (?) AND c MATCH(?) AND d HAVING (?) LIMIT
? OFFSET ?"
+ ps, err := Prepare(query)
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ g, errs := ParseQuery(query)
+ if errs != nil {
+ t.Fatalf("ParseQuery: %v", errs)
+ }
+ if got, want := len(ps.specs), countUnboundParams(g); got != want {
+ t.Fatalf("Prepare counted %d placeholders, binder counted %d",
got, want)
+ }
+}
+
+func TestBindResolvesEachKind(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default TIME > ? WHERE
service_id = ? AND codes IN (?) LIMIT ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ bound, err := ps.Bind([]*modelv1.TagValue{tvStr("-30m"),
tvStr("webapp"), tvStrArray("a", "b"), tvInt(10)})
+ if err != nil {
+ t.Fatalf("Bind: %v", err)
+ }
+ p := bound.values
+ if p[0].timeStr != "-30m" {
+ t.Errorf("time = %q, want -30m", p[0].timeStr)
+ }
+ if len(p[1].values) != 1 || *p[1].values[0].String != "webapp" {
+ t.Errorf("scalar = %+v, want [webapp]", p[1].values)
+ }
+ if len(p[2].values) != 2 || *p[2].values[0].String != "a" ||
*p[2].values[1].String != "b" {
+ t.Errorf("list = %+v, want [a b]", p[2].values)
+ }
+ if p[3].count != 10 {
+ t.Errorf("count = %d, want 10", p[3].count)
+ }
+}
+
+func TestBindTemplateStaysImmutable(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default WHERE service_id
= ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if _, err = ps.Bind([]*modelv1.TagValue{tvStr("first")}); err != nil {
+ t.Fatalf("Bind: %v", err)
+ }
+ node :=
ps.template.Select.Where.Expr.Left.Left.Binary.Tail.Compare.Value
+ if !node.Param || node.String != nil {
+ t.Fatalf("template mutated by Bind: Param=%v String=%v",
node.Param, node.String)
+ }
+ // A second bind with a different value must succeed on the same
template.
+ bound, err := ps.Bind([]*modelv1.TagValue{tvStr("second")})
+ if err != nil {
+ t.Fatalf("second Bind: %v", err)
+ }
+ if *bound.values[0].values[0].String != "second" {
+ t.Errorf("second bind value = %q, want second",
*bound.values[0].values[0].String)
+ }
+}
+
+func TestBindErrors(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default WHERE service_id
= ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if _, err = ps.Bind(nil); err == nil || !strings.Contains(err.Error(),
"count mismatch") {
+ t.Errorf("count mismatch: got %v", err)
+ }
+ if _, err = ps.Bind([]*modelv1.TagValue{tvStrArray("a", "b")}); err ==
nil || !strings.Contains(err.Error(), "parameter #1") {
+ t.Errorf("array in scalar: got %v", err)
+ }
+ if _, err = ps.Bind([]*modelv1.TagValue{{}}); err == nil ||
!strings.Contains(err.Error(), "has no value") {
+ t.Errorf("empty value: got %v", err)
+ }
+ // A nil parameter element must produce a clean error, not a panic: the
proto
+ // getter is nil-safe, so p.GetValue() returns nil and Bind reports
"has no value".
+ if _, err = ps.Bind([]*modelv1.TagValue{nil}); err == nil ||
!strings.Contains(err.Error(), "has no value") {
+ t.Errorf("nil element: got %v", err)
+ }
+}
+
+func TestTransformBoundRejectsNil(t *testing.T) {
+ var transformer Transformer
+ if _, err := transformer.TransformBound(context.Background(), nil); err
== nil {
+ t.Fatal("TransformBound(nil) must return an error, not panic")
+ }
+}
+
+func TestTransformBoundRejectsMismatchedOverlay(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default WHERE service_id
= ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ var transformer Transformer
+ // An overlay shorter than the placeholder count would otherwise panic
on an
+ // out-of-range ParamIndex; TransformBound must reject it up front.
+ bq := &BoundQuery{stmt: ps, values: nil}
+ if _, err := transformer.TransformBound(context.Background(), bq); err
== nil {
+ t.Fatal("TransformBound must reject an overlay that does not
match the placeholder count")
+ }
+}
+
+// BenchmarkBinding contrasts re-parsing per request with parse-once/bind-many.
+func BenchmarkBinding(b *testing.B) {
+ const query = "SELECT * FROM STREAM sw IN default TIME > ? WHERE
service_id = ? AND codes IN (?) LIMIT ?"
+ params := func() []*modelv1.TagValue {
+ return []*modelv1.TagValue{tvStr("-30m"), tvStr("webapp"),
tvStrArray("a", "b"), tvInt(10)}
+ }
+ b.Run("ParseAndBindEachRequest", func(b *testing.B) {
+ for i := 0; i < b.N; i++ {
+ g, errs := ParseQuery(query)
+ if errs != nil {
+ b.Fatal(errs)
+ }
+ if err := BindParams(g, params()); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ b.Run("PrepareOnceBindMany", func(b *testing.B) {
+ ps, err := Prepare(query)
+ if err != nil {
+ b.Fatal(err)
+ }
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ if _, err := ps.Bind(params()); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+}
+
+func TestNumPlaceholders(t *testing.T) {
+ cases := map[string]int{
+ "SELECT * FROM STREAM sw IN default":
0,
+ "SELECT * FROM STREAM sw IN default WHERE service_id = ?":
1,
+ "SELECT * FROM STREAM sw IN default TIME > ? WHERE a = ? LIMIT
?": 3,
+ "SELECT * FROM STREAM sw IN default WHERE a IN (?, ?) AND b
MATCH(?)": 3,
+ }
+ for query, want := range cases {
+ ps, err := Prepare(query)
+ if err != nil {
+ t.Fatalf("Prepare %q: %v", query, err)
+ }
+ if got := ps.NumPlaceholders(); got != want {
+ t.Errorf("NumPlaceholders(%q) = %d, want %d", query,
got, want)
+ }
+ }
+}
+
+func TestEstimatedSize(t *testing.T) {
+ ps, err := Prepare("SELECT * FROM STREAM sw IN default WHERE service_id
= ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if sz := ps.EstimatedSize(); sz <= 0 {
+ t.Fatalf("EstimatedSize = %d, want > 0", sz)
+ }
+ // The same query yields the same footprint.
+ ps2, err := Prepare("SELECT * FROM STREAM sw IN default WHERE
service_id = ?")
+ if err != nil {
+ t.Fatalf("Prepare: %v", err)
+ }
+ if ps.EstimatedSize() != ps2.EstimatedSize() {
+ t.Errorf("EstimatedSize not deterministic: %d vs %d",
ps.EstimatedSize(), ps2.EstimatedSize())
+ }
+ // A statement with more grammar nodes retains more memory.
+ simple, err := Prepare("SELECT * FROM STREAM sw IN default")
+ if err != nil {
+ t.Fatalf("Prepare simple: %v", err)
+ }
+ rich, err := Prepare("SELECT * FROM STREAM sw IN default WHERE a = ?
AND b IN (?, ?) AND c MATCH(?) LIMIT ? OFFSET ?")
+ if err != nil {
+ t.Fatalf("Prepare rich: %v", err)
+ }
+ if rich.EstimatedSize() <= simple.EstimatedSize() {
+ t.Errorf("rich size %d not larger than simple %d",
rich.EstimatedSize(), simple.EstimatedSize())
+ }
+}
+
+// TestDeepSizeBranches drives deepSize through real prepared statements, which
+// exercise all of its branches: nil and non-nil pointers (absent vs present
+// optional clauses), structs (grammar nodes), nil and non-nil slices (empty vs
+// populated IN lists and the groups list), strings (identifiers and
literals), and
+// scalar defaults (LIMIT ints, bool flags).
+func TestDeepSizeBranches(t *testing.T) {
+ sizeOf := func(query string) int {
+ ps, err := Prepare(query)
+ if err != nil {
+ t.Fatalf("Prepare %q: %v", query, err)
+ }
+ return ps.EstimatedSize()
+ }
+
+ // A minimal query already covers structs, absent/present pointers, the
groups
+ // slice, strings, and scalar fields; its footprint must be positive.
+ base := sizeOf("SELECT * FROM STREAM sw IN default")
+ if base <= 0 {
+ t.Fatalf("base size = %d, want > 0", base)
+ }
+ // A longer string literal adds out-of-line string bytes (String
branch).
+ shortStr := sizeOf("SELECT * FROM STREAM sw IN default WHERE service_id
= 'x'")
+ longStr := sizeOf("SELECT * FROM STREAM sw IN default WHERE service_id
= 'xxxxxxxxxxxxxxxxxxxxxxxx'")
+ if longStr <= shortStr {
+ t.Errorf("longer string literal did not grow size: %d <= %d",
longStr, shortStr)
+ }
+ // A longer IN list adds slice-backing and value-node bytes
(Slice/Pointer).
+ oneElem := sizeOf("SELECT * FROM STREAM sw IN default WHERE service_id
IN (?)")
+ manyElem := sizeOf("SELECT * FROM STREAM sw IN default WHERE service_id
IN (?, ?, ?, ?, ?)")
+ if manyElem <= oneElem {
+ t.Errorf("longer IN list did not grow size: %d <= %d",
manyElem, oneElem)
+ }
+ // Adding optional clauses turns nil pointers non-nil and adds int/bool
scalars.
+ withClauses := sizeOf("SELECT * FROM STREAM sw IN default TIME > ?
WHERE a = ? LIMIT ? OFFSET ?")
+ if withClauses <= base {
+ t.Errorf("added clauses did not grow size: %d <= %d",
withClauses, base)
+ }
+}
diff --git a/pkg/bydbql/transformer.go b/pkg/bydbql/transformer.go
index d1460b85e..2167a9283 100644
--- a/pkg/bydbql/transformer.go
+++ b/pkg/bydbql/transformer.go
@@ -99,18 +99,107 @@ func NewTransformer(registry metadata.Repo) *Transformer {
}
}
-// Transform transforms a Grammar into a native query request.
+// transformRun holds the per-request state of a single transformation: the
+// stateless Transformer plus the bound parameter overlay (nil for a literal or
+// in-place-bound grammar). The conversion methods hang off transformRun so the
+// shared Transformer stays safe for concurrent requests.
+type transformRun struct {
+ *Transformer
+ bound []resolvedParam
+}
+
+// resolveValue returns the effective literal value node for a scalar position:
+// the overlay's bound node for a placeholder, or the node itself for a literal
+// (also the in-place-bound path, where bound is nil and the node was mutated).
+func (r *transformRun) resolveValue(v *GrammarValue) *GrammarValue {
+ if v != nil && v.Param && r.bound != nil {
+ return r.bound[v.ParamIndex].values[0]
+ }
+ return v
+}
+
+// resolveValues expands a list of value nodes, splicing each placeholder's
bound
+// values (one, or several for an array) in place of the placeholder node. A
list
+// with no placeholder (a literal list, or a re-resolve of an already-expanded
+// one) is returned unchanged, so the bound path allocates only when it must.
+func (r *transformRun) resolveValues(values []*GrammarValue) []*GrammarValue {
+ if r.bound == nil {
+ return values
+ }
+ hasParam := false
+ for _, v := range values {
+ if v != nil && v.Param {
+ hasParam = true
+ break
+ }
+ }
+ if !hasParam {
+ return values
+ }
+ out := make([]*GrammarValue, 0, len(values))
+ for _, v := range values {
+ if v != nil && v.Param {
+ out = append(out, r.bound[v.ParamIndex].values...)
+ continue
+ }
+ out = append(out, v)
+ }
+ return out
+}
+
+// resolveTimeString returns a TIME value's effective string: the overlay's
bound
+// string for a placeholder, or the node's own literal.
+func (r *transformRun) resolveTimeString(v *GrammarTimeValue) string {
+ if v != nil && v.Param && r.bound != nil {
+ return r.bound[v.ParamIndex].timeStr
+ }
+ return v.ToString()
+}
+
+// resolveCount returns a count position's effective value: the overlay's bound
+// int for a placeholder, or the node's own literal.
+func (r *transformRun) resolveCount(value int, param bool, paramIndex int) int
{
+ if param && r.bound != nil {
+ return r.bound[paramIndex].count
+ }
+ return value
+}
+
+// Transform transforms a Grammar into a native query request. The grammar must
+// be free of unbound placeholders: either a literal query or one bound in
place
+// by BindParams. Use TransformBound for a reusable prepared statement.
func (t *Transformer) Transform(ctx context.Context, grammar *Grammar)
(*TransformResult, error) {
+ return (&transformRun{Transformer: t}).transform(ctx, grammar)
+}
+
+// TransformBound transforms a bound prepared statement, reading parameter
values
+// from the per-request overlay instead of the immutable template.
+func (t *Transformer) TransformBound(ctx context.Context, bq *BoundQuery)
(*TransformResult, error) {
+ if bq == nil || bq.stmt == nil {
+ return nil, errors.New("nil bound query; construct it with
PreparedStatement.Bind")
+ }
+ // The template's ParamIndex values index into bq.values, so a
mismatched
+ // overlay would panic on an out-of-range read; fail fast instead.
+ if len(bq.values) != len(bq.stmt.specs) {
+ return nil, fmt.Errorf("bound query has %d value(s) but the
statement has %d placeholder(s); construct it with PreparedStatement.Bind",
+ len(bq.values), len(bq.stmt.specs))
+ }
+ return (&transformRun{Transformer: t, bound: bq.values}).transform(ctx,
bq.stmt.template)
+}
+
+func (r *transformRun) transform(ctx context.Context, grammar *Grammar)
(*TransformResult, error) {
// Defense in depth: an unbound placeholder would otherwise transform
into an
- // empty string or a zero count silently instead of failing loudly. A
bound
- // grammar is guaranteed placeholder-free, so it skips the walk.
- if !grammar.paramsBound {
+ // empty string or a zero count silently instead of failing loudly. The
+ // overlay path (r.bound != nil) carries the values; the in-place path
sets
+ // paramsBound; a literal grammar has no placeholders to begin with.
+ if !grammar.paramsBound && r.bound == nil {
if unbound := countUnboundParams(grammar); unbound > 0 {
- return nil, fmt.Errorf("query contains %d unbound
placeholder(s); BindParams must be called before Transform", unbound)
+ return nil, fmt.Errorf("query contains %d unbound
placeholder(s); bind parameters before transforming", unbound)
}
}
// Literal counts get the same wrap guard as bound parameters, so LIMIT
-5
- // fails loudly instead of wrapping when narrowed to uint32.
+ // fails loudly instead of wrapping when narrowed to uint32. Overlay
counts
+ // were already range-checked at Bind time.
if err := validateGrammarCounts(grammar); err != nil {
return nil, err
}
@@ -119,13 +208,13 @@ func (t *Transformer) Transform(ctx context.Context,
grammar *Grammar) (*Transfo
resourceType := grammar.Select.From.ResourceType
switch strings.ToUpper(resourceType) {
case "STREAM":
- return t.transformStreamQuery(ctx, grammar)
+ return r.transformStreamQuery(ctx, grammar)
case "MEASURE":
- return t.transformMeasureQuery(ctx, grammar)
+ return r.transformMeasureQuery(ctx, grammar)
case "TRACE":
- return t.transformTraceQuery(ctx, grammar)
+ return r.transformTraceQuery(ctx, grammar)
case "PROPERTY":
- return t.transformPropertyQuery(ctx, grammar)
+ return r.transformPropertyQuery(ctx, grammar)
default:
return nil, fmt.Errorf("unsupported resource type in
select statement: %s", resourceType)
}
@@ -133,14 +222,14 @@ func (t *Transformer) Transform(ctx context.Context,
grammar *Grammar) (*Transfo
if grammar.TopN != nil {
resourceType := grammar.TopN.From.ResourceType
if strings.EqualFold(resourceType, "MEASURE") {
- return t.transformTopNMeasureQuery(ctx, grammar)
+ return r.transformTopNMeasureQuery(ctx, grammar)
}
return nil, fmt.Errorf("unsupported resource type in topn
statement: %s", resourceType)
}
return nil, errors.New("grammar must contain either Select or TopN
statement")
}
-func (t *Transformer) transformStreamQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
+func (r *transformRun) transformStreamQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
statement := grammar.Select
if statement == nil {
return nil, errors.New("stream query must be a select
statement")
@@ -151,15 +240,15 @@ func (t *Transformer) transformStreamQuery(ctx
context.Context, grammar *Grammar
resourceName := statement.From.ResourceName
// validate query
- if err := t.validateGroupOrResourceName(groups, resourceName); err !=
nil {
+ if err := r.validateGroupOrResourceName(groups, resourceName); err !=
nil {
return nil, err
}
// query schema for getting tags
- projection, _, allTags, _, err := t.convertTagAndField(
+ projection, _, allTags, _, err := r.convertTagAndField(
groups, resourceName,
func(group, name string) ([]*databasev1.TagFamilySpec,
[]*databasev1.FieldSpec, error) {
- stream, getErr :=
t.schemaRegistry.StreamRegistry().GetStream(ctx, &commonv1.Metadata{
+ stream, getErr :=
r.schemaRegistry.StreamRegistry().GetStream(ctx, &commonv1.Metadata{
Name: name,
Group: group,
})
@@ -173,26 +262,26 @@ func (t *Transformer) transformStreamQuery(ctx
context.Context, grammar *Grammar
}
// convert time range
- timeRange, err := t.convertTimeRange(time.Now(), statement.Time)
+ timeRange, err := r.convertTimeRange(time.Now(), statement.Time)
if err != nil {
return nil, fmt.Errorf("failed to convert time range: %w", err)
}
// convert order by
- orderBy := t.convertSelectOrderBy(statement.OrderBy)
+ orderBy := r.convertSelectOrderBy(statement.OrderBy)
// convert criteria
- criteria, err := t.convertSelectCriteria(statement.Where, allTags)
+ criteria, err := r.convertSelectCriteria(statement.Where, allTags)
if err != nil {
return nil, fmt.Errorf("failed to convert criteria: %w", err)
}
var offset, limit uint32
if statement.Offset != nil {
- offset = uint32(statement.Offset.Value)
+ offset = uint32(r.resolveCount(statement.Offset.Value,
statement.Offset.Param, statement.Offset.ParamIndex))
}
if statement.Limit != nil {
- limit = uint32(statement.Limit.Value)
+ limit = uint32(r.resolveCount(statement.Limit.Value,
statement.Limit.Param, statement.Limit.ParamIndex))
}
// extract stages
@@ -219,7 +308,7 @@ func (t *Transformer) transformStreamQuery(ctx
context.Context, grammar *Grammar
}, nil
}
-func (t *Transformer) transformMeasureQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
+func (r *transformRun) transformMeasureQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
statement := grammar.Select
if statement == nil {
return nil, errors.New("measure query must be a select
statement")
@@ -230,15 +319,15 @@ func (t *Transformer) transformMeasureQuery(ctx
context.Context, grammar *Gramma
resourceName := statement.From.ResourceName
// validate query
- if err := t.validateGroupOrResourceName(groups, resourceName); err !=
nil {
+ if err := r.validateGroupOrResourceName(groups, resourceName); err !=
nil {
return nil, err
}
// query schema for getting tags and fields
- projection, fields, allTags, allFields, err := t.convertTagAndField(
+ projection, fields, allTags, allFields, err := r.convertTagAndField(
groups, resourceName,
func(group, name string) ([]*databasev1.TagFamilySpec,
[]*databasev1.FieldSpec, error) {
- measure, getErr :=
t.schemaRegistry.MeasureRegistry().GetMeasure(ctx, &commonv1.Metadata{
+ measure, getErr :=
r.schemaRegistry.MeasureRegistry().GetMeasure(ctx, &commonv1.Metadata{
Name: name,
Group: group,
})
@@ -259,28 +348,28 @@ func (t *Transformer) transformMeasureQuery(ctx
context.Context, grammar *Gramma
}
// convert time range
- timeRange, err := t.convertTimeRange(time.Now(), statement.Time)
+ timeRange, err := r.convertTimeRange(time.Now(), statement.Time)
if err != nil {
return nil, fmt.Errorf("failed to convert time range: %w", err)
}
// convert order by
- orderBy := t.convertSelectOrderBy(statement.OrderBy)
+ orderBy := r.convertSelectOrderBy(statement.OrderBy)
// convert criteria
- criteria, err := t.convertSelectCriteria(statement.Where, allTags)
+ criteria, err := r.convertSelectCriteria(statement.Where, allTags)
if err != nil {
return nil, fmt.Errorf("failed to convert criteria: %w", err)
}
// convert aggregation
- agg, err := t.convertAggregation(statement.Projection, allFields)
+ agg, err := r.convertAggregation(statement.Projection, allFields)
if err != nil {
return nil, fmt.Errorf("failed to convert aggregation: %w", err)
}
// convert group by
- groupBy, err := t.convertGroupBy(statement.GroupBy, projection, fields)
+ groupBy, err := r.convertGroupBy(statement.GroupBy, projection, fields)
if err != nil {
return nil, fmt.Errorf("failed to convert group by: %w", err)
}
@@ -288,17 +377,17 @@ func (t *Transformer) transformMeasureQuery(ctx
context.Context, grammar *Gramma
return nil, errors.New("when aggregation and group by are both
present, group by must include a field")
}
- top, err := t.convertTOP(statement.Projection, allFields)
+ top, err := r.convertTOP(statement.Projection, allFields)
if err != nil {
return nil, fmt.Errorf("failed to convert top: %w", err)
}
var offset, limit uint32
if statement.Offset != nil {
- offset = uint32(statement.Offset.Value)
+ offset = uint32(r.resolveCount(statement.Offset.Value,
statement.Offset.Param, statement.Offset.ParamIndex))
}
if statement.Limit != nil {
- limit = uint32(statement.Limit.Value)
+ limit = uint32(r.resolveCount(statement.Limit.Value,
statement.Limit.Param, statement.Limit.ParamIndex))
}
// extract stages
@@ -329,7 +418,7 @@ func (t *Transformer) transformMeasureQuery(ctx
context.Context, grammar *Gramma
}, nil
}
-func (t *Transformer) transformTraceQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
+func (r *transformRun) transformTraceQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
statement := grammar.Select
if statement == nil {
return nil, errors.New("trace query must be a select statement")
@@ -340,11 +429,11 @@ func (t *Transformer) transformTraceQuery(ctx
context.Context, grammar *Grammar)
resourceName := statement.From.ResourceName
// validate query
- if err := t.validateGroupOrResourceName(groups, resourceName); err !=
nil {
+ if err := r.validateGroupOrResourceName(groups, resourceName); err !=
nil {
return nil, err
}
- timeRange, err := t.convertTimeRange(time.Now(), statement.Time)
+ timeRange, err := r.convertTimeRange(time.Now(), statement.Time)
if err != nil {
return nil, fmt.Errorf("failed to convert time range: %w", err)
}
@@ -352,7 +441,7 @@ func (t *Transformer) transformTraceQuery(ctx
context.Context, grammar *Grammar)
// query the trace schema for getting tags
allTags := make(map[string]*tagSpecWithFamily)
for _, g := range groups {
- trace, getErr := t.schemaRegistry.TraceRegistry().GetTrace(ctx,
&commonv1.Metadata{
+ trace, getErr := r.schemaRegistry.TraceRegistry().GetTrace(ctx,
&commonv1.Metadata{
Name: resourceName,
Group: g,
})
@@ -391,20 +480,20 @@ func (t *Transformer) transformTraceQuery(ctx
context.Context, grammar *Grammar)
}
// convert criteria
- criteria, err := t.convertSelectCriteria(statement.Where, allTags)
+ criteria, err := r.convertSelectCriteria(statement.Where, allTags)
if err != nil {
return nil, fmt.Errorf("failed to convert criteria: %w", err)
}
// convert order by
- orderBy := t.convertSelectOrderBy(statement.OrderBy)
+ orderBy := r.convertSelectOrderBy(statement.OrderBy)
var offset, limit uint32
if statement.Offset != nil {
- offset = uint32(statement.Offset.Value)
+ offset = uint32(r.resolveCount(statement.Offset.Value,
statement.Offset.Param, statement.Offset.ParamIndex))
}
if statement.Limit != nil {
- limit = uint32(statement.Limit.Value)
+ limit = uint32(r.resolveCount(statement.Limit.Value,
statement.Limit.Param, statement.Limit.ParamIndex))
}
// extract stages
@@ -431,7 +520,7 @@ func (t *Transformer) transformTraceQuery(ctx
context.Context, grammar *Grammar)
}, nil
}
-func (t *Transformer) transformPropertyQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
+func (r *transformRun) transformPropertyQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
statement := grammar.Select
if statement == nil {
return nil, errors.New("property query must be a select
statement")
@@ -442,14 +531,14 @@ func (t *Transformer) transformPropertyQuery(ctx
context.Context, grammar *Gramm
resourceName := statement.From.ResourceName
// validate query
- if err := t.validateGroupOrResourceName(groups, resourceName); err !=
nil {
+ if err := r.validateGroupOrResourceName(groups, resourceName); err !=
nil {
return nil, err
}
// get property schema to extract all tags
allTags := make(map[string]*tagSpecWithFamily)
for _, g := range groups {
- property, getErr :=
t.schemaRegistry.PropertyRegistry().GetProperty(ctx, &commonv1.Metadata{
+ property, getErr :=
r.schemaRegistry.PropertyRegistry().GetProperty(ctx, &commonv1.Metadata{
Name: resourceName,
Group: g,
})
@@ -499,7 +588,7 @@ func (t *Transformer) transformPropertyQuery(ctx
context.Context, grammar *Gramm
var criteria *modelv1.Criteria
if statement.Where != nil && statement.Where.Expr != nil {
var extractErr error
- ids, criteria, extractErr =
t.extractIDsAndCriteria(statement.Where.Expr, allTags)
+ ids, criteria, extractErr =
r.extractIDsAndCriteria(statement.Where.Expr, allTags)
if extractErr != nil {
return nil, fmt.Errorf("failed to convert criteria:
%w", extractErr)
}
@@ -508,13 +597,13 @@ func (t *Transformer) transformPropertyQuery(ctx
context.Context, grammar *Gramm
// handle limit
var limit uint32
if statement.Limit != nil {
- limit = uint32(statement.Limit.Value)
+ limit = uint32(r.resolveCount(statement.Limit.Value,
statement.Limit.Param, statement.Limit.ParamIndex))
}
// handle ORDER BY
var orderBy *propertyv1.QueryOrder
if statement.OrderBy != nil {
- modelQueryOrder := t.convertSelectOrderBy(statement.OrderBy)
+ modelQueryOrder := r.convertSelectOrderBy(statement.OrderBy)
if modelQueryOrder != nil {
orderBy = &propertyv1.QueryOrder{
TagName: modelQueryOrder.IndexRuleName,
@@ -539,7 +628,7 @@ func (t *Transformer) transformPropertyQuery(ctx
context.Context, grammar *Gramm
}, nil
}
-func (t *Transformer) transformTopNMeasureQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
+func (r *transformRun) transformTopNMeasureQuery(ctx context.Context, grammar
*Grammar) (*TransformResult, error) {
statement := grammar.TopN
if statement == nil {
return nil, errors.New("topn measure query must be a topn
statement")
@@ -550,12 +639,12 @@ func (t *Transformer) transformTopNMeasureQuery(ctx
context.Context, grammar *Gr
resourceName := statement.From.ResourceName
// validate query
- if err := t.validateGroupOrResourceName(groups, resourceName); err !=
nil {
+ if err := r.validateGroupOrResourceName(groups, resourceName); err !=
nil {
return nil, err
}
// convert time range
- timeRange, err := t.convertTimeRange(time.Now(), statement.Time)
+ timeRange, err := r.convertTimeRange(time.Now(), statement.Time)
if err != nil {
return nil, fmt.Errorf("failed to convert time range: %w", err)
}
@@ -563,20 +652,20 @@ func (t *Transformer) transformTopNMeasureQuery(ctx
context.Context, grammar *Gr
// convert agg
var aggFunc modelv1.AggregationFunction
if statement.AggregateBy != nil {
- aggFunc, err =
t.convertAggregationFunc(statement.AggregateBy.Function.Function)
+ aggFunc, err =
r.convertAggregationFunc(statement.AggregateBy.Function.Function)
if err != nil {
return nil, err
}
}
// convert conditions
- conditions, err := t.convertTopNAndConditions(ctx, statement.Where,
groups, resourceName)
+ conditions, err := r.convertTopNAndConditions(ctx, statement.Where,
groups, resourceName)
if err != nil {
return nil, fmt.Errorf("failed to convert criteria: %w", err)
}
// convert order by
- orderBy := t.convertTopNOrderBy(statement.OrderBy)
+ orderBy := r.convertTopNOrderBy(statement.OrderBy)
sort := modelv1.Sort_SORT_UNSPECIFIED
if orderBy != nil {
sort = orderBy.Sort
@@ -595,7 +684,7 @@ func (t *Transformer) transformTopNMeasureQuery(ctx
context.Context, grammar *Gr
Groups: groups,
Name: resourceName,
TimeRange: timeRange,
- TopN: int32(statement.N),
+ TopN: int32(r.resolveCount(statement.N,
statement.NParam, statement.NParamIndex)),
Agg: aggFunc,
Conditions: conditions,
FieldValueSort: sort,
@@ -605,22 +694,22 @@ func (t *Transformer) transformTopNMeasureQuery(ctx
context.Context, grammar *Gr
}, nil
}
-func (t *Transformer) convertSelectCriteria(where *GrammarSelectWhereClause,
allTags map[string]*tagSpecWithFamily) (*modelv1.Criteria, error) {
+func (r *transformRun) convertSelectCriteria(where *GrammarSelectWhereClause,
allTags map[string]*tagSpecWithFamily) (*modelv1.Criteria, error) {
if where == nil || where.Expr == nil {
return nil, nil
}
- return t.convertOrExpr(where.Expr, allTags, nil)
+ return r.convertOrExpr(where.Expr, allTags, nil)
}
-func (t *Transformer) convertTopNAndConditions(ctx context.Context, where
*GrammarTopNWhereClause, groups []string, resourceName string)
([]*modelv1.Condition, error) {
+func (r *transformRun) convertTopNAndConditions(ctx context.Context, where
*GrammarTopNWhereClause, groups []string, resourceName string)
([]*modelv1.Condition, error) {
if where == nil || where.Expr == nil {
return nil, nil
}
allTags := make(map[string]*tagSpecWithFamily)
for _, g := range groups {
- aggregation, getErr :=
t.schemaRegistry.TopNAggregationRegistry().GetTopNAggregation(ctx,
&commonv1.Metadata{
+ aggregation, getErr :=
r.schemaRegistry.TopNAggregationRegistry().GetTopNAggregation(ctx,
&commonv1.Metadata{
Name: resourceName,
Group: g,
})
@@ -628,7 +717,7 @@ func (t *Transformer) convertTopNAndConditions(ctx
context.Context, where *Gramm
return nil, fmt.Errorf("failed to get topn aggregation
%s/%s: %w", g, resourceName, getErr)
}
sourceMeasure := aggregation.SourceMeasure
- measure, getErr :=
t.schemaRegistry.MeasureRegistry().GetMeasure(ctx, sourceMeasure)
+ measure, getErr :=
r.schemaRegistry.MeasureRegistry().GetMeasure(ctx, sourceMeasure)
if getErr != nil {
return nil, fmt.Errorf("failed to get measure %s/%s:
%w", sourceMeasure.Group, sourceMeasure.Name, getErr)
}
@@ -644,7 +733,7 @@ func (t *Transformer) convertTopNAndConditions(ctx
context.Context, where *Gramm
}
var conditions []*modelv1.Condition
- _, err := t.convertAndExpr(where.Expr, allTags, func(c
*modelv1.Condition) {
+ _, err := r.convertAndExpr(where.Expr, allTags, func(c
*modelv1.Condition) {
conditions = append(conditions, c)
})
if err != nil {
@@ -654,7 +743,7 @@ func (t *Transformer) convertTopNAndConditions(ctx
context.Context, where *Gramm
return conditions, nil
}
-func (t *Transformer) convertGroupBy(g *GrammarGroupByClause, queryTags
*modelv1.TagProjection, queryFields []string) (*measurev1.QueryRequest_GroupBy,
error) {
+func (r *transformRun) convertGroupBy(g *GrammarGroupByClause, queryTags
*modelv1.TagProjection, queryFields []string) (*measurev1.QueryRequest_GroupBy,
error) {
if g == nil {
return nil, nil
}
@@ -754,7 +843,7 @@ func (t *Transformer) convertGroupBy(g
*GrammarGroupByClause, queryTags *modelv1
return groupBy, nil
}
-func (t *Transformer) convertAggregation(projection *GrammarProjection,
allFields map[string]*databasev1.FieldSpec)
(*measurev1.QueryRequest_Aggregation, error) {
+func (r *transformRun) convertAggregation(projection *GrammarProjection,
allFields map[string]*databasev1.FieldSpec)
(*measurev1.QueryRequest_Aggregation, error) {
var columns []*GrammarColumn
if projection != nil && len(projection.Columns) > 0 {
columns = append(columns, projection.Columns...)
@@ -789,7 +878,7 @@ func (t *Transformer) convertAggregation(projection
*GrammarProjection, allField
return nil, fmt.Errorf("field %s not found in schema",
aggColName)
}
- aggFunc, err := t.convertAggregationFunc(aggCol.Aggregate.Function)
+ aggFunc, err := r.convertAggregationFunc(aggCol.Aggregate.Function)
if err != nil {
return nil, err
}
@@ -800,7 +889,7 @@ func (t *Transformer) convertAggregation(projection
*GrammarProjection, allField
}, nil
}
-func (t *Transformer) convertAggregationFunc(f string)
(modelv1.AggregationFunction, error) {
+func (r *transformRun) convertAggregationFunc(f string)
(modelv1.AggregationFunction, error) {
switch strings.ToUpper(f) {
case "MEAN", "AVG":
return modelv1.AggregationFunction_AGGREGATION_FUNCTION_MEAN,
nil
@@ -817,12 +906,12 @@ func (t *Transformer) convertAggregationFunc(f string)
(modelv1.AggregationFunct
}
}
-func (t *Transformer) convertOrExpr(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
+func (r *transformRun) convertOrExpr(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
if expr == nil {
return nil, nil
}
- leftCriteria, err := t.convertAndExpr(expr.Left, allTags, accept)
+ leftCriteria, err := r.convertAndExpr(expr.Left, allTags, accept)
if err != nil {
return nil, err
}
@@ -833,7 +922,7 @@ func (t *Transformer) convertOrExpr(expr *GrammarOrExpr,
allTags map[string]*tag
// process all OR operations
for _, orRight := range expr.Right {
- rightCriteria, err := t.convertAndExpr(orRight.Right, allTags,
accept)
+ rightCriteria, err := r.convertAndExpr(orRight.Right, allTags,
accept)
if err != nil {
return nil, err
}
@@ -852,12 +941,12 @@ func (t *Transformer) convertOrExpr(expr *GrammarOrExpr,
allTags map[string]*tag
return leftCriteria, nil
}
-func (t *Transformer) convertAndExpr(expr *GrammarAndExpr, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
+func (r *transformRun) convertAndExpr(expr *GrammarAndExpr, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
if expr == nil {
return nil, nil
}
- leftCriteria, err := t.convertPredicate(expr.Left, allTags, accept)
+ leftCriteria, err := r.convertPredicate(expr.Left, allTags, accept)
if err != nil {
return nil, err
}
@@ -868,7 +957,7 @@ func (t *Transformer) convertAndExpr(expr *GrammarAndExpr,
allTags map[string]*t
// process all AND operations
for _, andRight := range expr.Right {
- rightCriteria, err := t.convertPredicate(andRight.Right,
allTags, accept)
+ rightCriteria, err := r.convertPredicate(andRight.Right,
allTags, accept)
if err != nil {
return nil, err
}
@@ -887,31 +976,31 @@ func (t *Transformer) convertAndExpr(expr
*GrammarAndExpr, allTags map[string]*t
return leftCriteria, nil
}
-func (t *Transformer) convertPredicate(pred *GrammarPredicate, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
+func (r *transformRun) convertPredicate(pred *GrammarPredicate, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
if pred == nil {
return nil, nil
}
if pred.Paren != nil {
- return t.convertOrExpr(pred.Paren, allTags, accept)
+ return r.convertOrExpr(pred.Paren, allTags, accept)
}
if pred.Binary != nil {
- return t.convertBinaryPredicate(pred.Binary, allTags, accept)
+ return r.convertBinaryPredicate(pred.Binary, allTags, accept)
}
if pred.In != nil {
- return t.convertInPredicate(pred.In, allTags, accept)
+ return r.convertInPredicate(pred.In, allTags, accept)
}
if pred.Having != nil {
- return t.convertHavingPredicate(pred.Having, allTags, accept)
+ return r.convertHavingPredicate(pred.Having, allTags, accept)
}
return nil, errors.New("empty predicate")
}
-func (t *Transformer) convertBinaryPredicate(
+func (r *transformRun) convertBinaryPredicate(
pred *GrammarBinaryPredicate,
allTags map[string]*tagSpecWithFamily,
accept func(c *modelv1.Condition),
@@ -927,17 +1016,17 @@ func (t *Transformer) convertBinaryPredicate(
}
if pred.Tail.Match != nil {
- return t.convertMatchPredicate(identifierName, pred.Tail.Match,
accept)
+ return r.convertMatchPredicate(identifierName, pred.Tail.Match,
accept)
}
if pred.Tail.Compare != nil {
- return t.convertComparePredicate(identifierName,
pred.Tail.Compare, tagSpec, accept)
+ return r.convertComparePredicate(identifierName,
pred.Tail.Compare, tagSpec, accept)
}
return nil, errors.New("empty binary predicate tail")
}
-func (t *Transformer) convertMatchPredicate(identifierName string, match
*GrammarMatchTail, accept func(c *modelv1.Condition)) (*modelv1.Criteria,
error) {
+func (r *transformRun) convertMatchPredicate(identifierName string, match
*GrammarMatchTail, accept func(c *modelv1.Condition)) (*modelv1.Criteria,
error) {
if match.Values == nil {
return nil, fmt.Errorf("MATCH operator requires values")
}
@@ -948,6 +1037,9 @@ func (t *Transformer) convertMatchPredicate(identifierName
string, match *Gramma
} else if match.Values.Array != nil {
values = match.Values.Array
}
+ // Expand any placeholder into its bound value(s); a single placeholder
+ // bound to an array becomes multiple values, matching the literal form.
+ values = r.resolveValues(values)
if len(values) == 0 {
return nil, fmt.Errorf("MATCH requires at least one value")
@@ -970,14 +1062,14 @@ func (t *Transformer)
convertMatchPredicate(identifierName string, match *Gramma
// single value: set as Str
cond.Value = &modelv1.TagValue{
Value: &modelv1.TagValue_Str{
- Str: &modelv1.Str{Value:
t.grammarValueToString(values[0])},
+ Str: &modelv1.Str{Value:
r.grammarValueToString(values[0])},
},
}
} else {
// multiple values: set as string array
strArr := make([]string, len(values))
for i, val := range values {
- strArr[i] = t.grammarValueToString(val)
+ strArr[i] = r.grammarValueToString(val)
}
cond.Value = &modelv1.TagValue{
Value: &modelv1.TagValue_StrArray{
@@ -1016,7 +1108,7 @@ func (t *Transformer)
convertMatchPredicate(identifierName string, match *Gramma
}, nil
}
-func (t *Transformer) convertComparePredicate(
+func (r *transformRun) convertComparePredicate(
identifierName string,
compare *GrammarCompareTail,
tagSpec *tagSpecWithFamily,
@@ -1024,10 +1116,10 @@ func (t *Transformer) convertComparePredicate(
) (*modelv1.Criteria, error) {
cond := &modelv1.Condition{
Name: identifierName,
- Op: t.convertCompareOp(compare.Operator),
+ Op: r.convertCompareOp(compare.Operator),
}
- if err := t.setGrammarConditionValue(cond, compare.Value, tagSpec); err
!= nil {
+ if err := r.setGrammarConditionValue(cond, compare.Value, tagSpec); err
!= nil {
return nil, fmt.Errorf("failed to set condition for tag %s:
%w", identifierName, err)
}
@@ -1042,7 +1134,7 @@ func (t *Transformer) convertComparePredicate(
}, nil
}
-func (t *Transformer) convertInPredicate(pred *GrammarInPredicate, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
+func (r *transformRun) convertInPredicate(pred *GrammarInPredicate, allTags
map[string]*tagSpecWithFamily, accept func(c *modelv1.Condition))
(*modelv1.Criteria, error) {
identifierName, nameErr := pred.Identifier.ToString(false)
if nameErr != nil {
return nil, fmt.Errorf("failed to parse identifier: %w",
nameErr)
@@ -1063,7 +1155,7 @@ func (t *Transformer) convertInPredicate(pred
*GrammarInPredicate, allTags map[s
Op: op,
}
- if err := t.setGrammarMultiValueCondition(cond, pred.Values, tagSpec);
err != nil {
+ if err := r.setGrammarMultiValueCondition(cond, pred.Values, tagSpec);
err != nil {
return nil, fmt.Errorf("failed to set condition for tag %s:
%w", identifierName, err)
}
@@ -1078,7 +1170,7 @@ func (t *Transformer) convertInPredicate(pred
*GrammarInPredicate, allTags map[s
}, nil
}
-func (t *Transformer) convertHavingPredicate(
+func (r *transformRun) convertHavingPredicate(
pred *GrammarHavingPredicate,
allTags map[string]*tagSpecWithFamily,
accept func(c *modelv1.Condition),
@@ -1103,11 +1195,21 @@ func (t *Transformer) convertHavingPredicate(
Op: op,
}
+ // Preserve the literal single-vs-list form: a single-value HAVING stays
+ // single unless a bound array expands it to several values, while a
+ // parenthesized list stays a list regardless of count.
+ wasSingle := pred.Values.Single != nil
+ values := pred.Values.Array
+ if wasSingle {
+ values = []*GrammarValue{pred.Values.Single}
+ }
+ values = r.resolveValues(values)
+
var err error
- if pred.Values.Single != nil {
- err = t.setGrammarConditionValue(cond, pred.Values.Single,
tagSpec)
- } else if pred.Values.Array != nil {
- err = t.setGrammarMultiValueCondition(cond, pred.Values.Array,
tagSpec)
+ if wasSingle && len(values) == 1 {
+ err = r.setGrammarConditionValue(cond, values[0], tagSpec)
+ } else {
+ err = r.setGrammarMultiValueCondition(cond, values, tagSpec)
}
if err != nil {
@@ -1125,7 +1227,8 @@ func (t *Transformer) convertHavingPredicate(
}, nil
}
-func (t *Transformer) setGrammarConditionValue(cond *modelv1.Condition, val
*GrammarValue, tagSpec *tagSpecWithFamily) error {
+func (r *transformRun) setGrammarConditionValue(cond *modelv1.Condition, val
*GrammarValue, tagSpec *tagSpecWithFamily) error {
+ val = r.resolveValue(val)
if val.Null {
cond.Value = &modelv1.TagValue{
Value: &modelv1.TagValue_Null{
@@ -1140,12 +1243,12 @@ func (t *Transformer) setGrammarConditionValue(cond
*modelv1.Condition, val *Gra
// Convert to Str
cond.Value = &modelv1.TagValue{
Value: &modelv1.TagValue_Str{
- Str: &modelv1.Str{Value:
t.grammarValueToString(val)},
+ Str: &modelv1.Str{Value:
r.grammarValueToString(val)},
},
}
case databasev1.TagType_TAG_TYPE_INT,
databasev1.TagType_TAG_TYPE_INT_ARRAY:
- intVal, err := t.grammarValueToInt64(val)
+ intVal, err := r.grammarValueToInt64(val)
if err != nil {
return err
}
@@ -1165,7 +1268,8 @@ func (t *Transformer) setGrammarConditionValue(cond
*modelv1.Condition, val *Gra
return nil
}
-func (t *Transformer) setGrammarMultiValueCondition(cond *modelv1.Condition,
values []*GrammarValue, tagSpec *tagSpecWithFamily) error {
+func (r *transformRun) setGrammarMultiValueCondition(cond *modelv1.Condition,
values []*GrammarValue, tagSpec *tagSpecWithFamily) error {
+ values = r.resolveValues(values)
switch tagSpec.tag.Type {
case databasev1.TagType_TAG_TYPE_STRING,
databasev1.TagType_TAG_TYPE_STRING_ARRAY:
strArr := make([]string, len(values))
@@ -1173,7 +1277,7 @@ func (t *Transformer) setGrammarMultiValueCondition(cond
*modelv1.Condition, val
if val.Null {
return fmt.Errorf("NULL is not allowed in array
values")
}
- strArr[i] = t.grammarValueToString(val)
+ strArr[i] = r.grammarValueToString(val)
}
cond.Value = &modelv1.TagValue{
Value: &modelv1.TagValue_StrArray{
@@ -1187,7 +1291,7 @@ func (t *Transformer) setGrammarMultiValueCondition(cond
*modelv1.Condition, val
if val.Null {
return fmt.Errorf("NULL is not allowed in array
values")
}
- intVal, err := t.grammarValueToInt64(val)
+ intVal, err := r.grammarValueToInt64(val)
if err != nil {
return err
}
@@ -1206,25 +1310,25 @@ func (t *Transformer)
setGrammarMultiValueCondition(cond *modelv1.Condition, val
return nil
}
-func (t *Transformer) convertTimeRange(now time.Time, timeClause
*GrammarTimeClause) (*modelv1.TimeRange, error) {
+func (r *transformRun) convertTimeRange(now time.Time, timeClause
*GrammarTimeClause) (*modelv1.TimeRange, error) {
if timeClause == nil {
return nil, nil
}
var begin, end time.Time
if timeClause.Between != nil {
- beginTS, err := t.parseTimestamp(now,
timeClause.Between.Begin.ToString())
+ beginTS, err := r.parseTimestamp(now,
r.resolveTimeString(timeClause.Between.Begin))
if err != nil {
return nil, err
}
- endTS, err := t.parseTimestamp(now,
timeClause.Between.End.ToString())
+ endTS, err := r.parseTimestamp(now,
r.resolveTimeString(timeClause.Between.End))
if err != nil {
return nil, err
}
begin = *beginTS
end = *endTS
} else if timeClause.Comparator != nil && timeClause.Value != nil {
- timestamp, err := t.parseTimestamp(now,
timeClause.Value.ToString())
+ timestamp, err := r.parseTimestamp(now,
r.resolveTimeString(timeClause.Value))
if err != nil {
return nil, err
}
@@ -1255,7 +1359,7 @@ func (t *Transformer) convertTimeRange(now time.Time,
timeClause *GrammarTimeCla
}, nil
}
-func (t *Transformer) parseTimestamp(now time.Time, timestamp string)
(*time.Time, error) {
+func (r *transformRun) parseTimestamp(now time.Time, timestamp string)
(*time.Time, error) {
// Try parsing as absolute time first (RFC3339)
if parsedTime, err := time.Parse(time.RFC3339, timestamp); err == nil {
return &parsedTime, nil
@@ -1275,7 +1379,7 @@ func (t *Transformer) parseTimestamp(now time.Time,
timestamp string) (*time.Tim
return &resultTime, nil
}
-func (t *Transformer) convertTagAndField(
+func (r *transformRun) convertTagAndField(
groups []string,
name string,
query func(group, name string) ([]*databasev1.TagFamilySpec,
[]*databasev1.FieldSpec, error),
@@ -1323,12 +1427,12 @@ func (t *Transformer) convertTagAndField(
if includeAll {
for _, spec := range allTags {
- if addErr := t.checkOrAddGrammarTagOrField(allTags,
allFields, &targetTagFamilies, &targetFields, columnExists, spec.tag.Name,
columnTypeTag, nil); addErr != nil {
+ if addErr := r.checkOrAddGrammarTagOrField(allTags,
allFields, &targetTagFamilies, &targetFields, columnExists, spec.tag.Name,
columnTypeTag, nil); addErr != nil {
return nil, nil, nil, nil, addErr
}
}
for _, f := range allFields {
- if addErr := t.checkOrAddGrammarTagOrField(allTags,
allFields, &targetTagFamilies, &targetFields, columnExists, f.Name,
columnTypeField, nil); addErr != nil {
+ if addErr := r.checkOrAddGrammarTagOrField(allTags,
allFields, &targetTagFamilies, &targetFields, columnExists, f.Name,
columnTypeField, nil); addErr != nil {
return nil, nil, nil, nil, addErr
}
}
@@ -1346,7 +1450,7 @@ func (t *Transformer) convertTagAndField(
if col.TypeSpec != nil {
colType = strings.ToUpper(*col.TypeSpec)
}
- if addErr := t.checkOrAddGrammarTagOrField(allTags, allFields,
&targetTagFamilies, &targetFields, columnExists, colName, colType,
col.Aggregate); addErr != nil {
+ if addErr := r.checkOrAddGrammarTagOrField(allTags, allFields,
&targetTagFamilies, &targetFields, columnExists, colName, colType,
col.Aggregate); addErr != nil {
return nil, nil, nil, nil, addErr
}
}
@@ -1358,7 +1462,7 @@ func (t *Transformer) convertTagAndField(
return tagProjection, targetFields, allTags, allFields, nil
}
-func (t *Transformer) findGrammarTagOrField(
+func (r *transformRun) findGrammarTagOrField(
sourceTags map[string]*tagSpecWithFamily,
sourceFields map[string]*databasev1.FieldSpec,
targetName string,
@@ -1378,7 +1482,7 @@ type tagSpecWithFamily struct {
family string
}
-func (t *Transformer) checkOrAddGrammarTagOrField(
+func (r *transformRun) checkOrAddGrammarTagOrField(
sourceTags map[string]*tagSpecWithFamily,
sourceFields map[string]*databasev1.FieldSpec,
targetTags *[]*modelv1.TagProjection_TagFamily,
@@ -1412,7 +1516,7 @@ func (t *Transformer) checkOrAddGrammarTagOrField(
return nil
}
- tag, field := t.findGrammarTagOrField(sourceTags, sourceFields,
colName, colType)
+ tag, field := r.findGrammarTagOrField(sourceTags, sourceFields,
colName, colType)
if tag == nil && field == nil {
return fmt.Errorf("column %s not found in schema", colName)
}
@@ -1472,7 +1576,7 @@ func (t *Transformer) checkOrAddGrammarTagOrField(
return nil
}
-func (t *Transformer) validateGroupOrResourceName(groups []string,
resourceName string) error {
+func (r *transformRun) validateGroupOrResourceName(groups []string,
resourceName string) error {
if len(groups) == 0 {
return errors.New("at least one group must be specified")
}
@@ -1483,7 +1587,7 @@ func (t *Transformer) validateGroupOrResourceName(groups
[]string, resourceName
return nil
}
-func (t *Transformer) convertSelectOrderBy(orderBy
*GrammarSelectOrderByClause) *modelv1.QueryOrder {
+func (r *transformRun) convertSelectOrderBy(orderBy
*GrammarSelectOrderByClause) *modelv1.QueryOrder {
if orderBy == nil {
return nil
}
@@ -1513,7 +1617,7 @@ func (t *Transformer) convertSelectOrderBy(orderBy
*GrammarSelectOrderByClause)
}
}
-func (t *Transformer) convertTopNOrderBy(orderBy *GrammarTopNOrderByClause)
*modelv1.QueryOrder {
+func (r *transformRun) convertTopNOrderBy(orderBy *GrammarTopNOrderByClause)
*modelv1.QueryOrder {
if orderBy == nil {
return nil
}
@@ -1528,7 +1632,7 @@ func (t *Transformer) convertTopNOrderBy(orderBy
*GrammarTopNOrderByClause) *mod
}
}
-func (t *Transformer) convertCompareOp(op string) modelv1.Condition_BinaryOp {
+func (r *transformRun) convertCompareOp(op string) modelv1.Condition_BinaryOp {
switch op {
case "=":
return modelv1.Condition_BINARY_OP_EQ
@@ -1547,7 +1651,7 @@ func (t *Transformer) convertCompareOp(op string)
modelv1.Condition_BinaryOp {
}
}
-func (t *Transformer) convertTOP(projection *GrammarProjection, fields
map[string]*databasev1.FieldSpec) (*measurev1.QueryRequest_Top, error) {
+func (r *transformRun) convertTOP(projection *GrammarProjection, fields
map[string]*databasev1.FieldSpec) (*measurev1.QueryRequest_Top, error) {
if projection == nil || projection.TopN == nil {
return nil, nil
}
@@ -1570,13 +1674,14 @@ func (t *Transformer) convertTOP(projection
*GrammarProjection, fields map[strin
}
return &measurev1.QueryRequest_Top{
- Number: int32(topn.N),
+ Number: int32(r.resolveCount(topn.N, topn.NParam,
topn.NParamIndex)),
FieldName: orderFieldName,
FieldValueSort: sort,
}, nil
}
-func (t *Transformer) grammarValueToString(val *GrammarValue) string {
+func (r *transformRun) grammarValueToString(val *GrammarValue) string {
+ val = r.resolveValue(val)
if val.String != nil {
return *val.String
}
@@ -1586,7 +1691,8 @@ func (t *Transformer) grammarValueToString(val
*GrammarValue) string {
return ""
}
-func (t *Transformer) grammarValueToInt64(val *GrammarValue) (int64, error) {
+func (r *transformRun) grammarValueToInt64(val *GrammarValue) (int64, error) {
+ val = r.resolveValue(val)
if val.Integer != nil {
return *val.Integer, nil
}
@@ -1602,13 +1708,13 @@ func (t *Transformer) grammarValueToInt64(val
*GrammarValue) (int64, error) {
// extractIDsAndCriteria separates ID conditions from other conditions in
property queries.
// ID conditions are extracted into a string array, while other conditions are
converted to criteria.
-func (t *Transformer) extractIDsAndCriteria(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
+func (r *transformRun) extractIDsAndCriteria(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
if expr == nil {
return nil, nil, nil
}
// Process the expression tree and collect IDs and criteria
- ids, criteria, err := t.extractIDsFromOrExpr(expr, allTags)
+ ids, criteria, err := r.extractIDsFromOrExpr(expr, allTags)
if err != nil {
return nil, nil, err
}
@@ -1617,8 +1723,8 @@ func (t *Transformer) extractIDsAndCriteria(expr
*GrammarOrExpr, allTags map[str
}
// extractIDsFromOrExpr processes OR expressions for ID extraction.
-func (t *Transformer) extractIDsFromOrExpr(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
- leftIDs, leftCriteria, err := t.extractIDsFromAndExpr(expr.Left,
allTags)
+func (r *transformRun) extractIDsFromOrExpr(expr *GrammarOrExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
+ leftIDs, leftCriteria, err := r.extractIDsFromAndExpr(expr.Left,
allTags)
if err != nil {
return nil, nil, err
}
@@ -1633,7 +1739,7 @@ func (t *Transformer) extractIDsFromOrExpr(expr
*GrammarOrExpr, allTags map[stri
currentCriteria := leftCriteria
for _, orRight := range expr.Right {
- rightIDs, rightCriteria, err :=
t.extractIDsFromAndExpr(orRight.Right, allTags)
+ rightIDs, rightCriteria, err :=
r.extractIDsFromAndExpr(orRight.Right, allTags)
if err != nil {
return nil, nil, err
}
@@ -1658,8 +1764,8 @@ func (t *Transformer) extractIDsFromOrExpr(expr
*GrammarOrExpr, allTags map[stri
}
// extractIDsFromAndExpr processes AND expressions for ID extraction.
-func (t *Transformer) extractIDsFromAndExpr(expr *GrammarAndExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
- leftIDs, leftCriteria, err := t.extractIDsFromPredicate(expr.Left,
allTags)
+func (r *transformRun) extractIDsFromAndExpr(expr *GrammarAndExpr, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
+ leftIDs, leftCriteria, err := r.extractIDsFromPredicate(expr.Left,
allTags)
if err != nil {
return nil, nil, err
}
@@ -1674,7 +1780,7 @@ func (t *Transformer) extractIDsFromAndExpr(expr
*GrammarAndExpr, allTags map[st
currentCriteria := leftCriteria
for _, andRight := range expr.Right {
- rightIDs, rightCriteria, err :=
t.extractIDsFromPredicate(andRight.Right, allTags)
+ rightIDs, rightCriteria, err :=
r.extractIDsFromPredicate(andRight.Right, allTags)
if err != nil {
return nil, nil, err
}
@@ -1699,9 +1805,9 @@ func (t *Transformer) extractIDsFromAndExpr(expr
*GrammarAndExpr, allTags map[st
}
// extractIDsFromPredicate processes predicates for ID extraction.
-func (t *Transformer) extractIDsFromPredicate(pred *GrammarPredicate, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
+func (r *transformRun) extractIDsFromPredicate(pred *GrammarPredicate, allTags
map[string]*tagSpecWithFamily) ([]string, *modelv1.Criteria, error) {
if pred.Paren != nil {
- return t.extractIDsFromOrExpr(pred.Paren, allTags)
+ return r.extractIDsFromOrExpr(pred.Paren, allTags)
}
if pred.Binary != nil {
@@ -1714,17 +1820,17 @@ func (t *Transformer) extractIDsFromPredicate(pred
*GrammarPredicate, allTags ma
if strings.EqualFold(identifierName, "ID") {
if pred.Binary.Tail.Compare != nil &&
pred.Binary.Tail.Compare.Operator == "=" {
// ID = 'value'
- if pred.Binary.Tail.Compare.Value.Null {
+ if
r.resolveValue(pred.Binary.Tail.Compare.Value).Null {
return nil, nil, fmt.Errorf("ID cannot
be NULL")
}
- idValue :=
t.grammarValueToString(pred.Binary.Tail.Compare.Value)
+ idValue :=
r.grammarValueToString(pred.Binary.Tail.Compare.Value)
return []string{idValue}, nil, nil
}
return nil, nil, fmt.Errorf("unsupported operator for
ID condition (only = is supported)")
}
// Not an ID condition, convert to criteria
- criteria, err := t.convertBinaryPredicate(pred.Binary, allTags,
nil)
+ criteria, err := r.convertBinaryPredicate(pred.Binary, allTags,
nil)
return nil, criteria, err
}
@@ -1740,27 +1846,28 @@ func (t *Transformer) extractIDsFromPredicate(pred
*GrammarPredicate, allTags ma
return nil, nil, fmt.Errorf("NOT IN is not
supported for ID condition")
}
// ID IN ('value1', 'value2', ...)
- if len(pred.In.Values) == 0 {
+ values := r.resolveValues(pred.In.Values)
+ if len(values) == 0 {
return nil, nil, fmt.Errorf("ID IN requires at
least one value")
}
- ids := make([]string, 0, len(pred.In.Values))
- for _, val := range pred.In.Values {
+ ids := make([]string, 0, len(values))
+ for _, val := range values {
if val.Null {
return nil, nil, fmt.Errorf("ID cannot
be NULL in IN clause")
}
- ids = append(ids, t.grammarValueToString(val))
+ ids = append(ids, r.grammarValueToString(val))
}
return ids, nil, nil
}
// not an ID condition, convert to criteria
- criteria, err := t.convertInPredicate(pred.In, allTags, nil)
+ criteria, err := r.convertInPredicate(pred.In, allTags, nil)
return nil, criteria, err
}
if pred.Having != nil {
// HAVING is not an ID condition
- criteria, err := t.convertHavingPredicate(pred.Having, allTags,
nil)
+ criteria, err := r.convertHavingPredicate(pred.Having, allTags,
nil)
return nil, criteria, err
}
