diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/AnalyzeClusteringTestSpark3_5.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/AnalyzeClusteringTestSpark3_5.java new file mode 100644 index 000000000..e1b6148c8 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/AnalyzeClusteringTestSpark3_5.java @@ -0,0 +1,150 @@ +package com.linkedin.openhouse.spark.catalogtest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linkedin.openhouse.tablestest.OpenHouseSparkITest; +import java.util.List; +import java.util.Optional; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; + +/** + * End-to-end integration tests for the Spark SQL {@code ANALYZE TABLE ... COMPUTE CLUSTERING + * QUALITY} command against a real OpenHouse catalog (embedded {@code OpenHouseLocalServer}, + * authenticated via the {@code spark.sql.catalog.openhouse.auth-token} catalog property configured + * by the test harness). + * + *

Unlike {@code VACUUM} / {@code OPTIMIZE} (thin {@code CALL} passthroughs), this command reads + * the table's {@code .files} / {@code .partitions} metadata tables and runs distributed aggregation + * SQL over them, then returns scalar quality metrics -- it never collects per-file rows to the + * driver. This suite is therefore the one that specifically exercises OpenHouse's metadata-table + * surface and catalog resolution through the read path (not just the procedure/auth path). The + * exhaustive metric matrix (coverage math, depth math, per-dimension breakdowns) lives in the + * Hadoop-catalog {@code OptimizeClusteringIcebergSuite} in the Spark repo; here we verify the + * command runs end-to-end through OpenHouse and returns coherent metrics. + * + *

Output schema is three string columns: {@code (metric, dimension, value)}. Table-level metrics + * carry a null {@code dimension}; per-key metrics (e.g. {@code depth_avg}) carry the key name. + */ +@TestMethodOrder(MethodOrderer.MethodName.class) +@Execution(ExecutionMode.SAME_THREAD) +public class AnalyzeClusteringTestSpark3_5 extends OpenHouseSparkITest { + + private static final String DATABASE = "d1_analyze_spark"; + private static final String ANALYZE_TEST_PREFIX = "analyze_test_"; + + private static String clustered(String keys) { + return "'optimize.cluster.keys'='" + + keys + + "', 'optimize.cluster.sort-mode'='zorder', " + + "'optimize.cluster.min-snapshot-age-minutes'='0'"; + } + + /** Value of a table-level metric (null dimension), or empty if absent. */ + private static Optional metric(List rows, String name) { + return rows.stream() + .filter(r -> name.equals(r.getString(0)) && r.isNullAt(1)) + .map(r -> r.getString(2)) + .findFirst(); + } + + /** Value of a per-dimension metric, or empty if absent. */ + private static Optional dim(List rows, String name, String dimension) { + return rows.stream() + .filter(r -> name.equals(r.getString(0)) && dimension.equals(r.getString(1))) + .map(r -> r.getString(2)) + .findFirst(); + } + + @Test + public void testAnalyzeClusteringQualityRunsOnOpenHouseTable() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableId = ANALYZE_TEST_PREFIX + System.currentTimeMillis(); + String tableName = "openhouse." + DATABASE + "." + tableId; + + spark.sql( + "CREATE TABLE " + + tableName + + " (ts int, val int) TBLPROPERTIES (" + + clustered("ts") + + ")"); + // Interleaved files (each spans the key range) -> measurable overlap depth > 1. + spark.sql("INSERT INTO " + tableName + " VALUES (1, 1), (6, 6)"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 2), (6, 7)"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 3), (6, 8)"); + + long snapsBefore = + spark.sql("SELECT * FROM " + tableName + ".snapshots").collectAsList().size(); + + List rows = + spark.sql("ANALYZE TABLE " + tableName + " COMPUTE CLUSTERING QUALITY").collectAsList(); + assertTrue(rows.size() > 0, "ANALYZE must return metric rows"); + + // The command read the OpenHouse table's metadata + configuration and reported it coherently. + assertEquals( + "true", + metric(rows, "clustering_configured").orElse(null), + "table is configured for clustering"); + assertEquals("ts", metric(rows, "keys").orElse(null), "configured clustering key"); + + // Coverage is a table-level percentage produced by a distributed aggregate over .files. + String coverage = metric(rows, "coverage_bytes_pct").orElse(null); + assertNotNull(coverage, "coverage_bytes_pct must be reported"); + double coveragePct = Double.parseDouble(coverage); + assertTrue( + coveragePct >= 0.0 && coveragePct <= 100.0, + "coverage_bytes_pct must be a valid percentage, got " + coveragePct); + + // Depth is the windowed stabbing-sweep over the key's intervals; interleaved data -> depth > + // 1. + String depthAvg = dim(rows, "depth_avg", "ts").orElse(null); + assertNotNull(depthAvg, "depth_avg for key 'ts' must be reported"); + assertTrue( + Double.parseDouble(depthAvg) >= 2.0, + "interleaved data should have overlap depth well above 1, got " + depthAvg); + + // ANALYZE is read-only: it must not commit a snapshot to the OpenHouse table. + long snapsAfter = + spark.sql("SELECT * FROM " + tableName + ".snapshots").collectAsList().size(); + assertEquals(snapsBefore, snapsAfter, "ANALYZE must not commit a snapshot"); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testAnalyzeClusteringQualityOnUnclusteredTableIsGraceful() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableId = ANALYZE_TEST_PREFIX + "unclustered_" + System.currentTimeMillis(); + String tableName = "openhouse." + DATABASE + "." + tableId; + + // No optimize.cluster.* properties -> the table is not configured for clustering. + spark.sql("CREATE TABLE " + tableName + " (ts int, val int)"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 1)"); + spark.sql("INSERT INTO " + tableName + " VALUES (2, 2)"); + + // Must not error on a non-clustered OpenHouse table; it should report clustering as + // unconfigured. + List rows = + spark.sql("ANALYZE TABLE " + tableName + " COMPUTE CLUSTERING QUALITY").collectAsList(); + assertTrue(rows.size() > 0, "ANALYZE must return a result even for unclustered tables"); + assertEquals( + "false", + metric(rows, "clustering_configured").orElse(null), + "table is not configured for clustering"); + + // Table remains fully readable. + List data = spark.sql("SELECT ts FROM " + tableName + " ORDER BY ts").collectAsList(); + assertEquals(2, data.size()); + + spark.sql("DROP TABLE " + tableName); + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/OptimizeTestSpark3_5.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/OptimizeTestSpark3_5.java new file mode 100644 index 000000000..9a04a0cde --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/OptimizeTestSpark3_5.java @@ -0,0 +1,136 @@ +package com.linkedin.openhouse.spark.catalogtest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linkedin.openhouse.tablestest.OpenHouseSparkITest; +import java.util.List; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; + +/** + * End-to-end integration tests for the Spark SQL {@code OPTIMIZE} command against a real OpenHouse + * catalog (embedded {@code OpenHouseLocalServer}, authenticated via the {@code + * spark.sql.catalog.openhouse.auth-token} catalog property configured by the test harness). + * + *

{@code OPTIMIZE} is thin sugar that resolves the target catalog and issues the equivalent + * {@code CALL openhouse.system.rewrite_data_files(...)} (and, under {@code REWRITE MANIFESTS}, + * {@code rewrite_manifests(...)}). These are the same procedure CALLs already exercised against + * OpenHouse by the jobs app ({@code Operations.rewriteDataFiles}); this suite verifies that the + * {@code OPTIMIZE} verb drives them end-to-end through the OpenHouse catalog and its auth path, + * that the clustering configuration carried on {@code optimize.cluster.*} table properties survives + * an OpenHouse round-trip, and that data is preserved. The exhaustive clustering correctness matrix + * (partition transforms, key types, sort modes, delete modes, incremental watermark) lives in the + * Hadoop-catalog {@code OptimizeClusteringIcebergSuite} in the Spark repo. + */ +@TestMethodOrder(MethodOrderer.MethodName.class) +@Execution(ExecutionMode.SAME_THREAD) +public class OptimizeTestSpark3_5 extends OpenHouseSparkITest { + + private static final String DATABASE = "d1_optimize_spark"; + private static final String OPTIMIZE_TEST_PREFIX = "optimize_test_"; + + // 'optimize.cluster.min-snapshot-age-minutes'='0' disables the conflict hold-back so a freshly + // inserted table is fully eligible for rewrite within the test (no real-time settle wait). + private static String clustered(String keys) { + return "'optimize.cluster.keys'='" + + keys + + "', 'optimize.cluster.sort-mode'='zorder', " + + "'optimize.cluster.min-snapshot-age-minutes'='0'"; + } + + private static long snapshotCount(SparkSession spark, String tableName) { + return spark.sql("SELECT snapshot_id FROM " + tableName + ".snapshots").collectAsList().size(); + } + + @Test + public void testOptimizeFullClustersAndPreservesDataOnOpenHouseTable() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableId = OPTIMIZE_TEST_PREFIX + System.currentTimeMillis(); + String tableName = "openhouse." + DATABASE + "." + tableId; + + spark.sql( + "CREATE TABLE " + + tableName + + " (ts int, val int) TBLPROPERTIES (" + + clustered("ts") + + ")"); + + // Assert the clustering configuration survived the OpenHouse create round-trip -- if + // OpenHouse + // strips unknown table properties, OPTIMIZE would silently degrade to plain compaction, so + // pin + // it explicitly here. + List keysProp = + spark + .sql("SHOW TBLPROPERTIES " + tableName + " ('optimize.cluster.keys')") + .collectAsList(); + assertEquals(1, keysProp.size(), "optimize.cluster.keys must persist on the OpenHouse table"); + assertEquals("ts", keysProp.get(0).getString(1)); + + // Several files, each spanning the whole key range -> interleaved -> a real rewrite to do. + spark.sql("INSERT INTO " + tableName + " VALUES (1, 1), (6, 6)"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 2), (6, 7)"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 3), (6, 8)"); + + long before = snapshotCount(spark, tableName); + spark.sql("OPTIMIZE " + tableName + " FULL"); + long after = snapshotCount(spark, tableName); + + assertTrue( + after > before, + "OPTIMIZE FULL should commit at least one rewrite snapshot (before=" + + before + + ", after=" + + after + + ")"); + + List rows = spark.sql("SELECT val FROM " + tableName + " ORDER BY val").collectAsList(); + assertEquals(6, rows.size(), "all data rows must remain readable after OPTIMIZE"); + assertEquals("1", rows.get(0).mkString()); + assertEquals("8", rows.get(5).mkString()); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testOptimizeRewriteManifestsRunsAndKeepsTableReadable() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableId = OPTIMIZE_TEST_PREFIX + "manifests_" + System.currentTimeMillis(); + String tableName = "openhouse." + DATABASE + "." + tableId; + + spark.sql( + "CREATE TABLE " + + tableName + + " (ts int, val int) TBLPROPERTIES (" + + clustered("ts") + + ")"); + spark.sql("INSERT INTO " + tableName + " VALUES (1, 1)"); + spark.sql("INSERT INTO " + tableName + " VALUES (2, 2)"); + spark.sql("INSERT INTO " + tableName + " VALUES (3, 3)"); + + // Exercise the full OPTIMIZE surface end-to-end against OpenHouse: rewrite_data_files + // followed + // by rewrite_manifests (two commits). Asserts the command drives both CALLs through the + // OpenHouse catalog + auth path without error and that data survives. + spark.sql("OPTIMIZE " + tableName + " REWRITE MANIFESTS"); + + List rows = spark.sql("SELECT ts FROM " + tableName + " ORDER BY ts").collectAsList(); + assertEquals(3, rows.size(), "table must remain fully readable after REWRITE MANIFESTS"); + assertEquals("1", rows.get(0).mkString()); + assertEquals("3", rows.get(2).mkString()); + + List manifests = + spark.sql("SELECT path FROM " + tableName + ".manifests").collectAsList(); + assertTrue(manifests.size() >= 1, "table must have at least one manifest after OPTIMIZE"); + + spark.sql("DROP TABLE " + tableName); + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/VacuumTestSpark3_5.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/VacuumTestSpark3_5.java new file mode 100644 index 000000000..6b4a7db23 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/VacuumTestSpark3_5.java @@ -0,0 +1,210 @@ +package com.linkedin.openhouse.spark.catalogtest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linkedin.openhouse.tablestest.OpenHouseSparkITest; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.junit.jupiter.api.MethodOrderer; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestMethodOrder; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; + +/** + * End-to-end integration tests for the Spark SQL {@code VACUUM} extension against a real OpenHouse + * catalog (embedded {@code OpenHouseLocalServer}, authenticated via the {@code + * spark.sql.catalog.openhouse.auth-token} catalog property configured by the test harness). + * + *

{@code VACUUM} resolves the target catalog and issues the equivalent {@code CALL + * openhouse.system.expire_snapshots(...)} / {@code remove_orphan_files(...)} statements. These are + * the same procedure CALLs already exercised against OpenHouse by {@code BranchTestSpark3_5} and by + * the jobs app ({@code Operations.rewriteDataFiles}); this suite verifies that the {@code VACUUM} + * verb drives them end-to-end through the OpenHouse catalog and its auth path. + * + *

It also pins the parts of the property contract that only a real server can exercise: the + * {@code maintenance.vacuum.enabled} opt-in has to be settable (an {@code openhouse.}-prefixed one + * would be rejected as a reserved key), and the default retention has to come from the {@code + * policies.history} the server persists, which is what the scheduled snapshot-expiration job reads. + */ +@TestMethodOrder(MethodOrderer.MethodName.class) +@Execution(ExecutionMode.SAME_THREAD) +public class VacuumTestSpark3_5 extends OpenHouseSparkITest { + + private static final String DATABASE = "d1_vacuum_spark"; + private static final String VACUUM_TEST_PREFIX = "vacuum_test_"; + + private static String createEnabledTable(SparkSession spark, String suffix) { + String tableName = "openhouse." + DATABASE + "." + VACUUM_TEST_PREFIX + suffix; + spark.sql("CREATE TABLE " + tableName + " (id int)"); + // The Alpha opt-in. This ALTER is itself part of what is under test: the property has to live + // outside the reserved `openhouse.` namespace for a user to be able to set it at all. + spark.sql( + "ALTER TABLE " + tableName + " SET TBLPROPERTIES ('maintenance.vacuum.enabled' = 'true')"); + return tableName; + } + + private static Map vacuum(SparkSession spark, String statement) { + Map metrics = new HashMap<>(); + for (Row row : spark.sql(statement).collectAsList()) { + metrics.put(row.getString(0), row.getString(1)); + } + return metrics; + } + + private static long snapshotCount(SparkSession spark, String tableName) { + return spark.sql("SELECT snapshot_id FROM " + tableName + ".snapshots").count(); + } + + @Test + public void testVacuumExpiresSnapshotsOnOpenHouseTable() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = createEnabledTable(spark, "" + System.currentTimeMillis()); + + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + spark.sql("INSERT INTO " + tableName + " VALUES (2)"); + spark.sql("INSERT INTO " + tableName + " VALUES (3)"); + + assertEquals( + 3, snapshotCount(spark, tableName), "three inserts should produce three snapshots"); + + // Snapshot expiration always runs; RETAIN 0 HOURS bounds older_than to the VACUUM instant so + // every non-current snapshot is expirable. + Map metrics = vacuum(spark, "VACUUM " + tableName + " RETAIN 0 HOURS"); + + assertEquals("RETAIN", metrics.get("snapshots_retain_source")); + assertEquals( + 1, snapshotCount(spark, tableName), "expiration should retain only the current snapshot"); + + List rows = spark.sql("SELECT id FROM " + tableName + " ORDER BY id").collectAsList(); + assertEquals(3, rows.size(), "all data rows should remain readable after VACUUM"); + assertEquals("1", rows.get(0).mkString()); + assertEquals("2", rows.get(1).mkString()); + assertEquals("3", rows.get(2).mkString()); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testVacuumRemoveOrphanFilesRunsAndKeepsTableReadable() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = createEnabledTable(spark, "ofd_" + System.currentTimeMillis()); + + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + spark.sql("INSERT INTO " + tableName + " VALUES (2)"); + + // Exercise the full VACUUM surface end-to-end against OpenHouse: orphan-file deletion + // followed by expiration. RETAIN 24 HOURS satisfies Iceberg's OFD safety window. This asserts + // the command drives both CALLs through the OpenHouse catalog + auth path without error and + // that referenced data survives (correct-files-preserved). Planting an aged orphan file to + // assert selective deletion requires direct access to the table's storage location, which the + // OpenHouse-managed layout does not expose to the client. + Map metrics = + vacuum(spark, "VACUUM " + tableName + " REMOVE ORPHAN FILES RETAIN 24 HOURS"); + assertEquals("24", metrics.get("orphan_files_retain_hours")); + + List rows = spark.sql("SELECT id FROM " + tableName + " ORDER BY id").collectAsList(); + assertEquals(2, rows.size(), "table must remain fully readable after orphan-file deletion"); + assertEquals("1", rows.get(0).mkString()); + assertEquals("2", rows.get(1).mkString()); + + List files = spark.sql("SELECT file_path FROM " + tableName + ".files").collectAsList(); + assertTrue(files.size() >= 1, "referenced data files must be preserved after VACUUM"); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testVacuumDefaultRetentionComesFromTheServerPersistedHistoryPolicy() + throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = createEnabledTable(spark, "policy_" + System.currentTimeMillis()); + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + + // The same knob the scheduled snapshot-expiration job reads back off the table. + spark.sql("ALTER TABLE " + tableName + " SET POLICY (HISTORY MAX_AGE=2D)"); + + Map metrics = vacuum(spark, "VACUUM " + tableName); + + assertEquals( + "48", + metrics.get("snapshots_retain_hours"), + "the default window must be the table's history policy, not the procedure default"); + assertTrue( + metrics.get("snapshots_retain_source").startsWith("policies.history"), + "resolved window should be reported as coming from the history policy, was " + + metrics.get("snapshots_retain_source")); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testVacuumDefaultRetentionFallsBackToTheJobDefault() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = createEnabledTable(spark, "nopolicy_" + System.currentTimeMillis()); + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + + Map metrics = vacuum(spark, "VACUUM " + tableName); + + assertEquals("72", metrics.get("snapshots_retain_hours")); + assertEquals("default (3 DAY)", metrics.get("snapshots_retain_source")); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testVacuumOptInPropertyMustBeOutsideTheReservedNamespace() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = + "openhouse." + DATABASE + "." + VACUUM_TEST_PREFIX + "gate_" + System.currentTimeMillis(); + spark.sql("CREATE TABLE " + tableName + " (id int)"); + + // Reserved: the /tables service rejects any attempt to set an `openhouse.`-prefixed property, + // so a gate in that namespace could never be turned on. + assertThrows( + Exception.class, + () -> + spark.sql( + "ALTER TABLE " + + tableName + + " SET TBLPROPERTIES ('openhouse.vacuum.enabled' = 'true')")); + + // Not yet opted in. + assertThrows( + UnsupportedOperationException.class, () -> spark.sql("VACUUM " + tableName).collect()); + + spark.sql( + "ALTER TABLE " + + tableName + + " SET TBLPROPERTIES ('maintenance.vacuum.enabled' = 'true')"); + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + vacuum(spark, "VACUUM " + tableName); + + spark.sql("DROP TABLE " + tableName); + } + } + + @Test + public void testVacuumRefusesWhenMaintenanceIsDisabled() throws Exception { + try (SparkSession spark = getSparkSession()) { + String tableName = createEnabledTable(spark, "disabled_" + System.currentTimeMillis()); + spark.sql("INSERT INTO " + tableName + " VALUES (1)"); + spark.sql( + "ALTER TABLE " + tableName + " SET TBLPROPERTIES ('maintenance.disabled' = 'true')"); + + assertThrows( + UnsupportedOperationException.class, () -> spark.sql("VACUUM " + tableName).collect()); + + spark.sql("DROP TABLE " + tableName); + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/AnalyzeClusteringQualityStatementTest.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/AnalyzeClusteringQualityStatementTest.java new file mode 100644 index 000000000..12bcebc96 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/AnalyzeClusteringQualityStatementTest.java @@ -0,0 +1,156 @@ +package com.linkedin.openhouse.spark.statementtest; + +import java.nio.file.Files; +import java.util.HashMap; +import java.util.Map; +import lombok.SneakyThrows; +import org.apache.hadoop.fs.Path; +import org.apache.spark.sql.Row; +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 AnalyzeClusteringQualityStatementTest { + + private static SparkSession spark = null; + + /** Collapse the (metric, dimension, value) output into metric[/dimension] -> value. */ + private Map analyze(String table) { + Map out = new HashMap<>(); + for (Row r : + spark.sql("ANALYZE TABLE " + table + " COMPUTE CLUSTERING QUALITY").collectAsList()) { + String metric = r.getString(0); + String dimension = r.isNullAt(1) ? null : r.getString(1); + out.put(dimension == null ? metric : metric + "/" + dimension, r.getString(2)); + } + return out; + } + + @Test + public void testAnalyzeUnconfiguredReportsNotConfigured() { + Map m = analyze("openhouse.db.table"); + Assertions.assertEquals("false", m.get("clustering_configured")); + // A not-configured table reports only the single flag row. + Assertions.assertEquals(1, m.size()); + } + + @Test + public void testAnalyzeAfterOptimizeReportsCoverageAndDepth() { + spark + .sql( + "ALTER TABLE openhouse.db.table SET TBLPROPERTIES (" + + "'optimize.cluster.keys' = 'id', " + + "'optimize.cluster.sort-mode' = 'sort', " + + "'optimize.cluster.min-snapshot-age-minutes' = '0')") + .show(); + spark.sql("OPTIMIZE openhouse.db.table").collect(); + + Map m = analyze("openhouse.db.table"); + Assertions.assertEquals("true", m.get("clustering_configured")); + Assertions.assertEquals("id", m.get("keys")); + Assertions.assertEquals("sort", m.get("sort_mode")); + Assertions.assertNotNull(m.get("config_id")); + // After a full-scope OPTIMIZE, all bytes fall inside the clustered interval. + Assertions.assertEquals("100.00", m.get("coverage_bytes_pct")); + // Per-key depth rows are emitted for the leading key. + Assertions.assertNotNull(m.get("depth_avg/id")); + Assertions.assertNotNull(m.get("depth_max/id")); + Assertions.assertNotNull(m.get("depth_avg_covered/id")); + // The persisted interval state is echoed back. + Assertions.assertTrue(m.get("state").trim().startsWith("[")); + } + + @Test + public void testAnalyzeIsReadOnly() { + spark + .sql( + "ALTER TABLE openhouse.db.table SET TBLPROPERTIES (" + + "'optimize.cluster.keys' = 'id', " + + "'optimize.cluster.min-snapshot-age-minutes' = '0')") + .show(); + long snapshotsBefore = spark.sql("SELECT * FROM openhouse.db.table.snapshots").count(); + analyze("openhouse.db.table"); + long snapshotsAfter = spark.sql("SELECT * FROM openhouse.db.table.snapshots").count(); + // A read-only probe commits nothing. + Assertions.assertEquals(snapshotsBefore, snapshotsAfter); + } + + @Test + public void testAnalyzeComputeStatisticsStillDelegatesToSpark() { + // COMPUTE STATISTICS must NOT be intercepted by the OpenHouse grammar; it is handled by Spark. + // For a v2 Iceberg table Spark rejects it with its own AnalysisException -- crucially NOT an + // OpenhouseParseException -- which confirms the extension did not claim the statement. + Exception e = + Assertions.assertThrows( + Exception.class, + () -> spark.sql("ANALYZE TABLE openhouse.db.table COMPUTE STATISTICS").collect()); + Assertions.assertFalse( + e + instanceof + com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseParseException, + "COMPUTE STATISTICS must delegate to Spark, not the OpenHouse parser"); + } + + @Test + public void testAnalyzeNonOpenhouseTableThrows() { + Assertions.assertThrows( + Exception.class, + () -> + spark + .sql("ANALYZE TABLE openhouse.db.not_openhouse COMPUTE CLUSTERING QUALITY") + .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(); + for (int i = 1; i <= 6; i++) { + spark + .sql("INSERT INTO openhouse.db.table VALUES (" + i + ", 'd" + i + "', '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-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/OptimizeStatementTest.java b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/OptimizeStatementTest.java new file mode 100644 index 000000000..96ad6ae70 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/OptimizeStatementTest.java @@ -0,0 +1,160 @@ +package com.linkedin.openhouse.spark.statementtest; + +import java.nio.file.Files; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import lombok.SneakyThrows; +import org.apache.hadoop.fs.Path; +import org.apache.spark.sql.Row; +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 OptimizeStatementTest { + + private static SparkSession spark = null; + + private Map optimize(String sql) { + Map metrics = new HashMap<>(); + for (Row r : spark.sql(sql).collectAsList()) { + metrics.put(r.getString(0), r.getString(1)); + } + return metrics; + } + + private long rowCount(String table) { + return spark.sql("SELECT * FROM " + table).count(); + } + + private String tableProperty(String table, String key) { + List rows = spark.sql("SHOW TBLPROPERTIES " + table + " ('" + key + "')").collectAsList(); + return rows.isEmpty() ? null : rows.get(0).getString(1); + } + + @Test + public void testOptimizeBinPackReturnsMetricsAndPreservesRows() { + Map m = optimize("OPTIMIZE openhouse.db.table"); + // Output surface: the four reduction metrics. + Assertions.assertTrue(m.containsKey("files_before")); + Assertions.assertTrue(m.containsKey("files_after")); + Assertions.assertTrue(m.containsKey("files_removed")); + Assertions.assertTrue(m.containsKey("snapshots_committed")); + // Bin-pack compaction never loses or duplicates rows. + Assertions.assertEquals(6, rowCount("openhouse.db.table")); + // Six single-row files are above the default min-input-files, so they compact down. + Assertions.assertTrue( + Long.parseLong(m.get("files_after")) < Long.parseLong(m.get("files_before"))); + } + + @Test + public void testOptimizeClusteringWritesDurableStateAndPreservesRows() { + spark + .sql( + "ALTER TABLE openhouse.db.table SET TBLPROPERTIES (" + + "'optimize.cluster.keys' = 'id', " + + "'optimize.cluster.sort-mode' = 'sort', " + + "'optimize.cluster.min-snapshot-age-minutes' = '0')") + .show(); + + Map m = optimize("OPTIMIZE openhouse.db.table"); + Assertions.assertEquals(6, rowCount("openhouse.db.table")); + // A clustered run committed a rewrite snapshot and advanced the durable metadata. + Assertions.assertTrue(Long.parseLong(m.get("snapshots_committed")) >= 1); + String state = tableProperty("openhouse.db.table", "optimize.cluster.state"); + Assertions.assertNotNull(state); + Assertions.assertTrue(state.trim().startsWith("[")); + Assertions.assertNotNull( + tableProperty("openhouse.db.table", "optimize.cluster.hwm-snapshot-id")); + } + + @Test + public void testOptimizeClusteringIncrementalSecondRunIsNoOp() { + spark + .sql( + "ALTER TABLE openhouse.db.table SET TBLPROPERTIES (" + + "'optimize.cluster.keys' = 'id', " + + "'optimize.cluster.sort-mode' = 'sort', " + + "'optimize.cluster.min-snapshot-age-minutes' = '0')") + .show(); + + optimize("OPTIMIZE openhouse.db.table"); + // No new data arrived, so the incremental run finds nothing above the last-clustered upper. + Map second = optimize("OPTIMIZE openhouse.db.table"); + Assertions.assertEquals("0", second.get("snapshots_committed")); + Assertions.assertEquals(6, rowCount("openhouse.db.table")); + } + + @Test + public void testOptimizeRewriteManifestsPreservesRows() { + Map m = optimize("OPTIMIZE openhouse.db.table REWRITE MANIFESTS"); + Assertions.assertTrue(m.containsKey("files_after")); + Assertions.assertEquals(6, rowCount("openhouse.db.table")); + } + + @Test + public void testOptimizeFullRewriteManifestsParsesAndRuns() { + Map m = optimize("OPTIMIZE openhouse.db.table FULL REWRITE MANIFESTS"); + Assertions.assertEquals(6, rowCount("openhouse.db.table")); + Assertions.assertTrue(m.containsKey("files_removed")); + } + + @Test + public void testOptimizeNonOpenhouseTableThrows() { + Assertions.assertThrows( + Exception.class, () -> spark.sql("OPTIMIZE openhouse.db.not_openhouse").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(); + for (int i = 1; i <= 6; i++) { + spark + .sql("INSERT INTO openhouse.db.table VALUES (" + i + ", 'd" + i + "', '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-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..4b59311e8 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/VacuumStatementTest.java @@ -0,0 +1,253 @@ +package com.linkedin.openhouse.spark.statementtest; + +import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseParseException; +import java.nio.file.Files; +import java.util.HashMap; +import java.util.Map; +import lombok.SneakyThrows; +import org.apache.hadoop.fs.Path; +import org.apache.spark.sql.Row; +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(); + } + + /** The (metric, value) rows VACUUM reports, as a map. */ + private Map vacuum(String statement) { + Map metrics = new HashMap<>(); + for (Row row : spark.sql(statement).collectAsList()) { + metrics.put(row.getString(0), row.getString(1)); + } + return metrics; + } + + private void setProperties(String table, String properties) { + spark.sql("ALTER TABLE " + table + " SET TBLPROPERTIES (" + properties + ")").show(); + } + + @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. + Map metrics = vacuum("VACUUM openhouse.db.table RETAIN 0 HOURS"); + + Assertions.assertEquals(1, snapshotCount("openhouse.db.table")); + Assertions.assertEquals(3, rowCount("openhouse.db.table")); + Assertions.assertEquals("0", metrics.get("snapshots_retain_hours")); + Assertions.assertEquals("RETAIN", metrics.get("snapshots_retain_source")); + } + + @Test + public void testVacuumWithDefaultRetentionSucceeds() { + // No RETAIN and no history policy: the snapshot-expiration job's own 3-day default applies. + Map metrics = vacuum("VACUUM openhouse.db.table"); + Assertions.assertEquals(3, rowCount("openhouse.db.table")); + Assertions.assertEquals("72", metrics.get("snapshots_retain_hours")); + Assertions.assertEquals("default (3 DAY)", metrics.get("snapshots_retain_source")); + } + + @Test + public void testVacuumDefaultRetentionComesFromTheHistoryPolicy() { + // The window the scheduled snapshot-expiration job would have used for this table. + setProperties( + "openhouse.db.table", + "'policies' = '{\"history\":{\"maxAge\":6,\"granularity\":\"HOUR\"}}'"); + + Map metrics = vacuum("VACUUM openhouse.db.table"); + + Assertions.assertEquals("6", metrics.get("snapshots_retain_hours")); + Assertions.assertEquals("policies.history (6 HOUR)", metrics.get("snapshots_retain_source")); + // Nothing is older than six hours, so every snapshot survives. + Assertions.assertEquals(3, snapshotCount("openhouse.db.table")); + } + + @Test + public void testVacuumAppliesTheHistoryPolicyVersionsCap() { + // `versions` caps how many snapshots survive regardless of age, as the job applies it. + setProperties( + "openhouse.db.table", + "'policies' = '{\"history\":{\"maxAge\":6,\"granularity\":\"HOUR\",\"versions\":2}}'"); + + Map metrics = vacuum("VACUUM openhouse.db.table"); + + Assertions.assertEquals("2", metrics.get("snapshots_retain_last")); + 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. + Map metrics = + vacuum("VACUUM openhouse.db.table REMOVE ORPHAN FILES RETAIN 24 HOURS"); + Assertions.assertEquals(3, rowCount("openhouse.db.table")); + Assertions.assertEquals("24", metrics.get("orphan_files_retain_hours")); + Assertions.assertEquals("RETAIN", metrics.get("orphan_files_retain_source")); + } + + @Test + public void testVacuumRemoveOrphanFilesDefaultWindowMatchesTheJob() { + Map metrics = vacuum("VACUUM openhouse.db.table REMOVE ORPHAN FILES"); + Assertions.assertEquals("168", metrics.get("orphan_files_retain_hours")); + Assertions.assertEquals("default", metrics.get("orphan_files_retain_source")); + + setProperties("openhouse.db.table", "'ofd.one_day_ttl.enabled' = 'true'"); + metrics = vacuum("VACUUM openhouse.db.table REMOVE ORPHAN FILES"); + Assertions.assertEquals("24", metrics.get("orphan_files_retain_hours")); + Assertions.assertEquals("ofd.one_day_ttl.enabled", metrics.get("orphan_files_retain_source")); + } + + @Test + public void testVacuumRemoveOrphanFilesOnBackupEnabledTableThrows() { + // The scheduled job moves orphans into the backup directory instead of deleting them; the + // stored procedure cannot, so it must not run here. + setProperties("openhouse.db.table", "'retention.backup.enabled' = 'true'"); + + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> spark.sql("VACUUM openhouse.db.table REMOVE ORPHAN FILES").collect()); + + // Expiration alone is unaffected. + vacuum("VACUUM openhouse.db.table RETAIN 0 HOURS"); + Assertions.assertEquals(1, snapshotCount("openhouse.db.table")); + } + + @Test + public void testVacuumOnReplicaTableThrows() { + // The scheduled snapshot-expiration job runs on primary tables only. + setProperties("openhouse.db.table", "'openhouse.tableType' = 'REPLICA_TABLE'"); + + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> spark.sql("VACUUM openhouse.db.table").collect()); + } + + @Test + public void testVacuumWhenMaintenanceIsDisabledThrows() { + setProperties("openhouse.db.table", "'maintenance.disabled' = 'true'"); + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> spark.sql("VACUUM openhouse.db.table").collect()); + } + + @Test + public void testVacuumWhenOnlyOrphanFileDeletionIsDisabledThrowsOnlyForThatStep() { + setProperties("openhouse.db.table", "'maintenance.ORPHAN_FILES_DELETION.disabled' = 'true'"); + + Assertions.assertThrows( + UnsupportedOperationException.class, + () -> spark.sql("VACUUM openhouse.db.table REMOVE ORPHAN FILES").collect()); + + // Snapshot expiration is a different job type and still runs. + Assertions.assertEquals(3, rowCount("openhouse.db.table")); + vacuum("VACUUM openhouse.db.table RETAIN 0 HOURS"); + Assertions.assertEquals(1, snapshotCount("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 + // maintenance.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', 'maintenance.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/build.gradle b/integrations/spark/spark-3.5/openhouse-spark-runtime/build.gradle index 5607c5a66..986507185 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/build.gradle +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/build.gradle @@ -8,6 +8,11 @@ plugins { ext { icebergVersion = rootProject.ext.iceberg_1_5_version sparkVersion = '3.5.2' + // Antlr grammar sources are owned by this module (spark-3.5 keeps its own copy of the + // grammar so it can evolve independently of spark-3.1). + antlrPackageDirPrefix = "com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/" + antlrMainDir = "${projectDir}/src/main/antlr/${antlrPackageDirPrefix}" + antlrMainGeneratedSrcDir = "${project.buildDir}/generated-src/antlr/main/" } configurations { @@ -16,18 +21,50 @@ configurations { exclude(group: 'org.mapstruct') } shadow.extendsFrom implementation + + antlr +} + +// The root build forces jackson-databind to 2.13.4; align jackson-module-scala to the same +// version so the runtime module's own unit tests can (de)serialize the clustering state. Spark +// provides a consistent jackson at itest/production runtime (this force only affects this module's +// dependency resolution, and jackson is not bundled into the shadow jar). +configurations.all { + resolutionStrategy { + force 'com.fasterxml.jackson.module:jackson-module-scala_2.12:2.13.4' + } } // Set source for antlr generated directory sourceSets { main { java { - srcDirs += "${project(':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12').buildDir}/generated-src/antlr/main" + srcDirs antlrMainGeneratedSrcDir } } } +// Task to generate java sources using Antlr tool +task runAntlr(type: JavaExec) { + inputs.dir antlrMainDir + outputs.dir antlrMainGeneratedSrcDir + + mainClass = "org.antlr.v4.Tool" + args = ["${antlrMainDir}/OpenhouseSqlExtensions.g4", + "-visitor", + "-o", "${antlrMainGeneratedSrcDir}/${antlrPackageDirPrefix}", + "-package", "com.linkedin.openhouse.spark.sql.catalyst.parser.extensions"] + maxHeapSize = "64m" + classpath = configurations.antlr +} + +compileJava.dependsOn runAntlr + dependencies { + // Required because we remove antlr plugin dependencies from the compile configuration + runtimeOnly "org.antlr:antlr4-runtime:4.7.1" + antlr "org.antlr:antlr4:4.7.1" + compileOnly(project(path: ':integrations:java:iceberg-1.5:openhouse-java-iceberg-1.5-runtime', configuration: 'shadow')) compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") { exclude group: 'org.apache.avro', module: 'avro' @@ -52,12 +89,6 @@ dependencies { fatJarPackagedDependencies(project(path: ':integrations:java:iceberg-1.5:openhouse-java-iceberg-1.5-runtime', configuration: 'shadow')) { transitive = false } - fatJarPackagedDependencies(project(path: ':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12', configuration: 'shadow')) { - transitive = false - } - implementation(project(path: ':integrations:spark:spark-3.1:openhouse-spark-runtime_2.12', configuration: 'shadow')) { - exclude group: "com.linkedin.iceberg", module: "iceberg-spark-runtime-3.1_2.12" - } implementation("com.linkedin.iceberg:iceberg-spark-runtime-3.5_2.12:" + icebergVersion) } diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/OPTIMIZE.md b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/OPTIMIZE.md new file mode 100644 index 000000000..55da89d14 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/OPTIMIZE.md @@ -0,0 +1,60 @@ +# OPTIMIZE + +`OPTIMIZE` is an OpenHouse Spark SQL extension that improves an OpenHouse Iceberg table's data +layout: bin-pack compaction by default, or a sort / z-order clustering rewrite when the table +configures clustering keys. It is sugar over the underlying Iceberg maintenance stored procedures. + +## Syntax + +```sql +OPTIMIZE [FULL] [REWRITE MANIFESTS] +``` + +- `
` — an OpenHouse table identifier (e.g. `openhouse.db.table`). +- `FULL` — *(optional)* recluster everything up to the age floor instead of only the slice added + since the last run. Has no effect when no clustering keys are configured. +- `REWRITE MANIFESTS` — *(optional)* also compact the table's manifests, in a second commit, after + the data rewrite. + +The command returns `files_before` / `files_after` / `files_removed` / `snapshots_committed`. + +## Behavior + +With no `optimize.cluster.keys` set, `OPTIMIZE` is a plain bin-pack compaction +(`rewrite_data_files` with defaults). With clustering configured it performs a scoped sort or +z-order rewrite that is **incremental by default**: it rewrites only the forward slice of the +leading key that has arrived since the previous run, bounded by an age floor that keeps it off the +partition a streaming writer is actively extending. + +After the data rewrite it compacts merge-on-read position delete files and drops deletes the +rewrite made dangling; on copy-on-write or delete-free tables that step is a no-op. + +Snapshot expiration is deliberately **not** part of `OPTIMIZE` — that is `VACUUM`'s job. + +## Table properties + +Clustering is configured with ordinary (user-settable) table properties: + +| Property | Meaning | Default | +| -------- | ------- | ------- | +| `optimize.cluster.keys` | Comma-separated clustering keys. Empty means plain bin-pack. | *(unset)* | +| `optimize.cluster.sort-mode` | `zorder` or `sort`. | `zorder` | +| `optimize.cluster.min-snapshot-age-minutes` | Age floor: snapshots younger than this are held back. | `30` | +| `optimize.cluster.max-commits` | Partial-progress commit budget for one run. | `10` | + +`OPTIMIZE` also writes back the state it needs to stay incremental — +`optimize.cluster.hwm-snapshot-id`, `optimize.cluster.config-id` and `optimize.cluster.state` — in +a single atomic property update, so they never disagree. `ANALYZE TABLE COMPUTE CLUSTERING +QUALITY` reads that same state to report how well the table is clustered. + +## Interaction with the scheduled maintenance jobs + +`OPTIMIZE` refuses to run on a table that has been opted out of platform maintenance, via +`maintenance.disabled = 'true'` or `maintenance.DATA_COMPACTION.disabled = 'true'` — the same +switches the jobs scheduler consults before dispatching work for a table. + +**Known gap.** The scheduled data-compaction job does not read `optimize.cluster.*`. It bin-packs +according to the data-layout strategies persisted for the table, so on a clustered table a +scheduled compaction can rewrite files without preserving the clustering `OPTIMIZE` established, +and it does not advance or respect the incremental watermark. Until that job learns this +configuration, treat clustered tables as owned by `OPTIMIZE` rather than by scheduled compaction. 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..00d248677 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/docs/VACUUM.md @@ -0,0 +1,126 @@ +# 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, and it reads the same table +properties as the scheduled snapshot-expiration and orphan-file-deletion jobs, so running +it by hand and letting the platform run it agree about the same table. + +## 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 step + falls back to what the corresponding maintenance job would have used for this table. + +The command returns the windows it resolved, so you can see what each step actually used: + +``` +metric value +--------------------------- ------------------------ +orphan_files_retain_hours 168 +orphan_files_retain_source default +snapshots_retain_hours 72 +snapshots_retain_source policies.history (3 DAY) +``` + +## 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. + +### Default retention + +With no `RETAIN`, each step uses the same window its scheduled job would have used: + +| Step | Window when `RETAIN` is omitted | +| ---- | ------------------------------- | +| Snapshot expiration | The table's history policy — `maxAge` x `granularity`. If the policy also sets `versions`, at most that many snapshots survive, regardless of age. With no history policy, the job default of **3 days** applies. | +| Orphan-file deletion | **7 days**, or **1 day** when the table sets `ofd.one_day_ttl.enabled = 'true'`. | + +Set the history policy the same way the scheduled job reads it: + +```sql +ALTER TABLE openhouse.db.table SET POLICY (HISTORY MAX_AGE=3D VERSIONS=10); +``` + +An explicit `RETAIN` overrides both, including below the defaults — that lever is deliberate, +so an operator can reclaim space in an emergency. + +## Enabling VACUUM + +`VACUUM` is Alpha and must be enabled on each table before use: + +```sql +ALTER TABLE openhouse.db.table + SET TBLPROPERTIES ('maintenance.vacuum.enabled' = 'true'); +``` + +| Property | Value | Meaning | +| ---------------------------- | -------- | ----------------------------------------------- | +| `maintenance.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. + +The property is in the `maintenance.` namespace rather than `openhouse.` because the /tables +service treats `openhouse.`-prefixed keys as reserved and rejects any attempt to set them. + +## When VACUUM refuses to run + +| Situation | Why | +| --------- | --- | +| `maintenance.disabled = 'true'`, or `maintenance.SNAPSHOTS_EXPIRATION.disabled` / `maintenance.ORPHAN_FILES_DELETION.disabled` for the step being run | The table has been opted out of platform maintenance; a manual `VACUUM` should not sidestep that. | +| The table is a replica (`openhouse.tableType = 'REPLICA_TABLE'`) | The scheduled expiration job runs on primary tables only. A replica's snapshots are replication state, and expiring them by hand can strand an incremental replication. Maintenance for replicas stays with the scheduled jobs. | +| `REMOVE ORPHAN FILES` on a table configured for orphan backups (`retention.backup.enabled` / `retention.backup.dir`) | On those tables the scheduled job *moves* orphans into the backup directory instead of deleting them. The stored procedure has no equivalent hook, so it would destroy files the platform expects to remain recoverable — and treat the backup directory's own contents as orphans. | + +## Examples + +Enable the feature, then expire snapshots older than 24 hours: + +```sql +ALTER TABLE openhouse.db.table + SET TBLPROPERTIES ('maintenance.vacuum.enabled' = 'true'); + +VACUUM openhouse.db.table RETAIN 24 HOURS; +``` + +Expire snapshots using the table's history policy: + +```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. +- **Expiration here also deletes files.** The scheduled expiration job deliberately leaves file + deletion to orphan-file deletion; `VACUUM` deletes the files the expired snapshots exclusively + referenced, because reclaiming that storage is the point of running it by hand. 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 new file mode 100644 index 000000000..e1641a05d --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 @@ -0,0 +1,264 @@ +grammar OpenhouseSqlExtensions; + +@lexer::members { + /** + * This method will be called when we see '/*' and try to match it as a bracketed comment. + * If the next character is '+', it should be parsed as hint later, and we cannot match + * it as a bracketed comment. + * + * Returns true if the next character is '+'. + */ + public boolean isHint() { + int nextChar = _input.LA(1); + if (nextChar == '+') { + return true; + } else { + return false; + } + } +} + +singleStatement + : statement EOF + ; + +statement + : ALTER TABLE multipartIdentifier SET POLICY '(' retentionPolicy (columnRetentionPolicy)? ')' #setRetentionPolicy + | ALTER TABLE multipartIdentifier SET POLICY '(' replicationPolicy ')' #setReplicationPolicy + | ALTER TABLE multipartIdentifier UNSET POLICY '(' replication ')' #unSetReplicationPolicy + | ALTER TABLE multipartIdentifier SET POLICY '(' sharingPolicy ')' #setSharingPolicy + | ALTER TABLE multipartIdentifier SET POLICY '(' historyPolicy ')' #setHistoryPolicy + | ALTER TABLE multipartIdentifier MODIFY columnNameClause SET columnPolicy #setColumnPolicyTag + | 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 + | OPTIMIZE multipartIdentifier (FULL)? (REWRITE MANIFESTS)? #optimizeTable + | ANALYZE TABLE multipartIdentifier COMPUTE CLUSTERING QUALITY #analyzeClusteringQuality + ; + +multipartIdentifier + : parts+=identifier ('.' parts+=identifier)* + ; + +privilege + : columnLevelPrivilege + | SELECT | DESCRIBE | ALTER | GRANT_REVOKE | CREATE_TABLE + ; + +columnLevelPrivilege + : SELECT policyTag + ; + +grantableResource + : TABLE multipartIdentifier + | DATABASE multipartIdentifier + ; + +principal + : identifier + ; + +identifier + : IDENTIFIER + | quotedIdentifier + | nonReserved + ; + +quotedIdentifier + : BACKQUOTED_IDENTIFIER + ; + +nonReserved + : ALTER | TABLE | SET | POLICY | RETENTION | SHARING | REPLICATION | HISTORY + | GRANT | REVOKE | ON | TO | SHOW | GRANTS | PATTERN | WHERE | COLUMN + | VACUUM | REMOVE | ORPHAN | FILES | RETAIN | HOURS + | OPTIMIZE | FULL | REWRITE | MANIFESTS + | ANALYZE | COMPUTE | CLUSTERING | QUALITY + ; + +sharingPolicy + : SHARING '=' BOOLEAN + ; + +BOOLEAN + : 'TRUE' | 'FALSE' + ; + +retentionPolicy + : RETENTION '=' duration + ; + +columnRetentionPolicy + : ON columnNameClause (columnRetentionPolicyPatternClause)? + ; + +replication + : REPLICATION + ; + +replicationPolicy + : replication '=' tableReplicationPolicy + ; + +tableReplicationPolicy + : '(' replicationPolicyClause (',' replicationPolicyClause)* ')' + ; + +replicationPolicyClause + : '{' replicationPolicyClusterClause (',' replicationPolicyIntervalClause)? '}' + ; + +replicationPolicyClusterClause + : DESTINATION ':' STRING + ; + +replicationPolicyIntervalClause + : INTERVAL ':' RETENTION_HOUR + | INTERVAL ':' RETENTION_DAY + ; + +columnRetentionPolicyPatternClause + : WHERE retentionColumnPatternClause + ; + +columnNameClause + : COLUMN identifier + ; + +retentionColumnPatternClause + : PATTERN '=' STRING + ; + +duration + : RETENTION_DAY + | RETENTION_YEAR + | RETENTION_MONTH + | RETENTION_HOUR + ; + +RETENTION_DAY + : POSITIVE_INTEGER 'D' + ; + +RETENTION_YEAR + : POSITIVE_INTEGER 'Y' + ; + +RETENTION_MONTH + : POSITIVE_INTEGER 'M' + ; + +RETENTION_HOUR + : POSITIVE_INTEGER 'H' + ; + +columnPolicy + : TAG '=' multiTagIdentifier + | TAG '=' '(' NONE ')' + ; + +multiTagIdentifier + : '(' policyTag (',' policyTag)* ')' + ; + +policyTag + : PII | HC + ; + +historyPolicy + : HISTORY maxAge? versions? + ; + +maxAge + : MAX_AGE'='duration + ; + +versions + : VERSIONS'='POSITIVE_INTEGER + ; + +ALTER: 'ALTER'; +TABLE: 'TABLE'; +SET: 'SET'; +UNSET: 'UNSET'; +POLICY: 'POLICY'; +RETENTION: 'RETENTION'; +REPLICATION: 'REPLICATION'; +HISTORY: 'HISTORY'; +SHARING: 'SHARING'; +GRANT: 'GRANT'; +REVOKE: 'REVOKE'; +ON: 'ON'; +TO: 'TO'; +FROM: 'FROM'; +SELECT: 'SELECT'; +DESCRIBE: 'DESCRIBE'; +GRANT_REVOKE: 'MANAGE GRANTS'; +CREATE_TABLE: 'CREATE TABLE'; +DATABASE: 'DATABASE'; +SHOW: 'SHOW'; +GRANTS: 'GRANTS'; +PATTERN: 'PATTERN'; +DESTINATION: 'DESTINATION'; +INTERVAL: 'INTERVAL'; +WHERE: 'WHERE'; +COLUMN: 'COLUMN'; +PII: 'PII'; +HC: 'HC'; +MODIFY: 'MODIFY'; +TAG: 'TAG'; +NONE: 'NONE'; +VERSIONS: 'VERSIONS'; +MAX_AGE: 'MAX_AGE'; +VACUUM: 'VACUUM'; +REMOVE: 'REMOVE'; +ORPHAN: 'ORPHAN'; +FILES: 'FILES'; +RETAIN: 'RETAIN'; +HOURS: 'HOURS'; +OPTIMIZE: 'OPTIMIZE'; +FULL: 'FULL'; +REWRITE: 'REWRITE'; +MANIFESTS: 'MANIFESTS'; +ANALYZE: 'ANALYZE'; +COMPUTE: 'COMPUTE'; +CLUSTERING: 'CLUSTERING'; +QUALITY: 'QUALITY'; + +POSITIVE_INTEGER + : DIGIT+ + ; + +STRING + : '\'' ( ~('\''|'\\') | ('\\' .) )* '\'' + | '"' ( ~('"'|'\\') | ('\\' .) )* '"' + ; + +IDENTIFIER + : (LETTER | DIGIT | '_')+ + ; + +BACKQUOTED_IDENTIFIER + : '`' ( ~'`' | '``' )* '`' + ; + +fragment DIGIT + : [0-9] + ; + +fragment LETTER + : [A-Z] + ; + +SIMPLE_COMMENT + : '--' ('\\\n' | ~[\r\n])* '\r'? '\n'? -> channel(HIDDEN) + ; + +BRACKETED_COMMENT + : '/*' {!isHint()}? (BRACKETED_COMMENT|.)*? '*/' -> channel(HIDDEN) + ; + +WS + : [ \r\n\t]+ -> channel(HIDDEN) + ; \ No newline at end of file diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/java/com/linkedin/openhouse/spark/OpenHouseCatalog.java b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/java/com/linkedin/openhouse/spark/OpenHouseCatalog.java new file mode 100644 index 000000000..69859fd13 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/java/com/linkedin/openhouse/spark/OpenHouseCatalog.java @@ -0,0 +1,18 @@ +package com.linkedin.openhouse.spark; + +/** + * Catalog implementation to create, read, update and delete tables in OpenHouse. This class + * leverages Openhouse tableclient to perform CRUD operations on Tables resource in the Catalog + * service. This implementation provides client side catalog implementation for Iceberg tables in + * Spark. + * + *

Catalog can be instantiated as a Iceberg catalog, with following configurations: + * spark.sql.catalog.openhouse=org.apache.iceberg.spark.SparkCatalog + * spark.sql.catalog.openhouse.catalog-impl=com.linkedin.openhouse.spark.OpenHouseCatalog + * spark.sql.catalog.openhouse.metrics-reporter-impl=com.linkedin.openhouse.javaclient.OpenHouseMetricsReporter + * spark.sql.catalog.openhouse.uri=http://[openhouse service host]:[openhouse service port] + * spark.sql.catalog.openhouse.cluster=[openhouse cluster name] + * + *

It can be used in spark shell as follows: spark.sql("USE openhouse") + */ +public class OpenHouseCatalog extends com.linkedin.openhouse.javaclient.OpenHouseCatalog {} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/extensions/OpenhouseSparkSessionExtensions.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/extensions/OpenhouseSparkSessionExtensions.scala new file mode 100644 index 000000000..c8d911dc2 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/extensions/OpenhouseSparkSessionExtensions.scala @@ -0,0 +1,12 @@ +package com.linkedin.openhouse.spark.extensions + +import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseSparkSqlExtensionsParser +import com.linkedin.openhouse.spark.sql.execution.datasources.v2.OpenhouseDataSourceV2Strategy +import org.apache.spark.sql.SparkSessionExtensions + +class OpenhouseSparkSessionExtensions extends (SparkSessionExtensions => Unit) { + override def apply(extensions: SparkSessionExtensions): Unit = { + extensions.injectParser { case (_, parser) => new OpenhouseSparkSqlExtensionsParser(parser) } + extensions.injectPlannerStrategy( spark => OpenhouseDataSourceV2Strategy(spark)) + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/constants/Principal.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/constants/Principal.scala new file mode 100644 index 000000000..7307ef5a8 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/constants/Principal.scala @@ -0,0 +1,18 @@ +package com.linkedin.openhouse.spark.sql.catalyst.constants + +/** + * This object is used to represent keyword global user group "PUBLIC" which maps to the acl policy representation "*" + */ +object Principal { + private val GLOBAL_USER_GROUP = "PUBLIC" + private val GLOBAL_USER_GROUP_ACL = "*" + def apply(principal: String): String = principal toUpperCase() match { + case GLOBAL_USER_GROUP => GLOBAL_USER_GROUP_ACL + case _ => principal + } + + def unapply(principalAcl: String): Option[String] = principalAcl match { + case GLOBAL_USER_GROUP_ACL => Some(GLOBAL_USER_GROUP) + case _ => Some(principalAcl) + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/enums/GrantableResourceTypes.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/enums/GrantableResourceTypes.scala new file mode 100644 index 000000000..88738993f --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/enums/GrantableResourceTypes.scala @@ -0,0 +1,6 @@ +package com.linkedin.openhouse.spark.sql.catalyst.enums + +private[sql] object GrantableResourceTypes extends Enumeration { + type GrantableResourceType = Value + val TABLE, DATABASE = Value +} 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..3ecf8bbe0 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 @@ -14,9 +14,26 @@ import org.apache.spark.sql.types.{DataType, StructType} import java.util.Locale -class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends ParserInterface { +import org.apache.iceberg.spark.ExtendedParser +import org.apache.iceberg.spark.ExtendedParser.RawOrderField + +class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends ParserInterface + with ExtendedParser { private lazy val astBuilder = new OpenhouseSqlExtensionsAstBuilder(delegate) + // Iceberg procedures that take a sort order (e.g. rewrite_data_files with strategy => 'sort') + // cast the session's outermost parser to Iceberg's ExtendedParser to parse the order string. + // This wrapper is outermost, so it must implement ExtendedParser and delegate to the underlying + // Iceberg parser; otherwise those procedures fail with "parser is not an Iceberg ExtendedParser". + override def parseSortOrder(sqlText: String): java.util.List[RawOrderField] = { + delegate match { + case extended: ExtendedParser => extended.parseSortOrder(sqlText) + case _ => + throw new UnsupportedOperationException( + "Parsing a sort order requires the Iceberg SQL extensions to be enabled") + } + } + override def parsePlan(sqlText: String): LogicalPlan = { if (isOpenhouseCommand(sqlText)) { parse(sqlText) { parser => astBuilder.visit(parser.singleStatement()) }.asInstanceOf[LogicalPlan] @@ -72,7 +89,11 @@ 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") || + normalized.startsWith("optimize") || + (normalized.startsWith("analyze table") && + normalized.contains("compute clustering quality")) } 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 new file mode 100644 index 000000000..6d3627180 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala @@ -0,0 +1,226 @@ +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.{AnalyzeClusteringQuality, GrantRevokeStatement, OptimizeTable, 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 +import org.apache.spark.sql.catalyst.parser.ParserInterface +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan + +import scala.collection.JavaConversions.iterableAsScalaIterable +import scala.collection.JavaConverters._ + +class OpenhouseSqlExtensionsAstBuilder (delegate: ParserInterface) extends OpenhouseSqlExtensionsBaseVisitor[AnyRef] { + override def visitSingleStatement(ctx: SingleStatementContext): LogicalPlan = { + typedVisit[LogicalPlan](ctx.statement) + } + + override def visitSetRetentionPolicy(ctx: SetRetentionPolicyContext): SetRetentionPolicy = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val (granularity, count) = typedVisit[(String, Int)](ctx.retentionPolicy()) + val (colName, colPattern) = + if (ctx.columnRetentionPolicy() != null) + typedVisit[(String, String)](ctx.columnRetentionPolicy()) + else (null, null) + SetRetentionPolicy(tableName, granularity, count, Option(colName), Option(colPattern)) + } + + override def visitSetReplicationPolicy(ctx: SetReplicationPolicyContext): SetReplicationPolicy = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val replicationPolicies = typedVisit[Seq[(String, Option[String])]](ctx.replicationPolicy()) + SetReplicationPolicy(tableName, replicationPolicies) + } + + override def visitUnSetReplicationPolicy(ctx: UnSetReplicationPolicyContext): UnSetReplicationPolicy = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val replicationPolicies = typedVisit[String](ctx.replication()) + UnSetReplicationPolicy(tableName, replicationPolicies) + } + + override def visitSetSharingPolicy(ctx: SetSharingPolicyContext): SetSharingPolicy = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val sharing = typedVisit[String](ctx.sharingPolicy()) + SetSharingPolicy(tableName, sharing) + } + + override def visitSetColumnPolicyTag(ctx: SetColumnPolicyTagContext): SetColumnPolicyTag = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val colName = ctx.columnNameClause().identifier().getText + val policyTags = typedVisit[Seq[String]](ctx.columnPolicy()) + SetColumnPolicyTag(tableName, colName, policyTags) + } + + override def visitGrantStatement(ctx: GrantStatementContext): GrantRevokeStatement = { + val (resourceType, resourceName) = typedVisit[(GrantableResourceType, Seq[String])](ctx.grantableResource()) + val principal = typedVisit[String](ctx.principal) + val privilege = typedVisit[String](ctx.privilege) + GrantRevokeStatement(isGrant = true, resourceType, resourceName, privilege, principal) + } + + override def visitRevokeStatement(ctx: RevokeStatementContext): GrantRevokeStatement = { + val (resourceType, resourceName) = typedVisit[(GrantableResourceType, Seq[String])](ctx.grantableResource()) + val privilege = typedVisit[String](ctx.privilege) + val principal = typedVisit[String](ctx.principal) + GrantRevokeStatement(isGrant = false, resourceType, resourceName, privilege, principal) + } + + override def visitShowGrantsStatement(ctx: ShowGrantsStatementContext): ShowGrantsStatement = { + val (resourceType, resourceName) = typedVisit[(GrantableResourceType, Seq[String])](ctx.grantableResource()) + ShowGrantsStatement(resourceType, resourceName) + } + + override def visitPrincipal(ctx: PrincipalContext): String = { + ctx.getText + } + + override def visitPrivilege(ctx: PrivilegeContext): String = { + ctx.getText.toUpperCase + } + + override def visitGrantableResource(ctx: GrantableResourceContext): (GrantableResourceType, Seq[String]) = { + val resourceName = typedVisit[Seq[String]](ctx.multipartIdentifier()) + val resourceType = if (ctx.DATABASE != null) { + GrantableResourceTypes.DATABASE + } else if (ctx.TABLE != null) { + GrantableResourceTypes.TABLE + } else { + throw new IllegalStateException("Unrecognized grantable resource: " + ctx.getText) + } + (resourceType, resourceName) + } + + override def visitMultipartIdentifier(ctx: MultipartIdentifierContext): Seq[String] = { + toSeq(ctx.parts).map(_.getText) + } + + override def visitRetentionPolicy(ctx: RetentionPolicyContext): (String, Int) = { + typedVisit[(String, Int)](ctx.duration()) + } + + override def visitReplicationPolicy(ctx: ReplicationPolicyContext): Seq[(String, Option[String])] = { + typedVisit[Seq[(String, Option[String])]](ctx.tableReplicationPolicy()) + } + + override def visitTableReplicationPolicy(ctx: TableReplicationPolicyContext): Seq[(String, Option[String])] = { + toSeq(ctx.replicationPolicyClause()).map(typedVisit[(String, Option[String])]) + } + + override def visitReplicationPolicyClause(ctx: ReplicationPolicyClauseContext): (String, Option[String]) = { + val cluster = typedVisit[String](ctx.replicationPolicyClusterClause()) + val interval = if (ctx.replicationPolicyIntervalClause() != null) + typedVisit[String](ctx.replicationPolicyIntervalClause()) + else + null + (cluster, Option(interval)) + } + + override def visitReplicationPolicyClusterClause(ctx: ReplicationPolicyClusterClauseContext): (String) = { + ctx.STRING().getText + } + + override def visitReplicationPolicyIntervalClause(ctx: ReplicationPolicyIntervalClauseContext): (String) = { + if (ctx.RETENTION_HOUR() != null) + ctx.RETENTION_HOUR().getText.toUpperCase() + else ctx.RETENTION_DAY().getText.toUpperCase() + } + + override def visitColumnRetentionPolicy(ctx: ColumnRetentionPolicyContext): (String, String) = { + if (ctx.columnRetentionPolicyPatternClause() != null) { + (ctx.columnNameClause().identifier().getText(), ctx.columnRetentionPolicyPatternClause().retentionColumnPatternClause().STRING().getText) + } else { + (ctx.columnNameClause().identifier().getText(), new String()) + } + } + + override def visitColumnRetentionPolicyPatternClause(ctx: ColumnRetentionPolicyPatternClauseContext): String = { + ctx.retentionColumnPatternClause().STRING().getText + } + + override def visitReplication(ctx: ReplicationContext): String = + { + ctx.REPLICATION().getText + } + + override def visitSharingPolicy(ctx: SharingPolicyContext): String = { + ctx.BOOLEAN().getText + } + + override def visitColumnPolicy(ctx: ColumnPolicyContext): Seq[String] = { + if (ctx.NONE() == null) { + typedVisit[Seq[String]](ctx.multiTagIdentifier()); + } else { + Seq.empty + } + } + + override def visitMultiTagIdentifier(ctx: MultiTagIdentifierContext): Seq[String] = { + toSeq(ctx.policyTag()).map(_.getText) + } + + override def visitDuration(ctx: DurationContext): (String, Int) = { + val granularity: String = if (ctx.RETENTION_DAY != null) { + TimePartitionSpec.GranularityEnum.DAY.getValue() + } else if (ctx.RETENTION_YEAR() != null) { + TimePartitionSpec.GranularityEnum.YEAR.getValue() + } else if (ctx.RETENTION_MONTH() != null) { + TimePartitionSpec.GranularityEnum.MONTH.getValue() + } else { + TimePartitionSpec.GranularityEnum.HOUR.getValue() + } + val count = ctx.getText.substring(0, ctx.getText.length - 1).toInt + (granularity, count) + } + + override def visitSetHistoryPolicy(ctx: SetHistoryPolicyContext): SetHistoryPolicy = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val (granularity, maxAge, versions) = typedVisit[(Option[String], Int, Int)](ctx.historyPolicy()) + SetHistoryPolicy(tableName, granularity, maxAge, versions) + } + override def visitHistoryPolicy(ctx: HistoryPolicyContext): (Option[String], Int, Int) = { + val maxAgePolicy = if (ctx.maxAge() != null) + typedVisit[(String, Int)](ctx.maxAge().duration()) + else (null, -1) + val versionPolicy = if (ctx.versions() != null) + typedVisit[Int](ctx.versions()) + else -1 + if (maxAgePolicy._2 == -1 && versionPolicy == -1) { + throw new OpenhouseParseException("At least one of MAX_AGE or VERSIONS must be specified in HISTORY policy, e.g. " + + "ALTER TABLE openhouse.db.table SET POLICY (HISTORY MAX_AGE=2D) or ALTER TABLE openhouse.db.table SET POLICY (HISTORY VERSIONS=3)", + ctx.start.getLine, ctx.start.getCharPositionInLine) + } + (Option(maxAgePolicy._1), maxAgePolicy._2, versionPolicy) + } + + override def visitVersions(ctx: VersionsContext): Integer = { + 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) + } + + override def visitOptimizeTable(ctx: OptimizeTableContext): OptimizeTable = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + val full = ctx.FULL() != null + val rewriteManifests = ctx.MANIFESTS() != null + OptimizeTable(tableName, full, rewriteManifests) + } + + override def visitAnalyzeClusteringQuality( + ctx: AnalyzeClusteringQualityContext): AnalyzeClusteringQuality = { + val tableName = typedVisit[Seq[String]](ctx.multipartIdentifier) + AnalyzeClusteringQuality(tableName) + } + + private def toBuffer[T](list: java.util.List[T]) = list.asScala + private def toSeq[T](list: java.util.List[T]) = toBuffer(list).toSeq + + private def typedVisit[T](ctx: ParseTree): T = { + ctx.accept(this).asInstanceOf[T] + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/AnalyzeClusteringQuality.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/AnalyzeClusteringQuality.scala new file mode 100644 index 000000000..04145515d --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/AnalyzeClusteringQuality.scala @@ -0,0 +1,25 @@ +package com.linkedin.openhouse.spark.sql.catalyst.plans.logical + +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.catalyst.plans.logical.LeafCommand +import org.apache.spark.sql.types.StringType + +/** + * The logical plan of `ANALYZE TABLE t COMPUTE CLUSTERING QUALITY`. + * + * A read-only probe (no commit, no property write) that reports how well a table is clustered to + * its CURRENT key selection, using only what OPTIMIZE persists (`optimize.cluster.state`) plus + * manifest metrics (`t.files`). Output rows: `(metric, dimension, value)` where `dimension` is set + * only for per-key depth rows. + */ +case class AnalyzeClusteringQuality(tableName: Seq[String]) extends LeafCommand { + + override lazy val output: Seq[Attribute] = Seq( + AttributeReference("metric", StringType, nullable = false)(), + AttributeReference("dimension", StringType, nullable = true)(), + AttributeReference("value", StringType, nullable = false)()) + + override def simpleString(maxFields: Int): String = { + s"AnalyzeClusteringQuality: ${tableName}" + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTable.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTable.scala new file mode 100644 index 000000000..eed38cc45 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTable.scala @@ -0,0 +1,150 @@ +package com.linkedin.openhouse.spark.sql.catalyst.plans.logical + +import java.nio.charset.StandardCharsets +import java.util.Locale +import java.util.zip.CRC32 + +import scala.util.control.NonFatal + +import com.fasterxml.jackson.annotation.JsonInclude +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule} + +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.catalyst.plans.logical.LeafCommand +import org.apache.spark.sql.types.StringType + +/** + * The logical plan of the OPTIMIZE command: + * {{{ + * OPTIMIZE multi_part_name [FULL] [REWRITE MANIFESTS] + * }}} + * + * Behavior depends on whether clustering keys are configured via the `optimize.cluster.*` table + * properties: + * + * - '''No `optimize.cluster.keys`''': plain bin-pack compaction (`system.rewrite_data_files` with + * defaults). Historical behavior, unaffected by `FULL`. + * - '''Clustering configured''': a sort / z-order rewrite of the configured keys, incremental by + * default (only the forward slice of the leading key since the last run, tracked by an + * `optimize.cluster.hwm-snapshot-id` watermark); `FULL` reclusters up to the age floor. + * + * `REWRITE MANIFESTS` (`system.rewrite_manifests`) is independent and runs after the data rewrite. + * Snapshot expiration is intentionally not part of OPTIMIZE -- that is the VACUUM command's job. + */ +case class OptimizeTable(tableName: Seq[String], full: Boolean, rewriteManifests: Boolean) + extends LeafCommand { + + override lazy val output: Seq[Attribute] = Seq( + AttributeReference("metric", StringType, nullable = false)(), + AttributeReference("value", StringType, nullable = false)()) + + override def simpleString(maxFields: Int): String = { + s"OptimizeTable: ${tableName} full=${full} rewriteManifests=${rewriteManifests}" + } +} + +object OptimizeTable { + + val KEYS_PROP = "optimize.cluster.keys" + val SORT_MODE_PROP = "optimize.cluster.sort-mode" + val MIN_SNAPSHOT_AGE_PROP = "optimize.cluster.min-snapshot-age-minutes" + val HWM_PROP = "optimize.cluster.hwm-snapshot-id" + val MAX_COMMITS_PROP = "optimize.cluster.max-commits" + val CONFIG_ID_PROP = "optimize.cluster.config-id" + val STATE_PROP = "optimize.cluster.state" + + val DEFAULT_SORT_MODE = "zorder" + val DEFAULT_MIN_SNAPSHOT_AGE_MINUTES = 30L + val DEFAULT_MAX_COMMITS = 10L + + /** + * Clustering configuration resolved from the `optimize.cluster.*` table properties, with every + * default applied and every value parsed to its type. + */ + case class ClusterConfig( + keys: Seq[String], + sortMode: String, + minAgeMinutes: Long, + maxCommits: Long, + hwm: Option[Long], + state: Seq[ClusterInterval]) + + /** Parse the `optimize.cluster.*` table properties into a typed [[ClusterConfig]]. */ + def parseClusterConfig(props: Map[String, String]): ClusterConfig = ClusterConfig( + keys = props.get(KEYS_PROP) + .map(_.split(",").map(_.trim).filter(_.nonEmpty).toSeq).getOrElse(Seq.empty), + sortMode = props.getOrElse(SORT_MODE_PROP, DEFAULT_SORT_MODE), + minAgeMinutes = props.get(MIN_SNAPSHOT_AGE_PROP).map(_.toLong) + .getOrElse(DEFAULT_MIN_SNAPSHOT_AGE_MINUTES), + maxCommits = props.get(MAX_COMMITS_PROP).map(_.toLong).getOrElse(DEFAULT_MAX_COMMITS), + hwm = props.get(HWM_PROP).map(_.toLong), + state = parseState(props.getOrElse(STATE_PROP, ""))) + + val stateMapper = { + val mapper = new ObjectMapper() with ClassTagExtensions + mapper.registerModule(DefaultScalaModule) + // Omit an absent `lower` (None) so the persisted JSON stays compact and stable. + mapper.setSerializationInclusion(JsonInclude.Include.NON_ABSENT) + mapper + } + + /** + * One clustered leading-key interval `(lower, upper]` under a specific key selection (`config`). + * `lower = None` means unbounded below (a FULL / first backfill). Persisted, alongside the + * watermark, in the `optimize.cluster.state` table property so it survives snapshot expiration. + */ + case class ClusterInterval( + config: String, keys: String, mode: String, lower: Option[String], upper: String) + + /** Stable, compact identity of a key selection: only a keys/mode change produces a new id. */ + def configId(keys: Seq[String], sortMode: String): String = { + val normalized = keys.map(_.trim).mkString(",") + "|" + sortMode.toLowerCase(Locale.ROOT) + val crc = new CRC32() + crc.update(normalized.getBytes(StandardCharsets.UTF_8)) + java.lang.Long.toHexString(crc.getValue) + } + + /** + * Parse interval state. Empty / absent input is no state (a fresh table). Non-empty but + * unparseable input is a corrupted property, not "no state" -- silently treating it as empty + * would make OPTIMIZE recluster from scratch and mis-report ANALYZE coverage, so it fails loudly + * naming the property and how to clear it. + */ + def parseState(json: String): Seq[ClusterInterval] = { + if (json == null || json.trim.isEmpty) return Seq.empty + try { + stateMapper.readValue[Seq[ClusterInterval]](json) + } catch { + case NonFatal(e) => + throw new IllegalStateException( + s"Malformed clustering state in table property '$STATE_PROP'; OPTIMIZE cannot tell " + + s"what is already clustered. Clear the clustering metadata and let the next OPTIMIZE " + + s"rebuild it: ALTER TABLE

UNSET TBLPROPERTIES " + + s"('$STATE_PROP', '$HWM_PROP', '$CONFIG_ID_PROP'). Value was: $json", e) + } + } + + /** + * Fold a completed run into the interval state. A same-config incremental run extends the current + * epoch's upper bound (keeping its lower); a config change appends a new epoch, retains the old + * ones (durable key-selection history); FULL collapses the current config to one unbounded-below + * interval. + */ + def advanceState( + existing: Seq[ClusterInterval], + cfgId: String, + keys: Seq[String], + mode: String, + lower: Option[String], + upper: String, + full: Boolean): Seq[ClusterInterval] = { + val keysStr = keys.mkString(",") + val others = existing.filterNot(_.config == cfgId) + (full, existing.find(_.config == cfgId)) match { + case (true, _) => others :+ ClusterInterval(cfgId, keysStr, mode, None, upper) + case (false, Some(cur)) => others :+ cur.copy(keys = keysStr, mode = mode, upper = upper) + case (false, None) => existing :+ ClusterInterval(cfgId, keysStr, mode, lower, upper) + } + } +} 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..edb7a596c --- /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,26 @@ +package com.linkedin.openhouse.spark.sql.catalyst.plans.logical + +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.catalyst.plans.logical.LeafCommand +import org.apache.spark.sql.types.StringType + +/** + * The logical plan of the VACUUM command: + * {{{ + * VACUUM multi_part_name [REMOVE ORPHAN FILES] [RETAIN n HOURS] + * }}} + * + * Reports the retention windows it resolved, so an operator can see which one each step actually + * used -- an explicit `RETAIN`, the table's `policies.history`, or the maintenance job's default. + */ +case class VacuumTable(tableName: Seq[String], removeOrphanFiles: Boolean, retainHours: Option[Int]) + extends LeafCommand { + + override lazy val output: Seq[Attribute] = Seq( + AttributeReference("metric", StringType, nullable = false)(), + AttributeReference("value", StringType, nullable = false)()) + + 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/AnalyzeClusteringQualityExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExec.scala new file mode 100644 index 000000000..35fcfd451 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExec.scala @@ -0,0 +1,215 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import scala.collection.JavaConverters._ +import scala.collection.mutable + +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.OptimizeTable +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, GenericInternalRow, Literal} +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 +import org.apache.spark.sql.functions.col +import org.apache.spark.unsafe.types.UTF8String + +/** + * Read-only probe for `ANALYZE TABLE t COMPUTE CLUSTERING QUALITY`: reports how well a table is + * clustered to its current key selection, using only what OPTIMIZE persists + * (`optimize.cluster.state`) plus manifest metrics (`t.files`). All metrics are computed with + * distributed SQL over metadata (an aggregate for coverage, a windowed sweep for depth), so the + * command is safe on tables with very large file counts. No commit and no property write. + */ +case class AnalyzeClusteringQualityExec( + output: Seq[Attribute], + spark: SparkSession, + catalog: TableCatalog, + ident: Identifier) extends LeafV2CommandExec { + + import AnalyzeClusteringQualityExec._ + import OptimizeTable.{KEYS_PROP, SORT_MODE_PROP, STATE_PROP, HWM_PROP, DEFAULT_SORT_MODE, + configId, parseState} + + private def outRow(metric: String, dimension: String, value: String): InternalRow = + new GenericInternalRow(Array[Any]( + UTF8String.fromString(metric), + if (dimension == null) null else UTF8String.fromString(dimension), + UTF8String.fromString(value))) + + override protected def run(): Seq[InternalRow] = { + val props = catalog.loadTable(ident) match { + case iceberg: SparkTable if iceberg.table().properties().containsKey("openhouse.tableId") => + iceberg.table().properties().asScala.toMap + case table => + throw new UnsupportedOperationException( + s"Cannot compute clustering quality for non-Openhouse table: $table") + } + + val cat = quoteIfNeeded(catalog.name()) + val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") + val qualifiedTableName = s"$cat.$tableArg" + + val keys = props.get(KEYS_PROP) + .map(_.split(",").map(_.trim).filter(_.nonEmpty).toSeq).getOrElse(Seq.empty) + + val out = mutable.ArrayBuffer[InternalRow]() + + if (keys.isEmpty) { + out += outRow("clustering_configured", null, "false") + return out.toSeq + } + out += outRow("clustering_configured", null, "true") + + val sortMode = props.getOrElse(SORT_MODE_PROP, DEFAULT_SORT_MODE) + val cfgId = configId(keys, sortMode) + out += outRow("config_id", null, cfgId) + out += outRow("keys", null, keys.mkString(",")) + out += outRow("sort_mode", null, sortMode) + + val leadKey = keys.head + val leadType = spark.table(qualifiedTableName).schema(leadKey).dataType.sql + val current = parseState(props.getOrElse(STATE_PROP, "")).filter(_.config == cfgId) + + // Coverage: a file is covered iff its leading-key range fits inside a current-config interval. + // Computed as one aggregate over manifest metrics -- no per-file collect. + val leadLo = metricExpr(leadKey, "lower_bound") + val leadHi = metricExpr(leadKey, "upper_bound") + val coveredExpr = coveragePredicate(leadLo, leadHi, current, leadType) + val a = spark.sql( + s"""SELECT count(*) AS files_total, + | coalesce(sum(file_size_in_bytes), 0) AS bytes_total, + | coalesce(sum(CASE WHEN cov THEN 1 ELSE 0 END), 0) AS files_covered, + | coalesce(sum(CASE WHEN cov THEN file_size_in_bytes ELSE 0 END), 0) AS bytes_covered, + | coalesce(sum(CASE WHEN lead_null THEN file_size_in_bytes ELSE 0 END), 0) AS null_bytes + |FROM (SELECT file_size_in_bytes, + | coalesce($coveredExpr, false) AS cov, + | ($leadLo IS NULL OR $leadHi IS NULL) AS lead_null + | FROM $qualifiedTableName.files)""".stripMargin).collect().head + val filesTotal = a.getLong(0) + val bytesTotal = a.getLong(1) + val filesCovered = a.getLong(2) + val bytesCovered = a.getLong(3) + val nullBytes = a.getLong(4) + out += outRow("files_total", null, filesTotal.toString) + out += outRow("files_covered", null, filesCovered.toString) + out += outRow("bytes_total", null, bytesTotal.toString) + out += outRow("bytes_covered", null, bytesCovered.toString) + out += outRow("coverage_bytes_pct", null, pct(bytesCovered, bytesTotal)) + out += outRow("coverage_files_pct", null, pct(filesCovered, filesTotal)) + out += outRow("null_bound_bytes_pct", null, pct(nullBytes, bytesTotal)) + + // Depth per clustering dimension: global and over the covered region only (the SLA input). + // Each is a windowed stabbing-count sweep over metadata, kept off the driver. + keys.foreach { k => + val kLo = metricExpr(k, "lower_bound") + val kHi = metricExpr(k, "upper_bound") + val g = depthStats(spark, qualifiedTableName, kLo, kHi, None) + val c = depthStats(spark, qualifiedTableName, kLo, kHi, Some(coveredExpr)) + out += outRow("depth_avg", k, fmt(g.avg)) + out += outRow("depth_p90", k, fmt(g.p90)) + out += outRow("depth_max", k, g.max.toString) + out += outRow("depth_avg_covered", k, fmt(c.avg)) + out += outRow("depth_p90_covered", k, fmt(c.p90)) + } + + val tail = tailHours(spark, qualifiedTableName, props.get(HWM_PROP)) + out += outRow("unclustered_tail_hours", null, tail) + out += outRow("state", null, props.getOrElse(STATE_PROP, "[]")) + out.toSeq + } + + override def simpleString(maxFields: Int): String = { + s"AnalyzeClusteringQualityExec: ${catalog} ${ident}" + } +} + +object AnalyzeClusteringQualityExec { + + final case class DepthStats(avg: Double, p90: Double, max: Long) + + /** SQL access to a per-file column metric, e.g. `readable_metrics.`ts`.lower_bound`. */ + def metricExpr(key: String, field: String): String = + s"readable_metrics.${quoteIfNeeded(key)}.$field" + + /** + * SQL boolean: the leading-key range `[lo, hi]` fits inside some current-config interval + * `(lower, upper]`. Interval bounds are CAST to the leading-key type; the value is embedded as a + * Catalyst literal so quotes survive. `null` bounds make the expression `null` (-> uncovered via + * the caller's `coalesce(..., false)`). + */ + def coveragePredicate( + loExpr: String, + hiExpr: String, + intervals: Seq[OptimizeTable.ClusterInterval], + castType: String): String = { + if (intervals.isEmpty) return "false" + intervals.map { iv => + val upper = s"($hiExpr <= CAST(${Literal(iv.upper).sql} AS $castType))" + val lower = iv.lower match { + case Some(lo) => s"($loExpr > CAST(${Literal(lo).sql} AS $castType))" + case None => "true" + } + s"($upper AND $lower)" + }.mkString(" OR ") + } + + private def pct(part: Long, total: Long): String = + if (total == 0) "0.0" else fmt(100.0 * part / total) + + private def fmt(d: Double): String = f"$d%.2f" + + /** + * Stabbing-depth stats over the `[lower, upper]` intervals of one dimension, optionally restricted + * to the covered region. Computed with a windowed running-sum sweep in SQL (`+1` at each lower + * bound, `-1` past each upper, sampled at start events) so nothing is collected to the driver. + * Depth `1` means no overlap (perfectly clustered); higher means more interleaving. + */ + private def depthStats( + spark: SparkSession, + qualifiedTableName: String, + loExpr: String, + hiExpr: String, + coveredFilter: Option[String]): DepthStats = { + val extra = coveredFilter.map(f => s"AND coalesce($f, false)").getOrElse("") + val where = s"$loExpr IS NOT NULL AND $hiExpr IS NOT NULL $extra" + val q = + s"""WITH ev AS ( + | SELECT $loExpr AS pt, 1 AS delta FROM $qualifiedTableName.files WHERE $where + | UNION ALL + | SELECT $hiExpr AS pt, -1 AS delta FROM $qualifiedTableName.files WHERE $where + |), + |running AS (SELECT delta, sum(delta) OVER (ORDER BY pt, delta DESC) AS depth FROM ev) + |SELECT coalesce(avg(CASE WHEN delta = 1 THEN CAST(depth AS DOUBLE) END), 0.0), + | coalesce(percentile_approx( + | CASE WHEN delta = 1 THEN CAST(depth AS DOUBLE) END, 0.9), 0.0), + | coalesce(max(depth), 0L) + |FROM running""".stripMargin + val r = spark.sql(q).collect().head + DepthStats(r.getDouble(0), r.getDouble(1), r.getLong(2)) + } + + /** + * Age in hours of the oldest not-yet-clustered data: the oldest non-replace snapshot committed + * after the watermark snapshot. `0` if nothing is newer than the watermark; `unknown` if the + * watermark is unset or has been expired (so an SLA breach is never hidden). + */ + private def tailHours( + spark: SparkSession, qualifiedTableName: String, hwm: Option[String]): String = { + hwm match { + case None => "unknown" + case Some(h) => + val floor = spark.table(s"$qualifiedTableName.snapshots") + .where(col("snapshot_id") === h.toLong).select("committed_at").collect() + if (floor.isEmpty) return "unknown" // expired watermark + val rows = spark.sql( + s"""SELECT CAST((unix_timestamp(current_timestamp()) - + | unix_timestamp(min(committed_at))) / 3600.0 AS DOUBLE) + |FROM $qualifiedTableName.snapshots + |WHERE operation != 'replace' + | AND committed_at > (SELECT committed_at FROM $qualifiedTableName.snapshots + | WHERE snapshot_id = $h)""".stripMargin).collect() + if (rows.isEmpty || rows.head.isNullAt(0)) "0.0" else fmt(rows.head.getDouble(0)) + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenanceProperties.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenanceProperties.scala new file mode 100644 index 000000000..108c2fcd2 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenanceProperties.scala @@ -0,0 +1,184 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.time.Duration +import java.time.temporal.ChronoUnit +import java.util.Locale + +import scala.util.control.NonFatal + +import com.fasterxml.jackson.databind.ObjectMapper + +/** + * The table-property contract shared by the interactive maintenance DDL and the scheduled + * OpenHouse maintenance jobs. + * + * The jobs and the Spark SQL extensions ship as separate artifacts and cannot share code, so the + * keys and defaults below are mirrored from the job sources and pinned by + * `MaintenancePropertiesTest`: + * + * - snapshot expiration (SE) is policy-driven -- `apps/spark/.../jobs/scheduler/tasks/ + * TableSnapshotsExpirationTask.java` reads `policies.history` and passes it to + * `SnapshotsExpirationSparkApp` / `Operations.expireSnapshots`; + * - orphan-file deletion (OFD) is not policy-driven -- `apps/spark/.../jobs/spark/ + * OrphanFilesDeletionSparkApp.java` reads raw table properties (see `AppConstants.java`); + * - `maintenance.*` is the user-settable channel the scheduler reads back as + * `jobExecutionProperties` (`apps/spark/.../jobs/client/TablesClient.java`), including the + * per-table and per-job-type disable switches (`TableMetadata.isMaintenanceJobDisabled`). + * + * Note that `openhouse.*` and `policies` are '''preserved''' keys: the /tables service rejects any + * attempt to set them with `ALTER TABLE ... SET TBLPROPERTIES`. Anything a user is expected to set + * therefore has to live outside those namespaces -- hence `maintenance.vacuum.enabled` rather than + * `openhouse.vacuum.enabled`, and `ALTER TABLE ... SET POLICY (HISTORY ...)` rather than a direct + * write to `policies`. + */ +object MaintenanceProperties { + + /** Present on every OpenHouse table; its absence means the table is not an OpenHouse table. */ + val TABLE_ID_PROP = "openhouse.tableId" + + /** Server-written JSON holding the table's OpenHouse policies, including `history`. */ + val POLICIES_PROP = "policies" + + /** Server-written table type. The scheduled SE job runs on primary tables only. */ + val TABLE_TYPE_PROP = "openhouse.tableType" + val REPLICA_TABLE_TYPE = "REPLICA_TABLE" + + /** Opts a table into the Alpha VACUUM command. Must not be `openhouse.`-prefixed (preserved). */ + val VACUUM_ENABLED_PROP = "maintenance.vacuum.enabled" + + /** OFD: forces a one-day orphan window when set to `true`. */ + val OFD_ONE_DAY_TTL_PROP = "ofd.one_day_ttl.enabled" + + /** OFD: when backups are on, the job moves orphans aside instead of deleting them. */ + val BACKUP_ENABLED_PROP = "retention.backup.enabled" + val BACKUP_DIR_PROP = "retention.backup.dir" + + /** Job types, as named by `JobConf.JobTypeEnum` in the disable switches. */ + val SNAPSHOTS_EXPIRATION_JOB = "SNAPSHOTS_EXPIRATION" + val ORPHAN_FILES_DELETION_JOB = "ORPHAN_FILES_DELETION" + val DATA_COMPACTION_JOB = "DATA_COMPACTION" + + /** `SnapshotsExpirationSparkApp.DEFAULT_CONFIGURATION`: a 3-day TTL is enforced even unset. */ + val DEFAULT_HISTORY_MAX_AGE = 3 + val DEFAULT_HISTORY_GRANULARITY = "DAY" + + /** + * `OrphanFilesDeletionSparkApp.createApp`: a 7-day default window, which that job additionally + * floors at one day because its window arrives as a CLI argument. Here the default is fixed, so + * the floor is only reachable through an explicit `RETAIN`, which is the operator's own call. + */ + val DEFAULT_ORPHAN_TTL: Duration = Duration.ofDays(7) + val ONE_DAY_ORPHAN_TTL: Duration = Duration.ofDays(1) + + private val policiesMapper = new ObjectMapper() + + /** + * The snapshot-expiration window a table is configured for. + * + * @param age snapshots older than this are expired + * @param versions retain at most this many snapshots, when the policy sets it (`versions > 0`) + * @param source where the window came from, for the command's output row + */ + case class SnapshotRetention(age: Duration, versions: Option[Int], source: String) + + /** + * Resolve the snapshot-expiration window exactly as the scheduled SE job does: from the table's + * `policies.history` (`maxAge` x `granularity`, plus `versions`), falling back to the job's own + * 3-day default when the table has no history policy. + */ + def snapshotRetention(props: Map[String, String]): SnapshotRetention = { + val history = policyNode(props, "history") + val maxAge = history.map(_.path("maxAge").asInt(0)).getOrElse(0) + val versions = history.map(_.path("versions").asInt(0)).getOrElse(0) + if (maxAge > 0) { + val granularity = history.map(_.path("granularity").asText(DEFAULT_HISTORY_GRANULARITY)) + .filter(_.nonEmpty).getOrElse(DEFAULT_HISTORY_GRANULARITY) + SnapshotRetention( + granularityUnit(granularity).getDuration.multipliedBy(maxAge.toLong), + Some(versions).filter(_ > 0), + s"$POLICIES_PROP.history ($maxAge ${granularity.toUpperCase(Locale.ROOT)})") + } else { + SnapshotRetention( + granularityUnit(DEFAULT_HISTORY_GRANULARITY).getDuration + .multipliedBy(DEFAULT_HISTORY_MAX_AGE.toLong), + Some(versions).filter(_ > 0), + s"default ($DEFAULT_HISTORY_MAX_AGE $DEFAULT_HISTORY_GRANULARITY)") + } + } + + /** + * Resolve the orphan-file window the way `OrphanFilesDeletionSparkApp` does: a 7-day default, + * dropped to one day by `ofd.one_day_ttl.enabled`. An explicit `RETAIN` is the operator's own + * call and is used as given, so the documented "lower it to handle an emergency" lever keeps + * working. + */ + def orphanRetention(props: Map[String, String], requested: Option[Int]): (Duration, String) = + requested match { + case Some(hours) => (Duration.ofHours(hours.toLong), "RETAIN") + case None if isEnabled(props, OFD_ONE_DAY_TTL_PROP) => + (ONE_DAY_ORPHAN_TTL, OFD_ONE_DAY_TTL_PROP) + case None => (DEFAULT_ORPHAN_TTL, "default") + } + + /** + * True when the platform has been told not to run maintenance on this table -- either wholesale + * (`maintenance.disabled`) or for one job type (`maintenance..disabled`). Mirrors + * `TableMetadata.isMaintenanceJobDisabled`, which the scheduler consults before dispatching. + */ + def isMaintenanceDisabled(props: Map[String, String], jobType: String): Boolean = + isEnabled(props, "maintenance.disabled") || isEnabled(props, s"maintenance.$jobType.disabled") + + /** + * True when the platform is configured to preserve orphans by moving them to a backup directory + * rather than deleting them (`Operations.deleteOrphanFiles`'s `deleteWith` hook). + */ + def isBackupConfigured(props: Map[String, String]): Boolean = + isEnabled(props, BACKUP_ENABLED_PROP) || props.get(BACKUP_DIR_PROP).exists(_.trim.nonEmpty) + + /** True when the table is a replica; the scheduled SE job skips non-primary tables. */ + def isReplica(props: Map[String, String]): Boolean = + props.get(TABLE_TYPE_PROP).exists(REPLICA_TABLE_TYPE.equalsIgnoreCase) + + private def isEnabled(props: Map[String, String], key: String): Boolean = + props.get(key).exists("true".equalsIgnoreCase) + + /** + * One sub-object of the server-written `policies` JSON. An absent or empty property is a table + * with no policies; a non-empty but unparseable one is corruption that must not be silently read + * as "no policy", since that would quietly expire snapshots on a different schedule than the one + * the table is configured for. + */ + private def policyNode(props: Map[String, String], name: String) = { + props.get(POLICIES_PROP).map(_.trim).filter(_.nonEmpty).flatMap { json => + val root = + try policiesMapper.readTree(json) + catch { + case NonFatal(e) => + throw new IllegalStateException( + s"Malformed '$POLICIES_PROP' table property; cannot resolve the maintenance " + + s"window the table is configured for. Value was: $json", e) + } + Option(root.get(name)) + } + } + + /** + * Map an OpenHouse policy granularity to its time unit, mirroring + * `SparkJobUtil.convertGranularityToChrono`: the `TimePartitionSpec.Granularity` names, with a + * fallback to the [[ChronoUnit]] name so a granularity already stored as e.g. `DAYS` resolves. + */ + private def granularityUnit(granularity: String): ChronoUnit = + granularity.toUpperCase(Locale.ROOT) match { + case "HOUR" => ChronoUnit.HOURS + case "DAY" => ChronoUnit.DAYS + case "MONTH" => ChronoUnit.MONTHS + case "YEAR" => ChronoUnit.YEARS + case other => + try ChronoUnit.valueOf(other) + catch { + case _: IllegalArgumentException => + throw new IllegalStateException( + s"Unrecognized granularity '$granularity' in the '$POLICIES_PROP' history policy.") + } + } +} 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 new file mode 100644 index 000000000..18e4701b8 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala @@ -0,0 +1,59 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{AnalyzeClusteringQuality, GrantRevokeStatement, OptimizeTable, 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 +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} +import org.apache.spark.sql.execution.SparkPlan + +import scala.collection.JavaConverters._ + +/* Strategy to convert a logical plan to physical plans */ +case class OpenhouseDataSourceV2Strategy(spark: SparkSession) extends Strategy with PredicateHelper { + override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match { + case SetRetentionPolicy(CatalogAndIdentifierExtractor(catalog, ident), granularity, count, colName, colPattern) => + SetRetentionPolicyExec(catalog, ident, granularity, count, colName, colPattern) :: Nil + case SetReplicationPolicy(CatalogAndIdentifierExtractor(catalog, ident), replicationPolicies) => + SetReplicationPolicyExec(catalog, ident, replicationPolicies) :: Nil + case UnSetReplicationPolicy(CatalogAndIdentifierExtractor(catalog, ident), replicationPolicies) => + UnSetReplicationPolicyExec(catalog, ident, replicationPolicies) :: Nil + case SetHistoryPolicy(CatalogAndIdentifierExtractor(catalog, ident), granularity, maxAge, versions) => + SetHistoryPolicyExec(catalog, ident, granularity, maxAge, versions) :: Nil + case SetSharingPolicy(CatalogAndIdentifierExtractor(catalog, ident), sharing) => + SetSharingPolicyExec(catalog, ident, sharing) :: Nil + case SetColumnPolicyTag(CatalogAndIdentifierExtractor(catalog, ident), policyTag, cols) => + SetColumnPolicyTagExec(catalog, ident, policyTag, cols) :: Nil + + case GrantRevokeStatement(isGrant, resourceType, CatalogAndIdentifierExtractor(catalog, ident), privilege, principal) => + GrantRevokeStatementExec(isGrant, resourceType, catalog, ident, privilege, principal) :: Nil + + case r @ ShowGrantsStatement(resourceType, CatalogAndIdentifierExtractor(catalog, ident)) => + ShowGrantsStatementExec(r.output, resourceType, catalog, ident) :: Nil + + case r @ VacuumTable(CatalogAndIdentifierExtractor(catalog, ident), removeOrphanFiles, retainHours) => + VacuumTableExec(r.output, spark, catalog, ident, removeOrphanFiles, retainHours) :: Nil + case r @ OptimizeTable(CatalogAndIdentifierExtractor(catalog, ident), full, rewriteManifests) => + OptimizeTableExec(r.output, spark, catalog, ident, full, rewriteManifests) :: Nil + + case r @ AnalyzeClusteringQuality(CatalogAndIdentifierExtractor(catalog, ident)) => + AnalyzeClusteringQualityExec(r.output, spark, catalog, ident) :: Nil + + case _ => Nil + } + + private object CatalogAndIdentifierExtractor { + def unapply(identifier: Seq[String]): Option[(TableCatalog, Identifier)] = { + val catalogAndIdentifier = Spark3Util.catalogAndIdentifier(spark, identifier.asJava) + catalogAndIdentifier.catalog match { + case icebergCatalog: SparkCatalog => + Some((icebergCatalog, catalogAndIdentifier.identifier)) + case icebergCatalog: SparkSessionCatalog[_] => + Some((icebergCatalog, catalogAndIdentifier.identifier)) + case _ => + None + } + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OptimizeTableExec.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OptimizeTableExec.scala new file mode 100644 index 000000000..13d270689 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OptimizeTableExec.scala @@ -0,0 +1,185 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.util.Locale + +import scala.collection.JavaConverters._ + +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.OptimizeTable +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, GenericInternalRow, Literal} +import org.apache.spark.sql.catalyst.util.quoteIfNeeded +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog, TableChange} +import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec +import org.apache.spark.sql.functions.{col, current_timestamp, expr, lit, max} +import org.apache.spark.unsafe.types.UTF8String + +/** + * Runs Iceberg data-layout maintenance for the OPTIMIZE command as thin sugar over the catalog's + * stored procedures. With no `optimize.cluster.keys` configured this is a plain bin-pack compaction + * (`system.rewrite_data_files`); with clustering configured it is a sort / z-order rewrite that is + * incremental by default (only the forward slice of the leading key since the last run, tracked by + * the `optimize.cluster.hwm-snapshot-id` watermark), with `FULL` reclustering up to the age floor. + * `REWRITE MANIFESTS` runs afterwards over the post-rewrite layout. + */ +case class OptimizeTableExec( + output: Seq[Attribute], + spark: SparkSession, + catalog: TableCatalog, + ident: Identifier, + full: Boolean, + rewriteManifests: Boolean) extends LeafV2CommandExec { + + private def row(metric: String, value: String): InternalRow = + new GenericInternalRow( + Array[Any](UTF8String.fromString(metric), UTF8String.fromString(value))) + + override protected def run(): Seq[InternalRow] = { + val props = catalog.loadTable(ident) match { + case iceberg: SparkTable + if iceberg.table().properties().containsKey(MaintenanceProperties.TABLE_ID_PROP) => + iceberg.table().properties().asScala.toMap + case table => + throw new UnsupportedOperationException(s"Cannot optimize non-Openhouse table: $table") + } + + // A table opted out of platform maintenance should not be compacted by hand either. + val compactionJob = MaintenanceProperties.DATA_COMPACTION_JOB + if (MaintenanceProperties.isMaintenanceDisabled(props, compactionJob)) { + throw new UnsupportedOperationException( + s"Maintenance is disabled for table '$ident' ('maintenance.disabled' or " + + s"'maintenance.$compactionJob.disabled'), so OPTIMIZE will not run on it.") + } + + val cat = quoteIfNeeded(catalog.name()) + val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") + val qualifiedTableName = s"$cat.$tableArg" + + // Snapshot the physical layout before doing any work so we can report the reduction. + val filesBefore = spark.table(s"$qualifiedTableName.files").count() + val snapshotsBefore = spark.table(s"$qualifiedTableName.snapshots").count() + + val config = OptimizeTable.parseClusterConfig(props) + config.keys match { + case Seq() => + // No clustering configured: plain bin-pack compaction (unchanged historical behavior). + spark.sql(s"CALL $cat.system.rewrite_data_files(table => '$tableArg')").collect() + case _ => + cluster(cat, tableArg, qualifiedTableName, config) + } + + if (rewriteManifests) { + // Independent manifest compaction; runs after the data rewrite so it sees the new layout. + spark.sql(s"CALL $cat.system.rewrite_manifests(table => '$tableArg')").collect() + } + + val filesAfter = spark.table(s"$qualifiedTableName.files").count() + val snapshotsAfter = spark.table(s"$qualifiedTableName.snapshots").count() + Seq( + row("files_before", filesBefore.toString), + row("files_after", filesAfter.toString), + row("files_removed", (filesBefore - filesAfter).toString), + row("snapshots_committed", (snapshotsAfter - snapshotsBefore).toString)) + } + + private def cluster( + cat: String, + tableArg: String, + qualifiedTableName: String, + config: OptimizeTable.ClusterConfig): Unit = { + import config.{hwm, keys, maxCommits, minAgeMinutes, sortMode, state} + + // Age floor: the newest snapshot at least `minAgeMinutes` old, by commit time. Everything + // younger is held back so we never rewrite files a concurrent streaming writer is extending. + val ageFloor = spark.table(s"$qualifiedTableName.snapshots") + .where(col("committed_at") <= current_timestamp() - expr(s"INTERVAL $minAgeMinutes MINUTES")) + .orderBy(col("committed_at").desc) + .limit(1) + .select("snapshot_id") + .collect().headOption.map(_.getLong(0)) + if (ageFloor.isEmpty) return // nothing old enough to consume yet -> no-op + + val floorId = ageFloor.get + // Incremental run whose watermark has not moved -> nothing new to do. + if (!full && hwm.contains(floorId)) return + + // The forward slice is bounded on the leading clustering key. Its upper bound is the max value + // present as of the age floor; Iceberg satisfies this from manifest metrics when the table has + // no deletes, so it is metadata-only. + val leadKey = keys.head + val floorMax = Option( + spark.read.option("snapshot-id", floorId).table(qualifiedTableName) + .agg(max(col(quoteIfNeeded(leadKey)))).head().get(0)) + if (floorMax.isEmpty) return // no data as of the age floor -> no-op + + // Lower bound: incremental runs skip what a prior run already clustered -- the last-clustered + // upper of the current key selection, read from the persisted interval state (a table property, + // so it survives snapshot expiration). FULL ignores the bound and reclusters everything up to + // the floor. + val cfgId = OptimizeTable.configId(keys, sortMode) + val lowerValue = state.find(_.config == cfgId).map(_.upper).filterNot(_ => full) + + // Cast the persisted (string) bound back to the leading key's type so a key promoted between + // runs (e.g. INT -> BIGINT) is compared after a cast, not across boxed types. + val leadType = spark.table(qualifiedTableName).schema(leadKey).dataType + val lead = col(quoteIfNeeded(leadKey)) + val lowerBound = lowerValue.map(u => lit(u).cast(leadType)) + + // No-op if the leading key has not advanced past the last-clustered upper. Evaluate the + // comparison in Catalyst (not in Scala) so the cast above governs the ordering. + val advanced = lowerBound.forall { lb => + spark.range(1).select(lit(floorMax.get) > lb).head().getBoolean(0) + } + if (!advanced) return + + // The `where` slice to recluster: `lead <= floorMax`, plus `lead > lowerBound` for an + // incremental run. Catalyst renders each key-type literal correctly, then the predicate is + // embedded as a SQL string literal so its own quotes survive the CALL. + val scope = lowerBound.map(lb => (lead > lb) && (lead <= lit(floorMax.get))) + .getOrElse(lead <= lit(floorMax.get)).expr.sql + val cols = keys.map(quoteIfNeeded).mkString(", ") + val sortOrder = sortMode.toLowerCase(Locale.ROOT) match { + case "zorder" => s"zorder($cols)" + case _ => cols + } + + // Scoped sort / z-order rewrite with partial progress: min-input-files=1 + rewrite-all=true + // cluster the region regardless of file count; use-starting-sequence-number keeps concurrent + // equality-deletes valid. rewrite-all forces a rewrite over the non-empty scope, so a healthy + // run always commits a snapshot; if none is committed the rewrite failed systemically (partial + // progress swallows per-group failures), so fail loudly and leave the watermark unadvanced. + val snapshotsBefore = spark.table(s"$qualifiedTableName.snapshots").count() + spark.sql( + s"CALL $cat.system.rewrite_data_files(" + + s"table => '$tableArg', " + + "strategy => 'sort', " + + s"sort_order => '$sortOrder', " + + s"where => ${Literal(scope).sql}, " + + "options => map(" + + "'min-input-files', '1', " + + "'rewrite-all', 'true', " + + "'use-starting-sequence-number', 'true', " + + "'partial-progress.enabled', 'true', " + + s"'partial-progress.max-commits', '$maxCommits'))").collect() + if (spark.table(s"$qualifiedTableName.snapshots").count() <= snapshotsBefore) { + throw new IllegalStateException( + s"OPTIMIZE clustered no data for '$qualifiedTableName': the scoped rewrite " + + s"(keys=[${keys.mkString(",")}], sort-mode=$sortMode) committed no snapshot despite a " + + s"non-empty scope. The watermark was left unadvanced so the run can be retried.") + } + + // Advance all clustering metadata in one atomic alterTable -- the watermark (the consumed age + // floor, not head), the config id, and the interval state -- so they never disagree. + val newState = OptimizeTable.advanceState( + state, cfgId, keys, sortMode, lowerValue, floorMax.get.toString, full) + catalog.alterTable(ident, + TableChange.setProperty(OptimizeTable.HWM_PROP, floorId.toString), + TableChange.setProperty(OptimizeTable.CONFIG_ID_PROP, cfgId), + TableChange.setProperty(OptimizeTable.STATE_PROP, OptimizeTable.stateMapper.writeValueAsString(newState))) + } + + override def simpleString(maxFields: Int): String = { + s"OptimizeTableExec: ${catalog} ${ident} full=${full} rewriteManifests=${rewriteManifests}" + } +} 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..5a0456a34 --- /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,161 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.time.{Duration, Instant, ZoneId} +import java.time.format.DateTimeFormatter + +import scala.collection.JavaConverters._ + +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, GenericInternalRow} +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 +import org.apache.spark.unsafe.types.UTF8String + +/** + * Runs Iceberg table maintenance for the VACUUM command as thin sugar over the catalog's stored + * procedures, using the same table-property contract as the scheduled maintenance jobs (see + * [[MaintenanceProperties]]). VACUUM is an '''Alpha''' feature and is opt-in per table via the + * `maintenance.vacuum.enabled` property. + * + * 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. + * + * `RETAIN n HOURS` bounds both operations. When it is omitted, each falls back to what the + * corresponding job would have used for this table: the `policies.history` window for expiration, + * and the orphan-file job's own default (or one day, under `ofd.one_day_ttl.enabled`) for orphan + * removal. The one deliberate divergence from the jobs is that expiration here also deletes the + * files the expired snapshots exclusively referenced -- reclaiming that storage is the point of + * running VACUUM by hand, whereas the scheduled job leaves it to orphan-file deletion. + */ +case class VacuumTableExec( + output: Seq[Attribute], + spark: SparkSession, + catalog: TableCatalog, + ident: Identifier, + removeOrphanFiles: Boolean, + retainHours: Option[Int]) extends LeafV2CommandExec { + + import MaintenanceProperties._ + + private def row(metric: String, value: String): InternalRow = + new GenericInternalRow( + Array[Any](UTF8String.fromString(metric), UTF8String.fromString(value))) + + override protected def run(): Seq[InternalRow] = { + val props = catalog.loadTable(ident) match { + case iceberg: SparkTable if iceberg.table().properties().containsKey(TABLE_ID_PROP) => + iceberg.table().properties().asScala.toMap + case table => + throw new UnsupportedOperationException(s"Cannot vacuum non-Openhouse table: $table") + } + + // VACUUM is an Alpha feature and is opt-in per table. The gate lives in the `maintenance.*` + // namespace because `openhouse.*` keys are preserved -- the /tables service rejects any attempt + // to set them -- so an `openhouse.`-prefixed gate could never be turned on. + if (!"true".equalsIgnoreCase(props.getOrElse(VACUUM_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 ('$VACUUM_ENABLED_PROP' = 'true').") + } + + // The scheduled snapshot-expiration job runs on primary tables only, so neither does VACUUM. + // A replica's snapshots are the replication protocol's state; expiring them by hand can strand + // an incremental replication mid-stream. Orphan cleanup for replicas stays with the scheduled + // orphan-file job, which applies its own replica-specific floor. + if (isReplica(props)) { + throw new UnsupportedOperationException( + s"Cannot vacuum replica table '$ident': snapshot expiration is not run on replica " + + s"tables. Maintenance for replicas is handled by the scheduled jobs.") + } + + requireMaintenanceEnabled(props, SNAPSHOTS_EXPIRATION_JOB) + if (removeOrphanFiles) { + requireMaintenanceEnabled(props, ORPHAN_FILES_DELETION_JOB) + } + + val quotedCatalog = quoteIfNeeded(catalog.name()) + val tableArg = (ident.namespace() :+ ident.name()).map(quoteIfNeeded).mkString(".") + val metrics = Seq.newBuilder[InternalRow] + + if (removeOrphanFiles) { + // Orphan-file deletion 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. + // + // On a table configured for orphan backups the scheduled job moves orphans into the backup + // directory instead of deleting them, via a delete hook the stored procedure has no + // equivalent of. Running the procedure would both destroy files the platform expects to + // remain recoverable and treat the backup directory's own contents as orphans, so refuse. + if (isBackupConfigured(props)) { + throw new UnsupportedOperationException( + s"Cannot remove orphan files on table '$ident': it is configured for orphan backups " + + s"('$BACKUP_ENABLED_PROP'/'$BACKUP_DIR_PROP'), which preserve orphans instead of " + + s"deleting them. Leave orphan-file cleanup to the scheduled job, or run VACUUM " + + s"without REMOVE ORPHAN FILES.") + } + val (age, source) = orphanRetention(props, retainHours) + metrics += row("orphan_files_retain_hours", age.toHours.toString) + metrics += row("orphan_files_retain_source", source) + spark.sql( + s"CALL $quotedCatalog.system.remove_orphan_files(" + + s"table => '$tableArg'${olderThanArg(age)})").collect() + } + + // Snapshot expiration always runs. An explicit RETAIN overrides the age the history policy + // configures, but not its `versions` cap: that is a separate policy dimension, and the + // scheduled job applies it independently of the age. + val configured = snapshotRetention(props) + val retention = retainHours + .map(h => configured.copy(age = Duration.ofHours(h.toLong), source = "RETAIN")) + .getOrElse(configured) + metrics += row("snapshots_retain_hours", retention.age.toHours.toString) + metrics += row("snapshots_retain_source", retention.source) + spark.sql( + s"CALL $quotedCatalog.system.expire_snapshots(" + + s"table => '$tableArg'${olderThanArg(retention.age)})").collect() + + // A `versions` history policy caps how many snapshots survive regardless of age. The job + // applies it as a second, separate expiration; mirror that rather than folding it into the + // call above, where `retain_last` would instead act as a floor on the age-based expiry. + retention.versions.foreach { versions => + metrics += row("snapshots_retain_last", versions.toString) + spark.sql( + s"CALL $quotedCatalog.system.expire_snapshots(" + + s"table => '$tableArg'${olderThanArg(Duration.ZERO)}, retain_last => $versions)").collect() + } + + metrics.result() + } + + /** + * Render an `older_than` argument for a retention window. Procedure arguments must be foldable, + * so the window is resolved here to a literal timestamp rather than an expression over + * `current_timestamp()`. 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. + */ + private def olderThanArg(age: Duration): 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.now().minus(age))}'" + } + + private def requireMaintenanceEnabled(props: Map[String, String], jobType: String): Unit = { + if (isMaintenanceDisabled(props, jobType)) { + throw new UnsupportedOperationException( + s"Maintenance is disabled for table '$ident' ('maintenance.disabled' or " + + s"'maintenance.$jobType.disabled'), so VACUUM will not run $jobType on it.") + } + } + + override def simpleString(maxFields: Int): String = { + s"VacuumTableExec: ${catalog} ${ident} removeOrphanFiles=${removeOrphanFiles} " + + s"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/mapper/IcebergCatalogMapper.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/mapper/IcebergCatalogMapper.scala new file mode 100644 index 000000000..6f3dd5748 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/mapper/IcebergCatalogMapper.scala @@ -0,0 +1,38 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2.mapper + +import org.apache.iceberg.CachingCatalog +import org.apache.iceberg.catalog.Catalog +import org.apache.iceberg.common.DynFields +import org.apache.iceberg.spark.{SparkCatalog, SparkSessionCatalog} +import org.apache.spark.sql.connector.catalog.TableCatalog + +object IcebergCatalogMapper { + + /** + * Convert Spark's {@link TableCatalog} to Iceberg's {@link Catalog} + * + * {@link Catalog} instance is a private field inside of a chain of wrapping {@link Catalog} classes in {@link TableCatalog}. + * To access the instance we need to access following private fields: + * {@link SparkSessionCatalog#icebergCatalog} -> {@link SparkCatalog} + * {@link SparkCatalog#icebergCatalog} -> {@link CachingCatalog} + * {@link CachingCatalog#catalog} -> {@link OpenHouseCatalog} + * + * @return null :if it is not iceberg based catalog + * catalog :if iceberg catalog + */ + def toIcebergCatalog(catalog: TableCatalog): Catalog = { + if (!(catalog.isInstanceOf[SparkCatalog] || catalog.isInstanceOf[SparkSessionCatalog[_]])) { + null + } else { + val sparkCatalog = if (catalog.isInstanceOf[SparkSessionCatalog[_]]) { + DynFields.builder.hiddenImpl(classOf[SparkSessionCatalog[_]], "icebergCatalog").build[TableCatalog](catalog).get + } else { + catalog + } + var icebergCatalog = DynFields.builder.hiddenImpl(classOf[SparkCatalog], "icebergCatalog").build[Catalog](sparkCatalog).get + val cacheEnabled = DynFields.builder.hiddenImpl(classOf[SparkCatalog], "cacheEnabled").build[Boolean](sparkCatalog).get + if (cacheEnabled) icebergCatalog = DynFields.builder.hiddenImpl(classOf[CachingCatalog], "catalog").build[Catalog](icebergCatalog).get + icebergCatalog + } + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTableTest.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTableTest.scala new file mode 100644 index 000000000..f7b75867c --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/OptimizeTableTest.scala @@ -0,0 +1,95 @@ +package com.linkedin.openhouse.spark.sql.catalyst.plans.logical + +import org.junit.jupiter.api.Assertions.{assertEquals, assertNotEquals, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.OptimizeTable._ + +class OptimizeTableTest { + + @Test + def parseClusterConfigResolvesDefaultsAndTypedValues(): Unit = { + val empty = parseClusterConfig(Map.empty) + assertTrue(empty.keys.isEmpty) + assertEquals(DEFAULT_SORT_MODE, empty.sortMode) + assertEquals(DEFAULT_MIN_SNAPSHOT_AGE_MINUTES, empty.minAgeMinutes) + assertEquals(DEFAULT_MAX_COMMITS, empty.maxCommits) + assertTrue(empty.hwm.isEmpty) + assertTrue(empty.state.isEmpty) + + val cfg = parseClusterConfig(Map( + KEYS_PROP -> " ts , uid ", + SORT_MODE_PROP -> "sort", + MIN_SNAPSHOT_AGE_PROP -> "5", + MAX_COMMITS_PROP -> "3", + HWM_PROP -> "42", + STATE_PROP -> """[{"config":"c1","keys":"ts","mode":"sort","upper":"20"}]""")) + assertEquals(Seq("ts", "uid"), cfg.keys) + assertEquals("sort", cfg.sortMode) + assertEquals(5L, cfg.minAgeMinutes) + assertEquals(3L, cfg.maxCommits) + assertEquals(Some(42L), cfg.hwm) + assertEquals(Seq(ClusterInterval("c1", "ts", "sort", None, "20")), cfg.state) + } + + @Test + def configIdStableAcrossWhitespaceChangesOnKeyOrMode(): Unit = { + assertEquals(configId(Seq("ts", "uid"), "zorder"), configId(Seq(" ts ", " uid "), "ZORDER")) + assertNotEquals(configId(Seq("ts", "uid"), "zorder"), configId(Seq("ts"), "zorder")) + assertNotEquals(configId(Seq("ts"), "zorder"), configId(Seq("ts"), "sort")) + } + + @Test + def parseStateRoundTripsWithAndWithoutLower(): Unit = { + val json = """[{"config":"c1","keys":"ts","mode":"sort","lower":"10","upper":"20"},""" + + """{"config":"c2","keys":"ts,uid","mode":"zorder","upper":"2026-01-06 00:00:00"}]""" + assertEquals(Seq( + ClusterInterval("c1", "ts", "sort", Some("10"), "20"), + ClusterInterval("c2", "ts,uid", "zorder", None, "2026-01-06 00:00:00")), parseState(json)) + } + + @Test + def parseStateEmptyOrNullIsNoState(): Unit = { + assertEquals(Seq.empty[ClusterInterval], parseState("")) + assertEquals(Seq.empty[ClusterInterval], parseState(null)) + } + + @Test + def parseStateMalformedFailsLoudly(): Unit = { + val e = assertThrows(classOf[IllegalStateException], () => parseState("not json")) + assertTrue(e.getMessage.contains(STATE_PROP)) + assertTrue(e.getMessage.contains("UNSET TBLPROPERTIES")) + } + + @Test + def advanceStateFirstRunCreatesInterval(): Unit = { + assertEquals( + Seq(ClusterInterval("c1", "ts", "sort", Some("5"), "10")), + advanceState(Seq.empty, "c1", Seq("ts"), "sort", Some("5"), "10", full = false)) + } + + @Test + def advanceStateSameConfigExtendsUpperKeepsLower(): Unit = { + val s0 = Seq(ClusterInterval("c1", "ts", "sort", Some("5"), "10")) + assertEquals( + Seq(ClusterInterval("c1", "ts", "sort", Some("5"), "20")), + advanceState(s0, "c1", Seq("ts"), "sort", Some("10"), "20", full = false)) + } + + @Test + def advanceStateFullCollapsesToUnbounded(): Unit = { + val s0 = Seq(ClusterInterval("c1", "ts", "sort", Some("5"), "20")) + assertEquals( + Seq(ClusterInterval("c1", "ts", "sort", None, "30")), + advanceState(s0, "c1", Seq("ts"), "sort", None, "30", full = true)) + } + + @Test + def advanceStateConfigChangeAppendsAndRetains(): Unit = { + val s0 = Seq(ClusterInterval("c1", "ts", "sort", None, "20")) + val s1 = advanceState(s0, "c2", Seq("ts", "uid"), "zorder", Some("20"), "40", full = false) + assertEquals(Seq( + ClusterInterval("c1", "ts", "sort", None, "20"), + ClusterInterval("c2", "ts,uid", "zorder", Some("20"), "40")), s1) + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExecTest.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExecTest.scala new file mode 100644 index 000000000..b66aea48c --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExecTest.scala @@ -0,0 +1,44 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue} +import org.junit.jupiter.api.Test + +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.OptimizeTable.ClusterInterval +import com.linkedin.openhouse.spark.sql.execution.datasources.v2.AnalyzeClusteringQualityExec._ + +class AnalyzeClusteringQualityExecTest { + + @Test + def metricExprAccessesPerFileMetricQuotingKey(): Unit = { + assertEquals("readable_metrics.ts.lower_bound", metricExpr("ts", "lower_bound")) + assertEquals("readable_metrics.`my-col`.upper_bound", metricExpr("my-col", "upper_bound")) + } + + @Test + def coveragePredicateNoIntervalsIsFalse(): Unit = { + assertEquals("false", coveragePredicate("lo", "hi", Seq.empty, "INT")) + } + + @Test + def coveragePredicateBoundedChecksBothSidesCastToKeyType(): Unit = { + val p = coveragePredicate("lo", "hi", + Seq(ClusterInterval("c", "ts", "sort", Some("5"), "20")), "INT") + assertEquals("((hi <= CAST('20' AS INT)) AND (lo > CAST('5' AS INT)))", p) + } + + @Test + def coveragePredicateUnboundedBelowDropsLowerCheck(): Unit = { + val p = coveragePredicate("lo", "hi", + Seq(ClusterInterval("c", "ts", "sort", None, "20")), "INT") + assertEquals("((hi <= CAST('20' AS INT)) AND true)", p) + } + + @Test + def coveragePredicateMultipleIntervalsAreOred(): Unit = { + val p = coveragePredicate("lo", "hi", Seq( + ClusterInterval("c", "ts", "sort", None, "10"), + ClusterInterval("c", "ts", "sort", Some("10"), "20")), "INT") + assertTrue(p.contains(" OR ")) + assertTrue(p.startsWith("((hi <= CAST('10' AS INT))")) + } +} diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenancePropertiesTest.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenancePropertiesTest.scala new file mode 100644 index 000000000..c7a07fa34 --- /dev/null +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/MaintenancePropertiesTest.scala @@ -0,0 +1,113 @@ +package com.linkedin.openhouse.spark.sql.execution.datasources.v2 + +import java.time.Duration + +import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertThrows, assertTrue} +import org.junit.jupiter.api.Test + +import com.linkedin.openhouse.spark.sql.execution.datasources.v2.MaintenanceProperties._ + +/** + * Pins the property contract the interactive maintenance DDL shares with the scheduled jobs. The + * expected values here are the job sources' own: a change that breaks one of these tests means the + * DDL and the job would disagree about the same table. + */ +class MaintenancePropertiesTest { + + private def policies(history: String): Map[String, String] = + Map(POLICIES_PROP -> s"""{"retention":{"count":1,"granularity":"DAY"},"history":$history}""") + + @Test + def snapshotRetentionFallsBackToTheSnapshotExpirationJobDefault(): Unit = { + // SnapshotsExpirationSparkApp enforces a 3-day TTL even when the table has no history policy. + Seq(Map.empty[String, String], Map(POLICIES_PROP -> ""), policies("""{"versions":0}""")) + .foreach { props => + val retention = snapshotRetention(props) + assertEquals(Duration.ofDays(3), retention.age) + assertTrue(retention.versions.isEmpty) + assertEquals("default (3 DAY)", retention.source) + } + } + + @Test + def snapshotRetentionUsesTheHistoryPolicyMaxAgeAndGranularity(): Unit = { + assertEquals(Duration.ofDays(5), + snapshotRetention(policies("""{"maxAge":5,"granularity":"DAY"}""")).age) + assertEquals(Duration.ofHours(12), + snapshotRetention(policies("""{"maxAge":12,"granularity":"HOUR"}""")).age) + // SparkJobUtil.convertGranularityToChrono also accepts an already-ChronoUnit granularity, as + // the expiration job's own default produces. + assertEquals(Duration.ofDays(2), + snapshotRetention(policies("""{"maxAge":2,"granularity":"DAYS"}""")).age) + } + + @Test + def snapshotRetentionGranularityIsCaseInsensitiveAndDefaultsToDays(): Unit = { + assertEquals(Duration.ofDays(4), + snapshotRetention(policies("""{"maxAge":4,"granularity":"day"}""")).age) + assertEquals(Duration.ofDays(4), snapshotRetention(policies("""{"maxAge":4}""")).age) + } + + @Test + def snapshotRetentionCarriesTheVersionsCapOnlyWhenSet(): Unit = { + val capped = snapshotRetention(policies("""{"maxAge":1,"granularity":"DAY","versions":10}""")) + assertEquals(Some(10), capped.versions) + assertEquals("policies.history (1 DAY)", capped.source) + assertTrue( + snapshotRetention(policies("""{"maxAge":1,"granularity":"DAY","versions":0}""")).versions + .isEmpty) + } + + @Test + def malformedPoliciesFailsRatherThanSilentlyReadingAsNoPolicy(): Unit = { + val e = assertThrows(classOf[IllegalStateException], + () => snapshotRetention(Map(POLICIES_PROP -> "not json"))) + assertTrue(e.getMessage.contains(POLICIES_PROP)) + } + + @Test + def unknownGranularityFailsLoudly(): Unit = { + assertThrows(classOf[IllegalStateException], + () => snapshotRetention(policies("""{"maxAge":1,"granularity":"FORTNIGHT"}"""))) + } + + @Test + def orphanRetentionMirrorsTheOrphanFilesDeletionJobDefaults(): Unit = { + assertEquals((Duration.ofDays(7), "default"), orphanRetention(Map.empty, None)) + assertEquals((Duration.ofDays(1), OFD_ONE_DAY_TTL_PROP), + orphanRetention(Map(OFD_ONE_DAY_TTL_PROP -> "true"), None)) + assertEquals((Duration.ofDays(7), "default"), + orphanRetention(Map(OFD_ONE_DAY_TTL_PROP -> "false"), None)) + } + + @Test + def orphanRetentionHonorsAnExplicitRetainEvenBelowTheDefaults(): Unit = { + assertEquals((Duration.ofHours(1), "RETAIN"), + orphanRetention(Map(OFD_ONE_DAY_TTL_PROP -> "true"), Some(1))) + } + + @Test + def maintenanceIsDisabledWholesaleOrPerJobType(): Unit = { + assertFalse(isMaintenanceDisabled(Map.empty, SNAPSHOTS_EXPIRATION_JOB)) + assertTrue(isMaintenanceDisabled(Map("maintenance.disabled" -> "true"), + SNAPSHOTS_EXPIRATION_JOB)) + val perJob = Map(s"maintenance.$ORPHAN_FILES_DELETION_JOB.disabled" -> "true") + assertTrue(isMaintenanceDisabled(perJob, ORPHAN_FILES_DELETION_JOB)) + assertFalse(isMaintenanceDisabled(perJob, SNAPSHOTS_EXPIRATION_JOB)) + } + + @Test + def backupIsConfiguredByEitherTheFlagOrTheDirectory(): Unit = { + assertFalse(isBackupConfigured(Map.empty)) + assertFalse(isBackupConfigured(Map(BACKUP_ENABLED_PROP -> "false", BACKUP_DIR_PROP -> " "))) + assertTrue(isBackupConfigured(Map(BACKUP_ENABLED_PROP -> "true"))) + assertTrue(isBackupConfigured(Map(BACKUP_DIR_PROP -> ".backup"))) + } + + @Test + def replicaIsRecognizedFromTheServerWrittenTableType(): Unit = { + assertFalse(isReplica(Map.empty)) + assertFalse(isReplica(Map(TABLE_TYPE_PROP -> "PRIMARY_TABLE"))) + assertTrue(isReplica(Map(TABLE_TYPE_PROP -> REPLICA_TABLE_TYPE))) + } +} diff --git a/tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/HouseTablesH2Repository.java b/tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/HouseTablesH2Repository.java index 6efb1649a..b70da7d2c 100644 --- a/tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/HouseTablesH2Repository.java +++ b/tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/HouseTablesH2Repository.java @@ -22,12 +22,13 @@ * communication to the implementation of {@link HouseTableRepository} is not needed. With {@link * Primary} annotation, this repository will be the default injection. * - *

The {@link ConditionalOnProperty} guard makes this in-memory stub the default (matchIfMissing = - * true, so every existing consumer is unaffected), but lets a consumer opt OUT of it by setting - * {@code openhouse.htsStub.enabled=false}. When opted out, the stub bean is not created and the real - * {@link com.linkedin.openhouse.internal.catalog.repository.HouseTableRepositoryImpl} (the HTTP client - * to a real House Table Service) becomes the sole {@link HouseTableRepository} — used by the - * delta-harness to test against an embedded real HTS. Test-only; no production/behavioral change. + *

The {@link ConditionalOnProperty} guard makes this in-memory stub the default (matchIfMissing + * = true, so every existing consumer is unaffected), but lets a consumer opt OUT of it by setting + * {@code openhouse.htsStub.enabled=false}. When opted out, the stub bean is not created and the + * real {@link com.linkedin.openhouse.internal.catalog.repository.HouseTableRepositoryImpl} (the + * HTTP client to a real House Table Service) becomes the sole {@link HouseTableRepository} — used + * by the delta-harness to test against an embedded real HTS. Test-only; no production/behavioral + * change. */ @Repository @Primary