“

Series: S3 Data Foundation · Article 8 | Level: Advanced | Reading Time: 20 min

Query Federation: Unified Cross-Engine Queries

Tags: #QueryFederation #CrossEngine #Doris #DuckDB #Elasticsearch #coomia-dip

#TL;DR

The coomia-dip platform distributes data across multiple storage engines: Doris (OLAP analytics), DuckDB (embedded lightweight queries), Elasticsearch (full-text search), and Iceberg (cold data archival). Users should not care where data is stored — a single OQL query should automatically route to the optimal engine and aggregate results. This article fully dissects the Query Federation architecture, query routing strategies, cross-engine JOIN implementations, data transfer optimizations, and consistency guarantees. By benchmarking against Palantir Foundry's unified data access layer, we demonstrate how to build transparent federated query capabilities atop heterogeneous storage.

#1. Why Query Federation

#1.1 The Heterogeneous Storage Reality

Code

coomia-dip Storage Engine Distribution:

+-------------+--------------------+--------------------+
| Engine       | Data Type           | Best For            |
+-------------+--------------------+--------------------+
| Apache Doris | Hot data, real-time | Sub-second OLAP     |
| DuckDB       | Ad-hoc, small sets  | Zero-latency embed  |
| Elasticsearch| Full-text, logs     | Fuzzy search, NLP   |
| Iceberg      | Cold data, archive  | Time travel, schema |
| MinIO (S3)   | Objects, files      | Large unstructured  |
+-------------+--------------------+--------------------+

Problem: A single Entity's properties may span 3 engines
  - Core properties    -> Doris entity_common table
  - Full-text content  -> Elasticsearch inverted index
  - Historical versions-> Iceberg snapshots

#1.2 Palantir Foundry's Approach

Code

Foundry's Unified Data Access Layer:

+------------------------------------------+
|              Foundry API                  |
|  (Users see only Dataset abstraction)     |
+------------------------------------------+
|          Unified Query Engine             |
|  (Auto-selects optimal storage backend)   |
+--------+--------+--------+---------------+
| Spark  | Trino  | ES     | Foundry SQL   |
+--------+--------+--------+---------------+

coomia-dip Mapping:
  Foundry Dataset   -> Ontology Entity Type
  Foundry API       -> OQL
  Unified Engine    -> Query Federation Layer
  Spark/Trino       -> Doris + DuckDB

#2. Federation Architecture

#2.1 Layered Architecture

Code

Query Federation Layered Architecture:

+------------------------------------------+
|           OQL Query Interface             |
|  (User submits OQL query)                 |
+------------------------------------------+
|           Query Planner                   |
|  (Parse -> AST -> Logical -> Physical)    |
+------------------------------------------+
|         Federation Coordinator            |
|  (Split -> Route -> Execute -> Merge)     |
+------------------------------------------+
|         Engine Adapters                   |
|  +--------+--------+--------+----------+ |
|  | Doris  | DuckDB | ES     | Iceberg  | |
|  |Adapter |Adapter |Adapter |Adapter   | |
|  +--------+--------+--------+----------+ |
+------------------------------------------+
|         Data Transfer Layer               |
|  (Arrow Flight high-perf data transfer)   |
+------------------------------------------+

#2.2 Query Routing Decisions

Python

