Full-Text Search

# Python example with psycopg 3. Credentials match the container from
# "Try it locally" above.
import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()

# Create the table this lesson works with, and seed three articles.
# Everything from here on queries these same rows.
cur.execute("""
    CREATE TABLE IF NOT EXISTS articles (
        id      SERIAL PRIMARY KEY,
        title   TEXT,
        content TEXT
    )
""")
cur.execute("SELECT count(*) FROM articles")
if cur.fetchone()[0] == 0:
    cur.execute("""
        INSERT INTO articles (title, content) VALUES
            ('Introduction to Databases',
             'Databases are systems for storing data...'),
            ('Running PostgreSQL',
             'Learn how to run and manage PostgreSQL instances...'),
            ('Database Design Patterns',
             'Common patterns for designing efficient databases...')
    """)
conn.commit()

# The naive approach: substring matching. Use ILIKE rather than LIKE,
# because case sensitivity is the one weakness SQL fixes for free and
# there is no point arguing against a strawman.
def ilike_search(term):
    cur.execute("""
        SELECT title
        FROM articles
        WHERE content ILIKE %s
        ORDER BY id
    """, (f'%{term}%',))
    return [row[0] for row in cur.fetchall()]


for search_term in ("database", "stored", "designs"):
    hits = ilike_search(search_term)
    print(f"ILIKE '%{search_term}%' -> {len(hits)} of 3")
    for title in hits:
        print(f"    {title}")

# Python with PostgreSQL FTS
import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()

# Search with ranking using @@ operator and ts_rank()
search_query = "database"
cur.execute("""
    SELECT
        title,
        ts_rank(search_vector, to_tsquery('english', %s)) AS rank
    FROM articles
    WHERE search_vector @@ to_tsquery('english', %s)
    -- Ties are common on short documents, and a tie makes the order
    -- arbitrary. Pin it with the primary key so the output is stable.
    ORDER BY rank DESC, id
    LIMIT 10
""", (search_query, search_query))

results = cur.fetchall()
for title, rank in results:
    print(f"{title} (score: {rank:.4f})")

import psycopg

conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
cur = conn.cursor()


def show(label, sql, arg):
    cur.execute(sql, (arg,))
    titles = [row[0] for row in cur.fetchall()]
    print(f"{label:<34} {len(titles)} hit(s)")
    for t in titles:
        print(f"    {t}")


TSQUERY = """
    SELECT title FROM articles
    WHERE search_vector @@ to_tsquery('english', %s)
    ORDER BY id"""

PHRASE = """
    SELECT title FROM articles
    WHERE search_vector @@ phraseto_tsquery('english', %s)
    ORDER BY id"""

# & both terms, | either term, ! exclude
show("AND    'database & design'",   TSQUERY, "database & design")
show("OR     'postgresql | mysql'",  TSQUERY, "postgresql | mysql")
show("NOT    'database & !design'",  TSQUERY, "database & !design")

# phraseto_tsquery requires the words ADJACENT and IN ORDER. The last
# two lines are the same two words either way round: as a phrase the
# reversed pair matches nothing, as an AND it still matches.
show("PHRASE 'design patterns'",     PHRASE,  "design patterns")
show("PHRASE 'patterns design'",     PHRASE,  "patterns design")
show("AND    'patterns & design'",   TSQUERY, "patterns & design")

from elasticsearch import Elasticsearch

# The container from "Try it locally" runs with xpack.security.enabled
# =false, so no credentials are needed and none are sent here.
es = Elasticsearch(['http://localhost:9200'])

# A real cluster does have security on, and then you pass credentials:
#   es = Elasticsearch(['https://localhost:9200'],
#                      basic_auth=('elastic', 'your-password'),
#                      ca_certs='/path/to/http_ca.crt')
# Passing basic_auth to a security-disabled cluster is harmless, it is
# simply ignored, which makes a wrong password look like it worked.

info = es.info()
print(f"Connected to: {info['cluster_name']}")

from elasticsearch import Elasticsearch

# Define index mapping (schema)
index_mapping = {
    "mappings": {
        "properties": {
            "title": {
                "type": "text",
                "analyzer": "english"  # Stemming, stop words
            },
            "content": {
                "type": "text",
                "analyzer": "english"
            },
            "author": {
                "type": "keyword"  # For exact match/filtering
            },
            "publish_date": {
                "type": "date"
            },
            "view_count": {
                "type": "integer"  # Used by the function_score example later
            },
            "tags": {
                "type": "keyword"  # For faceted search
            }
        }
    }
}

es = Elasticsearch(["http://localhost:9200"])
es.indices.create(index="articles", body=index_mapping)
print("Index created successfully")

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk


