[ASTERIXDB-3760][RT][STO] Handlers of vtree build on dimension mismatch

- user model changes: yes
- storage format changes: no
- interface changes: yes

Details:
By design, a vector index builds only over the embeddings whose
dimension matches the DDL and skips the rest, so a build is implicitly
partial. Each stage applied that rule on its own, which left the runtime
exception handling tangled: before this change, when no vector matched,
CREATE INDEX reported success while the bulk load dropped every record
and ANN queries on the finished index returned no rows; when only some
matched, training ran on vectors the index would never hold, one could
become a centroid, their components shaped the quantization constants,
and a vector longer than the declared dimension threw
ArrayIndexOutOfBoundsException in the distance loop.

Fix: apply the rule at every stage that consumes a sampled vector, and
reject a build that would leave a partition with nothing to index. The
rejection is raised in the first job, before any training work, and
names both the declared and the observed dimension.

- Training: test the dimension wherever a sampled vector is consumed,
  and seed k-means from the first indexable one.
- Quantization: skip non-indexable vectors so the constants describe the
  vectors the index will actually hold, and reject a partition whose
  whole sample is non-indexable.
- Reporting: the static-structure creator no longer builds after an
  upstream failure, which replaced the real error with "no training
  vectors were found"; the empty quantization payload now names its
  likely causes instead of a byte count.
- Metadata: drop the 384 fallback for a missing `dimension`.
- Storage: bind metaFrame in end(), where takeBlock needs it, so an
  empty load works. takeBlock reads getMaxPage() without binding the
  frame, unlike takePage.

Tests: operator coverage that off-dimension vectors do not shape the
centroids or the constants, for the partial and empty cases, and a
runtimets golden for the DDL error.

Ext-ref: MB-73219
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Change-Id: I44e13a38e10e09d0c439063582b2747942476f26
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21535
Integration-Tests: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Ian Maxon <imaxon@apache.org>
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>

Author: 2 people

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml

@@ -69,6 +69,19 @@
             <expected-error>cannot be used as a vector key field</expected-error>
         </compilation-unit>
     </test-case>
+    <test-case FilePath="vector">
+        <!-- Either build job can report a wholly non-indexable sample: the quantization job names the
+             declared and the found dimension, and the training job names the declared one among the causes
+             of an empty parameter block. Which one arrives first is not deterministic, so match the phrase
+             they share. Matching is a literal substring test under CancellationTestExecutor, so this cannot
+             be an alternation. -->
+        <compilation-unit name="create-index-vtree-dimension-mismatch">
+            <output-dir compare="Text">create-index-vtree-dimension-mismatch</output-dir>
+            <expected-error>the index declares</expected-error>
+            <!-- Raised by a build job on an NC, so it carries no source location. -->
+            <source-location>false</source-location>
+        </compilation-unit>
+    </test-case>
     <test-case FilePath="vector">
         <compilation-unit name="create-index-vtree-composite-pk">
             <output-dir compare="Clean-JSON">create-index-vtree-composite-pk</output-dir>

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/create-index-vtree-dimension-mismatch/create-index-vtree-dimension-mismatch.1.ddl.sqlpp

@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+// Negative test: the vectors in the data are not the dimension the index declares. Every stage of the
+// build skips a vector it cannot index, so without a guard the build "succeeds" over an empty index and
+// ANN queries then silently return nothing. It must fail instead, naming both dimensions.
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE MovieType AS {
+  id: int,
+  embedding: [double]
+};
+
+CREATE DATASET MovieSmall(MovieType) PRIMARY KEY id;

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/create-index-vtree-dimension-mismatch/create-index-vtree-dimension-mismatch.2.update.sqlpp

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+// Every embedding is 4-dimensional.
+INSERT INTO MovieSmall ([
+  {"id": 1, "embedding": [0.0, 0.0, 0.0, 0.1]},
+  {"id": 2, "embedding": [0.0, 0.0, 0.0, 0.2]},
+  {"id": 3, "embedding": [0.0, 0.0, 0.0, 0.3]},
+  {"id": 4, "embedding": [0.0, 0.0, 0.0, 0.4]},
+  {"id": 5, "embedding": [0.0, 0.0, 0.0, 0.5]},
+  {"id": 6, "embedding": [10.0, 10.0, 10.0, 0.1]},
+  {"id": 7, "embedding": [10.0, 10.0, 10.0, 0.2]},
+  {"id": 8, "embedding": [10.0, 10.0, 10.0, 0.3]},
+  {"id": 9, "embedding": [10.0, 10.0, 10.0, 0.4]},
+  {"id": 10, "embedding": [10.0, 10.0, 10.0, 0.5]}
+]);

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/create-index-vtree-dimension-mismatch/create-index-vtree-dimension-mismatch.3.ddl.sqlpp

@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+ANALYZE DATASET test.MovieSmall WITH {"sample-seed": 1000};

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/create-index-vtree-dimension-mismatch/create-index-vtree-dimension-mismatch.4.ddl.sqlpp

