mirror of
https://github.com/SigNoz/signoz.git
synced 2026-11-06 00:10:41 +00:00
Compare commits
2 Commits
nv/exec-st
...
feat/formu
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d8887db458 | ||
|
|
dd1cf90716 |
@@ -10088,8 +10088,10 @@ components:
|
||||
having:
|
||||
allOf:
|
||||
- $ref: '#/components/schemas/Querybuildertypesv5Having'
|
||||
description: Accepted but currently ignored; the computed values are returned
|
||||
unfiltered.
|
||||
description: A filter on the formula values, referencing the formula by
|
||||
its name or as __result, for example F1 > 0. A time series keeps only
|
||||
the points that match and drops a series with none left; a scalar result
|
||||
keeps only the rows that match.
|
||||
legend:
|
||||
description: The display name template for each series; {{field}} placeholders
|
||||
are replaced by label values.
|
||||
|
||||
@@ -7046,7 +7046,7 @@ export interface Querybuildertypesv5QueryBuilderFormulaDTO {
|
||||
*/
|
||||
functions?: Querybuildertypesv5FunctionDTO[] | null;
|
||||
/**
|
||||
* @description Accepted but currently ignored; the computed values are returned unfiltered.
|
||||
* @description A filter on the formula values, referencing the formula by its name or as __result, for example F1 > 0. A time series keeps only the points that match and drops a series with none left; a scalar result keeps only the rows that match.
|
||||
*/
|
||||
having?: Querybuildertypesv5HavingDTO;
|
||||
/**
|
||||
|
||||
@@ -26,9 +26,8 @@ const bucketCacheCellSteps = 60
|
||||
type BucketCache interface {
|
||||
// cached portion + list of gaps to fetch
|
||||
GetMissRanges(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step) (cached *qbtypes.Result, missing []*qbtypes.TimeRange)
|
||||
// store fresh buckets for future hits; only the ranges in the map were
|
||||
// queried for fresh, the rest of it came from the cache
|
||||
Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats)
|
||||
// store fresh buckets for future hits
|
||||
Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result)
|
||||
}
|
||||
|
||||
// bucketCache implements the BucketCache interface.
|
||||
@@ -117,7 +116,7 @@ func (bc *bucketCache) GetMissRanges(
|
||||
}
|
||||
|
||||
// Merge buckets into a single result
|
||||
mergedResult := bc.mergeBuckets(ctx, relevantBuckets, cachedWarnings, startMs, endMs)
|
||||
mergedResult := bc.mergeBuckets(ctx, relevantBuckets, cachedWarnings)
|
||||
|
||||
// Filter the merged result to only include values within the requested time range
|
||||
mergedResult = bc.filterResultToTimeRange(mergedResult, q, startMs, endMs, stepMs)
|
||||
@@ -126,7 +125,7 @@ func (bc *bucketCache) GetMissRanges(
|
||||
}
|
||||
|
||||
// Put stores fresh query results in the cache.
|
||||
func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) {
|
||||
func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Query, step qbtypes.Step, fresh *qbtypes.Result) {
|
||||
// Get query window
|
||||
startMs, endMs := q.Window()
|
||||
|
||||
@@ -189,7 +188,7 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
|
||||
|
||||
// Convert trimmed result to buckets, one per grid cell it spans
|
||||
cellWidthMs := bucketCacheCellSteps * stepMs
|
||||
freshBuckets := bc.resultToBuckets(ctx, trimmedResult, cachableStartMs, cachableEndMs, cellWidthMs, execStatsForQueriedTimeRanges)
|
||||
freshBuckets := bc.resultToBuckets(ctx, trimmedResult, cachableStartMs, cachableEndMs, cellWidthMs)
|
||||
|
||||
cellKeys := make([]string, len(freshBuckets))
|
||||
cellEntries := make([]cachetypes.Cacheable, len(freshBuckets))
|
||||
@@ -211,7 +210,7 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
|
||||
cellEntry = cellEntries[i].(*qbtypes.CachedData)
|
||||
}
|
||||
|
||||
cellBuckets, changed := bc.mergeFreshBucketIntoCell(ctx, cellEntry.Buckets, freshBucket, execStatsForQueriedTimeRanges)
|
||||
cellBuckets, changed := bc.mergeFreshBucketIntoCell(ctx, cellEntry.Buckets, freshBucket)
|
||||
if !changed {
|
||||
continue
|
||||
}
|
||||
@@ -464,7 +463,7 @@ func (bc *bucketCache) filterRelevantBuckets(buckets []*qbtypes.CachedBucket, st
|
||||
}
|
||||
|
||||
// mergeBuckets combines multiple cached buckets into a single result.
|
||||
func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.CachedBucket, warnings []string, startMs, endMs uint64) *qbtypes.Result {
|
||||
func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.CachedBucket, warnings []string) *qbtypes.Result {
|
||||
if len(buckets) == 0 {
|
||||
return &qbtypes.Result{}
|
||||
}
|
||||
@@ -472,13 +471,12 @@ func (bc *bucketCache) mergeBuckets(ctx context.Context, buckets []*qbtypes.Cach
|
||||
// All buckets should have the same type
|
||||
resultType := buckets[0].Type
|
||||
|
||||
// A bucket can reach past the window; only the part inside it is served
|
||||
// Aggregate stats
|
||||
var totalStats qbtypes.ExecStats
|
||||
for _, bucket := range buckets {
|
||||
servedStats := bucket.ProrateStatsToWindow(startMs, endMs)
|
||||
totalStats.RowsScanned += servedStats.RowsScanned
|
||||
totalStats.BytesScanned += servedStats.BytesScanned
|
||||
totalStats.DurationMS += servedStats.DurationMS
|
||||
totalStats.RowsScanned += bucket.Stats.RowsScanned
|
||||
totalStats.BytesScanned += bucket.Stats.BytesScanned
|
||||
totalStats.DurationMS += bucket.Stats.DurationMS
|
||||
}
|
||||
|
||||
// Merge values based on type
|
||||
@@ -676,7 +674,7 @@ func (bc *bucketCache) isEmptyResult(result *qbtypes.Result) (isEmpty bool, isFi
|
||||
}
|
||||
|
||||
// resultToBuckets converts a query result into time-based buckets.
|
||||
func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Result, startMs, endMs, cellWidthMs uint64, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) []*qbtypes.CachedBucket {
|
||||
func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Result, startMs, endMs, cellWidthMs uint64) []*qbtypes.CachedBucket {
|
||||
// Check if result is empty
|
||||
isEmpty, isFiltered := bc.isEmptyResult(result)
|
||||
|
||||
@@ -688,7 +686,7 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
|
||||
}
|
||||
|
||||
// Empty filtered results still become buckets, so the range is not re-queried
|
||||
perCellBuckets, err := result.SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs, execStatsForQueriedTimeRanges)
|
||||
perCellBuckets, err := result.SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs)
|
||||
if err != nil {
|
||||
bc.logger.ErrorContext(ctx, "failed to convert result to buckets", errors.Attr(err))
|
||||
return nil
|
||||
@@ -700,7 +698,7 @@ func (bc *bucketCache) resultToBuckets(ctx context.Context, result *qbtypes.Resu
|
||||
// and reports whether the cell changed: a bucket already covering the fresh
|
||||
// one leaves it as is, buckets inside the fresh one are replaced, buckets
|
||||
// overlapping or edge to edge with it are unioned, and the rest are kept apart.
|
||||
func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets []*qbtypes.CachedBucket, freshBucket *qbtypes.CachedBucket, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) ([]*qbtypes.CachedBucket, bool) {
|
||||
func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets []*qbtypes.CachedBucket, freshBucket *qbtypes.CachedBucket) ([]*qbtypes.CachedBucket, bool) {
|
||||
keptBuckets := make([]*qbtypes.CachedBucket, 0, len(cellBuckets)+1)
|
||||
touchingBuckets := []*qbtypes.CachedBucket{}
|
||||
for _, cellBucket := range cellBuckets {
|
||||
@@ -713,7 +711,7 @@ func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets
|
||||
keptBuckets = append(keptBuckets, cellBucket)
|
||||
}
|
||||
}
|
||||
keptBuckets = append(keptBuckets, bc.unionBuckets(ctx, freshBucket, touchingBuckets, execStatsForQueriedTimeRanges))
|
||||
keptBuckets = append(keptBuckets, bc.unionBuckets(ctx, freshBucket, touchingBuckets))
|
||||
|
||||
slices.SortFunc(keptBuckets, func(a, b *qbtypes.CachedBucket) int {
|
||||
return cmp.Compare(a.StartMs, b.StartMs)
|
||||
@@ -724,16 +722,7 @@ func (bc *bucketCache) mergeFreshBucketIntoCell(ctx context.Context, cellBuckets
|
||||
// unionBuckets widens the fresh bucket by the buckets it touches. Buckets
|
||||
// inside it are replaced outright, so only a genuine partial overlap decodes
|
||||
// anything.
|
||||
func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.CachedBucket, touchingBuckets []*qbtypes.CachedBucket, execStatsForQueriedTimeRanges map[qbtypes.TimeRange]qbtypes.ExecStats) *qbtypes.CachedBucket {
|
||||
// Data queried again takes the new query's cost, which the fresh bucket
|
||||
// carries; a touching bucket keeps its cost only for what was not queried again
|
||||
for _, touchingBucket := range touchingBuckets {
|
||||
keptStats := touchingBucket.ProrateStatsToUnqueried(execStatsForQueriedTimeRanges)
|
||||
freshBucket.Stats.RowsScanned += keptStats.RowsScanned
|
||||
freshBucket.Stats.BytesScanned += keptStats.BytesScanned
|
||||
freshBucket.Stats.DurationMS += keptStats.DurationMS
|
||||
}
|
||||
|
||||
func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.CachedBucket, touchingBuckets []*qbtypes.CachedBucket) *qbtypes.CachedBucket {
|
||||
bucketsToMerge := []*qbtypes.CachedBucket{freshBucket}
|
||||
for _, touchingBucket := range touchingBuckets {
|
||||
if !freshBucket.Contains(touchingBucket) {
|
||||
@@ -754,11 +743,14 @@ func (bc *bucketCache) unionBuckets(ctx context.Context, freshBucket *qbtypes.Ca
|
||||
EndMs: freshBucket.EndMs,
|
||||
Type: freshBucket.Type,
|
||||
Value: value,
|
||||
Stats: freshBucket.Stats,
|
||||
Stats: qbtypes.ExecStats{StepIntervals: freshBucket.Stats.StepIntervals},
|
||||
}
|
||||
for _, mergedBucket := range bucketsToMerge {
|
||||
unionBucket.StartMs = min(unionBucket.StartMs, mergedBucket.StartMs)
|
||||
unionBucket.EndMs = max(unionBucket.EndMs, mergedBucket.EndMs)
|
||||
unionBucket.Stats.RowsScanned += mergedBucket.Stats.RowsScanned
|
||||
unionBucket.Stats.BytesScanned += mergedBucket.Stats.BytesScanned
|
||||
unionBucket.Stats.DurationMS += mergedBucket.Stats.DurationMS
|
||||
}
|
||||
return unionBucket
|
||||
}
|
||||
|
||||
@@ -30,7 +30,7 @@ func BenchmarkBucketCache_GetMissRanges(b *testing.B) {
|
||||
endMs: uint64((i + 1) * 10000),
|
||||
}
|
||||
result := createBenchmarkResult(query.startMs, query.endMs, 1000)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
}
|
||||
|
||||
// Create test queries with varying cache hit patterns
|
||||
@@ -121,7 +121,7 @@ func BenchmarkBucketCache_Put(b *testing.B) {
|
||||
|
||||
for i := 0; i < b.N; i++ {
|
||||
for j := 0; j < tc.numQueries; j++ {
|
||||
bc.Put(ctx, orgID, queries[j], qbtypes.Step{Duration: 1000 * time.Millisecond}, results[j], map[qbtypes.TimeRange]qbtypes.ExecStats{{From: queries[j].startMs, To: queries[j].endMs}: results[j].Stats})
|
||||
bc.Put(ctx, orgID, queries[j], qbtypes.Step{Duration: 1000 * time.Millisecond}, results[j])
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -259,7 +259,7 @@ func BenchmarkBucketCache_ConcurrentOperations(b *testing.B) {
|
||||
endMs: uint64((i + 1) * 10000),
|
||||
}
|
||||
result := createBenchmarkResult(query.startMs, query.endMs, 1000)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
}
|
||||
|
||||
b.ResetTimer()
|
||||
@@ -284,7 +284,7 @@ func BenchmarkBucketCache_ConcurrentOperations(b *testing.B) {
|
||||
endMs: uint64((i + 1) * 10000),
|
||||
}
|
||||
result := createBenchmarkResult(query.startMs, query.endMs, 1000)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
case 2: // Partial read
|
||||
query := &mockQuery{
|
||||
fingerprint: fmt.Sprintf("concurrent-query-%d", i%100),
|
||||
|
||||
@@ -59,7 +59,7 @@ func TestBucketCacheStepAlignment(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put result in cache
|
||||
bc.Put(ctx, orgID, query, step, result, nil)
|
||||
bc.Put(ctx, orgID, query, step, result)
|
||||
|
||||
// Get cached data
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
|
||||
@@ -109,7 +109,7 @@ func TestBucketCacheNoStepInterval(t *testing.T) {
|
||||
}
|
||||
|
||||
// Should cache the entire range when step is 0
|
||||
bc.Put(ctx, orgID, query, step, result, nil)
|
||||
bc.Put(ctx, orgID, query, step, result)
|
||||
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
|
||||
assert.NotNil(t, cached)
|
||||
|
||||
@@ -156,7 +156,7 @@ func TestBucketCache_Put_And_Get(t *testing.T) {
|
||||
}
|
||||
|
||||
// Store in cache
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Wait a bit for cache to be written
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -189,7 +189,7 @@ func TestBucketCache_PartialHit(t *testing.T) {
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 1000, 3000, 1000),
|
||||
}
|
||||
bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1)
|
||||
|
||||
// Wait for cache write
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -225,7 +225,7 @@ func TestBucketCache_MultipleBuckets(t *testing.T) {
|
||||
bc.Put(context.Background(), valuer.UUID{}, query1, qbtypes.Step{Duration: 100 * time.Millisecond}, &qbtypes.Result{
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 1000, 2000, 100),
|
||||
}, nil)
|
||||
})
|
||||
|
||||
query2 := &mockQuery{
|
||||
fingerprint: "test-query",
|
||||
@@ -235,7 +235,7 @@ func TestBucketCache_MultipleBuckets(t *testing.T) {
|
||||
bc.Put(context.Background(), valuer.UUID{}, query2, qbtypes.Step{Duration: 100 * time.Millisecond}, &qbtypes.Result{
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 3000, 4000, 100),
|
||||
}, nil)
|
||||
})
|
||||
|
||||
// Wait for cache writes
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -280,7 +280,7 @@ func TestBucketCache_FluxInterval(t *testing.T) {
|
||||
}
|
||||
|
||||
// This should not be cached due to flux interval
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Wait a bit
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -358,7 +358,7 @@ func TestBucketCache_MergeTimeSeriesResults(t *testing.T) {
|
||||
{Series: []*qbtypes.TimeSeries{series1}},
|
||||
},
|
||||
},
|
||||
}, nil)
|
||||
})
|
||||
|
||||
// Cache second part
|
||||
query2 := &mockQuery{
|
||||
@@ -374,7 +374,7 @@ func TestBucketCache_MergeTimeSeriesResults(t *testing.T) {
|
||||
{Series: []*qbtypes.TimeSeries{series2}},
|
||||
},
|
||||
},
|
||||
}, nil)
|
||||
})
|
||||
|
||||
// Wait for cache writes
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -441,7 +441,7 @@ func TestBucketCache_RawData(t *testing.T) {
|
||||
Value: rawData,
|
||||
}
|
||||
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
cached, missing := bc.GetMissRanges(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
|
||||
@@ -481,7 +481,7 @@ func TestBucketCache_ScalarData(t *testing.T) {
|
||||
Value: scalarData,
|
||||
}
|
||||
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
cached, missing := bc.GetMissRanges(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
|
||||
@@ -509,7 +509,7 @@ func TestBucketCache_EmptyFingerprint(t *testing.T) {
|
||||
Value: createTestTimeSeries("A", 1000, 5000, 1000),
|
||||
}
|
||||
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Should still be able to retrieve
|
||||
@@ -564,7 +564,7 @@ func TestBucketCache_ConcurrentAccess(t *testing.T) {
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries(fmt.Sprintf("Q%d", id), query.startMs, query.endMs, 100),
|
||||
}
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 100 * time.Microsecond}, result, nil)
|
||||
bc.Put(context.Background(), valuer.UUID{}, query, qbtypes.Step{Duration: 100 * time.Microsecond}, result)
|
||||
done <- true
|
||||
}(i)
|
||||
}
|
||||
@@ -624,7 +624,7 @@ func TestBucketCache_GetMissRanges_FluxInterval(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, cachedResult, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, cachedResult)
|
||||
|
||||
// Get miss ranges
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
|
||||
@@ -686,7 +686,7 @@ func TestBucketCache_Put_FluxIntervalTrimming(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Retrieve cached data
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
|
||||
@@ -756,7 +756,7 @@ func TestBucketCache_Put_EntireRangeInFluxInterval(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result - should not cache anything
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Try to get cached data - should have no cached data
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond})
|
||||
@@ -862,7 +862,7 @@ func TestBucketCache_EmptyDataHandling(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, tt.result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, tt.result)
|
||||
|
||||
// Wait a bit for cache to be written
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -927,7 +927,7 @@ func TestBucketCache_PartialValues(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: query.startMs, To: query.endMs}: result.Stats})
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Wait for cache to be written
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -997,7 +997,7 @@ func TestBucketCache_AllPartialValues(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
|
||||
// Wait for cache to be written
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
@@ -1058,7 +1058,7 @@ func TestBucketCache_FilteredCachedResults(t *testing.T) {
|
||||
}
|
||||
|
||||
// Cache the wide range
|
||||
bc.Put(ctx, orgID, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1, nil)
|
||||
bc.Put(ctx, orgID, query1, qbtypes.Step{Duration: 1000 * time.Millisecond}, result1)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Now query for a smaller range (2000-3500ms)
|
||||
@@ -1125,7 +1125,7 @@ func TestBucketCache_PromQLKeepsTheValueAtTheWindowEnd(t *testing.T) {
|
||||
}},
|
||||
}},
|
||||
},
|
||||
}, nil)
|
||||
})
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
|
||||
@@ -1291,7 +1291,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Get cached data
|
||||
@@ -1345,7 +1345,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Get cached data
|
||||
@@ -1397,7 +1397,7 @@ func TestBucketCache_PartialValueDetection(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Get cached data
|
||||
@@ -1454,7 +1454,7 @@ func TestBucketCache_NoCache(t *testing.T) {
|
||||
}
|
||||
|
||||
// Put the result in cache
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result, nil)
|
||||
bc.Put(ctx, orgID, query, qbtypes.Step{Duration: 1000 * time.Millisecond}, result)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Verify data is cached
|
||||
@@ -1475,12 +1475,11 @@ func TestBucketCachePutSlicesAtCellBoundaries(t *testing.T) {
|
||||
|
||||
// a 1s step gives 60s cells, so 30s to 150s spans three of them
|
||||
query := &mockQuery{fingerprint: "sliced-query", startMs: 30000, endMs: 150000}
|
||||
result := &qbtypes.Result{
|
||||
bc.Put(ctx, orgID, query, step, &qbtypes.Result{
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 30000, 150000, 1000),
|
||||
Stats: qbtypes.ExecStats{RowsScanned: 120, BytesScanned: 1200, DurationMS: 12},
|
||||
}
|
||||
bc.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: 30000, To: 150000}: result.Stats})
|
||||
})
|
||||
|
||||
type bucketShape struct {
|
||||
cell, startMs, endMs, rowsScanned uint64
|
||||
@@ -1540,7 +1539,7 @@ func TestBucketCachePutAbsorbsWritesWithinACell(t *testing.T) {
|
||||
name: "OverlappingWrites_BecomeOne",
|
||||
firstWindow: qbtypes.TimeRange{From: 1000, To: 4000},
|
||||
secondWindow: qbtypes.TimeRange{From: 3000, To: 6000},
|
||||
expectedBuckets: []bucketShape{{1000, 6000, 5}},
|
||||
expectedBuckets: []bucketShape{{1000, 6000, 6}},
|
||||
expectedMissingRanges: []*qbtypes.TimeRange{},
|
||||
expectedValueCount: 5,
|
||||
},
|
||||
@@ -1571,13 +1570,12 @@ func TestBucketCachePutAbsorbsWritesWithinACell(t *testing.T) {
|
||||
|
||||
for _, window := range []qbtypes.TimeRange{testCase.firstWindow, testCase.secondWindow} {
|
||||
query := &mockQuery{fingerprint: "absorb-query", startMs: window.From, endMs: window.To}
|
||||
result := &qbtypes.Result{
|
||||
bc.Put(ctx, orgID, query, step, &qbtypes.Result{
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", window.From, window.To, 1000),
|
||||
// one row per second, so a bucket's stats read as the seconds of data it holds
|
||||
// one row per second, so a bucket's stats read as the seconds it was scanned for
|
||||
Stats: qbtypes.ExecStats{RowsScanned: (window.To - window.From) / 1000},
|
||||
}
|
||||
bc.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{window: result.Stats})
|
||||
})
|
||||
}
|
||||
|
||||
query := &mockQuery{
|
||||
@@ -1611,7 +1609,7 @@ func TestBucketCachePutLeavesAnUnchangedCellAlone(t *testing.T) {
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 1000, 10000, 1000),
|
||||
Warnings: []string{"from the wide write"},
|
||||
}, nil)
|
||||
})
|
||||
|
||||
// a write already covered by the cell's bucket must not rewrite the entry,
|
||||
// which shows as its warnings never arriving
|
||||
@@ -1620,7 +1618,7 @@ func TestBucketCachePutLeavesAnUnchangedCellAlone(t *testing.T) {
|
||||
Type: qbtypes.RequestTypeTimeSeries,
|
||||
Value: createTestTimeSeries("A", 3000, 5000, 1000),
|
||||
Warnings: []string{"from the inside write"},
|
||||
}, nil)
|
||||
})
|
||||
|
||||
var cellEntry qbtypes.CachedData
|
||||
require.NoError(t, bc.cache.Get(ctx, orgID, bc.generateCacheKey(wide, 0), &cellEntry))
|
||||
|
||||
@@ -415,6 +415,12 @@ func (q *querier) processTimeSeriesFormula(
|
||||
return nil
|
||||
}
|
||||
|
||||
formulaSeries, err = applyFormulaHaving(formula, formulaSeries)
|
||||
if err != nil {
|
||||
q.logger.ErrorContext(ctx, "failed to apply formula having", errors.Attr(err), slog.String("formula", formula.Name))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Create result for formula
|
||||
formulaResult := &qbtypes.TimeSeriesData{
|
||||
QueryName: formula.Name,
|
||||
@@ -650,6 +656,12 @@ func (q *querier) processScalarFormula(
|
||||
return nil
|
||||
}
|
||||
|
||||
formulaSeries, err = applyFormulaHaving(formula, formulaSeries)
|
||||
if err != nil {
|
||||
q.logger.ErrorContext(ctx, "failed to apply formula having", errors.Attr(err), slog.String("formula", formula.Name))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Apply ordering (and limit) before converting to scalar format.
|
||||
formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit)
|
||||
|
||||
@@ -700,6 +712,35 @@ func (q *querier) processScalarFormula(
|
||||
}
|
||||
}
|
||||
|
||||
// applyFormulaHaving keeps the values the formula having accepts, as SQL HAVING
|
||||
// keeps rows for a builder query, so a series with no values left is removed. It
|
||||
// runs before functions, order, and limit, as HAVING runs before them in SQL.
|
||||
func applyFormulaHaving(formula qbtypes.QueryBuilderFormula, series []*qbtypes.TimeSeries) ([]*qbtypes.TimeSeries, error) {
|
||||
if formula.Having == nil || strings.TrimSpace(formula.Having.Expression) == "" {
|
||||
return series, nil
|
||||
}
|
||||
|
||||
keep, err := querybuilder.CompileFormulaHaving(formula.Name, formula.Having.Expression)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
kept := series[:0]
|
||||
for _, s := range series {
|
||||
values := s.Values[:0]
|
||||
for _, value := range s.Values {
|
||||
if keep(value.Value) {
|
||||
values = append(values, value)
|
||||
}
|
||||
}
|
||||
if len(values) > 0 {
|
||||
s.Values = values
|
||||
kept = append(kept, s)
|
||||
}
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
|
||||
// filterDisabledQueries removes results for disabled queries.
|
||||
func (q *querier) filterDisabledQueries(results map[string]*qbtypes.Result, req *qbtypes.QueryRangeRequest) map[string]*qbtypes.Result {
|
||||
filtered := make(map[string]*qbtypes.Result)
|
||||
|
||||
@@ -212,6 +212,14 @@ func (q *querier) buildQueries(
|
||||
}
|
||||
|
||||
switch query.Type {
|
||||
case qbtypes.QueryTypeFormula:
|
||||
// the querier evaluates a formula after its queries run, so a bad
|
||||
// having must fail here, before they run
|
||||
if formula, ok := query.Spec.(qbtypes.QueryBuilderFormula); ok && formula.Having != nil && strings.TrimSpace(formula.Having.Expression) != "" {
|
||||
if _, err := querybuilder.CompileFormulaHaving(formula.Name, formula.Having.Expression); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
case qbtypes.QueryTypePromQL:
|
||||
promQuery, ok := query.Spec.(qbtypes.PromQuery)
|
||||
if !ok {
|
||||
@@ -823,7 +831,7 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
|
||||
return nil, err
|
||||
}
|
||||
// Store in cache for future use
|
||||
q.bucketCache.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: startMs, To: endMs}: result.Stats})
|
||||
q.bucketCache.Put(ctx, orgID, query, step, result)
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
@@ -879,19 +887,16 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
startMs, endMs := query.Window()
|
||||
q.bucketCache.Put(ctx, orgID, query, step, result, map[qbtypes.TimeRange]qbtypes.ExecStats{{From: startMs, To: endMs}: result.Stats})
|
||||
q.bucketCache.Put(ctx, orgID, query, step, result)
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Calculate total stats and filter out nil results
|
||||
validResults := make([]*qbtypes.Result, 0, len(freshResults))
|
||||
execStatsForQueriedTimeRanges := make(map[qbtypes.TimeRange]qbtypes.ExecStats, len(freshResults))
|
||||
for i, result := range freshResults {
|
||||
for _, result := range freshResults {
|
||||
if result != nil {
|
||||
validResults = append(validResults, result)
|
||||
execStatsForQueriedTimeRanges[*missingRanges[i]] = result.Stats
|
||||
totalStats.RowsScanned += result.Stats.RowsScanned
|
||||
totalStats.BytesScanned += result.Stats.BytesScanned
|
||||
totalStats.DurationMS += result.Stats.DurationMS
|
||||
@@ -906,7 +911,7 @@ func (q *querier) executeWithCache(ctx context.Context, orgID valuer.UUID, query
|
||||
mergedResult.Stats.DurationMS += totalStats.DurationMS
|
||||
|
||||
// Store merged result in cache
|
||||
q.bucketCache.Put(ctx, orgID, query, step, mergedResult, execStatsForQueriedTimeRanges)
|
||||
q.bucketCache.Put(ctx, orgID, query, step, mergedResult)
|
||||
|
||||
return mergedResult, nil
|
||||
}
|
||||
|
||||
166
pkg/querybuilder/formula_having.go
Normal file
166
pkg/querybuilder/formula_having.go
Normal file
@@ -0,0 +1,166 @@
|
||||
package querybuilder
|
||||
|
||||
import (
|
||||
"math"
|
||||
"slices"
|
||||
"strconv"
|
||||
|
||||
grammar "github.com/SigNoz/signoz/pkg/parser/havingexpression/grammar"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
)
|
||||
|
||||
// CompileFormulaHaving turns the having expression of a formula into a filter on
|
||||
// its values. The querier computes a formula in Go after ClickHouse returns, so the
|
||||
// having cannot run as SQL; it gets the same validation and messages as a
|
||||
// builder-query having, with the formula name and __result as the only references.
|
||||
func CompileFormulaHaving(formulaName, expression string) (func(value float64) bool, error) {
|
||||
rewriter := &HavingExpressionRewriter{columnMap: map[string]string{
|
||||
formulaName: qbtypes.DefaultOrderByKey,
|
||||
qbtypes.DefaultOrderByKey: qbtypes.DefaultOrderByKey,
|
||||
}}
|
||||
_, tree, err := rewriter.parseAndValidate(expression)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return compileHavingOr(tree.Expression().OrExpression()), nil
|
||||
}
|
||||
|
||||
func compileHavingOr(ctx grammar.IOrExpressionContext) func(float64) bool {
|
||||
andExpressions := ctx.AllAndExpression()
|
||||
conditions := make([]func(float64) bool, len(andExpressions))
|
||||
for i, andExpression := range andExpressions {
|
||||
conditions[i] = compileHavingAnd(andExpression)
|
||||
}
|
||||
|
||||
return func(value float64) bool {
|
||||
for _, condition := range conditions {
|
||||
if condition(value) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func compileHavingAnd(ctx grammar.IAndExpressionContext) func(float64) bool {
|
||||
primaries := ctx.AllPrimary()
|
||||
conditions := make([]func(float64) bool, len(primaries))
|
||||
for i, primary := range primaries {
|
||||
conditions[i] = compileHavingPrimary(primary)
|
||||
}
|
||||
|
||||
return func(value float64) bool {
|
||||
for _, condition := range conditions {
|
||||
if !condition(value) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func compileHavingPrimary(ctx grammar.IPrimaryContext) func(float64) bool {
|
||||
var condition func(float64) bool
|
||||
if ctx.OrExpression() != nil {
|
||||
condition = compileHavingOr(ctx.OrExpression())
|
||||
} else {
|
||||
condition = compileHavingComparison(ctx.Comparison())
|
||||
}
|
||||
|
||||
if ctx.NOT() == nil {
|
||||
return condition
|
||||
}
|
||||
return func(value float64) bool { return !condition(value) }
|
||||
}
|
||||
|
||||
func compileHavingComparison(ctx grammar.IComparisonContext) func(float64) bool {
|
||||
left := compileHavingOperand(ctx.Operand(0))
|
||||
|
||||
if ctx.IN() != nil {
|
||||
signedNumbers := ctx.InList().AllSignedNumber()
|
||||
list := make([]float64, len(signedNumbers))
|
||||
for i, signedNumber := range signedNumbers {
|
||||
list[i] = parseHavingNumber(signedNumber.GetText())
|
||||
}
|
||||
notIn := ctx.NOT() != nil
|
||||
return func(value float64) bool { return slices.Contains(list, left(value)) != notIn }
|
||||
}
|
||||
|
||||
right := compileHavingOperand(ctx.Operand(1))
|
||||
compOp := ctx.CompOp()
|
||||
switch {
|
||||
case compOp.EQUALS() != nil:
|
||||
return func(value float64) bool { return left(value) == right(value) }
|
||||
case compOp.NOT_EQUALS() != nil, compOp.NEQ() != nil:
|
||||
return func(value float64) bool { return left(value) != right(value) }
|
||||
case compOp.LT() != nil:
|
||||
return func(value float64) bool { return left(value) < right(value) }
|
||||
case compOp.LE() != nil:
|
||||
return func(value float64) bool { return left(value) <= right(value) }
|
||||
case compOp.GT() != nil:
|
||||
return func(value float64) bool { return left(value) > right(value) }
|
||||
default:
|
||||
return func(value float64) bool { return left(value) >= right(value) }
|
||||
}
|
||||
}
|
||||
|
||||
func compileHavingOperand(ctx grammar.IOperandContext) func(float64) float64 {
|
||||
if ctx.Operand() == nil {
|
||||
return compileHavingTerm(ctx.Term())
|
||||
}
|
||||
|
||||
left := compileHavingOperand(ctx.Operand())
|
||||
right := compileHavingTerm(ctx.Term())
|
||||
if ctx.MINUS() != nil {
|
||||
return func(value float64) float64 { return left(value) - right(value) }
|
||||
}
|
||||
return func(value float64) float64 { return left(value) + right(value) }
|
||||
}
|
||||
|
||||
func compileHavingTerm(ctx grammar.ITermContext) func(float64) float64 {
|
||||
if ctx.Term() == nil {
|
||||
return compileHavingFactor(ctx.Factor())
|
||||
}
|
||||
|
||||
left := compileHavingTerm(ctx.Term())
|
||||
right := compileHavingFactor(ctx.Factor())
|
||||
switch {
|
||||
case ctx.SLASH() != nil:
|
||||
return func(value float64) float64 { return left(value) / right(value) }
|
||||
case ctx.PERCENT() != nil:
|
||||
return func(value float64) float64 { return math.Mod(left(value), right(value)) }
|
||||
default:
|
||||
return func(value float64) float64 { return left(value) * right(value) }
|
||||
}
|
||||
}
|
||||
|
||||
func compileHavingFactor(ctx grammar.IFactorContext) func(float64) float64 {
|
||||
if ctx.Factor() != nil {
|
||||
inner := compileHavingFactor(ctx.Factor())
|
||||
if ctx.MINUS() != nil {
|
||||
return func(value float64) float64 { return -inner(value) }
|
||||
}
|
||||
return inner
|
||||
}
|
||||
|
||||
if ctx.Operand() != nil {
|
||||
return compileHavingOperand(ctx.Operand())
|
||||
}
|
||||
|
||||
if number := ctx.Atom().NUMBER(); number != nil {
|
||||
constant := parseHavingNumber(number.GetText())
|
||||
return func(float64) float64 { return constant }
|
||||
}
|
||||
|
||||
// validation rejects function calls, strings, and unknown names, so the atom
|
||||
// names the formula value
|
||||
return func(value float64) float64 { return value }
|
||||
}
|
||||
|
||||
// ParseFloat reads every form the lexer admits as a number; a literal outside the
|
||||
// float64 range becomes the nearest value ParseFloat returns.
|
||||
func parseHavingNumber(text string) float64 {
|
||||
number, _ := strconv.ParseFloat(text, 64)
|
||||
return number
|
||||
}
|
||||
85
pkg/querybuilder/formula_having_test.go
Normal file
85
pkg/querybuilder/formula_having_test.go
Normal file
@@ -0,0 +1,85 @@
|
||||
package querybuilder
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestCompileFormulaHaving_Matches(t *testing.T) {
|
||||
values := []float64{-2, 0, 1, 2.5, 3, 10}
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
expression string
|
||||
want []float64
|
||||
}{
|
||||
{name: "Name_GreaterThan", expression: "F1 > 2", want: []float64{2.5, 3, 10}},
|
||||
{name: "Result_LessOrEqual", expression: "__result <= 1", want: []float64{-2, 0, 1}},
|
||||
{name: "DoubleEquals", expression: "F1 == 3", want: []float64{3}},
|
||||
{name: "AngleNotEquals", expression: "F1 <> 0", want: []float64{-2, 1, 2.5, 3, 10}},
|
||||
{name: "In_Parentheses", expression: "F1 IN (0, 3, -2)", want: []float64{-2, 0, 3}},
|
||||
{name: "NotIn_Brackets", expression: "F1 NOT IN [0, 3, -2]", want: []float64{1, 2.5, 10}},
|
||||
{name: "AndBindsTighterThanOr", expression: "F1 > 0 AND F1 < 3 OR F1 = -2", want: []float64{-2, 1, 2.5}},
|
||||
{name: "ImplicitAnd", expression: "F1 > 0 F1 < 3", want: []float64{1, 2.5}},
|
||||
{name: "LowercaseKeywords", expression: "F1 > 2 and not F1 = 10", want: []float64{2.5, 3}},
|
||||
{name: "NotParenthesized", expression: "NOT (F1 >= 1 AND F1 <= 3)", want: []float64{-2, 0, 10}},
|
||||
{name: "Arithmetic", expression: "F1 * 2 - 1 > 4", want: []float64{3, 10}},
|
||||
{name: "Arithmetic_Parentheses", expression: "(F1 + 2) / 2 >= 2", want: []float64{2.5, 3, 10}},
|
||||
{name: "Modulo", expression: "F1 % 2 = 1", want: []float64{1, 3}},
|
||||
{name: "UnaryMinus", expression: "-F1 > 1", want: []float64{-2}},
|
||||
{name: "Number_Exponent", expression: "F1 >= 1e1", want: []float64{10}},
|
||||
{name: "Number_LeadingDot", expression: "F1 < .5", want: []float64{-2, 0}},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
keep, err := CompileFormulaHaving("F1", testCase.expression)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, keep)
|
||||
|
||||
got := []float64{}
|
||||
for _, value := range values {
|
||||
if keep(value) {
|
||||
got = append(got, value)
|
||||
}
|
||||
}
|
||||
assert.Equal(t, testCase.want, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompileFormulaHaving_Invalid(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
expression string
|
||||
wantErrMsg string
|
||||
wantAdditional []string
|
||||
}{
|
||||
{
|
||||
name: "QueryName_NotAReference",
|
||||
expression: "A > 0",
|
||||
wantErrMsg: "Invalid references in `Having` expression: [A]",
|
||||
wantAdditional: []string{"Valid references are: [F1, __result]"},
|
||||
},
|
||||
{
|
||||
name: "Aggregation_NotAReference",
|
||||
expression: "count() > 0",
|
||||
wantErrMsg: "Invalid references in `Having` expression: [count]",
|
||||
wantAdditional: []string{"Valid references are: [F1, __result]"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
keep, err := CompileFormulaHaving("F1", testCase.expression)
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, keep)
|
||||
assert.True(t, errors.Ast(err, errors.TypeInvalidInput))
|
||||
assert.ErrorContains(t, err, testCase.wantErrMsg)
|
||||
assert.Equal(t, testCase.wantAdditional, additionalMessages(err))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -237,7 +237,13 @@ func (v *havingExpressionRewriteVisitor) visitIdentifier(ctx grammar.IIdentifier
|
||||
return name
|
||||
}
|
||||
|
||||
// rewriteAndValidate is the single-pass implementation used by all RewriteFor* methods.
|
||||
func (r *HavingExpressionRewriter) rewriteAndValidate(expression string) (string, error) {
|
||||
rewritten, _, err := r.parseAndValidate(expression)
|
||||
return rewritten, err
|
||||
}
|
||||
|
||||
// parseAndValidate is the single-pass implementation used by all RewriteFor* methods.
|
||||
// It also returns the parse tree, which CompileFormulaHaving evaluates in Go.
|
||||
//
|
||||
// Validation layers:
|
||||
// 1. The visitor runs on the parse tree, rewriting and collecting invalid references.
|
||||
@@ -248,7 +254,7 @@ func (v *havingExpressionRewriteVisitor) visitIdentifier(ctx grammar.IIdentifier
|
||||
// 2. String literals in comparison-operand position → descriptive error
|
||||
// ("aggregator results are numeric").
|
||||
// 3. ANTLR syntax errors → error with messages referencing the original token names.
|
||||
func (r *HavingExpressionRewriter) rewriteAndValidate(expression string) (string, error) {
|
||||
func (r *HavingExpressionRewriter) parseAndValidate(expression string) (string, grammar.IQueryContext, error) {
|
||||
original := strings.TrimSpace(expression)
|
||||
|
||||
// Parse the expression once.
|
||||
@@ -281,7 +287,7 @@ func (r *HavingExpressionRewriter) rewriteAndValidate(expression string) (string
|
||||
// This is checked before invalid references so that "contains string literals" takes
|
||||
// priority when a bare string literal is also an unresolvable operand.
|
||||
if v.hasStringLiteral {
|
||||
return "", errors.NewInvalidInputf(
|
||||
return "", nil, errors.NewInvalidInputf(
|
||||
errors.CodeInvalidInput,
|
||||
"`Having` expression contains string literals",
|
||||
).WithAdditional("Aggregator results are numeric")
|
||||
@@ -317,7 +323,7 @@ func (r *HavingExpressionRewriter) rewriteAndValidate(expression string) (string
|
||||
).WithAdditional(
|
||||
"Valid references are: [" + strings.Join(validKeys, ", ") + "]",
|
||||
).WithSuggestions(suggestions...)
|
||||
return "", havingErr
|
||||
return "", nil, havingErr
|
||||
}
|
||||
|
||||
// Layer 3 – ANTLR syntax errors. We parse the original expression, so error messages
|
||||
@@ -343,12 +349,12 @@ func (r *HavingExpressionRewriter) rewriteAndValidate(expression string) (string
|
||||
return havingSuggestion(allSyntaxErrors[0], original)
|
||||
})
|
||||
|
||||
return "", havingErr.WithSuggestiveAdditional(msgs[0], suggestions...)
|
||||
return "", nil, havingErr.WithSuggestiveAdditional(msgs[0], suggestions...)
|
||||
}
|
||||
return "", havingErr.WithAdditional(msgs...)
|
||||
return "", nil, havingErr.WithAdditional(msgs...)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
return result, tree, nil
|
||||
}
|
||||
|
||||
// havingSuggestion returns a corrected expression string to show as a suggestion when
|
||||
|
||||
@@ -43,23 +43,6 @@ func (c *CachedBucket) Touches(other *CachedBucket) bool {
|
||||
return c.StartMs <= other.EndMs && other.StartMs <= c.EndMs
|
||||
}
|
||||
|
||||
// ProrateStatsToWindow returns the share of the stats for the part of the
|
||||
// bucket inside [startMs, endMs).
|
||||
func (c *CachedBucket) ProrateStatsToWindow(startMs, endMs uint64) ExecStats {
|
||||
return c.Stats.prorate(overlapMs(c.StartMs, c.EndMs, startMs, endMs), c.EndMs-c.StartMs)
|
||||
}
|
||||
|
||||
// ProrateStatsToUnqueried returns the share of the stats for the part of the
|
||||
// bucket no queried range covers.
|
||||
func (c *CachedBucket) ProrateStatsToUnqueried(execStatsForQueriedTimeRanges map[TimeRange]ExecStats) ExecStats {
|
||||
widthMs := c.EndMs - c.StartMs
|
||||
queriedMs := uint64(0)
|
||||
for queried := range execStatsForQueriedTimeRanges {
|
||||
queriedMs += overlapMs(c.StartMs, c.EndMs, queried.From, queried.To)
|
||||
}
|
||||
return c.Stats.prorate(widthMs-min(queriedMs, widthMs), widthMs)
|
||||
}
|
||||
|
||||
// CachedData represents the full cached data for a query.
|
||||
type CachedData struct {
|
||||
Buckets []*CachedBucket `json:"buckets"`
|
||||
@@ -105,7 +88,7 @@ func (c *CachedData) Cost() int64 {
|
||||
return size
|
||||
}
|
||||
|
||||
func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64, execStatsForQueriedTimeRanges map[TimeRange]ExecStats) ([]*CachedBucket, error) {
|
||||
func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64) ([]*CachedBucket, error) {
|
||||
tsData, ok := r.Value.(*TimeSeriesData)
|
||||
if !ok || tsData == nil {
|
||||
return nil, errors.NewInternalf(errors.CodeInternal, "result of type %s cannot be bucketed", r.Type.StringValue())
|
||||
@@ -128,39 +111,18 @@ func (r *Result) SplitIntoCacheCellBuckets(startMs, endMs, cellWidthMs uint64, e
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// A slice costs what was queried inside it; the rest of the result came
|
||||
// out of the cache and is costed there already
|
||||
sliceStats := ExecStats{StepIntervals: maps.Clone(r.Stats.StepIntervals)}
|
||||
for queried, stats := range execStatsForQueriedTimeRanges {
|
||||
queriedMs := overlapMs(from, to, queried.From, queried.To)
|
||||
if queriedMs == 0 {
|
||||
continue
|
||||
}
|
||||
share := stats.prorate(queriedMs, queried.To-queried.From)
|
||||
sliceStats.RowsScanned += share.RowsScanned
|
||||
sliceStats.BytesScanned += share.BytesScanned
|
||||
sliceStats.DurationMS += share.DurationMS
|
||||
}
|
||||
cellBuckets = append(cellBuckets, &CachedBucket{
|
||||
StartMs: from,
|
||||
EndMs: to,
|
||||
Type: r.Type,
|
||||
Value: value,
|
||||
Stats: sliceStats,
|
||||
Stats: r.Stats.prorate(to-from, endMs-startMs),
|
||||
})
|
||||
from = to
|
||||
}
|
||||
return cellBuckets, nil
|
||||
}
|
||||
|
||||
// overlapMs is the length of [aFromMs, aToMs) ∩ [bFromMs, bToMs).
|
||||
func overlapMs(aFromMs, aToMs, bFromMs, bToMs uint64) uint64 {
|
||||
if fromMs, toMs := max(aFromMs, bFromMs), min(aToMs, bToMs); fromMs < toMs {
|
||||
return toMs - fromMs
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// prorate returns the share of the stats that partMs out of wholeMs of the
|
||||
// scanned window accounts for.
|
||||
func (e ExecStats) prorate(partMs, wholeMs uint64) ExecStats {
|
||||
|
||||
@@ -26,7 +26,7 @@ type QueryBuilderFormula struct {
|
||||
|
||||
Limit int `json:"limit,omitempty" description:"The maximum number of series or rows to return. Unlimited when 0."`
|
||||
|
||||
Having *Having `json:"having,omitempty" description:"Accepted but currently ignored; the computed values are returned unfiltered."`
|
||||
Having *Having `json:"having,omitempty" description:"A filter on the formula values, referencing the formula by its name or as __result, for example F1 > 0. A time series keeps only the points that match and drops a series with none left; a scalar result keeps only the rows that match."`
|
||||
|
||||
Functions []Function `json:"functions,omitzero" description:"Post-processing functions applied in order to each result series."`
|
||||
|
||||
|
||||
3
tests/fixtures/querier.py
vendored
3
tests/fixtures/querier.py
vendored
@@ -348,6 +348,7 @@ def build_formula_query(
|
||||
disabled: bool = False,
|
||||
order: list[dict] | None = None,
|
||||
limit: int | None = None,
|
||||
having_expression: str | None = None,
|
||||
) -> dict:
|
||||
spec: dict[str, Any] = {
|
||||
"name": name,
|
||||
@@ -362,6 +363,8 @@ def build_formula_query(
|
||||
spec["order"] = order
|
||||
if limit is not None:
|
||||
spec["limit"] = limit
|
||||
if having_expression:
|
||||
spec["having"] = {"expression": having_expression}
|
||||
return {"type": "builder_formula", "spec": spec}
|
||||
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@ from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
@@ -12,6 +14,9 @@ from fixtures.querier import (
|
||||
build_order_by,
|
||||
build_scalar_query,
|
||||
find_named_result,
|
||||
get_all_series,
|
||||
get_error_message,
|
||||
index_series_by_label,
|
||||
make_query_request,
|
||||
)
|
||||
|
||||
@@ -191,3 +196,138 @@ def test_logs_formula_orderby_and_limit(
|
||||
assert len(f3_services) == 3, f"F3: expected 3 rows after limit, got {len(f3_services)}"
|
||||
assert f3_values == f4_values[:3], f"F3 values {f3_values} do not match F4[:3] values {f4_values[:3]}"
|
||||
assert set(f3_services) == set(f4_services[:3]), f"F3 services {f3_services} do not match F4[:3] services {f4_services[:3]}"
|
||||
|
||||
|
||||
# Logs per service at 5, 4, 3, 2, and 1 minutes before now. Totals: a 15, b 17, c 5, d 15.
|
||||
FORMULA_HAVING_COUNTS = {
|
||||
"service-a": [1, 4, 2, 5, 3],
|
||||
"service-b": [6, 1, 1, 2, 7],
|
||||
"service-c": [1, 1, 1, 1, 1],
|
||||
"service-d": [3, 3, 3, 3, 3],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"query_having,formula_having,expected_services",
|
||||
[
|
||||
pytest.param("count() > 4 AND count() < 16", "F1 > 4 AND F1 < 16", {"service-a", "service-b"}, id="and_range"),
|
||||
pytest.param("count() IN (1, 5, 17) OR count() > 15", "__result IN (1, 5, 17) OR __result > 15", {"service-a", "service-b", "service-c"}, id="in_or"),
|
||||
pytest.param("NOT (count() * 2 - 1 > 9)", "NOT (F1 * 2 - 1 > 9)", {"service-a", "service-b", "service-c", "service-d"}, id="not_arithmetic"),
|
||||
],
|
||||
)
|
||||
def test_logs_formula_having_time_series_matches_query_having(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
query_having: str,
|
||||
formula_having: str,
|
||||
expected_services: set[str],
|
||||
) -> None:
|
||||
"""
|
||||
The querier computes formulas in Go, so a formula having runs in Go. For the
|
||||
formula `A`, it must keep the same points that SQL HAVING keeps on query A.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_logs([Logs(timestamp=now - timedelta(minutes=5 - minute), resources={"service.name": service}, body=f"{service} log {minute}-{i}") for service, counts in FORMULA_HAVING_COUNTS.items() for minute, count in enumerate(counts) for i in range(count)])
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
group_by = [build_group_by_field("service.name", "string", "resource")]
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=15)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="time_series",
|
||||
queries=[
|
||||
build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
|
||||
build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
|
||||
build_formula_query("F1", "A", having_expression=formula_having),
|
||||
],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
query_series = index_series_by_label(get_all_series(response.json(), "B"), "service.name")
|
||||
formula_series = index_series_by_label(get_all_series(response.json(), "F1"), "service.name")
|
||||
assert set(formula_series) == expected_services
|
||||
assert set(query_series) == expected_services
|
||||
for service in expected_services:
|
||||
query_points = sorted((value["timestamp"], value["value"]) for value in query_series[service]["values"])
|
||||
formula_points = sorted((value["timestamp"], value["value"]) for value in formula_series[service]["values"])
|
||||
assert formula_points == query_points, service
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"query_having,formula_having,expected_services",
|
||||
[
|
||||
pytest.param("count() > 4 AND count() < 16", "F1 > 4 AND F1 < 16", {"service-a", "service-c", "service-d"}, id="and_range"),
|
||||
pytest.param("count() IN (1, 5, 17) OR count() > 15", "__result IN (1, 5, 17) OR __result > 15", {"service-b", "service-c"}, id="in_or"),
|
||||
pytest.param("NOT (count() * 2 - 1 > 9)", "NOT (F1 * 2 - 1 > 9)", {"service-c"}, id="not_arithmetic"),
|
||||
],
|
||||
)
|
||||
def test_logs_formula_having_scalar_matches_query_having(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
query_having: str,
|
||||
formula_having: str,
|
||||
expected_services: set[str],
|
||||
) -> None:
|
||||
"""
|
||||
For the formula `A` in a scalar request, the formula having must keep the
|
||||
same rows that SQL HAVING keeps on query A.
|
||||
"""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_logs([Logs(timestamp=now - timedelta(minutes=5 - minute), resources={"service.name": service}, body=f"{service} log {minute}-{i}") for service, counts in FORMULA_HAVING_COUNTS.items() for minute, count in enumerate(counts) for i in range(count)])
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
group_by = [build_group_by_field("service.name", "string", "resource")]
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=15)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="scalar",
|
||||
queries=[
|
||||
build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, disabled=True),
|
||||
build_scalar_query(name="B", signal="logs", aggregations=[build_aggregation("count()")], group_by=group_by, having_expression=query_having),
|
||||
build_formula_query("F1", "A", having_expression=formula_having),
|
||||
],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
|
||||
results = response.json()["data"]["data"]["results"]
|
||||
values_by_query = {}
|
||||
for name in ("B", "F1"):
|
||||
result = find_named_result(results, name)
|
||||
assert result is not None, name
|
||||
service_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "service.name")
|
||||
value_column = next(i for i, column in enumerate(result["columns"]) if column["columnType"] == "aggregation")
|
||||
values_by_query[name] = {row[service_column]: row[value_column] for row in result["data"]}
|
||||
assert set(values_by_query["F1"]) == expected_services
|
||||
assert values_by_query["F1"] == values_by_query["B"]
|
||||
|
||||
|
||||
def test_logs_formula_having_invalid_reference(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
) -> None:
|
||||
"""A formula having can reference only the formula value, so `count()` is a bad request."""
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=15)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type="time_series",
|
||||
queries=[
|
||||
build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], disabled=True),
|
||||
build_formula_query("F1", "A", having_expression="count() > 0"),
|
||||
],
|
||||
)
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
|
||||
assert get_error_message(response.json()) == "Invalid references in `Having` expression: [count]"
|
||||
|
||||
@@ -442,55 +442,3 @@ def test_builder_refreshing_a_sliding_time_range(
|
||||
assert uncached.status_code == HTTPStatus.OK, uncached.text
|
||||
|
||||
assert_results_equal(from_cache.json(), uncached.json(), "A", "the twentieth refresh")
|
||||
|
||||
|
||||
def test_builder_refreshing_does_not_inflate_rows_scanned(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
metric_name = f"cache_rows_scanned_{uuid4().hex[:8]}"
|
||||
|
||||
# 90 minutes back so even the last refresh closes clear of the flux
|
||||
# interval, which holds recent data out of the cache
|
||||
start_time = datetime.fromtimestamp(int((datetime.now(tz=UTC) - timedelta(minutes=90)).timestamp()) // 60 * 60, tz=UTC)
|
||||
start_time_ms = int(start_time.timestamp() * 1000)
|
||||
|
||||
query = [build_builder_query("A", metric_name, "max", "max")]
|
||||
|
||||
insert_metrics(
|
||||
[
|
||||
Metrics(
|
||||
metric_name=metric_name,
|
||||
labels={"service": "api"},
|
||||
timestamp=start_time + timedelta(minutes=minute),
|
||||
value=minute,
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
)
|
||||
for minute in range(70)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
# a dashboard left open on a thirty minute range, re-running a minute later
|
||||
# each time. rowsScanned is a cost indicator, so a refresh served mostly out
|
||||
# of the cache cannot claim more than every read made so far: the first
|
||||
# full read and then one new minute per refresh, measured with the cache off
|
||||
uncached_rows_scanned_so_far = 0
|
||||
for refresh in range(10):
|
||||
refresh_start_ms = start_time_ms + refresh * MINUTE_MS
|
||||
refresh_end_ms = refresh_start_ms + 30 * MINUTE_MS
|
||||
|
||||
uncached_start_ms = refresh_start_ms if refresh == 0 else refresh_end_ms - MINUTE_MS
|
||||
uncached_response = make_query_request(signoz, token, uncached_start_ms, refresh_end_ms, query, no_cache=True)
|
||||
assert uncached_response.status_code == HTTPStatus.OK, f"refresh {refresh}: uncached read failed: {uncached_response.text}"
|
||||
uncached_rows_scanned_so_far += uncached_response.json()["data"]["meta"]["rowsScanned"]
|
||||
assert uncached_rows_scanned_so_far > 0, f"refresh {refresh}: uncached read reported no rows scanned"
|
||||
|
||||
cached_response = make_query_request(signoz, token, refresh_start_ms, refresh_end_ms, query, no_cache=False)
|
||||
assert cached_response.status_code == HTTPStatus.OK, f"refresh {refresh}: cached read failed: {cached_response.text}"
|
||||
cached_rows_scanned = cached_response.json()["data"]["meta"]["rowsScanned"]
|
||||
assert cached_rows_scanned <= uncached_rows_scanned_so_far, f"refresh {refresh}: cached read reported {cached_rows_scanned} rows scanned, more than the {uncached_rows_scanned_so_far} rows all uncached reads scanned so far"
|
||||
|
||||
Reference in New Issue
Block a user