Re: [PR] Drain JDBC realm time-series data via a batched purge command [polaris]

["flyingImer (via GitHub)" <gi...@apache.org>]

flyingImer commented on code in PR #5089:
URL: https://github.com/apache/polaris/pull/5089#discussion_r3963604547


##########
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) {

Review Comment:
   If I didn't miss any context, this needs a migration path for existing databases and support in the current bootstrap schema. Existing v5 databases and the documented v3/v4-to-v5 upgrade do not create realm_purge, so the new enqueue fails and rolls back purge. Current main also bootstraps v6, whose scripts do not create this table. Could we install the queue on both paths and cover them with upgrade and fresh-bootstrap tests?
   



##########
persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/TableRealmDataPurger.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * 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.sql.SQLException;
+import java.util.List;
+import org.apache.polaris.persistence.relational.jdbc.QueryGenerator.PreparedQuery;
+
+/**
+ * A {@link RealmDataPurger} that deletes rows from a single realm-scoped table in bounded batches.
+ *
+ * <p>Uses a portable {@code DELETE ... WHERE realm_id = ? AND <key> IN (SELECT <key> ... WHERE
+ * realm_id = ? LIMIT ?)}: the {@code LIMIT} lives inside a subquery because {@code DELETE ...
+ * LIMIT} is not supported by PostgreSQL (it is by CockroachDB and H2), and the outer {@code
+ * realm_id} predicate keeps the delete scoped to the realm even when the key column is only unique
+ * within a realm (e.g. metrics {@code report_id}).
+ */
+public class TableRealmDataPurger implements RealmDataPurger {
+
+  private final DatasourceOperations datasourceOperations;
+  private final String tableName;
+  private final String keyColumn;
+
+  /**
+   * @param tableName realm-scoped table to purge
+   * @param keyColumn a column that is unique within a realm (used to bound each batch)
+   */
+  public TableRealmDataPurger(
+      DatasourceOperations datasourceOperations, String tableName, String keyColumn) {
+    this.datasourceOperations = datasourceOperations;
+    this.tableName = tableName;
+    this.keyColumn = keyColumn;
+  }
+
+  @Override
+  public String storeName() {
+    return tableName;
+  }
+
+  @Override
+  public long purgeBatch(String realmId, int batchSize) {
+    String fqTable = QueryGenerator.getFullyQualifiedTableName(tableName);
+    String sql =
+        "DELETE FROM "
+            + fqTable
+            + " WHERE realm_id = ? AND "

Review Comment:
   IIUC, this can also delete events and metrics from a realm bootstrapped after the purge. bootstrapRealms does not check the pending marker, so purge R, bootstrap R, new writes, then purge-drain reaches this DELETE with the same realm_id. 
   
   If so, could we preserve the new realm's data, either by preventing bootstrap while cleanup is outstanding or by scoping the task to the retired realm incarnation? A regression test for that sequence would make the lifecycle guarantee explicit IMO
   



##########
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)) {

Review Comment:
   IIRC, this lease does not enforce the documented one-drainer-per-realm guarantee. nowMs is reused for the whole run and the batch loop never renews or checks ownership, so another node can take an expired claim while the first keeps deleting, duplicating database work. The first node can then call complete, which removes the marker without checking the current owner. Could we use fresh claim times and retain ownership checks through draining and completion, with an expiry/takeover test?
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@polaris.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org