github-actions[bot] commented on code in PR #67630:
URL: https://github.com/apache/doris/pull/67630#discussion_r3985759769
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -4226,4 +4226,44 @@ public void handle(Field field, String value) throws
Exception {
+ "可降低导入内存压力并提升随机分桶表的吞吐量，覆盖所有导入类型。"})
public static boolean enable_adaptive_random_bucket_load = true;
+ @ConfField(mutable = true, masterOnly = true, varType =
VariableAnnotation.EXPERIMENTAL, description = {
Review Comment:
[P1] Keep this slice impossible to enable in production. The accepted v5.1
contract (sections 9.1 and 9.7) says this is disabled FE mutation control and
mutation cannot be enabled until hard worker isolation, finite possible-live
limits, and the remaining evidence gates exist. `mutable=true` (and ordinary
fe.conf loading) instead permits admission now; the warning confirms every
resulting job/fence is durable but unresolvable because dispatch and
FORCE_RELEASE are absent. Please use a test-only/internal seam for admission
coverage and expose an enableable runtime setting only with the complete safety
boundary.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java:
##########
@@ -87,6 +90,17 @@ public class CatalogMgr implements Writable,
GsonPostProcessable {
public static final String METADATA_REFRESH_INTERVAL_SEC =
"metadata_refresh_interval_sec";
public static final String CATALOG_TYPE_PROP = "type";
+ /**
+ * The Lance catalog properties whose change moves the persisted target
identity
+ * (provider, stable locator, or namespace mapping) out from under
unresolved index
+ * jobs. Credential and other properties are not target-changing and stay
unguarded.
+ * Keys match case-insensitively because the catalog property chain
performs no key
+ * normalization.
+ */
+ private static final Set<String> LANCE_TARGET_IDENTITY_KEYS =
ImmutableSet.of(
Review Comment:
[P1] Include storage-routing properties in the target identity guard.
`s3.endpoint` (and the OSS equivalent) is not a credential: changing it can
make the same `s3://bucket/path` resolve on a different MinIO/S3 service. Jobs
persist the URI but not the endpoint, and later reads/workers rebuild options
from the current catalog, while this five-key fingerprint also misses the
change during in-flight admission. Please fingerprint/guard the normalized
effective non-secret routing identity (including every accepted endpoint alias,
provider selector, and region-derived endpoint), while leaving access
keys/tokens rotatable.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -4226,4 +4226,44 @@ public void handle(Field field, String value) throws
Exception {
+ "可降低导入内存压力并提升随机分桶表的吞吐量，覆盖所有导入类型。"})
public static boolean enable_adaptive_random_bucket_load = true;
+ @ConfField(mutable = true, masterOnly = true, varType =
VariableAnnotation.EXPERIMENTAL, description = {
+ "是否启用 Lance 外表索引变更(CREATE/CREATE OR REPLACE/DROP INDEX)的
admission。默认关闭;"
+ + "启用前需确认未决 job 配额均为正值。注意:在 dispatch(后续版本)与
FORCE_RELEASE(后续版本)就绪前"
+ + "开启本开关会产生不可回收的 PENDING job 并冻结对应 catalog 的身份属性变更与 DROP
CATALOG。",
+ "Enable admission of Lance index mutations (CREATE/CREATE OR
REPLACE/DROP INDEX). "
+ + "Disabled by default; unresolved-job quotas must be
positive before enabling. "
+ + "WARNING: enabling before dispatch and FORCE_RELEASE
land in a later release creates "
+ + "PENDING jobs that cannot be resolved and freezes
catalog identity changes and DROP CATALOG."})
+ public static boolean enable_lance_index_mutation = false;
+
+ @ConfField(mutable = true, masterOnly = true,
+ callback =
LanceIndexConfigValidator.PositiveLongConfigHandler.class,
+ description = {"单个 Lance 数据表(locator 身份)允许的最大未决索引 job 数。",
+ "Max unresolved Lance index jobs per table (locator
identity)."})
+ public static long lance_index_job_max_unresolved_per_table = 8;
+
+ @ConfField(mutable = true, masterOnly = true,
+ callback =
LanceIndexConfigValidator.PositiveLongConfigHandler.class,
+ description = {"单个 Lance catalog 允许的最大未决索引 job 数。",
+ "Max unresolved Lance index jobs per catalog."})
+ public static long lance_index_job_max_unresolved_per_catalog = 64;
+
+ @ConfField(mutable = true, masterOnly = true,
+ callback =
LanceIndexConfigValidator.PositiveLongConfigHandler.class,
+ description = {"全部 catalog 合计允许的最大未决 Lance 索引 job 数。",
+ "Max unresolved Lance index jobs across all catalogs
(global)."})
+ public static long lance_index_job_max_unresolved_global = 256;
+
+ @ConfField(mutable = true, masterOnly = false,
Review Comment:
[P2] These admission limits should route to the admission owner. Lance
mutation validation runs only after `AlterTableCommand` is forwarded to the
master, but `masterOnly=false` lets `ADMIN SET FRONTEND CONFIG` on a follower
succeed locally without changing what the master enforces. Mark both ANN bounds
master-only (while retaining the explicit all-FE form when desired), or
otherwise make the update cluster-wide, and test follower routing instead of
locking in the false annotation.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceIndexAdmission.java:
##########
@@ -0,0 +1,504 @@
+// 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;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator;
+import org.apache.doris.datasource.lance.job.LanceIndexFenceKey;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationType;
+import org.apache.doris.datasource.lance.job.LanceIndexNameNormalizer;
+import org.apache.doris.datasource.lance.job.LanceIndexSchemaContract;
+import org.apache.doris.nereids.trees.plans.commands.info.IndexDefinition;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+
+/**
+ * Lance index admission (design sections 2.2 and 4.1): the single place where
a statically
+ * validated top-level CREATE [OR REPLACE]/DROP INDEX statement against a
Lance catalog table is
+ * turned into a durable job. The whole flow runs against one pinned admission
snapshot — the
+ * IF preflight never takes a second metadata read, and no catalog/db/table
metadata lock is held
+ * while the snapshot loader does its JNI work (design section 5.1).
+ *
+ * <p>The step order is the correctness contract: name normalization and
reserved prefix come
+ * first, before target capture and the snapshot read (fail cheap-first — a
reserved name
+ * rejected at admission depth costs no remote read), then case-only collision
analysis, IF
+ * preflight (including the two-stage {@code matches}: requested-algorithm
equality plus
+ * physical-family corroboration), schema contract from the stored column
name, locator
+ * normalization, deterministic properties JSON, positive-quota assertion, and
only then exactly
+ * one id allocation and the durable {@code createJob} transfer.
+ * Every rejection before {@code createJob} leaves no job, no fence, no quota
charge, no journal
+ * record, and no id allocation; the manager's own fence/quota rejections pass
through verbatim
+ * (an id burned by them is accepted — ids are never required to be
contiguous).
+ */
+public final class LanceIndexAdmission {
+
+ /**
+ * The snapshot read seam. Tests inject a prepared snapshot here so
admission runs without
+ * FE startup or JNI; the production default delegates to the catalog's
merged-snapshot read.
+ */
+ public interface SnapshotLoader {
+ LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog, String
dbName, String tableName)
+ throws Exception;
+ }
+
+ /** The admission result: the durable job id, or null for an IF no-op (no
job created). */
+ public static final class Outcome {
+ private final Long jobId;
+
+ private Outcome(Long jobId) {
+ this.jobId = jobId;
+ }
+
+ /**
+ * The admitted job id, or null when the IF preflight made the
statement an immediate
+ * no-op (design section 2.2: "returns an immediate no-op, without
creating a job").
+ */
+ @Nullable
+ public Long getJobId() {
+ return jobId;
+ }
+ }
+
+ private static final SnapshotLoader DEFAULT_LOADER = new SnapshotLoader() {
+ @Override
+ public LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog,
String dbName,
+ String tableName) throws Exception {
+ return catalog.loadTableIndexAdmissionSnapshot(dbName, tableName);
+ }
+ };
+
+ private LanceIndexAdmission() {
+ }
+
+ /**
+ * Admits a top-level CREATE [OR REPLACE] INDEX. Static validation
+ * ({@link LanceIndexMutationValidator#validateCreateIndex}) must already
have passed for
+ * {@code def}.
+ */
+ public static Outcome admitCreate(LanceExternalCatalog catalog,
LanceExternalDatabase db,
+ LanceExternalTable table, IndexDefinition def, boolean
ifNotExists) throws Exception {
+ return admitCreate(DEFAULT_LOADER, catalog, db, table, def,
ifNotExists);
+ }
+
+ static Outcome admitCreate(SnapshotLoader loader, LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table,
IndexDefinition def, boolean ifNotExists)
+ throws Exception {
+ // 1. Display/normalized names and the reserved system prefix, checked
before any metadata
+ // read so a reserved name rejected at admission depth costs no remote
snapshot read (fail
+ // cheap-first). The prefix is rejected for CREATE and REPLACE exactly
as for DROP; the
+ // static layer rejects it first and this is the defense-in-depth copy
at admission depth.
+ String displayName = def.getIndexName();
+ String normalizedName =
LanceIndexNameNormalizer.normalize(displayName);
+ LanceIndexMutationValidator.rejectIfReservedIndexName(displayName);
+ // 2. One pinned snapshot for every authoritative decision below.
+ CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
+ CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
+ LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
Review Comment:
[P1] Reject ambiguous top-level field identities in this pinned snapshot
before the IF preflight or contract selection. Lance accepts distinct `A` and
`a`, but `ExternalTable.getColumn()` returns the first `equalsIgnoreCase`
match; static validation and `storedColumnName()` can therefore select and
journal the wrong field ID/type, while CREATE IF may falsely match before the
builder runs. Detect collisions under the actual Doris lookup relation as well
as the persisted normalization (ROOT-lowercasing alone misses some Unicode
`equalsIgnoreCase` pairs), fail closed, and cover ASCII and quoted-Unicode
cases.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSchemaContractBuilder.java:
##########
@@ -0,0 +1,236 @@
+// 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;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.lance.job.LanceIndexSchemaContract;
+
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.lance.schema.LanceField;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.regex.Pattern;
+
+/**
+ * Builds schema contract v1 for one indexed column from the fresh LanceField
tree of the
+ * pinned admission snapshot. {@link LanceTypeConverter} is deliberately
bypassed because it
+ * erases fixed-size-list dimensions, float16-vs-float32, and timestamp
timezones.
+ *
+ * <p>The input is the stored column name (byte-identical to the LanceField
name), never the
+ * raw user input: callers resolve it via the table's column lookup first.
Matching is exact
+ * and top-level only; a missing field fails closed. The builder makes no
supportability
+ * judgment — every ArrowType yields a deterministic canonical string, so the
only failure is
+ * the indexed field not being found. The canonical vocabulary defined here is
the Java-side
+ * authority the Rust worker's golden fixtures align to (design section 4.2).
+ */
+final class LanceSchemaContractBuilder {
+ /** Timezones that fit the canonical {@code tz="…"} slot without any
escaping (IANA names). */
+ private static final Pattern SAFE_TIMEZONE =
Pattern.compile("[A-Za-z0-9+_/-]+");
+
+ private LanceSchemaContractBuilder() {
+ }
+
+ /**
+ * Builds the single-field contract for {@code storedColumnName}. Only
top-level fields are
+ * considered; nested subfields never enter the contract.
+ */
+ static LanceIndexSchemaContract build(List<LanceField> topLevelFields,
String storedColumnName)
+ throws AnalysisException {
+ if (topLevelFields == null) {
+ throw new IllegalArgumentException("Lance top-level schema fields
must not be null");
+ }
+ if (storedColumnName == null || storedColumnName.isEmpty()) {
+ throw new IllegalArgumentException("stored column name must not be
null or empty");
+ }
+ for (LanceField field : topLevelFields) {
+ if (field == null) {
+ throw new IllegalArgumentException("Lance top-level schema
field must not be null");
+ }
+ if (storedColumnName.equals(field.getName())) {
+ return new LanceIndexSchemaContract(
+ Collections.singletonList(indexedField(field)));
+ }
+ }
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_LANCE_INDEX_INVALID,
+ "unsupported schema contract: indexed field not found");
+ throw new IllegalStateException("unreachable");
+ }
+
+ private static LanceIndexSchemaContract.IndexedField
indexedField(LanceField field) {
+ ArrowType type = field.getType();
+ if (type == null) {
+ throw new IllegalArgumentException("Lance field type must not be
null");
+ }
+ Integer fixedSizeListDimension = null;
+ String vectorElementType = null;
+ Boolean vectorElementNullable = null;
+ if (type instanceof ArrowType.FixedSizeList) {
+ fixedSizeListDimension = ((ArrowType.FixedSizeList)
type).getListSize();
+ List<LanceField> children = field.getChildren();
+ if (children == null || children.size() != 1 || children.get(0) ==
null) {
+ throw new IllegalArgumentException(
+ "Lance fixed-size list field must have exactly one
child");
+ }
+ LanceField element = children.get(0);
+ vectorElementType = canonicalType(element.getType());
+ vectorElementNullable = element.isNullable();
+ }
+ return new LanceIndexSchemaContract.IndexedField(
Review Comment:
[P2] Validate provider-supplied numeric schema facts before journaling this
contract. The pinned manifest-read path reconstructs protobuf fields without
calling `Schema.validate()`, so a corrupt or historical manifest can reach Java
with `field.getId() < 0` or a fixed-list size `<= 0`; neither this builder nor
`IndexedField.validateForAdmission()` rejects them. Reject negative IDs (zero
is valid) and non-positive dimensions here, map the failure to the bounded
Lance admission error, and add manifest-shaped cases.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceIndexMetadataLoader.java:
##########
@@ -151,6 +152,86 @@ static List<LancePhysicalIndexEntry>
collectPhysicalEntries(Dataset dataset) {
return Collections.unmodifiableList(entries);
}
+ /**
+ * Loads one pinned latest-snapshot view of everything index admission
needs: the dataset
+ * version, the top-level schema fields, the logical indexes, and the
physical entries with
+ * their index types — all from a single {@code Dataset.open} so the
pieces cannot drift
+ * across snapshots (design section 3.1). Never calls {@code countRows()}
or
+ * {@code getIndexStatistics()}.
+ */
+ public static LanceIndexAdmissionSnapshot loadAdmissionSnapshot(String
datasetUri,
+ Map<String, String> javaStorageOptions, BufferAllocator allocator)
throws Exception {
+ try (Dataset dataset = openLatestDataset(datasetUri,
javaStorageOptions, allocator)) {
+ long datasetVersion = dataset.version();
+ List<LanceField> topLevelFields =
dataset.getLanceSchema().fields();
+ List<LanceLogicalIndex> logicalIndexes =
+ normalize(describeUserIndexes(dataset),
buildFieldNamesById(topLevelFields));
+ List<LanceIndexAdmissionSnapshot.PhysicalIndexInfo>
physicalIndexes =
+ collectPhysicalIndexInfos(dataset);
+ // LanceField is a pure POJO, so the materialized field list can
leave the open block
+ // with the snapshot; nothing here retains the Dataset or its
allocator.
+ return new LanceIndexAdmissionSnapshot(
+ datasetVersion, datasetUri, logicalIndexes,
physicalIndexes, topLevelFields);
+ }
+ }
+
+ private static Dataset openLatestDataset(String datasetUri,
+ Map<String, String> javaStorageOptions, BufferAllocator allocator)
{
+ return Dataset.open().allocator(allocator).uri(datasetUri)
+ .readOptions(LanceReadOptions.build(javaStorageOptions,
OptionalLong.empty())).build();
+ }
+
+ /**
+ * Collects the physical entries of the opened snapshot, applying the same
defenses as the
+ * logical path: the raw list is bounded before per-entry validation,
system entries are
+ * validated then filtered out, and duplicate UUID ownership fails closed.
+ */
+ static List<LanceIndexAdmissionSnapshot.PhysicalIndexInfo>
collectPhysicalIndexInfos(
+ Dataset dataset) {
+ List<Index> indexes = dataset.getIndexes();
+ if (indexes == null) {
+ throw new IllegalArgumentException("Lance physical index entries
must not be null");
+ }
+ if (indexes.size() > MAX_PHYSICAL_INDEX_ENTRIES) {
+ throw new IllegalArgumentException(
+ "Lance physical index entry count exceeds limit "
+ + MAX_PHYSICAL_INDEX_ENTRIES);
+ }
+
+ List<LanceIndexAdmissionSnapshot.PhysicalIndexInfo> entries = new
ArrayList<>(indexes.size());
+ Set<String> uuids = new HashSet<>();
+ for (Index index : indexes) {
+ if (index == null) {
+ throw new IllegalArgumentException("Lance physical index entry
must not be null");
+ }
+ String name = requireExternalString(index.name(), "Lance physical
index name");
+ if (index.uuid() == null) {
+ throw new IllegalArgumentException("Lance physical index uuid
must not be null");
+ }
+ String uuid = index.uuid().toString();
+ long indexDatasetVersion = index.datasetVersion();
+ if (indexDatasetVersion <= 0) {
+ throw new IllegalArgumentException(
+ "Lance physical index dataset version must be
positive");
+ }
+ IndexType indexType = index.indexType();
+ if (indexType == null) {
+ throw new IllegalArgumentException("Lance physical index type
must not be null");
+ }
+ if (SYSTEM_INDEX_NAMES.contains(name)) {
Review Comment:
[P2] Record the UUID before filtering system entries. As written,
`__lance_frag_reuse` or `__lance_mem_wal` can share a UUID with a later user
entry because the system entry reaches `continue` before `uuids.add(uuid)`.
That contradicts this method's fail-closed ownership contract and the existing
`collectPhysicalEntries` path, which checks UUID uniqueness before filtering.
Please move the duplicate check above this branch and cover a system/user
collision.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceIndexAdmission.java:
##########
@@ -0,0 +1,504 @@
+// 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;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator;
+import org.apache.doris.datasource.lance.job.LanceIndexFenceKey;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationType;
+import org.apache.doris.datasource.lance.job.LanceIndexNameNormalizer;
+import org.apache.doris.datasource.lance.job.LanceIndexSchemaContract;
+import org.apache.doris.nereids.trees.plans.commands.info.IndexDefinition;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+
+/**
+ * Lance index admission (design sections 2.2 and 4.1): the single place where
a statically
+ * validated top-level CREATE [OR REPLACE]/DROP INDEX statement against a
Lance catalog table is
+ * turned into a durable job. The whole flow runs against one pinned admission
snapshot — the
+ * IF preflight never takes a second metadata read, and no catalog/db/table
metadata lock is held
+ * while the snapshot loader does its JNI work (design section 5.1).
+ *
+ * <p>The step order is the correctness contract: name normalization and
reserved prefix come
+ * first, before target capture and the snapshot read (fail cheap-first — a
reserved name
+ * rejected at admission depth costs no remote read), then case-only collision
analysis, IF
+ * preflight (including the two-stage {@code matches}: requested-algorithm
equality plus
+ * physical-family corroboration), schema contract from the stored column
name, locator
+ * normalization, deterministic properties JSON, positive-quota assertion, and
only then exactly
+ * one id allocation and the durable {@code createJob} transfer.
+ * Every rejection before {@code createJob} leaves no job, no fence, no quota
charge, no journal
+ * record, and no id allocation; the manager's own fence/quota rejections pass
through verbatim
+ * (an id burned by them is accepted — ids are never required to be
contiguous).
+ */
+public final class LanceIndexAdmission {
+
+ /**
+ * The snapshot read seam. Tests inject a prepared snapshot here so
admission runs without
+ * FE startup or JNI; the production default delegates to the catalog's
merged-snapshot read.
+ */
+ public interface SnapshotLoader {
+ LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog, String
dbName, String tableName)
+ throws Exception;
+ }
+
+ /** The admission result: the durable job id, or null for an IF no-op (no
job created). */
+ public static final class Outcome {
+ private final Long jobId;
+
+ private Outcome(Long jobId) {
+ this.jobId = jobId;
+ }
+
+ /**
+ * The admitted job id, or null when the IF preflight made the
statement an immediate
+ * no-op (design section 2.2: "returns an immediate no-op, without
creating a job").
+ */
+ @Nullable
+ public Long getJobId() {
+ return jobId;
+ }
+ }
+
+ private static final SnapshotLoader DEFAULT_LOADER = new SnapshotLoader() {
+ @Override
+ public LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog,
String dbName,
+ String tableName) throws Exception {
+ return catalog.loadTableIndexAdmissionSnapshot(dbName, tableName);
+ }
+ };
+
+ private LanceIndexAdmission() {
+ }
+
+ /**
+ * Admits a top-level CREATE [OR REPLACE] INDEX. Static validation
+ * ({@link LanceIndexMutationValidator#validateCreateIndex}) must already
have passed for
+ * {@code def}.
+ */
+ public static Outcome admitCreate(LanceExternalCatalog catalog,
LanceExternalDatabase db,
+ LanceExternalTable table, IndexDefinition def, boolean
ifNotExists) throws Exception {
+ return admitCreate(DEFAULT_LOADER, catalog, db, table, def,
ifNotExists);
+ }
+
+ static Outcome admitCreate(SnapshotLoader loader, LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table,
IndexDefinition def, boolean ifNotExists)
+ throws Exception {
+ // 1. Display/normalized names and the reserved system prefix, checked
before any metadata
+ // read so a reserved name rejected at admission depth costs no remote
snapshot read (fail
+ // cheap-first). The prefix is rejected for CREATE and REPLACE exactly
as for DROP; the
+ // static layer rejects it first and this is the defense-in-depth copy
at admission depth.
+ String displayName = def.getIndexName();
+ String normalizedName =
LanceIndexNameNormalizer.normalize(displayName);
+ LanceIndexMutationValidator.rejectIfReservedIndexName(displayName);
+ // 2. One pinned snapshot for every authoritative decision below.
+ CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
+ CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
+ LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
+ // 3. Case-only analysis (design section 4.1): ambiguous external
collisions fail closed;
+ // a unique match resolves to the stored display name.
+ List<String> storedNames = logicalIndexNames(snapshot);
+ if (LanceIndexFamilies.isAmbiguousCaseCollision(storedNames,
normalizedName)) {
+ rejectInvalid("index name '" + displayName
+ + "' is ambiguous: multiple Lance indexes differ only by
case");
+ }
+ String storedName = LanceIndexFamilies.uniqueMatch(storedNames,
normalizedName);
+ // 4. IF preflight (design section 2.2).
+ boolean orReplace = def.isOrReplace();
+ if (!orReplace && storedName != null) {
+ if (!ifNotExists) {
+ rejectInvalid("index '" + displayName + "' already exists");
+ }
+ if (!matchesExistingDefinition(snapshot, storedName, def)) {
+ rejectInvalid("index '" + displayName + "' already exists with
a different definition");
+ }
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () ->
new Outcome(null));
+ }
+ // 5. Schema contract v1 from the stored column name (never the raw
user spelling).
+ String storedColumnName = storedColumnName(table,
def.getCols().get(0));
+ LanceIndexSchemaContract contract =
+ LanceSchemaContractBuilder.build(snapshot.getTopLevelFields(),
storedColumnName);
+ // 6. The fence locator is the normalized dataset uri of the same
pinned snapshot.
+ String locator = normalizeLocator(snapshot);
+ // 7. Deterministic normalized properties JSON for ANN; scalar
families persist null.
+ boolean ann = def.getLanceIndexType() == null;
+ String indexType = ann ? annIndexType(def) : def.getLanceIndexType();
+ String propertiesJson = ann ? buildAnnPropertiesJson(def) : null;
+ // D7 backstop: quota values from fe.conf bypass the ADMIN SET
callback, so admission
+ // re-asserts positivity before any id allocation or durable transfer.
+ assertPositiveQuotas();
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> {
+ // 8. Exactly one id allocation, after every preflight above has
passed.
+ long jobId = Env.getCurrentEnv().getNextId();
+ String creator = ConnectContext.get().getQualifiedUser();
+ // 9. REPLACE on an existing name persists the stored display name
(section 4.1) so the
+ // worker locates the case-sensitive target; a fresh REPLACE keeps
the user's spelling.
+ String persistedDisplayName = (orReplace && storedName != null) ?
storedName : displayName;
+ LanceIndexJob job;
+ try {
+ job = new LanceIndexJob(jobId, creator, catalog.getId(),
db.getFullName(), table.getName(),
+ LanceIndexFenceKey.PROVIDER_DIRECTORY, locator,
persistedDisplayName, normalizedName,
+ orReplace ? LanceIndexJobMutationType.REPLACE :
LanceIndexJobMutationType.CREATE,
+ ifNotExists, false, indexType, storedColumnName,
propertiesJson,
+ snapshot.getDatasetVersion(), contract);
+ } catch (IllegalArgumentException e) {
+ throw invalidAdmission(e.getMessage());
+ }
+ Env.getCurrentEnv().getLanceIndexJobManager().createJob(job,
+ Config.lance_index_job_max_unresolved_per_table,
+ Config.lance_index_job_max_unresolved_per_catalog,
+ Config.lance_index_job_max_unresolved_global);
+ // 10. The job and its fence are durable once createJob returns.
+ return new Outcome(jobId);
+ });
+ }
+
+ /**
+ * Admits a top-level DROP INDEX. The static name bounds
+ * ({@link LanceIndexMutationValidator#validateDropIndex}) must already
have passed.
+ */
+ public static Outcome admitDrop(LanceExternalCatalog catalog,
LanceExternalDatabase db,
+ LanceExternalTable table, String indexName, boolean ifExists)
throws Exception {
+ return admitDrop(DEFAULT_LOADER, catalog, db, table, indexName,
ifExists);
+ }
+
+ static Outcome admitDrop(SnapshotLoader loader, LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table, String
indexName, boolean ifExists)
+ throws Exception {
+ // Fail cheap-first: the reserved prefix is rejected before target
capture and the
+ // snapshot read, so it costs no remote read.
+ String normalizedName = LanceIndexNameNormalizer.normalize(indexName);
+ LanceIndexMutationValidator.rejectIfReservedIndexName(indexName);
+ CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
+ CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
+ LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
+ List<String> storedNames = logicalIndexNames(snapshot);
+ if (LanceIndexFamilies.isAmbiguousCaseCollision(storedNames,
normalizedName)) {
+ rejectInvalid("index name '" + indexName
+ + "' is ambiguous: multiple Lance indexes differ only by
case");
+ }
+ String storedName = LanceIndexFamilies.uniqueMatch(storedNames,
normalizedName);
+ if (storedName == null) {
+ if (ifExists) {
+ return catalogMgr.withLanceIndexAdmission(catalog, target, ()
-> new Outcome(null));
+ }
+ rejectInvalid("index '" + indexName + "' not found");
+ }
+ String locator = normalizeLocator(snapshot);
+ assertPositiveQuotas();
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> {
+ long jobId = Env.getCurrentEnv().getNextId();
+ String creator = ConnectContext.get().getQualifiedUser();
+ // DROP only runs past the preflight with a unique match, so the
stored display name is
+ // always persisted (section 4.1); definition fields stay null on
a DROP job record.
+ LanceIndexJob job;
+ try {
+ job = new LanceIndexJob(jobId, creator, catalog.getId(),
db.getFullName(), table.getName(),
+ LanceIndexFenceKey.PROVIDER_DIRECTORY, locator,
storedName, normalizedName,
+ LanceIndexJobMutationType.DROP, false, ifExists, null,
null, null,
+ snapshot.getDatasetVersion(), null);
+ } catch (IllegalArgumentException e) {
+ throw invalidAdmission(e.getMessage());
+ }
+ Env.getCurrentEnv().getLanceIndexJobManager().createJob(job,
+ Config.lance_index_job_max_unresolved_per_table,
+ Config.lance_index_job_max_unresolved_per_catalog,
+ Config.lance_index_job_max_unresolved_global);
+ return new Outcome(jobId);
+ });
+ }
+
+ /**
+ * The section 2.2 definition match, two stages: (a) the requested
algorithm must equal the
+ * stored logical algorithm under family normalization — a same-name
different-algorithm
+ * request is a mismatch, never a no-op; (b) the physical entry of the
same name must exist
+ * and back the logical algorithm (snapshot self-consistency, failing
closed); (c) the single
+ * normalized column must be equal; (d) whitelist properties are compared
per property — a
+ * value the request sets and the snapshot exposes must be equal, an
unexposed snapshot value
+ * is skipped, and a property the request omits is never compared.
+ */
+ private static boolean
matchesExistingDefinition(LanceIndexAdmissionSnapshot snapshot,
+ String storedName, IndexDefinition def) {
+ LanceLogicalIndex logical = null;
+ for (LanceLogicalIndex index : snapshot.getLogicalIndexes()) {
+ if (index.getName().equals(storedName)) {
+ logical = index;
+ break;
+ }
+ }
+ if (logical == null) {
+ return false;
+ }
+ String requestAlgorithm = requestedAlgorithm(def);
+ if (requestAlgorithm == null ||
!LanceIndexFamilies.normalize(logical.getIndexType())
+ .equals(LanceIndexFamilies.normalize(requestAlgorithm))) {
+ return false;
+ }
+ LanceIndexAdmissionSnapshot.PhysicalIndexInfo physical = null;
+ for (LanceIndexAdmissionSnapshot.PhysicalIndexInfo entry :
snapshot.getPhysicalIndexes()) {
+ if (entry.getName().equals(storedName)) {
+ physical = entry;
+ break;
+ }
+ }
+ if (physical == null
+ || !LanceIndexFamilies.isCompatible(logical.getIndexType(),
physical.getIndexTypeName())) {
+ return false;
+ }
+ if (logical.getColumns().size() != 1) {
+ return false;
+ }
+ String requestColumn =
LanceIndexNameNormalizer.normalize(def.getCols().get(0));
Review Comment:
[P2] Compare the same field-name representation on both sides. Logical
metadata escapes even a top-level name as a path segment (for example, a field
named `a.b` becomes a backtick-quoted path), while the parser strips identifier
quoting before putting raw `a.b` in `def.getCols()`. Thus a valid CREATE INDEX
IF NOT EXISTS is rejected as a different definition although table lookup and
schema-contract construction resolve the same field. Compare field identity/raw
stored names, or canonicalize the request with the same path formatter, and
test dots, spaces, and embedded backticks.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceSchemaContractBuilder.java:
##########
@@ -0,0 +1,236 @@
+// 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;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.lance.job.LanceIndexSchemaContract;
+
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.lance.schema.LanceField;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.regex.Pattern;
+
+/**
+ * Builds schema contract v1 for one indexed column from the fresh LanceField
tree of the
+ * pinned admission snapshot. {@link LanceTypeConverter} is deliberately
bypassed because it
+ * erases fixed-size-list dimensions, float16-vs-float32, and timestamp
timezones.
+ *
+ * <p>The input is the stored column name (byte-identical to the LanceField
name), never the
+ * raw user input: callers resolve it via the table's column lookup first.
Matching is exact
+ * and top-level only; a missing field fails closed. The builder makes no
supportability
+ * judgment — every ArrowType yields a deterministic canonical string, so the
only failure is
+ * the indexed field not being found. The canonical vocabulary defined here is
the Java-side
+ * authority the Rust worker's golden fixtures align to (design section 4.2).
+ */
+final class LanceSchemaContractBuilder {
+ /** Timezones that fit the canonical {@code tz="…"} slot without any
escaping (IANA names). */
+ private static final Pattern SAFE_TIMEZONE =
Pattern.compile("[A-Za-z0-9+_/-]+");
+
+ private LanceSchemaContractBuilder() {
+ }
+
+ /**
+ * Builds the single-field contract for {@code storedColumnName}. Only
top-level fields are
+ * considered; nested subfields never enter the contract.
+ */
+ static LanceIndexSchemaContract build(List<LanceField> topLevelFields,
String storedColumnName)
+ throws AnalysisException {
+ if (topLevelFields == null) {
+ throw new IllegalArgumentException("Lance top-level schema fields
must not be null");
+ }
+ if (storedColumnName == null || storedColumnName.isEmpty()) {
+ throw new IllegalArgumentException("stored column name must not be
null or empty");
+ }
+ for (LanceField field : topLevelFields) {
+ if (field == null) {
+ throw new IllegalArgumentException("Lance top-level schema
field must not be null");
+ }
+ if (storedColumnName.equals(field.getName())) {
+ return new LanceIndexSchemaContract(
+ Collections.singletonList(indexedField(field)));
+ }
+ }
+ ErrorReport.reportAnalysisException(ErrorCode.ERR_LANCE_INDEX_INVALID,
+ "unsupported schema contract: indexed field not found");
+ throw new IllegalStateException("unreachable");
+ }
+
+ private static LanceIndexSchemaContract.IndexedField
indexedField(LanceField field) {
+ ArrowType type = field.getType();
+ if (type == null) {
+ throw new IllegalArgumentException("Lance field type must not be
null");
+ }
+ Integer fixedSizeListDimension = null;
+ String vectorElementType = null;
+ Boolean vectorElementNullable = null;
+ if (type instanceof ArrowType.FixedSizeList) {
+ fixedSizeListDimension = ((ArrowType.FixedSizeList)
type).getListSize();
+ List<LanceField> children = field.getChildren();
Review Comment:
[P1] Resolve the primitive fixed-list contract against the pinned SDK. In
Lance 9.1.0-beta.3, `Field::try_from` leaves `children` empty for primitive
`FixedSizeList<Float16/Float32>`; JNI copies that empty list, so every such
column throws here while the unit test supplies an impossible mocked child. The
manifest logical type keeps element type/dimension but drops original item
nullability, and reconstructed Arrow views synthesize it as nullable; the
regression fixture also leaves that nested default. Thus the current pin cannot
supply the declared non-null-element contract as modeled. Please define a
contract the pinned reconstructed schema can actually provide (or use an
API/version that preserves the fact), then cover it with a real Dataset schema
fixture.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceIndexAdmission.java:
##########
@@ -0,0 +1,504 @@
+// 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;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.datasource.lance.job.LanceIndexDatasetLocator;
+import org.apache.doris.datasource.lance.job.LanceIndexFenceKey;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationType;
+import org.apache.doris.datasource.lance.job.LanceIndexNameNormalizer;
+import org.apache.doris.datasource.lance.job.LanceIndexSchemaContract;
+import org.apache.doris.nereids.trees.plans.commands.info.IndexDefinition;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+
+/**
+ * Lance index admission (design sections 2.2 and 4.1): the single place where
a statically
+ * validated top-level CREATE [OR REPLACE]/DROP INDEX statement against a
Lance catalog table is
+ * turned into a durable job. The whole flow runs against one pinned admission
snapshot — the
+ * IF preflight never takes a second metadata read, and no catalog/db/table
metadata lock is held
+ * while the snapshot loader does its JNI work (design section 5.1).
+ *
+ * <p>The step order is the correctness contract: name normalization and
reserved prefix come
+ * first, before target capture and the snapshot read (fail cheap-first — a
reserved name
+ * rejected at admission depth costs no remote read), then case-only collision
analysis, IF
+ * preflight (including the two-stage {@code matches}: requested-algorithm
equality plus
+ * physical-family corroboration), schema contract from the stored column
name, locator
+ * normalization, deterministic properties JSON, positive-quota assertion, and
only then exactly
+ * one id allocation and the durable {@code createJob} transfer.
+ * Every rejection before {@code createJob} leaves no job, no fence, no quota
charge, no journal
+ * record, and no id allocation; the manager's own fence/quota rejections pass
through verbatim
+ * (an id burned by them is accepted — ids are never required to be
contiguous).
+ */
+public final class LanceIndexAdmission {
+
+ /**
+ * The snapshot read seam. Tests inject a prepared snapshot here so
admission runs without
+ * FE startup or JNI; the production default delegates to the catalog's
merged-snapshot read.
+ */
+ public interface SnapshotLoader {
+ LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog, String
dbName, String tableName)
+ throws Exception;
+ }
+
+ /** The admission result: the durable job id, or null for an IF no-op (no
job created). */
+ public static final class Outcome {
+ private final Long jobId;
+
+ private Outcome(Long jobId) {
+ this.jobId = jobId;
+ }
+
+ /**
+ * The admitted job id, or null when the IF preflight made the
statement an immediate
+ * no-op (design section 2.2: "returns an immediate no-op, without
creating a job").
+ */
+ @Nullable
+ public Long getJobId() {
+ return jobId;
+ }
+ }
+
+ private static final SnapshotLoader DEFAULT_LOADER = new SnapshotLoader() {
+ @Override
+ public LanceIndexAdmissionSnapshot load(LanceExternalCatalog catalog,
String dbName,
+ String tableName) throws Exception {
+ return catalog.loadTableIndexAdmissionSnapshot(dbName, tableName);
+ }
+ };
+
+ private LanceIndexAdmission() {
+ }
+
+ /**
+ * Admits a top-level CREATE [OR REPLACE] INDEX. Static validation
+ * ({@link LanceIndexMutationValidator#validateCreateIndex}) must already
have passed for
+ * {@code def}.
+ */
+ public static Outcome admitCreate(LanceExternalCatalog catalog,
LanceExternalDatabase db,
+ LanceExternalTable table, IndexDefinition def, boolean
ifNotExists) throws Exception {
+ return admitCreate(DEFAULT_LOADER, catalog, db, table, def,
ifNotExists);
+ }
+
+ static Outcome admitCreate(SnapshotLoader loader, LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table,
IndexDefinition def, boolean ifNotExists)
+ throws Exception {
+ // 1. Display/normalized names and the reserved system prefix, checked
before any metadata
+ // read so a reserved name rejected at admission depth costs no remote
snapshot read (fail
+ // cheap-first). The prefix is rejected for CREATE and REPLACE exactly
as for DROP; the
+ // static layer rejects it first and this is the defense-in-depth copy
at admission depth.
+ String displayName = def.getIndexName();
+ String normalizedName =
LanceIndexNameNormalizer.normalize(displayName);
+ LanceIndexMutationValidator.rejectIfReservedIndexName(displayName);
+ // 2. One pinned snapshot for every authoritative decision below.
+ CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
+ CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
+ LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
+ // 3. Case-only analysis (design section 4.1): ambiguous external
collisions fail closed;
+ // a unique match resolves to the stored display name.
+ List<String> storedNames = logicalIndexNames(snapshot);
+ if (LanceIndexFamilies.isAmbiguousCaseCollision(storedNames,
normalizedName)) {
+ rejectInvalid("index name '" + displayName
+ + "' is ambiguous: multiple Lance indexes differ only by
case");
+ }
+ String storedName = LanceIndexFamilies.uniqueMatch(storedNames,
normalizedName);
+ // 4. IF preflight (design section 2.2).
+ boolean orReplace = def.isOrReplace();
+ if (!orReplace && storedName != null) {
+ if (!ifNotExists) {
+ rejectInvalid("index '" + displayName + "' already exists");
+ }
+ if (!matchesExistingDefinition(snapshot, storedName, def)) {
+ rejectInvalid("index '" + displayName + "' already exists with
a different definition");
+ }
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () ->
new Outcome(null));
+ }
+ // 5. Schema contract v1 from the stored column name (never the raw
user spelling).
+ String storedColumnName = storedColumnName(table,
def.getCols().get(0));
+ LanceIndexSchemaContract contract =
+ LanceSchemaContractBuilder.build(snapshot.getTopLevelFields(),
storedColumnName);
+ // 6. The fence locator is the normalized dataset uri of the same
pinned snapshot.
+ String locator = normalizeLocator(snapshot);
+ // 7. Deterministic normalized properties JSON for ANN; scalar
families persist null.
+ boolean ann = def.getLanceIndexType() == null;
+ String indexType = ann ? annIndexType(def) : def.getLanceIndexType();
+ String propertiesJson = ann ? buildAnnPropertiesJson(def) : null;
+ // D7 backstop: quota values from fe.conf bypass the ADMIN SET
callback, so admission
+ // re-asserts positivity before any id allocation or durable transfer.
+ assertPositiveQuotas();
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> {
+ // 8. Exactly one id allocation, after every preflight above has
passed.
+ long jobId = Env.getCurrentEnv().getNextId();
+ String creator = ConnectContext.get().getQualifiedUser();
+ // 9. REPLACE on an existing name persists the stored display name
(section 4.1) so the
+ // worker locates the case-sensitive target; a fresh REPLACE keeps
the user's spelling.
+ String persistedDisplayName = (orReplace && storedName != null) ?
storedName : displayName;
+ LanceIndexJob job;
+ try {
+ job = new LanceIndexJob(jobId, creator, catalog.getId(),
db.getFullName(), table.getName(),
+ LanceIndexFenceKey.PROVIDER_DIRECTORY, locator,
persistedDisplayName, normalizedName,
+ orReplace ? LanceIndexJobMutationType.REPLACE :
LanceIndexJobMutationType.CREATE,
+ ifNotExists, false, indexType, storedColumnName,
propertiesJson,
+ snapshot.getDatasetVersion(), contract);
+ } catch (IllegalArgumentException e) {
+ throw invalidAdmission(e.getMessage());
+ }
+ Env.getCurrentEnv().getLanceIndexJobManager().createJob(job,
+ Config.lance_index_job_max_unresolved_per_table,
+ Config.lance_index_job_max_unresolved_per_catalog,
+ Config.lance_index_job_max_unresolved_global);
+ // 10. The job and its fence are durable once createJob returns.
+ return new Outcome(jobId);
+ });
+ }
+
+ /**
+ * Admits a top-level DROP INDEX. The static name bounds
+ * ({@link LanceIndexMutationValidator#validateDropIndex}) must already
have passed.
+ */
+ public static Outcome admitDrop(LanceExternalCatalog catalog,
LanceExternalDatabase db,
+ LanceExternalTable table, String indexName, boolean ifExists)
throws Exception {
+ return admitDrop(DEFAULT_LOADER, catalog, db, table, indexName,
ifExists);
+ }
+
+ static Outcome admitDrop(SnapshotLoader loader, LanceExternalCatalog
catalog,
+ LanceExternalDatabase db, LanceExternalTable table, String
indexName, boolean ifExists)
+ throws Exception {
+ // Fail cheap-first: the reserved prefix is rejected before target
capture and the
+ // snapshot read, so it costs no remote read.
+ String normalizedName = LanceIndexNameNormalizer.normalize(indexName);
+ LanceIndexMutationValidator.rejectIfReservedIndexName(indexName);
+ CatalogMgr catalogMgr = Env.getCurrentEnv().getCatalogMgr();
+ CatalogMgr.LanceIndexTarget target =
catalogMgr.captureLanceIndexTarget(catalog);
+ LanceIndexAdmissionSnapshot snapshot = loader.load(catalog,
db.getRemoteName(), table.getRemoteName());
+ List<String> storedNames = logicalIndexNames(snapshot);
+ if (LanceIndexFamilies.isAmbiguousCaseCollision(storedNames,
normalizedName)) {
+ rejectInvalid("index name '" + indexName
+ + "' is ambiguous: multiple Lance indexes differ only by
case");
+ }
+ String storedName = LanceIndexFamilies.uniqueMatch(storedNames,
normalizedName);
+ if (storedName == null) {
+ if (ifExists) {
+ return catalogMgr.withLanceIndexAdmission(catalog, target, ()
-> new Outcome(null));
+ }
+ rejectInvalid("index '" + indexName + "' not found");
+ }
+ String locator = normalizeLocator(snapshot);
+ assertPositiveQuotas();
+ return catalogMgr.withLanceIndexAdmission(catalog, target, () -> {
+ long jobId = Env.getCurrentEnv().getNextId();
+ String creator = ConnectContext.get().getQualifiedUser();
+ // DROP only runs past the preflight with a unique match, so the
stored display name is
+ // always persisted (section 4.1); definition fields stay null on
a DROP job record.
+ LanceIndexJob job;
+ try {
+ job = new LanceIndexJob(jobId, creator, catalog.getId(),
db.getFullName(), table.getName(),
+ LanceIndexFenceKey.PROVIDER_DIRECTORY, locator,
storedName, normalizedName,
+ LanceIndexJobMutationType.DROP, false, ifExists, null,
null, null,
+ snapshot.getDatasetVersion(), null);
+ } catch (IllegalArgumentException e) {
+ throw invalidAdmission(e.getMessage());
+ }
+ Env.getCurrentEnv().getLanceIndexJobManager().createJob(job,
+ Config.lance_index_job_max_unresolved_per_table,
+ Config.lance_index_job_max_unresolved_per_catalog,
+ Config.lance_index_job_max_unresolved_global);
+ return new Outcome(jobId);
+ });
+ }
+
+ /**
+ * The section 2.2 definition match, two stages: (a) the requested
algorithm must equal the
+ * stored logical algorithm under family normalization — a same-name
different-algorithm
+ * request is a mismatch, never a no-op; (b) the physical entry of the
same name must exist
+ * and back the logical algorithm (snapshot self-consistency, failing
closed); (c) the single
+ * normalized column must be equal; (d) whitelist properties are compared
per property — a
+ * value the request sets and the snapshot exposes must be equal, an
unexposed snapshot value
+ * is skipped, and a property the request omits is never compared.
+ */
+ private static boolean
matchesExistingDefinition(LanceIndexAdmissionSnapshot snapshot,
+ String storedName, IndexDefinition def) {
+ LanceLogicalIndex logical = null;
+ for (LanceLogicalIndex index : snapshot.getLogicalIndexes()) {
+ if (index.getName().equals(storedName)) {
+ logical = index;
+ break;
+ }
+ }
+ if (logical == null) {
+ return false;
+ }
+ String requestAlgorithm = requestedAlgorithm(def);
+ if (requestAlgorithm == null ||
!LanceIndexFamilies.normalize(logical.getIndexType())
+ .equals(LanceIndexFamilies.normalize(requestAlgorithm))) {
+ return false;
+ }
+ LanceIndexAdmissionSnapshot.PhysicalIndexInfo physical = null;
+ for (LanceIndexAdmissionSnapshot.PhysicalIndexInfo entry :
snapshot.getPhysicalIndexes()) {
+ if (entry.getName().equals(storedName)) {
+ physical = entry;
+ break;
+ }
+ }
+ if (physical == null
+ || !LanceIndexFamilies.isCompatible(logical.getIndexType(),
physical.getIndexTypeName())) {
+ return false;
+ }
+ if (logical.getColumns().size() != 1) {
+ return false;
+ }
+ String requestColumn =
LanceIndexNameNormalizer.normalize(def.getCols().get(0));
+ if
(!LanceIndexNameNormalizer.normalize(logical.getColumns().get(0)).equals(requestColumn))
{
+ return false;
+ }
+ return whitelistPropertiesMatch(logical, def);
+ }
+
+ /**
+ * Per-property whitelist comparison (metric ↔ metric_type,
num_sub_vectors ↔
+ * compression.num_sub_vectors, num_bits ↔ compression.num_bits).
num_partitions is never
+ * compared (section 2.2). BTREE/BITMAP carry no user build properties, so
the comparison is
+ * vacuous for them.
+ */
+ private static boolean whitelistPropertiesMatch(LanceLogicalIndex logical,
IndexDefinition def) {
+ if (def.getLanceIndexType() != null) {
+ return true;
+ }
+ Map<String, String> request =
normalizedAnnProperties(def.getProperties());
+ JsonObject exposed = parseSnapshotProperties(logical.getProperties());
+ if (exposed == null && logical.getProperties() != null &&
!logical.getProperties().isEmpty()) {
+ // A malformed provider payload is not "nothing exposed": fail the
comparison closed
+ // rather than guess at a match (design section 3.4).
+ return false;
+ }
+ String metric = request.get("metric");
+ if (metric != null) {
+ JsonElement exposedMetric = exposed == null ? null :
exposed.get("metric_type");
+ // Lance stores the metric uppercased ("L2") while the validated
request vocabulary is
+ // lowercase ("l2"): both sides fold under the root locale before
comparison. An
+ // exposed but non-primitive metric is malformed provider data and
fails closed
+ // (design section 3.4), like an unparsable numeric property below.
+ if (exposedMetric != null && (!exposedMetric.isJsonPrimitive()
+ || !exposedMetric.getAsString().toLowerCase(Locale.ROOT)
+ .equals(metric.toLowerCase(Locale.ROOT)))) {
+ return false;
+ }
+ }
+ // A compression block that is present but not an object is malformed
provider data:
+ // fail closed (design section 3.4) rather than treat every numeric
property as
+ // unexposed. ANN requests always carry num_sub_vectors, so there is
always at least
+ // one numeric property to corroborate.
+ if (exposed != null && exposed.has("compression") &&
!exposed.get("compression").isJsonObject()) {
+ return false;
+ }
+ JsonObject compression = exposed == null || !exposed.has("compression")
+ ? null : exposed.getAsJsonObject("compression");
+ return numericPropertyMatches(request.get("num_sub_vectors"),
compression, "num_sub_vectors")
Review Comment:
[P2] Compare the effective fixed `num_bits` value here. Omitting this
property means 8: the validator accepts omission and every admitted job
unconditionally persists `num_bits=8`. Passing null here instead makes
`numericPropertyMatches` skip even an exposed `compression.num_bits=4`, so the
command can report a successful no-op for a definition Doris could not create.
Use 8 as the request-side default when omitted and cover an exposed non-8 value.
##########
regression-test/suites/external_table_p0/lance/test_lance_index_admission.groovy:
##########
@@ -0,0 +1,257 @@
+// 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.
+
+suite("test_lance_index_admission", "p0,external,nonConcurrent") {
+ // The Lance fixture is preinstalled in the MinIO container of the Iceberg
+ // external environment, so this suite deliberately shares its switch.
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Lance index admission test because the Iceberg
MinIO environment is disabled.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String lanceRestPort = context.config.otherConfigs.get("lance_rest_port")
+ // Admitted jobs are durable and stay PENDING forever in this delivery
slice: dispatch,
Review Comment:
[P2] This test is not repeatable on the shared FE state it explicitly
anticipates. A successful intended run leaves this catalog plus three
permanently PENDING jobs (normal CREATE, quota-case CREATE, and DROP); the
timestamp avoids name/fence collisions but every run still consumes the global
unresolved quota. With the default 256 limit, 85 runs leave 255 entries and the
next run cannot finish, with partial failures leaking sooner. Run this scenario
only in disposable/resettable FE state or defer it until an authorized
cleanup/release path exists.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLanceIndexJobsCommand.java:
##########
@@ -0,0 +1,292 @@
+// 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.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.trees.expressions.And;
+import org.apache.doris.nereids.trees.expressions.CompoundPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSet;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.qe.StmtExecutor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * SHOW LANCE INDEX JOBS [FROM [catalog.]db] [WHERE TableName = "tbl" [AND
State = "PENDING"]].
+ *
+ * <p>Lists the durable Lance index job records held by the master. Rows whose
persisted
+ * target no longer resolves (the catalog is gone, or the catalog is there but
the db or
+ * table no longer resolves) are visible to global ADMIN only; every other row
requires
+ * table-level SHOW on the persisted (catalog, db, table). Rows that fail the
check are
+ * omitted entirely, so non-ADMIN users see no orphan trace, not even a count.
The job
+ * locator, provider, normalized names, propertiesJson and schema contract are
never shown.
+ *
+ * <p>The WHERE clause is deliberately narrowed to EqualTo predicates combined
with AND
+ * over the case-insensitive keys TableName and State (no Like, unlike the
SHOW COPY
+ * precedent).
+ */
+public class ShowLanceIndexJobsCommand extends ShowCommand {
+ public static final ImmutableList<String> TITLE_NAMES = new
ImmutableList.Builder<String>()
+ .add("JobId")
+ .add("CatalogName")
+ .add("DbName")
+ .add("TableName")
+ .add("IndexName")
+ .add("Operation")
+ .add("State")
+ .add("RefreshState")
+ .add("PossibleLive")
+ .add("CreateTime")
+ .add("UpdateTime")
+ .add("Message")
+ .add("ForceReleased")
+ .add("ForceActor")
+ .add("ForceTime")
+ .build();
+
+ private static final String KEY_TABLE_NAME = "TableName";
+ private static final String KEY_STATE = "State";
+ private static final String WHERE_HINT = "Where clause should looks like:
TableName = \"your_table_name\""
+ + " or State =
\"PENDING|RUNNING|COMMITTED|NOT_COMMITTED|UNKNOWN\","
+ + " or compound predicate with operator AND";
+
+ private final List<String> nameParts;
+ private final Expression whereClause;
+
+ private String ctlName;
+ private String dbName;
+ private String tableNameValue;
+ private String stateValue;
+
+ public ShowLanceIndexJobsCommand(List<String> nameParts, Expression
whereClause) {
+ super(PlanType.SHOW_LANCE_INDEX_JOBS_COMMAND);
+ this.nameParts = nameParts;
+ this.whereClause = whereClause;
+ }
+
+ public List<String> getNameParts() {
+ return nameParts;
+ }
+
+ public Expression getWhereClause() {
+ return whereClause;
+ }
+
+ @Override
+ public ShowResultSetMetaData getMetaData() {
+ ShowResultSetMetaData.Builder builder =
ShowResultSetMetaData.builder();
+ for (String title : TITLE_NAMES) {
+ builder.addColumn(new Column(title, ScalarType.createVarchar(30)));
+ }
+ return builder.build();
+ }
+
+ private void validate(ConnectContext ctx) throws AnalysisException {
+ if (nameParts != null) {
+ if (nameParts.size() == 1) {
+ dbName = nameParts.get(0);
+ CatalogIf currentCatalog = ctx.getCurrentCatalog();
+ ctlName = currentCatalog == null ? null :
currentCatalog.getName();
+ } else if (nameParts.size() == 2) {
+ ctlName = nameParts.get(0);
+ dbName = nameParts.get(1);
+ } else {
+ throw new AnalysisException(
+ "Only support SHOW LANCE INDEX JOBS FROM
[catalog.]database, but get: " + nameParts);
+ }
+ }
+ analyzeWhereClause();
+ }
+
+ private void analyzeWhereClause() throws AnalysisException {
+ if (whereClause == null) {
+ return;
+ }
+ List<Expression> children = new ArrayList<>();
+ splitCompoundPredicate(whereClause, children);
+ Set<String> names = new HashSet<>();
+ for (Expression child : children) {
+ analyzeSubPredicate(child);
+ String name = ((UnboundSlot)
child.child(0)).getName().toLowerCase(Locale.ROOT);
+ if (!names.add(name)) {
+ throw new AnalysisException("column names on both sides of
operator AND should be different");
+ }
+ }
+ }
+
+ private void splitCompoundPredicate(Expression expr, List<Expression>
children) throws AnalysisException {
+ if (expr instanceof CompoundPredicate) {
+ if (!(expr instanceof And)) {
+ throw new AnalysisException("Only allow compound predicate
with operator AND");
+ }
+ splitCompoundPredicate(expr.child(0), children);
+ splitCompoundPredicate(expr.child(1), children);
+ } else {
+ children.add(expr);
+ }
+ }
+
+ private void analyzeSubPredicate(Expression expr) throws AnalysisException
{
+ if (!(expr instanceof EqualTo)
+ || !(expr.child(0) instanceof UnboundSlot)
+ || !(expr.child(1) instanceof StringLikeLiteral)) {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ String key = ((UnboundSlot) expr.child(0)).getName();
+ String value = ((StringLikeLiteral) expr.child(1)).getStringValue();
+ if (key.equalsIgnoreCase(KEY_TABLE_NAME)) {
+ tableNameValue = value;
+ } else if (key.equalsIgnoreCase(KEY_STATE)) {
+ stateValue = value.toUpperCase(Locale.ROOT);
+ try {
+ LanceIndexJobMutationState.valueOf(stateValue);
+ } catch (IllegalArgumentException e) {
+ throw new AnalysisException("Unknown Lance index job state: "
+ value + "; " + WHERE_HINT);
+ }
+ } else {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ }
+
+ @Override
+ public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ validate(ctx);
+ List<List<String>> rows = new ArrayList<>();
+ for (LanceIndexJob job :
Env.getCurrentEnv().getLanceIndexJobManager().getAllJobsSnapshot()) {
+ CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog =
+
Env.getCurrentEnv().getCatalogMgr().getCatalog(job.getCatalogId());
+ if (!matchesFilters(catalog, job)) {
+ continue;
+ }
+ if (!isAuthorized(ctx, catalog, job)) {
+ continue;
+ }
+ rows.add(renderRow(job, catalog));
+ }
+ return new ShowResultSet(getMetaData(), rows);
+ }
+
+ private boolean matchesFilters(CatalogIf<? extends DatabaseIf<? extends
TableIf>> catalog, LanceIndexJob job) {
+ if (ctlName != null && (catalog == null ||
!ctlName.equals(catalog.getName()))) {
+ return false;
+ }
+ if (dbName != null && !dbName.equals(job.getDbName())) {
Review Comment:
[P2] Apply the catalog's database-name semantics to this filter. Admission
persists the resolved local database name, but `FROM` keeps the user's
identifier spelling and this exact equality runs before resolution. With
`lower_case_database_names=1` or `2`, `SHOW ... FROM DB1` can return no rows
even though the catalog resolves `DB1` to stored `db1`. Resolve/canonicalize
the requested database through the selected catalog without weakening
authorization, and cover both case modes.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLanceIndexJobsCommand.java:
##########
@@ -0,0 +1,292 @@
+// 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.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.trees.expressions.And;
+import org.apache.doris.nereids.trees.expressions.CompoundPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSet;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.qe.StmtExecutor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * SHOW LANCE INDEX JOBS [FROM [catalog.]db] [WHERE TableName = "tbl" [AND
State = "PENDING"]].
+ *
+ * <p>Lists the durable Lance index job records held by the master. Rows whose
persisted
+ * target no longer resolves (the catalog is gone, or the catalog is there but
the db or
+ * table no longer resolves) are visible to global ADMIN only; every other row
requires
+ * table-level SHOW on the persisted (catalog, db, table). Rows that fail the
check are
+ * omitted entirely, so non-ADMIN users see no orphan trace, not even a count.
The job
+ * locator, provider, normalized names, propertiesJson and schema contract are
never shown.
+ *
+ * <p>The WHERE clause is deliberately narrowed to EqualTo predicates combined
with AND
+ * over the case-insensitive keys TableName and State (no Like, unlike the
SHOW COPY
+ * precedent).
+ */
+public class ShowLanceIndexJobsCommand extends ShowCommand {
+ public static final ImmutableList<String> TITLE_NAMES = new
ImmutableList.Builder<String>()
+ .add("JobId")
+ .add("CatalogName")
+ .add("DbName")
+ .add("TableName")
+ .add("IndexName")
+ .add("Operation")
+ .add("State")
+ .add("RefreshState")
+ .add("PossibleLive")
+ .add("CreateTime")
+ .add("UpdateTime")
+ .add("Message")
+ .add("ForceReleased")
+ .add("ForceActor")
+ .add("ForceTime")
+ .build();
+
+ private static final String KEY_TABLE_NAME = "TableName";
+ private static final String KEY_STATE = "State";
+ private static final String WHERE_HINT = "Where clause should looks like:
TableName = \"your_table_name\""
+ + " or State =
\"PENDING|RUNNING|COMMITTED|NOT_COMMITTED|UNKNOWN\","
+ + " or compound predicate with operator AND";
+
+ private final List<String> nameParts;
+ private final Expression whereClause;
+
+ private String ctlName;
+ private String dbName;
+ private String tableNameValue;
+ private String stateValue;
+
+ public ShowLanceIndexJobsCommand(List<String> nameParts, Expression
whereClause) {
+ super(PlanType.SHOW_LANCE_INDEX_JOBS_COMMAND);
+ this.nameParts = nameParts;
+ this.whereClause = whereClause;
+ }
+
+ public List<String> getNameParts() {
+ return nameParts;
+ }
+
+ public Expression getWhereClause() {
+ return whereClause;
+ }
+
+ @Override
+ public ShowResultSetMetaData getMetaData() {
+ ShowResultSetMetaData.Builder builder =
ShowResultSetMetaData.builder();
+ for (String title : TITLE_NAMES) {
+ builder.addColumn(new Column(title, ScalarType.createVarchar(30)));
+ }
+ return builder.build();
+ }
+
+ private void validate(ConnectContext ctx) throws AnalysisException {
+ if (nameParts != null) {
+ if (nameParts.size() == 1) {
+ dbName = nameParts.get(0);
+ CatalogIf currentCatalog = ctx.getCurrentCatalog();
+ ctlName = currentCatalog == null ? null :
currentCatalog.getName();
+ } else if (nameParts.size() == 2) {
+ ctlName = nameParts.get(0);
+ dbName = nameParts.get(1);
+ } else {
+ throw new AnalysisException(
+ "Only support SHOW LANCE INDEX JOBS FROM
[catalog.]database, but get: " + nameParts);
+ }
+ }
+ analyzeWhereClause();
+ }
+
+ private void analyzeWhereClause() throws AnalysisException {
+ if (whereClause == null) {
+ return;
+ }
+ List<Expression> children = new ArrayList<>();
+ splitCompoundPredicate(whereClause, children);
+ Set<String> names = new HashSet<>();
+ for (Expression child : children) {
+ analyzeSubPredicate(child);
+ String name = ((UnboundSlot)
child.child(0)).getName().toLowerCase(Locale.ROOT);
+ if (!names.add(name)) {
+ throw new AnalysisException("column names on both sides of
operator AND should be different");
+ }
+ }
+ }
+
+ private void splitCompoundPredicate(Expression expr, List<Expression>
children) throws AnalysisException {
+ if (expr instanceof CompoundPredicate) {
+ if (!(expr instanceof And)) {
+ throw new AnalysisException("Only allow compound predicate
with operator AND");
+ }
+ splitCompoundPredicate(expr.child(0), children);
+ splitCompoundPredicate(expr.child(1), children);
+ } else {
+ children.add(expr);
+ }
+ }
+
+ private void analyzeSubPredicate(Expression expr) throws AnalysisException
{
+ if (!(expr instanceof EqualTo)
+ || !(expr.child(0) instanceof UnboundSlot)
+ || !(expr.child(1) instanceof StringLikeLiteral)) {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ String key = ((UnboundSlot) expr.child(0)).getName();
+ String value = ((StringLikeLiteral) expr.child(1)).getStringValue();
+ if (key.equalsIgnoreCase(KEY_TABLE_NAME)) {
+ tableNameValue = value;
+ } else if (key.equalsIgnoreCase(KEY_STATE)) {
+ stateValue = value.toUpperCase(Locale.ROOT);
+ try {
+ LanceIndexJobMutationState.valueOf(stateValue);
+ } catch (IllegalArgumentException e) {
+ throw new AnalysisException("Unknown Lance index job state: "
+ value + "; " + WHERE_HINT);
+ }
+ } else {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ }
+
+ @Override
+ public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ validate(ctx);
+ List<List<String>> rows = new ArrayList<>();
+ for (LanceIndexJob job :
Env.getCurrentEnv().getLanceIndexJobManager().getAllJobsSnapshot()) {
+ CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog =
+
Env.getCurrentEnv().getCatalogMgr().getCatalog(job.getCatalogId());
+ if (!matchesFilters(catalog, job)) {
+ continue;
+ }
+ if (!isAuthorized(ctx, catalog, job)) {
+ continue;
+ }
+ rows.add(renderRow(job, catalog));
+ }
+ return new ShowResultSet(getMetaData(), rows);
+ }
+
+ private boolean matchesFilters(CatalogIf<? extends DatabaseIf<? extends
TableIf>> catalog, LanceIndexJob job) {
+ if (ctlName != null && (catalog == null ||
!ctlName.equals(catalog.getName()))) {
+ return false;
+ }
+ if (dbName != null && !dbName.equals(job.getDbName())) {
+ return false;
+ }
+ if (tableNameValue != null &&
!tableNameValue.equals(job.getTableName())) {
+ return false;
+ }
+ return stateValue == null
+ || job.getMutationState() != null &&
stateValue.equals(job.getMutationState().name());
+ }
+
+ /**
+ * Orphan and half-orphan rows (catalog gone, or persisted db/table no
longer resolvable)
+ * are visible to global ADMIN only; every other row needs table-level
SHOW on the
+ * persisted target. The caller omits the row when this returns false, so
non-ADMIN users
+ * see no trace of orphaned jobs, not even a count.
+ */
+ static boolean isAuthorized(ConnectContext ctx, CatalogIf<? extends
DatabaseIf<? extends TableIf>> catalog,
+ LanceIndexJob job) {
+ if (!targetResolves(catalog, job)) {
+ return Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx,
PrivPredicate.ADMIN);
+ }
+ return Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx,
catalog.getName(),
+ job.getDbName(), job.getTableName(), PrivPredicate.SHOW);
+ }
+
+ static boolean targetResolves(CatalogIf<? extends DatabaseIf<? extends
TableIf>> catalog, LanceIndexJob job) {
Review Comment:
[P1] Authorize against the job's durable target, not just today's
catalog/db/table names. After a terminal job releases the unresolved guard, a
legal warehouse, namespace, or routing ALTER can repoint the same catalog id
and names from locator A to dataset B. This helper then succeeds, so a user
with SHOW on B can read A's historical job message, executor, and FORCE audit
fields. Freshly resolve and revalidate the provider plus normalized effective
target (including non-secret routing identity) against the job; treat
mismatches or missing identity as ADMIN-only orphans while keeping credential
rotation allowed.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ShowLanceIndexJobsCommand.java:
##########
@@ -0,0 +1,292 @@
+// 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.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.lance.job.LanceIndexJob;
+import org.apache.doris.datasource.lance.job.LanceIndexJobMutationState;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.analyzer.UnboundSlot;
+import org.apache.doris.nereids.trees.expressions.And;
+import org.apache.doris.nereids.trees.expressions.CompoundPredicate;
+import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.literal.StringLikeLiteral;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSet;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.qe.StmtExecutor;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * SHOW LANCE INDEX JOBS [FROM [catalog.]db] [WHERE TableName = "tbl" [AND
State = "PENDING"]].
+ *
+ * <p>Lists the durable Lance index job records held by the master. Rows whose
persisted
+ * target no longer resolves (the catalog is gone, or the catalog is there but
the db or
+ * table no longer resolves) are visible to global ADMIN only; every other row
requires
+ * table-level SHOW on the persisted (catalog, db, table). Rows that fail the
check are
+ * omitted entirely, so non-ADMIN users see no orphan trace, not even a count.
The job
+ * locator, provider, normalized names, propertiesJson and schema contract are
never shown.
+ *
+ * <p>The WHERE clause is deliberately narrowed to EqualTo predicates combined
with AND
+ * over the case-insensitive keys TableName and State (no Like, unlike the
SHOW COPY
+ * precedent).
+ */
+public class ShowLanceIndexJobsCommand extends ShowCommand {
+ public static final ImmutableList<String> TITLE_NAMES = new
ImmutableList.Builder<String>()
+ .add("JobId")
+ .add("CatalogName")
+ .add("DbName")
+ .add("TableName")
+ .add("IndexName")
+ .add("Operation")
+ .add("State")
+ .add("RefreshState")
+ .add("PossibleLive")
+ .add("CreateTime")
+ .add("UpdateTime")
+ .add("Message")
+ .add("ForceReleased")
+ .add("ForceActor")
+ .add("ForceTime")
+ .build();
+
+ private static final String KEY_TABLE_NAME = "TableName";
+ private static final String KEY_STATE = "State";
+ private static final String WHERE_HINT = "Where clause should looks like:
TableName = \"your_table_name\""
+ + " or State =
\"PENDING|RUNNING|COMMITTED|NOT_COMMITTED|UNKNOWN\","
+ + " or compound predicate with operator AND";
+
+ private final List<String> nameParts;
+ private final Expression whereClause;
+
+ private String ctlName;
+ private String dbName;
+ private String tableNameValue;
+ private String stateValue;
+
+ public ShowLanceIndexJobsCommand(List<String> nameParts, Expression
whereClause) {
+ super(PlanType.SHOW_LANCE_INDEX_JOBS_COMMAND);
+ this.nameParts = nameParts;
+ this.whereClause = whereClause;
+ }
+
+ public List<String> getNameParts() {
+ return nameParts;
+ }
+
+ public Expression getWhereClause() {
+ return whereClause;
+ }
+
+ @Override
+ public ShowResultSetMetaData getMetaData() {
+ ShowResultSetMetaData.Builder builder =
ShowResultSetMetaData.builder();
+ for (String title : TITLE_NAMES) {
+ builder.addColumn(new Column(title, ScalarType.createVarchar(30)));
+ }
+ return builder.build();
+ }
+
+ private void validate(ConnectContext ctx) throws AnalysisException {
+ if (nameParts != null) {
+ if (nameParts.size() == 1) {
+ dbName = nameParts.get(0);
+ CatalogIf currentCatalog = ctx.getCurrentCatalog();
+ ctlName = currentCatalog == null ? null :
currentCatalog.getName();
+ } else if (nameParts.size() == 2) {
+ ctlName = nameParts.get(0);
+ dbName = nameParts.get(1);
+ } else {
+ throw new AnalysisException(
+ "Only support SHOW LANCE INDEX JOBS FROM
[catalog.]database, but get: " + nameParts);
+ }
+ }
+ analyzeWhereClause();
+ }
+
+ private void analyzeWhereClause() throws AnalysisException {
+ if (whereClause == null) {
+ return;
+ }
+ List<Expression> children = new ArrayList<>();
+ splitCompoundPredicate(whereClause, children);
+ Set<String> names = new HashSet<>();
+ for (Expression child : children) {
+ analyzeSubPredicate(child);
+ String name = ((UnboundSlot)
child.child(0)).getName().toLowerCase(Locale.ROOT);
+ if (!names.add(name)) {
+ throw new AnalysisException("column names on both sides of
operator AND should be different");
+ }
+ }
+ }
+
+ private void splitCompoundPredicate(Expression expr, List<Expression>
children) throws AnalysisException {
+ if (expr instanceof CompoundPredicate) {
+ if (!(expr instanceof And)) {
+ throw new AnalysisException("Only allow compound predicate
with operator AND");
+ }
+ splitCompoundPredicate(expr.child(0), children);
+ splitCompoundPredicate(expr.child(1), children);
+ } else {
+ children.add(expr);
+ }
+ }
+
+ private void analyzeSubPredicate(Expression expr) throws AnalysisException
{
+ if (!(expr instanceof EqualTo)
+ || !(expr.child(0) instanceof UnboundSlot)
+ || !(expr.child(1) instanceof StringLikeLiteral)) {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ String key = ((UnboundSlot) expr.child(0)).getName();
+ String value = ((StringLikeLiteral) expr.child(1)).getStringValue();
+ if (key.equalsIgnoreCase(KEY_TABLE_NAME)) {
+ tableNameValue = value;
+ } else if (key.equalsIgnoreCase(KEY_STATE)) {
+ stateValue = value.toUpperCase(Locale.ROOT);
+ try {
+ LanceIndexJobMutationState.valueOf(stateValue);
+ } catch (IllegalArgumentException e) {
+ throw new AnalysisException("Unknown Lance index job state: "
+ value + "; " + WHERE_HINT);
+ }
+ } else {
+ throw new AnalysisException(WHERE_HINT);
+ }
+ }
+
+ @Override
+ public ShowResultSet doRun(ConnectContext ctx, StmtExecutor executor)
throws Exception {
+ validate(ctx);
+ List<List<String>> rows = new ArrayList<>();
+ for (LanceIndexJob job :
Env.getCurrentEnv().getLanceIndexJobManager().getAllJobsSnapshot()) {
+ CatalogIf<? extends DatabaseIf<? extends TableIf>> catalog =
+
Env.getCurrentEnv().getCatalogMgr().getCatalog(job.getCatalogId());
+ if (!matchesFilters(catalog, job)) {
+ continue;
+ }
+ if (!isAuthorized(ctx, catalog, job)) {
+ continue;
+ }
+ rows.add(renderRow(job, catalog));
+ }
+ return new ShowResultSet(getMetaData(), rows);
+ }
+
+ private boolean matchesFilters(CatalogIf<? extends DatabaseIf<? extends
TableIf>> catalog, LanceIndexJob job) {
+ if (ctlName != null && (catalog == null ||
!ctlName.equals(catalog.getName()))) {
+ return false;
+ }
+ if (dbName != null && !dbName.equals(job.getDbName())) {
+ return false;
+ }
+ if (tableNameValue != null &&
!tableNameValue.equals(job.getTableName())) {
+ return false;
+ }
+ return stateValue == null
+ || job.getMutationState() != null &&
stateValue.equals(job.getMutationState().name());
+ }
+
+ /**
+ * Orphan and half-orphan rows (catalog gone, or persisted db/table no
longer resolvable)
+ * are visible to global ADMIN only; every other row needs table-level
SHOW on the
+ * persisted target. The caller omits the row when this returns false, so
non-ADMIN users
+ * see no trace of orphaned jobs, not even a count.
+ */
+ static boolean isAuthorized(ConnectContext ctx, CatalogIf<? extends
DatabaseIf<? extends TableIf>> catalog,
+ LanceIndexJob job) {
+ if (!targetResolves(catalog, job)) {
+ return Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx,
PrivPredicate.ADMIN);
+ }
+ return Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx,
catalog.getName(),
+ job.getDbName(), job.getTableName(), PrivPredicate.SHOW);
+ }
+
+ static boolean targetResolves(CatalogIf<? extends DatabaseIf<? extends
TableIf>> catalog, LanceIndexJob job) {
+ if (catalog == null) {
+ return false;
+ }
+ DatabaseIf<? extends TableIf> db =
catalog.getDbNullable(job.getDbName());
Review Comment:
[P1] Keep provider lookup failures inside the non-disclosing authorization
boundary. `getTableNullable()` can initialize, list, or load external metadata
and throw on expired credentials or an outage. For detail, that makes a valid
job ID return a provider-specific error while a missing or unauthorized ID
returns fixed 5103; for list, one bad target aborts every row. Treat lookup
failure as an unresolved/orphan target through an exception-safe lookup
boundary, so non-ADMIN callers learn nothing and ADMIN can still inspect the
orphan.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
