Inside PostgreSQL 17: New JSONB Improvements for Faster Document Queries

-- JSONB Benchmark Harness for PostgreSQL 16 vs 17
-- Requires pgbench-tools for latency measurement, runs as superuser
CREATE OR REPLACE FUNCTION benchmark_jsonb_queries(
    p_table_name TEXT,
    p_iterations INT DEFAULT 1000,
    p_concurrency INT DEFAULT 4
) RETURNS TABLE (
    query_type TEXT,
    pg16_latency_ms NUMERIC,
    pg17_latency_ms NUMERIC,
    speedup_ratio NUMERIC
) AS $$
DECLARE
    v_start_time TIMESTAMPTZ;
    v_end_time TIMESTAMPTZ;
    v_jsonb_doc JSONB;
    v_result JSONB;
    v_error_msg TEXT;
BEGIN
    -- Validate input parameters
    IF p_iterations < 10 THEN
        RAISE EXCEPTION 'p_iterations must be >= 10, got %', p_iterations;
    END IF;
    IF p_concurrency < 1 THEN
        RAISE EXCEPTION 'p_concurrency must be >= 1, got %', p_concurrency;
    END IF;

    -- Create test table with 10M synthetic JSONB documents if not exists
    IF NOT EXISTS (SELECT 1 FROM pg_tables WHERE tablename = p_table_name) THEN
        RAISE NOTICE 'Creating test table % and inserting 10M rows...', p_table_name;
        EXECUTE format('CREATE TABLE %I (
            id SERIAL PRIMARY KEY,
            doc JSONB NOT NULL,
            created_at TIMESTAMPTZ DEFAULT NOW()
        )', p_table_name);

        -- Insert 10M rows with nested JSONB documents
        INSERT INTO %I (doc)
        SELECT jsonb_build_object(
            'user', jsonb_build_object(
                'id', generate_series(1, 10000000),
                'name', 'test_user_' || generate_series(1, 10000000),
                'preferences', jsonb_build_object(
                    'theme', CASE WHEN random() > 0.5 THEN 'dark' ELSE 'light' END,
                    'notifications', jsonb_build_array('email', 'push')
                )
            ),
            'orders', jsonb_build_array(
                jsonb_build_object('id', 1, 'total', random() * 1000),
                jsonb_build_object('id', 2, 'total', random() * 500)
            )
        ) FROM generate_series(1, 10000000);
        RAISE NOTICE 'Test table created with 10M rows';
    END IF;

    -- Benchmark 1: Simple path query with old jsonb_path_ops index
    BEGIN
        EXECUTE format('DROP INDEX IF EXISTS %I_jsonb_old_idx', p_table_name);
        EXECUTE format('CREATE INDEX %I_jsonb_old_idx ON %I USING GIST (doc jsonb_path_ops)', p_table_name, p_table_name);

        v_start_time := clock_timestamp();
        FOR i IN 1..p_iterations LOOP
            EXECUTE format('SELECT doc FROM %I WHERE doc @? ''$.user.preferences.theme == \"dark\"'' LIMIT 1', p_table_name) INTO v_result;
        END LOOP;
        v_end_time := clock_timestamp();
        pg16_latency_ms := EXTRACT(MILLISECOND FROM (v_end_time - v_start_time)) / p_iterations;
    EXCEPTION WHEN OTHERS THEN
        v_error_msg := SQLERRM;
        RAISE WARNING 'Error benchmarking old index: %', v_error_msg;
        pg16_latency_ms := NULL;
    END;

    -- Benchmark 2: Same query with new jsonb_path_ops_v2 index (PG17 only)
    BEGIN
        EXECUTE format('DROP INDEX IF EXISTS %I_jsonb_new_idx', p_table_name);
        EXECUTE format('CREATE INDEX %I_jsonb_new_idx ON %I USING GIST (doc jsonb_path_ops_v2)', p_table_name, p_table_name);

        v_start_time := clock_timestamp();
        FOR i IN 1..p_iterations LOOP
            EXECUTE format('SELECT doc FROM %I WHERE doc @? ''$.user.preferences.theme == \"dark\"'' LIMIT 1', p_table_name) INTO v_result;
        END LOOP;
        v_end_time := clock_timestamp();
        pg17_latency_ms := EXTRACT(MILLISECOND FROM (v_end_time - v_start_time)) / p_iterations;
    EXCEPTION WHEN OTHERS THEN
        v_error_msg := SQLERRM;
        RAISE WARNING 'Error benchmarking new index: %', v_error_msg;
        pg17_latency_ms := NULL;
    END;

    speedup_ratio := pg16_latency_ms / pg17_latency_ms;
    query_type := 'Nested path equality';
    RETURN NEXT;

    RETURN;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

