SQLglot을 활용한 post‐process

Lang2SQL과 같은 Text2SQL 프로젝트는 자연어 질문을 SQL로 변환하는 데 중요한 역할을 합니다. 하지만 대규모 언어 모델(LLM)이 생성한 SQL은 구문, 의미, 효율성, 또는 특정 데이터베이스 방언(dialect) 호환성 측면에서 문제가 있을 수 있습니다. 따라서 생성된 SQL의 신뢰성과 효율성을 높이기 위해 후처리(post-processing) 단계가 필요합니다.

이 글에서는 SQL 파서(parser)이자 변환기(transpiler)인 SQLglot 라이브러리를 Lang2SQL 프로젝트의 Text2SQL 후처리 과정에 통합하는 방안을 제안하고자 합니다. SQLglot의 다양한 기능을 활용하여 생성된 SQL의 품질을 향상시키고, 다양한 데이터베이스 환경에서의 호환성을 확보하는 구체적인 방법을 논의합니다.

Text2SQL 후처리가 필요한 이유

Text2SQL로 생성된 SQL 쿼리는 사용자의 의도를 정확히 반영하고 대상 데이터베이스에서 올바르게 실행될 수 있도록 체계적인 검증 절차가 필요합니다. LLM이 생성하는 SQL은 다음과 같은 잠재적 문제점을 가질 수 있기 때문입니다.

• 부정확성: 구문 오류가 있거나, 존재하지 않는 테이블/컬럼을 참조하는 등 의미적으로 올바르지 않을 수 있습니다.

• 비호환성: 특정 데이터베이스 시스템의 고유한 SQL 방언이나 함수를 사용하여 다른 시스템에서는 실행되지 않을 수 있습니다.

• (비효율성): 최적화되지 않아 성능 저하를 유발할 수 있습니다.

특히 다양한 데이터베이스 환경(MySQL, PostgreSQL, BigQuery 등)을 지원해야 하는 경우, 각 DB에 최적화된 SQL을 LLM이 직접 생성하도록 유도하는 것은 프롬프트 복잡성을 크게 증가시키고 일관성 있는 품질을 보장하기 어렵게 만듭니다.

따라서 대안은 LLM이 특정 표준 SQL 방언(예: SQLite 또는 ANSI SQL)으로 쿼리를 생성하도록 하고, 후처리 단계에서 이 SQL을 검증한 뒤 목표 데이터베이스의 방언으로 변환(transpile)하는 방식입니다. 이 접근법은 SQL 생성 과정을 단순화하면서도, SQLglot과 같은 도구를 활용하여 구문/의미 검증 및 방언 호환성 문제를 효과적으로 해결할 수 있게 해줍니다.

SQLglot 소개

SQLglot은 파이썬으로 작성된 포괄적인 SQL 파서, 변환기, 생성기 및 최적화기입니다. SQL 쿼리를 **추상 구문 트리(Abstract Syntax Tree, AST)**로 파싱하여 구조적으로 분석하고 조작할 수 있게 해줍니다. 이러한 기능들은 실제 데이터베이스 관리 시스템(DBMS)에 연결할 필요 없이 독립적으로 작동하므로, 적은 리소스로도 SQL 처리 및 검증이 가능하다는 장점이 있습니다.

주요 기능

• 다양한 SQL 방언 지원: MySQL, PostgreSQL, BigQuery, Snowflake, SQLite, DuckDB 등 20개 이상의 주요 SQL 방언을 지원하며, 방언 간 변환(transpiling)이 가능합니다.

• SQL 파싱 및 검증: SQL 쿼리를 AST로 파싱하고, 구문 오류를 감지하며, 오류 위치와 메시지를 제공합니다. 데이터베이스 엔진에 직접 연결하지 않고도 구문 검증이 가능합니다.

• SQL 구조 분석: AST를 탐색하여 쿼리에서 사용된 테이블, 컬럼, 함수 등의 정보를 쉽게 추출할 수 있습니다.

• SQL 최적화: 규칙 기반의 최적화 기능을 제공하여 쿼리 구조를 개선하고 잠재적인 성능 향상을 도모할 수 있습니다. (이 기능은 아직 실험적인 면이 있으며 추가 연구가 필요합니다.)

• SQL 생성: 프로그래밍 방식으로 SQL 쿼리를 생성하거나 기존 쿼리를 수정할 수 있습니다.

Lang2SQL에서의 SQLglot 활용 방안

