SQL Logic Duplication

Anti-pattern in database-driven applications where SQL query generation logic is duplicated across multiple code paths instead of being centralized, creating maintenance complexity and consistency risks. Particularly problematic when combined with dead code that appears functional but is never executed.

The Problem

Dead Code with Live Duplication

Common pattern where a well-designed method exists but is bypassed by duplicated logic:

class PostgresRetriever:
    def _build_select_sql(self, table, columns, filters):
        """Well-designed, parameterized SQL builder - NEVER CALLED"""
        where_clauses = []
        params = []
        
        for field, value in filters.items():
            where_clauses.append(f"{field} = %s")
            params.append(value)
        
        sql = f"SELECT {', '.join(columns)} FROM {table}"
        if where_clauses:
            sql += f" WHERE {' AND '.join(where_clauses)}"
        
        return sql, params
    
    def search(self, query, filters):
        # _build_select_sql is never called!
        # Instead, inline duplication:
        
        if self.table == "service_public":
            if filters.get("ministry"):
                sql = "SELECT content, metadata FROM service_public WHERE ministry = %s"
                params = [filters["ministry"]]
            else:
                sql = "SELECT content, metadata FROM service_public"
                params = []
        elif self.table == "dgafp":
            if filters.get("ministry"):
                sql = "SELECT content, metadata FROM dgafp WHERE ministry = %s" 
                params = [filters["ministry"]]
            else:
                sql = "SELECT content, metadata FROM dgafp"
                params = []
        # ... more duplication

Problems Created

Maintenance Nightmare

• Multiple code paths requiring synchronization
• Changes must be applied to every branch
• High risk of inconsistent implementations
• Difficult to track all places requiring updates

Bug Multiplication

• Bugs in one branch don't automatically fix others
• Easy to miss edge cases in some branches
• Testing complexity grows exponentially
• Silent failures in unused branches

Cognitive Load

• Developers must understand multiple implementations
• Code reviews become more complex
• New team members face steep learning curve
• Architecture becomes opaque

Feature Addition Complexity

# Adding new column requires updating every branch
if self.table == "service_public":
    if filters.get("ministry"):
        # Must remember to add new_column here
        sql = "SELECT content, metadata FROM service_public WHERE ministry = %s"
    else:
        # And here
        sql = "SELECT content, metadata FROM service_public"
elif self.table == "dgafp":
    # And here  
    if filters.get("ministry"):
        sql = "SELECT content, metadata FROM dgafp WHERE ministry = %s"
    # And here
    else:
        sql = "SELECT content, metadata FROM dgafp"
# Easy to miss one branch!

Root Causes

Incremental Development

• Original method designed correctly
• Later additions bypass existing abstractions
• Copy-paste programming for "quick fixes"
• Technical debt accumulation over time

Poor Code Review

• Reviews focus on individual changes, miss patterns
• Abstraction violations not flagged
• No architectural guidance enforcement
• Missing integration perspective

Testing Gaps

• Unit tests per branch, no integration testing
• Dead code not detected by coverage tools
• Missing architectural constraint testing
• No refactoring validation

Solutions

Centralized SQL Generation

class PostgresRetriever:
    def _build_select_sql(self, columns=None, filters=None):
        columns = columns or ["content", "metadata", "embedding"]
        
        sql = f"SELECT {', '.join(columns)} FROM {self.table}"
        params = []
        
        if filters:
            where_clauses = []
            for field, value in filters.items():
                if value is not None:
                    where_clauses.append(f"{field} = %s")
                    params.append(value)
            
            if where_clauses:
                sql += f" WHERE {' AND '.join(where_clauses)}"
        
        return sql, params
    
    def search(self, query, filters=None):
        # Always use the centralized builder
        sql, params = self._build_select_sql(
            columns=["content", "metadata", "embedding"],
            filters=filters
        )
        # Execute with consistent logic

Query Builder Pattern

class QueryBuilder:
    def __init__(self, table):
        self.table = table
        self.columns = ["*"]
        self.conditions = []
        self.params = []
    
    def select(self, columns):
        self.columns = columns
        return self
    
    def where(self, field, value):
        if value is not None:
            self.conditions.append(f"{field} = %s")
            self.params.append(value)
        return self
    
    def build(self):
        sql = f"SELECT {', '.join(self.columns)} FROM {self.table}"
        if self.conditions:
            sql += f" WHERE {' AND '.join(self.conditions)}"
        return sql, self.params

# Usage
def search(self, query, filters=None):
    builder = QueryBuilder(self.table).select(["content", "metadata"])
    
    if filters:
        for field, value in filters.items():
            builder.where(field, value)
    
    sql, params = builder.build()

Testing Dead Code Detection

def test_all_code_paths_used():
    """Ensure no dead SQL generation methods"""
    import coverage
    
    cov = coverage.Coverage()
    cov.start()
    
    # Exercise all retriever operations
    retriever = PostgresRetriever("service_public")
    retriever.search("query", {"ministry": "test"})
    retriever.search("query", {})
    
    cov.stop()
    
    # Verify _build_select_sql was called
    analysis = cov.analysis("src/rag/retriever.py")
    missing_lines = analysis[2]
    
    # Fail if _build_select_sql lines are in missing
    build_sql_lines = get_method_lines("_build_select_sql")
    assert not set(build_sql_lines).intersection(missing_lines)

Case Study: Assistant-RH

The assistant-rh system demonstrated this anti-pattern with:

• Unused _build_select_sql method with proper parameterization
• Duplicated SQL logic across multiple conditional branches
• Different table handling requiring separate maintenance
• High risk of inconsistency when adding features or columns

This created a maintenance nightmare and was classified as contributing to Category 5 system reliability issues.

Prevention Strategies

1. Centralized SQL Generation: Single source of truth for query building
2. Code Coverage Analysis: Detect and eliminate dead code paths
3. Architectural Reviews: Flag abstraction violations in code review
4. Refactoring Discipline: Regular cleanup of duplicated logic
5. Builder Patterns: Use query builders for complex SQL construction
6. Integration Testing: Test that abstractions are actually used

See also

• assistant-rh
• Code Duplication
• Technical Debt
• Database Abstraction
• Query Builder Pattern
• category-5-bugs