>From Shahrzad Shirazi <[email protected]>: Shahrzad Shirazi has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21985?usp=email )
Change subject: [NO ISSUE][COMP][RT] CLUSTER BY: share the k-means vector store ...................................................................... [NO ISSUE][COMP][RT] CLUSTER BY: share the k-means vector store - user model changes: no - storage format changes: no - interface changes: no Details: Under kmeans_parallel init, the oversampling loop and the Lloyd loop each wrote every input vector to a run file of their own, so the input reached disk twice before the labelling copy. Both loops run on the same locations, and each partition's file set covers the input once, which is all the Lloyd update needs. The Lloyd loop now reads the oversampling loop's vector file instead of writing its own: - KMeansStageOperator names the shared store; the expansion rule sets it on both loops, so they agree by construction. - The oversampling loop keeps its file on success; the Lloyd loop drains its own vector input unread, reads that file and deletes it after its last iteration. - The random init is unchanged: there is no oversampling loop, and the Lloyd loop keeps its own store. Results and plans are unchanged; one input-sized temporary file per partition is gone. Ext-ref: MB-74142 Generated-by: Claude Opus 5.5 <[email protected]> Change-Id: I2d34e7e35ae889edbce294e96889d187186c9d14 --- M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansLloydLoopPOperator.java M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansOversampleLoopPOperator.java M asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/RewriteClusterByToKMeansRule.java M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCostControllerOperatorDescriptor.java M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLloydControllerOperatorDescriptor.java M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/KMeansStageOperator.java M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SubstituteVariableVisitor.java 9 files changed, 81 insertions(+), 21 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/85/21985/1 diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansLloydLoopPOperator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansLloydLoopPOperator.java index 76d521f..bf24563 100644 --- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansLloydLoopPOperator.java +++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansLloydLoopPOperator.java @@ -56,12 +56,15 @@ ILogicalOperator src0, ILogicalOperator src1) throws AlgebricksException { JobSpecification spec = builder.getJobSpec(); // Unique + stable per loop instance, and distinct from any oversampling loop's key in the same job. - String loopKey = "kmeansLloydLoop#" + kop.getCandidateVariable(); + String loopKey = "kmeansLloydLoop#" + kop.getCandidateVariable().getId(); + // The oversampling loop's vector store under k-means||; null under the random init. + String sharedVectorsKey = kop.getVectorStoreVariable() == null ? null + : KMeansOversampleLoopPOperator.loopKeyOf(kop.getVectorStoreVariable()); int participants = clusterLocations.length; KMeansLloydControllerOperatorDescriptor op1 = new KMeansLloydControllerOperatorDescriptor(spec, centroidRecDesc, - KMeansLoopIO.PARTIAL_RD, loopKey, vectorColumn, centroidColumn, kop.getLoopRounds(), kop.getTopCount(), - framesLimit(), kop.getDimension(), metricOf(kop)); + KMeansLoopIO.PARTIAL_RD, loopKey, sharedVectorsKey, vectorColumn, centroidColumn, kop.getLoopRounds(), + kop.getTopCount(), framesLimit(), kop.getDimension(), metricOf(kop)); contributeOpDesc(builder, op, op1); builder.contributeGraphEdge(src0, 0, op, 0); builder.contributeGraphEdge(src1, 0, op, 1); diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansOversampleLoopPOperator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansOversampleLoopPOperator.java index ce3b5b2..9215689 100644 --- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansOversampleLoopPOperator.java +++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/KMeansOversampleLoopPOperator.java @@ -30,6 +30,7 @@ import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException; import org.apache.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder; import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator; +import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable; import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag; import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator; import org.apache.hyracks.algebricks.core.algebra.operators.logical.KMeansStageOperator; @@ -53,6 +54,14 @@ return PhysicalOperatorTag.KMEANS_OVERSAMPLE_LOOP; } + /** + * The joblet-state key of the oversampling loop with this candidate variable, which the Lloyd loop also + * uses to find its vector store. Unique within a plan, which is enough: joblet state is per job. + */ + static String loopKeyOf(LogicalVariable candidateVar) { + return "kmeansSystolicLoop#" + candidateVar.getId(); + } + @Override protected void contributeLoop(IHyracksJobBuilder builder, KMeansStageOperator kop, AbstractLogicalOperator op, RecordDescriptor poolEnvelopeRecDesc, int vectorColumn, int seedColumn, String[] clusterLocations, @@ -60,14 +69,15 @@ JobSpecification spec = builder.getJobSpec(); // Unique + stable per loop instance (one per query); baked into all five descriptors so every partition // and NC agrees on the joblet-state keys. - String loopKey = "kmeansSystolicLoop#" + kop.getCandidateVariable(); + String loopKey = loopKeyOf(kop.getCandidateVariable()); + boolean handOffVectors = kop.getVectorStoreVariable() != null; int participants = clusterLocations.length; // Op1 Cost/Controller is the registered descriptor: inputs land here, and output 0 carries the // weighed partials the downstream RECLUSTER reduces. KMeansCostControllerOperatorDescriptor op1 = new KMeansCostControllerOperatorDescriptor(spec, - poolEnvelopeRecDesc, KMeansLoopIO.SIGMA_RD, loopKey, vectorColumn, seedColumn, kop.getLoopRounds(), - framesLimit(), kop.getDimension(), metricOf(kop)); + poolEnvelopeRecDesc, KMeansLoopIO.SIGMA_RD, loopKey, handOffVectors, vectorColumn, seedColumn, + kop.getLoopRounds(), framesLimit(), kop.getDimension(), metricOf(kop)); contributeOpDesc(builder, op, op1); builder.contributeGraphEdge(src0, 0, op, 0); builder.contributeGraphEdge(src1, 0, op, 1); diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/RewriteClusterByToKMeansRule.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/RewriteClusterByToKMeansRule.java index 23b8ce9..8b713f8 100644 --- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/RewriteClusterByToKMeansRule.java +++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/optimizer/rules/RewriteClusterByToKMeansRule.java @@ -387,14 +387,16 @@ Mutable<ILogicalOperator> centroidsIn = seedOf(seedInput.getLeft(), seedInput.getRight(), forgy ? kmeans(cop).getNumClusters() : 1, options(cop).getDimension(), seed, context, loc); LogicalVariable centroidsVar = seedInput.getRight(); + LogicalVariable vectorStore = null; if (!forgy) { KMeansStageOperator recluster = oversampleAndRecluster(cop, shared, centroidsIn, centroidsVar, seed, context, loc); centroidsIn = new MutableObject<>(recluster); centroidsVar = recluster.getCandidateVariable(); + vectorStore = ((KMeansStageOperator) recluster.getInputs().get(0).getValue()).getVectorStoreVariable(); } - KMeansStageOperator lloyd = refine(cop, shared, centroidsIn, centroidsVar, context, loc); + KMeansStageOperator lloyd = refine(cop, shared, centroidsIn, centroidsVar, vectorStore, context, loc); AggregateOperator finalSet = centroidList(lloyd, context, loc); LogicalVariable cFinal = finalSet.getVariables().get(0); @@ -420,19 +422,23 @@ ref(oversample.getCandidateVariable()), kmeans(cop).getNumClusters(), seedValue, 0); recluster.getInputs().add(new MutableObject<>(oversample)); finish(recluster, context); + // Kept for the refinement loop, which would otherwise write a copy of it. + oversample.setVectorStoreVariable(oversample.getCandidateVariable()); return recluster; } /** The refinement loop: emits the k final centroids and nothing else. */ private KMeansStageOperator refine(ClusterByOperator cop, ReplicateOperator shared, - Mutable<ILogicalOperator> centroidsIn, LogicalVariable centroidsVar, IOptimizationContext context, - SourceLocation loc) throws AlgebricksException { + Mutable<ILogicalOperator> centroidsIn, LogicalVariable centroidsVar, LogicalVariable vectorStore, + IOptimizationContext context, SourceLocation loc) throws AlgebricksException { Pair<Mutable<ILogicalOperator>, LogicalVariable> input = branchOf(shared, cop.getVectorVariable(), context, loc); KMeansStageOperator lloyd = stage(cop, KMeansStageOperator.Mode.LLOYD_LOOP, context, ref(input.getRight()), ref(centroidsVar), kmeans(cop).getNumClusters(), 0L, lloydIterations(cop)); lloyd.getInputs().add(input.getLeft()); lloyd.getInputs().add(centroidsIn); + // Null under the random init: the loop keeps its own store. + lloyd.setVectorStoreVariable(vectorStore); // The execution mode is derived from the input, as GROUP BY's is: a partitioned input gives one loop // instance per partition, an unpartitioned input a single instance. The physical operators read it // (AbstractKMeansStagePOperator.unpartitioned) to size and place the loop. diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCostControllerOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCostControllerOperatorDescriptor.java index 904a96d..18ca751 100644 --- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCostControllerOperatorDescriptor.java +++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCostControllerOperatorDescriptor.java @@ -68,6 +68,8 @@ * pool and emits it as {@link KMeansVectorCodec.PoolEnvelopeWriter KIND_POOL envelopes} on <b>output 0</b>. * Output 0 is idle during the loop, so the blocking consumer cannot back-pressure the iteration.</li> * </ul> + * With {@code handOffVectors} the vector run file outlives the stage: the Lloyd loop reads and deletes it. + * <p> * The loop is acyclic in the job graph, since Release's feedback to CostLoop is the shared permit plus the * pool run file and not a data edge. The per-round seed lives in Sample, so the draws depend only on the * data. The sub-graph works on any topology: the co-located Op1/Op3/Op5 share an NC's joblet state, and the @@ -84,6 +86,8 @@ private static final int OUT_SIGMA = 1; // per-round local potential -> PhiMerge (SCALAR_RD) private final String loopKey; + // Leave the vector run file for the Lloyd loop (KMeansStageOperator#getVectorStoreVariable). + private final boolean handOffVectors; private final int vectorColumn; // vector column in input 0 private final int seedColumn; // vector column in input 1 (the seed) private final int loopRounds; // N oversampling rounds @@ -95,10 +99,12 @@ private final VectorSimilarityMetric metric; public KMeansCostControllerOperatorDescriptor(IOperatorDescriptorRegistry spec, - RecordDescriptor poolEnvelopeRecDesc, RecordDescriptor sigmaRecDesc, String loopKey, int vectorColumn, - int seedColumn, int loopRounds, int framesLimit, int dimension, VectorSimilarityMetric metric) { + RecordDescriptor poolEnvelopeRecDesc, RecordDescriptor sigmaRecDesc, String loopKey, boolean handOffVectors, + int vectorColumn, int seedColumn, int loopRounds, int framesLimit, int dimension, + VectorSimilarityMetric metric) { super(spec, 2, 2); this.loopKey = loopKey; + this.handOffVectors = handOffVectors; this.vectorColumn = vectorColumn; this.seedColumn = seedColumn; this.loopRounds = loopRounds; @@ -277,6 +283,8 @@ // successful one -- otherwise they stay on disk until joblet cleanup. MaterializerTaskState poolState = null; MaterializerTaskState vectorState = null; + // Handed off only on success. + boolean vectorsHandedOff = false; try { // Registered by StoreActivity, which addBlockingEdge (contributeActivities) joins ahead // of this activity -- so these are already present; no wait is warranted. @@ -288,6 +296,7 @@ LoopControlState.vectorsStateId(loopKey, partition)); runLoop(control, poolState, vectorState, sigmaWriter); emitWeighPartials(poolState, vectorState, poolWriter); + vectorsHandedOff = handOffVectors; } catch (InterruptedException e) { Thread.currentThread().interrupt(); poolWriter.fail(); @@ -300,12 +309,15 @@ } finally { // Release has appended every round and Sample has read for the last time (the final // awaitTurn orders both before this), and the terminal weigh was their last reader here, so - // the pool and vector run files are dead: delete them now rather than at joblet cleanup. + // the pool run file is dead, and the vector run file too unless handed off: delete them now + // rather than at joblet cleanup. try { try { KMeansLoopIO.discard(poolState); } finally { - KMeansLoopIO.discard(vectorState); + if (!vectorsHandedOff) { + KMeansLoopIO.discard(vectorState); + } } } finally { poolWriter.close(); diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLloydControllerOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLloydControllerOperatorDescriptor.java index d1d9158..ba951aa 100644 --- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLloydControllerOperatorDescriptor.java +++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLloydControllerOperatorDescriptor.java @@ -48,10 +48,11 @@ import org.apache.hyracks.dataflow.std.base.AbstractOperatorNodePushable; import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable; import org.apache.hyracks.dataflow.std.misc.MaterializerTaskState; +import org.apache.hyracks.dataflow.std.misc.SinkOperatorNodePushable; /** - * The Lloyd loop head: materializes this partition's vectors once, then runs every refinement iteration - * against them without leaving the operator. + * The Lloyd loop head: materializes this partition's vectors once (or reads the oversampling loop's file), then + * runs every refinement iteration against them without leaving the operator. * <p> * An iteration is one assignment pass: each resident vector is charged to its nearest current centroid, and the * partition emits one {@code (count, sum)} partial per centroid that attracted anything. The reduce that turns @@ -59,7 +60,7 @@ * network twice — out as partials, back as centroids — on ordinary pipelined connectors. Iteration is paced by * this partition's permit: the head emits, then parks until the tail has published the new centroids. * <p> - * Iterating inside one operator keeps the plan a fixed size: the vectors are written to a single run file and + * Iterating inside one operator keeps the plan a fixed size: the vectors sit in a single run file that is * re-streamed each round, so neither the graph nor the materialization grows with the iteration count. It also * makes a data-dependent count expressible, since the graph is the same however many times it runs, though the * count passed in today is a constant. @@ -75,6 +76,9 @@ private static final int OUT_CENTROIDS = 0; // the final centroid set (plain vectors), downstream private static final int OUT_PARTIALS = 1; // per-iteration (count, sum) partials -> CentroidMerge private final String loopKey; + // Key of the oversampling loop whose vector file this loop reads, or null. Both loops run on the same + // locations; each partition file set covers the input once, which is all the update needs. + private final String sharedVectorsKey; private final int vectorColumn; // vector column in input 0 private final int centroidColumn; // vector column in input 1 (the initial centroids) private final int iterations; @@ -88,11 +92,13 @@ private final VectorSimilarityMetric metric; public KMeansLloydControllerOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor centroidRecDesc, - RecordDescriptor partialRecDesc, String loopKey, int vectorColumn, int centroidColumn, int iterations, - int numClusters, int framesLimit, int dimension, VectorSimilarityMetric metric) { + RecordDescriptor partialRecDesc, String loopKey, String sharedVectorsKey, int vectorColumn, + int centroidColumn, int iterations, int numClusters, int framesLimit, int dimension, + VectorSimilarityMetric metric) { super(spec, 2, 2); this.metric = metric; this.loopKey = loopKey; + this.sharedVectorsKey = sharedVectorsKey; this.vectorColumn = vectorColumn; this.centroidColumn = centroidColumn; this.iterations = iterations; @@ -120,7 +126,10 @@ builder.addBlockingEdge(storeCentroids, loop); } - /** Materializes the partition's resident vectors into a raw-double run file for repeated re-streaming. */ + /** + * Materializes the partition's resident vectors into a raw-double run file for repeated re-streaming. With a + * shared vector file, the input is drained unread. + */ private final class StoreVectorsActivity extends AbstractActivityNode { private static final long serialVersionUID = 1L; @@ -131,6 +140,9 @@ @Override public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx, IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) { + if (sharedVectorsKey != null) { + return new SinkOperatorNodePushable(); + } final RecordDescriptor inRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0); return new AbstractUnaryInputSinkOperatorNodePushable() { private final FrameTupleAccessor accessor = new FrameTupleAccessor(inRecDesc); @@ -346,8 +358,9 @@ // ahead of this activity -- so these are already present; no wait is warranted. control = (LoopControlState) LoopControlState.required(ctx, LoopControlState.controlStateId(loopKey, partition)); - vectorState = (MaterializerTaskState) LoopControlState.required(ctx, - LoopControlState.vectorsStateId(loopKey, partition)); + // A shared file is complete: the oversampling loop finished before it seeded this loop. + vectorState = (MaterializerTaskState) LoopControlState.required(ctx, LoopControlState + .vectorsStateId(sharedVectorsKey != null ? sharedVectorsKey : loopKey, partition)); runLoop(control, vectorState, partialWriter); // The last iteration was the vectors' last reader; the final emit reads only the // centroid store, so the (input-sized) vector file can go before it. diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/KMeansStageOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/KMeansStageOperator.java index cfa09dd..0c6913c 100644 --- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/KMeansStageOperator.java +++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/KMeansStageOperator.java @@ -101,6 +101,9 @@ private final int dimension; // The metric's canonical name; a String since the metric enum lives above Algebricks. private final String metric; + // The vector store shared by the two loops of one expansion, named by the OVERSAMPLE_LOOP's candidate + // variable: that loop keeps its file, the LLOYD_LOOP reads it. Null when a stage keeps its own store. + private LogicalVariable vectorStoreVar; public KMeansStageOperator(Mutable<ILogicalExpression> vectorRef, Mutable<ILogicalExpression> poolRef, LogicalVariable candidateVar, Object candidateVarType, int topCount, Mode mode, long seed, int loopRounds, @@ -199,6 +202,14 @@ this.candidateVar = v; } + public LogicalVariable getVectorStoreVariable() { + return vectorStoreVar; + } + + public void setVectorStoreVariable(LogicalVariable v) { + this.vectorStoreVar = v; + } + public int getTopCount() { return topCount; } diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java index b1afc6f..186b956 100644 --- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java +++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java @@ -546,6 +546,7 @@ new VariableReferenceExpression(deepCopyVariable(op.getPoolVariable()))), deepCopyVariable(op.getCandidateVariable()), op.getCandidateVarType(), op.getTopCount(), op.getMode(), op.getSeed(), op.getLoopRounds(), op.getDimension(), op.getMetric()); + opCopy.setVectorStoreVariable(deepCopyVariable(op.getVectorStoreVariable())); deepCopyInputsAnnotationsAndExecutionMode(op, arg, opCopy); return opCopy; } diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java index e153758..7ba1c20 100644 --- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java +++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/OperatorDeepCopyVisitor.java @@ -251,6 +251,7 @@ new MutableObject<ILogicalExpression>(new VariableReferenceExpression(op.getPoolVariable())), op.getCandidateVariable(), op.getCandidateVarType(), op.getTopCount(), op.getMode(), op.getSeed(), op.getLoopRounds(), op.getDimension(), op.getMetric()); + opCopy.setVectorStoreVariable(op.getVectorStoreVariable()); return opCopy; } diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SubstituteVariableVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SubstituteVariableVisitor.java index 32c57e3..5b65251 100644 --- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SubstituteVariableVisitor.java +++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/SubstituteVariableVisitor.java @@ -328,6 +328,9 @@ if (op.getCandidateVariable().equals(pair.getLeft())) { op.setCandidateVariable(pair.getRight()); } + if (pair.getLeft().equals(op.getVectorStoreVariable())) { + op.setVectorStoreVariable(pair.getRight()); + } return null; } -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21985?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: I2d34e7e35ae889edbce294e96889d187186c9d14 Gerrit-Change-Number: 21985 Gerrit-PatchSet: 1 Gerrit-Owner: Shahrzad Shirazi <[email protected]>
