physicalplan

type Aggregation ¶

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

Aggregation groups together some lower level primitives to for the column to be aggregated by its function.

type AggregationFunction ¶

type AggregationFunction interface {
	Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)
}

type AndAggregation ¶

type AndAggregation struct{}

func (a *AndAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)

type AndExpr ¶

type AndExpr struct {
	Left  BooleanExpression
	Right BooleanExpression
}

func (a *AndExpr) Eval(r arrow.Record) (*Bitmap, error)

func (a *AndExpr) String() string

type ArrayRef ¶

type ArrayRef struct {
	ColumnName string
}

func (a *ArrayRef) ArrowArray(r arrow.Record) (arrow.Array, bool, error)

func (a *ArrayRef) String() string

type ArrayReference ¶

type ArrayReference struct{}

type BinaryScalarExpr ¶

type BinaryScalarExpr struct {
	Left  *ArrayRef
	Op    logicalplan.Op
	Right scalar.Scalar
}

func (e BinaryScalarExpr) Eval(r arrow.Record) (*Bitmap, error)

func (e BinaryScalarExpr) String() string

type Bitmap ¶

type Bitmap = roaring.Bitmap

func ArrayScalarCompute(funcName string, left arrow.Array, right scalar.Scalar) (*Bitmap, error)

func ArrayScalarContains(arr arrow.Array, right scalar.Scalar, not bool) (*Bitmap, error)

func ArrayScalarRegexMatch(left arrow.Array, right *regexp.Regexp) (*Bitmap, error)

func ArrayScalarRegexNotMatch(left arrow.Array, right *regexp.Regexp) (*Bitmap, error)

func BinaryArrayScalarRegexMatch(left *array.Binary, right *regexp.Regexp) (*Bitmap, error)

func BinaryArrayScalarRegexNotMatch(left *array.Binary, right *regexp.Regexp) (*Bitmap, error)

func BinaryDictionaryArrayScalarRegexMatch(dict *array.Dictionary, left *array.Binary, right *regexp.Regexp) (*Bitmap, error)

func BinaryDictionaryArrayScalarRegexNotMatch(dict *array.Dictionary, left *array.Binary, right *regexp.Regexp) (*Bitmap, error)

func BinaryScalarOperation(left arrow.Array, right scalar.Scalar, operator logicalplan.Op) (*Bitmap, error)

func DictionaryArrayScalarContains(left *array.Dictionary, right scalar.Scalar, not bool) (*Bitmap, error)

func DictionaryArrayScalarEqual(left *array.Dictionary, right scalar.Scalar) (*Bitmap, error)

func DictionaryArrayScalarNotEqual(left *array.Dictionary, right scalar.Scalar) (*Bitmap, error)

func NewBitmap() *Bitmap

func StringArrayScalarRegexMatch(left *array.String, right *regexp.Regexp) (*Bitmap, error)

func StringArrayScalarRegexNotMatch(left *array.String, right *regexp.Regexp) (*Bitmap, error)

type BooleanExpression ¶

type BooleanExpression interface {
	Eval(r arrow.Record) (*Bitmap, error)
	String() string
}

type CountAggregation ¶

type CountAggregation struct{}

func (a *CountAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)

type Diagram ¶

type Diagram struct {
	Details string
	Child   *Diagram
}

func (d *Diagram) String() string

type Distinction ¶

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

func Distinct(pool memory.Allocator, tracer trace.Tracer, columns []logicalplan.Expr) *Distinction

func (d *Distinction) Callback(ctx context.Context, r arrow.Record) error

func (d *Distinction) Close()

func (d *Distinction) Draw() *Diagram

func (d *Distinction) Finish(ctx context.Context) error

func (d *Distinction) SetNext(plan PhysicalPlan)

type HashAggregate ¶

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

func NewHashAggregate(
	pool memory.Allocator,
	tracer trace.Tracer,
	aggregations []Aggregation,
	groupByColumnMatchers []logicalplan.Expr,
	seed maphash.Seed,
	finalStage bool,
) *HashAggregate

func (a *HashAggregate) Callback(_ context.Context, r arrow.Record) error

func (a *HashAggregate) Close()

func (a *HashAggregate) Draw() *Diagram

func (a *HashAggregate) Finish(ctx context.Context) error

func (a *HashAggregate) SetNext(next PhysicalPlan)

type IndexRange ¶

