refactor: declare hidden attributes in DataJoint notation

Platform columns are now written the way a user writes an attribute, and
appended to the table definition before it is parsed. The ordinary machinery
then does everything -- backend type mapping, the `:type:` comment, and the
column-comment bookkeeping PostgreSQL needs for its out-of-line COMMENT ON.
No Python construction, and no adapter method.

    _job_start_time = null : datetime(3) # when computation began
    _job_duration   = null : float32     # computation duration in seconds
    _job_version    = ""   : varchar(64) # code version
    _prov           = null : json        # extrinsic provenance ...
    _singleton      = 1    : bool        # singleton primary key

Grammar and policy are separated to make that possible. The attribute grammar
accepts a leading underscore, so the framework can spell its own columns; a
*user* declaring one is refused by `_reject_user_hidden_attributes`, called from
declare() on the user's definition before anything is appended. That ordering is
the whole mechanism: the guard never sees the framework's lines.

Placement matters and is not uniform:

- Job metadata and provenance are secondary, so they follow every user
  attribute. A `---` is inserted when the definition has none, because
  otherwise a table whose attributes are all primary key takes a nullable
  hidden column as a nullable key attribute and is rejected.
- `_singleton` is the exception: it *is* the primary key, so it goes into the
  key section of a table that declares none of its own. Detection is by the
  absence of attribute or foreign-key lines ahead of the separator, so a
  leading table comment does not mask it.

The test for the underscore ban now exercises the user-facing path rather than
compile_attribute, which deliberately accepts these names. It gains coverage for
per-tier placement -- Entry gets `_prov`, Computed and Imported get job
metadata, Lookup, parts and job tables get neither.

deploy.add_prov_column no longer reaches into the in-process heading cache. A
Heading is memoized from the database, and refreshing it meant poking private
state and guessing which class to poke; a deploy operation runs before the
workers that write through it, as set_replica_identity does. The requirement is
documented instead, and the test reloads the way a deployment would.

681 passed, 14 skipped.

Author: dimitri-yatsenko

src/datajoint/declare.py

@@ -12,7 +12,6 @@
 
 import pyparsing as pp
 
-from . import provenance
 from .codecs import lookup_codec
 from .condition import translate_attribute
 from .errors import DataJointError
