Database Federation

When database size and query workloads outgrow a single server, sharding (horizontal partitioning of tables) is one solution. However, sharding still requires all tables to share a common horizontal key structure. What happens when your database contains completely distinct functional domains, such as User Billing, Catalog Management, and Search Recommendations? Database Federation solves this by splitting a monolithic database vertically by function and placing them on separate servers, then providing a single query coordinator to query across them as if they were a single logical store.

1. Learning Objectives

• Differentiate between Database Federation (Functional Splitting) and Database Sharding.
• Understand the Wrapper/Adapter pattern and Foreign Data Wrappers (FDW).
• Master the execution of application-level joins and query aggregation.
• Analyze Query Pushdown optimization and why it is critical for performance.
• Evaluate the trade-offs of schema independence, transactions, and latency under federation.
• Implement a fully functional Federated Query Engine with query pushdown and in-memory hash joins in Java, Python, and C++.

2. Prerequisites

To get the most out of this lesson, you should review the following topics first:

• Databases and DBMS: Standard SQL schemas, queries, and join mechanics.
• Database Sharding: Horizontal partitioning by row ranges.
• ACID & BASE: Transactions and consistency guarantees in distributed environments.

3. Why This Topic Matters

In a monolithic system, all tables (e.g. users, orders, inventory, billing) reside in a single database instance. This setup makes writing query joins simple: a single database process accesses the physical data blocks on disk and joins them in memory.

However, as traffic scales, this monolith becomes a major bottleneck:

• Lock Contention: Heavy writes on the orders table lock database indices, blocking read-heavy queries on users or inventory.
• Index Bloat: The database must hold index nodes for all tables in RAM. As sizes grow, indexes spill to disk, reducing cache hit rates.
• Blast Radius: A corrupted table or CPU spike in billing takes down the entire database, crashing all application services.

Database Federation addresses this by splitting the monolith into functional databases (e.g., a Users DB, an Orders DB, and a Products DB). To keep the architecture clean, a Federated Coordinator or query engine sits in front of these databases. It exposes a single database endpoint, allowing clients to query across them without knowing that the data is stored in separate physical engines.

4. Real-world Analogy

Imagine a large national university system with separate colleges: the College of Engineering, the College of Medicine, and the College of Business.

Each college operates its own student record database with different software and schemas. The engineering dean manages engineering students, and the medical dean manages medical students.

When the university President needs a report on all students who are enrolled in double majors (e.g., Bio-Medical Engineering), they cannot query a single system. Instead, they send a request to a Central Registrar. The Central Registrar contacts the College of Engineering for its student list, contacts the College of Medicine for its student list, and merges the two lists in their office to find matching student names. To the President, the registrar acts as a single, unified coordinator.

5. Core Concepts