class QueryRouter:
    """Query router: decides which engine handles each query"""

    def __init__(self, catalog: DataCatalog):
        self._catalog = catalog

    def route(self, plan: LogicalPlan) -> RoutingDecision:
        """Select optimal engine based on query characteristics"""
        features = self._analyze_query_features(plan)

        # Rule 1: Full-text search -> Elasticsearch
        if features.has_full_text_search:
            return RoutingDecision(
                primary_engine=Engine.ELASTICSEARCH,
                reason="Query contains full-text search predicates"
            )

        # Rule 2: Time travel -> Iceberg (via Doris Catalog)
        if features.has_time_travel:
            return RoutingDecision(
                primary_engine=Engine.ICEBERG,
                reason="Query requires time-travel snapshots"
            )

        # Rule 3: Small dataset + complex computation -> DuckDB
        if features.estimated_rows < 100_000 and features.complexity > 0.7:
            return RoutingDecision(
                primary_engine=Engine.DUCKDB,
                reason="Small dataset with complex computation"
            )

        # Rule 4: Large dataset + OLAP -> Doris
        if features.estimated_rows >= 100_000:
            return RoutingDecision(
                primary_engine=Engine.DORIS,
                reason="Large dataset OLAP query"
            )

        # Default -> Doris
        return RoutingDecision(
            primary_engine=Engine.DORIS,
            reason="Default routing"
        )

    def route_federated(self, plan: LogicalPlan) -> FederatedPlan:
        """Handle queries requiring cross-engine execution"""
        sub_plans = self._split_plan(plan)
        assignments = {}

        for sub_plan in sub_plans:
            engine = self.route(sub_plan).primary_engine
            assignments[sub_plan.id] = engine

        return FederatedPlan(
            sub_plans=sub_plans,
            assignments=assignments,
            join_strategy=self._choose_join_strategy(sub_plans)
        )

#3. Engine Adapters

#3.1 Adapter Interface

Python

from abc import ABC, abstractmethod
from typing import AsyncIterator
import pyarrow as pa

class EngineAdapter(ABC):
    """Engine adapter base class"""

    @abstractmethod
    async def execute(
        self, plan: PhysicalPlan
    ) -> AsyncIterator[pa.RecordBatch]:
        """Execute physical plan, return Arrow RecordBatch stream"""
        ...

    @abstractmethod
    def translate(self, plan: LogicalPlan) -> str:
        """Translate logical plan to engine-native query language"""
        ...

    @abstractmethod
    def estimate_cost(self, plan: LogicalPlan) -> QueryCost:
        """Estimate query cost"""
        ...

    @abstractmethod
    def get_statistics(self, entity_type: str) -> TableStatistics:
        """Get table statistics"""
        ...

#3.2 Doris Adapter

Python

class DorisAdapter(EngineAdapter):
    """Apache Doris OLAP engine adapter"""

    def __init__(self, connection_pool: DorisConnectionPool):
        self._pool = connection_pool

    def translate(self, plan: LogicalPlan) -> str:
        compiler = DorisSQLCompiler()
        return plan.accept(compiler)

    async def execute(
        self, plan: PhysicalPlan
    ) -> AsyncIterator[pa.RecordBatch]:
        sql = self.translate(plan.logical_plan)

        async with self._pool.acquire() as conn:
            flight_client = await conn.get_flight_client()
            ticket = await flight_client.execute(sql)

            async for batch in flight_client.do_get(ticket):
                yield batch

    def estimate_cost(self, plan: LogicalPlan) -> QueryCost:
        sql = self.translate(plan)
        explain = self._pool.execute_sync(f"EXPLAIN {sql}")
        return self._parse_explain(explain)

#3.3 DuckDB Adapter

Python

class DuckDBAdapter(EngineAdapter):
    """DuckDB embedded analytics engine adapter"""

    def __init__(self, db_path: str = ":memory:"):
        self._conn = duckdb.connect(db_path)
        self._setup_iceberg_extension()

    def _setup_iceberg_extension(self):
        """Load Iceberg extension to read tables from MinIO"""
        self._conn.execute("INSTALL iceberg; LOAD iceberg;")
        self._conn.execute("INSTALL httpfs; LOAD httpfs;")
        self._conn.execute("""
            SET s3_endpoint='minio.internal:9000';
            SET s3_access_key_id='minioadmin';
            SET s3_secret_access_key='minioadmin';
            SET s3_use_ssl=false;
            SET s3_url_style='path';
        """)

    def translate(self, plan: LogicalPlan) -> str:
        compiler = DuckDBSQLCompiler()
        return plan.accept(compiler)

    async def execute(
        self, plan: PhysicalPlan
    ) -> AsyncIterator[pa.RecordBatch]:
        sql = self.translate(plan.logical_plan)
        result = self._conn.execute(sql)

        while True:
            batch = result.fetch_arrow_table()
            if batch.num_rows == 0:
                break
            for record_batch in batch.to_batches(max_chunksize=8192):
                yield record_batch

