This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new f58ea84310 [#13773] feat(tms): Add scheduler pools and soft-slot
occupancy claim (#13774)
f58ea84310 is described below
commit f58ea843106f95681e82c3b5093960de1389c67d
Author: MaSai <[email protected]>
AuthorDate: Wed Oct 14 13:55:12 2026 +0800
[#13773] feat(tms): Add scheduler pools and soft-slot occupancy claim
(#13774)
### What changes were proposed in this pull request?
Second subtask of Epic #13760 (depends on #13761 / #13762, merged).
- Add db-scheduler 16.12.0 and three isolated pools (expand / cron /
commit) over `scheduled_expand_tasks` / `scheduled_cron_tasks` /
`scheduled_commit_tasks`
- Add `TableMaintenanceConfig` for pool threads, poll / heartbeat /
missed-heartbeat, expand batch size, and unbound claim TTL
- Soft-slot occupancy on `table_maintenance_action`: `finished_at = 0`
as the exclusive in-flight slot (`UNIQUE(table_id, policy_id,
finished_at)`), unbound lease via `created_at` CAS reclaim, `bindJobId`
keeps the slot, `markFinished` sets `finished_at > 0`
- Claim / reclaim MetaService APIs and DDL for the soft slot
**Out of scope (follow-up):** `OccupancyGate`,
`TableMaintenanceRESTFeature` (pools do not auto-start until the Feature
lands), PolicyExpand / Spark submit, IRC commit-job listener, in-job S0
/ drift / recommend.
### Why are the changes needed?
Epic #13760 needs a multi-node-safe scheduler foundation and soft-slot
claim storage before occupancy gates and submit pipelines can land on
top.
Fix: #13773
### Does this PR introduce _any_ user-facing change?
Yes. New config keys:
- `gravitino.maintenance.scheduler.expand.threads` / `cron.threads` /
`commit.threads`
- `gravitino.maintenance.scheduler.pollingIntervalMs` /
`heartbeatIntervalMs` / `missedHeartbeatsLimit`
- `gravitino.maintenance.expand.enqueueBatchSize`
- `gravitino.maintenance.occupancy.claimTimeoutMs`
No public REST resources in this change. Enabling pools via
`gravitino.server.rest.extensionPackages` lands with the Feature
follow-up.
### How was this patch tested?
```bash
./gradlew :plugins:tms:spotlessApply :plugins:tms:test -PskipITs
```
---------
Co-authored-by: Cursor <[email protected]>
---
LICENSE.bin | 3 +
design-docs/table-maintenance-service-design.md | 51 ++--
gradle/libs.versions.toml | 2 +
plugins/tms/build.gradle.kts | 22 +-
.../maintenance/TableMaintenanceConfig.java | 263 +++++++++++++++++++++
.../scheduler/TableMaintenanceSchedulerPools.java | 152 ++++++++++++
.../mapper/TableMaintenanceActionMapper.java | 42 +++-
.../TableMaintenanceActionSQLProviderFactory.java | 64 ++++-
.../TableMaintenanceActionBaseSQLProvider.java | 131 +++++++++-
.../TableMaintenanceActionPostgreSQLProvider.java | 9 +-
.../storage/service/OccupancyClaim.java | 82 +++++++
.../service/TableMaintenanceActionMetaService.java | 183 +++++++++++++-
.../TestTableMaintenanceSchedulerPools.java | 55 +++++
.../TestTableMaintenanceActionMetaService.java | 44 +++-
scripts/h2/schema-2.0.0-h2.sql | 60 ++++-
scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql | 60 ++++-
scripts/mysql/schema-2.0.0-mysql.sql | 64 ++++-
scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql | 64 ++++-
scripts/postgresql/schema-2.0.0-postgresql.sql | 98 +++++++-
.../upgrade-1.3.0-to-2.0.0-postgresql.sql | 98 +++++++-
20 files changed, 1451 insertions(+), 96 deletions(-)
diff --git a/LICENSE.bin b/LICENSE.bin
index 9da550578c..899b0e1dbd 100644
--- a/LICENSE.bin
+++ b/LICENSE.bin
@@ -411,6 +411,9 @@
Fabric8 Kubernetes Client
Fabric8 Kubernetes Model
Fabric8 zjsonpatch
+ db-scheduler
+ cron-utils (shaded into db-scheduler)
+ JSpecify
This product bundles various third-party components also under the
Apache Software Foundation License 1.1
diff --git a/design-docs/table-maintenance-service-design.md
b/design-docs/table-maintenance-service-design.md
index f1b768e79b..d4487e8e12 100644
--- a/design-docs/table-maintenance-service-design.md
+++ b/design-docs/table-maintenance-service-design.md
@@ -91,10 +91,10 @@ submit). Industry options compared below:
| Cluster CAS + heartbeat | Yes | Yes
| Lock only | Yes (heavy) |
| Fits long-lived expand + one-shot submits | Yes | Yes
| No | Yes |
-**Chosen:** db-scheduler — Apache 2.0, one `scheduled_tasks` table with
built-in heartbeat, and it
-fits both long-lived expand leases and one-shot cron/commit submits (three
pools: expand / cron /
-commit). JobRunr is LGPL. ShedLock only provides a lock, not the scheduler TMS
needs. Quartz JDBC
-can do the job but is heavier than needed here.
+**Chosen:** db-scheduler — Apache 2.0, built-in heartbeat, and three isolated
pools each with its
+own task table (`scheduled_expand_tasks` / `scheduled_cron_tasks` /
`scheduled_commit_tasks`).
+JobRunr is LGPL. ShedLock only provides a lock, not the scheduler TMS needs.
Quartz JDBC can do
+the job but is heavier than needed here.
### 4.3 Snapshot drift
@@ -137,7 +137,7 @@ policy first, then scheduling.
| Store | Role
|
| -------------------------------------------------------------- |
------------------------------------------------------------ |
| `policy_meta` / `policy_relation_meta` / `policy_version_info` | Triggers +
`minIntervalMs` + `jobOptions` / `rewriteOptions` |
-| `scheduled_tasks` | Scheduler
rows: expand / cron-job / commit-job |
+| `scheduled_expand_tasks` / `scheduled_cron_tasks` / `scheduled_commit_tasks`
| Scheduler rows (one db-scheduler table per pool) |
| `table_maintenance_job` | Per-run
occupancy; before/after `snapshot_id` |
| `table_snapshot_metrics` | Metrics
keyed by Iceberg `snapshot_id` |
| `gravitino.conf` |
Cluster-static Gravitino / IRC client auth |
@@ -153,7 +153,8 @@ policy may enable both `onCommit` and `crontab`.
Triggers live in **`policy_version_info.content.schedule`**, next to
`jobOptions`,
`rewriteOptions`, and `minIntervalMs`. Operators edit that versioned content
through Policy
-APIs. `policy_meta` only holds identity and attachment. `scheduled_tasks` is
runtime state for db-scheduler — not where operators set triggers.
+APIs. `policy_meta` only holds identity and attachment. The
`scheduled_*_tasks` tables are runtime
+state for db-scheduler — not where operators set triggers.
TMS has two trigger modes. They never feed each other's queues:
@@ -224,16 +225,17 @@ each — it does not write secrets back into policy content:
### 5.2 Scheduling and commit
-db-scheduler runs **three** task kinds in `scheduled_tasks`:
+db-scheduler runs **three** isolated pools. Each pool needs its own physical
task table (upstream
+db-scheduler does not support sharing one table across Scheduler instances):
-| Kind | `task_name` | Role |
-| ---- | --------------- | ----------------------------------------------- |
-| 1 | `policy-expand` | Long-lived lease per policy; fans out cron-jobs |
-| 2 | `cron-job` | One-shot crontab Spark submit |
-| 3 | `commit-job` | One-shot commit-triggered chained Spark submit |
+| Pool | Table | `task_name` | Role
|
+| ------ | ------------------------- | --------------- |
----------------------------------------------- |
+| Expand | `scheduled_expand_tasks` | `policy-expand` | Long-lived lease per
policy; fans out cron-jobs |
+| Cron | `scheduled_cron_tasks` | `cron-job` | One-shot crontab
Spark submit |
+| Commit | `scheduled_commit_tasks` | `commit-job` | One-shot
commit-triggered chained Spark submit |
**policy-expand** stays after each pick. **cron-job** and **commit-job** are
deleted after a
-short `runJob`. Spark lifetime and in-flight checks use
`table_maintenance_job`, not those
+short `runJob`. Spark lifetime and in-flight checks use
`table_maintenance_action`, not those
one-shot rows.
Every node may poll; **exactly one claim wins** per due instance (CAS +
heartbeat). Missed
@@ -241,11 +243,12 @@ heartbeats free only a hung **short** callback — not a
long-running Spark Job.
| Pool | Registers | Default threads | conf key
(`gravitino.maintenance.`) |
| ------ | ------------- | --------------- |
----------------------------------- |
-| Expand | policy-expand | 4 | `scheduler.expand.threads`
|
-| Cron | cron-job | 8 | `scheduler.cron.threads`
|
-| Commit | commit-job | 4 | `scheduler.commit.threads`
|
+| Expand | policy-expand | 4 | `scheduler.expandThreads`
|
+| Cron | cron-job | 8 | `scheduler.cronThreads`
|
+| Commit | commit-job | 4 | `scheduler.commitThreads`
|
-PK is `(task_name, task_instance)`. `scheduled_tasks` uses upstream
db-scheduler DDL.
+PK is `(task_name, task_instance)`. Each table uses upstream db-scheduler DDL
(MySQL
+`DATETIME(6)` / PostgreSQL `TIMESTAMP WITH TIME ZONE` time columns).
**TMS tables used by cron-job and commit-job**
@@ -448,9 +451,9 @@ gravitino.server.rest.extensionPackages =
org.apache.gravitino.maintenance.web.r
gravitino.auxService.names = iceberg-rest
gravitino.eventListener.names = tms-commit-job
gravitino.eventListener.tms-commit-job.class =
org.apache.gravitino.maintenance.listener.IcebergUpdateTableEventListener
-gravitino.maintenance.scheduler.expand.threads = 4
-gravitino.maintenance.scheduler.cron.threads = 8
-gravitino.maintenance.scheduler.commit.threads = 4
+gravitino.maintenance.scheduler.expandThreads = 4
+gravitino.maintenance.scheduler.cronThreads = 8
+gravitino.maintenance.scheduler.commitThreads = 4
```
`extensionPackages` plus colocated IRC start the TMS Feature and db-scheduler
pools. The
@@ -462,9 +465,9 @@ onCommit does not. HTTP / Kafka commit ingress is out of
scope.
| Key | Default |
Description |
| ------------------------------------------------------- | ------- |
----------------------------- |
-| `gravitino.maintenance.scheduler.expand.threads` | `4` | Expand
pool |
-| `gravitino.maintenance.scheduler.cron.threads` | `8` | Cron
pool |
-| `gravitino.maintenance.scheduler.commit.threads` | `4` | Commit
pool |
+| `gravitino.maintenance.scheduler.expandThreads` | `4` | Expand
pool |
+| `gravitino.maintenance.scheduler.cronThreads` | `8` | Cron
pool |
+| `gravitino.maintenance.scheduler.commitThreads` | `4` | Commit
pool |
| `gravitino.maintenance.scheduler.pollingIntervalMs` | `10000` | Poll
interval |
| `gravitino.maintenance.scheduler.heartbeatIntervalMs` | `60000` |
Heartbeat while callback runs |
| `gravitino.maintenance.scheduler.missedHeartbeatsLimit` | `6` | Misses
before dead |
@@ -534,7 +537,7 @@ gravitino.maintenance.ircAuth.oauthCredential =
client_id:client_secret
| ----- |
---------------------------------------------------------------------------------
|
| 1 | `TableMaintenanceRESTFeature`; start/stop three db-scheduler pools
|
| 2 | `PolicyExpandPipeline` + `MaintenanceSparkSubmitPipeline` + gates
|
-| 3 | Migrations: `scheduled_tasks`, `table_maintenance_job`,
`table_snapshot_metrics` |
+| 3 | Migrations: `scheduled_*_tasks`, `table_maintenance_action`,
`table_snapshot_metrics` |
| 4 | IRC EventListener upserts commit-job; policy create → INSERT expand;
Job listener |
| 5 | Metrics, shutdown, docs; `gravitino.conf` client auth; privilege
docs |
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index cfe5bb7ce6..cb827cdacb 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -17,6 +17,7 @@
# under the License.
#
[versions]
+db-scheduler = "16.12.0"
awssdk = "2.31.73"
azure-identity = "1.13.1"
azure-storage-file-datalake = "12.20.0"
@@ -184,6 +185,7 @@ aspectj = "1.9.24"
ql-expression = "4.0.3"
[libraries]
+db-scheduler = { group = "com.github.kagkarlsson", name = "db-scheduler",
version.ref = "db-scheduler" }
aspectj-aspectjrt = { group = "org.aspectj", name = "aspectjrt", version.ref =
"aspectj" }
aws-dynamodb = { group = "software.amazon.awssdk", name = "dynamodb",
version.ref = "awssdk" }
aws-glue = { group = "software.amazon.awssdk", name = "glue", version.ref =
"awssdk" }
diff --git a/plugins/tms/build.gradle.kts b/plugins/tms/build.gradle.kts
index 1e6d92d9ad..835eb0fea6 100644
--- a/plugins/tms/build.gradle.kts
+++ b/plugins/tms/build.gradle.kts
@@ -30,6 +30,7 @@ dependencies {
implementation(project(":common"))
implementation(project(":core"))
+ implementation(libs.db.scheduler)
implementation(libs.guava)
implementation(libs.mybatis)
@@ -40,7 +41,6 @@ dependencies {
testImplementation(project(":core", "testArtifacts"))
testImplementation(project(":integration-test-common", "testArtifacts"))
testImplementation(project(":server"))
- testImplementation(project(":server-common"))
testImplementation(libs.junit.jupiter.api)
testImplementation(libs.junit.jupiter.params)
testImplementation(libs.mockito.core)
@@ -53,11 +53,19 @@ dependencies {
}
tasks {
+ val runtimeJars by registering(Copy::class) {
+ from(configurations.runtimeClasspath)
+ into("build/libs")
+ }
+
val copyLibs by registering(Copy::class) {
- dependsOn(jar)
- from(layout.buildDirectory.dir("libs")) {
- include("gravitino-tms-*.jar")
+ dependsOn(jar, runtimeJars)
+ from("build/libs") {
exclude("*-javadoc.jar", "*-sources.jar")
+ exclude("guava-*.jar")
+ exclude("log4j-*.jar")
+ exclude("slf4j-*.jar")
+ exclude("error_prone_annotations-*.jar")
}
into("$rootDir/distribution/package/libs")
duplicatesStrategy = DuplicatesStrategy.EXCLUDE
@@ -65,10 +73,14 @@ tasks {
register("copyLibAndConfigs", Copy::class) {
group = "gravitino distribution"
- description = "Copy tms plugin jar into distribution package libs"
+ description = "Copy tms plugin jar and runtime dependencies into
distribution package libs"
dependsOn(copyLibs)
}
+ jar {
+ dependsOn(runtimeJars)
+ }
+
test {
useJUnitPlatform()
environment("GRAVITINO_HOME", rootDir.path)
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/TableMaintenanceConfig.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/TableMaintenanceConfig.java
new file mode 100644
index 0000000000..a69c04d6a2
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/TableMaintenanceConfig.java
@@ -0,0 +1,263 @@
+/*
+ * 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.maintenance;
+
+import org.apache.gravitino.Config;
+import org.apache.gravitino.config.ConfigBuilder;
+import org.apache.gravitino.config.ConfigConstants;
+import org.apache.gravitino.config.ConfigEntry;
+
+/** Configuration for table maintenance scheduler pools (§7.2). */
+public final class TableMaintenanceConfig {
+
+ public static final ConfigEntry<Integer> SCHEDULER_EXPAND_THREADS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.expandThreads")
+ .doc("Thread count for the policy-expand db-scheduler pool")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(4);
+
+ public static final ConfigEntry<Integer> SCHEDULER_CRON_THREADS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.cronThreads")
+ .doc("Thread count for the cron-job db-scheduler pool")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(8);
+
+ public static final ConfigEntry<Integer> SCHEDULER_COMMIT_THREADS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.commitThreads")
+ .doc("Thread count for the commit-job db-scheduler pool")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(4);
+
+ public static final ConfigEntry<Long> SCHEDULER_POLLING_INTERVAL_MS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.pollingIntervalMs")
+ .doc("db-scheduler poll interval in milliseconds")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .longConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(10_000L);
+
+ public static final ConfigEntry<Long> SCHEDULER_HEARTBEAT_INTERVAL_MS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.heartbeatIntervalMs")
+ .doc("Heartbeat interval while a scheduler callback runs")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .longConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(60_000L);
+
+ public static final ConfigEntry<Integer> SCHEDULER_MISSED_HEARTBEATS_LIMIT =
+ new
ConfigBuilder("gravitino.maintenance.scheduler.missedHeartbeatsLimit")
+ .doc("Missed heartbeats before a picked task is considered dead")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(6);
+
+ public static final ConfigEntry<Integer> EXPAND_ENQUEUE_BATCH_SIZE =
+ new ConfigBuilder("gravitino.maintenance.expand.enqueueBatchSize")
+ .doc("Maximum cron-jobs enqueued per policy-expand pick")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(100);
+
+ /** Default unbound claim TTL ({@code job_id} still null), matching
ShedLock-style short locks. */
+ public static final long DEFAULT_CLAIM_TIMEOUT_MS = 120_000L;
+
+ public static final ConfigEntry<Long> OCCUPANCY_CLAIM_TIMEOUT_MS =
+ new ConfigBuilder("gravitino.maintenance.occupancy.claimTimeoutMs")
+ .doc(
+ "Max age of an unbound table_maintenance_action claim (job_id
null) before another"
+ + " submitter may reclaim it")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .longConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_CLAIM_TIMEOUT_MS);
+
+ private final int expandThreads;
+ private final int cronThreads;
+ private final int commitThreads;
+ private final long pollingIntervalMs;
+ private final long heartbeatIntervalMs;
+ private final int missedHeartbeatsLimit;
+ private final int expandEnqueueBatchSize;
+ private final long claimTimeoutMs;
+
+ private TableMaintenanceConfig(
+ int expandThreads,
+ int cronThreads,
+ int commitThreads,
+ long pollingIntervalMs,
+ long heartbeatIntervalMs,
+ int missedHeartbeatsLimit,
+ int expandEnqueueBatchSize,
+ long claimTimeoutMs) {
+ this.expandThreads = expandThreads;
+ this.cronThreads = cronThreads;
+ this.commitThreads = commitThreads;
+ this.pollingIntervalMs = pollingIntervalMs;
+ this.heartbeatIntervalMs = heartbeatIntervalMs;
+ this.missedHeartbeatsLimit = missedHeartbeatsLimit;
+ this.expandEnqueueBatchSize = expandEnqueueBatchSize;
+ this.claimTimeoutMs = claimTimeoutMs;
+ }
+
+ /**
+ * Reads scheduler settings from server configuration.
+ *
+ * @param config gravitino server config
+ * @return resolved maintenance config
+ */
+ public static TableMaintenanceConfig from(Config config) {
+ return new TableMaintenanceConfig(
+ config.get(SCHEDULER_EXPAND_THREADS),
+ config.get(SCHEDULER_CRON_THREADS),
+ config.get(SCHEDULER_COMMIT_THREADS),
+ config.get(SCHEDULER_POLLING_INTERVAL_MS),
+ config.get(SCHEDULER_HEARTBEAT_INTERVAL_MS),
+ config.get(SCHEDULER_MISSED_HEARTBEATS_LIMIT),
+ config.get(EXPAND_ENQUEUE_BATCH_SIZE),
+ config.get(OCCUPANCY_CLAIM_TIMEOUT_MS));
+ }
+
+ /**
+ * @return expand pool thread count
+ */
+ public int expandThreads() {
+ return expandThreads;
+ }
+
+ /**
+ * @return cron pool thread count
+ */
+ public int cronThreads() {
+ return cronThreads;
+ }
+
+ /**
+ * @return commit pool thread count
+ */
+ public int commitThreads() {
+ return commitThreads;
+ }
+
+ /**
+ * @return poll interval in milliseconds
+ */
+ public long pollingIntervalMs() {
+ return pollingIntervalMs;
+ }
+
+ /**
+ * @return heartbeat interval in milliseconds
+ */
+ public long heartbeatIntervalMs() {
+ return heartbeatIntervalMs;
+ }
+
+ /**
+ * @return missed heartbeat limit
+ */
+ public int missedHeartbeatsLimit() {
+ return missedHeartbeatsLimit;
+ }
+
+ /**
+ * @return expand enqueue batch size
+ */
+ public int expandEnqueueBatchSize() {
+ return expandEnqueueBatchSize;
+ }
+
+ /**
+ * @return unbound occupancy claim timeout in milliseconds
+ */
+ public long claimTimeoutMs() {
+ return claimTimeoutMs;
+ }
+
+ /**
+ * Test-only factory with explicit values.
+ *
+ * @param expandThreads expand pool threads
+ * @param cronThreads cron pool threads
+ * @param commitThreads commit pool threads
+ * @param pollingIntervalMs poll interval
+ * @param heartbeatIntervalMs heartbeat interval
+ * @param missedHeartbeatsLimit missed heartbeat limit
+ * @param expandEnqueueBatchSize expand batch size
+ * @return config instance
+ */
+ public static TableMaintenanceConfig forTest(
+ int expandThreads,
+ int cronThreads,
+ int commitThreads,
+ long pollingIntervalMs,
+ long heartbeatIntervalMs,
+ int missedHeartbeatsLimit,
+ int expandEnqueueBatchSize) {
+ return forTest(
+ expandThreads,
+ cronThreads,
+ commitThreads,
+ pollingIntervalMs,
+ heartbeatIntervalMs,
+ missedHeartbeatsLimit,
+ expandEnqueueBatchSize,
+ DEFAULT_CLAIM_TIMEOUT_MS);
+ }
+
+ /**
+ * Test-only factory with explicit values including claim timeout.
+ *
+ * @param expandThreads expand pool threads
+ * @param cronThreads cron pool threads
+ * @param commitThreads commit pool threads
+ * @param pollingIntervalMs poll interval
+ * @param heartbeatIntervalMs heartbeat interval
+ * @param missedHeartbeatsLimit missed heartbeat limit
+ * @param expandEnqueueBatchSize expand batch size
+ * @param claimTimeoutMs unbound claim timeout
+ * @return config instance
+ */
+ public static TableMaintenanceConfig forTest(
+ int expandThreads,
+ int cronThreads,
+ int commitThreads,
+ long pollingIntervalMs,
+ long heartbeatIntervalMs,
+ int missedHeartbeatsLimit,
+ int expandEnqueueBatchSize,
+ long claimTimeoutMs) {
+ return new TableMaintenanceConfig(
+ expandThreads,
+ cronThreads,
+ commitThreads,
+ pollingIntervalMs,
+ heartbeatIntervalMs,
+ missedHeartbeatsLimit,
+ expandEnqueueBatchSize,
+ claimTimeoutMs);
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java
new file mode 100644
index 0000000000..e60bbf9d18
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java
@@ -0,0 +1,152 @@
+/*
+ * 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.maintenance.scheduler;
+
+import com.github.kagkarlsson.scheduler.Scheduler;
+import com.github.kagkarlsson.scheduler.task.Task;
+import com.github.kagkarlsson.scheduler.task.helper.Tasks;
+import com.google.common.base.Preconditions;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.gravitino.maintenance.TableMaintenanceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Starts and stops three isolated db-scheduler pools (§5.2, §7.2), each with
its own task table
+ * ({@link #EXPAND_TASKS_TABLE}, {@link #CRON_TASKS_TABLE}, {@link
#COMMIT_TASKS_TABLE}).
+ *
+ * <p>db-scheduler requires a dedicated {@code tableName} per Scheduler
instance for multi-pool
+ * isolation; sharing one physical table across pools is unsupported (see
design §5.2).
+ *
+ * <p>Task handlers are intentionally minimal in this change; submit and
occupancy logic land in
+ * follow-up PRs.
+ */
+public class TableMaintenanceSchedulerPools implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TableMaintenanceSchedulerPools.class);
+
+ /** Expand-pool db-scheduler table. */
+ public static final String EXPAND_TASKS_TABLE = "scheduled_expand_tasks";
+
+ /** Cron-pool db-scheduler table. */
+ public static final String CRON_TASKS_TABLE = "scheduled_cron_tasks";
+
+ /** Commit-pool db-scheduler table. */
+ public static final String COMMIT_TASKS_TABLE = "scheduled_commit_tasks";
+
+ /** Long-lived policy expand lease ({@code task_name}). */
+ public static final String POLICY_EXPAND = "policy-expand";
+
+ /** One-shot crontab Spark submit ({@code task_name}). */
+ public static final String CRON_JOB = "cron-job";
+
+ /** One-shot on-commit chained Spark submit ({@code task_name}). */
+ public static final String COMMIT_JOB = "commit-job";
+
+ private final List<Scheduler> schedulers = new ArrayList<>();
+
+ /**
+ * Creates and starts expand, cron, and commit scheduler pools.
+ *
+ * <p>If any pool fails to start, already-started pools are stopped and this
instance returns to
+ * the idle state so {@link #start} may be retried.
+ *
+ * @param dataSource relational JDBC pool shared with the entity store
+ * @param config maintenance scheduler settings
+ */
+ public void start(DataSource dataSource, TableMaintenanceConfig config) {
+ Preconditions.checkState(schedulers.isEmpty(), "scheduler pools already
started");
+ Preconditions.checkNotNull(dataSource, "dataSource must not be null");
+
+ List<Scheduler> built =
+ List.of(
+ buildScheduler(
+ dataSource,
+ config,
+ config.expandThreads(),
+ EXPAND_TASKS_TABLE,
+ noopTask(POLICY_EXPAND)),
+ buildScheduler(
+ dataSource, config, config.cronThreads(), CRON_TASKS_TABLE,
noopTask(CRON_JOB)),
+ buildScheduler(
+ dataSource,
+ config,
+ config.commitThreads(),
+ COMMIT_TASKS_TABLE,
+ noopTask(COMMIT_JOB)));
+
+ try {
+ for (Scheduler scheduler : built) {
+ scheduler.start();
+ schedulers.add(scheduler);
+ }
+ } catch (RuntimeException e) {
+ stop();
+ throw e;
+ }
+ LOG.info("Started {} table maintenance scheduler pools",
schedulers.size());
+ }
+
+ /** Stops all scheduler pools. */
+ public void stop() {
+ for (Scheduler scheduler : schedulers) {
+ try {
+ scheduler.stop();
+ } catch (Exception e) {
+ LOG.warn("Failed to stop a table maintenance scheduler pool", e);
+ }
+ }
+ schedulers.clear();
+ LOG.info("Stopped table maintenance scheduler pools");
+ }
+
+ @Override
+ public void close() {
+ stop();
+ }
+
+ private static Scheduler buildScheduler(
+ DataSource dataSource,
+ TableMaintenanceConfig config,
+ int threads,
+ String tableName,
+ Task<Void> task) {
+ return Scheduler.create(dataSource, task)
+ .threads(threads)
+ .pollingInterval(Duration.ofMillis(config.pollingIntervalMs()))
+ .heartbeatInterval(Duration.ofMillis(config.heartbeatIntervalMs()))
+ .missedHeartbeatsLimit(config.missedHeartbeatsLimit())
+ .tableName(tableName)
+ .registerShutdownHook()
+ .build();
+ }
+
+ private static Task<Void> noopTask(String taskName) {
+ return Tasks.custom(taskName, Void.class)
+ .execute(
+ (taskInstance, executionContext) -> {
+ LOG.debug(
+ "TMS scheduler callback for {} instance {}", taskName,
taskInstance.getId());
+ return (executionComplete, executionOperations) -> {};
+ });
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
index ac14efa56a..2882964445 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.maintenance.storage.mapper;
+import java.util.List;
import javax.annotation.Nullable;
import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
import org.apache.ibatis.annotations.DeleteProvider;
@@ -36,8 +37,35 @@ public interface TableMaintenanceActionMapper {
@Options(useGeneratedKeys = true, keyProperty = "actionMeta.id", keyColumn =
"id")
void insertAction(@Param("actionMeta") TableMaintenanceActionPO actionMeta);
+ @UpdateProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "reclaimExpiredUnbound")
+ int reclaimExpiredUnbound(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs);
+
+ @Nullable
+ @SelectProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "selectUnboundByTableAndPolicy")
+ TableMaintenanceActionPO selectUnboundByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId);
+
+ @Nullable
+ @SelectProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "selectLiveUnboundByTableAndPolicy")
+ TableMaintenanceActionPO selectLiveUnboundByTableAndPolicy(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs);
+
@UpdateProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "bindJobId")
- int bindJobId(@Param("id") long id, @Param("jobId") long jobId);
+ int bindJobId(
+ @Param("id") long id,
+ @Param("jobId") long jobId,
+ @Param("expectedCreatedAt") long expectedCreatedAt);
@Nullable
@SelectProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "selectById")
@@ -64,4 +92,16 @@ public interface TableMaintenanceActionMapper {
@DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class,
method = "deleteByJobId")
int deleteByJobId(@Param("jobId") long jobId);
+
+ @SelectProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "selectInFlightByTableAndPolicy")
+ List<TableMaintenanceActionPO> selectInFlightByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId);
+
+ @Nullable
+ @SelectProvider(
+ type = TableMaintenanceActionSQLProviderFactory.class,
+ method = "selectMaxFinishedAt")
+ Long selectMaxFinishedAt(@Param("tableId") long tableId, @Param("policyId")
long policyId);
}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
index 2bfe0b66ee..891dd120d7 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java
@@ -56,13 +56,53 @@ public class TableMaintenanceActionSQLProviderFactory {
return getProvider().insertAction(actionMeta);
}
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL in milliseconds
+ * @return UPDATE statement
+ */
+ public static String reclaimExpiredUnbound(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs) {
+ return getProvider().reclaimExpiredUnbound(tableId, policyId,
claimTimeoutMs);
+ }
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement
+ */
+ public static String selectUnboundByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return getProvider().selectUnboundByTableAndPolicy(tableId, policyId);
+ }
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL in milliseconds
+ * @return SELECT statement
+ */
+ public static String selectLiveUnboundByTableAndPolicy(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs) {
+ return getProvider().selectLiveUnboundByTableAndPolicy(tableId, policyId,
claimTimeoutMs);
+ }
+
/**
* @param id action id
* @param jobId job run id
+ * @param expectedCreatedAt claim lease start observed at acquire time
* @return UPDATE statement
*/
- public static String bindJobId(@Param("id") long id, @Param("jobId") long
jobId) {
- return getProvider().bindJobId(id, jobId);
+ public static String bindJobId(
+ @Param("id") long id,
+ @Param("jobId") long jobId,
+ @Param("expectedCreatedAt") long expectedCreatedAt) {
+ return getProvider().bindJobId(id, jobId, expectedCreatedAt);
}
/**
@@ -120,6 +160,26 @@ public class TableMaintenanceActionSQLProviderFactory {
return getProvider().deleteByJobId(jobId);
}
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement
+ */
+ public static String selectInFlightByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return getProvider().selectInFlightByTableAndPolicy(tableId, policyId);
+ }
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement
+ */
+ public static String selectMaxFinishedAt(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return getProvider().selectMaxFinishedAt(tableId, policyId);
+ }
+
static TableMaintenanceActionBaseSQLProvider getProvider() {
String databaseId =
SqlSessionFactoryHelper.getInstance()
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
index 9b5c44b134..419a1987fb 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java
@@ -21,37 +21,112 @@ package
org.apache.gravitino.maintenance.storage.mapper.provider.base;
import static
org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceActionMapper.TABLE_NAME;
import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
+import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
import org.apache.ibatis.annotations.Param;
-/** Portable SQL for {@code table_maintenance_action}. */
+/** Portable SQL for {@code table_maintenance_action} (MySQL / H2 {@code
MODE=MYSQL}). */
public class TableMaintenanceActionBaseSQLProvider {
/**
- * Inserts an in-flight occupancy row before {@code runJob}. {@code job_id}
stays null until bound
- * after submission (§5.2).
+ * @return SQL expression for the database clock in epoch millis
+ */
+ protected String currentTimeMillisExpr() {
+ return DatabaseTimeSQL.MYSQL;
+ }
+
+ /**
+ * Inserts an unbound occupancy claim before {@code runJob}. {@code
finished_at = 0} holds the
+ * exclusive soft slot (same pattern as {@code deleted_at}); {@code
created_at} is the claim lease
+ * start from the database clock.
*
- * @param actionMeta row to insert
+ * @param actionMeta row to insert ({@code createdAt} is ignored; DB time is
used)
* @return INSERT statement
*/
public String insertAction(@Param("actionMeta") TableMaintenanceActionPO
actionMeta) {
return "INSERT INTO "
+ TABLE_NAME
- + " (table_id, policy_id, created_at)"
- + " VALUES (#{actionMeta.tableId}, #{actionMeta.policyId},
#{actionMeta.createdAt})";
+ + " (table_id, policy_id, created_at, finished_at)"
+ + " VALUES (#{actionMeta.tableId}, #{actionMeta.policyId}, "
+ + currentTimeMillisExpr()
+ + ", 0)";
}
/**
- * Binds the Gravitino job run id after {@code runJob} succeeds.
+ * Extends an expired unbound claim lease by CAS on {@code created_at}
(ShedLock-style). Lease age
+ * is evaluated against the database clock so every node shares one TTL
timeline.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL in milliseconds
+ * @return UPDATE statement
+ */
+ public String reclaimExpiredUnbound(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs) {
+ String now = currentTimeMillisExpr();
+ return "UPDATE "
+ + TABLE_NAME
+ + " SET created_at = "
+ + now
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId}"
+ + " AND job_id IS NULL AND finished_at = 0"
+ + " AND created_at <= ("
+ + now
+ + ") - #{claimTimeoutMs}";
+ }
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement for the unbound claim row
+ */
+ public String selectUnboundByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return selectColumns()
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId}"
+ + " AND job_id IS NULL AND finished_at = 0";
+ }
+
+ /**
+ * Live unbound claim still within TTL on the database clock.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL in milliseconds
+ * @return SELECT statement
+ */
+ public String selectLiveUnboundByTableAndPolicy(
+ @Param("tableId") long tableId,
+ @Param("policyId") long policyId,
+ @Param("claimTimeoutMs") long claimTimeoutMs) {
+ return selectColumns()
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId}"
+ + " AND job_id IS NULL AND finished_at = 0"
+ + " AND created_at > ("
+ + currentTimeMillisExpr()
+ + ") - #{claimTimeoutMs}";
+ }
+
+ /**
+ * Binds the Gravitino job run id after {@code runJob}. Keeps {@code
finished_at = 0} so the soft
+ * slot covers the whole run. {@code expectedCreatedAt} fences against a
concurrent reclaim that
+ * CAS-updated {@code created_at}.
*
* @param id action row id
* @param jobId {@code job_run_meta.job_run_id}
+ * @param expectedCreatedAt claim lease start observed at acquire time
* @return UPDATE statement
*/
- public String bindJobId(@Param("id") long id, @Param("jobId") long jobId) {
+ public String bindJobId(
+ @Param("id") long id,
+ @Param("jobId") long jobId,
+ @Param("expectedCreatedAt") long expectedCreatedAt) {
return "UPDATE "
+ TABLE_NAME
+ " SET job_id = #{jobId}"
- + " WHERE id = #{id} AND job_id IS NULL AND finished_at IS NULL";
+ + " WHERE id = #{id} AND job_id IS NULL AND finished_at = 0"
+ + " AND created_at = #{expectedCreatedAt}";
}
/**
@@ -82,15 +157,18 @@ public class TableMaintenanceActionBaseSQLProvider {
return "UPDATE "
+ TABLE_NAME
+ " SET before_snapshot_id = #{beforeSnapshotId}"
- + " WHERE job_id = #{jobId} AND before_snapshot_id IS NULL AND
finished_at IS NULL";
+ + " WHERE job_id = #{jobId} AND before_snapshot_id IS NULL AND
finished_at = 0";
}
/**
- * Sets terminal snapshot pointers when the Spark Job finishes.
+ * Sets terminal snapshot pointers when the Spark Job finishes and releases
the soft slot.
+ *
+ * <p>Only one in-flight row exists per {@code (table_id, policy_id)}
({@code finished_at = 0}),
+ * so finish timestamps do not contend across concurrent submitters.
*
* @param jobId Gravitino job run id
* @param afterSnapshotId terminal snapshot id
- * @param finishedAt finish time millis
+ * @param finishedAt finish time millis; must be {@code > 0}
* @return UPDATE statement
*/
public String markFinished(
@@ -100,7 +178,7 @@ public class TableMaintenanceActionBaseSQLProvider {
return "UPDATE "
+ TABLE_NAME
+ " SET after_snapshot_id = #{afterSnapshotId}, finished_at =
#{finishedAt}"
- + " WHERE job_id = #{jobId} AND finished_at IS NULL";
+ + " WHERE job_id = #{jobId} AND finished_at = 0";
}
/**
@@ -119,6 +197,33 @@ public class TableMaintenanceActionBaseSQLProvider {
return "DELETE FROM " + TABLE_NAME + " WHERE job_id = #{jobId}";
}
+ /**
+ * Returns the open row ({@code finished_at = 0}) for {@code (table_id,
policy_id)}, if any.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement
+ */
+ public String selectInFlightByTableAndPolicy(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return selectColumns()
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId} AND
finished_at = 0";
+ }
+
+ /**
+ * Returns {@code MAX(finished_at)} among completed rows for {@code
(table_id, policy_id)}.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return SELECT statement
+ */
+ public String selectMaxFinishedAt(
+ @Param("tableId") long tableId, @Param("policyId") long policyId) {
+ return "SELECT MAX(finished_at) FROM "
+ + TABLE_NAME
+ + " WHERE table_id = #{tableId} AND policy_id = #{policyId} AND
finished_at > 0";
+ }
+
private static String selectColumns() {
return "SELECT id, job_id AS jobId, table_id AS tableId, policy_id AS
policyId,"
+ " before_snapshot_id AS beforeSnapshotId, after_snapshot_id AS
afterSnapshotId,"
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
index 49f89e7827..bcc32f49ea 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/postgresql/TableMaintenanceActionPostgreSQLProvider.java
@@ -19,7 +19,14 @@
package org.apache.gravitino.maintenance.storage.mapper.provider.postgresql;
import
org.apache.gravitino.maintenance.storage.mapper.provider.base.TableMaintenanceActionBaseSQLProvider;
+import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL;
/** PostgreSQL SQL provider for {@code table_maintenance_action}. */
public class TableMaintenanceActionPostgreSQLProvider
- extends TableMaintenanceActionBaseSQLProvider {}
+ extends TableMaintenanceActionBaseSQLProvider {
+
+ @Override
+ protected String currentTimeMillisExpr() {
+ return DatabaseTimeSQL.POSTGRESQL;
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java
new file mode 100644
index 0000000000..adc4f5bcdd
--- /dev/null
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java
@@ -0,0 +1,82 @@
+/*
+ * 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.maintenance.storage.service;
+
+import java.util.Objects;
+
+/**
+ * Immutable occupancy claim returned by acquire.
+ *
+ * <p>Callers must pass {@link #createdAt()} to {@link
TableMaintenanceActionMetaService#bindJobId}
+ * without re-reading {@code created_at} from the row.
+ *
+ * <p>{@link #id()} is the durable claim token for OccupancyGate follow-ups:
make {@code runJob}
+ * idempotent on this id (or outbox / reconcile-before-reclaim). Binding
{@code job_id} alone cannot
+ * undo a submit that already succeeded after a crash between {@code runJob}
and bind.
+ */
+public final class OccupancyClaim {
+
+ private final long id;
+ private final long createdAt;
+
+ /**
+ * @param id {@code table_maintenance_action.id}
+ * @param createdAt claim lease start written at acquire / reclaim time
+ */
+ public OccupancyClaim(long id, long createdAt) {
+ this.id = id;
+ this.createdAt = createdAt;
+ }
+
+ /**
+ * @return action row id
+ */
+ public long id() {
+ return id;
+ }
+
+ /**
+ * @return {@code created_at} observed when this claim was acquired
+ */
+ public long createdAt() {
+ return createdAt;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof OccupancyClaim)) {
+ return false;
+ }
+ OccupancyClaim that = (OccupancyClaim) o;
+ return id == that.id && createdAt == that.createdAt;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, createdAt);
+ }
+
+ @Override
+ public String toString() {
+ return "OccupancyClaim{id=" + id + ", createdAt=" + createdAt + '}';
+ }
+}
diff --git
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
index ea5c570d3e..9814cc3c34 100644
---
a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
+++
b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java
@@ -21,7 +21,13 @@ package org.apache.gravitino.maintenance.storage.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
import com.google.common.base.Preconditions;
+import java.sql.SQLIntegrityConstraintViolationException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
import javax.annotation.Nullable;
+import org.apache.gravitino.maintenance.TableMaintenanceConfig;
import
org.apache.gravitino.maintenance.storage.mapper.TableMaintenanceActionMapper;
import org.apache.gravitino.maintenance.storage.po.TableMaintenanceActionPO;
import org.apache.gravitino.metrics.Monitored;
@@ -43,48 +49,146 @@ public class TableMaintenanceActionMetaService {
}
/**
- * 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}), using
+ * the database clock for lease age
+ * <li>Otherwise {@code INSERT} with {@code finished_at = 0} and DB {@code
created_at}; 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 from the database
+ * clock so multi-node TTL does not depend on JVM skew.
+ *
+ * <p><b>Submit contract:</b> {@link #bindJobId} only fences the DB bind. It
cannot undo a {@code
+ * runJob} that already succeeded. OccupancyGate / submit follow-ups must
make {@code runJob}
+ * idempotent on {@link OccupancyClaim#id()} (or use an outbox /
reconcile-before-reclaim) so a
+ * reclaim after a crash between submit and bind cannot double-submit Spark
work.
*
* @param tableId table entity id
* @param policyId policy entity id
- * @return generated action id
+ * @param claimTimeoutMs unbound claim TTL
+ * @return claim ({@code id} + DB {@code created_at}) 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 Optional<OccupancyClaim> tryAcquireInFlight(
+ long tableId, long policyId, long claimTimeoutMs) {
+ Preconditions.checkArgument(claimTimeoutMs > 0, "claimTimeoutMs must be
positive");
+ int reclaimed =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.reclaimExpiredUnbound(tableId, policyId,
claimTimeoutMs));
+ if (reclaimed == 1) {
+ return Optional.of(readUnboundClaim(tableId, policyId));
+ }
+
+ TableMaintenanceActionPO live =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectLiveUnboundByTableAndPolicy(tableId,
policyId, claimTimeoutMs));
+ if (live != null) {
+ return Optional.empty();
+ }
+
TableMaintenanceActionPO actionMeta =
TableMaintenanceActionPO.builder()
.withTableId(tableId)
.withPolicyId(policyId)
- .withCreatedAt(System.currentTimeMillis())
+ .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 Optional.empty();
+ }
+ throw e;
+ }
Preconditions.checkState(actionMeta.id() != null, "insert did not return
generated id");
- return actionMeta.id();
+ TableMaintenanceActionPO inserted =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.selectById(actionMeta.id()));
+ Preconditions.checkState(
+ inserted != null && inserted.createdAt() != null, "insert lost
created_at");
+ return Optional.of(new OccupancyClaim(inserted.id(),
inserted.createdAt()));
}
/**
- * Binds {@code job_run_meta.job_run_id} after a successful {@code runJob}.
+ * Acquires an unbound claim using {@link
TableMaintenanceConfig#claimTimeoutMs()} or fails.
*
- * @param id action row id
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param config maintenance config supplying the unbound claim TTL
+ * @return acquired claim
+ */
+ public OccupancyClaim insertInFlight(long tableId, long policyId,
TableMaintenanceConfig config) {
+ Preconditions.checkNotNull(config, "config must not be null");
+ return insertInFlight(tableId, policyId, config.claimTimeoutMs());
+ }
+
+ /**
+ * Acquires an unbound claim with the given claim timeout or fails.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL
+ * @return acquired claim
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertTableMaintenanceAction")
+ public OccupancyClaim insertInFlight(long tableId, long policyId, long
claimTimeoutMs) {
+ Optional<OccupancyClaim> acquired = tryAcquireInFlight(tableId, policyId,
claimTimeoutMs);
+ Preconditions.checkState(acquired.isPresent(), "failed to acquire unbound
occupancy claim");
+ return acquired.get();
+ }
+
+ /**
+ * Binds {@code job_run_meta.job_run_id} after a successful {@code runJob}.
Does not release the
+ * soft slot ({@code finished_at} stays {@code 0} until {@link
#markFinished}).
+ *
+ * <p>Pass the {@link OccupancyClaim} from acquire / reclaim as-is. Do not
re-read {@code
+ * created_at} from the row: after another node reclaims, that would observe
the new fence and
+ * defeat reclaim fencing. The CAS matches zero rows for a stale claim.
+ *
+ * <p>This CAS does <em>not</em> prevent duplicate Spark submissions if
{@code runJob} already
+ * ran; see {@link #tryAcquireInFlight} submit contract.
+ *
+ * @param claim claim returned by {@link #tryAcquireInFlight} / {@link
#insertInFlight}
* @param jobId Gravitino job run id
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "bindTableMaintenanceActionJobId")
- public void bindJobId(long id, long jobId) {
+ public void bindJobId(OccupancyClaim claim, long jobId) {
+ Preconditions.checkNotNull(claim, "claim must not be null");
int updated =
SessionUtils.doWithCommitAndFetchResult(
- TableMaintenanceActionMapper.class, mapper -> mapper.bindJobId(id,
jobId));
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.bindJobId(claim.id(), jobId, claim.createdAt()));
Preconditions.checkState(
updated == 1,
"expected to bind job_id on exactly one in-flight action, updated=%s",
updated);
}
+ private static OccupancyClaim readUnboundClaim(long tableId, long policyId) {
+ TableMaintenanceActionPO unbound =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectUnboundByTableAndPolicy(tableId, policyId));
+ Preconditions.checkState(
+ unbound != null && unbound.id() != null && unbound.createdAt() != null,
+ "reclaim lost unbound row");
+ return new OccupancyClaim(unbound.id(), unbound.createdAt());
+ }
+
/**
* @param id action row id
* @return action row or null
@@ -134,6 +238,10 @@ public class TableMaintenanceActionMetaService {
/**
* Marks the action finished with the terminal snapshot id.
*
+ * <p>{@code UNIQUE(table_id, policy_id, finished_at)} exists so {@code
finished_at = 0} is a
+ * single soft slot per pair (like {@code deleted_at}). Only one in-flight
row can finish that
+ * pair at a time, so two completed rows cannot race to the same millisecond
{@code finished_at}.
+ *
* @param jobId Gravitino job run id
* @param afterSnapshotId terminal snapshot id
* @param finishedAt finish timestamp millis
@@ -142,6 +250,7 @@ public class TableMaintenanceActionMetaService {
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "markTableMaintenanceActionFinished")
public void markFinished(long jobId, Long afterSnapshotId, long finishedAt) {
+ Preconditions.checkArgument(finishedAt > 0, "finishedAt must be positive");
int updated =
SessionUtils.doWithCommitAndFetchResult(
TableMaintenanceActionMapper.class,
@@ -175,4 +284,52 @@ public class TableMaintenanceActionMetaService {
SessionUtils.doWithCommit(
TableMaintenanceActionMapper.class, mapper ->
mapper.deleteByJobId(jobId));
}
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return open rows with {@code finished_at = 0} (at most one per
table/policy; may include a
+ * terminal job not yet marked finished)
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "listInFlightTableMaintenanceActions")
+ public List<TableMaintenanceActionPO> listInFlight(long tableId, long
policyId) {
+ List<TableMaintenanceActionPO> open =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectInFlightByTableAndPolicy(tableId,
policyId));
+ return open == null ? Collections.emptyList() : open;
+ }
+
+ /**
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @return latest {@code finished_at}, or null when none finished yet
+ */
+ @Nullable
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "getMaxFinishedAtTableMaintenanceAction")
+ public Long maxFinishedAt(long tableId, long policyId) {
+ return SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectMaxFinishedAt(tableId, policyId));
+ }
+
+ private static boolean isUniqueViolation(Throwable throwable) {
+ for (Throwable t = throwable; t != null; t = t.getCause()) {
+ if (t instanceof SQLIntegrityConstraintViolationException) {
+ return true;
+ }
+ String message = t.getMessage();
+ if (message != null) {
+ String lower = message.toLowerCase(Locale.ROOT);
+ if (lower.contains("unique") || lower.contains("duplicate")) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java
new file mode 100644
index 0000000000..5de4f541b1
--- /dev/null
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java
@@ -0,0 +1,55 @@
+/*
+ * 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.maintenance.scheduler;
+
+import javax.sql.DataSource;
+import org.apache.gravitino.maintenance.TableMaintenanceConfig;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.TestTemplate;
+
+class TestTableMaintenanceSchedulerPools extends TestJDBCBackend {
+
+ private TableMaintenanceSchedulerPools pools;
+
+ @AfterEach
+ void tearDown() {
+ if (pools != null) {
+ pools.stop();
+ }
+ }
+
+ @TestTemplate
+ void testThreePoolsStartAndStop() {
+ DataSource dataSource =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getEnvironment()
+ .getDataSource();
+ pools = new TableMaintenanceSchedulerPools();
+ Assertions.assertDoesNotThrow(
+ () ->
+ pools.start(dataSource, TableMaintenanceConfig.forTest(1, 1, 1,
500L, 1_000L, 4, 10)));
+ Assertions.assertDoesNotThrow(pools::stop);
+ }
+}
diff --git
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
index 2921255cd4..855b073d49 100644
---
a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
+++
b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.maintenance.storage.service;
+import java.util.Optional;
+import org.apache.gravitino.maintenance.TableMaintenanceConfig;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
@@ -32,15 +34,21 @@ class TestTableMaintenanceActionMetaService extends
TestJDBCBackend {
TableSnapshotMetricsMetaService metricsService =
TableSnapshotMetricsMetaService.getInstance();
long before = System.currentTimeMillis();
- long actionId = actionService.insertInFlight(10L, 20L);
+ OccupancyClaim claim =
+ actionService.insertInFlight(
+ 10L, 20L, TableMaintenanceConfig.forTest(1, 1, 1, 500L, 1_000L, 4,
10));
long after = System.currentTimeMillis();
- Assertions.assertNull(actionService.getById(actionId).jobId());
- Assertions.assertTrue(actionService.getById(actionId).createdAt() >=
before);
- Assertions.assertTrue(actionService.getById(actionId).createdAt() <=
after);
+ Assertions.assertNull(actionService.getById(claim.id()).jobId());
+ Assertions.assertEquals(0L,
actionService.getById(claim.id()).finishedAt());
+ // DB clock may skew slightly from the JVM; allow a small window.
+ Assertions.assertTrue(claim.createdAt() >= before - 5_000L);
+ Assertions.assertTrue(claim.createdAt() <= after + 5_000L);
+ Assertions.assertEquals(claim.createdAt(),
actionService.getById(claim.id()).createdAt());
long jobId = 999L;
- actionService.bindJobId(actionId, jobId);
+ actionService.bindJobId(claim, jobId);
Assertions.assertEquals(jobId, actionService.getByJobId(jobId).jobId());
+ Assertions.assertEquals(0L, actionService.getByJobId(jobId).finishedAt());
actionService.recordBeforeSnapshot(jobId, 100L);
Assertions.assertEquals(100L,
actionService.getByJobId(jobId).beforeSnapshotId());
@@ -62,4 +70,30 @@ class TestTableMaintenanceActionMetaService extends
TestJDBCBackend {
Assertions.assertEquals(
"{\"mse\":2}", metricsService.getMetrics(10L, 20L,
200L).metricsValue());
}
+
+ @TestTemplate
+ void testTryAcquireRejectsLiveUnboundAndReclaimsExpired() throws Exception {
+ TableMaintenanceActionMetaService actions =
TableMaintenanceActionMetaService.getInstance();
+ long tableId = 11L;
+ long policyId = 21L;
+ long timeoutMs = 200L;
+
+ Optional<OccupancyClaim> first = actions.tryAcquireInFlight(tableId,
policyId, timeoutMs);
+ Assertions.assertTrue(first.isPresent());
+ Assertions.assertEquals(0L,
actions.getById(first.get().id()).finishedAt());
+ Assertions.assertTrue(actions.tryAcquireInFlight(tableId, policyId,
timeoutMs).isEmpty());
+
+ Thread.sleep(timeoutMs + 50L);
+
+ Optional<OccupancyClaim> reclaimed = actions.tryAcquireInFlight(tableId,
policyId, timeoutMs);
+ Assertions.assertTrue(reclaimed.isPresent());
+ Assertions.assertEquals(first.get().id(), reclaimed.get().id());
+ Assertions.assertTrue(reclaimed.get().createdAt() >=
first.get().createdAt());
+ Assertions.assertEquals(0L,
actions.getById(reclaimed.get().id()).finishedAt());
+
+ Assertions.assertThrows(
+ IllegalStateException.class, () -> actions.bindJobId(first.get(),
1001L));
+ actions.bindJobId(reclaimed.get(), 1002L);
+ Assertions.assertEquals(1002L,
actions.getById(reclaimed.get().id()).jobId());
+ }
}
diff --git a/scripts/h2/schema-2.0.0-h2.sql b/scripts/h2/schema-2.0.0-h2.sql
index ac7fb308d1..901f3c99cd 100644
--- a/scripts/h2/schema-2.0.0-h2.sql
+++ b/scripts/h2/schema-2.0.0-h2.sql
@@ -681,19 +681,71 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
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`);
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance expand schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_expand_tasks_execution_time_idx` ON
`scheduled_expand_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_expand_tasks_last_heartbeat_idx` ON
`scheduled_expand_tasks`(`last_heartbeat`);
+
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance cron schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_cron_tasks_execution_time_idx` ON
`scheduled_cron_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_cron_tasks_last_heartbeat_idx` ON
`scheduled_cron_tasks`(`last_heartbeat`);
+
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance commit schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_commit_tasks_execution_time_idx` ON
`scheduled_commit_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_commit_tasks_last_heartbeat_idx` ON
`scheduled_commit_tasks`(`last_heartbeat`);
+
CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
`id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
- `job_id` BIGINT COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `job_id` BIGINT COMMENT 'job run id, maps to job_run_meta.job_run_id',
`table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
`policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
`before_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
`after_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id after maintenance,
null while pending',
- `finished_at` BIGINT COMMENT 'finish timestamp in milliseconds, null while
in flight',
+ `finished_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'finish millis, 0 = soft
in-flight slot (one per table/policy)',
`created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
PRIMARY KEY (`id`)
-) COMMENT='table maintenance action occupancy and run metadata';
+) COMMENT='table maintenance action metadata';
CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_job_id` ON
`table_maintenance_action`(`job_id`);
-CREATE INDEX IF NOT EXISTS `idx_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
+-- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
CREATE INDEX IF NOT EXISTS `idx_tma_before_sid` ON
`table_maintenance_action`(`table_id`, `before_snapshot_id`);
CREATE INDEX IF NOT EXISTS `idx_tma_after_sid` ON
`table_maintenance_action`(`table_id`, `after_snapshot_id`);
CREATE INDEX IF NOT EXISTS `idx_tma_created_at` ON
`table_maintenance_action`(`created_at`);
diff --git a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
index fb4f838a66..ffc5b9ca11 100644
--- a/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
+++ b/scripts/h2/upgrade-1.3.0-to-2.0.0-h2.sql
@@ -102,19 +102,71 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB COMMENT 'semantic model version information';
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance expand schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_expand_tasks_execution_time_idx` ON
`scheduled_expand_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_expand_tasks_last_heartbeat_idx` ON
`scheduled_expand_tasks`(`last_heartbeat`);
+
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance cron schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_cron_tasks_execution_time_idx` ON
`scheduled_cron_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_cron_tasks_last_heartbeat_idx` ON
`scheduled_cron_tasks`(`last_heartbeat`);
+
+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) 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) COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`)
+) COMMENT='table maintenance commit schedule metadata';
+CREATE INDEX IF NOT EXISTS `scheduled_commit_tasks_execution_time_idx` ON
`scheduled_commit_tasks`(`execution_time`);
+CREATE INDEX IF NOT EXISTS `scheduled_commit_tasks_last_heartbeat_idx` ON
`scheduled_commit_tasks`(`last_heartbeat`);
+
CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
`id` BIGINT AUTO_INCREMENT NOT NULL COMMENT 'auto increment id',
- `job_id` BIGINT COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `job_id` BIGINT COMMENT 'job run id, maps to job_run_meta.job_run_id',
`table_id` BIGINT NOT NULL COMMENT 'table id from table_meta',
`policy_id` BIGINT NOT NULL COMMENT 'policy id from policy_meta',
`before_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
`after_snapshot_id` BIGINT COMMENT 'Iceberg snapshot id after maintenance,
null while pending',
- `finished_at` BIGINT COMMENT 'finish timestamp in milliseconds, null while
in flight',
+ `finished_at` BIGINT NOT NULL DEFAULT 0 COMMENT 'finish millis, 0 = soft
in-flight slot (one per table/policy)',
`created_at` BIGINT NOT NULL COMMENT 'creation timestamp in milliseconds',
PRIMARY KEY (`id`)
-) COMMENT='table maintenance action occupancy and run metadata';
+) COMMENT='table maintenance action metadata';
CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_job_id` ON
`table_maintenance_action`(`job_id`);
-CREATE INDEX IF NOT EXISTS `idx_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
+-- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+CREATE UNIQUE INDEX IF NOT EXISTS `uk_tma_table_policy_finished` ON
`table_maintenance_action`(`table_id`, `policy_id`, `finished_at`);
CREATE INDEX IF NOT EXISTS `idx_tma_before_sid` ON
`table_maintenance_action`(`table_id`, `before_snapshot_id`);
CREATE INDEX IF NOT EXISTS `idx_tma_after_sid` ON
`table_maintenance_action`(`table_id`, `after_snapshot_id`);
CREATE INDEX IF NOT EXISTS `idx_tma_created_at` ON
`table_maintenance_action`(`created_at`);
diff --git a/scripts/mysql/schema-2.0.0-mysql.sql
b/scripts/mysql/schema-2.0.0-mysql.sql
index 786356bcf1..db39e065ba 100644
--- a/scripts/mysql/schema-2.0.0-mysql.sql
+++ b/scripts/mysql/schema-2.0.0-mysql.sql
@@ -668,23 +668,74 @@ 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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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
millis, 0 = soft in-flight slot (one per table/policy)',
`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`),
+ -- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+ UNIQUE KEY `uk_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
KEY `idx_tma_before_sid` (`table_id`, `before_snapshot_id`),
KEY `idx_tma_after_sid` (`table_id`, `after_snapshot_id`),
KEY `idx_tma_created_at` (`created_at`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
- COMMENT 'table maintenance action occupancy and run metadata';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance action metadata';
CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
@@ -696,5 +747,4 @@ CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tid_pid_sid` (`table_id`, `policy_id`, `snapshot_id`),
KEY `idx_tsm_created_at` (`created_at`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
- COMMENT 'table snapshot metrics metadata';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
snapshot metrics metadata';
diff --git a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
index c5108a79c9..26677f167e 100644
--- a/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
+++ b/scripts/mysql/upgrade-1.3.0-to-2.0.0-mysql.sql
@@ -549,23 +549,74 @@ CREATE TABLE IF NOT EXISTS `semantic_model_version_info` (
KEY `idx_smvi_sid` (`schema_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'semantic
model version information';
+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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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` DATETIME(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` DATETIME(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` DATETIME(6) NULL COMMENT 'last successful execution time',
+ `last_failure` DATETIME(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
millis, 0 = soft in-flight slot (one per table/policy)',
`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`),
+ -- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+ UNIQUE KEY `uk_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
KEY `idx_tma_before_sid` (`table_id`, `before_snapshot_id`),
KEY `idx_tma_after_sid` (`table_id`, `after_snapshot_id`),
KEY `idx_tma_created_at` (`created_at`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
- COMMENT 'table maintenance action occupancy and run metadata';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance action metadata';
CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
@@ -577,8 +628,7 @@ CREATE TABLE IF NOT EXISTS `table_snapshot_metrics` (
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tid_pid_sid` (`table_id`, `policy_id`, `snapshot_id`),
KEY `idx_tsm_created_at` (`created_at`)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin
- COMMENT 'table snapshot metrics metadata';
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
snapshot metrics metadata';
-- Merge duplicate live owners left by concurrent assignments: the newest live
row
-- (largest id) wins, and older ones are soft-deleted.
diff --git a/scripts/postgresql/schema-2.0.0-postgresql.sql
b/scripts/postgresql/schema-2.0.0-postgresql.sql
index b64a8d7fbb..19868d56db 100644
--- a/scripts/postgresql/schema-2.0.0-postgresql.sql
+++ b/scripts/postgresql/schema-2.0.0-postgresql.sql
@@ -1176,6 +1176,93 @@ 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 WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_expand_tasks_execution_time_idx ON
scheduled_expand_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_expand_tasks_last_heartbeat_idx ON
scheduled_expand_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_expand_tasks IS 'table maintenance expand schedule
metadata';
+COMMENT ON COLUMN scheduled_expand_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_expand_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_expand_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_expand_tasks.execution_time IS 'next execution
time';
+COMMENT ON COLUMN scheduled_expand_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_expand_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_expand_tasks.last_heartbeat IS 'last heartbeat
from the picker';
+COMMENT ON COLUMN scheduled_expand_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_expand_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_expand_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_expand_tasks.last_failure IS 'last failed
execution time';
+
+CREATE TABLE IF NOT EXISTS scheduled_cron_tasks (
+ task_name VARCHAR(100) NOT NULL,
+ task_instance VARCHAR(100) NOT NULL,
+ task_data BYTEA,
+ execution_time TIMESTAMP WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_cron_tasks_execution_time_idx ON
scheduled_cron_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_cron_tasks_last_heartbeat_idx ON
scheduled_cron_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_cron_tasks IS 'table maintenance cron schedule
metadata';
+COMMENT ON COLUMN scheduled_cron_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_cron_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_cron_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_cron_tasks.execution_time IS 'next execution time';
+COMMENT ON COLUMN scheduled_cron_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_cron_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_cron_tasks.last_heartbeat IS 'last heartbeat from
the picker';
+COMMENT ON COLUMN scheduled_cron_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_cron_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_cron_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_cron_tasks.last_failure IS 'last failed execution
time';
+
+CREATE TABLE IF NOT EXISTS scheduled_commit_tasks (
+ task_name VARCHAR(100) NOT NULL,
+ task_instance VARCHAR(100) NOT NULL,
+ task_data BYTEA,
+ execution_time TIMESTAMP WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_commit_tasks_execution_time_idx ON
scheduled_commit_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_commit_tasks_last_heartbeat_idx ON
scheduled_commit_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_commit_tasks IS 'table maintenance commit schedule
metadata';
+COMMENT ON COLUMN scheduled_commit_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_commit_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_commit_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_commit_tasks.execution_time IS 'next execution
time';
+COMMENT ON COLUMN scheduled_commit_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_commit_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_commit_tasks.last_heartbeat IS 'last heartbeat
from the picker';
+COMMENT ON COLUMN scheduled_commit_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_commit_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_commit_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_commit_tasks.last_failure IS 'last failed
execution time';
+
CREATE TABLE IF NOT EXISTS table_maintenance_action (
id BIGSERIAL PRIMARY KEY,
job_id BIGINT,
@@ -1183,22 +1270,23 @@ CREATE TABLE IF NOT EXISTS table_maintenance_action (
policy_id BIGINT NOT NULL,
before_snapshot_id BIGINT,
after_snapshot_id BIGINT,
- finished_at BIGINT,
+ finished_at BIGINT NOT NULL DEFAULT 0,
created_at BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_job_id ON table_maintenance_action
(job_id);
-CREATE INDEX IF NOT EXISTS idx_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
+-- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
CREATE INDEX IF NOT EXISTS idx_tma_before_sid ON table_maintenance_action
(table_id, before_snapshot_id);
CREATE INDEX IF NOT EXISTS idx_tma_after_sid ON table_maintenance_action
(table_id, after_snapshot_id);
CREATE INDEX IF NOT EXISTS idx_tma_created_at ON table_maintenance_action
(created_at);
-COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
occupancy and run metadata';
+COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
metadata';
COMMENT ON COLUMN table_maintenance_action.id IS 'auto increment id';
-COMMENT ON COLUMN table_maintenance_action.job_id IS 'optional job run id
mapped to job_run_meta.job_run_id, null until bound after runJob';
+COMMENT ON COLUMN table_maintenance_action.job_id IS 'job run id, maps to
job_run_meta.job_run_id';
COMMENT ON COLUMN table_maintenance_action.table_id IS 'table id from
table_meta';
COMMENT ON COLUMN table_maintenance_action.policy_id IS 'policy id from
policy_meta';
COMMENT ON COLUMN table_maintenance_action.before_snapshot_id IS 'Iceberg
snapshot id before maintenance, null until sampled';
COMMENT ON COLUMN table_maintenance_action.after_snapshot_id IS 'Iceberg
snapshot id after maintenance, null while pending';
-COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish timestamp in
milliseconds, null while in flight';
+COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish millis, 0 =
soft in-flight slot (one per table/policy)';
COMMENT ON COLUMN table_maintenance_action.created_at IS 'creation timestamp
in milliseconds';
CREATE TABLE IF NOT EXISTS table_snapshot_metrics (
diff --git a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
index 6fde2a8e27..b6deb5b273 100644
--- a/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
+++ b/scripts/postgresql/upgrade-1.3.0-to-2.0.0-postgresql.sql
@@ -151,6 +151,93 @@ COMMENT ON COLUMN semantic_model_version_info.properties
IS 'semantic model prop
COMMENT ON COLUMN semantic_model_version_info.audit_info IS 'semantic model
version audit info';
COMMENT ON COLUMN semantic_model_version_info.deleted_at IS 'version deleted
at';
+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 WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_expand_tasks_execution_time_idx ON
scheduled_expand_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_expand_tasks_last_heartbeat_idx ON
scheduled_expand_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_expand_tasks IS 'table maintenance expand schedule
metadata';
+COMMENT ON COLUMN scheduled_expand_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_expand_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_expand_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_expand_tasks.execution_time IS 'next execution
time';
+COMMENT ON COLUMN scheduled_expand_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_expand_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_expand_tasks.last_heartbeat IS 'last heartbeat
from the picker';
+COMMENT ON COLUMN scheduled_expand_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_expand_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_expand_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_expand_tasks.last_failure IS 'last failed
execution time';
+
+CREATE TABLE IF NOT EXISTS scheduled_cron_tasks (
+ task_name VARCHAR(100) NOT NULL,
+ task_instance VARCHAR(100) NOT NULL,
+ task_data BYTEA,
+ execution_time TIMESTAMP WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_cron_tasks_execution_time_idx ON
scheduled_cron_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_cron_tasks_last_heartbeat_idx ON
scheduled_cron_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_cron_tasks IS 'table maintenance cron schedule
metadata';
+COMMENT ON COLUMN scheduled_cron_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_cron_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_cron_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_cron_tasks.execution_time IS 'next execution time';
+COMMENT ON COLUMN scheduled_cron_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_cron_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_cron_tasks.last_heartbeat IS 'last heartbeat from
the picker';
+COMMENT ON COLUMN scheduled_cron_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_cron_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_cron_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_cron_tasks.last_failure IS 'last failed execution
time';
+
+CREATE TABLE IF NOT EXISTS scheduled_commit_tasks (
+ task_name VARCHAR(100) NOT NULL,
+ task_instance VARCHAR(100) NOT NULL,
+ task_data BYTEA,
+ execution_time TIMESTAMP WITH TIME ZONE NOT NULL,
+ picked BOOLEAN NOT NULL,
+ picked_by VARCHAR(50),
+ last_heartbeat TIMESTAMP WITH TIME ZONE,
+ version BIGINT NOT NULL,
+ consecutive_failures INT,
+ last_success TIMESTAMP WITH TIME ZONE,
+ last_failure TIMESTAMP WITH TIME ZONE,
+ PRIMARY KEY (task_name, task_instance)
+);
+CREATE INDEX IF NOT EXISTS scheduled_commit_tasks_execution_time_idx ON
scheduled_commit_tasks (execution_time);
+CREATE INDEX IF NOT EXISTS scheduled_commit_tasks_last_heartbeat_idx ON
scheduled_commit_tasks (last_heartbeat);
+COMMENT ON TABLE scheduled_commit_tasks IS 'table maintenance commit schedule
metadata';
+COMMENT ON COLUMN scheduled_commit_tasks.task_name IS 'task name';
+COMMENT ON COLUMN scheduled_commit_tasks.task_instance IS 'task instance';
+COMMENT ON COLUMN scheduled_commit_tasks.task_data IS 'serialized task data';
+COMMENT ON COLUMN scheduled_commit_tasks.execution_time IS 'next execution
time';
+COMMENT ON COLUMN scheduled_commit_tasks.picked IS 'whether the task is
currently picked';
+COMMENT ON COLUMN scheduled_commit_tasks.picked_by IS 'scheduler instance that
picked the task';
+COMMENT ON COLUMN scheduled_commit_tasks.last_heartbeat IS 'last heartbeat
from the picker';
+COMMENT ON COLUMN scheduled_commit_tasks.version IS 'optimistic lock version';
+COMMENT ON COLUMN scheduled_commit_tasks.consecutive_failures IS 'consecutive
failure count';
+COMMENT ON COLUMN scheduled_commit_tasks.last_success IS 'last successful
execution time';
+COMMENT ON COLUMN scheduled_commit_tasks.last_failure IS 'last failed
execution time';
+
CREATE TABLE IF NOT EXISTS table_maintenance_action (
id BIGSERIAL PRIMARY KEY,
job_id BIGINT,
@@ -158,22 +245,23 @@ CREATE TABLE IF NOT EXISTS table_maintenance_action (
policy_id BIGINT NOT NULL,
before_snapshot_id BIGINT,
after_snapshot_id BIGINT,
- finished_at BIGINT,
+ finished_at BIGINT NOT NULL DEFAULT 0,
created_at BIGINT NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_job_id ON table_maintenance_action
(job_id);
-CREATE INDEX IF NOT EXISTS idx_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
+-- Soft slot: finished_at=0 is exclusive per table/policy (one in-flight
finish).
+CREATE UNIQUE INDEX IF NOT EXISTS uk_tma_table_policy_finished ON
table_maintenance_action (table_id, policy_id, finished_at);
CREATE INDEX IF NOT EXISTS idx_tma_before_sid ON table_maintenance_action
(table_id, before_snapshot_id);
CREATE INDEX IF NOT EXISTS idx_tma_after_sid ON table_maintenance_action
(table_id, after_snapshot_id);
CREATE INDEX IF NOT EXISTS idx_tma_created_at ON table_maintenance_action
(created_at);
-COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
occupancy and run metadata';
+COMMENT ON TABLE table_maintenance_action IS 'table maintenance action
metadata';
COMMENT ON COLUMN table_maintenance_action.id IS 'auto increment id';
-COMMENT ON COLUMN table_maintenance_action.job_id IS 'optional job run id
mapped to job_run_meta.job_run_id, null until bound after runJob';
+COMMENT ON COLUMN table_maintenance_action.job_id IS 'job run id, maps to
job_run_meta.job_run_id';
COMMENT ON COLUMN table_maintenance_action.table_id IS 'table id from
table_meta';
COMMENT ON COLUMN table_maintenance_action.policy_id IS 'policy id from
policy_meta';
COMMENT ON COLUMN table_maintenance_action.before_snapshot_id IS 'Iceberg
snapshot id before maintenance, null until sampled';
COMMENT ON COLUMN table_maintenance_action.after_snapshot_id IS 'Iceberg
snapshot id after maintenance, null while pending';
-COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish timestamp in
milliseconds, null while in flight';
+COMMENT ON COLUMN table_maintenance_action.finished_at IS 'finish millis, 0 =
soft in-flight slot (one per table/policy)';
COMMENT ON COLUMN table_maintenance_action.created_at IS 'creation timestamp
in milliseconds';
CREATE TABLE IF NOT EXISTS table_snapshot_metrics (
