Consolidate error handling

Handle errors in the same way in NodeSubscriber and RelationshipSubscriber.

Author: pontusmelke

core/src/main/java/org/neo4j/gds/core/loading/CypherRelationshipLoader.java

@@ -125,6 +125,9 @@ BatchLoadResult loadSingleBatch(InternalTransaction tx, int bufferSize) {
         }
         subscriber.initialize(subscription.fieldNames(), propertyDefaultValueByName);
         CypherLoadingUtils.consume(subscription);
+        subscriber.error().ifPresent(e -> {
+            throw e;
+        });
         progressTracker.endSubTask("Relationships");
         return new BatchLoadResult(subscriber.rows(), -1L);
     }

core/src/main/java/org/neo4j/gds/core/loading/RelationshipSubscriber.java

@@ -25,11 +25,13 @@
 import org.neo4j.gds.core.loading.construction.RelationshipsBuilder;
 import org.neo4j.gds.core.utils.progress.tasks.ProgressTracker;
 import org.neo4j.graphdb.QueryStatistics;
+import org.neo4j.kernel.impl.query.QueryExecutionKernelException;
 import org.neo4j.kernel.impl.query.QuerySubscriber;
 import org.neo4j.values.AnyValue;
 import org.neo4j.values.storable.NumberValue;
 import org.neo4j.values.storable.TextValue;
 
+import java.util.Optional;
 import java.util.Set;
 
 import static org.neo4j.gds.RelationshipType.ALL_RELATIONSHIPS;
@@ -65,6 +67,8 @@ class RelationshipSubscriber implements QuerySubscriber {
 
     private RelationshipsBuilder allRelationshipsBuilder;
 
+    private Optional<RuntimeException> error = Optional.empty();
+
     RelationshipSubscriber(
         IdMap idMap,
         CypherRelationshipLoader.Context loaderContext,
@@ -105,6 +109,10 @@ void initialize(String[] fieldNames, ObjectDoubleMap<String> propertyDefaultValu
         this.propertyValueBuffer = new double[propertyCount];
     }
 
+    Optional<RuntimeException> error() {
+        return error;
+    }
+
     public long rows() {
         return rows;
     }
@@ -183,7 +191,13 @@ public void onRecordCompleted() {
     }
     @Override
     public void onError(Throwable throwable) {
-        throw new RuntimeException(throwable);
+        if (throwable instanceof RuntimeException) {
+            this.error = Optional.of((RuntimeException) throwable);
+        } else if (throwable instanceof QueryExecutionKernelException) {
+            this.error = Optional.of(((QueryExecutionKernelException) throwable).asUserException());
+        } else {
+            this.error = Optional.of(new RuntimeException(throwable));
+        }
     }
 
     @Override