Files
049323c40f Add Parquet block format (#1479)
* starting parquet support

starting parquet support. some things are still todo:  compaction, find trace by ID

Co-Authored-By: Annanay Agarwal <[email protected]>
Co-Authored-By: Martin Disibio <[email protected]>

* add parquet reusable streaming block

* Add parquet compaction

* Add findTraceByID logic and parquet -> proto conversion

Signed-off-by: Annanay <[email protected]>

* Switch findtracebyid to readrow until we get SeekToRow working

* Fix error handling in bufferedReaderAt. Add various instrumentation and log messages

* Use jsonpb to roundtrip event attributes

* Improve buffered reader when reading near end of file. More instrumentation and tidiness

* Add totalBlockBytes instrumentation. Fix inspectedBytes instrumentation

* parquet.FindTraceByID check bloom

* Make compator flush on trace count until we get the row group size target working

* Support maxDuration in parquet search

* Correct attribute translation for array/kvlist

Signed-off-by: Annanay <[email protected]>

* extend roundtrip test

Signed-off-by: Annanay <[email protected]>

* Use default page buffer size

* Fix size url param to be uint64

* Increase parquet searches to 1MB stripes

* Add test for findtracebyid

* make row group size configurable

Signed-off-by: Annanay <[email protected]>

* Record parquet footer size in block meta and short-circuit header/footer reads

* Add tests for SubstringPredicate

* Switch compaction to compact blocks of same version, use re-usable buffer in findtracebyid

Signed-off-by: Annanay <[email protected]>

* Add search support for remaining well-known attributes and span status, add search tests

* Benchmark substringPredicate

* Make block version configurable, default to v2

* memoize substringPredicate

* Update segmentio/parquet-go to catch SeekToRow fix, revert hack in FindTraceByID

Signed-off-by: Annanay <[email protected]>

* Add multi block prefetch iterator for use in compactor

Signed-off-by: Annanay <[email protected]>

* Add BufferedWriter with outgoing queue to flush traces async

Signed-off-by: Annanay <[email protected]>

Make compactor block complete async

Signed-off-by: Annanay <[email protected]>

* First pass CLI to dump schema and column sizes from block

Signed-off-by: Annanay <[email protected]>

* honor search start/end time range

* Improve error message, add blockID

Signed-off-by: Annanay <[email protected]>

* Fix race in InstrumentedPredicate with atomics

* Make bufferedReaderAt not single-threaded

* Debug SeekToRow panic, add test data from vulture tenant

Signed-off-by: Annanay <[email protected]>

* fix test

* repair tests

* Address println warnings

* Experiment with bookmark prefetch

Signed-off-by: Annanay <[email protected]>

Remove quitCh from bookmark

Signed-off-by: Annanay <[email protected]>

* fix prefetch on the bookmarks

Signed-off-by: Annanay <[email protected]>

* lint

* tracker -> RowNumber, lint

* delete unused predicates

* lint

* lint

* mod vendor

* remove pool from buffered writer queue

Signed-off-by: Annanay <[email protected]>

* post merge cleanup

Signed-off-by: Annanay <[email protected]>

* Close quitCh instead of attempting to signal to it

Signed-off-by: Annanay <[email protected]>

* split multiblock and prefetch iterators, refactor around interface

Signed-off-by: Annanay <[email protected]>

* Split search into 2 phases to avoid loading traceid/display columns unless there is a match

* Make columnIterator.SeekTo do something

* update to catch segmentio/parquet-go#220

Signed-off-by: Annanay <[email protected]>

* Fix column index to store entire 32-byte trace ID string. lint

* Make parquet honor the adjusted start/end times that come from the wal

* post merge cleanup: use compaction metric callbacks

Signed-off-by: Annanay <[email protected]>

* serverless sum/mods

Signed-off-by: Annanay <[email protected]>

* parquetquery predicates handle column index not present

* Make parquet compaction really honor input block start/end times, add universal test at tempodb-level

* parquet skip page indexes while searching to reduce latency on openfile

* More testing around iterators and combiner

Signed-off-by: Annanay <[email protected]>

* Update to latest parquet lib to catch rle bitpack fix

* Review feedback:  LatestEncoding -> DefaultEncoding

* Pass block footerSize through serverless

* mod vendor for serverless

* Fix tests

* add blockID to error message in block iterator

Signed-off-by: Annanay <[email protected]>

* increase readvalues buffer, add more span instrumentation

Signed-off-by: Annanay <[email protected]>

* changelog and docs

* Remove the use of async writer in vparquet compactor

Signed-off-by: Annanay <[email protected]>

Co-authored-by: Annanay Agarwal <[email protected]>
2022-06-23 16:00:18 +05:30

135 lines
4.4 KiB
Go

package brotli
/* Copyright 2013 Google Inc. All Rights Reserved.
Distributed under MIT license.
See file LICENSE for detail or copy at https://opensource.org/licenses/MIT
*/
/* A ringBuffer(window_bits, tail_bits) contains `1 << window_bits' bytes of
data in a circular manner: writing a byte writes it to:
`position() % (1 << window_bits)'.
For convenience, the ringBuffer array contains another copy of the
first `1 << tail_bits' bytes:
buffer_[i] == buffer_[i + (1 << window_bits)], if i < (1 << tail_bits),
and another copy of the last two bytes:
buffer_[-1] == buffer_[(1 << window_bits) - 1] and
buffer_[-2] == buffer_[(1 << window_bits) - 2]. */
type ringBuffer struct {
size_ uint32
mask_ uint32
tail_size_ uint32
total_size_ uint32
cur_size_ uint32
pos_ uint32
data_ []byte
buffer_ []byte
}
func ringBufferInit(rb *ringBuffer) {
rb.pos_ = 0
}
func ringBufferSetup(params *encoderParams, rb *ringBuffer) {
var window_bits int = computeRbBits(params)
var tail_bits int = params.lgblock
*(*uint32)(&rb.size_) = 1 << uint(window_bits)
*(*uint32)(&rb.mask_) = (1 << uint(window_bits)) - 1
*(*uint32)(&rb.tail_size_) = 1 << uint(tail_bits)
*(*uint32)(&rb.total_size_) = rb.size_ + rb.tail_size_
}
const kSlackForEightByteHashingEverywhere uint = 7
/* Allocates or re-allocates data_ to the given length + plus some slack
region before and after. Fills the slack regions with zeros. */
func ringBufferInitBuffer(buflen uint32, rb *ringBuffer) {
var new_data []byte
var i uint
size := 2 + int(buflen) + int(kSlackForEightByteHashingEverywhere)
if cap(rb.data_) < size {
new_data = make([]byte, size)
} else {
new_data = rb.data_[:size]
}
if rb.data_ != nil {
copy(new_data, rb.data_[:2+rb.cur_size_+uint32(kSlackForEightByteHashingEverywhere)])
}
rb.data_ = new_data
rb.cur_size_ = buflen
rb.buffer_ = rb.data_[2:]
rb.data_[1] = 0
rb.data_[0] = rb.data_[1]
for i = 0; i < kSlackForEightByteHashingEverywhere; i++ {
rb.buffer_[rb.cur_size_+uint32(i)] = 0
}
}
func ringBufferWriteTail(bytes []byte, n uint, rb *ringBuffer) {
var masked_pos uint = uint(rb.pos_ & rb.mask_)
if uint32(masked_pos) < rb.tail_size_ {
/* Just fill the tail buffer with the beginning data. */
var p uint = uint(rb.size_ + uint32(masked_pos))
copy(rb.buffer_[p:], bytes[:brotli_min_size_t(n, uint(rb.tail_size_-uint32(masked_pos)))])
}
}
/* Push bytes into the ring buffer. */
func ringBufferWrite(bytes []byte, n uint, rb *ringBuffer) {
if rb.pos_ == 0 && uint32(n) < rb.tail_size_ {
/* Special case for the first write: to process the first block, we don't
need to allocate the whole ring-buffer and we don't need the tail
either. However, we do this memory usage optimization only if the
first write is less than the tail size, which is also the input block
size, otherwise it is likely that other blocks will follow and we
will need to reallocate to the full size anyway. */
rb.pos_ = uint32(n)
ringBufferInitBuffer(rb.pos_, rb)
copy(rb.buffer_, bytes[:n])
return
}
if rb.cur_size_ < rb.total_size_ {
/* Lazily allocate the full buffer. */
ringBufferInitBuffer(rb.total_size_, rb)
/* Initialize the last two bytes to zero, so that we don't have to worry
later when we copy the last two bytes to the first two positions. */
rb.buffer_[rb.size_-2] = 0
rb.buffer_[rb.size_-1] = 0
}
{
var masked_pos uint = uint(rb.pos_ & rb.mask_)
/* The length of the writes is limited so that we do not need to worry
about a write */
ringBufferWriteTail(bytes, n, rb)
if uint32(masked_pos+n) <= rb.size_ {
/* A single write fits. */
copy(rb.buffer_[masked_pos:], bytes[:n])
} else {
/* Split into two writes.
Copy into the end of the buffer, including the tail buffer. */
copy(rb.buffer_[masked_pos:], bytes[:brotli_min_size_t(n, uint(rb.total_size_-uint32(masked_pos)))])
/* Copy into the beginning of the buffer */
copy(rb.buffer_, bytes[rb.size_-uint32(masked_pos):][:uint32(n)-(rb.size_-uint32(masked_pos))])
}
}
{
var not_first_lap bool = rb.pos_&(1<<31) != 0
var rb_pos_mask uint32 = (1 << 31) - 1
rb.data_[0] = rb.buffer_[rb.size_-2]
rb.data_[1] = rb.buffer_[rb.size_-1]
rb.pos_ = (rb.pos_ & rb_pos_mask) + uint32(uint32(n)&rb_pos_mask)
if not_first_lap {
/* Wrap, but preserve not-a-first-lap feature. */
rb.pos_ |= 1 << 31
}
}
}