Feature Request: Proactive server query killing based on Query cost or Strategy

[anuragrai16]

Motivation

Apache Pinot currently has mechanisms to kill runaway queries based on CPU time and memory pressure via QueryResourceAggregator. However, these mechanisms have limitations:

Reactive, not proactive: The background thread (QueryResourceAggregator) samples resource usage periodically (default 100ms). A query scanning billions of entries can cause significant damage before being detected.
Delayed detection: A full table scan query can process 100M+ entries between background thread samples, causing CPU spikes and impacting other queries before termination.
Gap in protection: The existing killMostExpensiveQuery() (memory-based) and killCPUTimeExceedQueries() (CPU-based) work well for memory-intensive and CPU-intensive queries respectively. However, a query doing a massive filter scan may not immediately trigger high memory usage but will consume significant CPU cycles processing entries that ultimately get filtered out.

Problem : In production environments with tables containing billions of rows, a single missing filter predicate or low-selectivity filter can trigger full segment scans. By the time the background thread detects high CPU usage, the query has already consumed significant resources and impacted cluster performance and increased tail latency for most other queries.

To solve this problem, I would like to propose a Proactive Query Killing Framework that terminates queries inline based on scan thresholds, complementing the existing background-based CPU/memory killing. This framework will plug into existing query killing mechanism, just introduce a proactive way to track and terminate the queries instead of relying on overall JVM memory or CPU usage. The strategy will be made pluggable so it can be extended with more cost measurements in the future.

Core Components

QueryCostContext Interface: Exposes real-time query metrics:

getNumEntriesScannedInFilter()
getNumDocsScanned()
getElapsedTimeMs()

QueryKillingStrategy Interface: Pluggable strategy for termination decisions:

  interface QueryKillingStrategy {
      boolean shouldTerminate(QueryCostContext context);
      String getTerminationReason();
  }

Configuration

pinot.server.query.killer.enabled=true
pinot.server.query.killer.threshold.entriesScannedInFilter=100000000
pinot.server.query.killer.threshold.docsScanned=10000000
# Can also be overridden with table specific overrides

####Non-Goals for now (Can be a future prospect)

• Broker-side query killing (future work)
• Query fingerprinting / pattern-based blocking
• Adaptive thresholds based on cluster load

####Similar design in other systems

• ClickHouse provides comprehensive query complexity settings that limit scan operations inline. eg. max_rows_to_read, max_rows_to_read_leaf, read_overflow_mode

• Trino provides query.max-scan-physical-bytes to limit data scanned:

"The maximum number of bytes that can be scanned by a query during its execution. When this limit is reached, query processing is terminated to prevent excessive resource usage."

Previous Work

• Runtime query killing on brokers and servers using Accounting

[mayankshriv]

Thanks @anuragrai16 for the PEP, this is a good idea. I am wondering though, should we take the kill route, or should we take the capping based on cost route? We already do that for number of groups, we can do the same thing for scans as well?

[anuragrai16]

@mayankshriv - Thanks. We can do both if needed by supporting a configuration like OVERFLOW_MODE which can take in KILL (kill the query if cost exceeds) or BREAK/TRIM (break the query and return if cost exceeds). But practically, not every cost strategy would work with BREAK/TRIM. Cap/Trim only makes semantic sense when partial results are meaningful. For GROUP BY with ORDER BY + LIMIT, returning top-N groups is valid. Whereas stopping a scan in the middle and returning partial result to be aggregated will yield wrong results. So, I'd like to keep the default as Kill for now, as its easier from a customer perspective to explain why your query was killed, but we can have a mode that Breaks/caps the query as well.

[anuragrai16]

[RFC] Pinot : Proactive server query killing based on Query cost or Strategy

This design layers a scan-cost-based query killing framework on top of Pinot's existing termination infrastructure. Queries that exceed configurable scan thresholds (entries scanned in filter, docs scanned) are terminated proactively -- before they exhaust server resources.

Goals

Proactive termination of expensive queries before they exhaust resources
Minimal performance overhead in hot paths (< 0.1% impact at 30K QPS)
Extensible architecture supporting multiple killing strategies
Production safety -- default OFF, dry-run mode, framework isolation, dynamic config
Debuggability -- actionable error messages for customers, structured logs for operators

Architecture

