refactor: how to add a node type in workflow (#558)

backend/domain/workflow/internal/schema/branch_schema.go

@@ -0,0 +1,196 @@
+/*
+ * Copyright 2025 coze-dev Authors
+ *
+ * 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 schema
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/cloudwego/eino/compose"
+
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
+)
+
+// Port type constants
+const (
+	PortDefault      = "default"
+	PortBranchError  = "branch_error"
+	PortBranchFormat = "branch_%d"
+)
+
+// BranchSchema defines the schema for workflow branches.
+type BranchSchema struct {
+	From             vo.NodeKey                `json:"from_node"`
+	DefaultMapping   map[string]bool           `json:"default_mapping,omitempty"`
+	ExceptionMapping map[string]bool           `json:"exception_mapping,omitempty"`
+	Mappings         map[int64]map[string]bool `json:"mappings,omitempty"`
+}
+
+// BuildBranches builds branch schemas from connections.
+func BuildBranches(connections []*Connection) (map[vo.NodeKey]*BranchSchema, error) {
+	var branchMap map[vo.NodeKey]*BranchSchema
+
+	for _, conn := range connections {
+		if conn.FromPort == nil || len(*conn.FromPort) == 0 {
+			continue
+		}
+
+		port := *conn.FromPort
+		sourceNodeKey := conn.FromNode
+
+		if branchMap == nil {
+			branchMap = map[vo.NodeKey]*BranchSchema{}
+		}
+
+		// Get or create branch schema for source node
+		branch, exists := branchMap[sourceNodeKey]
+		if !exists {
+			branch = &BranchSchema{
+				From: sourceNodeKey,
+			}
+			branchMap[sourceNodeKey] = branch
+		}
+
+		// Classify port type and add to appropriate mapping
+		switch {
+		case port == PortDefault:
+			if branch.DefaultMapping == nil {
+				branch.DefaultMapping = map[string]bool{}
+			}
+			branch.DefaultMapping[string(conn.ToNode)] = true
+		case port == PortBranchError:
+			if branch.ExceptionMapping == nil {
+				branch.ExceptionMapping = map[string]bool{}
+			}
+			branch.ExceptionMapping[string(conn.ToNode)] = true
+		default:
+			var branchNum int64
+			_, err := fmt.Sscanf(port, PortBranchFormat, &branchNum)
+			if err != nil || branchNum < 0 {
+				return nil, fmt.Errorf("invalid port format '%s' for connection %+v", port, conn)
+			}
+			if branch.Mappings == nil {
+				branch.Mappings = map[int64]map[string]bool{}
+			}
+			if _, exists := branch.Mappings[branchNum]; !exists {
+				branch.Mappings[branchNum] = make(map[string]bool)
+			}
+			branch.Mappings[branchNum][string(conn.ToNode)] = true
+		}
+	}
+
+	return branchMap, nil
+}
+
+func (bs *BranchSchema) OnlyException() bool {
+	return len(bs.Mappings) == 0 && len(bs.ExceptionMapping) > 0 && len(bs.DefaultMapping) > 0
+}
+
+func (bs *BranchSchema) GetExceptionBranch() *compose.GraphBranch {
+	condition := func(ctx context.Context, in map[string]any) (map[string]bool, error) {
+		isSuccess, ok := in["isSuccess"]
+		if ok && isSuccess != nil && !isSuccess.(bool) {
+			return bs.ExceptionMapping, nil
+		}
+
+		return bs.DefaultMapping, nil
+	}
+
+	// Combine ExceptionMapping and DefaultMapping into a new map
+	endNodes := make(map[string]bool)
+	for node := range bs.ExceptionMapping {
+		endNodes[node] = true
+	}
+	for node := range bs.DefaultMapping {
+		endNodes[node] = true
+	}
+
+	return compose.NewGraphMultiBranch(condition, endNodes)
+}
+
+func (bs *BranchSchema) GetFullBranch(ctx context.Context, bb BranchBuilder) (*compose.GraphBranch, error) {
+	extractor, hasBranch := bb.BuildBranch(ctx)
+	if !hasBranch {
+		return nil, fmt.Errorf("branch expected but BranchBuilder thinks not. BranchSchema: %v", bs)
+	}
+
+	if len(bs.ExceptionMapping) == 0 { // no exception, it's a normal branch
+		condition := func(ctx context.Context, in map[string]any) (map[string]bool, error) {
+			index, isDefault, err := extractor(ctx, in)
+			if err != nil {
+				return nil, err
+			}
+
+			if isDefault {
+				return bs.DefaultMapping, nil
+			}
+
+			if _, ok := bs.Mappings[index]; !ok {
+				return nil, fmt.Errorf("chosen index= %d, out of range", index)
+			}
+
+			return bs.Mappings[index], nil
+		}
+
+		// Combine DefaultMapping and normal mappings into a new map
+		endNodes := make(map[string]bool)
+		for node := range bs.DefaultMapping {
+			endNodes[node] = true
+		}
+		for _, ms := range bs.Mappings {
+			for node := range ms {
+				endNodes[node] = true
+			}
+		}
+
+		return compose.NewGraphMultiBranch(condition, endNodes), nil
+	}
+
+	condition := func(ctx context.Context, in map[string]any) (map[string]bool, error) {
+		isSuccess, ok := in["isSuccess"]
+		if ok && isSuccess != nil && !isSuccess.(bool) {
+			return bs.ExceptionMapping, nil
+		}
+
+		index, isDefault, err := extractor(ctx, in)
+		if err != nil {
+			return nil, err
+		}
+
+		if isDefault {
+			return bs.DefaultMapping, nil
+		}
+
+		return bs.Mappings[index], nil
+	}
+
+	// Combine ALL mappings into a new map
+	endNodes := make(map[string]bool)
+	for node := range bs.ExceptionMapping {
+		endNodes[node] = true
+	}
+	for node := range bs.DefaultMapping {
+		endNodes[node] = true
+	}
+	for _, ms := range bs.Mappings {
+		for node := range ms {
+			endNodes[node] = true
+		}
+	}
+
+	return compose.NewGraphMultiBranch(condition, endNodes), nil
+}

backend/domain/workflow/internal/schema/node_builder.go

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2025 coze-dev Authors
+ *
+ * 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 schema
+
+import (
+	"context"
+
+	"github.com/cloudwego/eino/compose"
+)
+
+type BuildOptions struct {
+	WS    *WorkflowSchema
+	Inner compose.Runnable[map[string]any, map[string]any]
+}
+
+func GetBuildOptions(opts ...BuildOption) *BuildOptions {
+	bo := &BuildOptions{}
+	for _, o := range opts {
+		o(bo)
+	}
+	return bo
+}
+
+type BuildOption func(options *BuildOptions)
+
+func WithWorkflowSchema(ws *WorkflowSchema) BuildOption {
+	return func(options *BuildOptions) {
+		options.WS = ws
+	}
+}
+
+func WithInnerWorkflow(inner compose.Runnable[map[string]any, map[string]any]) BuildOption {
+	return func(options *BuildOptions) {
+		options.Inner = inner
+	}
+}
+
+// NodeBuilder takes a NodeSchema and several BuildOption to build an executable node instance.
+// The result 'executable' MUST implement at least one of the execute interfaces:
+// - nodes.InvokableNode
+// - nodes.StreamableNode
+// - nodes.CollectableNode
+// - nodes.TransformableNode
+// - nodes.InvokableNodeWOpt
+// - nodes.StreamableNodeWOpt
+// - nodes.CollectableNodeWOpt
+// - nodes.TransformableNodeWOpt
+// NOTE: the 'normal' version does not take NodeOption, while the 'WOpt' versions take NodeOption.
+// NOTE: a node should either implement the 'normal' versions, or the 'WOpt' versions, not mix them up.
+type NodeBuilder interface {
+	Build(ctx context.Context, ns *NodeSchema, opts ...BuildOption) (
+		executable any, err error)
+}
+
+// BranchBuilder builds the extractor function that maps node output to port index.
+type BranchBuilder interface {
+	BuildBranch(ctx context.Context) (extractor func(ctx context.Context,
+		nodeOutput map[string]any) (int64, bool /*if is default branch*/, error), hasBranch bool)
+}

