Source Code Reading: QueryFederationService — Multi-Engine Query Routing

“

Series: S9 Source Code Reading · Article 5 | Level: Advanced | Reading Time: 25 min

#TL;DR

QueryFederationGrpcService is the core query federation service in Data Layer (Data Layer), built on Quarkus 3.x + gRPC. It receives OQL queries and processes them through a four-stage pipeline: parse, optimize, route, execute — routing queries to Doris (OLAP) or DuckDB (embedded analytics). It supports cache acceleration, streaming results, query plan explanation, vector search, and graph traversal. This article analyzes the eight-dependency injection architecture, complete query execution pipeline, StorageRouter federation logic, aggregation-aware caching, VirtualPropertyEnricher derived property filling, and async query state management.

#Table of Contents

Architecture and Eight Collaborators
Query Execution Pipeline: 8 Steps of executeQuery
StorageRouter: Federation Determination
Caching Strategy: Aggregation-Aware Query Cache
VirtualPropertyEnricher: Derived Property Filling
Streaming Queries: executeQueryStream
Query Validation and Plan Explanation
Vector Search and Graph Traversal: OQL Unified Abstraction
Time Travel Queries: Reserved Interface Design
Async Query State Management
Key Takeaways

#1. Architecture and Eight Collaborators

Java

@GrpcService
public class QueryFederationGrpcService
        extends QueryFederationServiceGrpc.QueryFederationServiceImplBase {

    private final OQLParserService parser;
    private final QueryOptimizer optimizer;
    private final QueryExecutor executor;
    private final StorageRouter router;
    private final QueryProtoConverter queryConverter;
    private final ResultProtoConverter resultConverter;
    private final QueryCacheService cacheService;
    private final VirtualPropertyEnricher virtualPropertyEnricher;

    private final Map<String, AsyncQueryState> asyncQueries = new ConcurrentHashMap<>();
}

Eight collaborators by responsibility:

Collaborator	Stage	Responsibility
`OQLParserService`	Parse	OQL text to AST
`QueryOptimizer`	Optimize	AST to physical execution plan
`QueryExecutor`	Execute	Single-engine query execution
`StorageRouter`	Route	Federation determination and cross-engine execution
`QueryProtoConverter`	Inbound	gRPC request to query context
`ResultProtoConverter`	Outbound	Query result to gRPC response
`QueryCacheService`	Cache	Redis caching of query results
`VirtualPropertyEnricher`	Enrich	Fill Reasoning & Decision Layer computed derived properties

This is one of the most dependency-heavy services in coomia-dip, reflecting query federation complexity.

#2. Query Execution Pipeline: 8 Steps of executeQuery

Java

@Override
public void executeQuery(ExecuteQueryRequest request,
        StreamObserver<ExecuteQueryResponse> responseObserver) {
    try {
        long startTime = System.currentTimeMillis();

        // Step 1: Build query context
        var context = queryConverter.toQueryContext(request);

        // Step 2: Cache check (early return on hit)
        if (options.enableCache() && cacheService.isEnabled()) {
            cacheKey = cacheService.generateCacheKey(worldId, query, params);
            var cached = cacheService.get(cacheKey);
            if (cached.isPresent()) { /* return cached */ return; }
        }

        // Step 3: Parse OQL to AST
        var ast = parser.parse(query);

        // Step 4: Optimize AST to physical plan
        var plan = optimizer.optimize(ast, context);

        // Step 5: Aggregation-aware cache policy
        boolean isAggregation = plan instanceof PhysicalAggregatePlan;
        boolean shouldCache = options.enableCache()
                && (!isAggregation || cacheService.isCacheAggregationsEnabled());

        // Step 6: Route and execute
        QueryResult result;
        if (router.requiresFederation(plan)) {
            result = router.executeFederatedQuery(plan, context);
        } else {
            result = executor.execute(plan, context);
        }

        // Step 6b: Virtual property enrichment (FEAT-005)
        result = virtualPropertyEnricher.enrich(result, context, ontologyType);

        // Step 7: Cache result
        if (shouldCache && cacheKey != null) {
            cacheService.put(cacheKey, cachedResult, options.cacheTtlSeconds());
        }

        // Step 8: Convert and return
        responseObserver.onNext(resultConverter.toProto(result));
        responseObserver.onCompleted();
    } catch (OQLSyntaxException e) {
        responseObserver.onError(Status.INVALID_ARGUMENT...);
    } catch (QueryExecutionException e) {
        responseObserver.onError(Status.INTERNAL...);
    }
}

