RAG Beyond Text: Tables, Code, and Structured Data

Why text-chunking breaks on structured data

Tables, code, and structured records have internal relationships that naive token-based chunking destroys. When you split by token count alone:

Tables fall apart:

Original:

| Customer | Plan | Renewal Date |
|----------|------|--------------|
| Acme Inc | Pro  | 2026-08-15   |
| Beta LLC | Free | 2026-12-20   |

Naive 100-token chunk after "Pro | 2026-" gets cut. The model later retrieves the header row separately from the data row, and loses the semantic link. It might answer "Who is renewing in August?" with "Pro" (the plan, not a customer name) because the context got fragmented.

Code breaks mid-function:

Original:

def calculate_tax(amount, rate):
    """Compute sales tax on an amount."""
    base_tax = amount * rate
    return base_tax

Naive 100-token chunk might split after amount * rate, cutting the function in half. Later, the model retrieves only the incomplete first half and can't understand what the function does. A query like "How do you compute tax?" might not match at all.

Structured records lose relationships:

A JSON document:

{
  "user_id": 42,
  "name": "Alice",
  "contact": {
    "email": "[email protected]",
    "phone": "+1-555-0123"
  }
}

Chunked as plain text, the email might land in one chunk and the user_id in another. A query like "What's Alice's contact?" retrieves email and phone separately, forcing the model to stitch them back together (it usually does, but it's error-prone).

Chunking strategy 1: Tables

Tables are fundamentally 2D; treating them as 1D text destroys their structure. Three practical approaches:

Approach 1A: Keep header + every row as a unit

Each chunk is one table with its full header, plus one or more complete data rows.

def chunk_table_with_headers(rows: list[dict], headers: list[str], chunk_size: int = 3) -> list[str]:
    """
    Chunk a table by grouping rows, keeping headers with each chunk.
    headers: list of column names
    rows: list of dicts, each with keys matching headers
    chunk_size: rows per chunk
    """
    chunks = []
    header_line = " | ".join(headers)

    for i in range(0, len(rows), chunk_size):
        chunk_rows = rows[i:i + chunk_size]
        chunk_text = header_line + "\n"

        for row in chunk_rows:
            row_line = " | ".join(str(row.get(h, "")) for h in headers)
            chunk_text += row_line + "\n"

        chunks.append(chunk_text)

    return chunks

# Example
headers = ["Customer", "Plan", "Renewal Date", "MRR"]
rows = [
    {"Customer": "Acme Inc", "Plan": "Pro", "Renewal Date": "2026-08-15", "MRR": 5000},
    {"Customer": "Beta LLC", "Plan": "Free", "Renewal Date": "2026-12-20", "MRR": 0},
    {"Customer": "Gamma Co", "Plan": "Enterprise", "Renewal Date": "2026-06-30", "MRR": 25000},
    {"Customer": "Delta Ltd", "Plan": "Pro", "Renewal Date": "2026-09-10", "MRR": 7500},
]

chunks = chunk_table_with_headers(rows, headers, chunk_size=2)
# Chunk 1: headers + Acme + Beta
# Chunk 2: headers + Gamma + Delta

Advantage: A query like "Which Pro customer renews in August?" retrieves a chunk containing both the header and the relevant row. No fragmentation.

Disadvantage: If a table has 1000 rows, even with chunk_size=10, you have 100 chunks. Retrieval returns the top-5, which might miss rows further down. Mitigate by also adding row-level filters (e.g., "Pro plan" as metadata).

Approach 1B: Serialize rows as key-value pairs

Instead of table format, convert each row to a prose snippet.

def serialize_table_row_as_prose(row: dict, table_name: str) -> str:
    """Convert a single table row to prose for better embedding relevance."""
    pairs = [f"{key}: {value}" for key, value in row.items()]
    return f"{table_name} record. " + ", ".join(pairs)

# Example
row = {"Customer": "Acme Inc", "Plan": "Pro", "Renewal Date": "2026-08-15", "MRR": 5000}
prose = serialize_table_row_as_prose(row, "Customers")
# Output: "Customers record. Customer: Acme Inc, Plan: Pro, Renewal Date: 2026-08-15, MRR: 5000"

Advantage: Prose chunks embed well because the model can use word associations (Pro + Acme + August). Queries like "Pro plan renewals" are more likely to match.

Disadvantage: You lose the visual structure. A query like "Show me the table" can't reconstruct it. Also, very long rows become very long prose snippets, wasting tokens.

Approach 1C: Store metadata separately

Index table cells as atomic records with rich metadata, then filter before retrieving.

# Each cell/row is a document with metadata
documents = [
    {
        "id": "customer_42",
        "content": "Acme Inc is a Pro plan customer",
        "metadata": {
            "table": "Customers",
            "customer_name": "Acme Inc",
            "plan": "Pro",
            "renewal_date": "2026-08-15",
            "mrr": 5000,
            "plan_tier": "paid"  # Categorical for filtering
        }
    },
    # ... more rows
]

# At retrieval time, filter first, then embed-search
def retrieve_table_with_filters(query: str, metadata_filters: dict, documents: list, vector_db) -> list:
    """
    Filter documents by metadata, then vector search within the filtered set.
    """
    # Apply metadata filters
    filtered_docs = [
        doc for doc in documents
        if all(doc["metadata"].get(k) == v for k, v in metadata_filters.items())
    ]

    # Vector search only over the filtered set
    results = vector_db.search(query, candidates=filtered_docs, top_k=5)
    return results

# Example: "Pro plan renewals in Q3"
# First filter by plan = "Pro"
# Then search for "renewal" and "Q3" within that subset
results = retrieve_table_with_filters(
    query="When do they renew?",
    metadata_filters={"plan": "Pro"},
    documents=documents,
    vector_db=vector_db
)

Advantage: Combines the efficiency of metadata filtering (fast, deterministic) with the flexibility of semantic search.

Disadvantage: Requires setting up a metadata index upfront. Overhead, but pays off at scale.

Chunking strategy 2: Code

Code has explicit structure: functions, classes, imports, docstrings. Chunking should respect these boundaries.

Naive approach (fails):

# Bad: chunk by token count, split mid-function
def process_user(user_data):
    id, name = user_data
    # Function chunk ends here (hit token limit)
    # ... rest of function in next chunk

Better: AST-based chunking

Use the Abstract Syntax Tree to identify function and class boundaries, then chunk along those boundaries.

import ast

def chunk_python_by_ast(code_text: str) -> list[str]:
    """
    Parse Python code as an AST and chunk by function/class definitions.
    Each chunk is a complete top-level function or class.
    """
    try:
        tree = ast.parse(code_text)
    except SyntaxError:
        # Fall back to naive chunking if parsing fails
        return chunk_naive(code_text, token_limit=500)

    chunks = []
    current_line_start = None
    current_line_end = None

    for node in tree.body:
        # ast nodes have lineno and end_lineno attributes
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            start = node.lineno - 1  # Convert to 0-indexed
            end = node.end_lineno    # Inclusive

            # Extract the source lines for this node
            lines = code_text.split('\n')
            chunk_text = '\n'.join(lines[start:end])
            chunks.append(chunk_text)

    return chunks

# Example
code = """
def greet(name):
    return f"Hello, {name}!"

def calculate(a, b):
    total = a + b
    squared = total ** 2
    return squared

class DataProcessor:
    def __init__(self):
        self.data = []

    def add(self, item):
        self.data.append(item)
"""

chunks = chunk_python_by_ast(code)
# Chunk 1: def greet(...)
# Chunk 2: def calculate(...)
# Chunk 3: class DataProcessor(...)

Advantage: A query like "How do you calculate something?" retrieves the calculate function as a complete, self-contained unit. No fragmentation, no confusion.

Disadvantage: If a function is huge (2000+ lines), the chunk might still exceed a typical embedding model's token limit. Mitigate by also respecting a maximum chunk size, then splitting large functions at sensible boundaries (e.g., by method within a class).

Also store metadata:

def chunk_python_with_metadata(code_text: str, filename: str = "unknown.py") -> list[dict]:
    """Chunk Python by AST, and attach metadata for filtering."""
    tree = ast.parse(code_text)
    chunks = []

    for node in tree.body:
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            name = node.name
            docstring = ast.get_docstring(node) or ""
            lines = code_text.split('\n')
            chunk_text = '\n'.join(lines[node.lineno - 1:node.end_lineno])

            chunks.append({
                "text": chunk_text,
                "metadata": {
                    "type": "function",
                    "name": name,
                    "file": filename,
                    "docstring": docstring
                }
            })

        elif isinstance(node, ast.ClassDef):
            name = node.name
            methods = [m.name for m in node.body if isinstance(m, ast.FunctionDef)]
            lines = code_text.split('\n')
            chunk_text = '\n'.join(lines[node.lineno - 1:node.end_lineno])

            chunks.append({
                "text": chunk_text,
                "metadata": {
                    "type": "class",
                    "name": name,
                    "methods": methods,
                    "file": filename
                }
            })

    return chunks

Now a query like "Get all functions that handle user login" can filter by metadata docstring contains "login", then vector search within that subset.

Chunking strategy 3: Structured records and semi-structured data

For JSON, XML, CSV with schemas, the goal is to keep logical units (records, objects) intact.

def chunk_json_by_objects(json_array: list[dict], max_chunk_size: int = 5) -> list[str]:
    """
    Chunk a JSON array by grouping objects, keeping them intact.
    """
    chunks = []

    for i in range(0, len(json_array), max_chunk_size):
        group = json_array[i:i + max_chunk_size]
        chunk_text = json.dumps(group, indent=2)
        chunks.append(chunk_text)

    return chunks

# Example
documents = [
    {
        "doc_id": "policy_001",
        "title": "Return Policy",
        "content": "Items may be returned within 30 days...",
        "version": "2.0"
    },
    {
        "doc_id": "policy_002",
        "title": "Shipping Policy",
        "content": "We ship within 2-3 business days...",
        "version": "1.5"
    },
]

chunks = chunk_json_by_objects(documents)

Alternative: Use document-level chunking with field-level metadata:

def chunk_document_with_field_metadata(doc: dict) -> list[dict]:
    """
    For a complex document, chunk by major section but preserve field-level metadata.
    """
    chunks = []

    # Assume the document has a "sections" field with subsections
    for section in doc.get("sections", []):
        chunks.append({
            "text": section["content"],
            "metadata": {
                "document_id": doc["id"],
                "document_title": doc["title"],
                "section_title": section.get("title"),
                "section_order": section.get("order"),
                "version": doc.get("version", "1.0")
            }
        })

    return chunks

When to route to text-to-SQL or metadata filters instead

Sometimes the question isn't really a "find the semantically relevant chunk" problem. It's a database query problem.

Recognize these patterns:

• "How many customers are on the Pro plan?" — This is a COUNT, not a semantic search.
• "List all invoices from Q2 2026 that are unpaid." — This is a filter + sort, not a semantic search.
• "What's the total MRR from the Enterprise tier?" — This is an aggregation.

For these queries, embedding search is overkill and error-prone. Instead, route to a text-to-SQL converter or a metadata query engine.

Simple router:

def route_query(user_query: str, has_sql_schema: bool = True) -> str:
    """
    Decide: "embedding search" or "sql conversion"?
    """
    sql_keywords = ["count", "total", "sum", "how many", "list all", "where", "filter", "which"]
    semantic_keywords = ["what is", "explain", "how do", "describe", "why"]

    query_lower = user_query.lower()

    # If the query has SQL-like keywords and we have a schema, route to SQL
    if has_sql_schema and any(kw in query_lower for kw in sql_keywords):
        return "sql"

    # Otherwise, use embedding search
    return "embedding_search"

# Example
print(route_query("How many Pro customers are there?"))  # "sql"
print(route_query("What does the Pro plan include?"))    # "embedding_search"

Text-to-SQL sketch:

def text_to_sql_and_execute(user_query: str, schema: str, db_connection) -> str:
    """
    Convert natural language query to SQL, execute, return results.
    """
    prompt = f"""
Given the database schema:
{schema}

Convert this question to a SQL query:
"{user_query}"

Respond with only the SQL query, no explanation.
"""

    sql = llm.generate(prompt)

    # Execute the SQL
    try:
        results = db_connection.execute(sql)
        return format_results(results)
    except Exception as e:
        return f"SQL error: {e}"

Hybrid approach:

Many systems use both. For instance, a fintech app might:

1. Check if the query is a SQL-style question → route to SQL
2. If not, but the query mentions a specific customer/account ID → filter by metadata, then embedding search
3. Otherwise → pure embedding search

def hybrid_retrieval(user_query: str, context: dict) -> str:
    """
    Routing logic for a hybrid system.
    """
    route = route_query(user_query, has_sql_schema=context.get("has_sql_schema", False))

    if route == "sql":
        return text_to_sql_and_execute(user_query, context["schema"], context["db"])

    # Extract any metadata filters from the query
    # (e.g., "Pro plan" -> filter by plan="Pro")
    filters = extract_metadata_filters(user_query)

    if filters:
        # Metadata-aware search
        results = vector_db.search(user_query, filters=filters, top_k=5)
    else:
        # Pure semantic search
        results = vector_db.search(user_query, top_k=5)

    return generate_answer(user_query, results)

Practice: A realistic example

Scenario: You have a code repository with multiple Python files. A user asks: "Show me all functions that validate user input."

Setup:

• You've chunked all Python files by AST, storing function name and docstring as metadata.
• You have a vector database with embeddings for each function.

Retrieval:

# Step 1: Filter by metadata
# Find functions with "validate", "input", or "user" in the docstring
candidate_functions = [
    func for func in all_functions
    if any(term in func["metadata"]["docstring"].lower()
           for term in ["validate", "input", "user"])
]

# Step 2: Vector search within candidates
results = vector_db.search(
    "functions that validate user input",
    candidates=candidate_functions,
    top_k=10
)

# Results might include:
# - validate_email(email)
# - validate_password(pwd)
# - check_username_format(name)
# - sanitize_user_input(data)

Without structured chunking and metadata:

• The query might retrieve fragments of functions, mixing implementation details with interface.
• Redundant results (same function appearing in multiple chunks).
• Context confusion (is this checking the input, or using the input?).

With structured chunking and metadata:

• Each function is atomic, complete, and clearly named in metadata.
• Filtering first narrows the search space, improving relevance.
• The answer is clear: "Here are 4 functions for input validation."

Common mistake

Over-applying text-based thinking to structured data. Just because something can be serialized as text doesn't mean it should be chunked like text. A CSV file chunked row-by-row, a JSON array split mid-way, or code split mid-function all look like valid text chunks, but they destroy the structure that made the data meaningful in the first place.

Before chunking, ask: "What is the atomic unit of meaning in this data?" For tables, it's a row (plus header). For code, it's a function or class. For structured records, it's a single object. Design your chunking around that unit, then chunk multiple units per chunk if needed for size, not the other way around.