backend/domain/workflow/internal/schema/node_schema.go

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2025 coze-dev Authors
+ *
+ * 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 schema
+
+import (
+	"github.com/cloudwego/eino/compose"
+
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity"
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
+)
+
+// NodeSchema is the universal description and configuration for a workflow Node.
+// It should contain EVERYTHING a node needs to instantiate.
+type NodeSchema struct {
+	// Key is the node key within the Eino graph.
+	// A node may need this information during execution,
+	// e.g.
+	// - using this Key to query workflow State for data belonging to current node.
+	Key vo.NodeKey `json:"key"`
+
+	// Name is the name for this node as specified on Canvas.
+	// A node may show this name on Canvas as part of this node's input/output.
+	Name string `json:"name"`
+
+	// Type is the NodeType for the node.
+	Type entity.NodeType `json:"type"`
+
+	// Configs are node specific configurations, with actual struct type defined by each Node Type.
+	// Will not hold information relating to field mappings, nor as node's static values.
+	// In a word, these Configs are INTERNAL to node's implementation, NOT related to workflow orchestration.
+	// Actual type of these Configs should implement two interfaces:
+	// - NodeAdaptor: to provide conversion from vo.Node to NodeSchema
+	// - NodeBuilder: to provide instantiation from NodeSchema to actual node instance.
+	Configs any `json:"configs,omitempty"`
+
+	// InputTypes are type information about the node's input fields.
+	InputTypes map[string]*vo.TypeInfo `json:"input_types,omitempty"`
+	// InputSources are field mapping information about the node's input fields.
+	InputSources []*vo.FieldInfo `json:"input_sources,omitempty"`
+
+	// OutputTypes are type information about the node's output fields.
+	OutputTypes map[string]*vo.TypeInfo `json:"output_types,omitempty"`
+	// OutputSources are field mapping information about the node's output fields.
+	// NOTE: only applicable to composite nodes such as NodeTypeBatch or NodeTypeLoop.
+	OutputSources []*vo.FieldInfo `json:"output_sources,omitempty"`
+
+	// ExceptionConfigs are about exception handling strategy of the node.
+	ExceptionConfigs *ExceptionConfig `json:"exception_configs,omitempty"`
+	// StreamConfigs are streaming characteristics of the node.
+	StreamConfigs *StreamConfig `json:"stream_configs,omitempty"`
+
+	// SubWorkflowBasic is basic information of the sub workflow if this node is NodeTypeSubWorkflow.
+	SubWorkflowBasic *entity.WorkflowBasic `json:"sub_workflow_basic,omitempty"`
+	// SubWorkflowSchema is WorkflowSchema of the sub workflow if this node is NodeTypeSubWorkflow.
+	SubWorkflowSchema *WorkflowSchema `json:"sub_workflow_schema,omitempty"`
+
+	// FullSources contains more complete information about a node's input fields' mapping sources,
+	// such as whether a field's source is a 'streaming field',
+	// or whether the field is an object that contains sub-fields with real mappings.
+	// Used for those nodes that need to process streaming input.
+	// Set InputSourceAware = true in NodeMeta to enable.
+	FullSources map[string]*SourceInfo
+
+	// Lambda directly sets the node to be an Eino Lambda.
+	// NOTE: not serializable, used ONLY for internal test.
+	Lambda *compose.Lambda
+}
+
+type RequireCheckpoint interface {
+	RequireCheckpoint() bool
+}
+
+type ExceptionConfig struct {
+	TimeoutMS   int64                `json:"timeout_ms,omitempty"`   // timeout in milliseconds, 0 means no timeout
+	MaxRetry    int64                `json:"max_retry,omitempty"`    // max retry times, 0 means no retry
+	ProcessType *vo.ErrorProcessType `json:"process_type,omitempty"` // error process type, 0 means throw error
+	DataOnErr   string               `json:"data_on_err,omitempty"`  // data to return when error, effective when ProcessType==Default occurs
+}
+
+type StreamConfig struct {
+	// whether this node has the ability to produce genuine streaming output.
+	// not include nodes that only passes stream down as they receives them
+	CanGeneratesStream bool `json:"can_generates_stream,omitempty"`
+	// whether this node prioritize streaming input over none-streaming input.
+	// not include nodes that can accept both and does not have preference.
+	RequireStreamingInput bool `json:"can_process_stream,omitempty"`
+}
+
+func (s *NodeSchema) SetConfigKV(key string, value any) {
+	if s.Configs == nil {
+		s.Configs = make(map[string]any)
+	}
+
+	s.Configs.(map[string]any)[key] = value
+}
+
+func (s *NodeSchema) SetInputType(key string, t *vo.TypeInfo) {
+	if s.InputTypes == nil {
+		s.InputTypes = make(map[string]*vo.TypeInfo)
+	}
+	s.InputTypes[key] = t
+}
+
+func (s *NodeSchema) AddInputSource(info ...*vo.FieldInfo) {
+	s.InputSources = append(s.InputSources, info...)
+}
+
+func (s *NodeSchema) SetOutputType(key string, t *vo.TypeInfo) {
+	if s.OutputTypes == nil {
+		s.OutputTypes = make(map[string]*vo.TypeInfo)
+	}
+	s.OutputTypes[key] = t
+}
+
+func (s *NodeSchema) AddOutputSource(info ...*vo.FieldInfo) {
+	s.OutputSources = append(s.OutputSources, info...)
+}

