diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java index b2c7f88c4cc..7f2baec5171 100644 --- a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java +++ b/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) { diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java index c67bbccb67b..a23992d9cb5 100644 --- a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java +++ b/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 realmDataPurgers; protected JdbcMetaStoreManagerFactory() {} @@ -226,6 +230,7 @@ public synchronized Map purgeRealms(Iterable 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 purgeRealms(Iterable 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 diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmDataPurger.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmDataPurger.java new file mode 100644 index 00000000000..c20f3f40fed --- /dev/null +++ b/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); +} diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmDataPurgerProducer.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmDataPurgerProducer.java new file mode 100644 index 00000000000..2cb61672a6c --- /dev/null +++ b/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}, so adding a table means adding a producer here, not editing the + * drainer. + * + *

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"); + } +} diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeService.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeService.java new file mode 100644 index 00000000000..b28ba17ec25 --- /dev/null +++ b/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 purgers; + + public RealmPurgeService(RealmPurgeStore store, List 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; + } + } + } + } +} diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeStore.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeStore.java new file mode 100644 index 00000000000..f279740411f --- /dev/null +++ b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeStore.java @@ -0,0 +1,134 @@ +/* + * 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.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.List; +import java.util.Map; +import org.apache.polaris.persistence.relational.jdbc.QueryGenerator.PreparedQuery; +import org.apache.polaris.persistence.relational.jdbc.models.Converter; + +/** + * DAO for the {@code realm_purge} queue of realms whose time-series data is pending async batched + * deletion. Cross-node coordination is via the {@link #claim} conditional update plus a lease. + */ +public class RealmPurgeStore { + + private static final String TABLE = "realm_purge"; + + private final DatasourceOperations datasourceOperations; + + public RealmPurgeStore(DatasourceOperations datasourceOperations) { + this.datasourceOperations = datasourceOperations; + } + + private PreparedQuery enqueueQuery(String realmId, long requestedAtMs) { + // ON CONFLICT DO NOTHING keeps enqueue idempotent (PostgreSQL, CockroachDB, H2). + String sql = + "INSERT INTO " + + QueryGenerator.getFullyQualifiedTableName(TABLE) + + " (realm_id, requested_at, status) VALUES (?, ?, 'pending') ON CONFLICT DO NOTHING"; + return new PreparedQuery(sql, List.of(realmId, requestedAtMs)); + } + + /** Records {@code realmId} as pending purge; a no-op if it is already queued. */ + public void enqueue(String realmId, long requestedAtMs) { + try { + datasourceOperations.executeUpdate(enqueueQuery(realmId, requestedAtMs)); + } catch (SQLException e) { + throw new RuntimeException( + String.format("Failed to enqueue realm %s for purge due to %s", realmId, e.getMessage()), + e); + } + } + + /** + * Enqueues on an existing transaction {@code connection}, committing atomically with the caller. + */ + public void enqueueInTransaction(Connection connection, String realmId, long requestedAtMs) + throws SQLException { + datasourceOperations.execute(connection, enqueueQuery(realmId, requestedAtMs)); + } + + /** + * Returns realms available to drain: {@code pending}, or {@code in_progress} with an expired + * lease. + */ + public List listClaimable(long staleClaimBeforeMs) { + String sql = + "SELECT realm_id FROM " + + QueryGenerator.getFullyQualifiedTableName(TABLE) + + " WHERE status = 'pending' OR (status = 'in_progress' AND claimed_at < ?)"; + try { + return datasourceOperations.executeSelect( + new PreparedQuery(sql, List.of(staleClaimBeforeMs)), new RealmIdConverter()); + } catch (SQLException e) { + throw new RuntimeException( + "Failed to list claimable realm purges due to " + e.getMessage(), e); + } + } + + /** + * Atomically claims {@code realmId} for {@code node}, succeeding only if it is pending or its + * previous claim has expired — one active drainer per realm without leader election. + */ + public boolean claim(String realmId, String node, long nowMs, long staleClaimBeforeMs) { + String sql = + "UPDATE " + + QueryGenerator.getFullyQualifiedTableName(TABLE) + + " SET status = 'in_progress', claimed_by = ?, claimed_at = ?" + + " WHERE realm_id = ? AND (status = 'pending' OR claimed_at < ?)"; + try { + return datasourceOperations.executeUpdate( + new PreparedQuery(sql, List.of(node, nowMs, realmId, staleClaimBeforeMs))) + > 0; + } catch (SQLException e) { + throw new RuntimeException( + String.format("Failed to claim realm %s for purge due to %s", realmId, e.getMessage()), + e); + } + } + + /** Removes the marker once a realm has been fully drained. */ + public void complete(String realmId) { + String sql = + "DELETE FROM " + QueryGenerator.getFullyQualifiedTableName(TABLE) + " WHERE realm_id = ?"; + try { + datasourceOperations.executeUpdate(new PreparedQuery(sql, List.of(realmId))); + } catch (SQLException e) { + throw new RuntimeException( + String.format("Failed to complete purge of realm %s due to %s", realmId, e.getMessage()), + e); + } + } + + private static final class RealmIdConverter implements Converter { + @Override + public String fromResultSet(ResultSet rs) throws SQLException { + return rs.getString("realm_id"); + } + + @Override + public Map toMap(DatabaseType databaseType) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/TableRealmDataPurger.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/TableRealmDataPurger.java new file mode 100644 index 00000000000..5c2a18611ab --- /dev/null +++ b/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. + * + *