@@ -157,7 +156,10 @@ def build_attribute_parser() -> pp.ParserElement:
     """
     quoted = pp.QuotedString('"') ^ pp.QuotedString("'")
     colon = pp.Literal(":").suppress()
-    attribute_name = pp.Word(pp.srange("[a-z]"), pp.srange("[a-z0-9_]")).set_results_name("name")
+    # A leading underscore is permitted by the grammar so the framework can
+    # declare its own columns in this notation. Whether a *user* may is policy,
+    # enforced in prepare_declare where a user's definition is parsed.
+    attribute_name = pp.Word(pp.srange("[a-z_]"), pp.srange("[a-z0-9_]")).set_results_name("name")
     data_type = (
         pp.Combine(pp.Word(pp.alphas) + pp.SkipTo("#", ignore=quoted))
         ^ pp.QuotedString("<", end_quote_char=">", unquote_results=False)
@@ -452,6 +454,84 @@ def _covered(candidate: list[str]) -> bool:
     )
 
 
+#: Primary key for a table that declares none of its own.
+SINGLETON_DEFINITION = "_singleton = 1 : bool # singleton primary key"
+
+
+def _reject_user_hidden_attributes(definition) -> None:
+    """Refuse a user-declared attribute whose name begins with an underscore.
+
+    Policy, not grammar: the parser accepts such a name so that the framework can
+    declare its own columns in the same notation. Only a definition written by a
+    user passes through here.
+    """
+    lines = definition.split("\n") if isinstance(definition, str) else definition
+    for line in lines:
+        stripped = line.strip()
+        if not stripped or stripped.startswith("#") or stripped.startswith("---"):
+            continue
+        if is_foreign_key(stripped):
+            continue
+        if stripped.startswith("_"):
+            raise DataJointError(
+                f'Attribute name in line "{line}" starts with an underscore. '
+                "Names with leading underscore are reserved for platform-managed "
+                "columns (e.g. _job_start_time, _singleton). Use a regular "
+                "attribute name; if you need to control visibility at the call "
+                "site, use proj()."
+            )
+
+
+def _append_platform_attributes(definition, table_name: str, config) -> str:
+    """Add the framework's own hidden attributes to a table definition.
+
+    They are written in DataJoint notation and parsed by the same machinery as
+    any user attribute, so the backend type mapping, the ``:type:`` comment and
+    the column-comment bookkeeping all come from the one path that owns them.
+
+    Placement matters. Job metadata and provenance are secondary, so they go
+    after every user attribute and need a ``---`` ahead of them -- without one,
+    a definition whose attributes are all primary key would take a nullable
+    hidden column as a nullable key attribute and be rejected. ``_singleton`` is
+    the exception: it *is* the primary key, so it goes into the key section of a
+    table that declares none of its own.
+    """
+    from .jobs import JOB_METADATA_DEFINITION
+    from .provenance import PROV_DEFINITION
+    from .user_tables import Manual
+
+    lines = list(definition) if not isinstance(definition, str) else definition.split("\n")
+
+    def is_attribute(line: str) -> bool:
+        stripped = line.strip()
+        return bool(stripped) and not stripped.startswith("#") and not stripped.startswith("---")
+
+    separator = next((i for i, line in enumerate(lines) if line.strip().startswith("---")), None)
+    key_lines = lines[:separator] if separator is not None else lines
+
+    secondary = []
+    # Computed (__) and Imported (_) tables, but not a part (__ in the middle).
+    is_computed = table_name.startswith("__") and "__" not in table_name[2:]
+    is_imported = table_name.startswith("_") and not table_name.startswith("__")
+    if config.jobs.add_job_metadata and (is_computed or is_imported):
+        secondary.extend(JOB_METADATA_DEFINITION)
+
+    # Entry tables, where rows enter from outside. Matched against the Manual
+    # tier itself rather than by excluding the other tiers' prefixes.
+    if config.provenance.capture and re.fullmatch(Manual.tier_regexp, table_name):
+        secondary.append(PROV_DEFINITION)
+
+    # A table that declares no primary key of its own gets the sentinel.
+    singleton = [] if any(is_attribute(line) for line in key_lines) else [SINGLETON_DEFINITION]
+
+    if not singleton and not secondary:
+        return definition if isinstance(definition, str) else "\n".join(lines)
+
+    if separator is None:
+        return "\n".join(key_lines + singleton + ["---"] + secondary)
+    return "\n".join(lines[:separator] + singleton + lines[separator:] + secondary)
+
+
 def declare(
     full_table_name: str, definition: str, context: dict, adapter, *, config=None
 ) -> tuple[str, list[str], list[str], dict[str, tuple[str, str]], list[str], list[str]]:
@@ -498,6 +578,14 @@ def declare(
             )
         )
 
+    if config is None:
+        from .settings import config as _config
+
+        config = _config
+
+    _reject_user_hidden_attributes(definition)
+    definition = _append_platform_attributes(definition, table_name, config)
+
     (
         table_comment,
         primary_key,
@@ -509,53 +597,6 @@ def declare(
         column_comments,
     ) = prepare_declare(definition, context, adapter)
 
-    # Add hidden job metadata for Computed/Imported tables (not parts)
-    if config is None:
-        from .settings import config as _config
-
-        config = _config
-    if config.jobs.add_job_metadata:
-        # Check if this is a Computed (__) or Imported (_) table, but not a Part (contains __ in middle)
-        is_computed = table_name.startswith("__") and "__" not in table_name[2:]
-        is_imported = table_name.startswith("_") and not table_name.startswith("__")
-        if is_computed or is_imported:
-            # Deferred import: jobs imports table, which imports this module.
-            from .jobs import JOB_METADATA_SPEC, job_metadata_column_definitions
-
-            attribute_sql.extend(job_metadata_column_definitions(adapter))
-            for name, core_type, _default, comment in JOB_METADATA_SPEC:
-                column_comments[name] = f":{core_type}:{comment}"
-
-    # Add the hidden extrinsic-provenance slot to Entry tables, where rows enter
-    # from outside the pipeline.  Computed and Imported tables have no use for
-    # it -- their provenance is entailed by the foreign-key graph -- and a part
-    # inherits its master's.
-    # Matched against the Manual tier itself, not by excluding the other tiers'
-    # prefixes: enumerating exclusions makes every tier added later an Entry
-    # table by default, which is how job tables (`~`) first acquired the slot.
-    # Imported here rather than at module scope: user_tables imports table,
-    # which imports this module.
-    from .user_tables import Manual
-
-    if config.provenance.capture and re.fullmatch(Manual.tier_regexp, table_name):
-        attribute_sql.append(provenance.column_definition(adapter))
-        column_comments[provenance.PROV_ATTRIBUTE] = provenance.PROV_COMMENT
-
-    if not primary_key:
-        # Singleton table: add hidden sentinel attribute
-        primary_key = ["_singleton"]
-        singleton_comment = ":bool:singleton primary key"
-        sql_type = adapter.core_type_to_sql("bool")
-        singleton_sql = adapter.format_column_definition(
-            name="_singleton",
-            sql_type=sql_type,
-            nullable=False,
-            default="NOT NULL DEFAULT TRUE",
-            comment=singleton_comment,
-        )
-        attribute_sql.insert(0, singleton_sql)
-        column_comments["_singleton"] = singleton_comment
-
     pre_ddl = []  # DDL to run BEFORE CREATE TABLE (e.g., CREATE TYPE for enums)
     post_ddl = []  # DDL to run AFTER CREATE TABLE (e.g., COMMENT ON)
 
@@ -964,14 +1005,6 @@ def compile_attribute(
     DataJointError
         If syntax is invalid, primary key is nullable, or blob has invalid default.
     """