type IndexRange struct {
	Start uint32
	End   uint32
}

type Limiter ¶

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

func Limit(pool memory.Allocator, tracer trace.Tracer, expr logicalplan.Expr) (*Limiter, error)

func (l *Limiter) Callback(ctx context.Context, r arrow.Record) error

func (l *Limiter) Close()

func (l *Limiter) Draw() *Diagram

func (l *Limiter) Finish(ctx context.Context) error

func (l *Limiter) SetNext(next PhysicalPlan)

type MaxAggregation ¶

type MaxAggregation struct{}

func (a *MaxAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)

type MinAggregation ¶

type MinAggregation struct{}

func (a *MinAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)

type Option ¶

type Option func(o *execOptions)

func WithOrderedAggregations() Option

func WithOverrideInput(input []PhysicalPlan) Option

func WithReadMode(m logicalplan.ReadMode) Option

type OrExpr ¶

type OrExpr struct {
	Left  BooleanExpression
	Right BooleanExpression
}

func (a *OrExpr) Eval(r arrow.Record) (*Bitmap, error)

func (a *OrExpr) String() string

type OrderedAggregate ¶

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

OrderedAggregate is an aggregation operator that supports aggregations on streams of data ordered by the group by columns. This is a more efficient aggregation than aggregating by hash since a group can be determined as completed once a different aggregation key is found in the ordered stream. OrderedAggregate also supports partially ordered aggregations. This means aggregating on keys that arrive in ordered sets of data that are not mutually exclusive. For example consider the group by columns: a, a, b, c, a, b, c. The OrderedAggregate will perform the aggregation on the first ordered set a, a, b, c and another one on the second a, b, c. The result of both aggregations is merged. Specifically, if the example is pushed to Callback in two records (a, a, b, c) followed by (a, b, c), and assuming that the aggregation values for each row are 1 for simplicity and we're using a sum aggregation, after the first call to Callback the OrderedAggregate will store [a, b, c], [2, 1, 1] but not emit anything. When the second record is pushed, the OrderedAggregate will realize that the first value in the new record (a) sorts before the "current group" (c), so will store the aggregation results of the second record as another ordered group [a, b, c], [1, 1, 1]. Only when Finish is called, will the OrderedAggregate be able to emit the merged aggregation results. The merged results should be: [a, b, c], [3, 2, 2].

func NewOrderedAggregate(
	pool memory.Allocator,
	tracer trace.Tracer,
	aggregation Aggregation,
	groupByColumnMatchers []logicalplan.Expr,
	finalStage bool,
) *OrderedAggregate

func (a *OrderedAggregate) Callback(_ context.Context, r arrow.Record) error

func (a *OrderedAggregate) Close()

func (a *OrderedAggregate) Draw() *Diagram

func (a *OrderedAggregate) Finish(ctx context.Context) error

func (a *OrderedAggregate) SetNext(next PhysicalPlan)

type OrderedSynchronizer ¶

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

OrderedSynchronizer implements synchronizing ordered input from multiple goroutines. The strategy used is that any input that calls Callback must wait for all the other inputs to call Callback, since an ordered result cannot be produced until all inputs have pushed data. Another strategy would be to store the pushed records, but that requires fully copying all the data for safety.

func NewOrderedSynchronizer(pool memory.Allocator, inputs int, orderByExprs []logicalplan.Expr) *OrderedSynchronizer

func (o *OrderedSynchronizer) Callback(ctx context.Context, r arrow.Record) error

func (o *OrderedSynchronizer) Close()

func (o *OrderedSynchronizer) Draw() *Diagram

func (o *OrderedSynchronizer) Finish(ctx context.Context) error

func (o *OrderedSynchronizer) SetNext(next PhysicalPlan)

type OutputPlan ¶

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

func Build(
	ctx context.Context,
	pool memory.Allocator,
	tracer trace.Tracer,
	s *dynparquet.Schema,
	plan *logicalplan.LogicalPlan,
	options ...Option,
) (*OutputPlan, error)

func (e *OutputPlan) Callback(ctx context.Context, r arrow.Record) error

func (e *OutputPlan) Close()

func (e *OutputPlan) Draw() *Diagram

func (e *OutputPlan) DrawString() string

func (e *OutputPlan) Execute(ctx context.Context, pool memory.Allocator, callback func(ctx context.Context, r arrow.Record) error) error

