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

169 lines
4.8 KiB
Go

package brotli
/* Copyright 2018 Google Inc. All Rights Reserved.
Distributed under MIT license.
See file LICENSE for detail or copy at https://opensource.org/licenses/MIT
*/
/* NOTE: this hasher does not search in the dictionary. It is used as
backup-hasher, the main hasher already searches in it. */
const kRollingHashMul32 uint32 = 69069
const kInvalidPosHashRolling uint32 = 0xffffffff
/* This hasher uses a longer forward length, but returning a higher value here
will hurt compression by the main hasher when combined with a composite
hasher. The hasher tests for forward itself instead. */
func (*hashRolling) HashTypeLength() uint {
return 4
}
func (*hashRolling) StoreLookahead() uint {
return 4
}
/* Computes a code from a single byte. A lookup table of 256 values could be
used, but simply adding 1 works about as good. */
func (*hashRolling) HashByte(b byte) uint32 {
return uint32(b) + 1
}
func (h *hashRolling) HashRollingFunctionInitial(state uint32, add byte, factor uint32) uint32 {
return uint32(factor*state + h.HashByte(add))
}
func (h *hashRolling) HashRollingFunction(state uint32, add byte, rem byte, factor uint32, factor_remove uint32) uint32 {
return uint32(factor*state + h.HashByte(add) - factor_remove*h.HashByte(rem))
}
/* Rolling hash for long distance long string matches. Stores one position
per bucket, bucket key is computed over a long region. */
type hashRolling struct {
hasherCommon
jump int
state uint32
table []uint32
next_ix uint
factor uint32
factor_remove uint32
}
func (h *hashRolling) Initialize(params *encoderParams) {
h.state = 0
h.next_ix = 0
h.factor = kRollingHashMul32
/* Compute the factor of the oldest byte to remove: factor**steps modulo
0xffffffff (the multiplications rely on 32-bit overflow) */
h.factor_remove = 1
for i := 0; i < 32; i += h.jump {
h.factor_remove *= h.factor
}
h.table = make([]uint32, 16777216)
for i := 0; i < 16777216; i++ {
h.table[i] = kInvalidPosHashRolling
}
}
func (h *hashRolling) Prepare(one_shot bool, input_size uint, data []byte) {
/* Too small size, cannot use this hasher. */
if input_size < 32 {
return
}
h.state = 0
for i := 0; i < 32; i += h.jump {
h.state = h.HashRollingFunctionInitial(h.state, data[i], h.factor)
}
}
func (*hashRolling) Store(data []byte, mask uint, ix uint) {
}
func (*hashRolling) StoreRange(data []byte, mask uint, ix_start uint, ix_end uint) {
}
func (h *hashRolling) StitchToPreviousBlock(num_bytes uint, position uint, ringbuffer []byte, ring_buffer_mask uint) {
var position_masked uint
/* In this case we must re-initialize the hasher from scratch from the
current position. */
var available uint = num_bytes
if position&uint(h.jump-1) != 0 {
var diff uint = uint(h.jump) - (position & uint(h.jump-1))
if diff > available {
available = 0
} else {
available = available - diff
}
position += diff
}
position_masked = position & ring_buffer_mask
/* wrapping around ringbuffer not handled. */
if available > ring_buffer_mask-position_masked {
available = ring_buffer_mask - position_masked
}
h.Prepare(false, available, ringbuffer[position&ring_buffer_mask:])
h.next_ix = position
}
func (*hashRolling) PrepareDistanceCache(distance_cache []int) {
}
func (h *hashRolling) FindLongestMatch(dictionary *encoderDictionary, data []byte, ring_buffer_mask uint, distance_cache []int, cur_ix uint, max_length uint, max_backward uint, gap uint, max_distance uint, out *hasherSearchResult) {
var cur_ix_masked uint = cur_ix & ring_buffer_mask
var pos uint = h.next_ix
if cur_ix&uint(h.jump-1) != 0 {
return
}
/* Not enough lookahead */
if max_length < 32 {
return
}
for pos = h.next_ix; pos <= cur_ix; pos += uint(h.jump) {
var code uint32 = h.state & ((16777216 * 64) - 1)
var rem byte = data[pos&ring_buffer_mask]
var add byte = data[(pos+32)&ring_buffer_mask]
var found_ix uint = uint(kInvalidPosHashRolling)
h.state = h.HashRollingFunction(h.state, add, rem, h.factor, h.factor_remove)
if code < 16777216 {
found_ix = uint(h.table[code])
h.table[code] = uint32(pos)
if pos == cur_ix && uint32(found_ix) != kInvalidPosHashRolling {
/* The cast to 32-bit makes backward distances up to 4GB work even
if cur_ix is above 4GB, despite using 32-bit values in the table. */
var backward uint = uint(uint32(cur_ix - found_ix))
if backward <= max_backward {
var found_ix_masked uint = found_ix & ring_buffer_mask
var len uint = findMatchLengthWithLimit(data[found_ix_masked:], data[cur_ix_masked:], max_length)
if len >= 4 && len > out.len {
var score uint = backwardReferenceScore(uint(len), backward)
if score > out.score {
out.len = uint(len)
out.distance = backward
out.score = score
out.len_code_delta = 0
}
}
}
}
}
}
h.next_ix = cur_ix + uint(h.jump)
}