mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
Zach LeslieandGitHub
a9f63d073b
* Update proto for v1 backend * Functioning blocklist with new package * Functioning blocklist with existing package * Generate proto to existing package * Add wrapper around google UUID for proto marshaling * Update blocklist for wrapper type * Test WriteTenantIndex * Implement json for uuid * Almost functional backend tests * Passing backend tests * Improve UUID coverage and utility * TempoDB updates for UUID and struct type changes * Update method style for uuid.New() * Update method style for uuid.MustParse() * Add uuid.From helper * Update modules for UUID and proto struct types * Update poller integration test for UUID * Update tempo-cli for UUID and proto struct type changes * Lint and clean up * Fix serverless * Fix cli * Fix cli * Add fixture tests for roundtrip assurance between versions * Clean up notes from proto definition * Update backend/test * Be more specific about unmarshal UUID * Include event for compacted block * Use better json tag for compacted meta * Implement UnmarshalJson on CompcatedBlockMeta for embedded struct * Test for CompactedMeta marshal/unmarshal * Drop benchmark index * Clean commented code * Extend mocks to capture multiple writes * Update backend/test fixtures * Write both proto and json to the backend for all three files in all backends * Fix lint * Back UUID with google.UUID to avoid struct * Relocate pkg/uuid to tempodb/backend * Re-plumb UUID through tempodb * Re-plumb UUID through cmd * Re-plumb UUID through modules * Re-plumb UUID through integration/ * Drop package spec from tempo.proto * Improve encoding marshal performance * Drop block proto and tenantindex json * Drop additional proto include * Implement zstd tenant index compression * Add note about use of json in the dedicatedcolumns * Fix ingester local block * Spell * Revert mock update * Drop json fallback from tenantindex read * Replace single-tenant test fixtures * Update fixture replacement test * Improve error handling and fix test * Extend test for tenantindex round-trip * Small fixes * Revert "Drop json fallback from tenantindex read" This reverts commit 834ff5b85d8a8a4d7eb560fbb67ba8f3a5a3959a. * Add back mocks to capture multiple backend writes * Add back proto & json writitng/reading for tenantindex * Change TotalObjects type from int32 -> int64 The original change was int -> int32, which on 64bit machines is a downgrade that I overlooked. Raising here to int64 to maintain the size. * Store write buffer at object path in mocks to improve readability in tests * Avoid allocation in interface validation * Update note to include intern * Pluralize metas and compacted_metas fields * Fallback to json only on ErrDoesNotExist * Lint * Drop debug * Revert "Pluralize metas and compacted_metas fields" This reverts commit 46a8c36e37ac33e1ffc104287d898930e02cdd97. * Import original tenantindex and test contents for backwards compatibility * Update changelog
150 lines
4.3 KiB
Go
150 lines
4.3 KiB
Go
package vparquet4
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"testing"
|
|
|
|
"github.com/parquet-go/parquet-go"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/grafana/tempo/pkg/cache"
|
|
"github.com/grafana/tempo/tempodb/backend"
|
|
"github.com/grafana/tempo/tempodb/backend/local"
|
|
)
|
|
|
|
var tenantID = "single-tenant"
|
|
|
|
type dummyReader struct {
|
|
r io.ReaderAt
|
|
footer bool
|
|
columnIndex bool
|
|
offsetIndex bool
|
|
}
|
|
|
|
func (d *dummyReader) ReadAt(p []byte, off int64) (int, error) { return d.r.ReadAt(p, off) }
|
|
|
|
func (d *dummyReader) SetFooterSection(_, _ int64) { d.footer = true }
|
|
func (d *dummyReader) SetColumnIndexSection(_, _ int64) { d.columnIndex = true }
|
|
func (d *dummyReader) SetOffsetIndexSection(_, _ int64) { d.offsetIndex = true }
|
|
|
|
// TestParquetGoSetsMetadataSections tests if the special metadata sections are set correctly for caching.
|
|
// It is the best way right now to ensure that the interface used by the underlying parquet-go library does not drift.
|
|
// If this test starts failing at some point, we should update the interface used by `parquetOptimizedReaderAt` to match
|
|
// the specification in parquet-go
|
|
func TestParquetGoSetsMetadataSections(t *testing.T) {
|
|
rawR, _, _, err := local.New(&local.Config{
|
|
Path: "./test-data",
|
|
})
|
|
require.NoError(t, err)
|
|
|
|
r := backend.NewReader(rawR)
|
|
ctx := context.Background()
|
|
|
|
blocks, _, err := r.Blocks(ctx, tenantID)
|
|
require.NoError(t, err)
|
|
require.Len(t, blocks, 1)
|
|
|
|
meta, err := r.BlockMeta(ctx, blocks[0], tenantID)
|
|
require.NoError(t, err)
|
|
|
|
br := NewBackendReaderAt(ctx, r, DataFileName, meta)
|
|
dr := &dummyReader{r: br}
|
|
_, err = parquet.OpenFile(dr, int64(meta.Size_))
|
|
require.NoError(t, err)
|
|
|
|
require.True(t, dr.footer)
|
|
require.True(t, dr.columnIndex)
|
|
require.True(t, dr.offsetIndex)
|
|
}
|
|
|
|
func TestCachingReaderShortcircuitsFooterHeader(t *testing.T) {
|
|
rr := &recordingReaderAt{}
|
|
pr := newCachedReaderAt(rr, 1000, 1000, 100)
|
|
|
|
expectedReads := []read{}
|
|
|
|
// magic number doesn't pass through
|
|
_, err := pr.ReadAt(make([]byte, 4), 0)
|
|
require.NoError(t, err)
|
|
|
|
// footer size doesn't pass through
|
|
_, err = pr.ReadAt(make([]byte, 8), 992)
|
|
require.NoError(t, err)
|
|
|
|
// other calls pass through
|
|
_, err = pr.ReadAt(make([]byte, 13), 25)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{13, 25, cache.RoleParquetPage})
|
|
|
|
_, err = pr.ReadAt(make([]byte, 97), 118)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{97, 118, cache.RoleParquetPage})
|
|
|
|
_, err = pr.ReadAt(make([]byte, 59), 421)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{59, 421, cache.RoleParquetPage})
|
|
|
|
require.Equal(t, expectedReads, rr.reads)
|
|
}
|
|
|
|
func TestCachingReaderAt(t *testing.T) {
|
|
rr := &recordingReaderAt{}
|
|
cr := newCachedReaderAt(rr, 1000, 100000, 10)
|
|
|
|
expectedReads := []read{}
|
|
|
|
// specially cached sections
|
|
cr.SetColumnIndexSection(1, 34)
|
|
_, err := cr.ReadAt(make([]byte, 34), 1)
|
|
expectedReads = append(expectedReads, read{34, 1, cache.RoleParquetColumnIdx})
|
|
require.NoError(t, err)
|
|
|
|
cr.SetFooterSection(14, 20)
|
|
_, err = cr.ReadAt(make([]byte, 20), 14)
|
|
expectedReads = append(expectedReads, read{20, 14, cache.RoleParquetFooter})
|
|
require.NoError(t, err)
|
|
|
|
cr.SetOffsetIndexSection(13, 12)
|
|
_, err = cr.ReadAt(make([]byte, 12), 13)
|
|
expectedReads = append(expectedReads, read{12, 13, cache.RoleParquetOffsetIdx})
|
|
require.NoError(t, err)
|
|
|
|
// everything else is a parquet page
|
|
_, err = cr.ReadAt(make([]byte, 13), 25)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{13, 25, cache.RoleParquetPage})
|
|
|
|
_, err = cr.ReadAt(make([]byte, 97), 118)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{97, 118, cache.RoleParquetPage})
|
|
|
|
// unless it's larger than the page size
|
|
_, err = cr.ReadAt(make([]byte, 1001), 421)
|
|
require.NoError(t, err)
|
|
expectedReads = append(expectedReads, read{1001, 421, cache.RoleNone})
|
|
|
|
require.Equal(t, expectedReads, rr.reads)
|
|
}
|
|
|
|
type read struct {
|
|
len int
|
|
off int64
|
|
role cache.Role
|
|
}
|
|
type recordingReaderAt struct {
|
|
reads []read
|
|
}
|
|
|
|
func (r *recordingReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
|
|
r.reads = append(r.reads, read{len(p), off, ""})
|
|
|
|
return len(p), nil
|
|
}
|
|
|
|
func (r *recordingReaderAt) ReadAtWithCache(p []byte, off int64, role cache.Role) (n int, err error) {
|
|
r.reads = append(r.reads, read{len(p), off, role})
|
|
|
|
return len(p), nil
|
|
}
|