>From Hongyu Shi <[email protected]>: 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 <[email protected]> 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 <[email protected]>