SQLglot의 강력한 기능들은 Lang2SQL의 Text2SQL 후처리 파이프라인에 다음과 같이 효과적으로 통합될 수 있습니다.

1. 구문 오류 발견 및 상세 오류 메시지 제공

LLM이 생성한 SQL 쿼리를 데이터베이스에서 직접 실행하기 전에 SQLglot을 사용하여 구문 오류를 미리 검증할 수 있습니다. sqlglot.parse_one 함수는 SQL 문자열을 파싱하여 AST를 생성하며, 이때 구문 오류가 발견되면 즉시 예외를 발생시켜 오류의 종류와 위치를 알려줍니다. 이는 실제 쿼리 실행 실패를 방지하고 디버깅을 용이하게 합니다.

import sqlglot
from sqlglot import Dialect

def validate_sql_syntax(sql_query: str, sql_dialect: str = "sqlite") -> tuple[bool, str | None]:
    """
    SQLglot을 사용하여 SQL 쿼리의 구문 오류를 검증합니다.

    Args:
        sql_query: 검증할 SQL 쿼리 문자열.
        sql_dialect: 대상 SQL 방언 (기본값: sqlite).

    Returns:
        (유효성 여부, 오류 메시지) 튜플. 오류가 없으면 (True, None), 있으면 (False, 오류 메시지 문자열).
    """
    try:
        # 지정된 방언으로 SQL 파싱 시도
        # error_level을 IMMEDIATE로 설정하여 첫 오류 발생 시 즉시 중단
        sqlglot.parse_one(
            sql=sql_query,
            read=sql_dialect.lower(), # 방언은 소문자로 지정
            error_level=sqlglot.ErrorLevel.IMMEDIATE,
        )
        print(f"'{sql_dialect}' 방언에 대한 구문 검증 성공.")
        return True, None
    except sqlglot.errors.ParseError as e:
        # 파싱 오류 발생 시 상세 정보 출력 및 반환
        error_message = f"구문 오류 발생:\n{e}"
        print(error_message)
        # 오류 발생 위치 강조 표시 (선택 사항)
        # print(f"오류 컨텍스트:\n{e.highlight(sql=sql_query)}")
        return False, str(e) # 오류 메시지만 반환
    except Exception as e:
        error_message = f"예상치 못한 오류 발생: {e}"
        print(error_message)
        return False, error_message

# 예시 사용 (기존 코드 주석 처리)%% 
initial_generated_sql = "SELECT name FROM users WHERE age > 30 ORDER BY" # LLM이 처음 생성한 SQL이라고 가정
target_dialect = "mysql"

is_valid, error_msg = validate_sql_syntax(initial_generated_sql, target_dialect)

if is_valid:
    print("생성된 SQL 쿼리가 유효합니다.")
else:
    print(f"생성된 SQL 쿼리에 구문 오류가 있습니다: {error_msg}")

generated_sql_correct = "SELECT name FROM users WHERE age > 30 ORDER BY name;"
is_valid, _ = validate_sql_syntax(generated_sql_correct, target_dialect)
if is_valid:
    print("수정된 SQL 쿼리가 유효합니다.")

위 예시 코드는 SQLglot 검증과 LLM 기반의 자동 수정 과정을 결합하여, Text2SQL 시스템의 견고성을 높이는 방법입니다.

2. 스키마 기반 의미적 검증 (테이블 및 컬럼 이름 추출 활용)

SQLglot은 파싱된 AST에서 쿼리에 사용된 모든 테이블과 컬럼 이름을 추출하는 기능을 제공합니다. ast.find_all(sqlglot.exp.Table) 및 ast.find_all(sqlglot.exp.Column)을 사용하면 이 정보들을 쉽게 얻을 수 있습니다.

Lang2SQL에서는 이 기능을 활용하여 다음과 같은 의미적 검증을 수행할 수 있습니다.

• 추출된 테이블 및 컬럼 이름이 실제 데이터베이스 스키마에 존재하는지 확인합니다.
• 사용자가 접근 권한이 있는 테이블/컬럼만 사용하는지 검증합니다. (스키마 정보와 권한 정보 결합 필요)

이는 LLM이 생성한 쿼리가 문법적으로는 맞지만 스키마 상 존재하지 않는 객체를 참조하는 '환각(hallucination)' 문제를 방지하는 데 큰 도움이 됩니다.

from sqlglot import parse_one, exp

