SQL Duplication

Anti-pattern where SQL query generation logic is duplicated across multiple code paths, creating maintenance nightmares and increasing the risk of inconsistencies. Particularly problematic in RAG systems where query complexity can be high and variations numerous.

Problem Pattern

The assistant-rh project demonstrated this anti-pattern with an abandoned _build_select_sql function while SQL generation was duplicated inline:

def _build_select_sql(self, table, filters):
    # UNUSED FUNCTION - logic duplicated elsewhere
    pass

def search(self, query, top_k=10, filters=None):
    # SQL logic duplicated here instead of using _build_select_sql
    if table == "service_public":
        sql = "SELECT chunk, embedding FROM service_public WHERE ..."
    elif table == "dgafp":
        sql = "SELECT chunk, embedding FROM dgafp WHERE ..."
    # Multiple conditional branches with similar but divergent SQL

Technical Debt Impact

• Maintenance Nightmare: Changes require updates in multiple locations
• Inconsistency Risk: Easy to forget updating one branch when modifying another
• Testing Complexity: Each duplicated path needs separate test coverage
• Feature Divergence: Branches gradually diverge as modifications accumulate

Common Causes

1. Rushed Development: Quick fixes that bypass proper abstraction
2. Feature Creep: Adding special cases without refactoring the base pattern
3. Legacy Migration: Incremental changes that leave old code paths active
4. Conditional Complexity: Complex branching logic that resists simple factoring

Refactoring Strategies

Query Builder Pattern

class QueryBuilder:
    def __init__(self, table):
        self.table = table
        self.columns = []
        self.conditions = []
    
    def select(self, *columns):
        self.columns.extend(columns)
        return self
    
    def where(self, condition):
        self.conditions.append(condition)
        return self
    
    def build(self):
        return f"SELECT {','.join(self.columns)} FROM {self.table} WHERE {' AND '.join(self.conditions)}"

Template-Based Generation

SQL_TEMPLATES = {
    "service_public": "SELECT chunk, embedding FROM service_public WHERE {filters}",
    "dgafp": "SELECT chunk, embedding FROM dgafp WHERE {filters}",
}

def build_query(table, filters):
    template = SQL_TEMPLATES[table]
    return template.format(filters=build_filter_clause(filters))

Parameterized Functions

def build_select_sql(table, columns=None, filters=None):
    columns = columns or ["chunk", "embedding"]
    base_sql = f"SELECT {','.join(columns)} FROM {table}"
    if filters:
        base_sql += f" WHERE {build_filter_clause(filters)}"
    return base_sql

Prevention Strategies

1. Code Review Focus: Flag SQL duplication in reviews
2. Abstraction First: Build query abstractions before adding variations
3. Testing Requirements: Require tests that exercise all query paths
4. Refactoring Discipline: Regular cleanup of duplicated query logic

See also

• query-building
• technical-debt
• code-maintenance
• postgres-retriever