Full-Text Search & Elasticsearch

from fastapi import FastAPI, HTTPException

app = FastAPI()

SQL = """
    SELECT id, name, description,
           ts_rank(search_vec, plainto_tsquery('english', %s)) AS rank
    FROM   products
    WHERE  search_vec @@ plainto_tsquery('english', %s)
    ORDER  BY rank DESC
    LIMIT  20
"""


@app.get("/search")
async def search(q: str = ""):
    if not q:
        raise HTTPException(400, "missing query param q")

    # plainto_tsquery converts plain text safely (no special chars needed);
    # websearch_to_tsquery also supports AND/OR/-term syntax.
    # The query text is a BOUND PARAMETER, never interpolated (chapter 08).
    async with pool.connection() as conn:
        cur = await conn.execute(SQL, (q, q))
        rows = await cur.fetchall()

    return [
        {"id": r[0], "name": r[1], "description": r[2], "rank": r[3]}
        for r in rows
    ]

const SQL = `
  SELECT id, name, description,
         ts_rank(search_vec, plainto_tsquery('english', $1)) AS rank
  FROM   products
  WHERE  search_vec @@ plainto_tsquery('english', $1)
  ORDER  BY rank DESC
  LIMIT  20
`

app.get('/search', async (req, res) => {
  const q = req.query.q
  if (!q) {
    return res.status(400).json({ error: 'missing query param q' })
  }

  try {
    // plainto_tsquery converts plain text safely (no special chars needed);
    // websearch_to_tsquery also supports AND/OR/-term syntax.
    const { rows } = await pool.query(SQL, [q])
    res.json(rows)
  } catch (err) {
    res.status(500).json({ error: err.message })
  }
})

import type { Pool } from 'pg'
import type { Request, Response } from 'express'

interface SearchResult {
  id: number
  name: string
  description: string
  rank: number
}

const SQL = `
  SELECT id, name, description,
         ts_rank(search_vec, plainto_tsquery('english', $1)) AS rank
  FROM   products
  WHERE  search_vec @@ plainto_tsquery('english', $1)
  ORDER  BY rank DESC
  LIMIT  20
`

export async function search(req: Request, res: Response): Promise<void> {
  // `req.query.q` is not a string: Express parses `?q=a&q=b` into an
  // ARRAY, so a naive cast would hand an array straight to the driver.
  // Narrowing it here is the difference between a 400 and a 500.
  const q = typeof req.query.q === 'string' ? req.query.q : ''
  if (q === '') {
    res.status(400).json({ error: 'missing query param q' })
    return
  }

  const { rows } = await pool.query<SearchResult>(SQL, [q])
  res.json(rows)
}

public record SearchResult(int id, String name, String description, double rank) {}

@RestController
class SearchController {
    private static final String SQL = """
        SELECT id, name, description,
               ts_rank(search_vec, plainto_tsquery('english', ?)) AS rank
        FROM   products
        WHERE  search_vec @@ plainto_tsquery('english', ?)
        ORDER  BY rank DESC
        LIMIT  20
        """;

    private final JdbcClient db;

    SearchController(JdbcClient db) {
        this.db = db;
    }

    @GetMapping("/search")
    ResponseEntity<?> search(@RequestParam(defaultValue = "") String q) {
        if (q.isBlank()) {
            return ResponseEntity.badRequest().body(Map.of("error", "missing query param q"));
        }

        // plainto_tsquery converts plain text safely (no special chars needed);
        // websearch_to_tsquery also supports AND/OR/-term syntax.
        // Both placeholders are bound, never concatenated (chapter 08).
        List<SearchResult> results = db.sql(SQL)
            .params(q, q)
            .query(SearchResult.class)
            .list();

        return ResponseEntity.ok(results);
    }
}

from elasticsearch import Elasticsearch, helpers
import csv

es = Elasticsearch(
    cloud_id="YOUR_CLOUD_ID",
    api_key="YOUR_API_KEY"
)

INDEX = "reviews"

# 1. Create index with explicit mapping
if es.indices.exists(index=INDEX):
    es.indices.delete(index=INDEX)

es.indices.create(index=INDEX, body={
    "mappings": {
        "properties": {
            "review":    {"type": "text"},     # analysed full-text
            "sentiment": {"type": "keyword"}  # exact: "positive" / "negative"
        }
    }
})

