feat(spark-3.5): [stacked] Add VACUUM SQL extension for Iceberg table maintenance

integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java

@@ -0,0 +1,159 @@
+package com.linkedin.openhouse.spark.statementtest;
+
+import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseParseException;
+import java.nio.file.Files;
+import lombok.SneakyThrows;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class VacuumStatementTest {
+
+  private static SparkSession spark = null;
+
+  private long snapshotCount(String table) {
+    return spark.sql("SELECT * FROM " + table + ".snapshots").count();
+  }
+
+  private long rowCount(String table) {
+    return spark.sql("SELECT * FROM " + table).count();
+  }
+
+  @Test
+  public void testVacuumExpiresSnapshots() {
+    // Three inserts create three snapshots.
+    Assertions.assertEquals(3, snapshotCount("openhouse.db.table"));
+
+    // RETAIN 0 HOURS expires everything but the current snapshot; the table stays readable.
+    spark.sql("VACUUM openhouse.db.table RETAIN 0 HOURS").collect();
+
+    Assertions.assertEquals(1, snapshotCount("openhouse.db.table"));
+    Assertions.assertEquals(3, rowCount("openhouse.db.table"));
+  }
+
+  @Test
+  public void testVacuumWithDefaultRetentionSucceeds() {
+    // No RETAIN: each procedure applies its own default retention. Table remains readable.
+    spark.sql("VACUUM openhouse.db.table").collect();
+    Assertions.assertEquals(3, rowCount("openhouse.db.table"));
+  }
+
+  @Test
+  public void testVacuumHonorsHistoryPolicyVersions() {
+    // The OpenHouse history policy (the `policies` property) sets versions=2. With no RETAIN,
+    // VACUUM must honor that via retain_last, keeping exactly the last 2 of the 3 snapshots.
+    spark
+        .sql(
+            "ALTER TABLE openhouse.db.table SET TBLPROPERTIES ("
+                + "'policies' = '{\"history\":{\"maxAge\":0,\"granularity\":\"DAY\",\"versions\":2}}')")
+        .show();
+    Assertions.assertEquals(3, snapshotCount("openhouse.db.table"));
+
+    spark.sql("VACUUM openhouse.db.table").collect();
+
+    Assertions.assertEquals(2, snapshotCount("openhouse.db.table"));
+    Assertions.assertEquals(3, rowCount("openhouse.db.table"));
+  }
+
+  @Test
+  public void testVacuumRemoveOrphanFilesPreservesLiveData() {
+    // A 24-hour window is safely above Iceberg's orphan-file removal floor and must not delete any
+    // file the table references, so all rows survive.
+    spark.sql("VACUUM openhouse.db.table REMOVE ORPHAN FILES RETAIN 24 HOURS").collect();
+    Assertions.assertEquals(3, rowCount("openhouse.db.table"));
+  }
+
+  @Test
+  public void testVacuumLowerCase() {
+    spark.sql("vacuum openhouse.db.table retain 0 hours").collect();
+    Assertions.assertEquals(1, snapshotCount("openhouse.db.table"));
+  }
+
+  @Test
+  public void testVacuumNonOpenhouseTableThrows() {
+    Assertions.assertThrows(
+        Exception.class, () -> spark.sql("VACUUM openhouse.db.not_openhouse").collect());
+  }
+
+  @Test
+  public void testVacuumNotEnabledThrows() {
+    // VACUUM is Alpha and opt-in: an OpenHouse table that has not set openhouse.vacuum.enabled=true
+    // is rejected.
+    Assertions.assertThrows(
+        UnsupportedOperationException.class,
+        () -> spark.sql("VACUUM openhouse.db.not_enabled").collect());
+  }
+
+  @Test
+  public void testVacuumInvalidSyntaxThrows() {
+    Assertions.assertThrows(
+        OpenhouseParseException.class,
+        () -> spark.sql("VACUUM openhouse.db.table RETAIN 5 DAYS").collect());
+  }
+
+  @SneakyThrows
+  @BeforeAll
+  public void setupSpark() {
+    Path unittest = new Path(Files.createTempDirectory("unittest").toString());
+    spark =
+        SparkSession.builder()
+            .master("local[2]")
+            .config(
+                "spark.sql.extensions",
+                ("org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,"
+                    + "com.linkedin.openhouse.spark.extensions.OpenhouseSparkSessionExtensions"))
+            .config("spark.sql.catalog.openhouse", "org.apache.iceberg.spark.SparkCatalog")
+            .config("spark.sql.catalog.openhouse.type", "hadoop")
+            .config("spark.sql.catalog.openhouse.warehouse", unittest.toString())
+            .getOrCreate();
+  }
+
+  @BeforeEach
+  public void setup() {
+    spark
+        .sql(
+            "CREATE TABLE openhouse.db.table (id bigint, data string, `openhouse.tableId` string) USING iceberg")
+        .show();
+    spark
+        .sql(
+            "ALTER TABLE openhouse.db.table SET TBLPROPERTIES ("
+                + "'openhouse.tableId' = 'tableid', 'openhouse.vacuum.enabled' = 'true')")
+        .show();
+    spark.sql("INSERT INTO openhouse.db.table VALUES (1, 'a', 'tableid')").show();
+    spark.sql("INSERT INTO openhouse.db.table VALUES (2, 'b', 'tableid')").show();
+    spark.sql("INSERT INTO openhouse.db.table VALUES (3, 'c', 'tableid')").show();
+
+    // OpenHouse table that has NOT opted into the Alpha VACUUM feature.
+    spark
+        .sql(
+            "CREATE TABLE openhouse.db.not_enabled (id bigint, data string, `openhouse.tableId` string) USING iceberg")
+        .show();
+    spark
+        .sql(
+            "ALTER TABLE openhouse.db.not_enabled SET TBLPROPERTIES ('openhouse.tableId' = 'tableid')")
+        .show();
+
+    spark
+        .sql("CREATE TABLE openhouse.db.not_openhouse (id bigint, data string) USING iceberg")
+        .show();
+  }
+
+  @AfterEach
+  public void tearDown() {
+    spark.sql("DROP TABLE IF EXISTS openhouse.db.table").show();
+    spark.sql("DROP TABLE IF EXISTS openhouse.db.not_enabled").show();
+    spark.sql("DROP TABLE IF EXISTS openhouse.db.not_openhouse").show();
+  }
+
+  @AfterAll
+  public void tearDownSpark() {
+    spark.close();
+  }
+}

integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md

@@ -0,0 +1,82 @@
+# VACUUM
+
+**Status: Alpha.** `VACUUM` is opt-in per table. Please See [Enabling VACUUM](#enabling-vacuum)).
+
+`VACUUM` is an OpenHouse Spark SQL extension that reclaims storage for an OpenHouse
+Iceberg table by removing files that are no longer needed. It is thin, ergonomic sugar
+over the underlying Iceberg maintenance stored procedures.
+
+## Syntax
+
+```sql
+VACUUM <table> [REMOVE ORPHAN FILES] [RETAIN <n> HOURS]
+```
+
+- `<table>` — an OpenHouse table identifier (e.g. `openhouse.db.table`).
+- `REMOVE ORPHAN FILES` — *(optional)* also delete orphaned files (see below). Off by default.
+- `RETAIN <n> HOURS` — *(optional)* retention window in whole hours. When omitted, each
+  underlying operation uses its own default retention.
+
+## Behavior
+
+Running `VACUUM` reclaims files beyond the retention window that are no longer referenced
+by the current version of the table.
+
+1. **Orphan-file deletion** (`REMOVE ORPHAN FILES`, opt-in). Orphan files are files under the table's location that are not referenced by any table metadata typically left behind by failed or aborted writes. This step only deletes files from storage; it does not commit table metadata, so it succeeds even when the table is out of write quota. 
+
+2. **Snapshot expiration** always runs. It removes snapshots older than the retention window and deletes the data, delete, manifest, and manifest-list files that those expired snapshots exclusively referenced. This command adds a commit and can conflict with in-flight transactions.
+
+3. **Retention** (`RETAIN <n> HOURS`) bounds both operations: only files older than `now - n hours` are eligible. The cutoff is resolved to a concrete timestamp in the session time zone at execution time. When `RETAIN` is omitted, snapshot expiration falls back to the table's configured snapshot-age retention and orphan-file deletion's configured default.
+
+
+## Enabling VACUUM
+
+`VACUUM` is Alpha and must be enabled on each table before use:
+
+```sql
+ALTER TABLE openhouse.db.table
+  SET TBLPROPERTIES ('openhouse.vacuum.enabled' = 'true');
+```
+
+| Property                   | Value    | Meaning                                     |
+| -------------------------- | -------- | ------------------------------------------- |
+| `openhouse.vacuum.enabled` | `'true'` | Opt this table into the Alpha `VACUUM` command. |
+
+Any other value (or the property being absent) leaves `VACUUM` disabled for the table, and
+running the command throws an `UnsupportedOperationException` that explains how to enable it.
+`VACUUM` is only supported on OpenHouse tables; running it on a non-OpenHouse table also
+throws.
+
+## Examples
+
+Enable the feature, then expire snapshots older than 24 hours:
+
+```sql
+ALTER TABLE openhouse.db.table
+  SET TBLPROPERTIES ('openhouse.vacuum.enabled' = 'true');
+
+VACUUM openhouse.db.table RETAIN 24 HOURS;
+```
+
+Expire snapshots using the table's default retention:
+
+```sql
+VACUUM openhouse.db.table;
+```
+
+Also remove orphaned files, retaining anything from the last 168 hours (7 days):
+
+```sql
+VACUUM openhouse.db.table REMOVE ORPHAN FILES RETAIN 168 HOURS;
+```
+
+## Notes and caveats
+
+- **`REMOVE ORPHAN FILES` is expensive.** It performs a recursive listing of the table's
+  location to find unreferenced files. On tables with very large file counts this can be
+  slow and memory-intensive, and may require a larger Spark driver to avoid running out of
+  memory.
+- **Low Retention causes in-flight operations to fail** A query sees the same snapshot of the table they start with, and expiring a snapshot that is in-use will cause transactions to fail. Deleting orphans of in-flight transactions can cause failure.  24 hours is the suggested minimum but can be lowered to mitigate emergency scenarios. 
+- **Snapshot expiration requires write quota**; orphan-file deletion does not. This is why
+  orphan-file deletion runs first — on a table that is out of quota, orphan cleanup still
+  proceeds even though expiration cannot commit.

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4

@@ -32,6 +32,7 @@ statement
   | GRANT privilege ON grantableResource TO principal                                                  #grantStatement
   | REVOKE privilege ON grantableResource FROM principal                                               #revokeStatement
   | SHOW GRANTS ON grantableResource                                                                   #showGrantsStatement
+  | VACUUM multipartIdentifier (REMOVE ORPHAN FILES)? (RETAIN POSITIVE_INTEGER HOURS)?                  #vacuumTable
   ;
 
 multipartIdentifier
@@ -69,6 +70,7 @@ quotedIdentifier
 nonReserved
     : ALTER | TABLE | SET | POLICY | RETENTION | SHARING | REPLICATION | HISTORY
     | GRANT | REVOKE | ON | TO | SHOW | GRANTS | PATTERN | WHERE | COLUMN
+    | VACUUM | REMOVE | ORPHAN | FILES | RETAIN | HOURS
     ;
 
 sharingPolicy
@@ -205,6 +207,12 @@ TAG: 'TAG';
 NONE: 'NONE';
 VERSIONS: 'VERSIONS';
 MAX_AGE: 'MAX_AGE';
+VACUUM: 'VACUUM';
+REMOVE: 'REMOVE';
+ORPHAN: 'ORPHAN';
+FILES: 'FILES';
+RETAIN: 'RETAIN';
+HOURS: 'HOURS';
 
 POSITIVE_INTEGER
     : DIGIT+

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala

@@ -72,7 +72,8 @@ class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends Pars
         normalized.contains("set tag"))) ||
       normalized.startsWith("grant") ||
       normalized.startsWith("revoke") ||
