Build unified data access layer across vector, graph, and relational stores

Introduction

When your application has to combine vector similarity search, knowledge-graph traversal, and relational lookups to answer a single user request, you face a choice: scatter pgvector / Neo4j / PostgreSQL calls across every service, or hide all three behind one access layer. Teams that pick the first path discover too late that adding a fourth store means rewriting connection handling, error mapping, and health checks in every consumer — turning a one-line feature into a multi-week migration and leaving outages to cascade store-by-store. This lesson shows how to build a unified data access layer that hides the three query languages, three connection pools, and three failure modes behind a single DataQuery abstraction, a capability-aware QueryRouter, and shared health/audit/metrics infrastructure. By the end you'll be able to add a new backing store without touching caller code and fall back gracefully when any individual store is unhealthy.

Key Terminology

• Polyglot persistence: an architecture that uses multiple specialized data stores (pgvector, Neo4j, PostgreSQL) side-by-side because no single engine optimally serves vector similarity, graph traversal, and relational access patterns.
• DataQuery abstraction: the typed Pydantic envelope that carries query_type, parameters, target_store, and limits so callers issue one shape regardless of which backend ultimately executes the query.
• Capability-based routing: the QueryRouter strategy of mapping each query_type (vector_similarity, graph_traversal, relational, hybrid) to the store whose get_capabilities() advertises support, with automatic fallback when DataStoreHealthChecker marks a store unhealthy.

Concepts

The Multi-Store Reality of GenAI Systems

Production GenAI systems almost never use a single data store. The access patterns demanded by modern AI applications are too diverse for any one database technology to handle optimally. Vector similarity search requires specialized indexing structures like HNSW or IVFFlat that relational databases were never designed for. Knowledge graph traversals need native graph storage with index-free adjacency for efficient multi-hop queries. Relational data demands ACID transactions, complex joins, and the mature query optimization that PostgreSQL has refined over decades.

The result is a polyglot persistence architecture where pgvector handles embedding storage and similarity search, Neo4j manages entity relationships and knowledge graphs, and PostgreSQL stores structured metadata, user records, and configuration. Each store excels at its specific access pattern, but the application code that ties them together faces a combinatorial complexity problem: every service that needs data must understand three different query languages, three connection pools, three error handling patterns, and three health monitoring strategies.

Health Monitoring Infrastructure

The DataStoreHealthChecker runs periodic probes against all stores. For pgvector, it executes a lightweight similarity query against a known canary embedding to verify both connectivity and vector index functionality. For Neo4j, it runs a simple connectivity check. For PostgreSQL, it executes SELECT 1 as a basic liveness probe.

Health status is stored in a dedicated table with columns for store name, type, status (healthy or unhealthy), latency in milliseconds, last checked timestamp, and consecutive failure count. The consecutive failure count is critical for avoiding false positives -- a single failed health check might indicate a transient network issue, but three consecutive failures indicate a genuine outage requiring alerting and fallback activation.

When a store's health check fails, the DataStoreFallbackRouter activates degraded-mode routing. Vector similarity queries fall back to PostgreSQL full-text search using tsvector and ts_rank(), which provides approximate semantic matching. Graph traversal queries fall back to relational JOIN queries on an entity_relationships table that mirrors the graph structure in tabular form. These fallbacks are slower and less precise, but they keep the system operational during store outages.

Prometheus Instrumentation

Every data query emits metrics:

• data_query_latency_seconds{store_type, query_type} histogram tracks the distribution of query execution times, enabling SLA monitoring and performance regression detection.
• data_query_results_count{store_type} histogram tracks result set sizes, useful for detecting queries that return too many or too few results.
• data_store_health_status{store_name} gauge reports the current health of each store as 1 (healthy) or 0 (unhealthy).
• data_query_routing_decisions_total{query_type, routed_to} counter tracks routing patterns, revealing which stores handle the most traffic.

These metrics feed into Alertmanager rules that fire when any store's health status drops to 0, when query latency exceeds SLA thresholds, or when routing patterns shift unexpectedly (indicating potential misconfiguration).

Query Audit Logging

The QueryAuditLogger records every data query in a dedicated audit log table. Each entry captures the query ID, a hash of the query template (for pattern grouping), the query template itself, the store type, execution latency, result count, execution timestamp, and a hash of the parameters. This audit log serves two purposes: compliance auditing (who queried what data and when) and performance analysis (which query patterns are most frequent and expensive). The pattern analysis capability feeds directly into the caching and materialization strategies covered later in this chapter.

Code Walkthrough

This walkthrough demonstrates the two pillars of the unified layer: the DataQuery envelope plus DataStoreInterface contract that every caller and adapter share, then the concrete adapters and QueryRouter that translate intent into store-native execution.

Foundational Abstractions: DataQuery and DataStoreInterface

The unified data access layer begins with a single typed envelope for every query, paired with the contract every store adapter must implement:

Code snippetpython
1from abc import ABC, abstractmethod
2from pydantic import BaseModel, Field
3from typing import Optional, Literal
4
5class DataQuery(BaseModel):
6    query_id: str = Field(..., description="Unique identifier for this query")
7    query_type: Literal["vector_similarity", "graph_traversal", "relational", "hybrid"] = Field(
8        ..., description="Type of query determining routing"
9    )
10    parameters: dict = Field(default_factory=dict, description="Query-specific parameters")
11    target_store: Optional[str] = Field(None, description="Explicit store target, auto-detected if None")
12    max_results: int = Field(default=10, ge=1, le=1000)
13    timeout_ms: int = Field(default=5000, ge=100, le=30000)
14    consistency_level: Literal["eventual", "strong"] = Field(default="eventual")
15
16class DataStoreInterface(ABC):
17    @abstractmethod
18    def execute_query(self, query: DataQuery) -> dict:
19        """Execute a query and return normalized results."""
20
21    @abstractmethod
22    def health_check(self) -> dict:
23        """Return health status including latency and connectivity."""
24
25    @abstractmethod
26    def get_capabilities(self) -> list[str]:
27        """Return list of supported query types."""
28
29    @abstractmethod
30    def estimate_latency(self, query: DataQuery) -> float:
31        """Estimate execution time in milliseconds for the given query."""

