Build Modern AI and Search Applications

1%pip install singlestoredb pandas --quiet

1import json, random, re, statistics, time   # repeated here so this cell works on its own2import pandas as pd3import singlestoredb as s24
5pd.set_option('display.max_colwidth', 120)6random.seed(42)7
8DB_NAME = 'adventure_inc'9
10USE_DB = None   # set below once we know which database to use11
12def connect():13    """Open a connection (again) and select the notebook's database."""14    global conn15    conn = s2.connect(database=USE_DB) if USE_DB else s2.connect()16    conn.autocommit(True)17    if USE_DB:18        with conn.cursor() as cur:19            cur.execute(f'USE {USE_DB}')20
21def _query(query, params=None):22    with conn.cursor() as cur:23        cur.execute(query, params)24        if cur.description:25            cols = [d[0] for d in cur.description]26            return pd.DataFrame(cur.fetchall(), columns=cols)27    return None28
29def run_sql(query, params=None):30    """Run a query and return a DataFrame (or None).31
32    Handles two temporary conditions for you:33    - an idle connection that was dropped: reconnects and retries once34    - the Full Text Service still starting up on first use: waits and retries35    """36    for attempt in range(13):37        try:38            return _query(query, params)39        except Exception as e:40            msg = str(e)41            if getattr(e, 'errno', None) in (2006, 2013) or 'Lost connection' in msg or 'gone away' in msg:42                print('(Connection was dropped, reconnecting...)')43                connect()44            elif 'Full Text Service' in msg and attempt < 12:45                if attempt == 0:46                    print('(Full Text Service is starting up, retrying in a few seconds...)')47                time.sleep(5)48            else:49                raise50
51connect()52
53def try_sql(query, params=None, label='This step'):54    """Like run_sql(), but explains and continues if the workspace doesn't support something."""55    try:56        return run_sql(query, params)57    except Exception as e:58        print(f'{label} was skipped: {e}')59        return None60
61shared = run_sql("SHOW VARIABLES LIKE 'is_shared_tier'")62IS_SHARED = shared is not None and len(shared) > 0 and shared.iloc[0, 1] == 'ON'63if not IS_SHARED:64    run_sql(f'CREATE DATABASE IF NOT EXISTS {DB_NAME}')65    USE_DB = DB_NAME66    run_sql(f'USE {DB_NAME}')67
68print('Connected to database:', run_sql('SELECT DATABASE() AS db').iloc[0, 0])69print('SingleStore version:  ', run_sql('SELECT @@memsql_version AS v').iloc[0, 0])70
71# ---- Feature check ----72import logging, os, warnings73os.environ['HF_HUB_VERBOSITY'] = 'error'              # quiet the model download messages74os.environ['HF_HUB_DISABLE_PROGRESS_BARS'] = '1'75for name in ['httpx', 'huggingface_hub', 'huggingface_hub.utils._http', 'sentence_transformers']:76    logging.getLogger(name).setLevel(logging.ERROR)77warnings.filterwarnings('ignore', category=FutureWarning)78warnings.filterwarnings('ignore', message='.*unauthenticated requests.*')79
80# 1. F16 vectors (half precision)81try:82    run_sql("SELECT '[0.5, -0.25]' :> VECTOR(2, F16) AS v")83    VEC_ELEM = 'F16'84    print('F16 vectors: supported')85except Exception as e:86    VEC_ELEM = 'F32'87    print('F16 vectors: not available on this workspace, falling back to F32 ->', e)88
89# 2. AI Functions for embeddings90local_model = None91try:92    probe = run_sql("SELECT LENGTH(cluster.EMBED_TEXT('hello from AdVenture Inc.')) AS n_bytes")93    DIM = int(probe.iloc[0, 0]) // 4   # EMBED_TEXT returns packed 32-bit floats94    run_sql(f"SELECT cluster.EMBED_TEXT('check') :> VECTOR({DIM}) :> VECTOR({DIM}, {VEC_ELEM}) AS v")95    EMBED_MODE = 'ai'96    print(f'AI Functions: available, embeddings have {DIM} dimensions')97except Exception as e:98    EMBED_MODE = 'local'99    print('AI Functions: not available, using a free open-source model instead ->', e)100    print('Downloading the model from Hugging Face (one time, about a minute)...')101    get_ipython().run_line_magic('pip', 'install sentence-transformers --quiet')102    from sentence_transformers import SentenceTransformer103    local_model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')104    DIM = (getattr(local_model, 'get_embedding_dimension', None)105           or local_model.get_sentence_embedding_dimension)()106
107VEC_TYPE = f'VECTOR({DIM}, {VEC_ELEM})'108print('Embedding column type:', VEC_TYPE, '| embeddings made by:', EMBED_MODE)

