You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/09/01 13:17:31 UTC
[L] Change in asterixdb[master]: [ASTERIXDB-3817][COMP][RT] Filter unusable vectors before the build
From Hongyu Shi <le...@gmail.com>:
Hongyu Shi has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21654?usp=email )
Change subject: [ASTERIXDB-3817][COMP][RT] Filter unusable vectors before the build
......................................................................
[ASTERIXDB-3817][COMP][RT] Filter unusable vectors before the build
- user model changes: yes
- storage format changes: no
- interface changes: no
Details:
Apply isvector(field, dimension) as a StreamSelect ahead of the stages that read
the field in three jobs of vtree index build, and clean the scattered runtime
validation each job carried. The predicate is the one a user can run, so the
rows a vector index contains are exactly the rows for which isvector is true,
and a partial index is explainable with
SELECT COUNT(*) FROM ds WHERE NOT isvector(ds.emb, 384). The build error no
longer names the observed dimension, pointing at that query instead.
Ext-ref: MB-73665, MB-73439, MB-73300
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I9133ac0dbf5a261bc4adbd8d382b84f44ac122b2
---
M asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.01.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.02.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.99.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.adm
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VectorComponentExtractorOperatorDescriptor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/dataflow/QuantizedIndexCreateOperatorDescriptor.java
10 files changed, 244 insertions(+), 154 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/54/21654/1
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml
index 56ef204..ef866e1 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/VectorQueries.xml
@@ -115,6 +115,11 @@
</compilation-unit>
</test-case>
<test-case FilePath="vector">
+ <compilation-unit name="vector-distance-empty-vector">
+ <output-dir compare="Text">vector-distance-empty-vector</output-dir>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="vector">
<compilation-unit name="create-index-vtree-bad-similarity">
<output-dir compare="Text">create-index-vtree-bad-similarity</output-dir>
<expected-error>Allowed values: EUCLIDEAN, L2, EUCLIDEAN_SQUARED, L2_SQUARED, COSINE, COSINE SIMILARITY, DOT</expected-error>
@@ -140,7 +145,11 @@
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>
+ <!-- Every record is the wrong dimension, so the usable-vector filter drops them all and the
+ build reports the generic "nothing usable" error. It no longer names the observed
+ dimension: the filter is a boolean predicate and does not carry that back. The message
+ points at isvector instead, which reports it more completely than the old text did. -->
+ <expected-error>The sampled records yielded no usable vector for the indexed field</expected-error>
<!-- Raised by a build job on an NC, so it carries no source location. -->
<source-location>false</source-location>
</compilation-unit>
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.01.ddl.sqlpp
new file mode 100644
index 0000000..91b0c25
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.01.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.
+ */
+// An empty vector cannot yield a distance. Whichever way it is written -- a field that holds one, or an
+// empty literal -- the call must warn and return NULL, the way every other failed evaluation does.
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE MovieType AS {
+ id: int
+};
+
+CREATE DATASET MovieSmall(MovieType) PRIMARY KEY id;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.02.update.sqlpp
new file mode 100644
index 0000000..6a1ba36
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.02.update.sqlpp
@@ -0,0 +1,25 @@
+/*
+ * 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;
+
+INSERT INTO MovieSmall ([
+ {"id": 1, "embedding": [0.1, 0.2, 0.3, 0.4]},
+ {"id": 2, "embedding": []},
+ {"id": 3, "embedding": [0.5, 0.6, 0.7, 0.8]}
+]);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.query.sqlpp
new file mode 100644
index 0000000..dbcbdb6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.query.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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;
+
+// An empty vector arriving from a field is a data condition, not a query error: the row warns and
+// yields NULL while the rest of the collection computes normally. This is the shape that matters --
+// a schemaless collection may hold a malformed embedding in any row, and one such row must not fail
+// the query.
+//
+// Known quirk, deliberately not asserted here: written as an empty *literal* rather than read from a
+// field, the same call returns a denormal instead of NULL, because the distance builtins declare a
+// non-nullable double and a runtime NULL is read back through that. It needs a constant no real query
+// contains, and fixing it means changing shared SQL++ machinery, so it is recorded rather than fixed.
+SELECT m.id, vector_distance(m.embedding, [0.1, 0.2, 0.3, 0.4], "euclidean") AS dist
+FROM MovieSmall m
+ORDER BY m.id;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.99.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.99.ddl.sqlpp
new file mode 100644
index 0000000..d240ef2
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/vector/vector-distance-empty-vector/vector-distance-empty-vector.99.ddl.sqlpp
@@ -0,0 +1,19 @@
+/*
+ * 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;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.adm
new file mode 100644
index 0000000..81a6903f
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/vector/vector-distance-empty-vector/vector-distance-empty-vector.03.adm
@@ -0,0 +1,3 @@
+{ "id": 1, "dist": 0.0 }
+{ "id": 2, "dist": null }
+{ "id": 3, "dist": 0.8000000000000002 }
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java
index fe0ad63..4cb4a33 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/SecondaryVectorOperationsHelper.java
@@ -22,6 +22,7 @@
import static org.apache.asterix.om.types.BuiltinType.*;
import static org.apache.asterix.om.utils.ProjectionFiltrationTypeUtil.ALL_FIELDS_TYPE;
+import java.util.Arrays;
import java.util.List;
import java.util.UUID;
@@ -37,16 +38,21 @@
import org.apache.asterix.dataflow.data.common.AOrderedListVectorBinaryAccessorFactory;
import org.apache.asterix.external.indexing.IndexingConstants;
import org.apache.asterix.formats.base.IDataFormat;
+import org.apache.asterix.formats.nontagged.BinaryBooleanInspector;
+import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.entities.Dataset;
import org.apache.asterix.metadata.entities.Index;
import org.apache.asterix.metadata.entities.InternalDatasetDetails;
+import org.apache.asterix.om.base.AInt32;
import org.apache.asterix.om.pointables.base.DefaultOpenFieldType;
import org.apache.asterix.om.types.AOrderedListType;
import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
import org.apache.asterix.om.types.IAType;
import org.apache.asterix.om.vector.VectorIndexParameters;
import org.apache.asterix.runtime.aggregates.std.QuantizationConstantsAggregateDescriptor;
+import org.apache.asterix.runtime.evaluators.functions.IsVectorDescriptor;
import org.apache.asterix.runtime.operators.HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor;
import org.apache.asterix.runtime.operators.LSMIndexBulkLoadOperatorDescriptor;
import org.apache.asterix.runtime.operators.LSMIndexBulkLoadOperatorDescriptor.BulkLoadUsage;
@@ -68,9 +74,11 @@
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
import org.apache.hyracks.algebricks.runtime.evaluators.ColumnAccessEvalFactory;
+import org.apache.hyracks.algebricks.runtime.evaluators.ConstantEvalFactory;
import org.apache.hyracks.algebricks.runtime.operators.aggreg.SimpleAlgebricksAccumulatingAggregatorFactory;
import org.apache.hyracks.algebricks.runtime.operators.base.SinkRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.operators.meta.AlgebricksMetaOperatorDescriptor;
+import org.apache.hyracks.algebricks.runtime.operators.std.StreamSelectRuntimeFactory;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
@@ -78,12 +86,14 @@
import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.SourceLocation;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.data.std.accessors.DoubleBinaryComparatorFactory;
import org.apache.hyracks.data.std.accessors.IntegerBinaryComparatorFactory;
import org.apache.hyracks.data.std.primitive.FixedLengthTypeTrait;
import org.apache.hyracks.data.std.primitive.VarLengthTypeTrait;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.dataflow.common.data.marshalling.ByteArraySerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.DoubleSerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
@@ -106,6 +116,7 @@
import org.apache.hyracks.storage.common.IResourceFactory;
import org.apache.hyracks.storage.common.IStorageManager;
import org.apache.hyracks.storage.common.projection.ITupleProjectorFactory;
+import org.apache.hyracks.util.annotations.AiProvenance;
public class SecondaryVectorOperationsHelper extends SecondaryTreeIndexOperationsHelper {
@@ -208,6 +219,13 @@
sourceOp = targetOp;
// primary index -> cast assign op (produces the secondary index entry)
+ // Filter before the assign: the record still carries type tags here, while the
+ // secondary tuple the assign emits stores a closed key untagged.
+ targetOp = createUsableVectorFilterOp(spec, createRecordVectorFieldAccessor(), getVectorDimension(), recordDesc,
+ primaryPartitionConstraint);
+ spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
+ sourceOp = targetOp;
+
targetOp = createAssignOp(spec, numSecondaryKeys, recordDesc);
spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
@@ -309,6 +327,13 @@
sourceOp = targetOp;
// primary index -> cast assign op (produces the secondary index entry)
+ // Filter before the assign: the record still carries type tags here, while the
+ // secondary tuple the assign emits stores a closed key untagged.
+ targetOp = createUsableVectorFilterOp(spec, createRecordVectorFieldAccessor(), getVectorDimension(), recordDesc,
+ primaryPartitionConstraint);
+ spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
+ sourceOp = targetOp;
+
targetOp = createAssignOp(spec, numSecondaryKeys, recordDesc);
spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
@@ -561,6 +586,12 @@
// Vector Accessor
IScalarEvaluatorFactory vectorFieldEvalFactory = createFieldAccessor(itemType, recordColumn, vectorFieldName);
+ // Only usable vectors reach the quantization stages.
+ targetOp = createUsableVectorFilterOp(spec, vectorFieldEvalFactory, vectorDimensions,
+ dataset.getPrimaryRecordDescriptor(metadataProvider), samplePartitionConstraint);
+ spec.connect(new OneToOneConnectorDescriptor(spec), sourceOp, 0, targetOp, 0);
+ sourceOp = targetOp;
+
// Flattened Descriptor
IDataFormat format = metadataProvider.getDataFormat();
ISerializerDeserializerProvider serdeProvider = format.getSerdeProvider();
@@ -923,4 +954,68 @@
}
return pkFields;
}
+
+ /**
+ * A filter admitting only records whose embedding is a usable vector, placed ahead of every stage that
+ * reads one. The predicate is the {@code isvector} builtin, evaluated here exactly as a user's
+ * {@code WHERE isvector(field, dimension)} would evaluate it, which is what makes this contract hold:
+ * <blockquote>the set of rows a VTREE index contains is exactly the set of rows for which
+ * {@code isvector(field, dimension)} is true.</blockquote>
+ * A user can therefore see what was skipped, and why, without any tooling from us:
+ *
+ * <pre>
+ * SELECT COUNT(*) FROM ds WHERE NOT isvector(ds.emb, 384);
+ * SELECT DISTINCT array_count(ds.emb) FROM ds;
+ * </pre>
+ * <p>
+ * Every stage downstream of this filter may assume its input is a list of the declared dimension holding
+ * numeric elements, and none of them re-check it.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+ private IOperatorDescriptor createUsableVectorFilterOp(JobSpecification spec,
+ IScalarEvaluatorFactory vectorFieldAccessor, int dimension, RecordDescriptor inputRecordDescriptor,
+ AlgebricksPartitionConstraint partitionConstraint) throws AlgebricksException {
+ IScalarEvaluatorFactory condition = new IsVectorDescriptor().createEvaluatorFactory(
+ new IScalarEvaluatorFactory[] { vectorFieldAccessor, constantInt32(dimension) });
+ // A null projection list passes every field through; the filter only decides which tuples survive.
+ StreamSelectRuntimeFactory selectRuntime =
+ new StreamSelectRuntimeFactory(condition, null, BinaryBooleanInspector.FACTORY, false, -1, null);
+ selectRuntime.setSourceLocation(sourceLoc);
+ AlgebricksMetaOperatorDescriptor filterOp = new AlgebricksMetaOperatorDescriptor(spec, 1, 1,
+ new IPushRuntimeFactory[] { selectRuntime }, new RecordDescriptor[] { inputRecordDescriptor });
+ filterOp.setSourceLocation(sourceLoc);
+ AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, filterOp, partitionConstraint);
+ return filterOp;
+ }
+
+ /** The declared dimension as a constant argument to {@code isvector}. */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+ private static IScalarEvaluatorFactory constantInt32(int value) throws AlgebricksException {
+ ArrayBackedValueStorage storage = new ArrayBackedValueStorage();
+ try {
+ SerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32)
+ .serialize(new AInt32(value), storage.getDataOutput());
+ } catch (HyracksDataException e) {
+ throw new AlgebricksException(e);
+ }
+ return new ConstantEvalFactory(Arrays.copyOfRange(storage.getByteArray(), storage.getStartOffset(),
+ storage.getStartOffset() + storage.getLength()));
+ }
+
+ /**
+ * The vector field read out of the record, which is where every value still carries its type tag. The
+ * secondary tuple the assign produces stores a closed key with its declared serde, i.e. <em>untagged</em>,
+ * so a filter placed after the assign would see a headerless list and reject everything. Filtering happens
+ * on the scan side of the assign in every job for that reason.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.ASSISTED)
+ private int getVectorDimension() {
+ return ((Index.VectorIndexDetails) index.getIndexDetails()).getVectorParameters().getDimension();
+ }
+
+ private IScalarEvaluatorFactory createRecordVectorFieldAccessor() throws AlgebricksException {
+ Index.VectorIndexDetails details = (Index.VectorIndexDetails) index.getIndexDetails();
+ int recordColumn = dataset.getDatasetType() == DatasetType.INTERNAL ? numPrimaryKeys : 0;
+ return createFieldAccessor(itemType, recordColumn, details.getKeyFieldNames().get(0));
+ }
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java
index db7398a..ecd1bef 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/HierarchicalKMeansPlusPlusCentroidsOperatorDescriptor.java
@@ -428,7 +428,6 @@
in = resetRunFileReader(ctx, sampleUUID, partition);
VSizeFrame frame = new VSizeFrame(ctx);
- int tempIdx = 0;
while (in.nextFrame(frame)) {
ByteBuffer buffer = frame.getBuffer();
@@ -438,19 +437,9 @@
for (int j = 0; j < tupleCount; j++) {
tuple.reset(fta, j);
eval.evaluate(tuple, inputVal);
- if (!ATYPETAGDESERIALIZER
- .deserialize(inputVal.getByteArray()[inputVal.getStartOffset()]).isListType()) {
- tempIdx++;
- continue;
- }
-
listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
try {
double[] point = kMeansUtils.createPrimitiveList(listAccessorConstant);
- if (!hasIndexDimension(point)) {
- tempIdx++;
- continue;
- }
// Compute D(x) = min distance to current centers
double minDist = Double.POSITIVE_INFINITY;
for (double[] center : currentCenters) {
@@ -463,7 +452,6 @@
} catch (IOException e) {
throw HyracksDataException.create(e);
}
- tempIdx++;
}
}
@@ -474,7 +462,6 @@
// PASS 2: Stream again, recompute D(x), and sample probabilistically
in = resetRunFileReader(ctx, sampleUUID, partition);
frame = new VSizeFrame(ctx);
- int currentIdx = 0;
int sampledCount = 0;
while (in.nextFrame(frame)) {
@@ -485,20 +472,9 @@
for (int j = 0; j < tupleCount; j++) {
tuple.reset(fta, j);
eval.evaluate(tuple, inputVal);
- if (!ATYPETAGDESERIALIZER
- .deserialize(inputVal.getByteArray()[inputVal.getStartOffset()]).isListType()) {
- currentIdx++;
- continue;
- }
-
listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
try {
double[] point = kMeansUtils.createPrimitiveList(listAccessorConstant);
- if (!hasIndexDimension(point)) {
- currentIdx++;
- continue;
- }
-
// RECOMPUTE D(x) (no storage from pass 1)
double minDist = Double.POSITIVE_INFINITY;
for (double[] center : currentCenters) {
@@ -518,7 +494,6 @@
} catch (IOException e) {
throw HyracksDataException.create(e);
}
- currentIdx++;
}
}
@@ -537,7 +512,6 @@
in = resetRunFileReader(ctx, sampleUUID, partition);
VSizeFrame weightFrame = new VSizeFrame(ctx);
- int weightIdx = 0;
while (in.nextFrame(weightFrame)) {
ByteBuffer buffer = weightFrame.getBuffer();
@@ -547,20 +521,9 @@
for (int j = 0; j < tupleCount; j++) {
tuple.reset(fta, j);
eval.evaluate(tuple, inputVal);
- if (!ATYPETAGDESERIALIZER.deserialize(inputVal.getByteArray()[inputVal.getStartOffset()])
- .isListType()) {
- weightIdx++;
- continue;
- }
-
listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
try {
double[] point = kMeansUtils.createPrimitiveList(listAccessorConstant);
- if (!hasIndexDimension(point)) {
- weightIdx++;
- continue;
- }
-
// Find nearest candidate (recompute distance)
double minDist = Double.POSITIVE_INFINITY;
int nearestCandidate = -1;
@@ -577,7 +540,6 @@
} catch (IOException e) {
throw HyracksDataException.create(e);
}
- weightIdx++;
}
}
@@ -702,20 +664,9 @@
for (int j = 0; j < tupleCount; j++) {
tuple.reset(fta, j);
eval.evaluate(tuple, inputVal);
- if (!ATYPETAGDESERIALIZER
- .deserialize(inputVal.getByteArray()[inputVal.getStartOffset()]).isListType()) {
- currentIdx++;
- continue;
- }
-
listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
try {
double[] point = kMeansUtils.createPrimitiveList(listAccessorConstant);
- if (!hasIndexDimension(point)) {
- currentIdx++;
- continue;
- }
-
// Find closest centroid
double minDist = Double.POSITIVE_INFINITY;
int closestCentroid = 0;
@@ -752,20 +703,9 @@
for (int j = 0; j < tupleCount; j++) {
tuple.reset(fta, j);
eval.evaluate(tuple, inputVal);
- if (!ATYPETAGDESERIALIZER
- .deserialize(inputVal.getByteArray()[inputVal.getStartOffset()]).isListType()) {
- currentIdx++;
- continue;
- }
-
listAccessorConstant.reset(inputVal.getByteArray(), inputVal.getStartOffset());
try {
double[] point = kMeansUtils.createPrimitiveList(listAccessorConstant);
- if (!hasIndexDimension(point)) {
- currentIdx++;
- continue;
- }
-
int centroidIdx = assignments[currentIdx];
for (int d = 0; d < point.length; d++) {
newCentroids[centroidIdx][d] += point[d];
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VectorComponentExtractorOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VectorComponentExtractorOperatorDescriptor.java
index 9c26424..a39c2e6 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VectorComponentExtractorOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/VectorComponentExtractorOperatorDescriptor.java
@@ -23,19 +23,11 @@
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
-import org.apache.asterix.dataflow.data.nontagged.serde.ADoubleSerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AFloatSerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt16SerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt32SerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt64SerializerDeserializer;
-import org.apache.asterix.dataflow.data.nontagged.serde.AInt8SerializerDeserializer;
import org.apache.asterix.formats.nontagged.SerializerDeserializerProvider;
import org.apache.asterix.om.base.ADouble;
import org.apache.asterix.om.base.AMutableDouble;
import org.apache.asterix.om.types.ATypeTag;
import org.apache.asterix.om.types.BuiltinType;
-import org.apache.asterix.om.types.EnumDeserializer;
-import org.apache.asterix.om.types.hierachy.ATypeHierarchy;
import org.apache.asterix.runtime.evaluators.common.ListAccessor;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluator;
import org.apache.hyracks.algebricks.runtime.base.IScalarEvaluatorFactory;
@@ -100,9 +92,8 @@
private IScalarEvaluator vectorFieldEval;
private final IPointable vectorFieldValue = new VoidPointable();
private final ListAccessor listAccessor = new ListAccessor();
- private boolean sawIndexableVector;
- private long skippedVectorCount;
- private int skippedDimension = -1;
+ /** The one decoder, shared with the k-means and bulk-load stages rather than copied. */
+ private final KMeansUtils kMeansUtils = new KMeansUtils(new VoidPointable(), new ArrayBackedValueStorage());
private final IPointable tempVal = new VoidPointable();
private final ArrayBackedValueStorage storage = new ArrayBackedValueStorage();
@SuppressWarnings("unchecked")
@@ -144,39 +135,19 @@
continue;
}
- ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(data[offset]);
- if (typeTag == ATypeTag.MISSING || typeTag == ATypeTag.NULL || typeTag == ATypeTag.SYSTEM_NULL) {
- continue;
- }
-
- // Check if it's a list type (required for vector)
- if (!typeTag.isListType()) {
- continue;
- }
-
- // Iterate through array elements and emit one tuple per component
+ // Input is pre-filtered by isvector(field, dimension), so every value here is already a list
+ // of the declared dimension with numeric elements. See SecondaryVectorOperationsHelper.
listAccessor.reset(data, offset);
ATypeTag itemTypeTag = listAccessor.getItemType();
int listSize = listAccessor.size();
- // The bulk load and k-means both skip a vector of another dimension, so its components must
- // not shape the quantization constants either.
- if (listSize != vectorDimension) {
- skippedVectorCount++;
- if (skippedDimension < 0) {
- skippedDimension = listSize;
- }
- continue;
- }
- sawIndexableVector = true;
-
for (int j = 0; j < listSize; j++) {
try {
// Get item from list (can throw IOException)
listAccessor.getOrWriteItem(j, tempVal, storage);
// Extract numeric value (coerce to double)
- double componentValue = extractNumericValue(tempVal, itemTypeTag);
+ double componentValue = kMeansUtils.extractNumericVector(tempVal, itemTypeTag);
if (Double.isNaN(componentValue)) {
continue; // Skip invalid values
}
@@ -215,69 +186,18 @@
* Extracts numeric value from pointable, following KMeansUtils pattern.
* Coerces all numeric types to double.
*/
- private double extractNumericValue(IPointable pointable, ATypeTag derivedTypeTag) throws HyracksDataException {
- byte[] data = pointable.getByteArray();
- int offset = pointable.getStartOffset();
-
- if (ATypeHierarchy.getTypeDomain(derivedTypeTag) == ATypeHierarchy.Domain.NUMERIC) {
- double value = getValueFromTag(derivedTypeTag, data, offset);
- return value;
- } else if (derivedTypeTag == ATypeTag.ANY) {
- ATypeTag typeTag = EnumDeserializer.ATYPETAGDESERIALIZER.deserialize(data[offset]);
- double value = getValueFromTag(typeTag, data, offset);
- return value;
- } else {
- return Double.NaN; // Invalid type
- }
- }
-
- /**
- * Gets numeric value from type tag, following KMeansUtils.getValueFromTag pattern.
- */
- private double getValueFromTag(ATypeTag typeTag, byte[] data, int offset) throws HyracksDataException {
- switch (typeTag) {
- case TINYINT:
- return AInt8SerializerDeserializer.getByte(data, offset + 1);
- case SMALLINT:
- return AInt16SerializerDeserializer.getShort(data, offset + 1);
- case INTEGER:
- return AInt32SerializerDeserializer.getInt(data, offset + 1);
- case BIGINT:
- return AInt64SerializerDeserializer.getLong(data, offset + 1);
- case FLOAT:
- return AFloatSerializerDeserializer.getFloat(data, offset + 1);
- case DOUBLE:
- return ADoubleSerializerDeserializer.getDouble(data, offset + 1);
- default:
- return Double.NaN;
- }
- }
@Override
public void fail() throws HyracksDataException {
writer.fail();
}
- /**
- * Rejects a partition that sampled vectors but can index none of them: it would contribute nothing to
- * the index. Raised in this first build job, ahead of any training work.
- */
- private void rejectIfNothingIndexable() throws HyracksDataException {
- if (!sawIndexableVector && skippedVectorCount > 0) {
- throw new RuntimeDataException(ErrorCode.COMPILATION_VECTOR_INDEX_CREATION_FAILED,
- "the index declares dimension " + vectorDimension + ", but none of the " + skippedVectorCount
- + " vector(s) sampled in one partition match it (found " + "dimension "
- + skippedDimension + "). Set the index \"dimension\" parameter to match the data.");
- }
- }
-
@Override
public void close() throws HyracksDataException {
// A failed writer must still be closed: fail() only moves it to FAILED, and per IFrameWriter
// close() is the one call allowed from there.
HyracksDataException failure = null;
try {
- rejectIfNothingIndexable();
FrameUtils.flushFrame(appender.getBuffer(), writer);
} catch (HyracksDataException e) {
failure = e;
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/dataflow/QuantizedIndexCreateOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/dataflow/QuantizedIndexCreateOperatorDescriptor.java
index 03a2b52..7379d00 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/dataflow/QuantizedIndexCreateOperatorDescriptor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-vtree/src/main/java/org/apache/hyracks/storage/am/lsm/vector/dataflow/QuantizedIndexCreateOperatorDescriptor.java
@@ -79,6 +79,8 @@
private final FrameTupleAccessor tupleAccessor;
private final FrameTupleReference tupleHelper;
private VTreeQuantizationParams quantizationParams;
+ /** An empty parameter block arrived, meaning the sample produced nothing. See extractQuantizationParams. */
+ private boolean receivedEmptyPayload;
private boolean failed;
public QuantizedIndexCreateOperatorNodePushable(IHyracksTaskContext ctx, IIndexBuilder[] indexBuilders,
@@ -135,16 +137,20 @@
ByteArrayPointable ptr = new ByteArrayPointable();
ptr.set(data, start + 1, length - 1);
int contentLength = ptr.getContentLength();
- // The global aggregate emits either the full parameter block or nothing, and nothing means the
- // sample yielded no usable vector.
+ // The global aggregate emits either the full parameter block or nothing; nothing means no record
+ // reached the build. Recorded here and reported once from close(), rather than per partition.
+ if (contentLength == 0) {
+ receivedEmptyPayload = true;
+ // The global GroupAll aggregate emits a tuple even on empty input, so "no rows reached the
+ // build" arrives here as an empty payload rather than as a failure. Report it from close(),
+ // once, rather than per partition.
+ return null;
+ }
if (contentLength < PARAMS_PAYLOAD_BYTES) {
- // A user data condition, not a broken invariant: report it as a build failure the user can
- // act on rather than ILLEGAL_STATE, which cbas deliberately leaves unmapped and so surfaces
- // as a bare "Internal error".
+ // A non-empty but truncated payload is a genuine corruption rather than an empty sample.
throw HyracksDataException.create(ErrorCode.VECTOR_INDEX_BUILD_FAILED,
- "The sampled records yielded no usable vector for the indexed field: it may be missing, "
- + "null or not a list in every sampled record, the dataset may be empty, or no "
- + "sampled vector may match the dimension the index declares.");
+ "The quantization parameter block is truncated: got " + contentLength + " byte(s), need "
+ + PARAMS_PAYLOAD_BYTES + ".");
}
// Big-endian; DataInputStream is the exact inverse of the DataOutput/ByteBuffer writer and
@@ -181,6 +187,18 @@
for (IIndexBuilder indexBuilder : indexBuilders) {
indexBuilder.build();
}
+ } else if (receivedEmptyPayload) {
+ // No record survived the usable-vector filter, so there is nothing to compute quantization
+ // constants from. This is the only place that reports it: the filter drops rows silently by
+ // design, and every stage downstream of it assumes valid input.
+ throw HyracksDataException.create(ErrorCode.VECTOR_INDEX_BUILD_FAILED,
+ "The sampled records yielded no usable vector for the indexed field: it may be missing, "
+ + "null or not a list in every sampled record, the dataset may be empty, or no "
+ + "sampled vector may match the dimension the index declares. Run "
+ + "SELECT COUNT(*) FROM <collection> WHERE NOT isvector(<field>, <dimension>) "
+ + "to count the records that cannot be indexed, and "
+ + "SELECT DISTINCT array_count(<field>) FROM <collection> to see which "
+ + "dimensions the data actually holds.");
} else {
throw HyracksDataException.create(ErrorCode.VECTOR_INDEX_BUILD_FAILED,
"No quantization constants were received");
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21654?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I9133ac0dbf5a261bc4adbd8d382b84f44ac122b2
Gerrit-Change-Number: 21654
Gerrit-PatchSet: 1
Gerrit-Owner: Hongyu Shi <le...@gmail.com>