fix(analytical): Resolve the issue of the Java app's results not being outputted by the context.

analytical_engine/core/context/java_context_base.h

@@ -132,7 +132,8 @@ class JavaContextBase : public grape::ContextBase {
           LOG(ERROR) << "Exception occurred when calling write back method";
         }
       } else {
-        VLOG(2) << "Not write back method found";
+        VLOG(2) << "Not write back method found, try to call output";
+        callJavaContextOutput();
       }
     }
   }
@@ -237,6 +238,31 @@ class JavaContextBase : public grape::ContextBase {
     }
   }
 
+  // We expect user to write the sync back logic in their java code.
+  // i.e. copy the data from java heap to cpp context heap.
+  void callJavaContextOutput() {
+    JNIEnvMark m;
+    if (m.env()) {
+      JNIEnv* env = m.env();
+
+      jclass context_class = env->GetObjectClass(this->context_object());
+      CHECK_NOTNULL(context_class);
+
+      const char* descriptor = "(Lcom/alibaba/graphscope/fragment/IFragment;)V";
+      jmethodID output_methodID =
+          env->GetMethodID(context_class, "Output", descriptor);
+      if (output_methodID) {
+        VLOG(1) << "Found output method in java context.";
+        env->CallVoidMethod(this->context_object(), output_methodID,
+                            this->fragment_object());
+      } else {
+        VLOG(1) << "Output method not found, skip.";
+      }
+    } else {
+      LOG(ERROR) << "JNI env not available.";
+    }
+  }
+
  private:
   /**
    * @brief Generate user class path, i.e. URLClassLoader class path, from lib

analytical_engine/core/context/java_pie_projected_context.h

@@ -71,30 +71,12 @@ class JavaPIEProjectedContext : public JavaContextBase<FRAG_T> {
   }
 
   void Output(std::ostream& os) override {
-    JNIEnvMark m;
-    if (m.env()) {
-      JNIEnv* env = m.env();
-
-      jclass context_class = env->GetObjectClass(this->context_object());
-      CHECK_NOTNULL(context_class);
-
-      const char* descriptor = "(Lcom/alibaba/graphscope/fragment/IFragment;)V";
-      jmethodID output_methodID =
-          env->GetMethodID(context_class, "Output", descriptor);
-      if (output_methodID) {
-        VLOG(1) << "Found output method in java context.";
-        env->CallVoidMethod(this->context_object(), output_methodID,
-                            this->fragment_object());
-      } else {
-        VLOG(1) << "Output method not found, skip.";
-      }
-    } else {
-      LOG(ERROR) << "JNI env not available.";
-    }
+    LOG(WARNING) << "Output is not supported for JavaPIEProjectedContext";
   }
 
   std::shared_ptr<gs::IContextWrapper> CreateInnerCtxWrapper(
       const std::string& id, std::shared_ptr<IFragmentWrapper> frag_wrapper) {
+    JavaContextBase<FRAG_T>::WriteBackJVMHeapToCppContext();
     std::string java_ctx_type_name = getJavaCtxTypeName(this->context_object());
     VLOG(1) << "Java ctx type name" << java_ctx_type_name;
     if (java_ctx_type_name == "VertexDataContext") {
@@ -144,6 +126,15 @@ class JavaPIEProjectedContext : public JavaContextBase<FRAG_T> {
         std::shared_ptr<inner_ctx_type> inner_ctx_impl_shared(inner_ctx_impl);
         return std::make_shared<inner_ctx_wrapper_type>(id, frag_wrapper,
                                                         inner_ctx_impl_shared);
+      } else if (data_type == "std::string") {
+        using inner_ctx_type = grape::VertexDataContext<FRAG_T, std::string>;
+        using inner_ctx_wrapper_type =
+            VertexDataContextWrapper<FRAG_T, std::string>;
+        auto inner_ctx_impl =
+            reinterpret_cast<inner_ctx_type*>(this->inner_context_addr());
+        std::shared_ptr<inner_ctx_type> inner_ctx_impl_shared(inner_ctx_impl);
+        return std::make_shared<inner_ctx_wrapper_type>(id, frag_wrapper,
+                                                        inner_ctx_impl_shared);
       } else {
         LOG(ERROR) << "Unrecognizable data type: " << data_type;
       }

analytical_engine/core/context/java_pie_property_context.h

@@ -94,6 +94,7 @@ class JavaPIEPropertyContext : public JavaContextBase<FRAG_T> {
 
   std::shared_ptr<gs::IContextWrapper> CreateInnerCtxWrapper(
       const std::string& id, std::shared_ptr<IFragmentWrapper> frag_wrapper) {
+    JavaContextBase<FRAG_T>::WriteBackJVMHeapToCppContext();
     std::string java_ctx_type_name = getJavaCtxTypeName(this->context_object());
     VLOG(1) << "Java ctx type name" << java_ctx_type_name;
     if (java_ctx_type_name == "LabeledVertexDataContext") {

analytical_engine/core/worker/default_worker.h

@@ -131,32 +131,13 @@ class DefaultWorker {
     MPI_Barrier(comm_spec_.comm());
 
     messages_.Finalize();
-    finishQuery();
   }
 
   std::shared_ptr<context_t> GetContext() { return context_; }
 
   void Output(std::ostream& os) { context_->Output(os); }
 
  private:
-  template <typename T = context_t>
-  typename std::enable_if<
-      std::is_base_of<JavaContextBase<fragment_t>, T>::value>::type
-  finishQuery() {
-    auto java_context =
-        std::dynamic_pointer_cast<JavaContextBase<fragment_t>>(context_);
-    if (java_context) {
-      VLOG(1) << "Write java heap data back to cpp context since it is java "
-                 "context";
-      java_context->WriteBackJVMHeapToCppContext();
-    }
-  }
-
-  template <typename T = context_t>
-  typename std::enable_if<
-      !std::is_base_of<JavaContextBase<fragment_t>, T>::value>::type
-  finishQuery() {}
-
   std::shared_ptr<APP_T> app_;
   std::shared_ptr<context_t> context_;
   message_manager_t messages_;

analytical_engine/frame/app_frame.cc

@@ -120,7 +120,7 @@ void Query(void* worker_handler, const gs::rpc::QueryArgs& query_args,
            const std::string& context_key,
            std::shared_ptr<gs::IFragmentWrapper> frag_wrapper,
            std::shared_ptr<gs::IContextWrapper>& ctx_wrapper,
-           bl::result<nullptr_t>& wrapper_error) {
+           bl::result<std::nullptr_t>& wrapper_error) {
   __FRAME_CATCH_AND_ASSIGN_GS_ERROR(
       wrapper_error, detail::Query(worker_handler, query_args, context_key,
                                    frag_wrapper, ctx_wrapper));

analytical_engine/java/grape-demo/src/main/java/com/alibaba/graphscope/example/giraph/circle/Circle.java

@@ -0,0 +1,181 @@
+/*
+ * Copyright 2022 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *  	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package com.alibaba.graphscope.example.giraph.circle;
+
+import com.google.common.collect.Lists;
+
+import org.apache.giraph.Algorithm;
+import org.apache.giraph.edge.Edge;
+import org.apache.giraph.graph.BasicComputation;
+import org.apache.giraph.graph.Vertex;
+import org.apache.hadoop.io.LongWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Algorithm(name = "Circle", description = "Finds Circle")
+public class Circle
+        extends BasicComputation<
+                LongWritable, VertexAttrWritable, LongWritable, VertexAttrWritable> {
+    private static final Logger logger = LoggerFactory.getLogger(Circle.class);
+    int maxIteration = 3;
+
+    public Circle() {}
+
+    public void preSuperstep() {
+        this.maxIteration = Integer.parseInt(this.getConf().get("max", "3"));
+        logger.info("[preSuperstep] max is {}", this.maxIteration);
+    }
+
+    public void compute(
+            Vertex<LongWritable, VertexAttrWritable, LongWritable> vertex,
+            Iterable<VertexAttrWritable> messages)
+            throws IOException {
+        this.maxIteration = Integer.parseInt(this.getConf().get("max", "3"));
+        long superStep = this.getSuperstep();
+
+        List<MsgWritable> writables = new ArrayList<MsgWritable>();
+        for (int i = 0; i < vertex.getId().get() % 5; ++i) {
+            writables.add(new MsgWritable(Arrays.asList(1L, 2L, 3L), Arrays.asList(12L, 23L)));
+        }
+        vertex.setValue(new VertexAttrWritable(writables));
+        vertex.voteToHalt();
+    }
+
+    private VertexAttrWritable mergeMsg(Iterable<VertexAttrWritable> messages) {
+        VertexAttrWritable merged = new VertexAttrWritable();
+
+        VertexAttrWritable mess;
+        for (Iterator var3 = messages.iterator();
+                var3.hasNext();
+                merged = this.merge(merged, mess)) {
+            mess = (VertexAttrWritable) var3.next();
+        }
+
+        return merged;
+    }
+
+    private void vprog(long vid, VertexAttrWritable vdata, VertexAttrWritable message) {
+        long superStep = this.getSuperstep();
+        List<MsgWritable> nodeAttr = vdata.getVertexAttr();
+        List<MsgWritable> messageAttr = message.getVertexAttr();
+        List<MsgWritable> processedMsg =
+                (List)
+                        messageAttr.stream()
+                                .peek(
+                                        (item) -> {
+                                            List<Long> vlist = item.getVertexPath();
+                                            this.addVertexToPath(vid, vlist);
+                                        })
+                                .collect(Collectors.toList());
+        nodeAttr.addAll(processedMsg);
+        List<MsgWritable> finalNodeAttr =
+                (List)
+                        nodeAttr.stream()
+                                .distinct()
+                                .filter(
+                                        (item) -> {
+                                            List<Long> vertexPath = item.getVertexPath();
+                                            return this.filterPathInVertexAttr(
+                                                    vertexPath, superStep + 1L);
+                                        })
+                                .collect(Collectors.toList());
+        vdata.setVertexAttr(finalNodeAttr);
+    }
+
+    private void addVertexToPath(long vid, List<Long> path) {
+        if (path.isEmpty()) {
+            path.add(vid);
+        } else {
+            int pathSize = path.size();
+            if (pathSize == 1 && !path.contains(vid)) {
+                path.add(vid);
+            } else {
+                if (!path.subList(1, pathSize).contains(vid)) {
+                    path.add(vid);
+                }
+            }
+        }
+    }
+
+    private boolean filterPathInVertexAttr(List<Long> vertexPath, long iteration) {
+        int pathSize = vertexPath.size();
+        return MsgWritable.isCircle(vertexPath) || (long) pathSize == iteration;
+    }
+
+    private VertexAttrWritable merge(VertexAttrWritable attr1, VertexAttrWritable attr2) {
+        List<MsgWritable> sp1 = attr1.getVertexAttr();
+        sp1.addAll(attr2.getVertexAttr());
+        return new VertexAttrWritable((List) sp1.stream().distinct().collect(Collectors.toList()));
+    }
+
+    private void sendMsg(Vertex<LongWritable, VertexAttrWritable, LongWritable> vertex) {
+        List<MsgWritable> currVertexAttr = ((VertexAttrWritable) vertex.getValue()).getVertexAttr();
+        long superStep = this.getSuperstep();
+        if (!currVertexAttr.isEmpty()) {
+            Iterator var5 = vertex.getEdges().iterator();
+
+            while (var5.hasNext()) {
+                Edge<LongWritable, LongWritable> edge = (Edge) var5.next();
+                long edgeId = ((LongWritable) edge.getValue()).get();
+                List<MsgWritable> finalMsgs =
+                        (List)
+                                currVertexAttr.stream()
+                                        .filter(
+                                                (path) -> {
+                                                    return this.meetMsgCondition(
+                                                            ((LongWritable) vertex.getId()).get(),
+                                                            edgeId,
+                                                            path,
+                                                            superStep);
+                                                })
+                                        .map(
+                                                (path) -> {
+                                                    List<Long> newEdgeSet =
+                                                            Lists.newArrayList(path.getEdgePath());
+                                                    newEdgeSet.add(edgeId);
+                                                    return new MsgWritable(
+                                                            path.getVertexPath(), newEdgeSet);
+                                                })
+                                        .collect(Collectors.toList());
+                if (!finalMsgs.isEmpty()) {
+                    this.sendMessage(edge.getTargetVertexId(), new VertexAttrWritable(finalMsgs));
+                }
+            }
+        }
+    }
+
+    private boolean meetMsgCondition(
+            long currVertexId,
+            long edgeIdToSendMsg,
+            MsgWritable onePathInCurrVertex,
+            long superStep) {
+        List<Long> vertexPath = onePathInCurrVertex.getVertexPath();
+        List<Long> edgePath = onePathInCurrVertex.getEdgePath();
+        int vertexPathSize = vertexPath.size();
+        return (long) vertexPathSize == superStep + 1L
+                && !MsgWritable.isCircle(vertexPath)
+                && currVertexId == (Long) vertexPath.get(vertexPathSize - 1)
+                && !edgePath.contains(edgeIdToSendMsg);
+    }
+}

analytical_engine/java/grape-demo/src/main/java/com/alibaba/graphscope/example/giraph/circle/CircleEdgeInputFormat.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2022 Alibaba Group Holding Limited.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *  	http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package com.alibaba.graphscope.example.giraph.circle;
+
+import org.apache.giraph.io.EdgeReader;
+import org.apache.giraph.io.formats.TextEdgeInputFormat;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+public class CircleEdgeInputFormat extends TextEdgeInputFormat<LongWritable, LongWritable> {
+    public CircleEdgeInputFormat() {}
+
+    public EdgeReader<LongWritable, LongWritable> createEdgeReader(
+            InputSplit split, TaskAttemptContext context) throws IOException {
+        return new CircleEdgeInputFormat.P2PEdgeReader();
+    }
+
+    public class P2PEdgeReader
+            extends TextEdgeInputFormat<LongWritable, LongWritable>
+                            .TextEdgeReaderFromEachLineProcessed<
+                    String[]> {
+        String SEPARATOR = " ";
+        private LongWritable srcId;
+        private LongWritable dstId;
+        private LongWritable edgeValue;
+
+        public P2PEdgeReader() {
+            super();
+        }
+
+        protected String[] preprocessLine(Text line) throws IOException {
+            String[] tokens = line.toString().split(this.SEPARATOR);
+            if (tokens.length != 3) {
+                throw new IllegalStateException("expect 3 ele in edge line");
+            } else {
+                this.srcId = new LongWritable(Long.parseLong(tokens[0]));
+                this.dstId = new LongWritable(Long.parseLong(tokens[1]));
+                this.edgeValue = new LongWritable(Long.parseLong(tokens[2]));
+                return tokens;
+            }
+        }
+
+        protected LongWritable getTargetVertexId(String[] line) throws IOException {
+            return this.dstId;
+        }
+
+        protected LongWritable getSourceVertexId(String[] line) throws IOException {
+            return this.srcId;
+        }
+
+        protected LongWritable getValue(String[] line) throws IOException {
+            return this.edgeValue;
+        }
+    }
+}