19619258155

Committed 23 Nov 2025 11:57PM UTC coverage: 91.472% (-0.7%) from 92.204%

Build # 19619258155

Build Type

push

github

Committed by

web-flow

Commit Message

AMQP 1.0 (#91)

Completely rewrote message broker comm layer, removed support for RabbitMQ AMQP 0.9.1, added support for AMQP 1.0 - tested with RabbitMQ 4.2.0, AciveMQ Classic 6.1.8, ActiveMQ Artemis 2.44.0.

Fixed a few bugs in/around message handler.

Improvements in TF deployment.

Removed k8s POC and Amazon Keyspaces deployment, they are now in https://github.com/capillariesio/capillaries-experimental.

Added experimental CapiMQ message broker.

Coverage Stats

242 of 312 new or added lines in 5 files covered. (77.56%)

5610 of 6133 relevant lines covered (91.47%)

1.09 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.57

/pkg/sc/script_node_def.go

package sc

import (
        "encoding/json"
        "errors"
        "fmt"
        "go/ast"
        "math"
        "regexp"
        "strings"

        "github.com/capillariesio/capillaries/pkg/eval"
)

const (
        HandlerExeTypeGeneric  string = "capi_daemon"
        HandlerExeTypeToolbelt string = "capi_toolbelt"
        HandlerExeTypeWebapi   string = "capi_webapi"
)

const MaxAcceptedBatchesByTableReader int = 1000000
const DefaultRowsetSize int = 1000 // 1000 seems to work on c7g.large without OOM, careful with using bigger values
const MaxRowsetSize int = 100000

type AggFinderVisitor struct {
        Error error
}

func (v *AggFinderVisitor) Visit(node ast.Node) ast.Visitor {
        switch callExp := node.(type) {
        case *ast.CallExpr:
                switch callIdentExp := callExp.Fun.(type) {
                case *ast.Ident:
                        if eval.StringToAggFunc(callIdentExp.Name) != eval.AggUnknown {
                                v.Error = fmt.Errorf("found aggregate function %s()", callIdentExp.Name)
                                return nil
                        }
                        return v
                default:
                        return v
                }
        default:
                return v
        }
}

type NodeType string

const (
        NodeTypeNone                NodeType = "none"
        NodeTypeFileTable           NodeType = "file_table"
        NodeTypeTableTable          NodeType = "table_table"
        NodeTypeTableLookupTable    NodeType = "table_lookup_table"
        NodeTypeTableFile           NodeType = "table_file"
        NodeTypeTableCustomTfmTable NodeType = "table_custom_tfm_table"
        NodeTypeDistinctTable       NodeType = "distinct_table"
)

func ValidateNodeType(nodeType NodeType) error {
        if nodeType == NodeTypeFileTable ||
                nodeType == NodeTypeTableTable ||
                nodeType == NodeTypeTableLookupTable ||
                nodeType == NodeTypeTableFile ||
                nodeType == NodeTypeDistinctTable ||
                nodeType == NodeTypeTableCustomTfmTable {
                return nil
        }
        return fmt.Errorf("invalid node type %s", nodeType)
}

const ReaderAlias string = "r"
const CreatorAlias string = "w"
const LookupAlias string = "l"
const CustomProcessorAlias string = "p"

type NodeRerunPolicy string

const (
        NodeRerun NodeRerunPolicy = "rerun" // Default
        NodeFail  NodeRerunPolicy = "fail"
)

func ValidateRerunPolicy(rerunPolicy NodeRerunPolicy) error {
        if rerunPolicy == NodeRerun ||
                rerunPolicy == NodeFail {
                return nil
        }
        return fmt.Errorf("invalid node rerun policy %s", rerunPolicy)
}

type NodeStartPolicy string

const (
        NodeStartManual NodeStartPolicy = "manual"
        NodeStartAuto   NodeStartPolicy = "auto" // Default
)

func ValidateStartPolicy(startPolicy NodeStartPolicy) error {
        if startPolicy == NodeStartManual ||
                startPolicy == NodeStartAuto {
                return nil
        }
        return fmt.Errorf("invalid node start policy %s", startPolicy)
}

type ScriptNodeDef struct {
        Name                   string          // Get it from the key
        Type                   NodeType        `json:"type" yaml:"type"`
        Desc                   string          `json:"desc" yaml:"desc"`
        StartPolicy            NodeStartPolicy `json:"start_policy" yaml:"start_policy"`
        RerunPolicy            NodeRerunPolicy `json:"rerun_policy,omitempty" yaml:"rerun_policy,omitempty"`
        CustomProcessorType    string          `json:"custom_proc_type,omitempty" yaml:"custom_proc_type,omitempty"`
        HandlerExeType         string          `json:"handler_exe_type,omitempty" yaml:"handler_exe_type,omitempty"`
        MaxBatchProcessingTime int             `json:"max_batch_processing_time,omitempty" yaml:"max_batch_processing_time,omitempty"`

        RawReader   json.RawMessage `json:"r" yaml:"r"` // This depends on tfm type
        TableReader TableReaderDef
        FileReader  FileReaderDef

        Lookup LookupDef `json:"l" yaml:"l"`

        RawProcessorDef json.RawMessage    `json:"p" yaml:"p"` // This depends on tfm type
        CustomProcessor CustomProcessorDef // Also should implement CustomProcessorRunner

        RawWriter            json.RawMessage `json:"w" yaml:"w"` // This depends on tfm type
        DependencyPolicyName string          `json:"dependency_policy" yaml:"dependency_policy"`
        TableCreator         TableCreatorDef
        TableUpdater         TableUpdaterDef
        FileCreator          FileCreatorDef
        DepPolDef            *DependencyPolicyDef
}

func (node *ScriptNodeDef) HasTableReader() bool {
        return node.Type == NodeTypeTableTable ||
                node.Type == NodeTypeTableLookupTable ||
                node.Type == NodeTypeTableFile ||
                node.Type == NodeTypeDistinctTable ||
                node.Type == NodeTypeTableCustomTfmTable
}
func (node *ScriptNodeDef) HasFileReader() bool {
        return node.Type == NodeTypeFileTable
}

func (node *ScriptNodeDef) HasLookup() bool {
        return node.Type == NodeTypeTableLookupTable
}

func (node *ScriptNodeDef) HasCustomProcessor() bool {
        return node.Type == NodeTypeTableCustomTfmTable
}

func (node *ScriptNodeDef) HasTableCreator() bool {
        return node.Type == NodeTypeFileTable ||
                node.Type == NodeTypeTableTable ||
                node.Type == NodeTypeDistinctTable ||
                node.Type == NodeTypeTableLookupTable ||
                node.Type == NodeTypeTableCustomTfmTable
}
func (node *ScriptNodeDef) HasFileCreator() bool {
        return node.Type == NodeTypeTableFile
}
func (node *ScriptNodeDef) GetTargetName() string {
        if node.HasTableCreator() {
                return node.TableCreator.Name
        } else if node.HasFileCreator() {
                return CreatorAlias
        }
        return "dev_error_uknown_target_name"
}

func (node *ScriptNodeDef) initReader() error {
        if node.HasTableReader() {
                if err := json.Unmarshal(node.RawReader, &node.TableReader); err != nil {
                        return fmt.Errorf("cannot unmarshal table reader: [%s]", err.Error())
                }
                foundErrors := make([]string, 0)
                if len(node.TableReader.TableName) == 0 {
                        foundErrors = append(foundErrors, "table reader cannot reference empty table name")
                }
                if node.TableReader.ExpectedBatchesTotal == 0 {
                        node.TableReader.ExpectedBatchesTotal = 1
                } else if node.TableReader.ExpectedBatchesTotal < 0 || node.TableReader.ExpectedBatchesTotal > MaxAcceptedBatchesByTableReader {
                        foundErrors = append(foundErrors, fmt.Sprintf("table reader can accept between 1 and %d batches, %d specified", MaxAcceptedBatchesByTableReader, node.TableReader.ExpectedBatchesTotal))
                }
                if node.TableReader.RowsetSize < 0 || MaxRowsetSize < node.TableReader.RowsetSize {
                        foundErrors = append(foundErrors, fmt.Sprintf("invalid rowset size %d, table reader can accept between 0 (defaults to %d) and %d", node.TableReader.RowsetSize, DefaultRowsetSize, MaxRowsetSize))
                }
                if node.TableReader.RowsetSize == 0 {
                        node.TableReader.RowsetSize = DefaultRowsetSize
                }
                if len(foundErrors) > 0 {
                        return fmt.Errorf("%s", strings.Join(foundErrors, "; "))
                }
        } else if node.HasFileReader() {
                if err := node.FileReader.Deserialize(node.RawReader); err != nil {
                        return fmt.Errorf("cannot deserialize file reader [%s]: [%s]", string(node.RawReader), err.Error())
                }
        }

        return nil
}

func (node *ScriptNodeDef) initCreator() error {
        if node.HasTableCreator() {
                if err := node.TableCreator.Deserialize(node.RawWriter); err != nil {
                        return fmt.Errorf("cannot deserialize table creator [%s]: [%s]", strings.ReplaceAll(string(node.RawWriter), "\n", " "), err.Error())
                }
        } else if node.HasFileCreator() {
                if err := node.FileCreator.Deserialize(node.RawWriter); err != nil {
                        return fmt.Errorf("cannot deserialize file creator [%s]: [%s]", strings.ReplaceAll(string(node.RawWriter), "\n", " "), err.Error())
                }
        }
        return nil
}

func (node *ScriptNodeDef) initCustomProcessor(customProcessorDefFactory CustomProcessorDefFactory, customProcessorsSettings map[string]json.RawMessage, scriptType ScriptType, caPath string, privateKeys map[string]string) error {
        if node.HasCustomProcessor() {
                if customProcessorDefFactory == nil {
                        return errors.New("undefined custom processor factory")
                }
                if customProcessorsSettings == nil {
                        return errors.New("missing custom processor settings section")
                }
                var ok bool
                node.CustomProcessor, ok = customProcessorDefFactory.Create(node.CustomProcessorType)
                if !ok {
                        return fmt.Errorf("cannot deserialize unknown custom processor %s", node.CustomProcessorType)
                }
                customProcSettings, ok := customProcessorsSettings[node.CustomProcessorType]
                if !ok {
                        return fmt.Errorf("cannot find custom processing settings for [%s] in the environment config file", node.CustomProcessorType)
                }
                if err := node.CustomProcessor.Deserialize(node.RawProcessorDef, customProcSettings, scriptType, caPath, privateKeys); err != nil {
                        re := regexp.MustCompile("[ \r\n]+")
                        return fmt.Errorf("cannot deserialize custom processor [%s]: [%s]", re.ReplaceAllString(string(node.RawProcessorDef), ""), err.Error())
                }
        }
        return nil
}

func (node *ScriptNodeDef) Deserialize(customProcessorDefFactory CustomProcessorDefFactory, customProcessorsSettings map[string]json.RawMessage, scriptType ScriptType, caPath string, privateKeys map[string]string) error {
        foundErrors := make([]string, 0)

        if err := ValidateNodeType(node.Type); err != nil {
                return err
        }

        // Defaults

        if node.MaxBatchProcessingTime == 0 {
                node.MaxBatchProcessingTime = 60000 // 1 min
        }

        if len(node.HandlerExeType) == 0 {
                node.HandlerExeType = HandlerExeTypeGeneric
        }

        if len(node.RerunPolicy) == 0 {
                node.RerunPolicy = NodeRerun
        } else if err := ValidateRerunPolicy(node.RerunPolicy); err != nil {
                return err
        }

        if len(node.StartPolicy) == 0 {
                node.StartPolicy = NodeStartAuto
        } else if err := ValidateStartPolicy(node.StartPolicy); err != nil {
                return err
        }

        // Reader
        if err := node.initReader(); err != nil {
                foundErrors = append(foundErrors, err.Error())
        }

        // Creator
        if err := node.initCreator(); err != nil {
                foundErrors = append(foundErrors, err.Error())
        }

        // Custom processor
        if err := node.initCustomProcessor(customProcessorDefFactory, customProcessorsSettings, scriptType, caPath, privateKeys); err != nil {
                foundErrors = append(foundErrors, err.Error())
        }

        // Distinct table
        if node.Type == NodeTypeDistinctTable {
                if node.RerunPolicy != NodeFail {
                        foundErrors = append(foundErrors, "distinct_table node must have fail policy, no reruns possible")
                }
                if _, _, err := node.TableCreator.GetSingleUniqueIndexDef(); err != nil {
                        foundErrors = append(foundErrors, err.Error())
                }
        }

        if len(foundErrors) > 0 {
                return fmt.Errorf("%s", strings.Join(foundErrors, "; "))
        }

        return nil
}

func (node *ScriptNodeDef) evalCreatorAndLookupExpressionsAndCheckType() error {
        foundErrors := make([]string, 0, 2)

        if node.HasLookup() && node.Lookup.UsesFilter() {
                if err := evalExpressionWithFieldRefsAndCheckType(node.Lookup.Filter, node.Lookup.UsedInFilterFields, FieldTypeBool); err != nil {
                        foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate lookup filter expression [%s]: [%s]", node.Lookup.RawFilter, err.Error()))
                }
        }

        if node.HasTableCreator() {
                // Having
                if err := evalExpressionWithFieldRefsAndCheckType(node.TableCreator.Having, node.TableCreator.UsedInHavingFields, FieldTypeBool); err != nil {
                        foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate table creator 'having' expression [%s]: [%s]", node.TableCreator.RawHaving, err.Error()))
                }

                // Target table fields
                for tgtFieldName, tgtFieldDef := range node.TableCreator.Fields {

                        // TODO: find a way to check field usage:
                        // - lookup fields must be used only within enclosing agg calls (sum etc), otherwise last one wins
                        // - src table fields are allowed within enclosing agg calls, and there is even a biz case for it (multiply src field by the number of lookup rows)

                        // If no grouping is used, no agg calls allowed
                        if node.HasLookup() && !node.Lookup.IsGroup || !node.HasLookup() {
                                v := AggFinderVisitor{}
                                if tgtFieldDef.ParsedExpression == nil {
                                        foundErrors = append(foundErrors, fmt.Sprintf("cannot parse node %s, target field %s expression [%s] was not parsed successfully", node.Name, tgtFieldName, tgtFieldDef.RawExpression))
                                } else {
                                        ast.Walk(&v, tgtFieldDef.ParsedExpression)
                                        if v.Error != nil {
                                                foundErrors = append(foundErrors, fmt.Sprintf("cannot use agg functions in [%s], lookup group flag is not set or no lookups used: [%s]", tgtFieldDef.RawExpression, v.Error.Error()))
                                        }
                                }
                        }

                        // Just eval with test values, agg functions will go through preserving the type no problem
                        if err := evalExpressionWithFieldRefsAndCheckType(tgtFieldDef.ParsedExpression, node.TableCreator.UsedInTargetExpressionsFields, tgtFieldDef.Type); err != nil {
                                foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate table creator target field %s expression [%s]: [%s]", tgtFieldName, tgtFieldDef.RawExpression, err.Error()))
                        }
                }
        }

        if node.HasFileCreator() {
                // Having
                if err := evalExpressionWithFieldRefsAndCheckType(node.FileCreator.Having, node.FileCreator.UsedInHavingFields, FieldTypeBool); err != nil {
                        foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate file creator 'having' expression [%s]: [%s]", node.FileCreator.RawHaving, err.Error()))
                }

                // Target table fields (yes, they are not just strings, we check the type)
                for i := 0; i < len(node.FileCreator.Columns); i++ {
                        colDef := &node.FileCreator.Columns[i]
                        if err := evalExpressionWithFieldRefsAndCheckType(colDef.ParsedExpression, node.FileCreator.UsedInTargetExpressionsFields, colDef.Type); err != nil {
                                foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate file creator target field %s expression [%s]: [%s]", colDef.Name, colDef.RawExpression, err.Error()))
                        }
                }
        }

        // NOTE: do not even try to eval expressions from the custom processor here,
        // they may contain custom stuff and are pretty much guaranteed to fail

        if len(foundErrors) > 0 {
                return fmt.Errorf("%s", strings.Join(foundErrors, "; "))
        }

        return nil
}

func (node *ScriptNodeDef) getSourceFieldRefs() (*FieldRefs, error) {
        if node.HasFileReader() {
                return node.FileReader.getFieldRefs(), nil
        } else if node.HasTableReader() {
                return node.TableReader.TableCreator.GetFieldRefsWithAlias(ReaderAlias), nil
        }

        return nil, fmt.Errorf("dev error, node of type %s has no file or table reader", node.Type)
}

func (node *ScriptNodeDef) GetUniqueIndexesFieldRefs() *FieldRefs {
        if !node.HasTableCreator() {
                return &FieldRefs{}
        }
        fieldTypeMap := map[string]TableFieldType{}
        for _, idxDef := range node.TableCreator.Indexes {
                if idxDef.Uniqueness == IdxUnique {
                        for _, idxComponentDef := range idxDef.Components {
                                fieldTypeMap[idxComponentDef.FieldName] = idxComponentDef.FieldType
                        }
                }
        }
        fieldRefs := make(FieldRefs, len(fieldTypeMap))
        fieldRefIdx := 0
        for fieldName, fieldType := range fieldTypeMap {
                fieldRefs[fieldRefIdx] = FieldRef{
                        FieldName: fieldName,
                        FieldType: fieldType,
                        TableName: node.TableCreator.Name}
                fieldRefIdx++
        }

        return &fieldRefs
}

func (node *ScriptNodeDef) GetTokenIntervalsByNumberOfBatches() ([][]int64, error) {
        if node.HasTableReader() || node.HasFileCreator() && node.TableReader.ExpectedBatchesTotal > 1 {
                if node.TableReader.ExpectedBatchesTotal == 1 {
                        return [][]int64{{int64(math.MinInt64), int64(math.MaxInt64)}}, nil
                }

                tokenIntervalPerBatch := int64(math.MaxInt64/node.TableReader.ExpectedBatchesTotal) - int64(math.MinInt64/node.TableReader.ExpectedBatchesTotal)

                intervals := make([][]int64, node.TableReader.ExpectedBatchesTotal)
                left := int64(math.MinInt64)
                for i := 0; i < len(intervals); i++ {
                        var right int64
                        if i == len(intervals)-1 {
                                right = math.MaxInt64
                        } else {
                                right = left + tokenIntervalPerBatch - 1
                        }
                        intervals[i] = []int64{left, right}
                        left = right + 1
                }
                return intervals, nil
                // } else if node.HasFileCreator() && node.TableReader.ExpectedBatchesTotal == 1 {
                //         // One output file - one batch, dummy intervals
                //         intervals := make([][]int64, 1)
                //         intervals[0] = []int64{int64(0), 0}
                //         return intervals, nil
        } else if node.HasFileReader() {
                // One input file - one batch
                intervals := make([][]int64, len(node.FileReader.SrcFileUrls))
                for i := 0; i < len(node.FileReader.SrcFileUrls); i++ {
                        intervals[i] = []int64{int64(i), int64(i)}
                }
                return intervals, nil
        }

        return nil, fmt.Errorf("cannot find implementation for intervals for node %s", node.Name)
}

func (node *ScriptNodeDef) isNodeUsesIdx(idxName string) bool {
        if node.HasLookup() && node.Lookup.IndexName == idxName {
                return true
        }

        distinctIdxCandidate, ok := node.TableCreator.Indexes[idxName]
        if ok {
                if node.Type == NodeTypeDistinctTable && distinctIdxCandidate.Uniqueness == IdxUnique {
                        return true
                }
        }

        return false
}

1	package sc
2
3	import (
4	"encoding/json"
5	"errors"
6	"fmt"
7	"go/ast"
8	"math"
9	"regexp"
10	"strings"
11
12	"github.com/capillariesio/capillaries/pkg/eval"
13	)
14
15	const (
16	HandlerExeTypeGeneric string = "capi_daemon"
17	HandlerExeTypeToolbelt string = "capi_toolbelt"
18	HandlerExeTypeWebapi string = "capi_webapi"
19	)
20
21	const MaxAcceptedBatchesByTableReader int = 1000000
22	const DefaultRowsetSize int = 1000 // 1000 seems to work on c7g.large without OOM, careful with using bigger values
23	const MaxRowsetSize int = 100000
24
25	type AggFinderVisitor struct {
26	Error error
27	}
28
29	func (v *AggFinderVisitor) Visit(node ast.Node) ast.Visitor {	1✔
30	switch callExp := node.(type) {	1✔
31	case *ast.CallExpr:	1✔
32	switch callIdentExp := callExp.Fun.(type) {	1✔
33	case *ast.Ident:	1✔
34	if eval.StringToAggFunc(callIdentExp.Name) != eval.AggUnknown {	2✔
35	v.Error = fmt.Errorf("found aggregate function %s()", callIdentExp.Name)	1✔
36	return nil	1✔
37	}	1✔
38	return v	×
39	default:	×
40	return v	×
41	}
42	default:	1✔
43	return v	1✔
44	}
45	}
46
47	type NodeType string
48
49	const (
50	NodeTypeNone NodeType = "none"
51	NodeTypeFileTable NodeType = "file_table"
52	NodeTypeTableTable NodeType = "table_table"
53	NodeTypeTableLookupTable NodeType = "table_lookup_table"
54	NodeTypeTableFile NodeType = "table_file"
55	NodeTypeTableCustomTfmTable NodeType = "table_custom_tfm_table"
56	NodeTypeDistinctTable NodeType = "distinct_table"
57	)
58
59	func ValidateNodeType(nodeType NodeType) error {	1✔
60	if nodeType == NodeTypeFileTable \|\|	1✔
61	nodeType == NodeTypeTableTable \|\|	1✔
62	nodeType == NodeTypeTableLookupTable \|\|	1✔
63	nodeType == NodeTypeTableFile \|\|	1✔
64	nodeType == NodeTypeDistinctTable \|\|	1✔
65	nodeType == NodeTypeTableCustomTfmTable {	2✔
66	return nil	1✔
67	}	1✔
68	return fmt.Errorf("invalid node type %s", nodeType)	×
69	}
70
71	const ReaderAlias string = "r"
72	const CreatorAlias string = "w"
73	const LookupAlias string = "l"
74	const CustomProcessorAlias string = "p"
75
76	type NodeRerunPolicy string
77
78	const (
79	NodeRerun NodeRerunPolicy = "rerun" // Default
80	NodeFail NodeRerunPolicy = "fail"
81	)
82
83	func ValidateRerunPolicy(rerunPolicy NodeRerunPolicy) error {	1✔
84	if rerunPolicy == NodeRerun \|\|	1✔
85	rerunPolicy == NodeFail {	2✔
86	return nil	1✔
87	}	1✔
88	return fmt.Errorf("invalid node rerun policy %s", rerunPolicy)	1✔
89	}
90
91	type NodeStartPolicy string
92
93	const (
94	NodeStartManual NodeStartPolicy = "manual"
95	NodeStartAuto NodeStartPolicy = "auto" // Default
96	)
97
98	func ValidateStartPolicy(startPolicy NodeStartPolicy) error {	1✔
99	if startPolicy == NodeStartManual \|\|	1✔
100	startPolicy == NodeStartAuto {	2✔
101	return nil	1✔
102	}	1✔
103	return fmt.Errorf("invalid node start policy %s", startPolicy)	×
104	}
105
106	type ScriptNodeDef struct {
107	Name string // Get it from the key
108	Type NodeType `json:"type" yaml:"type"`
109	Desc string `json:"desc" yaml:"desc"`
110	StartPolicy NodeStartPolicy `json:"start_policy" yaml:"start_policy"`
111	RerunPolicy NodeRerunPolicy `json:"rerun_policy,omitempty" yaml:"rerun_policy,omitempty"`
112	CustomProcessorType string `json:"custom_proc_type,omitempty" yaml:"custom_proc_type,omitempty"`
113	HandlerExeType string `json:"handler_exe_type,omitempty" yaml:"handler_exe_type,omitempty"`
114	MaxBatchProcessingTime int `json:"max_batch_processing_time,omitempty" yaml:"max_batch_processing_time,omitempty"`
115
116	RawReader json.RawMessage `json:"r" yaml:"r"` // This depends on tfm type
117	TableReader TableReaderDef
118	FileReader FileReaderDef
119
120	Lookup LookupDef `json:"l" yaml:"l"`
121
122	RawProcessorDef json.RawMessage `json:"p" yaml:"p"` // This depends on tfm type
123	CustomProcessor CustomProcessorDef // Also should implement CustomProcessorRunner
124
125	RawWriter json.RawMessage `json:"w" yaml:"w"` // This depends on tfm type
126	DependencyPolicyName string `json:"dependency_policy" yaml:"dependency_policy"`
127	TableCreator TableCreatorDef
128	TableUpdater TableUpdaterDef
129	FileCreator FileCreatorDef
130	DepPolDef *DependencyPolicyDef
131	}
132
133	func (node *ScriptNodeDef) HasTableReader() bool {	1✔
134	return node.Type == NodeTypeTableTable \|\|	1✔
135	node.Type == NodeTypeTableLookupTable \|\|	1✔
136	node.Type == NodeTypeTableFile \|\|	1✔
137	node.Type == NodeTypeDistinctTable \|\|	1✔
138	node.Type == NodeTypeTableCustomTfmTable	1✔
139	}	1✔
140	func (node *ScriptNodeDef) HasFileReader() bool {	1✔
141	return node.Type == NodeTypeFileTable	1✔
142	}	1✔
143
144	func (node *ScriptNodeDef) HasLookup() bool {	1✔
145	return node.Type == NodeTypeTableLookupTable	1✔
146	}	1✔
147
148	func (node *ScriptNodeDef) HasCustomProcessor() bool {	1✔
149	return node.Type == NodeTypeTableCustomTfmTable	1✔
150	}	1✔
151
152	func (node *ScriptNodeDef) HasTableCreator() bool {	1✔
153	return node.Type == NodeTypeFileTable \|\|	1✔
154	node.Type == NodeTypeTableTable \|\|	1✔
155	node.Type == NodeTypeDistinctTable \|\|	1✔
156	node.Type == NodeTypeTableLookupTable \|\|	1✔
157	node.Type == NodeTypeTableCustomTfmTable	1✔
158	}	1✔
159	func (node *ScriptNodeDef) HasFileCreator() bool {	1✔
160	return node.Type == NodeTypeTableFile	1✔
161	}	1✔
162	func (node *ScriptNodeDef) GetTargetName() string {	1✔
163	if node.HasTableCreator() {	2✔
164	return node.TableCreator.Name	1✔
165	} else if node.HasFileCreator() {	3✔
166	return CreatorAlias	1✔
167	}	1✔
168	return "dev_error_uknown_target_name"	×
169	}
170
171	func (node *ScriptNodeDef) initReader() error {	1✔
172	if node.HasTableReader() {	2✔
173	if err := json.Unmarshal(node.RawReader, &node.TableReader); err != nil {	1✔
174	return fmt.Errorf("cannot unmarshal table reader: [%s]", err.Error())	×
175	}	×
176	foundErrors := make([]string, 0)	1✔
177	if len(node.TableReader.TableName) == 0 {	1✔
178	foundErrors = append(foundErrors, "table reader cannot reference empty table name")	×
179	}	×
180	if node.TableReader.ExpectedBatchesTotal == 0 {	2✔
181	node.TableReader.ExpectedBatchesTotal = 1	1✔
182	} else if node.TableReader.ExpectedBatchesTotal < 0 \|\| node.TableReader.ExpectedBatchesTotal > MaxAcceptedBatchesByTableReader {	2✔
183	foundErrors = append(foundErrors, fmt.Sprintf("table reader can accept between 1 and %d batches, %d specified", MaxAcceptedBatchesByTableReader, node.TableReader.ExpectedBatchesTotal))	×
184	}	×
185	if node.TableReader.RowsetSize < 0 \|\| MaxRowsetSize < node.TableReader.RowsetSize {	1✔
186	foundErrors = append(foundErrors, fmt.Sprintf("invalid rowset size %d, table reader can accept between 0 (defaults to %d) and %d", node.TableReader.RowsetSize, DefaultRowsetSize, MaxRowsetSize))	×
187	}	×
188	if node.TableReader.RowsetSize == 0 {	2✔
189	node.TableReader.RowsetSize = DefaultRowsetSize	1✔
190	}	1✔
191	if len(foundErrors) > 0 {	1✔
192	return fmt.Errorf("%s", strings.Join(foundErrors, "; "))	×
193	}	×
194	} else if node.HasFileReader() {	2✔
195	if err := node.FileReader.Deserialize(node.RawReader); err != nil {	1✔
196	return fmt.Errorf("cannot deserialize file reader [%s]: [%s]", string(node.RawReader), err.Error())	×
197	}	×
198	}
199
200	return nil	1✔
201	}
202
203	func (node *ScriptNodeDef) initCreator() error {	1✔
204	if node.HasTableCreator() {	2✔
205	if err := node.TableCreator.Deserialize(node.RawWriter); err != nil {	2✔
206	return fmt.Errorf("cannot deserialize table creator [%s]: [%s]", strings.ReplaceAll(string(node.RawWriter), "\n", " "), err.Error())	1✔
207	}	1✔
208	} else if node.HasFileCreator() {	2✔
209	if err := node.FileCreator.Deserialize(node.RawWriter); err != nil {	2✔
210	return fmt.Errorf("cannot deserialize file creator [%s]: [%s]", strings.ReplaceAll(string(node.RawWriter), "\n", " "), err.Error())	1✔
211	}	1✔
212	}
213	return nil	1✔
214	}
215
216	func (node *ScriptNodeDef) initCustomProcessor(customProcessorDefFactory CustomProcessorDefFactory, customProcessorsSettings map[string]json.RawMessage, scriptType ScriptType, caPath string, privateKeys map[string]string) error {	1✔
217	if node.HasCustomProcessor() {	2✔
218	if customProcessorDefFactory == nil {	1✔
219	return errors.New("undefined custom processor factory")	×
220	}	×
221	if customProcessorsSettings == nil {	1✔
222	return errors.New("missing custom processor settings section")	×
223	}	×
224	var ok bool	1✔
225	node.CustomProcessor, ok = customProcessorDefFactory.Create(node.CustomProcessorType)	1✔
226	if !ok {	1✔
227	return fmt.Errorf("cannot deserialize unknown custom processor %s", node.CustomProcessorType)	×
228	}	×
229	customProcSettings, ok := customProcessorsSettings[node.CustomProcessorType]	1✔
230	if !ok {	1✔
231	return fmt.Errorf("cannot find custom processing settings for [%s] in the environment config file", node.CustomProcessorType)	×
232	}	×
233	if err := node.CustomProcessor.Deserialize(node.RawProcessorDef, customProcSettings, scriptType, caPath, privateKeys); err != nil {	1✔
234	re := regexp.MustCompile("[ \r\n]+")	×
235	return fmt.Errorf("cannot deserialize custom processor [%s]: [%s]", re.ReplaceAllString(string(node.RawProcessorDef), ""), err.Error())	×
236	}	×
237	}
238	return nil	1✔
239	}
240
241	func (node *ScriptNodeDef) Deserialize(customProcessorDefFactory CustomProcessorDefFactory, customProcessorsSettings map[string]json.RawMessage, scriptType ScriptType, caPath string, privateKeys map[string]string) error {	1✔
242	foundErrors := make([]string, 0)	1✔
243		1✔
244	if err := ValidateNodeType(node.Type); err != nil {	1✔
245	return err	×
246	}	×
247
248	// Defaults
249
250	if node.MaxBatchProcessingTime == 0 {	2✔
251	node.MaxBatchProcessingTime = 60000 // 1 min	1✔
252	}	1✔
253
254	if len(node.HandlerExeType) == 0 {	2✔
255	node.HandlerExeType = HandlerExeTypeGeneric	1✔
256	}	1✔
257
258	if len(node.RerunPolicy) == 0 {	2✔
259	node.RerunPolicy = NodeRerun	1✔
260	} else if err := ValidateRerunPolicy(node.RerunPolicy); err != nil {	3✔
261	return err	1✔
262	}	1✔
263
264	if len(node.StartPolicy) == 0 {	2✔
265	node.StartPolicy = NodeStartAuto	1✔
266	} else if err := ValidateStartPolicy(node.StartPolicy); err != nil {	2✔
267	return err	×
268	}	×
269
270	// Reader
271	if err := node.initReader(); err != nil {	1✔
272	foundErrors = append(foundErrors, err.Error())	×
273	}	×
274
275	// Creator
276	if err := node.initCreator(); err != nil {	2✔
277	foundErrors = append(foundErrors, err.Error())	1✔
278	}	1✔
279
280	// Custom processor
281	if err := node.initCustomProcessor(customProcessorDefFactory, customProcessorsSettings, scriptType, caPath, privateKeys); err != nil {	1✔
282	foundErrors = append(foundErrors, err.Error())	×
283	}	×
284
285	// Distinct table
286	if node.Type == NodeTypeDistinctTable {	2✔
287	if node.RerunPolicy != NodeFail {	2✔
288	foundErrors = append(foundErrors, "distinct_table node must have fail policy, no reruns possible")	1✔
289	}	1✔
290	if _, _, err := node.TableCreator.GetSingleUniqueIndexDef(); err != nil {	2✔
291	foundErrors = append(foundErrors, err.Error())	1✔
292	}	1✔
293	}
294
295	if len(foundErrors) > 0 {	2✔
296	return fmt.Errorf("%s", strings.Join(foundErrors, "; "))	1✔
297	}	1✔
298
299	return nil	1✔
300	}
301
302	func (node *ScriptNodeDef) evalCreatorAndLookupExpressionsAndCheckType() error {	1✔
303	foundErrors := make([]string, 0, 2)	1✔
304		1✔
305	if node.HasLookup() && node.Lookup.UsesFilter() {	2✔
306	if err := evalExpressionWithFieldRefsAndCheckType(node.Lookup.Filter, node.Lookup.UsedInFilterFields, FieldTypeBool); err != nil {	2✔
307	foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate lookup filter expression [%s]: [%s]", node.Lookup.RawFilter, err.Error()))	1✔
308	}	1✔
309	}
310
311	if node.HasTableCreator() {	2✔
312	// Having	1✔
313	if err := evalExpressionWithFieldRefsAndCheckType(node.TableCreator.Having, node.TableCreator.UsedInHavingFields, FieldTypeBool); err != nil {	2✔
314	foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate table creator 'having' expression [%s]: [%s]", node.TableCreator.RawHaving, err.Error()))	1✔
315	}	1✔
316
317	// Target table fields
318	for tgtFieldName, tgtFieldDef := range node.TableCreator.Fields {	2✔
319		1✔
320	// TODO: find a way to check field usage:	1✔
321	// - lookup fields must be used only within enclosing agg calls (sum etc), otherwise last one wins	1✔
322	// - src table fields are allowed within enclosing agg calls, and there is even a biz case for it (multiply src field by the number of lookup rows)	1✔
323		1✔
324	// If no grouping is used, no agg calls allowed	1✔
325	if node.HasLookup() && !node.Lookup.IsGroup \|\| !node.HasLookup() {	2✔
326	v := AggFinderVisitor{}	1✔
327	if tgtFieldDef.ParsedExpression == nil {	1✔
NEW 328	foundErrors = append(foundErrors, fmt.Sprintf("cannot parse node %s, target field %s expression [%s] was not parsed successfully", node.Name, tgtFieldName, tgtFieldDef.RawExpression))	×
329	} else {	1✔
330	ast.Walk(&v, tgtFieldDef.ParsedExpression)	1✔
331	if v.Error != nil {	2✔
332	foundErrors = append(foundErrors, fmt.Sprintf("cannot use agg functions in [%s], lookup group flag is not set or no lookups used: [%s]", tgtFieldDef.RawExpression, v.Error.Error()))	1✔
333	}	1✔
334	}
335	}
336
337	// Just eval with test values, agg functions will go through preserving the type no problem
338	if err := evalExpressionWithFieldRefsAndCheckType(tgtFieldDef.ParsedExpression, node.TableCreator.UsedInTargetExpressionsFields, tgtFieldDef.Type); err != nil {	1✔
339	foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate table creator target field %s expression [%s]: [%s]", tgtFieldName, tgtFieldDef.RawExpression, err.Error()))	×
340	}	×
341	}
342	}
343
344	if node.HasFileCreator() {	2✔
345	// Having	1✔
346	if err := evalExpressionWithFieldRefsAndCheckType(node.FileCreator.Having, node.FileCreator.UsedInHavingFields, FieldTypeBool); err != nil {	2✔
347	foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate file creator 'having' expression [%s]: [%s]", node.FileCreator.RawHaving, err.Error()))	1✔
348	}	1✔
349
350	// Target table fields (yes, they are not just strings, we check the type)
351	for i := 0; i < len(node.FileCreator.Columns); i++ {	2✔
352	colDef := &node.FileCreator.Columns[i]	1✔
353	if err := evalExpressionWithFieldRefsAndCheckType(colDef.ParsedExpression, node.FileCreator.UsedInTargetExpressionsFields, colDef.Type); err != nil {	1✔
354	foundErrors = append(foundErrors, fmt.Sprintf("cannot evaluate file creator target field %s expression [%s]: [%s]", colDef.Name, colDef.RawExpression, err.Error()))	×
355	}	×
356	}
357	}
358
359	// NOTE: do not even try to eval expressions from the custom processor here,
360	// they may contain custom stuff and are pretty much guaranteed to fail
361
362	if len(foundErrors) > 0 {	2✔
363	return fmt.Errorf("%s", strings.Join(foundErrors, "; "))	1✔
364	}	1✔
365
366	return nil	1✔
367	}
368
369	func (node ScriptNodeDef) getSourceFieldRefs() (FieldRefs, error) {	1✔
370	if node.HasFileReader() {	2✔
371	return node.FileReader.getFieldRefs(), nil	1✔
372	} else if node.HasTableReader() {	3✔
373	return node.TableReader.TableCreator.GetFieldRefsWithAlias(ReaderAlias), nil	1✔
374	}	1✔
375
376	return nil, fmt.Errorf("dev error, node of type %s has no file or table reader", node.Type)	×
377	}
378
379	func (node ScriptNodeDef) GetUniqueIndexesFieldRefs() FieldRefs {	1✔
380	if !node.HasTableCreator() {	1✔
381	return &FieldRefs{}	×
382	}	×
383	fieldTypeMap := map[string]TableFieldType{}	1✔
384	for _, idxDef := range node.TableCreator.Indexes {	2✔
385	if idxDef.Uniqueness == IdxUnique {	2✔
386	for _, idxComponentDef := range idxDef.Components {	2✔
387	fieldTypeMap[idxComponentDef.FieldName] = idxComponentDef.FieldType	1✔
388	}	1✔
389	}
390	}
391	fieldRefs := make(FieldRefs, len(fieldTypeMap))	1✔
392	fieldRefIdx := 0	1✔
393	for fieldName, fieldType := range fieldTypeMap {	2✔
394	fieldRefs[fieldRefIdx] = FieldRef{	1✔
395	FieldName: fieldName,	1✔
396	FieldType: fieldType,	1✔
397	TableName: node.TableCreator.Name}	1✔
398	fieldRefIdx++	1✔
399	}	1✔
400
401	return &fieldRefs	1✔
402	}
403
404	func (node *ScriptNodeDef) GetTokenIntervalsByNumberOfBatches() ([][]int64, error) {	1✔
405	if node.HasTableReader() \|\| node.HasFileCreator() && node.TableReader.ExpectedBatchesTotal > 1 {	2✔
406	if node.TableReader.ExpectedBatchesTotal == 1 {	2✔
407	return [][]int64{{int64(math.MinInt64), int64(math.MaxInt64)}}, nil	1✔
408	}	1✔
409
410	tokenIntervalPerBatch := int64(math.MaxInt64/node.TableReader.ExpectedBatchesTotal) - int64(math.MinInt64/node.TableReader.ExpectedBatchesTotal)	1✔
411		1✔
412	intervals := make([][]int64, node.TableReader.ExpectedBatchesTotal)	1✔
413	left := int64(math.MinInt64)	1✔
414	for i := 0; i < len(intervals); i++ {	2✔
415	var right int64	1✔
416	if i == len(intervals)-1 {	2✔
417	right = math.MaxInt64	1✔
418	} else {	2✔
419	right = left + tokenIntervalPerBatch - 1	1✔
420	}	1✔
421	intervals[i] = []int64{left, right}	1✔
422	left = right + 1	1✔
423	}
424	return intervals, nil	1✔
425	// } else if node.HasFileCreator() && node.TableReader.ExpectedBatchesTotal == 1 {
426	// // One output file - one batch, dummy intervals
427	// intervals := make([][]int64, 1)
428	// intervals[0] = []int64{int64(0), 0}
429	// return intervals, nil
430	} else if node.HasFileReader() {	2✔
431	// One input file - one batch	1✔
432	intervals := make([][]int64, len(node.FileReader.SrcFileUrls))	1✔
433	for i := 0; i < len(node.FileReader.SrcFileUrls); i++ {	2✔
434	intervals[i] = []int64{int64(i), int64(i)}	1✔
435	}	1✔
436	return intervals, nil	1✔
437	}
438
439	return nil, fmt.Errorf("cannot find implementation for intervals for node %s", node.Name)	×
440	}
441
442	func (node *ScriptNodeDef) isNodeUsesIdx(idxName string) bool {	1✔
443	if node.HasLookup() && node.Lookup.IndexName == idxName {	2✔
444	return true	1✔
445	}	1✔
446
447	distinctIdxCandidate, ok := node.TableCreator.Indexes[idxName]	1✔
448	if ok {	2✔
449	if node.Type == NodeTypeDistinctTable && distinctIdxCandidate.Uniqueness == IdxUnique {	2✔
450	return true	1✔
451	}	1✔
452	}
453
454	return false	1✔
455	}

capillariesio / capillaries / 19619258155

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous