SQL Validation for internal consistency

[macieyng]

Is it possible to say that SQL is not only syntacticly correct, but also makes sense semantically?

I know that without knowledge about the database schema it's difficult to say that specific SQL can be executed and will return results, but I believe with AST we can say that SQL is internally consistent/coherent.

Here are some examples

1. SELECT * FROM table - assume * is always correct
2. SELECT col FROM table - assume col is in table so it's correct
3. SELECT * from table as 't' - is correct, assume 1
4. SELECT 't'.col FROM table as 't' - is correct, assume 2
5. SELECT col FROM (SELECT * FROM table) - is correct, assume 1 and 2
6. SELECT 't'.col FROM (SELECT col FROM table) as 't' - is correct, assume 4
7. SELECT aliased_col FROM (SELECT col as 'aliased_col' FROM table) - assume is correct, assume 2
8. SELECT no_col FROM (SELECT col FROM table) - is incorrect, because subquery has no column no_col in it's result, set of columns from outter select is not a subset of columns in subquery/inner select.

And point 8 is what would be a great first step. I'm think that we could apply formal logic (https://pypi.org/project/mathesis/) to the AST and somehow figure out if the whole query make sense.

Anyway, I wanted to spark a discussion about validating SQL queries before executing them against db, since I haven't found a tool that could do such thing.

And here is a bit of something I wrote that partially do what I wanted as an effect, but not when it comes to the usage of formal logic.

import logging
import sqlglot
from sqlglot import exp
import pytest


# Configure the logger
logging.basicConfig(level=logging.INFO)  # Change to DEBUG for more details
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class SQLValidationError(Exception):
    """Custom exception to raise SQL validation errors with multiple messages."""
    def __init__(self, errors):
        self.errors = errors
        error_message = "\n".join(errors)
        super().__init__(f"SQL validation failed with the following errors:\n{error_message}")


def validate_ast(ast: exp.Expression):
    """
    Validate that columns in outer SELECTs exist in inner subqueries, 
    including named subqueries (aliases).
    
    :param ast: Parsed AST of SQL query (list of Select expressions)
    :return: True if valid, raises SQLValidationError if there are validation issues.
    """
    def validate_select(select_expr: exp.Select, errors=None):
        """Recursively validate a Select expression."""
        errors = errors or []
        columns = set()
        subquery_cols = set()
        
        from_ = select_expr.args.get("from").this
        if isinstance(from_.this, exp.Select):
            logger.debug(f"Validating subquery: {from_}")
            sub_cols_, errs_ = validate_select(from_.this, errors)
            subquery_cols.update(sub_cols_)
        elif isinstance(from_, exp.Table):
            logger.debug(f"Validating table: {from_}")
        else:
            logger.debug(f"Validating from: {from_}, type: {type(from_)}")
            
        joins_ = select_expr.args.get("joins")
        if joins_:
            for join in joins_:
                if isinstance(join.this, exp.Select):
                    logger.debug(f"Validating subquery join: {join}")
                    sub_cols_, errs_ = validate_select(join.this, errors)
                    subquery_cols.update(sub_cols_)
                elif isinstance(join.this, exp.Table):
                    logger.debug(f"Validating table join: {join}")
                else:
                    logger.debug(f"Validating join: {join}, type: {type(join)}")
                join_type = "on" if "on" in join.args else "using"
                if join_type == "using":
                    logger.debug(f"    Validating join 'using': {join}")
                    continue
                elif join_type == "on" and isinstance(join.this, exp.Subquery):
                    logger.debug(f"    Validating join 'on' subquery: {join}")
                    sub_cols_, errs_ = validate_select(join.this.this, errors)
                    subquery_cols.update(sub_cols_)
                else:
                    logger.debug(f"    Validating join 'on': {join}")
                    right_side: exp.Dot = join.args["on"].left
                    left_side: exp.Dot = join.args["on"].right
                    
                    available_tables = [join.this.this.this, from_.this.this]
                    print(available_tables)
                    logger.debug(f"Right side type: {type(right_side)}")
                    if isinstance(right_side, exp.Dot):
                        if right_side.this.this not in available_tables + [join.this.alias, from_.alias]:
                            logger.debug(f"    Validation error: {right_side.this} not found in available tables({','.join(available_tables)}).")
                            errors.append(f"Column '{right_side.this}' not found in available tables({''.join(available_tables)}).")
                    elif isinstance(right_side, exp.Column):
                        if right_side.table not in available_tables:
                            logger.debug(f"    Validation error: {right_side.table} not found in available tables({','.join(available_tables)}).")
                            errors.append(f"Column '{right_side.table}' not found in available tables({''.join(available_tables)}).")
                        
                    logger.debug(f"Left side type: {type(left_side)}")
                    if isinstance(left_side, exp.Dot):
                        if left_side.this.this not in available_tables + [join.this.alias, from_.alias]:
                            logger.debug(f"    Validation error: {left_side.this} not found in available tables({','.join(available_tables)}).")
                            errors.append(f"Column '{left_side.this}' not found in available tables({''.join(available_tables)}).")
                    elif isinstance(left_side, exp.Column):
                        if left_side.table not in available_tables:
                            logger.debug(f"    Validation error: {left_side.table} not found in available tables({','.join(available_tables)}).")
                            errors.append(f"Column '{left_side.table}' not found in available tables({''.join(available_tables)}).")
        
        for expression in select_expr.expressions:
            logger.debug(f"    Validating expression: {expression}")
            if isinstance(expression, exp.Star):
                if len(select_expr.expressions) != 1:
                    errors.append("Cannot use '*' with other columns in the same SELECT statement.")
                else:
                    columns.add(expression)
            elif isinstance(expression, exp.Alias):
                columns.add(expression.this.alias_or_name)
            else:
                columns.add(expression.this.this)
        if len(subquery_cols) > 0:
            try:
                assert columns.issubset(subquery_cols), f"In query '{select_expr}' there are columns '{','.join(list(columns))}' that cannot be found in subquery '{from_.this}'. Verify if columns: {','.join(list(columns - subquery_cols))} are correct and can be found in the subquery."
            except AssertionError as e:
                logger.debug(f"    Validation error: {e}")
                errors.append(str(e))
        return columns, errors
    
    errors = []
    
    for select_expr in ast:
        logger.debug(f"Validating select expression: {select_expr}")
        _, errors_ = validate_select(select_expr, errors)
        errors.extend(errors_)
    if errors:
        raise SQLValidationError(errors)
    return ast
    