#3.4 Elasticsearch Adapter

Python

class ElasticsearchAdapter(EngineAdapter):
    """Elasticsearch full-text search adapter"""

    def __init__(self, es_client: AsyncElasticsearch):
        self._es = es_client

    def translate(self, plan: LogicalPlan) -> dict:
        """Translate to Elasticsearch DSL"""
        compiler = ESDSLCompiler()
        return plan.accept(compiler)

    async def execute(
        self, plan: PhysicalPlan
    ) -> AsyncIterator[pa.RecordBatch]:
        es_query = self.translate(plan.logical_plan)
        index = self._resolve_index(plan.entity_type)

        async for hits in self._scroll_search(index, es_query):
            batch = self._hits_to_arrow(hits)
            yield batch

    def _hits_to_arrow(self, hits: list[dict]) -> pa.RecordBatch:
        """Convert ES hits to Arrow RecordBatch"""
        columns = {}
        for hit in hits:
            source = hit['_source']
            for key, value in source.items():
                columns.setdefault(key, []).append(value)

        arrays = [pa.array(values) for values in columns.values()]
        names = list(columns.keys())
        return pa.RecordBatch.from_arrays(arrays, names=names)

#4. Cross-Engine JOIN

#4.1 JOIN Strategy Selection

Code

Cross-Engine JOIN Strategy Selection:

+-----------------+--------------------+-------------------+
| Strategy         | Best For            | Implementation     |
+-----------------+--------------------+-------------------+
| Broadcast Join   | One side small      | Broadcast small    |
|                  | (<10MB)             | table to large     |
+-----------------+--------------------+-------------------+
| Semi-Join Push   | Post-filter small   | Execute filter     |
|                  |                     | first, push IDs    |
+-----------------+--------------------+-------------------+
| Hash Join        | Both sides medium   | Pull to coordinator|
| (at Coordinator) | (<1M rows each)    | for local hash join|
+-----------------+--------------------+-------------------+
| Sort-Merge Join  | Both sides large    | Streaming merge,   |
|                  | and sorted          | low memory         |
+-----------------+--------------------+-------------------+
| Materialized     | Frequent cross-     | Pre-compute into   |
| View             | engine JOINs        | single engine      |
+-----------------+--------------------+-------------------+

#4.2 Semi-Join Pushdown Optimization

Python

class SemiJoinPushdown:
    """Semi-Join pushdown optimizer"""

    async def execute(
        self,
        left_adapter: EngineAdapter,
        left_plan: PhysicalPlan,
        right_adapter: EngineAdapter,
        right_plan: PhysicalPlan,
        join_key: str
    ) -> AsyncIterator[pa.RecordBatch]:
        # Step 1: Execute query on smaller side, extract JOIN keys
        small_side = await self._determine_small_side(
            left_adapter, left_plan,
            right_adapter, right_plan
        )

        if small_side == 'left':
            keys = await self._extract_keys(
                left_adapter, left_plan, join_key
            )
            # Step 2: Push key list to right side as IN filter
            right_plan_filtered = self._inject_in_filter(
                right_plan, join_key, keys
            )
            # Step 3: Execute filtered query on right side
            right_results = right_adapter.execute(right_plan_filtered)
            # Step 4: Local hash join at coordinator
            left_results = left_adapter.execute(left_plan)
            async for batch in self._local_hash_join(
                left_results, right_results, join_key
            ):
                yield batch

#5. Federation Coordinator

#5.1 Query Execution Flow

Code

Federated Query Execution Flow:

