From 40ad62b4751046818076dd3bf6d437b39f77aa6d Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Fri, 24 Jul 2026 09:24:00 -0700 Subject: [PATCH 1/5] [spark-3.5] Add VACUUM SQL extension for Iceberg table maintenance Adds a VACUUM command to the OpenHouse spark-3.5 SQL extensions: VACUUM [REMOVE ORPHAN FILES] [RETAIN n HOURS] Snapshot expiration always runs; REMOVE ORPHAN FILES opts into orphan-file deletion, run after expiration so it cleans against the settled live-file set. RETAIN n HOURS bounds both operations via the procedures' older_than argument (resolved to a literal timestamp in the session time zone, since procedure arguments must be foldable); when omitted, each procedure applies its own default retention. Implemented as thin sugar over the Iceberg stored-procedure CALL path: VacuumTableExec validates the target is an OpenHouse table (openhouse.tableId property) and issues CALL .system.expire_snapshots / remove_orphan_files via sparkSession.sql(...), so procedure resolution and argument binding reuse the existing CALL path. New non-reserved keywords VACUUM/REMOVE/ORPHAN/FILES/RETAIN/ HOURS; existing identifiers with those names still parse. Tested by VacuumStatementTest (real Iceberg, Hadoop catalog): RETAIN 0 HOURS collapses three snapshots to one with rows intact, REMOVE ORPHAN FILES RETAIN 24 HOURS preserves live data, default retention, lower-case, non-OpenHouse-table rejection, and invalid-syntax parse errors. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../statementtest/VacuumStatementTest.java | 120 ++++++++++++++++++ .../extensions/OpenhouseSqlExtensions.g4 | 8 ++ .../OpenhouseSparkSqlExtensionsParser.scala | 3 +- .../OpenhouseSqlExtensionsAstBuilder.scala | 9 +- .../catalyst/plans/logical/VacuumTable.scala | 9 ++ .../v2/OpenhouseDataSourceV2Strategy.scala | 5 +- .../datasources/v2/VacuumTableExec.scala | 71 +++++++++++ 7 files changed, 222 insertions(+), 3 deletions(-) create mode 100644 integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/VacuumTable.scala create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java new file mode 100644 index 000000000..50f9bfa23 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java @@ -0,0 +1,120 @@ +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 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 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')") + .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(); + + 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_openhouse").show(); + } + + @AfterAll + public void tearDownSpark() { + spark.close(); + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 index 95be037b5..9faffdde7 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 +++ b/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+ diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala index ee5cef038..070a7bbbc 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala +++ b/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") } diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala index 95319f917..13419e9e6 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala +++ b/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 diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/VacuumTable.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/VacuumTable.scala new file mode 100644 index 000000000..d679e78b2 --- /dev/null +++ b/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")}" + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala index 06d9494db..0a68e28d6 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala +++ b/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 } diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala new file mode 100644 index 000000000..022de1d56 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala @@ -0,0 +1,71 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.time.{Instant, ZoneId} +import java.time.format.DateTimeFormatter +import java.time.temporal.ChronoUnit + +import org.apache.iceberg.spark.source.SparkTable +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.Attribute +import org.apache.spark.sql.catalyst.util.quoteIfNeeded +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} +import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec + +/** + * Runs Iceberg table maintenance for the VACUUM command as thin sugar over the catalog's stored + * procedures. Snapshot expiration always runs; when `REMOVE ORPHAN FILES` is given, orphan-file + * deletion runs afterwards so it cleans up against the settled live-file set. A `RETAIN n HOURS` + * window bounds both operations via the procedures' `older_than` argument; when omitted, each + * procedure falls back to its own default retention. + */ +case class VacuumTableExec( + spark: SparkSession, + catalog: TableCatalog, + ident: Identifier, + removeOrphanFiles: Boolean, + retainHours: Option[Int]) extends LeafV2CommandExec { + + override lazy val output: Seq[Attribute] = Nil + + override protected def run(): Seq[InternalRow] = { + catalog.loadTable(ident) match { + case iceberg: SparkTable if iceberg.table().properties().containsKey("openhouse.tableId") => + val quotedCatalog = quoteIfNeeded(catalog.name()) + val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") + + // Procedure arguments must be foldable, so a RETAIN window is resolved here to a literal + // `older_than` timestamp (now - n hours) rather than a `current_timestamp()` expression. + // The literal is rendered in the session time zone because the CALL's `TIMESTAMP '...'` + // literal is parsed back in that same zone, so the round-trip preserves the intended + // instant. When RETAIN is omitted, `older_than` is left off so each procedure applies its + // own default retention. + val olderThanArg = retainHours.map { hours => + val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS") + .withZone(ZoneId.of(spark.sessionState.conf.sessionLocalTimeZone)) + val cutoff = formatter.format(Instant.now().minus(hours.toLong, ChronoUnit.HOURS)) + s", older_than => TIMESTAMP '$cutoff'" + }.getOrElse("") + + // Snapshot expiration always runs. + spark.sql( + s"CALL $quotedCatalog.system.expire_snapshots(table => '$tableArg'$olderThanArg)").collect() + + if (removeOrphanFiles) { + // Runs after expiration so it deletes against the settled live-file set. + spark.sql( + s"CALL $quotedCatalog.system.remove_orphan_files(table => '$tableArg'$olderThanArg)").collect() + } + + case table => + throw new UnsupportedOperationException(s"Cannot vacuum non-Openhouse table: $table") + } + + Nil + } + + override def simpleString(maxFields: Int): String = { + s"VacuumTableExec: ${catalog} ${ident} removeOrphanFiles=${removeOrphanFiles} " + + s"retainHours=${retainHours.getOrElse("default")}" + } +} From 84bb609706d35a8aef2b260bdb0d5f4407cd1786 Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Fri, 24 Jul 2026 12:36:12 -0700 Subject: [PATCH 2/5] [spark-3.5] VACUUM: run REMOVE ORPHAN FILES before snapshot expiration Snapshot expiration commits table metadata and therefore cannot run on a table that is out of quota, whereas orphan-file deletion only removes unreferenced files from storage and always can. Running orphan removal first ensures it still executes in that case, and scanning against the pre-expiration referenced-file set means it can never delete a file a still-live snapshot references. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../datasources/v2/VacuumTableExec.scala | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala index 022de1d56..64d93f495 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala @@ -14,8 +14,9 @@ import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec /** * Runs Iceberg table maintenance for the VACUUM command as thin sugar over the catalog's stored - * procedures. Snapshot expiration always runs; when `REMOVE ORPHAN FILES` is given, orphan-file - * deletion runs afterwards so it cleans up against the settled live-file set. A `RETAIN n HOURS` + * procedures. When `REMOVE ORPHAN FILES` is given, orphan-file deletion runs first (it only removes + * unreferenced files from storage, so it works even when the table is out of quota, unlike snapshot + * expiration which commits metadata); snapshot expiration always runs afterwards. A `RETAIN n HOURS` * window bounds both operations via the procedures' `older_than` argument; when omitted, each * procedure falls back to its own default retention. */ @@ -47,16 +48,20 @@ case class VacuumTableExec( s", older_than => TIMESTAMP '$cutoff'" }.getOrElse("") - // Snapshot expiration always runs. - spark.sql( - s"CALL $quotedCatalog.system.expire_snapshots(table => '$tableArg'$olderThanArg)").collect() - if (removeOrphanFiles) { - // Runs after expiration so it deletes against the settled live-file set. + // Runs BEFORE expiration. Snapshot expiration commits table metadata, so it cannot run on + // a table that is out of quota; orphan-file deletion only removes unreferenced files from + // storage and always can, so doing it first ensures it still runs in that case. Running + // first also means it scans against the pre-expiration referenced-file set, so it can + // never delete a file that a still-live snapshot references. spark.sql( s"CALL $quotedCatalog.system.remove_orphan_files(table => '$tableArg'$olderThanArg)").collect() } + // Snapshot expiration always runs. + spark.sql( + s"CALL $quotedCatalog.system.expire_snapshots(table => '$tableArg'$olderThanArg)").collect() + case table => throw new UnsupportedOperationException(s"Cannot vacuum non-Openhouse table: $table") } From f3d7f07d3ba44a38c1ddd883b5f009f3911bc784 Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Fri, 24 Jul 2026 12:45:43 -0700 Subject: [PATCH 3/5] [spark-3.5] VACUUM: gate behind Alpha opt-in property + add public docs VACUUM is an Alpha feature and is now opt-in per table. A table must set 'openhouse.vacuum.enabled' = 'true'; otherwise VACUUM throws UnsupportedOperationException explaining how to enable it. - VacuumTableExec: add the openhouse.vacuum.enabled gate + ENABLED_PROP constant. - VacuumStatementTest: enable the property on the vacuumed table and add testVacuumNotEnabledThrows for an OpenHouse table that has not opted in. - Add public documentation (docs/VACUUM.md): behavior, syntax, the Alpha opt-in knob, merge-on-read handling, examples, and caveats. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../statementtest/VacuumStatementTest.java | 24 +++- .../openhouse-spark-runtime/docs/VACUUM.md | 115 ++++++++++++++++++ .../datasources/v2/VacuumTableExec.scala | 20 ++- 3 files changed, 157 insertions(+), 2 deletions(-) create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java index 50f9bfa23..5c6e8a23c 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java @@ -65,6 +65,15 @@ public void testVacuumNonOpenhouseTableThrows() { 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( @@ -96,12 +105,24 @@ public void setup() { "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')") + .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(); @@ -110,6 +131,7 @@ public void setup() { @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(); } diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md new file mode 100644 index 000000000..154551705 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md @@ -0,0 +1,115 @@ +# VACUUM + +> **Status: Alpha.** `VACUUM` is opt-in per table. A table must explicitly enable it +> (see [Enabling VACUUM](#enabling-vacuum)); running `VACUUM` on a table that has not +> enabled it fails with an `UnsupportedOperationException`. Behavior may change in future +> releases. + +`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
[REMOVE ORPHAN FILES] [RETAIN HOURS] +``` + +- `
` — an OpenHouse table identifier (e.g. `openhouse.db.table`). +- `REMOVE ORPHAN FILES` — *(optional)* also delete orphaned files (see below). Off by default. +- `RETAIN HOURS` — *(optional)* retention window in whole hours. When omitted, each + underlying operation uses its own default retention. + +The keywords `VACUUM`, `REMOVE`, `ORPHAN`, `FILES`, `RETAIN`, and `HOURS` are +non-reserved, so existing identifiers with those names continue to parse. + +## 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) runs **first**. 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. Running it before expiration also means it evaluates against the + pre-expiration set of referenced files, so it can never delete a file that a still-live + snapshot references. + +2. **Snapshot expiration** always runs, **after** orphan-file deletion. It removes + snapshots older than the retention window and deletes the data, delete, manifest, and + manifest-list files that those expired snapshots exclusively referenced. Expiration + commits table metadata. + +3. **Retention** (`RETAIN 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 + falls back to Iceberg's safe default. + +### Merge-on-read (MoR) delete files + +`VACUUM` handles merge-on-read delete files (position and equality deletes) the same way +it handles data files: + +- Delete files referenced only by expired snapshots are removed by **snapshot expiration**. +- Orphaned delete files are removed by **`REMOVE ORPHAN FILES`**. + +Dangling delete files that are still referenced by the current snapshot but no longer +apply to any live data (for example, because the data files they targeted were compacted +away) are **not** removed by `VACUUM`. Clearing those is a data-rewrite/compaction concern +handled by `OPTIMIZE`, not by `VACUUM`. + +## 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. Prefer a generous `RETAIN` window and run it less frequently. +- **Choose the retention window carefully.** Files newer than the retention window are + never deleted. Time travel and snapshot rollback are only possible for snapshots that + have not been expired, so retain enough history for your recovery needs. +- **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. diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala index 64d93f495..e3a39da5c 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala @@ -14,7 +14,10 @@ import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec /** * Runs Iceberg table maintenance for the VACUUM command as thin sugar over the catalog's stored - * procedures. When `REMOVE ORPHAN FILES` is given, orphan-file deletion runs first (it only removes + * procedures. VACUUM is an '''Alpha''' feature and is opt-in per table via the + * `openhouse.vacuum.enabled` property; it is rejected on tables that have not enabled it. + * + * When `REMOVE ORPHAN FILES` is given, orphan-file deletion runs first (it only removes * unreferenced files from storage, so it works even when the table is out of quota, unlike snapshot * expiration which commits metadata); snapshot expiration always runs afterwards. A `RETAIN n HOURS` * window bounds both operations via the procedures' `older_than` argument; when omitted, each @@ -32,6 +35,13 @@ case class VacuumTableExec( override protected def run(): Seq[InternalRow] = { catalog.loadTable(ident) match { case iceberg: SparkTable if iceberg.table().properties().containsKey("openhouse.tableId") => + // VACUUM is an Alpha feature and is opt-in per table: a table must explicitly enable it via + // the `openhouse.vacuum.enabled` property, otherwise the command is rejected. + if (!"true".equalsIgnoreCase(iceberg.table().properties().get(VacuumTableExec.ENABLED_PROP))) { + throw new UnsupportedOperationException( + s"VACUUM is an Alpha feature and must be enabled on the table before use. Enable it " + + s"with: ALTER TABLE
SET TBLPROPERTIES ('${VacuumTableExec.ENABLED_PROP}' = 'true').") + } val quotedCatalog = quoteIfNeeded(catalog.name()) val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") @@ -74,3 +84,11 @@ case class VacuumTableExec( s"retainHours=${retainHours.getOrElse("default")}" } } + +object VacuumTableExec { + /** + * Table property that opts a table into the Alpha VACUUM command. VACUUM throws + * [[UnsupportedOperationException]] on tables where this is not set to `true`. + */ + val ENABLED_PROP = "openhouse.vacuum.enabled" +} From 10a91b2899c506094ae177c192e3c7caab9cdfd5 Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Fri, 24 Jul 2026 13:17:10 -0700 Subject: [PATCH 4/5] Apply suggestions from code review Co-authored-by: Mike Kuchenbecker --- .../openhouse-spark-runtime/docs/VACUUM.md | 45 +++---------------- 1 file changed, 6 insertions(+), 39 deletions(-) diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md index 154551705..9cc8d204a 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md @@ -1,9 +1,6 @@ # VACUUM -> **Status: Alpha.** `VACUUM` is opt-in per table. A table must explicitly enable it -> (see [Enabling VACUUM](#enabling-vacuum)); running `VACUUM` on a table that has not -> enabled it fails with an `UnsupportedOperationException`. Behavior may change in future -> releases. +**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 @@ -20,45 +17,17 @@ VACUUM
[REMOVE ORPHAN FILES] [RETAIN HOURS] - `RETAIN HOURS` — *(optional)* retention window in whole hours. When omitted, each underlying operation uses its own default retention. -The keywords `VACUUM`, `REMOVE`, `ORPHAN`, `FILES`, `RETAIN`, and `HOURS` are -non-reserved, so existing identifiers with those names continue to parse. - ## 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) runs **first**. 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. Running it before expiration also means it evaluates against the - pre-expiration set of referenced files, so it can never delete a file that a still-live - snapshot references. - -2. **Snapshot expiration** always runs, **after** orphan-file deletion. It removes - snapshots older than the retention window and deletes the data, delete, manifest, and - manifest-list files that those expired snapshots exclusively referenced. Expiration - commits table metadata. - -3. **Retention** (`RETAIN 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 - falls back to Iceberg's safe default. - -### Merge-on-read (MoR) delete files +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. -`VACUUM` handles merge-on-read delete files (position and equality deletes) the same way -it handles data files: +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. -- Delete files referenced only by expired snapshots are removed by **snapshot expiration**. -- Orphaned delete files are removed by **`REMOVE ORPHAN FILES`**. +3. **Retention** (`RETAIN 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. -Dangling delete files that are still referenced by the current snapshot but no longer -apply to any live data (for example, because the data files they targeted were compacted -away) are **not** removed by `VACUUM`. Clearing those is a data-rewrite/compaction concern -handled by `OPTIMIZE`, not by `VACUUM`. ## Enabling VACUUM @@ -106,10 +75,8 @@ VACUUM openhouse.db.table REMOVE ORPHAN FILES RETAIN 168 HOURS; - **`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. Prefer a generous `RETAIN` window and run it less frequently. -- **Choose the retention window carefully.** Files newer than the retention window are - never deleted. Time travel and snapshot rollback are only possible for snapshots that - have not been expired, so retain enough history for your recovery needs. + 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. From 153fc943ccee1339bae96a8fd443a1f7635bd706 Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Mon, 27 Jul 2026 09:35:53 -0700 Subject: [PATCH 5/5] [spark-3.5] VACUUM: wire retention cutoffs to OpenHouse table properties When RETAIN is omitted, VACUUM previously fell back to the Iceberg procedure defaults (5-day snapshot expiration, 3-day orphan-file deletion). Wire the cutoffs to the correct OpenHouse table properties instead, with no change to what the command does (it still cleans reclaimed files via the CALL procedures' default behavior): - Snapshot expiration honors the history policy in the 'policies' property: older_than = now - maxAge x granularity, plus retain_last => versions when set (defaults maxAge=3, granularity=DAY, versions=0, matching OpenHouse). - Orphan-file deletion honors 'ofd.one_day_ttl.enabled' (1 day when set, else the 3-day default). An explicit RETAIN n HOURS still overrides both. Pure helpers (ofdRetainDays, parseHistoryRetention, granularityToChrono) are unit-tested in VacuumTableExecTest; VacuumStatementTest.testVacuumHonorsHistoryPolicyVersions proves end-to-end that a history policy of versions=2 keeps exactly 2 snapshots. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../statementtest/VacuumStatementTest.java | 17 +++ .../datasources/v2/VacuumTableExec.scala | 120 +++++++++++++++--- .../datasources/v2/VacuumTableExecTest.scala | 60 +++++++++ 3 files changed, 181 insertions(+), 16 deletions(-) create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExecTest.scala diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java index 5c6e8a23c..a2f2b4236 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java @@ -45,6 +45,23 @@ public void testVacuumWithDefaultRetentionSucceeds() { 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 diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala index e3a39da5c..3e5e5321b 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExec.scala @@ -4,6 +4,7 @@ import java.time.{Instant, ZoneId} import java.time.format.DateTimeFormatter import java.time.temporal.ChronoUnit +import com.fasterxml.jackson.databind.ObjectMapper import org.apache.iceberg.spark.source.SparkTable import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.InternalRow @@ -19,9 +20,14 @@ import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec * * When `REMOVE ORPHAN FILES` is given, orphan-file deletion runs first (it only removes * unreferenced files from storage, so it works even when the table is out of quota, unlike snapshot - * expiration which commits metadata); snapshot expiration always runs afterwards. A `RETAIN n HOURS` - * window bounds both operations via the procedures' `older_than` argument; when omitted, each - * procedure falls back to its own default retention. + * expiration which commits metadata); snapshot expiration always runs afterwards. Both procedures + * run with their default file-cleaning behavior, so VACUUM deletes the reclaimed files. + * + * The retention window comes from `RETAIN n HOURS` when given. When it is omitted, the cutoffs are + * derived from the table's OpenHouse properties rather than the Iceberg procedure defaults: snapshot + * expiration uses the history policy (the `policies` property's `history` block -- maxAge/granularity + * and, when set, versions -> retain_last), and orphan-file deletion uses `ofd.one_day_ttl.enabled` + * (1 day when enabled, otherwise the 3-day default). */ case class VacuumTableExec( spark: SparkSession, @@ -44,19 +50,18 @@ case class VacuumTableExec( } val quotedCatalog = quoteIfNeeded(catalog.name()) val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") + val props = iceberg.table().properties() + val now = Instant.now() - // Procedure arguments must be foldable, so a RETAIN window is resolved here to a literal - // `older_than` timestamp (now - n hours) rather than a `current_timestamp()` expression. - // The literal is rendered in the session time zone because the CALL's `TIMESTAMP '...'` - // literal is parsed back in that same zone, so the round-trip preserves the intended - // instant. When RETAIN is omitted, `older_than` is left off so each procedure applies its - // own default retention. - val olderThanArg = retainHours.map { hours => + // Procedure arguments must be foldable, so a retention window is resolved here to a literal + // `older_than` timestamp rather than a `current_timestamp()` expression. The literal is + // rendered in the session time zone because the CALL's `TIMESTAMP '...'` literal is parsed + // back in that same zone, so the round-trip preserves the intended instant. + def olderThanClause(instant: Instant): String = { val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS") .withZone(ZoneId.of(spark.sessionState.conf.sessionLocalTimeZone)) - val cutoff = formatter.format(Instant.now().minus(hours.toLong, ChronoUnit.HOURS)) - s", older_than => TIMESTAMP '$cutoff'" - }.getOrElse("") + s", older_than => TIMESTAMP '${formatter.format(instant)}'" + } if (removeOrphanFiles) { // Runs BEFORE expiration. Snapshot expiration commits table metadata, so it cannot run on @@ -64,13 +69,33 @@ case class VacuumTableExec( // storage and always can, so doing it first ensures it still runs in that case. Running // first also means it scans against the pre-expiration referenced-file set, so it can // never delete a file that a still-live snapshot references. + // + // Cutoff: an explicit RETAIN overrides; otherwise honor the OpenHouse OFD TTL property + // (1 day when ofd.one_day_ttl.enabled=true, else the 3-day default). + val ofdOlderThan = retainHours match { + case Some(hours) => olderThanClause(now.minus(hours.toLong, ChronoUnit.HOURS)) + case None => + olderThanClause(now.minus(VacuumTableExec.ofdRetainDays(props), ChronoUnit.DAYS)) + } spark.sql( - s"CALL $quotedCatalog.system.remove_orphan_files(table => '$tableArg'$olderThanArg)").collect() + s"CALL $quotedCatalog.system.remove_orphan_files(table => '$tableArg'$ofdOlderThan)").collect() } - // Snapshot expiration always runs. + // Snapshot expiration always runs. Cutoff: an explicit RETAIN overrides; otherwise honor the + // OpenHouse history policy (maxAge x granularity for the age cutoff, and versions -> + // retain_last when set). The procedure cleans reclaimed files by default. + val expireArgs = retainHours match { + case Some(hours) => olderThanClause(now.minus(hours.toLong, ChronoUnit.HOURS)) + case None => + val history = VacuumTableExec.parseHistoryRetention(props.get(VacuumTableExec.POLICIES_PROP)) + val ageMillis = VacuumTableExec.granularityToChrono(history.granularity) + .getDuration.multipliedBy(history.maxAge.toLong).toMillis + val older = olderThanClause(now.minusMillis(ageMillis)) + val retainLast = if (history.versions > 0) s", retain_last => ${history.versions}" else "" + older + retainLast + } spark.sql( - s"CALL $quotedCatalog.system.expire_snapshots(table => '$tableArg'$olderThanArg)").collect() + s"CALL $quotedCatalog.system.expire_snapshots(table => '$tableArg'$expireArgs)").collect() case table => throw new UnsupportedOperationException(s"Cannot vacuum non-Openhouse table: $table") @@ -91,4 +116,67 @@ object VacuumTableExec { * [[UnsupportedOperationException]] on tables where this is not set to `true`. */ val ENABLED_PROP = "openhouse.vacuum.enabled" + + /** OpenHouse table property holding the policies JSON (retention, history, etc.). */ + val POLICIES_PROP = "policies" + + /** OpenHouse property that opts a table into a 1-day orphan-file-deletion TTL. */ + val OFD_ONE_DAY_TTL_ENABLED_PROP = "ofd.one_day_ttl.enabled" + + /** Default orphan-file-deletion TTL in days when the 1-day opt-in is not set. */ + val DEFAULT_OFD_TTL_DAYS = 3L + + // History-policy defaults, matching OpenHouse's server-side defaults. + val DEFAULT_HISTORY_MAX_AGE = 3 + val DEFAULT_HISTORY_GRANULARITY = "DAY" + val DEFAULT_HISTORY_VERSIONS = 0 + + private val mapper = new ObjectMapper() + + /** The snapshot-retention values honored by VACUUM, read from the history policy. */ + case class HistoryRetention(maxAge: Int, granularity: String, versions: Int) + + /** + * Orphan-file-deletion retention in days: 1 when `ofd.one_day_ttl.enabled` is `true`, otherwise + * the 3-day default. Exposed for testing. + */ + def ofdRetainDays(props: java.util.Map[String, String]): Long = + if ("true".equalsIgnoreCase(props.get(OFD_ONE_DAY_TTL_ENABLED_PROP))) 1L else DEFAULT_OFD_TTL_DAYS + + /** + * Parse the history block of the OpenHouse `policies` JSON into the snapshot-retention values. + * Absent property, absent history block, or unparseable JSON all fall back to the OpenHouse + * defaults (maxAge=3, granularity=DAY, versions=0). Exposed for testing. + */ + def parseHistoryRetention(policiesJson: String): HistoryRetention = { + val default = HistoryRetention( + DEFAULT_HISTORY_MAX_AGE, DEFAULT_HISTORY_GRANULARITY, DEFAULT_HISTORY_VERSIONS) + if (policiesJson == null || policiesJson.trim.isEmpty) return default + try { + val history = mapper.readTree(policiesJson).path("history") + if (history.isMissingNode || history.isNull) return default + HistoryRetention( + maxAge = if (history.hasNonNull("maxAge")) history.get("maxAge").asInt(DEFAULT_HISTORY_MAX_AGE) + else DEFAULT_HISTORY_MAX_AGE, + granularity = if (history.hasNonNull("granularity")) history.get("granularity").asText(DEFAULT_HISTORY_GRANULARITY) + else DEFAULT_HISTORY_GRANULARITY, + versions = if (history.hasNonNull("versions")) history.get("versions").asInt(DEFAULT_HISTORY_VERSIONS) + else DEFAULT_HISTORY_VERSIONS) + } catch { + case _: Exception => default + } + } + + /** + * Convert an OpenHouse history-policy granularity to the ChronoUnit used to compute the age cutoff, + * matching the OpenHouse expiration job. Unknown values fall back to DAYS. Exposed for testing. + */ + def granularityToChrono(granularity: String): ChronoUnit = + granularity.toUpperCase match { + case "HOUR" => ChronoUnit.HOURS + case "DAY" => ChronoUnit.DAYS + case "MONTH" => ChronoUnit.MONTHS + case "YEAR" => ChronoUnit.YEARS + case _ => ChronoUnit.DAYS + } } diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExecTest.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExecTest.scala new file mode 100644 index 000000000..98c60e881 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/VacuumTableExecTest.scala @@ -0,0 +1,60 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.util.Collections + +import scala.collection.JavaConverters._ + +import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows} +import org.junit.jupiter.api.Test + +import com.linkedin.openhouse.spark.sql.execution.datasources.v2.VacuumTableExec._ + +class VacuumTableExecTest { + + private def props(pairs: (String, String)*): java.util.Map[String, String] = + pairs.toMap.asJava + + @Test + def ofdRetainDaysDefaultsToThree(): Unit = { + assertEquals(3L, ofdRetainDays(Collections.emptyMap())) + assertEquals(3L, ofdRetainDays(props(OFD_ONE_DAY_TTL_ENABLED_PROP -> "false"))) + } + + @Test + def ofdRetainDaysOneDayWhenEnabled(): Unit = { + assertEquals(1L, ofdRetainDays(props(OFD_ONE_DAY_TTL_ENABLED_PROP -> "true"))) + assertEquals(1L, ofdRetainDays(props(OFD_ONE_DAY_TTL_ENABLED_PROP -> "TRUE"))) + } + + @Test + def parseHistoryRetentionDefaultsWhenAbsent(): Unit = { + val expected = HistoryRetention(3, "DAY", 0) + assertEquals(expected, parseHistoryRetention(null)) + assertEquals(expected, parseHistoryRetention("")) + assertEquals(expected, parseHistoryRetention("""{"retention":{"count":5}}""")) + } + + @Test + def parseHistoryRetentionReadsHistoryBlock(): Unit = { + assertEquals( + HistoryRetention(30, "DAY", 10), + parseHistoryRetention("""{"history":{"maxAge":30,"granularity":"DAY","versions":10}}""")) + assertEquals( + HistoryRetention(6, "HOUR", 0), + parseHistoryRetention("""{"history":{"maxAge":6,"granularity":"HOUR"}}""")) + } + + @Test + def parseHistoryRetentionFallsBackOnMalformedJson(): Unit = { + assertEquals(HistoryRetention(3, "DAY", 0), parseHistoryRetention("not json")) + } + + @Test + def granularityToChronoMapsAllUnits(): Unit = { + assertEquals(java.time.temporal.ChronoUnit.HOURS, granularityToChrono("HOUR")) + assertEquals(java.time.temporal.ChronoUnit.DAYS, granularityToChrono("day")) + assertEquals(java.time.temporal.ChronoUnit.MONTHS, granularityToChrono("MONTH")) + assertEquals(java.time.temporal.ChronoUnit.YEARS, granularityToChrono("YEAR")) + assertEquals(java.time.temporal.ChronoUnit.DAYS, granularityToChrono("unknown")) + } +}