diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java index 43235d9ac51160..b1c17cec6aa786 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @@ -109,6 +109,7 @@ import org.apache.doris.datasource.iceberg.IcebergExternalTable; import org.apache.doris.datasource.iceberg.IcebergSysExternalTable; import org.apache.doris.datasource.jdbc.JdbcExternalTable; +import org.apache.doris.datasource.lance.job.LanceIndexJobManager; import org.apache.doris.datasource.paimon.PaimonExternalTable; import org.apache.doris.datasource.paimon.PaimonSysExternalTable; import org.apache.doris.deploy.DeployManager; @@ -568,6 +569,8 @@ public class Env { private InsertOverwriteManager insertOverwriteManager; + private LanceIndexJobManager lanceIndexJobManager; + private DNSCache dnsCache; private final NereidsSqlCacheManager sqlCacheManager; @@ -853,6 +856,7 @@ public Env(boolean isCheckpointCatalog) { this.mtmvService = new MTMVService(); this.eventProcessor = new EventProcessor(mtmvService); this.insertOverwriteManager = new InsertOverwriteManager(); + this.lanceIndexJobManager = new LanceIndexJobManager(); this.dnsCache = new DNSCache(); this.sqlCacheManager = new NereidsSqlCacheManager(); this.sortedPartitionsCacheManager = new NereidsSortedPartitionsCacheManager(); @@ -978,6 +982,10 @@ public InsertOverwriteManager getInsertOverwriteManager() { return insertOverwriteManager; } + public LanceIndexJobManager getLanceIndexJobManager() { + return lanceIndexJobManager; + } + public TabletScheduler getTabletScheduler() { return tabletScheduler; } @@ -1812,6 +1820,11 @@ private void transferToMaster() { insertOverwriteManager.allTaskFail(); + // A durable RUNNING Lance index job at this point may have lost its result with the + // old master: sweep it to UNKNOWN (and refresh RUNNING back to REQUIRED) before any + // master-only dispatcher could start. + lanceIndexJobManager.onTransferToMaster(); + toMasterProgress = "start daemon threads"; // coz current fe was not master fe and didn't get all fes' alive session report before, which cause @@ -2657,6 +2670,18 @@ public long saveDictionaryManager(CountingDataOutputStream out, long checksum) t return checksum; } + public long loadLanceIndexJobManager(DataInputStream in, long checksum) throws IOException { + this.lanceIndexJobManager = LanceIndexJobManager.read(in); + LOG.info("finished replay lance index job manager from image"); + return checksum; + } + + public long saveLanceIndexJobManager(CountingDataOutputStream out, long checksum) throws IOException { + this.lanceIndexJobManager.write(out); + LOG.info("finished save lance index job manager to image"); + return checksum; + } + // Only called by checkpoint thread // return the latest image file's absolute path public String saveImage() throws IOException { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexDatasetLocator.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexDatasetLocator.java new file mode 100644 index 00000000000000..bed1ae696d8aa2 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexDatasetLocator.java @@ -0,0 +1,143 @@ +// 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.doris.datasource.lance.job; + +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.Locale; + +/** + * Dataset locator normalization v1 for the durable fence key. The rules, in + * order: + * + *
The authority (host/bucket) and path keep their original case: bucket and + * path components are case-sensitive on the providers Doris supports, and + * normalization v1 deliberately does not define cross-alias equivalence. URI + * aliases and external writers replacing the dataset at the same URI are + * outside Doris serialization. + */ +public final class LanceIndexDatasetLocator { + private static final String SCHEME_SEPARATOR = "://"; + /** Finite bound for the stable locator persisted in every durable job record. */ + public static final int MAX_LOCATOR_BYTES = 4096; + + private LanceIndexDatasetLocator() { + } + + /** + * Normalize a raw dataset locator into its durable identity form. + * + * @throws IllegalArgumentException if the locator is null/empty/oversized, + * is not a valid hierarchical URI or absolute path, carries + * userinfo/query/fragment data, has neither an authority nor a path, + * or is a scheme-less relative path + */ + public static String normalize(String rawLocator) { + if (rawLocator == null) { + throw new IllegalArgumentException("dataset locator must not be null"); + } + String locator = rawLocator.trim(); + if (locator.isEmpty()) { + throw new IllegalArgumentException("dataset locator must not be empty"); + } + if (locator.getBytes(StandardCharsets.UTF_8).length > MAX_LOCATOR_BYTES) { + throw new IllegalArgumentException( + "dataset locator exceeds " + MAX_LOCATOR_BYTES + " UTF-8 bytes"); + } + + // Preserve the established spelling of the scheme-less filesystem root. + // java.net.URI rejects "//" as a network-path reference without an + // authority, while it is a valid absolute filesystem path here. + if (containsOnlySlashes(locator)) { + return "/"; + } + + URI uri; + try { + uri = new URI(locator); + } catch (URISyntaxException e) { + // Do not include the raw locator: it may contain credentials. + throw new IllegalArgumentException("dataset locator is not a valid URI or absolute path"); + } + if (uri.getRawUserInfo() != null) { + throw new IllegalArgumentException( + "credential-bearing dataset locators are never identity (userinfo is not allowed)"); + } + if (uri.getRawQuery() != null) { + // Presigned object-store URLs and SAS URLs carry credentials here. + throw new IllegalArgumentException("dataset locator query parameters are not allowed"); + } + if (uri.getRawFragment() != null) { + throw new IllegalArgumentException("dataset locator fragments are not allowed"); + } + + String scheme = uri.getScheme(); + if (scheme == null) { + if (!locator.startsWith("/")) { + throw new IllegalArgumentException("dataset locator without a scheme must be an absolute path"); + } + return stripTrailingSlashes(locator, 1); + } + if (uri.isOpaque() || !locator.regionMatches(scheme.length(), SCHEME_SEPARATOR, 0, + SCHEME_SEPARATOR.length())) { + throw new IllegalArgumentException("dataset locator scheme must use hierarchical '://' syntax"); + } + String authority = uri.getRawAuthority() == null ? "" : uri.getRawAuthority(); + String rawPath = uri.getRawPath() == null ? "" : uri.getRawPath(); + String path = stripTrailingSlashes(rawPath, 0); + if (authority.isEmpty() && path.isEmpty()) { + // "s3://" / "file://" carry no identity at all; "file:///x" (empty + // authority, non-empty path) is legal and does not reach this. + throw new IllegalArgumentException("dataset locator has neither an authority nor a path"); + } + return scheme.toLowerCase(Locale.ROOT) + SCHEME_SEPARATOR + authority + path; + } + + private static boolean containsOnlySlashes(String value) { + for (int index = 0; index < value.length(); index++) { + if (value.charAt(index) != '/') { + return false; + } + } + return true; + } + + private static String stripTrailingSlashes(String value, int minLength) { + int end = value.length(); + while (end > minLength && value.charAt(end - 1) == '/') { + end--; + } + return value.substring(0, end); + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexFenceKey.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexFenceKey.java new file mode 100644 index 00000000000000..19cc9efe80ad45 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexFenceKey.java @@ -0,0 +1,92 @@ +// 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.doris.datasource.lance.job; + +import java.util.Objects; + +/** + * The durable same-name target/fence key: + *
+ * ( persisted catalog identity, provider = DIRECTORY, + * normalized stable dataset locator, persisted normalized logical-index-name bytes ) + *+ * The display name is not part of the key; it is persisted on the job itself. + * This class is a derived in-memory index key and is not persisted directly. + */ +public final class LanceIndexFenceKey { + /** Provider of every job in this delivery slice, mapped from a filesystem (Directory) Lance catalog. */ + public static final String PROVIDER_DIRECTORY = "DIRECTORY"; + + private final long catalogId; + private final String provider; + private final String normalizedLocator; + private final String normalizedIndexName; + + public LanceIndexFenceKey(long catalogId, String provider, String normalizedLocator, String normalizedIndexName) { + this.catalogId = catalogId; + this.provider = Objects.requireNonNull(provider, "provider"); + this.normalizedLocator = Objects.requireNonNull(normalizedLocator, "normalizedLocator"); + this.normalizedIndexName = Objects.requireNonNull(normalizedIndexName, "normalizedIndexName"); + } + + public long getCatalogId() { + return catalogId; + } + + public String getProvider() { + return provider; + } + + public String getNormalizedLocator() { + return normalizedLocator; + } + + public String getNormalizedIndexName() { + return normalizedIndexName; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof LanceIndexFenceKey)) { + return false; + } + LanceIndexFenceKey that = (LanceIndexFenceKey) o; + return catalogId == that.catalogId + && provider.equals(that.provider) + && normalizedLocator.equals(that.normalizedLocator) + && normalizedIndexName.equals(that.normalizedIndexName); + } + + @Override + public int hashCode() { + return Objects.hash(catalogId, provider, normalizedLocator, normalizedIndexName); + } + + /** + * Deliberately omits the locator: fence-conflict messages may surface to + * users without target privileges and must not disclose it. + */ + @Override + public String toString() { + return "LanceIndexFenceKey{catalogId=" + catalogId + ", provider=" + provider + + ", normalizedIndexName=" + normalizedIndexName + '}'; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJob.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJob.java new file mode 100644 index 00000000000000..59386acbcdb94c --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJob.java @@ -0,0 +1,695 @@ +// 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.doris.datasource.lance.job; + +import org.apache.doris.common.io.Text; +import org.apache.doris.common.io.Writable; +import org.apache.doris.persist.gson.GsonUtils; + +import com.google.gson.annotations.SerializedName; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Objects; + +/** + * The minimal durable Lance index job record. It contains only what the + * lifecycle invariants require: job identity/creator/revision and bounded + * timestamps; the persisted target identity and normalized same-name fence + * key; the mutation intent; the admitted dataset version and schema-contract + * representation; the mutation outcome, independent refresh state, typed + * result, and a bounded sanitized message; the dispatch identity (selected BE, + * BE process epoch, immutable invocation ID, deadline), possible-live + * ownership, and any matching termination proof; and the FORCE audit fields + * (populated only by the FORCE_RELEASE slice; here they are carried for + * durability and replay). + * + *
The record never carries secrets, raw/opaque provider data, or unbounded + * values: name/message/note/properties fields are length-checked at + * construction/mutation time. Credentials are resolved at execution time from + * the current catalog and are never persisted here. + * + *
Serialization is the standard Gson stream: {@code Text.writeString} of + * the JSON form. Every durable field carries a short {@link SerializedName}; + * fields missing in old data keep the safe in-class defaults, so replay stays + * tolerant of additive evolution. An instance is effectively immutable while + * published in the manager: every durable transition is applied to a private + * copy that is logged and then swapped in, so the object written to the edit + * log is never mutated afterwards. + */ +public class LanceIndexJob implements Writable { + /** Bound on the user/build properties JSON snapshot. */ + public static final int MAX_PROPERTIES_JSON_BYTES = 4096; + /** Bound on the FORCE_RELEASE operator note and the late-commit warning text. */ + public static final int MAX_FORCE_TEXT_BYTES = 1024; + /** Bound on persisted creator, target-name, mutation, and FORCE-actor text. */ + public static final int MAX_DURABLE_TEXT_BYTES = 1024; + /** Dispatch identities are UUID-like tokens, not arbitrary worker output. */ + public static final int MAX_INVOCATION_ID_BYTES = 256; + + // ------------------------------------------------------------------ + // Identity + // ------------------------------------------------------------------ + @SerializedName(value = "jid") + private long jobId; + + @SerializedName(value = "cr") + private String creator; + + /** Bumped by +1 on every durable transition; callbacks and replay are revision-checked. */ + @SerializedName(value = "rev") + private long revision; + + @SerializedName(value = "ctm") + private long createTimeMs; + + @SerializedName(value = "utm") + private long updateTimeMs; + + // ------------------------------------------------------------------ + // Target / fence identity + // ------------------------------------------------------------------ + @SerializedName(value = "cid") + private long catalogId; + + /** Local database name, kept for privilege and FORCE resolution when the catalog still resolves. */ + @SerializedName(value = "dbn") + private String dbName; + + /** Local table name, kept for privilege and FORCE resolution when the catalog still resolves. */ + @SerializedName(value = "tbn") + private String tableName; + + @SerializedName(value = "prv") + private String provider = LanceIndexFenceKey.PROVIDER_DIRECTORY; + + @SerializedName(value = "loc") + private String normalizedLocator; + + @SerializedName(value = "din") + private String displayIndexName; + + @SerializedName(value = "nin") + private String normalizedIndexName; + + // ------------------------------------------------------------------ + // Mutation intent + // ------------------------------------------------------------------ + @SerializedName(value = "mt") + private LanceIndexJobMutationType mutationType = LanceIndexJobMutationType.CREATE; + + @SerializedName(value = "ine") + private boolean ifNotExists; + + @SerializedName(value = "ie") + private boolean ifExists; + + /** Logical Lance algorithm (IVF_PQ / BTREE / BITMAP); required for CREATE/REPLACE, nullable for DROP. */ + @SerializedName(value = "it") + private String indexType; + + @SerializedName(value = "cn") + private String columnName; + + @SerializedName(value = "pj") + private String propertiesJson; + + // ------------------------------------------------------------------ + // Admission snapshot + // ------------------------------------------------------------------ + @SerializedName(value = "adv") + private long admittedDatasetVersion; + + /** Nullable only for DROP, which does not revalidate an indexed-field contract. */ + @SerializedName(value = "sc") + private LanceIndexSchemaContract schemaContract; + + // ------------------------------------------------------------------ + // Dual state + // ------------------------------------------------------------------ + @SerializedName(value = "ms") + private LanceIndexJobMutationState mutationState = LanceIndexJobMutationState.UNKNOWN; + + /** + * Safe default for a corrupt record missing the key: REQUIRED holds the fence on a + * terminal record, mirroring the null fallback in {@link #isUnresolved()}. A legal + * record always carries the state explicitly (admission sets NOT_REQUIRED). + */ + @SerializedName(value = "rs") + private LanceIndexJobRefreshState refreshState = LanceIndexJobRefreshState.REQUIRED; + + // ------------------------------------------------------------------ + // Result + // ------------------------------------------------------------------ + @SerializedName(value = "res") + private LanceIndexJobResult result; + + // ------------------------------------------------------------------ + // Dispatch + // ------------------------------------------------------------------ + @SerializedName(value = "bid") + private Long backendId; + + @SerializedName(value = "bpe") + private Long beProcessEpoch; + + /** + * Immutable revision established by PENDING -> RUNNING. Result and + * termination-proof callbacks use this dispatch identity rather than racing + * on the record's global revision. Null only before dispatch or in old data. + */ + @SerializedName(value = "drv") + private Long dispatchRevision; + + /** Immutable per-dispatch UUID; callbacks must present the matching identity. */ + @SerializedName(value = "iid") + private String invocationId; + + /** Bounds wait/runtime only; never proves termination and never releases a possible-live slot. */ + @SerializedName(value = "dlm") + private Long deadlineMs; + + @SerializedName(value = "plo") + private boolean possibleLiveOwned; + + @SerializedName(value = "tp") + private LanceIndexTerminationProof terminationProof = LanceIndexTerminationProof.NONE; + + // ------------------------------------------------------------------ + // FORCE_RELEASE audit (populated only by the FORCE slice; durable + replayable here) + // ------------------------------------------------------------------ + @SerializedName(value = "fr") + private boolean forceReleased; + + @SerializedName(value = "fa") + private String forceActor; + + @SerializedName(value = "ftm") + private Long forceTimeMs; + + @SerializedName(value = "fn") + private String forceNote; + + @SerializedName(value = "fw") + private String forceWarning; + + /** + * No-arg constructor for Gson replay only; missing fields keep the safe + * defaults declared above (UNKNOWN mutation state holds the fence, never + * the redispatchable PENDING; REQUIRED refresh state owes a refresh rather + * than silently releasing the fence). + */ + public LanceIndexJob() { + } + + /** + * Admission constructor: identity, intent, and the admission snapshot. The + * manager initializes the lifecycle fields when the job is admitted. + */ + public LanceIndexJob(long jobId, String creator, long catalogId, String dbName, String tableName, + String provider, String normalizedLocator, String displayIndexName, String normalizedIndexName, + LanceIndexJobMutationType mutationType, boolean ifNotExists, boolean ifExists, String indexType, + String columnName, String propertiesJson, long admittedDatasetVersion, + LanceIndexSchemaContract schemaContract) { + this.jobId = jobId; + this.creator = checkRequiredBytes(creator, MAX_DURABLE_TEXT_BYTES, "creator"); + this.catalogId = catalogId; + this.dbName = checkRequiredBytes(dbName, MAX_DURABLE_TEXT_BYTES, "dbName"); + this.tableName = checkRequiredBytes(tableName, MAX_DURABLE_TEXT_BYTES, "tableName"); + this.provider = Objects.requireNonNull(provider, "provider"); + this.normalizedLocator = Objects.requireNonNull(normalizedLocator, "normalizedLocator"); + setDisplayIndexName(displayIndexName); + setNormalizedIndexName(normalizedIndexName); + this.mutationType = Objects.requireNonNull(mutationType, "mutationType"); + this.ifNotExists = ifNotExists; + this.ifExists = ifExists; + setIndexType(indexType); + setColumnName(columnName); + setPropertiesJson(propertiesJson); + this.admittedDatasetVersion = admittedDatasetVersion; + this.schemaContract = schemaContract; + validateForAdmission(); + } + + /** + * Copy used by the manager to stage a durable transition: the copy is + * mutated, written to the edit log, and then swapped in verbatim, so a + * published instance is never mutated after being logged. Every field is + * carried over; the {@code result} and {@code schemaContract} references + * are shared, which is safe because both are immutable values that are + * only ever replaced wholesale, never mutated in place. + */ + public LanceIndexJob(LanceIndexJob other) { + this.jobId = other.jobId; + this.creator = other.creator; + this.revision = other.revision; + this.createTimeMs = other.createTimeMs; + this.updateTimeMs = other.updateTimeMs; + this.catalogId = other.catalogId; + this.dbName = other.dbName; + this.tableName = other.tableName; + this.provider = other.provider; + this.normalizedLocator = other.normalizedLocator; + this.displayIndexName = other.displayIndexName; + this.normalizedIndexName = other.normalizedIndexName; + this.mutationType = other.mutationType; + this.ifNotExists = other.ifNotExists; + this.ifExists = other.ifExists; + this.indexType = other.indexType; + this.columnName = other.columnName; + this.propertiesJson = other.propertiesJson; + this.admittedDatasetVersion = other.admittedDatasetVersion; + this.schemaContract = other.schemaContract; + this.mutationState = other.mutationState; + this.refreshState = other.refreshState; + this.result = other.result; + this.backendId = other.backendId; + this.beProcessEpoch = other.beProcessEpoch; + this.dispatchRevision = other.dispatchRevision; + this.invocationId = other.invocationId; + this.deadlineMs = other.deadlineMs; + this.possibleLiveOwned = other.possibleLiveOwned; + this.terminationProof = other.terminationProof; + this.forceReleased = other.forceReleased; + this.forceActor = other.forceActor; + this.forceTimeMs = other.forceTimeMs; + this.forceNote = other.forceNote; + this.forceWarning = other.forceWarning; + } + + // ------------------------------------------------------------------ + // Derived helpers + // ------------------------------------------------------------------ + + /** + * The durable same-name fence key of this job. + */ + public LanceIndexFenceKey fenceKey() { + return new LanceIndexFenceKey(catalogId, provider, normalizedLocator, normalizedIndexName); + } + + /** + * The per persisted table/locator quota identity. + */ + public LanceIndexJobQuota.TableQuotaKey getTableQuotaKey() { + return new LanceIndexJobQuota.TableQuotaKey(catalogId, normalizedLocator); + } + + /** + * Whether this job still holds its same-name fence and unresolved quota. + * Fence and quota are released together: PENDING/RUNNING always hold; a + * known terminal job holds until its required refresh is DONE (FAILED + * still holds, refresh may retry); UNKNOWN holds until a durable + * FORCE_RELEASE. A null state from a corrupt record is treated as UNKNOWN, + * the safe direction (fence retained, never redispatched). + */ + public boolean isUnresolved() { + LanceIndexJobMutationState ms = mutationState == null ? LanceIndexJobMutationState.UNKNOWN : mutationState; + switch (ms) { + case PENDING: + case RUNNING: + return true; + case UNKNOWN: + return !forceReleased; + case COMMITTED: + case NOT_COMMITTED: + default: + LanceIndexJobRefreshState rs = + refreshState == null ? LanceIndexJobRefreshState.REQUIRED : refreshState; + return rs == LanceIndexJobRefreshState.REQUIRED + || rs == LanceIndexJobRefreshState.RUNNING + || rs == LanceIndexJobRefreshState.FAILED; + } + } + + /** + * Whether this job still owns a possible-live worker slot: released only + * by a matching termination proof or a durable FORCE_RELEASE, never by a + * deadline. + */ + public boolean holdsPossibleLiveSlot() { + return possibleLiveOwned + && (terminationProof == null || terminationProof == LanceIndexTerminationProof.NONE) + && !forceReleased; + } + + /** + * Validate the invariant-bearing and bounded fields before this record is + * admitted. This is intentionally separate from Gson replay, which must + * remain tolerant of old or corrupt records in the safe direction. + */ + public void validateForAdmission() { + checkRequiredBytes(creator, MAX_DURABLE_TEXT_BYTES, "creator"); + checkRequiredBytes(dbName, MAX_DURABLE_TEXT_BYTES, "dbName"); + checkRequiredBytes(tableName, MAX_DURABLE_TEXT_BYTES, "tableName"); + if (provider == null) { + throw new IllegalArgumentException("lance index job provider must not be null"); + } + if (!LanceIndexFenceKey.PROVIDER_DIRECTORY.equals(provider)) { + throw new IllegalArgumentException("lance index job provider must be DIRECTORY"); + } + if (normalizedLocator == null) { + throw new IllegalArgumentException("normalized dataset locator must not be null"); + } + String canonicalLocator = LanceIndexDatasetLocator.normalize(normalizedLocator); + if (!canonicalLocator.equals(normalizedLocator)) { + throw new IllegalArgumentException("dataset locator is not in canonical identity form"); + } + LanceIndexNameNormalizer.validateDisplayName(displayIndexName); + validateNormalizedIndexName(normalizedIndexName); + String expectedNormalizedName = LanceIndexNameNormalizer.normalize(displayIndexName); + if (!expectedNormalizedName.equals(normalizedIndexName)) { + throw new IllegalArgumentException("normalized index name does not match the display name"); + } + if (mutationType == null) { + throw new IllegalArgumentException("mutation type must not be null"); + } + checkBytes(indexType, MAX_DURABLE_TEXT_BYTES, "indexType"); + checkBytes(columnName, MAX_DURABLE_TEXT_BYTES, "columnName"); + checkBytes(propertiesJson, MAX_PROPERTIES_JSON_BYTES, "propertiesJson"); + checkBytes(invocationId, MAX_INVOCATION_ID_BYTES, "invocationId"); + checkBytes(forceActor, MAX_DURABLE_TEXT_BYTES, "forceActor"); + checkBytes(forceNote, MAX_FORCE_TEXT_BYTES, "forceNote"); + checkBytes(forceWarning, MAX_FORCE_TEXT_BYTES, "forceWarning"); + if (result != null) { + checkBytes(result.getSanitizedMessage(), LanceIndexJobResult.MAX_MESSAGE_BYTES, "sanitizedMessage"); + } + if (schemaContract != null) { + schemaContract.validateForAdmission(); + } + } + + // ------------------------------------------------------------------ + // Accessors. Setters for bounded text fields re-validate the bound. + // ------------------------------------------------------------------ + + public long getJobId() { + return jobId; + } + + public void setJobId(long jobId) { + this.jobId = jobId; + } + + public String getCreator() { + return creator; + } + + public void setCreator(String creator) { + this.creator = checkBytes(creator, MAX_DURABLE_TEXT_BYTES, "creator"); + } + + public long getRevision() { + return revision; + } + + public void setRevision(long revision) { + this.revision = revision; + } + + public long getCreateTimeMs() { + return createTimeMs; + } + + public void setCreateTimeMs(long createTimeMs) { + this.createTimeMs = createTimeMs; + } + + public long getUpdateTimeMs() { + return updateTimeMs; + } + + public void setUpdateTimeMs(long updateTimeMs) { + this.updateTimeMs = updateTimeMs; + } + + public long getCatalogId() { + return catalogId; + } + + public String getDbName() { + return dbName; + } + + public String getTableName() { + return tableName; + } + + public String getProvider() { + return provider; + } + + public String getNormalizedLocator() { + return normalizedLocator; + } + + public String getDisplayIndexName() { + return displayIndexName; + } + + public final void setDisplayIndexName(String displayIndexName) { + LanceIndexNameNormalizer.validateDisplayName(displayIndexName); + this.displayIndexName = displayIndexName; + } + + public String getNormalizedIndexName() { + return normalizedIndexName; + } + + public final void setNormalizedIndexName(String normalizedIndexName) { + validateNormalizedIndexName(normalizedIndexName); + this.normalizedIndexName = normalizedIndexName; + } + + private static void validateNormalizedIndexName(String normalizedIndexName) { + if (normalizedIndexName == null || normalizedIndexName.isEmpty()) { + throw new IllegalArgumentException("normalized index name must not be null or empty"); + } + if (normalizedIndexName.getBytes(StandardCharsets.UTF_8).length + > LanceIndexNameNormalizer.MAX_INDEX_NAME_BYTES) { + throw new IllegalArgumentException( + "normalized index name exceeds " + LanceIndexNameNormalizer.MAX_INDEX_NAME_BYTES + " UTF-8 bytes"); + } + } + + public LanceIndexJobMutationType getMutationType() { + return mutationType; + } + + public boolean isIfNotExists() { + return ifNotExists; + } + + public boolean isIfExists() { + return ifExists; + } + + public String getIndexType() { + return indexType; + } + + public void setIndexType(String indexType) { + this.indexType = checkBytes(indexType, MAX_DURABLE_TEXT_BYTES, "indexType"); + } + + public String getColumnName() { + return columnName; + } + + public void setColumnName(String columnName) { + this.columnName = checkBytes(columnName, MAX_DURABLE_TEXT_BYTES, "columnName"); + } + + public String getPropertiesJson() { + return propertiesJson; + } + + public final void setPropertiesJson(String propertiesJson) { + this.propertiesJson = checkBytes(propertiesJson, MAX_PROPERTIES_JSON_BYTES, "propertiesJson"); + } + + public long getAdmittedDatasetVersion() { + return admittedDatasetVersion; + } + + public LanceIndexSchemaContract getSchemaContract() { + return schemaContract; + } + + public LanceIndexJobMutationState getMutationState() { + return mutationState; + } + + public void setMutationState(LanceIndexJobMutationState mutationState) { + this.mutationState = Objects.requireNonNull(mutationState, "mutationState"); + } + + public LanceIndexJobRefreshState getRefreshState() { + return refreshState; + } + + public void setRefreshState(LanceIndexJobRefreshState refreshState) { + this.refreshState = Objects.requireNonNull(refreshState, "refreshState"); + } + + public LanceIndexJobResult getResult() { + return result; + } + + public void setResult(LanceIndexJobResult result) { + this.result = result; + } + + public Long getBackendId() { + return backendId; + } + + public void setBackendId(Long backendId) { + this.backendId = backendId; + } + + public Long getBeProcessEpoch() { + return beProcessEpoch; + } + + public void setBeProcessEpoch(Long beProcessEpoch) { + this.beProcessEpoch = beProcessEpoch; + } + + public Long getDispatchRevision() { + return dispatchRevision; + } + + public void setDispatchRevision(Long dispatchRevision) { + this.dispatchRevision = dispatchRevision; + } + + public String getInvocationId() { + return invocationId; + } + + public void setInvocationId(String invocationId) { + this.invocationId = checkBytes(invocationId, MAX_INVOCATION_ID_BYTES, "invocationId"); + } + + public Long getDeadlineMs() { + return deadlineMs; + } + + public void setDeadlineMs(Long deadlineMs) { + this.deadlineMs = deadlineMs; + } + + public boolean isPossibleLiveOwned() { + return possibleLiveOwned; + } + + public void setPossibleLiveOwned(boolean possibleLiveOwned) { + this.possibleLiveOwned = possibleLiveOwned; + } + + public LanceIndexTerminationProof getTerminationProof() { + return terminationProof; + } + + public void setTerminationProof(LanceIndexTerminationProof terminationProof) { + this.terminationProof = Objects.requireNonNull(terminationProof, "terminationProof"); + } + + public boolean isForceReleased() { + return forceReleased; + } + + public void setForceReleased(boolean forceReleased) { + this.forceReleased = forceReleased; + } + + public String getForceActor() { + return forceActor; + } + + public void setForceActor(String forceActor) { + this.forceActor = checkBytes(forceActor, MAX_DURABLE_TEXT_BYTES, "forceActor"); + } + + public Long getForceTimeMs() { + return forceTimeMs; + } + + public void setForceTimeMs(Long forceTimeMs) { + this.forceTimeMs = forceTimeMs; + } + + public String getForceNote() { + return forceNote; + } + + public void setForceNote(String forceNote) { + this.forceNote = checkBytes(forceNote, MAX_FORCE_TEXT_BYTES, "forceNote"); + } + + public String getForceWarning() { + return forceWarning; + } + + public void setForceWarning(String forceWarning) { + this.forceWarning = checkBytes(forceWarning, MAX_FORCE_TEXT_BYTES, "forceWarning"); + } + + private static String checkBytes(String value, int maxBytes, String fieldName) { + if (value != null && value.getBytes(StandardCharsets.UTF_8).length > maxBytes) { + throw new IllegalArgumentException(fieldName + " exceeds " + maxBytes + " UTF-8 bytes"); + } + return value; + } + + private static String checkRequiredBytes(String value, int maxBytes, String fieldName) { + if (value == null || value.isEmpty()) { + throw new IllegalArgumentException(fieldName + " must not be null or empty"); + } + return checkBytes(value, maxBytes, fieldName); + } + + // ------------------------------------------------------------------ + // Serialization (Gson stream style, see DropIndexPolicyLog) + // ------------------------------------------------------------------ + + @Override + public void write(DataOutput out) throws IOException { + Text.writeString(out, GsonUtils.GSON.toJson(this)); + } + + public static LanceIndexJob read(DataInput in) throws IOException { + return GsonUtils.GSON.fromJson(Text.readString(in), LanceIndexJob.class); + } + + /** + * Deliberately omits the locator: job lookups must not disclose target + * details to callers without privilege, and log lines reuse this form. + */ + @Override + public String toString() { + return "LanceIndexJob{jobId=" + jobId + ", revision=" + revision + ", catalogId=" + catalogId + + ", db=" + dbName + ", table=" + tableName + ", index=" + displayIndexName + + ", mutationType=" + mutationType + ", mutationState=" + mutationState + + ", refreshState=" + refreshState + ", possibleLiveSlot=" + holdsPossibleLiveSlot() + + ", forceReleased=" + forceReleased + '}'; + } +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobCompletionReason.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobCompletionReason.java new file mode 100644 index 00000000000000..94bfcd320538d3 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobCompletionReason.java @@ -0,0 +1,28 @@ +// 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.doris.datasource.lance.job; + +/** + * Why a job reached its terminal state. NOOP is not a state: a typed + * post-dispatch DROP-not-found for DROP IF EXISTS is NOT_COMMITTED with + * completion reason IF_CONDITION_NOOP. + */ +public enum LanceIndexJobCompletionReason { + NONE, + IF_CONDITION_NOOP +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobManager.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobManager.java new file mode 100644 index 00000000000000..8ece27bb62f338 --- /dev/null +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobManager.java @@ -0,0 +1,694 @@ +// 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.doris.datasource.lance.job; + +import org.apache.doris.catalog.Env; +import org.apache.doris.common.DdlException; +import org.apache.doris.common.io.Text; +import org.apache.doris.common.io.Writable; +import org.apache.doris.persist.gson.GsonPostProcessable; +import org.apache.doris.persist.gson.GsonUtils; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.Maps; +import com.google.gson.annotations.SerializedName; +import org.apache.commons.lang3.StringUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.io.DataInput; +import java.io.DataOutput; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.NavigableSet; +import java.util.Objects; +import java.util.TreeSet; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +/** + * Master-owned manager of the durable Lance index job records, the same-name + * fences, and the three-level unresolved-job quotas. It deliberately reuses + * neither the generic JobManager scheduling framework nor the internal + * IndexChangeJob machinery: the external one-shot CAS, no-redispatch rule, + * same-name fence, and possible-live ownership required here are not provided + * by either. + * + *
Every FE keeps the same in-memory image of {@link #jobs}: the master + * writes each durable transition to the edit log and then applies the same + * record locally; followers apply it from replay. All transitions share one + * write-path shape: + *
+ * writeLock -> validate (state, revision CAS, callback identity) + * -> writeEditLog(updated copy) // fails only by System.exit + * -> applyToMemory(updated copy) // verbatim swap + fence/quota accounting + *+ * A published {@link LanceIndexJob} is never mutated; a transition stages a + * copy, logs it, and swaps it in. {@link #replayUpsertJob(LanceIndexJob)} is a + * verbatim replace with a monotonic-revision guard and performs no state + * transformation at all: a follower tailing a live master must keep a fresh + * RUNNING record RUNNING. The only place a durable RUNNING without a complete + * matching terminal result becomes UNKNOWN is {@link #onTransferToMaster()}, + * the master-election sweep that runs after metadata replay and before any + * dispatcher could start. + * + *
Fence and quota live and die together ({@link LanceIndexJob#isUnresolved()}): + * released when a known terminal job's refresh is NOT_REQUIRED or DONE, or by + * a durable FORCE_RELEASE; an unresolved UNKNOWN holds both across failover, + * timeout, termination proof, and metadata observations. + * + *
The class starts no threads, so no checkpoint-thread guard is needed in
+ * the constructor; the derived fence index and quota counters are rebuilt in
+ * {@link #gsonPostProcess()} after image load.
+ */
+public class LanceIndexJobManager implements Writable, GsonPostProcessable {
+ private static final Logger LOG = LogManager.getLogger(LanceIndexJobManager.class);
+
+ @SerializedName(value = "jobs")
+ private ConcurrentMap This is also the channel for marking a job UNKNOWN on an ambiguous
+ * result (result code NO_TRUSTED_RESULT), including the master-transfer
+ * sweep; no separate markUnknown API exists.
+ *
+ * @return false (with a warning) when the callback is stale or the job is not RUNNING
+ */
+ public boolean completeWithResult(long jobId, long expectedDispatchRevision, String invocationId,
+ Long beProcessEpoch,
+ LanceIndexJobResult result) {
+ Objects.requireNonNull(result, "result");
+ writeLock();
+ try {
+ LanceIndexJob current = jobs.get(jobId);
+ if (current == null || dispatchRevisionOf(current) != expectedDispatchRevision) {
+ LOG.warn("reject stale lance index job callback for job {}: expected dispatch revision {}, current {}",
+ jobId, expectedDispatchRevision, current);
+ return false;
+ }
+ if (current.getMutationState() != LanceIndexJobMutationState.RUNNING) {
+ LOG.warn("reject lance index job callback for job {} in state {}: only RUNNING accepts a result",
+ jobId, current.getMutationState());
+ return false;
+ }
+ if (!Objects.equals(current.getInvocationId(), invocationId)
+ || !Objects.equals(current.getBeProcessEpoch(), beProcessEpoch)) {
+ LOG.warn("reject stale lance index job callback for job {}: invocation/epoch mismatch, current {}",
+ jobId, current);
+ return false;
+ }
+ LanceIndexJobResultCode.Classification classification = LanceIndexJobResultCode.classify(
+ current.getMutationType(), result.getResultCode(), current.isIfExists(),
+ result.isExternalMetadataAdvanced());
+ LanceIndexJob updated = new LanceIndexJob(current);
+ if (updated.getDispatchRevision() == null) {
+ // Backfill old RUNNING records before the global revision advances so
+ // an independent termination proof can still identify this dispatch.
+ updated.setDispatchRevision(expectedDispatchRevision);
+ }
+ updated.setMutationState(classification.getMutationState());
+ updated.setRefreshState(classification.getRefreshState());
+ updated.setResult(new LanceIndexJobResult(result.getResultCode(), classification.getCompletionReason(),
+ result.getSanitizedMessage(), result.isExternalMetadataAdvanced()));
+ updated.setRevision(current.getRevision() + 1);
+ updated.setUpdateTimeMs(System.currentTimeMillis());
+ writeEditLog(updated);
+ applyToMemory(updated);
+ return true;
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ /**
+ * Refresh REQUIRED -> RUNNING, or FAILED -> RUNNING for a retry
+ * through the idempotent external-table refresh path.
+ */
+ public boolean markRefreshRunning(long jobId, long expectedRevision) {
+ return transitionRefresh(jobId, expectedRevision, LanceIndexJobRefreshState.RUNNING);
+ }
+
+ /**
+ * Refresh RUNNING -> DONE. On a known terminal job this releases the
+ * same-name fence and the unresolved quota.
+ */
+ public boolean markRefreshDone(long jobId, long expectedRevision) {
+ return transitionRefresh(jobId, expectedRevision, LanceIndexJobRefreshState.DONE);
+ }
+
+ /**
+ * Refresh RUNNING -> FAILED. The job keeps fence and quota; a later
+ * {@link #markRefreshRunning} retries through the idempotent path.
+ */
+ public boolean markRefreshFailed(long jobId, long expectedRevision) {
+ return transitionRefresh(jobId, expectedRevision, LanceIndexJobRefreshState.FAILED);
+ }
+
+ private boolean transitionRefresh(long jobId, long expectedRevision, LanceIndexJobRefreshState target) {
+ writeLock();
+ try {
+ LanceIndexJob current = jobs.get(jobId);
+ if (current == null || current.getRevision() != expectedRevision) {
+ LOG.warn("reject refresh transition to {} for lance index job {}: expected revision {}, current {}",
+ target, jobId, expectedRevision, current);
+ return false;
+ }
+ if (current.getMutationState() == null || !current.getMutationState().isTerminal()) {
+ LOG.warn("reject refresh transition to {} for non-terminal lance index job {} in mutation state {}",
+ target, jobId, current.getMutationState());
+ return false;
+ }
+ LanceIndexJobRefreshState from = current.getRefreshState();
+ boolean legal = (target == LanceIndexJobRefreshState.RUNNING
+ && (from == LanceIndexJobRefreshState.REQUIRED || from == LanceIndexJobRefreshState.FAILED))
+ || (target == LanceIndexJobRefreshState.DONE && from == LanceIndexJobRefreshState.RUNNING)
+ || (target == LanceIndexJobRefreshState.FAILED && from == LanceIndexJobRefreshState.RUNNING);
+ if (!legal) {
+ LOG.warn("reject illegal refresh transition {} -> {} for lance index job {}",
+ from, target, jobId);
+ return false;
+ }
+ LanceIndexJob updated = new LanceIndexJob(current);
+ updated.setRefreshState(target);
+ updated.setRevision(current.getRevision() + 1);
+ updated.setUpdateTimeMs(System.currentTimeMillis());
+ writeEditLog(updated);
+ applyToMemory(updated);
+ return true;
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ /**
+ * Record a matching termination proof for a job that still owns a
+ * possible-live slot. Backend, BE process epoch, invocation id, and immutable
+ * dispatch revision must all match. This releases only the slot: it never
+ * changes the mutation state and never releases the fence or quota.
+ */
+ public boolean recordTerminationProof(long jobId, long expectedDispatchRevision, long backendId,
+ long beProcessEpoch, String invocationId, LanceIndexTerminationProof proof) {
+ Objects.requireNonNull(proof, "proof");
+ writeLock();
+ try {
+ LanceIndexJob current = jobs.get(jobId);
+ if (current == null || dispatchRevisionOf(current) != expectedDispatchRevision) {
+ LOG.warn("reject termination proof for lance index job {}: expected dispatch revision {}, current {}",
+ jobId, expectedDispatchRevision, current);
+ return false;
+ }
+ if (StringUtils.isBlank(invocationId)
+ || !Objects.equals(current.getBackendId(), backendId)
+ || !Objects.equals(current.getBeProcessEpoch(), beProcessEpoch)
+ || !Objects.equals(current.getInvocationId(), invocationId)) {
+ LOG.warn("reject stale termination proof for lance index job {}:"
+ + " dispatch identity mismatch, current {}", jobId, current);
+ return false;
+ }
+ if (proof == LanceIndexTerminationProof.NONE || !current.isPossibleLiveOwned()
+ || current.getTerminationProof() != LanceIndexTerminationProof.NONE) {
+ LOG.warn("reject termination proof {} for lance index job {}: no possible-live slot owned, current {}",
+ proof, jobId, current);
+ return false;
+ }
+ LanceIndexJob updated = new LanceIndexJob(current);
+ if (updated.getDispatchRevision() == null) {
+ updated.setDispatchRevision(expectedDispatchRevision);
+ }
+ updated.setTerminationProof(proof);
+ updated.setPossibleLiveOwned(false);
+ updated.setRevision(current.getRevision() + 1);
+ updated.setUpdateTimeMs(System.currentTimeMillis());
+ writeEditLog(updated);
+ applyToMemory(updated);
+ return true;
+ } finally {
+ writeUnlock();
+ }
+ }
+
+ /**
+ * Master-election sweep, hooked from {@code Env.transferToMaster()} after
+ * metadata replay and before any master-only dispatcher could start. A
+ * durable RUNNING at this point means the terminal result may have been
+ * lost with the old master: the job becomes UNKNOWN through the same
+ * completeWithResult channel (result code NO_TRUSTED_RESULT, fence/quota/
+ * possible-live ownership retained, never redispatched), and an in-flight
+ * refresh is downgraded to REQUIRED so the idempotent refresh retries.
+ * Both transitions are written to the edit log so followers converge.
+ */
+ public void onTransferToMaster() {
+ List This class only counts. Callers hold the manager write lock, so no
+ * internal synchronization exists. Config-gated limits are resolved by the
+ * admission layer and passed in as positive finite values. The counters are rebuilt from the durable jobs after
+ * replay/image load and are never persisted themselves.
+ */
+public class LanceIndexJobQuota {
+ private static final Logger LOG = LogManager.getLogger(LanceIndexJobQuota.class);
+
+ private long globalCount;
+ private final Map The manager admission path uses {@link #hasCapacity} plus
+ * {@link #charge} instead: the check and the charge deliberately straddle
+ * the edit-log write. This self-contained variant is reserved for direct
+ * use by the later admission slice and by tests.
+ */
+ public boolean tryAcquire(LanceIndexJob job, long tableLimit, long catalogLimit, long globalLimit) {
+ if (!hasCapacity(job, tableLimit, catalogLimit, globalLimit)) {
+ return false;
+ }
+ charge(job);
+ return true;
+ }
+
+ /**
+ * Pure check variant of {@link #tryAcquire}: true when incrementing would
+ * not exceed any positive finite limit. A non-positive limit is rejected.
+ */
+ public boolean hasCapacity(LanceIndexJob job, long tableLimit, long catalogLimit, long globalLimit) {
+ Objects.requireNonNull(job, "job");
+ if (tableLimit <= 0 || catalogLimit <= 0 || globalLimit <= 0) {
+ return false;
+ }
+ if (globalCount >= globalLimit) {
+ return false;
+ }
+ if (getCatalogCount(job.getCatalogId()) >= catalogLimit) {
+ return false;
+ }
+ return getTableCount(job.getTableQuotaKey()) < tableLimit;
+ }
+
+ /**
+ * Increment all three levels unconditionally. Used when a durable record
+ * (already admitted, or replayed) is applied to memory.
+ */
+ public void charge(LanceIndexJob job) {
+ globalCount++;
+ catalogCounts.merge(job.getCatalogId(), 1L, Long::sum);
+ tableCounts.merge(job.getTableQuotaKey(), 1L, Long::sum);
+ }
+
+ /**
+ * Decrement all three levels. Underflow is clamped at zero and warned
+ * about: it indicates a bookkeeping bug, not a reason to fail replay.
+ */
+ public void release(LanceIndexJob job) {
+ if (globalCount <= 0) {
+ LOG.warn("lance index job quota global underflow on release of job {}", job.getJobId());
+ } else {
+ globalCount--;
+ }
+ decrement(catalogCounts, job.getCatalogId(), job.getJobId());
+ decrement(tableCounts, job.getTableQuotaKey(), job.getJobId());
+ }
+
+ private static Building a contract from an Arrow schema belongs to admission (a later
+ * delivery slice); this class is only the durable, comparable representation.
+ */
+public class LanceIndexSchemaContract {
+ /** The only schema contract version defined. */
+ public static final int SCHEMA_CONTRACT_VERSION_V1 = 1;
+ /** A finite representation bound; 4.2 admission may impose a smaller type-specific arity. */
+ public static final int MAX_INDEXED_FIELDS = 64;
+ /** Bound for every persisted schema-contract string. */
+ public static final int MAX_FIELD_STRING_BYTES = 1024;
+
+ @SerializedName(value = "scv")
+ private int schemaContractVersion = SCHEMA_CONTRACT_VERSION_V1;
+
+ @SerializedName(value = "flds")
+ private List Key invariants pinned here: PRE_INVOCATION_* always prove NOT_COMMITTED and owe a
+ * refresh only when trusted revalidation observed external metadata advancement;
+ * NATIVE_NOT_FOUND has no attribution for CREATE/REPLACE (UNKNOWN) but is a clean
+ * NOT_COMMITTED for DROP; IF_CONDITION_NOOP exists only for DROP IF EXISTS + NOT_FOUND;
+ * every other ambiguous post-invocation outcome is UNKNOWN + NOT_REQUIRED.
+ */
+public class LanceIndexJobResultClassifyTest {
+
+ @Test
+ public void fullClassificationProductMatchesDesignTable() {
+ int combos = 0;
+ for (LanceIndexJobResultCode code : LanceIndexJobResultCode.values()) {
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ for (boolean ifExists : new boolean[]{false, true}) {
+ for (boolean advanced : new boolean[]{false, true}) {
+ LanceIndexJobResultCode.Classification classification =
+ LanceIndexJobResultCode.classify(type, code, ifExists, advanced);
+ String context = "code=" + code + ", type=" + type
+ + ", ifExists=" + ifExists + ", advanced=" + advanced;
+ Assertions.assertEquals(expectedMutationState(type, code),
+ classification.getMutationState(), context);
+ Assertions.assertEquals(expectedRefreshState(type, code, advanced),
+ classification.getRefreshState(), context);
+ Assertions.assertEquals(expectedCompletionReason(type, code, ifExists),
+ classification.getCompletionReason(), context);
+ combos++;
+ }
+ }
+ }
+ }
+ Assertions.assertEquals(13 * 3 * 2 * 2, combos);
+ }
+
+ @Test
+ public void ifConditionNoopExistsOnlyForDropIfExistsNotFound() {
+ int noopCombos = 0;
+ for (LanceIndexJobResultCode code : LanceIndexJobResultCode.values()) {
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ for (boolean ifExists : new boolean[]{false, true}) {
+ for (boolean advanced : new boolean[]{false, true}) {
+ LanceIndexJobResultCode.Classification classification =
+ LanceIndexJobResultCode.classify(type, code, ifExists, advanced);
+ if (classification.getCompletionReason() == LanceIndexJobCompletionReason.IF_CONDITION_NOOP) {
+ noopCombos++;
+ Assertions.assertEquals(LanceIndexJobMutationType.DROP, type);
+ Assertions.assertEquals(LanceIndexJobResultCode.NATIVE_NOT_FOUND, code);
+ Assertions.assertTrue(ifExists);
+ }
+ }
+ }
+ }
+ }
+ // Exactly the two advanced-flag variants of (DROP, NATIVE_NOT_FOUND, ifExists).
+ Assertions.assertEquals(2, noopCombos);
+ }
+
+ @Test
+ public void preInvocationRefreshOwesOnlyWhenExternalMetadataAdvanced() {
+ for (LanceIndexJobResultCode code : LanceIndexJobResultCode.values()) {
+ if (!code.isPreInvocation()) {
+ continue;
+ }
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED,
+ LanceIndexJobResultCode.classify(type, code, false, false).getRefreshState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED,
+ LanceIndexJobResultCode.classify(type, code, false, true).getRefreshState());
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED,
+ LanceIndexJobResultCode.classify(type, code, true, true).getMutationState());
+ }
+ }
+ }
+
+ @Test
+ public void nativeOkCommitsAndOwesRefreshForEveryMutationType() {
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ LanceIndexJobResultCode.Classification classification =
+ LanceIndexJobResultCode.classify(type, LanceIndexJobResultCode.NATIVE_OK, false, false);
+ Assertions.assertEquals(LanceIndexJobMutationState.COMMITTED, classification.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, classification.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, classification.getCompletionReason());
+ }
+ }
+
+ @Test
+ public void commitConflictIsNotCommittedWithRefreshOwed() {
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ LanceIndexJobResultCode.Classification classification = LanceIndexJobResultCode.classify(
+ type, LanceIndexJobResultCode.NATIVE_COMMIT_CONFLICT, false, false);
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, classification.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, classification.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, classification.getCompletionReason());
+ }
+ }
+
+ @Test
+ public void nativeNotFoundAttributionDependsOnMutationType() {
+ for (LanceIndexJobMutationType type : new LanceIndexJobMutationType[]{
+ LanceIndexJobMutationType.CREATE, LanceIndexJobMutationType.REPLACE}) {
+ LanceIndexJobResultCode.Classification classification = LanceIndexJobResultCode.classify(
+ type, LanceIndexJobResultCode.NATIVE_NOT_FOUND, false, false);
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, classification.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, classification.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, classification.getCompletionReason());
+ }
+ LanceIndexJobResultCode.Classification dropClassification = LanceIndexJobResultCode.classify(
+ LanceIndexJobMutationType.DROP, LanceIndexJobResultCode.NATIVE_NOT_FOUND, false, false);
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, dropClassification.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, dropClassification.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, dropClassification.getCompletionReason());
+ }
+
+ @Test
+ public void ambiguousPostInvocationOutcomesAreUnknownWithoutRefresh() {
+ for (LanceIndexJobResultCode code : new LanceIndexJobResultCode[]{
+ LanceIndexJobResultCode.NATIVE_INVALID_ARGUMENT, LanceIndexJobResultCode.NATIVE_NOT_SUPPORTED,
+ LanceIndexJobResultCode.NATIVE_INDEX, LanceIndexJobResultCode.NATIVE_IO,
+ LanceIndexJobResultCode.NATIVE_INTERNAL, LanceIndexJobResultCode.NO_TRUSTED_RESULT}) {
+ for (LanceIndexJobMutationType type : LanceIndexJobMutationType.values()) {
+ LanceIndexJobResultCode.Classification classification =
+ LanceIndexJobResultCode.classify(type, code, true, true);
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, classification.getMutationState(),
+ "code=" + code + ", type=" + type);
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, classification.getRefreshState(),
+ "code=" + code + ", type=" + type);
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, classification.getCompletionReason(),
+ "code=" + code + ", type=" + type);
+ }
+ }
+ }
+
+ @Test
+ public void nullResultCodeFallsBackToNoTrustedResult() {
+ LanceIndexJobResultCode.Classification classification =
+ LanceIndexJobResultCode.classify(LanceIndexJobMutationType.CREATE, null, false, false);
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, classification.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, classification.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, classification.getCompletionReason());
+ }
+
+ @Test
+ public void isPreInvocationCoversExactlyTheFourTrustedRejections() {
+ int preInvocationCount = 0;
+ for (LanceIndexJobResultCode code : LanceIndexJobResultCode.values()) {
+ if (code.isPreInvocation()) {
+ preInvocationCount++;
+ }
+ }
+ Assertions.assertEquals(4, preInvocationCount);
+ Assertions.assertEquals(13, LanceIndexJobResultCode.values().length);
+ }
+
+ /**
+ * Independent restatement of the design table, mutation-state column.
+ */
+ private static LanceIndexJobMutationState expectedMutationState(
+ LanceIndexJobMutationType type, LanceIndexJobResultCode code) {
+ if (isPreInvocationSpec(code)) {
+ return LanceIndexJobMutationState.NOT_COMMITTED;
+ }
+ switch (code) {
+ case NATIVE_OK:
+ return LanceIndexJobMutationState.COMMITTED;
+ case NATIVE_COMMIT_CONFLICT:
+ return LanceIndexJobMutationState.NOT_COMMITTED;
+ case NATIVE_NOT_FOUND:
+ return type == LanceIndexJobMutationType.DROP
+ ? LanceIndexJobMutationState.NOT_COMMITTED : LanceIndexJobMutationState.UNKNOWN;
+ default:
+ return LanceIndexJobMutationState.UNKNOWN;
+ }
+ }
+
+ /**
+ * Independent restatement of the design table, refresh-obligation column.
+ */
+ private static LanceIndexJobRefreshState expectedRefreshState(
+ LanceIndexJobMutationType type, LanceIndexJobResultCode code, boolean externalMetadataAdvanced) {
+ if (isPreInvocationSpec(code)) {
+ return externalMetadataAdvanced
+ ? LanceIndexJobRefreshState.REQUIRED : LanceIndexJobRefreshState.NOT_REQUIRED;
+ }
+ switch (code) {
+ case NATIVE_OK:
+ case NATIVE_COMMIT_CONFLICT:
+ return LanceIndexJobRefreshState.REQUIRED;
+ case NATIVE_NOT_FOUND:
+ return type == LanceIndexJobMutationType.DROP
+ ? LanceIndexJobRefreshState.REQUIRED : LanceIndexJobRefreshState.NOT_REQUIRED;
+ default:
+ return LanceIndexJobRefreshState.NOT_REQUIRED;
+ }
+ }
+
+ /**
+ * Independent restatement of the design table, completion-reason column.
+ */
+ private static LanceIndexJobCompletionReason expectedCompletionReason(
+ LanceIndexJobMutationType type, LanceIndexJobResultCode code, boolean ifExists) {
+ return type == LanceIndexJobMutationType.DROP && code == LanceIndexJobResultCode.NATIVE_NOT_FOUND && ifExists
+ ? LanceIndexJobCompletionReason.IF_CONDITION_NOOP : LanceIndexJobCompletionReason.NONE;
+ }
+
+ private static boolean isPreInvocationSpec(LanceIndexJobResultCode code) {
+ return code == LanceIndexJobResultCode.PRE_INVOCATION_STALE_ADMISSION
+ || code == LanceIndexJobResultCode.PRE_INVOCATION_UNSUPPORTED_SCHEMA_CONTRACT
+ || code == LanceIndexJobResultCode.PRE_INVOCATION_CREDENTIAL_EXPIRED
+ || code == LanceIndexJobResultCode.PRE_INVOCATION_RESOURCE_REJECTED;
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/job/LanceIndexJobStateMachineTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/job/LanceIndexJobStateMachineTest.java
new file mode 100644
index 00000000000000..54a5945eb79b16
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/lance/job/LanceIndexJobStateMachineTest.java
@@ -0,0 +1,640 @@
+// 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.doris.datasource.lance.job;
+
+import org.apache.doris.common.DdlException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Lifecycle state-machine coverage driven through the master write paths of
+ * {@link LanceIndexJobManager} with the edit-log seam captured in memory. Covers the
+ * legal mutation chain PENDING -> RUNNING -> (COMMITTED | NOT_COMMITTED |
+ * UNKNOWN), the impossibility of leaving UNKNOWN, the independent refresh chain
+ * REQUIRED -> RUNNING -> DONE / FAILED with FAILED -> RUNNING retry, the
+ * revision compare-and-set on every transition, and the fence/quota release timing
+ * (immediately on a terminal outcome with refresh NOT_REQUIRED, only at refresh DONE
+ * otherwise, never for UNKNOWN without FORCE).
+ */
+public class LanceIndexJobStateMachineTest {
+ private static final long CATALOG_ID = 10L;
+ private static final String LOCATOR = "s3://bucket/dataset";
+ private static final long BACKEND_ID = 1001L;
+ private static final long BE_EPOCH = 55L;
+ private static final String INVOCATION_ID = "invocation-1";
+ private static final long DEADLINE_MS = 9999L;
+
+ @Test
+ public void createInitializesPendingRecord() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, stored.getRefreshState());
+ Assertions.assertEquals(0L, stored.getRevision());
+ Assertions.assertFalse(stored.isPossibleLiveOwned());
+ Assertions.assertEquals(LanceIndexTerminationProof.NONE, stored.getTerminationProof());
+ Assertions.assertTrue(stored.getCreateTimeMs() > 0L);
+ Assertions.assertTrue(manager.isFenceHeld(stored.fenceKey()));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+ Assertions.assertEquals(1L, manager.getQuota().getCatalogCount(CATALOG_ID));
+ Assertions.assertEquals(1L, manager.getQuota().getTableCount(stored.getTableQuotaKey()));
+ Assertions.assertEquals(1, manager.getUnresolvedJobs().size());
+ Assertions.assertEquals(1, manager.editLog.size());
+ }
+
+ @Test
+ public void pendingToRunningToCommitted() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+ Assertions.assertTrue(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+
+ LanceIndexJob running = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.RUNNING, running.getMutationState());
+ Assertions.assertEquals(1L, running.getRevision());
+ Assertions.assertEquals(BACKEND_ID, running.getBackendId().longValue());
+ Assertions.assertEquals(BE_EPOCH, running.getBeProcessEpoch().longValue());
+ Assertions.assertEquals(INVOCATION_ID, running.getInvocationId());
+ Assertions.assertEquals(DEADLINE_MS, running.getDeadlineMs().longValue());
+ Assertions.assertTrue(running.holdsPossibleLiveSlot());
+
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ LanceIndexJob committed = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.COMMITTED, committed.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, committed.getRefreshState());
+ Assertions.assertEquals(2L, committed.getRevision());
+ Assertions.assertEquals(LanceIndexJobResultCode.NATIVE_OK, committed.getResult().getResultCode());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.NONE, committed.getResult().getCompletionReason());
+ Assertions.assertTrue(containsJob(manager.getJobsNeedingRefresh(), committed.getJobId()));
+ }
+
+ @Test
+ public void pendingToRunningToNotCommitted() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.PRE_INVOCATION_CREDENTIAL_EXPIRED)));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, stored.getRefreshState());
+ Assertions.assertEquals(2L, stored.getRevision());
+ }
+
+ @Test
+ public void pendingToRunningToUnknown() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NO_TRUSTED_RESULT)));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, stored.getRefreshState());
+ Assertions.assertEquals(2L, stored.getRevision());
+ }
+
+ @Test
+ public void dropIfExistsNotFoundCompletesWithIfConditionNoop() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newDropJob(1L, "IdxA", true), 100, 100, 100);
+ Assertions.assertTrue(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_NOT_FOUND)));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, stored.getRefreshState());
+ Assertions.assertEquals(LanceIndexJobCompletionReason.IF_CONDITION_NOOP,
+ stored.getResult().getCompletionReason());
+ }
+
+ @Test
+ public void markRunningRejectsSecondDispatch() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+
+ Assertions.assertFalse(manager.markRunning(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertFalse(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertEquals(LanceIndexJobMutationState.RUNNING, manager.getJob(1L).getMutationState());
+ Assertions.assertEquals(1L, manager.getJob(1L).getRevision());
+ }
+
+ @Test
+ public void markRunningRejectsWrongRevisionAndUnknownJob() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+
+ Assertions.assertFalse(manager.markRunning(1L, 5L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertFalse(manager.markRunning(404L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, manager.getJob(1L).getMutationState());
+ Assertions.assertEquals(1, manager.editLog.size());
+ }
+
+ @Test
+ public void completeRejectsNonRunningJob() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+
+ Assertions.assertFalse(manager.completeWithResult(1L, 0L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, manager.getJob(1L).getMutationState());
+
+ createAndRun(manager, 2L, "IdxB");
+ Assertions.assertTrue(manager.completeWithResult(2L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertFalse(manager.completeWithResult(2L, 2L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertEquals(LanceIndexJobMutationState.COMMITTED, manager.getJob(2L).getMutationState());
+ }
+
+ @Test
+ public void unknownHasNoOutgoingTransitions() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NO_TRUSTED_RESULT)));
+ int loggedRecords = manager.editLog.size();
+
+ Assertions.assertFalse(manager.markRunning(1L, 2L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertFalse(manager.completeWithResult(1L, 2L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertFalse(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertFalse(manager.markRefreshDone(1L, 2L));
+ Assertions.assertFalse(manager.markRefreshFailed(1L, 2L));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, stored.getMutationState());
+ Assertions.assertEquals(2L, stored.getRevision());
+ Assertions.assertEquals(loggedRecords, manager.editLog.size());
+ }
+
+ @Test
+ public void refreshLifecycleReleasesFenceAndQuotaAtDone() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+
+ LanceIndexFenceKey fenceKey = manager.getJob(1L).fenceKey();
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.RUNNING, manager.getJob(1L).getRefreshState());
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+
+ Assertions.assertTrue(manager.markRefreshDone(1L, 3L));
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobRefreshState.DONE, stored.getRefreshState());
+ Assertions.assertEquals(4L, stored.getRevision());
+ Assertions.assertFalse(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ Assertions.assertTrue(manager.getUnresolvedJobs().isEmpty());
+ Assertions.assertTrue(manager.getJobsNeedingRefresh().isEmpty());
+ }
+
+ @Test
+ public void refreshFailureKeepsFenceAndRetriesThroughRunning() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ LanceIndexFenceKey fenceKey = manager.getJob(1L).fenceKey();
+
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertTrue(manager.markRefreshFailed(1L, 3L));
+ LanceIndexJob failed = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobRefreshState.FAILED, failed.getRefreshState());
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+ Assertions.assertTrue(containsJob(manager.getUnresolvedJobs(), failed.getJobId()));
+
+ // FAILED -> RUNNING is the retry entry through the idempotent refresh path.
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 4L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.RUNNING, manager.getJob(1L).getRefreshState());
+ Assertions.assertTrue(manager.markRefreshDone(1L, 5L));
+ Assertions.assertFalse(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ }
+
+ @Test
+ public void refreshTransitionsRejectedFromNotRequired() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.PRE_INVOCATION_RESOURCE_REJECTED)));
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, manager.getJob(1L).getRefreshState());
+
+ Assertions.assertFalse(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertFalse(manager.markRefreshDone(1L, 2L));
+ Assertions.assertFalse(manager.markRefreshFailed(1L, 2L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, manager.getJob(1L).getRefreshState());
+ Assertions.assertEquals(2L, manager.getJob(1L).getRevision());
+ }
+
+ @Test
+ public void refreshTransitionsRejectedFromDone() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertTrue(manager.markRefreshDone(1L, 3L));
+
+ Assertions.assertFalse(manager.markRefreshRunning(1L, 4L));
+ Assertions.assertFalse(manager.markRefreshDone(1L, 4L));
+ Assertions.assertFalse(manager.markRefreshFailed(1L, 4L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.DONE, manager.getJob(1L).getRefreshState());
+ }
+
+ @Test
+ public void refreshTransitionIsRevisionGuarded() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+
+ Assertions.assertFalse(manager.markRefreshRunning(1L, 99L));
+ Assertions.assertFalse(manager.markRefreshRunning(404L, 2L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, manager.getJob(1L).getRefreshState());
+ Assertions.assertEquals(2L, manager.getJob(1L).getRevision());
+ }
+
+ @Test
+ public void fenceAndQuotaReleasedImmediatelyWhenRefreshNotRequired() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ LanceIndexFenceKey fenceKey = manager.getJob(1L).fenceKey();
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.PRE_INVOCATION_STALE_ADMISSION)));
+
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, manager.getJob(1L).getMutationState());
+ Assertions.assertFalse(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ Assertions.assertEquals(0L, manager.getQuota().getCatalogCount(CATALOG_ID));
+ Assertions.assertTrue(manager.getUnresolvedJobs().isEmpty());
+ }
+
+ @Test
+ public void fenceAndQuotaSurviveUnknown() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ LanceIndexFenceKey fenceKey = manager.getJob(1L).fenceKey();
+
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_IO)));
+
+ Assertions.assertEquals(LanceIndexJobMutationState.UNKNOWN, manager.getJob(1L).getMutationState());
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+ Assertions.assertEquals(1, manager.getUnresolvedJobs().size());
+ Assertions.assertTrue(manager.getJobsNeedingRefresh().isEmpty());
+ }
+
+ @Test
+ public void terminationProofReleasesSlotOnly() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ LanceIndexFenceKey fenceKey = manager.getJob(1L).fenceKey();
+ Assertions.assertTrue(manager.getJob(1L).holdsPossibleLiveSlot());
+
+ Assertions.assertFalse(manager.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.NONE));
+ Assertions.assertFalse(manager.recordTerminationProof(1L, 99L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertTrue(manager.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.RUNNING, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexTerminationProof.CHILD_REAPED, stored.getTerminationProof());
+ Assertions.assertFalse(stored.holdsPossibleLiveSlot());
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+
+ // A slot may be proven exactly once.
+ Assertions.assertFalse(manager.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.BE_PROCESS_EPOCH_GONE));
+ }
+
+ @Test
+ public void everyAcceptedTransitionWritesExactlyOneEditLogRecord() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+ manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS);
+ manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH, result(LanceIndexJobResultCode.NATIVE_OK));
+ manager.markRefreshRunning(1L, 2L);
+ manager.markRefreshDone(1L, 3L);
+ Assertions.assertEquals(5, manager.editLog.size());
+
+ // Rejected transitions never reach the journal.
+ Assertions.assertFalse(manager.markRefreshDone(1L, 4L));
+ Assertions.assertFalse(manager.markRunning(1L, 4L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertEquals(5, manager.editLog.size());
+ }
+
+ @Test
+ public void markRunningRejectsBlankInvocationId() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+
+ // A null/blank invocation identity would match a null field under Objects.equals
+ // in completeWithResult and silently defeat the stale-callback guard.
+ Assertions.assertFalse(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, null, DEADLINE_MS));
+ Assertions.assertFalse(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, "", DEADLINE_MS));
+ Assertions.assertFalse(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, " \t\n", DEADLINE_MS));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, stored.getMutationState());
+ Assertions.assertEquals(0L, stored.getRevision());
+ Assertions.assertNull(stored.getInvocationId());
+ Assertions.assertEquals(1, manager.editLog.size());
+
+ // A well-formed dispatch is still accepted afterwards.
+ Assertions.assertTrue(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertEquals(LanceIndexJobMutationState.RUNNING, manager.getJob(1L).getMutationState());
+ }
+
+ @Test
+ public void failedRefreshJobStaysVisibleToTheRefreshDriver() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertTrue(manager.markRefreshFailed(1L, 3L));
+
+ // FAILED still owes the idempotent retry: the driver must see the job.
+ Assertions.assertTrue(containsJob(manager.getJobsNeedingRefresh(), 1L));
+
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 4L));
+ Assertions.assertTrue(manager.markRefreshDone(1L, 5L));
+ Assertions.assertTrue(manager.getJobsNeedingRefresh().isEmpty());
+
+ // Terminal jobs with refresh DONE or NOT_REQUIRED never show up.
+ createAndRun(manager, 2L, "IdxB");
+ Assertions.assertTrue(manager.completeWithResult(2L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.PRE_INVOCATION_RESOURCE_REJECTED)));
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, manager.getJob(2L).getRefreshState());
+ Assertions.assertTrue(manager.getJobsNeedingRefresh().isEmpty());
+ }
+
+ @Test
+ public void createJobResetsLifecycleFieldsAndPublishesAPrivateCopy() throws DdlException {
+ LanceIndexJob dirty = newCreateJob(1L, "IdxA");
+ dirty.setRevision(9L);
+ dirty.setMutationState(LanceIndexJobMutationState.RUNNING);
+ dirty.setRefreshState(LanceIndexJobRefreshState.RUNNING);
+ dirty.setResult(result(LanceIndexJobResultCode.NATIVE_OK));
+ dirty.setBackendId(BACKEND_ID);
+ dirty.setBeProcessEpoch(BE_EPOCH);
+ dirty.setInvocationId(INVOCATION_ID);
+ dirty.setDeadlineMs(DEADLINE_MS);
+ dirty.setPossibleLiveOwned(true);
+ dirty.setTerminationProof(LanceIndexTerminationProof.CHILD_REAPED);
+ dirty.setForceReleased(true);
+ dirty.setForceActor("admin");
+ dirty.setForceTimeMs(7L);
+ dirty.setForceNote("note");
+ dirty.setForceWarning("warning");
+
+ TestManager manager = new TestManager();
+ manager.createJob(dirty, 100, 100, 100);
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertNotSame(dirty, stored);
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.NOT_REQUIRED, stored.getRefreshState());
+ Assertions.assertEquals(0L, stored.getRevision());
+ Assertions.assertNull(stored.getResult());
+ Assertions.assertNull(stored.getBackendId());
+ Assertions.assertNull(stored.getBeProcessEpoch());
+ Assertions.assertNull(stored.getInvocationId());
+ Assertions.assertNull(stored.getDeadlineMs());
+ Assertions.assertFalse(stored.isPossibleLiveOwned());
+ Assertions.assertEquals(LanceIndexTerminationProof.NONE, stored.getTerminationProof());
+ Assertions.assertFalse(stored.isForceReleased());
+ Assertions.assertNull(stored.getForceActor());
+ Assertions.assertNull(stored.getForceTimeMs());
+ Assertions.assertNull(stored.getForceNote());
+ Assertions.assertNull(stored.getForceWarning());
+
+ // Mutating the caller's object after admission touches nothing inside the manager.
+ dirty.setMutationState(LanceIndexJobMutationState.UNKNOWN);
+ dirty.setNormalizedIndexName("idxb");
+ Assertions.assertEquals(LanceIndexJobMutationState.PENDING, manager.getJob(1L).getMutationState());
+ Assertions.assertTrue(manager.isFenceHeld(newCreateJob(2L, "idxa").fenceKey()));
+ Assertions.assertFalse(manager.isFenceHeld(newCreateJob(3L, "IdxB").fenceKey()));
+ }
+
+ @Test
+ public void refreshDoneOrFailedRequiresAPrecedingRefreshRunning() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, manager.getJob(1L).getRefreshState());
+
+ // No direct REQUIRED -> DONE / FAILED: a refresh must actually run first.
+ Assertions.assertFalse(manager.markRefreshDone(1L, 2L));
+ Assertions.assertFalse(manager.markRefreshFailed(1L, 2L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, manager.getJob(1L).getRefreshState());
+ Assertions.assertEquals(2L, manager.getJob(1L).getRevision());
+ Assertions.assertEquals(3, manager.editLog.size());
+ }
+
+ @Test
+ public void terminationProofNeedsASlotAndStillLandsAfterTheTerminalResult() throws DdlException {
+ TestManager manager = new TestManager();
+ manager.createJob(newCreateJob(1L, "IdxA"), 100, 100, 100);
+
+ // PENDING owns no possible-live slot: a proof has nothing to release.
+ Assertions.assertFalse(manager.recordTerminationProof(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertEquals(0L, manager.getJob(1L).getRevision());
+ Assertions.assertEquals(1, manager.editLog.size());
+
+ Assertions.assertTrue(manager.markRunning(1L, 0L, BACKEND_ID, BE_EPOCH, INVOCATION_ID, DEADLINE_MS));
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ // The terminal outcome does not release the slot; the proof still lands afterwards.
+ LanceIndexJob committed = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.COMMITTED, committed.getMutationState());
+ Assertions.assertTrue(committed.holdsPossibleLiveSlot());
+ LanceIndexFenceKey fenceKey = committed.fenceKey();
+
+ Assertions.assertTrue(manager.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ LanceIndexJob proven = manager.getJob(1L);
+ Assertions.assertFalse(proven.holdsPossibleLiveSlot());
+ // Fence and quota still follow the refresh rule, not the proof.
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 3L));
+ Assertions.assertTrue(manager.markRefreshDone(1L, 4L));
+ Assertions.assertFalse(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ }
+
+ @Test
+ public void terminationProofAndResultUseTheImmutableDispatchRevisionInEitherOrder() throws DdlException {
+ TestManager proofFirst = new TestManager();
+ createAndRun(proofFirst, 1L, "IdxA");
+ Assertions.assertEquals(1L, proofFirst.getJob(1L).getDispatchRevision());
+
+ Assertions.assertFalse(proofFirst.recordTerminationProof(1L, 1L, BACKEND_ID + 1, BE_EPOCH,
+ INVOCATION_ID, LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertFalse(proofFirst.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH + 1,
+ INVOCATION_ID, LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertFalse(proofFirst.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH,
+ "wrong-invocation", LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertEquals(1L, proofFirst.getJob(1L).getRevision());
+
+ Assertions.assertTrue(proofFirst.recordTerminationProof(1L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertEquals(2L, proofFirst.getJob(1L).getRevision());
+ Assertions.assertEquals(1L, proofFirst.getJob(1L).getDispatchRevision());
+ Assertions.assertTrue(proofFirst.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertEquals(LanceIndexJobMutationState.COMMITTED,
+ proofFirst.getJob(1L).getMutationState());
+
+ TestManager resultFirst = new TestManager();
+ createAndRun(resultFirst, 2L, "IdxB");
+ Assertions.assertTrue(resultFirst.completeWithResult(2L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertTrue(resultFirst.recordTerminationProof(2L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertFalse(resultFirst.getJob(2L).holdsPossibleLiveSlot());
+ Assertions.assertEquals(1L, resultFirst.getJob(2L).getDispatchRevision());
+
+ TestManager legacyResultFirst = new TestManager();
+ createAndRun(legacyResultFirst, 3L, "IdxC");
+ LanceIndexJob legacyRunning = legacyResultFirst.getJob(3L);
+ legacyRunning.setDispatchRevision(null);
+ legacyResultFirst.replayUpsertJob(legacyRunning);
+ Assertions.assertTrue(legacyResultFirst.completeWithResult(3L, 1L, INVOCATION_ID, BE_EPOCH,
+ result(LanceIndexJobResultCode.NATIVE_OK)));
+ Assertions.assertEquals(1L, legacyResultFirst.getJob(3L).getDispatchRevision());
+ Assertions.assertTrue(legacyResultFirst.recordTerminationProof(3L, 1L, BACKEND_ID, BE_EPOCH, INVOCATION_ID,
+ LanceIndexTerminationProof.CHILD_REAPED));
+ Assertions.assertFalse(legacyResultFirst.getJob(3L).holdsPossibleLiveSlot());
+ }
+
+ @Test
+ public void refreshTransitionsRejectNonTerminalMutationState() {
+ TestManager manager = new TestManager();
+ LanceIndexJob corrupt = newCreateJob(1L, "IdxA");
+ corrupt.setMutationState(LanceIndexJobMutationState.PENDING);
+ corrupt.setRefreshState(LanceIndexJobRefreshState.REQUIRED);
+ manager.replayUpsertJob(corrupt);
+
+ Assertions.assertFalse(manager.markRefreshRunning(1L, 0L));
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, manager.getJob(1L).getRefreshState());
+ Assertions.assertTrue(manager.editLog.isEmpty());
+ }
+
+ @Test
+ public void admissionRejectsNonPositiveQuotaLimitsBeforeDurableWrite() {
+ TestManager manager = new TestManager();
+ Assertions.assertThrows(DdlException.class,
+ () -> manager.createJob(newCreateJob(1L, "IdxA"), 0, 1, 1));
+ Assertions.assertThrows(DdlException.class,
+ () -> manager.createJob(newCreateJob(1L, "IdxA"), 1, 0, 1));
+ Assertions.assertThrows(DdlException.class,
+ () -> manager.createJob(newCreateJob(1L, "IdxA"), 1, 1, -1));
+ Assertions.assertEquals(0, manager.getJobCount());
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ Assertions.assertTrue(manager.editLog.isEmpty());
+ }
+
+ @Test
+ public void notCommittedWithRefreshRequiredReleasesFenceAtRefreshDone() throws DdlException {
+ TestManager manager = new TestManager();
+ createAndRun(manager, 1L, "IdxA");
+ Assertions.assertTrue(manager.completeWithResult(1L, 1L, INVOCATION_ID, BE_EPOCH,
+ new LanceIndexJobResult(LanceIndexJobResultCode.NATIVE_COMMIT_CONFLICT,
+ LanceIndexJobCompletionReason.NONE, "commit conflict", false)));
+
+ LanceIndexJob stored = manager.getJob(1L);
+ Assertions.assertEquals(LanceIndexJobMutationState.NOT_COMMITTED, stored.getMutationState());
+ Assertions.assertEquals(LanceIndexJobRefreshState.REQUIRED, stored.getRefreshState());
+ LanceIndexFenceKey fenceKey = stored.fenceKey();
+ Assertions.assertTrue(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(1L, manager.getQuota().getGlobalCount());
+ Assertions.assertTrue(containsJob(manager.getJobsNeedingRefresh(), stored.getJobId()));
+
+ Assertions.assertTrue(manager.markRefreshRunning(1L, 2L));
+ Assertions.assertTrue(manager.markRefreshDone(1L, 3L));
+ Assertions.assertFalse(manager.isFenceHeld(fenceKey));
+ Assertions.assertEquals(0L, manager.getQuota().getGlobalCount());
+ Assertions.assertTrue(manager.getUnresolvedJobs().isEmpty());
+ }
+
+ private static LanceIndexJob newCreateJob(long jobId, String displayName) {
+ return new LanceIndexJob(jobId, "tester", CATALOG_ID, "db1", "tbl1",
+ LanceIndexFenceKey.PROVIDER_DIRECTORY, LOCATOR,
+ displayName, LanceIndexNameNormalizer.normalize(displayName),
+ LanceIndexJobMutationType.CREATE, false, false, "IVF_PQ", "v",
+ null, 7L, null);
+ }
+
+ private static boolean containsJob(List
+ * PENDING -> RUNNING -> COMMITTED
+ * -> NOT_COMMITTED
+ * -> UNKNOWN
+ *
+ * All three outcomes are terminal. UNKNOWN has no outgoing transition:
+ * metadata never changes it to a known outcome, and FORCE_RELEASE (a later
+ * delivery slice) only releases the fence/quota/slot without rewriting the
+ * outcome.
+ */
+public enum LanceIndexJobMutationState {
+ /** The request and fence are durable; no execute send may have occurred. */
+ PENDING,
+ /** The durable dispatch boundary has been crossed; the one-shot call may execute or may already have executed. */
+ RUNNING,
+ /** A complete identity-matched typed success proves this job committed. */
+ COMMITTED,
+ /** A complete trusted result proves this job did not commit. */
+ NOT_COMMITTED,
+ /** Doris cannot safely prove whether this job committed. */
+ UNKNOWN;
+
+ public boolean isTerminal() {
+ return this == COMMITTED || this == NOT_COMMITTED || this == UNKNOWN;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobMutationType.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobMutationType.java
new file mode 100644
index 00000000000000..4d5d3e893ff071
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobMutationType.java
@@ -0,0 +1,28 @@
+// 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.doris.datasource.lance.job;
+
+/**
+ * The one-shot Lance index mutation a durable job intends to perform.
+ * REPLACE is a full same-name rebuild (CREATE OR REPLACE INDEX).
+ */
+public enum LanceIndexJobMutationType {
+ CREATE,
+ REPLACE,
+ DROP
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobQuota.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobQuota.java
new file mode 100644
index 00000000000000..5634dc6f9355a9
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobQuota.java
@@ -0,0 +1,189 @@
+// 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.doris.datasource.lance.job;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Three-level unresolved-job quota counters: per persisted table/locator
+ * identity, per catalog, and globally. An unresolved job is any job that still
+ * holds its same-name fence (PENDING/RUNNING, an unforced UNKNOWN, or a known
+ * terminal job whose required refresh is not DONE); active-plus-UNKNOWN counts
+ * stay bounded at every level.
+ *
+ *
+ * NOT_REQUIRED | REQUIRED -> RUNNING -> DONE | FAILED, FAILED -> RUNNING (retry)
+ *
+ * Refresh replays through the existing idempotent external-table refresh path;
+ * a failed refresh may be retried and still holds the same-name fence.
+ */
+public enum LanceIndexJobRefreshState {
+ /** No refresh is owed (e.g. a proven pre-invocation failure with no relevant metadata change). */
+ NOT_REQUIRED,
+ /** A refresh of the authoritative metadata is owed before the fence may be released. */
+ REQUIRED,
+ /** A refresh attempt is in flight. */
+ RUNNING,
+ /** The required refresh finished; the fence may be released. */
+ DONE,
+ /** The refresh attempt failed; retry through the idempotent path is allowed. */
+ FAILED
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResult.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResult.java
new file mode 100644
index 00000000000000..49782f275b3c2d
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResult.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.doris.datasource.lance.job;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+
+/**
+ * The immutable typed result of the one-shot invocation, persisted on the job.
+ * Contains only what is needed to classify the invocation: the saved typed
+ * code, the classified completion reason, a bounded sanitized message, and
+ * whether trusted pre-invocation revalidation observed relevant external
+ * metadata advancement. Never carries secrets, raw provider responses, or
+ * unbounded text.
+ */
+public class LanceIndexJobResult {
+ public static final int MAX_MESSAGE_BYTES = 1024;
+
+ @SerializedName(value = "rc")
+ private LanceIndexJobResultCode resultCode = LanceIndexJobResultCode.NO_TRUSTED_RESULT;
+
+ @SerializedName(value = "cr")
+ private LanceIndexJobCompletionReason completionReason = LanceIndexJobCompletionReason.NONE;
+
+ @SerializedName(value = "msg")
+ private String sanitizedMessage;
+
+ @SerializedName(value = "ema")
+ private boolean externalMetadataAdvanced;
+
+ /**
+ * No-arg constructor for Gson replay only; missing fields keep the safe
+ * defaults declared above.
+ */
+ public LanceIndexJobResult() {
+ }
+
+ public LanceIndexJobResult(LanceIndexJobResultCode resultCode, LanceIndexJobCompletionReason completionReason,
+ String sanitizedMessage, boolean externalMetadataAdvanced) {
+ this.resultCode = Objects.requireNonNull(resultCode, "resultCode");
+ this.completionReason = completionReason == null ? LanceIndexJobCompletionReason.NONE : completionReason;
+ this.sanitizedMessage = checkMessageBytes(sanitizedMessage);
+ this.externalMetadataAdvanced = externalMetadataAdvanced;
+ }
+
+ private static String checkMessageBytes(String message) {
+ if (message != null && message.getBytes(StandardCharsets.UTF_8).length > MAX_MESSAGE_BYTES) {
+ throw new IllegalArgumentException(
+ "sanitized message exceeds " + MAX_MESSAGE_BYTES + " UTF-8 bytes");
+ }
+ return message;
+ }
+
+ public LanceIndexJobResultCode getResultCode() {
+ return resultCode;
+ }
+
+ public LanceIndexJobCompletionReason getCompletionReason() {
+ return completionReason;
+ }
+
+ public String getSanitizedMessage() {
+ return sanitizedMessage;
+ }
+
+ public boolean isExternalMetadataAdvanced() {
+ return externalMetadataAdvanced;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof LanceIndexJobResult)) {
+ return false;
+ }
+ LanceIndexJobResult that = (LanceIndexJobResult) o;
+ return externalMetadataAdvanced == that.externalMetadataAdvanced
+ && resultCode == that.resultCode
+ && completionReason == that.completionReason
+ && Objects.equals(sanitizedMessage, that.sanitizedMessage);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resultCode, completionReason, sanitizedMessage, externalMetadataAdvanced);
+ }
+
+ @Override
+ public String toString() {
+ return "LanceIndexJobResult{resultCode=" + resultCode + ", completionReason=" + completionReason
+ + ", externalMetadataAdvanced=" + externalMetadataAdvanced + '}';
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResultCode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResultCode.java
new file mode 100644
index 00000000000000..fad096b5087212
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexJobResultCode.java
@@ -0,0 +1,158 @@
+// 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.doris.datasource.lance.job;
+
+/**
+ * Typed result codes of the one-shot Lance index invocation. Only saved typed
+ * codes are ever classified; message text is never inspected to infer an
+ * outcome. Codes fall into three groups:
+ *
+ *
+ *
+ */
+public enum LanceIndexJobResultCode {
+ /** Pre-invocation: admitted dataset version / schema contract no longer matches. */
+ PRE_INVOCATION_STALE_ADMISSION,
+ /** Pre-invocation: the recomputed contract is not a supported contract v1. */
+ PRE_INVOCATION_UNSUPPORTED_SCHEMA_CONTRACT,
+ /** Pre-invocation: credentials are known to be expired. */
+ PRE_INVOCATION_CREDENTIAL_EXPIRED,
+ /** Pre-invocation: trusted busy / pre-FFI resource rejection. */
+ PRE_INVOCATION_RESOURCE_REJECTED,
+ /** LANCE_OK from the one native invocation. */
+ NATIVE_OK,
+ /** LANCE_ERR_COMMIT_CONFLICT: the commit lost a race; the external dataset advanced. */
+ NATIVE_COMMIT_CONFLICT,
+ /** LANCE_ERR_NOT_FOUND after invocation. */
+ NATIVE_NOT_FOUND,
+ /** LANCE_ERR_INVALID_ARGUMENT after invocation. */
+ NATIVE_INVALID_ARGUMENT,
+ /** LANCE_ERR_NOT_SUPPORTED after invocation. */
+ NATIVE_NOT_SUPPORTED,
+ /** LANCE_ERR_INDEX after invocation (also the coarse duplicate-CREATE error). */
+ NATIVE_INDEX,
+ /** LANCE_ERR_IO after invocation. */
+ NATIVE_IO,
+ /** LANCE_ERR_INTERNAL after invocation. */
+ NATIVE_INTERNAL,
+ /** No complete trusted result exists; commitment cannot be proven either way. */
+ NO_TRUSTED_RESULT;
+
+ public boolean isPreInvocation() {
+ return this == PRE_INVOCATION_STALE_ADMISSION
+ || this == PRE_INVOCATION_UNSUPPORTED_SCHEMA_CONTRACT
+ || this == PRE_INVOCATION_CREDENTIAL_EXPIRED
+ || this == PRE_INVOCATION_RESOURCE_REJECTED;
+ }
+
+ /**
+ * Classify a complete saved result into the durable (mutationState, refreshState,
+ * completionReason) triple, following the provider-result classification table:
+ *
+ *
+ *
+ */
+ public static Classification classify(LanceIndexJobMutationType mutationType, LanceIndexJobResultCode resultCode,
+ boolean ifExists, boolean externalMetadataAdvanced) {
+ if (resultCode == null) {
+ resultCode = NO_TRUSTED_RESULT;
+ }
+ switch (resultCode) {
+ case PRE_INVOCATION_STALE_ADMISSION:
+ case PRE_INVOCATION_UNSUPPORTED_SCHEMA_CONTRACT:
+ case PRE_INVOCATION_CREDENTIAL_EXPIRED:
+ case PRE_INVOCATION_RESOURCE_REJECTED:
+ return new Classification(LanceIndexJobMutationState.NOT_COMMITTED,
+ externalMetadataAdvanced
+ ? LanceIndexJobRefreshState.REQUIRED : LanceIndexJobRefreshState.NOT_REQUIRED,
+ LanceIndexJobCompletionReason.NONE);
+ case NATIVE_OK:
+ return new Classification(LanceIndexJobMutationState.COMMITTED, LanceIndexJobRefreshState.REQUIRED,
+ LanceIndexJobCompletionReason.NONE);
+ case NATIVE_COMMIT_CONFLICT:
+ return new Classification(LanceIndexJobMutationState.NOT_COMMITTED, LanceIndexJobRefreshState.REQUIRED,
+ LanceIndexJobCompletionReason.NONE);
+ case NATIVE_NOT_FOUND:
+ if (mutationType == LanceIndexJobMutationType.DROP) {
+ return new Classification(LanceIndexJobMutationState.NOT_COMMITTED,
+ LanceIndexJobRefreshState.REQUIRED,
+ ifExists ? LanceIndexJobCompletionReason.IF_CONDITION_NOOP
+ : LanceIndexJobCompletionReason.NONE);
+ }
+ return new Classification(LanceIndexJobMutationState.UNKNOWN, LanceIndexJobRefreshState.NOT_REQUIRED,
+ LanceIndexJobCompletionReason.NONE);
+ case NATIVE_INVALID_ARGUMENT:
+ case NATIVE_NOT_SUPPORTED:
+ case NATIVE_INDEX:
+ case NATIVE_IO:
+ case NATIVE_INTERNAL:
+ case NO_TRUSTED_RESULT:
+ default:
+ return new Classification(LanceIndexJobMutationState.UNKNOWN, LanceIndexJobRefreshState.NOT_REQUIRED,
+ LanceIndexJobCompletionReason.NONE);
+ }
+ }
+
+ /**
+ * The immutable outcome of {@link #classify}: the durable mutation state, the
+ * independent refresh obligation, and the completion reason.
+ */
+ public static final class Classification {
+ private final LanceIndexJobMutationState mutationState;
+ private final LanceIndexJobRefreshState refreshState;
+ private final LanceIndexJobCompletionReason completionReason;
+
+ private Classification(LanceIndexJobMutationState mutationState, LanceIndexJobRefreshState refreshState,
+ LanceIndexJobCompletionReason completionReason) {
+ this.mutationState = mutationState;
+ this.refreshState = refreshState;
+ this.completionReason = completionReason;
+ }
+
+ public LanceIndexJobMutationState getMutationState() {
+ return mutationState;
+ }
+
+ public LanceIndexJobRefreshState getRefreshState() {
+ return refreshState;
+ }
+
+ public LanceIndexJobCompletionReason getCompletionReason() {
+ return completionReason;
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexNameNormalizer.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexNameNormalizer.java
new file mode 100644
index 00000000000000..b467aa287f7f19
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexNameNormalizer.java
@@ -0,0 +1,69 @@
+// 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.doris.datasource.lance.job;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Locale;
+
+/**
+ * Logical index name normalization v1, the only definition: the UTF-8 result
+ * of Java {@code toLowerCase(Locale.ROOT)}. Both the display name and the
+ * normalized bytes are persisted on the job; Doris preserves display case,
+ * rejects new case-only duplicates, and fails mutation on ambiguous external
+ * case-only collisions. No normalization migration or mixed-version fence
+ * protocol exists.
+ */
+public final class LanceIndexNameNormalizer {
+ /** Bound on the persisted logical index name, aligned with the external string bound. */
+ public static final int MAX_INDEX_NAME_BYTES = 1024;
+
+ private LanceIndexNameNormalizer() {
+ }
+
+ /**
+ * Normalization v1. The result is the identity bytes of the same-name fence key.
+ */
+ public static String normalize(String displayName) {
+ if (displayName == null) {
+ throw new IllegalArgumentException("index name must not be null");
+ }
+ return displayName.toLowerCase(Locale.ROOT);
+ }
+
+ /**
+ * True when two display names differ only by case under normalization v1.
+ */
+ public static boolean isCaseOnlyDuplicate(String displayA, String displayB) {
+ if (displayA == null || displayB == null) {
+ return false;
+ }
+ return !displayA.equals(displayB) && normalize(displayA).equals(normalize(displayB));
+ }
+
+ /**
+ * Validate the persisted display name: non-empty and within the UTF-8 byte bound.
+ */
+ public static void validateDisplayName(String displayName) {
+ if (displayName == null || displayName.isEmpty()) {
+ throw new IllegalArgumentException("index name must not be null or empty");
+ }
+ if (displayName.getBytes(StandardCharsets.UTF_8).length > MAX_INDEX_NAME_BYTES) {
+ throw new IllegalArgumentException("index name exceeds " + MAX_INDEX_NAME_BYTES + " UTF-8 bytes");
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexSchemaContract.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexSchemaContract.java
new file mode 100644
index 00000000000000..1175de7223da46
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/lance/job/LanceIndexSchemaContract.java
@@ -0,0 +1,250 @@
+// 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.doris.datasource.lance.job;
+
+import com.google.gson.annotations.SerializedName;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Persisted representation of schema contract v1: the ordered list of indexed
+ * fields captured at admission. Before native invocation the worker reopens
+ * the admitted dataset version, independently recomputes contract v1, and
+ * compares the ordered representation; a mismatch or unavailable version is a
+ * complete pre-invocation NOT_COMMITTED result. Equality is order-sensitive.
+ *
+ *