PostgreSQL Full-Text Search: FTS vs pgvector vs Elasticsearch

Search is a core feature of almost every content application. The question is not whether to implement it, but how to balance setup complexity, operational overhead, search quality, and cost.

PostgreSQL Native FTS

PostgreSQL has built-in full-text search using tsvector and tsquery types. It supports stemming, ranking, and phrase search—sufficient for most applications without additional infrastructure.

-- Add FTS index to posts
ALTER TABLE posts ADD COLUMN fts_vector tsvector
  GENERATED ALWAYS AS (
    setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
    setweight(to_tsvector('english', coalesce(summary, '')), 'B') ||
    setweight(to_tsvector('english', coalesce(content, '')), 'C')
  ) STORED;

CREATE INDEX idx_posts_fts ON posts USING GIN(fts_vector);

-- Search query
SELECT title, ts_rank(fts_vector, query) AS rank
FROM posts, to_tsquery('english', 'kubernetes & deployment') query
WHERE fts_vector @@ query
ORDER BY rank DESC LIMIT 10;

pgvector: Semantic Search

Unlike keyword search, semantic search finds conceptually related content even when exact keywords don't match. A search for "scaling horizontally" should find articles about "adding more servers," "load balancing," and "distributed systems." pgvector makes this possible using embeddings.

# Generate embedding for search query
query_embedding = openai.embeddings.create(
    model="text-embedding-3-small",
    input="scaling horizontally"
).data[0].embedding

# Find semantically similar articles
sql = (
    "SELECT title, 1 - (embedding <=> %s::vector) AS similarity "
    "FROM posts WHERE status = 'Published' "
    "ORDER BY embedding <=> %s::vector LIMIT 10"
)
results = db.execute(sql, (query_embedding, query_embedding)).fetchall()

Hybrid Search: Best of Both

Combine keyword search (precision for specific terms) with semantic search (recall for concepts) using Reciprocal Rank Fusion. Weight keyword matches higher for technical queries; weight semantic similarity higher for natural language questions.

When to Choose Elasticsearch

Elasticsearch is warranted when: you need complex faceted search (filter by multiple fields simultaneously), very large document volumes (>50M), or multi-language search. Its operational overhead (dedicated cluster, index management) isn't worth it for most applications that can serve search from PostgreSQL.

Production-Grade Python Implementation Example

To demonstrate these concepts, here is a complete, production-grade Python block showing proper error boundary management, type safety annotations, and context lifecycle handling:

import logging
import time
from typing import Generator, Any, Dict, Optional
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("MirahLabs.ProductionTelemetry")

class ProductionServiceException(Exception):
    """Custom domain exception for pipeline operations."""
    pass