backend/domain/workflow/internal/schema/stream.go

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2025 coze-dev Authors
+ *
+ * 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 schema
+
+import (
+	"github.com/cloudwego/eino/compose"
+
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
+)
+
+type FieldStreamType string
+
+const (
+	FieldIsStream    FieldStreamType = "yes"     // absolutely a stream
+	FieldNotStream   FieldStreamType = "no"      // absolutely not a stream
+	FieldMaybeStream FieldStreamType = "maybe"   // maybe a stream, requires request-time resolution
+	FieldSkipped     FieldStreamType = "skipped" // the field source's node is skipped
+)
+
+type FieldSkipStatus string
+
+// SourceInfo contains stream type for a input field source of a node.
+type SourceInfo struct {
+	// IsIntermediate means this field is itself not a field source, but a map containing one or more field sources.
+	IsIntermediate bool
+	// FieldType the stream type of the field. May require request-time resolution in addition to compile-time.
+	FieldType FieldStreamType
+	// FromNodeKey is the node key that produces this field source. empty if the field is a static value or variable.
+	FromNodeKey vo.NodeKey
+	// FromPath is the path of this field source within the source node. empty if the field is a static value or variable.
+	FromPath compose.FieldPath
+	TypeInfo *vo.TypeInfo
+	// SubSources are SourceInfo for keys within this intermediate Map(Object) field.
+	SubSources map[string]*SourceInfo
+}
+
+func (s *SourceInfo) Skipped() bool {
+	if !s.IsIntermediate {
+		return s.FieldType == FieldSkipped
+	}
+
+	for _, sub := range s.SubSources {
+		if !sub.Skipped() {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (s *SourceInfo) FromNode(nodeKey vo.NodeKey) bool {
+	if !s.IsIntermediate {
+		return s.FromNodeKey == nodeKey
+	}
+
+	for _, sub := range s.SubSources {
+		if sub.FromNode(nodeKey) {
+			return true
+		}
+	}
+
+	return false
+}

backend/domain/workflow/internal/schema/workflow_schema.go

@@ -0,0 +1,350 @@
+/*
+ * Copyright 2025 coze-dev Authors
+ *
+ * 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 schema
+
+import (
+	"fmt"
+	"maps"
+	"reflect"
+	"sync"
+
+	"github.com/cloudwego/eino/compose"
+
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity"
+	"github.com/coze-dev/coze-studio/backend/domain/workflow/entity/vo"
+)
+
+type WorkflowSchema struct {
+	Nodes       []*NodeSchema                `json:"nodes"`
+	Connections []*Connection                `json:"connections"`
+	Hierarchy   map[vo.NodeKey]vo.NodeKey    `json:"hierarchy,omitempty"` // child node key-> parent node key
+	Branches    map[vo.NodeKey]*BranchSchema `json:"branches,omitempty"`
+
+	GeneratedNodes []vo.NodeKey `json:"generated_nodes,omitempty"` // generated nodes for the nodes in batch mode
+
+	nodeMap           map[vo.NodeKey]*NodeSchema // won't serialize this
+	compositeNodes    []*CompositeNode           // won't serialize this
+	requireCheckPoint bool                       // won't serialize this
+	requireStreaming  bool
+
+	once sync.Once
+}
+
+type Connection struct {
+	FromNode vo.NodeKey `json:"from_node"`
+	ToNode   vo.NodeKey `json:"to_node"`
+	FromPort *string    `json:"from_port,omitempty"`
+}
+
+func (c *Connection) ID() string {
+	if c.FromPort != nil {
+		return fmt.Sprintf("%s:%s:%v", c.FromNode, c.ToNode, *c.FromPort)
+	}
+	return fmt.Sprintf("%v:%v", c.FromNode, c.ToNode)
+}
+
+type CompositeNode struct {
+	Parent   *NodeSchema
+	Children []*NodeSchema
+}
+
+func (w *WorkflowSchema) Init() {
+	w.once.Do(func() {
+		w.nodeMap = make(map[vo.NodeKey]*NodeSchema)
+		for _, node := range w.Nodes {
+			w.nodeMap[node.Key] = node
+		}
+
+		w.doGetCompositeNodes()
+
+		for _, node := range w.Nodes {
+			if node.Type == entity.NodeTypeSubWorkflow {
+				node.SubWorkflowSchema.Init()
+				if node.SubWorkflowSchema.requireCheckPoint {
+					w.requireCheckPoint = true
+					break
+				}
+			}
+
+			if rc, ok := node.Configs.(RequireCheckpoint); ok {
+				if rc.RequireCheckpoint() {
+					w.requireCheckPoint = true
+					break
+				}
+			}
+		}
+
+		w.requireStreaming = w.doRequireStreaming()
+	})
+}
+
+func (w *WorkflowSchema) GetNode(key vo.NodeKey) *NodeSchema {
+	return w.nodeMap[key]
+}
+
+func (w *WorkflowSchema) GetAllNodes() map[vo.NodeKey]*NodeSchema {
+	return w.nodeMap // TODO: needs to calculate node count separately, considering batch mode nodes
+}
+
+func (w *WorkflowSchema) GetCompositeNodes() []*CompositeNode {
+	if w.compositeNodes == nil {
+		w.compositeNodes = w.doGetCompositeNodes()
+	}
+
+	return w.compositeNodes
+}
+
+func (w *WorkflowSchema) GetBranch(key vo.NodeKey) *BranchSchema {
+	if w.Branches == nil {
+		return nil
+	}
+
+	return w.Branches[key]
+}
+
+func (w *WorkflowSchema) RequireCheckpoint() bool {
+	return w.requireCheckPoint
+}
+
+func (w *WorkflowSchema) RequireStreaming() bool {
+	return w.requireStreaming
+}
+
+func (w *WorkflowSchema) doGetCompositeNodes() (cNodes []*CompositeNode) {
+	if w.Hierarchy == nil {
+		return nil
+	}
+
+	// Build parent to children mapping
+	parentToChildren := make(map[vo.NodeKey][]*NodeSchema)
+	for childKey, parentKey := range w.Hierarchy {
+		if parentSchema := w.nodeMap[parentKey]; parentSchema != nil {
+			if childSchema := w.nodeMap[childKey]; childSchema != nil {
+				parentToChildren[parentKey] = append(parentToChildren[parentKey], childSchema)
+			}
+		}
+	}
+
+	// Create composite nodes
+	for parentKey, children := range parentToChildren {
+		if parentSchema := w.nodeMap[parentKey]; parentSchema != nil {
+			cNodes = append(cNodes, &CompositeNode{
+				Parent:   parentSchema,
+				Children: children,
+			})
+		}
+	}
+
+	return cNodes
+}
+
+func IsInSameWorkflow(n map[vo.NodeKey]vo.NodeKey, nodeKey, otherNodeKey vo.NodeKey) bool {
+	if n == nil {
+		return true
+	}
+
+	myParents, myParentExists := n[nodeKey]
+	theirParents, theirParentExists := n[otherNodeKey]
+
+	if !myParentExists && !theirParentExists {
+		return true
+	}
+
+	if !myParentExists || !theirParentExists {
+		return false
+	}
+
+	return myParents == theirParents
+}
+
+func IsBelowOneLevel(n map[vo.NodeKey]vo.NodeKey, nodeKey, otherNodeKey vo.NodeKey) bool {
+	if n == nil {
+		return false
+	}
+	_, myParentExists := n[nodeKey]
+	_, theirParentExists := n[otherNodeKey]
+
+	return myParentExists && !theirParentExists
+}
+
+func IsParentOf(n map[vo.NodeKey]vo.NodeKey, nodeKey, otherNodeKey vo.NodeKey) bool {
+	if n == nil {
+		return false
+	}
+	theirParent, theirParentExists := n[otherNodeKey]
+
+	return theirParentExists && theirParent == nodeKey
+}
+
+func (w *WorkflowSchema) IsEqual(other *WorkflowSchema) bool {
+	otherConnectionsMap := make(map[string]bool, len(other.Connections))
+	for _, connection := range other.Connections {
+		otherConnectionsMap[connection.ID()] = true
+	}
+	connectionsMap := make(map[string]bool, len(other.Connections))
+	for _, connection := range w.Connections {
+		connectionsMap[connection.ID()] = true
+	}
+	if !maps.Equal(otherConnectionsMap, connectionsMap) {
+		return false
+	}
+	otherNodeMap := make(map[vo.NodeKey]*NodeSchema, len(other.Nodes))
+	for _, node := range other.Nodes {
+		otherNodeMap[node.Key] = node
+	}
+	nodeMap := make(map[vo.NodeKey]*NodeSchema, len(w.Nodes))
+
+	for _, node := range w.Nodes {
+		nodeMap[node.Key] = node
+	}
+
+	if !maps.EqualFunc(otherNodeMap, nodeMap, func(node *NodeSchema, other *NodeSchema) bool {
+		if node.Name != other.Name {
+			return false
+		}
+		if !reflect.DeepEqual(node.Configs, other.Configs) {
+			return false
+		}
+		if !reflect.DeepEqual(node.InputTypes, other.InputTypes) {
+			return false
+		}
+		if !reflect.DeepEqual(node.InputSources, other.InputSources) {
+			return false
+		}
+
+		if !reflect.DeepEqual(node.OutputTypes, other.OutputTypes) {
+			return false
+		}
+		if !reflect.DeepEqual(node.OutputSources, other.OutputSources) {
+			return false
+		}
+		if !reflect.DeepEqual(node.ExceptionConfigs, other.ExceptionConfigs) {
+			return false
+		}
+		if !reflect.DeepEqual(node.SubWorkflowBasic, other.SubWorkflowBasic) {
+			return false
+		}
+		return true
+
+	}) {
+		return false
+	}
+
+	return true
+
+}
+
+func (w *WorkflowSchema) NodeCount() int32 {
+	return int32(len(w.Nodes) - len(w.GeneratedNodes))
+}
+
+func (w *WorkflowSchema) doRequireStreaming() bool {
+	producers := make(map[vo.NodeKey]bool)
+	consumers := make(map[vo.NodeKey]bool)
+
+	for _, node := range w.Nodes {
+		if node.StreamConfigs != nil && node.StreamConfigs.CanGeneratesStream {
+			producers[node.Key] = true
+		}
+
+		if node.StreamConfigs != nil && node.StreamConfigs.RequireStreamingInput {
+			consumers[node.Key] = true
+		}
+
+	}
+
+	if len(producers) == 0 || len(consumers) == 0 {
+		return false
+	}
+
+	// Build data-flow graph from InputSources
+	adj := make(map[vo.NodeKey]map[vo.NodeKey]struct{})
+	for _, node := range w.Nodes {
+		for _, source := range node.InputSources {
+			if source.Source.Ref != nil && len(source.Source.Ref.FromNodeKey) > 0 {
+				if _, ok := adj[source.Source.Ref.FromNodeKey]; !ok {
+					adj[source.Source.Ref.FromNodeKey] = make(map[vo.NodeKey]struct{})
+				}
+				adj[source.Source.Ref.FromNodeKey][node.Key] = struct{}{}
+			}
+		}
+	}
+
+	// For each producer, traverse the graph to see if it can reach a consumer
+	for p := range producers {
+		q := []vo.NodeKey{p}
+		visited := make(map[vo.NodeKey]bool)
+		visited[p] = true
+
+		for len(q) > 0 {
+			curr := q[0]
+			q = q[1:]
+
+			if consumers[curr] {
+				return true
+			}
+
+			for neighbor := range adj[curr] {
+				if !visited[neighbor] {
+					visited[neighbor] = true
+					q = append(q, neighbor)
+				}
+			}
+		}
+	}
+
+	return false
+}
+
+func (w *WorkflowSchema) FanInMergeConfigs() map[string]compose.FanInMergeConfig {
+	// what we need to do is to see if the workflow requires streaming, if not, then no fan-in merge configs needed
+	// then we find those nodes that have 'transform' or 'collect' as streaming paradigm,
+	// and see if each of those nodes has multiple data predecessors, if so, it's a fan-in node.
+	// then, look up the NodeTypeMeta's ExecutableMeta info and see if it requires fan-in stream merge.
+	if !w.requireStreaming {
+		return nil
+	}
+
+	fanInNodes := make(map[vo.NodeKey]bool)
+	for _, node := range w.Nodes {
+		if node.StreamConfigs != nil && node.StreamConfigs.RequireStreamingInput {
+			var predecessor *vo.NodeKey
+			for _, source := range node.InputSources {
+				if source.Source.Ref != nil && len(source.Source.Ref.FromNodeKey) > 0 {
+					if predecessor != nil {
+						fanInNodes[node.Key] = true
+						break
+					}
+					predecessor = &source.Source.Ref.FromNodeKey
+				}
+			}
+		}
+	}
+
+	fanInConfigs := make(map[string]compose.FanInMergeConfig)
+	for nodeKey := range fanInNodes {
+		if m := entity.NodeMetaByNodeType(w.GetNode(nodeKey).Type); m != nil {
+			if m.StreamSourceEOFAware {
+				fanInConfigs[string(nodeKey)] = compose.FanInMergeConfig{
+					StreamMergeWithSourceEOF: true,
+				}
+			}
+		}
+	}
+
+	return fanInConfigs
+}