Python에서 간단한 SQL 파서 구현하기: 원리부터 실습까지

데이터 처리 분야에서 SQL(Structured Query Language)은 데이터베이스와 상호 작용하기 위한 핵심 도구입니다. 하지만 SELECT id, name FROM users WHERE age > 18과 같은 구문을 작성할 때 프로그램이 어떻게 이해하는지 궁금한 적이 있나요? 이 뒤에는 SQL 파서의 강력한 기능이 있습니다. SQL 파서는 사람이 읽을 수 있는 SQL 텍스트를 프로그램이 실행할 수 있는 구조화된 데이터(예: 추상 구문 트리)로 변환합니다.

SQL 파서는 ORM 프레임워크(예: Django ORM), SQL 감사 도구(예: Sqitch) 및 데이터베이스 미들웨어(예: Sharding-JDBC)에서 널리 사용됩니다. 이 기사에서는 파싱 원리부터 시작하여 핵심 SQL 구문(SELECT ... FROM ... WHERE)을 지원하는 간단한 SQL 파서를 Python으로 구현하여 파서의 핵심 논리를 이해하도록 안내합니다.

I. SQL 파싱의 핵심: 두 가지 주요 단계

SQL을 포함한 모든 언어를 파싱하는 데에는 "어휘 분석"과 "구문 분석"이라는 두 가지 필수 단계가 있습니다. 이 과정을 "영어 문장 읽기"에 비유할 수 있습니다. 먼저 개별 단어를 식별하고(어휘 분석), 문장의 문법 구조를 이해합니다(구문 분석).

1. 어휘 분석: "단어"로 분해

어휘 분석의 목표는 연속적인 SQL 텍스트를 명확한 의미를 가진 개별 토큰(어휘소)으로 분할하는 것입니다. 토큰은 SQL의 "단어"와 같습니다. 예를 들어:

• SQL 구문: SELECT id, name FROM users WHERE age > 18
• 분할 후 토큰 시퀀스: [SELECT, ID(id), COMMA, ID(name), FROM, ID(users), WHERE, ID(age), GT(>), INT(18)]

일반적인 SQL 토큰 유형은 다음과 같습니다.