def extract_tables_and_columns(sql_query: str, sql_dialect: str = "sqlite"):
    """
    SQL 쿼리에서 사용된 테이블과 컬럼 이름을 추출합니다.

    Args:
        sql_query: 분석할 SQL 쿼리 문자열.
        sql_dialect: 대상 SQL 방언.

    Returns:
        (tables, columns) 튜플. tables는 테이블 이름 집합, columns는 (테이블명, 컬럼명) 튜플 집합.
    """
    tables = set()
    columns = set()
    try:
        ast = parse_one(sql=sql_query, read=sql_dialect.lower())

        # 테이블 이름 추출
        for table_node in ast.find_all(exp.Table):
            tables.add(table_node.name)
            # print(f"Found Table: {table_node.name}")

        # 컬럼 이름 추출 (테이블 정보 포함)
        for column_node in ast.find_all(exp.Column):
            table_name = column_node.table if column_node.table else None # 컬럼에 명시된 테이블 이름 (없을 수 있음)
            column_name = column_node.name
            columns.add((table_name, column_name))
            # print(f"Found Column: {column_name} (Table: {table_name})")

        return tables, columns

    except sqlglot.errors.ParseError as e:
        print(f"구문 오류로 테이블/컬럼 추출 실패: {e}")
        return None, None
    except Exception as e:
        print(f"예상치 못한 오류 발생: {e}")
        return None, None

# 예시 사용
sql = "SELECT c.name, o.order_date FROM customers c JOIN orders o ON c.customer_id = o.cust_id WHERE c.city = 'Seoul'"
target_dialect = "postgres"

extracted_tables, extracted_columns = extract_tables_and_columns(sql, target_dialect)
# 이후 Data 카탈로그에서 Table, Column 을 매칭시켜보고 있는 없는 것이라면 맞는 정보들을 넣을 수 있도록

3. 생성된 SQL 최적화 (추가 연구 필요)

SQLglot은 AST를 조작하여 SQL 쿼리를 최적화하는 기능을 제공합니다. sqlglot.optimizer.optimize 함수를 사용하면 데이터베이스 스키마 정보를 바탕으로 다양한 최적화 규칙(예: 술어 푸시다운, 불필요한 조인 제거 등)을 적용할 수 있습니다.

구글의 ADK(Agent Development Kit) 샘플에서도 LLM이 생성한 SQL(초기에는 SQLite 방언)을 파싱하고 오류를 검증한 후, optimize 함수를 사용하여 BigQuery 방언으로 변환하기 전에 최적화를 수행하는 사례를 볼 수 있습니다. (참고 링크)

# 최적화 예시 (개념 설명, 실제 적용 시 스키마 필요)
# schema_dict = { ... } # 실제 데이터베이스 스키마 정보
db = "my_database"
catalog = "my_catalog"
generated_sql = "..." # 검증 완료된 SQL
target_dialect = "mysql"

try:
    sql_query_ast = sqlglot.parse_one(sql=generated_sql, read=target_dialect.lower()) # 또는 표준 방언으로 파싱
    optimized_ast = sqlglot.optimizer.optimize(
        sql_query_ast,
        dialect=target_dialect.lower(), # 최적화는 대상 DB 방언 기준
        schema=schema_dict,
        # db=db, # 데이터베이스 이름 지정 (필요시)
        # catalog=catalog, # 카탈로그 이름 지정 (필요시)
        error_level=sqlglot.ErrorLevel.IMMEDIATE,
    )
    optimized_sql = optimized_ast.sql(dialect=target_dialect.lower(), pretty=True)
    print("\n최적화된 SQL:\n", optimized_sql)
except Exception as e:
    print(f"\nSQL 최적화 중 오류 발생: {e}")

하지만 SQL 최적화는 복잡한 작업이며, SQLglot의 최적화 기능은 어떤 관점으로 최적화하거나 어떻게 최적화 되는지 파악되지 않아 추가적인 조사가 필요합니다. Lang2SQL에 이 기능을 도입하기 전에 충분한 테스트와 검증, 그리고 다양한 데이터베이스 및 쿼리 패턴에 대한 성능 비교 연구가 필요해 보입니다. 잘못된 최적화는 오히려 성능을 저하시키거나 잘못된 결과를 초래할 수 있기 때문입니다.

4. 다양한 데이터베이스 방언 지원 (SQL 변환)

sqlglot.transpile 함수는 특정 방언으로 작성된 SQL 쿼리를 다른 방언으로 자동 변환해 줍니다.

