abdullah alamoudi has uploaded a new change for review.
https://asterix-gerrit.ics.uci.edu/585
Change subject: Add flush() to IFrameWriter
......................................................................
Add flush() to IFrameWriter
This method is expected to be used with feeds to push
frames all the way to storage when needed. As of now, it is
needed in two cases:
1. No activities in ingestion node and need to push content
so it can be stored.
2. When the ingestion node needs to move the checkpoint ahead
if the at least once semantics are used.
Change-Id: Id862ce9e9b1360864c6976f2aea2137092f51203
---
M
asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
M asterix-app/src/main/java/org/apache/asterix/feeds/FeedMessageReceiver.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/DistributeFeedFrameWriter.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCache.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCollector.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameHandlers.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedRuntimeInputHandler.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/MessageReceiver.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/MonitoredBuffer.java
M
asterix-common/src/main/java/org/apache/asterix/common/feeds/StorageSideMonitoredBuffer.java
M
asterix-external-data/src/main/java/org/apache/asterix/external/dataset/adapter/LookupAdapter.java
M
asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalLookupOperatorDescriptor.java
M
asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/CollectTransformFeedFrameWriter.java
M
asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaComputeNodePushable.java
M
asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaNodePushable.java
M
asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaStoreNodePushable.java
M
asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/NoTupleSourceRuntimeFactory.java
17 files changed, 109 insertions(+), 28 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/85/585/1
diff --git
a/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
b/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
index 92cd61a..ce80291 100644
---
a/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
+++
b/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
@@ -139,4 +139,8 @@
public void setInputRecordDescriptor(int index, RecordDescriptor
recordDescriptor) {
this.frameTupleAccessor = new FrameTupleAccessor(recordDescriptor);
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ }
}
\ No newline at end of file
diff --git
a/asterix-app/src/main/java/org/apache/asterix/feeds/FeedMessageReceiver.java
b/asterix-app/src/main/java/org/apache/asterix/feeds/FeedMessageReceiver.java
index a3cd217..4fa7712 100644
---
a/asterix-app/src/main/java/org/apache/asterix/feeds/FeedMessageReceiver.java
+++
b/asterix-app/src/main/java/org/apache/asterix/feeds/FeedMessageReceiver.java
@@ -20,8 +20,6 @@
import java.util.logging.Level;
-import org.json.JSONObject;
-
import org.apache.asterix.common.feeds.FeedConstants;
import org.apache.asterix.common.feeds.FeedTupleCommitAckMessage;
import org.apache.asterix.common.feeds.MessageReceiver;
@@ -36,6 +34,8 @@
import org.apache.asterix.common.feeds.message.ThrottlingEnabledFeedMessage;
import org.apache.asterix.feeds.CentralFeedManager.AQLExecutor;
import org.apache.asterix.hyracks.bootstrap.FeedBootstrap;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.json.JSONObject;
public class FeedMessageReceiver extends MessageReceiver<String> {
@@ -89,4 +89,8 @@
}
}
+
+ @Override
+ public void emptyInbox() throws HyracksDataException {
+ }
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/DistributeFeedFrameWriter.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/DistributeFeedFrameWriter.java
index d0e371e..5652545 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/DistributeFeedFrameWriter.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/DistributeFeedFrameWriter.java
@@ -140,4 +140,11 @@
public FrameDistributor.DistributionMode getDistributionMode() {
return frameDistributor.getDistributionMode();
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // Doesn't forward anything. this writer is followed by a frame
distributor and a null sink.
+ // The frame distributor will send the buffer to collectors who have
their own algorithm for flushing
+ // frames forward
+ }
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCache.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCache.java
index 55a7fb8..b4bcb3b 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCache.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCache.java
@@ -100,7 +100,6 @@
/**
* Replay the frame from the tuple (inclusive) with recordId as specified.
- *
* @param recordId
* @param frame
* @throws HyracksDataException
@@ -131,7 +130,6 @@
/**
* Replay the frame
- *
* @param frame
* @throws HyracksDataException
*/
@@ -168,4 +166,8 @@
frameWriter.nextFrame(frame);
}
}
+
+ @Override
+ public void emptyInbox() {
+ }
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCollector.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCollector.java
index a8c0e8f..f06971f 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCollector.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameCollector.java
@@ -155,4 +155,9 @@
return connectionId.toString().hashCode();
}
+ @Override
+ public synchronized void emptyInbox() throws HyracksDataException {
+ frameWriter.flush();
+ }
+
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameHandlers.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameHandlers.java
index c4a2ce0..4f195dd 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameHandlers.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedFrameHandlers.java
@@ -270,6 +270,10 @@
};
}
+ @Override
+ public void emptyInbox() throws HyracksDataException {
+ }
+
}
@Override
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedRuntimeInputHandler.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedRuntimeInputHandler.java
index 6642df1..4d5ba79 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedRuntimeInputHandler.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/FeedRuntimeInputHandler.java
@@ -437,4 +437,12 @@
public void setBufferingEnabled(boolean bufferingEnabled) {
this.bufferingEnabled = bufferingEnabled;
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ // Only flush when in process mode.
+ if (mode == Mode.PROCESS) {
+ coreOperator.flush();
+ }
+ }
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/MessageReceiver.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/MessageReceiver.java
index 6490c6a..8350631 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/MessageReceiver.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/MessageReceiver.java
@@ -25,6 +25,7 @@
import java.util.logging.Logger;
import org.apache.asterix.common.feeds.api.IMessageReceiver;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
public abstract class MessageReceiver<T> implements IMessageReceiver<T> {
@@ -74,12 +75,17 @@
this.inbox = messageReceiver.inbox;
this.messageReceiver = messageReceiver;
}
+ // TODO: this should handle exceptions better
@Override
public void run() {
while (true) {
try {
- T message = inbox.take();
+ T message = inbox.poll();
+ if (message == null) {
+ messageReceiver.emptyInbox();
+ message = inbox.take();
+ }
messageReceiver.processMessage(message);
} catch (InterruptedException e) {
e.printStackTrace();
@@ -108,4 +114,6 @@
}
}
+ public abstract void emptyInbox() throws HyracksDataException;
+
}
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/MonitoredBuffer.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/MonitoredBuffer.java
index e5a22b5..1168002 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/MonitoredBuffer.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/MonitoredBuffer.java
@@ -385,4 +385,9 @@
return storageTimeTrackingRateTask;
}
+ @Override
+ public void emptyInbox() throws HyracksDataException {
+ inputHandler.flush();
+ }
+
}
\ No newline at end of file
diff --git
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/StorageSideMonitoredBuffer.java
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/StorageSideMonitoredBuffer.java
index 4027237..84e07a5 100644
---
a/asterix-common/src/main/java/org/apache/asterix/common/feeds/StorageSideMonitoredBuffer.java
+++
b/asterix-common/src/main/java/org/apache/asterix/common/feeds/StorageSideMonitoredBuffer.java
@@ -34,7 +34,7 @@
public class StorageSideMonitoredBuffer extends MonitoredBuffer {
private static final long STORAGE_TIME_TRACKING_FREQUENCY = 5000; // 10
- //
seconds
+ // seconds
private boolean ackingEnabled;
private final boolean timeTrackingEnabled;
@@ -202,5 +202,4 @@
protected boolean reportInflowRate() {
return false;
}
-
}
diff --git
a/asterix-external-data/src/main/java/org/apache/asterix/external/dataset/adapter/LookupAdapter.java
b/asterix-external-data/src/main/java/org/apache/asterix/external/dataset/adapter/LookupAdapter.java
index ba6f83c..28ee91b 100644
---
a/asterix-external-data/src/main/java/org/apache/asterix/external/dataset/adapter/LookupAdapter.java
+++
b/asterix-external-data/src/main/java/org/apache/asterix/external/dataset/adapter/LookupAdapter.java
@@ -159,4 +159,10 @@
writer.close();
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ appender.flush(writer, true);
+ writer.flush();
+ }
}
diff --git
a/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalLookupOperatorDescriptor.java
b/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalLookupOperatorDescriptor.java
index c8881a3..2cab3a7 100644
---
a/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalLookupOperatorDescriptor.java
+++
b/asterix-external-data/src/main/java/org/apache/asterix/external/operators/ExternalLookupOperatorDescriptor.java
@@ -66,7 +66,7 @@
final IRecordDescriptorProvider recordDescProvider, final int
partition, int nPartitions)
throws HyracksDataException {
// Create a file index accessor to be used for files lookup operations
- // Note that all file index accessors will use partition 0 since we
only have 1 files index per NC
+ // Note that all file index accessors will use partition 0 since we
only have 1 files index per NC
final ExternalFileIndexAccessor snapshotAccessor = new
ExternalFileIndexAccessor(
(ExternalBTreeDataflowHelper)
dataflowHelperFactory.createIndexDataflowHelper(this, ctx, partition),
this);
@@ -80,7 +80,7 @@
try {
adapter = adapterFactory.createAdapter(ctx, partition,
recordDescProvider.getInputRecordDescriptor(getActivityId(), 0),
snapshotAccessor, writer);
- //Open the file index accessor here
+ // Open the file index accessor here
snapshotAccessor.open();
indexOpen = true;
adapter.open();
@@ -127,6 +127,11 @@
throw new HyracksDataException(th);
}
}
+
+ @Override
+ public void flush() throws HyracksDataException {
+ adapter.flush();
+ }
};
}
}
diff --git
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/CollectTransformFeedFrameWriter.java
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/CollectTransformFeedFrameWriter.java
index 30b369a..1c59b3b 100644
---
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/CollectTransformFeedFrameWriter.java
+++
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/CollectTransformFeedFrameWriter.java
@@ -47,7 +47,7 @@
public CollectTransformFeedFrameWriter(IHyracksTaskContext ctx,
IFrameWriter downstreamWriter,
ISubscribableRuntime sourceRuntime, RecordDescriptor
outputRecordDescriptor, FeedConnectionId connectionId)
- throws HyracksDataException {
+ throws HyracksDataException {
this.downstreamWriter = downstreamWriter;
RecordDescriptor inputRecordDescriptor =
sourceRuntime.getRecordDescriptor();
inputFrameTupleAccessor = new
FrameTupleAccessor(inputRecordDescriptor);
@@ -64,6 +64,7 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+ // always project the first field only. why?
inputFrameTupleAccessor.reset(buffer);
int nTuple = inputFrameTupleAccessor.getTupleCount();
for (int t = 0; t < nTuple; t++) {
@@ -117,4 +118,12 @@
this.downstreamWriter = writer;
}
+ @Override
+ public void flush() throws HyracksDataException {
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.flush(downstreamWriter, true);
+ }
+ downstreamWriter.flush();
+ }
+
}
\ No newline at end of file
diff --git
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaComputeNodePushable.java
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaComputeNodePushable.java
index f833019..bd75841 100644
---
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaComputeNodePushable.java
+++
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaComputeNodePushable.java
@@ -34,8 +34,8 @@
import org.apache.asterix.common.feeds.api.IFeedManager;
import org.apache.asterix.common.feeds.api.IFeedRuntime.FeedRuntimeType;
import org.apache.asterix.common.feeds.api.IFeedRuntime.Mode;
-import org.apache.asterix.external.feeds.FeedPolicyEnforcer;
import org.apache.asterix.common.feeds.api.ISubscribableRuntime;
+import org.apache.asterix.external.feeds.FeedPolicyEnforcer;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IActivity;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
@@ -224,4 +224,9 @@
}
}
+ @Override
+ public void flush() throws HyracksDataException {
+ inputSideHandler.flush();
+ }
+
}
\ No newline at end of file
diff --git
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaNodePushable.java
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaNodePushable.java
index 86f8750..9d60dd9 100644
---
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaNodePushable.java
+++
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaNodePushable.java
@@ -99,8 +99,8 @@
this.partition = partition;
this.nPartitions = nPartitions;
this.connectionId = feedConnectionId;
- this.feedManager = ((IAsterixAppRuntimeContext)
(IAsterixAppRuntimeContext) ctx.getJobletContext()
-
.getApplicationContext().getApplicationObject()).getFeedManager();
+ this.feedManager = ((IAsterixAppRuntimeContext)
ctx.getJobletContext().getApplicationContext()
+ .getApplicationObject()).getFeedManager();
IAsterixAppRuntimeContext runtimeCtx = (IAsterixAppRuntimeContext)
ctx.getJobletContext()
.getApplicationContext().getApplicationObject();
this.feedManager = runtimeCtx.getFeedManager();
@@ -126,10 +126,8 @@
private void initializeNewFeedRuntime(FeedRuntimeId runtimeId) throws
Exception {
this.fta = new FrameTupleAccessor(recordDesc);
- this.inputSideHandler = new FeedRuntimeInputHandler(ctx, connectionId,
runtimeId,
- (AbstractUnaryInputUnaryOutputOperatorNodePushable)
coreOperator,
- policyEnforcer.getFeedPolicyAccessor(), false, fta,
recordDesc, feedManager,
- nPartitions);
+ this.inputSideHandler = new FeedRuntimeInputHandler(ctx, connectionId,
runtimeId, coreOperator,
+ policyEnforcer.getFeedPolicyAccessor(), false, fta,
recordDesc, feedManager, nPartitions);
setupBasicRuntime(inputSideHandler);
}
@@ -186,4 +184,9 @@
}
}
+ @Override
+ public void flush() throws HyracksDataException {
+ inputSideHandler.flush();
+ }
+
}
\ No newline at end of file
diff --git
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaStoreNodePushable.java
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaStoreNodePushable.java
index b409745..76ddaca 100644
---
a/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaStoreNodePushable.java
+++
b/asterix-metadata/src/main/java/org/apache/asterix/metadata/feeds/FeedMetaStoreNodePushable.java
@@ -98,8 +98,8 @@
this.partition = partition;
this.nPartitions = nPartitions;
this.connectionId = feedConnectionId;
- this.feedManager = ((IAsterixAppRuntimeContext)
(IAsterixAppRuntimeContext) ctx.getJobletContext()
-
.getApplicationContext().getApplicationObject()).getFeedManager();
+ this.feedManager = ((IAsterixAppRuntimeContext)
ctx.getJobletContext().getApplicationContext()
+ .getApplicationObject()).getFeedManager();
IAsterixAppRuntimeContext runtimeCtx = (IAsterixAppRuntimeContext)
ctx.getJobletContext()
.getApplicationContext().getApplicationObject();
this.feedManager = runtimeCtx.getFeedManager();
@@ -130,11 +130,10 @@
}
this.fta = new FrameTupleAccessor(recordDesc);
this.inputSideHandler = new FeedRuntimeInputHandler(ctx, connectionId,
runtimeId, coreOperator,
- policyEnforcer.getFeedPolicyAccessor(), true, fta, recordDesc,
feedManager,
- nPartitions);
- if(coreOperator instanceof AsterixLSMInsertDeleteOperatorNodePushable){
+ policyEnforcer.getFeedPolicyAccessor(), true, fta, recordDesc,
feedManager, nPartitions);
+ if (coreOperator instanceof
AsterixLSMInsertDeleteOperatorNodePushable) {
AsterixLSMInsertDeleteOperatorNodePushable indexOp =
(AsterixLSMInsertDeleteOperatorNodePushable) coreOperator;
- if(!indexOp.isPrimary()){
+ if (!indexOp.isPrimary()) {
inputSideHandler.setBufferingEnabled(false);
}
}
@@ -149,8 +148,8 @@
this.inputSideHandler.setCoreOperator(coreOperator);
feedRuntime.setMode(Mode.PROCESS);
if (LOGGER.isLoggable(Level.WARNING)) {
- LOGGER.warning("Retreived state from the zombie instance from
previous execution for " + runtimeType
- + " node.");
+ LOGGER.warning(
+ "Retreived state from the zombie instance from previous
execution for " + runtimeType + " node.");
}
}
@@ -158,7 +157,7 @@
coreOperator.setOutputFrameWriter(0, writer, recordDesc);
FeedRuntimeId runtimeId = new FeedRuntimeId(runtimeType, partition,
operandId);
feedRuntime = new FeedRuntime(runtimeId, inputHandler, writer);
-
feedManager.getFeedConnectionManager().registerFeedRuntime(connectionId,
(FeedRuntime) feedRuntime);
+
feedManager.getFeedConnectionManager().registerFeedRuntime(connectionId,
feedRuntime);
}
@Override
@@ -216,9 +215,13 @@
private void deregister() {
if (feedRuntime != null) {
-
feedManager.getFeedConnectionManager().deRegisterFeedRuntime(connectionId,
- ((FeedRuntime) feedRuntime).getRuntimeId());
+
feedManager.getFeedConnectionManager().deRegisterFeedRuntime(connectionId,
feedRuntime.getRuntimeId());
}
}
+ @Override
+ public void flush() throws HyracksDataException {
+ inputSideHandler.flush();
+ }
+
}
\ No newline at end of file
diff --git
a/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/NoTupleSourceRuntimeFactory.java
b/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/NoTupleSourceRuntimeFactory.java
index c3e2681..8e1a322 100644
---
a/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/NoTupleSourceRuntimeFactory.java
+++
b/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/std/NoTupleSourceRuntimeFactory.java
@@ -43,6 +43,10 @@
writer.close();
}
+ @Override
+ public void flush() throws HyracksDataException {
+ }
+
};
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/585
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Id862ce9e9b1360864c6976f2aea2137092f51203
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: abdullah alamoudi <[email protected]>
