roryqi commented on code in PR #13774:
URL: https://github.com/apache/gravitino/pull/13774#discussion_r4259085514
##########
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:
[P1] Return the fencing value with the acquired claim. This API returns only
the row id, while bindJobId requires the created_at value observed at
acquisition time. The caller therefore has to read the row again, as the test
does. If holder A pauses past the TTL, holder B reclaims by changing
created_at, and A then performs that read, A observes B's value and can
successfully bind using B's fence. That defeats the reclaim fencing and can
allow two submitted jobs. Please return an immutable claim containing both id
and the acquired created_at value, and pass that exact value to bindJobId
without rereading the row.
##########
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:
[P1] Use the db-scheduler 16.12 MySQL column type here. Upstream changed all
scheduler time columns from TIMESTAMP(6) to DATETIME(6) in 16.10 because
Schedule.NEVER is 2500-01-01, outside MySQL TIMESTAMP's supported range. With
this schema, any task rescheduled to NEVER fails with a data-truncation error.
Please mirror DATETIME(6) for execution_time, last_heartbeat, last_success, and
last_failure in all three tables and in the upgrade script.
##########
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:
[P1] Match this type to db-scheduler's JDBC timestamp mode. Version 16.12
uses TIMESTAMP WITH TIME ZONE in its PostgreSQL schema; the builder in
TableMaintenanceSchedulerPools also leaves alwaysPersistTimestampInUTC
disabled, so db-scheduler binds and reads OffsetDateTime. A zone-less
TIMESTAMP(6) is therefore inconsistent with the configured JDBC customization
and can fail conversion or make execution/heartbeat instants session-time-zone
dependent. Either use the upstream TIMESTAMP WITH TIME ZONE columns, or
explicitly enable UTC persistence if the zone-less schema is intentional; apply
the same choice to all scheduler time columns and the upgrade script.
##########
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:
[P2] Avoid making the historical finish timestamp unique. The uniqueness
needed for the soft slot is only for finished_at = 0, but this key also
requires every completed action for a table/policy pair to have a distinct
millisecond value forever. Two sequential fast/early-exit jobs can finish in
the same millisecond, and a wall-clock rollback can reuse an older value;
markFinished then hits this constraint and leaves the new action in-flight. The
design spec used a non-unique history index. Please model the active slot
separately or use a dialect-appropriate conditional/generated unique key so
completed timestamps remain non-unique.
--
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]
