| 54 | } |
| 55 | |
| 56 | func mergeStreams(left, right batchStream, result batchStream, size int) batchStream { |
| 57 | // Reset the Index and Length of existing batches. |
| 58 | for i := range result { |
| 59 | result[i].Index = 0 |
| 60 | result[i].Length = 0 |
| 61 | } |
| 62 | resultLen := 1 // Number of batches in the final result. |
| 63 | b := &result[0] |
| 64 | |
| 65 | // This function adds a new batch to the result. |
| 66 | nextBatch := func(valueType chunkenc.ValueType) { |
| 67 | // The Index is the place at which new sample |
| 68 | // has to be appended, hence it tells the length. |
| 69 | b.Length = b.Index |
| 70 | resultLen++ |
| 71 | if resultLen > len(result) { |
| 72 | // It is possible that result can grow longer |
| 73 | // then the one provided. |
| 74 | result = append(result, promchunk.Batch{}) |
| 75 | } |
| 76 | b = &result[resultLen-1] |
| 77 | b.ValType = valueType |
| 78 | } |
| 79 | |
| 80 | populateVal := func(bs batchStream, valueType chunkenc.ValueType) { |
| 81 | if b.Index == 0 { |
| 82 | b.ValType = valueType |
| 83 | } else if b.Index == size || b.ValType != valueType { |
| 84 | // The batch reached it intended size or a new value type is used. |
| 85 | // Add another batch to the result and use it for further appending. |
| 86 | nextBatch(valueType) |
| 87 | } |
| 88 | switch valueType { |
| 89 | case chunkenc.ValFloat: |
| 90 | b.Timestamps[b.Index], b.Values[b.Index] = bs.at() |
| 91 | case chunkenc.ValHistogram: |
| 92 | b.Timestamps[b.Index], b.Histograms[b.Index] = bs.atHistogram() |
| 93 | case chunkenc.ValFloatHistogram: |
| 94 | b.Timestamps[b.Index], b.FloatHistograms[b.Index] = bs.atFloatHistogram() |
| 95 | default: |
| 96 | panic("unsupported value type") |
| 97 | } |
| 98 | b.Index++ |
| 99 | } |
| 100 | |
| 101 | for leftValueType, rightValueType := left.hasNext(), right.hasNext(); leftValueType != chunkenc.ValNone && rightValueType != chunkenc.ValNone; { |
| 102 | t1, t2 := left.atTime(), right.atTime() |
| 103 | if t1 < t2 { |
| 104 | populateVal(left, leftValueType) |
| 105 | left.next() |
| 106 | leftValueType = left.hasNext() |
| 107 | } else if t1 > t2 { |
| 108 | populateVal(right, rightValueType) |
| 109 | right.next() |
| 110 | rightValueType = right.hasNext() |
| 111 | } else { |
| 112 | populateVal(left, leftValueType) |
| 113 | left.next() |