mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Martin DisibioandGitHub
8e57f1972c
* Experiments with next parquet version. Added rounded start time column * Vparquet5 use lower precision timestamp column when possible based on step interval * microoptimization * leftjoiniterator use dynamic branch reordering, plus other performance improvements * Add columns with other granularities * Add new functionality: nocompactflg and short trace ID search * Fix handling of large step intervals and split between backend and generator requests * Repair tests and reduce divergence from main * Cleanup, separate benchmark env vars * Set version string to preview * rename/comments * Oops undo change to default encoding * Fix tests and lint * Bring in more updates from vparquet4 * Improve accuracy of estimate function, cleanup debug statements * Reduce diff from main * lint * Add test parquet file * Fix to maintain nano-second precision for blocks overlapping query range edges and recent/backend handoff where things don't align * Fix tests after merging main, one more fix to maintain nanosecond precision when needed * changelog * Cleanup/reduce diff from main * Review feedback
104 lines
2.6 KiB
Go
104 lines
2.6 KiB
Go
package vparquet5
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/grafana/tempo/pkg/cache"
|
|
"github.com/grafana/tempo/tempodb/backend"
|
|
"github.com/grafana/tempo/tempodb/encoding/common"
|
|
)
|
|
|
|
func CopyBlock(ctx context.Context, fromMeta, toMeta *backend.BlockMeta, from backend.Reader, to backend.Writer) error {
|
|
// Copy streams, efficient but can't cache.
|
|
copyStream := func(name string) error {
|
|
reader, size, err := from.StreamReader(ctx, name, (uuid.UUID)(fromMeta.BlockID), fromMeta.TenantID)
|
|
if err != nil {
|
|
return fmt.Errorf("error reading %s: %w", name, err)
|
|
}
|
|
defer reader.Close()
|
|
|
|
return to.StreamWriter(ctx, name, (uuid.UUID)(toMeta.BlockID), toMeta.TenantID, reader, size)
|
|
}
|
|
|
|
// Read entire object and attempt to cache
|
|
cpy := func(name string, cacheInfo *backend.CacheInfo) error {
|
|
cacheInfo.Meta = fromMeta
|
|
b, err := from.Read(ctx, name, (uuid.UUID)(fromMeta.BlockID), fromMeta.TenantID, cacheInfo)
|
|
if err != nil {
|
|
return fmt.Errorf("error reading %s: %w", name, err)
|
|
}
|
|
|
|
cacheInfo.Meta = toMeta
|
|
return to.Write(ctx, name, (uuid.UUID)(toMeta.BlockID), toMeta.TenantID, b, cacheInfo)
|
|
}
|
|
|
|
// Data
|
|
err := copyStream(DataFileName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Bloom
|
|
cacheInfo := &backend.CacheInfo{Role: cache.RoleBloom}
|
|
for i := 0; i < common.ValidateShardCount(int(fromMeta.BloomShardCount)); i++ {
|
|
err = cpy(common.BloomName(i), cacheInfo)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Index (may not exist)
|
|
err = cpy(common.NameIndex, &backend.CacheInfo{Role: cache.RoleTraceIDIdx})
|
|
if err != nil && !errors.Is(err, backend.ErrDoesNotExist) {
|
|
return err
|
|
}
|
|
|
|
// Meta
|
|
err = to.WriteBlockMeta(ctx, toMeta)
|
|
return err
|
|
}
|
|
|
|
func writeBlockMeta(ctx context.Context, w backend.Writer, meta *backend.BlockMeta, bloom *common.ShardedBloomFilter, index *index) error {
|
|
// bloom
|
|
blooms, err := bloom.Marshal()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
cacheInfo := &backend.CacheInfo{
|
|
Meta: meta,
|
|
Role: cache.RoleBloom,
|
|
}
|
|
for i, bloom := range blooms {
|
|
nameBloom := common.BloomName(i)
|
|
err := w.Write(ctx, nameBloom, (uuid.UUID)(meta.BlockID), meta.TenantID, bloom, cacheInfo)
|
|
if err != nil {
|
|
return fmt.Errorf("unexpected error writing bloom-%d: %w", i, err)
|
|
}
|
|
}
|
|
|
|
// Index
|
|
i, err := index.Marshal()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = w.Write(ctx, common.NameIndex, (uuid.UUID)(meta.BlockID), meta.TenantID, i, &backend.CacheInfo{
|
|
Meta: meta,
|
|
Role: cache.RoleTraceIDIdx,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// meta
|
|
err = w.WriteBlockMeta(ctx, meta)
|
|
if err != nil {
|
|
return fmt.Errorf("unexpected error writing meta: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|