Drain JDBC realm time-series data via a batched purge command

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

@@ -404,6 +404,12 @@ public void deleteAll(@NonNull PolarisCallContext callCtx) {
                     ModelPolicyMappingRecord.ALL_COLUMNS,
                     ModelPolicyMappingRecord.TABLE_NAME,
                     params));
+            // Queue time-series data (events, metrics) for async batched purge instead of
+            // deleting inline. realm_purge exists from schema v5 only.
+            if (schemaVersion >= 5) {
+              new RealmPurgeStore(datasourceOperations)
+                  .enqueueInTransaction(connection, realmId, System.currentTimeMillis());
+            }
             return true;
           });
     } catch (SQLException e) {

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

@@ -25,12 +25,15 @@
 import jakarta.enterprise.inject.Instance;
 import jakarta.enterprise.inject.Produces;
 import jakarta.inject.Inject;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.sql.SQLException;
 import java.time.Clock;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import javax.sql.DataSource;
 import org.apache.polaris.core.PolarisCallContext;
@@ -87,6 +90,7 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   @Inject PolarisDiagnostics diagnostics;
   @Inject DatasourceOperations datasourceOperations;
   @Inject RealmConfigurationSource realmConfigurationSource;
+  @Inject Instance<RealmDataPurger> realmDataPurgers;
 
   protected JdbcMetaStoreManagerFactory() {}
 
@@ -226,6 +230,7 @@ public synchronized Map<String, BaseResult> purgeRealms(Iterable<String> realms)
         continue;
       }
 
+      // purge -> deleteAll enqueues the realm into realm_purge in the same transaction.
       BaseResult result = metaStoreManager.purge(callContext);
       results.put(realm, result);
 
@@ -238,6 +243,30 @@ public synchronized Map<String, BaseResult> purgeRealms(Iterable<String> realms)
     return Map.copyOf(results);
   }
 