# 2. Bulk insert from CSV
def generate_docs(path: str):
    with open(path) as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row.get("review") and row.get("sentiment"):
                yield {
                    "_index": INDEX,
                    "_source": {
                        "review":    row["review"],
                        "sentiment": row["sentiment"]
                    }
                }

success, errors = helpers.bulk(es, generate_docs("reviews.csv"))
print(f"Inserted {success} documents, {len(errors)} errors")

import { Client } from '@elastic/elasticsearch'
import { createReadStream } from 'node:fs'
import { parse } from 'csv-parse'

const es = new Client({
  cloud: { id: 'YOUR_CLOUD_ID' },
  auth: { apiKey: 'YOUR_API_KEY' },
})

const INDEX = 'reviews'

// 1. Create index with explicit mapping
if (await es.indices.exists({ index: INDEX })) {
  await es.indices.delete({ index: INDEX })
}

await es.indices.create({
  index: INDEX,
  mappings: {
    properties: {
      review: { type: 'text' },       // analysed full-text
      sentiment: { type: 'keyword' }, // exact: "positive" / "negative"
    },
  },
})

// 2. Bulk insert from CSV. `helpers.bulk` takes an async iterable, so the
// file is streamed rather than read into memory, which matters the first
// time somebody points this at a 2 GB export.
async function* generateDocs(path) {
  const parser = createReadStream(path).pipe(parse({ columns: true }))
  for await (const row of parser) {
    if (row.review && row.sentiment) {
      yield { review: row.review, sentiment: row.sentiment }
    }
  }
}

const result = await es.helpers.bulk({
  datasource: generateDocs('reviews.csv'),
  onDocument: () => ({ index: { _index: INDEX } }),
})

console.log(`Inserted ${result.successful} documents, ${result.failed} errors`)

import { Client } from '@elastic/elasticsearch'

// The document type is worth declaring even though Elasticsearch is
// schemaless-ish: it is the only thing keeping the indexer and the
// search code agreeing on field names, and a field name typo here is
// not an error at all in Elasticsearch, it just indexes a new field
// that nothing will ever search.
interface Review {
  review: string
  sentiment: 'positive' | 'negative'
}

const es = new Client({
  cloud: { id: process.env.ELASTIC_CLOUD_ID! },
  auth: { apiKey: process.env.ELASTIC_API_KEY! },
})

const INDEX = 'reviews'

// 1. Create index with explicit mapping
if (await es.indices.exists({ index: INDEX })) {
  await es.indices.delete({ index: INDEX })
}

await es.indices.create({
  index: INDEX,
  mappings: {
    properties: {
      review: { type: 'text' },       // analysed full-text
      sentiment: { type: 'keyword' }, // exact: "positive" / "negative"
    },
  },
})

// 2. Bulk insert, streamed
const result = await es.helpers.bulk<Review>({
  datasource: generateDocs('reviews.csv'),
  onDocument: () => ({ index: { _index: INDEX } }),
  onDrop: (doc) => console.error('dropped', doc.document, doc.error),
})

console.log(`Inserted ${result.successful} documents, ${result.failed} errors`)

public record Review(String review, String sentiment) {}

ElasticsearchClient es = new ElasticsearchClient(
    new RestClientTransport(
        RestClient.builder(HttpHost.create("https://...")).build(),
        new JacksonJsonpMapper()));

final String INDEX = "reviews";

// 1. Create index with explicit mapping.
// The Java client's builder lambdas mirror the JSON DSL one level per
// nested object, which is verbose but means a misspelled DSL key is a
// compile error rather than an index that silently ignores it.
if (es.indices().exists(e -> e.index(INDEX)).value()) {
    es.indices().delete(d -> d.index(INDEX));
}

es.indices().create(c -> c
    .index(INDEX)
    .mappings(m -> m
        .properties("review", p -> p.text(t -> t))        // analysed full-text
        .properties("sentiment", p -> p.keyword(k -> k))  // exact match
    ));

// 2. Bulk insert from CSV, in batches rather than one request per row.
BulkRequest.Builder br = new BulkRequest.Builder();

try (CSVParser rows = CSVFormat.DEFAULT.withFirstRecordAsHeader()
        .parse(new FileReader("reviews.csv"))) {
    for (CSVRecord row : rows) {
        String review = row.get("review");
        String sentiment = row.get("sentiment");
        if (review.isBlank() || sentiment.isBlank()) continue;

        br.operations(op -> op.index(idx -> idx
            .index(INDEX)
            .document(new Review(review, sentiment))));
    }
}