1vendors = pd.read_json('https://raw.githubusercontent.com/singlestore-labs/demo-realtime-digital-marketing/main/data/vendors.json')2vendors['domain'] = (vendors['vendor'].str.lower() + '.' + vendors['tld'])3vendors = vendors.drop_duplicates('domain')4
5run_sql('DROP TABLE IF EXISTS vendors')6run_sql('''7    CREATE ROWSTORE REFERENCE TABLE vendors (8        domain VARCHAR(200) PRIMARY KEY,9        vendor VARCHAR(100),10        category VARCHAR(50),11        popularity DOUBLE12    )13''')14with conn.cursor() as cur:15    cur.executemany(16        'INSERT INTO vendors VALUES (%s, %s, %s, %s)',17        vendors[['domain', 'vendor', 'category', 'popularity']].values.tolist(),18    )19run_sql('SELECT category, COUNT(*) AS vendors FROM vendors GROUP BY category ORDER BY vendors DESC LIMIT 8')

1# The copywriter's vibe book: three hooks, a story and a call to action per category.2VIBES = {3    'books':       ('Plot twist: you finally read it', 'Your reading nook called', 'Chapter one starts tonight',4                    'Rain on the window, tea in hand, a stack of new stories and nowhere to be on a slow Sunday.', 'Curl up and start reading.'),5    'games':       ('Press start on the weekend', 'Couch co-op, snacks ready', 'One more level, we promise',6                    'Stay-in weekends, board game nights with friends and cozy gaming marathons when the weather turns grey.', 'Grab a controller.'),7    'movies':      ('Popcorn is mandatory', 'Your movie night just got an upgrade', 'Roll credits on boredom',8                    'Blanket forts, film marathons and date nights on the sofa while it pours outside.', 'Hit play.'),9    'music':       ('Turn it up', 'Your road trip needs a soundtrack', 'Headphones on, world off',10                    'Wireless headphones, vinyl crates and playlists for long drives, workouts and quiet rainy afternoons.', 'Find your sound.'),11    'electronics': ('Upgrade everything', 'Wireless and fearless', 'Charged up and ready',12                    'Wireless headphones, smart speakers and gadgets that make your home, commute and workouts easier.', 'Plug into the future.'),13    'computers':   ('Built for deadline day', 'Your laptop deserves better', 'Faster than your group chat',14                    'Laptops, monitors and keyboards for working from home, studying late and gaming after hours.', 'Power up your desk.'),15    'sports':      ('Game day ready', 'Run further, recover faster', 'Your personal best is waiting',16                    'Running shoes, team jerseys and gear for runners, weekend leagues and big game watch parties.', 'Get in the game.'),17    'outdoors':    ('Adventure is calling', 'Tents up, phones down', 'Trail mix not included',18                    'Camping gear, hiking boots and everything for a road trip into the mountains or a lazy lake weekend.', 'Head outside.'),19    'shoes':       ('Step it up', 'Blisters are so last season', 'Run the city in comfort',20                    'Running shoes, sneakers and boots made for long walks, city commutes and marathon training.', 'Find your fit.'),21    'clothing':    ('New season, new you', 'Your closet called, it is bored', 'Layer up in style',22                    'Cozy sweaters for rainy days, outfits for date night and everyday basics that just work.', 'Dress the part.'),23    'beauty':      ('Glow like you mean it', 'Self-care Sunday, sorted', 'Your skin says thank you',24                    'Skincare, candles and bath bombs for a relaxing spa day at home or getting ready for a night out.', 'Treat yourself.'),25    'jewelry':     ('Say it with sparkle', 'Anniversary panic? Solved', 'Shine on',26                    'Rings, necklaces and gifts for anniversaries, date nights, proposals and holiday surprises.', 'Make it unforgettable.'),27    'health':      ('Feel good from the inside', 'Wellness, minus the lecture', 'Your body will notice',28                    'Vitamins, yoga mats and sleep aids for calm mornings, healthy habits and a fresh start.', 'Start feeling better.'),29    'home':        ('Make your place a sanctuary', 'Cozy mode: activated', 'Home is where the throw pillows are',30                    'Blankets, candles and furniture that turn a grey rainy day into the coziest day of the year.', 'Nest a little.'),31    'garden':      ('Grow something good', 'Green thumbs welcome', 'Your balcony, but lush',32                    'Seeds, planters and tools for spring planting, herb gardens and slow weekends outdoors.', 'Get your hands dirty.'),33    'grocery':     ('Dinner, solved', 'Snack attack approved', 'Fresh from the market',34                    'Fresh groceries, comfort food ingredients and snacks for movie nights and busy weeknights.', 'Fill your cart.'),35    'kids':        ('Keep the little ones busy', 'Rainy day rescue kit', 'Fun for the whole family',36                    'Crafts, puzzles and activities for rainy days inside, school breaks and family weekends.', 'Make some memories.'),37    'toys':        ('Playtime, upgraded', 'The favorite toy is here', 'Bored? Not anymore',38                    'Building sets, puzzles and games for birthdays, holidays and rainy afternoons at home.', 'Let the fun begin.'),39    'baby':        ('Tiny humans, big deals', 'Sleep more (maybe)', 'Everything for the new arrival',40                    'Soft blankets, strollers and nursery essentials for new parents and baby showers.', 'Welcome the little one.'),41    'automotive':  ('Road trip ready', 'Your car deserves a spa day', 'Built for the long haul',42                    'Car care, tires and accessories for road trips, daily commutes and winter driving.', 'Hit the road.'),43    'tools':       ('Weekend project? Nailed it', 'Fix it yourself', 'Measure twice, save once',44                    'Drills, toolkits and supplies for DIY weekends, home repairs and garage projects.', 'Build something.'),45    'industrial':  ('Heavy duty, light price', 'Built for the job site', 'Tough gear for tough days',46                    'Safety gear, equipment and supplies for warehouses, workshops and serious builders.', 'Get to work.'),47    'general':     ('Deals worth stopping for', 'You will want to see this', 'Something good nearby',48                    'Everyday deals from local shops around the city.', 'Take a look.'),49}50
51run_sql('DROP TABLE IF EXISTS category_vibes')52run_sql('''53    CREATE ROWSTORE REFERENCE TABLE category_vibes (54        category VARCHAR(50) PRIMARY KEY,55        hook1 TEXT, hook2 TEXT, hook3 TEXT, story TEXT, cta TEXT56    )57''')58with conn.cursor() as cur:59    cur.executemany('INSERT INTO category_vibes VALUES (%s, %s, %s, %s, %s, %s)',60                    [(k, *v) for k, v in VIBES.items()])61print(f'{len(VIBES)} categories in the vibe book')