@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+SET `compiler.vector.trainseed` "42";
+
+// The data is 4-dimensional, so no sampled vector can be indexed under dimension 8.
+CREATE INDEX idx_emb
+  ON MovieSmall(embedding VECTOR)
+  TYPE VTREE
+  WITH { "dimension": 8, "similarity": "euclidean", "num_clusters": 2, "train_list_fraction": 1.0 }
+  EXCLUDE UNKNOWN KEY;

asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/create-index-vtree-dimension-mismatch/create-index-vtree-dimension-mismatch.99.ddl.sqlpp

@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test IF EXISTS;

asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java

@@ -677,7 +677,7 @@ public JobSpecification buildCreationJobSpec() throws AlgebricksException {
         RecordDescriptor flattenedRecordDesc = new RecordDescriptor(flattenedSerDes, flattenedTypeTraits);
 
         targetOp = new VectorComponentExtractorOperatorDescriptor(spec, vectorFieldEvalFactory,
-                dataset.getPrimaryRecordDescriptor(metadataProvider), flattenedRecordDesc);
+                dataset.getPrimaryRecordDescriptor(metadataProvider), flattenedRecordDesc, vectorDimensions);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, targetOp, samplePartitionConstraint);
         spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
         sourceOp = targetOp;

asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/VTreeResourceFactoryProvider.java

@@ -92,9 +92,14 @@ public IResourceFactory getResourceFactory(MetadataProvider mdProvider, Dataset
                     vectorIndexDetails.getKeyFieldNames().size(), index.getIndexType(), 1);
         }
 
-        // Extract vector dimensions from WITH clause
+        // The index's physical dimension: it sizes the data frames and is what query vectors are checked
+        // against. Mandatory at CREATE INDEX and at persist, so a missing value here means a corrupt record.
         AdmObjectNode withObjectNode = vectorIndexDetails.getWithObjectNode();
-        int vectorDimensions = (withObjectNode != null) ? withObjectNode.getOptionalInt(WITH_KEY_DIMENSION, 384) : 384;
+        int vectorDimensions = (withObjectNode != null) ? withObjectNode.getOptionalInt(WITH_KEY_DIMENSION, -1) : -1;
+        if (vectorDimensions <= 0) {
+            throw new CompilationException(ErrorCode.COMPILATION_VECTOR_INDEX_CREATION_FAILED,
+                    "Index " + index.getIndexName() + " is missing the required positive `dimension` parameter");
+        }
 
         // Get INCLUDE fields count from index details (needed by factory)
         List<List<String>> includeFieldNames = vectorIndexDetails.getIncludeFieldNames();

asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java

@@ -178,6 +178,15 @@ public HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor(IOperatorDescriptor
         this.similarityMetric = similarityMetric;
     }
 