Exception layered mapping:

Exception	gRPC Status	Meaning
`OQLSyntaxException`	`INVALID_ARGUMENT`	OQL syntax error (client issue)
`OQLSemanticException`	`INVALID_ARGUMENT`	OQL semantic error (type mismatch)
`QueryExecutionException`	`INTERNAL`	Query execution failure (server issue)

#3. StorageRouter: Federation Determination

Java

if (router.requiresFederation(plan)) {
    result = router.executeFederatedQuery(plan, context);
} else {
    result = executor.execute(plan, context);
}

StorageRouter analyzes the physical execution plan to determine whether a query requires cross-engine federation:

Doris: Large-scale OLAP queries, aggregations, full-text search
DuckDB: Small dataset embedded analytics, ad-hoc queries
Federation: Cross-engine JOIN or UNION queries

The explainQuery method outputs which engines a query involves through the engines field — invaluable for performance estimation and debugging.

#4. Caching Strategy: Aggregation-Aware Query Cache

Java

boolean isAggregation = plan instanceof PhysicalAggregatePlan;
boolean shouldCache = options.enableCache()
        && cacheService.isEnabled()
        && (!isAggregation || cacheService.isCacheAggregationsEnabled());

Special handling for aggregation queries: Aggregation results may become stale as data changes, so isCacheAggregationsEnabled() provides an independent toggle. By default, regular queries are cacheable, but aggregation caching requires explicit enablement.

Cache key generation includes three dimensions: worldId (data isolation), query (query text), and params (pagination parameters), ensuring results from different worlds, queries, and pages never collide.

#5. VirtualPropertyEnricher: Derived Property Filling

Java

// Step 6b: Enrich with virtual properties (FEAT-005)
String ontologyType = ast.from().entityType();
result = virtualPropertyEnricher.enrich(result, context, ontologyType);

VirtualPropertyEnricher executes after query execution but before cache writing, filling virtual property values computed by Reasoning & Decision Layer's DerivedPropertyEngine. The comment "Zero overhead if no virtual properties are defined for this type" highlights the performance consideration.

#6. Streaming Queries: executeQueryStream

Java

@Override
public void executeQueryStream(ExecuteQueryRequest request,
        StreamObserver<QueryResultRow> responseObserver) {
    try {
        var ast = parser.parse(request.getQuery());
        var context = queryConverter.toQueryContext(request);
        var plan = optimizer.optimize(ast, context);
        var result = executor.execute(plan, context);

        long rowIndex = 0;
        for (var row : result.rows()) {
            var protoRow = resultConverter.toProtoResultRow(row, rowIndex++);
            responseObserver.onNext(protoRow);
        }
        responseObserver.onCompleted();
    } catch (Exception e) {
        responseObserver.onError(Status.INTERNAL...);
    }
}

Server Streaming RPC: Unlike executeQuery which returns the complete result set, executeQueryStream uses gRPC server-side streaming, sending QueryResultRow messages one at a time. This is ideal for large result sets — clients do not need to wait for all data to load into memory.

#7. Query Validation and Plan Explanation

#7.1 Query Validation

Java

@Override
public void validateQuery(ValidateQueryRequest request,
        StreamObserver<ValidateQueryResponse> responseObserver) {
    try {
        var ast = parser.parse(request.getQuery());
        responseBuilder.setValid(true)
                .setStructure(QueryStructure.newBuilder()
                        .setQueryType("SELECT")
                        .addFromTypes(ast.from().entityType()).build());
    } catch (OQLSyntaxException e) {
        responseBuilder.setValid(false)
                .addErrors(QueryValidationError.newBuilder()
                        .setErrorType("SYNTAX_ERROR")
                        .setLine(firstError.line())
                        .setColumn(firstError.column()).build());
    }
}

Precise error location: Syntax errors return line and column, enabling IDE/editor-level error pinpointing.

#7.2 Query Plan Explanation