Existing Killing Mechanisms
Query Timeout: QueryExecutionContext._activeDeadlineMs checked in checkTerminationInternal()
OOM Kill: QueryResourceAggregator kills queries when heap exceeds thresholds (panic/critical/alarming levels)
CPU Time Kill: QueryResourceAggregator.killCPUTimeExceedQueries() kills based on CPU time
Interrupt-based: Future.cancel(true) sends interrupt to worker threads

MSE Coverage

MSE (Multi-Stage Engine) queries are automatically covered. MSQE leaf stages use BaseOperator subclasses from pinot-core for segment scanning:
LeafOperator.execute()
-> QueryExecutor.execute(ServerQueryRequest)
-> [SSE operators: DocIdSetOperator, AggregationOperator, etc.]
-> BaseOperator.nextBlock() -> checkTermination()
Since the scan cost check is added to BaseOperator.checkTermination(), both SSE and MSQE queries participate.

Key Design Decisions

Scan cost check lives in BaseOperator.checkTermination() (pinot-core)
LongAdder-based cost accumulator for lock-free cross-thread aggregation
Config in CommonConstants.Accounting + QueryMonitorConfig (cluster-wide) with QueryConfig overrides (table-level)
Framework isolation: kill-evaluation bugs never crash queries
Log-only (dry-run) mode for safe production rollout
Distinct error code QUERY_SCAN_LIMIT_EXCEEDED (HTTP 400) -- tells clients "fix your query", not "retry later"

SPI Changes

QueryScanCostContext (pinot-spi)
Thread-safe accumulator using java.util.concurrent.atomic.LongAdder. One instance per query, shared across all segment worker threads.

public final class QueryScanCostContext {
    private final LongAdder _numEntriesScannedInFilter = new LongAdder();
    private final LongAdder _numDocsScanned = new LongAdder();
    private final LongAdder _numEntriesScannedPostFilter = new LongAdder();
    private final long _startTimeMs = System.currentTimeMillis();

    public void addEntriesScannedInFilter(long count) {
        _numEntriesScannedInFilter.add(count);
    }

    public void addDocsScanned(long count) {
        _numDocsScanned.add(count);
    }

    public void addEntriesScannedPostFilter(long count) {
        _numEntriesScannedPostFilter.add(count);
    }

    public long getNumEntriesScannedInFilter() { return _numEntriesScannedInFilter.sum(); }
    public long getNumDocsScanned() { return _numDocsScanned.sum(); }
    public long getNumEntriesScannedPostFilter() { return _numEntriesScannedPostFilter.sum(); }
    public long getElapsedTimeMs() { return System.currentTimeMillis() - _startTimeMs; }
}

QueryKillingStrategy Interface (pinot-core)

**public interface QueryKillingStrategy {

boolean shouldTerminate(QueryScanCostContext context);

QueryKillReport buildKillReport(QueryScanCostContext context,
                                 String queryId, String tableName,
                                 String configSource);

default QueryErrorCode getErrorCode() {
    return QueryErrorCode.QUERY_SCAN_LIMIT_EXCEEDED;
}

default int priority() { return 100; }

/**
 * Returns a query-specific variant with table-level overrides applied.
 * If no overrides are relevant, returns this (no allocation).
 * Default: returns this (strategy does not support table overrides).
 */
default QueryKillingStrategy forQuery(@Nullable QueryConfig queryConfig,
                                       QueryMonitorConfig clusterConfig) {
    return this;
}

}**

QueryKillingStrategyFactory Interface (pinot-core)

Loaded by class name from config, following the same pattern as ThreadAccountantFactory. Custom strategies provide their own factory that reads whatever config keys they need.

public interface QueryKillingStrategyFactory {
    @Nullable
    QueryKillingStrategy create(QueryMonitorConfig config);
    String getName();
}

ScanEntriesThresholdStrategy (pinot-core)
Default strategy. Kills if numEntriesScannedInFilter > threshold OR numDocsScanned > threshold. Threshold of Long.MAX_VALUE disables that metric.
Includes a nested Factory class implementing QueryKillingStrategyFactory. The factory validates config at init: if no thresholds are set (all Long.MAX_VALUE), returns null and logs a warning.

public class ScanEntriesThresholdStrategy implements QueryKillingStrategy {
    private final long _maxEntriesScannedInFilter;
    private final long _maxDocsScanned;

    @Override
    public boolean shouldTerminate(QueryScanCostContext ctx) { /* ... */ }

