Import FieldValues

This PR adds the ability to import field values into RangeEnabled frames
using the `pilosa import` command.

Example:
```
pilosa import -i i -f f --field foo sample-vals.csv
```
imports data from sample-vals.csv, which contains data in the format:
```
[ColumnID, Value]
```

Also fixes a bug where `frame.rangeEnabled` was not being set on frame creation.
This commit is contained in:
Travis 2017-09-13 21:59:29 -05:00
parent 1fb540f40b
commit 01fa368c6a
No known key found for this signature in database
GPG key ID: 7F08008DFD9314C9
9 changed files with 867 additions and 43 deletions

139
client.go
View file

@ -415,6 +415,94 @@ func (c *Client) importNode(ctx context.Context, node *Node, buf []byte) error {
return nil
}
// ImportValue bulk imports field values for a single slice to a host.
func (c *Client) ImportValue(ctx context.Context, index, frame, field string, slice uint64, vals []FieldValue) error {
if index == "" {
return ErrIndexRequired
} else if frame == "" {
return ErrFrameRequired
}
buf, err := MarshalImportValuePayload(index, frame, field, slice, vals)
if err != nil {
return fmt.Errorf("Error Creating Payload: %s", err)
}
// Retrieve a list of nodes that own the slice.
nodes, err := c.FragmentNodes(ctx, index, slice)
if err != nil {
return fmt.Errorf("slice nodes: %s", err)
}
// Import to each node.
for _, node := range nodes {
if err := c.importValueNode(ctx, node, buf); err != nil {
return fmt.Errorf("import node: host=%s, err=%s", node.Host, err)
}
}
return nil
}
// MarshalImportValuePayload marshalls the import parameters into a protobuf byte slice.
func MarshalImportValuePayload(index, frame, field string, slice uint64, vals []FieldValue) ([]byte, error) {
// Separate row and column IDs to reduce allocations.
columnIDs := FieldValues(vals).ColumnIDs()
values := FieldValues(vals).Values()
// Marshal bits to protobufs.
buf, err := proto.Marshal(&internal.ImportValueRequest{
Index: index,
Frame: frame,
Slice: slice,
Field: field,
ColumnIDs: columnIDs,
Values: values,
})
if err != nil {
return nil, fmt.Errorf("marshal import request: %s", err)
}
return buf, nil
}
// importValueNode sends a pre-marshaled import request to a node.
func (c *Client) importValueNode(ctx context.Context, node *Node, buf []byte) error {
// Create URL & HTTP request.
u := url.URL{Scheme: "http", Host: node.Host, Path: "/import-value"}
req, err := http.NewRequest("POST", u.String(), bytes.NewReader(buf))
if err != nil {
return err
}
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("Accept", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+Version)
// Execute request against the host.
resp, err := c.HTTPClient.Do(req.WithContext(ctx))
if err != nil {
return err
}
defer resp.Body.Close()
// Read body and unmarshal response.
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
} else if resp.StatusCode != http.StatusOK {
return errors.New(string(body))
}
var isresp internal.ImportResponse
if err := proto.Unmarshal(body, &isresp); err != nil {
return fmt.Errorf("unmarshal import response: %s", err)
} else if s := isresp.Err; s != "" {
return errors.New(s)
}
return nil
}
// ExportCSV bulk exports data for a single slice from a host to CSV format.
func (c *Client) ExportCSV(ctx context.Context, index, frame, view string, slice uint64, w io.Writer) error {
if index == "" {
@ -1090,6 +1178,57 @@ func (p Bits) GroupBySlice() map[uint64][]Bit {
return m
}
// FieldValues represents the value for a column within a
// range-encoded frame.
type FieldValue struct {
ColumnID uint64
Value uint64
}
// FieldValues represents a slice of field values.
type FieldValues []FieldValue
func (p FieldValues) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
func (p FieldValues) Len() int { return len(p) }
func (p FieldValues) Less(i, j int) bool {
return p[i].ColumnID < p[j].ColumnID
}
// ColumnIDs returns a slice of all the column IDs.
func (p FieldValues) ColumnIDs() []uint64 {
other := make([]uint64, len(p))
for i := range p {
other[i] = p[i].ColumnID
}
return other
}
// Values returns a slice of all the values.
func (p FieldValues) Values() []uint64 {
other := make([]uint64, len(p))
for i := range p {
other[i] = p[i].Value
}
return other
}
// GroupBySlice returns a map of field values by slice.
func (p FieldValues) GroupBySlice() map[uint64][]FieldValue {
m := make(map[uint64][]FieldValue)
for _, val := range p {
slice := val.ColumnID / SliceWidth
m[slice] = append(m[slice], val)
}
for slice, vals := range m {
sort.Sort(FieldValues(vals))
m[slice] = vals
}
return m
}
// BitsByPos represents a slice of bits sorted by internal position.
type BitsByPos []Bit

View file

@ -55,6 +55,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
flags.StringVarP(&Importer.Host, "host", "", "localhost:10101", "host:port of Pilosa.")
flags.StringVarP(&Importer.Index, "index", "i", "", "Pilosa index to import into.")
flags.StringVarP(&Importer.Frame, "frame", "f", "", "Frame to import into.")
flags.StringVarP(&Importer.Field, "field", "", "", "Field to import into.")
flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.")
flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.")
flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.")

View file

@ -45,6 +45,9 @@ type ImportCommand struct {
// CreateSchema ensures the schema exists before import
CreateSchema bool
// For Range-Encoded fields, name of the Field to import into.
Field string `json:"field"`
// Filenames to import from.
Paths []string `json:"paths"`
@ -122,6 +125,16 @@ func (cmd *ImportCommand) ensureSchema(ctx context.Context) error {
// importPath parses a path into bits and imports it to the server.
func (cmd *ImportCommand) importPath(ctx context.Context, path string) error {
// If a field is provided, treat the import data as values to be range-encoded.
if cmd.Field != "" {
return cmd.bufferFieldValues(ctx, path)
} else {
return cmd.bufferBits(ctx, path)
}
}
// bufferBits buffers slices of bits to be imported as a batch.
func (cmd *ImportCommand) bufferBits(ctx context.Context, path string) error {
a := make([]pilosa.Bit, 0, cmd.BufferSize)
var r *csv.Reader
@ -204,7 +217,7 @@ func (cmd *ImportCommand) importPath(ctx context.Context, path string) error {
return nil
}
// importPath parses a path into bits and imports it to the server.
// importBits sends batches of bits to the server.
func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
@ -224,5 +237,104 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err
}
}
return nil
}
// bufferFieldValues buffers slices of fieldValues to be imported as a batch.
func (cmd *ImportCommand) bufferFieldValues(ctx context.Context, path string) error {
a := make([]pilosa.FieldValue, 0, cmd.BufferSize)
var r *csv.Reader
if path != "-" {
// Open file for reading.
f, err := os.Open(path)
if err != nil {
return err
}
defer f.Close()
// Read rows as bits.
r = csv.NewReader(f)
} else {
r = csv.NewReader(cmd.Stdin)
}
r.FieldsPerRecord = -1
rnum := 0
for {
rnum++
// Read CSV row.
record, err := r.Read()
if err == io.EOF {
break
} else if err != nil {
return err
}
// Ignore blank rows.
if record[0] == "" {
continue
} else if len(record) < 2 {
return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record))
}
var val pilosa.FieldValue
// Parse column id.
columnID, err := strconv.ParseUint(record[0], 10, 64)
if err != nil {
return fmt.Errorf("invalid column id on row %d: %q", rnum, record[0])
}
val.ColumnID = columnID
// Parse field value.
value, err := strconv.ParseUint(record[1], 10, 64)
if err != nil {
return fmt.Errorf("invalid value on row %d: %q", rnum, record[1])
}
val.Value = value
a = append(a, val)
// If we've reached the buffer size then import field values.
if len(a) == cmd.BufferSize {
if err := cmd.importFieldValues(ctx, a); err != nil {
return err
}
a = a[:0]
}
}
// If there are still values in the buffer then flush them.
if err := cmd.importFieldValues(ctx, a); err != nil {
return err
}
return nil
}
// importFieldValues sends batches of fieldValues to the server.
func (cmd *ImportCommand) importFieldValues(ctx context.Context, vals []pilosa.FieldValue) error {
logger := log.New(cmd.Stderr, "", log.LstdFlags)
// Group vals by slice.
logger.Printf("grouping %d vals", len(vals))
valsBySlice := pilosa.FieldValues(vals).GroupBySlice()
// Parse path into field values.
for slice, vals := range valsBySlice {
if cmd.Sort {
sort.Sort(pilosa.FieldValues(vals))
}
logger.Printf("importing slice: %d, n=%d", slice, len(vals))
if err := cmd.Client.ImportValue(ctx, cmd.Index, cmd.Frame, cmd.Field, slice, vals); err != nil {
return err
}
}
return nil
}

