# Module `paroxython.derived_labels_db`

An in-memory database collecting the labels of a given program.

### Main table

The database is not really relational. It merely consists in a table `t` where each row is a distinct occurrence of every label featured by a given program. Initially, `t` is populated with the labels found by the queries specified by a regular expression of `spec.md`.

The SQL queries are then executed, and the resulting labels (known as derived) inserted into `t` as they are found.

Note that the results of a given SQL query may themselves be required by a subsequent SQL query.

#### Example

The following simple SQL query of `spec.md` finds the loops of a program. The resulting label names (e.g., `"loop:while"`) consist in a fixed prefix `"loop"` and a suffix denoting the category of the loop (either `"for"` or `"while"`).

    SELECT "loop",
        name_prefix,
        span,
        path
    FROM t
    WHERE name_prefix IN ("for", "while")
    GROUP BY path

The execution of this query comes after the evaluation of all regular expressions, in particular those collecting the features `"for"` and `"while"`. It simply gathers them under the prefix `"loop"`, and qualifies them by their category.

The resulting labels will later be used by other SQL queries to search for features such as `"loop_with_break"`, `"loop_with_late_exit"`, `"loop_with_else"`, and so on.

### Subtables

For simplicity and performance reasons, a bunch of subtables are extracted on the fly from `t`. Actually, every time a query relies on a non-existing table named `t_PREFIX`, this table is dynamically created as the set of rows of `t` whose value on column `name_prefix` is `PREFIX`.

#### Example

The following SQL query of `spec.md` relies on the previous one to find the loops featuring a `break` statement. It is more complicated, since it suffixes `"loop_with_break"` with the span of the smallest enclosing loop. Anyway, it relies on the results of both the previous SQL query and a trivial regular expression matching the `break` statements.

    SELECT "loop_with_break",
        l.name_suffix,
        max(l.span_start) || "-" || min(l.span_end),
        max(l.path)
    FROM t_loop l
    JOIN t_break b ON (b.path GLOB l.path || "*-")
    GROUP BY b.rowid

Instead of referring the corresponding rows in `t` (which would complicate the query), two subtables restricted to them are referred, `t_loop` and `t_break`. For example, if a `t_loop` subtable does not exist yet, the system extracts it from `t` by executing:

    CREATE TABLE t_loop AS
    SELECT *
    FROM t
    WHERE name_prefix = 'loop'

The newly created table is then available for all subsequent queries that may need it, which can significantly speed up their execution.

### Note

Using views instead of tables turns out to be slightly slower. Also, for whatever mysterious reason, several queries produce an incorrect result.

