[GitHub] [inlong] healchow commented on a diff in pull request #5106: [INLONG-5103][Manager] Add constant field support for stream source and transform node

[GitBox <gi...@apache.org>]

healchow commented on code in PR #5106:
URL: https://github.com/apache/inlong/pull/5106#discussion_r923408156


##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/util/ExtractNodeUtils.java:
##########
@@ -431,4 +385,16 @@ public static TubeMQExtractNode createExtractNode(TubeMQSource source) {
         );
     }
 
+    private static List<FieldInfo> parseFieldInfos(List<StreamField> streamFields, String nodeId) {

Review Comment:
   Please add Java docs for static methods, thanks.



##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sort/DefaultSortConfigOperator.java:
##########
@@ -130,44 +130,30 @@ private GroupInfo getLightweightGroupInfo(InlongGroupInfo groupInfo, List<Inlong
     }
 
     /**
-     * Get Sort GroupInfo of normal inlong group.
+     * Get constant field from stream fields
      *
-     * @see org.apache.inlong.sort.protocol.GroupInfo
+     * @param nodeId The node id
+     * @param fields The stream fields
+     * @param constantFieldMap The constant field map
      */
-    private GroupInfo getNormalGroupInfo(InlongGroupInfo groupInfo, List<InlongStreamInfo> streamInfoList) {
-        // get source info
-        Map<String, List<StreamSource>> sourceMap = sourceService.getSourcesMap(groupInfo, streamInfoList);
-        // get sink info
-        Map<String, List<StreamSink>> sinkMap = sinkService.getSinksMap(groupInfo, streamInfoList);
-
-        // create StreamInfo for Sort protocol
-        List<StreamInfo> sortStreamInfos = new ArrayList<>();
-        for (InlongStreamInfo inlongStream : streamInfoList) {
-            String streamId = inlongStream.getInlongStreamId();
-            List<StreamSource> sources = sourceMap.get(streamId);
-            List<StreamSink> sinks = sinkMap.get(streamId);
-            StreamInfo sortStream = new StreamInfo(streamId,
-                    this.createNodesForStream(sources, sinks),
-                    this.createNodeRelationsForStream(sources, sinks));
-            sortStreamInfos.add(sortStream);
+    private void parseConstantFieldMap(String nodeId, List<StreamField> fields,
+            Map<String, StreamField> constantFieldMap) {
+        if (fields != null && !fields.isEmpty()) {

Review Comment:
   Maybe can refactor like this:
   
   ```java
   if (Collections.isEmpty(fields)) {
       return;
   }
   
   // loop directly
   for (StreamField field : fields) {
       ...
   }
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org