-    if line.lstrip().startswith("_"):
-        raise DataJointError(
-            f'Attribute name in line "{line}" starts with an underscore. '
-            "Names with leading underscore are reserved for platform-managed "
-            "columns (e.g. _job_start_time, _singleton). Use a regular "
-            "attribute name; if you need to control visibility at the call "
-            "site, use proj()."
-        )
     try:
         match = attribute_parser.parse_string(line + "#", parse_all=True)
     except pp.ParseException as err:

src/datajoint/deploy.py

@@ -186,30 +186,6 @@ def set_replica_identity(
     return result
 
 
-def _refresh_heading(target, table_name: str) -> None:
-    """Drop a cached heading so a freshly added column is visible in-process.
-
-    `Heading` loads from the database once and memoizes. After an ALTER the
-    cached copy is stale, and for `_prov` that is silent: `_has_prov_attribute`
-    reports False and inserts record nothing.
-    """
-    from .schemas import _Schema
-
-    candidates = []
-    if isinstance(target, _Schema):
-        context = getattr(target, "context", None) or {}
-        candidates = [cls for cls in context.values() if hasattr(cls, "table_name") and cls.table_name == table_name]
-    else:
-        candidates = [target]
-
-    for candidate in candidates:
-        instance = candidate() if isinstance(candidate, type) else candidate
-        heading = getattr(instance, "_heading", None) or getattr(type(instance), "_heading", None)
-        if heading is not None:
-            heading._attributes = None
-            heading._table_status = None
-
-
 def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
     """
     Add the hidden ``_prov`` attribute to Entry (``dj.Manual``) tables that lack it.
@@ -250,6 +226,11 @@ def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
       a part table inherits its master's.
     - Rows already present keep ``NULL``.  Provenance is recorded at insert and
       is never reconstructed after the fact.
+    - **Takes effect on the next schema load.**  A ``Heading`` is read from the
+      database once and memoized, so a table already in use keeps its pre-ALTER
+      attributes for the life of the process and its inserts go on recording
+      nothing.  This is a deploy-time operation: run it before the workers that
+      will write through it, as with :func:`set_replica_identity`.
     """
     import re
 
@@ -282,7 +263,7 @@ def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
         raise DataJointError("Cannot add the provenance column: the target has no database.")
 
     adapter = connection.adapter
-    column_sql = provenance.column_definition(adapter)
+    column_sql, _prov_comment = provenance.column_definition(adapter)
 
     # `tables_modified` and `columns_added` count work actually done, as
     # set_replica_identity does; a dry run reports through `ddl` and `details`.
@@ -317,11 +298,6 @@ def add_prov_column(target: "TargetType", dry_run: bool = True) -> dict:
         result["details"].append({"table": f"{database}.{table_name}", "status": "pending" if dry_run else "added"})
         if not dry_run:
             connection.query(ddl)
-            # Invalidate the cached heading: without this the table keeps its
-            # pre-ALTER attributes for the life of the process, `_prov` stays
-            # invisible to the insert path, and every insert silently records
-            # nothing until something reconnects.
-            _refresh_heading(target, table_name)
             result["tables_modified"] += 1
             result["columns_added"] += 1
 

src/datajoint/jobs.py

@@ -24,31 +24,16 @@
 logger = logging.getLogger(__name__.split(".")[0])
 
 
-#: Job metadata columns, as (name, DataJoint core type, default clause, comment).
-#: Declared through the adapter's type system rather than as hand-written SQL so
-#: the backend mapping lives in the one place that owns it -- `datetime(3)`
-#: becomes `timestamp(3)` on PostgreSQL, which the hand-written form dropped.
-JOB_METADATA_SPEC = (
-    ("_job_start_time", "datetime(3)", "DEFAULT NULL", "when computation began"),
-    ("_job_duration", "float32", "DEFAULT NULL", "computation duration in seconds"),
-    ("_job_version", "varchar(64)", "DEFAULT ''", "code version"),
+#: Job metadata columns, in DataJoint definition notation. Parsed by the same
+#: machinery as any user attribute, so the backend type mapping and the `:type:`
+#: comment come from the one place that owns them.
+JOB_METADATA_DEFINITION = (
+    "_job_start_time = null : datetime(3) # when computation began",
+    "_job_duration = null : float32 # computation duration in seconds",
+    '_job_version = "" : varchar(64) # code version',
 )
 
 
-def job_metadata_column_definitions(adapter) -> list[str]:
-    """Return the DDL fragments declaring the hidden job-metadata columns."""
-    return [
-        adapter.format_column_definition(
-            name=name,
-            sql_type=adapter.core_type_to_sql(core_type),
-            nullable=True,
-            default=default,
-            comment=f":{core_type}:{comment}",
-        )
-        for name, core_type, default, comment in JOB_METADATA_SPEC
-    ]
-
-
 def _get_job_version(config=None) -> str:
     """
     Get version string based on config settings.

src/datajoint/provenance.py

@@ -89,25 +89,23 @@ def _jsonable(value):
     return str(value)
 
 
-#: Recorded in the column comment so `heading` reads the declared core type back
-#: as ``original_type``, exactly as it does for a user-declared attribute.
-PROV_COMMENT = ":json:extrinsic provenance for a row that entered from outside"
+#: The attribute, in DataJoint definition notation -- parsed by the same
+#: machinery as any user attribute.
+PROV_DEFINITION = "_prov = null : json # extrinsic provenance for a row that entered from outside"
 
 
 def column_definition(adapter):
-    """Return the DDL fragment declaring ``_prov``, in DataJoint's type system.
+    """Return (DDL fragment, comment) declaring ``_prov``.
 
-    Built from ``core_type_to_sql("json")`` and ``format_column_definition``
-    rather than hand-written per backend, so the json/jsonb choice stays in the
-    one place that already owns it and a new adapter needs nothing added.
+    Used by ``deploy.add_prov_column`` to ALTER an existing table; declaration
+    appends :data:`PROV_DEFINITION` to the table definition instead.
     """
-    return adapter.format_column_definition(
-        name=PROV_ATTRIBUTE,
-        sql_type=adapter.core_type_to_sql("json"),
-        nullable=True,
-        default="DEFAULT NULL",
-        comment=PROV_COMMENT,
+    from .declare import compile_attribute
+
+    _name, sql, _store, comment = compile_attribute(
+        PROV_DEFINITION, in_key=False, foreign_key_sql=[], context={}, adapter=adapter
     )
+    return sql, comment
 
 
 def build_payload(connection, config=None):

tests/integration/test_entry_provenance.py

@@ -205,13 +205,11 @@ class Legacy(dj.Manual):
         # idempotent
         assert add_prov_column(Legacy, dry_run=False)["columns_added"] == 0
 
-        # No manual _init_from_database(): add_prov_column invalidates the cached
-        # heading, and the next access reloads it -- the same way the insert path
-        # does. Reaching for _init_from_database() here is what previously hid
-        # that inserts kept recording nothing until the process reconnected.
-        heading = Legacy().heading
-        heading.attributes  # force the lazy reload, as _has_prov_attribute does
-        assert provenance.PROV_ATTRIBUTE in heading._attributes
+        # add_prov_column is a deploy-time operation and does not reach into the
+        # in-process cache; the schema is reloaded before anything writes through
+        # it. Reload here for the same reason a deployment would.
+        Legacy().heading._init_from_database()
+        assert provenance.PROV_ATTRIBUTE in Legacy().heading._attributes
 
         # the pre-existing row keeps NULL; a new row carries a record
         Legacy.insert1({"legacy_id": 2, "note": "after"})