-
Notifications
You must be signed in to change notification settings - Fork 0
/
madis.go
56 lines (47 loc) · 1.27 KB
/
madis.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
package master
import (
"github.com/benmizrahi/gobig/internal/protos"
"github.com/google/uuid"
"google.golang.org/protobuf/types/known/timestamppb"
)
type Mafream struct {
columns []string
plan []*protos.IPartition
context *Context
}
func NewDataFrame(c *Context, columns []string, numPartitions int) *Mafream {
partitions := make([]*protos.IPartition, numPartitions)
for i := 0; i < numPartitions; i++ {
partitions[i] = &protos.IPartition{}
}
return &Mafream{
columns: columns,
plan: partitions,
context: c,
}
}
func (w *Mafream) Show() *Mafream {
actions := []string{protos.TAKE, protos.LIMIT}
w.assignActions(actions)
results := w.context.DoAction(w.plan)
w.handleTasksResults(actions, results)
return w
}
func (w *Mafream) Count() *Mafream {
actions := []string{protos.COUNT}
w.assignActions([]string{protos.COUNT})
results := w.context.DoAction(w.plan)
w.handleTasksResults(actions, results)
return w
}
func (w *Mafream) assignActions(actions []string) {
for _, partition := range w.plan {
partition.Tasks = append(partition.Tasks, &protos.Task{
Uuid: uuid.New().String(),
Instactions: actions,
CreationTime: timestamppb.Now(),
})
}
}
func (w *Mafream) handleTasksResults(actions []string, res []*protos.IPartitionResult) {
}