[#13984] feat(tms): Expand policy-expand into cron-job rows (nearest-policy)

plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/CronJobData.java

@@ -0,0 +1,72 @@
+/*
+ * 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 java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Payload for {@code cron-job} rows. One table for {@code table:...} instance ids; many tables for
+ * batched snapshot-expiry {@code batch:...} instance ids (§5.2.2).
+ */
+public final class CronJobData implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  private final ArrayList<Long> tableIds;
+
+  /**
+   * @param tableIds table entity ids covered by this cron-job
+   */
+  public CronJobData(List<Long> tableIds) {
+    this.tableIds = new ArrayList<>(tableIds == null ? List.of() : tableIds);
+  }
+
+  /**
+   * @return unmodifiable table ids
+   */
+  public List<Long> tableIds() {
+    return Collections.unmodifiableList(tableIds);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof CronJobData)) {
+      return false;
+    }
+    CronJobData that = (CronJobData) o;
+    return Objects.equals(tableIds, that.tableIds);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(tableIds);
+  }
+
+  @Override
+  public String toString() {
+    return "CronJobData{tableIds=" + tableIds + '}';
+  }
+}

plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/CronJobIds.java

@@ -0,0 +1,79 @@
+/*
+ * 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 java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Builds db-scheduler instance ids for one-shot {@code cron-job} rows (§5.2.2).
+ *
+ * <p>Most types use {@code table:{table_id}:{policy_id}}. Batched snapshot-expiry uses {@code
+ * batch:{batch_key}:{policy_id}} where {@code batch_key} is a stable hash of the table id set.
+ */
+public final class CronJobIds {
+
+  private CronJobIds() {}
+
+  /**
+   * @param tableId table entity id
+   * @param policyId policy entity id
+   * @return cron-job instance id
+   */
+  public static String forTable(long tableId, long policyId) {
+    return "table:" + tableId + ":" + policyId;
+  }
+
+  /**
+   * @param policyId policy entity id
+   * @param tableIds tables covered by the batch (order ignored)
+   * @return cron-job instance id for a snapshot-expiry batch
+   */
+  public static String forBatch(long policyId, List<Long> tableIds) {
+    return "batch:" + batchKey(tableIds) + ":" + policyId;
+  }
+
+  /**
+   * Stable short hash of the sorted table-id set.
+   *
+   * @param tableIds table entity ids
+   * @return hex batch key
+   */
+  public static String batchKey(List<Long> tableIds) {
+    List<Long> sorted = new ArrayList<>(tableIds == null ? List.of() : tableIds);
+    Collections.sort(sorted);
+    String material = sorted.stream().map(String::valueOf).collect(Collectors.joining(","));
+    try {
+      MessageDigest digest = MessageDigest.getInstance("SHA-256");
+      byte[] hash = digest.digest(material.getBytes(StandardCharsets.UTF_8));
+      StringBuilder hex = new StringBuilder(16);
+      for (int i = 0; i < 8; i++) {
+        hex.append(String.format("%02x", hash[i]));
+      }
+      return hex.toString();
+    } catch (NoSuchAlgorithmException e) {
+      throw new IllegalStateException("SHA-256 not available", e);
+    }
+  }
+}

plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/CronJobScheduler.java

@@ -0,0 +1,48 @@
+/*
+ * 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 java.util.List;
+import org.apache.gravitino.policy.Policy;
+
+/**
+ * Inserts one-shot {@code cron-job} rows into the cron-pool db-scheduler table.
+ *
+ * <p>Existing instance ids should be left untouched so a repeated expand pick does not reset an
+ * already-queued submit. Snapshot-expiry winners attached at schema/catalog may be merged into
+ * {@code batch:{batch_key}:{policy_id}} rows (§5.2.2).
+ */
+@FunctionalInterface
+public interface CronJobScheduler {
+
+  /**
+   * Enqueues due-now cron jobs for the given expand candidates.
+   *
+   * @param policyType maintenance built-in type
+   * @param policyId policy id encoded in each instance id
+   * @param candidates enqueueable tables with winner levels
+   * @param batchSize max tables considered for this expand pick (caller may already truncate)
+   * @return number of newly scheduled cron-job rows
+   */
+  int enqueueDueNow(
+      Policy.BuiltInType policyType,
+      long policyId,
+      List<ExpandTableCandidate> candidates,
+      int batchSize);
+}

plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/ExpandOccupancyGate.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.maintenance.scheduler;
+
+/**
+ * Gate used by expand before inserting a {@code cron-job} row.
+ *
+ * <p>Production wires {@link org.apache.gravitino.maintenance.occupancy.OccupancyGate#shouldSkip}.
+ */
+@FunctionalInterface
+public interface ExpandOccupancyGate {
+
+  /**
+   * @param tableId table entity id
+   * @param policyId policy entity id
+   * @param minIntervalMs resolved cooldown for the policy type
+   * @param nowMs current epoch millis
+   * @return true when expand must not enqueue a cron-job for this pair
+   */
+  boolean shouldSkip(long tableId, long policyId, long minIntervalMs, long nowMs);
+}

plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/ExpandTableCandidate.java

@@ -0,0 +1,88 @@
+/*
+ * 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.google.common.base.Preconditions;
+import java.util.Objects;
+import org.apache.gravitino.MetadataObject;
+
+/**
+ * One enqueueable table after nearest-policy and occupancy gating, with the metadata level where
+ * the winning policy was selected.
+ */
+public final class ExpandTableCandidate {
+
+  private final long tableId;
+  private final MetadataObject.Type winnerLevel;
+
+  /**
+   * @param tableId table entity id
+   * @param winnerLevel {@link MetadataObject.Type#TABLE}, {@link MetadataObject.Type#SCHEMA}, or
+   *     {@link MetadataObject.Type#CATALOG}
+   */
+  public ExpandTableCandidate(long tableId, MetadataObject.Type winnerLevel) {
+    Preconditions.checkArgument(tableId > 0, "tableId must be positive");
+    Preconditions.checkNotNull(winnerLevel, "winnerLevel must not be null");
+    this.tableId = tableId;
+    this.winnerLevel = winnerLevel;
+  }
+
+  /**
+   * @return table entity id
+   */
+  public long tableId() {
+    return tableId;
+  }
+
+  /**
+   * @return level where the nearest winning policy was attached
+   */
+  public MetadataObject.Type winnerLevel() {
+    return winnerLevel;
+  }
+
+  /**
+   * @return true when snapshot-expiry may merge this table into a batch cron-job
+   */
+  public boolean batchableForSnapshotExpiry() {
+    return winnerLevel == MetadataObject.Type.SCHEMA || winnerLevel == MetadataObject.Type.CATALOG;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof ExpandTableCandidate)) {
+      return false;
+    }
+    ExpandTableCandidate that = (ExpandTableCandidate) o;
+    return tableId == that.tableId && winnerLevel == that.winnerLevel;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(tableId, winnerLevel);
+  }
+
+  @Override
+  public String toString() {
+    return "ExpandTableCandidate{tableId=" + tableId + ", winnerLevel=" + winnerLevel + '}';
+  }
+}