• DataQuery lines 5-13: The query_type literal constrains routing to four supported patterns (hybrid enables cross-store fan-out); parameters carries store-specific arguments; target_store allows explicit routing override; default caps put results at 10, timeout at 5 seconds, and consistency at eventual for optimal latency.
• DataStoreInterface lines 15-30: A deliberately minimal contract — every store must execute queries, report health, declare capabilities, and estimate latency. The latency estimate feeds the hybrid planner's choice between sequential and parallel execution.

The query_type field drives routing decisions. When a caller specifies vector_similarity, the router directs the query to pgvector; graph_traversal goes to Neo4j; hybrid enables cross-store queries that fan out and merge results. Leaving target_store as None lets the router make the optimal decision based on declared capabilities.

Store Adapters and the Query Router

Each concrete adapter translates DataQuery parameters into store-native syntax — SQL with vector operators for pgvector, Cypher for Neo4j, parameterized SQL for PostgreSQL — and normalizes results into a common shape (results, total_count, execution_time_ms, store_type) so callers never need to know which store produced them. The QueryRouter then dispatches each incoming query to the right adapter based on query_type, honoring an explicit target_store override when present:

Code snippetpython
1class PgVectorStore(DataStoreInterface):
2    def execute_query(self, query: DataQuery) -> dict:
3        embedding = query.parameters["embedding"]
4        collection = query.parameters.get("collection", "embeddings")
5        # SELECT *, embedding <=> $1 AS distance
6        # FROM {collection}
7        # ORDER BY embedding <=> $1
8        # LIMIT $2
9
10class Neo4jStore(DataStoreInterface):
11    def execute_query(self, query: DataQuery) -> dict:
12        # MATCH (n:{label})-[r:{relationship}]->(m)
13        # WHERE n.name = $name
14        # RETURN n, r, m LIMIT $limit
15
16class QueryRouter:
17    def __init__(self, stores: dict[str, DataStoreInterface]):
18        self.stores = stores
19        self.routing_table = {
20            "vector_similarity": "pgvector",
21            "graph_traversal": "neo4j",
22            "relational": "postgresql",
23        }
24
25    def route(self, query: DataQuery) -> DataStoreInterface:
26        if query.target_store:
27            return self.stores[query.target_store]
28        store_key = self.routing_table[query.query_type]
29        return self.stores[store_key]

• PgVectorStore (lines 1-8): Extracts the embedding and target collection from parameters; the commented SQL shows the cosine distance operator (<=>) ranking and limiting results.
• Neo4jStore (lines 10-14): Mirrors the same pattern with a Cypher MATCH that traverses named relationships, filters by property, and limits results.
• QueryRouter (lines 16-29): Initialised with a registry of named store instances and a routing table that maps each query type to its optimal store. The route method checks for an explicit target_store override first, then falls back to the routing table — letting new backends register without any caller-side change.

For hybrid queries, the router fans out to multiple stores. The HybridQueryPlanner analyzes the query parameters to choose between sequential execution (when graph traversal depends on vector results) and parallel execution via asyncio.gather() (when stores can be queried independently), reducing total latency to the maximum of the individual store latencies rather than their sum.

You'll know it works when a single DataQuery routes correctly across all three stores, hybrid queries fan out and merge results, and an unhealthy backend triggers the fallback router without taking down the platform.

Do's and Don'ts

Do's

1. ✓Do express every cross-store request as a DataQuery with an explicit query_type literal and leave target_store as None — the QueryRouter resolves the right adapter from its routing_table, so registering a new backing store requires only a new adapter entry there, with zero changes in any calling service.
2. ✓Do implement all four DataStoreInterface methods — including estimate_latency — in every adapter — the HybridQueryPlanner reads estimate_latency to decide between asyncio.gather() and sequential execution; an adapter that returns a stub or raises here breaks the parallel-vs-sequential decision for every hybrid query that touches that store.
3. ✓Do route dependent hybrid sub-queries through the sequential path and independent ones through asyncio.gather() — when a Cypher MATCH needs node IDs produced by the preceding embedding <=> vector lookup, sequential ordering is mandatory; reserving asyncio.gather() for truly independent fan-outs collapses total latency to the slowest individual store rather than the sum of all stores.

Don'ts

1. ✗Don't issue raw embedding <=> $1 SQL or Cypher MATCH statements directly from application services — bypassing the DataStoreInterface contract re-scatters store-specific query languages across every caller, so adding a fourth store or replacing pgvector requires hunting down and rewriting every direct connection rather than dropping in a new adapter.
2. ✗Don't hardcode target_store on every DataQuery as a permanent routing strategy — when an explicit target_store is set, QueryRouter.route() skips the routing_table entirely and cannot redirect to a healthy fallback; callers that always pin to "pgvector" or "neo4j" nullify the graceful-degradation behavior the router exists to provide.
3. ✗Don't treat get_capabilities() as a documentation nicety — the QueryRouter uses the capability list returned by each adapter to validate that a store can actually serve a given query_type; omitting a capability or returning an empty list causes the router to silently mis-route queries to an adapter that will reject them at execution time rather than at dispatch time.