Add flush() to IFrameWriter
This change introduces flush() method in frame writers. It is
intended to be used for long running jobs (Feeds) to flush contents
of frames all the way to storage.
Change-Id: I85424bab7965b71aac709280af066e1655457aa3
Reviewed-on: https://asterix-gerrit.ics.uci.edu/584
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <hubailmor@gmail.com>
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
index fefc72e..9f0960d 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansAccumulatingAggregatorFactory.java
@@ -54,8 +54,7 @@
public IAggregatorDescriptor createAggregator(IHyracksTaskContext ctx, RecordDescriptor inRecordDesc,
RecordDescriptor outRecordDescriptor, int[] keys, int[] partialKeys) throws HyracksDataException {
- final AggregatorOutput outputWriter = new AggregatorOutput(subplans, keyFieldIdx.length,
- decorFieldIdx.length);
+ final AggregatorOutput outputWriter = new AggregatorOutput(subplans, keyFieldIdx.length, decorFieldIdx.length);
final NestedTupleSourceRuntime[] pipelines = new NestedTupleSourceRuntime[subplans.length];
for (int i = 0; i < subplans.length; i++) {
try {
@@ -132,8 +131,8 @@
}
@Override
- public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor,
- int tIndex, AggregateState state) throws HyracksDataException {
+ public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
+ AggregateState state) throws HyracksDataException {
throw new IllegalStateException("this method should not be called");
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
index 2ca933a..4b63155 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/aggreg/NestedPlansRunningAggregatorFactory.java
@@ -127,8 +127,8 @@
}
@Override
- public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor,
- int tIndex, AggregateState state) throws HyracksDataException {
+ public boolean outputPartialResult(ArrayTupleBuilder tupleBuilder, IFrameTupleAccessor accessor, int tIndex,
+ AggregateState state) throws HyracksDataException {
throw new IllegalStateException("this method should not be called");
}
@@ -219,14 +219,14 @@
for (int f = 0; f < w; f++) {
tb.addField(accessor, tIndex, f);
}
- FrameUtils.appendToWriter(outputWriter, outputAppender, tb.getFieldEndOffsets(),
- tb.getByteArray(), 0, tb.getSize());
+ FrameUtils.appendToWriter(outputWriter, outputAppender, tb.getFieldEndOffsets(), tb.getByteArray(), 0,
+ tb.getSize());
}
}
@Override
public void close() throws HyracksDataException {
- outputAppender.flush(outputWriter, true);
+ outputAppender.write(outputWriter, true);
}
public void setInputIdx(int inputIdx) {
@@ -241,6 +241,11 @@
public void fail() throws HyracksDataException {
}
+ @Override
+ public void flush() throws HyracksDataException {
+ outputAppender.flush(outputWriter);
+ }
+
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
index e57a4ba..e94f4b7 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
@@ -60,7 +60,7 @@
protected void flushAndReset() throws HyracksDataException {
if (appender.getTupleCount() > 0) {
- appender.flush(writer, true);
+ appender.write(writer, true);
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/SinkRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/SinkRuntimeFactory.java
index 69876b0..d1fb703 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/SinkRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/SinkRuntimeFactory.java
@@ -57,6 +57,11 @@
@Override
public void close() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // flush() is meaningless for sink operators
+ }
};
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
index b7a73d4..2368a9c 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/group/MicroPreClusteredGroupRuntimeFactory.java
@@ -29,7 +29,6 @@
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.preclustered.PreclusteredGroupWriter;
@@ -90,6 +89,11 @@
public void close() throws HyracksDataException {
pgw.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ pgw.flush();
+ }
};
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
index 8322cdc..f2afe98 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
@@ -157,6 +157,11 @@
public void fail() throws HyracksDataException {
startOfPipeline.fail();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ startOfPipeline.flush();
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
index 25eb229..ac9bae2 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
@@ -136,7 +136,6 @@
tb.addFieldEndOffset();
}
}
-
}
IFrameWriter endPipe = new TupleOuterProduct();
@@ -166,6 +165,10 @@
}
}
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/AssignRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/AssignRuntimeFactory.java
index 79d9b7d1..1877d64 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/AssignRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/AssignRuntimeFactory.java
@@ -171,6 +171,11 @@
writer.fail();
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/EmptyTupleSourceRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/EmptyTupleSourceRuntimeFactory.java
index a2b9652..2b7c2da 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/EmptyTupleSourceRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/EmptyTupleSourceRuntimeFactory.java
@@ -52,13 +52,18 @@
if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
throw new IllegalStateException();
}
- appender.flush(writer, true);
+ appender.write(writer, true);
}
@Override
public void close() throws HyracksDataException {
writer.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/NestedTupleSourceRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/NestedTupleSourceRuntimeFactory.java
index 5fcb9ef..0f1e50d 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/NestedTupleSourceRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/NestedTupleSourceRuntimeFactory.java
@@ -69,8 +69,9 @@
writer.fail();
}
- public void forceFlush() throws HyracksDataException {
- appender.flush(writer, true);
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
}
}
}
\ No newline at end of file
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
index 3d1eb06..b7f11d8 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PartitioningSplitOperatorDescriptor.java
@@ -89,8 +89,7 @@
if (isOpen[i]) {
try {
tupleAppender.reset(writeBuffers[i], false);
- // ? by JF why didn't clear the buffer ?
- tupleAppender.flush(writers[i], false);
+ tupleAppender.write(writers[i], false);
} catch (Throwable th) {
if (hde == null) {
hde = new HyracksDataException();
@@ -114,6 +113,14 @@
}
@Override
+ public void flush() throws HyracksDataException {
+ for (int i = 0; i < outputArity; i++) {
+ tupleAppender.reset(writeBuffers[i], false);
+ tupleAppender.flush(writers[i]);
+ }
+ }
+
+ @Override
public void fail() throws HyracksDataException {
HyracksDataException hde = null;
for (int i = 0; i < outputArity; i++) {
@@ -172,6 +179,7 @@
} catch (IOException e) {
throw new HyracksDataException(e);
}
+ tupleAppender.reset(writeBuffers[outputIndex], false);
FrameUtils.appendToWriter(writers[outputIndex], tupleAppender, tupleBuilder.getFieldEndOffsets(),
tupleBuilder.getByteArray(), 0, tupleBuilder.getSize());
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PrinterRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PrinterRuntimeFactory.java
index b77f7b8..021784a 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PrinterRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/PrinterRuntimeFactory.java
@@ -59,6 +59,6 @@
public IPushRuntime createPushRuntime(IHyracksTaskContext ctx) {
IAWriter w = PrinterBasedWriterFactory.INSTANCE.createWriter(printColumns, System.out, printerFactories,
inputRecordDesc);
- return new SinkWriterRuntime(w, ctx, System.out, inputRecordDesc);
+ return new SinkWriterRuntime(w, System.out, inputRecordDesc);
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
index 8a5f38c..5a26f36 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/RunningAggregateRuntimeFactory.java
@@ -160,6 +160,10 @@
}
}
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntime.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntime.java
index c069e4c..2512c5e 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntime.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntime.java
@@ -24,14 +24,12 @@
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.data.IAWriter;
import org.apache.hyracks.algebricks.runtime.operators.base.AbstractOneInputSinkPushRuntime;
-import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
public class SinkWriterRuntime extends AbstractOneInputSinkPushRuntime {
- private final IHyracksTaskContext ctx;
private final PrintStream printStream;
private final IAWriter writer;
private RecordDescriptor inputRecordDesc;
@@ -39,18 +37,16 @@
private boolean autoClose = false;
private boolean first = true;
- public SinkWriterRuntime(IAWriter writer, IHyracksTaskContext ctx, PrintStream printStream,
- RecordDescriptor inputRecordDesc) {
+ public SinkWriterRuntime(IAWriter writer, PrintStream printStream, RecordDescriptor inputRecordDesc) {
this.writer = writer;
- this.ctx = ctx;
this.printStream = printStream;
this.inputRecordDesc = inputRecordDesc;
this.tAccess = new FrameTupleAccessor(inputRecordDesc);
}
- public SinkWriterRuntime(IAWriter writer, IHyracksTaskContext ctx, PrintStream printStream,
- RecordDescriptor inputRecordDesc, boolean autoClose) {
- this(writer, ctx, printStream, inputRecordDesc);
+ public SinkWriterRuntime(IAWriter writer, PrintStream printStream, RecordDescriptor inputRecordDesc,
+ boolean autoClose) {
+ this(writer, printStream, inputRecordDesc);
this.autoClose = autoClose;
}
@@ -95,4 +91,9 @@
@Override
public void fail() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // flush() makes no sense to sink operators
+ }
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntimeFactory.java
index 3f66e23..62e3542 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/SinkWriterRuntimeFactory.java
@@ -75,6 +75,6 @@
throw new AlgebricksException(e);
}
IAWriter w = writerFactory.createWriter(fields, filePrintStream, printerFactories, inputRecordDesc);
- return new SinkWriterRuntime(w, ctx, filePrintStream, inputRecordDesc, true);
+ return new SinkWriterRuntime(w, filePrintStream, inputRecordDesc, true);
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamLimitRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamLimitRuntimeFactory.java
index ef172c7..11f47ac 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamLimitRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamLimitRuntimeFactory.java
@@ -147,6 +147,11 @@
return lim;
}
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
+
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamProjectRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamProjectRuntimeFactory.java
index 2cea90d..001a5982 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamProjectRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamProjectRuntimeFactory.java
@@ -84,6 +84,11 @@
}
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
+
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamSelectRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamSelectRuntimeFactory.java
index 75c3d08..5eb4604 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamSelectRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StreamSelectRuntimeFactory.java
@@ -157,6 +157,11 @@
}
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
};
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StringStreamingRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StringStreamingRuntimeFactory.java
index 395d321..a9f5a21 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StringStreamingRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/StringStreamingRuntimeFactory.java
@@ -188,6 +188,11 @@
// close the following operator in the chain
super.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ ps.flush();
+ }
};
}
}
diff --git a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/UnnestRuntimeFactory.java b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/UnnestRuntimeFactory.java
index 6b21cda..2c04003 100644
--- a/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/UnnestRuntimeFactory.java
+++ b/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/UnnestRuntimeFactory.java
@@ -159,6 +159,11 @@
}
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
};
}
diff --git a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameAppender.java b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameAppender.java
index 18e44cc..e48e4d6 100644
--- a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameAppender.java
+++ b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameAppender.java
@@ -48,12 +48,22 @@
ByteBuffer getBuffer();
/**
- * Flush the frame content to the given writer.
- * Clear the inner buffer after flush if {@code clear} is <code>true</code>.
+ * Write the frame content to the given writer.
+ * Clear the inner buffer after write if {@code clear} is <code>true</code>.
*
* @param outWriter the output writer
- * @param clear indicate whether to clear the inside frame after flushed or not.
+ * @param clear indicate whether to clear the inside frame after writing or not.
* @throws HyracksDataException
*/
- void flush(IFrameWriter outWriter, boolean clear) throws HyracksDataException;
+ void write(IFrameWriter outWriter, boolean clear) throws HyracksDataException;
+
+ /**
+ * Write currently buffered records to {@code writer} then flushes {@code writer}. The inside frame is always cleared
+ * @param writer the FrameWriter to write to and flush
+ * @throws HyracksDataException
+ */
+ public default void flush(IFrameWriter writer) throws HyracksDataException {
+ write(writer, true);
+ writer.flush();
+ }
}
diff --git a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
index f6c3ad0b..87c5223 100644
--- a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
+++ b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
@@ -39,14 +39,13 @@
* <ul>
* <li>{@link IFrameWriter#close()} to give up any resources owned by the {@link IFrameWriter} and enter the CLOSED state.</li>
* <li>{@link IFrameWriter#nextFrame(ByteBuffer)} to provide data to the {@link IFrameWriter}. The call returns normally on success and the {@link IFrameWriter} remains in the OPENED state. On failure, the call throws a {@link HyracksDataException}, the {@link IFrameWriter} remains in the OPENED state.</li>
+ * <li>{@link IFrameWriter#flush()} to push tuples that are ready in the output frame. The {@link IFrameWriter} remains in the OPENED state.</li>
* <li>{@link IFrameWriter#fail()} to indicate that stream is to be aborted. The {@link IFrameWriter} enters the FAILED state.</li>
* </ul>
* In the FAILED state, the only call allowed is the {@link IFrameWriter#close()} to move the {@link IFrameWriter} into the CLOSED
* state and give up all resources.
* No calls are allowed when the {@link IFrameWriter} is in the CLOSED state.
* Note: If the call to {@link IFrameWriter#open()} failed, the {@link IFrameWriter#close()} must still be called by the producer.
- *
- * @author vinayakb
*/
public interface IFrameWriter {
/**
@@ -56,7 +55,6 @@
/**
* Provide data to the stream of this {@link IFrameWriter}.
- *
* @param buffer
* - Buffer containing data.
* @throws HyracksDataException
@@ -64,16 +62,22 @@
public void nextFrame(ByteBuffer buffer) throws HyracksDataException;
/**
+ * request the frame to push its content forward and flush its consumers
+ * @throws HyracksDataException
+ */
+ public default void flush() throws HyracksDataException {
+ throw new HyracksDataException("flush() is not supported in this IFrameWriter");
+ }
+
+ /**
* Indicate that a failure was encountered and the current stream is to be
* aborted.
- *
* @throws HyracksDataException
*/
public void fail() throws HyracksDataException;
/**
* Close this {@link IFrameWriter} and give up all resources.
- *
* @throws HyracksDataException
*/
public void close() throws HyracksDataException;
diff --git a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/IDataWriter.java b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/IDataWriter.java
index a5d3d4e..0b8ffc0 100644
--- a/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/IDataWriter.java
+++ b/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/IDataWriter.java
@@ -22,13 +22,10 @@
/**
* Accepts data from data producers.
- *
- * @author vinayakb
*/
public interface IDataWriter<T> {
/**
* Pushes data to the acceptor.
- *
* @param data
* - Data pushed to the acceptor. <code>null</code> indicates the
* end of stream.
@@ -38,14 +35,18 @@
/**
* Indicates that the stream has failed.
- *
* @throws HyracksDataException
*/
public void fail() throws HyracksDataException;
/**
+ * Request the writer to flush its content
+ * @throws HyracksDataException
+ */
+ public void flush() throws HyracksDataException;
+
+ /**
* Closes this writer.
- *
* @throws Exception
*/
public void close() throws HyracksDataException;
diff --git a/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java b/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
index ed4279d..b5f359c 100644
--- a/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
+++ b/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
@@ -118,4 +118,10 @@
}
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // At the network boundary.
+ // This frame writer always pushes its content
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/dataset/DatasetPartitionWriter.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/dataset/DatasetPartitionWriter.java
index 9fa511c..e007050 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/dataset/DatasetPartitionWriter.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/dataset/DatasetPartitionWriter.java
@@ -41,8 +41,6 @@
private final ResultSetId resultSetId;
- private final boolean asyncMode;
-
private final boolean orderedResult;
private final int partition;
@@ -63,7 +61,6 @@
this.manager = manager;
this.jobId = jobId;
this.resultSetId = rsId;
- this.asyncMode = asyncMode;
this.orderedResult = orderedResult;
this.partition = partition;
this.nPartitions = nPartitions;
@@ -139,5 +136,4 @@
}
}
}
-
}
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionInputChannel.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionInputChannel.java
index ef4c552..a1f4f5f 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionInputChannel.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionInputChannel.java
@@ -118,7 +118,7 @@
ByteBuffer destFrame = emptyQueue.poll();
buffer.position(0);
buffer.limit(buffer.capacity());
- if (destFrame.capacity() < buffer.capacity()){
+ if (destFrame.capacity() < buffer.capacity()) {
throw new HyracksDataException("should never happen");
}
destFrame.clear();
@@ -137,5 +137,10 @@
public void close() throws HyracksDataException {
monitor.notifyEndOfStream(MaterializedPartitionInputChannel.this);
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // materialize operators should only send their output once all of their input has been consumed. hence, this is a no op
+ }
}
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionWriter.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionWriter.java
index 291064d..09d62c5 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionWriter.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializedPartitionWriter.java
@@ -103,4 +103,9 @@
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // materialize writer is kind of a sink operator, hence, flush() is a no op.
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
index 07b5c53..40b719b 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
@@ -60,6 +60,8 @@
private boolean failed;
+ protected boolean flushRequest;
+
public MaterializingPipelinedPartition(IHyracksTaskContext ctx, PartitionManager manager, PartitionId pid,
TaskAttemptId taId, Executor executor) {
this.ctx = ctx;
@@ -93,8 +95,9 @@
MaterializingPipelinedPartition.this.wait();
}
}
- IFileHandle fh = fRef == null ? null : ioManager.open(fRef,
- IIOManager.FileReadWriteMode.READ_ONLY, IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
+ IFileHandle fh = fRef == null ? null
+ : ioManager.open(fRef, IIOManager.FileReadWriteMode.READ_ONLY,
+ IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
try {
writer.open();
try {
@@ -103,9 +106,14 @@
ByteBuffer buffer = ctx.allocateFrame();
boolean fail = false;
boolean done = false;
+ boolean flush = false;
while (!fail && !done) {
synchronized (MaterializingPipelinedPartition.this) {
- while (offset >= size && !eos && !failed) {
+ if (flushRequest) {
+ flushRequest = false;
+ flush = true;
+ }
+ while (offset >= size && !eos && !failed && !flush) {
try {
MaterializingPipelinedPartition.this.wait();
} catch (InterruptedException e) {
@@ -126,6 +134,10 @@
offset += readLen;
buffer.flip();
writer.nextFrame(buffer);
+ if (flush) {
+ writer.flush();
+ flush = false;
+ }
}
}
}
@@ -195,4 +207,10 @@
notifyAll();
}
}
+
+ @Override
+ public synchronized void flush() throws HyracksDataException {
+ flushRequest = true;
+ notifyAll();
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PipelinedPartition.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PipelinedPartition.java
index 2ca1e0f..b83e0f0 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PipelinedPartition.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PipelinedPartition.java
@@ -80,8 +80,8 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- ensureConnected();
if (!failed) {
+ ensureConnected();
delegate.nextFrame(buffer);
}
}
@@ -117,4 +117,12 @@
delegate.close();
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ if (!failed) {
+ ensureConnected();
+ delegate.flush();
+ }
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ConnectorSenderProfilingFrameWriter.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ConnectorSenderProfilingFrameWriter.java
index a46fa7b..9023120 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ConnectorSenderProfilingFrameWriter.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ConnectorSenderProfilingFrameWriter.java
@@ -69,4 +69,9 @@
public void fail() throws HyracksDataException {
writer.fail();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ProfilingPartitionWriterFactory.java b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ProfilingPartitionWriterFactory.java
index 5e543fa..54ef732 100644
--- a/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ProfilingPartitionWriterFactory.java
+++ b/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/profiling/ProfilingPartitionWriterFactory.java
@@ -80,10 +80,16 @@
@Override
public void close() throws HyracksDataException {
closeTime = System.currentTimeMillis();
- ((Task) ctx).setPartitionSendProfile(new PartitionProfile(new PartitionId(ctx.getJobletContext()
- .getJobId(), cd.getConnectorId(), senderIndex, receiverIndex), openTime, closeTime, mrep));
+ ((Task) ctx)
+ .setPartitionSendProfile(new PartitionProfile(new PartitionId(ctx.getJobletContext().getJobId(),
+ cd.getConnectorId(), senderIndex, receiverIndex), openTime, closeTime, mrep));
writer.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
};
}
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
index 7b18c3a..fd71716 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
@@ -47,11 +47,10 @@
protected boolean hasEnoughSpace(int fieldCount, int tupleLength) {
return tupleDataEndOffset + FrameHelper.calcSpaceInFrame(fieldCount, tupleLength)
- + tupleCount * FrameConstants.SIZE_LEN
- <= FrameHelper.getTupleCountOffset(frame.getFrameSize());
+ + tupleCount * FrameConstants.SIZE_LEN <= FrameHelper.getTupleCountOffset(frame.getFrameSize());
}
- private void reset(ByteBuffer buffer, boolean clear) {
+ protected void reset(ByteBuffer buffer, boolean clear) {
array = buffer.array();
if (clear) {
IntSerDeUtils.putInt(array, FrameHelper.getTupleCountOffset(frame.getFrameSize()), 0);
@@ -59,9 +58,8 @@
tupleDataEndOffset = FrameConstants.TUPLE_START_OFFSET;
} else {
tupleCount = IntSerDeUtils.getInt(array, FrameHelper.getTupleCountOffset(frame.getFrameSize()));
- tupleDataEndOffset = tupleCount == 0 ?
- FrameConstants.TUPLE_START_OFFSET :
- IntSerDeUtils.getInt(array, FrameHelper.getTupleCountOffset(frame.getFrameSize())
+ tupleDataEndOffset = tupleCount == 0 ? FrameConstants.TUPLE_START_OFFSET
+ : IntSerDeUtils.getInt(array, FrameHelper.getTupleCountOffset(frame.getFrameSize())
- tupleCount * FrameConstants.SIZE_LEN);
}
}
@@ -77,7 +75,7 @@
}
@Override
- public void flush(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
+ public void write(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
getBuffer().clear();
if (getTupleCount() > 0) {
outWriter.nextFrame(getBuffer());
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameDeserializingDataWriter.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameDeserializingDataWriter.java
index 7d9d643..3657b7e 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameDeserializingDataWriter.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameDeserializingDataWriter.java
@@ -58,4 +58,9 @@
@Override
public void fail() {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldAppender.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldAppender.java
index 996f296..21a7a71 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldAppender.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldAppender.java
@@ -50,8 +50,8 @@
}
@Override
- public void flush(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
- super.flush(outWriter, clearFrame);
+ public void write(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
+ super.write(outWriter, clearFrame);
if (clearFrame) {
if (leftOverSize > 0) {
if (!canHoldNewTuple(0, leftOverSize)) {
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldTupleAppender.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldTupleAppender.java
index 616b8ca..9b9e803 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldTupleAppender.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameFixedFieldTupleAppender.java
@@ -142,7 +142,7 @@
}
@Override
- public void flush(IFrameWriter outWriter, boolean clear) throws HyracksDataException {
- lastAppender.flush(outWriter, clear);
+ public void write(IFrameWriter outWriter, boolean clear) throws HyracksDataException {
+ lastAppender.write(outWriter, clear);
}
}
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameOutputStream.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameOutputStream.java
index 509fe6d..e8f826f 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameOutputStream.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameOutputStream.java
@@ -58,6 +58,6 @@
}
public void flush(IFrameWriter writer) throws HyracksDataException {
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
}
}
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java
index 847deb8..77495dd 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppenderWrapper.java
@@ -46,8 +46,12 @@
outputWriter.open();
}
+ public void write() throws HyracksDataException {
+ frameTupleAppender.write(outputWriter, true);
+ }
+
public void flush() throws HyracksDataException {
- frameTupleAppender.flush(outputWriter, true);
+ frameTupleAppender.flush(outputWriter);
}
public void close() throws HyracksDataException {
@@ -64,8 +68,7 @@
public void appendSkipEmptyField(int[] fieldSlots, byte[] bytes, int offset, int length)
throws HyracksDataException {
- FrameUtils.appendSkipEmptyFieldToWriter(outputWriter, frameTupleAppender,
- fieldSlots, bytes, offset, length);
+ FrameUtils.appendSkipEmptyFieldToWriter(outputWriter, frameTupleAppender, fieldSlots, bytes, offset, length);
}
public void append(byte[] bytes, int offset, int length) throws HyracksDataException {
@@ -88,8 +91,8 @@
public void appendConcat(IFrameTupleAccessor accessor0, int tIndex0, int[] fieldSlots1, byte[] bytes1, int offset1,
int dataLen1) throws HyracksDataException {
- FrameUtils.appendConcatToWriter(outputWriter, frameTupleAppender, accessor0, tIndex0,
- fieldSlots1, bytes1, offset1, dataLen1);
+ FrameUtils.appendConcatToWriter(outputWriter, frameTupleAppender, accessor0, tIndex0, fieldSlots1, bytes1,
+ offset1, dataLen1);
}
public void appendProjection(IFrameTupleAccessor accessor, int tIndex, int[] fields) throws HyracksDataException {
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/SerializingDataWriter.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/SerializingDataWriter.java
index 918ebd9..365b01c 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/SerializingDataWriter.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/SerializingDataWriter.java
@@ -62,11 +62,12 @@
if (!open) {
throw new HyracksDataException("Closing SerializingDataWriter that has not been opened");
}
- tupleAppender.flush(frameWriter, true);
+ tupleAppender.write(frameWriter, true);
frameWriter.close();
open = false;
}
+ @SuppressWarnings("unchecked")
@Override
public void writeData(Object[] data) throws HyracksDataException {
if (!open) {
@@ -88,4 +89,9 @@
public void fail() throws HyracksDataException {
frameWriter.fail();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ tupleAppender.flush(frameWriter);
+ }
}
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/util/FrameUtils.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/util/FrameUtils.java
index 56eef43..5aad232 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/util/FrameUtils.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/util/FrameUtils.java
@@ -65,7 +65,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.appendSkipEmptyField(fieldSlots, bytes, offset, length)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.appendSkipEmptyField(fieldSlots, bytes, offset, length)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -91,7 +91,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.append(bytes, offset, length)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.append(bytes, offset, length)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -114,7 +114,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.append(tupleAccessor, tStartOffset, tEndOffset)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.append(tupleAccessor, tStartOffset, tEndOffset)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -135,7 +135,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.append(tupleAccessor, tIndex)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.append(tupleAccessor, tIndex)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -159,7 +159,7 @@
if (!tupleAppender.append(fieldEndOffsets, byteArray, start, size)) {
flushedBytes = tupleAppender.getBuffer().capacity();
- tupleAppender.flush(writer, true);
+ tupleAppender.write(writer, true);
if (!tupleAppender.append(fieldEndOffsets, byteArray, start, size)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
@@ -184,7 +184,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.appendConcat(accessor0, tIndex0, accessor1, tIndex1)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.appendConcat(accessor0, tIndex0, accessor1, tIndex1)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -210,7 +210,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.appendConcat(accessor0, tIndex0, fieldSlots1, bytes1, offset1, dataLen1)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.appendConcat(accessor0, tIndex0, fieldSlots1, bytes1, offset1, dataLen1)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -232,7 +232,7 @@
int flushedBytes = 0;
if (!frameTupleAppender.appendProjection(accessor, tIndex, fields)) {
flushedBytes = frameTupleAppender.getBuffer().capacity();
- frameTupleAppender.flush(writer, true);
+ frameTupleAppender.write(writer, true);
if (!frameTupleAppender.appendProjection(accessor, tIndex, fields)) {
throw new HyracksDataException("The output cannot be fit into a frame.");
}
@@ -254,7 +254,7 @@
int flushedBytes = 0;
if (!appender.appendField(array, start, length)) {
flushedBytes = appender.getBuffer().capacity();
- appender.flush(writer, true);
+ appender.write(writer, true);
if (!appender.appendField(array, start, length)) {
throw new HyracksDataException("Could not write frame: the size of the tuple is too long");
}
@@ -276,7 +276,7 @@
int flushedBytes = 0;
if (!appender.appendField(accessor, tid, fid)) {
flushedBytes = appender.getBuffer().capacity();
- appender.flush(writer, true);
+ appender.write(writer, true);
if (!appender.appendField(accessor, tid, fid)) {
throw new HyracksDataException("Could not write frame: the size of the tuple is too long");
}
diff --git a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/RunFileWriter.java b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/RunFileWriter.java
index 2af40fd..722e9dc 100644
--- a/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/RunFileWriter.java
+++ b/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/RunFileWriter.java
@@ -86,4 +86,9 @@
}
return new RunFileReader(file, ioManager, size, true);
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // this is a kind of a sink operator and hence, flush() is a no op
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-common/src/test/java/org/apache/hyracks/dataflow/common/comm/io/largeobject/FrameFixedFieldTupleAppenderTest.java b/hyracks/hyracks-dataflow-common/src/test/java/org/apache/hyracks/dataflow/common/comm/io/largeobject/FrameFixedFieldTupleAppenderTest.java
index 05710ad..699674e 100644
--- a/hyracks/hyracks-dataflow-common/src/test/java/org/apache/hyracks/dataflow/common/comm/io/largeobject/FrameFixedFieldTupleAppenderTest.java
+++ b/hyracks/hyracks-dataflow-common/src/test/java/org/apache/hyracks/dataflow/common/comm/io/largeobject/FrameFixedFieldTupleAppenderTest.java
@@ -26,9 +26,6 @@
import java.nio.ByteBuffer;
import java.util.Arrays;
-import org.junit.Before;
-import org.junit.Test;
-
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.IFrameTupleAccessor;
import org.apache.hyracks.api.comm.IFrameTupleAppender;
@@ -44,6 +41,8 @@
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
+import org.junit.Before;
+import org.junit.Test;
public class FrameFixedFieldTupleAppenderTest {
@@ -51,22 +50,19 @@
static final int TEST_FRAME_SIZE = 256;
FrameFixedFieldAppender appender;
- static ISerializerDeserializer[] fields = new ISerializerDeserializer[] {
- IntegerSerializerDeserializer.INSTANCE,
- new UTF8StringSerializerDeserializer(),
- IntegerSerializerDeserializer.INSTANCE,
- new UTF8StringSerializerDeserializer(),
- };
+ static ISerializerDeserializer[] fields = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE,
+ new UTF8StringSerializerDeserializer(), IntegerSerializerDeserializer.INSTANCE,
+ new UTF8StringSerializerDeserializer(), };
static RecordDescriptor recordDescriptor = new RecordDescriptor(fields);
static ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(recordDescriptor.getFieldCount());
- class SequetialDataVerifier implements IFrameWriter {
+ class SequentialDataVerifier implements IFrameWriter {
private final IFrameTupleAccessor accessor;
private IFrameTupleAccessor innerAccessor;
private int tid;
- public SequetialDataVerifier(IFrameTupleAccessor accessor) {
+ public SequentialDataVerifier(IFrameTupleAccessor accessor) {
this.accessor = accessor;
this.innerAccessor = new FrameTupleAccessor(recordDescriptor);
}
@@ -87,7 +83,8 @@
private void validate(IFrameTupleAccessor innerAccessor, int i) {
assertTrue(tid < accessor.getTupleCount());
assertEquals(accessor.getTupleLength(tid), innerAccessor.getTupleLength(i));
- assertArrayEquals(Arrays.copyOfRange(accessor.getBuffer().array(), accessor.getTupleStartOffset(tid),
+ assertArrayEquals(
+ Arrays.copyOfRange(accessor.getBuffer().array(), accessor.getTupleStartOffset(tid),
accessor.getTupleEndOffset(tid)),
Arrays.copyOfRange(innerAccessor.getBuffer().array(), innerAccessor.getTupleStartOffset(i),
innerAccessor.getTupleEndOffset(i)));
@@ -103,6 +100,10 @@
public void close() throws HyracksDataException {
assertEquals(accessor.getTupleCount(), tid);
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
}
@Before
@@ -119,13 +120,13 @@
for (int tid = 0; tid < accessor.getTupleCount(); tid++) {
for (int fid = 0; fid < fields.length; fid++) {
if (!appender.appendField(accessor, tid, fid)) {
- appender.flush(writer, true);
+ appender.write(writer, true);
if (!appender.appendField(accessor, tid, fid)) {
}
}
}
}
- appender.flush(writer, true);
+ appender.write(writer, true);
writer.close();
}
@@ -143,7 +144,7 @@
}
private IFrameWriter prepareValidator(IFrameTupleAccessor accessor) throws HyracksDataException {
- return new SequetialDataVerifier(accessor);
+ return new SequentialDataVerifier(accessor);
}
enum DATA_TYPE {
@@ -154,8 +155,8 @@
private IFrameTupleAccessor prepareData(DATA_TYPE type) throws HyracksDataException {
IFrameTupleAccessor accessor = new FrameTupleAccessor(recordDescriptor);
- IFrameTupleAppender appender = new FrameTupleAppender(
- new VSizeFrame(new FrameManager(INPUT_BUFFER_SIZE)), true);
+ IFrameTupleAppender appender = new FrameTupleAppender(new VSizeFrame(new FrameManager(INPUT_BUFFER_SIZE)),
+ true);
int i = 0;
do {
switch (type) {
@@ -169,8 +170,8 @@
makeABigObjectTuple(tupleBuilder, i++);
break;
}
- } while (appender
- .append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0, tupleBuilder.getSize()));
+ } while (appender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
+ tupleBuilder.getSize()));
accessor.reset(appender.getBuffer());
return accessor;
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/base/AbstractUnaryInputSinkOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/base/AbstractUnaryInputSinkOperatorNodePushable.java
index 52b4abe..1aa48e5 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/base/AbstractUnaryInputSinkOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/base/AbstractUnaryInputSinkOperatorNodePushable.java
@@ -20,10 +20,16 @@
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
public abstract class AbstractUnaryInputSinkOperatorNodePushable extends AbstractUnaryInputOperatorNodePushable {
@Override
public final void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
throw new IllegalStateException();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // Sink operator, nothing to flush
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/LocalityAwarePartitionDataWriter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/LocalityAwarePartitionDataWriter.java
index ee8c656..830075c 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/LocalityAwarePartitionDataWriter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/LocalityAwarePartitionDataWriter.java
@@ -127,7 +127,7 @@
for (int i = 0; i < pWriters.length; ++i) {
if (isWriterOpen[i]) {
try {
- appenders[i].flush(pWriters[i], true);
+ appenders[i].write(pWriters[i], true);
} catch (Throwable th) {
if (closeException == null) {
closeException = new HyracksDataException(th);
@@ -151,4 +151,11 @@
throw closeException;
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ for (int i = 0; i < pWriters.length; ++i) {
+ appenders[i].flush(pWriters[i]);
+ }
+ }
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/MToNReplicatingConnectorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/MToNReplicatingConnectorDescriptor.java
index 4d73fa5..7d60ce5 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/MToNReplicatingConnectorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/MToNReplicatingConnectorDescriptor.java
@@ -113,6 +113,13 @@
epWriters[i].open();
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ for (IFrameWriter writer : epWriters) {
+ writer.flush();
+ }
+ }
};
}
@@ -126,4 +133,4 @@
NonDeterministicFrameReader frameReader = new NonDeterministicFrameReader(channelReader);
return new PartitionCollector(ctx, getConnectorId(), index, expectedPartitions, frameReader, channelReader);
}
-}
\ No newline at end of file
+}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/PartitionDataWriter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/PartitionDataWriter.java
index 336272c..646883f 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/PartitionDataWriter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/connectors/PartitionDataWriter.java
@@ -68,7 +68,7 @@
if (isOpen[i]) {
if (allocatedFrame) {
try {
- appenders[i].flush(pWriters[i], true);
+ appenders[i].write(pWriters[i], true);
} catch (Throwable th) {
if (closeException == null) {
closeException = new HyracksDataException(th);
@@ -112,7 +112,6 @@
for (int i = 0; i < tupleCount; ++i) {
int h = tpc.partition(tupleAccessor, i, consumerPartitionCount);
FrameUtils.appendToWriter(pWriters[h], appenders[h], tupleAccessor, i);
-
}
}
@@ -142,4 +141,11 @@
throw failException;
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ for (int i = 0; i < consumerPartitionCount; i++) {
+ appenders[i].flush(pWriters[i]);
+ }
+ }
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractDeserializedFileScanOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractDeserializedFileScanOperatorDescriptor.java
index 24b3bf9..35ae0da 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractDeserializedFileScanOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractDeserializedFileScanOperatorDescriptor.java
@@ -108,6 +108,11 @@
public void fail() throws HyracksDataException {
// do nothing
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // do nothing
+ }
}
@Override
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractFileWriteOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractFileWriteOperatorDescriptor.java
index 8228367..6e08238 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractFileWriteOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/AbstractFileWriteOperatorDescriptor.java
@@ -70,6 +70,11 @@
throw new HyracksDataException(e);
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // This is a kind of a sink operator and hence, flush() is a no op
+ }
}
private static final long serialVersionUID = 1L;
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/DelimitedDataTupleParserFactory.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/DelimitedDataTupleParserFactory.java
index 81b3c4e..0960927 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/DelimitedDataTupleParserFactory.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/file/DelimitedDataTupleParserFactory.java
@@ -87,7 +87,7 @@
FrameUtils.appendToWriter(writer, appender, tb.getFieldEndOffsets(), tb.getByteArray(), 0,
tb.getSize());
}
- appender.flush(writer, true);
+ appender.write(writer, true);
} catch (IOException e) {
throw new HyracksDataException(e);
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/DeserializedPreclusteredGroupOperator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/DeserializedPreclusteredGroupOperator.java
index d041d5e..e67752b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/DeserializedPreclusteredGroupOperator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/DeserializedPreclusteredGroupOperator.java
@@ -30,6 +30,7 @@
public class DeserializedPreclusteredGroupOperator implements IOpenableDataWriterOperator {
private final int[] groupFields;
+ @SuppressWarnings("rawtypes")
private final IComparator[] comparators;
private final IGroupAggregator aggregator;
@@ -42,6 +43,7 @@
private IOpenableDataReader<Object[]> reader;
+ @SuppressWarnings("rawtypes")
public DeserializedPreclusteredGroupOperator(int[] groupFields, IComparator[] comparators,
IGroupAggregator aggregator) {
this.groupFields = groupFields;
@@ -110,6 +112,7 @@
buffer.add(data);
}
+ @SuppressWarnings("unchecked")
private int compare(Object[] d1, Object[] d2) {
for (int i = 0; i < groupFields.length; ++i) {
int fIdx = groupFields[i];
@@ -126,4 +129,9 @@
// TODO Auto-generated method stub
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ throw new HyracksDataException("unsupported operation");
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/HashSpillableTableFactory.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/HashSpillableTableFactory.java
index 052fe8c..7fbde54 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/HashSpillableTableFactory.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/HashSpillableTableFactory.java
@@ -287,7 +287,7 @@
if (!outputAppender.appendSkipEmptyField(outputTupleBuilder.getFieldEndOffsets(),
outputTupleBuilder.getByteArray(), 0, outputTupleBuilder.getSize())) {
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
if (!outputAppender.appendSkipEmptyField(outputTupleBuilder.getFieldEndOffsets(),
outputTupleBuilder.getByteArray(), 0, outputTupleBuilder.getSize())) {
throw new HyracksDataException(
@@ -297,7 +297,7 @@
} while (true);
}
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
aggregator.close();
return;
}
@@ -330,14 +330,14 @@
if (!outputAppender.appendSkipEmptyField(outputTupleBuilder.getFieldEndOffsets(),
outputTupleBuilder.getByteArray(), 0, outputTupleBuilder.getSize())) {
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
if (!outputAppender.appendSkipEmptyField(outputTupleBuilder.getFieldEndOffsets(),
outputTupleBuilder.getByteArray(), 0, outputTupleBuilder.getSize())) {
throw new HyracksDataException("The output item is too large to be fit into a frame.");
}
}
}
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
aggregator.close();
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupMergeOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupMergeOperatorNodePushable.java
index 779c631..aca294a 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupMergeOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupMergeOperatorNodePushable.java
@@ -325,14 +325,14 @@
if (!writerAppender.appendSkipEmptyField(finalTupleBuilder.getFieldEndOffsets(),
finalTupleBuilder.getByteArray(), 0, finalTupleBuilder.getSize())) {
- writerAppender.flush(writer, true);
+ writerAppender.write(writer, true);
if (!writerAppender.appendSkipEmptyField(finalTupleBuilder.getFieldEndOffsets(),
finalTupleBuilder.getByteArray(), 0, finalTupleBuilder.getSize())) {
throw new HyracksDataException("Aggregation output is too large to be fit into a frame.");
}
}
}
- writerAppender.flush(writer, true);
+ writerAppender.write(writer, true);
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/hash/GroupingHashTable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/hash/GroupingHashTable.java
index 97cea25..d2342ea 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/hash/GroupingHashTable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/hash/GroupingHashTable.java
@@ -240,7 +240,7 @@
}
}
}
- appender.flush(writer, true);
+ appender.write(writer, true);
}
void close() throws HyracksDataException {
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
index 7539abf..515987a 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupOperatorNodePushable.java
@@ -73,4 +73,9 @@
public void close() throws HyracksDataException {
pgw.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ pgw.flush();
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
index 1c08b53..31f52ee 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
@@ -176,7 +176,7 @@
if (!isFailed && !first) {
assert (copyFrameAccessor.getTupleCount() > 0);
writeOutput(copyFrameAccessor, copyFrameAccessor.getTupleCount() - 1);
- appenderWrapper.flush();
+ appenderWrapper.write();
}
aggregator.close();
aggregateState.close();
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/InMemoryHashJoin.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/InMemoryHashJoin.java
index 2a66fb7..a139341 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/InMemoryHashJoin.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/InMemoryHashJoin.java
@@ -147,7 +147,7 @@
}
public void closeJoin(IFrameWriter writer) throws HyracksDataException {
- appender.flush(writer, true);
+ appender.write(writer, true);
int nFrames = buffers.size();
buffers.clear();
ctx.deallocateFrames(nFrames);
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/NestedLoopJoin.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/NestedLoopJoin.java
index 5ca7700..6746b50 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/NestedLoopJoin.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/NestedLoopJoin.java
@@ -204,7 +204,7 @@
outBuffers.clear();
currentMemSize = 0;
- appender.flush(writer, true);
+ appender.write(writer, true);
}
private int compare(FrameTupleAccessor accessor0, int tIndex0, FrameTupleAccessor accessor1, int tIndex1)
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
index c0c467a..a692674 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
@@ -138,8 +138,8 @@
boolean isLeftOuter, INullWriterFactory[] nullWriterFactories1) throws HyracksDataException {
super(spec, 2, 1);
- this.memsize = memsize;
this.inputsize0 = inputsize0;
+ this.memsize = memsize;
this.fudgeFactor = factor;
this.probeKeys = keys0;
this.buildKeys = keys1;
@@ -450,7 +450,7 @@
}
//Build Side is smaller
applyInMemHashJoin(buildKeys, probeKeys, tabSize, probeRd, buildRd, probeHpc, buildHpc,
- buildSideReader, probeSideReader, false, pid); //checked-confirmed
+ buildSideReader, probeSideReader, false, pid); // checked-confirmed
} else { //Case 1.2 - InMemHJ with Role Reversal
LOGGER.fine(
"\t>>>Case 1.2. (NoIsLeftOuter || probe<build) AND ApplyInMemHJ WITH RoleReversal - [Level "
@@ -463,7 +463,7 @@
}
//Probe Side is smaller
applyInMemHashJoin(probeKeys, buildKeys, tabSize, buildRd, probeRd, buildHpc, probeHpc,
- probeSideReader, buildSideReader, true, pid); //checked-confirmed
+ probeSideReader, buildSideReader, true, pid); // checked-confirmed
}
}
//Apply (Recursive) HHJ
@@ -515,7 +515,7 @@
continue;
}
- joinPartitionPair(rHHj, rbrfw, rprfw, rPid, afterMax, (level + 1), false); //checked-confirmed
+ joinPartitionPair(rHHj, rbrfw, rprfw, rPid, afterMax, (level + 1), false); // checked-confirmed
}
} else { //Case 2.1.2 - Switch to NLJ
@@ -636,7 +636,7 @@
}
bReader.close();
rPartbuff.reset();
- // probe
+ //probe
pReader.open();
while (pReader.nextFrame(rPartbuff)) {
joiner.join(rPartbuff.getBuffer(), writer);
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/map/DeserializedMapperOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/map/DeserializedMapperOperatorDescriptor.java
index 048cea2..e4b5bb3 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/map/DeserializedMapperOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/map/DeserializedMapperOperatorDescriptor.java
@@ -62,6 +62,10 @@
public void writeData(Object[] data) throws HyracksDataException {
mapper.map(data, writer);
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
}
private static final long serialVersionUID = 1L;
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/ConstantTupleSourceOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/ConstantTupleSourceOperatorNodePushable.java
index 0c647d7..af42168 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/ConstantTupleSourceOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/ConstantTupleSourceOperatorNodePushable.java
@@ -48,7 +48,7 @@
appender.append(fieldSlots, tupleData, 0, tupleSize);
writer.open();
try {
- appender.flush(writer, false);
+ appender.write(writer, false);
} catch (Throwable th) {
writer.fail();
throw new HyracksDataException(th);
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/IdentityOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/IdentityOperatorDescriptor.java
index 971f61d..817c6d2 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/IdentityOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/IdentityOperatorDescriptor.java
@@ -40,7 +40,7 @@
@Override
public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions)
- throws HyracksDataException {
+ throws HyracksDataException {
return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {
@Override
public void open() throws HyracksDataException {
@@ -61,6 +61,11 @@
public void close() throws HyracksDataException {
writer.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
};
}
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/LimitOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/LimitOperatorDescriptor.java
index 5c09f42..084c9ab 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/LimitOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/LimitOperatorDescriptor.java
@@ -46,7 +46,7 @@
@Override
public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions)
- throws HyracksDataException {
+ throws HyracksDataException {
return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {
private FrameTupleAccessor fta;
@@ -73,7 +73,7 @@
FrameUtils.appendToWriter(writer, partialAppender, fta, i);
currentSize++;
}
- partialAppender.flush(writer,false);
+ partialAppender.write(writer, false);
finished = true;
} else {
FrameUtils.flushFrame(buffer, writer);
@@ -92,6 +92,11 @@
public void close() throws HyracksDataException {
writer.close();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
};
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/MaterializingOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/MaterializingOperatorDescriptor.java
index 36bf919..db86980 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/MaterializingOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/MaterializingOperatorDescriptor.java
@@ -30,8 +30,6 @@
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
-import org.apache.hyracks.dataflow.common.util.IntSerDeUtils;
import org.apache.hyracks.dataflow.std.base.AbstractActivityNode;
import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
@@ -61,8 +59,8 @@
@Override
public void contributeActivities(IActivityGraphBuilder builder) {
if (isSingleActivity) {
- MaterializerReaderActivityNode mra = new MaterializerReaderActivityNode(new ActivityId(odId,
- MATERIALIZER_READER_ACTIVITY_ID));
+ MaterializerReaderActivityNode mra = new MaterializerReaderActivityNode(
+ new ActivityId(odId, MATERIALIZER_READER_ACTIVITY_ID));
builder.addActivity(this, mra);
builder.addSourceEdge(0, mra, 0);
@@ -97,8 +95,8 @@
@Override
public void open() throws HyracksDataException {
- state = new MaterializerTaskState(ctx.getJobletContext().getJobId(), new TaskId(getActivityId(),
- partition));
+ state = new MaterializerTaskState(ctx.getJobletContext().getJobId(),
+ new TaskId(getActivityId(), partition));
state.open(ctx);
}
@@ -116,7 +114,6 @@
state.close();
state.writeOut(writer, new VSizeFrame(ctx));
}
-
};
}
}
@@ -136,8 +133,8 @@
@Override
public void open() throws HyracksDataException {
- state = new MaterializerTaskState(ctx.getJobletContext().getJobId(), new TaskId(getActivityId(),
- partition));
+ state = new MaterializerTaskState(ctx.getJobletContext().getJobId(),
+ new TaskId(getActivityId(), partition));
state.open(ctx);
}
@@ -172,8 +169,8 @@
return new AbstractUnaryOutputSourceOperatorNodePushable() {
@Override
public void initialize() throws HyracksDataException {
- MaterializerTaskState state = (MaterializerTaskState) ctx.getStateObject(new TaskId(new ActivityId(
- getOperatorId(), MATERIALIZER_ACTIVITY_ID), partition));
+ MaterializerTaskState state = (MaterializerTaskState) ctx.getStateObject(
+ new TaskId(new ActivityId(getOperatorId(), MATERIALIZER_ACTIVITY_ID), partition));
state.writeOut(writer, new VSizeFrame(ctx));
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/PrinterOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/PrinterOperatorDescriptor.java
index 792a041..9ff5d9b 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/PrinterOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/PrinterOperatorDescriptor.java
@@ -62,6 +62,11 @@
public void setDataWriter(int index, IOpenableDataWriter<Object[]> writer) {
throw new IllegalArgumentException();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ System.err.flush();
+ }
}
@Override
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitOperatorDescriptor.java
index 82c62a5..dc368eb 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitOperatorDescriptor.java
@@ -138,6 +138,13 @@
}
@Override
+ public void flush() throws HyracksDataException {
+ for (int i = 0; i < numberOfNonMaterializedOutputs; i++) {
+ writers[i].flush();
+ }
+ }
+
+ @Override
public void close() throws HyracksDataException {
HyracksDataException hde = null;
try {
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitVectorOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitVectorOperatorDescriptor.java
index 4f12db1..41c9569 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitVectorOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/misc/SplitVectorOperatorDescriptor.java
@@ -88,8 +88,8 @@
@Override
public void open() throws HyracksDataException {
- state = new CollectTaskState(ctx.getJobletContext().getJobId(), new TaskId(getActivityId(),
- partition));
+ state = new CollectTaskState(ctx.getJobletContext().getJobId(),
+ new TaskId(getActivityId(), partition));
state.buffer = new ArrayList<Object[]>();
}
@@ -107,9 +107,14 @@
public void fail() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // flush() is a no op since the frame writer's whole job is to write state data to a buffer
+ }
};
- return new DeserializedOperatorNodePushable(ctx, op, recordDescProvider.getInputRecordDescriptor(
- getActivityId(), 0));
+ return new DeserializedOperatorNodePushable(ctx, op,
+ recordDescProvider.getInputRecordDescriptor(getActivityId(), 0));
}
}
@@ -138,8 +143,8 @@
@Override
public void open() throws HyracksDataException {
- state = (CollectTaskState) ctx.getStateObject(new TaskId(new ActivityId(getOperatorId(),
- COLLECT_ACTIVITY_ID), partition));
+ state = (CollectTaskState) ctx.getStateObject(
+ new TaskId(new ActivityId(getOperatorId(), COLLECT_ACTIVITY_ID), partition));
}
@Override
@@ -161,9 +166,14 @@
public void fail() throws HyracksDataException {
writer.fail();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
};
- return new DeserializedOperatorNodePushable(ctx, op, recordDescProvider.getOutputRecordDescriptor(
- getActivityId(), 0));
+ return new DeserializedOperatorNodePushable(ctx, op,
+ recordDescProvider.getOutputRecordDescriptor(getActivityId(), 0));
}
}
@@ -171,7 +181,8 @@
private final int splits;
- public SplitVectorOperatorDescriptor(IOperatorDescriptorRegistry spec, int splits, RecordDescriptor recordDescriptor) {
+ public SplitVectorOperatorDescriptor(IOperatorDescriptorRegistry spec, int splits,
+ RecordDescriptor recordDescriptor) {
super(spec, 1, 1);
this.splits = splits;
recordDescriptors[0] = recordDescriptor;
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
index f50d14c..e56d51d 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractFrameSorter.java
@@ -173,7 +173,7 @@
}
}
maxFrameSize = Math.max(maxFrameSize, outputFrame.getFrameSize());
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
if (LOGGER.isLoggable(Level.FINE)) {
LOGGER.fine(
"Flushed records:" + limit + " out of " + tupleCount + "; Flushed through " + (io + 1) + " frames");
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
index ea46682..c1e81ff 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/ExternalSortRunGenerator.java
@@ -29,9 +29,9 @@
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
import org.apache.hyracks.dataflow.std.sort.buffermanager.EnumFreeSlotPolicy;
-import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotSmallestFit;
import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotBiggestFirst;
import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotLastFit;
+import org.apache.hyracks.dataflow.std.sort.buffermanager.FrameFreeSlotSmallestFit;
import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameBufferManager;
import org.apache.hyracks.dataflow.std.sort.buffermanager.IFrameFreeSlotPolicy;
import org.apache.hyracks.dataflow.std.sort.buffermanager.VariableFrameMemoryManager;
@@ -53,7 +53,7 @@
public ExternalSortRunGenerator(IHyracksTaskContext ctx, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
RecordDescriptor recordDesc, Algorithm alg, EnumFreeSlotPolicy policy, int framesLimit)
- throws HyracksDataException {
+ throws HyracksDataException {
this(ctx, sortFields, firstKeyNormalizerFactory, comparatorFactories, recordDesc, alg, policy, framesLimit,
Integer.MAX_VALUE);
}
@@ -61,7 +61,7 @@
public ExternalSortRunGenerator(IHyracksTaskContext ctx, int[] sortFields,
INormalizedKeyComputerFactory firstKeyNormalizerFactory, IBinaryComparatorFactory[] comparatorFactories,
RecordDescriptor recordDesc, Algorithm alg, EnumFreeSlotPolicy policy, int framesLimit, int outputLimit)
- throws HyracksDataException {
+ throws HyracksDataException {
this.ctx = ctx;
maxSortFrames = framesLimit - 1;
@@ -98,12 +98,14 @@
}
}
+ @Override
protected RunFileWriter getRunFileWriter() throws HyracksDataException {
- FileReference file = ctx.getJobletContext().createManagedWorkspaceFile(
- ExternalSortRunGenerator.class.getSimpleName());
+ FileReference file = ctx.getJobletContext()
+ .createManagedWorkspaceFile(ExternalSortRunGenerator.class.getSimpleName());
return new RunFileWriter(file, ctx.getIOManager());
}
+ @Override
protected IFrameWriter getFlushableFrameWriter(RunFileWriter writer) throws HyracksDataException {
return writer;
}
@@ -112,5 +114,4 @@
public ISorter getSorter() {
return frameSorter;
}
-
}
\ No newline at end of file
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
index bf4f621..7f2847c 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/HeapSortRunGenerator.java
@@ -65,8 +65,7 @@
public void open() throws HyracksDataException {
IFramePool framePool = new VariableFramePool(ctx, (frameLimit - 1) * ctx.getInitialFrameSize());
ITupleBufferManager bufferManager = new VariableTupleMemoryManager(framePool, recordDescriptor);
- tupleSorter = new TupleSorterHeapSort(ctx, bufferManager, topK, sortFields, nmkFactory,
- comparatorFactories);
+ tupleSorter = new TupleSorterHeapSort(ctx, bufferManager, topK, sortFields, nmkFactory, comparatorFactories);
super.open();
}
@@ -77,8 +76,8 @@
@Override
protected RunFileWriter getRunFileWriter() throws HyracksDataException {
- FileReference file = ctx.getJobletContext().createManagedWorkspaceFile(
- HeapSortRunGenerator.class.getSimpleName());
+ FileReference file = ctx.getJobletContext()
+ .createManagedWorkspaceFile(HeapSortRunGenerator.class.getSimpleName());
return new RunFileWriter(file, ctx.getIOManager());
}
@@ -99,5 +98,4 @@
}
}
}
-
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
index 5c31f89..fc15bbc 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/TupleSorterHeapSort.java
@@ -262,7 +262,7 @@
}
}
maxFrameSize = Math.max(maxFrameSize, outputFrame.getFrameSize());
- outputAppender.flush(writer, true);
+ outputAppender.write(writer, true);
if (LOGGER.isLoggable(Level.INFO)) {
LOGGER.info(
"Flushed records:" + numEntries + "; Flushed through " + (io + 1) + " frames");
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/union/UnionAllOperatorDescriptor.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/union/UnionAllOperatorDescriptor.java
index 289b879..e76d2d3 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/union/UnionAllOperatorDescriptor.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/union/UnionAllOperatorDescriptor.java
@@ -121,6 +121,13 @@
}
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ synchronized (UnionOperator.this) {
+ writer.flush();
+ }
+ }
};
}
}
diff --git a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/util/DeserializedOperatorNodePushable.java b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/util/DeserializedOperatorNodePushable.java
index b656b46..92d18f0 100644
--- a/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/util/DeserializedOperatorNodePushable.java
+++ b/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/util/DeserializedOperatorNodePushable.java
@@ -76,4 +76,9 @@
public String getDisplayName() {
return "Deserialized(" + delegate + ")";
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ delegate.flush();
+ }
}
\ No newline at end of file
diff --git a/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/DataGenOperatorDescriptor.java b/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/DataGenOperatorDescriptor.java
index 431a4f4..5ef6417 100644
--- a/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/DataGenOperatorDescriptor.java
+++ b/hyracks/hyracks-examples/btree-example/btreehelper/src/main/java/org/apache/hyracks/examples/btree/helper/DataGenOperatorDescriptor.java
@@ -88,14 +88,14 @@
}
if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- appender.flush(writer, true);
+ appender.write(writer, true);
if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
throw new HyracksDataException("Record size (" + tb.getSize()
+ ") larger than frame size (" + appender.getBuffer().capacity() + ")");
}
}
}
- appender.flush(writer, true);
+ appender.write(writer, true);
} catch (Throwable th) {
writer.fail();
throw new HyracksDataException(th);
diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/comm/SerializationDeserializationTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/comm/SerializationDeserializationTest.java
index 41e63a0..995ae30 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/comm/SerializationDeserializationTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/comm/SerializationDeserializationTest.java
@@ -26,8 +26,6 @@
import java.util.logging.Level;
import java.util.logging.Logger;
-import org.junit.Test;
-
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.IFrameReader;
import org.apache.hyracks.api.comm.IFrameWriter;
@@ -45,6 +43,7 @@
import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import org.apache.hyracks.test.support.TestUtils;
+import org.junit.Test;
public class SerializationDeserializationTest {
private static final Logger LOGGER = Logger.getLogger(SerializationDeserializationTest.class.getName());
@@ -82,6 +81,10 @@
@Override
public void fail() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
});
}
diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/ErrorReportingTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/ErrorReportingTest.java
index fb816da..6b08c3e 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/ErrorReportingTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/ErrorReportingTest.java
@@ -187,6 +187,10 @@
public void close() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
};
}
diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/SuperActivityRewritingTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/SuperActivityRewritingTest.java
index 5d300f4..bba18b3 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/SuperActivityRewritingTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/rewriting/SuperActivityRewritingTest.java
@@ -173,6 +173,10 @@
public void close() throws HyracksDataException {
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
};
}
diff --git a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/TopKRunGeneratorTest.java b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/TopKRunGeneratorTest.java
index 43e7649..f621bf9 100644
--- a/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/TopKRunGeneratorTest.java
+++ b/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/unit/TopKRunGeneratorTest.java
@@ -34,8 +34,6 @@
import java.util.Map;
import java.util.TreeMap;
-import org.junit.Test;
-
import org.apache.hyracks.api.comm.FixedSizeFrame;
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.IFrameWriter;
@@ -45,8 +43,9 @@
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.std.sort.AbstractSortRunGenerator;
-import org.apache.hyracks.dataflow.std.sort.HybridTopKSortRunGenerator;
import org.apache.hyracks.dataflow.std.sort.HeapSortRunGenerator;
+import org.apache.hyracks.dataflow.std.sort.HybridTopKSortRunGenerator;
+import org.junit.Test;
public class TopKRunGeneratorTest {
@@ -90,14 +89,18 @@
public void close() throws HyracksDataException {
assertTrue(answer.isEmpty());
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
}
@Test
public void testReverseOrderedDataShouldNotGenerateAnyRuns() throws HyracksDataException {
int topK = 1;
IHyracksTaskContext ctx = AbstractRunGeneratorTest.testUtils.create(PAGE_SIZE);
- HeapSortRunGenerator sorter = new HeapSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK,
- SortFields, null, ComparatorFactories, RecordDesc);
+ HeapSortRunGenerator sorter = new HeapSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK, SortFields, null,
+ ComparatorFactories, RecordDesc);
testInMemoryOnly(ctx, topK, ORDER.REVERSE, sorter);
}
@@ -106,8 +109,8 @@
public void testAlreadySortedDataShouldNotGenerateAnyRuns() throws HyracksDataException {
int topK = SORT_FRAME_LIMIT;
IHyracksTaskContext ctx = AbstractRunGeneratorTest.testUtils.create(PAGE_SIZE);
- HeapSortRunGenerator sorter = new HeapSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK,
- SortFields, null, ComparatorFactories, RecordDesc);
+ HeapSortRunGenerator sorter = new HeapSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK, SortFields, null,
+ ComparatorFactories, RecordDesc);
testInMemoryOnly(ctx, topK, ORDER.INORDER, sorter);
}
@@ -116,8 +119,8 @@
public void testHybridTopKShouldNotGenerateAnyRuns() throws HyracksDataException {
int topK = 1;
IHyracksTaskContext ctx = AbstractRunGeneratorTest.testUtils.create(PAGE_SIZE);
- AbstractSortRunGenerator sorter = new HybridTopKSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK,
- SortFields, null, ComparatorFactories, RecordDesc);
+ AbstractSortRunGenerator sorter = new HybridTopKSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK, SortFields, null,
+ ComparatorFactories, RecordDesc);
testInMemoryOnly(ctx, topK, ORDER.REVERSE, sorter);
}
@@ -126,8 +129,8 @@
public void testHybridTopKShouldSwitchToFrameSorterWhenFlushed() {
int topK = 1;
IHyracksTaskContext ctx = AbstractRunGeneratorTest.testUtils.create(PAGE_SIZE);
- AbstractSortRunGenerator sorter = new HybridTopKSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK,
- SortFields, null, ComparatorFactories, RecordDesc);
+ AbstractSortRunGenerator sorter = new HybridTopKSortRunGenerator(ctx, SORT_FRAME_LIMIT, topK, SortFields, null,
+ ComparatorFactories, RecordDesc);
}
@@ -148,8 +151,8 @@
int minRecordSize = 16;
int maxRecordSize = 64;
- AbstractRunGeneratorTest
- .prepareData(ctx, frameList, minDataSize, minRecordSize, maxRecordSize, null, keyValuePair);
+ AbstractRunGeneratorTest.prepareData(ctx, frameList, minDataSize, minRecordSize, maxRecordSize, null,
+ keyValuePair);
assert topK > 0;
diff --git a/hyracks/hyracks-examples/text-example/texthelper/src/main/java/org/apache/hyracks/examples/text/WordTupleParserFactory.java b/hyracks/hyracks-examples/text-example/texthelper/src/main/java/org/apache/hyracks/examples/text/WordTupleParserFactory.java
index 964280b..eb2714f 100644
--- a/hyracks/hyracks-examples/text-example/texthelper/src/main/java/org/apache/hyracks/examples/text/WordTupleParserFactory.java
+++ b/hyracks/hyracks-examples/text-example/texthelper/src/main/java/org/apache/hyracks/examples/text/WordTupleParserFactory.java
@@ -59,7 +59,7 @@
FrameUtils.appendToWriter(writer, appender, tb.getFieldEndOffsets(), tb.getByteArray(), 0,
tb.getSize());
}
- appender.flush(writer, true);
+ appender.write(writer, true);
} catch (IOException e) {
throw new HyracksDataException(e);
}
diff --git a/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/TextKeyValueParserFactory.java b/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/TextKeyValueParserFactory.java
index 3e0bb14..f6bf66c 100644
--- a/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/TextKeyValueParserFactory.java
+++ b/hyracks/hyracks-hdfs/hyracks-hdfs-core/src/main/java/org/apache/hyracks/hdfs/lib/TextKeyValueParserFactory.java
@@ -60,7 +60,7 @@
@Override
public void close(IFrameWriter writer) throws HyracksDataException {
- appender.flush(writer, false);
+ appender.write(writer, false);
}
};
diff --git a/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java b/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
index a97441c..2226b03 100644
--- a/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/dataflow/BTreeSearchOperatorNodePushable.java
@@ -40,7 +40,8 @@
public BTreeSearchOperatorNodePushable(AbstractTreeIndexOperatorDescriptor opDesc, IHyracksTaskContext ctx,
int partition, IRecordDescriptorProvider recordDescProvider, int[] lowKeyFields, int[] highKeyFields,
- boolean lowKeyInclusive, boolean highKeyInclusive, int[] minFilterFieldIndexes, int[] maxFilterFieldIndexes) {
+ boolean lowKeyInclusive, boolean highKeyInclusive, int[] minFilterFieldIndexes,
+ int[] maxFilterFieldIndexes) {
super(opDesc, ctx, partition, recordDescProvider, minFilterFieldIndexes, maxFilterFieldIndexes);
this.lowKeyInclusive = lowKeyInclusive;
this.highKeyInclusive = highKeyInclusive;
@@ -75,8 +76,8 @@
ITreeIndex treeIndex = (ITreeIndex) index;
lowKeySearchCmp = BTreeUtils.getSearchMultiComparator(treeIndex.getComparatorFactories(), lowKey);
highKeySearchCmp = BTreeUtils.getSearchMultiComparator(treeIndex.getComparatorFactories(), highKey);
- return new RangePredicate(lowKey, highKey, lowKeyInclusive, highKeyInclusive, lowKeySearchCmp,
- highKeySearchCmp, minFilterKey, maxFilterKey);
+ return new RangePredicate(lowKey, highKey, lowKeyInclusive, highKeyInclusive, lowKeySearchCmp, highKeySearchCmp,
+ minFilterKey, maxFilterKey);
}
@Override
diff --git a/hyracks/hyracks-storage-am-btree/src/test/java/org/apache/hyracks/storage/am/btree/test/FramewriterTest.java b/hyracks/hyracks-storage-am-btree/src/test/java/org/apache/hyracks/storage/am/btree/test/FramewriterTest.java
index b5c14c1..6fc26a7 100644
--- a/hyracks/hyracks-storage-am-btree/src/test/java/org/apache/hyracks/storage/am/btree/test/FramewriterTest.java
+++ b/hyracks/hyracks-storage-am-btree/src/test/java/org/apache/hyracks/storage/am/btree/test/FramewriterTest.java
@@ -178,7 +178,7 @@
writer.nextFrame(EMPTY_BUFFER);
return null;
}
- }).when(appenders[0]).flush(Matchers.any(IFrameWriter.class), Matchers.anyBoolean());
+ }).when(appenders[0]).write(Matchers.any(IFrameWriter.class), Matchers.anyBoolean());
appenders[1] = Mockito.mock(FrameTupleAppender.class);
Mockito.doAnswer(new Answer<Object>() {
@@ -186,7 +186,7 @@
public Object answer(InvocationOnMock invocation) throws Throwable {
throw new HyracksDataException("couldn't flush frame");
}
- }).when(appenders[1]).flush(Matchers.any(IFrameWriter.class), Matchers.anyBoolean());
+ }).when(appenders[1]).write(Matchers.any(IFrameWriter.class), Matchers.anyBoolean());
return appenders;
}
diff --git a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
index 631b2f1..59d852e 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexBulkLoadOperatorNodePushable.java
@@ -91,7 +91,6 @@
}
}
FrameUtils.flushFrame(buffer, writer);
-
}
@Override
@@ -113,6 +112,11 @@
}
@Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
+
+ @Override
public void fail() throws HyracksDataException {
if (index != null) {
writer.fail();
diff --git a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
index 6c7e8d0..6736420 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
@@ -162,4 +162,9 @@
writer.fail();
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ writer.flush();
+ }
}
diff --git a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
index 568bde8..c4f3fca 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
@@ -188,12 +188,17 @@
}
@Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
+
+ @Override
public void close() throws HyracksDataException {
HyracksDataException closeException = null;
if (index != null) {
// if index == null, then the index open was not successful
try {
- appender.flush(writer, true);
+ appender.write(writer, true);
} catch (Throwable th) {
closeException = new HyracksDataException(th);
}
diff --git a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexDiskOrderScanOperatorNodePushable.java b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexDiskOrderScanOperatorNodePushable.java
index 08775bb..1b1bee0 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexDiskOrderScanOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexDiskOrderScanOperatorNodePushable.java
@@ -82,7 +82,7 @@
FrameUtils.appendToWriter(writer, appender, tb.getFieldEndOffsets(), tb.getByteArray(), 0,
tb.getSize());
}
- appender.flush(writer, true);
+ appender.write(writer, true);
} catch (Throwable th) {
writer.fail();
throw new HyracksDataException(th);
diff --git a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexStatsOperatorNodePushable.java b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexStatsOperatorNodePushable.java
index 48e65bb..f70df29 100644
--- a/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexStatsOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexStatsOperatorNodePushable.java
@@ -84,7 +84,7 @@
throw new HyracksDataException("Record size (" + tb.getSize() + ") larger than frame size ("
+ appender.getBuffer().capacity() + ")");
}
- appender.flush(writer, false);
+ appender.write(writer, false);
} catch (Exception e) {
writer.fail();
throw new HyracksDataException(e);
diff --git a/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java b/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
index 50f7be0..25f2382 100644
--- a/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
@@ -120,6 +120,6 @@
for (int i = startTupleIndex; i < endTupleIndex; i++) {
FrameUtils.appendToWriter(writer, appender, accessor, i);
}
- appender.flush(writer, true);
+ appender.write(writer, true);
}
}
diff --git a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/dataflow/BinaryTokenizerOperatorNodePushable.java b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/dataflow/BinaryTokenizerOperatorNodePushable.java
index c91aff7..002457b 100644
--- a/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/dataflow/BinaryTokenizerOperatorNodePushable.java
+++ b/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/dataflow/BinaryTokenizerOperatorNodePushable.java
@@ -154,7 +154,7 @@
@Override
public void close() throws HyracksDataException {
try {
- appender.flush(writer, true);
+ appender.write(writer, true);
} finally {
writer.close();
}
@@ -164,4 +164,9 @@
public void fail() throws HyracksDataException {
writer.fail();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer);
+ }
}