flux

package module
v0.18.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Feb 6, 2019 License: MIT Imports: 21 Imported by: 88

README

Flux - Influx data language

CircleCI

Flux is a lightweight scripting language for querying databases (like InfluxDB) and working with data. It's part of InfluxDB 1.7 and 2.0, but can be run independently of those. This repo represents the language definition and an implementation of the language core.

NOTE: We plan to provide a flux command line program that exposes a REPL and talks to various data sources. In the meantime see the influx command in this repo as it has working Flux installation against the 2.0 InfluxDB database.

Specification

A complete specification can be found in SPEC.md. The specification contains many examples to start learning Flux.

Getting Started

Currently Flux is only avaliable via InfluxDB, see http://docs.influxdata.com/flux/ for getting started.

Basic Syntax

Here are a few examples of the language to get an idea of the syntax.

// This line is a comment

// Support for traditional math operators
1 + 1

// Several data types are built-in
true                     // a boolean true value
1                        // an int
1.0                      // a float
"this is a string"       // a string literal
1h5m                     // a duration of time representing 1 hour and 5 minutes
2018-10-10               // a time starting at midnight for the default timezone on Oct 10th 2018
2018-10-10T10:05:00      // a time at 10:05 AM for the default timezone on Oct 10th 2018
[1,1,2]                  // an array of integers
{foo: "str", bar: false} // an object with two keys and their values

// Values can be assigned to identifers
x = 5.0
x + 3.0 // 8.0

// Import libraries
import "math"

// Call functions always using keyword arguments
math.pow(base: 5, exponent: 3) // 5^3 = 125

// Functions are defined by assigning them to identifers
add = (a, b) => a + b

// Call add using keyword arguments
add(a: 5, b: 3) // 8

// Functions are polymorphic
add(a: 5.5, b: 2.5) // 8.0

// Access data from a database and store it as an identifer
import "influxdb"
data = influxdb.from(bucket:"telegraf/autogen")

// Chain more transformation functions to further specify the desired data
cpu = data 
    // only get the last 5m of data
    |> range(start: -5m)
    // only get the "usage_user" data from the _measurement "cpu"
    |> filter(fn: (r) => r._measurement == "cpu" and r._field == "usage_user")

// Return the data to the client
cpu |> yield()

// Group an aggregate along different dimensions
cpu
    // organize data into groups by host and region
    |> group(columns:["host","region"])
    // compute the mean of each group
    |> mean()
    // yield this result to the client
    |> yield()

// Window an aggregate over time
cpu
    // organize data into groups of 1 minute
    // compute the mean of each group
    |> aggregateWindow(every: 1m, fn: mean)
    // yield this result to the client
    |> yield()

// Gather different data
mem = data 
    // only get the last 5m of data
    |> range(start: -5m)
    // only get the "used_percent" data from the _measurement "mem"
    |> filter(fn: (r) => r._measurement == "mem" and r._field == "used_percent")