1. User submits OQL:
   FETCH Person
   SELECT name, age, description, historical_role
   WHERE name LIKE '%Zhang%' AND age > 30
   AT TIME '2024-01-01'

2. Query Analyzer identifies data sources:
   - name, age        -> Doris (entity_common)
   - description      -> Elasticsearch (full-text)
   - LIKE '%Zhang%'   -> Elasticsearch (tokenized search)
   - AT TIME          -> Iceberg (time travel)

3. Query splitting:
   Sub-Query A (ES):    Search name LIKE '%Zhang%' -> entity_id list
   Sub-Query B (Doris): SELECT name, age FROM entity_common
                         WHERE entity_id IN (...) AND age > 30
   Sub-Query C (Iceberg): Read historical_role at 2024-01-01 snapshot

4. Execution plan:
   [ES: full-text search] --> entity_id list
                                 |
   [Doris: property query] <-- IN filter (Semi-Join Push)
                                 |
   [Iceberg: historical]   <-- entity_id list
                                 |
   [Coordinator: Hash Join] --> final result

#5.2 Parallel Execution Engine

Python

class FederationCoordinator:
    """Federated query coordinator"""

    def __init__(self, adapters: dict[Engine, EngineAdapter]):
        self._adapters = adapters

    async def execute(self, federated_plan: FederatedPlan) -> pa.Table:
        dag = self._build_execution_dag(federated_plan)
        execution_order = dag.topological_sort()

        results: dict[str, pa.Table] = {}

        for level in execution_order:
            # Sub-queries at same level execute in parallel
            tasks = []
            for sub_plan_id in level:
                sub_plan = federated_plan.sub_plans[sub_plan_id]
                engine = federated_plan.assignments[sub_plan_id]
                adapter = self._adapters[engine]

                enriched_plan = self._inject_upstream_results(
                    sub_plan, results
                )
                tasks.append(self._execute_sub_plan(
                    adapter, enriched_plan, sub_plan_id
                ))

            level_results = await asyncio.gather(*tasks)
            for sub_plan_id, result in zip(level, level_results):
                results[sub_plan_id] = result

        return self._merge_results(results, federated_plan)

#6. Data Transfer Optimization

#6.1 Arrow Flight Transport

Code

Cross-Engine Data Transfer Protocol Comparison:

+----------------+----------+----------+----------+
| Protocol        | Throughput| Latency  | Serializ |
+----------------+----------+----------+----------+
| JDBC/ODBC       | ~100 MB/s| Medium   | Row-wise |
| Arrow Flight    | ~2 GB/s  | Low      | Zero-copy|
| Arrow Flight SQL| ~1.5 GB/s| Low      | Zero-copy|
| gRPC Protobuf   | ~500 MB/s| Low      | Protobuf |
+----------------+----------+----------+----------+

Choice: Arrow Flight SQL
  - Native Doris integration
  - Zero-copy memory mapping
  - Columnar transfer naturally fits analytical queries

#6.2 Compression and Batching

Python

class DataTransferOptimizer:
    """Data transfer optimizer"""

    async def transfer_with_compression(
        self,
        source: AsyncIterator[pa.RecordBatch],
        target_engine: Engine
    ) -> AsyncIterator[pa.RecordBatch]:
        """Transfer data with engine-appropriate compression"""
        buffer = []
        buffer_size = 0

        async for batch in source:
            buffer.append(batch)
            buffer_size += batch.nbytes

            if buffer_size >= self._config.batch_size_bytes:
                merged = pa.Table.from_batches(buffer)

                if target_engine == Engine.DORIS:
                    compressed = self._compress_lz4(merged)
                elif target_engine == Engine.DUCKDB:
                    compressed = merged  # Local, no compression needed
                else:
                    compressed = self._compress_zstd(merged)

                for out_batch in compressed.to_batches(
                    max_chunksize=self._config.chunk_size
                ):
                    yield out_batch

                buffer = []
                buffer_size = 0

        if buffer:
            merged = pa.Table.from_batches(buffer)
            for out_batch in merged.to_batches(
                max_chunksize=self._config.chunk_size
            ):
                yield out_batch

