[spark-3.5] Consolidate the maintenance SQL extensions and wire them to the OFD/SE table properties

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).
+ *
+ * <p>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.
+ *
+ * <p>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<String> metric(List<Row> 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<String> dim(List<Row> 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<Row> 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<Row> 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<Row> data = spark.sql("SELECT ts FROM " + tableName + " ORDER BY ts").collectAsList();
+      assertEquals(2, data.size());
+
+      spark.sql("DROP TABLE " + tableName);
+    }
+  }
+}

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).
+ *
+ * <p>{@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<Row> 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<Row> 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<Row> 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<Row> 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);
+    }
+  }
+}