(feat: persistence) Add schema-metrics-v1.sql for metrics tables (#3337)

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatabaseType.java

@@ -60,4 +60,24 @@ public InputStream openInitScriptResource(int schemaVersion) {
     ClassLoader classLoader = DatasourceOperations.class.getClassLoader();
     return classLoader.getResourceAsStream(resourceName);
   }
+
+  /**
+   * Open an InputStream that contains data from the metrics schema init script. This stream should
+   * be closed by the caller.
+   *
+   * @param metricsSchemaVersion the metrics schema version (currently only 1 is supported)
+   * @return an InputStream for the metrics schema SQL file
+   */
+  public InputStream openMetricsSchemaResource(int metricsSchemaVersion) {
+    if (metricsSchemaVersion != 1) {
+      throw new IllegalArgumentException(
+          "Unknown or invalid metrics schema version " + metricsSchemaVersion);
+    }
+
+    final String resourceName =
+        String.format("%s/schema-metrics-v%d.sql", this.getDisplayName(), metricsSchemaVersion);
+
+    ClassLoader classLoader = DatasourceOperations.class.getClassLoader();
+    return classLoader.getResourceAsStream(resourceName);
+  }
 }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java

@@ -57,7 +57,9 @@ public class DatasourceOperations {
   private static final String RELATION_DOES_NOT_EXIST = "42P01";
 
   // H2 STATUS CODES
-  private static final String H2_RELATION_DOES_NOT_EXIST = "90079";
+  // 90079 = Schema not found, 42S02 = Table or view not found
+  private static final String H2_SCHEMA_DOES_NOT_EXIST = "90079";
+  private static final String H2_TABLE_DOES_NOT_EXIST = "42S02";
 
   // POSTGRES RETRYABLE EXCEPTIONS
   private static final String SERIALIZATION_FAILURE_SQL_CODE = "40001";
@@ -402,7 +404,9 @@ public boolean isConstraintViolation(SQLException e) {
   public boolean isRelationDoesNotExist(SQLException e) {
     return (RELATION_DOES_NOT_EXIST.equals(e.getSQLState())
             && databaseType == DatabaseType.POSTGRES)
-        || (H2_RELATION_DOES_NOT_EXIST.equals(e.getSQLState()) && databaseType == DatabaseType.H2);
+        || ((H2_SCHEMA_DOES_NOT_EXIST.equals(e.getSQLState())
+                || H2_TABLE_DOES_NOT_EXIST.equals(e.getSQLState()))
+            && databaseType == DatabaseType.H2);
   }
 
   private Connection borrowConnection() throws SQLException {

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java

@@ -777,6 +777,25 @@ static boolean entityTableExists(DatasourceOperations datasourceOperations) {
     }
   }
 
+  /**
+   * Checks if the metrics tables have been bootstrapped by querying the metrics_version table.
+   *
+   * @param datasourceOperations the datasource operations to use for the check
+   * @return true if the metrics_version table exists and contains data, false otherwise
+   */
+  public static boolean metricsTableExists(DatasourceOperations datasourceOperations) {
+    PreparedQuery query = QueryGenerator.generateMetricsVersionQuery();
+    try {
+      List<SchemaVersion> versions = datasourceOperations.executeSelect(query, new SchemaVersion());
+      return versions != null && !versions.isEmpty();
+    } catch (SQLException e) {
+      if (datasourceOperations.isRelationDoesNotExist(e)) {
+        return false;
+      }
+      throw new IllegalStateException("Failed to check if metrics tables exist", e);
+    }
+  }
+
   /** {@inheritDoc} */
   @Override
   public <T extends PolarisEntity & LocationBasedEntity>

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBootstrapUtils.java

@@ -88,4 +88,18 @@ public static int getRequestedSchemaVersion(BootstrapOptions bootstrapOptions) {
     }
     return -1;
   }
+
+  /**
+   * Determines whether the metrics schema should be included during bootstrap.
+   *
+   * @param bootstrapOptions The bootstrap options containing schema information.
+   * @return true if the metrics schema should be included, false otherwise.
+   */
+  public static boolean shouldIncludeMetrics(BootstrapOptions bootstrapOptions) {
+    SchemaOptions schemaOptions = bootstrapOptions.schemaOptions();
+    if (schemaOptions != null) {
+      return schemaOptions.includeMetrics();
+    }
+    return false;
+  }
 }

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -172,6 +172,13 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
               datasourceOperations
                   .getDatabaseType()
                   .openInitScriptResource(effectiveSchemaVersion));