Expand source code Browse GitHub

    import sqlite3
    from collections import defaultdict
    from typing import Any, Set, Callable, Dict

    import regex  # type: ignore

    from .goodies import couple_to_string, print_warning
    from .user_types import Label, LabelName, Labels, Query, Span

    class DerivedLabelsDatabase:

        columns = (
            # use rowid as primary key:
            "name TEXT",
            "name_prefix TEXT",
            "name_suffix TEXT",
            "span TEXT",
            "span_start INT",
            "span_end INT",
            "path TEXT",
        )
        def __init__(self):
            self.c = sqlite3.connect(":memory:")
            self.c.create_function("regexp", narg=2, func=lambda rex, s: bool(regex.match(rex, s)))
            # References:
            # - https://www.sqlite.org/inmemorydb.html
            # - https://www.sqlite.org/lang_expr.html#regexp
            # - https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function

        def create(
            self,
            labels: Labels,
            create_main_table: str = f"CREATE TABLE t ({','.join(columns)})",  # never provided
        ) -> None:
            self.c.execute(create_main_table)
            self.update(labels)
            self.subtables: Set[LabelName] = set()

        def read(
            self,
            query: Query,
            prerequisites: Callable = regex.compile(r"(?m)\b(?:FROM|JOIN) t_(\w+)").findall,
            create_subtable: str = "CREATE TABLE t_{0} AS SELECT * FROM t WHERE name_prefix = '{0}'",
        ) -> Labels:
            # Parse the prerequisites of the query, and create the corresponding subtables if needed.
            for label_name in prerequisites(query):
                if label_name not in self.subtables:
                    self.c.execute(create_subtable.format(label_name))
                self.subtables.add(label_name)
            self.c.commit()
            # Execute the query and return the resulting labels, deduplicated by (label_name, path).
            labels_spans: Dict[LabelName, Dict[Span, Any]] = defaultdict(dict)
            try:
                row_iterator = self.c.execute(query)
            except Exception as exception:  # pragma: no cover
                print_warning(f"problem in the following query:\n\n{query}\n")
                raise exception
            for (name_prefix, name_suffix, span_string, path) in row_iterator:
                label_name = f"{name_prefix}:{name_suffix}" if name_suffix != "" else name_prefix
                span = span_string.split("-")
                labels_spans[label_name][Span(int(span[0]), int(span[-1]), path)] = None
            return [Label(name, list(spans)) for (name, spans) in labels_spans.items()]

        def update(
            self,
            labels: Labels,
            update_query: str = f"INSERT INTO t VALUES ({','.join('?' * len(columns))})",
        ) -> None:
            rows = []
            for (name, spans) in labels:
                for span in spans:
                    (prefix, _, suffix) = name.partition(":")  # split on the first colon
                    span_string = couple_to_string(span)
                    rows.append((name, prefix, suffix, span_string, span.start, span.end, span.path))
            self.c.executemany(update_query, rows)

        def delete(self) -> None:
            self.c.execute("DROP TABLE t")
            for label_name in self.subtables:
                self.c.execute(f"DROP TABLE t_{label_name}")

        def __str__(self) -> str:  # pragma: no cover
            rows = ["name name_prefix name_suffix span span_start span_end path".split()]
            rows.extend(sorted(list(map(str, row)) for row in self.c.execute("SELECT * FROM t")))
            widths = list(map(len, [max(column, key=len) for column in zip(*rows)]))
            result = [""]
            for row in rows:
                result.append(" | ".join(s + " " * (w - len(s)) for (w, s) in zip(widths, row)))
            result[1:1] = ["-+-".join("-" * w for w in widths)]
            return "\n".join(result)

## Classes