es = Elasticsearch(["http://localhost:9200"])

# Index single document
doc = {
    "title": "Introduction to Databases",
    "content": "Databases are systems for storing and retrieving data efficiently...",
    "author": "John Doe",
    "publish_date": "2025-01-15",
    "view_count": 1500,
    "tags": ["databases", "tutorial", "beginner"]
}

es.index(index="articles", id=1, document=doc)

# Bulk indexing for better performance (10-100x faster).
# Every document carries publish_date and view_count: the function_score
# example later scores on both, and Elasticsearch rejects the query
# outright if a field it references is missing from the mapping.
docs = [
    {
        "_index": "articles",
        "_id": i,
        "_source": {
            "title": f"Article {i}",
            "content": f"Content for article {i} about databases...",
            "author": "Jane Smith",
            "publish_date": "2025-01-01",
            "view_count": (i * 7) % 500,
            "tags": ["python", "databases"]
        }
    }
    for i in range(2, 1000)
]

success, failed = bulk(es, docs)
es.indices.refresh(index="articles")   # make them searchable immediately
print(f"Indexed {success} documents")

from elasticsearch import Elasticsearch

# Simple match query
query = {
    "query": {
        "match": {
            "content": "database systems"
        }
    }
}

es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="articles", body=query)

print(f"Found {response['hits']['total']['value']} results\n")
for hit in response['hits']['hits']:
    print(f"Score: {hit['_score']:.2f}")
    print(f"Title: {hit['_source']['title']}")
    print(f"Snippet: {hit['_source']['content'][:80]}...\n")

from elasticsearch import Elasticsearch


# Search with typos - "databse systms" instead of "database systems"
fuzzy_query = {
    "query": {
        "match": {
            "content": {
                "query": "databse systms",  # Two typos!
                "fuzziness": "AUTO"         # Levenshtein distance edit
            }
        }
    }
}

es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="articles", body=fuzzy_query)

print(f"Found {response['hits']['total']['value']} results despite typos")
for hit in response['hits']['hits'][:3]:
    print(f"  - {hit['_source']['title']} (score: {hit['_score']:.2f})")

from elasticsearch import Elasticsearch


# Boost title matches over content matches
boosted_query = {
    "query": {
        "multi_match": {
            "query": "database design",
            "fields": [
                "title^3",      # 3x boost for title matches
                "content^1"     # Normal weight for content
            ]
        }
    }
}

es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="articles", body=boosted_query)

for hit in response['hits']['hits'][:4]:
    print(f"{hit['_score']:.2f}: {hit['_source']['title']}")

from elasticsearch import Elasticsearch


# Combine text relevance with popularity/recency
function_score_query = {"query": {
    "function_score": {
        "query": {
            "match": {"content": "database"}
        },
        "functions": [
            {
                # Boost newer articles (exponential decay)
                "exp": {
                    "publish_date": {
                        "origin": "2025-02-05",
                        "scale": "30d",
                        "decay": 0.5
                    }
                },
                "weight": 2
            },
            {
                # Boost articles with more views (logarithmic)
                "field_value_factor": {
                    "field": "view_count",
                    "modifier": "log1p",
                    "factor": 0.1
                }
            }
        ],
        "score_mode": "sum",  # Add all function scores
        "boost_mode": "multiply"  # Multiply with text score
    }
}, "size": 3}

es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="articles", body=function_score_query)

for hit in response['hits']['hits']:
    src = hit['_source']
    print(f"{hit['_score']:.4f}  {src['title']:<28} views={src['view_count']}")

from elasticsearch import Elasticsearch


# Search with facets (like Amazon's sidebar filters)
faceted_query = {
    "query": {
        "match": {"content": "database"}
    },
    "aggs": {
        "by_author": {
            "terms": {"field": "author"}
        },
        "by_tags": {
            "terms": {"field": "tags", "size": 10}
        },
        "by_year": {
            "date_histogram": {
                "field": "publish_date",
                "calendar_interval": "year"
            }
        }
    },
    "size": 10  # Return top 10 results
}

es = Elasticsearch(["http://localhost:9200"])
response = es.search(index="articles", body=faceted_query)

# Display facets
print("Authors:")
for bucket in response['aggregations']['by_author']['buckets']:
    print(f"  {bucket['key']}: {bucket['doc_count']} articles")

print("\nTags:")
for bucket in response['aggregations']['by_tags']['buckets']:
    print(f"  {bucket['key']}: {bucket['doc_count']} articles")

# Write to both PostgreSQL and Elasticsearch
import psycopg
from elasticsearch import Elasticsearch

pg_conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
es = Elasticsearch(['http://localhost:9200'])