BulkResponse result = es.bulk(br.build());
long failed = result.items().stream().filter(i -> i.error() != null).count();
System.out.printf("Inserted %d documents, %d errors%n",
    result.items().size() - failed, failed);

def search_reviews(query: str, sentiment_filter: str = None) -> list:
    must_clauses = [{
        "multi_match": {
            "query":     query.lower(),
            "fields":   ["review"],
            "fuzziness": "AUTO",
            "operator":  "and"   # all terms must appear
        }
    }]

    filter_clauses = []
    if sentiment_filter:
        filter_clauses.append({
            "term": {"sentiment": sentiment_filter}
        })

    body = {
        "query": {"bool": {"must": must_clauses, "filter": filter_clauses}},
        "size": 20
    }

    resp = es.search(index=INDEX, body=body)
    return [
        {"score": hit["_score"], **hit["_source"]}
        for hit in resp["hits"]["hits"]
    ]

# Usage
results = search_reviews("gret product", sentiment_filter="positive")
# "gret" (typo for "great") matched via fuzziness=AUTO (edit distance 1)

async function searchReviews(query, sentimentFilter) {
  const must = [{
    multi_match: {
      query: query.toLowerCase(),
      fields: ['review'],
      fuzziness: 'AUTO', // edit distance scaled to term length
      operator: 'and',   // all terms must appear
    },
  }]

  // `filter`, not `must`. A filter is a yes/no gate: it does not
  // contribute to the score and Elasticsearch caches the bitset, so it
  // is both more correct for exact matches and faster.
  const filter = sentimentFilter
    ? [{ term: { sentiment: sentimentFilter } }]
    : []

  const resp = await es.search({
    index: INDEX,
    query: { bool: { must, filter } },
    size: 20,
  })

  return resp.hits.hits.map((hit) => ({ score: hit._score, ...hit._source }))
}

// Usage
const results = await searchReviews('gret product', 'positive')
// "gret" (typo for "great") matched via fuzziness=AUTO (edit distance 1)

import type { QueryDslQueryContainer } from '@elastic/elasticsearch/lib/api/types'

// The client's own DSL types are the reason to use TypeScript here: the
// query DSL is a deeply nested untyped blob in every other language, and
// a misspelled key like `fuzzines` is not an error to Elasticsearch, it
// is a silently ignored clause that quietly turns typo tolerance off.
async function searchReviews(
  query: string,
  sentimentFilter?: 'positive' | 'negative',
): Promise<Array<Review & { score: number }>> {
  const must: QueryDslQueryContainer[] = [{
    multi_match: {
      query: query.toLowerCase(),
      fields: ['review'],
      fuzziness: 'AUTO', // edit distance scaled to term length
      operator: 'and',   // all terms must appear
    },
  }]

  const filter: QueryDslQueryContainer[] = sentimentFilter
    ? [{ term: { sentiment: sentimentFilter } }] // exact gate, unscored
    : []

  const resp = await es.search<Review>({
    index: INDEX,
    query: { bool: { must, filter } },
    size: 20,
  })

  return resp.hits.hits.map((hit) => ({
    score: hit._score ?? 0, // _score is null on a pure filter query
    ...(hit._source as Review),
  }))
}

List<Hit<Review>> searchReviews(String query, String sentimentFilter) throws IOException {
    Query multiMatch = MultiMatchQuery.of(m -> m
        .query(query.toLowerCase())
        .fields("review")
        .fuzziness("AUTO")                 // edit distance scaled to term length
        .operator(Operator.And)            // all terms must appear
    )._toQuery();

    // `filter`, not `must`: an exact term is a yes/no gate that should
    // not influence the score, and Elasticsearch caches it.
    List<Query> filters = (sentimentFilter == null) ? List.of() : List.of(
        TermQuery.of(t -> t.field("sentiment").value(sentimentFilter))._toQuery());

    SearchResponse<Review> resp = es.search(s -> s
        .index("reviews")
        .size(20)
        .query(q -> q.bool(b -> b.must(multiMatch).filter(filters))),
        Review.class);

    // "gret" (typo for "great") matches via fuzziness=AUTO (edit distance 1)
    return resp.hits().hits();
}

import asyncio
import json

from fastapi import Request
from fastapi.responses import StreamingResponse


