asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From il...@apache.org
Subject incubator-asterixdb-hyracks git commit: ASTERIXDB-1002: Fix exception handling in EmptyTupleSourceRuntimeFactory
Date Fri, 09 Oct 2015 07:37:27 GMT
Repository: incubator-asterixdb-hyracks
Updated Branches:
  refs/heads/master c4118bd0c -> 87428ba48


ASTERIXDB-1002: Fix exception handling in EmptyTupleSourceRuntimeFactory

Revisiting the previous fix by calling fail() on pipeline

Change-Id: I19f8c8485e483e4d4efeff939e6bd82c7a04a101
Reviewed-on: https://asterix-gerrit.ics.uci.edu/443
Tested-by: Jenkins <jenkins@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <hubailmor@gmail.com>
Reviewed-by: Ian Maxon <imaxon@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/commit/87428ba4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/tree/87428ba4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/diff/87428ba4

Branch: refs/heads/master
Commit: 87428ba48ec358ef6eb5d5e6875bf13b886b8f66
Parents: c4118bd
Author: Ildar Absalyamov <ildar.absalyamov@gmail.com>
Authored: Fri Oct 9 00:06:31 2015 -0700
Committer: Ildar Absalyamov <ildar.absalyamov@gmail.com>
Committed: Fri Oct 9 00:28:04 2015 -0700

----------------------------------------------------------------------
 .../meta/AlgebricksMetaOperatorDescriptor.java  | 27 ++++++++++++--------
 .../std/EmptyTupleSourceRuntimeFactory.java     | 19 +++++++-------
 .../nc/partitions/PipelinedPartition.java       |  4 ++-
 3 files changed, 29 insertions(+), 21 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/87428ba4/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
----------------------------------------------------------------------
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 6618326..1a7150e 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
@@ -20,9 +20,6 @@ package org.apache.hyracks.algebricks.runtime.operators.meta;
 
 import java.nio.ByteBuffer;
 
-import org.json.JSONException;
-import org.json.JSONObject;
-
 import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
 import org.apache.hyracks.algebricks.runtime.base.AlgebricksPipeline;
 import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
@@ -36,6 +33,8 @@ import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
 import org.apache.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
 import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
 import org.apache.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
+import org.json.JSONException;
+import org.json.JSONObject;
 
 public class AlgebricksMetaOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor
{
 
@@ -93,10 +92,11 @@ public class AlgebricksMetaOperatorDescriptor extends AbstractSingleActivityOper
             final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions)
{
         return new AbstractUnaryOutputSourceOperatorNodePushable() {
 
+            @Override
             public void initialize() throws HyracksDataException {
                 IFrameWriter startOfPipeline;
-                RecordDescriptor pipelineOutputRecordDescriptor = outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.recordDescriptors[0]
-                        : null;
+                RecordDescriptor pipelineOutputRecordDescriptor = outputArity > 0
+                        ? AlgebricksMetaOperatorDescriptor.this.recordDescriptors[0] : null;
 
                 PipelineAssembler pa = new PipelineAssembler(pipeline, inputArity, outputArity,
null,
                         pipelineOutputRecordDescriptor);
@@ -105,8 +105,13 @@ public class AlgebricksMetaOperatorDescriptor extends AbstractSingleActivityOper
                 } catch (AlgebricksException e) {
                     throw new HyracksDataException(e);
                 }
-                startOfPipeline.open();
-                startOfPipeline.close();
+                try {
+                    startOfPipeline.open();
+                } catch (HyracksDataException e) {
+                    startOfPipeline.fail();
+                } finally {
+                    startOfPipeline.close();
+                }
             }
         };
     }
@@ -120,10 +125,10 @@ public class AlgebricksMetaOperatorDescriptor extends AbstractSingleActivityOper
             @Override
             public void open() throws HyracksDataException {
                 if (startOfPipeline == null) {
-                    RecordDescriptor pipelineOutputRecordDescriptor = outputArity > 0
? AlgebricksMetaOperatorDescriptor.this.recordDescriptors[0]
-                            : null;
-                    RecordDescriptor pipelineInputRecordDescriptor = recordDescProvider.getInputRecordDescriptor(
-                            AlgebricksMetaOperatorDescriptor.this.getActivityId(), 0);
+                    RecordDescriptor pipelineOutputRecordDescriptor = outputArity > 0
+                            ? AlgebricksMetaOperatorDescriptor.this.recordDescriptors[0]
: null;
+                    RecordDescriptor pipelineInputRecordDescriptor = recordDescProvider
+                            .getInputRecordDescriptor(AlgebricksMetaOperatorDescriptor.this.getActivityId(),
0);
                     PipelineAssembler pa = new PipelineAssembler(pipeline, inputArity, outputArity,
                             pipelineInputRecordDescriptor, pipelineOutputRecordDescriptor);
                     try {

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/87428ba4/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/std/EmptyTupleSourceRuntimeFactory.java
----------------------------------------------------------------------
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 5b66736..a2b9652 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
@@ -43,20 +43,21 @@ public class EmptyTupleSourceRuntimeFactory implements IPushRuntimeFactory
{
     public IPushRuntime createPushRuntime(final IHyracksTaskContext ctx) throws HyracksDataException
{
         return new AbstractOneInputSourcePushRuntime() {
 
-            private ArrayTupleBuilder tb = new ArrayTupleBuilder(0);
-            private FrameTupleAppender appender = new FrameTupleAppender(new VSizeFrame(ctx));
+            private final ArrayTupleBuilder tb = new ArrayTupleBuilder(0);
+            private final FrameTupleAppender appender = new FrameTupleAppender(new VSizeFrame(ctx));
 
             @Override
             public void open() throws HyracksDataException {
                 writer.open();
-                try {
-                    if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize()))
{
-                        throw new IllegalStateException();
-                    }
-                    appender.flush(writer, true);
-                } finally {
-                    writer.close();
+                if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize()))
{
+                    throw new IllegalStateException();
                 }
+                appender.flush(writer, true);
+            }
+
+            @Override
+            public void close() throws HyracksDataException {
+                writer.close();
             }
         };
     }

http://git-wip-us.apache.org/repos/asf/incubator-asterixdb-hyracks/blob/87428ba4/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/PipelinedPartition.java
----------------------------------------------------------------------
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 7f6dd10..2ca1e0f 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
@@ -81,7 +81,9 @@ public class PipelinedPartition implements IFrameWriter, IPartition {
     @Override
     public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
         ensureConnected();
-        delegate.nextFrame(buffer);
+        if (!failed) {
+            delegate.nextFrame(buffer);
+        }
     }
 
     private void ensureConnected() throws HyracksDataException {


Mime
View raw message