Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
influxdb/tsdb/engine/tsm1/compact_test.go

4810 lines
114 KiB
Go
Raw Permalink Blame History

package tsm1_test
import (
"errors"
"fmt"
"io/fs"
"math"
"os"
"path/filepath"
"sort"
"sync/atomic"
"testing"
"time"
"github.com/influxdata/influxdb/tsdb"
"github.com/influxdata/influxdb/tsdb/engine/tsm1"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
)
// Tests compacting a Cache snapshot into a single TSM file
func TestCompactor_Snapshot(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(1, float64(1))
v2 := tsm1.NewValue(1, float64(1))
v3 := tsm1.NewValue(2, float64(2))
points1 := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v1},
"cpu,host=B#!~#value": {v2, v3},
}
c := tsm1.NewCache(0)
for k, v := range points1 {
if err := c.Write([]byte(k), v); err != nil {
t.Fatalf("failed to write key foo to cache: %s", err.Error())
}
}
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = &fakeFileStore{}
files, err := compactor.WriteSnapshot(c, zap.NewNop())
if err == nil {
t.Fatalf("expected error writing snapshot: %v", err)
}
if len(files) > 0 {
t.Fatalf("no files should be compacted: got %v", len(files))
}
compactor.Open()
files, err = compactor.WriteSnapshot(c, zap.NewNop())
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 2; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{v1}},
{"cpu,host=B#!~#value", []tsm1.Value{v2, v3}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
func TestCompactor_CompactFullLastTimestamp(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
var vals tsm1.Values
ts := int64(1e9)
for i := 0; i < 120; i++ {
vals = append(vals, tsm1.NewIntegerValue(ts, 1))
ts += 1e9
}
// 121st timestamp skips a second
ts += 1e9
vals = append(vals, tsm1.NewIntegerValue(ts, 1))
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": vals[:100],
}
f1 := MustWriteTSM(dir, 1, writes)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": vals[100:],
}
f2 := MustWriteTSM(dir, 2, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFull([]string{f1, f2}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %#v", err)
}
r := MustOpenTSMReader(files[0])
entries := r.Entries([]byte("cpu,host=A#!~#value"))
_, b, err := r.ReadBytes(&entries[0], nil)
if err != nil {
t.Fatalf("ReadBytes: unexpected error %v", err)
}
var a tsdb.IntegerArray
err = tsm1.DecodeIntegerArrayBlock(b, &a)
if err != nil {
t.Fatalf("DecodeIntegerArrayBlock: unexpected error %v", err)
}
if a.MaxTime() != entries[0].MaxTime {
t.Fatalf("expected MaxTime == a.MaxTime()")
}
}
// Ensures that a compaction will properly merge multiple TSM files
func TestCompactor_CompactFull(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1},
}
f1 := MustWriteTSM(dir, 1, writes)
a2 := tsm1.NewValue(2, 1.2)
b1 := tsm1.NewValue(1, 2.1)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a2},
"cpu,host=B#!~#value": {b1},
}
f2 := MustWriteTSM(dir, 2, writes)
a3 := tsm1.NewValue(1, 1.3)
c1 := tsm1.NewValue(1, 3.1)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a3},
"cpu,host=C#!~#value": {c1},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
if err == nil {
t.Fatalf("expected error writing snapshot: %v", err)
}
if len(files) > 0 {
t.Fatalf("no files should be compacted: got %v", len(files))
}
compactor.Open()
files, err = compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f3)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 3; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a3, a2}},
{"cpu,host=B#!~#value", []tsm1.Value{b1}},
{"cpu,host=C#!~#value", []tsm1.Value{c1}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
// Ensures that a compaction will properly merge multiple TSM files
func TestCompactor_DecodeError(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1},
}
f1 := MustWriteTSM(dir, 1, writes)
a2 := tsm1.NewValue(2, 1.2)
b1 := tsm1.NewValue(1, 2.1)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a2},
"cpu,host=B#!~#value": {b1},
}
f2 := MustWriteTSM(dir, 2, writes)
a3 := tsm1.NewValue(1, 1.3)
c1 := tsm1.NewValue(1, 3.1)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a3},
"cpu,host=C#!~#value": {c1},
}
f3 := MustWriteTSM(dir, 3, writes)
f, err := os.OpenFile(f3, os.O_RDWR, os.ModePerm)
if err != nil {
panic(err)
}
f.WriteAt([]byte("ffff"), 10) // skip over header
f.Close()
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
require.Error(t, err, "expected error writing snapshot")
require.Zero(t, len(files), "no files should be compacted")
compactor.Open()
_, err = compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
require.ErrorContains(t, err, "decode error: unable to decompress block type float for key 'cpu,host=A#!~#value': unpackBlock: not enough data for timestamp")
tsm1.MoveTsmOnReadErr(err, zap.NewNop(), func(strings []string, strings2 []string) error {
require.Equal(t, 1, len(strings))
require.Equal(t, strings[0], f3)
return nil
})
}
// Ensures that a compaction will properly merge multiple TSM files
func TestCompactor_Compact_OverlappingBlocks(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(4, 1.1)
a2 := tsm1.NewValue(5, 1.1)
a3 := tsm1.NewValue(7, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2, a3},
}
f1 := MustWriteTSM(dir, 1, writes)
c1 := tsm1.NewValue(3, 1.2)
c2 := tsm1.NewValue(8, 1.2)
c3 := tsm1.NewValue(9, 1.2)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {c1, c2, c3},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFast([]string{f1, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{c1, a1, a2, a3, c2, c3}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
// Ensures that a compaction will properly merge multiple TSM files
func TestCompactor_Compact_OverlappingBlocksMultiple(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(4, 1.1)
a2 := tsm1.NewValue(5, 1.1)
a3 := tsm1.NewValue(7, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2, a3},
}
f1 := MustWriteTSM(dir, 1, writes)
b1 := tsm1.NewValue(1, 1.2)
b2 := tsm1.NewValue(2, 1.2)
b3 := tsm1.NewValue(6, 1.2)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {b1, b2, b3},
}
f2 := MustWriteTSM(dir, 2, writes)
c1 := tsm1.NewValue(3, 1.2)
c2 := tsm1.NewValue(8, 1.2)
c3 := tsm1.NewValue(9, 1.2)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {c1, c2, c3},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFast([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{b1, b2, c1, a1, a2, b3, a3, c2, c3}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
func TestCompactor_Compact_UnsortedBlocks(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 2 TSM files with different data and one new point
a1 := tsm1.NewValue(4, 1.1)
a2 := tsm1.NewValue(5, 1.1)
a3 := tsm1.NewValue(6, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2, a3},
}
f1 := MustWriteTSM(dir, 1, writes)
b1 := tsm1.NewValue(1, 1.2)
b2 := tsm1.NewValue(2, 1.2)
b3 := tsm1.NewValue(3, 1.2)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {b1, b2, b3},
}
f2 := MustWriteTSM(dir, 2, writes)
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = &fakeFileStore{}
compactor.Open()
files, err := compactor.CompactFast([]string{f1, f2}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{b1, b2, b3, a1, a2, a3}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
func TestCompactor_Compact_UnsortedBlocksOverlapping(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files where two blocks are overlapping and with unsorted order
a1 := tsm1.NewValue(1, 1.1)
a2 := tsm1.NewValue(2, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2},
}
f1 := MustWriteTSM(dir, 1, writes)
b1 := tsm1.NewValue(3, 1.2)
b2 := tsm1.NewValue(4, 1.2)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {b1, b2},
}
f2 := MustWriteTSM(dir, 2, writes)
c1 := tsm1.NewValue(1, 1.1)
c2 := tsm1.NewValue(2, 1.1)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {c1, c2},
}
f3 := MustWriteTSM(dir, 3, writes)
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = &fakeFileStore{}
compactor.Open()
files, err := compactor.CompactFast([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a1, a2, b1, b2}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
}
// Ensures that a compaction will properly merge multiple TSM files
func TestCompactor_CompactFull_SkipFullBlocks(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
a2 := tsm1.NewValue(2, 1.2)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2},
}
f1 := MustWriteTSM(dir, 1, writes)
a3 := tsm1.NewValue(3, 1.3)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a3},
}
f2 := MustWriteTSM(dir, 2, writes)
a4 := tsm1.NewValue(4, 1.4)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a4},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f3)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a1, a2, a3, a4}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
if got, exp := len(r.Entries([]byte("cpu,host=A#!~#value"))), 2; got != exp {
t.Fatalf("block count mismatch: got %v, exp %v", got, exp)
}
}
// Ensures that a full compaction will skip over blocks that have the full
// range of time contained in the block tombstoned
func TestCompactor_CompactFull_TombstonedSkipBlock(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
a2 := tsm1.NewValue(2, 1.2)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2},
}
f1 := MustWriteTSM(dir, 1, writes)
ts := tsm1.NewTombstoner(f1, nil)
ts.AddRange([][]byte{[]byte("cpu,host=A#!~#value")}, math.MinInt64, math.MaxInt64)
if err := ts.Flush(); err != nil {
t.Fatalf("unexpected error flushing tombstone: %v", err)
}
a3 := tsm1.NewValue(3, 1.3)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a3},
}
f2 := MustWriteTSM(dir, 2, writes)
a4 := tsm1.NewValue(4, 1.4)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a4},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f3)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a3, a4}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
if got, exp := len(r.Entries([]byte("cpu,host=A#!~#value"))), 1; got != exp {
t.Fatalf("block count mismatch: got %v, exp %v", got, exp)
}
}
// Ensures that a full compaction will decode and combine blocks with
// partial tombstoned values
func TestCompactor_CompactFull_TombstonedPartialBlock(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
a2 := tsm1.NewValue(2, 1.2)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2},
}
f1 := MustWriteTSM(dir, 1, writes)
ts := tsm1.NewTombstoner(f1, nil)
// a1 should remain after compaction
ts.AddRange([][]byte{[]byte("cpu,host=A#!~#value")}, 2, math.MaxInt64)
if err := ts.Flush(); err != nil {
t.Fatalf("unexpected error flushing tombstone: %v", err)
}
a3 := tsm1.NewValue(3, 1.3)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a3},
}
f2 := MustWriteTSM(dir, 2, writes)
a4 := tsm1.NewValue(4, 1.4)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a4},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f3)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a1, a3, a4}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
if got, exp := len(r.Entries([]byte("cpu,host=A#!~#value"))), 2; got != exp {
t.Fatalf("block count mismatch: got %v, exp %v", got, exp)
}
}
// Ensures that a full compaction will decode and combine blocks with
// multiple tombstoned ranges within the block e.g. (t1, t2, t3, t4)
// having t2 and t3 removed
func TestCompactor_CompactFull_TombstonedMultipleRanges(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
// write 3 TSM files with different data and one new point
a1 := tsm1.NewValue(1, 1.1)
a2 := tsm1.NewValue(2, 1.2)
a3 := tsm1.NewValue(3, 1.3)
a4 := tsm1.NewValue(4, 1.4)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a1, a2, a3, a4},
}
f1 := MustWriteTSM(dir, 1, writes)
ts := tsm1.NewTombstoner(f1, nil)
// a1, a3 should remain after compaction
ts.AddRange([][]byte{[]byte("cpu,host=A#!~#value")}, 2, 2)
ts.AddRange([][]byte{[]byte("cpu,host=A#!~#value")}, 4, 4)
if err := ts.Flush(); err != nil {
t.Fatalf("unexpected error flushing tombstone: %v", err)
}
a5 := tsm1.NewValue(5, 1.5)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a5},
}
f2 := MustWriteTSM(dir, 2, writes)
a6 := tsm1.NewValue(6, 1.6)
writes = map[string][]tsm1.Value{
"cpu,host=A#!~#value": {a6},
}
f3 := MustWriteTSM(dir, 3, writes)
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
files, err := compactor.CompactFull([]string{f1, f2, f3}, zap.NewNop(), 2)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 1; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f3)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
r := MustOpenTSMReader(files[0])
if got, exp := r.KeyCount(), 1; got != exp {
t.Fatalf("keys length mismatch: got %v, exp %v", got, exp)
}
var data = []struct {
key string
points []tsm1.Value
}{
{"cpu,host=A#!~#value", []tsm1.Value{a1, a3, a5, a6}},
}
for _, p := range data {
values, err := r.ReadAll([]byte(p.key))
if err != nil {
t.Fatalf("unexpected error reading: %v", err)
}
if got, exp := len(values), len(p.points); got != exp {
t.Fatalf("values length mismatch %s: got %v, exp %v", p.key, got, exp)
}
for i, point := range p.points {
assertValueEqual(t, values[i], point)
}
}
if got, exp := len(r.Entries([]byte("cpu,host=A#!~#value"))), 2; got != exp {
t.Fatalf("block count mismatch: got %v, exp %v", got, exp)
}
}
// Ensures that a compaction will properly rollover to a new file when the
// max keys per blocks is exceeded
func TestCompactor_CompactFull_MaxKeys(t *testing.T) {
// This test creates a lot of data and causes timeout failures for these envs
if testing.Short() || os.Getenv("CI") != "" || os.Getenv("GORACE") != "" {
t.Skip("Skipping max keys compaction test")
}
dir := MustTempDir()
defer os.RemoveAll(dir)
// write two files where the first contains a single key with the maximum
// number of full blocks that can fit in a TSM file
f1, f1Name := MustTSMWriter(dir, 1)
values := make([]tsm1.Value, 1000)
for i := 0; i < 65534; i++ {
values = values[:0]
for j := 0; j < 1000; j++ {
values = append(values, tsm1.NewValue(int64(i*1000+j), int64(1)))
}
if err := f1.Write([]byte("cpu,host=A#!~#value"), values); err != nil {
t.Fatalf("write tsm f1: %v", err)
}
}
if err := f1.WriteIndex(); err != nil {
t.Fatalf("write index f1: %v", err)
}
f1.Close()
// Write a new file with 2 blocks that when compacted would exceed the max
// blocks
f2, f2Name := MustTSMWriter(dir, 2)
for i := 0; i < 2; i++ {
lastTimeStamp := values[len(values)-1].UnixNano() + 1
values = values[:0]
for j := lastTimeStamp; j < lastTimeStamp+1000; j++ {
values = append(values, tsm1.NewValue(int64(j), int64(1)))
}
if err := f2.Write([]byte("cpu,host=A#!~#value"), values); err != nil {
t.Fatalf("write tsm f1: %v", err)
}
}
if err := f2.WriteIndex(); err != nil {
t.Fatalf("write index f2: %v", err)
}
f2.Close()
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
// Compact both files, should get 2 files back
files, err := compactor.CompactFull([]string{f1Name, f2Name}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
if err != nil {
t.Fatalf("unexpected error writing snapshot: %v", err)
}
if got, exp := len(files), 2; got != exp {
t.Fatalf("files length mismatch: got %v, exp %v", got, exp)
}
expGen, expSeq, err := tsm1.DefaultParseFileName(f2Name)
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
expSeq = expSeq + 1
gotGen, gotSeq, err := tsm1.DefaultParseFileName(files[0])
if err != nil {
t.Fatalf("unexpected error parsing file name: %v", err)
}
if gotGen != expGen {
t.Fatalf("wrong generation for new file: got %v, exp %v", gotGen, expGen)
}
if gotSeq != expSeq {
t.Fatalf("wrong sequence for new file: got %v, exp %v", gotSeq, expSeq)
}
}
func TestCompactor_CompactFull_InProgress(t *testing.T) {
// This test creates a lot of data and causes timeout failures for these envs
if testing.Short() || os.Getenv("CI") != "" || os.Getenv("GORACE") != "" {
t.Skip("Skipping in progress compaction test")
}
dir := MustTempDir()
defer os.RemoveAll(dir)
f2Name := func() string {
values := make([]tsm1.Value, 1000)
// Write a new file with 2 blocks
f2, f2Name := MustTSMWriter(dir, 2)
defer func() {
assert.NoError(t, f2.Close(), "closing TSM file %s", f2Name)
}()
for i := 0; i < 2; i++ {
values = values[:0]
for j := 0; j < 1000; j++ {
values = append(values, tsm1.NewValue(int64(i*1000+j), int64(1)))
}
assert.NoError(t, f2.Write([]byte("cpu,host=A#!~#value"), values), "writing TSM file: %s", f2Name)
}
assert.NoError(t, f2.WriteIndex(), "writing TSM file index for %s", f2Name)
return f2Name
}()
ffs := &fakeFileStore{}
defer ffs.Close()
compactor := tsm1.NewCompactor()
compactor.Dir = dir
compactor.FileStore = ffs
compactor.Open()
expGen, expSeq, err := tsm1.DefaultParseFileName(f2Name)
assert.NoError(t, err, "unexpected error parsing file name %s", f2Name)
expSeq = expSeq + 1
fileName := filepath.Join(compactor.Dir, tsm1.DefaultFormatFileName(expGen, expSeq)+"."+tsm1.TSMFileExtension+"."+tsm1.TmpTSMFileExtension)
// Create a temp file to simulate an in progress compaction
f, err := os.Create(fileName)
assert.NoError(t, err, "creating in-progress compaction file %s", fileName)
defer func() {
assert.NoError(t, f.Close(), "closing in-progress compaction file %s", fileName)
}()
_, err = compactor.CompactFull([]string{f2Name}, zap.NewNop(), tsdb.DefaultMaxPointsPerBlock)
assert.Errorf(t, err, "expected an error writing snapshot for %s", f2Name)
assert.ErrorContainsf(t, err, "file exists", "unexpected error writing snapshot for %s", f2Name)
assert.Truef(t, errors.Is(err, fs.ErrExist), "error did not indicate file existence: %v", err)
pathErr := &os.PathError{}
assert.Truef(t, errors.As(err, &pathErr), "expected path error, got %v", err)
assert.Truef(t, errors.Is(pathErr, fs.ErrExist), "error did not indicate file existence: %v", pathErr)
}
func newTSMKeyIterator(size int, fast bool, interrupt chan struct{}, readers ...*tsm1.TSMReader) (tsm1.KeyIterator, error) {
files := []string{}
for _, r := range readers {
files = append(files, r.Path())
}
return tsm1.NewTSMBatchKeyIterator(size, fast, 0, interrupt, files, readers...)
}
// Tests that a single TSM file can be read and iterated over
func TestTSMKeyIterator_Single(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(1, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v1},
}
r := MustTSMReader(dir, 1, writes)
iter, err := newTSMKeyIterator(1, false, nil, r)
if err != nil {
t.Fatalf("unexpected error creating WALKeyIterator: %v", err)
}
var readValues bool
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if got, exp := string(key), "cpu,host=A#!~#value"; got != exp {
t.Fatalf("key mismatch: got %v, exp %v", got, exp)
}
if got, exp := len(values), len(writes); got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
for _, v := range values {
readValues = true
assertValueEqual(t, v, v1)
}
}
if !readValues {
t.Fatalf("failed to read any values")
}
}
// Tests that duplicate point values are merged. There is only one case
// where this could happen and that is when a compaction completed and we replace
// the old TSM file with a new one and we crash just before deleting the old file.
// No data is lost but the same point time/value would exist in two files until
// compaction corrects it.
func TestTSMKeyIterator_Duplicate(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(1, int64(1))
v2 := tsm1.NewValue(1, int64(2))
writes1 := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v1},
}
r1 := MustTSMReader(dir, 1, writes1)
writes2 := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v2},
}
r2 := MustTSMReader(dir, 2, writes2)
iter, err := newTSMKeyIterator(1, false, nil, r1, r2)
if err != nil {
t.Fatalf("unexpected error creating WALKeyIterator: %v", err)
}
var readValues bool
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if got, exp := string(key), "cpu,host=A#!~#value"; got != exp {
t.Fatalf("key mismatch: got %v, exp %v", got, exp)
}
if got, exp := len(values), 1; got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
readValues = true
assertValueEqual(t, values[0], v2)
}
if !readValues {
t.Fatalf("failed to read any values")
}
}
// Tests that deleted keys are not seen during iteration with
// TSM files.
func TestTSMKeyIterator_MultipleKeysDeleted(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(2, int64(1))
points1 := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v1},
}
r1 := MustTSMReader(dir, 1, points1)
if e := r1.Delete([][]byte{[]byte("cpu,host=A#!~#value")}); nil != e {
t.Fatal(e)
}
v2 := tsm1.NewValue(1, float64(1))
v3 := tsm1.NewValue(1, float64(1))
points2 := map[string][]tsm1.Value{
"cpu,host=A#!~#count": {v2},
"cpu,host=B#!~#value": {v3},
}
r2 := MustTSMReader(dir, 2, points2)
r2.Delete([][]byte{[]byte("cpu,host=A#!~#count")})
iter, err := newTSMKeyIterator(1, false, nil, r1, r2)
if err != nil {
t.Fatalf("unexpected error creating WALKeyIterator: %v", err)
}
var readValues bool
var data = []struct {
key string
value tsm1.Value
}{
{"cpu,host=B#!~#value", v3},
}
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if got, exp := string(key), data[0].key; got != exp {
t.Fatalf("key mismatch: got %v, exp %v", got, exp)
}
if got, exp := len(values), 1; got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
readValues = true
assertValueEqual(t, values[0], data[0].value)
data = data[1:]
}
if !readValues {
t.Fatalf("failed to read any values")
}
}
// Tests that deleted keys are not seen during iteration with
// TSM files.
func TestTSMKeyIterator_SingleDeletes(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(10, int64(1))
v2 := tsm1.NewValue(20, int64(1))
v3 := tsm1.NewValue(30, int64(1))
v4 := tsm1.NewValue(40, int64(1))
v5 := tsm1.NewValue(50, int64(1))
v6 := tsm1.NewValue(60, int64(1))
points1 := map[string][]tsm1.Value{
"cpu,host=0#!~#value": {v1, v2},
"cpu,host=A#!~#value": {v5, v6},
"cpu,host=B#!~#value": {v3, v4},
"cpu,host=C#!~#value": {v1, v2},
"cpu,host=D#!~#value": {v1, v2},
}
r1 := MustTSMReader(dir, 1, points1)
if e := r1.DeleteRange([][]byte{[]byte("cpu,host=A#!~#value")}, 50, 50); nil != e {
t.Fatal(e)
}
if e := r1.DeleteRange([][]byte{[]byte("cpu,host=A#!~#value")}, 60, 60); nil != e {
t.Fatal(e)
}
if e := r1.DeleteRange([][]byte{[]byte("cpu,host=C#!~#value")}, 10, 10); nil != e {
t.Fatal(e)
}
if e := r1.DeleteRange([][]byte{[]byte("cpu,host=C#!~#value")}, 60, 60); nil != e {
t.Fatal(e)
}
if e := r1.DeleteRange([][]byte{[]byte("cpu,host=C#!~#value")}, 20, 20); nil != e {
t.Fatal(e)
}
iter, err := newTSMKeyIterator(1, false, nil, r1)
if err != nil {
t.Fatalf("unexpected error creating WALKeyIterator: %v", err)
}
var readValues int
var data = []struct {
key string
value tsm1.Value
}{
{"cpu,host=0#!~#value", v1},
{"cpu,host=B#!~#value", v3},
{"cpu,host=D#!~#value", v1},
}
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if exp, got := string(key), data[0].key; exp != got {
t.Fatalf("key mismatch: got %v, exp %v", exp, got)
}
if exp, got := len(values), 2; exp != got {
t.Fatalf("values length mismatch: exp %v, got %v", exp, got)
}
readValues++
assertValueEqual(t, values[0], data[0].value)
data = data[1:]
}
if exp, got := 3, readValues; exp != got {
t.Fatalf("failed to read expected values: exp %v, got %v", exp, got)
}
}
// Tests that the TSMKeyIterator will abort if the interrupt channel is closed
func TestTSMKeyIterator_Abort(t *testing.T) {
dir := MustTempDir()
defer os.RemoveAll(dir)
v1 := tsm1.NewValue(1, 1.1)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v1},
}
r := MustTSMReader(dir, 1, writes)
intC := make(chan struct{})
iter, err := newTSMKeyIterator(1, false, intC, r)
if err != nil {
t.Fatalf("unexpected error creating WALKeyIterator: %v", err)
}
var aborted bool
for iter.Next() {
// Abort
close(intC)
_, _, _, _, err := iter.Read()
if err == nil {
t.Fatalf("unexpected error read: %v", err)
}
aborted = err != nil
}
if !aborted {
t.Fatalf("iteration not aborted")
}
}
func TestCacheKeyIterator_Single(t *testing.T) {
v0 := tsm1.NewValue(1, 1.0)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v0},
}
c := tsm1.NewCache(0)
for k, v := range writes {
if err := c.Write([]byte(k), v); err != nil {
t.Fatalf("failed to write key foo to cache: %s", err.Error())
}
}
iter := tsm1.NewCacheKeyIterator(c, 1, nil)
var readValues bool
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if got, exp := string(key), "cpu,host=A#!~#value"; got != exp {
t.Fatalf("key mismatch: got %v, exp %v", got, exp)
}
if got, exp := len(values), len(writes); got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
for _, v := range values {
readValues = true
assertValueEqual(t, v, v0)
}
}
if !readValues {
t.Fatalf("failed to read any values")
}
}
func TestCacheKeyIterator_Chunked(t *testing.T) {
v0 := tsm1.NewValue(1, 1.0)
v1 := tsm1.NewValue(2, 2.0)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v0, v1},
}
c := tsm1.NewCache(0)
for k, v := range writes {
if err := c.Write([]byte(k), v); err != nil {
t.Fatalf("failed to write key foo to cache: %s", err.Error())
}
}
iter := tsm1.NewCacheKeyIterator(c, 1, nil)
var readValues bool
var chunk int
for iter.Next() {
key, _, _, block, err := iter.Read()
if err != nil {
t.Fatalf("unexpected error read: %v", err)
}
values, err := tsm1.DecodeBlock(block, nil)
if err != nil {
t.Fatalf("unexpected error decode: %v", err)
}
if got, exp := string(key), "cpu,host=A#!~#value"; got != exp {
t.Fatalf("key mismatch: got %v, exp %v", got, exp)
}
if got, exp := len(values), 1; got != exp {
t.Fatalf("values length mismatch: got %v, exp %v", got, exp)
}
for _, v := range values {
readValues = true
assertValueEqual(t, v, writes["cpu,host=A#!~#value"][chunk])
}
chunk++
}
if !readValues {
t.Fatalf("failed to read any values")
}
}
// Tests that the CacheKeyIterator will abort if the interrupt channel is closed
func TestCacheKeyIterator_Abort(t *testing.T) {
v0 := tsm1.NewValue(1, 1.0)
writes := map[string][]tsm1.Value{
"cpu,host=A#!~#value": {v0},
}
c := tsm1.NewCache(0)
for k, v := range writes {
if err := c.Write([]byte(k), v); err != nil {
t.Fatalf("failed to write key foo to cache: %s", err.Error())
}
}
intC := make(chan struct{})
iter := tsm1.NewCacheKeyIterator(c, 1, intC)
var aborted bool
for iter.Next() {
//Abort
close(intC)
_, _, _, _, err := iter.Read()
if err == nil {
t.Fatalf("unexpected error read: %v", err)
}
aborted = err != nil
}
if !aborted {
t.Fatalf("iteration not aborted")
}
}
func normalizeExtFileStat(efs []tsm1.ExtFileStat, defaultBlockCount int) []tsm1.ExtFileStat {
efsNorm := make([]tsm1.ExtFileStat, 0, len(efs))
for _, f := range efs {
if f.FirstBlockCount == 0 {
f.FirstBlockCount = defaultBlockCount
}
efsNorm = append(efsNorm, f)
}
return efsNorm
}
type ffsOpt func(ffs *fakeFileStore)
func withExtFileStats(efs []tsm1.ExtFileStat) ffsOpt {
return func(ffs *fakeFileStore) {
ffs.PathsFn = func() []tsm1.ExtFileStat { return normalizeExtFileStat(efs, ffs.defaultBlockCount) }
}
}
func withFileStats(fs []tsm1.FileStat) ffsOpt {
return withExtFileStats(tsm1.FileStats(fs).ToExtFileStats())
}
func withDefaultBlockCount(blockCount int) ffsOpt {
return func(ffs *fakeFileStore) {
ffs.defaultBlockCount = blockCount
}
}
func newFakeFileStore(opts ...ffsOpt) *fakeFileStore {
ffs := &fakeFileStore{}
for _, o := range opts {
o(ffs)
}
return ffs
}
func TestDefaultPlanner_Plan_Min(t *testing.T) {
fileStats := []tsm1.FileStat{
{
Path: "01-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "02-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-1.tsm1",
Size: 251 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(fileStats)), tsdb.DefaultCompactFullWriteColdDuration,
)
tsm, pLen := cp.Plan(time.Now())
if exp, got := 0, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
// Ensure that if there are older files that can be compacted together but a newer
// file that is in a larger step, the older ones will get compacted.
func TestDefaultPlanner_Plan_CombineSequence(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-04.tsm1",
Size: 128 * 1024 * 1024,
},
{
Path: "02-04.tsm1",
Size: 128 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 128 * 1024 * 1024,
},
{
Path: "04-04.tsm1",
Size: 128 * 1024 * 1024,
},
{
Path: "06-02.tsm1",
Size: 67 * 1024 * 1024,
},
{
Path: "07-02.tsm1",
Size: 128 * 1024 * 1024,
},
{
Path: "08-01.tsm1",
Size: 251 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[0], data[1], data[2], data[3]}
tsm, pLen := cp.Plan(time.Now())
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
// Ensure that the planner grabs the smallest compaction step
func TestDefaultPlanner_Plan_MultipleGroups(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "02-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "04-04.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "05-04.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "06-04.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "07-04.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "08-04.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "09-04.tsm1", // should be skipped
Size: 129 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration)
expFiles := []tsm1.FileStat{data[0], data[1], data[2], data[3],
data[4], data[5], data[6], data[7]}
tsm, pLen := cp.Plan(time.Now())
if got, exp := len(tsm), 2; got != exp {
t.Fatalf("compaction group length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
if exp, got := len(expFiles[:4]), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
}
if exp, got := len(expFiles[4:]), len(tsm[1]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
}
for i, p := range expFiles[:4] {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
for i, p := range expFiles[4:] {
if got, exp := tsm[1][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
// Ensure that the planner grabs the smallest compaction step
func TestDefaultPlanner_PlanLevel_SmallestCompactionStep(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-03.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "04-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "05-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "07-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "08-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "09-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "10-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "11-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "12-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[4], data[5], data[6], data[7], data[8], data[9], data[10], data[11]}
tsm, pLen := cp.PlanLevel(1)
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_PlanLevel_SplitFile(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-03.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "04-03.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[0], data[1], data[2], data[3], data[4]}
tsm, pLen := cp.PlanLevel(3)
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_PlanLevel_IsolatedHighLevel(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-02.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-02.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "04-02.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-02.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-02.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{}
tsm, pLen := cp.PlanLevel(3)
if exp, got := len(expFiles), len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
func TestDefaultPlanner_PlanLevel3_MinFiles(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-03.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-01.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "04-01.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-02.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{}
tsm, pLen := cp.PlanLevel(3)
if exp, got := len(expFiles), len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
func TestDefaultPlanner_PlanLevel2_MinFiles(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "02-04.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "03-02.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{}
tsm, pLen := cp.PlanLevel(2)
if exp, got := len(expFiles), len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
func TestDefaultPlanner_PlanLevel_Tombstone(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-03.tsm1",
Size: 251 * 1024 * 1024,
HasTombstone: true,
},
{
Path: "02-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-01.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "04-01.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-02.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[0], data[1]}
tsm, pLen := cp.PlanLevel(3)
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_PlanLevel_Multiple(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-01.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-01.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "04-01.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "07-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "08-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles1 := []tsm1.FileStat{data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7]}
tsm, pLen := cp.PlanLevel(1)
if exp, got := len(expFiles1), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles1 {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_PlanLevel_InUse(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-01.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-01.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
{
Path: "04-01.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "05-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "07-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "08-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "09-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "10-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "11-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "12-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "13-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "14-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "15-01.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "16-01.tsm1",
Size: 1 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles1 := data[0:8]
expFiles2 := data[8:16]
tsm, pLen := cp.PlanLevel(1)
if exp, got := len(expFiles1), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles1 {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
if exp, got := len(expFiles2), len(tsm[1]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
}
for i, p := range expFiles2 {
if got, exp := tsm[1][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
cp.Release(tsm[1:])
tsm, pLen = cp.PlanLevel(1)
if exp, got := len(expFiles2), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles2 {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_PlanOptimize_NoLevel4(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-03.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "02-03.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{}
tsm, pLen, gLen := cp.PlanOptimize(time.Now().Add(-tsdb.DefaultCompactFullWriteColdDuration + 1))
if exp, got := len(expFiles), len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
} else if gLen != int64(3) {
t.Fatalf("generation len plan mismatch: got %v, exp %v", gLen, 3)
}
}
func TestDefaultPlanner_PlanOptimize_Tombstones(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-04.tsm1",
Size: 251 * 1024 * 1024,
},
{
Path: "01-05.tsm1",
Size: 1 * 1024 * 1024,
HasTombstone: true,
},
{
Path: "02-06.tsm1",
Size: 2 * 1024 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[0], data[1], data[2]}
tsm, pLen, _ := cp.PlanOptimize(time.Now().Add(-tsdb.DefaultCompactFullWriteColdDuration + 1))
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
// Ensure that the planner will compact all files if no writes
// have happened in some interval
func TestDefaultPlanner_Plan_FullOnCold(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-01.tsm1",
Size: 513 * 1024 * 1024,
},
{
Path: "02-02.tsm1",
Size: 129 * 1024 * 1024,
},
{
Path: "03-02.tsm1",
Size: 33 * 1024 * 1024,
},
{
Path: "04-02.tsm1",
Size: 1 * 1024 * 1024,
},
{
Path: "05-02.tsm1",
Size: 10 * 1024 * 1024,
},
{
Path: "06-01.tsm1",
Size: 2 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
time.Nanosecond,
)
tsm, pLen := cp.Plan(time.Now().Add(-time.Second))
if exp, got := len(data), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range data {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
// Ensure that the planner will not return files that are over the max
// allowable size
func TestDefaultPlanner_Plan_SkipMaxSizeFiles(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-01.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "02-02.tsm1",
Size: 2049 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
tsm, pLen := cp.Plan(time.Now())
if exp, got := 0, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
// Ensure that the planner will not return files that are over the max
// allowable size
func TestDefaultPlanner_Plan_SkipPlanningAfterFull(t *testing.T) {
testSet := []tsm1.FileStat{
{
Path: "01-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "02-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "03-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "04-04.tsm1",
Size: 256 * 1024 * 1024,
},
}
ffs := newFakeFileStore(withFileStats(testSet), withDefaultBlockCount(tsdb.DefaultMaxPointsPerBlock))
cp := tsm1.NewDefaultPlanner(ffs, time.Nanosecond)
plan, pLen := cp.Plan(time.Now().Add(-time.Second))
// first verify that our test set would return files
if exp, got := 4, len(plan[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(plan)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(plan)
// skip planning if all files are over the limit
over := []tsm1.FileStat{
{
Path: "01-05.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "02-05.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "03-05.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "04-05.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "05-05.tsm1",
Size: 2049 * 1024 * 1024,
},
}
overFs := newFakeFileStore(withFileStats(over), withDefaultBlockCount(tsdb.DefaultMaxPointsPerBlock))
cp.FileStore = overFs
plan, pLen = cp.Plan(time.Now().Add(-time.Second))
if exp, got := 0, len(plan); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(plan)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(plan)
plan, pLen, _ = cp.PlanOptimize(time.Now().Add(-tsdb.DefaultCompactFullWriteColdDuration + 1))
// ensure the optimize planner would pick this up
if exp, got := 1, len(plan); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(plan)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(plan)
cp.FileStore = ffs
// ensure that it will plan if last modified has changed
ffs.lastModified = time.Now()
cGroups, pLen := cp.Plan(time.Now())
if exp, got := 4, len(cGroups[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(cGroups)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
// Tests that 2 generations, each over 2 GB and the second in level 2 does
// not return just the first generation. This was a case where full planning
// would get repeatedly plan the same files and never stop.
func TestDefaultPlanner_Plan_TwoGenLevel3(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "000002245-000001666.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001667.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001668.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001669.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001670.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001671.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001672.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002245-000001673.tsm",
Size: 192631258,
},
{
Path: "000002246-000000002.tsm",
Size: 2049 * 1024 * 1024,
},
{
Path: "000002246-000000003.tsm",
Size: 192631258,
},
}
fs := newFakeFileStore(withFileStats(data), withDefaultBlockCount(tsdb.DefaultMaxPointsPerBlock))
cp := tsm1.NewDefaultPlanner(fs, time.Hour)
tsm, pLen := cp.Plan(time.Now().Add(-24 * time.Hour))
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
// Ensure that the planner will return files over the max file
// size, but do not contain full blocks
func TestDefaultPlanner_Plan_NotFullOverMaxsize(t *testing.T) {
testSet := []tsm1.FileStat{
{
Path: "01-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "02-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "03-05.tsm1",
Size: 256 * 1024 * 1024,
},
{
Path: "04-04.tsm1",
Size: 256 * 1024 * 1024,
},
}
ffs := newFakeFileStore(withFileStats(testSet), withDefaultBlockCount(100))
cp := tsm1.NewDefaultPlanner(
ffs,
time.Nanosecond,
)
plan, pLen := cp.Plan(time.Now().Add(-time.Second))
// first verify that our test set would return files
if exp, got := 4, len(plan[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(plan)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(plan)
// skip planning if all files are over the limit
over := []tsm1.FileStat{
{
Path: "01-05.tsm1",
Size: 2049 * 1024 * 1024,
},
{
Path: "02-05.tsm1",
Size: 2049 * 1024 * 1024,
},
}
overFs := newFakeFileStore(withFileStats(over), withDefaultBlockCount(100))
cp.FileStore = overFs
cGroups, pLen := cp.Plan(time.Now().Add(-time.Second))
if exp, got := 1, len(cGroups); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(cGroups)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
// Ensure that the planner will compact files that are past the smallest step
// size even if there is a single file in the smaller step size
func TestDefaultPlanner_Plan_CompactsMiddleSteps(t *testing.T) {
data := []tsm1.FileStat{
{
Path: "01-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "02-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "04-04.tsm1",
Size: 64 * 1024 * 1024,
},
{
Path: "05-02.tsm1",
Size: 2 * 1024 * 1024,
},
}
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(data)),
tsdb.DefaultCompactFullWriteColdDuration,
)
expFiles := []tsm1.FileStat{data[0], data[1], data[2], data[3]}
tsm, pLen := cp.Plan(time.Now())
if exp, got := len(expFiles), len(tsm[0]); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
for i, p := range expFiles {
if got, exp := tsm[0][i], p.Path; got != exp {
t.Fatalf("tsm file mismatch: got %v, exp %v", got, exp)
}
}
}
func TestDefaultPlanner_Plan_LargeGeneration(t *testing.T) {
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(
[]tsm1.FileStat{
{
Path: "000000278-000000006.tsm",
Size: 2148340232,
},
{
Path: "000000278-000000007.tsm",
Size: 2148356556,
},
{
Path: "000000278-000000008.tsm",
Size: 167780181,
},
{
Path: "000000278-000047040.tsm",
Size: 2148728539,
},
{
Path: "000000278-000047041.tsm",
Size: 701863692,
},
})),
tsdb.DefaultCompactFullWriteColdDuration,
)
tsm, pLen := cp.Plan(time.Now())
if exp, got := 0, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
}
func TestDefaultPlanner_Plan_ForceFull(t *testing.T) {
cp := tsm1.NewDefaultPlanner(
newFakeFileStore(withFileStats(
[]tsm1.FileStat{
{
Path: "000000001-000000001.tsm",
Size: 2148340232,
},
{
Path: "000000002-000000001.tsm",
Size: 2148356556,
},
{
Path: "000000003-000000001.tsm",
Size: 167780181,
},
{
Path: "000000004-000000001.tsm",
Size: 2148728539,
},
{
Path: "000000005-000000001.tsm",
Size: 2148340232,
},
{
Path: "000000006-000000001.tsm",
Size: 2148356556,
},
{
Path: "000000007-000000001.tsm",
Size: 167780181,
},
{
Path: "000000008-000000001.tsm",
Size: 2148728539,
},
{
Path: "000000009-000000002.tsm",
Size: 701863692,
},
{
Path: "000000010-000000002.tsm",
Size: 701863692,
},
{
Path: "000000011-000000002.tsm",
Size: 701863692,
},
{
Path: "000000012-000000002.tsm",
Size: 701863692,
},
{
Path: "000000013-000000002.tsm",
Size: 701863692,
},
})),
tsdb.DefaultCompactFullWriteColdDuration,
)
tsm, pLen := cp.PlanLevel(1)
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
tsm, pLen = cp.PlanLevel(2)
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
cp.ForceFull()
// Level plans should not return any plans
tsm, pLen = cp.PlanLevel(1)
if exp, got := 0, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
tsm, pLen = cp.PlanLevel(2)
if exp, got := 0, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
tsm, pLen = cp.Plan(time.Now())
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
if got, exp := len(tsm[0]), 13; got != exp {
t.Fatalf("plan length mismatch: got %v, exp %v", got, exp)
}
cp.Release(tsm)
// Level plans should return plans now that Plan has been called
tsm, pLen = cp.PlanLevel(1)
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
tsm, pLen = cp.PlanLevel(2)
if exp, got := 1, len(tsm); got != exp {
t.Fatalf("tsm file length mismatch: got %v, exp %v", got, exp)
} else if pLen != int64(len(tsm)) {
t.Fatalf("tsm file plan length mismatch: got %v, exp %v", pLen, exp)
}
cp.Release(tsm)
}
func TestIsGroupOptimized(t *testing.T) {
testSetNoExt := []tsm1.FileStat{
{
Path: "01-05.tsm",
Size: 256 * 1024 * 1024,
},
{
Path: "02-05.tsm",
Size: 256 * 1024 * 1024,
},
{
Path: "03-05.tsm",
Size: 256 * 1024 * 1024,
},
{
Path: "04-04.tsm",
Size: 256 * 1024 * 1024,
},
}
testSet := tsm1.FileStats(testSetNoExt).ToExtFileStats()
blockCounts := []struct {
blockCounts []int
optimizedName string
}{
{
blockCounts: []int{
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
},
optimizedName: "",
},
{
blockCounts: []int{
tsdb.DefaultAggressiveMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
},
optimizedName: "01-05.tsm",
},
{
blockCounts: []int{
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultAggressiveMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
},
optimizedName: "02-05.tsm",
},
{
blockCounts: []int{
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultMaxPointsPerBlock,
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
optimizedName: "04-04.tsm",
},
}
ffs := newFakeFileStore(withExtFileStats(testSet))
cp := tsm1.NewDefaultPlanner(ffs, tsdb.DefaultCompactFullWriteColdDuration)
e := MustOpenEngine(tsdb.InmemIndexName)
e.CompactionPlan = cp
e.Compactor = tsm1.NewCompactor()
e.Compactor.FileStore = ffs
fileGroup := make([]string, 0, len(testSet))
for j := 0; j < len(testSet); j++ {
fileGroup = append(fileGroup, testSet[j].Path)
}
for testIdx := 0; testIdx < len(blockCounts); testIdx++ {
require.Lenf(t, blockCounts[testIdx].blockCounts, len(testSet), "incorrect block count for case %d", testIdx)
for statIdx := range testSet {
testSet[statIdx].FirstBlockCount = blockCounts[testIdx].blockCounts[statIdx]
}
ok, fName, _ := e.IsGroupOptimized(fileGroup)
require.Equal(t, blockCounts[testIdx].optimizedName != "", ok, "unexpected result for optimization check")
require.Equal(t, blockCounts[testIdx].optimizedName, fName, "unexpected file name in optimization check")
}
}
func TestEnginePlanCompactions(t *testing.T) {
type testLevelResults struct {
level1Groups []tsm1.PlannedCompactionGroup
level2Groups []tsm1.PlannedCompactionGroup
level3Groups []tsm1.PlannedCompactionGroup
level4Groups []tsm1.PlannedCompactionGroup
level5Groups []tsm1.PlannedCompactionGroup
}
type testEnginePlanCompactionsRunner struct {
name string
files []tsm1.ExtFileStat
defaultBlockCount int // Default block count if member of files has FirstBlockCount of 0.
// This is specifically used to adjust the modification time
// so we can simulate the passage of time in tests
testShardTime time.Duration
// Each result is for the different plantypes
getResultByPlanType func(planType tsm1.PlanType) testLevelResults
}
tests := []testEnginePlanCompactionsRunner{
{
name: "many generations under 2GB",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-04.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm", "02-05.tsm", "03-05.tsm", "04-04.tsm"},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Many generations with files over 2GB",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-06.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-07.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-08.tsm1",
Size: 1048 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-06.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-07.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-08.tsm1",
Size: 1048 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "03-04.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: 10,
},
{
FileStat: tsm1.FileStat{
Path: "03-05.tsm1",
Size: 512 * 1024 * 1024,
},
FirstBlockCount: 5,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm1", "01-06.tsm1", "01-07.tsm1", "01-08.tsm1", "02-05.tsm1", "02-06.tsm1", "02-07.tsm1", "02-08.tsm1", "03-04.tsm1", "03-05.tsm1"},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Small group size with single generation",
/* These files are supposed to have 0 block counts */
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm1",
Size: 300 * 1024 * 1024,
},
},
{
FileStat: tsm1.FileStat{
Path: "01-06.tsm1",
Size: 200 * 1024 * 1024,
},
},
{
FileStat: tsm1.FileStat{
Path: "01-07.tsm1",
Size: 100 * 1024 * 1024,
},
},
{
FileStat: tsm1.FileStat{
Path: "01-08.tsm1",
Size: 50 * 1024 * 1024,
},
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm1",
"01-06.tsm1",
"01-07.tsm1",
"01-08.tsm1",
},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Small group size with single generation and levels under 4",
/* These files are supposed to have block counts of 0 */
files: tsm1.FileStats{
{
Path: "01-02.tsm1",
Size: 300 * 1024 * 1024,
},
{
Path: "01-03.tsm1",
Size: 200 * 1024 * 1024,
},
{
Path: "01-04.tsm1",
Size: 100 * 1024 * 1024,
},
}.ToExtFileStats(),
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-02.tsm1",
"01-03.tsm1",
"01-04.tsm1",
},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Small group size with single generation all at DefaultMaxPointsPerBlock",
files: tsm1.FileStats{
{
Path: "01-05.tsm1",
Size: 300 * 1024 * 1024,
},
{
Path: "01-06.tsm1",
Size: 200 * 1024 * 1024,
},
{
Path: "01-07.tsm1",
Size: 100 * 1024 * 1024,
},
{
Path: "01-08.tsm1",
Size: 50 * 1024 * 1024,
},
}.ToExtFileStats(),
defaultBlockCount: tsdb.DefaultMaxPointsPerBlock,
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm1",
"01-06.tsm1",
"01-07.tsm1",
"01-08.tsm1",
},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Small group size with single generation 50% at DefaultMaxPointsPerBlock and 50% at DefaultAggressiveMaxPointsPerBlock",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm1",
Size: 700 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-06.tsm1",
Size: 500 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-07.tsm1",
Size: 400 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-08.tsm1",
Size: 300 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-09.tsm1",
Size: 200 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-10.tsm1",
Size: 100 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-11.tsm1",
Size: 50 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-12.tsm1",
Size: 25 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm1",
"01-06.tsm1",
"01-07.tsm1",
"01-08.tsm1",
"01-09.tsm1",
"01-10.tsm1",
"01-11.tsm1",
"01-12.tsm1",
},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Group size over 2GB with single generation",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-13.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-14.tsm1",
Size: 650 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-15.tsm1",
Size: 450 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-13.tsm1",
"01-14.tsm1",
"01-15.tsm1",
},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
// This test is added to account for halting state after
// TestDefaultPlanner_FullyCompacted_SmallSingleGeneration
// will need to ensure that once we have single TSM file under 2 GB we stop
name: "Single TSM file",
/* These files are supposed to have 0 blocks */
files: tsm1.FileStats{
{
Path: "01-09.tsm1",
Size: 650 * 1024 * 1024,
},
}.ToExtFileStats(),
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
return testLevelResults{}
},
},
{
// This test is added to account for a single generation that has a group size
// over 2 GB with 1 file under 2 GB all at max points per block with aggressive compaction.
// It should not compact any further.
name: "TSM files at DefaultAggressiveMaxPointsPerBlock",
files: tsm1.FileStats{
{
Path: "01-13.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "01-14.tsm1",
Size: 691 * 1024 * 1024,
},
}.ToExtFileStats(),
defaultBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
return testLevelResults{}
},
},
{
// This test is added to account for a single generation that has a group size
// over 2 GB at max points per block with aggressive compaction, and, 1 file
// under 2 GB at default max points per block.
// It should not compact any further.
name: "TSM files cannot compact further, single file under 2G and at DefaultMaxPointsPerBlock",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-13.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-14.tsm1",
Size: 691 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
return testLevelResults{}
},
},
{
// This test is added to account for a single generation that has a group size
// over 2 GB and multiple files under 2 GB all at max points per block for aggressive compaction.
name: "Group size over 2 with multiple files under 2GB and at DefaultAggressiveMaxPointsPerBlock",
files: tsm1.FileStats{
{
Path: "01-13.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "01-14.tsm1",
Size: 650 * 1024 * 1024,
},
{
Path: "01-15.tsm1",
Size: 450 * 1024 * 1024,
},
}.ToExtFileStats(),
defaultBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
return testLevelResults{}
},
},
{
name: "Generations with files under level 4",
// These files are supposed to have block counts of 0
files: tsm1.FileStats{
{
Path: "01-05.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "01-06.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "01-07.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "01-08.tsm1",
Size: 1048 * 1024 * 1024,
},
{
Path: "02-05.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "02-06.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "02-07.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "02-08.tsm1",
Size: 1048 * 1024 * 1024,
},
{
Path: "03-03.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "03-04.tsm1",
Size: 2048 * 1024 * 1024,
},
{
Path: "03-05.tsm1",
Size: 600 * 1024 * 1024,
},
{
Path: "03-06.tsm1",
Size: 500 * 1024 * 1024,
},
}.ToExtFileStats(),
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm1",
"01-06.tsm1",
"01-07.tsm1",
"01-08.tsm1",
"02-05.tsm1",
"02-06.tsm1",
"02-07.tsm1",
"02-08.tsm1",
"03-03.tsm1",
"03-04.tsm1",
"03-05.tsm1",
"03-06.tsm1"},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
// This test will mock a 'backfill' condition where we have a single
// shard with many generations. The initial generation should be fully
// compacted, but we have some new generations that are not. We need to ensure
// the optimize planner will pick these up and compact everything together.
name: "Backfill mock condition",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-06.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "01-07.tsm1",
Size: 2048 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-04.tsm1",
Size: 700 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm1",
Size: 500 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-06.tsm1",
Size: 400 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-02.tsm1",
Size: 700 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-03.tsm1",
Size: 500 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-04.tsm1",
Size: 400 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-01.tsm1",
Size: 700 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-02.tsm1",
Size: 500 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "03-03.tsm1",
Size: 400 * 1024 * 1024,
},
FirstBlockCount: 10,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{
"01-05.tsm1",
"01-06.tsm1",
"01-07.tsm1",
"02-04.tsm1",
"02-05.tsm1",
"02-06.tsm1",
"03-02.tsm1",
"03-03.tsm1",
"03-04.tsm1",
"03-03.tsm1",
"04-01.tsm1",
"04-02.tsm1",
}, tsdb.DefaultAggressiveMaxPointsPerBlock},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "1.12.0 RC0 Planner issue mock data from cluster",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "000029202-000000004.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000005.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000006.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000007.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000008.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000009.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000010.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000011.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000012.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000013.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000014.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000015.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000016.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000017.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000018.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000019.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000020.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000021.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000022.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000023.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000024.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000025.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000026.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000027.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000028.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000029.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000030.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000031.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000032.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000033.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000034.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000035.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000036.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000037.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000038.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000039.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000040.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000041.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000042.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000043.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000044.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000045.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029202-000000046.tsm",
Size: 161480704,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029235-000000003.tsm",
Size: 96468992,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029267-000000003.tsm",
Size: 109051904,
},
FirstBlockCount: 224,
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000001.tsm",
Size: 3040870,
},
FirstBlockCount: 413,
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000002.tsm",
Size: 2254857830,
},
FirstBlockCount: 561,
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000003.tsm",
Size: 2254857830,
},
FirstBlockCount: 402,
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000004.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000005.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000006.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000007.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000008.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000009.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000010.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000011.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000012.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000013.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000014.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000015.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000016.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000017.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000018.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000019.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000020.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000021.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000022.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000023.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000024.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000025.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000026.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000027.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000028.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000029.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000030.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000031.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000032.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000033.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000034.tsm",
Size: 2254857830,
},
},
{
FileStat: tsm1.FileStat{
Path: "000029268-000000035.tsm",
Size: 1717986918,
},
FirstBlockCount: 368,
},
},
defaultBlockCount: tsdb.DefaultMaxPointsPerBlock,
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
common := testLevelResults{
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{
"000029202-000000004.tsm",
"000029202-000000005.tsm",
"000029202-000000006.tsm",
"000029202-000000007.tsm",
"000029202-000000008.tsm",
"000029202-000000009.tsm",
"000029202-000000010.tsm",
"000029202-000000011.tsm",
"000029202-000000012.tsm",
"000029202-000000013.tsm",
"000029202-000000014.tsm",
"000029202-000000015.tsm",
"000029202-000000016.tsm",
"000029202-000000017.tsm",
"000029202-000000018.tsm",
"000029202-000000019.tsm",
"000029202-000000020.tsm",
"000029202-000000021.tsm",
"000029202-000000022.tsm",
"000029202-000000023.tsm",
"000029202-000000024.tsm",
"000029202-000000025.tsm",
"000029202-000000026.tsm",
"000029202-000000027.tsm",
"000029202-000000028.tsm",
"000029202-000000029.tsm",
"000029202-000000030.tsm",
"000029202-000000031.tsm",
"000029202-000000032.tsm",
"000029202-000000033.tsm",
"000029202-000000034.tsm",
"000029202-000000035.tsm",
"000029202-000000036.tsm",
"000029202-000000037.tsm",
"000029202-000000038.tsm",
"000029202-000000039.tsm",
"000029202-000000040.tsm",
"000029202-000000041.tsm",
"000029202-000000042.tsm",
"000029202-000000043.tsm",
"000029202-000000044.tsm",
"000029202-000000045.tsm",
"000029202-000000046.tsm",
"000029235-000000003.tsm",
"000029267-000000003.tsm",
"000029268-000000001.tsm",
"000029268-000000002.tsm",
"000029268-000000003.tsm",
"000029268-000000004.tsm",
"000029268-000000005.tsm",
"000029268-000000006.tsm",
"000029268-000000007.tsm",
"000029268-000000008.tsm",
"000029268-000000009.tsm",
"000029268-000000010.tsm",
"000029268-000000011.tsm",
"000029268-000000012.tsm",
"000029268-000000013.tsm",
"000029268-000000014.tsm",
"000029268-000000015.tsm",
"000029268-000000016.tsm",
"000029268-000000017.tsm",
"000029268-000000018.tsm",
"000029268-000000019.tsm",
"000029268-000000020.tsm",
"000029268-000000021.tsm",
"000029268-000000022.tsm",
"000029268-000000023.tsm",
"000029268-000000024.tsm",
"000029268-000000025.tsm",
"000029268-000000026.tsm",
"000029268-000000027.tsm",
"000029268-000000028.tsm",
"000029268-000000029.tsm",
"000029268-000000030.tsm",
"000029268-000000031.tsm",
"000029268-000000032.tsm",
"000029268-000000033.tsm",
"000029268-000000034.tsm",
"000029268-000000035.tsm",
},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard, tsm1.PT_SmartOptimize:
return common
case tsm1.PT_NoOptimize:
return testLevelResults{}
}
return testLevelResults{}
},
},
{
name: "Mock another planned level inside scheduler",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-04.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "05-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "06-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "07-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "08-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "09-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "10-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "11-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "12-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "13-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "14-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
standard := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm", "02-05.tsm", "03-05.tsm", "04-04.tsm"},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
common := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard:
return standard
case tsm1.PT_SmartOptimize, tsm1.PT_NoOptimize:
return common
}
return testLevelResults{}
},
},
{
name: "Mock another planned level inside scheduler aggress blocks middle",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "05-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "06-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "07-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "08-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "09-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "10-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "11-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-04.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "12-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "13-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "14-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
standard := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm", "02-05.tsm", "03-05.tsm", "04-04.tsm"},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
common := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard:
return standard
case tsm1.PT_SmartOptimize, tsm1.PT_NoOptimize:
return common
}
return testLevelResults{}
},
},
{
name: "Mock another planned level inside scheduler aggress blocks end",
files: []tsm1.ExtFileStat{
{
FileStat: tsm1.FileStat{
Path: "05-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 100,
},
{
FileStat: tsm1.FileStat{
Path: "06-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "07-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "08-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "09-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "10-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "11-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "12-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "13-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "14-01.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 50,
},
{
FileStat: tsm1.FileStat{
Path: "01-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: 200,
},
{
FileStat: tsm1.FileStat{
Path: "02-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultAggressiveMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "03-05.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
{
FileStat: tsm1.FileStat{
Path: "04-04.tsm",
Size: 256 * 1024 * 1024,
},
FirstBlockCount: tsdb.DefaultMaxPointsPerBlock,
},
},
testShardTime: -1,
getResultByPlanType: func(planType tsm1.PlanType) testLevelResults {
standard := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
level5Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"01-05.tsm", "02-05.tsm", "03-05.tsm", "04-04.tsm"},
tsdb.DefaultAggressiveMaxPointsPerBlock,
},
},
}
common := testLevelResults{
level1Groups: []tsm1.PlannedCompactionGroup{
{
tsm1.CompactionGroup{"05-01.tsm", "06-01.tsm", "07-01.tsm", "08-01.tsm", "09-01.tsm", "10-01.tsm", "11-01.tsm", "12-01.tsm"},
tsdb.DefaultMaxPointsPerBlock,
},
},
}
switch planType {
case tsm1.PT_Standard:
return standard
case tsm1.PT_SmartOptimize, tsm1.PT_NoOptimize:
return common
}
return testLevelResults{}
},
},
}
e, err := NewEngine(tsdb.InmemIndexName)
require.NoError(t, err, "create engine")
e.SetEnabled(false)
require.NoError(t, e.Open(), "open engine")
defer func() { require.NoError(t, e.Close(), "close engine") }()
e.Compactor = tsm1.NewCompactor()
defer e.Compactor.Close()
for i := 0; i < 3; i++ {
for _, test := range tests {
var testName string
switch i {
case 0:
testName = test.name + "_PT_Standard"
case 1:
testName = test.name + "_PT_SmartOptimize"
case 2:
testName = test.name + "_PT_NoOptimize"
}
t.Run(testName, func(t *testing.T) {
ffs := newFakeFileStore(withExtFileStats(test.files), withDefaultBlockCount(test.defaultBlockCount))
cp := tsm1.NewDefaultPlanner(ffs, test.testShardTime)
e.MaxPointsPerBlock = tsdb.DefaultMaxPointsPerBlock
e.CompactionPlan = cp
e.Compactor.FileStore = ffs
// Arbitrary group length to use in Scheduler.SetDepth
mockGroupLen := 5
// Set the scheduler depth for our lower level groups.
// During PT_Standard this should still plan a level5 compaction group
// but during PT_SmartOptimize this should not.
e.Scheduler.SetDepth(1, mockGroupLen)
e.Scheduler.SetDepth(2, mockGroupLen)
// Normally this is called within PlanCompactions but because we want to simulate already running
// some compactions we will set them manually here.
atomic.StoreInt64(&e.Stats.TSMCompactionsActive[0], int64(mockGroupLen))
atomic.StoreInt64(&e.Stats.TSMCompactionsActive[1], int64(mockGroupLen))
// Should use PlanType 0 (PT_Standard), 1(PT_SmartOptimize), 2(PT_NoOptimize)
planType := tsm1.PlanType(i)
level1Groups, level2Groups, Level3Groups, Level4Groups, Level5Groups := e.PlanCompactions(planType)
compareLevelGroups(t, test.getResultByPlanType(planType).level1Groups, level1Groups, "unexpected level 1 Group")
compareLevelGroups(t, test.getResultByPlanType(planType).level2Groups, level2Groups, "unexpected level 2 Group")
compareLevelGroups(t, test.getResultByPlanType(planType).level3Groups, Level3Groups, "unexpected level 3 Group")
compareLevelGroups(t, test.getResultByPlanType(planType).level4Groups, Level4Groups, "unexpected level 4 Group")
compareLevelGroups(t, test.getResultByPlanType(planType).level5Groups, Level5Groups, "unexpected level 5 Group")
})
}
}
}
func compareLevelGroups(t *testing.T, exp, got []tsm1.PlannedCompactionGroup, message string) {
require.Lenf(t, got, len(exp), "%s %s", message, " collection length mismatch")
for i, group := range exp {
require.Equal(t, group.Group, got[i].Group, message)
require.Equal(t, group.PointsPerBlock, got[i].PointsPerBlock, message)
}
}
func assertValueEqual(t *testing.T, a, b tsm1.Value) {
if got, exp := a.UnixNano(), b.UnixNano(); got != exp {
t.Fatalf("time mismatch: got %v, exp %v", got, exp)
}
if got, exp := a.Value(), b.Value(); got != exp {
t.Fatalf("value mismatch: got %v, exp %v", got, exp)
}
}
func MustTSMWriter(dir string, gen int) (tsm1.TSMWriter, string) {
f := MustTempFile(dir)
oldName := f.Name()
// Windows can't rename a file while it's open. Close first, rename and
// then re-open
if err := f.Close(); err != nil {
panic(fmt.Sprintf("close temp file: %v", err))
}
newName := filepath.Join(filepath.Dir(oldName), tsm1.DefaultFormatFileName(gen, 1)+".tsm")
if err := os.Rename(oldName, newName); err != nil {
panic(fmt.Sprintf("create tsm file: %v", err))
}
var err error
f, err = os.OpenFile(newName, os.O_RDWR, 0666)
if err != nil {
panic(fmt.Sprintf("open tsm files: %v", err))
}
w, err := tsm1.NewTSMWriter(f)
if err != nil {
panic(fmt.Sprintf("create TSM writer: %v", err))
}
return w, newName
}
func MustWriteTSM(dir string, gen int, values map[string][]tsm1.Value) string {
w, name := MustTSMWriter(dir, gen)
keys := make([]string, 0, len(values))
for k := range values {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
if err := w.Write([]byte(k), values[k]); err != nil {
panic(fmt.Sprintf("write TSM value: %v", err))
}
}
if err := w.WriteIndex(); err != nil {
panic(fmt.Sprintf("write TSM index: %v", err))
}
if err := w.Close(); err != nil {
panic(fmt.Sprintf("write TSM close: %v", err))
}
return name
}
func MustTSMReader(dir string, gen int, values map[string][]tsm1.Value) *tsm1.TSMReader {
return MustOpenTSMReader(MustWriteTSM(dir, gen, values))
}
func MustOpenTSMReader(name string) *tsm1.TSMReader {
f, err := os.Open(name)
if err != nil {
panic(fmt.Sprintf("open file: %v", err))
}
r, err := tsm1.NewTSMReader(f)
if err != nil {
panic(fmt.Sprintf("new reader: %v", err))
}
return r
}
type fakeFileStore struct {
PathsFn func() []tsm1.ExtFileStat
defaultBlockCount int
lastModified time.Time
// fakeFileStore blockCount holds a map of file paths from
// PathsFn.FileStat to a mock block count as an integer.
blockCount map[string]int
readers []*tsm1.TSMReader
}
func (w *fakeFileStore) Stats() []tsm1.ExtFileStat {
return w.PathsFn()
}
func (w *fakeFileStore) NextGeneration() int {
return 1
}
func (w *fakeFileStore) LastModified() time.Time {
return w.lastModified
}
func (w *fakeFileStore) TSMReader(path string) (*tsm1.TSMReader, error) {
r := MustOpenTSMReader(path)
w.readers = append(w.readers, r)
r.Ref()
return r, nil
}
func (w *fakeFileStore) Close() {
for _, r := range w.readers {
r.Close()
}
w.readers = nil
}
func (w *fakeFileStore) ParseFileName(path string) (int, int, error) {
return tsm1.DefaultParseFileName(path)
}
Reference in New Issue View Git Blame Copy Permalink