Skip to content

Batch select query in CleanupExcessQueryResultRows #40491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nulmete merged 3 commits into from
Feb 25, 2026
Merged

Batch select query in CleanupExcessQueryResultRows #40491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e2053ef
Batch query in CleanupExcessQueryResultRows
nulmete Feb 25, 2026
fbd0ebc
Add saved = 1 filter
nulmete Feb 25, 2026
9b0db93
Adjust changes file name
nulmete Feb 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changes/40476-fix-query-results-cleanup-placeholder-limit
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
1 change: 1 addition & 0 deletions changes/fix-query-results-cleanup-placeholder-limit
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
41 changes: 27 additions & 14 deletions server/datastore/mysql/query_results.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package mysql
import (
"context"
"fmt"
"slices"
"strings"

"github.com/fleetdm/fleet/v4/server/contexts/ctxerr"
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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)
Expand Down
49 changes: 49 additions & 0 deletions server/datastore/mysql/query_results_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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)
}
Loading