[FB-1776] support for marshaling ExtractedIDMatrixSorted (#2319)

support for marshaling ExtractedIDMatrixSorted

encoding/proto/proto.go

@@ -557,6 +557,9 @@ func (s Serializer) encodeQueryResponse(m *pilosa.QueryResponse) *pb.QueryRespon
 		case arrow.Table:
 			resp.Results[i].Type = queryResultTypeArrowTable
 			resp.Results[i].ArrowTable = s.encodeArrowTable(result)
+		case pilosa.ExtractedIDMatrixSorted:
+			resp.Results[i].Type = queryResultTypeExtractedIDMatrixSorted
+			resp.Results[i].ExtractedIDMatrixSorted = s.endcodeExtractedIDMatrixSorted(result)
 		default:
 			panic(fmt.Errorf("unknown type: %T", m.Results[i]))
 		}
@@ -1338,6 +1341,7 @@ const (
 	queryResultTypeDistinctTimestamp
 	queryResultTypeDataFrame
 	queryResultTypeArrowTable
+	queryResultTypeExtractedIDMatrixSorted
 )
 
 func (s Serializer) decodeQueryResult(pb *pb.QueryResult) interface{} {
@@ -1380,6 +1384,8 @@ func (s Serializer) decodeQueryResult(pb *pb.QueryResult) interface{} {
 		return s.decodeDataFrame(pb.DataFrame)
 	case queryResultTypeArrowTable:
 		return s.decodeArrowTable(pb.ArrowTable)
+	case queryResultTypeExtractedIDMatrixSorted:
+		return s.decodeExtractedIDMatrixSorted(pb.ExtractedIDMatrixSorted)
 	}
 	panic(fmt.Sprintf("unknown type: %d", pb.Type))
 }
@@ -1462,24 +1468,8 @@ func (s Serializer) decodeExtractedTable(t *pb.ExtractedTable) pilosa.ExtractedT
 		}
 
 		rows := make([]interface{}, len(c.Values))
-		for j, v := range rows {
-			var val interface{}
-			switch v := v.(type) {
-			case *pb.ExtractedTableValue_IDs:
-				val = v.IDs.IDs
-			case *pb.ExtractedTableValue_Keys:
-				val = v.Keys.Keys
-			case *pb.ExtractedTableValue_BSIValue:
-				val = v.BSIValue
-			case *pb.ExtractedTableValue_MutexID:
-				val = v.MutexID
-			case *pb.ExtractedTableValue_MutexKey:
-				val = v.MutexKey
-			case *pb.ExtractedTableValue_Bool:
-				val = v.Bool
-			}
-
-			rows[j] = val
+		for j, v := range c.Values {
+			rows[j] = castExtractedTableValue(v)
 		}
 
 		columns[i] = pilosa.ExtractedTableColumn{
@@ -1878,3 +1868,117 @@ func (s Serializer) encodeDecimal(p *pql.Decimal) *pb.Decimal {
 	}
 	return retval
 }
+
+func (s Serializer) endcodeExtractedIDMatrixSorted(m pilosa.ExtractedIDMatrixSorted) *pb.ExtractedIDMatrixSorted {
+	cols := make([]*pb.ExtractedIDColumn, len(m.ExtractedIDMatrix.Columns))
+	for i, v := range m.ExtractedIDMatrix.Columns {
+		vals := make([]*pb.IDList, len(v.Rows))
+		for j, f := range v.Rows {
+			vals[j] = &pb.IDList{IDs: f}
+		}
+		cols[i] = &pb.ExtractedIDColumn{
+			ID:   v.ColumnID,
+			Vals: vals,
+		}
+	}
+	rowKeyV := s.encodeRowKVs(m.RowKVs)
+	return &pb.ExtractedIDMatrixSorted{
+		ExtractedIDMatrix: &pb.ExtractedIDMatrix{
+			Fields:  m.ExtractedIDMatrix.Fields,
+			Columns: cols,
+		},
+		RowKVs: rowKeyV,
+	}
+}
+
+func (s Serializer) encodeRowKVs(kvs []pilosa.RowKV) []*pb.RowKV {
+	result := make([]*pb.RowKV, len(kvs))
+	for i, v := range kvs {
+		kv := &pb.RowKV{}
+		kv.RowID = v.RowID
+		switch val := v.Value.(type) {
+		case []uint64:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_IDs{
+					IDs: &pb.IDList{
+						IDs: val,
+					},
+				},
+			}
+		case []string:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_Keys{
+					Keys: &pb.KeyList{
+						Keys: val,
+					},
+				},
+			}
+		case int64:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_BSIValue{
+					BSIValue: val,
+				},
+			}
+		case uint64:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_MutexID{
+					MutexID: val,
+				},
+			}
+		case string:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_MutexKey{
+					MutexKey: val,
+				},
+			}
+		case bool:
+			kv.Value = &pb.ExtractedTableValue{
+				Value: &pb.ExtractedTableValue_Bool{
+					Bool: val,
+				},
+			}
+		}
+		result[i] = kv
+
+	}
+	return result
+}
+
+func (s Serializer) decodeExtractedIDMatrixSorted(m *pb.ExtractedIDMatrixSorted) pilosa.ExtractedIDMatrixSorted {
+	mat := s.decodeExtractedIDMatrix(m.ExtractedIDMatrix)
+	kvs := s.decodeRowKVs(m.RowKVs)
+	return pilosa.ExtractedIDMatrixSorted{
+		ExtractedIDMatrix: &mat,
+		RowKVs:            kvs,
+	}
+}
+
+func castExtractedTableValue(v *pb.ExtractedTableValue) interface{} {
+	switch val := v.Value.(type) {
+	case *pb.ExtractedTableValue_IDs:
+		return val.IDs.IDs
+	case *pb.ExtractedTableValue_Keys:
+		return val.Keys.Keys
+	case *pb.ExtractedTableValue_BSIValue:
+		return val.BSIValue
+	case *pb.ExtractedTableValue_MutexID:
+		return val.MutexID
+	case *pb.ExtractedTableValue_MutexKey:
+		return val.MutexKey
+	case *pb.ExtractedTableValue_Bool:
+		return val.Bool
+	}
+	// Shouldn't happen, but i don't think we should panic
+	return v
+}
+
+func (s Serializer) decodeRowKVs(m []*pb.RowKV) []pilosa.RowKV {
+	rows := make([]pilosa.RowKV, len(m))
+	for i, v := range m {
+		kv := pilosa.RowKV{}
+		kv.RowID = v.RowID
+		kv.Value = castExtractedTableValue(v.Value)
+		rows[i] = kv
+	}
+	return rows
+}

