Add ADMINISTER command to vacuum.

Calling `administer vacuum(SomeType, OtherType.ptr, full := true)` will
vacuum the table for `SomeType` and all its descendants, as well as the
link table associated with the `ptr` (assuming that `ptr` is multi) and
all its descendants. The named only argument `full` being set to `true`
indicates the `FULL` option.

All of these arguments can be omitted. Not specifying any types or
pointers to vacuum will result in vacuuming everything that is
accessible to the user.

Co-authored-by: Michael J. Sullivan <[email protected]>

Fixes #6663

edb/server/compiler/compiler.py

@@ -1738,6 +1738,8 @@ def _compile_ql_administer(
         return ddl.administer_repair_schema(ctx, ql)
     elif ql.expr.func == 'reindex':
         return ddl.administer_reindex(ctx, ql)
+    elif ql.expr.func == 'vacuum':
+        return ddl.administer_vacuum(ctx, ql)
     else:
         raise errors.QueryError(
             'Unknown ADMINISTER function',

edb/server/compiler/ddl.py

@@ -1404,3 +1404,153 @@ def administer_reindex(
     return dbstate.MaintenanceQuery(
         sql=tuple(q.encode('utf-8') for q in commands)
     )
+
+
+def administer_vacuum(
+    ctx: compiler.CompileContext,
+    ql: qlast.AdministerStmt,
+) -> dbstate.BaseQuery:
+    from edb.ir import ast as irast
+    from edb.ir import typeutils as irtypeutils
+    from edb.schema import objtypes as s_objtypes
+    from edb.schema import pointers as s_pointers
+
+    # check that the kwargs are valid
+    kwargs: Dict[str, str] = {}
+    for name, val in ql.expr.kwargs.items():
+        if name != 'full':
+            raise errors.QueryError(
+                f'unrecognized keyword argument {name!r} for vacuum()',
+                context=val.context,
+            )
+        elif not isinstance(val, qlast.BooleanConstant):
+            raise errors.QueryError(
+                f'argument {name!r} for vacuum() must be a boolean literal',
+                context=val.context,
+            )
+        kwargs[name] = val.value
+
+    # Next go over the args (if any) and convert paths to tables/columns
+    args: List[Tuple[irast.Pointer | None, s_objtypes.ObjectType]] = []
+    current_tx = ctx.state.current_tx()
+    schema = current_tx.get_schema(ctx.compiler_state.std_schema)
+    modaliases = current_tx.get_modaliases()
+
+    for arg in ql.expr.args:
+        match arg:
+            case qlast.Path(
+                steps=[qlast.ObjectRef()],
+                partial=False,
+            ):
+                ptr = False
+            case qlast.Path(
+                steps=[qlast.ObjectRef(), qlast.Ptr()],
+                partial=False,
+            ):
+                ptr = True
+            case _:
+                raise errors.QueryError(
+                    'argument to vacuum() must be an object type '
+                    'or a link or property reference',
+                    context=arg.context,
+                )
+
+        ir: irast.Statement = qlcompiler.compile_ast_to_ir(
+            arg,
+            schema=schema,
+            options=qlcompiler.CompilerOptions(
+                modaliases=modaliases
+            ),
+        )
+        expr = ir.expr
+        if ptr:
+            if (
+                not expr.expr
+                or not isinstance(expr.expr, irast.SelectStmt)
+                or not expr.expr.result.rptr
+            ):
+                raise errors.QueryError(
+                    'invalid pointer argument to vacuum()',
+                    context=arg.context,
+                )
+            rptr = expr.expr.result.rptr
+            source = rptr.source
+        else:
+            rptr = None
+            source = expr
+        schema, obj = irtypeutils.ir_typeref_to_type(schema, source.typeref)
+
+        if (
+            not isinstance(obj, s_objtypes.ObjectType)
+            or not obj.is_material_object_type(schema)
+        ):
+            raise errors.QueryError(
+                'argument to vacuum() must be an object type '
+                'or a link or property reference',
+                context=arg.context,
+            )
+        args.append((rptr, obj))
+
+    tables: set[s_pointers.Pointer | s_objtypes.ObjectType] = set()
+
+    for arg, (rptr, obj) in zip(ql.expr.args, args):
+        if not rptr:
+            # On a type, we just vacuum the type and its descendants
+            tables.update({obj} | {
+                desc for desc in obj.descendants(schema)
+                if desc.is_material_object_type(schema)
+            })
+        else:
+            # On a pointer, we must go over the pointer and its descendants
+            # so that we may retrieve any link talbes if necessary.
+            if not isinstance(rptr.ptrref, irast.PointerRef):
+                raise errors.QueryError(
+                    'invalid pointer argument to vacuum()',
+                    context=arg.context,
+                )
+            schema, ptrcls = irtypeutils.ptrcls_from_ptrref(
+                rptr.ptrref, schema=schema)
+
+            card = ptrcls.get_cardinality(schema)
+            if not (
+                card.is_multi() or ptrcls.has_user_defined_properties(schema)
+            ):
+                vn = ptrcls.get_verbosename(schema, with_parent=True)
+                if ptrcls.is_property(schema):
+                    raise errors.QueryError(
+                        f'{vn} is not a valid argument to vacuum() '
+                        f'because it is not a multi property',
+                        context=arg.context,
+                    )
+                else:
+                    raise errors.QueryError(
+                        f'{vn} is not a valid argument to vacuum() '
+                        f'because it is neither a multi link nor '
+                        f'does it have link properties',
+                        context=arg.context,
+                    )
+
+            ptrclses = {ptrcls} | {
+                desc for desc in ptrcls.descendants(schema)
+                if isinstance(
+                    (src := desc.get_source(schema)), s_objtypes.ObjectType)
+                and src.is_material_object_type(schema)
+            }
+            tables.update(ptrclses)
+
+    tables_and_columns = [
+        pg_common.get_backend_name(schema, table)
+        for table in tables
+    ]
+
+    if kwargs.get('full', '').lower() == 'true':
+        options = 'FULL'
+    else:
+        options = ''
+
+    command = f'VACUUM {options} ' + ', '.join(tables_and_columns)
+
+    return dbstate.MaintenanceQuery(
+        sql=(command.encode('utf-8'),),
+        is_transactional=False,
+    )