• Functional Splitting (Federation): Splitting database tables vertically by business domain. The resulting databases (e.g. Users DB, Products DB) run on completely separate physical servers.
• Federated Database System (FDBS): A system that consolidates multiple autonomous, heterogeneous database instances into a single logical database interface. The underlying databases can run different engines (e.g., Postgres, MySQL, MongoDB).
• Wrappers & Adapters (Foreign Data Wrappers): Standardized connector interfaces (like PostgreSQL's postgres_fdw) that translate queries from the federated coordinator into the native syntax of the target database.
• Application-Level Join: Performing a relational join in application memory because the target tables reside on different physical database instances that cannot talk directly to each other.
• Query Pushdown: An optimization technique where the federated coordinator pushes filters (WHERE clauses) and projections (SELECT columns) down to the target database nodes. This filters out irrelevant data at the source, minimizing network latency.

6. Visualizations

Monolithic vs. Federated Architecture

Federated Query Lifecycle (With Pushdown)

7. How It Works Step-by-Step

Let's walk through how a federated query engine resolves a join across two separate databases:

1. Query Submission: The client sends a SQL query to the Federated Coordinator:

Query Parsing & Analysis: The coordinator parses the SQL statement. By checking its schema catalog, it recognizes that users is hosted on physical database server A (Postgres) and orders is on server B (MySQL).

Pushdown Planning: The optimizer extracts the filter WHERE users.active = true. Instead of fetching all users and filtering them in the coordinator, it decides to push the filter down to Server A.

Sub-query Execution: The coordinator formats and executes two independent sub-queries in parallel:

• To Server A: SELECT id, username FROM users WHERE active = true;
• To Server B: SELECT user_id, price FROM orders;

In-Memory Hash Join: The coordinator reads the results from Server A and builds an in-memory hash table of active users indexed by their id. It then streams the records from Server B, matching each order's user_id against the hash table.

Client Return: The coordinator aggregates the joined rows and streams the formatted result set back to the client application.

8. Internal Architecture

A federated query system is composed of the following key components:

• Query Parser & Planner: Validates SQL queries, parses the abstract syntax tree (AST), and identifies the destination node for each table.
• Federated Catalog: A database tracking schema metadata. It maps logical database tables to their physical connection strings and target database types.
• Adapters / Connectors: Translates standardized coordinator commands into the native dialects of the target databases (e.g., translating SQL to MongoDB query language).
• Execution Engine: Spawns parallel worker threads to fetch remote data, manages memory buffers, and executes in-memory joins and sorting.

9. Request Lifecycle

Let's examine how write and read request lifecycles behave under a federated database setup:

Write Request Lifecycle

When an application client registers a user and places an initial order, it cannot run a single ACID transaction. The writes must be routed separately:

1. The application sends a write request containing user data to the Federated Coordinator.
2. The coordinator writes the record to the Users DB locally, returning a new user_id.
3. The application then issues a write query containing the order data to the coordinator.
4. The coordinator writes the order record (incorporating the user_id) to the Orders DB and returns success.

Note: If the second write fails, the application must handle the failure and rollback the first write manually (using compensating transactions) since there is no native cross-database transaction locking.

10. Deep Dive

A. Query Pushdown Optimization Mechanics

Without pushdown optimization, database federation is slow. Consider joining a users table (10 million rows) and an orders table (100 million rows) on separate servers to find orders placed by a single user named "Bob":

If the coordinator cannot perform a pushdown, it must fetch all 10 million user rows and all 100 million order rows across the network into its own memory space, perform the filter name = 'Bob', and then join them. This will likely crash the coordinator due to memory exhaustion.

With Pushdown: The coordinator pushes the filter name = 'Bob' down to the Users DB. The Users DB executes the query locally using its indexes and returns only 1 row (Bob's user ID). The coordinator then queries the Orders DB for orders matching Bob's user_id, reducing network transfer and processing overhead.

B. Application-Level Hash Joins

When joining tables from separate databases, the coordinator must execute the join in memory. The most efficient algorithm for this is the Hash Join:

• Build Phase: The coordinator fetches the rows from the smaller table (e.g., the filtered users list) and builds a hash map in memory using the join key (user_id) as the hash key.
• Probe Phase: The coordinator fetches rows from the larger table (e.g., the orders list) and streams them one by one. For each order row, it looks up the user_id in the hash map. If a match is found, it merges the data and yields the joined row.

C. Schema Mapping & Drift

Because the databases in a federated system are independent, they are prone to schema drift. If a developer updates a table column name in the Orders DB (e.g., renaming user_id to customer_id) without updating the coordinator's catalog registry, queries will fail. Federated systems require strict tooling (like schema registries or shared migrations) to coordinate updates across nodes.

11. Production Examples

• Presto / Trino: A distributed SQL query engine designed to run fast queries against heterogeneous data sources. A single Presto query can join data from a Hive data lake, a PostgreSQL operational store, and an Elasticsearch index in a single statement.
• PostgreSQL Foreign Data Wrappers (FDW): PostgreSQL supports SQL/MED (SQL Management of External Data), allowing a Postgres server to mount tables from external Postgres, MySQL, or Oracle servers as if they were local tables.
• MySQL Federated Storage Engine: A MySQL storage engine that points to tables residing on a remote MySQL database server, translating local writes and reads into remote SQL socket calls.

12. Advantages

• Zero Lock Contention Across Domains: Writes to the Orders DB cannot block reads from the Users DB since they run on different engines.
• Heterogeneous Database Support: Allows you to use the best tool for the job: storing user relations in PostgreSQL, order logs in Cassandra, and catalog data in MongoDB, while keeping a unified query interface.
• Clean Domain Boundaries: Simplifies the transition to microservices by enforcing functional database isolation early.

13. Limitations

• High Network Latency: Queries joining large tables across the network suffer from network transit bottlenecks.
• Loss of Cross-Database Integrity: Foreign key constraints cannot be checked across database servers by the engines themselves.
• Single Point of Failure (SPOF): If the coordinator node crashes, the client cannot query any of the underlying databases, even if they are all healthy.

14. Trade-offs

Database Federation vs. Database Sharding

Federation (Vertical Splitting by Function) splits the schema by table purpose (e.g. Users DB vs Orders DB). This is best when query patterns map to isolated domains and rarely cross boundaries.

Sharding (Horizontal Splitting by Rows) replicates the exact same schema across all nodes, splitting rows by a key range or modulo. This is best when a single table is too large for one server's disk space. High-scale systems often use both: federating the database by domain, and then sharding the orders database horizontally across multiple servers.

15. Performance Considerations

• Network Bandwidth Exhaustion: Streaming millions of records to a central coordinator can saturate network interfaces. Always use filters to limit query sizes.
• Coordinator Memory Limits: In-memory hash joins store the smaller table in RAM. If both tables are large, the coordinator can run out of memory. When this happens, the coordinator must spill data to disk (using merge-sort joins), which degrades performance.

16. Failure Scenarios

• Tail Latency Bottlenecks: If a federated query hits 3 databases and one node is running a backup (experiencing high disk wait times), the entire coordinator query is delayed.
• Schema Drift Crashes: If a downstream database drops a column that the coordinator is configured to read, queries will fail with SQL execution errors.

17. Best Practices

• Design application domain boundaries carefully to minimize queries that join tables across databases.
• Replicate small lookup tables (e.g. countries or currency_rates) to all databases to allow joins to be processed locally on individual nodes.
• Use query pushdown optimizations (like filtering and projections) on all queries.
• Implement circuit breakers to fail fast if a downstream database node is slow or unresponsive.

18. Common Mistakes

• Over-federating: Splitting databases too early can lead to high network latency and complex query logic.
• Assuming Distributed Transactions work: Do not rely on multi-database transactions. Build systems to tolerate eventual consistency.

19. Implementation (Database Federation Engine)

Below is a complete, production-grade simulation of a Federated Query Engine Coordinator. It connects to two mock databases (Users DB and Orders DB), analyzes incoming queries, applies query pushdown filters to the sources, and performs an in-memory Hash Join to merge the results.

20. Interview Questions & Answers

Q1. What is Query Pushdown in database federation, and why is it important?

Answer: Query Pushdown is an optimization where the federated coordinator shifts query filters, columns selections, and sorting logic down to the target database nodes.

Without pushdown, the coordinator must stream all remote table rows across the network and process them in its own memory. This degrades query performance and risks memory exhaustion. Pushing filters down enables the remote engines to filter rows locally using indexes, transferring only relevant results across the network.

Q2. How does Database Federation differ from Database Sharding?

Answer:

• Database Federation splits the system vertically by business function. Tables with different schemas reside on different physical servers (e.g. Users DB vs. Products DB).
• Database Sharding splits the system horizontally by row range. The same schema is duplicated across all physical instances, but each instance contains a unique subset of rows.

Q3. How do you handle database joins across a federated database?

Answer: Standard SQL engines cannot run joins across separate servers. Joins must be implemented in application memory:

1. Query the smaller table (e.g. users) and build a hash map of records keyed by the join column in coordinator memory (Build Phase).
2. Stream the larger table (e.g. orders) and match each row against the hash map (Probe Phase).
3. Aggregate and format the matching rows to return the joined result set.

21. Practice Exercises

• Exercise 1 (Easy): Draw a diagram representing the query execution path for joining a MongoDB table and a MySQL table using a federated coordinator.
  Answer: The coordinator connects to MongoDB using a MongoDB connector (translating SQL to JSON query strings) and to MySQL using a JDBC/ODBC adapter. It executes sub-queries in parallel, receives the documents and SQL rows, converts them to a common schema format, and joins them in memory.
• Exercise 2 (Medium): Write a SQL query demonstrating Postgres FDW setup to mount a remote table named remote_orders from server 192.168.1.50 to a local schema.
  Answer:

Exercise 3 (Hard): Implement a Python script simulating a federated merge-join algorithm. It reads pre-sorted user rows from UserDB and pre-sorted order rows from OrderDB, and merges them using two pointers with $O(1)$ auxiliary space complexity.

22. Challenge Problem

Federated Join Optimization: You are designing a federated query optimizer. The system needs to join Table A (1,000 rows) and Table B (5,000,000 rows) on separate servers.

Write an analysis detailing:

• Why a standard Broadcast Hash Join (fetching Table A, caching it, and streaming Table B) is preferred over a Shuffled Hash Join.
• What the coordinator should do if Table A grows to 5,000,000 rows, making a Broadcast Hash Join trigger an Out-Of-Memory (OOM) crash.
• How the optimizer should use historical query execution statistics to dynamically select the correct join algorithm (Nested Loops vs. Hash vs. Merge Join).

23. Summary

Database Federation is a vertical splitting strategy that divides monolithic databases into functionally isolated instances. While federation reduces CPU and write contention and simplifies database schema management, it moves the query join logic to the application layer. Implementing query pushdown optimization is critical to keep query performance fast and minimize network latency.

24. Cheat Sheet

Metric	Monolithic DB	Sharded DB	Federated DB
Splitting Axis	None (Single node)	Horizontal (Rows split by key)	Vertical (Functional schema split)
Join Performance	Excellent (Local engine)	Poor (Cross-shard joins needed)	Moderate (Requires coordinator joins)
Referential Integrity	Strict SQL Foreign Keys	Loss across shards	Loss across domains
Write Contention	High lock contention	Distributed across shard nodes	Isolated by business domain

25. Quiz

26. Further Reading

• Presto: SQL on Everything (2019) — Trino/Presto Core Engineering Team.
• PostgreSQL postgres_fdw Documentation: Postgres Foreign Data Wrapper guide.
• Designing Data-Intensive Applications (Chapter 10: Batch Processing & Federated Engines) — Martin Kleppmann.

27. Next Lesson Preview

We have completed Module 3: Databases & Data Modeling, learning how to scale, sharding, replication, consistent hashing, and database federation. In the next module, we move on to Module 4: Architecture & Communication. We will start with the foundational building block of modern web systems: Client-Server Architecture.