1if 'VEC_TYPE' not in globals():2    raise RuntimeError('Run the setup cell in Chapter 0 first (it ends by printing "Embedding column type").')3
4MAX_CREATIVES = 3000   # caps AI Function calls per batch, keep it modest on a trial5SCALE = '100k-4p'      # dataset size in the bucket: 1k-2p, 10k-2p, 100k-4p, 1m-16p, ... 10m-80p6
7try_sql('DROP PIPELINE IF EXISTS purchases_pipeline', label='Dropping an old pipeline')8for t in ['purchases', 'offers', 'ad_creatives', 'creatives_staging']:9    run_sql(f'DROP TABLE IF EXISTS {t}')10
11run_sql('''12    CREATE TABLE purchases (13        subscriber_id BIGINT,14        vendor VARCHAR(100),15        ts DATETIME(6) DEFAULT NOW(6),16        SORT KEY (ts),17        KEY (vendor) USING HASH18    )19''')20
21run_sql('''22    CREATE ROWSTORE TABLE offers (23        offer_id BIGINT PRIMARY KEY,24        customer VARCHAR(100),25        enabled BOOL,26        notification_zone TEXT,27        segment_ids JSON,28        notification_content TEXT,29        notification_target VARCHAR(200),30        maximum_bid_cents BIGINT31    )32''')33
34run_sql('''35    CREATE ROWSTORE TABLE creatives_staging (36        offer_id BIGINT PRIMARY KEY,37        customer VARCHAR(100), category VARCHAR(50), domain VARCHAR(200),38        headline TEXT, body TEXT, max_bid_cents BIGINT39    )40''')41
42run_sql(f'''43    CREATE TABLE ad_creatives (44        offer_id BIGINT NOT NULL,45        customer VARCHAR(100),46        category VARCHAR(50),47        domain VARCHAR(200),48        headline TEXT,49        body TEXT,50        max_bid_cents BIGINT,51        embedding {VEC_TYPE},52        created_at DATETIME(6) DEFAULT NOW(6),53        SORT KEY (created_at),54        SHARD KEY (offer_id)55    )56''')57print('Tables ready.')

