featurebase/idk/header.go
souhailanoor 2bdf1f5081
FB-1597: unifying idk and featurebase (#2160)
* unifying idk and featurebase: first pass

* resolved conflict with master for gitignore & dockerignore

* deleted binaries that were accidentally pushed to git

* combined gitlab jobs for idk & featurebase

* run go fmt for idk

* updated ssh env variable, and made docker password variable in gitlab env variables

* fixed typo assigning variable name

* trying to fix docker login error

* trying a different solution for docker password

* pass registry

* fixed docker login

* updated paths for idk

* exclude idk tests from featurebase test run

* fix vendor error

* update certificates

* grpc needs to be in version 1.38
genproto, which is imported by big query updates the grpc version to 1.47.0
grpc 1.47.0 causes etcd to deadlock when calling etcd.Close()
the fix is to have a replace in go.mod to specify a specific grpc version

* run go mod tidy

* go mod

* run go mod tidy

* exclude bigquery since it is causing issues and undo grpc replace in go.mod

* fix grpc version

* fix formatting error

* update formatting

* attempt to fix formatting

* update path for code coverage

* update to use current branch binaries, not master

* fix for building idk - path updates

* udpate path for binaries

* update job dependecies

* update docker idk tests to use the current branch registry

* update stages for jobs

* updated job dependencies

* not allow idk s3 dump to fail since it is a dependency for integration tests

* update dependecy for idk tests

* update paths for idk build and code coverage

* download featurebase binary from s3

* pass branch name to all setup scripts

* change to current branch instead of master

* updated sonarcloud

* sonarcloud fix and branch name fix

* trying to speed up pipeline run time

* update stage

* branch name fix + sonar cloud

* sonarcloud
2022-07-28 17:23:16 -05:00

611 lines
16 KiB
Go

package idk
import (
"encoding/json"
"strconv"
"strings"
"time"
pilosa "github.com/molecula/featurebase/v3"
"github.com/molecula/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}")
)
// 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:
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:])
}
field = idField
case BoolType:
field = BoolField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to BoolField %s: %v", headerField, fieldspec[1:])
}
case StringType:
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:])
}
field = strField
case LookupTextType:
lTextField := LookupTextField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to LookupTextField %s: %v", headerField, fieldspec[1:])
}
field = lTextField
case IntType:
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:])
}
field = intField
case ForeignKeyType:
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:])
}
field = fkField
case DecimalType:
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:])
}
field = decField
case StringArrayType:
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:])
}
field = strArrField
case IDArrayType:
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:])
}
field = idArrField
case DateIntType:
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, "parsing epoch for '%s'", 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:])
}
field = dateField
case TimestampType:
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, "parsing epoch for '%s'", headerField)
}
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:])
}
field = tsField
case RecordTimeType:
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, "parsing epoch for '%s'", 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:])
}
field = rtField
case SignedIntBoolKeyType:
field = SignedIntBoolKeyField{
NameVal: sourceName,
DestNameVal: destName,
}
if len(fieldspec) > 1 {
log.Printf("ignoring extra arguments to SignedIntBoolKeyField %s: %v", headerField, fieldspec[1:])
}
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, 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
}
func ParseHeader(raw []byte) ([]Field, PathTable, error) {
var rawSchema []struct {
Name string `json:"name"`
Path []string `json:"path"`
Type string `json:"type"`
Config json.RawMessage
}
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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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, "decoding config for field %q", 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
}