fix: prevent differing field types in the same shard (#26025)

pkg/data/gensyncmap/gensyncmap.go

@@ -0,0 +1,47 @@
+package gensyncmap
+
+import "sync"
+
+type Map[K comparable, V any] struct {
+	m sync.Map
+}
+
+func (m *Map[K, V]) Delete(key K) {
+	m.m.Delete(key)
+}
+
+func (m *Map[K, V]) Load(key K) (value V, ok bool) {
+	v, ok := m.m.Load(key)
+	if !ok {
+		return value, ok
+	}
+	return v.(V), ok
+}
+
+func (m *Map[K, V]) LoadAndDelete(key K) (value V, loaded bool) {
+	v, loaded := m.m.LoadAndDelete(key)
+	if !loaded {
+		return value, loaded
+	}
+	return v.(V), loaded
+}
+
+func (m *Map[K, V]) LoadOrStore(key K, value V) (actual V, loaded bool) {
+	a, loaded := m.m.LoadOrStore(key, value)
+	return a.(V), loaded
+}
+
+func (m *Map[K, V]) Range(f func(key K, value V) bool) {
+	m.m.Range(func(key, value any) bool { return f(key.(K), value.(V)) })
+}
+
+func (m *Map[K, V]) Store(key K, value V) { m.m.Store(key, value) }
+
+func (m *Map[K, V]) Len() int {
+	var n int
+	m.m.Range(func(_, _ any) bool {
+		n++
+		return true
+	})
+	return n
+}

tsdb/engine/tsm1/engine.go

@@ -1264,7 +1264,7 @@ func (e *Engine) addToIndexFromKey(keys [][]byte, fieldTypes []influxql.DataType
 		keys[i], field = SeriesAndFieldFromCompositeKey(keys[i])
 		name := models.ParseName(keys[i])
 		mf := e.fieldset.CreateFieldsIfNotExists(name)
-		if err := mf.CreateFieldIfNotExists(field, fieldTypes[i]); err != nil {
+		if _, err := mf.CreateFieldIfNotExists(field, fieldTypes[i]); err != nil {
 			return err
 		}
 

tsdb/field_validator.go

@@ -13,17 +13,19 @@ const MaxFieldValueLength = 1048576
 // ValidateFields will return a PartialWriteError if:
 //   - the point has inconsistent fields, or
 //   - the point has fields that are too long
-func ValidateFields(mf *MeasurementFields, point models.Point, skipSizeValidation bool) error {
+func ValidateFields(mf *MeasurementFields, point models.Point, skipSizeValidation bool) ([]*FieldCreate, error) {
 	pointSize := point.StringSize()
 	iter := point.FieldIterator()
+	var fieldsToCreate []*FieldCreate
+
 	for iter.Next() {
 		if !skipSizeValidation {
 			// Check for size of field too large. Note it is much cheaper to check the whole point size
 			// than checking the StringValue size (StringValue potentially takes an allocation if it must
 			// unescape the string, and must at least parse the string)
 			if pointSize > MaxFieldValueLength && iter.Type() == models.String {
 				if sz := len(iter.StringValue()); sz > MaxFieldValueLength {
-					return PartialWriteError{
+					return nil, PartialWriteError{
 						Reason: fmt.Sprintf(
 							"input field \"%s\" on measurement \"%s\" is too long, %d > %d",
 							iter.FieldKey(), point.Name(), sz, MaxFieldValueLength),
@@ -33,14 +35,9 @@ func ValidateFields(mf *MeasurementFields, point models.Point, skipSizeValidatio
 			}
 		}
 
+		fieldKey := iter.FieldKey()
 		// Skip fields name "time", they are illegal.
-		if bytes.Equal(iter.FieldKey(), timeBytes) {
-			continue
-		}
-
-		// If the fields is not present, there cannot be a conflict.
-		f := mf.FieldBytes(iter.FieldKey())
-		if f == nil {
+		if bytes.Equal(fieldKey, timeBytes) {
 			continue
 		}
 
@@ -49,18 +46,26 @@ func ValidateFields(mf *MeasurementFields, point models.Point, skipSizeValidatio
 			continue
 		}
 
-		// If the types are not the same, there is a conflict.
-		if f.Type != dataType {
-			return PartialWriteError{
+		// If the field is not present, remember to create it.
+		f := mf.FieldBytes(fieldKey)
+		if f == nil {
+			fieldsToCreate = append(fieldsToCreate, &FieldCreate{
+				Measurement: point.Name(),
+				Field: &Field{
+					Name: string(fieldKey),
+					Type: dataType,
+				}})
+		} else if f.Type != dataType {
+			// If the types are not the same, there is a conflict.
+			return nil, PartialWriteError{
 				Reason: fmt.Sprintf(
 					"%s: input field \"%s\" on measurement \"%s\" is type %s, already exists as type %s",
-					ErrFieldTypeConflict, iter.FieldKey(), point.Name(), dataType, f.Type),
+					ErrFieldTypeConflict, fieldKey, point.Name(), dataType, f.Type),
 				Dropped: 1,
 			}
 		}
 	}
-
-	return nil
+	return fieldsToCreate, nil
 }
 
 // dataTypeFromModelsFieldType returns the influxql.DataType that corresponds to the