Repository navigation
Expand file tree
/
Copy pathtable_buffer.go
More file actions
93 lines (81 loc) · 2.21 KB
/
Copy pathtable_buffer.go
File metadata and controls
93 lines (81 loc) · 2.21 KB
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
package arrow
import (
"github.com/apache/arrow/go/arrow/array"
"github.com/influxdata/flux"
"github.com/influxdata/flux/codes"
"github.com/influxdata/flux/internal/errors"
)
// TableBuffer represents the buffered component of an arrow table.
//
// TableBuffer is a low-level structure for creating
// a table that implements the flux.ColReader interface.
// It does not have very many guiding blocks to ensure it is
// used correctly.
//
// A valid TableBuffer will have a number of columns that
// is equal in length to the number of values arrays.
// All of the values arrays will have the same length.
type TableBuffer struct {
GroupKey flux.GroupKey
Columns []flux.ColMeta
Values []array.Interface
}
var _ flux.ColReader = (*TableBuffer)(nil)
func (t *TableBuffer) Len() int {
if len(t.Columns) == 0 {
return 0
}
return t.Values[0].Len()
}
func (t *TableBuffer) Bools(j int) *array.Boolean {
return t.Values[j].(*array.Boolean)
}
func (t *TableBuffer) Ints(j int) *array.Int64 {
return t.Values[j].(*array.Int64)
}
func (t *TableBuffer) UInts(j int) *array.Uint64 {
return t.Values[j].(*array.Uint64)
}
func (t *TableBuffer) Floats(j int) *array.Float64 {
return t.Values[j].(*array.Float64)
}
func (t *TableBuffer) Strings(j int) *array.Binary {
return t.Values[j].(*array.Binary)
}
func (t *TableBuffer) Times(j int) *array.Int64 {
return t.Values[j].(*array.Int64)
}
func (t *TableBuffer) Retain() {
for _, vs := range t.Values {
vs.Retain()
}
}
func (t *TableBuffer) Release() {
for _, vs := range t.Values {
vs.Release()
}
}
func (t *TableBuffer) Key() flux.GroupKey {
return t.GroupKey
}
func (t *TableBuffer) Cols() []flux.ColMeta {
return t.Columns
}
// Validate will validate that this TableBuffer has the
// proper structure.
func (t *TableBuffer) Validate() error {
if len(t.Columns) != len(t.Values) {
return errors.Newf(codes.Internal, "mismatched number of columns and arrays: %d != %d", len(t.Columns), len(t.Values))
}
// If a table has no columns, do not validate the length.
if len(t.Columns) == 0 {
return nil
}
sz := t.Values[0].Len()
for i := 1; i < len(t.Values); i++ {
if t.Values[i].Len() != sz {
return errors.New(codes.Internal, "column size mismatch")
}
}
return nil
}