From b7030ebc149dd25c030a6c1676a7316f30a57b90 Mon Sep 17 00:00:00 2001 From: Mike Kuchenbecker Date: Fri, 24 Jul 2026 11:20:43 -0700 Subject: [PATCH] [spark-3.5] Add ANALYZE TABLE ... COMPUTE CLUSTERING QUALITY SQL extension Adds a read-only clustering-quality probe to the OpenHouse spark-3.5 SQL extensions: ANALYZE TABLE 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). No commit and no property write. Emits (metric, dimension, value) rows: clustering_configured; config_id/keys/sort_mode; coverage_bytes_pct / coverage_files_pct (fraction whose leading-key range was clustered under the current config); per-key depth_avg/p90/max and the _covered variants (Snowflake-style stabbing depth, the SLA quality input); null_bound_bytes_pct; unclustered_tail_hours; state. Coverage is an aggregate over metadata and depth is a windowed sweep, both in distributed SQL so the command is safe on tables with very large file counts. Implemented as AnalyzeClusteringQualityExec, reusing the optimize.cluster.* property contract from OptimizeTable, with an OpenHouse-table guard. Routing intercepts only the COMPUTE CLUSTERING QUALITY variant, so ordinary ANALYZE TABLE ... COMPUTE STATISTICS still delegates to Spark. New non-reserved keywords ANALYZE/COMPUTE/CLUSTERING/QUALITY. Tested by AnalyzeClusteringQualityExecTest (metricExpr/coveragePredicate) and AnalyzeClusteringQualityStatementTest (real Iceberg, Hadoop catalog: unconfigured flag, post-OPTIMIZE coverage/depth, read-only invariant, COMPUTE STATISTICS delegation, non-OpenHouse rejection). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- ...AnalyzeClusteringQualityStatementTest.java | 156 +++++++++++++ .../extensions/OpenhouseSqlExtensions.g4 | 6 + .../OpenhouseSparkSqlExtensionsParser.scala | 4 +- .../OpenhouseSqlExtensionsAstBuilder.scala | 8 +- .../logical/AnalyzeClusteringQuality.scala | 25 ++ .../v2/AnalyzeClusteringQualityExec.scala | 215 ++++++++++++++++++ .../v2/OpenhouseDataSourceV2Strategy.scala | 5 +- .../v2/AnalyzeClusteringQualityExecTest.scala | 44 ++++ 8 files changed, 460 insertions(+), 3 deletions(-) create mode 100644 integrations/spark/spark-3.5/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/statementtest/AnalyzeClusteringQualityStatementTest.java create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/plans/logical/AnalyzeClusteringQuality.scala create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExec.scala create mode 100644 integrations/spark/spark-3.5/openhouse-spark-runtime/src/test/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/AnalyzeClusteringQualityExecTest.scala 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-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 index bb1af5ac6..39391575a 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/antlr/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensions.g4 @@ -33,6 +33,7 @@ statement | REVOKE privilege ON grantableResource FROM principal #revokeStatement | SHOW GRANTS ON grantableResource #showGrantsStatement | OPTIMIZE multipartIdentifier (FULL)? (REWRITE MANIFESTS)? #optimizeTable + | ANALYZE TABLE multipartIdentifier COMPUTE CLUSTERING QUALITY #analyzeClusteringQuality ; multipartIdentifier @@ -71,6 +72,7 @@ nonReserved : ALTER | TABLE | SET | POLICY | RETENTION | SHARING | REPLICATION | HISTORY | GRANT | REVOKE | ON | TO | SHOW | GRANTS | PATTERN | WHERE | COLUMN | OPTIMIZE | FULL | REWRITE | MANIFESTS + | ANALYZE | COMPUTE | CLUSTERING | QUALITY ; sharingPolicy @@ -211,6 +213,10 @@ OPTIMIZE: 'OPTIMIZE'; FULL: 'FULL'; REWRITE: 'REWRITE'; MANIFESTS: 'MANIFESTS'; +ANALYZE: 'ANALYZE'; +COMPUTE: 'COMPUTE'; +CLUSTERING: 'CLUSTERING'; +QUALITY: 'QUALITY'; POSITIVE_INTEGER : DIGIT+ diff --git a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSparkSqlExtensionsParser.scala index 2e2b11985..111db976a 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 @@ -90,7 +90,9 @@ class OpenhouseSparkSqlExtensionsParser (delegate: ParserInterface) extends Pars normalized.startsWith("grant") || normalized.startsWith("revoke") || normalized.startsWith("show grants") || - normalized.startsWith("optimize") + 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 index c5c68f0bc..ffa19f753 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/catalyst/parser/extensions/OpenhouseSqlExtensionsAstBuilder.scala @@ -2,7 +2,7 @@ package com.linkedin.openhouse.spark.sql.catalyst.parser.extensions import com.linkedin.openhouse.spark.sql.catalyst.enums.GrantableResourceTypes import com.linkedin.openhouse.spark.sql.catalyst.parser.extensions.OpenhouseSqlExtensionsParser._ -import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, OptimizeTable, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy} +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{AnalyzeClusteringQuality, GrantRevokeStatement, OptimizeTable, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy} 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 @@ -204,6 +204,12 @@ class OpenhouseSqlExtensionsAstBuilder (delegate: ParserInterface) extends Openh 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 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/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/OpenhouseDataSourceV2Strategy.scala b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala index 70ceb7859..6850c192b 100644 --- a/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala +++ b/integrations/spark/spark-3.5/openhouse-spark-runtime/src/main/scala/com/linkedin/openhouse/spark/sql/execution/datasources/v2/OpenhouseDataSourceV2Strategy.scala @@ -1,6 +1,6 @@ package com.linkedin.openhouse.spark.sql.execution.datasources.v2 -import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{GrantRevokeStatement, OptimizeTable, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy} +import com.linkedin.openhouse.spark.sql.catalyst.plans.logical.{AnalyzeClusteringQuality, GrantRevokeStatement, OptimizeTable, SetColumnPolicyTag, SetHistoryPolicy, SetReplicationPolicy, SetRetentionPolicy, SetSharingPolicy, ShowGrantsStatement, UnSetReplicationPolicy} import org.apache.iceberg.spark.{Spark3Util, SparkCatalog, SparkSessionCatalog} import org.apache.spark.sql.{SparkSession, Strategy} import org.apache.spark.sql.catalyst.expressions.PredicateHelper @@ -35,6 +35,9 @@ case class OpenhouseDataSourceV2Strategy(spark: SparkSession) extends Strategy w 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 } 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))")) + } +}