Remove multiplicity from QueryUnit.sql (#7985)

Currently, `QueryUnit.sql` is a tuple representing, possibly, multiple
SQL statements corresponding to the original EdgeQL statement.  This
introduces significant complexity to consumers of `QueryUnit`, which are
mostly unprepared to handle more than one SQL statement anyway.

Originally the SQL tuple was used to represent multiple EdgeQL
statements,
a task which is now handled by the `QueryUnitGroup` stuff.  Another use
case are the non-transactional commands (`CREATE BRANCH` and friends).
Finally, since this facility was available, more uses of it were added
without actually _needing_ be executed as multiple SQL statements
with no other recourse: those are mostly maintenance commands and the
DDL type_id readback.

I think the above uses are no longer a sufficient reason to keep the
tuple complexity and so I'm ripping it out here, making `QueryUnit.sql`
a `bytes` property with the invariant of _always_ containing exactly one
SQL statement and thus not needing any special handling.  The users are
fixed up as follows:

- Non-transactional branch units encode the extra SQL needed to
  represent them in the new `QueryUnit.db_op_trailer` property which is
  a tuple of SQL.  Branch commands have special handling for them
  already, so this is not a nuisance.

- The newly-added type id mappings produced by DDL commands are now
  communicated via the new "indirect return" mechanism, whereby a DDL
  PLBlock can communicate a return value via a specially-formatted
  `NoticeResponse` message.

- All other unwitting users of multi-statement `QueryUnit` are converted
to use a single statement instead (primarily by converting them to use a
  `DO` block).

edb/buildmeta.py

@@ -60,7 +60,7 @@
 # The merge conflict there is a nice reminder that you probably need
 # to write a patch in edb/pgsql/patches.py, and then you should preserve
 # the old value.
-EDGEDB_CATALOG_VERSION = 2024_11_08_01_00
+EDGEDB_CATALOG_VERSION = 2024_11_12_00_00
 EDGEDB_MAJOR_VERSION = 6
 
 

edb/edgeql/ast.py

@@ -674,6 +674,11 @@ class DDLCommand(Command, DDLOperation):
     __abstract_node__ = True
 
 
+class NonTransactionalDDLCommand(DDLCommand):
+    __abstract_node__ = True
+    __rust_ignore__ = True
+
+
 class AlterAddInherit(DDLOperation):
     position: typing.Optional[Position] = None
     bases: typing.List[TypeName]
@@ -868,7 +873,7 @@ class BranchType(s_enum.StrEnum):
     TEMPLATE = 'TEMPLATE'
 
 
-class DatabaseCommand(ExternalObjectCommand):
+class DatabaseCommand(ExternalObjectCommand, NonTransactionalDDLCommand):
 
     __abstract_node__ = True
     __rust_ignore__ = True

edb/pgsql/dbops/base.py

@@ -504,6 +504,10 @@ def __repr__(self) -> str:
         return f'<Query {self.text!r}>'
 
 
+class PLQuery(Query):
+    pass
+
+
 class DefaultMeta(type):
     def __bool__(cls):
         return False

edb/pgsql/delta.py

@@ -2953,7 +2953,7 @@ def _alter_finalize(
         return schema
 
 
-def drop_dependant_func_cache(pg_type: Tuple[str, ...]) -> dbops.Query:
+def drop_dependant_func_cache(pg_type: Tuple[str, ...]) -> dbops.PLQuery:
     if len(pg_type) == 1:
         types_cte = f'''
                     SELECT
@@ -2980,7 +2980,6 @@ def drop_dependant_func_cache(pg_type: Tuple[str, ...]) -> dbops.Query:
                         )\
         '''
     drop_func_cache_sql = textwrap.dedent(f'''
-        DO $$
         DECLARE
             qc RECORD;
         BEGIN
@@ -3014,9 +3013,9 @@ class AS (
             LOOP
                 PERFORM edgedb_VER."_evict_query_cache"(qc.key);
             END LOOP;
-        END $$;
+        END;
     ''')
-    return dbops.Query(drop_func_cache_sql)
+    return dbops.PLQuery(drop_func_cache_sql)
 
 
 class DeleteScalarType(ScalarTypeMetaCommand,

edb/pgsql/metaschema.py

@@ -105,13 +105,13 @@ class PGConnection(Protocol):
 
     async def sql_execute(
         self,
-        sql: bytes | tuple[bytes, ...],
+        sql: bytes,
     ) -> None:
         ...
 
     async def sql_fetch(
         self,
-        sql: bytes | tuple[bytes, ...],
+        sql: bytes,
         *,
         args: tuple[bytes, ...] | list[bytes] = (),
     ) -> list[tuple[bytes, ...]]:
@@ -1497,6 +1497,83 @@ def __init__(self) -> None:
         )
 
 
+class RaiseNoticeFunction(trampoline.VersionedFunction):
+    text = '''
+    BEGIN
+        RAISE NOTICE USING
+            MESSAGE = "msg",
+            DETAIL = COALESCE("detail", ''),
+            HINT = COALESCE("hint", ''),
+            COLUMN = COALESCE("column", ''),
+            CONSTRAINT = COALESCE("constraint", ''),
+            DATATYPE = COALESCE("datatype", ''),
+            TABLE = COALESCE("table", ''),
+            SCHEMA = COALESCE("schema", '');
+        RETURN "rtype";
+    END;
+    '''
+
+    def __init__(self) -> None:
+        super().__init__(
+            name=('edgedb', 'notice'),
+            args=[
+                ('rtype', ('anyelement',)),
+                ('msg', ('text',), "''"),
+                ('detail', ('text',), "''"),
+                ('hint', ('text',), "''"),
+                ('column', ('text',), "''"),
+                ('constraint', ('text',), "''"),
+                ('datatype', ('text',), "''"),
+                ('table', ('text',), "''"),
+                ('schema', ('text',), "''"),
+            ],
+            returns=('anyelement',),
+            # NOTE: The main reason why we don't want this function to be
+            # immutable is that immutable functions can be
+            # pre-evaluated by the query planner once if they have
+            # constant arguments. This means that using this function
+            # as the second argument in a COALESCE will raise a
+            # notice regardless of whether the first argument is
+            # NULL or not.
+            volatility='stable',
+            language='plpgsql',
+            text=self.text,
+        )
+
+
+# edgedb.indirect_return() to be used to return values from
+# anonymous code blocks or other contexts that have no return
+# data channel.
+class IndirectReturnFunction(trampoline.VersionedFunction):
+    text = """
+    SELECT
+        edgedb_VER.notice(
+            NULL::text,
+            msg => 'edb:notice:indirect_return',
+            detail => "value"
+        )
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            name=('edgedb', 'indirect_return'),
+            args=[
+                ('value', ('text',)),
+            ],
+            returns=('text',),
+            # NOTE: The main reason why we don't want this function to be
+            # immutable is that immutable functions can be
+            # pre-evaluated by the query planner once if they have
+            # constant arguments. This means that using this function
+            # as the second argument in a COALESCE will raise a
+            # notice regardless of whether the first argument is
+            # NULL or not.
+            volatility='stable',
+            language='sql',
+            text=self.text,
+        )
+
+
 class RaiseExceptionFunction(trampoline.VersionedFunction):
     text = '''
     BEGIN
@@ -4980,6 +5057,8 @@ def get_bootstrap_commands(
         dbops.CreateFunction(GetSharedObjectMetadata()),
         dbops.CreateFunction(GetDatabaseMetadataFunction()),
         dbops.CreateFunction(GetCurrentDatabaseFunction()),
+        dbops.CreateFunction(RaiseNoticeFunction()),
+        dbops.CreateFunction(IndirectReturnFunction()),
         dbops.CreateFunction(RaiseExceptionFunction()),
         dbops.CreateFunction(RaiseExceptionOnNullFunction()),
         dbops.CreateFunction(RaiseExceptionOnNotNullFunction()),

edb/server/bootstrap.py

@@ -173,15 +173,15 @@ async def _retry_conn_errors(
 
         return result
 
-    async def sql_execute(self, sql: bytes | tuple[bytes, ...]) -> None:
+    async def sql_execute(self, sql: bytes) -> None:
         async def _task() -> None:
             assert self._conn is not None
             await self._conn.sql_execute(sql)
         return await self._retry_conn_errors(_task)
 
     async def sql_fetch(
         self,
-        sql: bytes | tuple[bytes, ...],
+        sql: bytes,
         *,
         args: tuple[bytes, ...] | list[bytes] = (),
     ) -> list[tuple[bytes, ...]]:
@@ -634,8 +634,8 @@ def compile_single_query(
 ) -> str:
     ql_source = edgeql.Source.from_string(eql)
     units = edbcompiler.compile(ctx=compilerctx, source=ql_source).units
-    assert len(units) == 1 and len(units[0].sql) == 1
-    return units[0].sql[0].decode()
+    assert len(units) == 1
+    return units[0].sql.decode()
 
 
 def _get_all_subcommands(
@@ -687,7 +687,7 @@ def prepare_repair_patch(
     schema_class_layout: s_refl.SchemaClassLayout,
     backend_params: params.BackendRuntimeParams,
     config: Any,
-) -> tuple[bytes, ...]:
+) -> bytes:
     compiler = edbcompiler.new_compiler(
         std_schema=stdschema,
         reflection_schema=reflschema,
@@ -701,7 +701,7 @@ def prepare_repair_patch(
     )
     res = edbcompiler.repair_schema(compilerctx)
     if not res:
-        return ()
+        return b""
     sql, _, _ = res
 
     return sql
@@ -2111,10 +2111,10 @@ def compile_sys_queries(
         ),
         source=edgeql.Source.from_string(report_configs_query),
     ).units
-    assert len(units) == 1 and len(units[0].sql) == 1
+    assert len(units) == 1
 
     report_configs_typedesc_2_0 = units[0].out_type_id + units[0].out_type_data
-    queries['report_configs'] = units[0].sql[0].decode()
+    queries['report_configs'] = units[0].sql.decode()
 
     units = edbcompiler.compile(
         ctx=edbcompiler.new_compiler_context(
@@ -2128,7 +2128,7 @@ def compile_sys_queries(
         ),
         source=edgeql.Source.from_string(report_configs_query),
     ).units
-    assert len(units) == 1 and len(units[0].sql) == 1
+    assert len(units) == 1
     report_configs_typedesc_1_0 = units[0].out_type_id + units[0].out_type_data
 
     return (