Query Engine Architecture

+----------------------------------------------------------------+
|                      Query Engine Service                       |
|                                                                 |
|  +------------------+  +------------------+  +---------------+  |
|  | REST Controllers |  | Security Layer   |  | Cache Layer   |  |
|  | - QueryController|  | - JWT Filter     |  | - L1 Memory   |  |
|  | - ScheduleCtrl   |  | - RLS Service    |  | - L2 Redis    |  |
|  | - CacheCtrl      |  | - Masking Svc    |  | - Semantic    |  |
|  | - CostEstCtrl    |  | - SecurityConfig |  | - Plan Cache  |  |
|  +--------+---------+  +--------+---------+  +-------+-------+  |
|           |                      |                    |          |
|  +--------v----------------------v--------------------v-------+  |
|  |                  Query Execution Service                   |  |
|  |  - Sync execution    - Async submission                    |  |
|  |  - State machine     - Result pagination                   |  |
|  +----------------------------+-------------------------------+  |
|                               |                                  |
|  +----------------------------v-------------------------------+  |
|  |                  Smart Query Router                        |  |
|  |  - Query analysis     - Cost estimation                    |  |
|  |  - Engine selection   - Workload routing                   |  |
|  +-----+------------------+------------------+----------------+  |
|        |                  |                  |                   |
|  +-----v------+   +------v-------+   +------v--------+         |
|  | Trino      |   | ClickHouse   |   | Spark Async   |         |
|  | Strategy   |   | Strategy     |   | Strategy      |         |
|  +------------+   +--------------+   +---------------+         |
+----------------------------------------------------------------+

// Security filter chain ordering
@Configuration
public class SecurityConfig {
    // 1. JWT Authentication Filter - validates token, extracts tenant context
    // 2. RLS Filter Service - injects row-level security predicates
    // 3. Data Masking Service - applies column masking rules
    // 4. Query Audit Service - logs access for compliance
}

@Service
@RequiredArgsConstructor
public class QueryExecutionService {
 
    private final SmartQueryRouter queryRouter;
    private final Map<EngineType, QueryEngineStrategy> strategies;
    private final RlsFilterService rlsFilterService;
    private final QueryCacheService cacheService;
    private final QueryAuditService auditService;
 
    public QueryResponse executeSync(UUID tenantId, UUID userId, QueryRequest request) {
        // 1. Check cache for existing result
        // 2. Apply RLS filters
        // 3. Apply data masking rules
        // 4. Route to optimal engine
        // 5. Execute via strategy
        // 6. Cache result
        // 7. Record audit entry
        // 8. Return response
    }
 
    public AsyncQueryResponse submitAsync(UUID tenantId, UUID userId, QueryRequest request) {
        // 1. Validate query
        // 2. Estimate cost
        // 3. Apply RLS and masking
        // 4. Submit to execution queue
        // 5. Return execution ID and estimated completion time
    }
}

public interface QueryEngineStrategy {
    QueryResponse execute(QueryRequest request, int timeoutSeconds);
    EngineType getEngineType();
    boolean isAvailable();
    void healthCheck();
}

Query arrives
    |
    v
[Contains real-time tables?] -- Yes --> ClickHouse
    |
    No
    v
[Estimated scan > 100GB?] -- Yes --> Spark Async
    |
    No
    v
[Estimated rows < 1M AND no joins AND no window functions?] -- Yes --> ClickHouse
    |
    No
    v
[Has window functions OR > 2 joins?] -- Yes --> Trino
    |
    No
    v
Default --> Trino

private QueryAnalysis analyzeQuery(QueryRequest request) {
    QueryAnalysis analysis = new QueryAnalysis();
    String sql = request.getSql().toLowerCase();
 
    // 1. Check for real-time table patterns (events, clicks, pageviews, metrics, logs, sessions)
    // 2. Detect aggregate functions (COUNT, SUM, AVG, MIN, MAX, PERCENTILE, APPROX_*)
    // 3. Detect window functions (OVER, PARTITION BY, ROW_NUMBER, RANK, DENSE_RANK, LAG, LEAD)
    // 4. Count JOIN operations
    // 5. Parse SQL via JSqlParser for GROUP BY, ORDER BY, and subquery detection
    // 6. Estimate row count and scan size
}

                 +----------+
                 | RECEIVED |
                 +----+-----+
                      |
                 +----v-----+
                 |  QUEUED   |
                 +----+-----+
                      |
              +-------+--------+
              |                |
         +----v-----+    +----v-------+
         | RUNNING  |    | CANCELLED  |
         +----+-----+    +------------+
              |
      +-------+--------+
      |                |
 +----v-----+    +----v------+
 | COMPLETED|    |  FAILED   |
 +----------+    +-----------+

public class QueryRequest {
    private String sql;                        // The SQL query to execute
    private String catalog;                    // Trino catalog (optional, defaults to 'delta')
    private String schema;                     // Trino schema (optional, defaults to 'default')
    private Integer limit;                     // Result row limit (default: 10000, max: 100000)
    private Integer timeoutSeconds;            // Execution timeout (default: 300)
    private String enginePreference;           // Preferred engine (optional, overrides router)
    private Map<String, String> sessionProperties;  // Trino session properties
    private Map<String, Object> metadata;      // Billing context, tags, correlation IDs
    private UUID tenantId;                     // Injected from header
    private UUID queryId;                      // Generated if not provided
}

public class QueryResponse {
    private UUID executionId;
    private QueryStatus status;                // RECEIVED, QUEUED, RUNNING, COMPLETED, FAILED, CANCELLED
    private EngineType engineType;             // TRINO, CLICKHOUSE, SPARK_ASYNC
    private List<ColumnMetadata> columns;      // Column names, types, nullability
    private List<Map<String, Object>> data;    // Result rows
    private long rowCount;                     // Rows returned in this page
    private long totalRows;                    // Total rows in result set
    private long executionTimeMs;              // Total execution time
    private boolean hasMore;                   // Whether more rows are available
    private String cursorToken;                // Cursor for next page
    private Instant submittedAt;
    private Instant completedAt;
    private String errorMessage;               // Error details if FAILED
}

public enum DataSourceType {
    POSTGRESQL,
    MYSQL,
    MONGODB,
    S3,
    GCS,
    AZURE_BLOB,
    TRINO_CATALOG,
    CLICKHOUSE,
    SNOWFLAKE,
    BIGQUERY
}

query.execute (root)
  |-- query.cache.lookup
  |-- query.rls.apply
  |-- query.masking.apply
  |-- query.router.decide
  |-- query.engine.execute
  |     |-- trino.connection.acquire
  |     |-- trino.session.configure
  |     |-- trino.query.submit
  |     |-- trino.results.fetch
  |-- query.cache.store
  |-- query.audit.record

# Key Helm values
replicaCount: 2
image:
  repository: matih/query-engine
  tag: latest
service:
  port: 8080
resources:
  requests:
    cpu: 500m
    memory: 1Gi
  limits:
    cpu: 2
    memory: 4Gi
env:
  QUERY_TRINO_URL:
    value: "jdbc:trino://trino.matih-data-plane.svc.cluster.local:8080"
  QUERY_TRINO_CATALOG:
    value: "delta"
  SPRING_REDIS_HOST:
    valueFrom:
      secretKeyRef:
        name: query-engine-redis
        key: host
  SPRING_DATASOURCE_URL:
    valueFrom:
      secretKeyRef:
        name: query-engine-db
        key: url