feat: Add production GCN infer pipeline

geaflow/geaflow-common/src/main/java/org/apache/geaflow/common/config/InferConfigHelper.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.common.config;
+
+import static org.apache.geaflow.common.config.keys.FrameworkConfigKeys.INFER_ENV_USER_TRANSFORM_CLASSNAME;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public final class InferConfigHelper {
+
+    private InferConfigHelper() {
+    }
+
+    public static boolean hasTransformClassName(Configuration configuration) {
+        return getTransformClassName(configuration) != null;
+    }
+
+    public static String getTransformClassName(Configuration configuration) {
+        if (configuration == null || !configuration.contains(INFER_ENV_USER_TRANSFORM_CLASSNAME)) {
+            return null;
+        }
+        String transformClassName = configuration.getString(
+            INFER_ENV_USER_TRANSFORM_CLASSNAME.getKey(), null);
+        if (transformClassName == null) {
+            return null;
+        }
+        String normalizedClassName = transformClassName.trim();
+        return normalizedClassName.isEmpty() ? null : normalizedClassName;
+    }
+
+    public static Configuration buildInferConfiguration(Configuration baseConfig,
+                                                        String pythonTransformClassName) {
+        if (baseConfig == null || pythonTransformClassName == null) {
+            return baseConfig;
+        }
+        String normalizedClassName = pythonTransformClassName.trim();
+        if (normalizedClassName.isEmpty()) {
+            return baseConfig;
+        }
+        if (normalizedClassName.equals(getTransformClassName(baseConfig))) {
+            return baseConfig;
+        }
+        Map<String, String> configMap = new HashMap<>(baseConfig.getConfigMap());
+        configMap.put(INFER_ENV_USER_TRANSFORM_CLASSNAME.getKey(), normalizedClassName);
+        Configuration configuration = new Configuration(configMap);
+        configuration.setMasterId(baseConfig.getMasterId());
+        return configuration;
+    }
+}

geaflow/geaflow-core/geaflow-api/src/main/java/org/apache/geaflow/api/graph/base/algo/AbstractIncVertexCentricComputeAlgo.java

@@ -32,6 +32,10 @@ public AbstractIncVertexCentricComputeAlgo(long iterations, String name) {
         super(iterations, name);
     }
 
+    public String getPythonTransformClassName() {
+        return null;
+    }
+
     public abstract FUNC getIncComputeFunction();
 
 }

geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/DynamicGraphVertexCentricComputeOp.java

@@ -30,6 +30,7 @@
 import org.apache.geaflow.api.graph.function.vc.base.IncGraphInferContext;
 import org.apache.geaflow.collector.ICollector;
 import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.config.InferConfigHelper;
 import org.apache.geaflow.common.config.keys.FrameworkConfigKeys;
 import org.apache.geaflow.common.exception.GeaflowRuntimeException;
 import org.apache.geaflow.infer.InferContext;
@@ -164,7 +165,8 @@ class IncGraphInferComputeContextImpl<OUT> extends IncGraphComputeContextImpl im
         public IncGraphInferComputeContextImpl() {
             if (clientLocal.get() == null) {
                 try {
-                    inferContext = new InferContext<>(runtimeContext.getConfiguration());
+                    inferContext = new InferContext<>(buildInferConfiguration(runtimeContext.getConfiguration(),
+                        function.getPythonTransformClassName()));
                 } catch (Exception e) {
                     throw new GeaflowRuntimeException(e);
                 }
@@ -191,4 +193,8 @@ public void close() throws IOException {
             }
         }
     }
+
+    static Configuration buildInferConfiguration(Configuration baseConfig, String pythonTransformClassName) {
+        return InferConfigHelper.buildInferConfiguration(baseConfig, pythonTransformClassName);
+    }
 }

geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/test/java/org/apache/geaflow/operator/impl/graph/compute/dynamic/DynamicGraphVertexCentricComputeOpTest.java

