package idk import ( "encoding/json" "strconv" "strings" "time" pilosa "github.com/featurebasedb/featurebase/v3" "github.com/featurebasedb/featurebase/v3/logger" "github.com/pkg/errors" ) type FieldType string const ( IDType FieldType = "id" BoolType FieldType = "bool" StringType FieldType = "string" LookupTextType FieldType = "lookuptext" IntType FieldType = "int" ForeignKeyType FieldType = "foreignkey" DecimalType FieldType = "decimal" StringArrayType FieldType = "stringarray" IDArrayType FieldType = "idarray" DateIntType FieldType = "dateint" RecordTimeType FieldType = "recordtime" SignedIntBoolKeyType FieldType = "signedintboolkey" IgnoreType FieldType = "ignore" TimestampType FieldType = "timestamp" ) var ( ErrNoFieldSpec = errors.New("no field spec in this header") ErrInvalidFieldName = errors.New("field name must match [a-z][a-z0-9Θ_-]{0,229}") ErrParsingEpoch = "parsing epoch for " ErrDecodingConfig = "decoding config for field " ) // HeaderToField takes a header specification which looks like // sourcename___destname__FieldType_Arg_Arg2 (note that sourcename and // destname are separated by "___", triple underscore) and converts it // to an idk Field like: // // FieldTypeField { // NameVal: sourcename, // DestNameVal: destname, // Thing1: Arg, // Thing2: Arg2, // } // // It does this using a variety of reflective magic. The unwritten // rules are that all idk Fields must be structs and have their first // member be `NameVal string`. The arguments are applied in order to // exported fields. func HeaderToField(headerField string, log logger.Logger) (field Field, _ error) { if log == nil { log = logger.NopLogger } sourceName, destName, fieldspecstr, err := splitHeader(headerField) if err != nil { return nil, err } fieldspec := strings.Split(fieldspecstr, "_") if len(fieldspec) == 0 { return nil, errors.Errorf("no fieldspec-impossible? headerfield: %s, fieldspec: %s", headerField, fieldspec) } fieldType := FieldType(strings.ToLower(fieldspec[0])) if sourceName == "" && fieldType != RecordTimeType { return nil, errors.Errorf("field '%s' has no sourceName", headerField) } switch fieldType { case IDType: field, err = headerToIDField(headerField, sourceName, destName, fieldspec, log) case BoolType: field, err = headerToBoolField(headerField, sourceName, destName, fieldspec, log) case StringType: field, err = headerToStringField(headerField, sourceName, destName, fieldspec, log) case LookupTextType: field, err = headerToLookupTextField(headerField, sourceName, destName, fieldspec, log) case IntType: field, err = headerToIntField(headerField, sourceName, destName, fieldspec, log) case ForeignKeyType: field, err = headerToForeignKeyField(headerField, sourceName, destName, fieldspec, log) case DecimalType: field, err = headerToDecimalField(headerField, sourceName, destName, fieldspec, log) case StringArrayType: field, err = headerToStringArrayField(headerField, sourceName, destName, fieldspec, log) case IDArrayType: field, err = headerToIDArrayField(headerField, sourceName, destName, fieldspec, log) case DateIntType: field, err = headerToDateIntField(headerField, sourceName, destName, fieldspec, log) case TimestampType: field, err = headerToTimestampField(headerField, sourceName, destName, fieldspec, log) case RecordTimeType: field, err = headerToRecordTimeField(headerField, sourceName, destName, fieldspec, log) case SignedIntBoolKeyType: field, err = headerToSignedIntBoolKeyField(headerField, sourceName, destName, fieldspec, log) case IgnoreType: field = IgnoreField{} if len(fieldspec) > 1 { log.Printf("ignoring extra arguments to IgnoreField %s: %v", headerField, fieldspec[1:]) } default: return nil, errors.Errorf("unknown field '%s' for '%s'", fieldspec[0], headerField) } return field, err } func headerToIDField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { idField := IDField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { if fieldspec[1] == "T" { idField.Mutex = true } else if fieldspec[1] != "F" { return nil, errors.Errorf("can't interpret '%s' for IDField.Mutex for field '%s'", fieldspec[1], sourceName) } } if len(fieldspec) > 2 { idField.Quantum = fieldspec[2] } if len(fieldspec) > 3 { idField.TTL = fieldspec[3] } if len(fieldspec) > 4 { log.Printf("ignoring extra arguments to IDField %s: %v", headerField, fieldspec[4:]) } return idField, nil } func headerToBoolField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { field := BoolField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { log.Printf("ignoring extra arguments to BoolField %s: %v", headerField, fieldspec[1:]) } return field, nil } func headerToStringField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { strField := StringField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { if fieldspec[1] == "T" { strField.Mutex = true } else if fieldspec[1] != "F" { return nil, errors.Errorf("can't interpret '%s' for StringField.Mutex for field '%s'", fieldspec[1], sourceName) } } if len(fieldspec) > 2 { strField.Quantum = fieldspec[2] } if len(fieldspec) > 3 { strField.TTL = fieldspec[3] } if len(fieldspec) > 4 { log.Printf("ignoring extra arguments to StringField %s: %v", headerField, fieldspec[4:]) } return strField, nil } func headerToLookupTextField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { lTextField := LookupTextField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { log.Printf("ignoring extra arguments to LookupTextField %s: %v", headerField, fieldspec[1:]) } return lTextField, nil } func headerToIntField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { intField := IntField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { min, err := strconv.ParseInt(fieldspec[1], 10, 64) if err != nil { return nil, errors.Wrapf(err, "parsing min for %s", sourceName) } intField.Min = &min } if len(fieldspec) > 2 { max, err := strconv.ParseInt(fieldspec[2], 10, 64) if err != nil { return nil, errors.Wrapf(err, "parsing max for %s", sourceName) } intField.Max = &max } if len(fieldspec) > 3 { intField.ForeignIndex = fieldspec[3] } if len(fieldspec) > 4 { log.Printf("ignoring extra arguments to IntField %s: %v", headerField, fieldspec[4:]) } return intField, nil } func headerToForeignKeyField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { fkField := IntField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { fkField.ForeignIndex = fieldspec[1] } else { return nil, errors.Errorf("need foreign index for foreign key field: %s", headerField) } if len(fieldspec) > 2 { log.Printf("ignoring extra arguments to ForeignKey Field %s: %v", headerField, fieldspec[2:]) } return fkField, nil } func headerToDecimalField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { decField := DecimalField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { scale, err := strconv.ParseInt(fieldspec[1], 10, 64) if err != nil { return nil, errors.Wrapf(err, "parsing scale for %s", sourceName) } decField.Scale = scale } if len(fieldspec) > 2 { log.Printf("ignoring extra arguments to DecimalField %s: %v", headerField, fieldspec[2:]) } return decField, nil } func headerToStringArrayField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { strArrField := StringArrayField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { strArrField.Quantum = fieldspec[1] } if len(fieldspec) > 2 { strArrField.TTL = fieldspec[2] } if len(fieldspec) > 3 { log.Printf("ignoring extra arguments to StringArrayField %s: %v", headerField, fieldspec[3:]) } return strArrField, nil } func headerToIDArrayField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { idArrField := IDArrayField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { idArrField.Quantum = fieldspec[1] } if len(fieldspec) > 2 { idArrField.TTL = fieldspec[2] } if len(fieldspec) > 3 { log.Printf("ignoring extra arguments to IDArrayField %s: %v", headerField, fieldspec[3:]) } return idArrField, nil } func headerToDateIntField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { dateField := DateIntField{ NameVal: sourceName, DestNameVal: destName, } layout := time.RFC3339 if len(fieldspec) > 1 { layout = fieldspec[1] } dateField.Layout = layout if len(fieldspec) > 2 { epoch, err := time.Parse(layout, fieldspec[2]) if err != nil { return nil, errors.Wrapf(err, ErrParsingEpoch, headerField) } dateField.Epoch = epoch } if len(fieldspec) > 3 { dateField.Unit = Unit(fieldspec[3]).unit() if len(fieldspec) > 4 && (dateField.Unit.IsCustom()) { if _, err := time.ParseDuration(fieldspec[4]); err != nil { return nil, errors.Wrapf(err, "parsing custom unit %s", fieldspec[4]) } dateField.CustomUnit = fieldspec[4] } else { if _, err := dateField.Unit.Duration(); err != nil { return nil, err } } } if len(fieldspec) > 5 { log.Printf("ignoring extra arguments to DateIntField %s: %v", headerField, fieldspec[5:]) } return dateField, nil } func headerToTimestampField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { tsField := TimestampField{ NameVal: sourceName, DestNameVal: destName, } granularity := "s" layout := time.RFC3339Nano if len(fieldspec) > 1 { granularity = fieldspec[1] } tsField.Granularity = granularity if len(fieldspec) > 2 { layout = fieldspec[2] } tsField.Layout = layout if len(fieldspec) > 3 { epoch, err := time.Parse(layout, fieldspec[3]) if err != nil { return nil, errors.Wrapf(err, ErrParsingEpoch, headerField) } if epoch.IsZero() { tsField.Epoch = time.Unix(0, 0) } else { tsField.Epoch = epoch } } if len(fieldspec) > 4 { unit := Unit(fieldspec[4]).unit() if _, err := unit.Duration(); err != nil { return nil, errors.Wrapf(err, "invalid unit for TimestampField %s", headerField) } tsField.Unit = unit } if len(fieldspec) > 5 { log.Printf("ignoring extra arguments to TimestampField %s: %v", headerField, fieldspec[5:]) } return tsField, nil } func headerToRecordTimeField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { rtField := RecordTimeField{ NameVal: sourceName, DestNameVal: destName, } // We used to use a default of "" here, and then // (RecordTimeType).layout() would treat that as RFC3339. // This is now more parallel to the handling for DateIntType. layout := time.RFC3339 if len(fieldspec) > 1 { layout = fieldspec[1] } rtField.Layout = layout if len(fieldspec) > 2 { epoch, err := time.Parse(layout, fieldspec[2]) if err != nil { return nil, errors.Wrapf(err, ErrParsingEpoch, headerField) } rtField.Epoch = epoch } if len(fieldspec) > 3 { unit := Unit(fieldspec[3]).unit() _, err := unit.Duration() if err != nil { return nil, err } rtField.Unit = unit } if len(fieldspec) > 4 { log.Printf("ignoring extra arguments to RecordTimeField %s: %v", headerField, fieldspec[4:]) } return rtField, nil } func headerToSignedIntBoolKeyField(headerField string, sourceName string, destName string, fieldspec []string, log logger.Logger) (Field, error) { field := SignedIntBoolKeyField{ NameVal: sourceName, DestNameVal: destName, } if len(fieldspec) > 1 { log.Printf("ignoring extra arguments to SignedIntBoolKeyField %s: %v", headerField, fieldspec[1:]) } return field, nil } var nameSpecDelimiter = "___" // Triple Underscore // splitHeader splits a header string. // Allowed formats: // - SourceName__FieldType_Arg_Arg2 // - SourceName___DestName__FieldType_Arg_Arg2 // where // - SourceName may be an arbitrary string // - DestName may be specified as a valid pilosa fieldname // - DestName MUST be specified if SourceName is NOT a valid pilosa field name // - DestName MUST NOT include any triple-underscores // - FieldType MUST be specified // - Args may or may not be required, depending on the FieldType func splitHeader(s string) (sourceName, destName, typeSpec string, err error) { // One dunder is required, to separate name from fieldspec. // One trunder is optional, to separate sourceName from destName. var idx2 int idx3 := strings.LastIndex(s, nameSpecDelimiter) if idx3 == -1 { // simple namespec: SourceName__FieldType_Arg_Arg2 s := s idx2 = strings.LastIndex(s, "__") if idx2 == -1 { return "", "", "", ErrNoFieldSpec } destName = s[:idx2] sourceName = destName typeSpec = s[idx2+2:] } else { // DestName namespec: SourceName___DestName__FieldType_Arg_Arg2 sourceName = s[:idx3] rest := s[idx3+len(nameSpecDelimiter):] idx2 = strings.LastIndex(rest, "__") if idx2 == -1 { return "", "", "", ErrNoFieldSpec } destName = rest[:idx2] typeSpec = rest[idx2+2:] } err = pilosa.ValidateName(destName) if destName != "" && err != nil { // "" is valid for RecordTimeField and IgnoreField return "", "", "", err } return sourceName, destName, typeSpec, nil } type PathTable [][]string type DeleteSentinel int const ( DELETE_SENTINEL DeleteSentinel = 1 ) func (t PathTable) Lookup(root interface{}, allowMissingFields bool) ([]interface{}, error) { data := make([]interface{}, len(t)) for i, path := range t { node := root for j, branch := range path { obj, ok := node.(map[string]interface{}) if !ok { return nil, errors.Errorf("failed to lookup path %v: element at %v is not an object", path, path[:j]) } next, ok := obj[branch] if !ok { if allowMissingFields { node = nil break } else { return nil, errors.Errorf("failed to lookup path %v: element %q at %v is missing", path, branch, path[:j]) } } else { if next == nil { node = DELETE_SENTINEL break } node = next } } data[i] = node } return data, nil } func (t PathTable) FlatMap() map[string]int { m, i := make(map[string]int), 0 for _, arr := range t { for _, str := range arr { if _, ok := m[str]; !ok { m[str] = i } i++ } } return m } // RawField is used in cases where header fields are configured as json, // typically read from a file. But this type is also used by the kafka runner in // fbsql. type RawField struct { Name string `json:"name"` Path []string `json:"path"` Type string `json:"type"` Config json.RawMessage } func ParseHeader(raw []byte) ([]Field, PathTable, error) { var rawSchema []RawField err := json.Unmarshal(raw, &rawSchema) if err != nil { return nil, nil, errors.Wrap(err, "parsing schema") } fields, paths := make([]Field, len(rawSchema)), make(PathTable, len(rawSchema)) fieldNameSet := make(map[string]int) for i, s := range rawSchema { if s.Path == nil { return nil, nil, errors.Errorf("field %q is missing a path", s.Name) } paths[i] = s.Path switch s.Type { case "id": var field IDField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "ids": var field IDArrayField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "string": var field StringField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "strings": var field StringArrayField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "bool": var field BoolField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "recordTime": var field RecordTimeField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "int": var field IntField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "decimal": var field DecimalField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "signedIntBoolKey": var field SignedIntBoolKeyField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "dateInt": var field DateIntField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "timestamp": var field TimestampField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field case "lookupText": var field LookupTextField if s.Config != nil { err := json.Unmarshal(s.Config, &field) if err != nil { return nil, nil, errors.Wrapf(err, ErrDecodingConfig, s.Name) } } field.NameVal = s.Name fields[i] = field default: return nil, nil, errors.Errorf("failed to initialize schema for field %q: unsupported type %q", s.Name, s.Type) } fieldName := fields[i].Name() if prev, ok := fieldNameSet[fieldName]; ok { return nil, nil, errors.Errorf("schema field %d duplicates name of field %d (%s)", i, prev, fieldName) } fieldNameSet[fieldName] = i } return fields, paths, nil }