<commit_before>/*
 * Copyright (C) 2016 Red Hat, Inc.
 *
 * 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 traversal

import (
	"encoding/json"
	"errors"
	"fmt"

	"github.com/skydive-project/skydive/common"
	"github.com/skydive-project/skydive/flow"
	"github.com/skydive-project/skydive/logging"
	"github.com/skydive-project/skydive/storage"
	"github.com/skydive-project/skydive/topology/graph"
	"github.com/skydive-project/skydive/topology/graph/traversal"
)

const (
	FLOW_TOKEN      traversal.Token = 1001
	METRICS_TOKEN   traversal.Token = 1002
	BANDWIDTH_TOKEN traversal.Token = 1003
	HOPS_TOKEN      traversal.Token = 1004
	NODES_TOKEN     traversal.Token = 1005
)

type FlowTraversalExtension struct {
	FlowToken      traversal.Token
	MetricsToken   traversal.Token
	BandwidthToken traversal.Token
	HopsToken      traversal.Token
	NodesToken     traversal.Token
	TableClient    *flow.TableClient
	Storage        storage.Storage
}

type FlowGremlinTraversalStep struct {
	TableClient     *flow.TableClient
	Storage         storage.Storage
	context         traversal.GremlinTraversalContext
	metricsNextStep bool
}

type FlowTraversalStep struct {
	GraphTraversal  *traversal.GraphTraversal
	flowset         *flow.FlowSet
	flowSearchQuery flow.FlowSearchQuery
	error           error
}

type MetricsGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type MetricsTraversalStep struct {
	GraphTraversal *traversal.GraphTraversal
	metrics        map[string][]*flow.FlowMetric
}

type BandwidthGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type BandwidthTraversalStep struct {
	GraphTraversal *traversal.GraphTraversal
	bandwidth      *flow.FlowSetBandwidth
}

type HopsGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type NodesGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

func (f *FlowTraversalStep) Out(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) In(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Both(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Nodes(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		logging.GetLogger().Critical(err)
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.NodeUUID != "" && flow.NodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.NodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}
	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Count(s ...interface{}) *traversal.GraphTraversalValue {
	if f.Error() != nil {
		return traversal.NewGraphTraversalValue(f.GraphTraversal, 0, f.Error())
	}

	return traversal.NewGraphTraversalValue(f.GraphTraversal, len(f.flowset.Flows))
}

func paramsToFilter(s ...interface{}) (*flow.Filter, error) {
	if len(s) < 2 {
		return nil, errors.New("At least two parameters must be provided")
	}

	if len(s)%2 != 0 {
		return nil, fmt.Errorf("slice must be defined by pair k,v: %v", s)
	}

	andFilter := &flow.BoolFilter{Op: flow.BoolFilterOp_AND}

	for i := 0; i < len(s); i += 2 {
		k, ok := s[i].(string)
		if !ok {
			return nil, errors.New("keys should be of string type")
		}

		switch v := s[i+1].(type) {
		case *traversal.RegexMetadataMatcher:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					RegexFilter: &flow.RegexFilter{Key: k, Value: v.Value().(string)},
				})
		case *traversal.NEMetadataMatcher:
			notFilters := &flow.Filter{}
			switch t := v.Value().(type) {
			case string:
				notFilters.TermStringFilter = &flow.TermStringFilter{Key: k, Value: t}
			case int64:
				notFilters.TermInt64Filter = &flow.TermInt64Filter{Key: k, Value: t}
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op:      flow.BoolFilterOp_NOT,
						Filters: []*flow.Filter{notFilters},
					},
				},
			)
		case *traversal.LTMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("LT values should be of int64 type")
			}
		case *traversal.GTMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("GT values should be of int64 type")
			}
		case *traversal.GTEMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						GteInt64Filter: &flow.GteInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("GTE values should be of int64 type")
			}
		case *traversal.LTEMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						LteInt64Filter: &flow.LteInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("LTE values should be of int64 type")
			}
		case *traversal.InsideMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Inside values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case *traversal.OutsideMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Outside values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case *traversal.BetweenMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Between values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								GteInt64Filter: &flow.GteInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case string:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					TermStringFilter: &flow.TermStringFilter{Key: k, Value: v},
				},
			)
		case int64:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					TermInt64Filter: &flow.TermInt64Filter{Key: k, Value: v},
				},
			)
		default:
			return nil, fmt.Errorf("value type unknown: %v", v)
		}
	}

	return &flow.Filter{BoolFilter: andFilter}, nil
}

func (f *FlowTraversalStep) Has(s ...interface{}) *FlowTraversalStep {
	if f.error != nil {
		return f
	}

	filter, err := paramsToFilter(s...)
	if err != nil {
		return &FlowTraversalStep{error: err}
	}

	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset.Filter(filter)}
}

func (f *FlowTraversalStep) Dedup() *FlowTraversalStep {
	if f.Error() != nil {
		return &FlowTraversalStep{error: f.Error()}
	}

	f.flowset.Dedup()
	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset}
}

func (f *FlowTraversalStep) Sort() *FlowTraversalStep {
	if f.Error() != nil {
		return &FlowTraversalStep{error: f.Error()}
	}

	f.flowset.Sort()
	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset}
}

func (f *FlowTraversalStep) propertyInt64Values(field string) *traversal.GraphTraversalValue {
	var s []interface{}
	for _, fl := range f.flowset.Flows {
		if v, err := fl.GetFieldInt64(field); err == nil {
			s = append(s, v)
		}
	}

	return traversal.NewGraphTraversalValue(f.GraphTraversal, s, nil)
}

func (f *FlowTraversalStep) propertyStringValues(field string) *traversal.GraphTraversalValue {
	var s []interface{}
	for _, fl := range f.flowset.Flows {

		v, err := fl.GetFieldString(field)
		if err != nil {
			return traversal.NewGraphTraversalValue(f.GraphTraversal, nil, err)<commit_msg>gremlin: add Sum() step for flows<commit_after>/*
 * Copyright (C) 2016 Red Hat, Inc.
 *
 * 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 traversal

import (
	"encoding/json"
	"errors"
	"fmt"

	"github.com/skydive-project/skydive/common"
	"github.com/skydive-project/skydive/flow"
	"github.com/skydive-project/skydive/logging"
	"github.com/skydive-project/skydive/storage"
	"github.com/skydive-project/skydive/topology/graph"
	"github.com/skydive-project/skydive/topology/graph/traversal"
)

const (
	FLOW_TOKEN      traversal.Token = 1001
	METRICS_TOKEN   traversal.Token = 1002
	BANDWIDTH_TOKEN traversal.Token = 1003
	HOPS_TOKEN      traversal.Token = 1004
	NODES_TOKEN     traversal.Token = 1005
)

type FlowTraversalExtension struct {
	FlowToken      traversal.Token
	MetricsToken   traversal.Token
	BandwidthToken traversal.Token
	HopsToken      traversal.Token
	NodesToken     traversal.Token
	TableClient    *flow.TableClient
	Storage        storage.Storage
}

type FlowGremlinTraversalStep struct {
	TableClient     *flow.TableClient
	Storage         storage.Storage
	context         traversal.GremlinTraversalContext
	metricsNextStep bool
}

type FlowTraversalStep struct {
	GraphTraversal  *traversal.GraphTraversal
	flowset         *flow.FlowSet
	flowSearchQuery flow.FlowSearchQuery
	error           error
}

type MetricsGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type MetricsTraversalStep struct {
	GraphTraversal *traversal.GraphTraversal
	metrics        map[string][]*flow.FlowMetric
}

type BandwidthGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type BandwidthTraversalStep struct {
	GraphTraversal *traversal.GraphTraversal
	bandwidth      *flow.FlowSetBandwidth
}

type HopsGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

type NodesGremlinTraversalStep struct {
	TableClient *flow.TableClient
	Storage     storage.Storage
	context     traversal.GremlinTraversalContext
}

func (f *FlowTraversalStep) Out(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) In(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Both(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}

	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Nodes(s ...interface{}) *traversal.GraphTraversalV {
	var nodes []*graph.Node

	if f.Error() != nil {
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, f.Error())
	}

	m, err := traversal.SliceToMetadata(s...)
	if err != nil {
		logging.GetLogger().Critical(err)
		return traversal.NewGraphTraversalV(f.GraphTraversal, nodes, err)
	}

	for _, flow := range f.flowset.Flows {
		if flow.NodeUUID != "" && flow.NodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.NodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.ANodeUUID != "" && flow.ANodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.ANodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
		if flow.BNodeUUID != "" && flow.BNodeUUID != "*" {
			if node := f.GraphTraversal.Graph.GetNode(graph.Identifier(flow.BNodeUUID)); node != nil && node.MatchMetadata(m) {
				nodes = append(nodes, node)
			}
		}
	}
	return traversal.NewGraphTraversalV(f.GraphTraversal, nodes)
}

func (f *FlowTraversalStep) Count(s ...interface{}) *traversal.GraphTraversalValue {
	if f.Error() != nil {
		return traversal.NewGraphTraversalValue(f.GraphTraversal, 0, f.Error())
	}

	return traversal.NewGraphTraversalValue(f.GraphTraversal, len(f.flowset.Flows))
}

func paramsToFilter(s ...interface{}) (*flow.Filter, error) {
	if len(s) < 2 {
		return nil, errors.New("At least two parameters must be provided")
	}

	if len(s)%2 != 0 {
		return nil, fmt.Errorf("slice must be defined by pair k,v: %v", s)
	}

	andFilter := &flow.BoolFilter{Op: flow.BoolFilterOp_AND}

	for i := 0; i < len(s); i += 2 {
		k, ok := s[i].(string)
		if !ok {
			return nil, errors.New("keys should be of string type")
		}

		switch v := s[i+1].(type) {
		case *traversal.RegexMetadataMatcher:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					RegexFilter: &flow.RegexFilter{Key: k, Value: v.Value().(string)},
				})
		case *traversal.NEMetadataMatcher:
			notFilters := &flow.Filter{}
			switch t := v.Value().(type) {
			case string:
				notFilters.TermStringFilter = &flow.TermStringFilter{Key: k, Value: t}
			case int64:
				notFilters.TermInt64Filter = &flow.TermInt64Filter{Key: k, Value: t}
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op:      flow.BoolFilterOp_NOT,
						Filters: []*flow.Filter{notFilters},
					},
				},
			)
		case *traversal.LTMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("LT values should be of int64 type")
			}
		case *traversal.GTMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("GT values should be of int64 type")
			}
		case *traversal.GTEMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						GteInt64Filter: &flow.GteInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("GTE values should be of int64 type")
			}
		case *traversal.LTEMetadataMatcher:
			switch t := v.Value().(type) {
			case int64:
				andFilter.Filters = append(andFilter.Filters,
					&flow.Filter{
						LteInt64Filter: &flow.LteInt64Filter{Key: k, Value: t},
					},
				)
			default:
				return nil, errors.New("LTE values should be of int64 type")
			}
		case *traversal.InsideMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Inside values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case *traversal.OutsideMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Outside values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								GtInt64Filter: &flow.GtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case *traversal.BetweenMetadataMatcher:
			from, to := v.Value()

			f64, fok := from.(int64)
			t64, tok := to.(int64)

			if !fok || !tok {
				return nil, errors.New("Between values should be of int64 type")
			}

			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					BoolFilter: &flow.BoolFilter{
						Op: flow.BoolFilterOp_AND,
						Filters: []*flow.Filter{
							&flow.Filter{
								GteInt64Filter: &flow.GteInt64Filter{Key: k, Value: f64},
							},
							&flow.Filter{
								LtInt64Filter: &flow.LtInt64Filter{Key: k, Value: t64},
							},
						},
					},
				},
			)
		case string:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					TermStringFilter: &flow.TermStringFilter{Key: k, Value: v},
				},
			)
		case int64:
			andFilter.Filters = append(andFilter.Filters,
				&flow.Filter{
					TermInt64Filter: &flow.TermInt64Filter{Key: k, Value: v},
				},
			)
		default:
			return nil, fmt.Errorf("value type unknown: %v", v)
		}
	}

	return &flow.Filter{BoolFilter: andFilter}, nil
}

func (f *FlowTraversalStep) Has(s ...interface{}) *FlowTraversalStep {
	if f.error != nil {
		return f
	}

	filter, err := paramsToFilter(s...)
	if err != nil {
		return &FlowTraversalStep{error: err}
	}

	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset.Filter(filter)}
}

func (f *FlowTraversalStep) Dedup() *FlowTraversalStep {
	if f.Error() != nil {
		return &FlowTraversalStep{error: f.Error()}
	}

	f.flowset.Dedup()
	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset}
}

func (f *FlowTraversalStep) Sort() *FlowTraversalStep {
	if f.Error() != nil {
		return &FlowTraversalStep{error: f.Error()}
	}

	f.flowset.Sort()
	return &FlowTraversalStep{GraphTraversal: f.GraphTraversal, flowset: f.flowset}
}

// Sum aggregates integer values mapped by 'key' cross flows
func (f *FlowTraversalStep) Sum(keys ...interface{}) *traversal.GraphTraversalValue {
	if f.error != nil {
		return traversal.NewGraphTraversalValue(f.GraphTraversal, nil, f.error)
	}
	var s float64
	key := keys[0].(string)
	for _, fl := range f.flowset.Flows {
		if v, err := fl.GetFieldInt64(key); err == nil {
			s += float64(v)
		} else {
			return traversal.NewGraphTraversalValue(f.GraphTraversal, s, err)
		}
	}
	return traversal.NewGraphTraversalValue(f.GraphTraversal, s, nil)
}

func (f *FlowTraversalStep) propertyInt64Values(field string) *traversal.GraphTraversalValue {
	var s []interface{}
	for _, fl := range f.flowset.Flows {
		if v, err := fl.GetFieldInt64(field); err == nil {
			s = append(s, v)
		}
	}

	return traversal.NewGraphTraversalValue(f.GraphTraversal, s, nil)
}

func (f *FlowTraversalStep) propertyStringValues(field string) *traversal.GraphTraversalValue {
	var s []interface{}
	for _, fl := range f.flowset.Flows {

		v, err := fl.GetFieldString(field)
		if err != nil {
			return traversal.NewGraphTraversalValue(f.GraphTraversal, nil, err)
{'fraction_non_alphanumeric': 0.13050277582297287, 'fraction_numerical': 0.011924309953866605, 'mean_word_length': 3.551423487544484, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6583441', 'n_tokens_mistral': 10300, 'n_tokens_neox': 8772, 'n_words': 1928}