query

Documentation

Index

• func GroupByUserWindow(records []arrow.Record, window time.Duration) (map[int64]float64, error)
• type Aggregation
• type Condition
• type Filter
• type GroupBy
• type Operator
• type Plan
• type Planner
  • func NewPlanner(im *index.IndexManager) *Planner
  • func (p *Planner) OptimizeQuery(ctx context.Context, q *Query) (*Plan, error)
• type Query
  • func NewQuery(records []arrow.Record) *Query
  • func (q *Query) Aggregate(column string, agg Aggregation) *Query
  • func (q *Query) Desc(desc bool) *Query
  • func (q *Query) Execute() (*QueryResult, error)
  • func (q *Query) GroupByColumns(columns ...string) *Query
  • func (q *Query) Having(column string, op Operator, value interface{}) *Query
  • func (q *Query) Limit(limit int64) *Query
  • func (q *Query) OrderBy(columns ...string) *Query
  • func (q *Query) Select(columns ...string) *Query
  • func (q *Query) Where(column string, op Operator, value interface{}) *Query
  • func (q *Query) WithWindow(window time.Duration) *Query
• type QueryResult
• type StreamingQuery
  • func NewStreamingQuery(q *Query, p *Planner) (*StreamingQuery, error)
  • func (sq *StreamingQuery) Start(ctx context.Context, db *db.DB) error
• type Window

Functions

func GroupByUserWindow

func GroupByUserWindow(records []arrow.Record, window time.Duration) (map[int64]float64, error)

GroupByUserWindow performs a query on Arrow records to compute SUM(purchase_amount) grouped by user_id for records with a timestamp within the given window.

Types

type Aggregation

type Aggregation string

Aggregation represents a SQL aggregation function

const (
	Sum   Aggregation = "SUM"
	Avg   Aggregation = "AVG"
	Count Aggregation = "COUNT"
	Min   Aggregation = "MIN"
	Max   Aggregation = "MAX"
)

type Condition

type Condition struct {
	Column   string
	Operator Operator
	Value    interface{}
}

Condition represents a WHERE clause condition

type Filter

type Filter struct {
	Column   string
	Operator Operator // Use the existing Operator type
	Value    interface{}
}

Filter represents a query filter condition

type GroupBy

type GroupBy struct {
	Columns []string
	Having  *Condition
}

GroupBy represents a GROUP BY clause

type Operator

type Operator string

Operator represents a SQL comparison operator

const (
	Eq  Operator = "="
	Neq Operator = "!="
	Gt  Operator = ">"
	Lt  Operator = "<"
	Gte Operator = ">="
	Lte Operator = "<="
)

type Plan

type Plan struct {
	ColumnPruning   []string      // Columns needed
	Filters         []Filter      // Push-down filters
	Aggregations    []Aggregation // Early aggregations
	IndexStrategies map[string]index.Strategy
}

Plan represents a query execution plan

type Planner

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

Planner optimizes query execution

func NewPlanner

func NewPlanner(im *index.IndexManager) *Planner

NewPlanner creates a query planner

func (*Planner) OptimizeQuery

func (p *Planner) OptimizeQuery(ctx context.Context, q *Query) (*Plan, error)

OptimizeQuery creates an execution plan for the query

type Query

type Query struct {
	Columns    []string               // Columns to select
	Aggregates map[string]Aggregation // Column -> Aggregation type
	GroupBy    *GroupBy
	Window     *Window
	// contains filtered or unexported fields
}

Query represents a SQL-like query structure

func NewQuery

func NewQuery(records []arrow.Record) *Query

NewQuery creates a new Query builder

func (*Query) Aggregate

func (q *Query) Aggregate(column string, agg Aggregation) *Query

Aggregate adds an aggregation

func (*Query) Desc

func (q *Query) Desc(desc bool) *Query

Desc sets descending order

func (*Query) Execute

func (q *Query) Execute() (*QueryResult, error)

Execute runs the query and returns the results

func (*Query) GroupByColumns

func (q *Query) GroupByColumns(columns ...string) *Query

GroupByColumns adds group by columns

func (*Query) Having

func (q *Query) Having(column string, op Operator, value interface{}) *Query

Having adds a having condition

func (*Query) Limit

func (q *Query) Limit(limit int64) *Query

Limit sets the limit

func (*Query) OrderBy

func (q *Query) OrderBy(columns ...string) *Query

OrderBy adds order by columns

func (*Query) Select

func (q *Query) Select(columns ...string) *Query

Select adds columns to select

func (*Query) Where

func (q *Query) Where(column string, op Operator, value interface{}) *Query

Where adds a where condition

func (*Query) WithWindow

func (q *Query) WithWindow(window time.Duration) *Query

WithWindow sets the time window

type QueryResult

type QueryResult struct {
	Records []arrow.Record
	Groups  map[string]interface{}
}

QueryResult holds the result of a query execution

type StreamingQuery

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

StreamingQuery represents a continuous query

func NewStreamingQuery

func NewStreamingQuery(q *Query, p *Planner) (*StreamingQuery, error)

NewStreamingQuery creates a new streaming query

func (*StreamingQuery) Start

func (sq *StreamingQuery) Start(ctx context.Context, db *db.DB) error

Start begins executing the streaming query

type Window

type Window struct {
	Duration time.Duration
	Offset   time.Duration
}