def create_article(title, content, author):
    """Insert into PostgreSQL and index in Elasticsearch"""

    # 1. Insert into PostgreSQL (source of truth)
    with pg_conn.cursor() as cur:
        cur.execute("""
            INSERT INTO articles (title, content, author, created_at)
            VALUES (%s, %s, %s, NOW())
            RETURNING id
        """, (title, content, author))
        article_id = cur.fetchone()[0]
        pg_conn.commit()

    # 2. Index in Elasticsearch (for search)
    es.index(
        index="articles",
        id=article_id,
        document={
            "title": title,
            "content": content,
            "author": author
        }
    )

    return article_id

# Usage
article_id = create_article(
    "New Database Tutorial",
    "Learn PostgreSQL from scratch...",
    "John Doe"
)
print(f"Created article {article_id} in both systems")

# Better approach: Use CDC to sync PostgreSQL -> Elasticsearch.
# PostgreSQL stays the single source of truth, Debezium streams its
# write-ahead log to Kafka, and this consumer applies changes to
# Elasticsearch. The application writes to ONE system.
import json

from elasticsearch import Elasticsearch
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "postgres.public.articles",                 # Debezium topic: prefix.schema.table
    bootstrap_servers=["localhost:29092"],
    auto_offset_reset="earliest",               # replay from the start of the topic
    value_deserializer=lambda m: json.loads(m.decode("utf-8")) if m else None,
)

es = Elasticsearch(["http://localhost:9200"])

# A SEPARATE index, and not by accident. cdc-pg is a different database
# with its own id sequence, so writing these events into the "articles"
# index used earlier would have two unrelated id spaces colliding in one
# place: cdc-pg's row 1 would overwrite article 1, and deleting cdc-pg's
# row 1 would delete article 1 outright. Mirror one source per index.
INDEX = "articles_cdc"

for message in consumer:
    if message.value is None:                   # tombstone record, sent after a delete
        continue

    # Debezium wraps events as {"schema": ..., "payload": ...} unless you
    # disable schemas. Reading change["op"] directly raises KeyError.
    change = message.value["payload"]
    op = change["op"]

    if op in ("c", "r"):                        # c = insert, r = snapshot read
        after = change["after"]
        es.index(index=INDEX, id=after["id"], document=after)
        label = "r (snapshot)" if op == "r" else "c (create)"
        print(f"op={label} id={after['id']}  title={after['title']!r}")

    elif op == "u":
        after = change["after"]
        es.update(index=INDEX, id=after["id"], doc=after)
        print(f"op=u (update)  id={after['id']}  title={after['title']!r}")

    elif op == "d":
        before = change["before"]
        es.delete(index=INDEX, id=before["id"], ignore=[404])
        print(f"op=d (delete)  id={before['id']}")

import psycopg
from elasticsearch import Elasticsearch

# Both clients are created once, at module level, not per request. A
# psycopg connection and an Elasticsearch client are both reusable, and
# rebuilding either inside the function would add a TCP handshake to
# every search.
pg_conn = psycopg.connect("host=localhost dbname=demo user=demo password=demo")
es = Elasticsearch(["http://localhost:9200"])


def search_articles_with_full_data(search_query, user_id):
    """
    1. Search in Elasticsearch (fast full-text search)
    2. Get full data from PostgreSQL (with user-specific data)
    """

    # Step 1: Get IDs from Elasticsearch
    es_query = {
        "query": {"match": {"content": search_query}},
        "_source": False,  # Don't return content, just IDs
        "size": 50
    }

    response = es.search(index="articles", body=es_query)
    article_ids = [hit['_id'] for hit in response['hits']['hits']]

    if not article_ids:
        return []

    # Step 2: Get full data from PostgreSQL with user context
    with pg_conn.cursor() as cur:
        cur.execute("""
            SELECT
                a.id,
                a.title,
                a.author,
                a.created_at,
                b.title AS bookmark_title
            FROM articles a
            LEFT JOIN bookmarks b ON b.article_id = a.id AND b.user_id = %s
            WHERE a.id = ANY(%s)
            ORDER BY array_position(%s, a.id)  -- Preserve ES ranking
        """, (user_id, article_ids, article_ids))

        return cur.fetchall()

# Usage. Note the search term: "database" appears in every document in
# this index, so its IDF collapses to zero and it contributes nothing to
# the score. "systems" appears in one, which is what actually ranks.
results = search_articles_with_full_data("database systems", user_id=42)
print(f"Elasticsearch ranked {len(results)} of them back from PostgreSQL:\n")
for article in results:
    bookmark = "*" if article[4] else " "
    print(f"{bookmark} {article[1]} by {article[2]}")