Overview

Download the code:

• SQL code
• Python Notebook

What You'll Need

What You'll Learn

What You'll Build

What is Batch Cortex Search?

NOTE: The throughput of the batch search function may vary depending on the amount of data indexed in the queried Cortex Search Service and the complexity of the search queries.

The throughput of the batch search function (the number of search queries processed per second) is not influenced by the size of the warehouse used to query it and there is no limit to the number of concurrent batch queries that can be run at a given time on a given service.

Use Cases

1. Entity Resolution

---

2. Catalog Mapping

---

3. Deduplication

---

4. Audience Matching

---

5. Content Tagging & Classification

---

6. Similar Item Discovery

---

Getting Data

NOTE: Skip this if you have your own data and prefer to use it for this demo instead of the marketplace data.

USE ROLE ACCOUNTADMIN;

CALL SYSTEM$ACCEPT_LEGAL_TERMS('DATA_EXCHANGE_LISTING', 'GZT0Z4C8RF3FT');
CREATE DATABASE IF NOT EXISTS AI_TRAINING_DATASET_FROM_WIKIPEDIA
FROM LISTING 'GZT0Z4C8RF3FT';

-- Set up environment
USE ROLE ACCOUNTADMIN;

-- Create database and warehouse
CREATE DATABASE IF NOT EXISTS BATCH_DEMO;
CREATE WAREHOUSE IF NOT EXISTS BATCH_SEARCH_WH WITH WAREHOUSE_SIZE = 'SMALL' AUTO_SUSPEND = 300 AUTO_RESUME = TRUE;

-- Set context
USE DATABASE BATCH_DEMO;
CREATE SCHEMA IF NOT EXISTS PUBLIC;
USE SCHEMA PUBLIC;
USE WAREHOUSE BATCH_SEARCH_WH;

-- Create a table based on 100k articles from the source data
CREATE OR REPLACE TABLE wikipedia_articles AS
SELECT 
    URL,
    TITLE,
    RAW_TEXT,
    TABLE_OF_CONTENTS,
    SEE_ALSO
FROM AI_TRAINING_DATASET_FROM_WIKIPEDIA.PUBLIC.WIKIPEDIA
LIMIT 100000;

-- Verify row count
SELECT COUNT(*) AS row_count FROM wikipedia_articles;

-- Preview sample data
SELECT TITLE, LEFT(RAW_TEXT, 200) AS text_preview 
FROM wikipedia_articles 
LIMIT 5;

Setting Up a Cortex Search Service

NOTE: This will take around 8 minutes. If you want something faster, simply reduce the table wikipedia_articles to a smaller amount of data, such as 50k.

-- Create a Cortex Search Service
CREATE OR REPLACE CORTEX SEARCH SERVICE wikipedia_search_service
ON RAW_TEXT
ATTRIBUTES TITLE
WAREHOUSE = BATCH_SEARCH_WH
TARGET_LAG = '1 hour'
AS
SELECT 
    TITLE,
    RAW_TEXT
FROM wikipedia_articles;

-- Verify service creation
SHOW CORTEX SEARCH SERVICES;

-- Single search against the Cortex Search service
SELECT SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{
        "query": "machine learning artificial intelligence",
        "columns": ["TITLE", "RAW_TEXT"],
        "limit": 3
    }'
) AS test_result;

-- A single result is a little hard to parse as JSON. We can wrap SEARCH_PREVIEW result into a table.
SELECT 
    r.value:TITLE::VARCHAR AS title,
    LEFT(r.value:RAW_TEXT::VARCHAR, 200) AS raw_text_preview
FROM TABLE(FLATTEN(
    input => PARSE_JSON(
        SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
            'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
            '{
                "query": "machine learning artificial intelligence",
                "columns": ["TITLE", "RAW_TEXT"],
                "limit": 5
            }'
        )
    ):results
)) AS r;

The Problem: No Elegant Way to Do Multiple Searches