func (e *OutputPlan) Finish(_ context.Context) error

func (e *OutputPlan) SetNext(_ PhysicalPlan)

func (e *OutputPlan) SetNextCallback(next func(ctx context.Context, r arrow.Record) error)

type PhysicalPlan ¶

type PhysicalPlan interface {
	Callback(ctx context.Context, r arrow.Record) error
	Finish(ctx context.Context) error
	SetNext(next PhysicalPlan)
	Draw() *Diagram
	Close()
}

func Aggregate(
	pool memory.Allocator,
	tracer trace.Tracer,
	agg *logicalplan.Aggregation,
	final bool,
	ordered bool,
	seed maphash.Seed,
) (PhysicalPlan, error)

type PostPlanVisitorFunc ¶

type PostPlanVisitorFunc func(plan *logicalplan.LogicalPlan) bool

func (f PostPlanVisitorFunc) PostVisit(plan *logicalplan.LogicalPlan) bool

func (f PostPlanVisitorFunc) PreVisit(_ *logicalplan.LogicalPlan) bool

type PreExprVisitorFunc ¶

type PreExprVisitorFunc func(expr logicalplan.Expr) bool

func (f PreExprVisitorFunc) PostVisit(_ logicalplan.Expr) bool

func (f PreExprVisitorFunc) PreVisit(expr logicalplan.Expr) bool

func (f PreExprVisitorFunc) Visit(_ logicalplan.Expr) bool

type PrePlanVisitorFunc ¶

type PrePlanVisitorFunc func(plan *logicalplan.LogicalPlan) bool

func (f PrePlanVisitorFunc) PostVisit(_ *logicalplan.LogicalPlan) bool

func (f PrePlanVisitorFunc) PreVisit(plan *logicalplan.LogicalPlan) bool

type PredicateFilter ¶

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

func Filter(pool memory.Allocator, tracer trace.Tracer, filterExpr logicalplan.Expr) (*PredicateFilter, error)

func (f *PredicateFilter) Callback(ctx context.Context, r arrow.Record) error

func (f *PredicateFilter) Close()

func (f *PredicateFilter) Draw() *Diagram

func (f *PredicateFilter) Finish(ctx context.Context) error

func (f *PredicateFilter) SetNext(next PhysicalPlan)

type Projection ¶

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

func Project(mem memory.Allocator, tracer trace.Tracer, exprs []logicalplan.Expr) (*Projection, error)

func (p *Projection) Callback(ctx context.Context, r arrow.Record) error

func (p *Projection) Close()

func (p *Projection) Draw() *Diagram

func (p *Projection) Finish(ctx context.Context) error

func (p *Projection) Project(_ context.Context, r arrow.Record) (arrow.Record, error)

func (p *Projection) SetNext(next PhysicalPlan)

type RegExpFilter ¶

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

func (f *RegExpFilter) Eval(r arrow.Record) (*Bitmap, error)

func (f *RegExpFilter) String() string

type ScanPhysicalPlan ¶

type ScanPhysicalPlan interface {
	Execute(ctx context.Context, pool memory.Allocator) error
	Draw() *Diagram
}

type SchemaScan ¶

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

func (s *SchemaScan) Draw() *Diagram

func (s *SchemaScan) Execute(ctx context.Context, pool memory.Allocator) error

type SumAggregation ¶

type SumAggregation struct{}

func (a *SumAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)

type Synchronizer ¶

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

Synchronizer is used to combine the results of multiple parallel streams into a single stream concurrent stream. It also forms a barrier on the finishers, by waiting to call next plan's finish until all previous parallel stages have finished.

func Synchronize(concurrency int) *Synchronizer

func (m *Synchronizer) Callback(ctx context.Context, r arrow.Record) error

func (m *Synchronizer) Close()

func (m *Synchronizer) Draw() *Diagram

func (m *Synchronizer) Finish(ctx context.Context) error

func (m *Synchronizer) SetNext(next PhysicalPlan)

func (m *Synchronizer) SetNextPlan(nextPlan PhysicalPlan)

type TableScan ¶

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

func (s *TableScan) Draw() *Diagram

func (s *TableScan) Execute(ctx context.Context, pool memory.Allocator) error

type UniqueAggregation ¶

type UniqueAggregation struct{}

func (a *UniqueAggregation) Aggregate(pool memory.Allocator, arrs []arrow.Array) (arrow.Array, error)