encoding/proto/proto_test.go

@@ -158,3 +158,225 @@ func CreateArrowTable(pool memory.Allocator) arrow.Table {
 	records := []arrow.Record{rec1, rec2, rec3}
 	return array.NewTableFromRecords(schema, records)
 }
+
+func areEqualKV(a, b []pilosa.RowKV) bool {
+	return reflect.DeepEqual(a, b)
+}
+
+func TestExtractedIDMatrixSorted(t *testing.T) {
+	areEqualEIM := func(a, b *pilosa.ExtractedIDMatrix) bool {
+		if !reflect.DeepEqual(a.Fields, b.Fields) {
+			return false
+		}
+		if len(a.Columns) != len(b.Columns) {
+			return false
+		}
+		for i := range a.Columns {
+			if a.Columns[i].ColumnID != b.Columns[i].ColumnID {
+				return false
+			}
+			for x := range a.Columns[i].Rows {
+				if len(a.Columns[i].Rows[x]) != len(b.Columns[i].Rows[x]) {
+					return false
+				}
+				for y := range a.Columns[i].Rows[x] {
+					if a.Columns[i].Rows[x][y] != b.Columns[i].Rows[x][y] {
+						return false
+					}
+				}
+			}
+		}
+		return true
+	}
+	compare := func(a, b pilosa.ExtractedIDMatrixSorted) bool {
+		return areEqualKV(a.RowKVs, b.RowKVs) && areEqualEIM(a.ExtractedIDMatrix, b.ExtractedIDMatrix)
+	}
+	t.Run("ExtractedIDMatrixSorted", func(t *testing.T) {
+		// not correct ExtractedIDMatrix contents, but will test the typing appropriately
+		em := &pilosa.ExtractedIDMatrix{
+			Fields: []string{"A", "B", "C"},
+			Columns: []pilosa.ExtractedIDColumn{
+				{ColumnID: 1, Rows: [][]uint64{{1}}},
+				{ColumnID: 2, Rows: [][]uint64{{2}}},
+				{ColumnID: 3, Rows: [][]uint64{{3}}},
+			},
+		}
+		rows := []pilosa.RowKV{
+			{RowID: 1, Value: int64(10)},
+			{RowID: 2, Value: int64(20)},
+			{RowID: 3, Value: int64(30)},
+		}
+		table := pilosa.ExtractedIDMatrixSorted{
+			ExtractedIDMatrix: em,
+			RowKVs:            rows,
+		}
+		s := Serializer{}
+		before := s.endcodeExtractedIDMatrixSorted(table)
+		q := &pb.QueryResult{Type: queryResultTypeExtractedIDMatrixSorted, ExtractedIDMatrixSorted: before}
+		decoded := s.decodeQueryResult(q).(pilosa.ExtractedIDMatrixSorted)
+
+		if !compare(table, decoded) {
+			t.Errorf("failed to decode ExtractedIDMatrixSorted. expected\n %#v\n got\n %#v", table, decoded)
+		}
+	})
+}
+
+func TestSerializer_RowKVs(t *testing.T) {
+	tests := []struct {
+		name string
+		args []pilosa.RowKV
+	}{
+		{
+			name: "IDs",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: []uint64{10}},
+			},
+		},
+		{
+			name: "Keys",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: []string{"aaaa"}},
+			},
+		},
+		{
+			name: "BSIValue",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: int64(52)},
+			},
+		},
+		{
+			name: "MutexID",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: uint64(42)},
+			},
+		},
+		{
+			name: "MutexKey",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: "aaaa"},
+			},
+		},
+		{
+			name: "Bool",
+			args: []pilosa.RowKV{
+				{RowID: 1, Value: true},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := Serializer{}
+			encoded := s.encodeRowKVs(tt.args)
+			after := s.decodeRowKVs(encoded)
+			if !areEqualKV(tt.args, after) {
+				t.Fatalf("expected: %v got: %v", tt.args, after)
+			}
+		})
+	}
+}
+
+func TestSerializer_ExtractedTable(t *testing.T) {
+	compare := func(a, b pilosa.ExtractedTable) bool {
+		return reflect.DeepEqual(a, b)
+		// return true
+	}
+	tests := []struct {
+		name string
+		args pilosa.ExtractedTable
+	}{
+		{
+			name: "uint64",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "uint64"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{uint64(1)},
+					},
+				},
+			},
+		},
+		{
+			name: "int64",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "int64"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{int64(1)},
+					},
+				},
+			},
+		},
+		{
+			name: "string",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "string"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{"a"},
+					},
+				},
+			},
+		},
+		{
+			name: "[]string",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "[]string"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{[]string{"a"}},
+					},
+				},
+			},
+		},
+		{
+			name: "bool",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "bool"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{true},
+					},
+				},
+			},
+		},
+		{
+			name: "[]uint64",
+			args: pilosa.ExtractedTable{
+				Fields: []pilosa.ExtractedTableField{
+					{Name: "a", Type: "[]uint64"},
+				},
+				Columns: []pilosa.ExtractedTableColumn{
+					{
+						Column: pilosa.KeyOrID{ID: 0},
+						Rows:   []interface{}{[]uint64{2}},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := Serializer{}
+			encoded := s.encodeExtractedTable(tt.args)
+			after := s.decodeExtractedTable(encoded)
+			if !compare(tt.args, after) {
+				t.Fatalf("expected: %v got: %v", tt.args, after)
+			}
+		})
+	}
+}

