RAG on Parquet: How to Build Retrieval Over Your Data Lake Without Embeddings

# Step 1: Read Parquet files
import pandas as pd
import pyarrow.parquet as pq

df = pq.read_table("s3://my-bucket/curated/sales/").to_pandas()

# Step 2: Convert rows to text chunks
chunks = []
for _, row in df.iterrows():
    chunk = f"Sale: {row['product']} in {row['region']} for ${row['amount']} on {row['date']}"
    chunks.append(chunk)

# Step 3: Embed chunks (this is expensive)
from openai import OpenAI
client = OpenAI()

embeddings = []
# 50M rows × $0.0001/1K tokens = ~$500+ just for embeddings
for batch in batched(chunks, 100):
    resp = client.embeddings.create(model="text-embedding-3-small", input=batch)
    embeddings.extend([e.embedding for e in resp.data])

# Step 4: Store in vector database
import chromadb
chroma = chromadb.Client()
collection = chroma.create_collection("sales")
collection.add(documents=chunks, embeddings=embeddings, ids=[str(i) for i in range(len(chunks))])

# Step 5: Query
results = collection.query(query_texts=["total sales last quarter"], n_results=10)
# Returns 10 individual sale records, not a SUM

from dataspoc_lens import LensClient

lens = LensClient()

# No embedding. No vector store. Just SQL over your existing Parquet files.
answer = lens.ask("What were total sales last quarter?")
print(answer)
# → Executed: SELECT SUM(amount) FROM curated_sales WHERE sale_date >= '2026-01-01'
# → Result: $4,230,000

"""RAG-based data Q&A over Parquet files."""
import pandas as pd
import chromadb
from openai import OpenAI

client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")


def build_rag_index(parquet_path: str, collection_name: str):
    """Embed Parquet data into ChromaDB. Run once (or on every update)."""
    df = pd.read_parquet(parquet_path)
    collection = chroma_client.get_or_create_collection(collection_name)

    # Convert each row to a text document
    documents = []
    for _, row in df.iterrows():
        doc = " | ".join(f"{col}: {val}" for col, val in row.items())
        documents.append(doc)

    # Batch embed and store
    batch_size = 100
    for i in range(0, len(documents), batch_size):
        batch = documents[i : i + batch_size]
        ids = [str(j) for j in range(i, i + len(batch))]
        collection.add(documents=batch, ids=ids)

    return collection


def rag_query(question: str, collection_name: str) -> str:
    """Retrieve relevant rows and ask LLM to answer."""
    collection = chroma_client.get_collection(collection_name)
    results = collection.query(query_texts=[question], n_results=20)

    context = "\n".join(results["documents"][0])

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Answer based on the data provided. If you cannot determine the exact answer from the context, say so."},
            {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {question}"},
        ],
    )
    return response.choices[0].message.content


# Usage
collection = build_rag_index("s3://bucket/curated/sales/data.parquet", "sales")
answer = rag_query("What were total sales last quarter?", "sales")
print(answer)
# → "Based on the retrieved records, I can see several sales entries...
#    I cannot provide an exact total as I only have a sample of 20 records."

"""SQL-based data Q&A over Parquet files with DataSpoc Lens."""
from dataspoc_lens import LensClient

lens = LensClient()


def sql_query(question: str) -> str:
    """Ask a question, get a precise answer via SQL."""
    return lens.ask(question)


# Direct natural language
answer = sql_query("What were total sales last quarter?")
print(answer)
# → "$4,230,512.00 — Query: SELECT SUM(amount) FROM curated_sales
#    WHERE sale_date >= '2026-01-01' AND sale_date < '2026-04-01'"

# Or explicit SQL for full control
df = lens.query("""
    SELECT
        region,
        DATE_TRUNC('month', sale_date) AS month,
        SUM(amount) AS total_sales,
        COUNT(*) AS num_orders,
        AVG(amount) AS avg_order_value
    FROM curated_sales
    WHERE sale_date >= '2025-01-01'
    GROUP BY region, month
    ORDER BY region, month
""")
print(df)

# Pre-warm the cache for tables you know will be queried
lens.cache_refresh("curated_sales")
lens.cache_refresh("curated_customers")

# Check cache status
status = lens.cache_status()
print(status)
# {'curated_sales': {'cached': True, 'rows': 10000000, 'size_mb': 450}}

# Hybrid: Let the agent choose between RAG and SQL
from dataspoc_lens import LensClient

lens = LensClient()

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_structured_data",
            "description": "Query structured data (sales, orders, metrics) using SQL.",
            "parameters": {
                "type": "object",
                "properties": {"sql": {"type": "string"}},
                "required": ["sql"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_documents",
            "description": "Search unstructured documents (policies, manuals, tickets).",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
]