lasdf1234 commented on code in PR #13774:
URL: https://github.com/apache/gravitino/pull/13774#discussion_r4259163549
##########
plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java:
##########
@@ -43,42 +49,123 @@ public static TableMaintenanceActionMetaService
getInstance() {
}
/**
- * Inserts an in-flight occupancy row before {@code runJob}. {@code job_id}
is bound later via
- * {@link #bindJobId(long, long)}.
+ * Tries to acquire an unbound occupancy claim before {@code runJob}.
+ *
+ * <ol>
+ * <li>CAS-{@code UPDATE created_at} on an expired unbound row ({@code
finished_at = 0})
+ * <li>Otherwise {@code INSERT} with {@code finished_at = 0}; unique
conflict means lost race
+ * </ol>
+ *
+ * <p>{@code finished_at = 0} is the soft exclusive slot (like {@code
deleted_at}); it stays until
+ * {@link #markFinished}. {@code created_at} is the unbound claim lease
start.
*
* @param tableId table entity id
* @param policyId policy entity id
- * @return generated action id
+ * @param nowMs current epoch millis
+ * @param claimTimeoutMs unbound claim TTL
+ * @return action id when acquired, empty when another live claim or open
run holds the slot
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
- baseMetricName = "insertTableMaintenanceAction")
- public long insertInFlight(long tableId, long policyId) {
+ baseMetricName = "tryAcquireTableMaintenanceAction")
+ public OptionalLong tryAcquireInFlight(
+ long tableId, long policyId, long nowMs, long claimTimeoutMs) {
+ Preconditions.checkArgument(claimTimeoutMs > 0, "claimTimeoutMs must be
positive");
+ long expireBefore = nowMs - claimTimeoutMs;
+ int reclaimed =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.reclaimExpiredUnbound(tableId, policyId, nowMs,
expireBefore));
+ if (reclaimed == 1) {
+ TableMaintenanceActionPO unbound =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectUnboundByTableAndPolicy(tableId,
policyId));
+ Preconditions.checkState(unbound != null && unbound.id() != null,
"reclaim lost unbound row");
+ return OptionalLong.of(unbound.id());
+ }
+
+ TableMaintenanceActionPO existing =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectUnboundByTableAndPolicy(tableId, policyId));
+ if (existing != null
+ && existing.createdAt() != null
+ && nowMs - existing.createdAt() < claimTimeoutMs) {
+ return OptionalLong.empty();
+ }
+
TableMaintenanceActionPO actionMeta =
TableMaintenanceActionPO.builder()
.withTableId(tableId)
.withPolicyId(policyId)
- .withCreatedAt(System.currentTimeMillis())
+ .withCreatedAt(nowMs)
+ .withFinishedAt(0L)
.build();
- SessionUtils.doWithCommit(
- TableMaintenanceActionMapper.class, mapper ->
mapper.insertAction(actionMeta));
+ try {
+ SessionUtils.doWithCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.insertAction(actionMeta));
+ } catch (RuntimeException e) {
+ if (isUniqueViolation(e)) {
+ return OptionalLong.empty();
+ }
+ throw e;
+ }
Preconditions.checkState(actionMeta.id() != null, "insert did not return
generated id");
- return actionMeta.id();
+ return OptionalLong.of(actionMeta.id());
Review Comment:
Fixed in the latest commit: `tryAcquireInFlight` / `insertInFlight` now
return an immutable `OccupancyClaim(id, createdAt)` captured at acquire/reclaim
time (for reclaim we fence with the `nowMs` we wrote, not a re-read of
`created_at`). `bindJobId(OccupancyClaim, jobId)` uses that value only.
##########
scripts/mysql/schema-2.0.0-mysql.sql:
##########
@@ -668,23 +668,73 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
+CREATE TABLE IF NOT EXISTS `scheduled_expand_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
Review Comment:
Fixed: all three MySQL `scheduled_*` tables (schema + upgrade) now use
`DATETIME(6)` for `execution_time` / `last_heartbeat` / `last_success` /
`last_failure`, matching db-scheduler 16.12.
##########
scripts/postgresql/schema-2.0.0-postgresql.sql:
##########
@@ -1176,29 +1176,116 @@ COMMENT ON COLUMN iceberg_cleanup_job.heartbeat_at IS
'last heartbeat from the w
COMMENT ON COLUMN iceberg_cleanup_job.created_by IS 'principal that requested
the drop (audit)';
COMMENT ON COLUMN iceberg_cleanup_job.updated_at IS 'last state change, drives
poll ordering and old finished-job cleanup';
+CREATE TABLE IF NOT EXISTS scheduled_expand_tasks (
+ task_name VARCHAR(100) NOT NULL,
+ task_instance VARCHAR(100) NOT NULL,
+ task_data BYTEA,
+ execution_time TIMESTAMP(6) NOT NULL,
Review Comment:
Fixed: PostgreSQL `scheduled_*` time columns are now `TIMESTAMP WITH TIME
ZONE` in schema + upgrade, matching db-scheduler 16.12 / `OffsetDateTime` JDBC
binding.
##########
scripts/mysql/schema-2.0.0-mysql.sql:
##########
@@ -668,23 +668,73 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
+CREATE TABLE IF NOT EXISTS `scheduled_expand_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance expand schedule metadata';
+
+CREATE TABLE IF NOT EXISTS `scheduled_cron_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance cron schedule metadata';
+
+CREATE TABLE IF NOT EXISTS `scheduled_commit_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance commit schedule metadata';
+
CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
- `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'job run id, maps to
job_run_meta.job_run_id',
`table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
`policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
`before_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
`after_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id after
maintenance, null while pending',
- `finished_at` BIGINT(20) UNSIGNED NULL COMMENT 'finish timestamp in
milliseconds, null while in flight',
+ `finished_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'finish
timestamp in milliseconds, 0 while in flight',
`created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tma_job_id` (`job_id`),
- KEY `idx_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
+ UNIQUE KEY `uk_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
Review Comment:
Thanks — we are keeping `UNIQUE(table_id, policy_id, finished_at)`
intentionally.
The uniqueness that matters is `finished_at = 0` as the soft in-flight slot
(one open row per table/policy). Because that slot admits only one in-flight
action, only one submitter can call `markFinished` for that pair at a time, so
two completed rows cannot race to the same millisecond `finished_at`. Clock
rollback leaving a durable collision is outside the failure modes we model here.
Added SQL / Javadoc comments documenting this. Happy to revisit a
partial/conditional unique key later if we see a concrete collision in practice.
--
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]