    @Override
    public QueryKillReport buildKillReport(QueryScanCostContext ctx,
            String queryId, String tableName, String configSource) { /* ... */ }

    @Override
    public int priority() { return 10; }

    /**
     * Returns a query-specific variant with table-level threshold overrides.
     * If table QueryConfig has non-null threshold fields, they replace this
     * strategy's thresholds. Otherwise returns this (no allocation).
     */
    @Override
    public QueryKillingStrategy forQuery(@Nullable QueryConfig queryConfig,
            QueryMonitorConfig clusterConfig) {
        if (queryConfig == null) return this;
        Long tableEntries = queryConfig.getMaxEntriesScannedInFilter();
        Long tableDocs = queryConfig.getMaxDocsScanned();
        if (tableEntries == null && tableDocs == null) return this;
        return new ScanEntriesThresholdStrategy(
            tableEntries != null ? tableEntries : _maxEntriesScannedInFilter,
            tableDocs != null ? tableDocs : _maxDocsScanned);
    }

    /**
     * Factory that creates a ScanEntriesThresholdStrategy from QueryMonitorConfig.
     * Default factory when no custom strategy is configured.
     * Returns null if no thresholds are set (logs warning).
     */
    public static class Factory implements QueryKillingStrategyFactory {
        @Override
        public QueryKillingStrategy create(QueryMonitorConfig config) {
            long maxEntries = config.getScanBasedKillingMaxEntriesScannedInFilter();
            long maxDocs = config.getScanBasedKillingMaxDocsScanned();
            if (maxEntries == Long.MAX_VALUE && maxDocs == Long.MAX_VALUE) {
                LOGGER.warn("Scan-based killing is enabled but no thresholds configured. "
                    + "Set at least one of: max.entries.scanned.in.filter, max.docs.scanned");
                return null;
            }
            return new ScanEntriesThresholdStrategy(maxEntries, maxDocs);
        }
        @Override
        public String getName() { return "ScanEntriesThresholdStrategyFactory"; }
    }
}

QueryKillingManager (pinot-core)

Init: loads factory from config (or default), builds strategy, validates config
Per-query: resolves table overrides via strategy.forQuery(), delegates to strategy.shouldTerminate()
Guard rails: enabled check, duplicate-kill prevention, log-only mode, framework isolation (try-catch)
Config change: rebuildStrategy() rebuilds from updated QueryMonitorConfig (wired to ResourceUsageAccountantFactory.onChange())
The manager never hardcodes threshold logic. All kill decisions flow through the strategy interface.

Config Examples

**Step 1: Enable with log-only **
  accounting.scan.based.killing.mode=logOnly
  accounting.scan.based.killing.max.entries.scanned.in.filter=500000000

**Step 2: Monitor logs and metrics (QUERIES_KILLED_SCAN_DRY_RUN)**
  - Validate which queries would be killed
  - Tune thresholds if needed

**Step 3: Switch to enforce mode (ZK, no restart)**
  accounting.scan.based.killing.mode=enforce

**Step 4: Emergency disable if needed (ZK, no restart)**
  accounting.scan.based.killing.mode=disabled

Table-Level Overrides
Per-table thresholds in QueryConfig (under the "query" key in table config JSON):

{ "query": { "maxEntriesScannedInFilter": 1000000000, "maxDocsScanned": 100000000 } }

Resolution: table config > cluster config > default (MAX_VALUE). Null in table config means "use cluster default."

Safety

• Killing disabled: _strategy == null means immediate return, zero overhead
• All thresholds MAX_VALUE: QueryScanCostContext not allocated, strategy not built
• Framework or Strategy bug: try-catch(Exception) around all evaluation — query continues normally
• Duplicate kills: getTerminateException() != null check prevents re-evaluation of already-terminated queries

Future Extensions

The design enables all of these without future operator changes:

1. Cost-Based Model: Add getWeightedCost() to QueryScanCostContext; new strategy implementation
2. Selectivity-Ratio Strategy: Kill if scanned / passed < threshold
3. Adaptive Thresholds: ThresholdProvider interface that reads server CPU/heap
4. Broker-Driven Per-Query Limits: Pass limits in QueryOptions
5. Query Fingerprinting: Per-pattern thresholds
6. Broker-Side Aggregated Killing: DataTable metadata already carries scan stats