+  @Override
+  public int drainRealmPurges(int batchSize, long claimLeaseMs) {
+    RealmPurgeService service =
+        new RealmPurgeService(
+            new RealmPurgeStore(datasourceOperations), realmDataPurgers.stream().toList());
+    return service.drainOnce(nodeId(), batchSize, claimLeaseMs, clock.millis());
+  }
+
+  // Resolved once: getLocalHost() can block for seconds on a misconfigured resolver.
+  private static final String NODE_HOST = resolveHost();
+
+  private static String resolveHost() {
+    try {
+      return InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      return "unknown-host";
+    }
+  }
+
+  // Claim owner: hostname (so a stuck lease points at a node) plus a UUID for per-run uniqueness.
+  private static String nodeId() {
+    return NODE_HOST + "-" + UUID.randomUUID();
+  }
+
   @Override
   public PolarisMetaStoreManager getOrCreateMetaStoreManager(RealmContext realmContext) {
     // Stateless — create a fresh instance on every call, no caching needed

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

@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+package org.apache.polaris.persistence.relational.jdbc;
+
+/**
+ * Deletes a single realm's rows from one purgeable store in bounded batches. Implementations are
+ * discovered by the background purge drainer, so a new store (e.g. relocated metrics persistence)
+ * can plug in by providing an implementation without changing the drainer.
+ */
+public interface RealmDataPurger {
+
+  /** Stable identifier for the store this purger drains, used for logging and metrics. */
+  String storeName();
+
+  /**
+   * Deletes up to {@code batchSize} rows belonging to {@code realmId}.
+   *
+   * @return the number of rows deleted; {@code 0} means the store is fully drained for the realm.
+   */
+  long purgeBatch(String realmId, int batchSize);
+}

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

@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+package org.apache.polaris.persistence.relational.jdbc;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import org.apache.polaris.persistence.relational.jdbc.models.ModelEvent;
+
+/**
+ * Produces a {@link RealmDataPurger} for each realm-scoped time-series table defined in this
+ * module's schema (see {@code schema-v*.sql}). The drainer discovers them via {@code
+ * Instance<RealmDataPurger>}, so adding a table means adding a producer here, not editing the
+ * drainer.
+ *
+ * <p>The metrics report tables' write path moved to the metrics extension (Apache Polaris #5068),
+ * but the tables themselves remain in this schema, so purging them stays a concern of this module.
+ */
+@ApplicationScoped
+public class RealmDataPurgerProducer {
+
+  // Time-series tables owned by this module's schema. events has a ModelEvent constant; the metrics
+  // report tables no longer have a Model class here (moved in #5068), so they are named directly.
+  static final String SCAN_METRICS_REPORT = "scan_metrics_report";
+  static final String COMMIT_METRICS_REPORT = "commit_metrics_report";
+
+  @Produces
+  @ApplicationScoped
+  RealmDataPurger eventsPurger(DatasourceOperations ops) {
+    return new TableRealmDataPurger(ops, ModelEvent.TABLE_NAME, "event_id");
+  }
+
+  @Produces
+  @ApplicationScoped
+  RealmDataPurger scanMetricsReportPurger(DatasourceOperations ops) {
+    return new TableRealmDataPurger(ops, SCAN_METRICS_REPORT, "report_id");
+  }
+
+  @Produces
+  @ApplicationScoped
+  RealmDataPurger commitMetricsReportPurger(DatasourceOperations ops) {
+    return new TableRealmDataPurger(ops, COMMIT_METRICS_REPORT, "report_id");
+  }
+}

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

@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+package org.apache.polaris.persistence.relational.jdbc;
+
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Drives the async, batched purge of realm-scoped time-series data. Realms are enqueued into {@code
+ * realm_purge} on purge; this service later drains them in bounded batches. Per-store deletion is
+ * delegated to pluggable {@link RealmDataPurger}s.
+ */
+public class RealmPurgeService {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(RealmPurgeService.class);
+
+  private final RealmPurgeStore store;
+  private final List<RealmDataPurger> purgers;
+
+  public RealmPurgeService(RealmPurgeStore store, List<RealmDataPurger> purgers) {
+    this.store = store;
+    this.purgers = List.copyOf(purgers);
+  }
+
+  /** Records that {@code realmId}'s time-series data should be purged asynchronously. */
+  public void enqueue(String realmId, long nowMs) {
+    store.enqueue(realmId, nowMs);
+  }
+
+  /**
+   * Drains all currently claimable realms, deleting each realm's data in {@code batchSize} batches
+   * until empty, then removing its marker. A realm owned by another node is skipped; a realm whose
+   * drain is interrupted resumes on a later run once its claim lease expires.
+   *
+   * @param node identifier of the draining node, recorded on the claim
+   * @param batchSize maximum rows deleted per statement
+   * @param claimLeaseMs how long a claim is honored before another node may steal it
+   * @param nowMs current wall-clock time in millis
+   * @return the number of realms fully drained during this run
+   */
+  public int drainOnce(String node, int batchSize, long claimLeaseMs, long nowMs) {
+    // A non-positive batch would delete nothing yet still clear the marker, discarding data.
+    if (batchSize <= 0) {
+      throw new IllegalArgumentException("batchSize must be strictly positive, got " + batchSize);
+    }
+    long staleClaimBeforeMs = nowMs - claimLeaseMs;
+    int drained = 0;
+    for (String realmId : store.listClaimable(staleClaimBeforeMs)) {
+      if (!store.claim(realmId, node, nowMs, staleClaimBeforeMs)) {
+        continue; // another node owns this realm
+      }
+      // Isolate each realm: a failure must not abort the run. The failed realm keeps its marker
+      // and is retried once its claim lease expires.
+      try {
+        drainRealm(realmId, batchSize);
+        store.complete(realmId);
+        drained++;
+      } catch (RuntimeException e) {
+        LOGGER.warn(
+            "Failed to drain realm {} for purge; leaving it queued for retry on a later run",
+            realmId,
+            e);
+      }
+    }
+    return drained;
+  }
+
+  private void drainRealm(String realmId, int batchSize) {
+    boolean anyRemaining = true;
+    while (anyRemaining) {
+      anyRemaining = false;
+      for (RealmDataPurger purger : purgers) {
+        if (purger.purgeBatch(realmId, batchSize) > 0) {
+          anyRemaining = true;
+        }
+      }
+    }
+  }
+}