From a56410a70b28393d9726a892ef9094bd51b7fbe1 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Thu, 18 Jan 2018 16:08:02 -0600 Subject: [PATCH 1/3] WIP: Modify `pilosa import` to support string rows/columns This PR adds a flag `pilosa import --string-keys=true` which treats the payload CSV as comma separated strings. --- client.go | 85 +++++++ cmd/import.go | 1 + ctl/import.go | 104 ++++++++- internal/public.pb.go | 523 ++++++++++++++++++++++++++++++++++++------ internal/public.proto | 10 +- 5 files changed, 655 insertions(+), 68 deletions(-) diff --git a/client.go b/client.go index 8f4f282f8..073a4629c 100644 --- a/client.go +++ b/client.go @@ -303,6 +303,32 @@ func (c *InternalHTTPClient) Import(ctx context.Context, index, frame string, sl return nil } +// ImportK bulk imports bitKs to a host. +func (c *InternalHTTPClient) ImportK(ctx context.Context, index, frame string, bitKs []BitK) error { + if index == "" { + return ErrIndexRequired + } else if frame == "" { + return ErrFrameRequired + } + + buf, err := marshalImportKPayload(index, frame, bitKs) + if err != nil { + return fmt.Errorf("Error Creating Payload: %s", err) + } + + node := &Node{ + Scheme: c.defaultURI.Scheme(), + Host: c.defaultURI.HostPort(), + } + + // Import to node. + if err := c.importNode(ctx, node, buf); err != nil { + return fmt.Errorf("import node: host=%s, err=%s", node.Host, err) + } + + return nil +} + func (c *InternalHTTPClient) EnsureIndex(ctx context.Context, name string, options IndexOptions) error { err := c.CreateIndex(ctx, name, options) if err == nil || err == ErrIndexExists { @@ -341,6 +367,27 @@ func marshalImportPayload(index, frame string, slice uint64, bits []Bit) ([]byte return buf, nil } +// marshalImportKPayload marshalls the import parameters into a protobuf byte slice. +func marshalImportKPayload(index, frame string, bitKs []BitK) ([]byte, error) { + // Separate row and column IDs to reduce allocations. + rowKeys := BitKs(bitKs).RowKeys() + columnKeys := BitKs(bitKs).ColumnKeys() + timestamps := BitKs(bitKs).Timestamps() + + // Marshal bits to protobufs. + buf, err := proto.Marshal(&internal.ImportKRequest{ + Index: index, + Frame: frame, + RowKeys: rowKeys, + ColumnKeys: columnKeys, + Timestamps: timestamps, + }) + if err != nil { + return nil, fmt.Errorf("marshal import request: %s", err) + } + return buf, nil +} + // importNode sends a pre-marshaled import request to a node. func (c *InternalHTTPClient) importNode(ctx context.Context, node *Node, buf []byte) error { // Create URL & HTTP request. @@ -1177,6 +1224,43 @@ func (p Bits) GroupBySlice() map[uint64][]Bit { return m } +// BitK represents the location of a single bit as string keys. +type BitK struct { + RowKey string + ColumnKey string + Timestamp int64 +} + +// BitKs represents a slice of bitKs. +type BitKs []BitK + +// RowKeys returns a slice of all the row Keys. +func (p BitKs) RowKeys() []string { + other := make([]string, len(p)) + for i := range p { + other[i] = p[i].RowKey + } + return other +} + +// ColumnKeys returns a slice of all the column Keys. +func (p BitKs) ColumnKeys() []string { + other := make([]string, len(p)) + for i := range p { + other[i] = p[i].ColumnKey + } + return other +} + +// Timestamps returns a slice of all the timestamps. +func (p BitKs) Timestamps() []int64 { + other := make([]int64, len(p)) + for i := range p { + other[i] = p[i].Timestamp + } + return other +} + // FieldValues represents the value for a column within a // range-encoded frame. type FieldValue struct { @@ -1271,6 +1355,7 @@ type InternalClient interface { FragmentNodes(ctx context.Context, index string, slice uint64) ([]*Node, error) ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) Import(ctx context.Context, index, frame string, slice uint64, bits []Bit) error + ImportK(ctx context.Context, index, frame string, bitKs []BitK) error EnsureIndex(ctx context.Context, name string, options IndexOptions) error EnsureFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) error ImportValue(ctx context.Context, index, frame, field string, slice uint64, vals []FieldValue) error diff --git a/cmd/import.go b/cmd/import.go index 250dccfd3..8692fa318 100644 --- a/cmd/import.go +++ b/cmd/import.go @@ -56,6 +56,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM. 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.BoolVar(&Importer.StringKeys, "string-keys", false, "Treat payload as string keys.") 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.") diff --git a/ctl/import.go b/ctl/import.go index 179385db5..b6c1b6f50 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -48,6 +48,9 @@ type ImportCommand struct { // For Range-Encoded fields, name of the Field to import into. Field string `json:"field"` + // Indicates that the payload should be treated as string keys. + StringKeys bool `json:"StringKeys"` + // Filenames to import from. Paths []string `json:"paths"` @@ -131,7 +134,11 @@ func (cmd *ImportCommand) importPath(ctx context.Context, path string) error { if cmd.Field != "" { return cmd.bufferFieldValues(ctx, path) } else { - return cmd.bufferBits(ctx, path) + if cmd.StringKeys { + return cmd.bufferBitKs(ctx, path) + } else { + return cmd.bufferBits(ctx, path) + } } } @@ -240,7 +247,102 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err } return nil +} +// bufferBitKs buffers slices of keys to be imported as a batch. +func (cmd *ImportCommand) bufferBitKs(ctx context.Context, path string) error { + a := make([]pilosa.BitK, 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 bitK pilosa.BitK + + // Parse row key. + if record[0] == "" { + return fmt.Errorf("invalid row key on row %d: %q", rnum, record[0]) + } + bitK.RowKey = record[0] + + // Parse column key. + if record[1] == "" { + return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) + } + bitK.ColumnKey = record[1] + + // Parse time, if exists. + if len(record) > 2 && record[2] != "" { + t, err := time.Parse(pilosa.TimeFormat, record[2]) + if err != nil { + return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2]) + } + bitK.Timestamp = t.UnixNano() + } + + a = append(a, bitK) + + // If we've reached the buffer size then import bitKs. + if len(a) == cmd.BufferSize { + if err := cmd.importBitKs(ctx, a); err != nil { + return err + } + a = a[:0] + } + } + + // If there are still bitKs in the buffer then flush them. + if err := cmd.importBitKs(ctx, a); err != nil { + return err + } + + return nil +} + +// importBitKs sends batches of bitKs to the server. +func (cmd *ImportCommand) importBitKs(ctx context.Context, bitKs []pilosa.BitK) error { + logger := log.New(cmd.Stderr, "", log.LstdFlags) + + // TODO: does it help to sort the rowKeys? + + logger.Printf("importing keys: n=%d", len(bitKs)) + if err := cmd.Client.ImportK(ctx, cmd.Index, cmd.Frame, bitKs); err != nil { + return err + } + + return nil } // bufferFieldValues buffers slices of fieldValues to be imported as a batch. diff --git a/internal/public.pb.go b/internal/public.pb.go index 1676d4ae5..83a402247 100644 --- a/internal/public.pb.go +++ b/internal/public.pb.go @@ -20,6 +20,7 @@ QueryResponse QueryResult ImportRequest + ImportKRequest ImportValueRequest */ package internal @@ -465,6 +466,54 @@ func (m *ImportRequest) GetTimestamps() []int64 { return nil } +type ImportKRequest 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"` + RowKeys []string `protobuf:"bytes,3,rep,name=RowKeys" json:"RowKeys,omitempty"` + ColumnKeys []string `protobuf:"bytes,4,rep,name=ColumnKeys" json:"ColumnKeys,omitempty"` + Timestamps []int64 `protobuf:"varint,5,rep,packed,name=Timestamps" json:"Timestamps,omitempty"` +} + +func (m *ImportKRequest) Reset() { *m = ImportKRequest{} } +func (m *ImportKRequest) String() string { return proto.CompactTextString(m) } +func (*ImportKRequest) ProtoMessage() {} +func (*ImportKRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} } + +func (m *ImportKRequest) GetIndex() string { + if m != nil { + return m.Index + } + return "" +} + +func (m *ImportKRequest) GetFrame() string { + if m != nil { + return m.Frame + } + return "" +} + +func (m *ImportKRequest) GetRowKeys() []string { + if m != nil { + return m.RowKeys + } + return nil +} + +func (m *ImportKRequest) GetColumnKeys() []string { + if m != nil { + return m.ColumnKeys + } + return nil +} + +func (m *ImportKRequest) GetTimestamps() []int64 { + if m != nil { + return m.Timestamps + } + return nil +} + 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"` @@ -477,7 +526,7 @@ type ImportValueRequest struct { 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 (*ImportValueRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{12} } func (m *ImportValueRequest) GetIndex() string { if m != nil { @@ -533,6 +582,7 @@ func init() { proto.RegisterType((*QueryResponse)(nil), "internal.QueryResponse") proto.RegisterType((*QueryResult)(nil), "internal.QueryResult") proto.RegisterType((*ImportRequest)(nil), "internal.ImportRequest") + proto.RegisterType((*ImportKRequest)(nil), "internal.ImportKRequest") proto.RegisterType((*ImportValueRequest)(nil), "internal.ImportValueRequest") } func (m *Bitmap) Marshal() (dAtA []byte, err error) { @@ -1104,6 +1154,84 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) { return i, nil } +func (m *ImportKRequest) 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 *ImportKRequest) 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 len(m.RowKeys) > 0 { + for _, s := range m.RowKeys { + dAtA[i] = 0x1a + i++ + l = len(s) + for l >= 1<<7 { + dAtA[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ + } + dAtA[i] = uint8(l) + i++ + i += copy(dAtA[i:], s) + } + } + if len(m.ColumnKeys) > 0 { + for _, s := range m.ColumnKeys { + dAtA[i] = 0x22 + i++ + l = len(s) + for l >= 1<<7 { + dAtA[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ + } + dAtA[i] = uint8(l) + i++ + i += copy(dAtA[i:], s) + } + } + if len(m.Timestamps) > 0 { + dAtA14 := make([]byte, len(m.Timestamps)*10) + var j13 int + for _, num1 := range m.Timestamps { + num := uint64(num1) + 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]) + } + return i, nil +} + func (m *ImportValueRequest) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -1143,27 +1271,9 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { 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) + dAtA16 := make([]byte, len(m.ColumnIDs)*10) var j15 int - for _, num1 := range m.Values { - num := uint64(num1) + for _, num := range m.ColumnIDs { for num >= 1<<7 { dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 @@ -1172,11 +1282,29 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { dAtA16[j15] = uint8(num) j15++ } - dAtA[i] = 0x32 + dAtA[i] = 0x2a i++ i = encodeVarintPublic(dAtA, i, uint64(j15)) i += copy(dAtA[i:], dAtA16[:j15]) } + if len(m.Values) > 0 { + dAtA18 := make([]byte, len(m.Values)*10) + var j17 int + for _, num1 := range m.Values { + num := uint64(num1) + for num >= 1<<7 { + dAtA18[j17] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j17++ + } + dAtA18[j17] = uint8(num) + j17++ + } + dAtA[i] = 0x32 + i++ + i = encodeVarintPublic(dAtA, i, uint64(j17)) + i += copy(dAtA[i:], dAtA18[:j17]) + } return i, nil } @@ -1450,6 +1578,39 @@ func (m *ImportRequest) Size() (n int) { return n } +func (m *ImportKRequest) 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 len(m.RowKeys) > 0 { + for _, s := range m.RowKeys { + l = len(s) + n += 1 + l + sovPublic(uint64(l)) + } + } + if len(m.ColumnKeys) > 0 { + for _, s := range m.ColumnKeys { + l = len(s) + n += 1 + l + sovPublic(uint64(l)) + } + } + if len(m.Timestamps) > 0 { + l = 0 + for _, e := range m.Timestamps { + l += sovPublic(uint64(e)) + } + n += 1 + sovPublic(uint64(l)) + l + } + return n +} + func (m *ImportValueRequest) Size() (n int) { var l int _ = l @@ -3256,6 +3417,234 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error { } return nil } +func (m *ImportKRequest) 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: ImportKRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ImportKRequest: 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 != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field RowKeys", 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.RowKeys = append(m.RowKeys, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex + case 4: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ColumnKeys", 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.ColumnKeys = append(m.ColumnKeys, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex + case 5: + if wireType == 0 { + var v int64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (int64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.Timestamps = append(m.Timestamps, v) + } else 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 int64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (int64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.Timestamps = append(m.Timestamps, v) + } + } else { + return fmt.Errorf("proto: wrong wireType = %d for field Timestamps", 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 (m *ImportValueRequest) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 @@ -3644,48 +4033,50 @@ var ( func init() { proto.RegisterFile("public.proto", fileDescriptorPublic) } var fileDescriptorPublic = []byte{ - // 678 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, 0x54, 0xd5, 0x08, 0x8a, 0x85, 0x50, 0x64, 0x59, 0x2c, 0xbc, - 0x4a, 0xa5, 0xb0, 0x07, 0x91, 0x3e, 0xa4, 0xa8, 0xa2, 0x82, 0x9b, 0x52, 0xd6, 0x6e, 0x3b, 0x2a, - 0x96, 0xfc, 0xc2, 0x1e, 0x8b, 0xe6, 0x3b, 0xd8, 0xb0, 0x66, 0x03, 0x1f, 0xc1, 0x07, 0xc0, 0x8e, - 0x4f, 0x40, 0xe1, 0x47, 0xd0, 0x9d, 0xf1, 0xd8, 0x0e, 0x95, 0x80, 0x05, 0xbb, 0x39, 0xe7, 0xcc, - 0x5c, 0xdf, 0xc7, 0xb9, 0x86, 0x71, 0x56, 0x5e, 0x44, 0xe1, 0xe5, 0x2c, 0xcb, 0x53, 0x99, 0xf2, - 0x41, 0x98, 0x48, 0x91, 0x27, 0x41, 0xe4, 0x9d, 0x43, 0x6f, 0x11, 0xca, 0x38, 0xc8, 0x38, 0x07, - 0x7b, 0x11, 0xca, 0xc2, 0x61, 0xae, 0xe5, 0xdb, 0xa8, 0xce, 0xfc, 0x11, 0x74, 0x9f, 0x49, 0x99, - 0x17, 0x4e, 0xc7, 0xb5, 0xfc, 0xd1, 0x7c, 0x67, 0x66, 0xde, 0xcd, 0x88, 0x46, 0x2d, 0xd2, 0xcb, - 0x13, 0xb1, 0x2e, 0x1c, 0xcb, 0xb5, 0xfc, 0x21, 0xaa, 0xb3, 0xf7, 0x04, 0xec, 0x17, 0x41, 0x98, - 0xf3, 0x1d, 0xe8, 0x2c, 0x0f, 0x1d, 0xe6, 0x32, 0xdf, 0xc6, 0xce, 0xf2, 0x90, 0xdf, 0x85, 0xee, - 0x41, 0x5a, 0x26, 0xd2, 0xe9, 0x28, 0x4a, 0x03, 0xbe, 0x0b, 0xd6, 0x89, 0x58, 0x3b, 0x96, 0xcb, - 0xfc, 0x21, 0xd2, 0xd1, 0x9b, 0xc3, 0x60, 0x55, 0xc6, 0xb5, 0xba, 0x2a, 0x63, 0x15, 0xc4, 0x42, - 0x3a, 0x6e, 0x47, 0xb1, 0xaa, 0x28, 0xde, 0x2b, 0xb0, 0x16, 0xa1, 0x24, 0x11, 0xd3, 0x77, 0xf5, - 0x57, 0x35, 0xe0, 0x0f, 0x60, 0x70, 0x90, 0x46, 0x65, 0x9c, 0x2c, 0x0f, 0xab, 0x6f, 0xd7, 0x98, - 0x3f, 0x84, 0xe1, 0x59, 0x18, 0x8b, 0x42, 0x06, 0x71, 0xa6, 0x92, 0xb0, 0xb0, 0x21, 0xbc, 0xd7, - 0x30, 0xd1, 0x37, 0xa9, 0xda, 0x95, 0x90, 0xb7, 0x6a, 0xfa, 0xb7, 0x2e, 0xdd, 0xae, 0xf1, 0x33, - 0x03, 0x9b, 0x34, 0x23, 0xb1, 0x5a, 0xa2, 0x96, 0x9e, 0xad, 0x33, 0x51, 0x65, 0xaa, 0xce, 0xdc, - 0x85, 0xd1, 0x4a, 0xe6, 0x61, 0x72, 0x7d, 0x1e, 0x44, 0xa5, 0xa8, 0x02, 0xb5, 0x29, 0xaa, 0x71, - 0x99, 0x48, 0x2d, 0xdb, 0xaa, 0x8c, 0x1a, 0x53, 0x8d, 0x8b, 0x34, 0x8d, 0xb4, 0xd8, 0x75, 0x99, - 0x3f, 0xc0, 0x86, 0xe0, 0x53, 0x80, 0xe3, 0x28, 0x0d, 0xaa, 0xb7, 0x3d, 0x97, 0xf9, 0x0c, 0x5b, - 0x8c, 0xb7, 0x0f, 0x7d, 0xca, 0xf4, 0x79, 0x90, 0x35, 0xd5, 0xb2, 0x3f, 0x54, 0xeb, 0x7d, 0x61, - 0x30, 0x7e, 0x59, 0x8a, 0x7c, 0x8d, 0xe2, 0x6d, 0x29, 0x0a, 0x35, 0x15, 0x85, 0xab, 0x2a, 0x35, - 0xe0, 0x7b, 0xd0, 0x5b, 0x45, 0xe1, 0xa5, 0xd0, 0xbd, 0xb3, 0xb1, 0x42, 0x54, 0x6b, 0xd3, 0xf3, - 0x42, 0xd5, 0x3a, 0xc0, 0x36, 0x45, 0x2f, 0x51, 0xc4, 0xa9, 0x34, 0xc5, 0x54, 0x88, 0x7b, 0x30, - 0x3e, 0xba, 0xb9, 0x8c, 0xca, 0x2b, 0xa1, 0x9f, 0xf6, 0x94, 0xba, 0xc5, 0x51, 0xf4, 0x0a, 0x2b, - 0xc7, 0xf7, 0x75, 0xf4, 0x16, 0xe5, 0xbd, 0x67, 0x30, 0xa9, 0xd2, 0x2f, 0xb2, 0x34, 0x29, 0x04, - 0xcd, 0xe8, 0x28, 0xcf, 0xcd, 0x8c, 0x8e, 0xf2, 0x9c, 0xef, 0x43, 0x1f, 0x45, 0x51, 0x46, 0xd2, - 0x0c, 0xfe, 0x5e, 0xd3, 0x0a, 0xf3, 0xb6, 0x8c, 0x24, 0x9a, 0x5b, 0xfc, 0x29, 0xec, 0x6c, 0x19, - 0x49, 0x6f, 0xcc, 0x68, 0x7e, 0xbf, 0x79, 0xb7, 0xa5, 0xe3, 0x6f, 0xd7, 0xbd, 0x6f, 0x0c, 0x46, - 0xad, 0xc8, 0xdc, 0x37, 0xcb, 0xab, 0xd2, 0x1a, 0xcd, 0x77, 0x9b, 0x40, 0x9a, 0x47, 0xb3, 0xdc, - 0x63, 0x60, 0xa7, 0x95, 0x99, 0xd8, 0x29, 0x8d, 0x90, 0x96, 0xd3, 0x7c, 0xbf, 0x35, 0x42, 0xa2, - 0x51, 0x8b, 0xdc, 0x81, 0xfe, 0xc1, 0x9b, 0x20, 0xb9, 0x16, 0x57, 0xca, 0x4c, 0x03, 0x34, 0x90, - 0xcf, 0x9a, 0xe5, 0x54, 0xdd, 0x1f, 0xcd, 0x79, 0x13, 0xc2, 0x28, 0xd8, 0x2c, 0xb0, 0x71, 0x33, - 0xcd, 0x62, 0xa2, 0xdd, 0xec, 0x7d, 0x62, 0x30, 0x59, 0xc6, 0x59, 0x9a, 0xcb, 0x96, 0x43, 0x96, - 0xc9, 0x95, 0xb8, 0x31, 0x0e, 0x51, 0x80, 0xd8, 0xe3, 0x3c, 0x88, 0xf5, 0x2a, 0x0c, 0x51, 0x03, - 0x62, 0x95, 0x53, 0x94, 0x33, 0x6c, 0xd4, 0x40, 0x79, 0x82, 0x96, 0xbd, 0x70, 0x6c, 0xed, 0x26, - 0x8d, 0xc8, 0xfb, 0x66, 0xd7, 0x0b, 0xa7, 0xab, 0xa4, 0x86, 0x20, 0xef, 0xd7, 0xcb, 0x4e, 0x7e, - 0xb1, 0x7c, 0x0b, 0x5b, 0x8c, 0xf7, 0x91, 0x01, 0xd7, 0x99, 0xaa, 0x5d, 0xf8, 0x7f, 0xe9, 0xd2, - 0xdd, 0x50, 0x44, 0xba, 0xbd, 0x74, 0x97, 0xc0, 0x5f, 0x92, 0xdd, 0x83, 0x9e, 0xca, 0xc2, 0x24, - 0x5a, 0xa1, 0xc5, 0xee, 0xd7, 0xcd, 0x94, 0x7d, 0xdf, 0x4c, 0xd9, 0x8f, 0xcd, 0x94, 0x7d, 0xf8, - 0x39, 0xbd, 0x73, 0xd1, 0x53, 0xbf, 0xfa, 0xc7, 0xbf, 0x02, 0x00, 0x00, 0xff, 0xff, 0xe7, 0x6f, - 0xfd, 0x3f, 0xfa, 0x05, 0x00, 0x00, + // 712 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcb, 0x6a, 0x14, 0x41, + 0x14, 0xb5, 0xa6, 0x7b, 0x5e, 0x77, 0x1e, 0x84, 0x42, 0x63, 0x23, 0x32, 0x34, 0x8d, 0x8b, 0x5e, + 0x4d, 0x60, 0xdc, 0x2b, 0x4e, 0x1e, 0x30, 0x04, 0x83, 0xd6, 0xc4, 0xb8, 0xee, 0x24, 0x45, 0x6c, + 0xe8, 0x97, 0xdd, 0xd5, 0x24, 0xf3, 0x1d, 0x82, 0xb8, 0x76, 0xa3, 0x1f, 0xe1, 0x07, 0xe8, 0xce, + 0x4f, 0x90, 0xf8, 0x23, 0x72, 0x6f, 0x75, 0x4d, 0xf7, 0x24, 0xa0, 0x22, 0xee, 0xea, 0x9c, 0x53, + 0x75, 0xfb, 0x3e, 0xce, 0x9d, 0x81, 0x61, 0x56, 0x9e, 0x46, 0xe1, 0xd9, 0x34, 0xcb, 0x53, 0x95, + 0xf2, 0x5e, 0x98, 0x28, 0x99, 0x27, 0x41, 0xe4, 0x9d, 0x40, 0x67, 0x1e, 0xaa, 0x38, 0xc8, 0x38, + 0x07, 0x7b, 0x1e, 0xaa, 0xc2, 0x61, 0xae, 0xe5, 0xdb, 0x82, 0xce, 0xfc, 0x11, 0xb4, 0x9f, 0x29, + 0x95, 0x17, 0x4e, 0xcb, 0xb5, 0xfc, 0xc1, 0x6c, 0x3c, 0x35, 0xef, 0xa6, 0x48, 0x0b, 0x2d, 0xe2, + 0xcb, 0x43, 0xb9, 0x2a, 0x1c, 0xcb, 0xb5, 0xfc, 0xbe, 0xa0, 0xb3, 0xf7, 0x04, 0xec, 0x17, 0x41, + 0x98, 0xf3, 0x31, 0xb4, 0x16, 0x7b, 0x0e, 0x73, 0x99, 0x6f, 0x8b, 0xd6, 0x62, 0x8f, 0xdf, 0x85, + 0xf6, 0x6e, 0x5a, 0x26, 0xca, 0x69, 0x11, 0xa5, 0x01, 0xdf, 0x02, 0xeb, 0x50, 0xae, 0x1c, 0xcb, + 0x65, 0x7e, 0x5f, 0xe0, 0xd1, 0x9b, 0x41, 0x6f, 0x59, 0xc6, 0x6b, 0x75, 0x59, 0xc6, 0x14, 0xc4, + 0x12, 0x78, 0xdc, 0x8c, 0x62, 0x55, 0x51, 0xbc, 0x57, 0x60, 0xcd, 0x43, 0x85, 0xa2, 0x48, 0x2f, + 0xd7, 0x5f, 0xd5, 0x80, 0x3f, 0x80, 0xde, 0x6e, 0x1a, 0x95, 0x71, 0xb2, 0xd8, 0xab, 0xbe, 0xbd, + 0xc6, 0xfc, 0x21, 0xf4, 0x8f, 0xc3, 0x58, 0x16, 0x2a, 0x88, 0x33, 0x4a, 0xc2, 0x12, 0x35, 0xe1, + 0xbd, 0x86, 0x91, 0xbe, 0x89, 0xd5, 0x2e, 0xa5, 0xba, 0x55, 0xd3, 0xdf, 0x75, 0xe9, 0x76, 0x8d, + 0x9f, 0x19, 0xd8, 0xa8, 0x19, 0x89, 0xad, 0x25, 0x6c, 0xe9, 0xf1, 0x2a, 0x93, 0x55, 0xa6, 0x74, + 0xe6, 0x2e, 0x0c, 0x96, 0x2a, 0x0f, 0x93, 0x8b, 0x93, 0x20, 0x2a, 0x65, 0x15, 0xa8, 0x49, 0x61, + 0x8d, 0x8b, 0x44, 0x69, 0xd9, 0xa6, 0x32, 0xd6, 0x18, 0x6b, 0x9c, 0xa7, 0x69, 0xa4, 0xc5, 0xb6, + 0xcb, 0xfc, 0x9e, 0xa8, 0x09, 0x3e, 0x01, 0x38, 0x88, 0xd2, 0xa0, 0x7a, 0xdb, 0x71, 0x99, 0xcf, + 0x44, 0x83, 0xf1, 0x76, 0xa0, 0x8b, 0x99, 0x3e, 0x0f, 0xb2, 0xba, 0x5a, 0xf6, 0x9b, 0x6a, 0xbd, + 0x2f, 0x0c, 0x86, 0x2f, 0x4b, 0x99, 0xaf, 0x84, 0x7c, 0x5b, 0xca, 0x82, 0xa6, 0x42, 0xb8, 0xaa, + 0x52, 0x03, 0xbe, 0x0d, 0x9d, 0x65, 0x14, 0x9e, 0x49, 0xdd, 0x3b, 0x5b, 0x54, 0x08, 0x6b, 0xad, + 0x7b, 0x5e, 0x50, 0xad, 0x3d, 0xd1, 0xa4, 0xf0, 0xa5, 0x90, 0x71, 0xaa, 0x4c, 0x31, 0x15, 0xe2, + 0x1e, 0x0c, 0xf7, 0xaf, 0xce, 0xa2, 0xf2, 0x5c, 0xea, 0xa7, 0x1d, 0x52, 0x37, 0x38, 0x8c, 0x5e, + 0x61, 0x72, 0x7c, 0x57, 0x47, 0x6f, 0x50, 0xde, 0x3b, 0x06, 0xa3, 0x2a, 0xfd, 0x22, 0x4b, 0x93, + 0x42, 0xe2, 0x8c, 0xf6, 0xf3, 0xdc, 0xcc, 0x68, 0x3f, 0xcf, 0xf9, 0x0e, 0x74, 0x85, 0x2c, 0xca, + 0x48, 0x99, 0xc1, 0xdf, 0xab, 0x5b, 0x61, 0xde, 0x96, 0x91, 0x12, 0xe6, 0x16, 0x7f, 0x0a, 0xe3, + 0x0d, 0x23, 0xe9, 0x8d, 0x19, 0xcc, 0xee, 0xd7, 0xef, 0x36, 0x74, 0x71, 0xe3, 0xba, 0xf7, 0x8d, + 0xc1, 0xa0, 0x11, 0x99, 0xfb, 0x66, 0x79, 0x29, 0xad, 0xc1, 0x6c, 0xab, 0x0e, 0xa4, 0x79, 0x61, + 0x96, 0x7b, 0x08, 0xec, 0xa8, 0x32, 0x13, 0x3b, 0xc2, 0x11, 0xe2, 0x72, 0x9a, 0xef, 0x37, 0x46, + 0x88, 0xb4, 0xd0, 0x22, 0x77, 0xa0, 0xbb, 0xfb, 0x26, 0x48, 0x2e, 0xe4, 0x39, 0x99, 0xa9, 0x27, + 0x0c, 0xe4, 0xd3, 0x7a, 0x39, 0xa9, 0xfb, 0x83, 0x19, 0xaf, 0x43, 0x18, 0x45, 0xd4, 0x0b, 0x6c, + 0xdc, 0x8c, 0xb3, 0x18, 0x69, 0x37, 0x7b, 0x9f, 0x18, 0x8c, 0x16, 0x71, 0x96, 0xe6, 0xaa, 0xe1, + 0x90, 0x45, 0x72, 0x2e, 0xaf, 0x8c, 0x43, 0x08, 0x20, 0x7b, 0x90, 0x07, 0xb1, 0x5e, 0x85, 0xbe, + 0xd0, 0x00, 0x59, 0x72, 0x0a, 0x39, 0xc3, 0x16, 0x1a, 0x90, 0x27, 0x70, 0xd9, 0x0b, 0xc7, 0xd6, + 0x6e, 0xd2, 0x08, 0xbd, 0x6f, 0x76, 0xbd, 0x70, 0xda, 0x24, 0xd5, 0x04, 0x7a, 0x7f, 0xbd, 0xec, + 0xe8, 0x17, 0xcb, 0xb7, 0x44, 0x83, 0xf1, 0xde, 0x33, 0x18, 0xeb, 0x4c, 0x0f, 0xff, 0x25, 0x55, + 0x07, 0xba, 0x22, 0xbd, 0x6c, 0xfc, 0x40, 0x1a, 0x88, 0x1f, 0xd6, 0x59, 0x90, 0x68, 0x93, 0xd8, + 0x60, 0x6e, 0x24, 0xd6, 0xbe, 0x95, 0xd8, 0x47, 0x06, 0x5c, 0x27, 0x46, 0x4b, 0xfa, 0xff, 0xfa, + 0x88, 0x77, 0x43, 0x19, 0xe9, 0xb9, 0xe3, 0x5d, 0x04, 0x7f, 0xe8, 0xe2, 0x36, 0x74, 0x28, 0x0b, + 0xd3, 0xc1, 0x0a, 0xcd, 0xb7, 0xbe, 0x5e, 0x4f, 0xd8, 0xf7, 0xeb, 0x09, 0xfb, 0x71, 0x3d, 0x61, + 0x1f, 0x7e, 0x4e, 0xee, 0x9c, 0x76, 0xe8, 0x3f, 0xe8, 0xf1, 0xaf, 0x00, 0x00, 0x00, 0xff, 0xff, + 0xf3, 0x62, 0xbb, 0x7d, 0x93, 0x06, 0x00, 0x00, } diff --git a/internal/public.proto b/internal/public.proto index e5fc833fd..649fc0e3d 100644 --- a/internal/public.proto +++ b/internal/public.proto @@ -41,7 +41,7 @@ message Attr { } message AttrMap { - repeated Attr Attrs = 1; + repeated Attr Attrs = 1; } message QueryRequest { @@ -77,6 +77,14 @@ message ImportRequest { repeated int64 Timestamps = 6; } +message ImportKRequest { + string Index = 1; + string Frame = 2; + repeated string RowKeys = 3; + repeated string ColumnKeys = 4; + repeated int64 Timestamps = 5; +} + message ImportValueRequest { string Index = 1; string Frame = 2; From d8559ae469486064e667c6c2e23285188e70c5db Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Fri, 19 Jan 2018 17:47:03 -0600 Subject: [PATCH 2/3] refactor to remove BitK (in favor of Bit) --- client.go | 77 +++---- ctl/import.go | 32 +-- internal/public.pb.go | 511 ++++++++++++------------------------------ internal/public.proto | 11 +- 4 files changed, 197 insertions(+), 434 deletions(-) diff --git a/client.go b/client.go index 073a4629c..bbf118399 100644 --- a/client.go +++ b/client.go @@ -303,15 +303,15 @@ func (c *InternalHTTPClient) Import(ctx context.Context, index, frame string, sl return nil } -// ImportK bulk imports bitKs to a host. -func (c *InternalHTTPClient) ImportK(ctx context.Context, index, frame string, bitKs []BitK) error { +// ImportK bulk imports bits to a host. +func (c *InternalHTTPClient) ImportK(ctx context.Context, index, frame string, bits []Bit) error { if index == "" { return ErrIndexRequired } else if frame == "" { return ErrFrameRequired } - buf, err := marshalImportKPayload(index, frame, bitKs) + buf, err := marshalImportPayloadK(index, frame, bits) if err != nil { return fmt.Errorf("Error Creating Payload: %s", err) } @@ -367,15 +367,15 @@ func marshalImportPayload(index, frame string, slice uint64, bits []Bit) ([]byte return buf, nil } -// marshalImportKPayload marshalls the import parameters into a protobuf byte slice. -func marshalImportKPayload(index, frame string, bitKs []BitK) ([]byte, error) { +// marshalImportPayloadK marshalls the import parameters into a protobuf byte slice. +func marshalImportPayloadK(index, frame string, bits []Bit) ([]byte, error) { // Separate row and column IDs to reduce allocations. - rowKeys := BitKs(bitKs).RowKeys() - columnKeys := BitKs(bitKs).ColumnKeys() - timestamps := BitKs(bitKs).Timestamps() + rowKeys := Bits(bits).RowKeys() + columnKeys := Bits(bits).ColumnKeys() + timestamps := Bits(bits).Timestamps() // Marshal bits to protobufs. - buf, err := proto.Marshal(&internal.ImportKRequest{ + buf, err := proto.Marshal(&internal.ImportRequest{ Index: index, Frame: frame, RowKeys: rowKeys, @@ -1162,6 +1162,8 @@ func (c *InternalHTTPClient) NodeID(uri *URI) (string, error) { type Bit struct { RowID uint64 ColumnID uint64 + RowKey string + ColumnKey string Timestamp int64 } @@ -1199,6 +1201,24 @@ func (p Bits) ColumnIDs() []uint64 { return other } +// RowKeys returns a slice of all the row keys. +func (p Bits) RowKeys() []string { + other := make([]string, len(p)) + for i := range p { + other[i] = p[i].RowKey + } + return other +} + +// ColumnKeys returns a slice of all the column keys. +func (p Bits) ColumnKeys() []string { + other := make([]string, len(p)) + for i := range p { + other[i] = p[i].ColumnKey + } + return other +} + // Timestamps returns a slice of all the timestamps. func (p Bits) Timestamps() []int64 { other := make([]int64, len(p)) @@ -1224,43 +1244,6 @@ func (p Bits) GroupBySlice() map[uint64][]Bit { return m } -// BitK represents the location of a single bit as string keys. -type BitK struct { - RowKey string - ColumnKey string - Timestamp int64 -} - -// BitKs represents a slice of bitKs. -type BitKs []BitK - -// RowKeys returns a slice of all the row Keys. -func (p BitKs) RowKeys() []string { - other := make([]string, len(p)) - for i := range p { - other[i] = p[i].RowKey - } - return other -} - -// ColumnKeys returns a slice of all the column Keys. -func (p BitKs) ColumnKeys() []string { - other := make([]string, len(p)) - for i := range p { - other[i] = p[i].ColumnKey - } - return other -} - -// Timestamps returns a slice of all the timestamps. -func (p BitKs) Timestamps() []int64 { - other := make([]int64, len(p)) - for i := range p { - other[i] = p[i].Timestamp - } - return other -} - // FieldValues represents the value for a column within a // range-encoded frame. type FieldValue struct { @@ -1355,7 +1338,7 @@ type InternalClient interface { FragmentNodes(ctx context.Context, index string, slice uint64) ([]*Node, error) ExecuteQuery(ctx context.Context, index string, queryRequest *internal.QueryRequest) (*internal.QueryResponse, error) Import(ctx context.Context, index, frame string, slice uint64, bits []Bit) error - ImportK(ctx context.Context, index, frame string, bitKs []BitK) error + ImportK(ctx context.Context, index, frame string, bits []Bit) error EnsureIndex(ctx context.Context, name string, options IndexOptions) error EnsureFrame(ctx context.Context, indexName string, frameName string, options FrameOptions) error ImportValue(ctx context.Context, index, frame, field string, slice uint64, vals []FieldValue) error diff --git a/ctl/import.go b/ctl/import.go index b6c1b6f50..8103bf673 100644 --- a/ctl/import.go +++ b/ctl/import.go @@ -135,7 +135,7 @@ func (cmd *ImportCommand) importPath(ctx context.Context, path string) error { return cmd.bufferFieldValues(ctx, path) } else { if cmd.StringKeys { - return cmd.bufferBitKs(ctx, path) + return cmd.bufferBitsK(ctx, path) } else { return cmd.bufferBits(ctx, path) } @@ -249,9 +249,9 @@ func (cmd *ImportCommand) importBits(ctx context.Context, bits []pilosa.Bit) err return nil } -// bufferBitKs buffers slices of keys to be imported as a batch. -func (cmd *ImportCommand) bufferBitKs(ctx context.Context, path string) error { - a := make([]pilosa.BitK, 0, cmd.BufferSize) +// bufferBitsK buffers slices of keys to be imported as a batch. +func (cmd *ImportCommand) bufferBitsK(ctx context.Context, path string) error { + a := make([]pilosa.Bit, 0, cmd.BufferSize) var r *csv.Reader @@ -289,19 +289,19 @@ func (cmd *ImportCommand) bufferBitKs(ctx context.Context, path string) error { return fmt.Errorf("bad column count on row %d: col=%d", rnum, len(record)) } - var bitK pilosa.BitK + var bit pilosa.Bit // Parse row key. if record[0] == "" { return fmt.Errorf("invalid row key on row %d: %q", rnum, record[0]) } - bitK.RowKey = record[0] + bit.RowKey = record[0] // Parse column key. if record[1] == "" { return fmt.Errorf("invalid column id on row %d: %q", rnum, record[1]) } - bitK.ColumnKey = record[1] + bit.ColumnKey = record[1] // Parse time, if exists. if len(record) > 2 && record[2] != "" { @@ -309,14 +309,14 @@ func (cmd *ImportCommand) bufferBitKs(ctx context.Context, path string) error { if err != nil { return fmt.Errorf("invalid timestamp on row %d: %q", rnum, record[2]) } - bitK.Timestamp = t.UnixNano() + bit.Timestamp = t.UnixNano() } - a = append(a, bitK) + a = append(a, bit) - // If we've reached the buffer size then import bitKs. + // If we've reached the buffer size then import bits. if len(a) == cmd.BufferSize { - if err := cmd.importBitKs(ctx, a); err != nil { + if err := cmd.importBitsK(ctx, a); err != nil { return err } a = a[:0] @@ -324,21 +324,21 @@ func (cmd *ImportCommand) bufferBitKs(ctx context.Context, path string) error { } // If there are still bitKs in the buffer then flush them. - if err := cmd.importBitKs(ctx, a); err != nil { + if err := cmd.importBitsK(ctx, a); err != nil { return err } return nil } -// importBitKs sends batches of bitKs to the server. -func (cmd *ImportCommand) importBitKs(ctx context.Context, bitKs []pilosa.BitK) error { +// importBitsK sends batches of bitKs to the server. +func (cmd *ImportCommand) importBitsK(ctx context.Context, bits []pilosa.Bit) error { logger := log.New(cmd.Stderr, "", log.LstdFlags) // TODO: does it help to sort the rowKeys? - logger.Printf("importing keys: n=%d", len(bitKs)) - if err := cmd.Client.ImportK(ctx, cmd.Index, cmd.Frame, bitKs); err != nil { + logger.Printf("importing keys: n=%d", len(bits)) + if err := cmd.Client.ImportK(ctx, cmd.Index, cmd.Frame, bits); err != nil { return err } diff --git a/internal/public.pb.go b/internal/public.pb.go index 83a402247..dfb9a819e 100644 --- a/internal/public.pb.go +++ b/internal/public.pb.go @@ -20,7 +20,6 @@ QueryResponse QueryResult ImportRequest - ImportKRequest ImportValueRequest */ package internal @@ -416,6 +415,8 @@ type ImportRequest struct { Slice uint64 `protobuf:"varint,3,opt,name=Slice,proto3" json:"Slice,omitempty"` RowIDs []uint64 `protobuf:"varint,4,rep,packed,name=RowIDs" json:"RowIDs,omitempty"` ColumnIDs []uint64 `protobuf:"varint,5,rep,packed,name=ColumnIDs" json:"ColumnIDs,omitempty"` + RowKeys []string `protobuf:"bytes,7,rep,name=RowKeys" json:"RowKeys,omitempty"` + ColumnKeys []string `protobuf:"bytes,8,rep,name=ColumnKeys" json:"ColumnKeys,omitempty"` Timestamps []int64 `protobuf:"varint,6,rep,packed,name=Timestamps" json:"Timestamps,omitempty"` } @@ -459,6 +460,20 @@ func (m *ImportRequest) GetColumnIDs() []uint64 { return nil } +func (m *ImportRequest) GetRowKeys() []string { + if m != nil { + return m.RowKeys + } + return nil +} + +func (m *ImportRequest) GetColumnKeys() []string { + if m != nil { + return m.ColumnKeys + } + return nil +} + func (m *ImportRequest) GetTimestamps() []int64 { if m != nil { return m.Timestamps @@ -466,67 +481,20 @@ func (m *ImportRequest) GetTimestamps() []int64 { return nil } -type ImportKRequest struct { +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"` - RowKeys []string `protobuf:"bytes,3,rep,name=RowKeys" json:"RowKeys,omitempty"` - ColumnKeys []string `protobuf:"bytes,4,rep,name=ColumnKeys" json:"ColumnKeys,omitempty"` - Timestamps []int64 `protobuf:"varint,5,rep,packed,name=Timestamps" json:"Timestamps,omitempty"` -} - -func (m *ImportKRequest) Reset() { *m = ImportKRequest{} } -func (m *ImportKRequest) String() string { return proto.CompactTextString(m) } -func (*ImportKRequest) ProtoMessage() {} -func (*ImportKRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} } - -func (m *ImportKRequest) GetIndex() string { - if m != nil { - return m.Index - } - return "" -} - -func (m *ImportKRequest) GetFrame() string { - if m != nil { - return m.Frame - } - return "" -} - -func (m *ImportKRequest) GetRowKeys() []string { - if m != nil { - return m.RowKeys - } - return nil -} - -func (m *ImportKRequest) GetColumnKeys() []string { - if m != nil { - return m.ColumnKeys - } - return nil -} - -func (m *ImportKRequest) GetTimestamps() []int64 { - if m != nil { - return m.Timestamps - } - return nil -} - -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 []int64 `protobuf:"varint,6,rep,packed,name=Values" json:"Values,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"` + ColumnKeys []string `protobuf:"bytes,7,rep,name=ColumnKeys" json:"ColumnKeys,omitempty"` + Values []int64 `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{12} } +func (*ImportValueRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} } func (m *ImportValueRequest) GetIndex() string { if m != nil { @@ -563,6 +531,13 @@ func (m *ImportValueRequest) GetColumnIDs() []uint64 { return nil } +func (m *ImportValueRequest) GetColumnKeys() []string { + if m != nil { + return m.ColumnKeys + } + return nil +} + func (m *ImportValueRequest) GetValues() []int64 { if m != nil { return m.Values @@ -582,7 +557,6 @@ func init() { proto.RegisterType((*QueryResponse)(nil), "internal.QueryResponse") proto.RegisterType((*QueryResult)(nil), "internal.QueryResult") proto.RegisterType((*ImportRequest)(nil), "internal.ImportRequest") - proto.RegisterType((*ImportKRequest)(nil), "internal.ImportKRequest") proto.RegisterType((*ImportValueRequest)(nil), "internal.ImportValueRequest") } func (m *Bitmap) Marshal() (dAtA []byte, err error) { @@ -1151,39 +1125,9 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) { i = encodeVarintPublic(dAtA, i, uint64(j11)) i += copy(dAtA[i:], dAtA12[:j11]) } - return i, nil -} - -func (m *ImportKRequest) 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 *ImportKRequest) 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 len(m.RowKeys) > 0 { for _, s := range m.RowKeys { - dAtA[i] = 0x1a + dAtA[i] = 0x3a i++ l = len(s) for l >= 1<<7 { @@ -1198,7 +1142,7 @@ func (m *ImportKRequest) MarshalTo(dAtA []byte) (int, error) { } if len(m.ColumnKeys) > 0 { for _, s := range m.ColumnKeys { - dAtA[i] = 0x22 + dAtA[i] = 0x42 i++ l = len(s) for l >= 1<<7 { @@ -1211,24 +1155,6 @@ func (m *ImportKRequest) MarshalTo(dAtA []byte) (int, error) { i += copy(dAtA[i:], s) } } - if len(m.Timestamps) > 0 { - dAtA14 := make([]byte, len(m.Timestamps)*10) - var j13 int - for _, num1 := range m.Timestamps { - num := uint64(num1) - 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]) - } return i, nil } @@ -1271,9 +1197,27 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { i += copy(dAtA[i:], m.Field) } if len(m.ColumnIDs) > 0 { - dAtA16 := make([]byte, len(m.ColumnIDs)*10) - var j15 int + 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 _, num1 := range m.Values { + num := uint64(num1) for num >= 1<<7 { dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 @@ -1282,28 +1226,25 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { dAtA16[j15] = uint8(num) j15++ } - dAtA[i] = 0x2a + dAtA[i] = 0x32 i++ i = encodeVarintPublic(dAtA, i, uint64(j15)) i += copy(dAtA[i:], dAtA16[:j15]) } - if len(m.Values) > 0 { - dAtA18 := make([]byte, len(m.Values)*10) - var j17 int - for _, num1 := range m.Values { - num := uint64(num1) - for num >= 1<<7 { - dAtA18[j17] = uint8(uint64(num)&0x7f | 0x80) - num >>= 7 - j17++ + if len(m.ColumnKeys) > 0 { + for _, s := range m.ColumnKeys { + dAtA[i] = 0x3a + i++ + l = len(s) + for l >= 1<<7 { + dAtA[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ } - dAtA18[j17] = uint8(num) - j17++ + dAtA[i] = uint8(l) + i++ + i += copy(dAtA[i:], s) } - dAtA[i] = 0x32 - i++ - i = encodeVarintPublic(dAtA, i, uint64(j17)) - i += copy(dAtA[i:], dAtA18[:j17]) } return i, nil } @@ -1575,20 +1516,6 @@ func (m *ImportRequest) Size() (n int) { } n += 1 + sovPublic(uint64(l)) + l } - return n -} - -func (m *ImportKRequest) 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 len(m.RowKeys) > 0 { for _, s := range m.RowKeys { l = len(s) @@ -1601,13 +1528,6 @@ func (m *ImportKRequest) Size() (n int) { n += 1 + l + sovPublic(uint64(l)) } } - if len(m.Timestamps) > 0 { - l = 0 - for _, e := range m.Timestamps { - l += sovPublic(uint64(e)) - } - n += 1 + sovPublic(uint64(l)) + l - } return n } @@ -1643,6 +1563,12 @@ func (m *ImportValueRequest) Size() (n int) { } n += 1 + sovPublic(uint64(l)) + l } + if len(m.ColumnKeys) > 0 { + for _, s := range m.ColumnKeys { + l = len(s) + n += 1 + l + sovPublic(uint64(l)) + } + } return n } @@ -3396,115 +3322,7 @@ func (m *ImportRequest) Unmarshal(dAtA []byte) error { } else { return fmt.Errorf("proto: wrong wireType = %d for field Timestamps", 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 (m *ImportKRequest) 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: ImportKRequest: wiretype end group for non-group") - } - if fieldNum <= 0 { - return fmt.Errorf("proto: ImportKRequest: 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: + case 7: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field RowKeys", wireType) } @@ -3533,7 +3351,7 @@ func (m *ImportKRequest) Unmarshal(dAtA []byte) error { } m.RowKeys = append(m.RowKeys, string(dAtA[iNdEx:postIndex])) iNdEx = postIndex - case 4: + case 8: if wireType != 2 { return fmt.Errorf("proto: wrong wireType = %d for field ColumnKeys", wireType) } @@ -3562,68 +3380,6 @@ func (m *ImportKRequest) Unmarshal(dAtA []byte) error { } m.ColumnKeys = append(m.ColumnKeys, string(dAtA[iNdEx:postIndex])) iNdEx = postIndex - case 5: - if wireType == 0 { - var v int64 - for shift := uint(0); ; shift += 7 { - if shift >= 64 { - return ErrIntOverflowPublic - } - if iNdEx >= l { - return io.ErrUnexpectedEOF - } - b := dAtA[iNdEx] - iNdEx++ - v |= (int64(b) & 0x7F) << shift - if b < 0x80 { - break - } - } - m.Timestamps = append(m.Timestamps, v) - } else 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 int64 - for shift := uint(0); ; shift += 7 { - if shift >= 64 { - return ErrIntOverflowPublic - } - if iNdEx >= l { - return io.ErrUnexpectedEOF - } - b := dAtA[iNdEx] - iNdEx++ - v |= (int64(b) & 0x7F) << shift - if b < 0x80 { - break - } - } - m.Timestamps = append(m.Timestamps, v) - } - } else { - return fmt.Errorf("proto: wrong wireType = %d for field Timestamps", wireType) - } default: iNdEx = preIndex skippy, err := skipPublic(dAtA[iNdEx:]) @@ -3904,6 +3660,35 @@ func (m *ImportValueRequest) Unmarshal(dAtA []byte) error { } else { return fmt.Errorf("proto: wrong wireType = %d for field Values", wireType) } + case 7: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ColumnKeys", 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.ColumnKeys = append(m.ColumnKeys, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPublic(dAtA[iNdEx:]) @@ -4033,50 +3818,50 @@ var ( func init() { proto.RegisterFile("public.proto", fileDescriptorPublic) } var fileDescriptorPublic = []byte{ - // 712 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcb, 0x6a, 0x14, 0x41, - 0x14, 0xb5, 0xa6, 0x7b, 0x5e, 0x77, 0x1e, 0x84, 0x42, 0x63, 0x23, 0x32, 0x34, 0x8d, 0x8b, 0x5e, - 0x4d, 0x60, 0xdc, 0x2b, 0x4e, 0x1e, 0x30, 0x04, 0x83, 0xd6, 0xc4, 0xb8, 0xee, 0x24, 0x45, 0x6c, - 0xe8, 0x97, 0xdd, 0xd5, 0x24, 0xf3, 0x1d, 0x82, 0xb8, 0x76, 0xa3, 0x1f, 0xe1, 0x07, 0xe8, 0xce, - 0x4f, 0x90, 0xf8, 0x23, 0x72, 0x6f, 0x75, 0x4d, 0xf7, 0x24, 0xa0, 0x22, 0xee, 0xea, 0x9c, 0x53, - 0x75, 0xfb, 0x3e, 0xce, 0x9d, 0x81, 0x61, 0x56, 0x9e, 0x46, 0xe1, 0xd9, 0x34, 0xcb, 0x53, 0x95, - 0xf2, 0x5e, 0x98, 0x28, 0x99, 0x27, 0x41, 0xe4, 0x9d, 0x40, 0x67, 0x1e, 0xaa, 0x38, 0xc8, 0x38, - 0x07, 0x7b, 0x1e, 0xaa, 0xc2, 0x61, 0xae, 0xe5, 0xdb, 0x82, 0xce, 0xfc, 0x11, 0xb4, 0x9f, 0x29, - 0x95, 0x17, 0x4e, 0xcb, 0xb5, 0xfc, 0xc1, 0x6c, 0x3c, 0x35, 0xef, 0xa6, 0x48, 0x0b, 0x2d, 0xe2, - 0xcb, 0x43, 0xb9, 0x2a, 0x1c, 0xcb, 0xb5, 0xfc, 0xbe, 0xa0, 0xb3, 0xf7, 0x04, 0xec, 0x17, 0x41, - 0x98, 0xf3, 0x31, 0xb4, 0x16, 0x7b, 0x0e, 0x73, 0x99, 0x6f, 0x8b, 0xd6, 0x62, 0x8f, 0xdf, 0x85, - 0xf6, 0x6e, 0x5a, 0x26, 0xca, 0x69, 0x11, 0xa5, 0x01, 0xdf, 0x02, 0xeb, 0x50, 0xae, 0x1c, 0xcb, - 0x65, 0x7e, 0x5f, 0xe0, 0xd1, 0x9b, 0x41, 0x6f, 0x59, 0xc6, 0x6b, 0x75, 0x59, 0xc6, 0x14, 0xc4, - 0x12, 0x78, 0xdc, 0x8c, 0x62, 0x55, 0x51, 0xbc, 0x57, 0x60, 0xcd, 0x43, 0x85, 0xa2, 0x48, 0x2f, - 0xd7, 0x5f, 0xd5, 0x80, 0x3f, 0x80, 0xde, 0x6e, 0x1a, 0x95, 0x71, 0xb2, 0xd8, 0xab, 0xbe, 0xbd, - 0xc6, 0xfc, 0x21, 0xf4, 0x8f, 0xc3, 0x58, 0x16, 0x2a, 0x88, 0x33, 0x4a, 0xc2, 0x12, 0x35, 0xe1, - 0xbd, 0x86, 0x91, 0xbe, 0x89, 0xd5, 0x2e, 0xa5, 0xba, 0x55, 0xd3, 0xdf, 0x75, 0xe9, 0x76, 0x8d, - 0x9f, 0x19, 0xd8, 0xa8, 0x19, 0x89, 0xad, 0x25, 0x6c, 0xe9, 0xf1, 0x2a, 0x93, 0x55, 0xa6, 0x74, - 0xe6, 0x2e, 0x0c, 0x96, 0x2a, 0x0f, 0x93, 0x8b, 0x93, 0x20, 0x2a, 0x65, 0x15, 0xa8, 0x49, 0x61, - 0x8d, 0x8b, 0x44, 0x69, 0xd9, 0xa6, 0x32, 0xd6, 0x18, 0x6b, 0x9c, 0xa7, 0x69, 0xa4, 0xc5, 0xb6, - 0xcb, 0xfc, 0x9e, 0xa8, 0x09, 0x3e, 0x01, 0x38, 0x88, 0xd2, 0xa0, 0x7a, 0xdb, 0x71, 0x99, 0xcf, - 0x44, 0x83, 0xf1, 0x76, 0xa0, 0x8b, 0x99, 0x3e, 0x0f, 0xb2, 0xba, 0x5a, 0xf6, 0x9b, 0x6a, 0xbd, - 0x2f, 0x0c, 0x86, 0x2f, 0x4b, 0x99, 0xaf, 0x84, 0x7c, 0x5b, 0xca, 0x82, 0xa6, 0x42, 0xb8, 0xaa, - 0x52, 0x03, 0xbe, 0x0d, 0x9d, 0x65, 0x14, 0x9e, 0x49, 0xdd, 0x3b, 0x5b, 0x54, 0x08, 0x6b, 0xad, - 0x7b, 0x5e, 0x50, 0xad, 0x3d, 0xd1, 0xa4, 0xf0, 0xa5, 0x90, 0x71, 0xaa, 0x4c, 0x31, 0x15, 0xe2, - 0x1e, 0x0c, 0xf7, 0xaf, 0xce, 0xa2, 0xf2, 0x5c, 0xea, 0xa7, 0x1d, 0x52, 0x37, 0x38, 0x8c, 0x5e, - 0x61, 0x72, 0x7c, 0x57, 0x47, 0x6f, 0x50, 0xde, 0x3b, 0x06, 0xa3, 0x2a, 0xfd, 0x22, 0x4b, 0x93, - 0x42, 0xe2, 0x8c, 0xf6, 0xf3, 0xdc, 0xcc, 0x68, 0x3f, 0xcf, 0xf9, 0x0e, 0x74, 0x85, 0x2c, 0xca, - 0x48, 0x99, 0xc1, 0xdf, 0xab, 0x5b, 0x61, 0xde, 0x96, 0x91, 0x12, 0xe6, 0x16, 0x7f, 0x0a, 0xe3, - 0x0d, 0x23, 0xe9, 0x8d, 0x19, 0xcc, 0xee, 0xd7, 0xef, 0x36, 0x74, 0x71, 0xe3, 0xba, 0xf7, 0x8d, - 0xc1, 0xa0, 0x11, 0x99, 0xfb, 0x66, 0x79, 0x29, 0xad, 0xc1, 0x6c, 0xab, 0x0e, 0xa4, 0x79, 0x61, - 0x96, 0x7b, 0x08, 0xec, 0xa8, 0x32, 0x13, 0x3b, 0xc2, 0x11, 0xe2, 0x72, 0x9a, 0xef, 0x37, 0x46, - 0x88, 0xb4, 0xd0, 0x22, 0x77, 0xa0, 0xbb, 0xfb, 0x26, 0x48, 0x2e, 0xe4, 0x39, 0x99, 0xa9, 0x27, - 0x0c, 0xe4, 0xd3, 0x7a, 0x39, 0xa9, 0xfb, 0x83, 0x19, 0xaf, 0x43, 0x18, 0x45, 0xd4, 0x0b, 0x6c, - 0xdc, 0x8c, 0xb3, 0x18, 0x69, 0x37, 0x7b, 0x9f, 0x18, 0x8c, 0x16, 0x71, 0x96, 0xe6, 0xaa, 0xe1, - 0x90, 0x45, 0x72, 0x2e, 0xaf, 0x8c, 0x43, 0x08, 0x20, 0x7b, 0x90, 0x07, 0xb1, 0x5e, 0x85, 0xbe, - 0xd0, 0x00, 0x59, 0x72, 0x0a, 0x39, 0xc3, 0x16, 0x1a, 0x90, 0x27, 0x70, 0xd9, 0x0b, 0xc7, 0xd6, - 0x6e, 0xd2, 0x08, 0xbd, 0x6f, 0x76, 0xbd, 0x70, 0xda, 0x24, 0xd5, 0x04, 0x7a, 0x7f, 0xbd, 0xec, - 0xe8, 0x17, 0xcb, 0xb7, 0x44, 0x83, 0xf1, 0xde, 0x33, 0x18, 0xeb, 0x4c, 0x0f, 0xff, 0x25, 0x55, - 0x07, 0xba, 0x22, 0xbd, 0x6c, 0xfc, 0x40, 0x1a, 0x88, 0x1f, 0xd6, 0x59, 0x90, 0x68, 0x93, 0xd8, - 0x60, 0x6e, 0x24, 0xd6, 0xbe, 0x95, 0xd8, 0x47, 0x06, 0x5c, 0x27, 0x46, 0x4b, 0xfa, 0xff, 0xfa, - 0x88, 0x77, 0x43, 0x19, 0xe9, 0xb9, 0xe3, 0x5d, 0x04, 0x7f, 0xe8, 0xe2, 0x36, 0x74, 0x28, 0x0b, - 0xd3, 0xc1, 0x0a, 0xcd, 0xb7, 0xbe, 0x5e, 0x4f, 0xd8, 0xf7, 0xeb, 0x09, 0xfb, 0x71, 0x3d, 0x61, - 0x1f, 0x7e, 0x4e, 0xee, 0x9c, 0x76, 0xe8, 0x3f, 0xe8, 0xf1, 0xaf, 0x00, 0x00, 0x00, 0xff, 0xff, - 0xf3, 0x62, 0xbb, 0x7d, 0x93, 0x06, 0x00, 0x00, + // 705 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcb, 0x6e, 0xd3, 0x40, + 0x14, 0x65, 0x62, 0x27, 0x71, 0x6e, 0x92, 0xaa, 0x1a, 0x41, 0xb1, 0x10, 0x8a, 0x2c, 0x8b, 0x85, + 0x57, 0xa9, 0x14, 0xf6, 0x20, 0xd2, 0x87, 0x14, 0x55, 0x54, 0x30, 0x29, 0x65, 0xed, 0xb6, 0xa3, + 0x62, 0xc9, 0x2f, 0xec, 0xb1, 0xda, 0x7c, 0x07, 0x1b, 0x3e, 0x81, 0x8f, 0x60, 0xc5, 0x0a, 0x76, + 0x7c, 0x02, 0x94, 0x1f, 0x41, 0xf7, 0x8e, 0x27, 0x76, 0x5a, 0x09, 0x58, 0xb0, 0x9b, 0x73, 0xce, + 0xcc, 0xf5, 0x9c, 0xb9, 0xe7, 0x26, 0x30, 0xca, 0xab, 0xb3, 0x38, 0x3a, 0x9f, 0xe6, 0x45, 0xa6, + 0x32, 0xee, 0x44, 0xa9, 0x92, 0x45, 0x1a, 0xc6, 0xfe, 0x29, 0xf4, 0xe6, 0x91, 0x4a, 0xc2, 0x9c, + 0x73, 0xb0, 0xe7, 0x91, 0x2a, 0x5d, 0xe6, 0x59, 0x81, 0x2d, 0x68, 0xcd, 0x9f, 0x40, 0xf7, 0x85, + 0x52, 0x45, 0xe9, 0x76, 0x3c, 0x2b, 0x18, 0xce, 0xb6, 0xa6, 0xe6, 0xdc, 0x14, 0x69, 0xa1, 0x45, + 0x3c, 0x79, 0x24, 0x57, 0xa5, 0x6b, 0x79, 0x56, 0x30, 0x10, 0xb4, 0xf6, 0x9f, 0x81, 0xfd, 0x2a, + 0x8c, 0x0a, 0xbe, 0x05, 0x9d, 0xc5, 0xbe, 0xcb, 0x3c, 0x16, 0xd8, 0xa2, 0xb3, 0xd8, 0xe7, 0xf7, + 0xa1, 0xbb, 0x97, 0x55, 0xa9, 0x72, 0x3b, 0x44, 0x69, 0xc0, 0xb7, 0xc1, 0x3a, 0x92, 0x2b, 0xd7, + 0xf2, 0x58, 0x30, 0x10, 0xb8, 0xf4, 0x67, 0xe0, 0x2c, 0xab, 0x64, 0xad, 0x2e, 0xab, 0x84, 0x8a, + 0x58, 0x02, 0x97, 0x9b, 0x55, 0xac, 0xba, 0x8a, 0xff, 0x06, 0xac, 0x79, 0xa4, 0x50, 0x14, 0xd9, + 0xd5, 0xfa, 0xab, 0x1a, 0xf0, 0x47, 0xe0, 0xec, 0x65, 0x71, 0x95, 0xa4, 0x8b, 0xfd, 0xfa, 0xdb, + 0x6b, 0xcc, 0x1f, 0xc3, 0xe0, 0x24, 0x4a, 0x64, 0xa9, 0xc2, 0x24, 0xa7, 0x4b, 0x58, 0xa2, 0x21, + 0xfc, 0xb7, 0x30, 0xd6, 0x3b, 0xd1, 0xed, 0x52, 0xaa, 0x3b, 0x9e, 0xfe, 0xed, 0x95, 0xee, 0x7a, + 0xfc, 0xc4, 0xc0, 0x46, 0xcd, 0x48, 0x6c, 0x2d, 0xe1, 0x93, 0x9e, 0xac, 0x72, 0x59, 0xdf, 0x94, + 0xd6, 0xdc, 0x83, 0xe1, 0x52, 0x15, 0x51, 0x7a, 0x79, 0x1a, 0xc6, 0x95, 0xac, 0x0b, 0xb5, 0x29, + 0xf4, 0xb8, 0x48, 0x95, 0x96, 0x6d, 0xb2, 0xb1, 0xc6, 0xe8, 0x71, 0x9e, 0x65, 0xb1, 0x16, 0xbb, + 0x1e, 0x0b, 0x1c, 0xd1, 0x10, 0x7c, 0x02, 0x70, 0x18, 0x67, 0x61, 0x7d, 0xb6, 0xe7, 0xb1, 0x80, + 0x89, 0x16, 0xe3, 0xef, 0x42, 0x1f, 0x6f, 0xfa, 0x32, 0xcc, 0x1b, 0xb7, 0xec, 0x0f, 0x6e, 0xfd, + 0xcf, 0x0c, 0x46, 0xaf, 0x2b, 0x59, 0xac, 0x84, 0x7c, 0x5f, 0xc9, 0x92, 0xba, 0x42, 0xb8, 0x76, + 0xa9, 0x01, 0xdf, 0x81, 0xde, 0x32, 0x8e, 0xce, 0xa5, 0x7e, 0x3b, 0x5b, 0xd4, 0x08, 0xbd, 0x36, + 0x6f, 0x5e, 0x92, 0x57, 0x47, 0xb4, 0x29, 0x3c, 0x29, 0x64, 0x92, 0x29, 0x63, 0xa6, 0x46, 0xdc, + 0x87, 0xd1, 0xc1, 0xf5, 0x79, 0x5c, 0x5d, 0x48, 0x7d, 0xb4, 0x47, 0xea, 0x06, 0x87, 0xd5, 0x6b, + 0x4c, 0x89, 0xef, 0xeb, 0xea, 0x2d, 0xca, 0xff, 0xc0, 0x60, 0x5c, 0x5f, 0xbf, 0xcc, 0xb3, 0xb4, + 0x94, 0xd8, 0xa3, 0x83, 0xa2, 0x30, 0x3d, 0x3a, 0x28, 0x0a, 0xbe, 0x0b, 0x7d, 0x21, 0xcb, 0x2a, + 0x56, 0xa6, 0xf1, 0x0f, 0x9a, 0xa7, 0x30, 0x67, 0xab, 0x58, 0x09, 0xb3, 0x8b, 0x3f, 0x87, 0xad, + 0x8d, 0x20, 0xe9, 0x89, 0x19, 0xce, 0x1e, 0x36, 0xe7, 0x36, 0x74, 0x71, 0x6b, 0xbb, 0xff, 0x8d, + 0xc1, 0xb0, 0x55, 0x99, 0x07, 0x66, 0x78, 0xe9, 0x5a, 0xc3, 0xd9, 0x76, 0x53, 0x48, 0xf3, 0xc2, + 0x0c, 0xf7, 0x08, 0xd8, 0x71, 0x1d, 0x26, 0x76, 0x8c, 0x2d, 0xc4, 0xe1, 0x34, 0xdf, 0x6f, 0xb5, + 0x10, 0x69, 0xa1, 0x45, 0xee, 0x42, 0x7f, 0xef, 0x5d, 0x98, 0x5e, 0xca, 0x0b, 0x0a, 0x93, 0x23, + 0x0c, 0xe4, 0xd3, 0x66, 0x38, 0xe9, 0xf5, 0x87, 0x33, 0xde, 0x94, 0x30, 0x8a, 0x68, 0x06, 0xd8, + 0xa4, 0x19, 0x7b, 0x31, 0xd6, 0x69, 0xf6, 0x7f, 0x32, 0x18, 0x2f, 0x92, 0x3c, 0x2b, 0x54, 0x2b, + 0x21, 0x8b, 0xf4, 0x42, 0x5e, 0x9b, 0x84, 0x10, 0x40, 0xf6, 0xb0, 0x08, 0x13, 0x3d, 0x0a, 0x03, + 0xa1, 0x01, 0xb2, 0x94, 0x14, 0x4a, 0x86, 0x2d, 0x34, 0xa0, 0x4c, 0xe0, 0xb0, 0x97, 0xae, 0xad, + 0xd3, 0xa4, 0x11, 0x66, 0xdf, 0xcc, 0x7a, 0xe9, 0x76, 0x49, 0x6a, 0x08, 0xcc, 0xfe, 0x7a, 0xd8, + 0x31, 0x2f, 0x56, 0x60, 0x89, 0x16, 0x83, 0xef, 0x20, 0xb2, 0x2b, 0xfa, 0x85, 0xeb, 0xd3, 0x2f, + 0x9c, 0x81, 0x78, 0x52, 0x97, 0x21, 0xd1, 0x21, 0xb1, 0xc5, 0xf8, 0x5f, 0x18, 0x70, 0xed, 0x91, + 0xa6, 0xe8, 0xff, 0x19, 0xc5, 0xbd, 0x91, 0x8c, 0x75, 0x63, 0x70, 0x2f, 0x82, 0xbf, 0xd8, 0xdc, + 0x81, 0x1e, 0xdd, 0xc2, 0x58, 0xac, 0xd1, 0x2d, 0x13, 0xfd, 0xdb, 0x26, 0xe6, 0xdb, 0x5f, 0x6f, + 0x26, 0xec, 0xfb, 0xcd, 0x84, 0xfd, 0xb8, 0x99, 0xb0, 0x8f, 0xbf, 0x26, 0xf7, 0xce, 0x7a, 0xf4, + 0x27, 0xf2, 0xf4, 0x77, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa3, 0xa0, 0xd2, 0x51, 0x54, 0x06, 0x00, + 0x00, } diff --git a/internal/public.proto b/internal/public.proto index 649fc0e3d..5fa40cfab 100644 --- a/internal/public.proto +++ b/internal/public.proto @@ -74,22 +74,17 @@ message ImportRequest { uint64 Slice = 3; repeated uint64 RowIDs = 4; repeated uint64 ColumnIDs = 5; + repeated string RowKeys = 7; + repeated string ColumnKeys = 8; repeated int64 Timestamps = 6; } -message ImportKRequest { - string Index = 1; - string Frame = 2; - repeated string RowKeys = 3; - repeated string ColumnKeys = 4; - repeated int64 Timestamps = 5; -} - message ImportValueRequest { string Index = 1; string Frame = 2; uint64 Slice = 3; string Field = 4; repeated uint64 ColumnIDs = 5; + repeated string ColumnKeys = 7; repeated int64 Values = 6; } From 5fa7dfd63d59c5ab092e03df1491f688e6f5f415 Mon Sep 17 00:00:00 2001 From: Travis Turner Date: Mon, 22 Jan 2018 18:04:41 -0600 Subject: [PATCH 3/3] allow for QueryResultTypeNil --- handler.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/handler.go b/handler.go index b81b8e124..e5cce4648 100644 --- a/handler.go +++ b/handler.go @@ -1627,7 +1627,8 @@ func (h *Handler) logger() *log.Logger { // QueryResult types. const ( - QueryResultTypeBitmap uint32 = iota + QueryResultTypeNil uint32 = iota + QueryResultTypeBitmap QueryResultTypePairs QueryResultTypeSumCount QueryResultTypeUint64 @@ -1727,6 +1728,8 @@ func encodeQueryResponse(resp *QueryResponse) *internal.QueryResponse { case bool: pb.Results[i].Type = QueryResultTypeBool pb.Results[i].Changed = result + case nil: + pb.Results[i].Type = QueryResultTypeNil } }