-      normalized.startsWith("show grants")
+      normalized.startsWith("show grants") ||
+      normalized.startsWith("vacuum")
 
   }
 

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala

@@ -2,7 +2,7 @@ package com.linkedin.openhouse.spark.sql.catalyst.parser.extensions
 
 import com.linkedin.openhouse.spark.sql.catalyst.enums.GrantableResourceTypes
 import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseSqlExtensionsParser._
-import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy}
+import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy, VacuumTable}
 import com.linkedin.openhouse.spark.sql.catalyst.enums.GrantableResourceTypes.GrantableResourceType
 import com.linkedin.openhouse.gen.tables.client.model.TimePartitionSpec
 import org.antlr.v4.runtime.tree.ParseTree
@@ -197,6 +197,13 @@ class OpenhouseSqlExtensionsAstBuilder (delegate: ParserInterface) extends Openh
     ctx.POSITIVE_INTEGER().getText.toInt
   }
 
+  override def visitVacuumTable(ctx: VacuumTableContext): VacuumTable = {
+    val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier)
+    val removeOrphanFiles = ctx.REMOVE() != null
+    val retainHours = Option(ctx.POSITIVE_INTEGER()).map(_.getText.toInt)
+    VacuumTable(tableName, removeOrphanFiles, retainHours)
+  }
+
   private def toBuffer[T](list: java.util.List[T]) = list.asScala
   private def toSeq[T](list: java.util.List[T]) = toBuffer(list).toSeq
 

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/VacuumTable.scala

@@ -0,0 +1,9 @@
+package com.linkedin.openhouse.spark.sql.catalyst.plans.logical
+
+import org.apache.spark.sql.catalyst.plans.logical.LeafCommand
+
+case class VacuumTable(tableName: Seq[String], removeOrphanFiles: Boolean, retainHours: Option[Int]) extends LeafCommand {
+  override def simpleString(maxFields: Int): String = {
+    s"VacuumTable: ${tableName} removeOrphanFiles=${removeOrphanFiles} retainHours=${retainHours.getOrElse("default")}"
+  }
+}

integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala

@@ -1,6 +1,6 @@
 package com.linkedin.openhouse.spark.sql.execution.datasources.v2
 
-import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy}
+import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy, VacuumTable}
 import org.apache.iceberg.spark.{Spark3Util, SparkCatalog, SparkSessionCatalog}
 import org.apache.spark.sql.{SparkSession, Strategy}
 import org.apache.spark.sql.catalyst.expressions.PredicateHelper
@@ -32,6 +32,9 @@ case class OpenhouseDataSourceV2Strategy(spark: SparkSession) extends Strategy w
     case r @ ShowGrantsStatement(resourceType, CatalogAndIdentifierExtractor(catalog, ident)) =>
       ShowGrantsStatementExec(r.output, resourceType, catalog, ident) :: Nil
 
+    case VacuumTable(CatalogAndIdentifierExtractor(catalog, ident), removeOrphanFiles, retainHours) =>
+      VacuumTableExec(spark, catalog, ident, removeOrphanFiles, retainHours) :: Nil
+
     case _ => Nil
   }