-- Example: 10 searches using UNION ALL (verbose and doesn't scale)
SELECT 'quantum physics' AS query, PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "quantum physics", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR AS top_result
UNION ALL
SELECT 'world war history', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "world war history", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'climate change', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "climate change", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'solar system', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "solar system", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'ancient rome', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "ancient rome", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'machine learning', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "machine learning", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'french revolution', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "french revolution", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'deep ocean', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "deep ocean", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'renewable energy', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "renewable energy", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR
UNION ALL
SELECT 'human brain', PARSE_JSON(SNOWFLAKE.CORTEX.SEARCH_PREVIEW(
    'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE',
    '{"query": "human brain", "columns": ["TITLE"], "limit": 1}'
)):results[0]:TITLE::VARCHAR;

Batch Cortex Search to the Rescue

Batch the Same 10 Searches

NOTE: Run all three statements together as one block.

CREATE OR REPLACE TEMPORARY TABLE ten_searches (query_text VARCHAR);

INSERT INTO ten_searches VALUES 
    ('quantum physics'),
    ('world war history'),
    ('climate change'),
    ('solar system'),
    ('ancient rome'),
    ('machine learning'),
    ('french revolution'),
    ('deep ocean'),
    ('renewable energy'),
    ('human brain');

SELECT 
    q.query_text, 
    s.* 
FROM ten_searches AS q,
LATERAL CORTEX_SEARCH_BATCH(
    service_name => 'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE', 
    query => q.query_text, 
    limit => 1
) AS s;

TIP: The throughput of CORTEX_SEARCH_BATCH is not influenced by the size of the warehouse used to query it. A small warehouse works just as well as a large one for batch search operations.

Scale to 1000 Searches

NOTE: Run these statements together as one block.

-- Create a table of 1000 diverse search queries from Wikipedia titles
CREATE OR REPLACE TABLE thousand_searches AS 
SELECT DISTINCT TITLE AS query_text 
FROM wikipedia_articles 
ORDER BY RANDOM() 
LIMIT 1000;

-- Verify
SELECT COUNT(*) AS total_queries FROM thousand_searches;

-- Run batch search on all 1000 queries (returns up to 5000 results: 1000 queries x 5 results each)
SELECT 
    q.query_text, 
    s.* 
FROM thousand_searches AS q,
LATERAL CORTEX_SEARCH_BATCH(
    service_name => 'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE', 
    query => q.query_text, 
    limit => 5
) AS s;

TIP: You can run interactive and batch queries concurrently on the same Cortex Search Service without any degradation to interactive query performance. Separate compute resources are used to serve each type.

Bonus: Python SDK Comparison

NOTE: This section requires running in a Snowflake Notebook environment. Create a new notebook in Workspaces and copy cells below.

Setup

import time
import pandas as pd
from snowflake.snowpark.context import get_active_session
from snowflake.core import Root

# Get active Snowsight session
session = get_active_session()

# Set database context
session.sql("USE DATABASE BATCH_DEMO").collect()
session.sql("USE SCHEMA PUBLIC").collect()

# Connect to Cortex Search service via SDK
root = Root(session)
search_service = root.databases["BATCH_DEMO"].schemas["PUBLIC"].cortex_search_services["WIKIPEDIA_SEARCH_SERVICE"]

SERVICE_NAME = 'BATCH_DEMO.PUBLIC.WIKIPEDIA_SEARCH_SERVICE'

print(f"Connected to: {SERVICE_NAME}")

Test 1: 100 Queries

NUM_QUERIES_100 = 100

queries_df_100 = session.sql(f"""
    SELECT TITLE AS query_text 
    FROM BATCH_DEMO.PUBLIC.WIKIPEDIA_ARTICLES 
    ORDER BY RANDOM() 
    LIMIT {NUM_QUERIES_100}
""").to_pandas()

search_queries_100 = queries_df_100['QUERY_TEXT'].tolist()
print(f"Generated {len(search_queries_100)} search queries")
print(f"Sample: {search_queries_100[:3]}")

print("Running 100 interactive searches...")

interactive_results_100 = []
test_start = time.time()

for i, query in enumerate(search_queries_100):
    try:
        user_start = time.time()
        resp = search_service.search(query=query, columns=["RAW_TEXT"], limit=5)
        _ = resp.to_json()
        user_latency = (time.time() - user_start) * 1000
        interactive_results_100.append({'query_num': i + 1, 'latency_ms': user_latency})
        if (i + 1) % 20 == 0:
            print(f"  Completed {i + 1}/100 queries...")
    except Exception as e:
        print(f"  Error on query {i + 1}: {str(e)[:50]}")

interactive_total_100 = time.time() - test_start
df_interactive_100 = pd.DataFrame(interactive_results_100)

print(f"\nINTERACTIVE (100 queries):")
print(f"  Total time: {interactive_total_100:.1f}s")
print(f"  Avg latency: {df_interactive_100['latency_ms'].mean():.0f}ms")
print(f"  Throughput: {NUM_QUERIES_100 / interactive_total_100:.1f} queries/sec")

print("Running 100 batch searches in ONE call...")

session.sql("CREATE OR REPLACE TEMPORARY TABLE temp_100_queries (query_text VARCHAR)").collect()
values = ",".join([f"('{q.replace(chr(39), chr(39)+chr(39))}')" for q in search_queries_100])
session.sql(f"INSERT INTO temp_100_queries VALUES {values}").collect()

batch_start = time.time()
batch_sql = f"""
    SELECT q.query_text, s.* 
    FROM temp_100_queries AS q, 
    LATERAL CORTEX_SEARCH_BATCH(
        service_name => '{SERVICE_NAME}', 
        query => q.query_text, 
        limit => 5
    ) AS s
"""
batch_results_100 = session.sql(batch_sql).collect()
batch_total_100 = time.time() - batch_start

print(f"\nBATCH (100 queries):")
print(f"  Total time: {batch_total_100:.1f}s")
print(f"  Results returned: {len(batch_results_100)} rows")
print(f"  Throughput: {NUM_QUERIES_100 / batch_total_100:.1f} queries/sec")

Test 2: 1,000 Queries

NUM_QUERIES_1000 = 1000

queries_df_1000 = session.sql(f"""
    SELECT TITLE AS query_text 
    FROM BATCH_DEMO.PUBLIC.WIKIPEDIA_ARTICLES 
    ORDER BY RANDOM() 
    LIMIT {NUM_QUERIES_1000}
""").to_pandas()

search_queries_1000 = queries_df_1000['QUERY_TEXT'].tolist()
print(f"Generated {len(search_queries_1000)} search queries")

print("Running 1000 interactive searches...")

interactive_results_1000 = []
test_start = time.time()

for i, query in enumerate(search_queries_1000):
    try:
        user_start = time.time()
        resp = search_service.search(query=query, columns=["RAW_TEXT"], limit=5)
        _ = resp.to_json()
        user_latency = (time.time() - user_start) * 1000
        interactive_results_1000.append({'query_num': i + 1, 'latency_ms': user_latency})
        if (i + 1) % 200 == 0:
            print(f"  Completed {i + 1}/1000 queries...")
    except Exception as e:
        print(f"  Error on query {i + 1}: {str(e)[:50]}")

interactive_total_1000 = time.time() - test_start
df_interactive_1000 = pd.DataFrame(interactive_results_1000)

print(f"\nINTERACTIVE (1000 queries):")
print(f"  Total time: {interactive_total_1000:.1f}s")
print(f"  Avg latency: {df_interactive_1000['latency_ms'].mean():.0f}ms")
print(f"  Throughput: {NUM_QUERIES_1000 / interactive_total_1000:.1f} queries/sec")

print("Running 1000 batch searches in ONE call...")

session.sql("CREATE OR REPLACE TEMPORARY TABLE temp_1000_queries (query_text VARCHAR)").collect()

# Insert in chunks to avoid SQL size limits
chunk_size = 500
for i in range(0, len(search_queries_1000), chunk_size):
    chunk = search_queries_1000[i:i+chunk_size]
    values = ",".join([f"('{q.replace(chr(39), chr(39)+chr(39))}')" for q in chunk])
    session.sql(f"INSERT INTO temp_1000_queries VALUES {values}").collect()

batch_start = time.time()
batch_sql = f"""
    SELECT q.query_text, s.* 
    FROM temp_1000_queries AS q, 
    LATERAL CORTEX_SEARCH_BATCH(
        service_name => '{SERVICE_NAME}', 
        query => q.query_text, 
        limit => 5
    ) AS s
"""
batch_results_1000 = session.sql(batch_sql).collect()
batch_total_1000 = time.time() - batch_start

print(f"\nBATCH (1000 queries):")
print(f"  Total time: {batch_total_1000:.1f}s")
print(f"  Results returned: {len(batch_results_1000)} rows")
print(f"  Throughput: {NUM_QUERIES_1000 / batch_total_1000:.1f} queries/sec")

Test 3: 10,000 Queries

WARNING: We do not recommend running interactive search tests for 5,000+ queries. Based on our benchmarks, interactive search runs at ~350ms per query, meaning:

• 5,000 queries → ~29 minutes
• 10,000 queries → ~58 minutes

The batch search results below complete in under 2 minutes total. Trust the benchmark data in the next section!

NUM_QUERIES_10000 = 10000

print("Generating 10,000 search queries...")
queries_df_10000 = session.sql(f"""
    SELECT TITLE AS query_text 
    FROM BATCH_DEMO.PUBLIC.WIKIPEDIA_ARTICLES 
    ORDER BY RANDOM() 
    LIMIT {NUM_QUERIES_10000}
""").to_pandas()

search_queries_10000 = queries_df_10000['QUERY_TEXT'].tolist()
print(f"Generated {len(search_queries_10000)} search queries")

print("Running 10,000 batch searches in ONE call...")

session.sql("CREATE OR REPLACE TEMPORARY TABLE temp_10000_queries (query_text VARCHAR)").collect()

# Insert in chunks to avoid SQL size limits
chunk_size = 500
for i in range(0, len(search_queries_10000), chunk_size):
    chunk = search_queries_10000[i:i+chunk_size]
    values = ",".join([f"('{q.replace(chr(39), chr(39)+chr(39))}')" for q in chunk])
    session.sql(f"INSERT INTO temp_10000_queries VALUES {values}").collect()
    if (i + chunk_size) % 2000 == 0:
        print(f"  Inserted {min(i + chunk_size, len(search_queries_10000))}/{len(search_queries_10000)} queries...")

batch_start = time.time()
batch_sql = f"""
    SELECT q.query_text, s.* 
    FROM temp_10000_queries AS q, 
    LATERAL CORTEX_SEARCH_BATCH(
        service_name => '{SERVICE_NAME}', 
        query => q.query_text, 
        limit => 5
    ) AS s
"""
batch_results_10000 = session.sql(batch_sql).collect()
batch_total_10000 = time.time() - batch_start

print(f"\nBATCH (10,000 queries):")
print(f"  Total time: {batch_total_10000:.1f}s")
print(f"  Results returned: {len(batch_results_10000)} rows")
print(f"  Throughput: {NUM_QUERIES_10000 / batch_total_10000:.1f} queries/sec")

# Estimate how long interactive would take
estimated_interactive = NUM_QUERIES_10000 * (df_interactive_1000['latency_ms'].mean() / 1000)
print(f"\n  Estimated interactive time: {estimated_interactive:.0f}s ({estimated_interactive/60:.1f} minutes)")
print(f"  Batch speedup: ~{estimated_interactive / batch_total_10000:.0f}x faster!")

Final Summary

print("="*70)
print("FINAL SUMMARY: Interactive vs Batch")
print("="*70)

speedup_100 = interactive_total_100 / batch_total_100 if batch_total_100 > 0 else 0
speedup_1000 = interactive_total_1000 / batch_total_1000 if batch_total_1000 > 0 else 0

summary = pd.DataFrame({
    'Queries': [100, 1000, 10000],
    'Interactive (s)': [
        f"{interactive_total_100:.1f}", 
        f"{interactive_total_1000:.1f}", 
        f"~{estimated_interactive:.0f} (est)"
    ],
    'Batch (s)': [
        f"{batch_total_100:.1f}", 
        f"{batch_total_1000:.1f}", 
        f"{batch_total_10000:.1f}"
    ],
    'Speedup': [f"{speedup_100:.1f}x", f"{speedup_1000:.1f}x", f"~{estimated_interactive/batch_total_10000:.0f}x"]
})

print(summary.to_string(index=False))

print("\n" + "="*70)
print("RECOMMENDATION:")
print("="*70)
print("  - For real-time, user-facing search: Use Interactive SDK")
print("  - For batch processing (100+ queries): Use CORTEX_SEARCH_BATCH")
print("  - Batch has a small bootstrap cost - amortize it with larger batches!")

Benchmark Results

NOTE: These results are provided for illustrative purposes only and should not be considered official performance figures.

The Crossover Point

Throughput: The Real Story

Speedup Factor

When to Use Each Approach

NOTE: You can run multiple batch queries on a given service in parallel, achieving higher throughput than with a single Batch Cortex Search call.

Cleanup

-- Set context
USE ROLE ACCOUNTADMIN;
USE DATABASE BATCH_DEMO;
USE SCHEMA PUBLIC;

-- Step 1: Drop the Cortex Search service
DROP CORTEX SEARCH SERVICE IF EXISTS WIKIPEDIA_SEARCH_SERVICE;

-- Step 2: Drop tables
DROP TABLE IF EXISTS wikipedia_articles;
DROP TABLE IF EXISTS thousand_searches;
DROP TABLE IF EXISTS ten_searches;

-- Step 3: Drop database (optional - only if you created it for this demo)
-- DROP DATABASE IF EXISTS BATCH_DEMO;

-- Step 4: Drop warehouse (optional - only if you created it for this demo)
-- DROP WAREHOUSE IF EXISTS BATCH_SEARCH_WH;

-- Step 5: Unsubscribe from the Wikipedia Marketplace listing
DROP DATABASE IF EXISTS AI_TRAINING_DATASET_FROM_WIKIPEDIA;

Conclusion And Resources

What You Learned

Related Resources