clickhouse_types.lark

WS: /[ \t\f\r\n]/+
DIGIT: "0".."9"
INT: DIGIT+
FLOAT: DIGIT? "." DIGIT+
     | DIGIT+ "."

int: INT
float: FLOAT

// simple types
SIMPLE: "String"
      | ("Int" | "UInt") ("8" | "16" | "32" | "64")
      | "Float" ("32" | "64")
      | "UUID"
      | "IPv4" | "IPv6"
      | "DateTime"
      | "Date"

simple: SIMPLE

// parametric types
PRECISION: "32" | "64" | "128"
precision: PRECISION -> int
scale: INT -> int
decimal: "Decimal" precision "(" scale ")"
       | "Decimal" "(" precision "," scale ")"
?int_arg: "(" INT ")" -> int
fixed_string: "FixedString" int_arg
?parametric: fixed_string | decimal

array: "Array" "(" _type ")"
tuple: "Tuple" "(" _type ("," _type)* ")"
?composite: array | tuple

// TODO: extend aggregate_function_decl
quantiles: "quantiles" "(" float ("," float)* ")"
FUNCTION_MODIFIER: ("ArrayIf" | "If" | "Array" | "ForEach" )? ("OrDefault" | "OrNull")?
FUNCTION: ("uniq" | "any" | "quantiles") FUNCTION_MODIFIER
// XXX: what to do with `Resample`?
_agg_func_param: int | float | _type
agg_func_params: "(" _agg_func_param ("," _agg_func_param)* ")"
aggregate_function: "AggregateFunction" "(" FUNCTION agg_func_params? "," _type ("," _type)* ")"

_ENUM_INNER: /.*?/
_ENUM_ESC_INNER: _ENUM_INNER /(?<!\\)(\\\\)*?/
ENUM_KEY: "'" _ENUM_ESC_INNER "'"
ENUM_TYPE: "Enum" ("8" | "16")?
enum: ENUM_TYPE "(" ENUM_KEY "=" int ("," ENUM_KEY "=" int)* ")"

nullable: "Nullable" "(" (simple | parametric | enum) ")"

_type: simple
     | parametric
     | composite
     | enum
     | aggregate_function
     | nullable

ESCAPED_IDENTIFIER: ("`" /.*/ "`") | ("\"" /.*/ "\"")
IDENTIFIER: /[a-zA-Z_][0-9a-zA-Z_]*/ | ESCAPED_IDENTIFIER
_nested_inner: IDENTIFIER _type ("," IDENTIFIER _type)*
nested: "Nested" "(" _nested_inner ")"

?start: _type
      | nested

%ignore WS

clickhouse_types.py

from pathlib import Path
from ast import literal_eval
import enum

from clickhouse_sqlalchemy import types

from lark import Lark, Transformer, v_args


@v_args(inline=True)
class TreeToClickhouseType(Transformer):
    def simple(self, t):
        return getattr(types, t)()

    def decimal(self, p, s):
        return types.Decimal(p, s)

    def fixed_string(self, l):
        try:
            return types.FixedString(l)
        except AttributeError:
            raise NotImplementedError("clickhouse_sqlalchemy.types doesn't implement FixedString yet")

    def int(self, v):
        return int(v)

    def array(self, *args):
        return types.Array(*args)

    def tuple(self, *args):
        return types.Tuple(*args)

    def aggregate_function(self, *args):
        raise NotImplementedError("clickhouse_sqlalchemy.types doesn't implement AggregateFunction yet")

    def enum(self, t, *args):
        return getattr(types, t)(enum.Enum(t, zip(map(literal_eval, args[::2]), args[1::2])))


parser = Lark(
    Path(__file__).with_suffix(".lark").read_text(),
    parser="lalr",
    lexer="contextual",
    transformer=TreeToClickhouseType(),
)

test_clickhouse_types.py

from unittest import TestCase

from clickhouse_sqlalchemy.drivers.base import ClickHouseDialect
from clickhouse_types import parser


class TestClickhouseTypes(TestCase):
    def test(self):
        tests = [
            i + j
            for i in ["UInt", "Int"]
            for j in ["8", "16", "32", "64"]
        ] + [
            "Float32", "Float64",
            "Array(UInt8)",
            "Enum('foo' = 100, 'bar' = 500)",
            "Enum8('foo' = 100, 'bar' = 500)",
            "Enum16('foo' = 100, 'bar' = 500)",
            "Enum8(' \\' t = ' = 1, 'test' = 2)",
        ]
        dialect = ClickHouseDialect()
        for i in tests:
            with self.subTest(type=i):
                parsed = parser.parse(i)
                compiled = parsed.compile(dialect)
                self.assertEqual(compiled, i)