Copilot commented on code in PR #11266:
URL: https://github.com/apache/gravitino/pull/11266#discussion_r3323583453
##########
scripts/postgresql/schema-1.3.0-postgresql.sql:
##########
@@ -1054,3 +1054,21 @@ COMMENT ON COLUMN entity_change_log.entity_type IS
'METALAKE | CATALOG | SCHEMA
COMMENT ON COLUMN entity_change_log.entity_full_name IS 'Dot-separated full
name of the affected entity. For ALTER, stores the old name. For DROP, stores
the entity name.';
COMMENT ON COLUMN entity_change_log.operate_type IS 'Operate type code:
1=ALTER, 2=DROP, 3=INSERT. Codes are stable and never re-used.';
COMMENT ON COLUMN entity_change_log.created_at IS 'timestamp of the change in
millis';
+
+CREATE TABLE IF NOT EXISTS iceberg_cleanup_job (
+ id BIGINT NOT NULL PRIMARY KEY,
+ catalog_id BIGINT NOT NULL,
+ namespace VARCHAR(512) NOT NULL,
+ table_name VARCHAR(256) NOT NULL,
+ metadata_location TEXT NOT NULL,
+ file_io_impl VARCHAR(256) NOT NULL,
+ file_io_props TEXT NOT NULL,
+ state VARCHAR(16) NOT NULL,
+ attempts INT NOT NULL DEFAULT 0,
+ last_error VARCHAR(2048),
+ heartbeat_at BIGINT NOT NULL DEFAULT 0,
+ created_by VARCHAR(128) NOT NULL,
+ updated_at BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_state_updated ON iceberg_cleanup_job (state,
updated_at);
+CREATE INDEX IF NOT EXISTS idx_object ON iceberg_cleanup_job (catalog_id,
namespace, table_name, state);
Review Comment:
The PostgreSQL schema script adds COMMENTs for most tables/columns, but
`iceberg_cleanup_job` is missing COMMENT ON TABLE/COLUMN statements. This makes
the new table harder to inspect and inconsistent with the rest of the
PostgreSQL schema (and the MySQL script which includes detailed comments).
##########
scripts/postgresql/upgrade-1.2.0-to-1.3.0-postgresql.sql:
##########
@@ -151,3 +151,21 @@ COMMENT ON COLUMN idp_user_group_rel.group_id IS 'idp
group id';
COMMENT ON COLUMN idp_user_group_rel.current_version IS 'idp relation current
version';
COMMENT ON COLUMN idp_user_group_rel.last_version IS 'idp relation last
version';
COMMENT ON COLUMN idp_user_group_rel.deleted_at IS 'idp relation deleted at';
+
+CREATE TABLE IF NOT EXISTS iceberg_cleanup_job (
+ id BIGINT NOT NULL PRIMARY KEY,
+ catalog_id BIGINT NOT NULL,
+ namespace VARCHAR(512) NOT NULL,
+ table_name VARCHAR(256) NOT NULL,
+ metadata_location TEXT NOT NULL,
+ file_io_impl VARCHAR(256) NOT NULL,
+ file_io_props TEXT NOT NULL,
+ state VARCHAR(16) NOT NULL,
+ attempts INT NOT NULL DEFAULT 0,
+ last_error VARCHAR(2048),
+ heartbeat_at BIGINT NOT NULL DEFAULT 0,
+ created_by VARCHAR(128) NOT NULL,
+ updated_at BIGINT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_state_updated ON iceberg_cleanup_job (state,
updated_at);
+CREATE INDEX IF NOT EXISTS idx_object ON iceberg_cleanup_job (catalog_id,
namespace, table_name, state);
Review Comment:
The PostgreSQL upgrade script documents other new tables with COMMENT ON
TABLE/COLUMN, but the new `iceberg_cleanup_job` table is missing those
comments. Adding them here keeps the upgrade output self-describing and
consistent with `schema-1.3.0-postgresql.sql`.
##########
scripts/h2/upgrade-1.2.0-to-1.3.0-h2.sql:
##########
@@ -100,3 +100,22 @@ CREATE TABLE IF NOT EXISTS `idp_user_group_rel` (
KEY `idx_iuig_uid` (`user_id`),
KEY `idx_iuig_gid` (`group_id`)
) ENGINE=InnoDB;
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT NOT NULL,
+ `catalog_id` BIGINT NOT NULL,
+ `namespace` VARCHAR(512) NOT NULL,
+ `table_name` VARCHAR(256) NOT NULL,
+ `metadata_location` CLOB NOT NULL,
+ `file_io_impl` VARCHAR(256) NOT NULL,
+ `file_io_props` CLOB NOT NULL,
+ `state` VARCHAR(16) NOT NULL,
+ `attempts` INT NOT NULL DEFAULT 0,
+ `last_error` VARCHAR(2048) NULL,
+ `heartbeat_at` BIGINT NOT NULL DEFAULT 0,
+ `created_by` VARCHAR(128) NOT NULL,
+ `updated_at` BIGINT NOT NULL,
+ PRIMARY KEY (`id`)
+);
Review Comment:
In the H2 1.2.0->1.3.0 upgrade script, the new `iceberg_cleanup_job` table
is created without the inline table/column `COMMENT` metadata that the
surrounding tables use. Adding comments here keeps the upgrade script
consistent with the rest of the H2 DDL and makes the schema self-describing.
##########
iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java:
##########
@@ -0,0 +1,209 @@
+/*
+ * 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.gravitino.iceberg.service.cleanup;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.List;
+import java.util.Optional;
+import
org.apache.gravitino.iceberg.service.cleanup.mapper.IcebergCleanupJobMapper;
+import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+
+/**
+ * Persistence for {@code iceberg_cleanup_job}, layered on the Gravitino
entity store's shared
+ * relational backend. Async cleanup reuses the entity store's connection
pool, transaction
+ * management, and per-backend SQL dispatch instead of opening its own JDBC
connections. Row ids and
+ * timestamps are supplied by the application, keeping the SQL portable across
H2, MySQL, and
+ * PostgreSQL.
+ */
+public class IcebergCleanupJobStore {
+
+ private static final int MAX_ERROR_LENGTH = 2048;
+
+ private final IdGenerator idGenerator;
+
+ /**
+ * Creates a cleanup job store.
+ *
+ * @param idGenerator generator for new row ids
+ */
+ public IcebergCleanupJobStore(IdGenerator idGenerator) {
+ this.idGenerator = idGenerator;
+ }
+
+ /**
+ * Persists a new PENDING job.
+ *
+ * @param job job to persist
+ * @return generated id
+ */
+ public long addJob(IcebergCleanupJob job) {
+ long id = idGenerator.nextId();
+ long now = System.currentTimeMillis();
+ IcebergCleanupJobPO po = IcebergCleanupJobPO.fromCleanupJob(job, id, now);
+ SessionUtils.doWithCommit(IcebergCleanupJobMapper.class, mapper ->
mapper.insertCleanupJob(po));
+ return id;
+ }
+
+ /**
+ * Scans a small candidate window and takes the first available row via
compare-and-swap.
+ *
+ * @param now current epoch millis, written as the initial heartbeat
+ * @param heartbeatTimeoutMs age past which a RUNNING heartbeat is stale
+ * @param window max candidates to consider
+ * @return the taken job, or {@link Optional#empty()} if nothing was
available
+ */
+ public Optional<IcebergCleanupJob> takePendingJob(long now, long
heartbeatTimeoutMs, int window) {
+ long heartbeatExpiry = now - heartbeatTimeoutMs;
+ List<IcebergCleanupJobPO> candidates =
+ SessionUtils.getWithoutCommit(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.selectCandidateJobs(heartbeatExpiry, window));
+ for (IcebergCleanupJobPO po : candidates) {
+ long id = po.getId();
+ int marked =
+ SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markRunning(id, now, heartbeatExpiry));
+ if (marked == 1) {
+ // The claim only flips mutable columns (state, heartbeat_at,
updated_at); everything
+ // toCleanupJob reads was fixed at enqueue, so the candidate snapshot
is still accurate.
+ return Optional.of(po.toCleanupJob());
+ }
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Marks a RUNNING job SUCCEEDED.
+ *
+ * @param id job id
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean markSucceeded(long id) {
+ long now = System.currentTimeMillis();
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.SUCCEEDED.name(), null, now))
+ > 0;
+ }
+
+ /**
+ * Marks a RUNNING job FAILED immediately.
+ *
+ * @param id job id
+ * @param reason failure text
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean markFailed(long id, String reason) {
+ long now = System.currentTimeMillis();
+ String err = truncate(reason);
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.FAILED.name(), err, now))
+ > 0;
+ }
+
+ /**
+ * Records a transient failure: {@code attempts++}, then FAILED at the
ceiling else PENDING.
+ *
+ * @param id job id
+ * @param reason failure text
+ * @param maxAttempts ceiling from config
+ * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
+ * the job)
+ */
+ public boolean recordFailure(long id, String reason, int maxAttempts) {
+ long now = System.currentTimeMillis();
+ String err = truncate(reason);
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.recordFailure(id, err, maxAttempts, now))
+ > 0;
+ }
+
+ /**
+ * Refreshes a heartbeat with compare-and-swap ownership check.
+ *
+ * @param id job id
+ * @param lastHeartbeat previous heartbeat value
+ * @param now new heartbeat value
+ * @return {@code true} iff the row was still owned by the caller
+ */
+ public boolean heartbeat(long id, long lastHeartbeat, long now) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class, mapper -> mapper.heartbeat(id,
lastHeartbeat, now))
+ > 0;
+ }
+
+ /**
+ * Finds the id of an unfinished (PENDING or RUNNING) cleanup job for the
identifier, if any.
+ *
+ * @param catalogId globally unique id of the owning catalog
+ * @param namespace table namespace
+ * @param table table name
+ * @return the unfinished job id, or {@link Optional#empty()} if none exists
+ */
+ public Optional<Long> findUnfinishedJobId(long catalogId, String namespace,
String table) {
+ return Optional.ofNullable(
+ SessionUtils.getWithoutCommit(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.selectUnfinishedJobId(catalogId, namespace,
table)));
+ }
+
+ /**
+ * Deletes finished (SUCCEEDED or FAILED) jobs whose last update predates
the timeline.
+ *
+ * @param legacyTimeline cutoff epoch millis; rows updated before this are
removed
+ * @return rows deleted
+ */
+ public int deleteFinishedJobsByLegacyTimeline(long legacyTimeline) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ IcebergCleanupJobMapper.class,
+ mapper -> mapper.deleteFinishedJobsByLegacyTimeline(legacyTimeline));
+ }
+
+ /**
+ * Reads a job state for tests.
+ *
+ * @param id job id
+ * @return its current state
+ * @throws IllegalStateException if the row is gone
+ */
+ @VisibleForTesting
+ public IcebergCleanupJob.State stateOf(long id) {
+ String state =
Review Comment:
`stateOf` is annotated `@VisibleForTesting` and only used by tests in the
same package, but it is declared `public`, which unnecessarily expands the
production API surface of `IcebergCleanupJobStore`. Make it package-private to
keep the test-only helper accessible without exporting it as part of the
store's public API.
##########
scripts/h2/schema-1.3.0-h2.sql:
##########
@@ -612,3 +612,22 @@ CREATE TABLE IF NOT EXISTS `entity_change_log` (
PRIMARY KEY (`id`),
KEY `idx_ecl_created_at` (`created_at`)
) ENGINE=InnoDB COMMENT='Append-only log of entity structural changes for
targeted metadataIdCache invalidation';
+
+CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
+ `id` BIGINT NOT NULL,
+ `catalog_id` BIGINT NOT NULL,
+ `namespace` VARCHAR(512) NOT NULL,
+ `table_name` VARCHAR(256) NOT NULL,
+ `metadata_location` CLOB NOT NULL,
+ `file_io_impl` VARCHAR(256) NOT NULL,
+ `file_io_props` CLOB NOT NULL,
+ `state` VARCHAR(16) NOT NULL,
+ `attempts` INT NOT NULL DEFAULT 0,
+ `last_error` VARCHAR(2048) NULL,
+ `heartbeat_at` BIGINT NOT NULL DEFAULT 0,
+ `created_by` VARCHAR(128) NOT NULL,
+ `updated_at` BIGINT NOT NULL,
+ PRIMARY KEY (`id`)
+);
Review Comment:
In the H2 schema script, most tables/columns include inline `COMMENT`
metadata (for example `entity_change_log` just above), but the new
`iceberg_cleanup_job` table is missing both table and column comments. Adding
comments keeps the H2 schema consistent and makes troubleshooting easier.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