class TestValidateAST:
    def test_validate_ast_valid_query(self):
        assert validate_ast(sqlglot.parse("SELECT exist FROM table"))

    def test_validate_ast_valid_subquery(self):
        assert validate_ast(sqlglot.parse("SELECT exist FROM (SELECT exist FROM my_table)"))

    def test_validate_ast_star_with_alias(self):
        assert validate_ast(sqlglot.parse("SELECT * from TABLE as 't'"))

    def test_validate_ast_star_without_alias(self):
        assert validate_ast(sqlglot.parse("SELECT * from TABLE"))

    def test_validate_ast_column_with_alias(self):
        assert validate_ast(sqlglot.parse("SELECT 't'.COL as 'col' from TABLE as 't'"))
        
    def test_validate_ast_join(self):
        assert validate_ast(sqlglot.parse("SELECT TABLE.COL from TABLE join TABLE2 on TABLE.COL = TABLE2.COL"))
        
    def test_validate_ast_join_with_alias_left(self):
        assert validate_ast(sqlglot.parse("SELECT 't'.COL as 'col' from TABLE as 't' join TABLE2 on 't'.COL = TABLE2.COL"))
        
    def test_validate_ast_join_with_alias_right(self):
        assert validate_ast(sqlglot.parse("SELECT TABLE.COL from TABLE join TABLE2 as 't2' on TABLE.COL = 't2'.COL"))
        
    def test_validate_ast_join_with_alias_both(self):
        assert validate_ast(sqlglot.parse("SELECT 't'.COL as 'col' from TABLE as 't' join TABLE2 as 't2' on 't'.COL = 't2'.COL"))
        
    def test_validate_ast_invalid_join_with_alias(self):
        with pytest.raises(SQLValidationError):
            validate_ast(sqlglot.parse("SELECT 't'.COL as 'col' from TABLE as 't' join TABLE2 as 't2' on 't'.COL = 't3'.COL"))

    def test_validate_ast_invalid_column_in_subquery(self):
        with pytest.raises(SQLValidationError):
            validate_ast(sqlglot.parse("SELECT not_exist FROM (SELECT exist FROM my_table)"))

    def test_validate_ast_invalid_column_with_alias_in_subquery(self):
        with pytest.raises(SQLValidationError):
            validate_ast(sqlglot.parse("SELECT q.not_exist FROM (SELECT exist FROM my_table) as 'q'"))

    def test_validate_ast_multiple_invalid_columns_in_nested_subquery(self):
        with pytest.raises(SQLValidationError):
            validate_ast(sqlglot.parse(
                "select q2.not_exist1, q2.not_exist2 from (SELECT q1.not_exist1 as 'not_exist2' FROM (SELECT exist FROM my_table) as 'q1') as 'q2'"
            ))


if __name__ == '__main__':
    pytest.main(args=[__file__, '--disable-warnings'])

[georgesittas]

Hiya,

I replied in the corresponding Slack thread as well, just mirroring the answer here. Implementing analyses like the one you described where you infer & propagate constraints incrementally should be possible, SQLGlot is very powerful for these use cases. However, this is not something we plan to work on (i.e. the core team).

Also note that syntax validation is not in scope, so there are no strong guarantees there. SQLGlot mostly detects simple errors like unbalanced parentheses, etc.