You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by sh...@apache.org on 2026/09/30 20:39:16 UTC
(asterixdb) branch master updated: [NO ISSUE][RT] Delete CLUSTER BY run files
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new dcc5f7fb6e [NO ISSUE][RT] Delete CLUSTER BY run files
dcc5f7fb6e is described below
commit dcc5f7fb6ea16ba1ff6a9834806efde89e074353
Author: Shahrzad Shirazi <Sh...@couchbase.com>
AuthorDate: Tue Sep 29 14:57:55 2026 -0700
[NO ISSUE][RT] Delete CLUSTER BY run files
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
The k-means loops materialize their vectors, candidate pool
and per-round columns as managed workspace files, and the
merge stages spill sort runs. Several of these were only
reclaimed at joblet cleanup, so a long loop over a large
input held its whole footprint on disk until the job ended.
Delete each file as soon as its last reader is done: the
loop heads discard the vector and pool files after the final
round, the sort owners discard their runs after each fold,
and a store activity that fails discards its half-written
file instead of leaving it behind. Sort runs are the subtle
case: a delete-on-close reader only deletes if it was
opened, so runs a merge never reached stayed on disk.
RECLUSTER partitions other than 0 no longer read and sort an
input whose result they never emit.
Ext-ref: MB-74142
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
Change-Id: I737896be5bac77759570b3f0e5c3b9ee56262fc4
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21938
Reviewed-by: Hongyu Shi <le...@gmail.com>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
---
.../KMeansCentroidMergeOperatorDescriptor.java | 36 +++++--
.../KMeansCostControllerOperatorDescriptor.java | 49 ++++-----
.../KMeansLloydControllerOperatorDescriptor.java | 44 ++++++---
.../runtime/operators/kmeans/KMeansLoopIO.java | 40 ++++++++
.../kmeans/KMeansPoolMergeOperatorDescriptor.java | 32 ++++--
.../kmeans/KMeansReclusterOperatorDescriptor.java | 50 +++++++---
.../operators/kmeans/KMeansStageRuntime.java | 9 ++
.../kmeans/KMeansReclusterOperatorTest.java | 110 ++++++++++++++++++---
.../kmeans/KMeansRunFileLifecycleTest.java | 71 ++++++++++++-
9 files changed, 361 insertions(+), 80 deletions(-)
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCentroidMergeOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCentroidMergeOperatorDescriptor.java
index 1dac9bef2e..fee238490a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCentroidMergeOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansCentroidMergeOperatorDescriptor.java
@@ -163,23 +163,33 @@ public class KMeansCentroidMergeOperatorDescriptor extends AbstractSingleActivit
}
}
+ /** Deletes the current iteration's sort runs and forgets the sort. */
+ private void discardSort() {
+ KMeansLoopIO.discardRuns(partialSort);
+ partialSort = null;
+ }
+
private void emitCentroids(int iter) throws HyracksDataException {
if (partialSort != null) {
- if (sortAppender.getTupleCount() > 0) {
- partialSort.nextFrame(sortFrame.getBuffer());
- }
- partialSort.close();
long[] weights = new long[maxSeq + 1];
double[][] sums = new double[maxSeq + 1][];
- foldSorted(weights, sums);
+ try {
+ if (sortAppender.getTupleCount() > 0) {
+ partialSort.nextFrame(sortFrame.getBuffer());
+ }
+ partialSort.close();
+ foldSorted(weights, sums);
+ } finally {
+ // Folded (or failed): the iteration's runs have no further reader.
+ discardSort();
+ }
int emittedSeq = 0;
for (int i = 0; i < weights.length; i++) {
if (weights[i] > 0) {
emitCentroid(iter, emittedSeq++, KMeansLoopIO.centroidOf(sums[i], weights[i], metric));
}
}
- partialSort = null; // next iteration gets a fresh sort
- maxSeq = -1;
+ maxSeq = -1; // next iteration gets a fresh sort
}
emitEnd(iter);
appender.write(writer, true);
@@ -282,12 +292,20 @@ public class KMeansCentroidMergeOperatorDescriptor extends AbstractSingleActivit
@Override
public void fail() throws HyracksDataException {
- writer.fail();
+ try {
+ discardSort(); // an iteration cut short never folds, so its spilled partials have no reader
+ } finally {
+ writer.fail();
+ }
}
@Override
public void close() throws HyracksDataException {
- writer.close();
+ try {
+ discardSort(); // an iteration still open at close never folds either
+ } finally {
+ writer.close();
+ }
}
};
}
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 ff9581403e..904a96dae6 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
@@ -211,6 +211,9 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
@Override
public void close() throws HyracksDataException {
+ if (state == null) {
+ return; // open() threw, or fail() already deleted the run file: nothing to publish
+ }
flushToState();
if (vectors) {
// The vector file is fully written; close the writer handle now (readers open their own
@@ -224,10 +227,9 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
@Override
public void fail() throws HyracksDataException {
// The run file is created in open() and only registered in close(), so on this path nothing
- // else holds a reference and nothing else will close it.
- if (state != null) {
- state.close();
- }
+ // else holds a reference and nothing else will close or delete it.
+ KMeansLoopIO.discard(state);
+ state = null;
}
};
}
@@ -271,9 +273,10 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
final IFrameWriter sigmaWriter = writers[OUT_SIGMA];
poolWriter.open();
sigmaWriter.open();
- // Held outside the try so the finally can close the pool handle on every path, not just the
- // successful one -- an aborted loop otherwise leaves this writer open until joblet cleanup.
+ // Held outside the try so the finally can delete both run files on every path, not just the
+ // successful one -- otherwise they stay on disk until joblet cleanup.
MaterializerTaskState poolState = null;
+ MaterializerTaskState vectorState = null;
try {
// Registered by StoreActivity, which addBlockingEdge (contributeActivities) joins ahead
// of this activity -- so these are already present; no wait is warranted.
@@ -281,7 +284,7 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
LoopControlState.controlStateId(loopKey, partition));
poolState = (MaterializerTaskState) LoopControlState.required(ctx,
LoopControlState.poolStateId(loopKey, partition));
- MaterializerTaskState vectorState = (MaterializerTaskState) LoopControlState.required(ctx,
+ vectorState = (MaterializerTaskState) LoopControlState.required(ctx,
LoopControlState.vectorsStateId(loopKey, partition));
runLoop(control, poolState, vectorState, sigmaWriter);
emitWeighPartials(poolState, vectorState, poolWriter);
@@ -295,14 +298,19 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
sigmaWriter.fail();
throw HyracksDataException.create(e);
} finally {
- // Release has appended every round and Sample/Cost have read for the last time, so the
- // pool writer handle can go (StoreVectors already closed the vector writer). The managed
- // workspace files themselves are reclaimed at joblet cleanup.
- if (poolState != null) {
- poolState.close();
+ // 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.
+ try {
+ try {
+ KMeansLoopIO.discard(poolState);
+ } finally {
+ KMeansLoopIO.discard(vectorState);
+ }
+ } finally {
+ poolWriter.close();
+ sigmaWriter.close();
}
- poolWriter.close();
- sigmaWriter.close();
}
}
@@ -328,10 +336,7 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
for (int r = 0; r < loopRounds; r++) {
// A fresh column per round; the previous one is dead once Op3 consumed it, which the
// loop's ordering guarantees has happened before this round begins.
- if (scoreState != null) {
- scoreState.close();
- scoreState.deleteFile();
- }
+ KMeansLoopIO.discard(scoreState);
scoreState = LoopControlState.sharedRunFile(ctx,
LoopControlState.scoreStateId(loopKey, partition));
ctx.setStateObject(scoreState);
@@ -376,10 +381,7 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
}
}
} finally {
- if (scoreState != null) {
- scoreState.close();
- scoreState.deleteFile();
- }
+ KMeansLoopIO.discard(scoreState);
}
}
@@ -459,8 +461,7 @@ public class KMeansCostControllerOperatorDescriptor extends AbstractOperatorDesc
index, count, sum));
envelope.flush();
} finally {
- weighColumn.close();
- weighColumn.deleteFile();
+ KMeansLoopIO.discard(weighColumn);
}
}
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 09d109d664..d1d91583aa 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
@@ -187,6 +187,9 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
@Override
public void close() throws HyracksDataException {
+ if (state == null) {
+ return; // open() threw, or fail() already deleted the run file: nothing to publish
+ }
flushToState();
// Fully written; readers open their own independent handles via createReader.
state.close();
@@ -195,10 +198,10 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
@Override
public void fail() throws HyracksDataException {
- // Created in open(), registered only in close(): on this path nothing else will close it.
- if (state != null) {
- state.close();
- }
+ // Created in open(), registered only in close(): on this path nothing else will close or
+ // delete it, and the loop that would read it never runs.
+ KMeansLoopIO.discard(state);
+ state = null;
}
};
}
@@ -225,6 +228,7 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
private final KMeansVectorCodec.ListVectorDecoder decoder =
new KMeansVectorCodec.ListVectorDecoder(dimension);
private boolean building;
+ private boolean failed;
private LoopControlState control;
@Override
@@ -268,8 +272,8 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
public void close() throws HyracksDataException {
// No end marker here -- close IS the end of the seed set. An empty seed still publishes,
// as an empty set, which is what the loop's first iteration then reads.
- if (control == null) {
- return; // open() threw; no seed set to publish
+ if (control == null || failed) {
+ return; // open() threw, or fail() discarded the store: no seed set to publish
}
CentroidStore store = control.getCentroids();
if (!building) {
@@ -283,8 +287,12 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
public void fail() throws HyracksDataException {
// The seed set never lands, so the loop body cannot run and the tail will never release.
// Guarded like close(): both still run after an open() that threw.
+ failed = true;
if (control != null) {
control.abort();
+ // Nothing will read the half-built set: the loop sits behind this activity's blocking
+ // edge and the tail only receives what the loop emits. Delete it now.
+ control.getCentroids().destroy();
}
}
};
@@ -332,14 +340,18 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
// Held outside the try so the finally can release the centroid handoff on every path. The
// loop is over by then either way, and nothing reads the store after this activity ends.
LoopControlState control = null;
+ MaterializerTaskState vectorState = null;
try {
// Registered by the store activities, which addBlockingEdge (contributeActivities) joins
// ahead of this activity -- so these are already present; no wait is warranted.
control = (LoopControlState) LoopControlState.required(ctx,
LoopControlState.controlStateId(loopKey, partition));
- MaterializerTaskState vectorState = (MaterializerTaskState) LoopControlState.required(ctx,
+ vectorState = (MaterializerTaskState) LoopControlState.required(ctx,
LoopControlState.vectorsStateId(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.
+ KMeansLoopIO.discard(vectorState);
emitFinalCentroids(control, centroidWriter, partition);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -351,11 +363,18 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
partialWriter.fail();
throw HyracksDataException.create(e);
} finally {
- if (control != null) {
- control.getCentroids().destroy();
+ try {
+ try {
+ KMeansLoopIO.discard(vectorState); // no-op when already deleted above
+ } finally {
+ if (control != null) {
+ control.getCentroids().destroy();
+ }
+ }
+ } finally {
+ centroidWriter.close();
+ partialWriter.close();
}
- centroidWriter.close();
- partialWriter.close();
}
}
@@ -395,8 +414,7 @@ public class KMeansLloydControllerOperatorDescriptor extends AbstractOperatorDes
iter, index, count, sum));
}
} finally {
- column.close();
- column.deleteFile();
+ KMeansLoopIO.discard(column);
}
emitEnd(partialWriter, appender, tb, it);
appender.write(partialWriter, true);
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLoopIO.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLoopIO.java
index 37e89b5336..50a72e8a8a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLoopIO.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansLoopIO.java
@@ -46,8 +46,10 @@ import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import org.apache.hyracks.dataflow.common.data.marshalling.DoubleSerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.common.io.RunFileReader;
import org.apache.hyracks.dataflow.std.misc.MaterializerTaskState;
+import org.apache.hyracks.dataflow.std.sort.AbstractSortRunGenerator;
import org.apache.hyracks.storage.am.vector.api.IVTreeDistanceFunction;
/**
@@ -381,6 +383,44 @@ public final class KMeansLoopIO {
return sink -> streamRawVectors(state, ctx, sink);
}
+ /**
+ * Deletes a run file the moment its last reader is done with it, instead of leaving it in the joblet
+ * workspace until the job ends. Closes the writer handle first. Null-safe and idempotent, so it can sit in
+ * both the success path and a {@code finally}.
+ */
+ public static void discard(MaterializerTaskState state) throws HyracksDataException {
+ if (state == null) {
+ return;
+ }
+ try {
+ state.close();
+ } finally {
+ state.deleteFile();
+ }
+ }
+
+ /**
+ * Deletes every run file a sort generated. The runs are delete-on-close readers, but a reader only deletes
+ * if it was opened, so runs a merge never reached -- because it failed part-way, or never ran -- would
+ * otherwise stay on disk until the job ends. Runs the merge already consumed are simply gone. Null-safe and
+ * idempotent; never throws, so it cannot mask the error that sent a caller to its {@code finally}.
+ */
+ public static void discardRuns(AbstractSortRunGenerator sort) {
+ if (sort == null) {
+ return;
+ }
+ List<GeneratedRunFileReader> runs = sort.getRuns();
+ for (GeneratedRunFileReader run : runs) {
+ try {
+ run.close();
+ } catch (HyracksDataException e) {
+ // Still delete below: a handle that would not close must not keep the file.
+ }
+ run.getFile().delete();
+ }
+ runs.clear();
+ }
+
/**
* An ordered sequence of vectors that stays in the heap while it fits the frame budget and moves to a run
* file when it does not.
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansPoolMergeOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansPoolMergeOperatorDescriptor.java
index a368e1f772..fd0e6f90f2 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansPoolMergeOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansPoolMergeOperatorDescriptor.java
@@ -146,18 +146,28 @@ public class KMeansPoolMergeOperatorDescriptor extends AbstractSingleActivityOpe
// no sort was ever opened; the end marker still has to go out so Release wakes Cost for the next
// round.
if (drawSort != null) {
- if (sortAppender.getTupleCount() > 0) {
- drawSort.nextFrame(sortFrame.getBuffer());
+ try {
+ if (sortAppender.getTupleCount() > 0) {
+ drawSort.nextFrame(sortFrame.getBuffer());
+ }
+ drawSort.close();
+ emitSorted(round);
+ } finally {
+ // The round's union is out (or failed): its runs have no further reader.
+ discardSort();
}
- drawSort.close();
- emitSorted(round);
- drawSort = null; // next round gets a fresh sort
}
emitEnd(round);
appender.write(writer, true);
writer.flush();
}
+ /** Deletes the current round's sort runs and forgets the sort; the next round gets a fresh one. */
+ private void discardSort() {
+ KMeansLoopIO.discardRuns(drawSort);
+ drawSort = null;
+ }
+
private void ensureSort() throws HyracksDataException {
if (drawSort == null) {
drawSort = new ExternalSortRunGenerator(ctx, KMeansLoopIO.DRAW_SORT_FIELDS, null,
@@ -247,12 +257,20 @@ public class KMeansPoolMergeOperatorDescriptor extends AbstractSingleActivityOpe
@Override
public void fail() throws HyracksDataException {
- writer.fail();
+ try {
+ discardSort(); // a round cut short never emits, so its spilled draws have no reader
+ } finally {
+ writer.fail();
+ }
}
@Override
public void close() throws HyracksDataException {
- writer.close();
+ try {
+ discardSort(); // a round still open at close never emits either
+ } finally {
+ writer.close();
+ }
}
};
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorDescriptor.java
index 49240af6fe..2858a5845b 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorDescriptor.java
@@ -113,6 +113,7 @@ public final class KMeansReclusterOperatorDescriptor extends AbstractOperatorDes
means.add(KMeansLoopIO.centroidOf(sum, weight, metric));
memberWeights[meanCount[0]++] = weight;
});
+ rt.discard(); // the partials are folded into the means; their sort runs are dead from here on
means.seal();
// Fewer means than requested: the pool holds fewer distinct candidates than k, because the input has
@@ -224,6 +225,7 @@ public final class KMeansReclusterOperatorDescriptor extends AbstractOperatorDes
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
return new AbstractUnaryInputSinkOperatorNodePushable() {
private MaterializerTaskState state;
+ private boolean failed;
@Override
public void open() throws HyracksDataException {
@@ -240,16 +242,24 @@ public final class KMeansReclusterOperatorDescriptor extends AbstractOperatorDes
@Override
public void close() throws HyracksDataException {
// state can be null if open() failed; close() is called either way.
- if (state != null) {
- state.close();
- ctx.setStateObject(state);
+ if (state == null) {
+ return;
}
+ if (failed) {
+ // A partial input is never scored (Score sits behind the blocking edge), so the file
+ // has no reader: delete it rather than publish it.
+ KMeansLoopIO.discard(state);
+ state = null;
+ return;
+ }
+ state.close();
+ ctx.setStateObject(state);
}
@Override
public void fail() throws HyracksDataException {
- // Nothing to do: close() always follows and closes the run file. (The oversampling pool
- // store closes here instead, because its writer stays open across the loop.)
+ // close() always follows, and deletes the run file.
+ failed = true;
}
};
}
@@ -270,20 +280,34 @@ public final class KMeansReclusterOperatorDescriptor extends AbstractOperatorDes
@Override
public void initialize() throws HyracksDataException {
writer.open();
+ MaterializerTaskState poolState = null;
+ KMeansStageRuntime rt = null;
try {
- MaterializerTaskState poolState = (MaterializerTaskState) ctx.getStateObject(
+ poolState = (MaterializerTaskState) ctx.getStateObject(
new TaskId(new ActivityId(getOperatorId(), STORE_POOL_ACTIVITY_ID), partition));
- KMeansStageRuntime rt =
- new KMeansStageRuntime(ctx, writer, vecRecDesc, poolColumn, framesLimit);
- rt.readInput(poolState);
- KMeansStageRuntime.Emitter emitter = rt.newEmitter();
- emitRecluster(ctx, new TaskId(getActivityId(), partition), rt, emitter, partition);
- emitter.flush();
+ if (partition == 0) {
+ // Only partition 0 speaks (see emitRecluster). Elsewhere, reading the input would
+ // only spill sort runs nothing merges, so the input is dropped unread.
+ rt = new KMeansStageRuntime(ctx, writer, vecRecDesc, poolColumn, framesLimit);
+ rt.readInput(poolState);
+ KMeansStageRuntime.Emitter emitter = rt.newEmitter();
+ emitRecluster(ctx, new TaskId(getActivityId(), partition), rt, emitter, partition);
+ emitter.flush();
+ }
} catch (Exception e) {
writer.fail();
throw HyracksDataException.create(e);
} finally {
- writer.close();
+ try {
+ // readInput already deleted the input after its one read; this covers the paths
+ // that never got there. Then any sort runs the fold left behind.
+ KMeansLoopIO.discard(poolState);
+ } finally {
+ if (rt != null) {
+ rt.discard();
+ }
+ writer.close();
+ }
}
}
};
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansStageRuntime.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansStageRuntime.java
index d16d64b1d9..d01deb4725 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansStageRuntime.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/kmeans/KMeansStageRuntime.java
@@ -123,6 +123,15 @@ final class KMeansStageRuntime {
return poolSize;
}
+ /**
+ * Deletes the partial sort's run files. The fold's merge consumes them, but only the ones it opened, so a
+ * failure part-way (or before the fold) would leave the rest on disk until the job ends. Idempotent.
+ */
+ void discard() {
+ KMeansLoopIO.discardRuns(partialSort);
+ partialSort = null;
+ }
+
/** Re-emits one weigh partial into the sort, flat, keyed by (seq, part). */
private void spillPartial(Row r) throws HyracksDataException {
partialTb.reset();
diff --git a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorTest.java b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorTest.java
index d62a785792..8971db8d63 100644
--- a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorTest.java
+++ b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansReclusterOperatorTest.java
@@ -19,6 +19,7 @@
package org.apache.asterix.runtime.operators.kmeans;
+import java.io.File;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -39,6 +40,7 @@ import org.apache.hyracks.api.dataflow.IActivity;
import org.apache.hyracks.api.dataflow.IActivityGraphBuilder;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
+import org.apache.hyracks.api.dataflow.TaskId;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
@@ -253,7 +255,28 @@ public class KMeansReclusterOperatorTest {
*/
@Test
public void reclusterFoldIsIdenticalWhetherOrNotTheSortSpills() throws Exception {
- // Enough partials to exceed a 4-frame budget (3 usable x 32 KB) but sit comfortably in a large one.
+ double[][] rows = spillingRows();
+ List<double[]> spilled = runRecluster(rows, 4);
+ List<double[]> inMemory = runRecluster(rows, 4096);
+
+ Assert.assertEquals("centroid count differed between budgets", inMemory.size(), spilled.size());
+ Assert.assertFalse("fixture produced no centroids; the test would prove nothing", spilled.isEmpty());
+ for (int i = 0; i < spilled.size(); i++) {
+ double[] a = spilled.get(i);
+ double[] b = inMemory.get(i);
+ Assert.assertEquals("centroid " + i + " width", b.length, a.length);
+ for (int d = 0; d < a.length; d++) {
+ Assert.assertEquals("centroid " + i + " component " + d + " differed with the frame budget",
+ Double.doubleToLongBits(b[d]), Double.doubleToLongBits(a[d]));
+ }
+ }
+ }
+
+ /**
+ * Enough partials to exceed a 4-frame budget (3 usable x 32 KB) but sit comfortably in a large one. Partials
+ * arrive scrambled across partitions, as a concurrent M-to-1 connector would deliver them.
+ */
+ private static double[][] spillingRows() {
final int members = 600;
final int partitionsPerMember = 5;
double[][] rows = new double[members + members * partitionsPerMember][];
@@ -261,26 +284,87 @@ public class KMeansReclusterOperatorTest {
for (int m = 0; m < members; m++) {
rows[r++] = new double[] { 0, 0, m, 0, m, m + 1 }; // pool member m
}
- // Partials arrive scrambled across partitions, as a concurrent M-to-1 connector would deliver them.
for (int p = partitionsPerMember - 1; p >= 0; p--) {
for (int m = members - 1; m >= 0; m--) {
rows[r++] = new double[] { 2, p, m, 1, 0.1d * (p + 1), 1.0e8d * (p + 1) };
}
}
- List<double[]> spilled = runRecluster(rows, 4);
- List<double[]> inMemory = runRecluster(rows, 4096);
+ return rows;
+ }
- Assert.assertEquals("centroid count differed between budgets", inMemory.size(), spilled.size());
- Assert.assertFalse("fixture produced no centroids; the test would prove nothing", spilled.isEmpty());
- for (int i = 0; i < spilled.size(); i++) {
- double[] a = spilled.get(i);
- double[] b = inMemory.get(i);
- Assert.assertEquals("centroid " + i + " width", b.length, a.length);
- for (int d = 0; d < a.length; d++) {
- Assert.assertEquals("centroid " + i + " component " + d + " differed with the frame budget",
- Double.doubleToLongBits(b[d]), Double.doubleToLongBits(a[d]));
+ /** Workspace files RECLUSTER can create: its stored input, and the partial sort's runs and merge files. */
+ private static final String[] WORKSPACE_PREFIXES =
+ { "MaterializerTaskState", "ExternalSortRunGenerator", "ExternalSortRunMerger" };
+
+ /** Those files currently on disk; the test context keeps managed workspace files in the JVM temp dir. */
+ private static int liveFiles() {
+ File[] files = new File(System.getProperty("java.io.tmpdir")).listFiles();
+ int n = 0;
+ for (File f : files == null ? new File[0] : files) {
+ for (String prefix : WORKSPACE_PREFIXES) {
+ if (f.getName().startsWith(prefix)) {
+ n++;
+ break;
+ }
}
}
+ return n;
+ }
+
+ /**
+ * A spilling RECLUSTER deletes everything it wrote by the time it returns: the stored input after its
+ * single read, and every sort run once the fold is done with it -- not at job end.
+ */
+ @Test
+ public void aSpillingReclusterLeavesNoFilesBehind() throws Exception {
+ int before = liveFiles();
+ List<double[]> out = runRecluster(spillingRows(), 4);
+ Assert.assertFalse("fixture produced no centroids; the test would prove nothing", out.isEmpty());
+ Assert.assertEquals("RECLUSTER left workspace files behind", before, liveFiles());
+ }
+
+ /**
+ * Only partition 0 speaks, so any other partition's copy of the broadcast input is dead on arrival: it is
+ * deleted unread, and no sort runs are spilled for a fold that would never happen.
+ */
+ @Test
+ public void aNonSpeakingPartitionDeletesItsInputUnread() throws Exception {
+ IHyracksTaskContext ctx = TestUtils.create(32768);
+ KMeansReclusterOperatorDescriptor op = new KMeansReclusterOperatorDescriptor(new JobSpecification(),
+ VEC_REC_DESC, 8, 0, 4, DEFAULT_RECLUSTER_SEED, VectorSimilarityMetric.EUCLIDEAN_SQUARED);
+ List<IActivity> activities = collectActivities(op);
+ IRecordDescriptorProvider rdp = recordDescProvider();
+ int before = liveFiles();
+ IOperatorNodePushable poolStore = activities.get(0).createPushRuntime(ctx, rdp, 1, 2);
+ poolStore.getInputFrameWriter(0).open();
+ poolStore.getInputFrameWriter(0)
+ .nextFrame(envelopesFrame(ctx, new double[][] { { 0, 0, 0, 0, 1, 2 }, { 2, 1, 0, 1, 1, 2 } }));
+ poolStore.getInputFrameWriter(0).close();
+ Assert.assertEquals("the stored input is one file", before + 1, liveFiles());
+
+ List<double[]> out = collectOutput(activities.get(1).createPushRuntime(ctx, rdp, 1, 2), false);
+ Assert.assertTrue("only partition 0 emits", out.isEmpty());
+ Assert.assertEquals("the input must be deleted, not kept to job end", before, liveFiles());
+ }
+
+ /** An input store that failed part-way is never scored, so it deletes its file instead of publishing it. */
+ @Test
+ public void aFailedInputStoreLeavesNoFile() throws Exception {
+ IHyracksTaskContext ctx = TestUtils.create(32768);
+ KMeansReclusterOperatorDescriptor op =
+ new KMeansReclusterOperatorDescriptor(new JobSpecification(), VEC_REC_DESC, 2, 0, TEST_FRAMES_LIMIT,
+ DEFAULT_RECLUSTER_SEED, VectorSimilarityMetric.EUCLIDEAN_SQUARED);
+ List<IActivity> activities = collectActivities(op);
+ int before = liveFiles();
+ IOperatorNodePushable poolStore = activities.get(0).createPushRuntime(ctx, recordDescProvider(), 0, 1);
+ IFrameWriter in = poolStore.getInputFrameWriter(0);
+ in.open();
+ in.nextFrame(envelopesFrame(ctx, new double[][] { { 0, 0, 0, 0, 1, 2 } }));
+ in.fail();
+ in.close(); // the framework always closes after fail
+ Assert.assertEquals("a failed input store must not leave its file", before, liveFiles());
+ Assert.assertNull("a failed input store must not publish its state",
+ ctx.getStateObject(new TaskId(activities.get(0).getActivityId(), 0)));
}
/** Drives RECLUSTER over the given envelope rows at a chosen frame budget. */
diff --git a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansRunFileLifecycleTest.java b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansRunFileLifecycleTest.java
index b06ade560a..27e4796fd9 100644
--- a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansRunFileLifecycleTest.java
+++ b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/operators/kmeans/KMeansRunFileLifecycleTest.java
@@ -24,13 +24,20 @@ import java.util.List;
import java.util.Random;
import org.apache.asterix.runtime.utils.VectorDistanceCalculation;
+import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.ActivityId;
import org.apache.hyracks.api.dataflow.OperatorDescriptorId;
import org.apache.hyracks.api.dataflow.TaskId;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.marshalling.DoubleSerializerDeserializer;
+import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import org.apache.hyracks.dataflow.std.misc.MaterializerTaskState;
+import org.apache.hyracks.dataflow.std.sort.Algorithm;
+import org.apache.hyracks.dataflow.std.sort.ExternalSortRunGenerator;
import org.apache.hyracks.storage.am.vector.api.IVTreeDistanceFunction;
import org.apache.hyracks.test.support.TestUtils;
import org.junit.Assert;
@@ -54,13 +61,18 @@ public class KMeansRunFileLifecycleTest {
/** Managed workspace files currently on disk. */
private static int liveFiles() {
+ return liveFiles(WORKSPACE_PREFIX);
+ }
+
+ /** Workspace files with the given prefix currently on disk. */
+ private static int liveFiles(String prefix) {
File[] files = new File(System.getProperty("java.io.tmpdir")).listFiles();
if (files == null) {
return 0;
}
int n = 0;
for (File f : files) {
- if (f.getName().startsWith(WORKSPACE_PREFIX)) {
+ if (f.getName().startsWith(prefix)) {
n++;
}
}
@@ -154,6 +166,63 @@ public class KMeansRunFileLifecycleTest {
Assert.assertEquals(before, liveFiles());
}
+ /** discard() deletes the file on the spot, and is safe to repeat and to call on nothing. */
+ @Test
+ public void discardDeletesTheFileAndIsIdempotent() throws Exception {
+ IHyracksTaskContext ctx = TestUtils.create(FRAME_SIZE);
+ int before = liveFiles();
+ MaterializerTaskState state = newState(ctx, 5);
+ KMeansLoopIO.ScoreColumnWriter w = new KMeansLoopIO.ScoreColumnWriter(state, ctx);
+ w.append(new double[] { 1.0d }, new int[] { 0 }, 1);
+ w.finish();
+ Assert.assertEquals(1, liveFiles() - before);
+ KMeansLoopIO.discard(state);
+ Assert.assertEquals("deleted when discarded, not at job end", before, liveFiles());
+ KMeansLoopIO.discard(state); // a finally repeating the success path's discard
+ KMeansLoopIO.discard(null); // a state that was never created
+ Assert.assertEquals(before, liveFiles());
+ }
+
+ /**
+ * A sort's runs are delete-on-close, but a run only deletes once opened, so runs no merge reached -- a merge
+ * that failed, or one that never ran -- would stay on disk to job end. discardRuns() removes them.
+ */
+ @Test
+ public void discardRunsDeletesRunsNoMergeOpened() throws Exception {
+ IHyracksTaskContext ctx = TestUtils.create(FRAME_SIZE);
+ String prefix = ExternalSortRunGenerator.class.getSimpleName();
+ int before = liveFiles(prefix);
+ ExternalSortRunGenerator sort = new ExternalSortRunGenerator(ctx, KMeansLoopIO.PARTIAL_FLAT_SORT_FIELDS, null,
+ KMeansLoopIO.PARTIAL_FLAT_COMPARATORS, KMeansLoopIO.PARTIAL_FLAT_RD, Algorithm.MERGE_SORT, 4);
+ sort.open();
+ VSizeFrame frame = new VSizeFrame(ctx);
+ FrameTupleAppender appender = new FrameTupleAppender(frame);
+ ArrayTupleBuilder tb = new ArrayTupleBuilder(4);
+ // Well past a 4-frame budget, so the generator has to spill several runs.
+ for (int i = 0; i < 20000; i++) {
+ tb.reset();
+ tb.addField(IntegerSerializerDeserializer.INSTANCE, 20000 - i);
+ tb.addField(IntegerSerializerDeserializer.INSTANCE, i % 7);
+ tb.addField(DoubleSerializerDeserializer.INSTANCE, 1.0d);
+ KMeansLoopIO.writeRawVector(tb, new double[] { i, i + 1 });
+ if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+ sort.nextFrame(frame.getBuffer());
+ appender.reset(frame, true);
+ Assert.assertTrue(appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize()));
+ }
+ }
+ sort.nextFrame(frame.getBuffer());
+ sort.close();
+ Assert.assertTrue("the fixture must spill more than one run", sort.getRuns().size() > 1);
+ Assert.assertEquals(sort.getRuns().size(), liveFiles(prefix) - before);
+
+ KMeansLoopIO.discardRuns(sort); // no merge ever opened a run
+ Assert.assertEquals("unmerged runs must be deleted", before, liveFiles(prefix));
+ Assert.assertTrue(sort.getRuns().isEmpty());
+ KMeansLoopIO.discardRuns(sort); // idempotent
+ KMeansLoopIO.discardRuns(null); // a sort that was never started
+ }
+
/** A partition with no rows must not fault the scan or the accumulator -- the width filter can empty one. */
@Test
public void anEmptyPartitionProducesNothingAndFaultsNothing() throws Exception {