+    /**
+     * Whether the index can hold this vector. The bulk load applies the same rule, so one that fails here is
+     * never indexed, and must not shape the centroids either: not as a seed, candidate, padding or mean.
+     */
+    @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+    private boolean hasIndexDimension(double[] point) {
+        return point.length == vectorDimension;
+    }
+
     @Override
     public void contributeActivities(IActivityGraphBuilder builder) {
         // Activity 1: Store centroids and materialize data
@@ -332,7 +341,7 @@ public void initialize() throws HyracksDataException {
 
                     } catch (Throwable e) {
                         writer.fail();
-                        throw new RuntimeException(e);
+                        throw HyracksDataException.create(e);
                     } finally {
                         // Close the LAST reader opened by any nested reset, not just the first one.
                         closeCurrentSampleReader();
@@ -395,6 +404,12 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                     int firstIdx = rand.nextInt(totalTupleCount);
                     double[] firstCentroid = getPointAtIndex(in, fta, tuple, eval, inputVal, listAccessorConstant,
                             kMeansUtils, firstIdx, ctx);
+                    if (firstCentroid != null && !hasIndexDimension(firstCentroid)) {
+                        // Every pass below skips non-indexable vectors, so seeding with one would train
+                        // around a centroid nothing can be assigned to.
+                        firstCentroid = seedFromFirstIndexableVector(ctx, fta, tuple, eval, inputVal,
+                                listAccessorConstant, kMeansUtils, partition);
+                    }
                     if (firstCentroid == null) {
                         return new ClusteringResult(new ArrayList<>(), assignments);
                     }
@@ -432,10 +447,11 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                                 listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
                                 try {
                                     double[] point = kMeansUtils.createPrimitveList(listAccessorConstant);
-                                    // Compute D(x) = min distance to current centers
-                                    if (point.length != vectorDimension) {
+                                    if (!hasIndexDimension(point)) {
+                                        tempIdx++;
                                         continue;
                                     }
+                                    // Compute D(x) = min distance to current centers
                                     double minDist = Double.POSITIVE_INFINITY;
                                     for (double[] center : currentCenters) {
                                         double dist = distanceFunction.apply(point, center);
@@ -478,6 +494,10 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                                 listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
                                 try {
                                     double[] point = kMeansUtils.createPrimitveList(listAccessorConstant);
+                                    if (!hasIndexDimension(point)) {
+                                        currentIdx++;
+                                        continue;
+                                    }
 
                                     // RECOMPUTE D(x) (no storage from pass 1)
                                     double minDist = Double.POSITIVE_INFINITY;
@@ -536,6 +556,10 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                             listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
                             try {
                                 double[] point = kMeansUtils.createPrimitveList(listAccessorConstant);
+                                if (!hasIndexDimension(point)) {
+                                    weightIdx++;
+                                    continue;
+                                }
 
                                 // Find nearest candidate (recompute distance)
                                 double minDist = Double.POSITIVE_INFINITY;
@@ -623,7 +647,8 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                             List<double[]> additionalPoints = getPointsAtSortedIndices(in, fta, tuple, eval, inputVal,
                                     listAccessorConstant, kMeansUtils, additionalIndices, ctx);
                             for (double[] additionalPoint : additionalPoints) {
-                                if (additionalPoint != null) {
+                                // Non-indexable vectors must not become centroids either.
+                                if (additionalPoint != null && hasIndexDimension(additionalPoint)) {
                                     // Avoid duplicates - only add if not too close to existing centroids
                                     boolean isDuplicate = false;
                                     for (double[] existingCentroid : centroids) {
@@ -686,6 +711,10 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                                 listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
                                 try {
                                     double[] point = kMeansUtils.createPrimitveList(listAccessorConstant);
+                                    if (!hasIndexDimension(point)) {
+                                        currentIdx++;
+                                        continue;
+                                    }
 
                                     // Find closest centroid
                                     double minDist = Double.POSITIVE_INFINITY;
@@ -732,6 +761,10 @@ private ClusteringResult performKMeansParallel(IHyracksTaskContext ctx, Generate
                                 listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
                                 try {
                                     double[] point = kMeansUtils.createPrimitveList(listAccessorConstant);
+                                    if (!hasIndexDimension(point)) {
+                                        currentIdx++;
+                                        continue;
+                                    }
 
                                     int centroidIdx = assignments[currentIdx];
                                     for (int d = 0; d < point.length; d++) {
@@ -1012,6 +1045,38 @@ private int selectWeightedRandomIndex(List<double[]> candidates, int[] weights,
                     return candidates.size() - 1; // Fallback
                 }
 
+                /**
+                 * Returns the first indexable vector, to seed k-means when the random pick was not one, or
+                 * {@code null} if this partition holds none. The first build job already rejects that case, so
+                 * it is unreachable here for a dimension mismatch.
+                 */
+                @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+                private double[] seedFromFirstIndexableVector(IHyracksTaskContext ctx, FrameTupleAccessor fta,
+                        FrameTupleReference tuple, IScalarEvaluator eval, IPointable inputVal,
+                        ListAccessor listAccessor, KMeansUtils kMeansUtils, int partition)
+                        throws HyracksDataException, IOException {
+                    GeneratedRunFileReader reader = resetRunFileReader(ctx, sampleUUID, partition);
+                    VSizeFrame frame = new VSizeFrame(ctx);
+                    while (reader.nextFrame(frame)) {
+                        fta.reset(frame.getBuffer());
+                        int tupleCount = fta.getTupleCount();
+                        for (int i = 0; i < tupleCount; i++) {
+                            tuple.reset(fta, i);
+                            eval.evaluate(tuple, inputVal);
+                            if (!ATYPETAGDESERIALIZER.deserialize(inputVal.getByteArray()[inputVal.getStartOffset()])
+                                    .isListType()) {
+                                continue;
+                            }
+                            listAccessor.reset(inputVal.getByteArray(), inputVal.getStartOffset());
+                            double[] point = kMeansUtils.createPrimitveList(listAccessor);
+                            if (hasIndexDimension(point)) {
+                                return point;
+                            }
+                        }
+                    }
+                    return null;
+                }
+
                 /**
                  * Get a specific point by index from the run file.
                  */

asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VTreeStaticStructureCreatorOperatorDescriptor.java

@@ -145,6 +145,10 @@ public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
                 private int sampleCount = 0;
                 private boolean quantizationParamsLoaded = false;
 
+                /** Set on producer failure, so {@code close()} does not build over a truncated stream. */
+                @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+                private boolean upstreamFailed = false;
+
                 // Get all storage partitions for this compute partition
                 private final int[] storagePartitions = partitionsMap[partition];
 
@@ -250,7 +254,7 @@ public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
 
                 @Override
                 public void fail() throws HyracksDataException {
-                    // no op
+                    upstreamFailed = true;
                 }
 
                 private void processTuple(ITupleReference tuple) throws HyracksDataException {
@@ -440,7 +444,11 @@ private ITupleReference createLeafTupleWithQuantization(int centroidId, double[]
 
                 @Override
                 public void close() throws HyracksDataException {
-
+                    if (upstreamFailed) {
+                        // Only a prefix of the centroids arrived, if any. Building on that would fail here
+                        // and replace the producer's real error.
+                        return;
+                    }
                     try {
                         createStaticStructure();
                     } finally {