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
commit be91ee91e7487d68fc900394539d97a2a28692b4 Author: MaSai <[email protected]> AuthorDate: Fri Oct 16 10:01:18 2026 +0800 [#13800] feat(policy): Maintenance execution peers on Iceberg contents (#13885) Split from #13801 (1/2). **Domain** execution peers for Epic #13764 / - `PolicyExecutionSpec` / `ExecutablePolicyContent` / `PolicyExecutionFields` - Four Iceberg maintenance contents carry peers at construction - TMS design notes on peers / `BuiltInType` identity **Follow-up (stack):** #13888 / #13887 — wire/DTO/SerDe + `minIntervalMs` registry. TMS needs shared trigger / cooldown / option-bag concepts without freezing them onto every `PolicyContent`. Fix: #13800 API/domain surface for optional peers on Iceberg maintenance contents (REST wire lands in #13887). ```bash ./gradlew :api:test --tests 'org.apache.gravitino.policy.*' -PskipITs ``` --------- Co-authored-by: Cursor <[email protected]> --- .../policy/AbstractExecutablePolicyContent.java | 112 +++++++ .../gravitino/policy/ExecutablePolicyContent.java | 35 ++ .../policy/IcebergDataCompactionContent.java | 78 +++-- .../policy/IcebergOrphanFileRemovalContent.java | 23 +- .../policy/IcebergRewriteManifestsContent.java | 46 ++- .../policy/IcebergSnapshotExpirationContent.java | 2 +- .../org/apache/gravitino/policy/PolicyContent.java | 11 +- .../apache/gravitino/policy/PolicyContents.java | 156 ++++++++- .../gravitino/policy/PolicyExecutionSpec.java | 283 ++++++++++++++++ .../policy/TestIcebergRewriteManifestsContent.java | 13 +- .../gravitino/policy/TestPolicyContents.java | 67 +++- .../gravitino/policy/TestPolicyExecutionSpec.java | 103 ++++++ .../json/ExecutablePolicyContentSerde.java | 370 +++++++++++++++++++++ .../java/org/apache/gravitino/json/JsonUtils.java | 49 ++- .../storage/relational/utils/TestPOConverters.java | 2 + design-docs/table-maintenance-service-design.md | 92 +++-- 16 files changed, 1355 insertions(+), 87 deletions(-) diff --git a/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java b/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java new file mode 100644 index 0000000000..c24a827699 --- /dev/null +++ b/api/src/main/java/org/apache/gravitino/policy/AbstractExecutablePolicyContent.java @@ -0,0 +1,112 @@ +/* + * 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.policy; + +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Base for built-in Iceberg maintenance policy contents that carry optional execution peers. + * + * <p>Peers are supplied at construction as an immutable {@link PolicyExecutionSpec}. Subclasses + * keep strategy fields separate and pass the spec through {@link + * #AbstractExecutablePolicyContent(PolicyExecutionSpec)}. Wire and persistence JSON keep peer keys + * flat; mapping to and from those keys is handled in {@code common} (custom Jackson serdes), not by + * flattening the five fields onto this type. + */ +public abstract class AbstractExecutablePolicyContent implements ExecutablePolicyContent { + + private final PolicyExecutionSpec executionSpec; + + /** + * Creates content with the given execution peers. + * + * @param executionSpec immutable peers; {@code null} is treated as {@link + * PolicyExecutionSpec#empty()} + */ + protected AbstractExecutablePolicyContent(@Nullable PolicyExecutionSpec executionSpec) { + this.executionSpec = executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec; + } + + /** Creates content with unset execution peers. */ + protected AbstractExecutablePolicyContent() { + this(PolicyExecutionSpec.empty()); + } + + @Override + public PolicyExecutionSpec executionSpec() { + return executionSpec; + } + + /** + * Validates universal {@link PolicyContent} rules plus shared execution-peer checks. + * + * <p>Does not enforce type-specific maintenance contracts (for example orphan cleanup forbidding + * {@code commitTrigger}, or cron parsing). Those belong on concrete subclasses or the server + * create/alter path with the real {@link Policy.BuiltInType}. + */ + @Override + public void validate() throws IllegalArgumentException { + ExecutablePolicyContent.super.validate(); + executionSpec().validate(); + } + + /** + * Whether execution fields match. + * + * @param that other content + * @return true when execution fields are equal + */ + protected boolean executionFieldsEqual(AbstractExecutablePolicyContent that) { + return Objects.equals(executionSpec(), that.executionSpec()); + } + + /** + * @return hash of execution fields + */ + protected int executionFieldsHashCode() { + return Objects.hashCode(executionSpec()); + } + + /** + * Appends execution fields to a subclass {@code toString} body. + * + * <p>Option map <em>values</em> are never rendered (keys only) so accidental credentials are not + * exposed through logs or exception messages. + * + * <p>Callers that already wrote subclass fields should append {@code ", "} before invoking this + * method. + * + * @param sb builder already containing an opening brace and any subclass fields + */ + protected void appendExecutionFields(StringBuilder sb) { + PolicyExecutionSpec spec = executionSpec(); + sb.append("timeTrigger='") + .append(spec.timeTrigger()) + .append('\'') + .append(", commitTrigger=") + .append(spec.commitTrigger()) + .append(", minIntervalMs=") + .append(spec.minIntervalMs()) + .append(", jobOptionsKeys=") + .append(spec.jobOptions().keySet()) + .append(", updaterOptionsKeys=") + .append(spec.updaterOptions().keySet()); + } +} diff --git a/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java b/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java new file mode 100644 index 0000000000..7a6fa84296 --- /dev/null +++ b/api/src/main/java/org/apache/gravitino/policy/ExecutablePolicyContent.java @@ -0,0 +1,35 @@ +/* + * 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.policy; + +/** + * Policy content that carries immutable execution peers for automated maintenance. + * + * <p>Row-filter, column-mask, and custom contents are not executable in this sense. Callers that + * need peers should use {@code instanceof ExecutablePolicyContent} (or a concrete maintenance + * content type) and read {@link #executionSpec()} rather than assuming every {@link PolicyContent} + * exposes triggers or option bags. + */ +public interface ExecutablePolicyContent extends PolicyContent { + + /** + * @return immutable execution peers; never {@code null} + */ + PolicyExecutionSpec executionSpec(); +} diff --git a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java index 011aec1ba1..902700bb62 100644 --- a/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/IcebergDataCompactionContent.java @@ -33,7 +33,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.MetadataObject; /** Built-in policy content for Iceberg compaction strategy. */ -public class IcebergDataCompactionContent implements PolicyContent { +public class IcebergDataCompactionContent extends AbstractExecutablePolicyContent { /** Property key for strategy type. */ public static final String STRATEGY_TYPE_KEY = "strategy.type"; /** Strategy type value for iceberg data compaction. */ @@ -162,7 +162,7 @@ public class IcebergDataCompactionContent implements PolicyContent { /** Default constructor for Jackson deserialization only. */ private IcebergDataCompactionContent() { - this(null, null, null, null, null, null, null, null); + this(null, null, null, null, null, null, null, null, PolicyExecutionSpec.empty()); } IcebergDataCompactionContent( @@ -174,6 +174,29 @@ public class IcebergDataCompactionContent implements PolicyContent { String rewriteStrategy, String sortOrder, Map<String, String> rewriteOptions) { + this( + minDataFileMse, + minDeleteFileNumber, + dataFileMseWeight, + deleteFileNumberWeight, + maxPartitionNum, + rewriteStrategy, + sortOrder, + rewriteOptions, + PolicyExecutionSpec.empty()); + } + + IcebergDataCompactionContent( + Long minDataFileMse, + Long minDeleteFileNumber, + Long dataFileMseWeight, + Long deleteFileNumberWeight, + Long maxPartitionNum, + String rewriteStrategy, + String sortOrder, + Map<String, String> rewriteOptions, + PolicyExecutionSpec executionSpec) { + super(executionSpec); // Nullable inputs are treated as "use default" to simplify policy creation. this.minDataFileMse = minDataFileMse == null ? DEFAULT_MIN_DATA_FILE_MSE : minDataFileMse; this.minDeleteFileNumber = @@ -299,7 +322,7 @@ public class IcebergDataCompactionContent implements PolicyContent { @Override public void validate() throws IllegalArgumentException { - PolicyContent.super.validate(); + super.validate(); // All fields are defaulted in the constructor, so only range checks are needed here. Preconditions.checkArgument(minDataFileMse >= 0, "minDataFileMse must be >= 0"); Preconditions.checkArgument(minDeleteFileNumber >= 0, "minDeleteFileNumber must be >= 0"); @@ -333,7 +356,8 @@ public class IcebergDataCompactionContent implements PolicyContent { return false; } IcebergDataCompactionContent that = (IcebergDataCompactionContent) o; - return Objects.equals(minDataFileMse, that.minDataFileMse) + return executionFieldsEqual(that) + && Objects.equals(minDataFileMse, that.minDataFileMse) && Objects.equals(minDeleteFileNumber, that.minDeleteFileNumber) && Objects.equals(dataFileMseWeight, that.dataFileMseWeight) && Objects.equals(deleteFileNumberWeight, that.deleteFileNumberWeight) @@ -346,6 +370,7 @@ public class IcebergDataCompactionContent implements PolicyContent { @Override public int hashCode() { return Objects.hash( + executionFieldsHashCode(), minDataFileMse, minDeleteFileNumber, dataFileMseWeight, @@ -358,26 +383,31 @@ public class IcebergDataCompactionContent implements PolicyContent { @Override public String toString() { - return "IcebergDataCompactionContent{" - + "minDataFileMse=" - + minDataFileMse - + ", minDeleteFileNumber=" - + minDeleteFileNumber - + ", dataFileMseWeight=" - + dataFileMseWeight - + ", deleteFileNumberWeight=" - + deleteFileNumberWeight - + ", maxPartitionNum=" - + maxPartitionNum - + ", rewriteStrategy='" - + rewriteStrategy - + '\'' - + ", sortOrder='" - + sortOrder - + '\'' - + ", rewriteOptions=" - + rewriteOptions - + '}'; + StringBuilder sb = + new StringBuilder("IcebergDataCompactionContent{") + .append("minDataFileMse=") + .append(minDataFileMse) + .append(", minDeleteFileNumber=") + .append(minDeleteFileNumber) + .append(", dataFileMseWeight=") + .append(dataFileMseWeight) + .append(", deleteFileNumberWeight=") + .append(deleteFileNumberWeight) + .append(", maxPartitionNum=") + .append(maxPartitionNum) + .append(", rewriteStrategy='") + .append(rewriteStrategy) + .append('\'') + .append(", sortOrder='") + .append(sortOrder) + .append('\'') + // Never render rewriteOptions values; same redaction as jobOptions / updaterOptions. + .append(", rewriteOptionsKeys=") + .append(rewriteOptions.keySet()) + .append(", "); + appendExecutionFields(sb); + sb.append('}'); + return sb.toString(); } private static void validateRewriteOption(String key, String value) { diff --git a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java index d10f6e2071..ba184fd115 100644 --- a/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/IcebergOrphanFileRemovalContent.java @@ -31,7 +31,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.MetadataObject; /** Configuration for table-level Iceberg orphan cleanup on an optimizer invocation. */ -public class IcebergOrphanFileRemovalContent implements PolicyContent { +public class IcebergOrphanFileRemovalContent extends AbstractExecutablePolicyContent { /** Property key for strategy type. */ public static final String STRATEGY_TYPE_KEY = "strategy.type"; /** Property key for the job template. */ @@ -64,10 +64,19 @@ public class IcebergOrphanFileRemovalContent implements PolicyContent { private final boolean dryRun; private IcebergOrphanFileRemovalContent() { - this(DEFAULT_OLDER_THAN_DAYS, null, DEFAULT_DRY_RUN); + this(DEFAULT_OLDER_THAN_DAYS, null, DEFAULT_DRY_RUN, PolicyExecutionSpec.empty()); } IcebergOrphanFileRemovalContent(long olderThanDays, @Nullable String location, boolean dryRun) { + this(olderThanDays, location, dryRun, PolicyExecutionSpec.empty()); + } + + IcebergOrphanFileRemovalContent( + long olderThanDays, + @Nullable String location, + boolean dryRun, + @Nullable PolicyExecutionSpec executionSpec) { + super(executionSpec); this.olderThanDays = olderThanDays; this.location = location; this.dryRun = dryRun; @@ -136,7 +145,10 @@ public class IcebergOrphanFileRemovalContent implements PolicyContent { @Override public void validate() { - PolicyContent.super.validate(); + super.validate(); + Preconditions.checkArgument( + !Boolean.TRUE.equals(executionSpec().commitTrigger()), + "commitTrigger is not allowed for orphan file removal"); validateOlderThanDays(olderThanDays); Preconditions.checkArgument( location == null || StringUtils.isNotBlank(location), "location must not be blank"); @@ -164,13 +176,14 @@ public class IcebergOrphanFileRemovalContent implements PolicyContent { return false; } IcebergOrphanFileRemovalContent that = (IcebergOrphanFileRemovalContent) other; - return olderThanDays == that.olderThanDays + return executionFieldsEqual(that) + && olderThanDays == that.olderThanDays && dryRun == that.dryRun && Objects.equals(location, that.location); } @Override public int hashCode() { - return Objects.hash(olderThanDays, location, dryRun); + return Objects.hash(executionFieldsHashCode(), olderThanDays, location, dryRun); } } diff --git a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java index 5a00e4e490..395b94007e 100644 --- a/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/IcebergRewriteManifestsContent.java @@ -30,7 +30,7 @@ import javax.annotation.Nullable; import org.apache.gravitino.MetadataObject; /** Typed thresholds and target options for Iceberg manifest rewriting. */ -public class IcebergRewriteManifestsContent implements PolicyContent { +public class IcebergRewriteManifestsContent extends AbstractExecutablePolicyContent { /** Property key for strategy type. */ public static final String STRATEGY_TYPE_KEY = "strategy.type"; /** Property key for the job template. */ @@ -69,7 +69,7 @@ public class IcebergRewriteManifestsContent implements PolicyContent { /** Creates default thresholds targeting Iceberg's current default partition spec. */ public IcebergRewriteManifestsContent() { - this(null, null, null, null, null); + this(null, null, null, null, null, PolicyExecutionSpec.empty()); } IcebergRewriteManifestsContent( @@ -78,6 +78,23 @@ public class IcebergRewriteManifestsContent implements PolicyContent { @Nullable Long avgManifestSizeThresholdBytes, @Nullable Integer specId, @Nullable Boolean useCaching) { + this( + manifestCountCritical, + manifestCountWarning, + avgManifestSizeThresholdBytes, + specId, + useCaching, + PolicyExecutionSpec.empty()); + } + + IcebergRewriteManifestsContent( + @Nullable Long manifestCountCritical, + @Nullable Long manifestCountWarning, + @Nullable Long avgManifestSizeThresholdBytes, + @Nullable Integer specId, + @Nullable Boolean useCaching, + @Nullable PolicyExecutionSpec executionSpec) { + super(executionSpec); this.manifestCountCritical = manifestCountCritical == null ? DEFAULT_MANIFEST_COUNT_CRITICAL : manifestCountCritical; this.manifestCountWarning = @@ -167,7 +184,7 @@ public class IcebergRewriteManifestsContent implements PolicyContent { @Override public void validate() { - PolicyContent.super.validate(); + super.validate(); Preconditions.checkArgument(manifestCountWarning > 0, "manifest_count_warning must be > 0"); Preconditions.checkArgument( manifestCountCritical >= manifestCountWarning, @@ -183,7 +200,8 @@ public class IcebergRewriteManifestsContent implements PolicyContent { return false; } IcebergRewriteManifestsContent that = (IcebergRewriteManifestsContent) other; - return Objects.equals(manifestCountCritical, that.manifestCountCritical) + return executionFieldsEqual(that) + && Objects.equals(manifestCountCritical, that.manifestCountCritical) && Objects.equals(manifestCountWarning, that.manifestCountWarning) && Objects.equals(avgManifestSizeThresholdBytes, that.avgManifestSizeThresholdBytes) && Objects.equals(specId, that.specId) @@ -193,10 +211,30 @@ public class IcebergRewriteManifestsContent implements PolicyContent { @Override public int hashCode() { return Objects.hash( + executionFieldsHashCode(), manifestCountCritical, manifestCountWarning, avgManifestSizeThresholdBytes, specId, useCaching); } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("IcebergRewriteManifestsContent{"); + sb.append("manifestCountCritical=") + .append(manifestCountCritical) + .append(", manifestCountWarning=") + .append(manifestCountWarning) + .append(", avgManifestSizeThresholdBytes=") + .append(avgManifestSizeThresholdBytes) + .append(", specId=") + .append(specId) + .append(", useCaching=") + .append(useCaching) + .append(", "); + appendExecutionFields(sb); + sb.append('}'); + return sb.toString(); + } } diff --git a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java index 6c6a62f2df..a655b93af1 100644 --- a/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/IcebergSnapshotExpirationContent.java @@ -36,7 +36,7 @@ import org.apache.gravitino.MetadataObject; * from {@link #olderThanDays()} by the job adapter; {@code retain_last} and {@code stream_results} * are forwarded). */ -public class IcebergSnapshotExpirationContent implements PolicyContent { +public class IcebergSnapshotExpirationContent extends AbstractExecutablePolicyContent { /** Property key for strategy type. */ public static final String STRATEGY_TYPE_KEY = "strategy.type"; /** Property key for the job template. */ diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java b/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java index 03f6168679..dfe48d9e50 100644 --- a/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java +++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContent.java @@ -24,7 +24,13 @@ import java.util.Set; import org.apache.commons.collections4.CollectionUtils; import org.apache.gravitino.MetadataObject; -/** The interface of the content of the policy. */ +/** + * The interface of the content of the policy. + * + * <p>Execution peers ({@code timeTrigger}, {@code commitTrigger}, {@code minIntervalMs}, {@code + * jobOptions}, {@code updaterOptions}) are not part of this universal interface. Built-in Iceberg + * maintenance contents carry them via {@link ExecutablePolicyContent}/{@link PolicyExecutionSpec}. + */ public interface PolicyContent { /** @@ -50,6 +56,9 @@ public interface PolicyContent { /** * Validates the policy content. * + * <p>Checks universal fields such as {@link #supportedObjectTypes()}. Maintenance execution-field + * validation lives on {@link ExecutablePolicyContent} implementations. + * * @throws IllegalArgumentException if the content is invalid. */ default void validate() throws IllegalArgumentException { diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java index 640424a64b..df994c4b0c 100644 --- a/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java +++ b/api/src/main/java/org/apache/gravitino/policy/PolicyContents.java @@ -136,6 +136,55 @@ public class PolicyContents { String rewriteStrategy, String sortOrder, Map<String, String> rewriteOptions) { + return icebergDataCompaction( + minDataFileMse, + minDeleteFileNumber, + dataFileMseWeight, + deleteFileNumberWeight, + maxPartitionNum, + rewriteStrategy, + sortOrder, + rewriteOptions, + null, + null, + null, + null, + null); + } + + /** + * Creates an iceberg compaction policy content with rewrite strategy, sort order, and execution + * peers. + * + * @param minDataFileMse minimum threshold for custom-data-file-mse + * @param minDeleteFileNumber minimum threshold for custom-delete-file-number + * @param dataFileMseWeight weight used for custom-data-file-mse score contribution + * @param deleteFileNumberWeight weight used for custom-delete-file-number score contribution + * @param maxPartitionNum maximum partition number selected for compaction + * @param rewriteStrategy Iceberg rewrite_data_files top-level strategy + * @param sortOrder Iceberg rewrite_data_files top-level sort_order + * @param rewriteOptions rewrite options forwarded as job.options.* + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return iceberg compaction policy content + */ + public static PolicyContent icebergDataCompaction( + long minDataFileMse, + long minDeleteFileNumber, + long dataFileMseWeight, + long deleteFileNumberWeight, + long maxPartitionNum, + String rewriteStrategy, + String sortOrder, + Map<String, String> rewriteOptions, + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { return new IcebergDataCompactionContent( minDataFileMse, minDeleteFileNumber, @@ -144,7 +193,9 @@ public class PolicyContents { maxPartitionNum, rewriteStrategy, sortOrder, - rewriteOptions); + rewriteOptions, + PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)); } /** @@ -167,7 +218,37 @@ public class PolicyContents { */ public static PolicyContent icebergOrphanFileRemoval( long olderThanDays, @Nullable String location, boolean dryRun) { - return new IcebergOrphanFileRemovalContent(olderThanDays, location, dryRun); + return icebergOrphanFileRemoval(olderThanDays, location, dryRun, null, null, null, null, null); + } + + /** + * Creates orphan cleanup policy content with execution peers. + * + * @param olderThanDays minimum file age in days, from 1 to 36500 inclusive + * @param location optional scan location within the table + * @param dryRun whether to list candidates without deleting them + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag (must not be {@code true}) + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return orphan cleanup policy content + */ + public static PolicyContent icebergOrphanFileRemoval( + long olderThanDays, + @Nullable String location, + boolean dryRun, + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + return new IcebergOrphanFileRemovalContent( + olderThanDays, + location, + dryRun, + PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)); } /** @@ -176,8 +257,9 @@ public class PolicyContents { * * @return manifest rewrite policy */ - public static PolicyContent icebergRewriteManifests() { - return icebergRewriteManifests(null, null, null, null, null); + public static IcebergRewriteManifestsContent icebergRewriteManifests() { + return new IcebergRewriteManifestsContent( + null, null, null, null, null, PolicyExecutionSpec.empty()); } /** @@ -197,12 +279,76 @@ public class PolicyContents { @Nullable Long avgManifestSizeThresholdBytes, @Nullable Integer specId, @Nullable Boolean useCaching) { + return icebergRewriteManifests( + manifestCountCritical, + manifestCountWarning, + avgManifestSizeThresholdBytes, + specId, + useCaching, + null, + null, + null, + null, + null); + } + + /** + * Creates Iceberg rewrite-manifests policy content with default thresholds and the given + * execution peers. + * + * <p>Prefer this over a peers-only five-arg overload so it does not clash with the thresholds + * factory {@code (Long, Long, Long, Integer, Boolean)}. + * + * @param executionSpec execution peers, or null for empty peers + * @return rewrite-manifests policy content + */ + public static IcebergRewriteManifestsContent icebergRewriteManifests( + @Nullable PolicyExecutionSpec executionSpec) { + return new IcebergRewriteManifestsContent( + null, + null, + null, + null, + null, + executionSpec == null ? PolicyExecutionSpec.empty() : executionSpec); + } + + /** + * Creates a manifest rewrite policy with thresholds and execution peers. Null thresholds use the + * built-in defaults. + * + * @param manifestCountCritical count that triggers regardless of size + * @param manifestCountWarning minimum count for the size trigger + * @param avgManifestSizeThresholdBytes exclusive average size threshold in bytes + * @param specId requested existing spec, or null to resolve Iceberg's current default at + * evaluation + * @param useCaching caching option, or null for the Iceberg default + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return manifest rewrite policy + */ + public static IcebergRewriteManifestsContent icebergRewriteManifests( + @Nullable Long manifestCountCritical, + @Nullable Long manifestCountWarning, + @Nullable Long avgManifestSizeThresholdBytes, + @Nullable Integer specId, + @Nullable Boolean useCaching, + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { return new IcebergRewriteManifestsContent( manifestCountCritical, manifestCountWarning, avgManifestSizeThresholdBytes, specId, - useCaching); + useCaching, + PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)); } /** diff --git a/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java new file mode 100644 index 0000000000..24ab7ba572 --- /dev/null +++ b/api/src/main/java/org/apache/gravitino/policy/PolicyExecutionSpec.java @@ -0,0 +1,283 @@ +/* + * 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.policy; + +import com.google.common.base.Preconditions; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Objects; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; + +/** + * Immutable execution peer fields for executable policy contents: triggers, cooldown, and option + * bags. + * + * <p>This type is Jackson-independent in-memory composition for {@link ExecutablePolicyContent}. + * Wire and persistence JSON keep the flat peer keys ({@code timeTrigger}, {@code commitTrigger}, + * {@code minIntervalMs}, {@code jobOptions}, {@code updaterOptions}); {@code common} maps those + * keys to and from this value object via custom Jackson serdes (same idea as {@code + * DistributionSerializer}). + * + * <p>Invoked from {@link AbstractExecutablePolicyContent}, not from universal {@link + * PolicyContent#validate()}. Type-specific rules (for example rejecting {@code commitTrigger} for + * orphan cleanup) belong on the concrete content or the server create/alter path. + * + * <p>{@code timeTrigger} cron syntax is <strong>not</strong> validated here. The scheduler + * registration path (db-scheduler / runtime cron parser) is the source of truth; validating with a + * regex or a second library in {@code api} would drift from that runtime and pull scheduler + * dependencies into the public API module. + */ +public final class PolicyExecutionSpec { + + /** Spec with every peer unset. */ + public static final PolicyExecutionSpec EMPTY = + new PolicyExecutionSpec(null, null, null, null, null); + + @Nullable private final String timeTrigger; + @Nullable private final Boolean commitTrigger; + @Nullable private final Long minIntervalMs; + private final Map<String, String> jobOptions; + private final Map<String, String> updaterOptions; + + private PolicyExecutionSpec( + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + this.timeTrigger = StringUtils.isBlank(timeTrigger) ? null : timeTrigger; + this.commitTrigger = commitTrigger; + this.minIntervalMs = minIntervalMs; + this.jobOptions = copyOptions(jobOptions); + this.updaterOptions = copyOptions(updaterOptions); + } + + /** + * Creates an immutable execution spec. + * + * @param timeTrigger optional cron expression + * @param commitTrigger optional commit-trigger flag; any non-null value (including {@code false}) + * is retained + * @param minIntervalMs optional cooldown ms + * @param jobOptions optional job options + * @param updaterOptions optional updater options + * @return immutable spec, or {@link #EMPTY} when every peer is unset + */ + public static PolicyExecutionSpec of( + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + if (!hasAny(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions)) { + return EMPTY; + } + return new PolicyExecutionSpec( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @return empty spec + */ + public static PolicyExecutionSpec empty() { + return EMPTY; + } + + /** + * @return cron expression, or {@code null} when unset + */ + @Nullable + public String timeTrigger() { + return timeTrigger; + } + + /** + * @return commit-trigger flag when explicitly supplied (including {@code false}), or {@code null} + * when unset + */ + @Nullable + public Boolean commitTrigger() { + return commitTrigger; + } + + /** + * @return minimum interval in milliseconds, or {@code null} when unset + */ + @Nullable + public Long minIntervalMs() { + return minIntervalMs; + } + + /** + * @return job options; never {@code null} + */ + public Map<String, String> jobOptions() { + return jobOptions; + } + + /** + * @return updater options; never {@code null} + */ + public Map<String, String> updaterOptions() { + return updaterOptions; + } + + /** + * @return {@code true} when any peer is set + */ + public boolean hasAny() { + return hasAny(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @param timeTrigger replacement cron expression + * @return new spec with updated {@code timeTrigger} + */ + public PolicyExecutionSpec withTimeTrigger(@Nullable String timeTrigger) { + return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @param commitTrigger replacement commit-trigger flag + * @return new spec with updated {@code commitTrigger} + */ + public PolicyExecutionSpec withCommitTrigger(@Nullable Boolean commitTrigger) { + return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @param minIntervalMs replacement cooldown + * @return new spec with updated {@code minIntervalMs} + */ + public PolicyExecutionSpec withMinIntervalMs(@Nullable Long minIntervalMs) { + return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @param jobOptions replacement job options + * @return new spec with updated {@code jobOptions} + */ + public PolicyExecutionSpec withJobOptions(@Nullable Map<String, String> jobOptions) { + return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * @param updaterOptions replacement updater options + * @return new spec with updated {@code updaterOptions} + */ + public PolicyExecutionSpec withUpdaterOptions(@Nullable Map<String, String> updaterOptions) { + return of(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * Validates shared peer constraints (cooldown and option maps). + * + * <p>Does not parse {@code timeTrigger}; cron validity is checked when the expression is + * registered with the scheduler. + */ + public void validate() { + if (!hasAny()) { + return; + } + if (minIntervalMs != null) { + Preconditions.checkArgument(minIntervalMs >= 0, "minIntervalMs must be >= 0"); + } + validateOptionMap(jobOptions, "jobOptions"); + validateOptionMap(updaterOptions, "updaterOptions"); + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof PolicyExecutionSpec)) { + return false; + } + PolicyExecutionSpec that = (PolicyExecutionSpec) other; + return Objects.equals(timeTrigger, that.timeTrigger) + && Objects.equals(commitTrigger, that.commitTrigger) + && Objects.equals(minIntervalMs, that.minIntervalMs) + && Objects.equals(jobOptions, that.jobOptions) + && Objects.equals(updaterOptions, that.updaterOptions); + } + + @Override + public int hashCode() { + return Objects.hash(timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + @Override + public String toString() { + // Never render option map values; credentials may appear before sensitive-key rejection. + return "PolicyExecutionSpec{" + + "timeTrigger='" + + timeTrigger + + '\'' + + ", commitTrigger=" + + commitTrigger + + ", minIntervalMs=" + + minIntervalMs + + ", jobOptionsKeys=" + + jobOptions.keySet() + + ", updaterOptionsKeys=" + + updaterOptions.keySet() + + '}'; + } + + /** + * Whether any execution peer field is set. + * + * <p>Every non-null {@code commitTrigger} (including {@code false}) counts as explicitly + * supplied, so JSON round-trips and {@code equals}/{@code hashCode} do not depend on other peers. + */ + private static boolean hasAny( + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger, + @Nullable Long minIntervalMs, + @Nullable Map<String, String> jobOptions, + @Nullable Map<String, String> updaterOptions) { + return StringUtils.isNotBlank(timeTrigger) + || commitTrigger != null + || minIntervalMs != null + || (jobOptions != null && !jobOptions.isEmpty()) + || (updaterOptions != null && !updaterOptions.isEmpty()); + } + + private static Map<String, String> copyOptions(@Nullable Map<String, String> options) { + if (options == null || options.isEmpty()) { + return Collections.emptyMap(); + } + return Collections.unmodifiableMap(new LinkedHashMap<>(options)); + } + + private static void validateOptionMap(Map<String, String> options, String mapName) { + if (options == null || options.isEmpty()) { + return; + } + options.forEach( + (key, value) -> { + Preconditions.checkArgument(StringUtils.isNotBlank(key), "%s key is blank", mapName); + Preconditions.checkArgument( + StringUtils.isNotBlank(value), "%s '%s' must have non-empty value", mapName, key); + }); + } +} diff --git a/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java b/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java index bbdd801abd..cd014c0fdf 100644 --- a/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java +++ b/api/src/test/java/org/apache/gravitino/policy/TestIcebergRewriteManifestsContent.java @@ -45,10 +45,12 @@ class TestIcebergRewriteManifestsContent { void testValidationAndOptions() { for (IcebergRewriteManifestsContent content : new IcebergRewriteManifestsContent[] { - PolicyContents.icebergRewriteManifests(99L, 100L, null, null, null), - PolicyContents.icebergRewriteManifests(null, 0L, null, null, null), - PolicyContents.icebergRewriteManifests(null, null, 0L, null, null), - PolicyContents.icebergRewriteManifests(null, null, null, -1, null) + PolicyContents.icebergRewriteManifests( + 99L, 100L, (Long) null, (Integer) null, (Boolean) null), + PolicyContents.icebergRewriteManifests( + null, 0L, (Long) null, (Integer) null, (Boolean) null), + PolicyContents.icebergRewriteManifests(null, null, 0L, (Integer) null, (Boolean) null), + PolicyContents.icebergRewriteManifests(null, null, (Long) null, -1, (Boolean) null) }) { Assertions.assertThrows(IllegalArgumentException.class, content::validate); } @@ -59,7 +61,8 @@ class TestIcebergRewriteManifestsContent { Assertions.assertEquals(false, content.rules().get("use_caching")); Assertions.assertNotEquals(content, PolicyContents.icebergRewriteManifests()); IcebergRewriteManifestsContent largestSpec = - PolicyContents.icebergRewriteManifests(null, null, null, Integer.MAX_VALUE, null); + PolicyContents.icebergRewriteManifests( + null, null, (Long) null, Integer.MAX_VALUE, (Boolean) null); largestSpec.validate(); Assertions.assertEquals(Integer.MAX_VALUE, largestSpec.specId()); } diff --git a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java index e378f4c7fd..b03541c567 100644 --- a/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java +++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyContents.java @@ -19,6 +19,7 @@ package org.apache.gravitino.policy; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import java.util.Arrays; import java.util.Collections; @@ -176,7 +177,71 @@ public class TestPolicyContents { IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class, content::validate); - Assertions.assertTrue(exception.getMessage().contains("maxPartitionNum")); + Assertions.assertTrue(exception.g @Test + void testValidateRejectsNegativeMinIntervalMsOnExecutableContent() { + PolicyContent content = + PolicyContents.icebergDataCompaction( + IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE, + IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER, + IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT, + IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT, + IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM, + IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY, + IcebergDataCompactionContent.DEFAULT_SORT_ORDER, + null, + null, + null, + -1L, + null, + null); + + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, content::validate); + Assertions.assertTrue(exception.getMessage().contains("minIntervalMs")); + } + + @Test + void testExecutionPeersOnlyOnExecutablePolicyContent() { + Assertions.assertInstanceOf( + ExecutablePolicyContent.class, PolicyContents.icebergDataCompaction()); + Assertions.assertInstanceOf( + ExecutablePolicyContent.class, PolicyContents.icebergOrphanFileRemoval()); + Assertions.assertFalse( + PolicyContents.custom( + ImmutableMap.of("rule", "v"), ImmutableSet.of(MetadataObject.Type.TABLE), null) + instanceof ExecutablePolicyContent); + } + + @Test + void testValidateAcceptsBuiltInContentsWithoutExecutionPeers() { + Assertions.assertDoesNotThrow(() -> PolicyContents.icebergDataCompaction().validate()); + Assertions.assertDoesNotThrow(() -> PolicyContents.icebergOrphanFileRemoval().validate()); + Assertions.assertDoesNotThrow( + () -> + PolicyContents.custom( + ImmutableMap.of("rule", "v"), ImmutableSet.of(MetadataObject.Type.TABLE), null) + .validate()); + } + + @Test + void testOrphanFileRemovalRejectsCommitTrigger() { + PolicyContent content = + PolicyContents.icebergOrphanFileRemoval( + IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS, + null, + IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN, + null, + true, + null, + null, + null); + + IllegalArgumentException exception = + Assertions.assertThrows(IllegalArgumentException.class, content::validate); + Assertions.assertTrue(exception.getMessage().contains("commitTrigger")); + } + +etMessage().contains("maxPartitionNum")); } @Test diff --git a/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java new file mode 100644 index 0000000000..cf36a06d66 --- /dev/null +++ b/api/src/test/java/org/apache/gravitino/policy/TestPolicyExecutionSpec.java @@ -0,0 +1,103 @@ +/* + * 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.policy; + +import com.google.common.collect.ImmutableMap; +import java.util.Collections; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestPolicyExecutionSpec { + + @Test + void emptyPeersReuseSingleton() { + Assertions.assertSame( + PolicyExecutionSpec.EMPTY, PolicyExecutionSpec.of(null, null, null, null, null)); + Assertions.assertSame(PolicyExecutionSpec.EMPTY, PolicyExecutionSpec.empty()); + } + + @Test + void commitTriggerFalseIsRetainedWhenAlone() { + PolicyExecutionSpec spec = PolicyExecutionSpec.of(null, false, null, null, null); + Assertions.assertNotSame(PolicyExecutionSpec.EMPTY, spec); + Assertions.assertEquals(Boolean.FALSE, spec.commitTrigger()); + Assertions.assertTrue(spec.hasAny()); + } + + @Test + void toStringOmitsOptionValues() { + PolicyExecutionSpec spec = + PolicyExecutionSpec.of( + "0 0 2 * * ?", + true, + 1000L, + ImmutableMap.of("password", "secret"), + ImmutableMap.of("oauth_credential", "id:secret")); + + String text = spec.toString(); + Assertions.assertTrue(text.contains("jobOptionsKeys=[password]")); + Assertions.assertTrue(text.contains("updaterOptionsKeys=[oauth_credential]")); + Assertions.assertFalse(text.contains("secret")); + Assertions.assertFalse(text.contains("id:secret")); + } + + @Test + void validateAcceptsScalarTriggers() { + PolicyExecutionSpec spec = + PolicyExecutionSpec.of( + "0 0 2 * * ?", + true, + 100L, + ImmutableMap.of("uri", "http://irc";), + ImmutableMap.of("metalake", "ml")); + Assertions.assertDoesNotThrow(spec::validate); + } + + @Test + void validateAcceptsCommitTriggerTrueAlone() { + PolicyExecutionSpec spec = + PolicyExecutionSpec.of(null, true, null, Collections.emptyMap(), Collections.emptyMap()); + Assertions.assertDoesNotThrow(spec::validate); + } + + @Test + void doesNotParseTimeTriggerCron() { + // Cron syntax is owned by the scheduler registration path, not this API type. + PolicyExecutionSpec spec = + PolicyExecutionSpec.of( + "0 2 * *", null, null, Collections.emptyMap(), Collections.emptyMap()); + Assertions.assertDoesNotThrow(spec::validate); + } + + @Test + void blankTimeTriggerTreatedAsUnset() { + Assertions.assertSame( + PolicyExecutionSpec.EMPTY, + PolicyExecutionSpec.of(" ", null, null, Collections.emptyMap(), Collections.emptyMap())); + Assertions.assertDoesNotThrow(PolicyExecutionSpec.EMPTY::validate); + } + + @Test + void validateRejectsNegativeMinInterval() { + PolicyExecutionSpec spec = PolicyExecutionSpec.of(null, null, -1L, null, null); + IllegalArgumentException ex = + Assertions.assertThrows(IllegalArgumentException.class, spec::validate); + Assertions.assertTrue(ex.getMessage().contains("minIntervalMs")); + } +} diff --git a/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java new file mode 100644 index 0000000000..a033614e57 --- /dev/null +++ b/common/src/main/java/org/apache/gravitino/json/ExecutablePolicyContentSerde.java @@ -0,0 +1,370 @@ +/* + * 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.json; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.DeserializationContext; +import com.fasterxml.jackson.databind.JsonDeserializer; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.module.SimpleModule; +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.Map; +import javax.annotation.Nullable; +import org.apache.gravitino.policy.IcebergDataCompactionContent; +import org.apache.gravitino.policy.IcebergOrphanFileRemovalContent; +import org.apache.gravitino.policy.IcebergRewriteManifestsContent; +import org.apache.gravitino.policy.IcebergSnapshotExpirationContent; +import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.policy.PolicyExecutionSpec; + +/** + * Jackson serdes for executable policy contents and {@link PolicyExecutionSpec}. + * + * <p>Follows the same idea as {@link JsonUtils.DistributionSerializer}: keep an immutable composed + * value in memory, and map it to or from the flat wire/persistence keys ({@code timeTrigger}, + * {@code commitTrigger}, {@code minIntervalMs}, {@code jobOptions}, {@code updaterOptions}) without + * nesting an {@code executionSpec} object. + */ +public final class ExecutablePolicyContentSerde { + + private ExecutablePolicyContentSerde() {} + + /** + * Registers serdes used when persisting domain {@link org.apache.gravitino.policy.PolicyContent} + * via {@link JsonUtils#anyFieldMapper()}. + * + * @param module module to register on + */ + public static void registerDomainSerdes(SimpleModule module) { + module.addSerializer( + IcebergDataCompactionContent.class, new IcebergDataCompactionContentSerializer()); + module.addDeserializer( + IcebergDataCompactionContent.class, new IcebergDataCompactionContentDeserializer()); + module.addSerializer( + IcebergOrphanFileRemovalContent.class, new IcebergOrphanFileRemovalContentSerializer()); + module.addDeserializer( + IcebergOrphanFileRemovalContent.class, new IcebergOrphanFileRemovalContentDeserializer()); + module.addSerializer( + IcebergRewriteManifestsContent.class, new IcebergRewriteManifestsContentSerializer()); + module.addDeserializer( + IcebergRewriteManifestsContent.class, new IcebergRewriteManifestsContentDeserializer()); + module.addSerializer( + IcebergSnapshotExpirationContent.class, new IcebergSnapshotExpirationContentSerializer()); + module.addDeserializer( + IcebergSnapshotExpirationContent.class, new IcebergSnapshotExpirationContentDeserializer()); + } + + /** + * Writes execution peer fields as siblings of other content properties (no object wrapper). + * + * @param spec execution peers; {@code null} or empty writes nothing + * @param gen JSON generator already inside a content object + * @throws IOException if writing JSON fails + */ + public static void writeExecutionPeerFields(@Nullable PolicyExecutionSpec spec, JsonGenerator gen) + throws IOException { + if (spec == null || !spec.hasAny()) { + return; + } + if (spec.timeTrigger() != null) { + gen.writeStringField("timeTrigger", spec.timeTrigger()); + } + if (spec.commitTrigger() != null) { + gen.writeBooleanField("commitTrigger", spec.commitTrigger()); + } + if (spec.minIntervalMs() != null) { + gen.writeNumberField("minIntervalMs", spec.minIntervalMs()); + } + if (!spec.jobOptions().isEmpty()) { + gen.writeObjectField("jobOptions", spec.jobOptions()); + } + if (!spec.updaterOptions().isEmpty()) { + gen.writeObjectField("updaterOptions", spec.updaterOptions()); + } + } + + /** + * Reads flat execution peer keys from a content JSON object. + * + * @param node content object node + * @return immutable spec (possibly {@link PolicyExecutionSpec#empty()}) + */ + public static PolicyExecutionSpec readExecutionSpec(JsonNode node) { + Preconditions.checkArgument( + node != null && node.isObject(), + "Cannot parse execution peers from invalid JSON: %s", + node); + String timeTrigger = textOrNull(node, "timeTrigger"); + Boolean commitTrigger = booleanOrNull(node, "commitTrigger"); + Long minIntervalMs = longOrNull(node, "minIntervalMs"); + Map<String, String> jobOptions = stringMapOrNull(node, "jobOptions"); + Map<String, String> updaterOptions = stringMapOrNull(node, "updaterOptions"); + return PolicyExecutionSpec.of( + timeTrigger, commitTrigger, minIntervalMs, jobOptions, updaterOptions); + } + + /** + * Serializer for {@link PolicyExecutionSpec} as a JSON object with the flat peer property names. + * + * <p>Same role as {@link JsonUtils.DistributionSerializer}: owns the wire shape of the composed + * value. Domain content serdes call {@link #writeExecutionPeerFields} so peers stay un-nested on + * the content object. + */ + public static class PolicyExecutionSpecSerializer extends JsonSerializer<PolicyExecutionSpec> { + @Override + public void serialize( + PolicyExecutionSpec value, JsonGenerator gen, SerializerProvider serializers) + throws IOException { + gen.writeStartObject(); + writeExecutionPeerFields(value, gen); + gen.writeEndObject(); + } + } + + /** + * Deserializer for {@link PolicyExecutionSpec} from a JSON object with the flat peer property + * names. + */ + public static class PolicyExecutionSpecDeserializer + extends JsonDeserializer<PolicyExecutionSpec> { + @Override + public PolicyExecutionSpec deserialize(JsonParser p, DeserializationContext ctxt) + throws IOException { + JsonNode node = p.getCodec().readTree(p); + return readExecutionSpec(node); + } + } + + /** Domain persistence serde for Iceberg compaction content. */ + public static class IcebergDataCompactionContentSerializer + extends JsonSerializer<IcebergDataCompactionContent> { + @Override + public void serialize( + IcebergDataCompactionContent value, JsonGenerator gen, SerializerProvider serializers) + throws IOException { + gen.writeStartObject(); + gen.writeNumberField("minDataFileMse", value.minDataFileMse()); + gen.writeNumberField("minDeleteFileNumber", value.minDeleteFileNumber()); + gen.writeNumberField("dataFileMseWeight", value.dataFileMseWeight()); + gen.writeNumberField("deleteFileNumberWeight", value.deleteFileNumberWeight()); + gen.writeNumberField("maxPartitionNum", value.maxPartitionNum()); + gen.writeStringField("rewriteStrategy", value.rewriteStrategy()); + gen.writeStringField("sortOrder", value.sortOrder()); + if (value.rewriteOptions() != null && !value.rewriteOptions().isEmpty()) { + gen.writeObjectField("rewriteOptions", value.rewriteOptions()); + } + writeExecutionPeerFields(value.executionSpec(), gen); + gen.writeEndObject(); + } + } + + /** Domain persistence serde for Iceberg compaction content. */ + public static class IcebergDataCompactionContentDeserializer + extends JsonDeserializer<IcebergDataCompactionContent> { + @Override + public IcebergDataCompactionContent deserialize(JsonParser p, DeserializationContext ctxt) + throws IOException { + JsonNode node = p.getCodec().readTree(p); + Preconditions.checkArgument( + node != null && node.isObject(), + "Cannot parse iceberg compaction content from invalid JSON: %s", + node); + return (IcebergDataCompactionContent) + PolicyContents.icebergDataCompaction( + longOrNull(node, "minDataFileMse"), + longOrNull(node, "minDeleteFileNumber"), + longOrNull(node, "dataFileMseWeight"), + longOrNull(node, "deleteFileNumberWeight"), + longOrNull(node, "maxPartitionNum"), + textOrNull(node, "rewriteStrategy"), + textOrNull(node, "sortOrder"), + stringMapOrNull(node, "rewriteOptions"), + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); + } + } + + /** Domain persistence serde for orphan-file removal content. */ + public static class IcebergOrphanFileRemovalContentSerializer + extends JsonSerializer<IcebergOrphanFileRemovalContent> { + @Override + public void serialize( + IcebergOrphanFileRemovalContent value, JsonGenerator gen, SerializerProvider serializers) + throws IOException { + gen.writeStartObject(); + gen.writeNumberField("olderThanDays", value.olderThanDays()); + if (value.location() != null) { + gen.writeStringField("location", value.location()); + } + gen.writeBooleanField("dryRun", value.dryRun()); + writeExecutionPeerFields(value.executionSpec(), gen); + gen.writeEndObject(); + } + } + + /** Domain persistence serde for orphan-file removal content. */ + public static class IcebergOrphanFileRemovalContentDeserializer + extends JsonDeserializer<IcebergOrphanFileRemovalContent> { + @Override + public IcebergOrphanFileRemovalContent deserialize(JsonParser p, DeserializationContext ctxt) + throws IOException { + JsonNode node = p.getCodec().readTree(p); + Preconditions.checkArgument( + node != null && node.isObject(), + "Cannot parse orphan file removal content from invalid JSON: %s", + node); + Long olderThanDaysNode = longOrNull(node, "olderThanDays"); + long olderThanDays = + olderThanDaysNode == null + ? IcebergOrphanFileRemovalContent.DEFAULT_OLDER_THAN_DAYS + : olderThanDaysNode; + Boolean dryRunNode = booleanOrNull(node, "dryRun"); + boolean dryRun = + dryRunNode == null ? IcebergOrphanFileRemovalContent.DEFAULT_DRY_RUN : dryRunNode; + return (IcebergOrphanFileRemovalContent) + PolicyContents.icebergOrphanFileRemoval( + olderThanDays, + textOrNull(node, "location"), + dryRun, + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); + } + } + + /** Domain persistence serde for rewrite-manifests content. */ + public static class IcebergRewriteManifestsContentSerializer + extends JsonSerializer<IcebergRewriteManifestsContent> { + @Override + public void serialize( + IcebergRewriteManifestsContent value, JsonGenerator gen, SerializerProvider serializers) + throws IOException { + gen.writeStartObject(); + writeExecutionPeerFields(value.executionSpec(), gen); + gen.writeEndObject(); + } + } + + /** Domain persistence serde for rewrite-manifests content. */ + public static class IcebergRewriteManifestsContentDeserializer + extends JsonDeserializer<IcebergRewriteManifestsContent> { + @Override + public IcebergRewriteManifestsContent deserialize(JsonParser p, DeserializationContext ctxt) + throws IOException { + JsonNode node = p.getCodec().readTree(p); + Preconditions.checkArgument( + node != null && node.isObject(), + "Cannot parse rewrite-manifests content from invalid JSON: %s", + node); + return (IcebergRewriteManifestsContent) + PolicyContents.icebergRewriteManifests( + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); + } + } + + /** Domain persistence serde for snapshot-expiration content. */ + public static class IcebergSnapshotExpirationContentSerializer + extends JsonSerializer<IcebergSnapshotExpirationContent> { + @Override + public void serialize( + IcebergSnapshotExpirationContent value, JsonGenerator gen, SerializerProvider serializers) + throws IOException { + gen.writeStartObject(); + writeExecutionPeerFields(value.executionSpec(), gen); + gen.writeEndObject(); + } + } + + /** Domain persistence serde for snapshot-expiration content. */ + public static class IcebergSnapshotExpirationContentDeserializer + extends JsonDeserializer<IcebergSnapshotExpirationContent> { + @Override + public IcebergSnapshotExpirationContent deserialize(JsonParser p, DeserializationContext ctxt) + throws IOException { + JsonNode node = p.getCodec().readTree(p); + Preconditions.checkArgument( + node != null && node.isObject(), + "Cannot parse snapshot-expiration content from invalid JSON: %s", + node); + return (IcebergSnapshotExpirationContent) + PolicyContents.icebergSnapshotExpiration( + textOrNull(node, "timeTrigger"), + booleanOrNull(node, "commitTrigger"), + longOrNull(node, "minIntervalMs"), + stringMapOrNull(node, "jobOptions"), + stringMapOrNull(node, "updaterOptions")); + } + } + + @Nullable + private static String textOrNull(JsonNode node, String property) { + if (!node.has(property) || node.get(property).isNull()) { + return null; + } + return node.get(property).asText(); + } + + @Nullable + private static Boolean booleanOrNull(JsonNode node, String property) { + if (!node.has(property) || node.get(property).isNull()) { + return null; + } + return node.get(property).asBoolean(); + } + + @Nullable + private static Long longOrNull(JsonNode node, String property) { + if (!node.has(property) || node.get(property).isNull()) { + return null; + } + return node.get(property).asLong(); + } + + @Nullable + private static Map<String, String> stringMapOrNull(JsonNode node, String property) { + if (!node.has(property) || node.get(property).isNull()) { + return null; + } + JsonNode mapNode = node.get(property); + Preconditions.checkArgument( + mapNode.isObject(), "Cannot parse string map property %s from: %s", property, mapNode); + Map<String, String> map = new LinkedHashMap<>(); + Iterator<Map.Entry<String, JsonNode>> fields = mapNode.fields(); + while (fields.hasNext()) { + Map.Entry<String, JsonNode> entry = fields.next(); + map.put(entry.getKey(), entry.getValue().isNull() ? null : entry.getValue().asText()); + } + return map; + } +} diff --git a/common/src/main/java/org/apache/gravitino/json/JsonUtils.java b/common/src/main/java/org/apache/gravitino/json/JsonUtils.java index e98bc7e387..4757df5e14 100644 --- a/common/src/main/java/org/apache/gravitino/json/JsonUtils.java +++ b/common/src/main/java/org/apache/gravitino/json/JsonUtils.java @@ -82,6 +82,7 @@ import org.apache.gravitino.dto.rel.partitions.IdentityPartitionDTO; import org.apache.gravitino.dto.rel.partitions.ListPartitionDTO; import org.apache.gravitino.dto.rel.partitions.PartitionDTO; import org.apache.gravitino.dto.rel.partitions.RangePartitionDTO; +import org.apache.gravitino.policy.PolicyExecutionSpec; import org.apache.gravitino.rel.Column; import org.apache.gravitino.rel.TableChange; import org.apache.gravitino.rel.expressions.Expression; @@ -312,16 +313,28 @@ public class JsonUtils { .setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY) .registerModule(new JavaTimeModule()) .registerModule(new Jdk8Module()) - .registerModule( - new SimpleModule() - .addDeserializer(Type.class, new TypeDeserializer()) - .addSerializer(Type.class, new TypeSerializer()) - .addDeserializer(Partitioning.class, new PartitioningDeserializer()) - .addSerializer(Partitioning.class, new PartitioningSerializer()) - .addDeserializer(Expression.class, new ColumnDefaultValueDeserializer()) - .addSerializer(Expression.class, new ColumnDefaultValueSerializer()) - .addDeserializer(StatisticValue.class, new StatisticValueDeserializer()) - .addSerializer(StatisticValue.class, new StatisticValueSerializer())); + .registerModule(anyFieldModule()); + } + + private static SimpleModule anyFieldModule() { + SimpleModule module = + new SimpleModule() + .addDeserializer(Type.class, new TypeDeserializer()) + .addSerializer(Type.class, new TypeSerializer()) + .addDeserializer(Partitioning.class, new PartitioningDeserializer()) + .addSerializer(Partitioning.class, new PartitioningSerializer()) + .addDeserializer(Expression.class, new ColumnDefaultValueDeserializer()) + .addSerializer(Expression.class, new ColumnDefaultValueSerializer()) + .addDeserializer(StatisticValue.class, new StatisticValueDeserializer()) + .addSerializer(StatisticValue.class, new StatisticValueSerializer()) + .addSerializer( + PolicyExecutionSpec.class, + new ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer()) + .addDeserializer( + PolicyExecutionSpec.class, + new ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer()); + ExecutablePolicyContentSerde.registerDomainSerdes(module); + return module; } /** @@ -1309,6 +1322,22 @@ public class JsonUtils { } } + /** + * Custom JSON serializer for {@link PolicyExecutionSpec}. + * + * @see ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer + */ + public static class PolicyExecutionSpecSerializer + extends ExecutablePolicyContentSerde.PolicyExecutionSpecSerializer {} + + /** + * Custom JSON deserializer for {@link PolicyExecutionSpec}. + * + * @see ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer + */ + public static class PolicyExecutionSpecDeserializer + extends ExecutablePolicyContentSerde.PolicyExecutionSpecDeserializer {} + /** Custom JSON serializer for DistributionDTO objects. */ public static class DistributionSerializer extends JsonSerializer<DistributionDTO> { @Override diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java index 759bbdc059..464485f111 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java @@ -68,9 +68,11 @@ import org.apache.gravitino.meta.TableStatisticEntity; import org.apache.gravitino.meta.TagEntity; import org.apache.gravitino.meta.TopicEntity; import org.apache.gravitino.meta.UserEntity; +import org.apache.gravitino.policy.IcebergDataCompactionContent; import org.apache.gravitino.policy.Policy; import org.apache.gravitino.policy.PolicyContent; import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.policy.PolicyExecutionSpec; import org.apache.gravitino.rel.expressions.Expression; import org.apache.gravitino.rel.expressions.NamedReference; import org.apache.gravitino.rel.expressions.distributions.Distributions; diff --git a/design-docs/table-maintenance-service-design.md b/design-docs/table-maintenance-service-design.md index d4487e8e12..5ec6be113f 100644 --- a/design-docs/table-maintenance-service-design.md +++ b/design-docs/table-maintenance-service-design.md @@ -125,6 +125,31 @@ parts: 1. **Policy** — what to maintain, when to trigger, and non-sensitive job settings. 2. **Scheduling and commit** — how work is enqueued, claimed, and recorded. +### 5.0 Terminology: policy identity vs scheduler task kind + +Do **not** overload “task type”. TMS uses two different identities: + +| Concept | Meaning | Identity today | +| ------- | ------- | -------------- | +| **Policy / activity type** | Which maintenance the policy describes (compaction, manifest-rewrite, snapshot-expiry, orphan-cleanup) | `Policy.BuiltInType` (e.g. `ICEBERG_COMPACTION`) | +| **Scheduler task kind** | Which db-scheduler row / pool owns the work | `policy-expand`, `cron-job`, `commit-job` | + +Rules: + +- Prefer **`Policy.BuiltInType`** as the single policy/activity identity. Server defaults + (for example per-type `minIntervalMs`) and content registries key off it. Do **not** + introduce a parallel enum that maps 1:1 to `BuiltInType` (the former + `TableMaintenanceTaskType` pattern). +- Keep **scheduler task kinds** separate. They are control-plane scheduling units, not + policy types. One policy may produce expand / cron / commit work without changing its + `BuiltInType`. +- If a future runtime “activity” ever diverges from policy identity (non-1:1), introduce + that concept then with an explicit contract. Until then, do not invent a second + activity layer. + +In prose, prefer “maintenance type” / `BuiltInType` for policy activity, and “scheduler +task kind” for expand / cron-job / commit-job. + TMS has two trigger paths that stay separate: - **crontab** — expand a policy into one-shot Spark submits (one type per submit). @@ -147,31 +172,37 @@ policy first, then scheduling. TMS reuses the existing metalake Policy APIs and does **not** add `/api/maintenance/table/policies`. Each policy covers **one** maintenance type. The same -policy may enable both `onCommit` and `crontab`. +policy may enable both commit and time triggers (`commitTrigger` and `timeTrigger`). -#### 5.1.1 Evaluate triggers (`onCommit` + `crontab`) +#### 5.1.1 Evaluate triggers (`commitTrigger` + `timeTrigger`) -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. The `scheduled_*_tasks` tables are runtime +Triggers are **flat** peers on policy content (same JSON object as `jobOptions` / +`updaterOptions` / `rewriteOptions` / `minIntervalMs`): `commitTrigger` (boolean) and +`timeTrigger` (cron string). Operators edit that versioned content through Policy 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: -| Trigger | Path | Which types | -| ---------- | ---------------------------------------- | ----------------------------------------------- | -| `onCommit` | IRC → **commit-job** only (skips expand) | Compaction / manifest-rewrite / snapshot-expiry | -| `crontab` | **policy-expand** → cron-job | All four types | +| Wire field | Runtime path | Which types | +| ---------------- | ----------------------------------------- | ----------------------------------------------- | +| `commitTrigger` | IRC → **commit-job** only (skips expand) | Compaction / manifest-rewrite / snapshot-expiry | +| `timeTrigger` | **policy-expand** → cron-job | All four types | -**`onCommit`** runs after a successful Iceberg commit through IRC. That fits types that +**`commitTrigger`** runs after a successful Iceberg commit through IRC. That fits types that rewrite or expire data and metadata from the commit stream. **`orphan-cleanup` does not:** it scans storage for unreferenced files on a schedule. Running it on every commit would be -too heavy. Create/alter rejects `schedule.onCommit = true` for orphan-cleanup; that type is -**crontab-only**. +too heavy. Create/alter rejects `commitTrigger = true` for orphan-cleanup; that type is +**timeTrigger-only** when automation is enabled. -**`minIntervalMs`** sits in policy content **beside** `schedule` (not inside it). It is the -cooldown for **that policy's type** after `MAX(finished_at)` on `table_maintenance_job` for -`(table, policy)`. Server conf and code defaults are lower-priority fallbacks. +**`minIntervalMs`** is the cooldown for **that policy's type** after `MAX(finished_at)` on +`table_maintenance_job` for `(table, policy)`. Server conf and code defaults are +lower-priority fallbacks. + +**`timeTrigger` cron:** the string is stored as-is on the policy API. Syntax and time-zone +rules are owned by the TMS / db-scheduler registration path (not re-validated with a second +parser in `api`). Examples below use a typical six-field Quartz-style form; align examples +with the scheduler contract when registration validation is wired. Example content: @@ -183,28 +214,27 @@ Example content: "uri": "http://host:9001/iceberg";, "type": "rest" }, - "schedule": { "onCommit": true, "crontab": "0 2 * * *" }, + "commitTrigger": true, + "timeTrigger": "0 0 2 * * ?", "minIntervalMs": 3600000 } ``` -| Field | Runtime effect | -| ------------------- | ----------------------------------------------------------------- | -| `schedule.onCommit` | IRC upserts commit-job; commit pool submits one chained Job | -| `schedule.crontab` | Sets / refreshes next `policy-expand` due time | -| `minIntervalMs` | Cooldown for this policy's type; highest when set | -| `jobOptions` | Non-sensitive Spark / IRC client props (`uri`, `type`, memory, …) | - -Automated maintenance should set at least one of `onCommit` or `crontab` (orphan-cleanup: -**`crontab` only**). +| Field | Runtime effect | +| ---------------- | ----------------------------------------------------------------- | +| `commitTrigger` | IRC upserts commit-job; commit pool submits one chained Job | +| `timeTrigger` | Sets / refreshes next `policy-expand` due time | +| `minIntervalMs` | Cooldown for this policy's type; highest when set | +| `jobOptions` | Non-sensitive Spark / IRC client props (`uri`, `type`, memory, …) | +| `updaterOptions` | Non-sensitive Gravitino updater props (`gravitino_uri`, `metalake`, …) | #### 5.1.2 Sensitive credentials Policy content is versioned, edited through Policy APIs, and readable with `VIEW_POLICY`. **Secrets must not live there.** Put credential-shaped keys in `jobOptions` / -`rewriteOptions` and create/alter **rejects** them. Non-sensitive Spark / IRC client props -(`uri`, `type`, memory, …) stay on the nearest policy; auth is cluster-static in -`gravitino.conf` (values may be SecretManager references). Conf keys: §7.4. +`rewriteOptions` / `updaterOptions` and create/alter **rejects** them. Non-sensitive Spark / +IRC client props (`uri`, `type`, memory, …) stay on the nearest policy; auth is +cluster-static in `gravitino.conf` (values may be SecretManager references). Conf keys: §7.4. TMS talks to **two** backends with **two** identities, so there are two prefixes: @@ -220,7 +250,7 @@ each — it does not write secrets back into policy content: | Runtime bag | Non-sensitive (examples) | Auth overlay | | ------------------- | ------------------------------------------------------------- | --------------------------------------- | -| **`updateOptions`** | `gravitino_uri`, `metalake`, updater impl names, … | `gravitino.maintenance.gravitinoAuth.*` | +| **`updaterOptions`** | `gravitino_uri`, `metalake`, updater impl names, … | `gravitino.maintenance.gravitinoAuth.*` | | **`jobOptions`** | nearest policy `jobOptions` (`uri`, `type`, Spark resources…) | `gravitino.maintenance.ircAuth.*` | ### 5.2 Scheduling and commit @@ -325,7 +355,7 @@ flowchart TD B --> C[Keep expand row; set next crontab] ``` -Example with crontab `0 2 * * *`: create inserts expand due next 02:00; at 02:00 expand inserts +Example with `timeTrigger` `0 0 2 * * ?`: create inserts expand due next 02:00; at 02:00 expand inserts many cron-jobs due now, then schedules itself for the next 02:00. Expand's due time is when **enqueue** happens — not when Spark runs. @@ -500,7 +530,7 @@ Cluster-static — **not** in policy content and **not** a DB table. | Prefix | Role | Merged into `runJob` bag | | -------------------------------------- | ------------------------------------------------------------ | ----------------------------------------------- | -| `gravitino.maintenance.gravitinoAuth.` | Expand / submit / RBAC client; also jobs that call Gravitino | `updateOptions` (+ non-sensitive updater props) | +| `gravitino.maintenance.gravitinoAuth.` | Expand / submit / RBAC client; also jobs that call Gravitino | `updaterOptions` (+ non-sensitive updater props) | | `gravitino.maintenance.ircAuth.` | Iceberg REST client auth for Spark Jobs → IRC | `jobOptions` (+ policy non-sensitive props) | Illustrative auth keys (same shape under each prefix; values may be SecretManager references where