• 키워드: SELECT, FROM, WHERE (대소문자 구분 없음)
• 식별자: 테이블 이름(예: users), 열 이름(예: id)
• 리터럴: 숫자(18), 문자열('Alice')`
• 연산자: = (같음), > (보다 큼), < (보다 작음)
• 구두점: , (쉼표), * (별표, 모든 열을 나타냄)

2. 구문 분석: "문장 구조" 구성

구문 분석은 SQL 문법 규칙(예: "SELECT 다음에는 열 이름 또는 *가 와야 하고, FROM 다음에는 테이블 이름이 와야 함")을 사용하여 토큰 시퀀스를 **추상 구문 트리(AST)**로 변환합니다. AST는 SQL 구문의 논리적 계층 구조를 명확하게 나타내는 트리 구조입니다. 예를 들어 위의 SQL 구문의 AST는 다음과 같이 단순화할 수 있습니다.

Query
├─ select_clause: [id, name]
├─ from_clause: users
└─ where_clause:
   ├─ column: age
   ├─ operator: >
   └─ value: 18

AST의 가치는 구조화되지 않은 텍스트를 구조화된 데이터로 변환하는 능력에 있습니다. AST를 탐색하여 프로그램은 "어떤 열을 쿼리할지", "어떤 테이블에서 쿼리할지", "어떤 필터 조건을 적용할지"와 같은 핵심 정보를 쉽게 검색할 수 있습니다.

II. 실습: Python에서 간단한 SQL 파서 구현하기

Python의 ply 라이브러리(Python Lex-Yacc)를 사용하여 파서를 구현할 것입니다. ply는 기존의 컴파일러 구축 도구인 lex(어휘 분석용)와 yacc(구문 분석용)를 시뮬레이션하는 라이브러리입니다. 시작하기 쉽고 파서의 핵심 논리와 밀접하게 일치합니다.

1. 환경 준비

먼저 ply 라이브러리를 설치합니다.

pip install ply

2. 1단계: 어휘 분석기(Lexer) 구현

어휘 분석기의 핵심은 정규 표현식을 사용하여 다양한 유형의 토큰을 일치시키고 공백 및 주석과 같은 관련 없는 문자를 무시하는 것입니다.

코드 구현 (Lexer)

import ply.lex as lex

# 1. 토큰 유형 정의(오류 방지를 위해 먼저 정의해야 함)
tokens = (
    'SELECT', 'FROM', 'WHERE',  # 키워드
    'ID', 'INT', 'STRING',      # 식별자 및 리터럴
    'EQ', 'GT', 'LT',           # 연산자(같음, 보다 큼, 보다 작음)
    'COMMA', 'STAR'             # 구두점(쉼표, 별표)
)

# 2. 키워드에 대한 일치 규칙 정의(키워드도 문자로 구성되므로 식별자보다 우선순위가 높음)
reserved = {
    'select': 'SELECT',
    'from': 'FROM',
    'where': 'WHERE'
}

# 3. 토큰에 대한 정규 표현식 정의(우선순위가 가장 높은 것부터 가장 낮은 것 순서로)
# 문자열 리터럴: 작은따옴표로 묶음, 예: 'Alice'
def t_STRING(t):
    r"'[^']*'"  # Regex: 작은따옴표 안의 모든 문자(작은따옴표 제외)와 일치
    t.value = t.value[1:-1]  # 실제 내용을 유지하기 위해 작은따옴표 제거
    return t

# 정수 리터럴: 숫자 시퀀스
def t_INT(t):
    r'\d+'
    t.value = int(t.value)  # 정수 유형으로 변환
    return t

# 식별자(테이블 이름, 열 이름): 문자로 시작하고 문자/숫자/밑줄이 뒤따름
def t_ID(t):
    r'[a-zA-Z_][a-zA-Z0-9_]*'
    # 식별자가 키워드인지 확인(예: 'select'는 ID가 아닌 SELECT로 인식되어야 함)
    t.type = reserved.get(t.value.lower(), 'ID')
    return t

# 연산자
t_EQ = r'='    # 같음
t_GT = r'>'    # 보다 큼
t_LT = r'<'    # 보다 작음

# 구두점
t_COMMA = r',' # 쉼표
t_STAR = r'\*' # 별표(regex에서 *는 특별한 의미를 가지므로 이스케이프해야 함)

# 4. 관련 없는 문자 무시(공백, 탭, 줄 바꿈)
t_ignore = ' \t\n'

# 5. 오류 처리(인식할 수 없는 문자가 발생할 때 트리거됨)
def t_error(t):
    print(f"Illegal character: '{t.value[0]}'"
    t.lexer.skip(1)  # 잘못된 문자 건너뛰고 다음 내용 파싱 계속

# 6. Lexer 인스턴스 생성
lexer = lex.lex()

# Lexer 테스트: SQL 텍스트 입력하고 토큰 시퀀스 출력
def test_lexer(sql):
    lexer.input(sql)
    print("Lexical Analysis Result (Token Sequence):")
    while True:
        tok = lexer.token()
        if not tok:
            break
        print(f"Type: {tok.type:10}, Value: {tok.value}")

# 테스트 케이스
test_sql = "SELECT id, name FROM users WHERE age > 18 AND name = 'Alice'"
test_lexer(test_sql)

실행 결과

Lexical Analysis Result (Token Sequence):
Type: SELECT    , Value: select
Type: ID        , Value: id
Type: COMMA     , Value: ,
Type: ID        , Value: name
Type: FROM      , Value: from
Type: ID        , Value: users
Type: WHERE     , Value: where
Type: ID        , Value: age
Type: GT        , Value: >
Type: INT       , Value: 18
Type: ID        , Value: AND  # 참고: AND 키워드가 아직 정의되지 않았으므로 임시로 ID로 인식됨
Type: ID        , Value: name
Type: EQ        , Value: =
Type: STRING    , Value: Alice

3. 2단계: 구문 분석기(Parser) 구현

구문 분석기의 핵심은 SQL 문법 규칙을 정의하고 토큰 시퀀스를 AST로 변환하는 것입니다. 가장 핵심적인 쿼리 구문을 지원할 것입니다.

SELECT [column_list/*] FROM table_name [WHERE condition (column operator value)]

코드 구현 (Parser)

import ply.yacc as yacc
from lexer import tokens  # 1단계에서 정의한 토큰 유형 가져오기

# 1. AST 노드 정의(간단하고 명확하게 하기 위해 딕셔너리로 표현)
def create_ast(node_type, **kwargs):
    return {'type': node_type, **kwargs}

# 2. 문법 규칙 정의(우선순위가 가장 낮은 것부터 가장 높은 것 순서로, 시작 규칙은 'query'임)
# 시작 규칙: 쿼리 구문 = SELECT 절 + FROM 절 + [WHERE 절]
def p_query(p):
    '''query : select_clause from_clause where_clause_opt'''
    # p[0]은 규칙의 반환 값이고, p[1]은 select_clause, p[2]는 from_clause, p[3]은 where_clause_opt임
    p[0] = create_ast(
        'Query',
        select=p[1],
        from_clause=p[2],
        where_clause=p[3] if p[3] else None  # 선택적 절, 존재하지 않으면 None으로 설정
    )

# 선택적 WHERE 절: 있거나 없음
def p_where_clause_opt(p):
    '''where_clause_opt : WHERE condition
                        | empty'''
    if len(p) == 3:  # "WHERE condition"과 일치
        p[0] = p[2]
    else:  # "empty"와 일치(WHERE 절 없음)
        p[0] = None

# SELECT 절: SELECT + (별표 / 열 목록)
def p_select_clause(p):
    '''select_clause : SELECT STAR
                     | SELECT column_list'''
    if p[2] == '*':  # "SELECT *"와 일치
        p[0] = create_ast('SelectClause', columns=['*'])
    else:  # "SELECT column_list"와 일치
        p[0] = create_ast('SelectClause', columns=p[2])

# 열 목록: 쉼표로 구분된 여러 ID(예: id, name, age)
def p_column_list(p):
    '''column_list : ID
                   | column_list COMMA ID'''
    if len(p) == 2:  # 단일 열(예: id)
        p[0] = [p[1]]
    else:  # 여러 열(예: column_list, ID)
        p[0] = p[1] + [p[3]]

# FROM 절: FROM + 테이블 이름(예: FROM users)
def p_from_clause(p):
    '''from_clause : FROM ID'''
    p[0] = create_ast('FromClause', table=p[2])

# 조건: 열 + 연산자 + 값(예: age > 18 또는 name = 'Alice')
def p_condition(p):
    '''condition : ID EQ INT
                 | ID EQ STRING
                 | ID GT INT
                 | ID LT INT'''
    p[0] = create_ast(
        'Condition',
        column=p[1],
        operator=p[2],
        value=p[3]
    )

# 빈 규칙(선택적 절에 사용됨)
def p_empty(p):
    '''empty :'''
    p[0] = None

# 구문 오류 처리
def p_error(p):
    if p:
        print(f"Syntax error: Near Token {p.type} (Value: {p.value})")
    else:
        print("Syntax error: Unexpected end of input")

# Parser 인스턴스 생성
parser = yacc.yacc()

# Parser 테스트: SQL 텍스트 입력하고 AST 출력
def parse_sql(sql):
    ast = parser.parse(sql)
    print("\nSyntactic Analysis Result (AST):")
    import json  # 더 나은 가독성을 위해 JSON을 사용하여 출력 형식 지정
    print(json.dumps(ast, indent=2))

# 테스트 케이스(별표, 여러 열, INT/STRING 조건 지원)
test_sql1 = "SELECT id, name FROM users WHERE age > 18"
test_sql2 = "SELECT * FROM orders WHERE product = 'phone'"
parse_sql(test_sql1)
parse_sql(test_sql2)

실행 결과 (AST)

첫 번째 테스트 케이스의 AST(SELECT id, name FROM users WHERE age > 18):

{
  "type": "Query",
  "select": {
    "type": "SelectClause",
    "columns": ["id", "name"]
  },
  "from_clause": {
    "type": "FromClause",
    "table": "users"
  },
  "where_clause": {
    "type": "Condition",
    "column": "age",
    "operator": ">",
    "value": 18
  }
}

두 번째 테스트 케이스의 AST(SELECT * FROM orders WHERE product = 'phone'):

{
  "type": "Query",
  "select": {
    "type": "SelectClause",
    "columns": ["*"]
  },
  "from_clause": {
    "type": "FromClause",
    "table": "orders"
  },
  "where_clause": {
    "type": "Condition",
    "column": "product",
    "operator": "=",
    "value": "phone"
  }
}

III. 파싱 결과 적용: AST의 가치

AST가 있으면 많은 작업을 수행할 수 있습니다. 예를 들어 AST를 자연어 설명으로 변환하는 간단한 "쿼리 인터프리터"를 작성할 수 있습니다.

def interpret_ast(ast):
    if ast['type'] != 'Query':
        return "Unsupported statement type"
    
    # SELECT 절 파싱
    select_cols = ', '.join(ast['select']['columns'])
    select_desc = f"Query columns: {select_cols}"
    
    # FROM 절 파싱
    from_desc = f"From table: {ast['from_clause']['table']}"
    
    # WHERE 절 파싱
    where_desc = ""
    if ast['where_clause']:
        cond = ast['where_clause']
        where_desc = f", Filter condition: {cond['column']} {cond['operator']} {cond['value']}"
    
    return f"Execution logic: {select_desc} {from_desc}{where_desc}"

# 인터프리터 테스트
ast1 = parser.parse(test_sql1)
print(interpret_ast(ast1))  # 출력: Execution logic: Query columns: id, name From table: users, Filter condition: age > 18

IV. 제한 사항 및 고급 방향

이 기사에서 구현한 파서는 가장 기본적인 SQL 구문만 지원하며 명백한 제한 사항이 있습니다.

• 다중 테이블 조인(JOIN), 집계 함수(COUNT, SUM), 그룹화(GROUP BY)와 같은 복잡한 구문을 지원하지 않습니다.
• 의미 분석(예: 테이블/열이 존재하는지 또는 데이터 유형이 일치하는지 확인)을 지원하지 않습니다.
• SQL 주석 및 혼합 대소문자(예: Select)와 같은 에지 케이스를 처리하지 않습니다.

보다 실용적인 SQL 파싱 기능을 위해 다음과 같은 고급 방향을 참조할 수 있습니다.

1. 성숙한 라이브러리 사용: 산업 등급 시나리오의 경우 sqlparse(Python) 또는 antlr4(크로스 언어)와 같은 라이브러리를 사용하는 것을 우선시합니다. sqlparse는 복잡한 SQL을 직접 파싱하고 AST를 생성할 수 있습니다.
2. 문법 규칙 확장: ply를 기반으로 JOIN, GROUP BY 등에 대한 규칙을 추가합니다. 문법 우선순위(예: AND가 OR보다 우선순위가 높음)에 주의합니다.
3. 의미 분석: 구문 분석 후 "테이블 구조 확인" 단계를 추가하여 SELECT 절의 열이 FROM 절에 지정된 테이블에 존재하는지 확인합니다.
4. 쿼리 최적화: AST를 기반으로 쿼리를 최적화합니다(예: 필터 조건 푸시다운, 인덱스 선택). 이는 데이터베이스 커널의 핵심 기능입니다.

V. 결론

SQL 파싱의 본질은 "텍스트를 구조화된 데이터로 변환"하는 것이며, 이는 "어휘 분석을 통해 토큰으로 분할"하고 "구문 분석을 통해 AST를 구축"하는 두 가지 핵심 단계에 의존합니다. ply를 사용하여 이 기사에서 구현한 간단한 파서는 프로덕션 환경에 충분하지 않지만 파서의 작동 원리를 이해하는 데 도움이 될 수 있습니다.

Leapcell: 최고의 서버리스 웹 호스팅

마지막으로 Python 서비스를 배포하기 위한 훌륭한 플랫폼인 **Leapcell**을 추천합니다.

🚀 선호하는 언어로 구축

JavaScript, Python, Go 또는 Rust로 손쉽게 개발할 수 있습니다.

🌍 무제한 프로젝트를 무료로 배포

사용한 만큼만 지불하세요. 요청도 없고 요금도 없습니다.

⚡ 사용량에 따라 지불, 숨겨진 비용 없음

유휴 요금이 없고 원활한 확장성만 있습니다.

📖 문서 탐색

🔹 트위터에서 팔로우하세요: @LeapcellHQ