// Join data to create wider tables and map a function over the result
join(tables: {cpu:cpu, mem:mem}, on:["_time", "host"])
    // compute the ratio of cpu usage to mem used_percent
    |> map(fn:(r) => {_time: r._time, _value: r._value_cpu / r._value_mem)
    // again yield this result to the client
    |> yield()

The above examples give only a taste of what is possible with Flux. See the complete documentation for more complete examples and installation instructions.

Documentation

Overview

Example (Option)

Example_option demonstrates retrieving an option value from a scope object 

package main

import (
	"fmt"
	"os"

	"github.com/influxdata/flux"

	_ "github.com/influxdata/flux/stdlib"
)

func main() {

	// Instantiate a new universe block
	universe := flux.Prelude()

	// Retrieve the default value for the now option
	nowFunc, _ := universe.Lookup("now")

	// The now option is a function value whose default behavior is to return
	// the current system time when called. The function now() doesn't take
	// any arguments so can be called with nil.
	nowTime, _ := nowFunc.Function().Call(nil)
	fmt.Fprintf(os.Stderr, "The current system time (UTC) is: %v\n", nowTime)
}

Output:
Example (OverrideDefaultOptionExternally)

Example_overrideDefaultOptionExternally demonstrates how declaring an option in a Flux script will change that option's binding globally. 

package main

import (
	"fmt"

	"github.com/influxdata/flux"
	"github.com/influxdata/flux/interpreter"
	"github.com/influxdata/flux/parser"
	"github.com/influxdata/flux/semantic"

	_ "github.com/influxdata/flux/stdlib"
)

func main() {
	queryString := `
		option now = () => 2018-07-13T00:00:00Z
		what_time_is_it = now()`

	itrp := interpreter.NewInterpreter()
	universe := flux.Prelude()

	astPkg := parser.ParseSource(queryString)
	semPkg, _ := semantic.New(astPkg)

	// Evaluate package
	_, err := itrp.Eval(semPkg, universe, nil)
	if err != nil {
		fmt.Println(err)
	}

	// After evaluating the package, lookup the value of what_time_is_it
	now, _ := universe.Lookup("what_time_is_it")

	// what_time_is_it? Why it's ....
	fmt.Printf("The new current time (UTC) is: %v", now)
}

Output:

The new current time (UTC) is: 2018-07-13T00:00:00.000000000Z
Example (OverrideDefaultOptionInternally)

Example_overrideDefaultOptionInternally demonstrates how one can override a default option that is used in a query before that query is evaluated by the interpreter. 

package main

import (
	"fmt"
	"time"

	"github.com/influxdata/flux"
	"github.com/influxdata/flux/interpreter"
	"github.com/influxdata/flux/parser"
	"github.com/influxdata/flux/semantic"
	"github.com/influxdata/flux/values"

	_ "github.com/influxdata/flux/stdlib"
)

func main() {
	queryString := `what_time_is_it = now()`

	itrp := interpreter.NewInterpreter()
	universe := flux.Prelude()

	astPkg := parser.ParseSource(queryString)
	semPkg, _ := semantic.New(astPkg)

	// Define a new now function which returns a static time value of 2018-07-13T00:00:00.000000000Z
	timeValue := time.Date(2018, 7, 13, 0, 0, 0, 0, time.UTC)
	functionName := "newTime"
	functionType := semantic.NewFunctionType(semantic.FunctionSignature{
		Return: semantic.Time,
	})
	functionCall := func(args values.Object) (values.Value, error) {
		return values.NewTime(values.ConvertTime(timeValue)), nil
	}
	sideEffect := false

	newNowFunc := values.NewFunction(functionName, functionType, functionCall, sideEffect)

	// Override the default now function with the new one
	universe.Set("now", newNowFunc)

	// Evaluate package
	_, err := itrp.Eval(semPkg, universe, nil)
	if err != nil {
		fmt.Println(err)
	}

	// After evaluating the package, lookup the value of what_time_is_it
	now, _ := universe.Lookup("what_time_is_it")

	// what_time_is_it? Why it's ....
	fmt.Printf("The new current time (UTC) is: %v", now)
}

Output:

The new current time (UTC) is: 2018-07-13T00:00:00.000000000Z
Example (SetOption)

Example_setOption demonstrates setting an option value on a scope object 

package main

import (
	"fmt"

	"github.com/influxdata/flux"
	"github.com/influxdata/flux/values"

	_ "github.com/influxdata/flux/stdlib"
)

func main() {

	// Instantiate a new universe block
	universe := flux.Prelude()

	// Create a new option binding
	universe.Set("dummy_option", values.NewInt(3))

	v, _ := universe.Lookup("dummy_option")

	fmt.Printf("dummy_option = %d", v.Int())
}

Output:

dummy_option = 3

Index

Examples

Constants

View Source 
const (
	TablesParameter = "tables"
)

Variables

View Source 
var (
	MinTime = Time{
		Absolute: time.Unix(0, math.MinInt64),
	}
	MaxTime = Time{
		Absolute: time.Unix(0, math.MaxInt64),
	}
	Now = Time{
		IsRelative: true,
	}
)
View Source 
var DefaultTrigger = AfterWatermarkTriggerSpec{}
View Source 
var TableObjectMonoType semantic.Type
View Source 
var TableObjectType = semantic.NewObjectPolyType(

	map[string]semantic.PolyType{
		tableKindKey: semantic.String,
	},
	nil,

	semantic.LabelSet{tableKindKey},
)

Functions

func BuiltIns 

func BuiltIns() map[string]values.Value

BuiltIns returns a copy of the builtin values and their declarations.

func Eval 

func Eval(flux string, opts ...ScopeMutator) ([]values.Value, interpreter.Scope, error)

Eval accepts a Flux script and evaluates it to produce a set of side effects (as a slice of values) and a scope.

func EvalAST added in v0.18.0 

func EvalAST(astPkg *ast.Package, opts ...ScopeMutator) ([]values.Value, interpreter.Scope, error)

EvalAST accepts a Flux AST and evaluates it to produce a set of side effects (as a slice of values) and a scope.

func FinalizeBuiltIns 

func FinalizeBuiltIns()

FinalizeBuiltIns must be called to complete registration. Future calls to RegisterFunction or RegisterPackageValue will panic.

func FmtJSON 

func FmtJSON(f *formatter)

func Formatted 

func Formatted(q *Spec, opts ...FormatOption) fmt.Formatter

func FunctionSignature 

func FunctionSignature(parameters map[string]semantic.PolyType, required []string) semantic.FunctionPolySignature

FunctionSignature returns a standard functions signature which accepts a table piped argument, with any additional arguments.

func FunctionValue added in v0.14.0 

func FunctionValue(name string, c CreateOperationSpec, sig semantic.FunctionPolySignature) values.Value

FunctionValue creates a values.Value from the operation spec and signature. Name is the name of the function as it would be called. c is a function reference of type CreateOperationSpec sig is a function signature type that specifies the names and types of each argument for the function.

func FunctionValueWithSideEffect added in v0.14.0 

func FunctionValueWithSideEffect(name string, c CreateOperationSpec, sig semantic.FunctionPolySignature) values.Value

FunctionValueWithSideEffect creates a values.Value from the operation spec and signature. Name is the name of the function as it would be called. c is a function reference of type CreateOperationSpec sig is a function signature type that specifies the names and types of each argument for the function.

func IsEncoderError 

func IsEncoderError(err error) bool

IsEncoderError reports whether or not the underlying cause of an error is a valid EncoderError.

func NumberOfOperations 

func NumberOfOperations() int

func Parse added in v0.18.0 

func Parse(flux string) (*ast.Package, error)

Parse parses a Flux script and produces an ast.Package.

func Prelude added in v0.14.0 

func Prelude() interpreter.Scope

Prelude returns a scope object representing the Flux universe block

func RegisterOpSpec 

func RegisterOpSpec(k OperationKind, c NewOperationSpec)

RegisterOpSpec registers an operation spec with a given kind. k is a label that uniquely identifies this operation. If the kind has already been registered the call panics. c is a function reference that creates a new, default-initialized opSpec for the given kind. TODO:(nathanielc) make this part of RegisterMethod/RegisterFunction

func RegisterPackage added in v0.14.0 

func RegisterPackage(pkg *ast.Package)

RegisterPackage adds a builtin package

func RegisterPackageValue added in v0.14.0 

func RegisterPackageValue(pkgpath, name string, value values.Value)

RegisterPackageValue adds a value for an identifier in a builtin package

func ReplacePackageValue added in v0.14.0 

func ReplacePackageValue(pkgpath, name string, value values.Value)

ReplacePackageValue replaces a value for an identifier in a builtin package

func SemanticType added in v0.14.0 

func SemanticType(typ ColType) semantic.Type

func StdLib added in v0.14.0 

func StdLib() interpreter.Importer

StdLib returns an importer for the Flux standard library.

Types

type Administration 

type Administration struct {
	// contains filtered or unexported fields
}

func (*Administration) AddParent 

func (a *Administration) AddParent(np *TableObject)

AddParent instructs the evaluation Context that a new edge should be created from the parent to the current operation. Duplicate parents will be removed, so the caller need not concern itself with which parents have already been added.

func (*Administration) AddParentFromArgs 

func (a *Administration) AddParentFromArgs(args Arguments) error

AddParentFromArgs reads the args for the `table` argument and adds the value as a parent.

type AfterAtLeastCountTriggerSpec 

type AfterAtLeastCountTriggerSpec struct {
	Count int
}

func (AfterAtLeastCountTriggerSpec) Kind 

func (AfterAtLeastCountTriggerSpec) Kind() TriggerKind

type AfterProcessingTimeTriggerSpec 

type AfterProcessingTimeTriggerSpec struct {
	Duration Duration
}

func (AfterProcessingTimeTriggerSpec) Kind 

func (AfterProcessingTimeTriggerSpec) Kind() TriggerKind

type AfterWatermarkTriggerSpec 

type AfterWatermarkTriggerSpec struct {
	AllowedLateness Duration
}

func (AfterWatermarkTriggerSpec) Kind 

func (AfterWatermarkTriggerSpec) Kind() TriggerKind

type Arguments 

type Arguments struct {
	interpreter.Arguments
}

func (Arguments) GetDuration 

func (a Arguments) GetDuration(name string) (Duration, bool, error)

func (Arguments) GetRequiredDuration 

func (a Arguments) GetRequiredDuration(name string) (Duration, error)

func (Arguments) GetRequiredTime 

func (a Arguments) GetRequiredTime(name string) (Time, error)

func (Arguments) GetTime 

func (a Arguments) GetTime(name string) (Time, bool, error)

type Bounds 

type Bounds struct {
	Start Time
	Stop  Time
	Now   time.Time
}

func (Bounds) HasZero 

func (b Bounds) HasZero() bool

HasZero returns true if the given bounds contain a Go zero time value as either Start or Stop.

func (Bounds) IsEmpty 

func (b Bounds) IsEmpty() bool

IsEmpty reports whether the given bounds are empty, i.e., if start >= stop.

type ColMeta 

type ColMeta struct {
	// Label is the name of the column. The label is unique per table.
	Label string
	// Type is the type of the column. Only basic types are allowed.
	Type ColType
}

ColMeta contains the information about the column metadata.

type ColReader 

type ColReader interface {
	Key() GroupKey
	// Cols returns a list of column metadata.
	Cols() []ColMeta
	// Len returns the length of the slices.
	// All slices will have the same length.
	Len() int
	Bools(j int) *array.Boolean
	Ints(j int) *array.Int64
	UInts(j int) *array.Uint64
	Floats(j int) *array.Float64
	Strings(j int) *array.Binary
	Times(j int) *array.Int64
}

ColReader allows access to reading arrow buffers of column data. All data the ColReader exposes is guaranteed to be in memory. Once an ColReader goes out of scope, all slices are considered invalid.

type ColType 

type ColType int

ColType is the type for a column. This covers only basic data types. 

const (
	TInvalid ColType = iota
	TBool
	TInt
	TUInt
	TFloat
	TString
	TTime
)

func ColumnType 

func ColumnType(typ semantic.Type) ColType

ColumnType returns the column type when given a semantic.Type. It returns flux.TInvalid if the Type is not a valid column type.

func (ColType) String 

func (t ColType) String() string

String returns a string representation of the column type.

type Compiler 

type Compiler interface {
	// Compile produces a specification for the query.
	Compile(ctx context.Context) (*Spec, error)
	CompilerType() CompilerType
}

Compiler produces a specification for the query.

type CompilerMappings 

type CompilerMappings map[CompilerType]CreateCompiler

func (CompilerMappings) Add 

func (m CompilerMappings) Add(t CompilerType, c CreateCompiler) error

type CompilerType 

type CompilerType string

CompilerType is the name of a query compiler.

type CreateCompiler 

type CreateCompiler func() Compiler

type CreateDialect 

type CreateDialect func() Dialect

type CreateOperationSpec 

type CreateOperationSpec func(args Arguments, a *Administration) (OperationSpec, error)

type DelimitedMultiResultEncoder 

type DelimitedMultiResultEncoder struct {
	Delimiter []byte
	Encoder   interface {
		ResultEncoder
		// EncodeError encodes an error on the writer.
		EncodeError(w io.Writer, err error) error
	}
}

DelimitedMultiResultEncoder encodes multiple results using a trailing delimiter. The delimiter is written after every result.

If an error is encountered when iterating and the error is an encoder error, the error will be returned. Otherwise, the error is assumed to have arisen from query execution, and said error will be encoded with the EncodeError method of the Encoder field.

If the io.Writer implements flusher, it will be flushed after each delimiter.

func (*DelimitedMultiResultEncoder) Encode 

func (e *DelimitedMultiResultEncoder) Encode(w io.Writer, results ResultIterator) (int64, error)

type Dialect 

type Dialect interface {
	// Encoder creates an encoder for the results
	Encoder() MultiResultEncoder
	// DialectType report the type of the dialect
	DialectType() DialectType
}

Dialect describes how to encode results.

type DialectMappings 

type DialectMappings map[DialectType]CreateDialect

func (DialectMappings) Add 

func (m DialectMappings) Add(t DialectType, c CreateDialect) error

type DialectType 

type DialectType string

DialectType is the name of a query result dialect.

type Duration 

type Duration time.Duration

Duration is a marshalable duration type. TODO make this the real duration parsing not just time.ParseDuration

func (Duration) MarshalText 

func (d Duration) MarshalText() ([]byte, error)

func (*Duration) UnmarshalText 

func (d *Duration) UnmarshalText(data []byte) error

type Edge 

type Edge struct {
	Parent OperationID `json:"parent"`
	Child  OperationID `json:"child"`
}

Edge is a data flow relationship between a parent and a child

type EncoderError 

type EncoderError interface {
	IsEncoderError() bool
}

EncoderError is an interface that any error produced from a ResultEncoder implementation should conform to. It allows for differentiation between errors that occur in results, and errors that occur while encoding results.

type FormatOption 

type FormatOption func(*formatter)

TODO(nathanielc): Add better options for formatting plans as Graphviz dot format.

type GroupKey 

type GroupKey interface {
	Cols() []ColMeta
	Values() []values.Value

	HasCol(label string) bool
	LabelValue(label string) values.Value

	IsNull(j int) bool
	ValueBool(j int) bool
	ValueUInt(j int) uint64
	ValueInt(j int) int64
	ValueFloat(j int) float64
	ValueString(j int) string
	ValueDuration(j int) values.Duration
	ValueTime(j int) values.Time
	Value(j int) values.Value

	Equal(o GroupKey) bool
	Less(o GroupKey) bool
	String() string
}

type GroupKeys added in v0.15.0 

type GroupKeys []GroupKey

GroupKeys provides a sortable collection of group keys.

func (GroupKeys) Len added in v0.15.0 

func (a GroupKeys) Len() int

func (GroupKeys) Less added in v0.15.0 

func (a GroupKeys) Less(i, j int) bool

func (GroupKeys) String added in v0.15.0 

func (a GroupKeys) String() string

String returns a string representation of the keys

func (GroupKeys) Swap added in v0.15.0 

func (a GroupKeys) Swap(i, j int)

type GroupMode added in v0.14.0 

type GroupMode int

GroupMode represents the method for grouping data 

const (
	// GroupModeNone indicates that no grouping action is specified
	GroupModeNone GroupMode = 0
	// GroupModeBy produces a table for each unique value of the specified GroupKeys.
	GroupModeBy GroupMode = 1 << iota
	// GroupModeExcept produces a table for the unique values of all keys, except those specified by GroupKeys.
	GroupModeExcept
)

type IDer 

type IDer interface {
	ID(*TableObject) OperationID
}

IDer produces the mapping of table Objects to OpertionIDs

type IDerOpSpec 

type IDerOpSpec interface {
	IDer(ider IDer)
}

IDerOpSpec is the interface any operation spec that needs access to OperationIDs in the query spec must implement.

type MultiResultDecoder 

type MultiResultDecoder interface {
	// Decode decodes multiple results from r.
	Decode(r io.ReadCloser) (ResultIterator, error)
}

MultiResultDecoder can decode multiple results from a reader.

type MultiResultEncoder 

type MultiResultEncoder interface {
	// Encode writes multiple results from r into w.
	// Returns the number of bytes written to w and any error resulting from the encoding process.
	// Errors obtained from the results object should be encoded to w and then discarded.
	Encode(w io.Writer, results ResultIterator) (int64, error)
}

MultiResultEncoder can encode multiple results into a writer.

type NewOperationSpec 

type NewOperationSpec func() OperationSpec

func OperationSpecNewFn 

func OperationSpecNewFn(k OperationKind) NewOperationSpec

type Operation 

type Operation struct {
	ID   OperationID   `json:"id"`
	Spec OperationSpec `json:"spec"`
}

Operation denotes a single operation in a query.

func (Operation) MarshalJSON 

func (o Operation) MarshalJSON() ([]byte, error)

func (*Operation) UnmarshalJSON 

func (o *Operation) UnmarshalJSON(data []byte) error

type OperationID 

type OperationID string

OperationID is a unique ID within a query for the operation.

type OperationKind 

type OperationKind string

OperationKind denotes the kind of operations.

type OperationSpec 

type OperationSpec interface {
	// Kind returns the kind of the operation.
	Kind() OperationKind
}

OperationSpec specifies an operation as part of a query.

type Option 

type Option func(*options)

func Verbose 

func Verbose(v bool) Option

type OrFinallyTriggerSpec 

type OrFinallyTriggerSpec struct {
	Main    TriggerSpec
	Finally TriggerSpec
}

func (OrFinallyTriggerSpec) Kind 

func (OrFinallyTriggerSpec) Kind() TriggerKind

type Priority 

type Priority int32

Priority is an integer that represents the query priority. Any positive 32bit integer value may be used. Special constants are provided to represent the extreme high and low priorities. 

const (
	// High is the highest possible priority = 0
	High Priority = 0
	// Low is the lowest possible priority = MaxInt32
	Low Priority = math.MaxInt32
)

func (Priority) MarshalText 

func (p Priority) MarshalText() ([]byte, error)

func (*Priority) UnmarshalText 

func (p *Priority) UnmarshalText(txt []byte) error

type Query 

type Query interface {
	// Spec returns the spec used to execute this query.
	// Spec must not be modified.
	Spec() *Spec

	// Ready returns a channel that will deliver the query results.
	// Its possible that the channel is closed before any results arrive,
	// in which case the query should be inspected for an error using Err().
	Ready() <-chan map[string]Result

	// Done must always be called to free resources. It is safe to call Done
	// multiple times.
	Done()

	// Cancel will signal that query execution should stop.
	// Done must still be called to free resources.
	// It is safe to call Cancel multiple times.
	Cancel()

	// Err reports any error the query may have encountered.
	Err() error

	Statisticser
}

Query represents an active query.

type RepeatedTriggerSpec 

type RepeatedTriggerSpec struct {
	Trigger TriggerSpec
}

func (RepeatedTriggerSpec) Kind 

func (RepeatedTriggerSpec) Kind() TriggerKind

type ResourceManagement 

type ResourceManagement struct {
	// Priority or the query.
	// Queries with a lower value will move to the front of the priority queue.
	// A zero value indicates the highest priority.
	Priority Priority `json:"priority"`
	// ConcurrencyQuota is the number of concurrency workers allowed to process this query.
	// A zero value indicates the planner can pick the optimal concurrency.
	ConcurrencyQuota int `json:"concurrency_quota"`
	// MemoryBytesQuota is the number of bytes of RAM this query may consume.
	// There is a small amount of overhead memory being consumed by a query that will not be counted towards this limit.
	// A zero value indicates unlimited.
	MemoryBytesQuota int64 `json:"memory_bytes_quota"`
}

ResourceManagement defines how the query should consume avaliable resources.

type Result 

type Result interface {
	Name() string
	// Tables returns a TableIterator for iterating through results
	Tables() TableIterator
	// Statistics returns statistics collected the processing of the result.
	Statistics() Statistics
}

type ResultDecoder 

type ResultDecoder interface {
	// Decode decodes data from r into a result.
	Decode(r io.Reader) (Result, error)
}

ResultDecoder can decode a result from a reader.

type ResultEncoder 

type ResultEncoder interface {
	// Encode encodes data from the result into w.
	// Returns the number of bytes written to w and any error.
	Encode(w io.Writer, result Result) (int64, error)
}

ResultEncoder can encode a result into a writer.

type ResultIterator 

type ResultIterator interface {
	// More indicates if there are more results.
	More() bool

	// Next returns the next result.
	// If More is false, Next panics.
	Next() Result

	// Release discards the remaining results and frees the currently used resources.
	// It must always be called to free resources. It can be called even if there are
	// more results. It is safe to call Release multiple times.
	Release()

	// Err reports the first error encountered.
	// Err will not report anything unless More has returned false,
	// or the query has been cancelled.
	Err() error

	// Statistics returns any statistics computed by the resultset.
	Statistics() Statistics
}

ResultIterator allows iterating through all results synchronously. A ResultIterator is not thread-safe and all of the methods are expected to be called within the same goroutine. A ResultIterator may implement Statisticser.

func NewMapResultIterator 

func NewMapResultIterator(results map[string]Result) ResultIterator

func NewResultIteratorFromQuery 

func NewResultIteratorFromQuery(q Query) ResultIterator

func NewSliceResultIterator 

func NewSliceResultIterator(results []Result) ResultIterator

type ScopeMutator added in v0.14.0 

type ScopeMutator = func(interpreter.Scope)

ScopeMutator is any function that mutates the scope of an identifier.

func SetOption added in v0.14.0 

func SetOption(name string, v values.Value) ScopeMutator

SetOption returns a func that adds a var binding to a scope.

type Spec 

type Spec struct {
	Operations []*Operation       `json:"operations"`
	Edges      []Edge             `json:"edges"`
	Resources  ResourceManagement `json:"resources"`
	Now        time.Time          `json:"now"`
	// contains filtered or unexported fields
}

Spec specifies a query.

func Compile 

func Compile(ctx context.Context, q string, now time.Time, opts ...Option) (*Spec, error)

Compile evaluates a Flux script producing a query Spec. now parameter must be non-zero, that is the default now time should be set before compiling.

func CompileAST added in v0.18.0 

func CompileAST(ctx context.Context, astPkg *ast.Package, now time.Time, opts ...Option) (*Spec, error)

CompileAST evaluates a Flux AST and produces a query Spec. now parameter must be non-zero, that is the default now time should be set before compiling.

func ToSpec 

func ToSpec(functionCalls []values.Value, now time.Time) (*Spec, error)

func (*Spec) Children 

func (q *Spec) Children(id OperationID) []*Operation

Children returns a list of children for a given operation. If the query is invalid no children will be returned.

func (*Spec) Functions 

func (q *Spec) Functions() ([]string, error)

Functions return the names of all functions used in the plan

func (*Spec) Parents 

func (q *Spec) Parents(id OperationID) []*Operation

Parents returns a list of parents for a given operation. If the query is invalid no parents will be returned.

func (*Spec) Validate 

func (q *Spec) Validate() error

Validate ensures the query is a valid DAG.

func (*Spec) Walk 

func (q *Spec) Walk(f func(o *Operation) error) error

Walk calls f on each operation exactly once. The function f will be called on an operation only after all of its parents have already been passed to f.

type Statistics 

type Statistics struct {
	// TotalDuration is the total amount of time in nanoseconds spent.
	TotalDuration time.Duration `json:"total_duration"`
	// CompileDuration is the amount of time in nanoseconds spent compiling the query.
	CompileDuration time.Duration `json:"compile_duration"`
	// QueueDuration is the amount of time in nanoseconds spent queueing.
	QueueDuration time.Duration `json:"queue_duration"`
	// PlanDuration is the amount of time in nanoseconds spent in plannig the query.
	PlanDuration time.Duration `json:"plan_duration"`
	// RequeueDuration is the amount of time in nanoseconds spent requeueing.
	RequeueDuration time.Duration `json:"requeue_duration"`
	// ExecuteDuration is the amount of time in nanoseconds spent in executing the query.
	ExecuteDuration time.Duration `json:"execute_duration"`

	// Concurrency is the number of goroutines allocated to process the query
	Concurrency int `json:"concurrency"`
	// MaxAllocated is the maximum number of bytes the query allocated.
	MaxAllocated int64 `json:"max_allocated"`

	// ScannedValues is the number of values scanned.
	ScannedValues int `json:"scanned_values"`
	// ScannedBytes number of uncompressed bytes scanned.
	ScannedBytes int `json:"scanned_bytes"`
}

Statistics is a collection of statisitcs about the processing of a query.

func (Statistics) Add added in v0.7.1 

func (s Statistics) Add(other Statistics) Statistics

Add returns the sum of s and other.

type Statisticser 

type Statisticser interface {
	// Statistics reports the statisitcs for the query.
	// The statisitcs are not complete until the query is finished.
	Statistics() Statistics
}

Statisticser reports statisitcs about query processing.

type Table 

type Table interface {
	Key() GroupKey

	Cols() []ColMeta

	// Do calls f to process the data contained within the table.
	// It uses the arrow buffers.
	Do(f func(ColReader) error) error

	// RefCount modifies the reference count on the table by n.
	// When the RefCount goes to zero, the table is freed.
	RefCount(n int)

	// Empty returns whether the table contains no records.
	Empty() bool

	// Stats returns collected statistics about this table during processing.
	Statistics() Statistics
}

type TableIterator 

type TableIterator interface {
	Do(f func(Table) error) error
	Statistics() Statistics
}

type TableObject 

type TableObject struct {
	Kind    OperationKind
	Spec    OperationSpec
	Parents values.Array
	// contains filtered or unexported fields
}

func (*TableObject) Array 

func (t *TableObject) Array() values.Array

func (*TableObject) Bool 

func (t *TableObject) Bool() bool

func (*TableObject) Duration 

func (t *TableObject) Duration() values.Duration

func (*TableObject) Equal 

func (t *TableObject) Equal(rhs values.Value) bool

func (*TableObject) Float 

func (t *TableObject) Float() float64

func (*TableObject) Function 

func (t *TableObject) Function() values.Function

func (*TableObject) Get 

func (t *TableObject) Get(name string) (values.Value, bool)

func (*TableObject) Int 

func (t *TableObject) Int() int64

func (*TableObject) IsNull added in v0.14.0 

func (t *TableObject) IsNull() bool

func (*TableObject) Len 

func (t *TableObject) Len() int

func (*TableObject) Object 

func (t *TableObject) Object() values.Object

func (*TableObject) Operation 

func (t *TableObject) Operation(ider IDer) *Operation

func (*TableObject) PolyType 

func (t *TableObject) PolyType() semantic.PolyType

func (*TableObject) Range 

func (t *TableObject) Range(f func(name string, v values.Value))

func (*TableObject) Regexp 

func (t *TableObject) Regexp() *regexp.Regexp

func (*TableObject) Set 

func (t *TableObject) Set(name string, v values.Value)

func (*TableObject) Str 

func (t *TableObject) Str() string

func (*TableObject) String 

func (t *TableObject) String() string

func (*TableObject) Time 

func (t *TableObject) Time() values.Time

func (*TableObject) ToSpec 

func (t *TableObject) ToSpec() *Spec

func (*TableObject) Type 

func (t *TableObject) Type() semantic.Type

func (*TableObject) UInt 

func (t *TableObject) UInt() uint64

type Time 

type Time struct {
	IsRelative bool
	Relative   time.Duration
	Absolute   time.Time
}

Time represents either a relative or absolute time. If Time is its zero value then it represents a time.Time{}. To represent the now time you must set IsRelative to true.

func ToQueryTime 

func ToQueryTime(value values.Value) (Time, error)

func (Time) IsZero 

func (t Time) IsZero() bool

func (Time) MarshalText 

func (t Time) MarshalText() ([]byte, error)

func (Time) Time 

func (t Time) Time(now time.Time) time.Time

Time returns the time specified relative to now.

func (*Time) UnmarshalText 

func (t *Time) UnmarshalText(data []byte) error

type TriggerKind 

type TriggerKind int
const (
	AfterWatermark TriggerKind = iota
	Repeated
	AfterProcessingTime
	AfterAtLeastCount
	OrFinally
)

type TriggerSpec 

type TriggerSpec interface {
	Kind() TriggerKind
}

Source Files

View all Source files

Directories

Path Synopsis
ast
Package ast declares the types used to represent the syntax tree for Flux source code.
 Package ast declares the types used to represent the syntax tree for Flux source code.
asttest
Package asttest implements utilities for testing the abstract syntax tree.
 Package asttest implements utilities for testing the abstract syntax tree.
asttest/cmpgen
cmpgen generates comparison options for the asttest package.
 cmpgen generates comparison options for the asttest package.
edit
Package builtin contains all packages related to Flux built-ins are imported and initialized.
 Package builtin contains all packages related to Flux built-ins are imported and initialized.
cmd
flux
flux/cmd
The compiler package provides a compiler and Go runtime for a subset of the Flux language.
 The compiler package provides a compiler and Go runtime for a subset of the Flux language.
Package complete provides types to aid with auto-completion of Flux scripts in editors.
 Package complete provides types to aid with auto-completion of Flux scripts in editors.
Package control controls which resources a query may consume.
 Package control controls which resources a query may consume.
controltest
Package controltest provides a controller for use in tests.
 Package controltest provides a controller for use in tests.
Package csv contains the csv result encoders and decoders.
 Package csv contains the csv result encoders and decoders.
Package execute contains the implementation of the execution phase in the query engine.
 Package execute contains the implementation of the execution phase in the query engine.
executetest
Package executetest contains utilities for testing the query execution phase.
 Package executetest contains utilities for testing the query execution phase.
internal
cmd/builtin
cmd/builtin/cmd
cmd/changelog
cmd/changelog/cmd
gen
parser
pkg/syncutil
scanner
token
tools Module
Package interpreter provides the implementation of the Flux interpreter.
 Package interpreter provides the implementation of the Flux interpreter.
interptest
Package mock contains mock implementations of the query package interfaces for testing.
 Package mock contains mock implementations of the query package interfaces for testing.
Package parser implements a parser for Flux source files.
 Package parser implements a parser for Flux source files.
plantest
Package plantest contains utilities for testing each query planning phase
 Package plantest contains utilities for testing each query planning phase
Package querytest contains utilities for testing the query end-to-end.
 Package querytest contains utilities for testing the query end-to-end.
Package repl implements the read-eval-print-loop for the command line flux query console.
 Package repl implements the read-eval-print-loop for the command line flux query console.
The semantic package provides a graph structure that represents the meaning of a Flux script.
 The semantic package provides a graph structure that represents the meaning of a Flux script.
semantictest
Package semantictest contains utilities for testing the semantic package.
 Package semantictest contains utilities for testing the semantic package.
Package stdlib represents the Flux standard library.
 Package stdlib represents the Flux standard library.
csv
generate
http
influxdata/influxdb
From is an operation that mocks the real implementation of InfluxDB's from.
 From is an operation that mocks the real implementation of InfluxDB's from.
influxdata/influxdb/v1
kafka
sql
strings
system
testing
universe
Package transformations contains the implementations for the builtin transformation functions.
 Package transformations contains the implementations for the builtin transformation functions.
Package values declares the flux data types and implements them.
 Package values declares the flux data types and implements them.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.