`class DerivedLabelsDatabase`
  Expand source code Browse GitHub

      class DerivedLabelsDatabase:

          columns = (
              # use rowid as primary key:
              "name TEXT",
              "name_prefix TEXT",
              "name_suffix TEXT",
              "span TEXT",
              "span_start INT",
              "span_end INT",
              "path TEXT",
          )
          def __init__(self):
              self.c = sqlite3.connect(":memory:")
              self.c.create_function("regexp", narg=2, func=lambda rex, s: bool(regex.match(rex, s)))
              # References:
              # - https://www.sqlite.org/inmemorydb.html
              # - https://www.sqlite.org/lang_expr.html#regexp
              # - https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.create_function

          def create(
              self,
              labels: Labels,
              create_main_table: str = f"CREATE TABLE t ({','.join(columns)})",  # never provided
          ) -> None:
              self.c.execute(create_main_table)
              self.update(labels)
              self.subtables: Set[LabelName] = set()

          def read(
              self,
              query: Query,
              prerequisites: Callable = regex.compile(r"(?m)\b(?:FROM|JOIN) t_(\w+)").findall,
              create_subtable: str = "CREATE TABLE t_{0} AS SELECT * FROM t WHERE name_prefix = '{0}'",
          ) -> Labels:
              # Parse the prerequisites of the query, and create the corresponding subtables if needed.
              for label_name in prerequisites(query):
                  if label_name not in self.subtables:
                      self.c.execute(create_subtable.format(label_name))
                  self.subtables.add(label_name)
              self.c.commit()
              # Execute the query and return the resulting labels, deduplicated by (label_name, path).
              labels_spans: Dict[LabelName, Dict[Span, Any]] = defaultdict(dict)
              try:
                  row_iterator = self.c.execute(query)
              except Exception as exception:  # pragma: no cover
                  print_warning(f"problem in the following query:\n\n{query}\n")
                  raise exception
              for (name_prefix, name_suffix, span_string, path) in row_iterator:
                  label_name = f"{name_prefix}:{name_suffix}" if name_suffix != "" else name_prefix
                  span = span_string.split("-")
                  labels_spans[label_name][Span(int(span[0]), int(span[-1]), path)] = None
              return [Label(name, list(spans)) for (name, spans) in labels_spans.items()]

          def update(
              self,
              labels: Labels,
              update_query: str = f"INSERT INTO t VALUES ({','.join('?' * len(columns))})",
          ) -> None:
              rows = []
              for (name, spans) in labels:
                  for span in spans:
                      (prefix, _, suffix) = name.partition(":")  # split on the first colon
                      span_string = couple_to_string(span)
                      rows.append((name, prefix, suffix, span_string, span.start, span.end, span.path))
              self.c.executemany(update_query, rows)

          def delete(self) -> None:
              self.c.execute("DROP TABLE t")
              for label_name in self.subtables:
                  self.c.execute(f"DROP TABLE t_{label_name}")

          def __str__(self) -> str:  # pragma: no cover
              rows = ["name name_prefix name_suffix span span_start span_end path".split()]
              rows.extend(sorted(list(map(str, row)) for row in self.c.execute("SELECT * FROM t")))
              widths = list(map(len, [max(column, key=len) for column in zip(*rows)]))
              result = [""]
              for row in rows:
                  result.append(" | ".join(s + " " * (w - len(s)) for (w, s) in zip(widths, row)))
              result[1:1] = ["-+-".join("-" * w for w in widths)]
              return "\n".join(result)

  ### Class variables

  `var columns`

    The structure of the main table `t`, consisting of the following columns:

    Column | Type | Description and example
    --- | --- | ---
    `rowid` | `INT` | Primary key (automatically generated)
    `name` | `TEXT` | Complete name of the label, e.g.
    |  | `"loop_with_early_exit:for:break"`
    `name_prefix` | `TEXT` | Its prefix (part before the first colon, here
    |  | `"loop_with_early_exit"`
    `name_suffix` | `TEXT` | Its suffix (may contain one or more colons),
    |  | here `"for:break"`
    `span` | `TEXT` | Numbers of the first and last lines of the
    |  | feature as a hyphen-separated string, e.g. `"2-8"`
    `span_start` | `INT` | Its first line number, here `2`
    `span_end` | `INT` | Its first line number, here `8`
    `path` | `TEXT` | String encoding the beginning of the feature
    |  | as its path from the root of the AST, e.g.
    |  | `"3-5-3-0-1-2-2-1-"`

  ### Methods

  `def create(self,
  labels: List[Label],
  create_main_table: str = 'CREATE TABLE t (name TEXT,name_prefix TEXT,name_suffix TEXT,span TEXT,span_start INT,span_end INT,path TEXT)'
  ) ‑> NoneType`

    Create the main table and populate it with regex-specified labels.

    ## Args

    `labels` :`Labels`
      The label names and spans featured by a certain program. Only those specified by a regular expression in `spec.md` are included. It is the very purpose of this module to add the remaining ones, specified by a SQL query.
    `create_main_table` :`str`, optional
      The SQL query creating the table. Not to be explicitly provided. Defaults to `f"CREATE TABLE t ({','.join(columns)})"`.
    Expand source code Browse GitHub

        def create(
            self,
            labels: Labels,
            create_main_table: str = f"CREATE TABLE t ({','.join(columns)})",  # never provided
        ) -> None:
            self.c.execute(create_main_table)
            self.update(labels)
            self.subtables: Set[LabelName] = set()

  `def read(self,
  query: Query,
  prerequisites: Callable = <built-in method findall of regex>,
  create_subtable: str = "CREATE TABLE t_{0} AS SELECT * FROM t WHERE name_prefix = '{0}'"
  ) ‑> List[Label]`

    Apply the given SQL query to the currently known labels, and returns the new labels.

    ## Args

    `query` :`Query`
      The SQL specification of a certain feature, as defined in `spec.md`.
    `prerequisites` :`Callable`, optional
      A function taking an SQL query, and returning the table names referred to in the FROM clause. Not to be explicitly provided. Defaults to `regex.compile(r"(?m)\b(?:FROM|JOIN) t_(\w+)").findall`.
    `create_subtable` :`str`, optional
      The SQL query creating a subtable from a label prefix. Not to be explicitly provided. Defaults to `"CREATE TABLE t_{0} AS SELECT * FROM t WHERE name_prefix = '{0}'"`.

    ## Returns

    `Labels`
      The additional labels found by the given SQL query.

    ## Note

    As a side-effect, create in the database any non-existing subtable of `t` referred to as a prerequisite.

    Expand source code Browse GitHub

        def read(
            self,
            query: Query,
            prerequisites: Callable = regex.compile(r"(?m)\b(?:FROM|JOIN) t_(\w+)").findall,
            create_subtable: str = "CREATE TABLE t_{0} AS SELECT * FROM t WHERE name_prefix = '{0}'",
        ) -> Labels:
            # Parse the prerequisites of the query, and create the corresponding subtables if needed.
            for label_name in prerequisites(query):
                if label_name not in self.subtables:
                    self.c.execute(create_subtable.format(label_name))
                self.subtables.add(label_name)
            self.c.commit()
            # Execute the query and return the resulting labels, deduplicated by (label_name, path).
            labels_spans: Dict[LabelName, Dict[Span, Any]] = defaultdict(dict)
            try:
                row_iterator = self.c.execute(query)
            except Exception as exception:  # pragma: no cover
                print_warning(f"problem in the following query:\n\n{query}\n")
                raise exception
            for (name_prefix, name_suffix, span_string, path) in row_iterator:
                label_name = f"{name_prefix}:{name_suffix}" if name_suffix != "" else name_prefix
                span = span_string.split("-")
                labels_spans[label_name][Span(int(span[0]), int(span[-1]), path)] = None
            return [Label(name, list(spans)) for (name, spans) in labels_spans.items()]

  `def update(self,
  labels: List[Label],
  update_query: str = 'INSERT INTO t VALUES (?,?,?,?,?,?,?)'
  ) ‑> NoneType`

    Insert a set of labels into the main table `t`.

    ## Args

    `labels` :`Labels`
      The labels to be inserted.
    `update_query` :`str`, optional
      The SQL query inserting a row in `t`. Not to be explicitly provided. Defaults to `f"INSERT INTO t VALUES ({','.join('?' * len(columns))})"`.
    Expand source code Browse GitHub

        def update(
            self,
            labels: Labels,
            update_query: str = f"INSERT INTO t VALUES ({','.join('?' * len(columns))})",
        ) -> None:
            rows = []
            for (name, spans) in labels:
                for span in spans:
                    (prefix, _, suffix) = name.partition(":")  # split on the first colon
                    span_string = couple_to_string(span)
                    rows.append((name, prefix, suffix, span_string, span.start, span.end, span.path))
            self.c.executemany(update_query, rows)

  `def delete(self) ‑> NoneType`

    Empty the database after having executed all the SQL queries.

    Expand source code Browse GitHub

        def delete(self) -> None:
            self.c.execute("DROP TABLE t")
            for label_name in self.subtables:
                self.c.execute(f"DROP TABLE t_{label_name}")