feat: DSL parser + AST implementation

[AlSchlo]

Problem

We want a DSL to be able to write rules & operators in a declarative fashion. This makes the code more maintainable, maximizes compatibility, and speeds up the writing of newer rules.

Summary of changes

Wrote a parser of the OPTD-DSL using Pest. The syntax is highly functional and inspired from Scala. Some snippets below:

Next steps

Semantic analysis, type checking, and low-level IR generation.
Also switching to Chumsky as parser, since Pest is not very maintainable.

// Logical Properties
Logical Props(schema_len: Int64)

// Scalar Operators
Scalar ColumnRef(idx: Int64)
Scalar Mult(left: Int64, right: Int64)
Scalar Add(left: Int64, right: Int64)
Scalar And(children: [Scalar])
Scalar Or(children: [Scalar])
Scalar Not(child: Scalar)

// Logical Operators
Logical Scan(table_name: String) derive {
  schema_len = 5 // <call catalog>
}

Logical Filter(child: Logical, cond: Scalar) derive {
  schema_len = input.schema_len
}

Logical Project(child: Logical, exprs: [Scalar]) derive {
  schema_len = exprs.len
}

Logical Join(
    left: Logical,
    right: Logical,
    typ: String,
    cond: Scalar
) derive {
  schema_len = left.schema_len + right.schema_len
}

Logical Sort(child: Logical, keys: [(Scalar, String)]) derive {
  schema_len = input.schema_len
}

Logical Aggregate(child: Logical, group_keys: [Scalar], aggs: [(Scalar, String)]) 
  derive {
    schema_len = group_keys.len + aggs.len
}

// Rules
def rewrite_column_refs(predicate: Scalar, map: Map[Int64, Int64]): Scalar =
  match predicate
    case ColumnRef(idx) => ColumnRef(map(idx)),
    case other @ _ => predicate.apply_children(child => rewrite_column_refs(child, map))

@rule(Logical)
def join_commute(expr: Logical): Logical =
  match expr
    case Join("Inner", left, right, cond) =>
      val left_len = left.schema_len;
      val right_len = right.schema_len;
      
      val right_indices = 0..right_len;
      val left_indices = 0..left_len;
      
      val remapping = (left_indices.map(i => (i, i + right_len)) ++
        right_indices.map(i => (left_len + i, i))).to_map();

      Project(
        Join("Inner", right, left, rewrite_column_refs(cond, remapping)),
          left_indices.map(i => ColumnRef(i + right_len)) ++ 
          right_indices.map(i => ColumnRef(i)
        )
      )

def has_refs_in_range(cond: Scalar, from: Int64, to: Int64): Bool =
  match predicate
    case ColumnRef(idx) => from <= idx && idx < to,
    case _ => predicate.children.any(child => has_refs_in_range(child, from, to))
  

@rule(Logical)
def join_associate(expr: Logical): Logical =
  match expr
    case op @ Join("Inner", Join("Inner", a, b, cond_inner), c, cond_outer) =>
      val a_len = a.schema_len;
      if !has_refs_in_range(cond_outer, 0, a_len) then
        val remap_inner = (a.schema_len..op.schema_len).map(i => (i, i - a_len)).to_map();
        Join(
          "Inner", a, 
          Join("Inner", b, c, rewrite_column_refs(cond_outer, remap_inner), 
            cond_inner)
        )
      else fail("")

@rule(Scalar)
def conjunctive_normal_form(expr: Scalar): Scalar = fail("unimplemented")

def with_optional_filter(key: String, old: Scalar, grouped: Map[String, [Scalar]]): Scalar = 
  match grouped(key)
     case Some(conds) => Filter(conds, old),
     case _ => old

@rule(Logical)
def filter_pushdown_join(expr: Logical): Logical =
  match expr
    case op @ Filter(Join(join_type, left, right, join_cond), cond) =>
      val cnf = conjunctive_normal_form(cond);
      val grouped = cnf.children.groupBy(cond => {
        if has_refs_in_range(cond, 0, left.schema_len) && 
           !has_refs_in_range(cond, left.schema_len, op.schema_len) then
          "left"
        else if !has_refs_in_range(cond, 0, left.schema_len) && 
                has_refs_in_range(cond, left.schema_len, op.schema_len) then
          "right"
        else 
          "remain"
      });
 
      with_optional_filter("remain", 
        Join(join_type, 
          with_optional_filter("left", grouped), 
          with_optional_filter("right", grouped), 
          join_cond
        )
      )

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 89.22222% with 97 lines in your changes missing coverage. Please review.

Project coverage is 78.0%. Comparing base (a94829a) to head (cfb02bd).

Files with missing lines	Patch %	Lines
optd-core/src/dsl/parser/expr.rs	86.3%	30 Missing ⚠️
optd-core/src/dsl/parser/patterns.rs	83.0%	21 Missing ⚠️
optd-core/src/dsl/parser/operators.rs	90.4%	16 Missing ⚠️
optd-core/src/dsl/parser/functions.rs	90.3%	15 Missing ⚠️
optd-core/src/dsl/parser/types.rs	89.3%	11 Missing ⚠️
optd-core/src/dsl/parser/mod.rs	96.9%	4 Missing ⚠️

Additional details and impacted files

Files with missing lines	Coverage Δ
optd-core/src/cascades/mod.rs	99.2% <100.0%> (ø)
optd-core/src/dsl/parser/mod.rs	96.9% <96.9%> (ø)
optd-core/src/dsl/parser/types.rs	89.3% <89.3%> (ø)
optd-core/src/dsl/parser/functions.rs	90.3% <90.3%> (ø)
optd-core/src/dsl/parser/operators.rs	90.4% <90.4%> (ø)
optd-core/src/dsl/parser/patterns.rs	83.0% <83.0%> (ø)
optd-core/src/dsl/parser/expr.rs	86.3% <86.3%> (ø)

... and 1 file with indirect coverage changes

[yliang412]

nit: remember to change these to rust comments.

[skyzh]

LGTM, seems like we can parse the DSL now, and we need some more work to make the interpreter to work with it.

[skyzh]

rm unused files?

[AlSchlo]

ah damn i pushed name analyzer in here.

[skyzh]

not sure if this could be simplified with https://github.com/pest-parser/ast in the future, basically it's a conversion from pest::Pair -> AST

[skyzh]

⬇️ more reviews on the rule correctness

[skyzh]

why aggs [Scalar, String]? should be something like [Scalar] or [AggFunction]

[AlSchlo]

string is the aggfunction for now

we don't support UDTs yet

[skyzh]

need handle the case where map doesn't contain a mapping of a column referenced, throw an error that it can't rewrite in such cases

[AlSchlo]

i was planning to do that in the map implementation

[skyzh]

grammar wise, is the order required here? i.e., would Join("Inner", right, left, cond) produce a correct match?

proposing syntex,

Join { type: "Inner", left, right, cond }

[AlSchlo]

yes order is required

[skyzh]

right_indices don't seem to be set correctly, should be i -> i - left_len?

[AlSchlo]

no this is correct, the pair is reversed

[skyzh]

should be right -> right + left_len, left -> left - right_len?

[AlSchlo]

yeah there seems to be smt wrong here

[skyzh]

s/remap_inner/remap_outer?