@app.get("/search/stream")
async def stream_search(request: Request):
    term = request.query_params.get("q", "")

    async def generate():
        # Fire BOTH queries concurrently, don't await sequentially.
        # create_task starts the coroutine now; awaiting them in turn
        # would run the second only after the first had finished.
        pending = {
            asyncio.create_task(pg_search(term)): "postgres",
            asyncio.create_task(es_search(term)): "elasticsearch",
        }

        while pending:
            # Wake on whichever task finishes FIRST, so the fast source
            # is never held hostage by the slow one.
            done, _ = await asyncio.wait(
                pending, return_when=asyncio.FIRST_COMPLETED)

            for task in done:
                source = pending.pop(task)
                try:
                    yield json.dumps({"source": source, "data": task.result()}) + "\n"
                except Exception as e:
                    # One source failing degrades the page, it does not
                    # blank it: the other source's results still ship.
                    yield json.dumps({"source": source, "error": str(e)}) + "\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

import { Client } from '@elastic/elasticsearch'
import { neon }     from '@neondatabase/serverless'

export async function GET(req) {
  const { searchParams } = new URL(req.url)
  const term = searchParams.get('q')

  const encoder = new TextEncoder()
  const stream = new ReadableStream({
    async start(controller) {
      // Fire BOTH queries concurrently, don't await sequentially
      const [pgResult, esResult] = await Promise.allSettled([
        pgSearch(term),
        esSearch(term),
      ])

      controller.enqueue(encoder.encode(JSON.stringify({
        source: 'postgres',
        ...pgResult
      })))
      controller.enqueue(encoder.encode(JSON.stringify({
        source: 'elasticsearch',
        ...esResult
      })))
      controller.close()
    }
  })
  return new Response(stream, { headers: { 'Content-Type': 'text/event-stream' } })
}

interface StreamChunk<T> {
  source: 'postgres' | 'elasticsearch'
  status: 'fulfilled' | 'rejected'
  data?: T
  error?: string
}

// allSettled, never all: `Promise.all` rejects the whole response the
// moment ONE source fails, which is precisely the wrong behaviour here.
// Elasticsearch being down should degrade the page to Postgres results,
// not blank it. The union type above is what forces the consumer to
// handle the rejected case rather than assuming `data` is there.
export async function GET(req: Request): Promise<Response> {
  const term = new URL(req.url).searchParams.get('q') ?? ''
  const encoder = new TextEncoder()

  const stream = new ReadableStream({
    async start(controller) {
      const send = <T>(chunk: StreamChunk<T>): void => {
        controller.enqueue(encoder.encode(JSON.stringify(chunk) + '\n'))
      }

      // Fire BOTH queries concurrently, and send each as it settles
      await Promise.all([
        pgSearch(term).then(
          (data) => send({ source: 'postgres', status: 'fulfilled', data }),
          (err: Error) => send({ source: 'postgres', status: 'rejected', error: err.message }),
        ),
        esSearch(term).then(
          (data) => send({ source: 'elasticsearch', status: 'fulfilled', data }),
          (err: Error) => send({ source: 'elasticsearch', status: 'rejected', error: err.message }),
        ),
      ])

      controller.close()
    },
  })

  return new Response(stream, { headers: { 'Content-Type': 'text/event-stream' } })
}

@RestController
class StreamController {

    // SseEmitter is Spring's streaming response: the request thread is
    // released immediately and events are pushed as they are produced.
    @GetMapping(value = "/search/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    SseEmitter streamSearch(@RequestParam(defaultValue = "") String q) {
        SseEmitter emitter = new SseEmitter(30_000L); // always set a timeout

        // Fire BOTH queries concurrently, don't await sequentially.
        // supplyAsync starts the work now; the thenAccept fires per source.
        CompletableFuture<Void> pg = CompletableFuture
            .supplyAsync(() -> pgSearch(q))
            .handle((data, err) -> send(emitter, "postgres", data, err))
            .thenAccept(x -> { });

        CompletableFuture<Void> es = CompletableFuture
            .supplyAsync(() -> esSearch(q))
            .handle((data, err) -> send(emitter, "elasticsearch", data, err))
            .thenAccept(x -> { });

        // Complete only once BOTH have reported, success or failure.
        CompletableFuture.allOf(pg, es)
            .whenComplete((v, err) -> emitter.complete());

        return emitter;
    }

    private Object send(SseEmitter emitter, String source, Object data, Throwable err) {
        try {
            emitter.send(Map.of(
                "source", source,
                err == null ? "data" : "error",
                err == null ? data : err.getMessage()));
        } catch (IOException io) {
            emitter.completeWithError(io); // client hung up mid-stream
        }
        return null;
    }
}