From 3878eba8b5b2e6b8c838e5d823940a6f0b4682f1 Mon Sep 17 00:00:00 2001 From: leecha <22087646+leecha@users.noreply.github.com> Date: Thu, 23 Jul 2026 10:39:02 +0800 Subject: [PATCH 1/2] perf: speed up point key checks --- point/check.go | 53 +++++++++- point/check_test.go | 235 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 287 insertions(+), 1 deletion(-) diff --git a/point/check.go b/point/check.go index 0eeed6d1..1a9019a7 100644 --- a/point/check.go +++ b/point/check.go @@ -10,15 +10,27 @@ import ( "math" "reflect" "strings" + "sync" ) +const ( + // A scratch map is faster than repeatedly scanning KVs once a point is this wide. + keyConflictMapThreshold = 16 + // Bound scratch-map retention when callers disable the default field and tag limits. + keyConflictMapMaxRetainedKeys = 2048 +) + +var keyConflictMapPool sync.Pool + type checker struct { *cfg - warns []*Warn + warns []*Warn + skipKeyConflicts bool } func (c *checker) reset() { c.warns = c.warns[:0] + c.skipKeyConflicts = false } func (c *checker) check(pt *Point) *Point { @@ -75,6 +87,8 @@ func (c *checker) checkKVs(kvs KVs) KVs { kvs = kvs.TrimTags(c.cfg.maxTags) } + c.skipKeyConflicts = canSkipKeyConflicts(kvs) + // check each kv valid idx := 0 for _, kv := range kvs { @@ -84,6 +98,7 @@ func (c *checker) checkKVs(kvs KVs) KVs { } else if defaultPTPool != nil { // When point-pool enabled, on drop f, we should put-back to pool. defaultPTPool.PutKV(x) + c.skipKeyConflicts = false } } @@ -99,6 +114,37 @@ func (c *checker) checkKVs(kvs KVs) KVs { return kvs } +// canSkipKeyConflicts keeps duplicate inputs on the legacy scan path because +// in-place compaction and PointPool mutations make conflict order observable. +func canSkipKeyConflicts(kvs KVs) bool { + if len(kvs) < keyConflictMapThreshold { + return false + } + + var keys map[string]struct{} + if pooled := keyConflictMapPool.Get(); pooled == nil { + keys = make(map[string]struct{}, min(len(kvs), keyConflictMapMaxRetainedKeys)) + } else { + keys = pooled.(map[string]struct{}) + } + + unique := true + for _, kv := range kvs { + if _, ok := keys[kv.Key]; ok { + unique = false + break + } + keys[kv.Key] = struct{}{} + } + + clear(keys) + if len(kvs) <= keyConflictMapMaxRetainedKeys { + keyConflictMapPool.Put(keys) + } + + return unique +} + // Remove all `\` suffix on key/val // Replace all `\n` with ` `. func adjustKV(x string) string { @@ -122,6 +168,10 @@ func (c *checker) checkKV(f *Field, kvs KVs) (*Field, bool) { } func (c *checker) keyConflict(key string, kvs KVs) bool { + if c.skipKeyConflicts { + return false + } + i := 0 for _, kv := range kvs { if kv.Key == key { @@ -271,6 +321,7 @@ func (c *checker) checkField(f *Field, kvs KVs) (*Field, bool) { // `abc,tag=1 f1=32i` if defaultPTPool != nil { defaultPTPool.PutKV(f) + c.skipKeyConflicts = false f = defaultPTPool.GetKV(f.Key, int64(x.U)) } else { f.Val = &Field_I{I: int64(x.U)} diff --git a/point/check_test.go b/point/check_test.go index b5c906dd..e9c46855 100644 --- a/point/check_test.go +++ b/point/check_test.go @@ -128,6 +128,221 @@ func TestCheckPoints(t *T.T) { }) } +func TestCheckPointsDuplicateKeys(t *T.T) { + previousPool := GetPointPool() + ClearPointPool() + t.Cleanup(func() { SetPointPool(previousPool) }) + + wideSeparatedKeys := []string{"duplicate"} + wideSeparatedWant := make([]string, 0, keyConflictMapThreshold-1) + for i := 0; i < keyConflictMapThreshold-2; i++ { + key := fmt.Sprintf("keep_%02d", i) + wideSeparatedKeys = append(wideSeparatedKeys, key) + wideSeparatedWant = append(wideSeparatedWant, key) + } + wideSeparatedKeys = append(wideSeparatedKeys, "duplicate") + wideSeparatedWant = append(wideSeparatedWant, "duplicate") + + wideAdjacentKeys := append([]string{"duplicate", "duplicate"}, wideSeparatedKeys[1:keyConflictMapThreshold-1]...) + wideAdjacentWant := wideSeparatedWant[:keyConflictMapThreshold-2] + + tests := []struct { + name string + keys []string + wantKeys []string + wantWarns int + withPool bool + }{ + { + name: "adjacent", + keys: []string{"duplicate", "duplicate", "keep"}, + wantKeys: []string{"keep"}, + wantWarns: 2, + }, + { + name: "separated", + keys: []string{"duplicate", "keep", "duplicate"}, + wantKeys: []string{"keep", "duplicate"}, + wantWarns: 1, + }, + { + name: "adjacent-with-point-pool", + keys: []string{"duplicate", "duplicate", "keep"}, + wantKeys: []string{"duplicate", "keep"}, + wantWarns: 1, + withPool: true, + }, + { + name: "wide-separated", + keys: wideSeparatedKeys, + wantKeys: wideSeparatedWant, + wantWarns: 1, + }, + { + name: "wide-adjacent-with-point-pool", + keys: wideAdjacentKeys, + wantKeys: append([]string{"duplicate"}, wideAdjacentWant...), + wantWarns: 1, + withPool: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *T.T) { + if tc.withPool { + SetPointPool(NewReservedCapPointPool(16)) + t.Cleanup(ClearPointPool) + } + + kvs := make(KVs, 0, len(tc.keys)) + for i, key := range tc.keys { + kvs = append(kvs, NewKV(key, i)) + } + + pt := NewPoint("measurement", kvs, WithPrecheck(false), WithKeySorted(false)) + pts := CheckPoints([]*Point{pt}) + require.Len(t, pts, 1) + + gotKeys := make([]string, 0, len(pts[0].pt.Fields)) + for _, kv := range pts[0].pt.Fields { + gotKeys = append(gotKeys, kv.Key) + } + assert.Equal(t, tc.wantKeys, gotKeys) + require.Len(t, pts[0].Warns(), tc.wantWarns) + for _, warn := range pts[0].Warns() { + assert.Equal(t, &Warn{Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, warn) + } + }) + } + + t.Run("tag-and-field", func(t *T.T) { + kvs := KVs{ + NewKV("duplicate", "tag", WithKVTagSet(true)), + NewKV("duplicate", 1), + NewKV("keep", 2), + } + + pt := NewPoint("measurement", kvs, WithPrecheck(false), WithKeySorted(false)) + pts := CheckPoints([]*Point{pt}) + require.Len(t, pts, 1) + require.Len(t, pts[0].pt.Fields, 1) + assert.Equal(t, "keep", pts[0].pt.Fields[0].Key) + assert.Equal(t, []*Warn{ + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, + }, pts[0].Warns()) + }) + + t.Run("new-point", func(t *T.T) { + pt := NewPoint("measurement", KVs{ + NewKV("duplicate", 1), + NewKV("duplicate", 2), + NewKV("keep", 3), + }, WithKeySorted(false)) + + require.Len(t, pt.pt.Fields, 1) + assert.Equal(t, "keep", pt.pt.Fields[0].Key) + assert.Equal(t, []*Warn{ + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, + }, pt.Warns()) + }) + + t.Run("adjusted-key-conflict", func(t *T.T) { + pt := NewPoint("measurement", KVs{ + NewKV("field.1", 1), + NewKV("field_1", 2), + }, WithDotInKey(false), WithKeySorted(false)) + + require.Len(t, pt.pt.Fields, 1) + assert.Equal(t, "field_1", pt.pt.Fields[0].Key) + assert.Equal(t, []*Warn{ + {Type: WarnDotInkey, Msg: "invalid field key `field.1': found `.'"}, + {Type: WarnInvalidTagKey, Msg: `field key "field_1" exist after adjust`}, + }, pt.Warns()) + }) + + t.Run("wide-adjusted-key-conflict", func(t *T.T) { + kvs := KVs{NewKV("field.1", 1), NewKV("field_1", 2)} + for i := 0; i < keyConflictMapThreshold-2; i++ { + kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) + } + + pt := NewPoint("measurement", kvs, WithDotInKey(false), WithKeySorted(false)) + + require.Len(t, pt.pt.Fields, keyConflictMapThreshold-1) + assert.Equal(t, "field_1", pt.pt.Fields[0].Key) + assert.Equal(t, []*Warn{ + {Type: WarnDotInkey, Msg: "invalid field key `field.1': found `.'"}, + {Type: WarnInvalidTagKey, Msg: `field key "field_1" exist after adjust`}, + }, pt.Warns()) + }) + + t.Run("wide-aliased-field", func(t *T.T) { + shared := NewKV("duplicate.", 1) + kvs := KVs{shared, shared} + for i := 0; i < keyConflictMapThreshold-2; i++ { + kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) + } + + pt := NewPoint("measurement", kvs, WithDotInKey(false), WithKeySorted(false)) + + require.Len(t, pt.pt.Fields, keyConflictMapThreshold-2) + assert.Equal(t, []*Warn{ + {Type: WarnDotInkey, Msg: "invalid field key `duplicate.': found `.'"}, + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate_")`}, + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate_")`}, + }, pt.Warns()) + }) + + t.Run("wide-point-before-narrow-duplicate", func(t *T.T) { + wideKVs := make(KVs, 0, keyConflictMapThreshold) + for i := 0; i < keyConflictMapThreshold; i++ { + wideKVs = append(wideKVs, NewKV(fmt.Sprintf("wide_%02d", i), i)) + } + + pts := CheckPoints([]*Point{ + NewPoint("wide", wideKVs, WithPrecheck(false), WithKeySorted(false)), + NewPoint("duplicate", KVs{ + NewKV("duplicate", 1), + NewKV("keep", 2), + NewKV("duplicate", 3), + }, WithPrecheck(false), WithKeySorted(false)), + }) + + require.Len(t, pts, 2) + assert.Empty(t, pts[0].Warns()) + require.Len(t, pts[1].pt.Fields, 2) + assert.Equal(t, "keep", pts[1].pt.Fields[0].Key) + assert.Equal(t, "duplicate", pts[1].pt.Fields[1].Key) + assert.Equal(t, []*Warn{ + {Type: WarnKeyNameConflict, Msg: `same key ("duplicate")`}, + }, pts[1].Warns()) + }) + + t.Run("wide-unique-with-point-pool-drop", func(t *T.T) { + SetPointPool(NewReservedCapPointPool(16)) + t.Cleanup(ClearPointPool) + + kvs := KVs{NewKV("drop", "value")} + for i := 1; i < keyConflictMapThreshold; i++ { + kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) + } + + pt := NewPoint("measurement", kvs, WithPrecheck(false), WithKeySorted(false)) + pts := CheckPoints([]*Point{pt}, WithStrField(false)) + + require.Len(t, pts, 1) + require.Len(t, pts[0].pt.Fields, keyConflictMapThreshold-1) + assert.Equal(t, []*Warn{ + { + Type: WarnInvalidFieldValueType, + Msg: "field(drop) dropped with string value, when [DisableStringField] enabled", + }, + }, pts[0].Warns()) + }) +} + func TestCheckTags(t *T.T) { cases := []struct { name string @@ -669,3 +884,23 @@ func BenchmarkCheckPoints(b *T.B) { } }) } + +func BenchmarkCheckPointsWide(b *T.B) { + for _, fields := range []int{16, 32, 64, 256, 1024} { + b.Run(fmt.Sprintf("%d-fields", fields), func(b *T.B) { + kvs := make(KVs, fields) + for i := range kvs { + kvs[i] = NewKV(fmt.Sprintf("field_%04d", i), i) + } + + pt := NewPoint("measurement", kvs, WithPrecheck(false), WithKeySorted(false)) + pts := []*Point{pt} + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + CheckPoints(pts) + } + }) + } +} From b8b2dd5161b0820b52e8f37c327544c13e85cac1 Mon Sep 17 00:00:00 2001 From: leecha <22087646+leecha@users.noreply.github.com> Date: Thu, 23 Jul 2026 16:13:41 +0800 Subject: [PATCH 2/2] fix(point): isolate key conflict map pools --- point/check.go | 89 +++++++++++++++++++++++++++------------------ point/check_test.go | 54 +++++++++++++++++++++------ 2 files changed, 95 insertions(+), 48 deletions(-) diff --git a/point/check.go b/point/check.go index 1a9019a7..178b21ce 100644 --- a/point/check.go +++ b/point/check.go @@ -13,24 +13,16 @@ import ( "sync" ) -const ( - // A scratch map is faster than repeatedly scanning KVs once a point is this wide. - keyConflictMapThreshold = 16 - // Bound scratch-map retention when callers disable the default field and tag limits. - keyConflictMapMaxRetainedKeys = 2048 -) - -var keyConflictMapPool sync.Pool +// One pool per power-of-two size from 16 through 2048 keys. +var keyConflictMapPools [8]sync.Pool type checker struct { *cfg - warns []*Warn - skipKeyConflicts bool + warns []*Warn } func (c *checker) reset() { c.warns = c.warns[:0] - c.skipKeyConflicts = false } func (c *checker) check(pt *Point) *Point { @@ -87,18 +79,19 @@ func (c *checker) checkKVs(kvs KVs) KVs { kvs = kvs.TrimTags(c.cfg.maxTags) } - c.skipKeyConflicts = canSkipKeyConflicts(kvs) + // PointPool operations may mutate fields still referenced by kvs, so retain + // the legacy conflict scans when pooling is enabled. + skipKeyConflicts := defaultPTPool == nil && canSkipKeyConflicts(kvs) // check each kv valid idx := 0 for _, kv := range kvs { - if x, ok := c.checkKV(kv, kvs); ok { + if x, ok := c.checkKV(kv, kvs, skipKeyConflicts); ok { kvs[idx] = x idx++ } else if defaultPTPool != nil { // When point-pool enabled, on drop f, we should put-back to pool. defaultPTPool.PutKV(x) - c.skipKeyConflicts = false } } @@ -115,19 +108,13 @@ func (c *checker) checkKVs(kvs KVs) KVs { } // canSkipKeyConflicts keeps duplicate inputs on the legacy scan path because -// in-place compaction and PointPool mutations make conflict order observable. +// in-place compaction makes conflict order observable. func canSkipKeyConflicts(kvs KVs) bool { - if len(kvs) < keyConflictMapThreshold { + keys, pool := getKeyConflictMap(len(kvs)) + if keys == nil { return false } - var keys map[string]struct{} - if pooled := keyConflictMapPool.Get(); pooled == nil { - keys = make(map[string]struct{}, min(len(kvs), keyConflictMapMaxRetainedKeys)) - } else { - keys = pooled.(map[string]struct{}) - } - unique := true for _, kv := range kvs { if _, ok := keys[kv.Key]; ok { @@ -137,14 +124,45 @@ func canSkipKeyConflicts(kvs KVs) bool { keys[kv.Key] = struct{}{} } - clear(keys) - if len(kvs) <= keyConflictMapMaxRetainedKeys { - keyConflictMapPool.Put(keys) + if pool != nil { + clear(keys) + pool.Put(keys) } return unique } +func getKeyConflictMap(size int) (map[string]struct{}, *sync.Pool) { + const ( + // A scratch map is faster than repeatedly scanning KVs once a point is this wide. + KEY_CONFLICT_MAP_THRESHOLD = 16 + // Bound scratch-map retention when callers disable the default field and tag limits. + KEY_CONFLICT_MAP_MAX_RETAINED_KEYS = 2048 + ) + + if size < KEY_CONFLICT_MAP_THRESHOLD { + return nil, nil + } + + if size > KEY_CONFLICT_MAP_MAX_RETAINED_KEYS { + return make(map[string]struct{}, KEY_CONFLICT_MAP_MAX_RETAINED_KEYS), nil + } + + poolIndex := 0 + poolCapacity := KEY_CONFLICT_MAP_THRESHOLD + for poolCapacity < size { + poolIndex++ + poolCapacity *= 2 + } + + pool := &keyConflictMapPools[poolIndex] + if pooled := pool.Get(); pooled != nil { + return pooled.(map[string]struct{}), pool + } + + return make(map[string]struct{}, poolCapacity), pool +} + // Remove all `\` suffix on key/val // Replace all `\n` with ` `. func adjustKV(x string) string { @@ -159,16 +177,16 @@ func adjustKV(x string) string { return x } -func (c *checker) checkKV(f *Field, kvs KVs) (*Field, bool) { +func (c *checker) checkKV(f *Field, kvs KVs, skipKeyConflicts bool) (*Field, bool) { if f.IsTag { - return c.checkTag(f, kvs) + return c.checkTag(f, kvs, skipKeyConflicts) } else { - return c.checkField(f, kvs) + return c.checkField(f, kvs, skipKeyConflicts) } } -func (c *checker) keyConflict(key string, kvs KVs) bool { - if c.skipKeyConflicts { +func (c *checker) keyConflict(key string, kvs KVs, skip bool) bool { + if skip { return false } @@ -189,7 +207,7 @@ func (c *checker) keyConflict(key string, kvs KVs) bool { // checkTag try to auto modify the f. If we need to drop // f, we return false. -func (c *checker) checkTag(f *Field, kvs KVs) (*Field, bool) { +func (c *checker) checkTag(f *Field, kvs KVs, skipKeyConflicts bool) (*Field, bool) { if c.cfg.maxTagKeyLen > 0 && len(f.Key) > c.cfg.maxTagKeyLen { c.addWarn(WarnMaxTagKeyLen, fmt.Sprintf("exceed max tag key length(%d), got %d, key truncated", @@ -234,7 +252,7 @@ func (c *checker) checkTag(f *Field, kvs KVs) (*Field, bool) { return f, false } - if c.keyConflict(f.Key, kvs) { + if c.keyConflict(f.Key, kvs, skipKeyConflicts) { return f, false } @@ -262,7 +280,7 @@ func (c *checker) checkTag(f *Field, kvs KVs) (*Field, bool) { // checkField try to auto modify the f. If we need to drop // f, we return false. -func (c *checker) checkField(f *Field, kvs KVs) (*Field, bool) { +func (c *checker) checkField(f *Field, kvs KVs, skipKeyConflicts bool) (*Field, bool) { // trim key if c.cfg.maxFieldKeyLen > 0 && len(f.Key) > c.cfg.maxFieldKeyLen { c.addWarn(WarnMaxFieldKeyLen, @@ -303,7 +321,7 @@ func (c *checker) checkField(f *Field, kvs KVs) (*Field, bool) { return f, false } - if c.keyConflict(f.Key, kvs) { + if c.keyConflict(f.Key, kvs, skipKeyConflicts) { return f, false } @@ -321,7 +339,6 @@ func (c *checker) checkField(f *Field, kvs KVs) (*Field, bool) { // `abc,tag=1 f1=32i` if defaultPTPool != nil { defaultPTPool.PutKV(f) - c.skipKeyConflicts = false f = defaultPTPool.GetKV(f.Key, int64(x.U)) } else { f.Val = &Field_I{I: int64(x.U)} diff --git a/point/check_test.go b/point/check_test.go index e9c46855..ce1ae339 100644 --- a/point/check_test.go +++ b/point/check_test.go @@ -129,13 +129,15 @@ func TestCheckPoints(t *T.T) { } func TestCheckPointsDuplicateKeys(t *T.T) { + const KEY_CONFLICT_MAP_THRESHOLD = 16 + previousPool := GetPointPool() ClearPointPool() t.Cleanup(func() { SetPointPool(previousPool) }) wideSeparatedKeys := []string{"duplicate"} - wideSeparatedWant := make([]string, 0, keyConflictMapThreshold-1) - for i := 0; i < keyConflictMapThreshold-2; i++ { + wideSeparatedWant := make([]string, 0, KEY_CONFLICT_MAP_THRESHOLD-1) + for i := 0; i < KEY_CONFLICT_MAP_THRESHOLD-2; i++ { key := fmt.Sprintf("keep_%02d", i) wideSeparatedKeys = append(wideSeparatedKeys, key) wideSeparatedWant = append(wideSeparatedWant, key) @@ -143,8 +145,8 @@ func TestCheckPointsDuplicateKeys(t *T.T) { wideSeparatedKeys = append(wideSeparatedKeys, "duplicate") wideSeparatedWant = append(wideSeparatedWant, "duplicate") - wideAdjacentKeys := append([]string{"duplicate", "duplicate"}, wideSeparatedKeys[1:keyConflictMapThreshold-1]...) - wideAdjacentWant := wideSeparatedWant[:keyConflictMapThreshold-2] + wideAdjacentKeys := append([]string{"duplicate", "duplicate"}, wideSeparatedKeys[1:KEY_CONFLICT_MAP_THRESHOLD-1]...) + wideAdjacentWant := wideSeparatedWant[:KEY_CONFLICT_MAP_THRESHOLD-2] tests := []struct { name string @@ -264,13 +266,13 @@ func TestCheckPointsDuplicateKeys(t *T.T) { t.Run("wide-adjusted-key-conflict", func(t *T.T) { kvs := KVs{NewKV("field.1", 1), NewKV("field_1", 2)} - for i := 0; i < keyConflictMapThreshold-2; i++ { + for i := 0; i < KEY_CONFLICT_MAP_THRESHOLD-2; i++ { kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) } pt := NewPoint("measurement", kvs, WithDotInKey(false), WithKeySorted(false)) - require.Len(t, pt.pt.Fields, keyConflictMapThreshold-1) + require.Len(t, pt.pt.Fields, KEY_CONFLICT_MAP_THRESHOLD-1) assert.Equal(t, "field_1", pt.pt.Fields[0].Key) assert.Equal(t, []*Warn{ {Type: WarnDotInkey, Msg: "invalid field key `field.1': found `.'"}, @@ -281,13 +283,13 @@ func TestCheckPointsDuplicateKeys(t *T.T) { t.Run("wide-aliased-field", func(t *T.T) { shared := NewKV("duplicate.", 1) kvs := KVs{shared, shared} - for i := 0; i < keyConflictMapThreshold-2; i++ { + for i := 0; i < KEY_CONFLICT_MAP_THRESHOLD-2; i++ { kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) } pt := NewPoint("measurement", kvs, WithDotInKey(false), WithKeySorted(false)) - require.Len(t, pt.pt.Fields, keyConflictMapThreshold-2) + require.Len(t, pt.pt.Fields, KEY_CONFLICT_MAP_THRESHOLD-2) assert.Equal(t, []*Warn{ {Type: WarnDotInkey, Msg: "invalid field key `duplicate.': found `.'"}, {Type: WarnKeyNameConflict, Msg: `same key ("duplicate_")`}, @@ -296,8 +298,8 @@ func TestCheckPointsDuplicateKeys(t *T.T) { }) t.Run("wide-point-before-narrow-duplicate", func(t *T.T) { - wideKVs := make(KVs, 0, keyConflictMapThreshold) - for i := 0; i < keyConflictMapThreshold; i++ { + wideKVs := make(KVs, 0, KEY_CONFLICT_MAP_THRESHOLD) + for i := 0; i < KEY_CONFLICT_MAP_THRESHOLD; i++ { wideKVs = append(wideKVs, NewKV(fmt.Sprintf("wide_%02d", i), i)) } @@ -325,7 +327,7 @@ func TestCheckPointsDuplicateKeys(t *T.T) { t.Cleanup(ClearPointPool) kvs := KVs{NewKV("drop", "value")} - for i := 1; i < keyConflictMapThreshold; i++ { + for i := 1; i < KEY_CONFLICT_MAP_THRESHOLD; i++ { kvs = append(kvs, NewKV(fmt.Sprintf("keep_%02d", i), i)) } @@ -333,7 +335,7 @@ func TestCheckPointsDuplicateKeys(t *T.T) { pts := CheckPoints([]*Point{pt}, WithStrField(false)) require.Len(t, pts, 1) - require.Len(t, pts[0].pt.Fields, keyConflictMapThreshold-1) + require.Len(t, pts[0].pt.Fields, KEY_CONFLICT_MAP_THRESHOLD-1) assert.Equal(t, []*Warn{ { Type: WarnInvalidFieldValueType, @@ -886,6 +888,8 @@ func BenchmarkCheckPoints(b *T.B) { } func BenchmarkCheckPointsWide(b *T.B) { + const KEY_CONFLICT_MAP_THRESHOLD = 16 + for _, fields := range []int{16, 32, 64, 256, 1024} { b.Run(fmt.Sprintf("%d-fields", fields), func(b *T.B) { kvs := make(KVs, fields) @@ -903,4 +907,30 @@ func BenchmarkCheckPointsWide(b *T.B) { } }) } + + for _, fields := range []int{1024, 1280, 2048} { + b.Run(fmt.Sprintf("threshold-after-%d-fields", fields), func(b *T.B) { + wideKVs := make(KVs, fields) + for i := range wideKVs { + wideKVs[i] = NewKV(fmt.Sprintf("wide_%04d", i), i) + } + CheckPoints([]*Point{ + NewPoint("wide", wideKVs, WithPrecheck(false), WithKeySorted(false)), + }, WithMaxFields(0)) + + kvs := make(KVs, KEY_CONFLICT_MAP_THRESHOLD) + for i := range kvs { + kvs[i] = NewKV(fmt.Sprintf("field_%04d", i), i) + } + pts := []*Point{ + NewPoint("measurement", kvs, WithPrecheck(false), WithKeySorted(false)), + } + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + CheckPoints(pts) + } + }) + } }