#7. Consistency and Transactions

#7.1 Snapshot Isolation

Code

Federated Query Consistency Model:

Problem: Data may change during cross-engine query execution
  - t0: Read Person list from Doris
  - t1: Person data is updated
  - t2: Read description from ES
  -> Inconsistent: Person list is t0 version, description is t2 version

Solution: Snapshot Isolation

+------------------------------------------+
| Federation Snapshot Manager               |
|                                           |
| 1. Acquire global snapshot timestamp      |
|    T_snap at query start                  |
| 2. All sub-queries read at T_snap         |
|                                           |
| Doris:   Read MVCC version at T_snap      |
| Iceberg: Use snapshot_id for T_snap       |
| ES:      Use Point-in-Time API            |
| DuckDB:  Direct read (embedded, no conc.) |
+------------------------------------------+

#7.2 Snapshot Management

Python

class SnapshotManager:
    """Federated query snapshot manager"""

    async def create_snapshot(self) -> FederatedSnapshot:
        """Create cross-engine consistent snapshot"""
        timestamp = datetime.utcnow()

        doris_snapshot = await self._doris.get_snapshot_at(timestamp)
        iceberg_snapshot = await self._iceberg.get_snapshot_at(timestamp)
        es_pit = await self._es.open_point_in_time(
            index="entity_*",
            keep_alive="5m"
        )

        return FederatedSnapshot(
            timestamp=timestamp,
            doris_snapshot=doris_snapshot,
            iceberg_snapshot_id=iceberg_snapshot.snapshot_id,
            es_point_in_time=es_pit,
            ttl=timedelta(minutes=5)
        )

    async def release_snapshot(self, snapshot: FederatedSnapshot):
        """Release snapshot resources"""
        await self._es.close_point_in_time(snapshot.es_point_in_time)

#8. Query Cache and Materialization

#8.1 Result Cache

Python

class FederatedQueryCache:
    """Federated query result cache"""

    def __init__(self, redis_client: Redis, max_cache_size_mb: int = 512):
        self._redis = redis_client
        self._max_size = max_cache_size_mb * 1024 * 1024

    async def get_or_execute(
        self,
        query_hash: str,
        executor: Callable[[], Awaitable[pa.Table]]
    ) -> pa.Table:
        cached = await self._redis.get(f"fed_cache:{query_hash}")
        if cached:
            return self._deserialize(cached)

        result = await executor()

        serialized = self._serialize(result)
        if len(serialized) < self._max_size // 100:
            await self._redis.setex(
                f"fed_cache:{query_hash}",
                timedelta(minutes=5),
                serialized
            )

        return result

#8.2 Automatic Materialization Recommendations

Code

Materialized View Auto-Recommendation:

Monitoring system detects:
  - Query Q1 (Doris + ES JOIN) executes 500 times/day
  - Average latency 2.3 seconds
  - Result set < 50MB

Recommendation:
  CREATE MATERIALIZED VIEW mv_person_with_desc AS
  SELECT p.entity_id, p.name, p.age, es.description
  FROM entity_common p
  JOIN es_entity_index es ON p.entity_id = es.entity_id
  WHERE p.entity_type = 'Person'
  REFRESH EVERY 5 MINUTES;

Impact:
  - Query time: 2.3s -> 50ms
  - Cross-engine JOIN -> single-engine query
  - Cost: 50MB extra storage + refresh every 5 minutes

#9. Monitoring and Observability

#9.1 Federation Query Tracing

Code

Distributed Trace Example:

Trace ID: fed-20240301-abc123
Total Duration: 1.2s