Uses a portable {@code DELETE ... WHERE realm_id = ? AND IN (SELECT ... 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 " + + keyColumn + + " IN (SELECT " + + keyColumn + + " FROM " + + fqTable + + " WHERE realm_id = ? LIMIT ?)"; + try { + return datasourceOperations.executeUpdate( + new PreparedQuery(sql, List.of(realmId, realmId, batchSize))); + } catch (SQLException e) { + throw new RuntimeException( + String.format( + "Failed to purge batch from %s for realm %s due to %s", + tableName, realmId, e.getMessage()), + e); + } + } +} diff --git a/persistence/relational-jdbc/src/main/resources/cockroachdb/schema-v5.sql b/persistence/relational-jdbc/src/main/resources/cockroachdb/schema-v5.sql index 3450329d5b8..6d3cbc7ae65 100644 --- a/persistence/relational-jdbc/src/main/resources/cockroachdb/schema-v5.sql +++ b/persistence/relational-jdbc/src/main/resources/cockroachdb/schema-v5.sql @@ -278,3 +278,14 @@ CREATE INDEX IF NOT EXISTS idx_commit_report_lookup ON commit_metrics_report(rea -- CockroachDB requires explicit INT4 type declarations to correctly map columns to Java's Integer type. -- Using generic INTEGER or INT types causes type mapping failures in CockroachDB's JDBC driver. -- INT4 is equivalent to INTEGER in PostgreSQL, ensuring compatibility with both databases. + +-- Queue of realms whose time-series data (events, metrics reports) is pending asynchronous, +-- batched deletion by the background purge drainer. One row per realm being purged. +CREATE TABLE IF NOT EXISTS realm_purge ( + realm_id TEXT NOT NULL, + requested_at BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + claimed_by TEXT, + claimed_at BIGINT, + PRIMARY KEY (realm_id) +); diff --git a/persistence/relational-jdbc/src/main/resources/h2/schema-v5.sql b/persistence/relational-jdbc/src/main/resources/h2/schema-v5.sql index de7fb2484bd..e44aaf7adab 100644 --- a/persistence/relational-jdbc/src/main/resources/h2/schema-v5.sql +++ b/persistence/relational-jdbc/src/main/resources/h2/schema-v5.sql @@ -272,3 +272,14 @@ CREATE INDEX IF NOT EXISTS idx_commit_report_timestamp ON commit_metrics_report( -- Index for query lookups by catalog_id and table_id CREATE INDEX IF NOT EXISTS idx_commit_report_lookup ON commit_metrics_report(realm_id, catalog_id, table_id, timestamp_ms); + +-- Queue of realms whose time-series data (events, metrics reports) is pending asynchronous, +-- batched deletion by the background purge drainer. One row per realm being purged. +CREATE TABLE IF NOT EXISTS realm_purge ( + realm_id TEXT NOT NULL, + requested_at BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + claimed_by TEXT, + claimed_at BIGINT, + PRIMARY KEY (realm_id) +); diff --git a/persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql b/persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql index 7b4e7a07196..698d1c4a052 100644 --- a/persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql +++ b/persistence/relational-jdbc/src/main/resources/postgres/schema-v5.sql @@ -272,3 +272,14 @@ CREATE INDEX IF NOT EXISTS idx_commit_report_timestamp ON commit_metrics_report( -- Index for query lookups by catalog_id and table_id CREATE INDEX IF NOT EXISTS idx_commit_report_lookup ON commit_metrics_report(realm_id, catalog_id, table_id, timestamp_ms); + +-- Queue of realms whose time-series data (events, metrics reports) is pending asynchronous, +-- batched deletion by the background purge drainer. One row per realm being purged. +CREATE TABLE IF NOT EXISTS realm_purge ( + realm_id TEXT NOT NULL, + requested_at BIGINT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + claimed_by TEXT, + claimed_at BIGINT, + PRIMARY KEY (realm_id) +); diff --git a/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImplTest.java b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImplTest.java index b41f131b3a9..fc74bb1651b 100644 --- a/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImplTest.java +++ b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImplTest.java @@ -30,18 +30,26 @@ import java.io.IOException; import java.io.InputStream; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.UUID; import org.apache.polaris.core.PolarisCallContext; import org.apache.polaris.core.PolarisDefaultDiagServiceImpl; import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.entity.EventEntity; import org.apache.polaris.core.entity.PolarisBaseEntity; import org.apache.polaris.core.entity.PolarisChangeTrackingVersions; import org.apache.polaris.core.entity.PolarisEntityId; import org.apache.polaris.core.entity.PolarisEntitySubType; import org.apache.polaris.core.entity.PolarisEntityType; import org.apache.polaris.core.entity.PolarisGrantRecord; +import org.apache.polaris.core.policy.PolarisPolicyMappingRecord; +import org.apache.polaris.core.policy.PredefinedPolicyTypes; import org.h2.jdbcx.JdbcConnectionPool; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -57,6 +65,7 @@ class JdbcBasePersistenceImplTest { private static final long GRANTEE_CATALOG_ID = 3L; private static final long GRANTEE_ID = 4L; private static final int PRIVILEGE_CODE = 21; + private static final int TIME_SERIES_ROWS = 5; @ParameterizedTest @ValueSource(ints = {1, 2, 3, 4, 5}) @@ -256,6 +265,192 @@ void hasChildrenUsesExistsQueryNotFullScan(int schemaVersion) throws SQLExceptio assertThat(sql).doesNotContain("internal_properties"); } + @ParameterizedTest + @ValueSource(ints = {4, 5}) // v4: realm_purge absent (guard skips enqueue); v5: real path + void deleteAllPurgesMetastoreMetadataButNotTimeSeriesDataForCurrentRealmOnly(int schemaVersion) + throws SQLException, IOException { + JdbcConnectionPool dataSource = + JdbcConnectionPool.create( + "jdbc:h2:mem:delete_all_v" + + schemaVersion + + "_" + + System.nanoTime() + + ";DB_CLOSE_DELAY=-1;MODE=PostgreSQL", + "sa", + ""); + DatasourceOperations datasourceOperations = + new DatasourceOperations(dataSource, new TestJdbcConfiguration()); + try (InputStream script = DatabaseType.H2.openInitScriptResource(schemaVersion)) { + datasourceOperations.executeScript(script); + } + + RealmContext otherRealm = () -> "OTHER_REALM"; + + // Populate every realm-scoped table for both the target realm and another realm. + populateAllTables(datasourceOperations, REALM_CONTEXT, schemaVersion); + populateAllTables(datasourceOperations, otherRealm, schemaVersion); + + List tables = expectedRealmScopedTables(schemaVersion); + for (String table : tables) { + assertThat(countRows(dataSource, table, REALM_CONTEXT.getRealmIdentifier())) + .as("rows in %s for target realm before deleteAll", table) + .isPositive(); + } + + JdbcBasePersistenceImpl impl = + new JdbcBasePersistenceImpl( + new PolarisDefaultDiagServiceImpl(), + datasourceOperations, + RANDOM_SECRETS, + REALM_CONTEXT.getRealmIdentifier(), + schemaVersion); + impl.deleteAll(new PolarisCallContext(REALM_CONTEXT, impl)); + + // First four tables are metastore metadata (purged); the rest are time-series data (left). + List metastoreTables = tables.subList(0, 4); + List timeSeriesTables = tables.subList(4, tables.size()); + + for (String table : metastoreTables) { + assertThat(countRows(dataSource, table, REALM_CONTEXT.getRealmIdentifier())) + .as("metastore rows in %s for target realm after deleteAll", table) + .isZero(); + assertThat(countRows(dataSource, table, otherRealm.getRealmIdentifier())) + .as("rows in %s for other realm must be untouched", table) + .isPositive(); + } + + for (String table : timeSeriesTables) { + assertThat(countRows(dataSource, table, REALM_CONTEXT.getRealmIdentifier())) + .as("time-series rows in %s must survive deleteAll (purged asynchronously)", table) + .isPositive(); + assertThat(countRows(dataSource, table, otherRealm.getRealmIdentifier())) + .as("rows in %s for other realm must be untouched", table) + .isPositive(); + } + } + + private static List expectedRealmScopedTables(int schemaVersion) { + List tables = + new ArrayList<>( + List.of( + "entities", + "grant_records", + "principal_authentication_data", + "policy_mapping_record")); + if (schemaVersion >= 3) { + tables.add("events"); + } + if (schemaVersion >= 4) { + tables.add("scan_metrics_report"); + tables.add("commit_metrics_report"); + } + return tables; + } + + private static void populateAllTables( + DatasourceOperations datasourceOperations, RealmContext realm, int schemaVersion) { + JdbcBasePersistenceImpl impl = + new JdbcBasePersistenceImpl( + new PolarisDefaultDiagServiceImpl(), + datasourceOperations, + RANDOM_SECRETS, + realm.getRealmIdentifier(), + schemaVersion); + PolarisCallContext callCtx = new PolarisCallContext(realm, impl); + + PolarisBaseEntity entity = + new PolarisBaseEntity.Builder() + .id(1L) + .catalogId(0L) + .parentId(0L) + .typeCode(PolarisEntityType.PRINCIPAL.getCode()) + .subTypeCode(PolarisEntitySubType.NULL_SUBTYPE.getCode()) + .name("entity-" + realm.getRealmIdentifier()) + .entityVersion(1) + .grantRecordsVersion(1) + .createTimestamp(System.currentTimeMillis()) + .build(); + impl.writeEntity(callCtx, entity, false, null); + + impl.writeToGrantRecords( + callCtx, + new PolarisGrantRecord( + SECURABLE_CATALOG_ID, SECURABLE_ID, GRANTEE_CATALOG_ID, GRANTEE_ID, PRIVILEGE_CODE)); + + impl.generateNewPrincipalSecrets(callCtx, "principal-" + realm.getRealmIdentifier(), 1L); + + impl.writeToPolicyMappingRecords( + callCtx, + new PolarisPolicyMappingRecord( + 1L, 2L, 3L, 4L, PredefinedPolicyTypes.DATA_COMPACTION.getCode(), "{}")); + + if (schemaVersion >= 3) { + for (int i = 0; i < TIME_SERIES_ROWS; i++) { + impl.writeEvents( + List.of( + new EventEntity( + "catalog-1", + UUID.randomUUID().toString(), + "req-" + i, + "TEST_EVENT", + System.currentTimeMillis(), + "principal-1", + EventEntity.ResourceType.CATALOG, + "resource-1"))); + } + } + + if (schemaVersion < 4) { + return; + } + + // Metrics report tables live in the relational-jdbc schema, but their write API moved to the + // metrics extension (#5068), so rows are inserted directly here. + String realmId = realm.getRealmIdentifier(); + for (int i = 0; i < TIME_SERIES_ROWS; i++) { + insertRow( + datasourceOperations, + "scan_metrics_report", + "(report_id, realm_id, catalog_id, table_id, timestamp_ms) VALUES (?, ?, ?, ?, ?)", + List.of(UUID.randomUUID().toString(), realmId, 1L, 2L, 1L)); + insertRow( + datasourceOperations, + "commit_metrics_report", + "(report_id, realm_id, catalog_id, table_id, timestamp_ms, snapshot_id, operation)" + + " VALUES (?, ?, ?, ?, ?, ?, ?)", + List.of(UUID.randomUUID().toString(), realmId, 1L, 2L, 1L, 3L, "append")); + } + } + + private static void insertRow( + DatasourceOperations ops, String table, String columnsAndValues, List params) { + try { + ops.executeUpdate( + new QueryGenerator.PreparedQuery( + "INSERT INTO " + + QueryGenerator.getFullyQualifiedTableName(table) + + " " + + columnsAndValues, + params)); + } catch (SQLException e) { + throw new RuntimeException(e); + } + } + + private static long countRows(JdbcConnectionPool dataSource, String table, String realmId) + throws SQLException { + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = + connection.prepareStatement( + "SELECT COUNT(*) FROM POLARIS_SCHEMA." + table + " WHERE realm_id = ?")) { + statement.setString(1, realmId); + try (ResultSet rs = statement.executeQuery()) { + rs.next(); + return rs.getLong(1); + } + } + } + private static final class TestJdbcConfiguration implements RelationalJdbcConfiguration { @Override public Optional maxRetries() { diff --git a/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeTest.java b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeTest.java new file mode 100644 index 00000000000..297d8c2516f --- /dev/null +++ b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/RealmPurgeTest.java @@ -0,0 +1,289 @@ +/* + * 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 static org.apache.polaris.core.persistence.PrincipalSecretsGenerator.RANDOM_SECRETS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.io.InputStream; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import org.apache.polaris.core.PolarisCallContext; +import org.apache.polaris.core.PolarisDefaultDiagServiceImpl; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.entity.EventEntity; +import org.apache.polaris.persistence.relational.jdbc.QueryGenerator.PreparedQuery; +import org.apache.polaris.persistence.relational.jdbc.models.Converter; +import org.apache.polaris.persistence.relational.jdbc.models.ModelEvent; +import org.h2.jdbcx.JdbcConnectionPool; +import org.junit.jupiter.api.Test; + +class RealmPurgeTest { + + private static final int SCHEMA_VERSION = 5; + private static final String SCAN_METRICS_TABLE = RealmDataPurgerProducer.SCAN_METRICS_REPORT; + private static final String COMMIT_METRICS_TABLE = RealmDataPurgerProducer.COMMIT_METRICS_REPORT; + + private DatasourceOperations newH2(String label) throws IOException, SQLException { + JdbcConnectionPool ds = + JdbcConnectionPool.create( + "jdbc:h2:mem:" + label + "_" + System.nanoTime() + ";DB_CLOSE_DELAY=-1;MODE=PostgreSQL", + "sa", + ""); + DatasourceOperations ops = new DatasourceOperations(ds, new TestJdbcConfiguration()); + try (InputStream script = DatabaseType.H2.openInitScriptResource(SCHEMA_VERSION)) { + ops.executeScript(script); + } + return ops; + } + + private void writeEvents(DatasourceOperations ops, String realmId, int count) { + JdbcBasePersistenceImpl impl = + new JdbcBasePersistenceImpl( + new PolarisDefaultDiagServiceImpl(), ops, RANDOM_SECRETS, realmId, SCHEMA_VERSION); + for (int i = 0; i < count; i++) { + impl.writeEvents( + List.of( + new EventEntity( + "catalog-1", + UUID.randomUUID().toString(), + "req-" + i, + "TEST_EVENT", + System.currentTimeMillis(), + "principal-1", + EventEntity.ResourceType.CATALOG, + "resource-1"))); + } + } + + // Metrics report tables live in the relational-jdbc schema, but their write API moved to the + // metrics extension (#5068), so rows are inserted directly (only the row's presence matters + // here). + private void writeMetrics(DatasourceOperations ops, String realmId, int count) + throws SQLException { + for (int i = 0; i < count; i++) { + ops.executeUpdate( + new PreparedQuery( + "INSERT INTO " + + QueryGenerator.getFullyQualifiedTableName(SCAN_METRICS_TABLE) + + " (report_id, realm_id, catalog_id, table_id, timestamp_ms) VALUES (?, ?, ?, ?, ?)", + List.of(UUID.randomUUID().toString(), realmId, 1L, 2L, 1L))); + ops.executeUpdate( + new PreparedQuery( + "INSERT INTO " + + QueryGenerator.getFullyQualifiedTableName(COMMIT_METRICS_TABLE) + + " (report_id, realm_id, catalog_id, table_id, timestamp_ms, snapshot_id, operation)" + + " VALUES (?, ?, ?, ?, ?, ?, ?)", + List.of(UUID.randomUUID().toString(), realmId, 1L, 2L, 1L, 3L, "append"))); + } + } + + private long countRows(DatasourceOperations ops, String table, String realmId) + throws SQLException { + String sql = + "SELECT COUNT(*) AS c FROM " + + QueryGenerator.getFullyQualifiedTableName(table) + + " WHERE realm_id = ?"; + return ops.executeSelect(new PreparedQuery(sql, List.of(realmId)), new CountConverter()).get(0); + } + + // The standard set of purgers, mirroring RealmDataPurgerProducer (which supplies them via CDI in + // production). + private RealmPurgeService newService(DatasourceOperations ops) { + return new RealmPurgeService( + new RealmPurgeStore(ops), + List.of( + new TableRealmDataPurger(ops, ModelEvent.TABLE_NAME, "event_id"), + new TableRealmDataPurger(ops, SCAN_METRICS_TABLE, "report_id"), + new TableRealmDataPurger(ops, COMMIT_METRICS_TABLE, "report_id"))); + } + + @Test + void markerStoreEnqueueClaimComplete() throws IOException, SQLException { + DatasourceOperations ops = newH2("marker"); + RealmPurgeStore store = new RealmPurgeStore(ops); + + store.enqueue("R1", 1000L); + store.enqueue("R1", 2000L); // idempotent, still one row + assertThat(store.listClaimable(Long.MAX_VALUE)).containsExactly("R1"); + + // First claim wins; a second claim within the same lease window loses. + assertThat(store.claim("R1", "nodeA", 5000L, 4000L)).isTrue(); + assertThat(store.claim("R1", "nodeB", 5001L, 4000L)).isFalse(); + + // Not claimable while the claim is fresh (claimed_at=5000 is not < 4000)... + assertThat(store.listClaimable(4000L)).isEmpty(); + // ...but reclaimable once the lease is considered expired (claimed_at=5000 < 6000). + assertThat(store.listClaimable(6000L)).containsExactly("R1"); + assertThat(store.claim("R1", "nodeB", 7000L, 6000L)).isTrue(); + + store.complete("R1"); + assertThat(store.listClaimable(Long.MAX_VALUE)).isEmpty(); + } + + @Test + void purgeBatchDrainsOnlyTargetRealmInBatches() throws IOException, SQLException { + DatasourceOperations ops = newH2("purge"); + writeEvents(ops, "R1", 25); + writeEvents(ops, "R2", 7); + + TableRealmDataPurger purger = new TableRealmDataPurger(ops, ModelEvent.TABLE_NAME, "event_id"); + + long total = 0; + int batches = 0; + long deleted; + while ((deleted = purger.purgeBatch("R1", 10)) > 0) { + assertThat(deleted).isLessThanOrEqualTo(10); + total += deleted; + batches++; + } + + assertThat(total).isEqualTo(25); + assertThat(batches).isEqualTo(3); // 10 + 10 + 5 + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R1")).isZero(); + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R2")).isEqualTo(7); // other realm untouched + } + + @Test + void drainOnceRemovesTimeSeriesDataAndClearsMarker() throws IOException, SQLException { + DatasourceOperations ops = newH2("drain"); + writeEvents(ops, "R1", 23); + writeMetrics(ops, "R1", 4); + writeEvents(ops, "R2", 5); + + RealmPurgeService service = newService(ops); + service.enqueue("R1", 1000L); + + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R1")).isEqualTo(23); + assertThat(countRows(ops, SCAN_METRICS_TABLE, "R1")).isEqualTo(4); + assertThat(countRows(ops, COMMIT_METRICS_TABLE, "R1")).isEqualTo(4); + + int drained = service.drainOnce("nodeA", 10, 60_000L, 100_000L); + + assertThat(drained).isEqualTo(1); + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R1")).isZero(); + assertThat(countRows(ops, SCAN_METRICS_TABLE, "R1")).isZero(); + assertThat(countRows(ops, COMMIT_METRICS_TABLE, "R1")).isZero(); + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R2")).isEqualTo(5); // other realm untouched + assertThat(new RealmPurgeStore(ops).listClaimable(Long.MAX_VALUE)).isEmpty(); // marker cleared + } + + @Test + void drainOnceRejectsNonPositiveBatchSizeInsteadOfClearingMarkers() + throws IOException, SQLException { + DatasourceOperations ops = newH2("batch-guard"); + RealmPurgeStore store = new RealmPurgeStore(ops); + store.enqueue("R1", 1000L); + + RealmPurgeService service = newService(ops); + assertThatThrownBy(() -> service.drainOnce("nodeA", 0, 60_000L, 100_000L)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("batchSize"); + + assertThat(store.listClaimable(Long.MAX_VALUE)).containsExactly("R1"); // marker survives + } + + @Test + void drainOnceIsolatesAFailingRealmAndDrainsTheRest() throws IOException, SQLException { + DatasourceOperations ops = newH2("drain-isolate"); + RealmPurgeStore store = new RealmPurgeStore(ops); + store.enqueue("BAD", 1000L); + store.enqueue("GOOD", 1000L); + + // A purger that fails only for realm "BAD". + RealmDataPurger flaky = + new RealmDataPurger() { + @Override + public String storeName() { + return "flaky"; + } + + @Override + public long purgeBatch(String realmId, int batchSize) { + if ("BAD".equals(realmId)) { + throw new RuntimeException("boom"); + } + return 0; // "GOOD" is already empty + } + }; + RealmPurgeService service = new RealmPurgeService(store, List.of(flaky)); + + int drained = service.drainOnce("nodeA", 10, 60_000L, 100_000L); + + assertThat(drained).isEqualTo(1); // healthy realm drained + assertThat(store.listClaimable(Long.MAX_VALUE)).containsExactly("BAD"); // failed realm retried + } + + @Test + void deleteAllAtomicallyEnqueuesRealmForPurgeAndLeavesTimeSeriesData() + throws IOException, SQLException { + DatasourceOperations ops = newH2("deleteall-enqueue"); + writeEvents(ops, "R1", 3); + + RealmContext realm = () -> "R1"; + JdbcBasePersistenceImpl impl = + new JdbcBasePersistenceImpl( + new PolarisDefaultDiagServiceImpl(), ops, RANDOM_SECRETS, "R1", SCHEMA_VERSION); + impl.deleteAll(new PolarisCallContext(realm, impl)); + + assertThat(new RealmPurgeStore(ops).listClaimable(Long.MAX_VALUE)) + .containsExactly("R1"); // queued + assertThat(countRows(ops, ModelEvent.TABLE_NAME, "R1")).isEqualTo(3); // data left for drainer + } + + private static final class CountConverter implements Converter { + @Override + public Long fromResultSet(ResultSet rs) throws SQLException { + return rs.getLong("c"); + } + + @Override + public Map toMap(DatabaseType databaseType) { + throw new UnsupportedOperationException(); + } + } + + private static final class TestJdbcConfiguration implements RelationalJdbcConfiguration { + @Override + public Optional maxRetries() { + return Optional.of(1); + } + + @Override + public Optional maxDurationInMs() { + return Optional.of(100L); + } + + @Override + public Optional initialDelayInMs() { + return Optional.of(100L); + } + + @Override + public Optional databaseType() { + return Optional.of("h2"); + } + } +} diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/MetaStoreManagerFactory.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/MetaStoreManagerFactory.java index 4a32a88591d..c4eea9cd376 100644 --- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/MetaStoreManagerFactory.java +++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/MetaStoreManagerFactory.java @@ -45,4 +45,18 @@ default Map bootstrapRealms(BootstrapOptions boo /** Purge all metadata for the realms provided */ Map purgeRealms(Iterable realms); + + /** + * Drains queued realm purges: deletes the time-series data (events, metrics reports) of already + * purged realms, in bounded batches, resuming any interrupted drains. Backends without an + * asynchronous realm-purge queue (e.g. those that purge such data through their own maintenance) + * return 0. + * + * @param batchSize maximum rows deleted per statement + * @param claimLeaseMs how long a drain claim is honored before another node may resume it + * @return the number of realms fully drained during this run + */ + default int drainRealmPurges(int batchSize, long claimLeaseMs) { + return 0; + } } diff --git a/runtime/admin/src/main/java/org/apache/polaris/admintool/PolarisAdminTool.java b/runtime/admin/src/main/java/org/apache/polaris/admintool/PolarisAdminTool.java index dcb427a8ec5..13b2fcef2d5 100644 --- a/runtime/admin/src/main/java/org/apache/polaris/admintool/PolarisAdminTool.java +++ b/runtime/admin/src/main/java/org/apache/polaris/admintool/PolarisAdminTool.java @@ -32,6 +32,7 @@ HelpCommand.class, BootstrapCommand.class, PurgeCommand.class, + PurgeDrainCommand.class, NoSqlCommand.class, }) public class PolarisAdminTool extends BaseCommand { diff --git a/runtime/admin/src/main/java/org/apache/polaris/admintool/PurgeDrainCommand.java b/runtime/admin/src/main/java/org/apache/polaris/admintool/PurgeDrainCommand.java new file mode 100644 index 00000000000..c734fc12b95 --- /dev/null +++ b/runtime/admin/src/main/java/org/apache/polaris/admintool/PurgeDrainCommand.java @@ -0,0 +1,62 @@ +/* + * 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.admintool; + +import picocli.CommandLine; + +@CommandLine.Command( + name = "purge-drain", + mixinStandardHelpOptions = true, + description = { + "Drain queued realm purges (relational-JDBC backend).", + "Deletes purged realms' time-series data (events and metrics reports) in bounded batches. " + + "Safe to run repeatedly and concurrently; intended to be scheduled (e.g. as a " + + "periodic job)." + }) +public class PurgeDrainCommand extends BaseMetaStoreCommand { + + @CommandLine.Option( + names = "--batch-size", + defaultValue = "1000", + description = "Maximum rows deleted per statement (default: ${DEFAULT-VALUE}).") + int batchSize; + + @CommandLine.Option( + names = "--claim-lease-ms", + defaultValue = "600000", + description = + "Milliseconds a drain claim is honored before another run may resume an interrupted " + + "realm (default: ${DEFAULT-VALUE}).") + long claimLeaseMs; + + @Override + public Integer call() { + if (batchSize <= 0) { + throw new CommandLine.ParameterException( + spec.commandLine(), "--batch-size must be strictly positive."); + } + if (claimLeaseMs <= 0) { + throw new CommandLine.ParameterException( + spec.commandLine(), "--claim-lease-ms must be strictly positive."); + } + int drained = metaStoreManagerFactory.drainRealmPurges(batchSize, claimLeaseMs); + spec.commandLine().getOut().printf("Drained %d realm purge(s).%n", drained); + return 0; + } +} diff --git a/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTest.java b/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTest.java new file mode 100644 index 00000000000..068fcc7942c --- /dev/null +++ b/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTest.java @@ -0,0 +1,135 @@ +/* + * 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.admintool; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.PrintWriter; +import java.io.StringWriter; +import java.util.Map; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.persistence.BasePersistence; +import org.apache.polaris.core.persistence.MetaStoreManagerFactory; +import org.apache.polaris.core.persistence.PolarisMetaStoreManager; +import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet; +import org.apache.polaris.core.persistence.cache.EntityCache; +import org.apache.polaris.core.persistence.dao.entity.BaseResult; +import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult; +import org.junit.jupiter.api.Test; +import picocli.CommandLine; + +class PurgeDrainCommandTest { + + @Test + void drainsWithDefaultsAndReportsCount() { + RecordingFactory factory = new RecordingFactory(3); + PurgeDrainCommand command = new PurgeDrainCommand(); + command.metaStoreManagerFactory = factory; + + CommandLine commandLine = new CommandLine(command); + StringWriter out = new StringWriter(); + commandLine.setOut(new PrintWriter(out)); + + int exitCode = commandLine.execute(); + + assertThat(exitCode).isZero(); + assertThat(out.toString()).contains("Drained 3 realm purge(s)."); + // Default options are passed through to the factory. + assertThat(factory.batchSize).isEqualTo(1000); + assertThat(factory.claimLeaseMs).isEqualTo(600_000L); + } + + @Test + void drainsWithProvidedOptions() { + RecordingFactory factory = new RecordingFactory(0); + PurgeDrainCommand command = new PurgeDrainCommand(); + command.metaStoreManagerFactory = factory; + + CommandLine commandLine = new CommandLine(command); + commandLine.setOut(new PrintWriter(new StringWriter())); + + int exitCode = commandLine.execute("--batch-size", "250", "--claim-lease-ms", "5000"); + + assertThat(exitCode).isZero(); + assertThat(factory.batchSize).isEqualTo(250); + assertThat(factory.claimLeaseMs).isEqualTo(5000L); + } + + @Test + void rejectsNonPositiveBatchSizeWithoutDraining() { + RecordingFactory factory = new RecordingFactory(0); + PurgeDrainCommand command = new PurgeDrainCommand(); + command.metaStoreManagerFactory = factory; + + CommandLine commandLine = new CommandLine(command); + StringWriter err = new StringWriter(); + commandLine.setErr(new PrintWriter(err)); + + int exitCode = commandLine.execute("--batch-size", "0"); + + assertThat(exitCode).isNotZero(); + assertThat(err.toString()).contains("--batch-size must be strictly positive"); + assertThat(factory.batchSize).isEqualTo(-1); // never reached the drainer + } + + /** Minimal {@link MetaStoreManagerFactory} recording the arguments passed to the drain. */ + private static final class RecordingFactory implements MetaStoreManagerFactory { + private final int drainResult; + int batchSize = -1; + long claimLeaseMs = -1; + + private RecordingFactory(int drainResult) { + this.drainResult = drainResult; + } + + @Override + public int drainRealmPurges(int batchSize, long claimLeaseMs) { + this.batchSize = batchSize; + this.claimLeaseMs = claimLeaseMs; + return drainResult; + } + + @Override + public Map purgeRealms(Iterable realms) { + throw new UnsupportedOperationException(); + } + + @Override + public PolarisMetaStoreManager getOrCreateMetaStoreManager(RealmContext realmContext) { + throw new UnsupportedOperationException(); + } + + @Override + public BasePersistence getOrCreateSession(RealmContext realmContext) { + throw new UnsupportedOperationException(); + } + + @Override + public EntityCache getOrCreateEntityCache(RealmContext realmContext, RealmConfig realmConfig) { + throw new UnsupportedOperationException(); + } + + @Override + public Map bootstrapRealms( + Iterable realms, RootCredentialsSet rootCredentialsSet) { + throw new UnsupportedOperationException(); + } + } +} diff --git a/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTestBase.java b/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTestBase.java new file mode 100644 index 00000000000..7e433e7cb21 --- /dev/null +++ b/runtime/admin/src/test/java/org/apache/polaris/admintool/PurgeDrainCommandTestBase.java @@ -0,0 +1,50 @@ +/* + * 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.admintool; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.quarkus.runtime.StartupEvent; +import io.quarkus.test.junit.main.Launch; +import io.quarkus.test.junit.main.LaunchResult; +import io.quarkus.test.junit.main.QuarkusMainTest; +import jakarta.enterprise.event.Observes; +import java.util.List; +import org.apache.polaris.core.persistence.MetaStoreManagerFactory; +import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet; +import org.junit.jupiter.api.Test; + +@QuarkusMainTest +public abstract class PurgeDrainCommandTestBase { + private static final String REALM1 = "purge-drain-test-realm1"; + private static final String REALM2 = "purge-drain-test-realm2"; + + // Purging a realm queues it into realm_purge; the launched purge-drain then drains the queue. + void preBootstrapAndPurge( + @Observes StartupEvent event, MetaStoreManagerFactory metaStoreManagerFactory) { + metaStoreManagerFactory.bootstrapRealms(List.of(REALM1, REALM2), RootCredentialsSet.EMPTY); + metaStoreManagerFactory.purgeRealms(List.of(REALM1, REALM2)); + } + + @Test + @Launch(value = {"purge-drain"}) + public void testPurgeDrain(LaunchResult result) { + assertThat(result.getOutput()).contains("Drained 2 realm purge(s)."); + } +} diff --git a/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/CockroachJdbcPurgeDrainCommandTest.java b/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/CockroachJdbcPurgeDrainCommandTest.java new file mode 100644 index 00000000000..a7175484814 --- /dev/null +++ b/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/CockroachJdbcPurgeDrainCommandTest.java @@ -0,0 +1,26 @@ +/* + * 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.admintool.relational.jdbc; + +import io.quarkus.test.junit.TestProfile; +import org.apache.polaris.admintool.AdminProfiles; +import org.apache.polaris.admintool.PurgeDrainCommandTestBase; + +@TestProfile(AdminProfiles.CockroachJdbc.class) +public class CockroachJdbcPurgeDrainCommandTest extends PurgeDrainCommandTestBase {} diff --git a/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/RelationalJdbcPurgeDrainCommandTest.java b/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/RelationalJdbcPurgeDrainCommandTest.java new file mode 100644 index 00000000000..bb6a2de5924 --- /dev/null +++ b/runtime/admin/src/test/java/org/apache/polaris/admintool/relational/jdbc/RelationalJdbcPurgeDrainCommandTest.java @@ -0,0 +1,26 @@ +/* + * 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.admintool.relational.jdbc; + +import io.quarkus.test.junit.TestProfile; +import org.apache.polaris.admintool.AdminProfiles; +import org.apache.polaris.admintool.PurgeDrainCommandTestBase; + +@TestProfile(AdminProfiles.RelationalJdbc.class) +public class RelationalJdbcPurgeDrainCommandTest extends PurgeDrainCommandTestBase {} diff --git a/site/content/in-dev/unreleased/admin-tool.md b/site/content/in-dev/unreleased/admin-tool.md index b7ea65eb990..41c7fa3237e 100644 --- a/site/content/in-dev/unreleased/admin-tool.md +++ b/site/content/in-dev/unreleased/admin-tool.md @@ -59,10 +59,11 @@ Polaris administration & maintenance tool -h, --help Show this help message and exit. -V, --version Print version information and exit. Commands: - help Display help information about the specified command. - bootstrap Bootstraps realms and root principal credentials. - purge Purge realms and all associated entities. - nosql Sub-commands specific to NoSQL persistence. + help Display help information about the specified command. + bootstrap Bootstraps realms and root principal credentials. + purge Purge realms and all associated entities. + purge-drain Drain queued realm purges (relational-JDBC backend). + nosql Sub-commands specific to NoSQL persistence. ``` ## Configuration @@ -207,6 +208,63 @@ docker run --rm -it \ Again, the Polaris Admin Tool must be executed with appropriate configuration to connect to the same database used by the Polaris server. The configuration can be done via environment variables (as above) or system properties. +## Draining Queued Realm Purges + +On the relational-JDBC backend, purging a realm does not delete its time-series data (events and +metrics reports) inline. Instead, the realm is queued for asynchronous, batched deletion, and the +`purge-drain` command processes that queue. + +{{< alert note >}} +The `purge-drain` command applies to the relational-JDBC backend only. On the NoSQL backend this data +is reclaimed by [NoSQL maintenance](#running-nosql-maintenance) instead, and `purge-drain` is a no-op. +{{< /alert >}} + +The command is safe to run repeatedly and concurrently across nodes: each queued realm is claimed for +the duration of a lease, so only one drain works a given realm at a time, and a realm left unfinished +(e.g. because a node stopped) is resumed once its claim lease expires. It is intended to be scheduled +to run periodically, for example once per day. + +If you have downloaded the [binary distribution]({{% ref "getting-started/binary-distribution" %}}), you can run the `purge-drain` command as follows: + +```shell +java -jar polaris-bin-/admin/quarkus-run.jar purge-drain --help +``` + +You can also use the Docker image to run the `purge-drain` command: + +```shell +docker run apache/polaris-admin-tool:latest purge-drain --help +``` + +The basic usage of the `purge-drain` command is outlined below: + +``` +Usage: polaris-admin-tool.jar purge-drain [-hV] [--batch-size=] + [--claim-lease-ms=] +Drain queued realm purges (relational-JDBC backend). +Deletes purged realms' time-series data (events and metrics reports) in bounded +batches. Safe to run repeatedly and concurrently; intended to be scheduled (e.g. +as a periodic job). + --batch-size= + Maximum rows deleted per statement (default: 1000). + --claim-lease-ms= + Milliseconds a drain claim is honored before another run may + resume an interrupted realm (default: 600000). + -h, --help Show this help message and exit. + -V, --version Print version information and exit. +``` + +For example, to drain queued realm purges on the PostgreSQL backend: + +```bash +docker run --rm -it \ + --env="polaris.persistence.type=relational-jdbc" \ + --env="quarkus.datasource.username=" \ + --env="quarkus.datasource.password=" \ + --env="quarkus.datasource.jdbc.url=" \ + apache/polaris-admin-tool:latest purge-drain +``` + ## NoSQL Specific Operations The `nosql` admin tool command is used to perform NoSQL metastore specific operations.