u70b3 commented on code in PR #67630: URL: https://github.com/apache/doris/pull/67630#discussion_r4003110648
########## 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: Fixed in de4daa43f7 — an omitted request `num_bits` now compares as its effective value 8 against any exposed `compression.num_bits`. ########## 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: Fixed in e28543a241 — the duplicate-UUID check now runs before the system-entry filter, so a UUID shared between a system entry and a user entry fails closed; covered by a system/user shared-UUID test. ########## 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: Fixed in 4f887ace6b — element facts now come from the reconstructed Arrow view (`asArrowField`'s synthesized child), verified against a real on-disk dataset on the pinned 9.1.0-beta.3: https://github.com/apache/doris/pull/67630#issuecomment-5637228386. ########## 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: Fixed in d192e4d2de — negative field ids and non-positive fixed-list dimensions are rejected before the contract is journaled. -- 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]
