From 4c6cbd41001c3a1915cd50a49404d60af196837f Mon Sep 17 00:00:00 2001 From: Yuriy Tsemashko Date: Mon, 29 Jun 2026 22:49:18 +0300 Subject: [PATCH] Implement basic execution pipeline with Sequence Scanner --- engine/executor/.keep | 0 engine/executor/executor.go | 51 +++++++++ engine/executor/filter.go | 54 ++++++++++ engine/executor/filter_test.go | 118 ++++++++++++++++++++ engine/executor/limit.go | 43 ++++++++ engine/executor/limit_test.go | 137 +++++++++++++++++++++++ engine/executor/mock_test.go | 70 ++++++++++++ engine/executor/offset.go | 47 ++++++++ engine/executor/offset_test.go | 114 ++++++++++++++++++++ engine/executor/operator.go | 14 +++ engine/executor/seq_scan.go | 77 +++++++++++++ engine/executor/seq_scan_test.go | 180 +++++++++++++++++++++++++++++++ engine/plan/.keep | 0 engine/plan/physical_plan.go | 10 ++ engine/plan/scan.go | 9 ++ engine/utils/colect.go | 26 +++++ 16 files changed, 950 insertions(+) delete mode 100644 engine/executor/.keep create mode 100644 engine/executor/executor.go create mode 100644 engine/executor/filter.go create mode 100644 engine/executor/filter_test.go create mode 100644 engine/executor/limit.go create mode 100644 engine/executor/limit_test.go create mode 100644 engine/executor/mock_test.go create mode 100644 engine/executor/offset.go create mode 100644 engine/executor/offset_test.go create mode 100644 engine/executor/operator.go create mode 100644 engine/executor/seq_scan.go create mode 100644 engine/executor/seq_scan_test.go delete mode 100644 engine/plan/.keep create mode 100644 engine/plan/physical_plan.go create mode 100644 engine/plan/scan.go create mode 100644 engine/utils/colect.go diff --git a/engine/executor/.keep b/engine/executor/.keep deleted file mode 100644 index e69de29..0000000 diff --git a/engine/executor/executor.go b/engine/executor/executor.go new file mode 100644 index 0000000..60ecf56 --- /dev/null +++ b/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 +} diff --git a/engine/executor/filter.go b/engine/executor/filter.go new file mode 100644 index 0000000..7c2a82c --- /dev/null +++ b/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 +} diff --git a/engine/executor/filter_test.go b/engine/executor/filter_test.go new file mode 100644 index 0000000..de72d4f --- /dev/null +++ b/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) + }) +} diff --git a/engine/executor/limit.go b/engine/executor/limit.go new file mode 100644 index 0000000..f0b1222 --- /dev/null +++ b/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 +} diff --git a/engine/executor/limit_test.go b/engine/executor/limit_test.go new file mode 100644 index 0000000..fa546a9 --- /dev/null +++ b/engine/executor/limit_test.go @@ -0,0 +1,137 @@ +package executor + +import ( + "context" + "errors" + "testing" + + "github.com/azzz/strata/engine/types" + "github.com/azzz/strata/engine/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLimit(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("no limit", func(t *testing.T) { + input := newInput() + + op := NewLimit(input, 0) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Len(t, retrieved, 3) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + newRow(types.NewInt64Value(15)), + }, retrieved) + }) + + t.Run("limit is higher than number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewLimit(input, 5) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Len(t, retrieved, 3) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + newRow(types.NewInt64Value(15)), + }, retrieved) + }) + + t.Run("limit is equal to number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewLimit(input, 3) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Len(t, retrieved, 3) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + newRow(types.NewInt64Value(15)), + }, retrieved) + assert.Equal(t, 3, input.cursor) + }) + + t.Run("limit is lower than number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewLimit(input, 2) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Len(t, retrieved, 2) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + }, retrieved) + assert.Equal(t, 2, input.cursor, "limit should stop reading upstream once enough rows were emitted") + }) + + t.Run("next after reaching the limit does not read upstream", func(t *testing.T) { + input := newInput() + + op := NewLimit(input, 2) + + require.True(t, op.Next(context.Background())) + assert.Equal(t, newRow(types.NewInt64Value(4)), op.Row()) + require.True(t, op.Next(context.Background())) + assert.Equal(t, newRow(types.NewInt64Value(8)), op.Row()) + require.False(t, op.Next(context.Background())) + require.False(t, op.Next(context.Background())) + + assert.NoError(t, op.Err()) + assert.Equal(t, 2, input.cursor, "limit should not read upstream after reaching the bound") + }) + + t.Run("upstream error before reaching the limit is returned", func(t *testing.T) { + expectedErr := errors.New("boom") + input := new(MockOperator). + addRow(newRow(types.NewInt64Value(4))). + addErr(expectedErr) + + op := NewLimit(input, 5) + retrieved, err := utils.Collect(context.Background(), op) + + require.ErrorIs(t, err, expectedErr) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + }, retrieved) + }) + + t.Run("upstream error after reaching the limit is ignored", func(t *testing.T) { + expectedErr := errors.New("boom") + input := new(MockOperator). + addRow(newRow(types.NewInt64Value(4))). + addRow(newRow(types.NewInt64Value(8))). + addErr(expectedErr) + + op := NewLimit(input, 2) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + }, retrieved) + assert.Equal(t, 2, input.cursor, "limit should not read rows beyond the configured bound") + }) +} diff --git a/engine/executor/mock_test.go b/engine/executor/mock_test.go new file mode 100644 index 0000000..d471def --- /dev/null +++ b/engine/executor/mock_test.go @@ -0,0 +1,70 @@ +package executor + +import ( + "context" + + "github.com/azzz/strata/engine/types" +) + +type MockOperator struct { + source []struct { + row types.Row + err error + } + cursor int + + row types.Row + err error +} + +func (m *MockOperator) addRow(row types.Row) *MockOperator { + m.source = append(m.source, + struct { + row types.Row + err error + }{row: row}, + ) + + return m +} + +func (m *MockOperator) addErr(err error) *MockOperator { + m.source = append(m.source, + struct { + row types.Row + err error + }{err: err}, + ) + + return m +} + +func (m *MockOperator) Row() types.Row { return m.row } +func (m *MockOperator) Err() error { return m.err } +func (m *MockOperator) Close() error { return nil } + +func (m *MockOperator) Next(ctx context.Context) bool { + for m.cursor < len(m.source) { + fetcher := m.source[m.cursor] + row, err := fetcher.row, fetcher.err + if err != nil { + m.err = err + return false + } else { + m.row = row + m.cursor++ + return true + } + } + + return false +} + +func newRow(values ...types.Value) types.Row { + row := types.NewRow(len(values)) + for i, v := range values { + row.Set(i, v) + } + + return row +} diff --git a/engine/executor/offset.go b/engine/executor/offset.go new file mode 100644 index 0000000..05289ed --- /dev/null +++ b/engine/executor/offset.go @@ -0,0 +1,47 @@ +package executor + +import ( + "context" + + "github.com/azzz/strata/engine/types" +) + +type Offset struct { + input Operator + + offset int + skipped int + + row types.Row + err error +} + +func NewOffset(input Operator, offset int) *Offset { + return &Offset{ + input: input, + offset: offset, + } +} + +func (o *Offset) Row() types.Row { return o.row } +func (o *Offset) Err() error { return o.err } +func (o *Offset) Close() error { return o.input.Close() } + +func (o *Offset) Next(ctx context.Context) bool { + for o.skipped < o.offset { + if !o.input.Next(ctx) { + o.err = o.input.Err() + return false + } + + o.skipped++ + } + + if !o.input.Next(ctx) { + o.err = o.input.Err() + return false + } + + o.row = o.input.Row() + return true +} diff --git a/engine/executor/offset_test.go b/engine/executor/offset_test.go new file mode 100644 index 0000000..a625e7a --- /dev/null +++ b/engine/executor/offset_test.go @@ -0,0 +1,114 @@ +package executor + +import ( + "context" + "errors" + "testing" + + "github.com/azzz/strata/engine/types" + "github.com/azzz/strata/engine/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestOffset(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("no offset", func(t *testing.T) { + input := newInput() + + op := NewOffset(input, 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("offset is higher than number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewOffset(input, 5) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Empty(t, retrieved) + assert.Equal(t, 3, input.cursor) + }) + + t.Run("offset is equal to number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewOffset(input, 3) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Empty(t, retrieved) + assert.Equal(t, 3, input.cursor) + }) + + t.Run("offset is lower than number of rows in the source", func(t *testing.T) { + input := newInput() + + op := NewOffset(input, 2) + retrieved, err := utils.Collect(context.Background(), op) + + require.NoError(t, err) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(15)), + }, retrieved) + assert.Equal(t, 3, input.cursor) + }) + + t.Run("next after source is exhausted does not change state", func(t *testing.T) { + input := newInput() + + op := NewOffset(input, 3) + + require.False(t, op.Next(context.Background())) + require.False(t, op.Next(context.Background())) + + assert.NoError(t, op.Err()) + assert.Equal(t, 3, input.cursor, "offset should not read past the end of the source") + }) + + t.Run("upstream error while skipping rows is returned", func(t *testing.T) { + expectedErr := errors.New("boom") + input := new(MockOperator). + addRow(newRow(types.NewInt64Value(4))). + addErr(expectedErr) + + op := NewOffset(input, 2) + retrieved, err := utils.Collect(context.Background(), op) + + require.ErrorIs(t, err, expectedErr) + assert.Empty(t, retrieved) + }) + + t.Run("upstream error after offset 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 := NewOffset(input, 1) + retrieved, err := utils.Collect(context.Background(), op) + + require.ErrorIs(t, err, expectedErr) + assert.Equal(t, + []types.Row{ + newRow(types.NewInt64Value(8)), + }, retrieved) + }) +} diff --git a/engine/executor/operator.go b/engine/executor/operator.go new file mode 100644 index 0000000..e2c4daa --- /dev/null +++ b/engine/executor/operator.go @@ -0,0 +1,14 @@ +package executor + +import ( + "context" + + "github.com/azzz/strata/engine/types" +) + +type Operator interface { + Next(ctx context.Context) bool + Row() types.Row + Err() error + Close() error +} diff --git a/engine/executor/seq_scan.go b/engine/executor/seq_scan.go new file mode 100644 index 0000000..8ac6a42 --- /dev/null +++ b/engine/executor/seq_scan.go @@ -0,0 +1,77 @@ +package executor + +import ( + "context" + "fmt" + + "github.com/azzz/strata/engine/storage" + "github.com/azzz/strata/engine/types" +) + +type SeqScan struct { + storage storage.Storage + requests []storage.ScanRequest + + current storage.Scanner + idx int + + row types.Row + err error +} + +func NewSeqScan(storage storage.Storage, requests []storage.ScanRequest) *SeqScan { + return &SeqScan{ + storage: storage, + requests: requests, + } +} + +func (s *SeqScan) Next(ctx context.Context) bool { + for s.idx < len(s.requests) { + if s.current == nil { + scanner, err := s.storage.NewScanner(ctx, s.requests[s.idx]) + if err != nil { + s.err = fmt.Errorf("failed to open scanner (%d): %w", s.idx, err) + return false + } + + if err := scanner.Open(); err != nil { + s.err = fmt.Errorf("failed to open scanner (%d): %w", s.idx, err) + return false + } + + s.current = scanner + } + + // everything is good + if s.current.Next(ctx) { + s.row = s.current.Row() + return true + } + + // handle current scanner error + if err := s.current.Err(); err != nil { + s.err = fmt.Errorf("scanner failed (%d): %w", s.idx, err) + return false + } + + // no error? Scanner finished correctly => close current scanner + if err := s.current.Close(); err != nil { + s.err = fmt.Errorf("failed to close current scanner (%d): %w", s.idx, err) + return false + } + + // update cursor + s.idx++ + s.current = nil + } + + return false +} + +func (r *SeqScan) Row() types.Row { return r.row } +func (r *SeqScan) Err() error { return r.err } + +func (r *SeqScan) Close() error { + return r.current.Close() +} diff --git a/engine/executor/seq_scan_test.go b/engine/executor/seq_scan_test.go new file mode 100644 index 0000000..c90b365 --- /dev/null +++ b/engine/executor/seq_scan_test.go @@ -0,0 +1,180 @@ +package executor + +import ( + "context" + "errors" + "testing" + + "github.com/azzz/strata/engine/storage" + "github.com/azzz/strata/engine/types" + "github.com/azzz/strata/engine/utils" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type mockStorage struct { + scanners []storage.Scanner + errs []error + calls int + requests []storage.ScanRequest +} + +func (m *mockStorage) NewScanner(ctx context.Context, req storage.ScanRequest) (storage.Scanner, error) { + m.requests = append(m.requests, req) + + idx := m.calls + m.calls++ + + if idx < len(m.errs) && m.errs[idx] != nil { + return nil, m.errs[idx] + } + + return m.scanners[idx], nil +} + +type mockScanner struct { + rows []types.Row + nextErr error + openErr error + closeErr error + cursor int + openCalls int + closeCalls int + row types.Row + err error +} + +func (m *mockScanner) Open() error { + m.openCalls++ + return m.openErr +} + +func (m *mockScanner) Next(ctx context.Context) bool { + if m.cursor >= len(m.rows) { + if m.nextErr != nil { + m.err = m.nextErr + } + return false + } + + m.row = m.rows[m.cursor] + m.cursor++ + + return true +} + +func (m *mockScanner) Err() error { return m.err } +func (m *mockScanner) Row() types.Row { return m.row } + +func (m *mockScanner) Close() error { + m.closeCalls++ + return m.closeErr +} + +func TestSeqScan(t *testing.T) { + requests := []storage.ScanRequest{ + {URI: "first.jsonl", Format: storage.FormatJSONL}, + {URI: "second.jsonl", Format: storage.FormatJSONL}, + } + + t.Run("reads rows from all requests sequentially", func(t *testing.T) { + first := &mockScanner{ + rows: []types.Row{ + newRow(types.NewInt64Value(4)), + newRow(types.NewInt64Value(8)), + }, + } + second := &mockScanner{ + rows: []types.Row{ + newRow(types.NewInt64Value(15)), + }, + } + input := &mockStorage{scanners: []storage.Scanner{first, second}} + + op := NewSeqScan(input, requests) + 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) + assert.Equal(t, requests, input.requests) + assert.Equal(t, 1, first.openCalls) + assert.Equal(t, 1, second.openCalls) + assert.Equal(t, 1, first.closeCalls) + assert.Equal(t, 1, second.closeCalls) + }) + + t.Run("new scanner error is returned", func(t *testing.T) { + expectedErr := errors.New("boom") + input := &mockStorage{ + errs: []error{expectedErr}, + } + + op := NewSeqScan(input, requests[:1]) + retrieved, err := utils.Collect(context.Background(), op) + + require.ErrorIs(t, err, expectedErr) + assert.Empty(t, retrieved) + assert.Equal(t, 1, input.calls) + }) + + t.Run("open error is returned", func(t *testing.T) { + expectedErr := errors.New("boom") + scanner := &mockScanner{openErr: expectedErr} + input := &mockStorage{scanners: []storage.Scanner{scanner}} + + op := NewSeqScan(input, requests[:1]) + retrieved, err := utils.Collect(context.Background(), op) + + require.ErrorIs(t, err, expectedErr) + assert.Empty(t, retrieved) + assert.Equal(t, 1, scanner.openCalls) + assert.Equal(t, 0, scanner.closeCalls) + }) + + t.Run("scanner error is returned with collected rows", func(t *testing.T) { + expectedErr := errors.New("boom") + scanner := &mockScanner{ + rows: []types.Row{ + newRow(types.NewInt64Value(4)), + }, + nextErr: expectedErr, + } + input := &mockStorage{scanners: []storage.Scanner{scanner}} + + op := NewSeqScan(input, requests[:1]) + 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, 0, scanner.closeCalls) + }) + + t.Run("close error is returned after scanner is exhausted", func(t *testing.T) { + expectedErr := errors.New("boom") + scanner := &mockScanner{ + rows: []types.Row{ + newRow(types.NewInt64Value(4)), + }, + closeErr: expectedErr, + } + input := &mockStorage{scanners: []storage.Scanner{scanner}} + + op := NewSeqScan(input, requests[:1]) + 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, 1, scanner.closeCalls) + }) +} diff --git a/engine/plan/.keep b/engine/plan/.keep deleted file mode 100644 index e69de29..0000000 diff --git a/engine/plan/physical_plan.go b/engine/plan/physical_plan.go new file mode 100644 index 0000000..cfd12e9 --- /dev/null +++ b/engine/plan/physical_plan.go @@ -0,0 +1,10 @@ +package plan + +import "github.com/azzz/strata/engine/expr" + +type PhysicalPlan struct { + Scans []Scan + Filter expr.Filter + Limit int + Offset int +} diff --git a/engine/plan/scan.go b/engine/plan/scan.go new file mode 100644 index 0000000..05c4597 --- /dev/null +++ b/engine/plan/scan.go @@ -0,0 +1,9 @@ +package plan + +import "github.com/azzz/strata/engine/storage" + +type Scan struct { + URI string + Format storage.Format + Columns []int +} diff --git a/engine/utils/colect.go b/engine/utils/colect.go new file mode 100644 index 0000000..ff45172 --- /dev/null +++ b/engine/utils/colect.go @@ -0,0 +1,26 @@ +package utils + +import ( + "context" + + "github.com/azzz/strata/engine/types" +) + +type RowScanner interface { + Next(context.Context) bool + Row() types.Row + Err() error +} + +func Collect(ctx context.Context, sc RowScanner) ([]types.Row, error) { + result := make([]types.Row, 0, 100) + for sc.Next(ctx) { + result = append(result, sc.Row()) + } + + if err := sc.Err(); err != nil { + return result, err + } + + return result, nil +}