Java

@Override
public void explainQuery(ExplainQueryRequest request,
        StreamObserver<ExplainQueryResponse> responseObserver) {
    var ast = parser.parse(request.getQuery());
    var plan = optimizer.optimize(ast, context);

    // Plan text + engines + cost estimate
    var cost = QueryCostEstimate.newBuilder()
            .setEstimatedResultRows(plan.estimatedRows())
            .setCostScore(plan.estimatedCost())
            .build();
}

Three-dimensional output: explainQuery returns execution plan text (planText), involved storage engines (engines), and cost estimation (estimatedRows + costScore).

#8. Vector Search and Graph Traversal: OQL Unified Abstraction

#8.1 Vector Search

Java

private String buildVectorSearchOQL(VectorSearchRequest request) {
    // SELECT * FROM Entity WHERE SIMILAR_TO(field, [vector], threshold) LIMIT topK
    sb.append(" WHERE SIMILAR_TO(").append(request.getVectorField())
      .append(", [").append(vectorStr).append("], ")
      .append(request.getMinSimilarity()).append(")");
    sb.append(" LIMIT ").append(request.getTopK());
    return sb.toString();
}

OQL unified abstraction: Vector search is not a separate execution path — it is converted to an OQL query with a SIMILAR_TO function, reusing the entire parse-optimize-route-execute pipeline.

#8.2 Graph Traversal

Java

private String buildGraphTraversalOQL(GraphTraversalRequest request) {
    // SELECT * FROM Entity WHERE CONNECTED_TO(startId, relationType, minDepth, maxDepth)
    sb.append(" WHERE CONNECTED_TO('").append(request.getStartIds(0))
      .append("', '").append(pattern.getRelationTypes(0))
      .append("', 1, ").append(request.getMaxDepth()).append(")");
    return sb.toString();
}

Graph traversal is similarly converted to OQL with CONNECTED_TO(startId, relationType, minDepth, maxDepth), covering all scenarios from direct relationships to multi-hop traversal.

#9. Time Travel Queries: Reserved Interface Design

Java

@Override
public void queryAtTimestamp(QueryAtTimestampRequest request,
        StreamObserver<ExecuteQueryResponse> responseObserver) {
    responseObserver.onError(Status.UNIMPLEMENTED
            .withDescription("Time travel queries not yet implemented")
            .asRuntimeException());
}

Three time-travel methods (queryAtTimestamp, queryAtVersion, queryDiff) currently return UNIMPLEMENTED. This is standard practice for gRPC interface reservation — Proto definitions and gRPC stubs are ready, awaiting Nessie integration completion.

#10. Async Query State Management

Java

private record AsyncQueryState(
    QueryStatus status, QueryResult result, String error
) {}

private enum QueryStatus {
    PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
}

AsyncQueryState is managed via ConcurrentHashMap, supporting status tracking for long-running queries. Five states form a complete lifecycle: PENDING -> RUNNING -> COMPLETED/FAILED, with CANCELLED triggerable at any stage.

#11. Key Takeaways

Four-stage pipeline: Parse (OQL to AST), Optimize (AST to Plan), Route (Doris/DuckDB/Federation), Execute — each stage handled by an independent collaborator
Aggregation-aware caching: Regular queries cacheable by default; aggregation caching requires explicit enablement to avoid stale data
OQL unified abstraction: Vector search (SIMILAR_TO) and graph traversal (CONNECTED_TO) are converted to OQL queries, reusing the complete pipeline
Derived property enrichment: VirtualPropertyEnricher inserted after execution, before caching — zero overhead when no virtual properties exist
Streaming results: executeQueryStream uses gRPC server streaming for row-by-row delivery, ideal for large result sets
Precise error location: Syntax errors return line and column numbers for IDE-level positioning
Interface reservation: Time travel queries return UNIMPLEMENTED with Proto and stubs ready

#Next Article

S9-06: OQL Parser — From Text to Execution Plan. We will dive into OQL's hand-written lexer and recursive descent parser to see how it transforms query text into an AST supporting 30+ keywords and 6 condition types.

Tags: #coomia-dip #source-code-reading #data-Layer #query-federation #doris #duckdb #oql #caching #vector-search #graph-traversal