View file

@ -1259,6 +1259,31 @@ func (f *Fragment) Import(rowIDs, columnIDs []uint64) error {
return nil
}
// ImportValue bulk imports a set of range-encoded values.
func (f *Fragment) ImportValue(columnIDs, values []uint64, bitDepth uint) error {
// Verify that there are an equal number of column ids and values.
if len(columnIDs) != len(values) {
return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values))
}
// Process every value.
// If an error occurs then reopen the storage.
if err := func() error {
for i := range columnIDs {
columnID, value := columnIDs[i], values[i]
_, err := f.SetFieldValue(columnID, bitDepth, value)
if err != nil {
return err
}
}
return nil
}(); err != nil {
return err
}
return nil
}
// incrementOpN increase the operation count by one.
// If the count exceeds the maximum allowed then a snapshot is performed.
func (f *Fragment) incrementOpN() error {

View file

@ -425,7 +425,7 @@ func (f *Frame) CreateField(field *Field) error {
defer f.mu.Unlock()
// Ensure frame supports fields.
if f.rangeEnabled {
if !f.RangeEnabled() {
return ErrFrameFieldsNotAllowed
}
@ -445,7 +445,7 @@ func (f *Frame) DeleteField(name string) error {
defer f.mu.Unlock()
// Ensure frame supports fields.
if f.rangeEnabled {
if !f.RangeEnabled() {
return ErrFrameFieldsNotAllowed
}
@ -856,6 +856,58 @@ func (f *Frame) Import(rowIDs, columnIDs []uint64, timestamps []*time.Time) erro
return nil
}
// ImportValue bulk imports range-encoded value data.
func (f *Frame) ImportValue(fieldName string, columnIDs, values []uint64) error {
// Verify that this frame is range-encoded.
if !f.RangeEnabled() {
return fmt.Errorf("Frame not RangeEnabled: %s", f.name)
}
viewName := ViewFieldPrefix + fieldName
// Get the field so we know bitDepth.
field := f.Field(fieldName)
if field == nil {
return fmt.Errorf("Field does not exist: %s", fieldName)
}
// Split import data by fragment.
dataByFragment := make(map[importKey]importValueData)
for i := range columnIDs {
columnID, value := columnIDs[i], values[i]
// Attach value to each field view.
for _, name := range []string{viewName} {
key := importKey{View: name, Slice: columnID / SliceWidth}
data := dataByFragment[key]
data.ColumnIDs = append(data.ColumnIDs, columnID)
data.Values = append(data.Values, value)
dataByFragment[key] = data
}
}
// Import into each fragment.
for key, data := range dataByFragment {
// The view must already exist (i.e. we can't create it)
// because we need to know bitDepth (based on min/max value).
view, err := f.CreateViewIfNotExists(key.View)
if err != nil {
return err
}
frag, err := view.CreateFragmentIfNotExists(key.Slice)
if err != nil {
return err
}
if err := frag.ImportValue(data.ColumnIDs, data.Values, field.BitDepth()); err != nil {
return err
}
}
return nil
}
// encodeFrames converts a into its internal representation.
func encodeFrames(a []*Frame) []*internal.Frame {
other := make([]*internal.Frame, len(a))

View file

@ -106,6 +106,7 @@ func NewRouter(handler *Handler) *mux.Router {
router.HandleFunc("/fragment/data", handler.handlePostFragmentData).Methods("POST")
router.HandleFunc("/fragment/nodes", handler.handleGetFragmentNodes).Methods("GET")
router.HandleFunc("/import", handler.handlePostImport).Methods("POST")
router.HandleFunc("/import-value", handler.handlePostImportValue).Methods("POST")
router.HandleFunc("/index", handler.handleGetIndexes).Methods("GET")
router.HandleFunc("/index/{index}", handler.handleGetIndex).Methods("GET")
router.HandleFunc("/index/{index}", handler.handlePostIndex).Methods("POST")
@ -1173,6 +1174,76 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
w.Write(buf)
}
// handlePostImportValue handles /import-value requests.
func (h *Handler) handlePostImportValue(w http.ResponseWriter, r *http.Request) {
// Verify that request is only communicating over protobufs.
if r.Header.Get("Content-Type") != "application/x-protobuf" {
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
} else if r.Header.Get("Accept") != "application/x-protobuf" {
http.Error(w, "Not acceptable", http.StatusNotAcceptable)
return
}
// Read entire body.
body, err := ioutil.ReadAll(r.Body)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Marshal into request object.
var req internal.ImportValueRequest
if err := proto.Unmarshal(body, &req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// Validate that this handler owns the slice.
if !h.Cluster.OwnsFragment(h.Host, req.Index, req.Slice) {
mesg := fmt.Sprintf("host does not own slice %s-%s slice:%d", h.Host, req.Index, req.Slice)
http.Error(w, mesg, http.StatusPreconditionFailed)
return
}
// Find the Index.
h.logger().Println("importing:", req.Index, req.Frame, req.Slice)
index := h.Holder.Index(req.Index)
if index == nil {
h.logger().Printf("fragment error: index=%s, frame=%s, slice=%d, err=%s", req.Index, req.Frame, req.Slice, ErrIndexNotFound.Error())
http.Error(w, ErrIndexNotFound.Error(), http.StatusNotFound)
return
}
// Retrieve frame.
f := index.Frame(req.Frame)
if f == nil {
h.logger().Printf("frame error: index=%s, frame=%s, slice=%d, err=%s", req.Index, req.Frame, req.Slice, ErrFrameNotFound.Error())
http.Error(w, ErrFrameNotFound.Error(), http.StatusNotFound)
return
}
// Import into fragment.
err = f.ImportValue(req.Field, req.ColumnIDs, req.Values)
if err != nil {
h.logger().Printf("import error: index=%s, frame=%s, slice=%d, field=%s, bits=%d, err=%s", req.Index, req.Frame, req.Slice, req.Field, len(req.ColumnIDs), err)
return
}
// Marshal response object.
buf, e := proto.Marshal(&internal.ImportResponse{Err: errorString(err)})
if e != nil {
http.Error(w, fmt.Sprintf("marshal import response: %s", err), http.StatusInternalServerError)
return
}
// Write response.
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
w.Write(buf)
}
// handleGetExport handles /export requests.
func (h *Handler) handleGetExport(w http.ResponseWriter, r *http.Request) {
switch r.Header.Get("Accept") {

View file

@ -485,6 +485,8 @@ func (i *Index) createFrame(name string, opt FrameOptions) (*Frame, error) {
}
f.inverseEnabled = opt.InverseEnabled
f.rangeEnabled = opt.RangeEnabled
if err := f.saveMeta(); err != nil {
f.Close()
return nil, err
@ -657,6 +659,11 @@ type importData struct {
ColumnIDs []uint64
}
type importValueData struct {
ColumnIDs []uint64
Values []uint64
}
// CreateInputDefinition creates a new input definition.
func (i *Index) CreateInputDefinition(pb *internal.InputDefinition) (*InputDefinition, error) {
// Ensure input definition doesn't already exist.

View file

@ -20,6 +20,7 @@
QueryResponse
QueryResult
ImportRequest
ImportValueRequest
*/
package internal
@ -222,6 +223,20 @@ func (m *ImportRequest) String() string { return proto.CompactTextStr
func (*ImportRequest) ProtoMessage() {}
func (*ImportRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{10} }
type ImportValueRequest struct {
Index string `protobuf:"bytes,1,opt,name=Index,proto3" json:"Index,omitempty"`
Frame string `protobuf:"bytes,2,opt,name=Frame,proto3" json:"Frame,omitempty"`
Slice uint64 `protobuf:"varint,3,opt,name=Slice,proto3" json:"Slice,omitempty"`
Field string `protobuf:"bytes,4,opt,name=Field,proto3" json:"Field,omitempty"`
ColumnIDs []uint64 `protobuf:"varint,5,rep,packed,name=ColumnIDs" json:"ColumnIDs,omitempty"`
Values []uint64 `protobuf:"varint,6,rep,packed,name=Values" json:"Values,omitempty"`
}
func (m *ImportValueRequest) Reset() { *m = ImportValueRequest{} }
func (m *ImportValueRequest) String() string { return proto.CompactTextString(m) }
func (*ImportValueRequest) ProtoMessage() {}
func (*ImportValueRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} }
func init() {
proto.RegisterType((*Bitmap)(nil), "internal.Bitmap")
proto.RegisterType((*Pair)(nil), "internal.Pair")
@ -234,6 +249,7 @@ func init() {
proto.RegisterType((*QueryResponse)(nil), "internal.QueryResponse")
proto.RegisterType((*QueryResult)(nil), "internal.QueryResult")
proto.RegisterType((*ImportRequest)(nil), "internal.ImportRequest")
proto.RegisterType((*ImportValueRequest)(nil), "internal.ImportValueRequest")
}
func (m *Bitmap) Marshal() (dAtA []byte, err error) {
size := m.Size()
@ -772,6 +788,81 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) {
return i, nil
}
func (m *ImportValueRequest) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA)
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) {
var i int
_ = i
var l int
_ = l
if len(m.Index) > 0 {
dAtA[i] = 0xa
i++
i = encodeVarintPublic(dAtA, i, uint64(len(m.Index)))
i += copy(dAtA[i:], m.Index)
}
if len(m.Frame) > 0 {
dAtA[i] = 0x12
i++
i = encodeVarintPublic(dAtA, i, uint64(len(m.Frame)))
i += copy(dAtA[i:], m.Frame)
}
if m.Slice != 0 {
dAtA[i] = 0x18
i++
i = encodeVarintPublic(dAtA, i, uint64(m.Slice))
}
if len(m.Field) > 0 {
dAtA[i] = 0x22
i++
i = encodeVarintPublic(dAtA, i, uint64(len(m.Field)))
i += copy(dAtA[i:], m.Field)
}
if len(m.ColumnIDs) > 0 {
dAtA14 := make([]byte, len(m.ColumnIDs)*10)
var j13 int
for _, num := range m.ColumnIDs {
for num >= 1<<7 {
dAtA14[j13] = uint8(uint64(num)&0x7f | 0x80)
num >>= 7
j13++
}
dAtA14[j13] = uint8(num)
j13++
}
dAtA[i] = 0x2a
i++
i = encodeVarintPublic(dAtA, i, uint64(j13))
i += copy(dAtA[i:], dAtA14[:j13])
}
if len(m.Values) > 0 {
dAtA16 := make([]byte, len(m.Values)*10)
var j15 int
for _, num := range m.Values {
for num >= 1<<7 {
dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80)
num >>= 7
j15++
}
dAtA16[j15] = uint8(num)
j15++
}
dAtA[i] = 0x32
i++
i = encodeVarintPublic(dAtA, i, uint64(j15))
i += copy(dAtA[i:], dAtA16[:j15])
}
return i, nil
}
func encodeFixed64Public(dAtA []byte, offset int, v uint64) int {
dAtA[offset] = uint8(v)
dAtA[offset+1] = uint8(v >> 8)
@ -1025,6 +1116,41 @@ func (m *ImportRequest) Size() (n int) {
return n
}
func (m *ImportValueRequest) Size() (n int) {
var l int
_ = l
l = len(m.Index)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
l = len(m.Frame)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
if m.Slice != 0 {
n += 1 + sovPublic(uint64(m.Slice))
}
l = len(m.Field)
if l > 0 {
n += 1 + l + sovPublic(uint64(l))
}
if len(m.ColumnIDs) > 0 {
l = 0
for _, e := range m.ColumnIDs {
l += sovPublic(uint64(e))
}
n += 1 + sovPublic(uint64(l)) + l
}
if len(m.Values) > 0 {
l = 0
for _, e := range m.Values {
l += sovPublic(uint64(e))
}
n += 1 + sovPublic(uint64(l)) + l
}
return n
}
func sovPublic(x uint64) (n int) {
for {
n++
@ -2690,6 +2816,286 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *ImportValueRequest) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: ImportValueRequest: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: ImportValueRequest: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Index", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Index = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Frame", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Frame = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 3:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Slice", wireType)
}
m.Slice = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.Slice |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
case 4:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Field", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Field = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 5:
if wireType == 2 {
var packedLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
packedLen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if packedLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + packedLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
for iNdEx < postIndex {
var v uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.ColumnIDs = append(m.ColumnIDs, v)
}
} else if wireType == 0 {
var v uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.ColumnIDs = append(m.ColumnIDs, v)
} else {
return fmt.Errorf("proto: wrong wireType = %d for field ColumnIDs", wireType)
}
case 6:
if wireType == 2 {
var packedLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
packedLen |= (int(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
if packedLen < 0 {
return ErrInvalidLengthPublic
}
postIndex := iNdEx + packedLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
for iNdEx < postIndex {
var v uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.Values = append(m.Values, v)
}
} else if wireType == 0 {
var v uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowPublic
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
v |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
m.Values = append(m.Values, v)
} else {
return fmt.Errorf("proto: wrong wireType = %d for field Values", wireType)
}
default:
iNdEx = preIndex
skippy, err := skipPublic(dAtA[iNdEx:])
if err != nil {
return err
}
if skippy < 0 {
return ErrInvalidLengthPublic
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func skipPublic(dAtA []byte) (n int, err error) {
l := len(dAtA)
iNdEx := 0
@ -2798,44 +3204,46 @@ var (
func init() { proto.RegisterFile("public.proto", fileDescriptorPublic) }
var fileDescriptorPublic = []byte{
// 621 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0x4b, 0x8e, 0xd3, 0x40,
0x10, 0xa5, 0x63, 0x27, 0x71, 0x2a, 0x99, 0x51, 0xd4, 0xe2, 0x63, 0x21, 0x14, 0x59, 0x16, 0x0b,
0xaf, 0x32, 0x52, 0x38, 0x00, 0xc2, 0x49, 0x46, 0xb2, 0x10, 0x23, 0xe8, 0x0c, 0xec, 0x3d, 0x49,
0x6b, 0xb0, 0xe4, 0x1f, 0xed, 0xb6, 0x98, 0x9c, 0x83, 0x0d, 0x37, 0x80, 0x0b, 0xb0, 0xe3, 0x00,
0x2c, 0x39, 0x02, 0x0a, 0x17, 0x41, 0xd5, 0xed, 0x8e, 0x1d, 0x16, 0x88, 0x5d, 0xbd, 0x57, 0x5d,
0xe5, 0xfa, 0xbc, 0x32, 0x4c, 0xca, 0xfa, 0x26, 0x4d, 0xb6, 0xf3, 0x52, 0x14, 0xb2, 0xa0, 0x4e,
0x92, 0x4b, 0x2e, 0xf2, 0x38, 0xf5, 0x43, 0x18, 0x84, 0x89, 0xcc, 0xe2, 0x92, 0x52, 0xb0, 0xc3,
0x44, 0x56, 0x2e, 0xf1, 0xac, 0xc0, 0x66, 0xca, 0xa6, 0x4f, 0xa1, 0xff, 0x42, 0x4a, 0x51, 0xb9,
0x3d, 0xcf, 0x0a, 0xc6, 0x8b, 0xf3, 0xb9, 0x89, 0x9b, 0x23, 0xcd, 0xb4, 0xd3, 0x9f, 0x83, 0xfd,
0x3a, 0x4e, 0x04, 0x9d, 0x82, 0xf5, 0x92, 0xef, 0x5d, 0xe2, 0x91, 0xc0, 0x66, 0x68, 0xd2, 0xfb,
0xd0, 0x5f, 0x16, 0x75, 0x2e, 0xdd, 0x9e, 0xe2, 0x34, 0xf0, 0x17, 0xe0, 0x6c, 0xea, 0x4c, 0xd9,
0x18, 0xb3, 0xa9, 0x33, 0x15, 0x63, 0x31, 0x34, 0x4f, 0x63, 0x2c, 0x13, 0xf3, 0x16, 0xac, 0x30,
0x91, 0xe8, 0x64, 0xc5, 0xc7, 0x68, 0xd5, 0x7c, 0x44, 0x03, 0xfa, 0x18, 0x9c, 0x65, 0x91, 0xd6,
0x59, 0x1e, 0xad, 0x9a, 0x2f, 0x1d, 0x31, 0x7d, 0x02, 0xa3, 0xeb, 0x24, 0xe3, 0x95, 0x8c, 0xb3,
0xd2, 0xb5, 0x54, 0xca, 0x96, 0xf0, 0xd7, 0x70, 0xa6, 0x5f, 0x62, 0x27, 0x1b, 0x2e, 0xe9, 0x39,
0xf4, 0x8e, 0xd9, 0x7b, 0xd1, 0xea, 0x3f, 0x27, 0xf0, 0x95, 0x80, 0x8d, 0x56, 0x77, 0x04, 0x23,
0x3d, 0x02, 0x0a, 0xf6, 0xf5, 0xbe, 0xe4, 0x4d, 0x5d, 0xca, 0xa6, 0x1e, 0x8c, 0x37, 0x52, 0x24,
0xf9, 0xed, 0xbb, 0x38, 0xad, 0xb9, 0xaa, 0x6a, 0xc4, 0xba, 0x14, 0x76, 0x14, 0xe5, 0x52, 0xbb,
0x6d, 0x55, 0xf4, 0x11, 0x63, 0x47, 0x61, 0x51, 0xa4, 0xda, 0xd9, 0xf7, 0x48, 0xe0, 0xb0, 0x96,
0xa0, 0x33, 0x80, 0xcb, 0xb4, 0x88, 0x9b, 0xd8, 0x81, 0x47, 0x02, 0xc2, 0x3a, 0x8c, 0x7f, 0x01,
0x43, 0xac, 0xf4, 0x55, 0x5c, 0xb6, 0xbd, 0x91, 0x7f, 0xf5, 0xf6, 0x9d, 0xc0, 0xe4, 0x4d, 0xcd,
0xc5, 0x9e, 0xf1, 0x0f, 0x35, 0xaf, 0xd4, 0x0e, 0x14, 0x6e, 0xba, 0xd4, 0x80, 0x3e, 0x84, 0xc1,
0x26, 0x4d, 0xb6, 0x5c, 0x4f, 0xca, 0x66, 0x0d, 0xc2, 0x5e, 0xdb, 0x09, 0x57, 0xaa, 0x57, 0x87,
0x75, 0x29, 0x8c, 0x64, 0x3c, 0x2b, 0xa4, 0x69, 0xa6, 0x41, 0xd4, 0x87, 0xc9, 0xfa, 0x6e, 0x9b,
0xd6, 0x3b, 0xae, 0x43, 0x07, 0xca, 0x7b, 0xc2, 0x61, 0xf6, 0x06, 0x2b, 0xed, 0x0e, 0x75, 0xf6,
0x0e, 0xe5, 0x7f, 0x22, 0x70, 0xd6, 0x94, 0x5f, 0x95, 0x45, 0x5e, 0x71, 0xdc, 0xd1, 0x5a, 0x08,
0xb3, 0xa3, 0xb5, 0x10, 0xf4, 0x02, 0x86, 0x8c, 0x57, 0x75, 0x2a, 0xcd, 0x9a, 0x1f, 0xb4, 0xa3,
0x30, 0xb1, 0x75, 0x2a, 0x99, 0x79, 0x45, 0x9f, 0xc3, 0xf9, 0x89, 0x6c, 0xb0, 0x2f, 0x8c, 0x7b,
0xd4, 0xc6, 0x9d, 0xf8, 0xd9, 0x5f, 0xcf, 0xfd, 0x6f, 0x04, 0xc6, 0x9d, 0xcc, 0x34, 0x30, 0x67,
0xa8, 0xca, 0x1a, 0x2f, 0xa6, 0x6d, 0x22, 0xcd, 0x33, 0x73, 0xa6, 0x13, 0x20, 0x57, 0x8d, 0x98,
0xc8, 0x15, 0xae, 0x10, 0x4f, 0xcf, 0x7c, 0xbf, 0xb3, 0x42, 0xa4, 0x99, 0x76, 0x52, 0x17, 0x86,
0xcb, 0xf7, 0x71, 0x7e, 0xcb, 0x77, 0x4a, 0x4c, 0x0e, 0x33, 0x90, 0xce, 0xdb, 0x53, 0x54, 0xd3,
0x1f, 0x2f, 0x68, 0x9b, 0xc2, 0x78, 0xd8, 0xf1, 0x8d, 0xff, 0x85, 0xc0, 0x59, 0x94, 0x95, 0x85,
0x90, 0x1d, 0x35, 0x44, 0xf9, 0x8e, 0xdf, 0x19, 0x35, 0x28, 0x80, 0xec, 0xa5, 0x88, 0x33, 0x2d,
0xfb, 0x11, 0xd3, 0x00, 0x59, 0xa5, 0x0a, 0xa5, 0x02, 0x9b, 0x69, 0xa0, 0xf6, 0x8f, 0x67, 0x5c,
0xb9, 0xb6, 0x56, 0x8e, 0x46, 0xa8, 0x73, 0x73, 0xc5, 0x95, 0xdb, 0x57, 0xae, 0x96, 0x40, 0x9d,
0x1f, 0xcf, 0x18, 0xb5, 0x61, 0x05, 0x16, 0xeb, 0x30, 0xe1, 0xf4, 0xc7, 0x61, 0x46, 0x7e, 0x1e,
0x66, 0xe4, 0xd7, 0x61, 0x46, 0x3e, 0xff, 0x9e, 0xdd, 0xbb, 0x19, 0xa8, 0x7f, 0xdf, 0xb3, 0x3f,
0x01, 0x00, 0x00, 0xff, 0xff, 0x27, 0x5d, 0xef, 0xb2, 0x0b, 0x05, 0x00, 0x00,
// 653 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x54, 0xcb, 0x6e, 0xd3, 0x40,
0x14, 0x65, 0x62, 0xe7, 0x75, 0x93, 0x56, 0xd5, 0x08, 0x8a, 0x85, 0x50, 0x14, 0x59, 0x2c, 0xbc,
0x4a, 0xa5, 0xf0, 0x01, 0x08, 0xb7, 0xa9, 0x64, 0x21, 0x2a, 0x98, 0x14, 0xf6, 0x6e, 0x3b, 0x2a,
0x96, 0xfc, 0x62, 0x3c, 0x16, 0xed, 0x77, 0xb0, 0x61, 0xcd, 0x06, 0x7e, 0x80, 0x1d, 0x1f, 0xc0,
0x92, 0x4f, 0x40, 0xe1, 0x47, 0xd0, 0xbd, 0xe3, 0x89, 0x1d, 0x16, 0xc0, 0x82, 0xdd, 0x9c, 0x73,
0x1f, 0xbe, 0x8f, 0x73, 0x0d, 0xd3, 0xb2, 0xbe, 0x48, 0x93, 0xcb, 0x45, 0xa9, 0x0a, 0x5d, 0xf0,
0x51, 0x92, 0x6b, 0xa9, 0xf2, 0x38, 0xf5, 0x43, 0x18, 0x84, 0x89, 0xce, 0xe2, 0x92, 0x73, 0x70,
0xc3, 0x44, 0x57, 0x1e, 0x9b, 0x3b, 0x81, 0x2b, 0xe8, 0xcd, 0x1f, 0x41, 0xff, 0xa9, 0xd6, 0xaa,
0xf2, 0x7a, 0x73, 0x27, 0x98, 0x2c, 0xf7, 0x17, 0x36, 0x6e, 0x81, 0xb4, 0x30, 0x46, 0x7f, 0x01,
0xee, 0x8b, 0x38, 0x51, 0xfc, 0x00, 0x9c, 0x67, 0xf2, 0xd6, 0x63, 0x73, 0x16, 0xb8, 0x02, 0x9f,
0xfc, 0x2e, 0xf4, 0x8f, 0x8b, 0x3a, 0xd7, 0x5e, 0x8f, 0x38, 0x03, 0xfc, 0x25, 0x8c, 0xd6, 0x75,
0x46, 0x6f, 0x8c, 0x59, 0xd7, 0x19, 0xc5, 0x38, 0x02, 0x9f, 0xbb, 0x31, 0x8e, 0x8d, 0x79, 0x05,
0x4e, 0x98, 0x68, 0x34, 0x8a, 0xe2, 0x5d, 0x74, 0xd2, 0x7c, 0xc4, 0x00, 0xfe, 0x00, 0x46, 0xc7,
0x45, 0x5a, 0x67, 0x79, 0x74, 0xd2, 0x7c, 0x69, 0x8b, 0xf9, 0x43, 0x18, 0x9f, 0x27, 0x99, 0xac,
0x74, 0x9c, 0x95, 0x9e, 0x43, 0x29, 0x5b, 0xc2, 0x5f, 0xc1, 0x9e, 0xf1, 0xc4, 0x4e, 0xd6, 0x52,
0xf3, 0x7d, 0xe8, 0x6d, 0xb3, 0xf7, 0xa2, 0x93, 0x7f, 0x9c, 0xc0, 0x67, 0x06, 0x2e, 0xbe, 0xba,
0x23, 0x18, 0x9b, 0x11, 0x70, 0x70, 0xcf, 0x6f, 0x4b, 0xd9, 0xd4, 0x45, 0x6f, 0x3e, 0x87, 0xc9,
0x5a, 0xab, 0x24, 0xbf, 0x7e, 0x1d, 0xa7, 0xb5, 0xa4, 0xaa, 0xc6, 0xa2, 0x4b, 0x61, 0x47, 0x51,
0xae, 0x8d, 0xd9, 0xa5, 0xa2, 0xb7, 0x18, 0x3b, 0x0a, 0x8b, 0x22, 0x35, 0xc6, 0xfe, 0x9c, 0x05,
0x23, 0xd1, 0x12, 0x7c, 0x06, 0x70, 0x9a, 0x16, 0x71, 0x13, 0x3b, 0x98, 0xb3, 0x80, 0x89, 0x0e,
0xe3, 0x1f, 0xc1, 0x10, 0x2b, 0x7d, 0x1e, 0x97, 0x6d, 0x6f, 0xec, 0x4f, 0xbd, 0x7d, 0x65, 0x30,
0x7d, 0x59, 0x4b, 0x75, 0x2b, 0xe4, 0xdb, 0x5a, 0x56, 0xb4, 0x03, 0xc2, 0x4d, 0x97, 0x06, 0xf0,
0x43, 0x18, 0xac, 0xd3, 0xe4, 0x52, 0x9a, 0x49, 0xb9, 0xa2, 0x41, 0xd8, 0x6b, 0x3b, 0xe1, 0x8a,
0x7a, 0x1d, 0x89, 0x2e, 0x85, 0x91, 0x42, 0x66, 0x85, 0xb6, 0xcd, 0x34, 0x88, 0xfb, 0x30, 0x5d,
0xdd, 0x5c, 0xa6, 0xf5, 0x95, 0x34, 0xa1, 0x03, 0xb2, 0xee, 0x70, 0x98, 0xbd, 0xc1, 0xa4, 0xdd,
0xa1, 0xc9, 0xde, 0xa1, 0xfc, 0xf7, 0x0c, 0xf6, 0x9a, 0xf2, 0xab, 0xb2, 0xc8, 0x2b, 0x89, 0x3b,
0x5a, 0x29, 0x65, 0x77, 0xb4, 0x52, 0x8a, 0x1f, 0xc1, 0x50, 0xc8, 0xaa, 0x4e, 0xb5, 0x5d, 0xf3,
0xbd, 0x76, 0x14, 0x36, 0xb6, 0x4e, 0xb5, 0xb0, 0x5e, 0xfc, 0x09, 0xec, 0xef, 0xc8, 0x06, 0xfb,
0xc2, 0xb8, 0xfb, 0x6d, 0xdc, 0x8e, 0x5d, 0xfc, 0xe6, 0xee, 0x7f, 0x61, 0x30, 0xe9, 0x64, 0xe6,
0x81, 0x3d, 0x43, 0x2a, 0x6b, 0xb2, 0x3c, 0x68, 0x13, 0x19, 0x5e, 0xd8, 0x33, 0x9d, 0x02, 0x3b,
0x6b, 0xc4, 0xc4, 0xce, 0x70, 0x85, 0x78, 0x7a, 0xf6, 0xfb, 0x9d, 0x15, 0x22, 0x2d, 0x8c, 0x91,
0x7b, 0x30, 0x3c, 0x7e, 0x13, 0xe7, 0xd7, 0xf2, 0x8a, 0xc4, 0x34, 0x12, 0x16, 0xf2, 0x45, 0x7b,
0x8a, 0x34, 0xfd, 0xc9, 0x92, 0xb7, 0x29, 0xac, 0x45, 0x6c, 0x7d, 0xfc, 0x4f, 0x0c, 0xf6, 0xa2,
0xac, 0x2c, 0x94, 0xee, 0xa8, 0x21, 0xca, 0xaf, 0xe4, 0x8d, 0x55, 0x03, 0x01, 0x64, 0x4f, 0x55,
0x9c, 0x19, 0xd9, 0x8f, 0x85, 0x01, 0xc8, 0x92, 0x2a, 0x48, 0x05, 0xae, 0x30, 0x80, 0xf6, 0x8f,
0x67, 0x5c, 0x79, 0xae, 0x51, 0x8e, 0x41, 0xa8, 0x73, 0x7b, 0xc5, 0x95, 0xd7, 0x27, 0x53, 0x4b,
0xa0, 0xce, 0xb7, 0x67, 0x8c, 0xda, 0x70, 0x02, 0x47, 0x74, 0x18, 0xff, 0x23, 0x03, 0x6e, 0x2a,
0x25, 0xdd, 0xff, 0xbf, 0x72, 0xd1, 0x37, 0x91, 0xa9, 0x19, 0x25, 0xfa, 0x22, 0xf8, 0x4b, 0xb1,
0x87, 0x30, 0xa0, 0x2a, 0x4c, 0xa1, 0xae, 0x68, 0x50, 0x78, 0xf0, 0x6d, 0x33, 0x63, 0xdf, 0x37,
0x33, 0xf6, 0x63, 0x33, 0x63, 0x1f, 0x7e, 0xce, 0xee, 0x5c, 0x0c, 0xe8, 0x07, 0xfd, 0xf8, 0x57,
0x00, 0x00, 0x00, 0xff, 0xff, 0xf6, 0x73, 0x96, 0xb9, 0xb0, 0x05, 0x00, 0x00,
}

View file

@ -72,3 +72,12 @@ message ImportRequest {
repeated uint64 ColumnIDs = 5;
repeated int64 Timestamps = 6;
}
message ImportValueRequest {
string Index = 1;
string Frame = 2;
uint64 Slice = 3;
string Field = 4;
repeated uint64 ColumnIDs = 5;
repeated uint64 Values = 6;
}