Implement basic execution pipeline with Sequence Scanner

engine/executor/executor.go

@@ -0,0 +1,51 @@
+package executor
+
+import (
+	"context"
+
+	"github.com/azzz/strata/engine/plan"
+	"github.com/azzz/strata/engine/storage"
+	"github.com/azzz/strata/engine/types"
+)
+
+// Executor builds execution pipeline using Plan.
+// Basic execution pipeline looks like:
+// Limit => Offset => Filter => SeqScan => StorageScanner
+type Executor struct {
+	Plan    plan.PhysicalPlan
+	Schema  types.Schema
+	Storage storage.Storage
+}
+
+func (e Executor) Exec(ctx context.Context) Operator {
+	requests := make([]storage.ScanRequest, 0, len(e.Plan.Scans))
+
+	for _, scan := range e.Plan.Scans {
+		req := storage.ScanRequest{
+			URI:    storage.URI(scan.URI),
+			Format: scan.Format,
+			Schema: e.Schema,
+		}
+
+		requests = append(requests, req)
+	}
+
+	var root Operator = &SeqScan{
+		requests: requests,
+		storage:  e.Storage,
+	}
+
+	if e.Plan.Filter != nil {
+		root = NewFilter(root, e.Plan.Filter)
+	}
+
+	if e.Plan.Offset > 0 {
+		root = NewOffset(root, e.Plan.Offset)
+	}
+
+	if e.Plan.Limit > 9 {
+		root = NewLimit(root, e.Plan.Limit)
+	}
+
+	return root
+}

engine/executor/filter.go

@@ -0,0 +1,54 @@
+package executor
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/azzz/strata/engine/expr"
+	"github.com/azzz/strata/engine/types"
+)
+
+type Filter struct {
+	input  Operator
+	filter expr.Filter
+
+	row types.Row
+	err error
+}
+
+func NewFilter(input Operator, filter expr.Filter) *Filter {
+	return &Filter{
+		input:  input,
+		filter: filter,
+	}
+}
+
+func (f *Filter) Row() types.Row { return f.row }
+func (f *Filter) Err() error     { return f.err }
+func (f *Filter) Close() error   { return f.input.Close() }
+
+func (f *Filter) Next(ctx context.Context) bool {
+	for f.input.Next(ctx) {
+		row := f.input.Row()
+
+		ok, err := f.filter.Match(row)
+		if err != nil {
+			f.err = fmt.Errorf("failed to apply filter: %w", err)
+			return false
+		}
+
+		if !ok {
+			continue
+		}
+
+		f.row = row
+		return true
+	}
+
+	if err := f.input.Err(); err != nil {
+		f.err = f.input.Err()
+		return false
+	}
+
+	return false
+}

engine/executor/filter_test.go

@@ -0,0 +1,118 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/azzz/strata/engine/expr"
+	"github.com/azzz/strata/engine/types"
+	"github.com/azzz/strata/engine/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type mockFilter struct {
+	match func(types.Row) (bool, error)
+	calls int
+}
+
+func (m *mockFilter) Match(row types.Row) (bool, error) {
+	m.calls++
+	return m.match(row)
+}
+
+func TestFilter(t *testing.T) {
+	newInput := func() *MockOperator {
+		return new(MockOperator).
+			addRow(newRow(types.NewInt64Value(4))).
+			addRow(newRow(types.NewInt64Value(8))).
+			addRow(newRow(types.NewInt64Value(15)))
+	}
+
+	t.Run("all rows match", func(t *testing.T) {
+		input := newInput()
+
+		op := NewFilter(input, &expr.GreaterThan{Col: 0, Value: types.NewInt64Value(0)})
+		retrieved, err := utils.Collect(context.Background(), op)
+
+		require.NoError(t, err)
+		assert.Equal(t,
+			[]types.Row{
+				newRow(types.NewInt64Value(4)),
+				newRow(types.NewInt64Value(8)),
+				newRow(types.NewInt64Value(15)),
+			}, retrieved)
+	})
+
+	t.Run("non matching rows are skipped", func(t *testing.T) {
+		input := newInput()
+
+		op := NewFilter(input, &expr.GreaterThan{Col: 0, Value: types.NewInt64Value(8)})
+		retrieved, err := utils.Collect(context.Background(), op)
+
+		require.NoError(t, err)
+		assert.Equal(t,
+			[]types.Row{
+				newRow(types.NewInt64Value(15)),
+			}, retrieved)
+	})
+
+	t.Run("no rows match", func(t *testing.T) {
+		input := newInput()
+
+		op := NewFilter(input, &expr.GreaterThan{Col: 0, Value: types.NewInt64Value(20)})
+		retrieved, err := utils.Collect(context.Background(), op)
+
+		require.NoError(t, err)
+		assert.Empty(t, retrieved)
+	})
+
+	t.Run("upstream error is returned with collected rows", func(t *testing.T) {
+		expectedErr := errors.New("boom")
+		input := new(MockOperator).
+			addRow(newRow(types.NewInt64Value(4))).
+			addRow(newRow(types.NewInt64Value(8))).
+			addErr(expectedErr)
+
+		op := NewFilter(input, &expr.GreaterThan{Col: 0, Value: types.NewInt64Value(0)})
+		retrieved, err := utils.Collect(context.Background(), op)
+
+		require.ErrorIs(t, err, expectedErr)
+		assert.Equal(t,
+			[]types.Row{
+				newRow(types.NewInt64Value(4)),
+				newRow(types.NewInt64Value(8)),
+			}, retrieved)
+	})
+
+	t.Run("filter error stops scan immediately", func(t *testing.T) {
+		expectedErr := errors.New("boom")
+		input := newInput()
+		filter := &mockFilter{
+			match: func(row types.Row) (bool, error) {
+				value, ok := row.Get(0)
+				if !ok {
+					return false, errors.New("missing value")
+				}
+
+				if value == types.NewInt64Value(8) {
+					return false, expectedErr
+				}
+
+				return true, nil
+			},
+		}
+
+		op := NewFilter(input, filter)
+		retrieved, err := utils.Collect(context.Background(), op)
+
+		require.ErrorIs(t, err, expectedErr)
+		assert.Equal(t,
+			[]types.Row{
+				newRow(types.NewInt64Value(4)),
+			}, retrieved)
+		assert.Equal(t, 2, input.cursor, "filter should stop reading upstream after evaluation error")
+		assert.Equal(t, 2, filter.calls)
+	})
+}

engine/executor/limit.go

@@ -0,0 +1,43 @@
+package executor
+
+import (
+	"context"
+
+	"github.com/azzz/strata/engine/types"
+)
+
+type Limit struct {
+	input Operator
+
+	limit   int
+	emitted int
+
+	row types.Row
+	err error
+}
+
+func NewLimit(input Operator, limit int) *Limit {
+	return &Limit{
+		input: input,
+		limit: limit,
+	}
+}
+func (l *Limit) Row() types.Row { return l.row }
+func (l *Limit) Err() error     { return l.err }
+func (l *Limit) Close() error   { return l.input.Close() }
+
+func (l *Limit) Next(ctx context.Context) bool {
+	if l.limit > 0 && l.emitted >= l.limit {
+		return false
+	}
+
+	if !l.input.Next(ctx) {
+		l.err = l.input.Err()
+		return false
+	}
+
+	l.row = l.input.Row()
+	l.emitted++
+
+	return true
+}