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..a2f2b4236 --- /dev/null +++ b/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(); + } +} 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..9cc8d204a --- /dev/null +++ b/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 [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. + +## 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 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. 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..3e5e5321b --- /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,182 @@ +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 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 +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. 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. 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, + 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") => + // 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(".") + val props = iceberg.table().properties() + val now = Instant.now() + + // 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)) + s", older_than => TIMESTAMP '${formatter.format(instant)}'" + } + + if (removeOrphanFiles) { + // 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. + // + // 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'$ofdOlderThan)").collect() + } + + // 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'$expireArgs)").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")}" + } +} + +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" + + /** 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")) + } +}