executor.go

@@ -344,6 +344,8 @@ func safeCopy(resp QueryResponse) (out QueryResponse) {
 		case *basicTable:
 			// dumpTable(x)
 			out.Results = append(out.Results, x)
+		case ExtractedIDMatrixSorted:
+			out.Results = append(out.Results, x)
 		default:
 			panic(fmt.Sprintf("handle %T here", v))
 		}
@@ -4362,27 +4364,12 @@ type TimeArgs struct {
 	To   time.Time
 }
 
-func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (ExtractedIDMatrix, error) {
-	// Extract the column filter call.
-	if len(c.Children) < 1 {
-		return ExtractedIDMatrix{}, errors.New("missing column filter in Extract")
-	}
-	filter := c.Children[0]
-	var sort_desc bool
-	if filter.Name == "Sort" {
-		sd, _, err := filter.BoolArg("sort-desc")
-		if err != nil {
-			return ExtractedIDMatrix{}, errors.Wrap(err, "sort field error")
-		}
-		sort_desc = sd
-	}
-
-	// Extract fields from rows calls.
+func extractFieldsFromRowsCalls(c *pql.Call) ([]string, []TimeArgs, error) {
 	fields := make([]string, len(c.Children)-1)
 	timeArgs := make([]TimeArgs, len(c.Children)-1)
 	for i, rows := range c.Children[1:] {
 		if rows.Name != "Rows" {
-			return ExtractedIDMatrix{}, errors.Errorf("child call of Extract is %q but expected Rows", rows.Name)
+			return fields, timeArgs, errors.Errorf("child call of Extract is %q but expected Rows", rows.Name)
 		}
 		var fieldName string
 		var ok bool
@@ -4395,33 +4382,30 @@ func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c
 			case "from":
 				fromTime, err := parseTime(v)
 				if err != nil {
-					return ExtractedIDMatrix{}, errors.Wrap(err, "parsing from time")
+					return fields, timeArgs, errors.Wrap(err, "parsing from time")
 				}
 				timeArg.From = fromTime
 			case "to":
 				toTime, err := parseTime(v)
 				if err != nil {
-					return ExtractedIDMatrix{}, errors.Wrap(err, "parsing from time")
+					return fields, timeArgs, errors.Wrap(err, "parsing from time")
 				}
 				timeArg.To = toTime
 			default:
-				return ExtractedIDMatrix{}, errors.Errorf("unsupported Rows argument for Extract: %q", k)
+				return fields, timeArgs, errors.Errorf("unsupported Rows argument for Extract: %q", k)
 			}
 		}
 		if !ok {
-			return ExtractedIDMatrix{}, errors.New("missing field specification in Rows")
+			return fields, timeArgs, errors.New("missing field specification in Rows")
 		}
 		fields[i] = fieldName
 		timeArgs[i] = timeArg
 	}
