diff --git a/point/check.go b/point/check.go index 0eeed6d1..178b21ce 100644 --- a/point/check.go +++ b/point/check.go @@ -10,8 +10,12 @@ import ( "math" "reflect" "strings" + "sync" ) +// One pool per power-of-two size from 16 through 2048 keys. +var keyConflictMapPools [8]sync.Pool + type checker struct { *cfg warns []*Warn @@ -75,10 +79,14 @@ func (c *checker) checkKVs(kvs KVs) KVs { kvs = kvs.TrimTags(c.cfg.maxTags) } + // 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 { @@ -99,6 +107,62 @@ func (c *checker) checkKVs(kvs KVs) KVs { return kvs } +// canSkipKeyConflicts keeps duplicate inputs on the legacy scan path because +// in-place compaction makes conflict order observable. +func canSkipKeyConflicts(kvs KVs) bool { + keys, pool := getKeyConflictMap(len(kvs)) + if keys == nil { + return false + } + + unique := true + for _, kv := range kvs { + if _, ok := keys[kv.Key]; ok { + unique = false + break + } + keys[kv.Key] = struct{}{} + } + + 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 { @@ -113,15 +177,19 @@ 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 { +func (c *checker) keyConflict(key string, kvs KVs, skip bool) bool { + if skip { + return false + } + i := 0 for _, kv := range kvs { if kv.Key == key { @@ -139,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", @@ -184,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 } @@ -212,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, @@ -253,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 } diff --git a/point/check_test.go b/point/check_test.go index b5c906dd..ce1ae339 100644 --- a/point/check_test.go +++ b/point/check_test.go @@ -128,6 +128,223 @@ 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, 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) + } + wideSeparatedKeys = append(wideSeparatedKeys, "duplicate") + wideSeparatedWant = append(wideSeparatedWant, "duplicate") + + wideAdjacentKeys := append([]string{"duplicate", "duplicate"}, wideSeparatedKeys[1:KEY_CONFLICT_MAP_THRESHOLD-1]...) + wideAdjacentWant := wideSeparatedWant[:KEY_CONFLICT_MAP_THRESHOLD-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 < 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, 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 `.'"}, + {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 < 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, KEY_CONFLICT_MAP_THRESHOLD-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, KEY_CONFLICT_MAP_THRESHOLD) + for i := 0; i < KEY_CONFLICT_MAP_THRESHOLD; 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 < KEY_CONFLICT_MAP_THRESHOLD; 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, KEY_CONFLICT_MAP_THRESHOLD-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 +886,51 @@ 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) + 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) + } + }) + } + + 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) + } + }) + } +}