1# Embed the ad's meaning, not the made-up brand names: strip the vendor name and domain first.2embed_text_sql = "REPLACE(REPLACE(CONCAT(s.headline, '. ', s.body), s.domain, ''), s.customer, '')"3embed_expr = (f"cluster.EMBED_TEXT({embed_text_sql}) :> VECTOR({DIM}) :> {VEC_TYPE}"4              if EMBED_MODE == 'ai' else 'NULL')5
6run_sql(f'''7CREATE OR REPLACE PROCEDURE build_creatives() AS8BEGIN9    INSERT INTO creatives_staging10    SELECT o.offer_id,11           o.customer,12           COALESCE(v.category, 'general'),13           o.notification_target,14           CONCAT(o.notification_content, ': ', ELT(1 + MOD(o.offer_id, 3), cv.hook1, cv.hook2, cv.hook3)),15           CONCAT(cv.story, ' ', o.customer, ' (', o.notification_target, ') bids up to ',16                  o.maximum_bid_cents, ' cents to reach you. ', cv.cta),17           o.maximum_bid_cents18    FROM offers o19    LEFT JOIN vendors v ON v.domain = o.notification_target20    JOIN category_vibes cv ON cv.category = COALESCE(v.category, 'general')21    WHERE NOT EXISTS (SELECT 1 FROM ad_creatives a WHERE a.offer_id = o.offer_id)22    LIMIT {MAX_CREATIVES};23
24    INSERT INTO ad_creatives (offer_id, customer, category, domain, headline, body, max_bid_cents, embedding)25    SELECT s.offer_id, s.customer, s.category, s.domain, s.headline, s.body, s.max_bid_cents,26           {embed_expr}27    FROM creatives_staging s;28
29    DELETE FROM creatives_staging;30END31''')32
33run_sql('''34CREATE OR REPLACE PROCEDURE refresh_ads() AS35BEGIN36    -- One offer per advertiser domain, generated the same way as the original demo app.37    INSERT IGNORE INTO offers38    SELECT CRC32(v.domain),39           v.vendor,40           TRUE,41           NULL,42           '[]',43           CONCAT(10 + MOD(CRC32(v.domain), 41), '% off at ', v.vendor),44           v.domain,45           2 + MOD(CRC32(v.domain), 14)46    FROM vendors v47    WHERE v.vendor IN (SELECT DISTINCT vendor FROM purchases);48
49    CALL build_creatives();50END51''')52
53run_sql('''54CREATE OR REPLACE PROCEDURE ingest_purchases(batch QUERY(subscriber_id BIGINT, vendor VARCHAR(100))) AS55BEGIN56    INSERT INTO purchases (subscriber_id, vendor) SELECT subscriber_id, vendor FROM batch;57    CALL refresh_ads();58END59''')60
61def embed_pending():62    """Local-model fallback: fill any NULL embeddings from Python."""63    if EMBED_MODE != 'local':64        return65    todo = run_sql('SELECT offer_id, customer, domain, headline, body FROM ad_creatives WHERE embedding IS NULL')66    if todo is None or todo.empty:67        return68    texts = [(h + '. ' + b).replace(d, '').replace(c, '')      # same brand-name stripping as the SQL path69             for h, b, d, c in zip(todo['headline'], todo['body'], todo['domain'], todo['customer'])]70    vecs = local_model.encode(texts, normalize_embeddings=True)71    with conn.cursor() as cur:72        cur.executemany('UPDATE ad_creatives SET embedding = %s WHERE offer_id = %s',73                        [(json.dumps([float(x) for x in v]), int(i)) for v, i in zip(vecs, todo['offer_id'])])74    print(f'Embedded {len(todo)} ads locally.')75
76print('Procedures ready.')

1DEMO_DB = 'martech'   # the database created by the Real-Time Digital Marketing demo2
3def load_from_s3():4    BUCKET_URL = 'https://singlestore-realtime-digital-marketing.s3.amazonaws.com'5
6    try:7        run_sql(f'''8            CREATE OR REPLACE PIPELINE purchases_pipeline AS9            LOAD DATA S3 'singlestore-realtime-digital-marketing/v2/{SCALE}/purchases.*'10            CREDENTIALS '{{}}'11            CONFIG '{{"region": "us-east-1"}}'12            INTO PROCEDURE ingest_purchases13            FORMAT PARQUET (14                subscriber_id <- subscriberid,15                vendor <- vendor16            )17        ''')18        t0 = time.perf_counter()19        run_sql('START PIPELINE purchases_pipeline FOREGROUND')20        print(f'Pipeline finished in {time.perf_counter() - t0:.1f}s')21    except Exception as e:22        # Fallback: read a few of the same files over HTTPS and run the same procedure logic.23        import io, urllib.request24        print('Pipeline not available here, loading over HTTPS instead ->', e)25        listing = urllib.request.urlopen(f'{BUCKET_URL}/?list-type=2&prefix=v2/{SCALE}/purchases.').read().decode()26        for key in re.findall(r'<Key>(.*?)</Key>', listing)[:10]:27            df = pd.read_parquet(io.BytesIO(urllib.request.urlopen(f'{BUCKET_URL}/{key}').read()))28            with conn.cursor() as cur:29                cur.executemany('INSERT INTO purchases (subscriber_id, vendor) VALUES (%s, %s)',30                                df[['subscriberid', 'vendor']].values.tolist())31        run_sql('CALL refresh_ads()')32
33
34def load_from_demo():35    """Copy up to MAX_CREATIVES offers and recent purchases from the demo's database, then build the ads.36
37    Copying no more offers than build_creatives() turns into ads keeps offers and ads in step,38    so no leftover offers sneak into the Chapter 6 waves."""39    try:40        run_sql(f'''41            INSERT IGNORE INTO offers42            SELECT offer_id, customer, enabled, notification_zone, segment_ids,43                   notification_content, notification_target, maximum_bid_cents44            FROM {DEMO_DB}.offers45            ORDER BY offer_id LIMIT {MAX_CREATIVES}46        ''')47    except Exception:48        # Older demo versions store the zone as GEOGRAPHY; we don't need it, so skip it.49        run_sql(f'''50            INSERT IGNORE INTO offers51            SELECT offer_id, customer, enabled, NULL, segment_ids,52                   notification_content, notification_target, maximum_bid_cents53            FROM {DEMO_DB}.offers54            ORDER BY offer_id LIMIT {MAX_CREATIVES}55        ''')56    run_sql(f'''57        INSERT INTO purchases (subscriber_id, vendor, ts)58        SELECT subscriber_id, vendor, ts FROM {DEMO_DB}.purchases59        ORDER BY ts DESC LIMIT 10000060    ''')61    run_sql('CALL build_creatives()')62
63found = run_sql('''64    SELECT COUNT(*) AS n FROM information_schema.TABLES65    WHERE TABLE_SCHEMA = %s AND TABLE_NAME IN ('offers', 'purchases')66''', (DEMO_DB,)).iloc[0, 0]67demo_offers = run_sql(f'SELECT COUNT(*) FROM {DEMO_DB}.offers').iloc[0, 0] if found == 2 else 068
69if demo_offers > 0:70    print(f'Found the demo database `{DEMO_DB}` with {demo_offers} offers. Using its live data.')71    t0 = time.perf_counter()72    load_from_demo()73    print(f'Loaded in {time.perf_counter() - t0:.1f}s')74else:75    print(f'No demo database `{DEMO_DB}` found (or it has no offers yet). Streaming purchases from S3 instead.')76    load_from_s3()77
78embed_pending()79run_sql('''80    SELECT (SELECT COUNT(*) FROM purchases)    AS purchases_loaded,81           (SELECT COUNT(*) FROM offers)       AS offers_created,82           (SELECT COUNT(*) FROM ad_creatives) AS ads_created83''')

1run_sql('''2    SELECT offer_id, customer, category, headline, body3    FROM ad_creatives4    ORDER BY RAND()5    LIMIT 56''')

1if not IS_SHARED:2    run_sql(f'USE {DB_NAME}')   # stay in the notebook's database, whatever the drop-down says3
4run_sql('''5    ALTER TABLE ad_creatives6    ADD FULLTEXT USING VERSION 2 ft_ads (customer, headline, body)7    INDEX_OPTIONS '{"analyzer": "standard"}'8''')9run_sql('OPTIMIZE TABLE ad_creatives FLUSH')   # make the newest rows searchable right away10print('Full-text index added.')

1def fts_status():2    """One-line summary of SHOW FULLTEXT INDEX STATUS. Returns the lowest build progress (%)."""3    df = try_sql('SHOW FULLTEXT INDEX STATUS FOR ad_creatives', label='SHOW FULLTEXT INDEX STATUS')4    if df is None or df.empty:5        return None6    progress, broken = pd.to_numeric(df['Progress (%)']), pd.to_numeric(df['Broken (%)'])7    print(f"Full-text index: {progress.min():.0f}% built on all {df['Partition'].nunique()} partitions, "8          f"{broken.max():.0f}% broken")9    return progress.min()10
11fts_status()

1q = 'body:"wireless headphones" OR headline:"wireless headphones"'2run_sql('''3    SELECT customer, category, headline, ROUND(BM25(ad_creatives, %(q)s), 3) AS bm254    FROM ad_creatives5    WHERE BM25(ad_creatives, %(q)s)6    ORDER BY bm25 DESC7    LIMIT 88''', {'q': q})

1run_sql('''2    SELECT customer, headline3    FROM ad_creatives4    WHERE MATCH (TABLE ad_creatives) AGAINST ('body:hedphones~1')5    LIMIT 56''')

1run_sql('''2    SELECT COUNT(*) AS ads_matching_run3    FROM ad_creatives4    WHERE MATCH (TABLE ad_creatives) AGAINST ('body:run')5''')

1phrase = 'Running shoes for runners who ran the marathon'2
3def tokens(options=None):4    """Return the tokens ANALYZE FULLTEXT produces for `phrase` (one row per token)."""5    q = 'ANALYZE FULLTEXT %s' + (' OPTIONS %s' if options else '')6    df = try_sql(q, (phrase, options) if options else (phrase,), label='ANALYZE FULLTEXT')7    if df is None or df.empty:8        return None9    if 'token' in df.columns:10        return df['token'].tolist()11    # Some versions return a single JSON document instead12    return [t['token'] for t in json.loads(df.iloc[0, 0])['tokens']]13
14print('standard:', tokens())15print('english: ', tokens('{"analyzer": "english"}'))

1run_sql('DROP INDEX ft_ads ON ad_creatives')2run_sql('''3    ALTER TABLE ad_creatives4    ADD FULLTEXT USING VERSION 2 ft_ads (customer, headline, body)5    INDEX_OPTIONS '{"analyzer": "english", "mappings": {"customer": {"analyzer": "standard"}}}'6''')7run_sql('OPTIMIZE TABLE ad_creatives FLUSH')8
9for _ in range(15):   # wait for the rebuild to finish10    done = fts_status()11    if done is None or done >= 100:12        break13    time.sleep(2)

1run_sql('''2    SELECT customer, category, headline3    FROM ad_creatives4    WHERE MATCH (TABLE ad_creatives) AGAINST ('body:run')5    LIMIT 56''')

1try_sql('''2    ALTER TABLE ad_creatives3    ADD VECTOR INDEX ivx (embedding)4    INDEX_OPTIONS '{"index_type": "HNSW_FLAT", "metric_type": "DOT_PRODUCT"}'5''', label='Vector index')6
7def set_query_vector(text):8    """Embed text into the session variable @qv, using the same model as the ads."""9    if EMBED_MODE == 'ai':10        run_sql(f'SET @qv = cluster.EMBED_TEXT(%s) :> VECTOR({DIM}) :> {VEC_TYPE}', (text,))11    else:12        v = local_model.encode([text], normalize_embeddings=True)[0]13        run_sql(f'SET @qv = %s :> {VEC_TYPE}', (json.dumps([float(x) for x in v]),))14
15def vibe_search(text, k=8):16    set_query_vector(text)17    return run_sql(f'''18        SELECT customer, category, headline, ROUND(embedding <*> @qv, 3) AS similarity19        FROM ad_creatives20        ORDER BY similarity DESC21        LIMIT {k}22    ''')23
24vibe_search('a rainy Sunday')

1for mood in ['getting ready for a romantic date night', 'the car is packed and the playlist is ready', 'I need to fix something this weekend']:2    print('Mood:', mood)3    display(vibe_search(mood, k=4))

1REPEAT = 20   # copies of each ad; raise for bigger tests2
3for t in ['bench_f16', 'bench_f32']:4    run_sql(f'DROP TABLE IF EXISTS {t}')5run_sql(f'CREATE TABLE bench_f16 (id BIGINT, embedding {VEC_TYPE}, SHARD KEY (id))')6run_sql(f'CREATE TABLE bench_f32 (id BIGINT, embedding VECTOR({DIM}), SHARD KEY (id))')7
8for r in range(REPEAT):9    run_sql(f'INSERT INTO bench_f16 SELECT offer_id * 1000 + {r}, embedding FROM ad_creatives WHERE embedding IS NOT NULL')10run_sql(f'INSERT INTO bench_f32 SELECT id, embedding :> VECTOR({DIM}) FROM bench_f16')11run_sql('OPTIMIZE TABLE bench_f16 FULL')12run_sql('OPTIMIZE TABLE bench_f32 FULL')13
14sizes = try_sql('''15    SELECT TABLE_NAME AS table_name,16           SUM(ROWS_COUNT) AS rows_stored,17           ROUND(SUM(COMPRESSED_SIZE) / 1024 / 1024, 2) AS embedding_mb_on_disk18    FROM information_schema.COLUMNAR_SEGMENTS19    WHERE DATABASE_NAME = DATABASE() AND TABLE_NAME IN ('bench_f16', 'bench_f32') AND COLUMN_NAME = 'embedding'20    GROUP BY TABLE_NAME21''', label='Storage comparison')22sizes

1def median_ms(query, runs=7):2    times = []3    for _ in range(runs):4        t0 = time.perf_counter()5        run_sql(query)6        times.append((time.perf_counter() - t0) * 1000)7    return round(statistics.median(times), 1)8
9moods = ['a rainy Sunday', 'date night', 'road trip', 'home workout', 'birthday gift for a kid']10overlap, t16, t32 = [], [], []11for mood in moods:12    set_query_vector(mood)13    run_sql(f'SET @qv32 = @qv :> VECTOR({DIM})')14    # speed: exact kNN over every row (no vector index on the bench tables)15    t16.append(median_ms('SELECT id FROM bench_f16 ORDER BY embedding <*> @qv DESC LIMIT 10'))16    t32.append(median_ms('SELECT id FROM bench_f32 ORDER BY embedding <*> @qv32 DESC LIMIT 10'))17    # quality: do F16 and F32 pick the same 10 distinct ads? (each ad is repeated REPEAT times)18    top = lambda t, v: set(run_sql(f'SELECT id DIV 1000 AS ad, MAX(embedding <*> {v}) AS s FROM {t} '19                               f'GROUP BY ad ORDER BY s DESC LIMIT 10')['ad'])20    overlap.append(len(top('bench_f16', '@qv') & top('bench_f32', '@qv32')) / 10)21
22if VEC_ELEM != 'F16':23    print('Note: F16 is not available on this workspace, so both tables below use F32. '24          'Rerun on a workspace with F16 enabled to see the real difference.')25pd.DataFrame({26    'metric': ['median exact kNN latency (ms)', 'top-10 agreement with F32'],27    f'{VEC_ELEM} (bench_f16)': [statistics.median(t16), f'{statistics.mean(overlap):.0%}'],28    'F32 (bench_f32)': [statistics.median(t32), '100% (baseline)'],29})

1def lucene(text):2    """Turn free text into a safe Lucene query over the indexed columns."""3    words = ' '.join(re.findall(r'[A-Za-z0-9]+', text))4    return f'headline:({words}) OR body:({words})'5
6HYBRID_SQL = '''7WITH kw AS (8    SELECT offer_id, BM25(ad_creatives, %(lq)s) AS s9    FROM ad_creatives10    WHERE BM25(ad_creatives, %(lq)s)11    ORDER BY s DESC LIMIT 5012),13vec AS (14    SELECT offer_id, embedding <*> @qv AS s15    FROM ad_creatives16    ORDER BY s DESC LIMIT 5017),18ranked AS (19    SELECT offer_id, ROW_NUMBER() OVER (ORDER BY s DESC) AS r, 'keyword' AS src FROM kw20    UNION ALL21    SELECT offer_id, ROW_NUMBER() OVER (ORDER BY s DESC) AS r, 'vector' AS src FROM vec22),23fused AS (24    SELECT offer_id,25           SUM(1.0 / (60 + r)) AS rrf,26           GROUP_CONCAT(CONCAT(src, ' #', r) SEPARATOR ', ') AS found_by27    FROM ranked GROUP BY offer_id28)29SELECT a.customer, a.category, a.headline, a.max_bid_cents, ROUND(f.rrf, 4) AS rrf, f.found_by30FROM fused f JOIN ad_creatives a ON a.offer_id = f.offer_id31ORDER BY f.rrf DESC32LIMIT 1033'''34
35def hybrid_search(text):36    set_query_vector(text)37    return run_sql(HYBRID_SQL, {'lq': lucene(text)})38
39hybrid_search('cozy headphones for a rainy day')

1set_query_vector('cozy headphones for a rainy day')2run_sql(HYBRID_SQL.replace('ORDER BY f.rrf DESC', 'WHERE a.max_bid_cents >= 10\nORDER BY f.rrf DESC'),3    {'lq': lucene('cozy headphones for a rainy day')})

1def index_segments():2    """Each columnstore segment carries its own vector and full-text index,3    so counting segments shows how many small indexes a query has to probe."""4    df = try_sql('''5        SELECT COUNT(DISTINCT PARTITION, SEGMENT_ID) AS segments,6               SUM(ROWS_COUNT - DELETED_ROWS_COUNT) AS live_rows7        FROM information_schema.COLUMNAR_SEGMENTS8        WHERE DATABASE_NAME = DATABASE() AND TABLE_NAME = 'ad_creatives' AND COLUMN_NAME = 'offer_id'9    ''', label='Segment counts')10    if df is not None:11        display(df)12
13print('Before kickoff:')14index_segments()

1WAVES, ADS_PER_WAVE = 12, 100   # 1,200 new ads; lower these to save AI Function quota2
3templates = run_sql('SELECT notification_zone, segment_ids FROM offers LIMIT 1').iloc[0]4next_id = int(run_sql('SELECT MAX(offer_id) AS m FROM offers').iloc[0, 0]) + 15vendor_rows = vendors[['vendor', 'domain']].values.tolist()6
7for wave in range(1, WAVES + 1):8    rows = []9    for _ in range(ADS_PER_WAVE):10        vendor, domain = random.choice(vendor_rows)11        rows.append((next_id, vendor, True, templates['notification_zone'], json.dumps([]),12                     f'{random.randint(10, 70)}% off at {vendor} during the big game', domain,13                     random.randint(5, 40)))14        next_id += 115    with conn.cursor() as cur:16        cur.executemany('INSERT INTO offers VALUES (%s, %s, %s, %s, %s, %s, %s, %s)', rows)17    run_sql('CALL build_creatives()')18    embed_pending()19    run_sql('OPTIMIZE TABLE ad_creatives FLUSH')   # each wave becomes its own segment20    print(f'Wave {wave}: {ADS_PER_WAVE} ads live')21
22print('\nAfter the rush:')23index_segments()

1QUERY = 'snacks and a big screen for the game'2set_query_vector(QUERY)3LQ = lucene(QUERY)4
5VECTOR_ONLY = 'SELECT offer_id FROM ad_creatives ORDER BY embedding <*> @qv DESC LIMIT 10'6KEYWORD_ONLY = '''7    SELECT offer_id, customer, BM25(ad_creatives, %(lq)s) AS s8    FROM ad_creatives WHERE BM25(ad_creatives, %(lq)s)9    ORDER BY s DESC LIMIT 1010'''11
12def latency_ms(query, params=None, runs=9):13    times = []14    for _ in range(runs):15        t0 = time.perf_counter()16        run_sql(query, params)17        times.append((time.perf_counter() - t0) * 1000)18    return round(statistics.median(times), 1)19
20results = {'vector search (ms)': [latency_ms(VECTOR_ONLY)],21           'keyword search (ms)': [latency_ms(KEYWORD_ONLY, {'lq': LQ})]}22top_keyword_before = run_sql(KEYWORD_ONLY, {'lq': LQ})23pd.DataFrame(results, index=['after the rush'])

1try_sql('OPTIMIZE TABLE ad_creatives VECTOR_INDEX ivx', label='Vector index merge')2results['vector search (ms)'].append(latency_ms(VECTOR_ONLY))3print(f"Vector search: {results['vector search (ms)'][0]} ms -> {results['vector search (ms)'][1]} ms")

1settings = run_sql("SHOW VARIABLES WHERE Variable_name IN ('enable_fts2_merger', 'background_vector_index_merger')")2display(settings)3
4merged = try_sql('''5    SELECT PARTITION, FTS_INDEX_ID, NUM_SEGMENTS, NUM_ROWS, SIZE6    FROM information_schema.MV_FTS_MERGED_INDEXES7    WHERE DATABASE_NAME = DATABASE() AND TABLE_NAME = 'ad_creatives'8    ORDER BY PARTITION9''', label='MV_FTS_MERGED_INDEXES')10
11if merged is not None and len(merged) > 0:12    print(f'The full-text merger combined per-segment indexes into {len(merged)} merged index(es):')13    display(merged)14else:15    print('No merged full-text indexes on this workspace yet. The full-text index merger is new and still rolling out; '16          'once it is active, this view fills in automatically as new segments arrive. Nothing to change in the notebook.')17
18fts_status()19
20results['keyword search (ms)'].append(latency_ms(KEYWORD_ONLY, {'lq': LQ}))21print(f"Keyword search: {results['keyword search (ms)'][0]} ms -> {results['keyword search (ms)'][1]} ms")22print('(Background merges of data segments and warm caches also affect this number, so compare several runs.)')

1top_keyword_after = run_sql(KEYWORD_ONLY, {'lq': LQ})2pd.DataFrame({3    'before: advertiser': top_keyword_before['customer'].values,4    'before: bm25': top_keyword_before['s'].round(3).values,5    'after: advertiser': top_keyword_after['customer'].values,6    'after: bm25': top_keyword_after['s'].round(3).values,7})

1try_sql('OPTIMIZE TABLE ad_creatives', label='Columnstore merge')2index_segments()3pd.DataFrame(results, index=['after the rush', 'after merging'])

1try_sql('DROP PIPELINE IF EXISTS purchases_pipeline', label='Drop pipeline')2if not IS_SHARED:3    run_sql(f'DROP DATABASE IF EXISTS {DB_NAME}')4else:5    for t in ['bench_f16', 'bench_f32', 'ad_creatives', 'creatives_staging', 'offers', 'purchases', 'category_vibes', 'vendors']:6        run_sql(f'DROP TABLE IF EXISTS {t}')7    for p in ['ingest_purchases', 'refresh_ads', 'build_creatives']:8        try_sql(f'DROP PROCEDURE IF EXISTS {p}', label=f'Drop {p}')9print('All clean. Morgan will never know.')