Lang2SQL은 내부적으로 LLM에게 특정 표준 방언(예: SQLite 또는 ANSI SQL)으로 SQL을 생성하도록 지시하고, 후처리 단계에서 사용자가 실제로 사용하는 데이터베이스(MySQL, PostgreSQL, BigQuery 등)의 방언으로 변환할 수 있습니다. 이는 LLM 프롬프트를 단순화하고, 다양한 데이터베이스 환경에 대한 이식성을 크게 향상시키는 효과를 가져옵니다.

import sqlglot

def transpile_sql(sql_query: str, source_dialect: str, target_dialect: str) -> str | None:
    """
    SQLglot을 사용하여 SQL 쿼리를 한 방언에서 다른 방언으로 변환합니다.

    Args:
        sql_query: 변환할 원본 SQL 쿼리 문자열.
        source_dialect: 원본 SQL 방언.
        target_dialect: 대상 SQL 방언.

    Returns:
        변환된 SQL 쿼리 문자열 또는 오류 발생 시 None.
    """
    try:
        # transpile 함수는 변환된 SQL 문자열 리스트를 반환
        transpiled_sql_list = sqlglot.transpile(
            sql=sql_query,
            read=source_dialect.lower(),
            write=target_dialect.lower(),
            pretty=True # 가독성을 위해 SQL 포맷팅 적용
        )
        # 일반적으로 하나의 쿼리를 변환하므로 첫 번째 결과 사용
        if transpiled_sql_list:
            return transpiled_sql_list[0]
        else:
            print("변환 결과가 없습니다.")
            return None
    except sqlglot.errors.ParseError as e:
        print(f"원본 SQL 파싱 오류: {e}")
        return None
    except sqlglot.errors.UnsupportedError as e:
        # 변환 중 지원되지 않는 기능이나 함수가 있을 경우 발생
        print(f"변환 오류 (지원되지 않는 기능): {e}")
        return None
    except Exception as e:
        print(f"예상치 못한 변환 오류 발생: {e}")
        return None

# 예시 사용: DuckDB의 EPOCH_MS 함수를 Hive의 FROM_UNIXTIME 함수로 변환
original_sql_duckdb = "SELECT EPOCH_MS(1618088028295) AS event_time;"
source = "duckdb"
target = "hive"

transpiled_sql_hive = transpile_sql(original_sql_duckdb, source, target)

if transpiled_sql_hive:
    print(f"\n원본 SQL ({source}):\n{original_sql_duckdb}\n")
    print(f"변환된 SQL ({target}):\n{transpiled_sql_hive}")

# 예시 사용: SQLite의 DATE 함수를 PostgreSQL의 DATE 함수로 변환 (동일하지만 방언 명시)
original_sql_sqlite = "SELECT DATE('now') AS today;"
source = "sqlite"
target = "postgres"

transpiled_sql_postgres = transpile_sql(original_sql_sqlite, source, target)

if transpiled_sql_postgres:
    print(f"\n원본 SQL ({source}):\n{original_sql_sqlite}\n")
    print(f"변환된 SQL ({target}):\n{transpiled_sql_postgres}")

결론

SQLglot은 Lang2SQL과 같은 Text2SQL 프로젝트에서 LLM이 생성한 SQL 쿼리의 품질과 신뢰성을 향상시키는 데 유용한 도구입니다. SQLglot을 후처리 파이프라인에 통합함으로써 다음과 같은 이점을 얻을 수 있습니다.

• 신뢰성 향상: 구문 오류 검증 및 스키마 기반 의미 검증을 통해 잘못된 쿼리 실행을 사전에 방지합니다. LLM을 활용한 자동 오류 수정 메커니즘을 추가하여 견고성을 더욱 높일 수 있습니다.

• 유연성 및 이식성 증대: SQL 변환 기능을 통해 특정 데이터베이스 방언에 대한 종속성을 줄이고 다양한 백엔드를 지원할 수 있습니다.

• 잠재적 성능 개선: 최적화 기능을 통해 (주의 깊은 검증 하에) 쿼리 실행 효율을 높일 수 있습니다.

물론 SQLglot을 효과적으로 활용하기 위해서는 데이터베이스 스키마 정보와의 연동, 최적화 기능에 대한 심층적인 연구, LLM 연동 자동 수정 로직의 구체화 및 테스트, 그리고 다양한 예외 상황 처리 등 추가적인 노력이 필요합니다.