Discharge7443/fleet
1
0
Fork 0
Code Issues Pull Requests Actions 22 Packages Projects Releases Wiki Activity

Batch select query in CleanupExcessQueryResultRows (#40491)

Browse Source 
<!-- Add the related story/sub-task/bug number, like Resolves #123, or
remove if NA -->
**Related issue:** Resolves #40476

# Checklist for submitter

- [x] Changes file added for user-visible changes in `changes/`,
`orbit/changes/` or `ee/fleetd-chrome/changes`.
See [Changes
files](https://github.com/fleetdm/fleet/blob/main/docs/Contributing/guides/committing-changes.md#changes-files)
for more information.

## Testing

- [x] Added/updated automated tests

- [x] QA'd all new/changed functionality manually

Before:
- inserted 70k queries to my local DB, saw the cron failing:

<img width="864" height="120" alt="Screenshot 2026-02-25 at 12 54 31 PM"
src="https://github.com/user-attachments/assets/d1e19aa8-56aa-46a2-a437-7ae5da1e5b1e"
/>

- ran new test without code fix, it failed with the same error in the
issue:

<img width="920" height="324" alt="Screenshot 2026-02-25 at 12 45 41 PM"
src="https://github.com/user-attachments/assets/c7342d81-f223-449e-a861-c7bae58bbe9e"
/>

After: ran test again, it passed

<img width="1556" height="174" alt="Screenshot 2026-02-25 at 12 45
04 PM"
src="https://github.com/user-attachments/assets/9eed3e6e-3ce6-4d69-aa70-9ebcfcf07623"
/>
This commit is contained in:
Nico
2026-02-25 18:01:47 -03:00
committed by GitHub
parent e7043b45dc
commit 09d86aa5f0
4 changed files with 78 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changes/40476-fix-query-results-cleanup-placeholder-limit
+1
View File
@@ -0,0 +1 @@
* Fixed query results cleanup cron failing with "too many placeholders" error by filtering to only saved queries and batching the SQL IN clause.
changes/fix-query-results-cleanup-placeholder-limit
+1
View File
@@ -0,0 +1 @@
* Fixed query results cleanup cron failing with "too many placeholders" error by filtering to only saved queries and batching the SQL IN clause.
server/datastore/mysql/query_results.go
+27 -14
View File
@@ -3,6 +3,7 @@ package mysql
import (
"context"
"fmt"
"slices"
"strings"
"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
@@ -155,12 +156,14 @@ func (ds *Datastore) CleanupExcessQueryResultRows(ctx context.Context, maxQueryR
batchSize = opts[0].BatchSize
}
// Get all distinct query_ids that have results and are scheduled queries with discard_data = false
// Get all saved query IDs that could have query results to clean up.
// Only saved queries (scheduled reports) store rows in query_results;
// live queries do not, so there's nothing to clean up for them.
var queryIDs []uint
selectStmt := `
SELECT id
FROM queries
WHERE discard_data = false AND logging_type = 'snapshot'
WHERE saved = 1 AND discard_data = false AND logging_type = 'snapshot'
`
if err := sqlx.SelectContext(ctx, ds.reader(ctx), &queryIDs, selectStmt); err != nil {
return nil, ctxerr.Wrap(ctx, err, "selecting query IDs for cleanup")
@@ -188,12 +191,18 @@ func (ds *Datastore) CleanupExcessQueryResultRows(ctx context.Context, maxQueryR
) cutoff
WHERE rn = ?
`
query, args, err := sqlx.In(cutoffStmt, queryIDs, maxQueryReportRows)
if err != nil {
return nil, ctxerr.Wrap(ctx, err, "building cutoff query")
}
if err := sqlx.SelectContext(ctx, ds.reader(ctx), &queryCutoffs, ds.reader(ctx).Rebind(query), args...); err != nil {
return nil, ctxerr.Wrap(ctx, err, "selecting cutoffs")
// Batch the IN clause to avoid MySQL's 65,535 placeholder limit.
const queryIDBatchSize = 50000
for batch := range slices.Chunk(queryIDs, queryIDBatchSize) {
var batchCutoffs []cutoffRow
query, args, err := sqlx.In(cutoffStmt, batch, maxQueryReportRows)
if err != nil {
return nil, ctxerr.Wrap(ctx, err, "building cutoff query")
}
if err := sqlx.SelectContext(ctx, ds.reader(ctx), &batchCutoffs, ds.reader(ctx).Rebind(query), args...); err != nil {
return nil, ctxerr.Wrap(ctx, err, "selecting cutoffs")
}
queryCutoffs = append(queryCutoffs, batchCutoffs...)
}
// Delete excess rows from each query, in batches.
@@ -230,12 +239,16 @@ func (ds *Datastore) CleanupExcessQueryResultRows(ctx context.Context, maxQueryR
WHERE query_id IN (?) AND data IS NOT NULL
GROUP BY query_id
`
query, args, err = sqlx.In(countStmt, queryIDs)
if err != nil {
return nil, ctxerr.Wrap(ctx, err, "building count query")
}
if err := sqlx.SelectContext(ctx, ds.reader(ctx), &counts, ds.reader(ctx).Rebind(query), args...); err != nil {
return nil, ctxerr.Wrap(ctx, err, "selecting counts")
for batch := range slices.Chunk(queryIDs, queryIDBatchSize) {
var batchCounts []countRow
query, args, err := sqlx.In(countStmt, batch)
if err != nil {
return nil, ctxerr.Wrap(ctx, err, "building count query")
}
if err := sqlx.SelectContext(ctx, ds.reader(ctx), &batchCounts, ds.reader(ctx).Rebind(query), args...); err != nil {
return nil, ctxerr.Wrap(ctx, err, "selecting counts")
}
counts = append(counts, batchCounts...)
}
queryCounts := make(map[uint]int)
server/datastore/mysql/query_results_test.go
+49
View File
@@ -31,6 +31,7 @@ func TestQueryResults(t *testing.T) {
{"QueryResultRowsFilter", testQueryResultRowsTeamFilter},
{"CleanupQueryResultRows", testCleanupQueryResultRows},
{"CleanupExcessQueryResultRows", testCleanupExcessQueryResultRows},
{"CleanupExcessQueryResultRowsManyQueries", testCleanupExcessQueryResultRowsManyQueries},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
@@ -767,3 +768,51 @@ func testCleanupExcessQueryResultRows(t *testing.T, ds *Datastore) {
return nil
})
}
// testCleanupExcessQueryResultRowsManyQueries verifies that CleanupExcessQueryResultRows
// works when there are more queries than MySQL's prepared statement placeholder limit (65,535).
func testCleanupExcessQueryResultRowsManyQueries(t *testing.T, ds *Datastore) {
const numQueries = 70000
ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error {
_, err := q.ExecContext(t.Context(),
`INSERT INTO users (name, email, password, salt) VALUES ('test', 'bulk@test.com', 'x', 'x')`)
if err != nil {
return err
}
_, err = q.ExecContext(t.Context(), `
INSERT INTO queries (name, description, query, author_id, logging_type, discard_data, saved)
SELECT
CONCAT('bulk_query_', seq),
'',
'SELECT 1',
(SELECT id FROM users LIMIT 1),
'snapshot',
false,
1
FROM (
SELECT a.N + b.N*10 + c.N*100 + d.N*1000 + e.N*10000 as seq
FROM
(SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) a,
(SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) b,
(SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) c,
(SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) d,
(SELECT 0 AS N UNION SELECT 1 UNION SELECT 2 UNION SELECT 3 UNION SELECT 4 UNION SELECT 5 UNION SELECT 6 UNION SELECT 7 UNION SELECT 8 UNION SELECT 9) e
) numbers
WHERE seq < ?
`, numQueries)
return err
})
var count int
ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error {
return sqlx.GetContext(t.Context(), q, &count,
`SELECT COUNT(*) FROM queries WHERE discard_data = false AND logging_type = 'snapshot'`)
})
require.Equal(t, numQueries, count)
queryCounts, err := ds.CleanupExcessQueryResultRows(t.Context(), fleet.DefaultMaxQueryReportRows)
require.NoError(t, err)
require.Len(t, queryCounts, numQueries)
}