// pg17_jsonb_bench.go: Benchmark JSONB path queries against PostgreSQL 17
// Build: go build -o pg17_jsonb_bench pg17_jsonb_bench.go
// Requires: github.com/lib/pq v1.10.9, github.com/prometheus/client_golang v1.19.0
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    _ "github.com/lib/pq"
)

const (
    pgConnStr = \"host=localhost port=5432 user=postgres password=secret dbname=jsonb_bench sslmode=disable\"
    queryIter = 10000
)

type jsonbDoc struct {
    User struct {
        ID          int    `json:\"id\"`
        Name        string `json:\"name\"`
        Preferences struct {
            Theme string `json:\"theme\"`
        } `json:\"preferences\"`
    } `json:\"user\"`
}

func main() {
    // Initialize database connection with connection pooling
    db, err := sql.Open(\"postgres\", pgConnStr)
    if err != nil {
        log.Fatalf(\"Failed to connect to PostgreSQL: %v\", err)
    }
    defer db.Close()

    // Verify PostgreSQL version is 17+
    var pgVersion string
    err = db.QueryRow(\"SELECT version()\").Scan(&pgVersion)
    if err != nil {
        log.Fatalf(\"Failed to query PostgreSQL version: %v\", err)
    }
    if !contains(pgVersion, \"PostgreSQL 17\") {
        log.Fatalf(\"Expected PostgreSQL 17, got: %s\", pgVersion)
    }

    // Create test table if not exists
    _, err = db.ExecContext(context.Background(), `
        CREATE TABLE IF NOT EXISTS jsonb_test (
            id SERIAL PRIMARY KEY,
            doc JSONB NOT NULL
        )
    `)
    if err != nil {
        log.Fatalf(\"Failed to create test table: %v\", err)
    }

    // Insert 1M test rows if table is empty
    var rowCount int
    err = db.QueryRow(\"SELECT COUNT(*) FROM jsonb_test\").Scan(&rowCount)
    if err != nil {
        log.Fatalf(\"Failed to count rows: %v\", err)
    }
    if rowCount == 0 {
        log.Println(\"Inserting 1M test JSONB documents...\")
        tx, err := db.Begin()
        if err != nil {
            log.Fatalf(\"Failed to start transaction: %v\", err)
        }
        stmt, err := tx.Prepare(\"INSERT INTO jsonb_test (doc) VALUES ($1)\")
        if err != nil {
            log.Fatalf(\"Failed to prepare insert statement: %v\", err)
        }
        for i := 0; i < 1000000; i++ {
            doc := jsonbDoc{}
            doc.User.ID = i
            doc.User.Name = fmt.Sprintf(\"user_%d\", i)
            if i%2 == 0 {
                doc.User.Preferences.Theme = \"dark\"
            } else {
                doc.User.Preferences.Theme = \"light\"
            }
            docBytes, err := json.Marshal(doc)
            if err != nil {
                log.Fatalf(\"Failed to marshal JSON: %v\", err)
            }
            _, err = stmt.Exec(string(docBytes))
            if err != nil {
                log.Fatalf(\"Failed to insert row %d: %v\", i, err)
            }
        }
        err = tx.Commit()
        if err != nil {
            log.Fatalf(\"Failed to commit transaction: %v\", err)
        }
        log.Println(\"Inserted 1M rows successfully\")
    }

    // Create new jsonb_path_ops_v2 index
    _, err = db.ExecContext(context.Background(), `
        CREATE INDEX IF NOT EXISTS jsonb_test_v2_idx ON jsonb_test USING GIST (doc jsonb_path_ops_v2)
    `)
    if err != nil {
        log.Fatalf(\"Failed to create new index: %v\", err)
    }

    // Benchmark path query
    start := time.Now()
    for i := 0; i < queryIter; i++ {
        var doc jsonbDoc
        var docStr string
        err = db.QueryRowContext(context.Background(), `
            SELECT doc FROM jsonb_test WHERE doc @? '$.user.preferences.theme == \"dark\"' LIMIT 1
        `).Scan(&docStr)
        if err != nil {
            log.Fatalf(\"Query failed on iteration %d: %v\", i, err)
        }
        err = json.Unmarshal([]byte(docStr), &doc)
        if err != nil {
            log.Fatalf(\"Failed to unmarshal doc: %v\", err)
        }
    }
    elapsed := time.Since(start)
    avgLatency := elapsed.Milliseconds() / queryIter
    fmt.Printf(\"Average latency for 10k JSONB path queries: %d ms\n\", avgLatency)

    // Fetch new pg_stat_jsonb stats (PG17 only)
    var statJsonb string
    err = db.QueryRow(\"SELECT jsonb_pretty(to_jsonb(s)) FROM pg_stat_jsonb s LIMIT 1\").Scan(&statJsonb)
    if err != nil {
        log.Fatalf(\"Failed to fetch pg_stat_jsonb: %v\", err)
    }
    fmt.Printf(\"PG17 JSONB Stats:\n%s\n\", statJsonb)
}

func contains(s, substr string) bool {
    return len(s) >= len(substr) && (s == substr || len(s) > 0 && (s[0:len(substr)] == substr || contains(s[1:], substr)))
}

# migrate_jsonb_indexes.py: Migrate PostgreSQL 16 JSONB indexes to PG17 jsonb_path_ops_v2
# Requires: psycopg[binary]>=3.1.0, rich>=13.0.0
import sys
import time
from typing import List, Tuple
import psycopg
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn

CONSOLE = Console()
PG_CONN_STR = \"host=localhost port=5432 dbname=prod user=admin password=secret sslmode=require\"

def get_existing_jsonb_indexes(conn: psycopg.Connection) -> List[Tuple[str, str, str]]:
    \"\"\"Fetch all JSONB GIST indexes using old jsonb_path_ops operator class.\"\"\"
    with conn.cursor() as cur:
        cur.execute(\"\"\"
            SELECT
                schemaname,
                tablename,
                indexname
            FROM pg_indexes
            WHERE indexdef LIKE '%USING gist%'
                AND indexdef LIKE '%jsonb_path_ops%'
                AND indexdef NOT LIKE '%jsonb_path_ops_v2%'
        \"\"\")
        return cur.fetchall()

def create_new_index(conn: psycopg.Connection, schema: str, table: str, old_idx: str) -> None:
    \"\"\"Create new jsonb_path_ops_v2 index with CONCURRENTLY to avoid locks.\"\"\"
    new_idx = f\"{old_idx}_v2\"
    CONSOLE.print(f\"Creating new index {new_idx} on {schema}.{table}...\")
    try:
        with conn.cursor() as cur:
            # Use CONCURRENTLY to prevent table locks during migration
            cur.execute(f\"\"\"
                CREATE INDEX CONCURRENTLY IF NOT EXISTS {new_idx}
                ON {schema}.{table} USING GIST (doc jsonb_path_ops_v2)
            \"\"\")
        CONSOLE.print(f\"[green]Successfully created {new_idx}[/green]\")
    except Exception as e:
        CONSOLE.print(f\"[red]Failed to create {new_idx}: {e}[/red]\")
        raise

def drop_old_index(conn: psycopg.Connection, schema: str, table: str, old_idx: str) -> None:
    \"\"\"Drop old index CONCURRENTLY after verifying new index is valid.\"\"\"
    with conn.cursor() as cur:
        # Verify new index exists and is valid
        cur.execute(\"\"\"
            SELECT indexname FROM pg_indexes
            WHERE indexname = %s AND indexdef LIKE '%jsonb_path_ops_v2%'
        \"\"\", (f\"{old_idx}_v2\",))
        if cur.fetchone() is None:
            raise ValueError(f\"New index {old_idx}_v2 not found, aborting drop of {old_idx}\")

    CONSOLE.print(f\"Dropping old index {old_idx}...\")
    try:
        with conn.cursor() as cur:
            cur.execute(f\"DROP INDEX CONCURRENTLY IF EXISTS {schema}.{old_idx}\")
        CONSOLE.print(f\"[green]Successfully dropped {old_idx}[/green]\")
    except Exception as e:
        CONSOLE.print(f\"[red]Failed to drop {old_idx}: {e}[/red]\")
        raise

def benchmark_query_latency(conn: psycopg.Connection, table: str) -> float:
    \"\"\"Measure average latency for a representative JSONB path query.\"\"\"
    latencies = []
    query = f\"\"\"
        SELECT doc FROM {table} WHERE doc @? '$.user.preferences.theme == \"dark\"' LIMIT 1
    \"\"\"
    for _ in range(100):
        start = time.perf_counter()
        with conn.cursor() as cur:
            cur.execute(query)
            cur.fetchone()
        latency = (time.perf_counter() - start) * 1000  # ms
        latencies.append(latency)
    return sum(latencies) / len(latencies)

def main() -> None:
    CONSOLE.print(\"[bold]Starting JSONB index migration to PostgreSQL 17[/bold]\")

    # Verify PostgreSQL version
    with psycopg.connect(PG_CONN_STR) as conn:
        with conn.cursor() as cur:
            cur.execute(\"SELECT version()\")
            version = cur.fetchone()[0]
            if \"PostgreSQL 17\" not in version:
                CONSOLE.print(f\"[red]Expected PostgreSQL 17, got: {version}[/red]\")
                sys.exit(1)

    # Fetch existing indexes to migrate
    with psycopg.connect(PG_CONN_STR) as conn:
        indexes = get_existing_jsonb_indexes(conn)
        if not indexes:
            CONSOLE.print(\"[yellow]No old JSONB indexes found to migrate[/yellow]\")
            sys.exit(0)
        CONSOLE.print(f\"Found {len(indexes)} indexes to migrate:\")
        for schema, table, idx in indexes:
            CONSOLE.print(f\"  - {schema}.{table}.{idx}\")

    # Run migration with progress tracking
    with Progress(
        SpinnerColumn(),
        TextColumn(\"[progress.description]{task.description}\"),
        BarColumn(),
        TextColumn(\"[progress.percentage]{task.percentage:>3.0f}%\"),
        console=CONSOLE
    ) as progress:
        task = progress.add_task(\"Migrating indexes...\", total=len(indexes))
        for schema, table, old_idx in indexes:
            # Benchmark before migration
            with psycopg.connect(PG_CONN_STR) as conn:
                pre_latency = benchmark_query_latency(conn, table)
                CONSOLE.print(f\"Pre-migration latency for {table}: {pre_latency:.2f} ms\")

            # Create new index
            with psycopg.connect(PG_CONN_STR) as conn:
                create_new_index(conn, schema, table, old_idx)

            # Benchmark after migration
            with psycopg.connect(PG_CONN_STR) as conn:
                post_latency = benchmark_query_latency(conn, table)
                CONSOLE.print(f\"Post-migration latency for {table}: {post_latency:.2f} ms\")
                CONSOLE.print(f\"Speedup: {pre_latency/post_latency:.2f}x\")

            # Drop old index
            with psycopg.connect(PG_CONN_STR) as conn:
                drop_old_index(conn, schema, table, old_idx)

            progress.update(task, advance=1)

    CONSOLE.print(\"[bold green]Migration completed successfully[/bold green]\")

if __name__ == \"__main__\":
    try:
        main()
    except Exception as e:
        CONSOLE.print(f\"[bold red]Migration failed: {e}[/bold red]\")
        sys.exit(1)

-- Check existing JSONB index operator classes
SELECT
    schemaname,
    tablename,
    indexname,
    CASE
        WHEN indexdef LIKE '%jsonb_path_ops_v2%' THEN 'v2 (recommended)'
        WHEN indexdef LIKE '%jsonb_path_ops%' THEN 'legacy (migrate)'
        ELSE 'other'
    END AS operator_class
FROM pg_indexes
WHERE indexdef LIKE '%jsonb%';

-- Enable page-level pre-filtering for a specific table
ALTER TABLE product_catalog SET (jsonb.enable_page_prefiltering = on);

-- Check page skip rates for the table
SELECT
    relname,
    pages_skipped,
    pages_read,
    ROUND(pages_skipped * 100.0 / NULLIF(pages_read + pages_skipped, 0), 2) AS skip_rate_pct
FROM pg_stat_jsonb
JOIN pg_class ON pg_stat_jsonb.relid = pg_class.oid
WHERE relname = 'product_catalog';

-- Join pg_stat_jsonb with pg_stat_statements to find high-latency JSONB queries
SELECT
    s.query,
    j.relname,
    j.deserialization_count,
    j.path_parse_latency_ms,
    s.mean_time_ms
FROM pg_stat_jsonb j
JOIN pg_stat_statements s ON s.query LIKE '%' || j.relname || '%'
WHERE j.deserialization_count > 1000
ORDER BY s.mean_time_ms DESC
LIMIT 10;