"""Google Cloud Spanner SQL dialect (GoogleSQL variant).
Extends the BigQuery dialect with Spanner-only DDL features: ``INTERLEAVE IN
[PARENT]`` for interleaved tables and ``ROW DELETION POLICY`` for row-level
time-to-live policies. The PostgreSQL-interface ``TTL INTERVAL ... ON`` form
is accepted on parse and normalized to the canonical row deletion policy so
generation always emits valid GoogleSQL.
"""
from typing import TYPE_CHECKING, Any
from sqlglot.dialects.bigquery import BigQuery
from sqlglot.tokenizer_core import TokenType
from sqlspec.dialects.spanner._generators import SpannerGenerator
from sqlspec.dialects.spanner._parsers import SpannerParser, attach_hints, normalize_spanner_tokens
if TYPE_CHECKING:
from sqlglot import exp
from sqlglot.tokenizer_core import Token
__all__ = ("Spanner",)
class SpannerTokenizer(BigQuery.Tokenizer):
"""Tokenizer for Spanner GoogleSQL string literal escapes and brace hints."""
STRING_ESCAPES = ["'", "\\"]
KEYWORDS = {**BigQuery.Tokenizer.KEYWORDS, "FLOAT32": TokenType.FLOAT, "TOKENLIST": TokenType.USERDEFINED}
def tokenize(self, sql: str) -> "list[Token]":
"""Tokenize Spanner SQL and normalize ``@{...}`` hints into token comments."""
return normalize_spanner_tokens(super().tokenize(sql), sql)
[docs]
class Spanner(BigQuery):
"""Google Cloud Spanner SQL dialect."""
Tokenizer = SpannerTokenizer
Parser = SpannerParser
Generator = SpannerGenerator
[docs]
def parse(self, sql: str, **opts: Any) -> "list[exp.Expr | None]":
"""Parse Spanner SQL statements and attach hints."""
expressions = super().parse(sql, **opts)
for expression in expressions:
if expression is not None:
attach_hints(expression)
return expressions
[docs]
def parse_into(self, expression_type: Any, sql: str, **opts: Any) -> "list[exp.Expr | None]":
"""Parse into specific expression type with attached hints."""
expressions = super().parse_into(expression_type, sql, **opts)
for expression in expressions:
if expression is not None:
attach_hints(expression)
return expressions
