Fix edgeql+schema patching (again) (#8302)

As usual, I broke it in between major versions. This time mostly
because of trampolines, though also because of a wrong assert
that supported an incorrect typing.

Also, add a +testmode kind modifier that only applies the patch in
testmode.

edb/pgsql/metaschema.py

@@ -8140,6 +8140,17 @@ def get_synthetic_type_views(
     return commands
 
 
+def _get_wrapper_views() -> dbops.CommandGroup:
+    # Create some trampolined wrapper views around _Schema types we need
+    # to reference from functions.
+    wrapper_commands = dbops.CommandGroup()
+    wrapper_commands.add_command(
+        dbops.CreateView(ObjectAncestorsView(), or_replace=True))
+    wrapper_commands.add_command(
+        dbops.CreateView(LinksView(), or_replace=True))
+    return wrapper_commands
+
+
 def get_support_views(
     schema: s_schema.Schema,
     backend_params: params.BackendRuntimeParams,
@@ -8169,13 +8180,7 @@ def get_support_views(
     synthetic_types = get_synthetic_type_views(schema, backend_params)
     commands.add_command(synthetic_types)
 
-    # Create some trampolined wrapper views around _Schema types we need
-    # to reference from functions.
-    wrapper_commands = dbops.CommandGroup()
-    wrapper_commands.add_command(
-        dbops.CreateView(ObjectAncestorsView(), or_replace=True))
-    wrapper_commands.add_command(
-        dbops.CreateView(LinksView(), or_replace=True))
+    wrapper_commands = _get_wrapper_views()
     commands.add_command(wrapper_commands)
 
     sys_alias_views = _generate_schema_alias_views(
@@ -8239,10 +8244,9 @@ async def generate_support_functions(
     return trampoline_functions(cmds)
 
 
-async def regenerate_config_support_functions(
-    conn: PGConnection,
+def _get_regenerated_config_support_functions(
     config_spec: edbconfig.Spec,
-) -> None:
+) -> dbops.CommandGroup:
     # Regenerate functions dependent on config spec.
     commands = dbops.CommandGroup()
 
@@ -8254,6 +8258,15 @@ async def regenerate_config_support_functions(
     cmds = [dbops.CreateFunction(func, or_replace=True) for func in funcs]
     commands.add_commands(cmds)
 
+    return commands
+
+
+async def regenerate_config_support_functions(
+    conn: PGConnection,
+    config_spec: edbconfig.Spec,
+) -> None:
+    # Regenerate functions dependent on config spec.
+    commands = _get_regenerated_config_support_functions(config_spec)
     block = dbops.PLTopBlock()
     commands.generate(block)
     await _execute_block(conn, block)

edb/pgsql/patches.py

@@ -57,6 +57,7 @@ def get_version_key(num_patches: int):
  * ext-pkg - installs an extension package given a name
  * repair - fix up inconsistencies in *user* schemas
  * sql-introspection - refresh all sql introspection views
+ * ...+testmode - only run the patch in testmode. Works with any patch kind.
 """
 PATCHES: list[tuple[str, str]] = [
 ]

edb/pgsql/trampoline.py

@@ -97,6 +97,10 @@ class Trampoline:
     def make(self) -> dbops.Command:
         pass
 
+    @abc.abstractmethod
+    def drop(self) -> dbops.Command:
+        pass
+
 
 @dataclasses.dataclass
 class TrampolineFunction(Trampoline):
@@ -105,6 +109,14 @@ class TrampolineFunction(Trampoline):
     def make(self) -> dbops.Command:
         return dbops.CreateFunction(self.func, or_replace=True)
 
+    def drop(self) -> dbops.Command:
+        return dbops.DropFunction(
+            self.func.name,
+            args=self.func.args or (),
+            has_variadic=bool(self.func.has_variadic),
+            if_exists=True,
+        )
+
 
 @dataclasses.dataclass
 class TrampolineView(Trampoline):
@@ -116,6 +128,12 @@ def make(self) -> dbops.Command:
                 {ql(q(*self.old_name))}, {ql(self.name[0])}, {ql(self.name[1])})
         ''')
 
+    def drop(self) -> dbops.Command:
+        return dbops.DropView(
+            self.name,
+            conditions=[dbops.ViewExists(self.name)],
+        )
+
 
 def make_trampoline(func: dbops.Function) -> TrampolineFunction:
     new_func = copy.copy(func)

edb/server/bootstrap.py

@@ -31,8 +31,10 @@
     Dict,
     List,
     Set,
+    Sequence,
     NamedTuple,
     TYPE_CHECKING,
+    cast,
 )
 
 import dataclasses
@@ -108,6 +110,7 @@ class ClusterMode(enum.IntEnum):
 
 
 _T = TypeVar("_T")
+_T_Schema = TypeVar("_T_Schema", bound=s_schema.Schema)
 
 
 # A simple connection proxy that reconnects and retries queries
@@ -555,11 +558,11 @@ async def _store_static_json_cache(
 
 
 def _process_delta_params(
-    delta, schema: s_schema.Schema, params,
+    delta, schema: _T_Schema, params,
     stdmode: bool=True,
     **kwargs,
 ) -> tuple[
-    s_schema.ChainedSchema,
+    _T_Schema,
     delta_cmds.MetaCommand,
     delta_cmds.CreateTrampolines,
 ]:
@@ -581,13 +584,15 @@ def _process_delta_params(
         backend_runtime_params=params,
         **kwargs,
     )
-    schema = sd.apply(delta, schema=schema, context=context)
+    schema = cast(
+        _T_Schema,
+        sd.apply(delta, schema=schema, context=context),
+    )
 
     if debug.flags.delta_pgsql_plan:
         debug.header('PgSQL Delta Plan')
         debug.dump(delta, schema=schema)
 
-    assert isinstance(schema, s_schema.ChainedSchema)
     if isinstance(delta, delta_cmds.DeltaRoot):
         out = delta.create_trampolines
     else:
@@ -744,7 +749,7 @@ async def gather_patch_info(
 
 
 def _generate_drop_views(
-    group: dbops.CommandGroup,
+    group: Sequence[dbops.Command | trampoline.Trampoline],
     preblock: dbops.PLBlock,
 ) -> None:
     for cv in reversed(list(group)):
@@ -770,6 +775,8 @@ def _generate_drop_views(
                 has_variadic=bool(cv.function.has_variadic),
                 if_exists=True,
             )
+        elif isinstance(cv, trampoline.Trampoline):
+            dv = cv.drop()
         else:
             raise AssertionError(f'unsupported support view command {cv}')
         dv.generate(preblock)
@@ -801,6 +808,12 @@ def prepare_patch(
 
     existing_view_columns = patch_info
 
+    if '+testmode' in kind:
+        if schema.get('cfg::TestSessionConfig', default=None):
+            kind = kind.replace('+testmode', '')
+        else:
+            return (update,), (), {}
+
     # Pure SQL patches are simple
     if kind == 'sql':
         return (patch, update), (), {}
@@ -949,7 +962,7 @@ def prepare_patch(
         )
         support_view_commands.generate(subblock)
 
-        _generate_drop_views(support_view_commands, preblock)
+        _generate_drop_views(list(support_view_commands), preblock)
 
         metadata_user_schema = reflschema
 
@@ -1004,8 +1017,17 @@ def prepare_patch(
                 backend_params.instance_params.version
             )
         )
+        wrapper_views = metaschema._get_wrapper_views()
+        support_view_commands.add_commands(list(wrapper_views))
+        trampolines = metaschema.trampoline_command(wrapper_views)
+
+        _generate_drop_views(
+            tuple(support_view_commands) + tuple(trampolines),
+            preblock,
+        )
 
-        _generate_drop_views(support_view_commands, preblock)
+        # Now add the trampolines to support_view_commands
+        support_view_commands.add_commands([t.make() for t in trampolines])
 
         # We want to limit how much unconditional work we do, so only recreate
         # extension views if requested.
@@ -1014,15 +1036,21 @@ def prepare_patch(
                 support_view_commands.add_command(
                     dbops.CreateView(extview, or_replace=True))
 
+        # Though we always update the instdata for the config system,
+        # because it is currently the most convenient way to make sure
+        # all the versioned fields get updated.
+        config_spec = config.load_spec_from_schema(schema)
+
         # Similarly, only do config system updates if requested.
         if '+config' in kind:
             support_view_commands.add_command(
                 metaschema.get_config_views(schema, existing_view_columns))
+            support_view_commands.add_command(
+                metaschema._get_regenerated_config_support_functions(
+                    config_spec
+                )
+            )
 
-        # Though we always update the instdata for the config system,
-        # because it is currently the most convenient way to make sure
-        # all the versioned fields get updated.
-        config_spec = config.load_spec_from_schema(schema)
         (
             sysqueries,
             report_configs_typedesc_1_0,