Documentation ¶
Overview ¶
Package executetest contains utilities for testing the query execution phase.
Index ¶
- Constants
- Variables
- func AggFuncBenchmarkHelper(b *testing.B, agg execute.Aggregate, data *array.Float64, want interface{})
- func AggFuncTestHelper(t *testing.T, agg execute.Aggregate, data *array.Float64, want interface{})
- func CreateAllocatingFromSource(spec plan.ProcedureSpec, id execute.DatasetID, a execute.Administration) (execute.Source, error)
- func CreateFromSource(spec plan.ProcedureSpec, id execute.DatasetID, a execute.Administration) (execute.Source, error)
- func CreateToTransformation(id execute.DatasetID, mode execute.AccumulationMode, spec plan.ProcedureSpec, ...) (execute.Transformation, execute.Dataset, error)
- func EqualResult(w, g flux.Result) error
- func EqualResultIterators(want, got flux.ResultIterator) error
- func EqualResults(want, got []flux.Result) error
- func IndexSelectorFuncBenchmarkHelper(b *testing.B, selector execute.IndexSelector, data flux.BufferedTable)
- func IndexSelectorFuncTestHelper(t *testing.T, selector execute.IndexSelector, data flux.Table, want [][]int)
- func MustCopyTable(tbl flux.Table) flux.Table
- func NewTestExecuteDependencies() flux.Dependencies
- func NewToProcedure(flux.OperationSpec, plan.Administration) (plan.ProcedureSpec, error)
- func NewYieldProcedureSpec(name string) plan.PhysicalProcedureSpec
- func NewYieldTransformation(d execute.Dataset, cache execute.TableBuilderCache) *yieldTransformation
- func NormalizeTables(bs []*Table)
- func ProcessTestHelper(t *testing.T, data []flux.Table, want []*Table, wantErr error, ...)
- func ProcessTestHelper2(t *testing.T, data []flux.Table, want []*Table, wantErr error, ...)
- func RandomDatasetID() execute.DatasetID
- func RowSelectorFuncBenchmarkHelper(b *testing.B, selector execute.RowSelector, data flux.Table)
- func RowSelectorFuncTestHelper(t *testing.T, selector execute.RowSelector, data flux.Table, ...)
- func RunTableTests(t *testing.T, tt TableTest)
- func TransformationPassThroughTestHelper(t *testing.T, newTr NewTransformation)
- type AllocatingFromProcedureSpec
- func (s *AllocatingFromProcedureSpec) AddTransformation(t execute.Transformation)
- func (s *AllocatingFromProcedureSpec) Copy() plan.ProcedureSpec
- func (AllocatingFromProcedureSpec) Cost(inStats []plan.Statistics) (cost plan.Cost, outStats plan.Statistics)
- func (AllocatingFromProcedureSpec) Kind() plan.ProcedureKind
- func (s *AllocatingFromProcedureSpec) Run(ctx context.Context)
- type ColReader
- func (cr *ColReader) Bools(j int) *array.Boolean
- func (cr *ColReader) Cols() []flux.ColMeta
- func (cr *ColReader) Floats(j int) *array.Float64
- func (cr *ColReader) Ints(j int) *array.Int64
- func (cr *ColReader) Key() flux.GroupKey
- func (cr *ColReader) Len() int
- func (cr *ColReader) Release()
- func (cr *ColReader) Retain()
- func (cr *ColReader) Strings(j int) *array.Binary
- func (cr *ColReader) Times(j int) *array.Int64
- func (cr *ColReader) UInts(j int) *array.Uint64
- type CreateNewTransformationWithDeps
- type Dataset
- func (d *Dataset) AddTransformation(t execute.Transformation)
- func (d *Dataset) Finish(err error)
- func (d *Dataset) RetractTable(key flux.GroupKey) error
- func (d *Dataset) SetTriggerSpec(t plan.TriggerSpec)
- func (d *Dataset) UpdateProcessingTime(t execute.Time) error
- func (d *Dataset) UpdateWatermark(mark execute.Time) error
- type FromProcedureSpec
- func (src *FromProcedureSpec) AddTransformation(t execute.Transformation)
- func (src *FromProcedureSpec) Copy() plan.ProcedureSpec
- func (src *FromProcedureSpec) Cost(inStats []plan.Statistics) (plan.Cost, plan.Statistics)
- func (src *FromProcedureSpec) Kind() plan.ProcedureKind
- func (src *FromProcedureSpec) Run(ctx context.Context)
- type NewTransformation
- type Result
- type RowWiseTable
- type SortedTables
- type SourceUrlValidationTestCases
- type Table
- type TableIterator
- type TableTest
- type TfUrlValidationTest
- type TfUrlValidationTestCase
- type ToProcedureSpec
- type ToTransformation
- func (t *ToTransformation) Finish(id execute.DatasetID, err error)
- func (t *ToTransformation) Process(id execute.DatasetID, tbl flux.Table) error
- func (t *ToTransformation) RetractTable(id execute.DatasetID, key flux.GroupKey) error
- func (t *ToTransformation) UpdateProcessingTime(id execute.DatasetID, pt execute.Time) error
- func (t *ToTransformation) UpdateWatermark(id execute.DatasetID, pt execute.Time) error
- type YieldProcedureSpec
Constants ¶
const AllocatingFromTestKind = "allocating-from-test"
const FromTestKind = "from-test"
const ToTestKind = "to-test"
ToTestKind represents an side-effect producing kind for testing
const YieldKind = "yield-test"
Variables ¶
var UnlimitedAllocator = &memory.Allocator{}
Functions ¶
func AggFuncBenchmarkHelper ¶
func AggFuncBenchmarkHelper(b *testing.B, agg execute.Aggregate, data *array.Float64, want interface{})
AggFuncBenchmarkHelper benchmarks the aggregate function over data and compares to wantValue
func AggFuncTestHelper ¶
AggFuncTestHelper splits the data in half, runs Do over each split and compares the Value to want.
func CreateAllocatingFromSource ¶ added in v0.28.3
func CreateAllocatingFromSource(spec plan.ProcedureSpec, id execute.DatasetID, a execute.Administration) (execute.Source, error)
func CreateFromSource ¶
func CreateFromSource(spec plan.ProcedureSpec, id execute.DatasetID, a execute.Administration) (execute.Source, error)
func CreateToTransformation ¶ added in v0.8.0
func CreateToTransformation(id execute.DatasetID, mode execute.AccumulationMode, spec plan.ProcedureSpec, a execute.Administration) (execute.Transformation, execute.Dataset, error)
func EqualResult ¶ added in v0.36.2
EqualResult compares to results for equality
func EqualResultIterators ¶ added in v0.36.2
func EqualResultIterators(want, got flux.ResultIterator) error
EqualResultIterators compares two ResultIterators for equlity
func EqualResults ¶ added in v0.7.1
EqualResults compares two lists of Flux Results for equlity
func IndexSelectorFuncBenchmarkHelper ¶
func IndexSelectorFuncBenchmarkHelper(b *testing.B, selector execute.IndexSelector, data flux.BufferedTable)
func NewTestExecuteDependencies ¶ added in v0.41.0
func NewTestExecuteDependencies() flux.Dependencies
func NewToProcedure ¶ added in v0.8.0
func NewToProcedure(flux.OperationSpec, plan.Administration) (plan.ProcedureSpec, error)
func NewYieldProcedureSpec ¶
func NewYieldProcedureSpec(name string) plan.PhysicalProcedureSpec
func NewYieldTransformation ¶ added in v0.21.0
func NewYieldTransformation(d execute.Dataset, cache execute.TableBuilderCache) *yieldTransformation
func NormalizeTables ¶
func NormalizeTables(bs []*Table)
NormalizeTables ensures that each table is normalized
func ProcessTestHelper ¶
func ProcessTestHelper2 ¶ added in v0.49.0
func RandomDatasetID ¶
func RunTableTests ¶ added in v0.36.0
RunTableTests will run the common table tests over each table in the returned TableIterator. The function will be called for each test.
func TransformationPassThroughTestHelper ¶
func TransformationPassThroughTestHelper(t *testing.T, newTr NewTransformation)
Types ¶
type AllocatingFromProcedureSpec ¶ added in v0.28.3
type AllocatingFromProcedureSpec struct { ByteCount int // contains filtered or unexported fields }
AllocatingFromProcedureSpec is a procedure spec AND an execution node that allocates ByteCount bytes during execution.
func (*AllocatingFromProcedureSpec) AddTransformation ¶ added in v0.28.3
func (s *AllocatingFromProcedureSpec) AddTransformation(t execute.Transformation)
func (*AllocatingFromProcedureSpec) Copy ¶ added in v0.28.3
func (s *AllocatingFromProcedureSpec) Copy() plan.ProcedureSpec
func (AllocatingFromProcedureSpec) Cost ¶ added in v0.28.3
func (AllocatingFromProcedureSpec) Cost(inStats []plan.Statistics) (cost plan.Cost, outStats plan.Statistics)
func (AllocatingFromProcedureSpec) Kind ¶ added in v0.28.3
func (AllocatingFromProcedureSpec) Kind() plan.ProcedureKind
func (*AllocatingFromProcedureSpec) Run ¶ added in v0.28.3
func (s *AllocatingFromProcedureSpec) Run(ctx context.Context)
type CreateNewTransformationWithDeps ¶ added in v0.51.0
type CreateNewTransformationWithDeps func(d execute.Dataset, deps flux.Dependencies, cache execute.TableBuilderCache, spec plan.ProcedureSpec) (execute.Transformation, error)
sql.createToSQLTransformation() and kafka.createToKafkaTransformation() converts plan.ProcedureSpec to their struct implementations ToSQLProcedureSpec and ToKafkaProcedureSpec respectively. This complicated the test harness requiring us to provide CreateNewTransformationWithDeps functions to do the plan.ProcedureSpec conversion and call the subsequent factory method namely: kafka.NewToKafkaTransformation() and sql.NewToSQLTransformation() See also: sql/to_test.go/TestToSql_NewTransformation and kafka/to_test.go/TestToKafka_NewTransformation
type Dataset ¶
type Dataset struct { ID execute.DatasetID Retractions []flux.GroupKey ProcessingTimeUpdates []execute.Time WatermarkUpdates []execute.Time Finished bool FinishedErr error }
func NewDataset ¶
func (*Dataset) AddTransformation ¶
func (d *Dataset) AddTransformation(t execute.Transformation)
func (*Dataset) SetTriggerSpec ¶
func (d *Dataset) SetTriggerSpec(t plan.TriggerSpec)
type FromProcedureSpec ¶
type FromProcedureSpec struct {
// contains filtered or unexported fields
}
FromProcedureSpec is a procedure spec AND an execution Node. It simulates the execution of a basic physical scan operation.
func NewFromProcedureSpec ¶
func NewFromProcedureSpec(data []*Table) *FromProcedureSpec
NewFromProcedureSpec specifies a from-test procedure with source data
func (*FromProcedureSpec) AddTransformation ¶
func (src *FromProcedureSpec) AddTransformation(t execute.Transformation)
func (*FromProcedureSpec) Copy ¶
func (src *FromProcedureSpec) Copy() plan.ProcedureSpec
func (*FromProcedureSpec) Cost ¶
func (src *FromProcedureSpec) Cost(inStats []plan.Statistics) (plan.Cost, plan.Statistics)
func (*FromProcedureSpec) Kind ¶
func (src *FromProcedureSpec) Kind() plan.ProcedureKind
func (*FromProcedureSpec) Run ¶
func (src *FromProcedureSpec) Run(ctx context.Context)
type NewTransformation ¶
type NewTransformation func(execute.Dataset, execute.TableBuilderCache) execute.Transformation
type Result ¶
func ConvertResult ¶ added in v0.36.2
ConvertResult produces a result object from any flux.Result type.
func (*Result) Tables ¶
func (r *Result) Tables() flux.TableIterator
type RowWiseTable ¶ added in v0.14.0
type RowWiseTable struct {
*Table
}
RowWiseTable is a flux Table implementation that calls f once for each row in its Do method.
type SortedTables ¶
type SortedTables []*Table
func (SortedTables) Len ¶
func (b SortedTables) Len() int
func (SortedTables) Swap ¶
func (b SortedTables) Swap(i int, j int)
type SourceUrlValidationTestCases ¶ added in v0.51.0
type SourceUrlValidationTestCases []struct { Name string Spec plan.ProcedureSpec V url.Validator ErrMsg string }
Some sources are located by a URL. e.g. sql.from, socket.from the URL/DSN supplied by the user need to be validated by a URLValidator{} before we can establish the connection. This struct (as well as the Run() method) acts as a test harness for that.
func (*SourceUrlValidationTestCases) Run ¶ added in v0.51.0
func (testCases *SourceUrlValidationTestCases) Run(t *testing.T, fn execute.CreateSource)
type Table ¶
type Table struct { // GroupKey of the table. Does not need to be set explicitly. GroupKey flux.GroupKey // KeyCols is a list of column that are part of the group key. // The column type is deduced from the ColMeta slice. KeyCols []string // KeyValues is a list of values for the group key columns. // Only needs to be set when no data is present on the table. KeyValues []interface{} // ColMeta is a list of columns of the table. ColMeta []flux.ColMeta // Data is a list of rows, i.e. Data[row][col] // Each row must be a list with length equal to len(ColMeta) Data [][]interface{} // Err contains the error that should be returned // by this table when calling Do. Err error // Alloc is the allocator used to create the column readers. // Memory is not tracked unless this is set. Alloc *memory.Allocator // IsDone indicates if this table has been used. IsDone bool }
Table is an implementation of execute.Table It is designed to make it easy to statically declare the data within the table. Not all fields need to be set. See comments on each field. Use Normalize to ensure that all fields are set before equality comparisons.
type TableIterator ¶
type TableTest ¶ added in v0.36.0
type TableTest struct { // NewFn returns a new TableIterator that can be processed. // The table iterator that is produced should have multiple // tables of different shapes and sizes to get coverage of // as much of the code as possible. The TableIterator will // be created once for each subtest. NewFn func(ctx context.Context, alloc *memory.Allocator) flux.TableIterator // IsDone will report if the table is considered done for reading. // The call to Done should force this to be true, but it is possible // for this to return true before the table has been processed. IsDone func(flux.Table) bool }
type TfUrlValidationTest ¶ added in v0.51.0
type TfUrlValidationTest struct { CreateFn CreateNewTransformationWithDeps Cases []TfUrlValidationTestCase }
func (*TfUrlValidationTest) Run ¶ added in v0.51.0
func (test *TfUrlValidationTest) Run(t *testing.T)
type TfUrlValidationTestCase ¶ added in v0.51.0
type ToProcedureSpec ¶ added in v0.8.0
type ToProcedureSpec struct{}
ToProcedureSpec defines an output operation. That is, an operation that does not transform its input data but performs a side effect while passing its input data through to the next op.
func (*ToProcedureSpec) Copy ¶ added in v0.8.0
func (s *ToProcedureSpec) Copy() plan.ProcedureSpec
func (*ToProcedureSpec) Cost ¶ added in v0.8.0
func (s *ToProcedureSpec) Cost(inStats []plan.Statistics) (plan.Cost, plan.Statistics)
func (*ToProcedureSpec) Kind ¶ added in v0.8.0
func (s *ToProcedureSpec) Kind() plan.ProcedureKind
type ToTransformation ¶ added in v0.8.0
type ToTransformation struct {
// contains filtered or unexported fields
}
ToTransformation simulates an output or an identity transformation
func (*ToTransformation) Finish ¶ added in v0.8.0
func (t *ToTransformation) Finish(id execute.DatasetID, err error)
func (*ToTransformation) RetractTable ¶ added in v0.8.0
func (*ToTransformation) UpdateProcessingTime ¶ added in v0.8.0
func (*ToTransformation) UpdateWatermark ¶ added in v0.8.0
type YieldProcedureSpec ¶
type YieldProcedureSpec struct { plan.DefaultCost // contains filtered or unexported fields }
func (YieldProcedureSpec) Copy ¶
func (y YieldProcedureSpec) Copy() plan.ProcedureSpec
func (YieldProcedureSpec) Kind ¶
func (YieldProcedureSpec) Kind() plan.ProcedureKind
func (YieldProcedureSpec) YieldName ¶
func (y YieldProcedureSpec) YieldName() string