@@ -34,6 +34,7 @@
 import org.apache.geaflow.collector.ICollector;
 import org.apache.geaflow.common.config.Configuration;
 import org.apache.geaflow.common.config.keys.ExecutionConfigKeys;
+import org.apache.geaflow.common.config.keys.FrameworkConfigKeys;
 import org.apache.geaflow.common.task.TaskArgs;
 import org.apache.geaflow.common.type.primitive.IntegerType;
 import org.apache.geaflow.common.utils.ReflectionUtil;
@@ -105,6 +106,21 @@ public VertexCentricCombineFunction getCombineFunction() {
         Assert.assertEquals(3L, ((RuntimeContext) ReflectionUtil.getField(operator, "runtimeContext")).getWindowId());
     }
 
+    @Test
+    public void testBuildInferConfigurationOverride() {
+        Configuration config = new Configuration();
+        config.put(FrameworkConfigKeys.INFER_ENV_USER_TRANSFORM_CLASSNAME, "GlobalTransform");
+        Configuration overridden = DynamicGraphVertexCentricComputeOp.buildInferConfiguration(config,
+            "AlgoTransform");
+        Assert.assertEquals(config.getString(FrameworkConfigKeys.INFER_ENV_USER_TRANSFORM_CLASSNAME),
+            "GlobalTransform");
+        Assert.assertEquals(overridden.getString(FrameworkConfigKeys.INFER_ENV_USER_TRANSFORM_CLASSNAME),
+            "AlgoTransform");
+
+        Configuration unchanged = DynamicGraphVertexCentricComputeOp.buildInferConfiguration(config, null);
+        Assert.assertSame(unchanged, config);
+    }
+
     public class TestRuntimeContext extends AbstractRuntimeContext {
 
         public TestRuntimeContext() {

geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/AlgorithmRuntimeContext.java

@@ -151,6 +151,13 @@ public interface AlgorithmRuntimeContext<K, M> {
      */
     Configuration getConfig();
 
+    /**
+     * Invoke model inference when runtime infer support is enabled.
+     */
+    default <OUT> OUT infer(Object... modelInputs) {
+        throw new UnsupportedOperationException("Inference is not enabled. Set INFER_ENV_ENABLE=true to enable inference.");
+    }
+
     /**
      * Sends a termination vote to the coordinator to signal algorithm completion.
      * This method allows vertices to vote for algorithm termination when they
@@ -160,4 +167,4 @@ public interface AlgorithmRuntimeContext<K, M> {
      * @param voteValue The vote value (typically 1 for termination vote)
      */
     void voteToTerminate(String terminationReason, Object voteValue);
-}
+}

geaflow/geaflow-dsl/geaflow-dsl-common/src/main/java/org/apache/geaflow/dsl/common/algo/DynamicVertexRuntimeContext.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.dsl.common.algo;
+
+import org.apache.geaflow.dsl.common.data.RowVertex;
+
+public interface DynamicVertexRuntimeContext<K, M> extends AlgorithmRuntimeContext<K, M> {
+
+    void setVertexId(K vertexId);
+
+    RowVertex loadVertex();
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java

@@ -39,6 +39,7 @@
 import org.apache.geaflow.dsl.udf.graph.ClusterCoefficient;
 import org.apache.geaflow.dsl.udf.graph.CommonNeighbors;
 import org.apache.geaflow.dsl.udf.graph.ConnectedComponents;
+import org.apache.geaflow.dsl.udf.graph.GCN;
 import org.apache.geaflow.dsl.udf.graph.IncKHopAlgorithm;
 import org.apache.geaflow.dsl.udf.graph.IncMinimumSpanningTree;
 import org.apache.geaflow.dsl.udf.graph.IncWeakConnectedComponents;
@@ -241,6 +242,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
             .add(GeaFlowFunction.of(LabelPropagation.class))
             .add(GeaFlowFunction.of(ConnectedComponents.class))
             .add(GeaFlowFunction.of(Louvain.class))
+            .add(GeaFlowFunction.of(GCN.class))
             .build();
 
     public BuildInSqlFunctionTable(GQLJavaTypeFactory typeFactory) {