+	return fields, timeArgs, nil
+}
 
-	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
-		return e.executeExtractShard(ctx, qcx, index, fields, filter, shard, mopt, timeArgs)
-	}
-
-	// Merge returned results at coordinating node.
-	reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
+func makeReduceFunc(sort_desc bool) func(ctx context.Context, prev, v interface{}) interface{} {
+	return func(ctx context.Context, prev, v interface{}) interface{} {
 		if err := ctx.Err(); err != nil {
 			return err
 		}
@@ -4452,13 +4436,9 @@ func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c
 			return ExtractedIDMatrix{}
 		}
 	}
+}
 
-	// Get full result set.
-	other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
-	if err != nil {
-		return ExtractedIDMatrix{}, err
-	}
-
+func handleExtractResults(other interface{}, filter *pql.Call, opt *ExecOptions) (interface{}, error) {
 	switch results := other.(type) {
 	case ExtractedIDMatrix:
 		sort.Slice(results.Columns, func(i, j int) bool {
@@ -4480,12 +4460,52 @@ func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c
 		if hasLimit && limit < uint64(len(results.RowKVs)) {
 			results.ExtractedIDMatrix.Columns = results.ExtractedIDMatrix.Columns[:limit]
 		}
+		// need to reture sorted to originating node in order to be able to sort properly
+		if opt.Remote {
+			return results, nil
+		}
 		return *results.ExtractedIDMatrix, nil
 	default:
 		return ExtractedIDMatrix{}, errors.New("Extract, unexpected result type found")
 	}
 }
 
+func (e *executor) executeExtract(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (interface{}, error) {
+	// Extract the column filter call.
+	if len(c.Children) < 1 {
+		return ExtractedIDMatrix{}, errors.New("missing column filter in Extract")
+	}
+	filter := c.Children[0]
+	var sort_desc bool
+	if filter.Name == "Sort" {
+		sd, _, err := filter.BoolArg("sort-desc")
+		if err != nil {
+			return ExtractedIDMatrix{}, errors.Wrap(err, "sort field error")
+		}
+		sort_desc = sd
+	}
+
+	// Extract fields from rows calls.
+	fields, timeArgs, err := extractFieldsFromRowsCalls(c)
+	if err != nil {
+		return ExtractedIDMatrix{}, errors.Wrap(err, "sort field error")
+	}
+
+	// Execute calls in bulk on each remote node and merge.
+	mapFn := func(ctx context.Context, shard uint64, mopt *mapOptions) (_ interface{}, err error) {
+		return e.executeExtractShard(ctx, qcx, index, fields, filter, shard, mopt, timeArgs)
+	}
+
+	// Merge returned results at coordinating node.
+	reduceFn := makeReduceFunc(sort_desc)
+	// Get full result set.
+	other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
+	if err != nil {
+		return ExtractedIDMatrix{}, err
+	}
+	return handleExtractResults(other, filter, opt)
+}
+
 func mergeBits(bits *Row, mask uint64, out map[uint64]uint64) {
 	for _, v := range bits.Columns() {
 		out[v] |= mask

pb/private.pb.go

@@ -6792,10 +6792,7 @@ func (m *IndexMeta) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -7264,10 +7261,7 @@ func (m *FieldOptions) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -7350,10 +7344,7 @@ func (m *ImportResponse) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -7538,10 +7529,7 @@ func (m *BlockDataRequest) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -7744,10 +7732,7 @@ func (m *BlockDataResponse) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -7874,10 +7859,7 @@ func (m *Cache) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8024,7 +8006,7 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error {
 					if err != nil {
 						return err
 					}
-					if skippy < 0 {
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
 						return ErrInvalidLengthPrivate
 					}
 					if (iNdEx + skippy) > postIndex {
@@ -8041,10 +8023,7 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8178,10 +8157,7 @@ func (m *CreateShardMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8264,10 +8240,7 @@ func (m *DeleteIndexMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8437,10 +8410,7 @@ func (m *CreateIndexMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8642,10 +8612,7 @@ func (m *CreateFieldMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8768,10 +8735,7 @@ func (m *UpdateFieldMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -8886,10 +8850,7 @@ func (m *FieldUpdate) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9004,10 +8965,7 @@ func (m *DeleteFieldMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9141,10 +9099,7 @@ func (m *DeleteAvailableShardMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9314,10 +9269,7 @@ func (m *Field) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9402,10 +9354,7 @@ func (m *Schema) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9577,10 +9526,7 @@ func (m *Index) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9714,10 +9660,7 @@ func (m *URI) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -9924,10 +9867,7 @@ func (m *Node) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10042,10 +9982,7 @@ func (m *NodeStateMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10151,10 +10088,7 @@ func (m *NodeEventMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10311,10 +10245,7 @@ func (m *NodeStatus) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10450,10 +10381,7 @@ func (m *IndexStatus) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10631,10 +10559,7 @@ func (m *FieldStatus) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10819,10 +10744,7 @@ func (m *ClusterStatus) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -10975,10 +10897,7 @@ func (m *BSIGroup) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -11125,10 +11044,7 @@ func (m *CreateViewMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -11275,10 +11191,7 @@ func (m *DeleteViewMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -11560,10 +11473,7 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -11765,10 +11675,7 @@ func (m *ResizeSource) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -11906,10 +11813,7 @@ func (m *TranslationResizeSource) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12047,10 +11951,7 @@ func (m *ResizeInstructionComplete) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12165,10 +12066,7 @@ func (m *Topology) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12219,10 +12117,7 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12273,10 +12168,7 @@ func (m *LoadSchemaMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12395,10 +12287,7 @@ func (m *TransactionMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12595,10 +12484,7 @@ func (m *Transaction) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12649,10 +12535,7 @@ func (m *TransactionStats) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12703,10 +12586,7 @@ func (m *ResizeAbortMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -12821,10 +12701,7 @@ func (m *ResizeNodeMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -13103,10 +12980,7 @@ func (m *FieldOperation) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -13409,7 +13283,7 @@ func (m *ShardIngestOperation) Unmarshal(dAtA []byte) error {
 					if err != nil {
 						return err
 					}
-					if skippy < 0 {
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
 						return ErrInvalidLengthPrivate
 					}
 					if (iNdEx + skippy) > postIndex {
@@ -13426,10 +13300,7 @@ func (m *ShardIngestOperation) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -13514,10 +13385,7 @@ func (m *ShardIngestOperations) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -13666,7 +13534,7 @@ func (m *ShardedIngestRequest) Unmarshal(dAtA []byte) error {
 					if err != nil {
 						return err
 					}
-					if skippy < 0 {
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
 						return ErrInvalidLengthPrivate
 					}
 					if (iNdEx + skippy) > postIndex {
@@ -13683,10 +13551,7 @@ func (m *ShardedIngestRequest) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {
@@ -13769,10 +13634,7 @@ func (m *DeleteDataframeMessage) Unmarshal(dAtA []byte) error {
 			if err != nil {
 				return err
 			}
-			if skippy < 0 {
-				return ErrInvalidLengthPrivate
-			}
-			if (iNdEx + skippy) < 0 {
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
 				return ErrInvalidLengthPrivate
 			}
 			if (iNdEx + skippy) > l {

pb/public.proto

@@ -32,7 +32,14 @@ message ExtractedIDColumn {
 	uint64 ID = 1;
 	repeated IDList Vals = 2;
 }
-
+message ExtractedIDMatrixSorted{
+	ExtractedIDMatrix ExtractedIDMatrix =1;
+	repeated RowKV RowKVs=2;
+}
+message RowKV{
+	uint64 RowID =1;
+	ExtractedTableValue Value =2;
+}
 message ExtractedIDMatrix {
 	repeated string Fields = 1;
 	repeated ExtractedIDColumn Columns = 2;
@@ -167,6 +174,7 @@ message QueryResult {
     DistinctTimestamp DistinctTimestamp = 17;
 	DataFrame DataFrame = 18;
 	ArrowTable ArrowTable = 19;
+	ExtractedIDMatrixSorted ExtractedIDMatrixSorted = 20;
 }
 
 message ImportRequest {