+
+          // Run the metrics schema script if requested
+          if (JdbcBootstrapUtils.shouldIncludeMetrics(bootstrapOptions)) {
+            LOGGER.info("Including metrics schema for realm: {}", realm);
+            datasourceOperations.executeScript(
+                datasourceOperations.getDatabaseType().openMetricsSchemaResource(1));
+          }
         } catch (SQLException e) {
           throw new RuntimeException(
               String.format("Error executing sql script: %s", e.getMessage()), e);

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/QueryGenerator.java

@@ -369,6 +369,13 @@ static PreparedQuery generateEntityTableExistQuery() {
         List.of());
   }
 
+  @VisibleForTesting
+  static PreparedQuery generateMetricsVersionQuery() {
+    return new PreparedQuery(
+        "SELECT version_value FROM POLARIS_SCHEMA.metrics_version WHERE version_key = 'metrics_version'",
+        List.of());
+  }
+
   /**
    * Generate a SELECT query to find any entities that have a given realm & parent and that may
    * overlap with a given location. The check is performed without consideration for the scheme, so

persistence/relational-jdbc/src/main/resources/h2/schema-metrics-v1.sql

@@ -0,0 +1,166 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements.  See the NOTICE file--
+--  distributed with this work for additional information
+-- regarding copyright ownership.  The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"). You may not use this file except in compliance
+-- with the License.  You may obtain a copy of the License at
+--
+--  http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied.  See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+--
+
+-- This schema is SEPARATE from the entity schema and can evolve independently.
+-- It contains tables for storing Iceberg metrics reports.
+--
+-- Tables:
+--   * `metrics_version` - Version tracking for the metrics schema
+--   * `scan_metrics_report` - Scan metrics reports
+--   * `commit_metrics_report` - Commit metrics reports
+-- ============================================================================
+
+CREATE SCHEMA IF NOT EXISTS POLARIS_SCHEMA;
+SET SCHEMA POLARIS_SCHEMA;
+
+-- Metrics schema version tracking (separate from entity schema version)
+CREATE TABLE IF NOT EXISTS metrics_version (
+    version_key VARCHAR PRIMARY KEY,
+    version_value INTEGER NOT NULL
+);
+
+MERGE INTO metrics_version (version_key, version_value)
+    KEY (version_key)
+    VALUES ('metrics_version', 1);
+
+COMMENT ON TABLE metrics_version IS 'the version of the metrics schema in use';
+
+-- ============================================================================
+-- SCAN METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS scan_metrics_report (
+    report_id TEXT NOT NULL,
+    realm_id TEXT NOT NULL,
+    catalog_id BIGINT NOT NULL,
+    table_id BIGINT NOT NULL,
+
+    -- Report metadata
+    timestamp_ms BIGINT NOT NULL,
+    principal_name TEXT,
+    request_id TEXT,
+
+    -- Trace correlation
+    otel_trace_id TEXT,
+    otel_span_id TEXT,
+    report_trace_id TEXT,
+
+    -- Scan context
+    snapshot_id BIGINT,
+    schema_id INTEGER,
+    filter_expression TEXT,
+    projected_field_ids TEXT,
+    projected_field_names TEXT,
+
+    -- Scan metrics
+    result_data_files BIGINT DEFAULT 0,
+    result_delete_files BIGINT DEFAULT 0,
+    total_file_size_bytes BIGINT DEFAULT 0,
+    total_data_manifests BIGINT DEFAULT 0,
+    total_delete_manifests BIGINT DEFAULT 0,
+    scanned_data_manifests BIGINT DEFAULT 0,
+    scanned_delete_manifests BIGINT DEFAULT 0,
+    skipped_data_manifests BIGINT DEFAULT 0,
+    skipped_delete_manifests BIGINT DEFAULT 0,
+    skipped_data_files BIGINT DEFAULT 0,
+    skipped_delete_files BIGINT DEFAULT 0,
+    total_planning_duration_ms BIGINT DEFAULT 0,
+
+    -- Equality/positional delete metrics
+    equality_delete_files BIGINT DEFAULT 0,
+    positional_delete_files BIGINT DEFAULT 0,
+    indexed_delete_files BIGINT DEFAULT 0,
+    total_delete_file_size_bytes BIGINT DEFAULT 0,
+
+    -- Additional metadata (for extensibility)
+    metadata TEXT DEFAULT '{}',
+
+    PRIMARY KEY (realm_id, report_id)
+);
+
+COMMENT ON TABLE scan_metrics_report IS 'Scan metrics reports as first-class entities';
+
+-- Index for retention cleanup by timestamp
+CREATE INDEX IF NOT EXISTS idx_scan_report_timestamp ON scan_metrics_report(realm_id, timestamp_ms);
+
+-- ============================================================================
+-- COMMIT METRICS REPORT TABLE
+-- ============================================================================
+
+CREATE TABLE IF NOT EXISTS commit_metrics_report (
+    report_id TEXT NOT NULL,
+    realm_id TEXT NOT NULL,
+    catalog_id BIGINT NOT NULL,
+    table_id BIGINT NOT NULL,
+
+    -- Report metadata
+    timestamp_ms BIGINT NOT NULL,
+    principal_name TEXT,
+    request_id TEXT,
+
+    -- Trace correlation
+    otel_trace_id TEXT,
+    otel_span_id TEXT,
+    report_trace_id TEXT,
+
+    -- Commit context
+    snapshot_id BIGINT NOT NULL,
+    sequence_number BIGINT,
+    operation TEXT NOT NULL,
+
+    -- File metrics
+    added_data_files BIGINT DEFAULT 0,
+    removed_data_files BIGINT DEFAULT 0,
+    total_data_files BIGINT DEFAULT 0,
+    added_delete_files BIGINT DEFAULT 0,
+    removed_delete_files BIGINT DEFAULT 0,
+    total_delete_files BIGINT DEFAULT 0,
+
+    -- Equality delete files
+    added_equality_delete_files BIGINT DEFAULT 0,
+    removed_equality_delete_files BIGINT DEFAULT 0,
+
+    -- Positional delete files
+    added_positional_delete_files BIGINT DEFAULT 0,
+    removed_positional_delete_files BIGINT DEFAULT 0,
+
+    -- Record metrics
+    added_records BIGINT DEFAULT 0,
+    removed_records BIGINT DEFAULT 0,
+    total_records BIGINT DEFAULT 0,
+
+    -- Size metrics
+    added_file_size_bytes BIGINT DEFAULT 0,
+    removed_file_size_bytes BIGINT DEFAULT 0,
+    total_file_size_bytes BIGINT DEFAULT 0,
+
+    -- Duration and attempts
+    total_duration_ms BIGINT DEFAULT 0,
+    attempts INTEGER DEFAULT 1,
+
+    -- Additional metadata (for extensibility)
+    metadata TEXT DEFAULT '{}',
+
+    PRIMARY KEY (realm_id, report_id)
+);
+
+COMMENT ON TABLE commit_metrics_report IS 'Commit metrics reports as first-class entities';
+
+-- Index for retention cleanup by timestamp
+CREATE INDEX IF NOT EXISTS idx_commit_report_timestamp ON commit_metrics_report(realm_id, timestamp_ms);