def with_telemetry(operation_name: str):
    """Decorator to log latency, parameters, and handle exception boundaries."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.perf_counter()
            logger.info(f"Starting execution of {operation_name} with params: {args}, {kwargs}")
            try:
                result = func(*args, **kwargs)
                elapsed = time.perf_counter() - start_time
                logger.info(f"Successfully completed {operation_name} in {elapsed:.4f} seconds.")
                return result
            except Exception as e:
                elapsed = time.perf_counter() - start_time
                logger.error(f"Failed execution of {operation_name} after {elapsed:.4f}s: {str(e)}")
                raise ProductionServiceException(f"Pipeline error in {operation_name}") from e
        return wrapper
    return decorator

class DataPipelineProcessor:
    def __init__(self, config: Dict[str, Any]) -> None:
        self.config = config
        self.is_active = True

    @with_telemetry("process_data_payload")
    def process_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        if not self.is_active:
            raise ProductionServiceException("Processor is deactivated.")
        if "id" not in payload:
            raise ValueError("Payload missing mandatory key: 'id'")
        
        # Simulating domain-specific calculations
        processed_data = {**payload, "status": "processed", "timestamp": time.time()}
        return processed_data

# Example Usage
if __name__ == "__main__":
    pipeline = DataPipelineProcessor(config={"mode": "production"})
    try:
        pipeline.process_payload({"id": "evt_10928a", "value": 42.0})
    except ProductionServiceException:
        pass

Production Trade-offs & Implementation Decisions

Deploying this solution in production environments requires a careful analysis of the trade-offs involved. For instance, focusing purely on consistency (such as ACID compliance) can limit network throughput and horizontal scalability. On the other hand, adopting an eventual consistency model can lead to dirty reads and requires complex conflict resolution strategies in the application layer.

At MirahLabs, our engineering teams balance these architectural constraints by separating critical transaction paths from analytics workloads. We apply message-driven architectures with idempotent consumer systems to guarantee that network failures or retries do not result in double processing or state contamination.

Real-World Benchmarks & Resource Planning

Below is a typical performance comparison profile compiled by our engineering team in staging environments under simulated loads (10k concurrent virtual users):

When capacity planning, we recommend scaling out horizontally using containerized workloads rather than vertically upgrading underlying instance models. This maximizes uptime and provides cost efficiency through dynamic scaling policies.

Security Considerations & Vulnerability Mitigations

No production blueprint is complete without addressing security. Ensure that all data paths utilize encryption in transit (TLS 1.3) and at rest (using AES-256). Furthermore, implement strict Role-Based Access Control (RBAC) to limit operations. For APIs, always enforce rate limits (e.g. using token bucket algorithms in Redis) and run continuous static application security testing (SAST) in your CI pipeline.

How MirahLabs Applies This in Practice

Our experience building high-volume solutions like MirahCare.ai and Ayurveda.ai has taught us that early optimization is often a trap, but ignoring structural security and data design early leads to fatal development blocks. We design all client products from day one to support modular extensions, robust query indexing, and standard schema definitions, ensuring rapid iteration without technical debt growth.

Production-Grade Python Implementation Example

To demonstrate these concepts, here is a complete, production-grade Python block showing proper error boundary management, type safety annotations, and context lifecycle handling:

import logging
import time
from typing import Generator, Any, Dict, Optional
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("MirahLabs.ProductionTelemetry")

class ProductionServiceException(Exception):
    """Custom domain exception for pipeline operations."""
    pass

def with_telemetry(operation_name: str):
    """Decorator to log latency, parameters, and handle exception boundaries."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.perf_counter()
            logger.info(f"Starting execution of {operation_name} with params: {args}, {kwargs}")
            try:
                result = func(*args, **kwargs)
                elapsed = time.perf_counter() - start_time
                logger.info(f"Successfully completed {operation_name} in {elapsed:.4f} seconds.")
                return result
            except Exception as e:
                elapsed = time.perf_counter() - start_time
                logger.error(f"Failed execution of {operation_name} after {elapsed:.4f}s: {str(e)}")
                raise ProductionServiceException(f"Pipeline error in {operation_name}") from e
        return wrapper
    return decorator

class DataPipelineProcessor:
    def __init__(self, config: Dict[str, Any]) -> None:
        self.config = config
        self.is_active = True

    @with_telemetry("process_data_payload")
    def process_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
        if not self.is_active:
            raise ProductionServiceException("Processor is deactivated.")
        if "id" not in payload:
            raise ValueError("Payload missing mandatory key: 'id'")
        
        # Simulating domain-specific calculations
        processed_data = {**payload, "status": "processed", "timestamp": time.time()}
        return processed_data

# Example Usage
if __name__ == "__main__":
    pipeline = DataPipelineProcessor(config={"mode": "production"})
    try:
        pipeline.process_payload({"id": "evt_10928a", "value": 42.0})
    except ProductionServiceException:
        pass

Production Trade-offs & Implementation Decisions

Deploying this solution in production environments requires a careful analysis of the trade-offs involved. For instance, focusing purely on consistency (such as ACID compliance) can limit network throughput and horizontal scalability. On the other hand, adopting an eventual consistency model can lead to dirty reads and requires complex conflict resolution strategies in the application layer.

At MirahLabs, our engineering teams balance these architectural constraints by separating critical transaction paths from analytics workloads. We apply message-driven architectures with idempotent consumer systems to guarantee that network failures or retries do not result in double processing or state contamination.

Real-World Benchmarks & Resource Planning

Below is a typical performance comparison profile compiled by our engineering team in staging environments under simulated loads (10k concurrent virtual users):

When capacity planning, we recommend scaling out horizontally using containerized workloads rather than vertically upgrading underlying instance models. This maximizes uptime and provides cost efficiency through dynamic scaling policies.

Security Considerations & Vulnerability Mitigations

No production blueprint is complete without addressing security. Ensure that all data paths utilize encryption in transit (TLS 1.3) and at rest (using AES-256). Furthermore, implement strict Role-Based Access Control (RBAC) to limit operations. For APIs, always enforce rate limits (e.g. using token bucket algorithms in Redis) and run continuous static application security testing (SAST) in your CI pipeline.

How MirahLabs Applies This in Practice

Our experience building high-volume solutions like MirahCare.ai and Ayurveda.ai has taught us that early optimization is often a trap, but ignoring structural security and data design early leads to fatal development blocks. We design all client products from day one to support modular extensions, robust query indexing, and standard schema definitions, ensuring rapid iteration without technical debt growth.