Allow DML in REPEATABLE READ transactions but perform safety checks (#8561)

- In conflicts.py, implement analysis for what inserts and updates
   affect a cross-table constraint
 - Plumb that information out into the QueryUnit and to the server
 - Report the conflicts as part of CommandDataDescription in an
   annotation called `unsafe_isolation_dangers`, so clients can
   potentially make proactive decisions.
 - Raise an error when attempting to execute something with isolation
   dangers in REPEATABLE READ mode.

There are still a few follow-ups before I can merge, but this is
reviewable now:
 - [x] Write tests
 - [x] Follow-up on some possible bugs I found in the conflict mechanism
       that might have wider scope once this is implemented.
 - [x] Improve some documentation

Advances #8458.

Author: msullivan

edb/api/errors.txt

@@ -44,6 +44,7 @@
 0x_03_04_00_00   CapabilityError
 0x_03_04_01_00   UnsupportedCapabilityError
 0x_03_04_02_00   DisabledCapabilityError
+0x_03_04_03_00   UnsafeIsolationLevelError
 
 
 ####

edb/edgeql/compiler/__init__.py

@@ -366,6 +366,7 @@ def compile_ast_fragment_to_ir(
         singletons=[],
         triggers=(),
         warnings=tuple(ctx.env.warnings),
+        unsafe_isolation_dangers=tuple(ctx.env.unsafe_isolation_dangers),
     )
 
 

edb/edgeql/compiler/conflicts.py

@@ -887,3 +887,100 @@ def compile_inheritance_conflict_checks(
         )
 
     return conflicters or None
+
+
+def check_for_isolation_conflicts(
+    stmt: irast.MutatingStmt,
+    typ: s_objtypes.ObjectType,
+    update_typ: Optional[s_objtypes.ObjectType] = None,
+    *, ctx: context.ContextLevel,
+) -> None:
+    """Check for conflicts on a DML stmt that cause isolation dangers.
+
+    Cross-table exclusive constraints are implemented by triggers that
+    read the other tables looking for conflicting rows. This works
+    fine in SERIALIZABLE mode, but in REPEATABLE READ mode, this can
+    miss two concurrent transactions creating conflicting objects.
+
+    Analyze the type involved in `stmt` to see if there are isolation
+    dangers, and log them if so.  These will be reported to the client
+    and will generate an error if the query is executed in REPEATABLE
+    READ mode.
+
+    This function is called for every subtype in an UPDATE.  In that
+    case, `typ` is the subtype and `update_typ` is the base type being
+    UDPATEd.
+    """
+
+    schema = ctx.env.schema
+
+    entries = _get_type_conflict_constraint_entries(stmt, typ, ctx=ctx)
+    constrs = [cnstr for cnstr, _ in entries]
+
+    op = 'INSERT' if isinstance(stmt, irast.InsertStmt) else 'UPDATE'
+    base_msg = f'{op} to {typ.get_verbosename(schema)} '
+
+    for constr in constrs:
+        subject = constr.get_subject(schema)
+        assert subject
+
+        # Find the origin type; if we are the only origin type and we
+        # don't have children, then we are good.
+        all_objs = []
+        match constr.get_constraint_origins(schema):
+            case []:
+                continue
+            case [root_constr, *_]:
+                root_subject = root_constr.get_subject(schema)
+                if isinstance(root_subject, s_pointers.Pointer):
+                    root_subject_obj = root_subject.get_source(schema)
+                else:
+                    root_subject_obj = root_subject
+
+                if isinstance(root_subject_obj, s_objtypes.ObjectType):
+                    all_objs = list(
+                        schemactx.get_all_concrete(root_subject_obj, ctx=ctx)
+                    )
+                    if root_subject_obj == typ and len(all_objs) == 1:
+                        continue
+                    if root_subject_obj == typ and constr.get_delegated(schema):
+                        continue
+                    # If this is an UPDATE and we are processing some
+                    # subtype, and the actual type being updated is
+                    # covered by this same constraint, don't report it for
+                    # the child: it will be reported for an ancestor,
+                    # which is less noisy.
+                    if (
+                        update_typ
+                        and update_typ != typ
+                        and update_typ in all_objs
+                    ):
+                        continue
+
+        subj_vn = subject.get_verbosename(schema, with_parent=True)
+        vn = f'{base_msg}affects an exclusive constraint on {subj_vn}'
+        if expr := constr.get_subjectexpr(schema):
+            vn += f" with expression '{expr.text}'"
+
+        if not root_subject_obj:
+            msg = (
+                f"{vn} that is defined on "
+                f"{root_subject.get_verbosename(schema)}"
+            )
+        elif root_subject_obj != typ:
+            msg = (
+                f"{vn} that is defined in ancestor "
+                f"{root_subject_obj.get_verbosename(schema)}"
+            )
+        else:
+            all_objs_s = ', '.join(sorted(
+                f"'{o.get_displayname(schema)}'" for o in all_objs if o != typ
+            ))
+            msg = (
+                f"{vn} that is shared with "
+                f"descendant types: {all_objs_s}"
+            )
+
+        ctx.log_repeatable_read_danger(
+            errors.UnsafeIsolationLevelError(msg, span=stmt.span)
+        )

edb/edgeql/compiler/context.py

@@ -308,6 +308,9 @@ class Environment:
     warnings: list[errors.EdgeDBError]
     """List of warnings to emit"""
 
+    unsafe_isolation_dangers: list[errors.UnsafeIsolationLevelError]
+    """List of repeatable read DML dangers"""
+
     def __init__(
         self,
         *,
@@ -358,6 +361,7 @@ def __init__(
         self.path_scope_map = {}
         self.dml_rewrites = {}
         self.warnings = []
+        self.unsafe_isolation_dangers = []
 
     def add_schema_ref(
         self, sobj: s_obj.Object, expr: Optional[qlast.Base]
@@ -836,6 +840,11 @@ def get_security_context(self) -> object:
     def log_warning(self, warning: errors.EdgeDBError) -> None:
         self.env.warnings.append(warning)
 
+    def log_repeatable_read_danger(
+        self, d: errors.UnsafeIsolationLevelError
+    ) -> None:
+        self.env.unsafe_isolation_dangers.append(d)
+
     def allow_factoring(self) -> None:
         self.no_factoring = self.warn_factoring = False
 

edb/edgeql/compiler/stmt.py

@@ -544,6 +544,9 @@ def compile_InsertQuery(
                 stmt.on_conflict = conflicts.compile_insert_unless_conflict(
                     stmt, stmt_subject_stype, ctx=ictx)
 
+        conflicts.check_for_isolation_conflicts(
+            stmt, stmt_subject_stype, ctx=ictx)
+
         mat_stype = schemactx.get_material_type(stmt_subject_stype, ctx=ctx)
         result = setgen.class_set(
             mat_stype, path_id=stmt.subject.path_id, ctx=ctx
@@ -705,6 +708,9 @@ def compile_UpdateQuery(
             ):
                 stmt.write_policies[dtype.id] = write_pol
 
+            conflicts.check_for_isolation_conflicts(
+                stmt, dtype, mat_stype, ctx=ictx)
+
         stmt.conflict_checks = conflicts.compile_inheritance_conflict_checks(
             stmt, mat_stype, ctx=ictx)
 

edb/edgeql/compiler/stmtctx.py

@@ -360,6 +360,7 @@ def fini_expression(
         singletons=ctx.env.singletons,
         triggers=ir_triggers,
         warnings=tuple(ctx.env.warnings),
+        unsafe_isolation_dangers=tuple(ctx.env.unsafe_isolation_dangers),
     )
     return result
 

edb/errors/__init__.py

@@ -23,6 +23,7 @@
     'CapabilityError',
     'UnsupportedCapabilityError',
     'DisabledCapabilityError',
+    'UnsafeIsolationLevelError',
     'QueryError',
     'InvalidSyntaxError',
     'EdgeQLSyntaxError',
@@ -158,6 +159,10 @@ class DisabledCapabilityError(CapabilityError):
     _code = 0x_03_04_02_00
 
 
+class UnsafeIsolationLevelError(CapabilityError):
+    _code = 0x_03_04_03_00
+
+
 class QueryError(EdgeDBError):
     _code = 0x_04_00_00_00
 

edb/ir/ast.py

@@ -806,6 +806,7 @@ class Statement(Command):
     singletons: list[PathId]
     triggers: tuple[tuple[Trigger, ...], ...]
     warnings: tuple[errors.EdgeDBError, ...]
+    unsafe_isolation_dangers: tuple[errors.UnsafeIsolationLevelError, ...]
 
 
 class TypeIntrospection(ImmutableExpr):

edb/server/compiler/compiler.py

@@ -2005,6 +2005,7 @@ def _compile_ql_query(
         has_dml=bool(ir.dml_exprs),
         query_asts=query_asts,
         warnings=ir.warnings,
+        unsafe_isolation_dangers=ir.unsafe_isolation_dangers,
     )
 
 
@@ -2165,6 +2166,7 @@ def _compile_ql_transaction(
     final_global_schema: Optional[s_schema.Schema] = None
     sp_name = None
     sp_id = None
+    iso = None
 
     if ctx.expect_rollback and not isinstance(
         ql, (qlast.RollbackTransaction, qlast.RollbackToSavepoint)
@@ -2190,25 +2192,10 @@ def _compile_ql_transaction(
         # Compute the effective access mode
         access = ql.access
         if access is None:
-            if default_iso is qltypes.TransactionIsolationLevel.SERIALIZABLE:
-                access_mode: statypes.TransactionAccessMode = _get_config_val(
-                    ctx, "default_transaction_access_mode"
-                )
-                access = access_mode.to_qltypes()
-            else:
-                access = qltypes.TransactionAccessMode.READ_ONLY
-
-        # Guard against unsupported isolation + access combinations
-        if (
-            iso is not qltypes.TransactionIsolationLevel.SERIALIZABLE
-            and access is not qltypes.TransactionAccessMode.READ_ONLY
-        ):
-            raise errors.TransactionError(
-                f"{iso.value} transaction isolation level is only "
-                "supported in read-only transactions",
-                span=ql.span,
-                hint=f"specify READ ONLY access mode",
+            access_mode: statypes.TransactionAccessMode = _get_config_val(
+                ctx, "default_transaction_access_mode"
             )
+            access = access_mode.to_qltypes()
 
         sqls = f'START TRANSACTION ISOLATION LEVEL {iso.value} {access.value}'
         if ql.deferrable is not None:
@@ -2284,6 +2271,7 @@ def _compile_ql_transaction(
         global_schema=final_global_schema,
         sp_name=sp_name,
         sp_id=sp_id,
+        isolation_level=iso,
         feature_used_metrics=(
             ddl.produce_feature_used_metrics(
                 ctx.compiler_state, final_user_schema
@@ -3003,6 +2991,7 @@ def _make_query_unit(
         cache_key=ctx.cache_key,
         user_schema_version=schema_version,
         warnings=comp.warnings,
+        unsafe_isolation_dangers=comp.unsafe_isolation_dangers,
     )
 
     if not comp.is_transactional:
@@ -3093,6 +3082,7 @@ def _make_query_unit(
             )
         unit.sql = comp.sql
         unit.cacheable = comp.cacheable
+        unit.tx_isolation_level = comp.isolation_level
 
         if not ctx.dump_restore_mode:
             if comp.user_schema is not None:
@@ -3226,6 +3216,9 @@ def _make_query_unit(
     if unit.warnings:
         for warning in unit.warnings:
             warning.__traceback__ = None
+    if unit.unsafe_isolation_dangers:
+        for warning in unit.unsafe_isolation_dangers:
+            warning.__traceback__ = None
 
     return unit, final_user_schema
 

edb/server/compiler/dbstate.py

@@ -89,6 +89,9 @@ class BaseQuery:
     warnings: tuple[errors.EdgeDBError, ...] = dataclasses.field(
         kw_only=True, default=()
     )
+    unsafe_isolation_dangers: tuple[errors.UnsafeIsolationLevelError, ...] = (
+        dataclasses.field(kw_only=True, default=())
+    )
 
 
 @dataclasses.dataclass(frozen=True, kw_only=True)
@@ -179,6 +182,8 @@ class TxControlQuery(BaseQuery):
 
     modaliases: Optional[immutables.Map[Optional[str], str]]
 
+    isolation_level: Optional[qltypes.TransactionIsolationLevel] = None
+
     user_schema: Optional[s_schema.Schema] = None
     global_schema: Optional[s_schema.Schema] = None
     cached_reflection: Any = None
@@ -255,6 +260,9 @@ class QueryUnit:
     # starts a new transaction.
     tx_id: Optional[int] = None
 
+    # If this is the start of the transaction, the isolation level of it.
+    tx_isolation_level: Optional[qltypes.TransactionIsolationLevel] = None
+
     # True if this unit is single 'COMMIT' command.
     # 'COMMIT' is always compiled to a separate QueryUnit.
     tx_commit: bool = False
@@ -316,6 +324,7 @@ class QueryUnit:
     server_param_conversions: Optional[list[ServerParamConversion]] = None
 
     warnings: tuple[errors.EdgeDBError, ...] = ()
+    unsafe_isolation_dangers: tuple[errors.UnsafeIsolationLevelError, ...] = ()
 
     # Set only when this unit contains a CONFIGURE INSTANCE command.
     system_config: bool = False
@@ -430,6 +439,9 @@ class QueryUnitGroup:
     unit_converted_param_indexes: Optional[dict[int, list[int]]] = None
 
     warnings: Optional[list[errors.EdgeDBError]] = None
+    unsafe_isolation_dangers: (
+        Optional[list[errors.UnsafeIsolationLevelError]]
+    ) = None
 
     # Cacheable QueryUnit is serialized in the compiler, so that the I/O server
     # doesn't need to serialize it again for persistence.
@@ -522,6 +534,11 @@ def append(
             if self.warnings is None:
                 self.warnings = []
             self.warnings.extend(query_unit.warnings)
+        if query_unit.unsafe_isolation_dangers is not None:
+            if self.unsafe_isolation_dangers is None:
+                self.unsafe_isolation_dangers = []
+            self.unsafe_isolation_dangers.extend(
+                query_unit.unsafe_isolation_dangers)
 
         if not serialize or query_unit.cache_sql is None:
             self._units.append(query_unit)