+-- [0-50ms]   Query Parse & Plan
|   +-- Lexer: 2ms
|   +-- Parser: 5ms
|   +-- Semantic Analysis: 15ms
|   +-- Federation Planning: 28ms
|
+-- [50-800ms] Sub-Query Execution (parallel)
|   +-- [50-200ms] ES: full-text search -> 150 entity_ids
|   +-- [50-600ms] Doris: SELECT ... WHERE IN (...) -> 120 rows
|   +-- [50-800ms] Iceberg: time-travel read -> 120 rows
|
+-- [800-1100ms] Cross-Engine Join
|   +-- Hash Build: 50ms (ES + Iceberg results)
|   +-- Hash Probe: 200ms (Doris results)
|   +-- Projection: 50ms
|
+-- [1100-1200ms] Result Serialization & Transfer
    +-- Arrow Flight: 100ms (120 rows, 48KB)

#9.2 Performance Metrics

Code

Federation Query Key Metrics:

+--------------------+----------+----------+----------+
| Metric              | P50      | P95      | P99      |
+--------------------+----------+----------+----------+
| Single-engine       | 50ms     | 200ms    | 500ms    |
| Two-engine federated| 200ms    | 800ms    | 2s       |
| Three-engine feder. | 500ms    | 1.5s     | 3s       |
| Cross-engine xfer   | 10ms     | 100ms    | 500ms    |
| JOIN overhead        | 50ms     | 200ms    | 1s       |
| Cache hit rate       | 65%      | -        | -        |
+--------------------+----------+----------+----------+

#10. Testing Strategy

#10.1 Engine Mock Testing

Python

class TestQueryFederation:

    @pytest.fixture
    def mock_adapters(self):
        doris = MockDorisAdapter(data={
            'Person': [
                {'entity_id': '1', 'name': 'Zhang San', 'age': 35},
                {'entity_id': '2', 'name': 'Li Si', 'age': 28},
            ]
        })
        es = MockESAdapter(data={
            'Person': [
                {'entity_id': '1', 'description': 'Senior Engineer'},
                {'entity_id': '2', 'description': 'Product Manager'},
            ]
        })
        return {'doris': doris, 'es': es}

    async def test_cross_engine_join(self, mock_adapters):
        coordinator = FederationCoordinator(mock_adapters)
        result = await coordinator.execute(
            parse("FETCH Person SELECT name, description WHERE age > 30")
        )
        assert len(result) == 1
        assert result[0]['name'] == 'Zhang San'
        assert result[0]['description'] == 'Senior Engineer'

    async def test_routing_full_text_to_es(self, mock_adapters):
        router = QueryRouter(catalog)
        plan = parse("FETCH Person WHERE description LIKE '%Engineer%'")
        decision = router.route(plan)
        assert decision.primary_engine == Engine.ELASTICSEARCH

#Key Takeaways

Query Federation hides storage heterogeneity from users: a single OQL query automatically routes to Doris, DuckDB, Elasticsearch, or Iceberg — users never need to know where data physically resides.
Query routing strategy is the performance key: automatic routing based on data volume, query complexity, and storage characteristics outperforms manual engine selection.
Cross-engine JOINs require careful strategy selection: Broadcast Join, Semi-Join Push, Hash Join, and Sort-Merge Join each suit different scenarios; the Federation Coordinator auto-selects based on statistics.
Arrow Flight is the optimal cross-engine data transfer protocol: zero-copy columnar transfer delivers 15-20x the throughput of traditional JDBC.
Snapshot consistency is the foundation of federated queries: cross-engine queries must use a unified snapshot timestamp, otherwise results will be inconsistent.

#Next Article

Next up: S3-09 "Query Optimization: From Logical Plan to Physical Execution" will dive into single-engine query optimization, including predicate pushdown, column pruning, partition pruning, and Doris materialized view auto-matching.

Tags: #QueryFederation #CrossEngine #Doris #DuckDB #Elasticsearch #Iceberg #ArrowFlight #SnapshotIsolation #FederatedJoin #coomia-dip #DataFoundation