From aad32f1dbd5e2c4855f3fb72d20165dc810bf617 Mon Sep 17 00:00:00 2001 From: jacob Date: Sun, 5 Mar 2023 23:03:52 -0600 Subject: [PATCH] adding kafka delete functionality --- idk/api/source.go | 4 + idk/cmd/molecula-consumer-github/main.go | 8 + idk/csv/source.go | 2 + idk/datagen/cmd.go | 2 + idk/idallocator_test.go | 4 + idk/ingest.go | 553 ++++++++---- idk/ingest_test.go | 4 + idk/interfaces.go | 37 +- idk/kafka/cmd_test.go | 804 +++++++++++++++--- idk/kafka/source.go | 56 +- idk/kafka/source_test.go | 2 +- idk/kafka/testdata/records/alltypes.json | 10 + .../records/alltypes_delete_fields.json | 5 + .../records/alltypes_delete_records.json | 2 + .../records/alltypes_delete_records_ids.json | 2 + .../records/alltypes_delete_value.json | 2 + .../records/alltypes_delete_value_ids.json | 2 + idk/kafka/testdata/records/alltypes_null.json | 2 + idk/kafka/testdata/schemas/alltypes.json | 41 + .../schemas/alltypes_delete_fields.json | 12 + .../schemas/alltypes_delete_records.json | 12 + .../schemas/alltypes_delete_value.json | 21 + idk/kafka_sasl/source.go | 2 + idk/kafka_static/source.go | 2 + idk/kinesis/source.go | 2 + idk/sql/source.go | 4 + 26 files changed, 1293 insertions(+), 304 deletions(-) create mode 100644 idk/kafka/testdata/records/alltypes.json create mode 100644 idk/kafka/testdata/records/alltypes_delete_fields.json create mode 100644 idk/kafka/testdata/records/alltypes_delete_records.json create mode 100644 idk/kafka/testdata/records/alltypes_delete_records_ids.json create mode 100644 idk/kafka/testdata/records/alltypes_delete_value.json create mode 100644 idk/kafka/testdata/records/alltypes_delete_value_ids.json create mode 100644 idk/kafka/testdata/records/alltypes_null.json create mode 100644 idk/kafka/testdata/schemas/alltypes.json create mode 100644 idk/kafka/testdata/schemas/alltypes_delete_fields.json create mode 100644 idk/kafka/testdata/schemas/alltypes_delete_records.json create mode 100644 idk/kafka/testdata/schemas/alltypes_delete_value.json diff --git a/idk/api/source.go b/idk/api/source.go index 8b81b8a51..bf83e2892 100644 --- a/idk/api/source.go +++ b/idk/api/source.go @@ -475,3 +475,7 @@ func (r idkRec) Commit(ctx context.Context) error { func (r idkRec) Data() []interface{} { return r } + +func (r idkRec) Schema() interface{} ( + return nil +) \ No newline at end of file diff --git a/idk/cmd/molecula-consumer-github/main.go b/idk/cmd/molecula-consumer-github/main.go index edc664c7d..504a86a37 100644 --- a/idk/cmd/molecula-consumer-github/main.go +++ b/idk/cmd/molecula-consumer-github/main.go @@ -316,6 +316,8 @@ func (r EventRecord) Data() []interface{} { func (r EventRecord) Commit(ctx context.Context) error { return nil } +func (r EventRecord) Schema() interface{} { return nil } + type UserRecord Event func (r UserRecord) Data() []interface{} { @@ -324,6 +326,8 @@ func (r UserRecord) Data() []interface{} { func (r UserRecord) Commit(ctx context.Context) error { return nil } +func (r UserRecord) Schema() interface{} { return nil } + type RepoRecord Event func (r RepoRecord) Data() []interface{} { @@ -332,12 +336,16 @@ func (r RepoRecord) Data() []interface{} { func (r RepoRecord) Commit(ctx context.Context) error { return nil } +func (r RepoRecord) Schema() interface{} { return nil } + type IssueRecord Event func (r IssueRecord) Valid() bool { return r.Type == "IssuesEvent" || r.Type == "IssueCommentEvent" } +func (r IssueRecord) Schema() interface{} { return nil } + func (r IssueRecord) Data() []interface{} { var issue Issue switch r.Type { diff --git a/idk/csv/source.go b/idk/csv/source.go index 23f01def5..89cd254b7 100644 --- a/idk/csv/source.go +++ b/idk/csv/source.go @@ -50,6 +50,8 @@ func (r Record) Data() []interface{} { } func (r Record) Commit(ctx context.Context) error { return nil } // TODO do +func (r Record) Schema() interface{} { return nil } + func (s *Source) Schema() []idk.Field { s.schemaLock.Lock() defer s.schemaLock.Unlock() diff --git a/idk/datagen/cmd.go b/idk/datagen/cmd.go index d99fcb968..8bb81b27d 100644 --- a/idk/datagen/cmd.go +++ b/idk/datagen/cmd.go @@ -632,6 +632,8 @@ func (r record) Data() []interface{} { return r } +func (r record) Schema() interface{} { return nil } + type startEnd struct { start uint64 end uint64 diff --git a/idk/idallocator_test.go b/idk/idallocator_test.go index 0524d4cd2..652396be8 100644 --- a/idk/idallocator_test.go +++ b/idk/idallocator_test.go @@ -198,6 +198,10 @@ func (r *offsetRecord) Commit(ctx context.Context) error { return nil } +func (r *offsetRecord) Schema() interface{} { + return nil +} + func (r *offsetRecord) StreamOffset() (string, uint64) { return r.groupKey, uint64(r.offset) } diff --git a/idk/ingest.go b/idk/ingest.go index aa2c2e4f4..00542afa8 100644 --- a/idk/ingest.go +++ b/idk/ingest.go @@ -35,6 +35,7 @@ import ( "github.com/featurebasedb/featurebase/v3/pql" proto "github.com/featurebasedb/featurebase/v3/proto" "github.com/felixge/fgprof" + "github.com/go-avro/avro" "github.com/pkg/errors" prom "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -263,20 +264,22 @@ func (m *Main) run() error { } l := &msgCounter{MaxMsgs: m.MaxMsgs} - for c := 0; c < m.Concurrency; c++ { - c := c - eg.Go(func() error { - var err error - if m.Delete { - err = m.runDeleter(c, l) - } else { - err = m.runIngester(c, l) - } - if err != nil && err != io.EOF { - return err - } - return nil - }) + if m.Delete { + err := m.runDeleter(l) + if err != nil { + return err + } + } else { + for c := 0; c < m.Concurrency; c++ { + c := c + eg.Go(func() error { + err := m.runIngester(c, l) + if err != nil && err != io.EOF { + return err + } + return nil + }) + } } return errors.Wrap(eg.Wait(), "idk.Main.Run") } @@ -798,6 +801,9 @@ func (m *Main) Setup() (onFinishRun func(), err error) { return nil, errors.Wrap(err, "creating featurebase client") } m.grpcClient = grpcClient + if m.Concurrency > 1 { + return nil, errors.New("delete consumers does not support concurrency > 1") + } } if m.LookupDBDSN != "" { @@ -1087,12 +1093,16 @@ func (h metricsJSONHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } } -func (m *Main) runDeleter(c int, limitCounter *msgCounter) error { +// runDeleter pulls records from the source associated with a consumer +// and deletes data from featurebase based on the format of that record +// it is currently mainly used with and tested on the kafka / avro consumer +// built on IDK +func (m *Main) runDeleter(limitCounter *msgCounter) error { m, err := m.clone() if err != nil { return errors.Wrap(err, "cloning *Main before delete") } - m.log.Printf("start deleter %d", c) + m.log.Printf("starting the delete consumer...") source, err := m.NewSource() if err != nil { return errors.Wrap(err, "getting source") @@ -1121,181 +1131,377 @@ func (m *Main) runDeleter(c int, limitCounter *msgCounter) error { client := m.PilosaClient() index := m.index + // Pull records one by one from kafka for ; !limitCounter.IsDone(); rec, err = source.Record() { if err == ErrFlush { continue + } else if err != nil && err != ErrSchemaChange { + return errors.Wrap(err, "getting record") } - m.log.Debugf("deleter record: %v\n", rec) - if err != nil { - if err == ErrSchemaChange { - schema := source.Schema() - recordizers, _, row, _, err = m.batchFromSchema(schema) - if err != nil { - return errors.Wrap(err, "batchFromSchema") - } + + bq := index.BatchQuery() + recSchema := rec.Schema() + avroRecord := false + deleteType := "" + switch recSchema.(type) { + case avro.Schema: + // record was encoded using avro + // in runDeleter, that avro.RecordSchema should have a delete property + // if it doesn't, it defaults to "fields" which runs older logic + // (i.e. this new code won't break old integrations) + deleteProp, ok := recSchema.(avro.Schema).Prop("delete") + if !ok || deleteProp == nil { + deleteType = "fields" } else { - break - } - } - data := rec.Data() - m.log.Debugf("deleter data: %+v %+v %+v", data, data[0], data[1]) - for _, rdz := range recordizers { - err = rdz(data, row) - if err != nil { - return errors.Wrap(err, "recordizing") - } - } - - var columnIDs []uint64 - var columnKeys []string - switch rowIdent := row.ID.(type) { - case int64: - columnIDs = []uint64{uint64(rowIdent)} - case uint64: - columnIDs = []uint64{rowIdent} - case string: - columnKeys = []string{rowIdent} - case []byte: - columnKeys = []string{string(rowIdent)} - default: - return errors.Errorf("recordizing primary key, got type: %T", row.ID) - } - - // TODO: sentinel value in field list to delete entire record - directives, ok := row.Values[len(row.Values)-1].([]string) - if !ok { - if row.Values[len(row.Values)-1] == nil { - continue - } - return errors.Errorf("directives should be a string slice but got: %+v of %[1]T", row.Values[len(row.Values)-1]) - } - if len(directives) == 0 { - continue - } - - // TODO: `directives` is not necessarily equivalent to a list of fieldNames - rr, err := inspect(m.grpcClient, m.Index, columnIDs, columnKeys, directives) - if err != nil { - return errors.Wrap(err, "retrieving values for delete") - } - - trns, err := m.SchemaManager.StartTransaction("", time.Minute, false, time.Hour) - if err != nil { - return errors.Wrap(err, "starting transaction") - } - for _, directive := range directives { - var recordID interface{} = row.ID - if idbytes, ok := recordID.([]byte); ok { - recordID = string(idbytes) - } - var fieldName string - if nameval := strings.SplitN(directive, "|", 2); len(nameval) == 2 { - // special handling for packed bools — may generalize this in future - fieldName = nameval[0] - value := nameval[1] - if m.PackBools == "" || fieldName != m.PackBools { - return errors.Errorf("unsupported directive '%s' field name must be equal to packed bools field: '%s'", directive, m.PackBools) - } - boolsField := index.Field(m.PackBools) - boolsExists := index.Field(m.PackBools + Exists) - _, err := client.Query(index.BatchQuery( - boolsField.Clear(value, recordID), - boolsExists.Clear(value, recordID), - )) - if err != nil { - return errors.Wrap(err, "clearing bools") - } - CounterDeleterRowsAdded.With(prom.Labels{"type": "packed-bool"}).Inc() - continue - } else { - fieldName = directive - } - - // get field, refreshing schema if needed - field, ok := index.Fields()[fieldName] - if !ok { - schema, err := m.SchemaManager.Schema() - if err != nil { - return errors.Wrap(err, "unknown field, getting new schema") - } - index = schema.Index(m.Index) - field, ok = index.Fields()[fieldName] + deleteType, ok = deleteProp.(string) if !ok { - return errors.Errorf("field '%s' not found", fieldName) + return errors.Errorf("delete property of avro delete record should be a string") } } + avroRecord = true + } - val, err := rr.Val(fieldName) + if !avroRecord || deleteType == "fields" { + // records without avro schemas or records with avro schemas + // with deletes property of "fields" uses historical logic + m.log.Debugf("deleter record: %v\n", rec) if err != nil { - return errors.Wrap(err, "getting value from inspect") - } - - switch field.Options().Type() { - case pilosaclient.FieldTypeDefault, pilosaclient.FieldTypeSet: - bq := index.BatchQuery() - if field.Options().Keys() { - valStrs, ok := val.([]string) - if !ok { - return errors.Errorf("unexpected value type for set field with keys, not []string but %T", val) - } - for _, valS := range valStrs { - bq.Add(field.Clear(valS, recordID)) + if err == ErrSchemaChange { + schema := source.Schema() + recordizers, _, row, _, err = m.batchFromSchema(schema) + if err != nil { + return errors.Wrap(err, "batchFromSchema") } } else { - valIDs, ok := val.([]uint64) - if !ok { - return errors.Errorf("unexpected value type for set field, not []uint64 but %T", val) - } - for _, valID := range valIDs { - bq.Add(field.Clear(valID, recordID)) - } + // shouldn't be able to get here + break } - _, err := client.Query(bq) + } + data := rec.Data() + m.log.Debugf("deleter data: %+v %+v %+v", data, data[0], data[1]) + for _, rdz := range recordizers { + err = rdz(data, row) if err != nil { - return errors.Wrap(err, "clearing set") + return errors.Wrap(err, "recordizing") } - CounterDeleterRowsAdded.With(prom.Labels{"type": "set"}).Inc() - case pilosaclient.FieldTypeMutex: - if val == "" { + } + + var columnIDs []uint64 + var columnKeys []string + switch rowIdent := row.ID.(type) { + case int64: + columnIDs = []uint64{uint64(rowIdent)} + case uint64: + columnIDs = []uint64{rowIdent} + case string: + columnKeys = []string{rowIdent} + case []byte: + columnKeys = []string{string(rowIdent)} + default: + return errors.Errorf("recordizing primary key, got type: %T", row.ID) + } + + // TODO: sentinel value in field list to delete entire record + directives, ok := row.Values[len(row.Values)-1].([]string) + if !ok { + if row.Values[len(row.Values)-1] == nil { continue } - _, err := client.Query(index.BatchQuery( - field.Clear(val, recordID), - )) - if err != nil { - return errors.Wrap(err, "clearing mutex") - } - CounterDeleterRowsAdded.With(prom.Labels{"type": "mutex"}).Inc() - case pilosaclient.FieldTypeBool: - _, err := client.Query(index.BatchQuery( - field.Clear(0, recordID), - field.Clear(1, recordID), - )) - if err != nil { - return errors.Wrap(err, "clearing bool") - } - CounterDeleterRowsAdded.With(prom.Labels{"type": "bool"}).Inc() - case pilosaclient.FieldTypeInt: - _, err := client.Query(field.Clear(0, recordID)) - if err != nil { - return errors.Wrap(err, "clearing int") - } - CounterDeleterRowsAdded.With(prom.Labels{"type": "int"}).Inc() - case pilosaclient.FieldTypeDecimal: - _, err := client.Query(field.Clear(0, recordID)) - if err != nil { - return errors.Wrap(err, "clearing decimal") - } - CounterDeleterRowsAdded.With(prom.Labels{"type": "decimal"}).Inc() - case pilosaclient.FieldTypeTime: - return errors.Errorf("deletion on time fields unimplemented") - default: - return errors.Errorf("unhandled field type %s", field.Options().Type()) + return errors.Errorf("directives should be a string slice but got: %+v of %[1]T", row.Values[len(row.Values)-1]) + } + if len(directives) == 0 { + continue + } + + // TODO: `directives` is not necessarily equivalent to a list of fieldNames + rr, err := inspect(m.grpcClient, m.Index, columnIDs, columnKeys, directives) + if err != nil { + return errors.Wrap(err, "retrieving values for delete") + } + + trns, err := m.SchemaManager.StartTransaction("", time.Minute, false, time.Hour) + if err != nil { + return errors.Wrap(err, "starting transaction") + } + for _, directive := range directives { + var recordID interface{} = row.ID + if idbytes, ok := recordID.([]byte); ok { + recordID = string(idbytes) + } + var fieldName string + if nameval := strings.SplitN(directive, "|", 2); len(nameval) == 2 { + // special handling for packed bools — may generalize this in future + fieldName = nameval[0] + value := nameval[1] + if m.PackBools == "" || fieldName != m.PackBools { + return errors.Errorf("unsupported directive '%s' field name must be equal to packed bools field: '%s'", directive, m.PackBools) + } + boolsField := index.Field(m.PackBools) + boolsExists := index.Field(m.PackBools + Exists) + _, err := client.Query(index.BatchQuery( + boolsField.Clear(value, recordID), + boolsExists.Clear(value, recordID), + )) + if err != nil { + return errors.Wrap(err, "clearing bools") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "packed-bool"}).Inc() + continue + } else { + fieldName = directive + } + + // get field, refreshing schema if needed + field, ok := index.Fields()[fieldName] + if !ok { + schema, err := m.SchemaManager.Schema() + if err != nil { + return errors.Wrap(err, "unknown field, getting new schema") + } + index = schema.Index(m.Index) + field, ok = index.Fields()[fieldName] + if !ok { + return errors.Errorf("field '%s' not found", fieldName) + } + } + + val, err := rr.Val(fieldName) + if err != nil { + return errors.Wrap(err, "getting value from inspect") + } + + switch field.Options().Type() { + case pilosaclient.FieldTypeDefault, pilosaclient.FieldTypeSet: + if field.Options().Keys() { + valStrs, ok := val.([]string) + if !ok { + return errors.Errorf("unexpected value type for set field with keys, not []string but %T", val) + } + for _, valS := range valStrs { + bq.Add(field.Clear(valS, recordID)) + } + } else { + valIDs, ok := val.([]uint64) + if !ok { + return errors.Errorf("unexpected value type for set field, not []uint64 but %T", val) + } + for _, valID := range valIDs { + bq.Add(field.Clear(valID, recordID)) + } + } + _, err := client.Query(bq) + if err != nil { + return errors.Wrap(err, "clearing set") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "set"}).Inc() + case pilosaclient.FieldTypeMutex: + if val == "" { + continue + } + _, err := client.Query(index.BatchQuery( + field.Clear(val, recordID), + )) + if err != nil { + return errors.Wrap(err, "clearing mutex") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "mutex"}).Inc() + case pilosaclient.FieldTypeBool: + _, err := client.Query(index.BatchQuery( + field.Clear(0, recordID), + field.Clear(1, recordID), + )) + if err != nil { + return errors.Wrap(err, "clearing bool") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "bool"}).Inc() + case pilosaclient.FieldTypeInt: + _, err := client.Query(field.Clear(0, recordID)) + if err != nil { + return errors.Wrap(err, "clearing int") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "int"}).Inc() + case pilosaclient.FieldTypeDecimal: + _, err := client.Query(field.Clear(0, recordID)) + if err != nil { + return errors.Wrap(err, "clearing decimal") + } + CounterDeleterRowsAdded.With(prom.Labels{"type": "decimal"}).Inc() + case pilosaclient.FieldTypeTime: + return errors.Errorf("deletion on time fields unimplemented") + default: + return errors.Errorf("unhandled field type %s", field.Options().Type()) + } + } + if len(directives) == 0 { + continue + } + _, err = m.SchemaManager.FinishTransaction(trns.ID) + if err != nil { + return errors.Wrap(err, "finishing transaction") + } + } else { + // here we have an record encoded by avro and it's delete type is + // "values", "records", or some unacceptable input + recRecordSchema, ok := recSchema.(*avro.RecordSchema) + if !ok { + return errors.Errorf("got data of type %T but wanted avro.RecordSchema", recSchema) + } + + // map values to fields or _id + avroFields := recRecordSchema.Fields + var recordID interface{} + fieldValues := make(map[string]interface{}) + for i, value := range rec.Data() { + name := avroFields[i].Name + if name == "_id" { + if m.index.Opts().Keys() == false { + recordID, err = toUint64(value) + if err != nil { + return errors.Errorf("unable convert _id to uint64 for index %s which is has keys set to false", m.index.Name()) + } + } else { + recordID, err = toString(value) + if err != nil { + return errors.Errorf("unable convert _id to string for index %s which is has keys set to true", m.index.Name()) + } + } + + } else { + fieldValues[name] = value + } + } + + switch deleteType { + case "values": + // find featurebase field based on avro / record field name + indexFields := index.Fields() + for key, value := range fieldValues { + field, ok := indexFields[key] + if !ok { + return errors.Errorf("unable to find field %s in index %s", key, index.Name()) + } + if value == nil { + // don't delete anything for this field if value is null + continue + } + switch fType := field.Options().Type(); fType { + case pilosaclient.FieldTypeSet, pilosaclient.FieldTypeMutex: + if key == m.PackBools { + // value should be list of bools to clear if avro field name + // is equal the name of the packed bools field + if arrayValue, err := toStringArray(value); err == nil { + boolsField := index.Field(m.PackBools) + boolsExists := index.Field(m.PackBools + Exists) + for _, v := range arrayValue { + m.log.Debugf("clearing %s and %s for %s bool field", m.PackBools, m.PackBools+Exists, v) + bq.Add(boolsField.Clear(v, recordID)) + bq.Add(boolsExists.Clear(v, recordID)) + } + } else { + return errors.Errorf("packed bools field %s should be a list of boolean values to delete", field.Name()) + } + } else { + // not packed bools, check for string vs ID field keys + switch keys := field.Options().Keys(); keys { + case true: + if arrayValue, err := toStringArray(value); err == nil { + for _, v := range arrayValue { + bq.Add(field.Clear(v, recordID)) + } + } else { + return errors.Errorf("value of keyed %s field %s should be a string or array of strings but was %T", fType, field.Name(), value) + } + case false: + if singleValue, err := toUint64(value); err == nil { + bq.Add(field.Clear(singleValue, recordID)) + } else if arrayValue, err := toUint64Array(value); err == nil { + for _, v := range arrayValue { + bq.Add(field.Clear(v, recordID)) + } + } else { + return errors.Errorf("value of non keyed %s field %s should be an int or array of ints but was %T", fType, field.Name(), value) + } + default: + return errors.Errorf("set field %s should have keys true or false", field.Name()) + } + } + case pilosaclient.FieldTypeInt, pilosaclient.FieldTypeDecimal, pilosaclient.FieldTypeTimestamp: + if boolVal, ok := value.(bool); ok { + if boolVal { + bq.Add(field.Clear(0, recordID)) + } + } else { + return errors.Errorf("%s fields should have a boolean value set to rue if value is to be deleted, false otherwise", fType) + } + case pilosaclient.FieldTypeBool: + if boolVal, ok := value.(bool); ok { + if boolVal { + bq.Add(field.Clear(0, recordID)) + bq.Add(field.Clear(1, recordID)) + } + } else { + return errors.Errorf("%s fields should have a boolean value set to rue if value is to be deleted, false otherwise", fType) + } + default: + // pilosa.FieldTypeTime is the only other field type at time of coding + return errors.Errorf("unable to handle values from fields with type: %s", fType) + } + } + m.log.Debugf("Delete consumer running the follow delete queries: %s", bq.Serialize()) + + resp, err := client.Query(bq, nil) + if err != nil || resp.Success != true { + return errors.Wrap(err, "error deleting values") + } + case "records": + // deleting a record + var rawQueries []string + if fieldValues["keys"] != nil { + // if keys set, delete list of record keys + keysAsStrings, err := toStringArray(fieldValues["keys"]) + if err != nil { + return errors.Errorf("unable to convert 'keys' value to an array of strings") + } + columnKeys := "'" + strings.Join(keysAsStrings, "','") + "'" + rawQueries = append(rawQueries, fmt.Sprintf("Delete(ConstRow(columns=[%s]))", columnKeys)) + + } + if fieldValues["ids"] != nil { + // if ids set, delete list of record IDs + idsAsInts, err := toUint64Array(fieldValues["ids"]) + if err != nil { + return errors.Errorf("unable to convert 'ids' value to an array of int64s") + } + keysAsStrings := make([]string, len(idsAsInts)) + for i, v := range idsAsInts { + keysAsStrings[i] = fmt.Sprint(v) + } + columnKeys := strings.Join(keysAsStrings, ",") + rawQueries = append(rawQueries, fmt.Sprintf("Delete(ConstRow(columns=[%s]))", columnKeys)) + } + if fieldValues["filter"] != nil { + // if filter set, use it as filter in delete query + filter, ok := fieldValues["filter"].(string) + if !ok { + return errors.Errorf("unable to convert fitler into string") + } + rawQueries = append(rawQueries, fmt.Sprintf("Delete(%s)", filter)) + } + if len(rawQueries) == 0 { + m.log.Infof("delete record doesn't contain any delete queries: confirm 'keys', 'ids', or 'filter' key has a value") + } + for _, query := range rawQueries { + baseQuery := index.RawQuery(query) + bq.Add(baseQuery) + } + + _, err := client.Query(bq, nil) + if err != nil { + return errors.Errorf("running delete query: %s", err) + } + default: + return errors.Errorf("unable to process delete where record is avro encoded & the delete property is not empty, 'records', 'fields', or 'values'") } - } - _, err = m.SchemaManager.FinishTransaction(trns.ID) - if err != nil { - return errors.Wrap(err, "finishing transaction") } err = m.commitRecord(context.Background(), rec, limitCounter, 1) if err != nil { @@ -1304,6 +1510,7 @@ func (m *Main) runDeleter(c int, limitCounter *msgCounter) error { if limitCounter.IsDone() { return nil } + } if !errors.Is(err, io.EOF) { diff --git a/idk/ingest_test.go b/idk/ingest_test.go index 40253ba18..03394719f 100644 --- a/idk/ingest_test.go +++ b/idk/ingest_test.go @@ -1325,6 +1325,10 @@ func (s *sliceRecord) Data() []interface{} { return s.data } +func (s *sliceRecord) Schema() interface{} { + return nil +} + func (s *testSource) Record() (Record, error) { s.i++ if s.i <= len(s.records) { diff --git a/idk/interfaces.go b/idk/interfaces.go index 186abb226..a721694c6 100644 --- a/idk/interfaces.go +++ b/idk/interfaces.go @@ -69,6 +69,8 @@ type ( Commit(ctx context.Context) error Data() []interface{} + + Schema() interface{} } // OffsetStreamRecord is an extension of the record type which also tracks offsets within streams. @@ -840,6 +842,26 @@ func (t TimestampField) PilosafyVal(val interface{}) (interface{}, error) { tsAsVal := TimestampToVal(t.granularity(), ts) + dur = tsAsVal - epochAsVal + } else if _, ok := val.([]byte); ok { + valAsString := string(val.([]byte)[:]) + // try to convert as time string + ts, err := timeFromTimestring(valAsString, t.layout()) + if err != nil { + // if that doesn't work, maybe it's an int represented as a string + valAsInt, err := strconv.ParseInt(valAsString, 0, 64) + if err == nil { + return valAsInt - epochAsVal, nil + } else { + return nil, errors.Wrap(err, "converting TimestampField") + } + } + if err := validateTimestamp(t.granularity(), ts); err != nil { + return nil, errors.Wrap(ErrTimestampOutOfRange, "validating timestamp") + } + + tsAsVal := TimestampToVal(t.granularity(), ts) + dur = tsAsVal - epochAsVal } else { valAsInt, err := toInt64(val) @@ -1195,6 +1217,8 @@ func toInt64(val interface{}) (int64, error) { return 0, err } return v, nil + case []byte: + return toInt64(string(vt[:])) default: return 0, errors.Errorf("couldn't convert %v of %[1]T to int64", vt) } @@ -1228,11 +1252,16 @@ func toStringArray(val interface{}) ([]string, error) { case []interface{}: ret := make([]string, len(vt)) for i, v := range vt { - vs, ok := v.(string) - if !ok { - return nil, errors.Errorf("couldn't convert []interface{} to []string, value %v of type %[1]T at %d", v, i) + switch v.(type) { + case []byte: + ret[i] = string(v.([]byte)[:]) + default: + vs, ok := v.(string) + if !ok { + return nil, errors.Errorf("couldn't convert []interface{} to []string, value %v of type %[1]T at %d", v, i) + } + ret[i] = vs } - ret[i] = vs } return ret, nil default: diff --git a/idk/kafka/cmd_test.go b/idk/kafka/cmd_test.go index 2cba2a680..994f27ae6 100644 --- a/idk/kafka/cmd_test.go +++ b/idk/kafka/cmd_test.go @@ -36,21 +36,38 @@ var ( ) func init() { + local := false var ok bool if pilosaHost, ok = os.LookupEnv("IDK_TEST_PILOSA_HOST"); !ok { - pilosaHost = "pilosa:10101" + if local { + pilosaHost = "localhost:10101" + } else { + pilosaHost = "pilosa:10101" + } } if pilosaTLSHost, ok = os.LookupEnv("IDK_TEST_PILOSA_TLS_HOST"); !ok { pilosaTLSHost = "https://pilosa-tls:10111" } if pilosaGrpcHost, ok = os.LookupEnv("IDK_TEST_PILOSA_GRPC_HOST"); !ok { - pilosaGrpcHost = "pilosa:20101" + if local { + pilosaGrpcHost = "localhost:20101" + } else { + pilosaGrpcHost = "pilosa:20101" + } } if kafkaHost, ok = os.LookupEnv("IDK_TEST_KAFKA_HOST"); !ok { - kafkaHost = "kafka:9092" + if local { + kafkaHost = "localhost:9092" + } else { + kafkaHost = "kafka:9092" + } } if registryHost, ok = os.LookupEnv("IDK_TEST_REGISTRY_HOST"); !ok { - registryHost = "schema-registry:8081" + if local { + registryHost = "localhost:8081" + } else { + registryHost = "schema-registry:8081" + } } if certPath, ok = os.LookupEnv("IDK_TEST_CERT_PATH"); !ok { certPath = "/certs" @@ -640,132 +657,721 @@ func TestCmdSchemaChange(t *testing.T) { } } -func TestTimeQuantums(t *testing.T) { - t.Parallel() +type ConsumerTestConfig struct { + idType string // must be "generated", "id", or "string" + keyFields []string // field used for "id" or "string" record keys + topic string + delete bool +} + +/* +Struct that captures data required to run a test that meets the following conditions: + - There is a JSON file will plain text records to write to kafka + - There is an Avro Schema that will be used to record records + above before writing them to kafka + - The data written above will be consumed and set to FeatureBase + - There is a list of PQL queries that can be used to determine if + above went correctly or incorrectly + +For test that require multiple iterations of " consumer data from kafka, write it to +FeatureBase, and run queries against it, " +*/ +type ConsumerTest struct { + name string + pathsToAvroSchema []string // code currently prepends values with ./testdata/schema/ + pathsToRecords []string + consumerConfigs []ConsumerTestConfig + index string + queries [][]string + expectedResults [][]string + pilosaHosts string + kafkaHost string + registryURL string +} + +func TestAddingRemovingData(t *testing.T) { + //t.Parallel() /* at a high level, a test here represents - an avro schema - a set of records to ingest to kafka - an ingest configuration - query to run to confirm the data was ingest properly + + see test */ - tests := []struct { - name string - pathToAvroSchema string - pathToRecords string - idType string // must be "generated", "id", or "string" - keyField string // field used for "id" or "string" record keys - pilosaHosts string - kafkaHost string - registryURL string - topic string - index string - queries []string - expectedResults []string - }{ + tests := []ConsumerTest{ + { // confirm time quantums are being ingested - name: "time quantums exist", - pathToAvroSchema: "timeQuantum.json", - pathToRecords: "./testdata/records/timeQuantum.json", - idType: "string", - keyField: "device", - pilosaHosts: pilosaHost, - registryURL: registryHost, - kafkaHost: kafkaHost, - topic: "timequantums", - index: "timequantums", - queries: []string{ - "Row(segment_ts='7R83')", - "Row(segment_ts='7R83', from=\"2023-02-17T00:00\", to=\"2023-02-18T00:00\")", - "Row(segment_ts='7R83', from=\"2023-02-16T00:00\", to=\"2023-02-17T00:00\")", + name: "time quantums exist", + pathsToAvroSchema: []string{"timeQuantum.json"}, + pathsToRecords: []string{"./testdata/records/timeQuantum.json"}, + pilosaHosts: pilosaHost, + registryURL: registryHost, + kafkaHost: kafkaHost, + consumerConfigs: []ConsumerTestConfig{ + { + idType: "string", + keyFields: []string{"device"}, //can be multiple for idType: string but a single value otherwise + topic: "timequantums", + delete: false, + }, }, - expectedResults: []string{ - "{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n", - "{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n", - "{\"results\":[{\"columns\":[]}]}\n", + index: "timequantums", + queries: [][]string{ + { + "Row(segment_ts='7R83')", + "Row(segment_ts='7R83', from=\"2023-02-17T00:00\", to=\"2023-02-18T00:00\")", + "Row(segment_ts='7R83', from=\"2023-02-16T00:00\", to=\"2023-02-17T00:00\")", + }, + }, + expectedResults: [][]string{ + { + "{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"0QKtSTqJYXMZWvVe\"]}]}\n", + "{\"results\":[{\"columns\":[]}]}\n", + }, + }, + }, + { // confirm all types are being ingest as they should + name: "all values can be inserted", + pathsToAvroSchema: []string{"alltypes.json"}, + pathsToRecords: []string{"./testdata/records/alltypes.json"}, + pilosaHosts: pilosaHost, + registryURL: registryHost, + kafkaHost: kafkaHost, + consumerConfigs: []ConsumerTestConfig{ + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes", + delete: false, + }, + }, + index: "alltypes", + queries: [][]string{ + { + "Count(All())", + "Count(ConstRow(columns=['u2Yr4|sHaUv|x5z8P', 'DY2Ui|kUbdU|pjxqm']))", + "Count(Row(stringset_string='58KIR'))", + "Count(Row(string_string='8MGwy'))", + "Count(Row(stringtq_string='ivWWb'))", + "Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))", + "Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))", + "Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))", + "Row(string_bytes='vTwn4')", + "Row(stringsettq_bytes='798ka')", + "Row(stringsettq_bytes='798ka', from=\"2023-02-18\")", + "Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\")", + "Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))", + "Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")", + "Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))", + "Count(Row(idset_long=839))", + "Count(Row(id_long=809))", + "Count(Row(idtq_long=533))", + "Count(Row(idtq_long=533, from=\"2020-01-01\"))", + "Count(Row(idset_int=533))", + "Count(Row(id_int=168))", + "Count(Row(idsettq_int=113))", + "Row(idsettq_int=113, to=\"2024-01-01\")", + "Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))", + "Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))", + "Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))", + "Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))", + "Count(Row(int_long > 500))", + "Count(Row(int_int > 500))", + "Count(Row(decimal_bytes > 1000.00))", + "Count(Row(decimal_float > 3.05))", + "Count(Row(decimal_double > 4.11))", + "Count(Row(dateint_bytes_ts > 1675163490))", + "Count(Row(bools=bool_bool))", + "Count(Not(Row(bools=bool_bool)))", + "Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))", + "Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))", + }, + }, + expectedResults: [][]string{ + { + "{\"results\":[10]}\n", + "{\"results\":[2]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[0]}\n", + "{\"results\":[1]}\n", + "{\"results\":[2]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"DY2Ui|kUbdU|pjxqm\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[{\"columns\":[]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[4]}\n", + "{\"results\":[5]}\n", + "{\"results\":[4]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + "{\"results\":[8]}\n", + "{\"results\":[4]}\n", + "{\"results\":[6]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + }, + }, + }, + { // confirm behavior of nulls.. when all null values passed, nothing should change + name: "all values can be inserted", + pathsToAvroSchema: []string{"alltypes.json"}, + pathsToRecords: []string{"./testdata/records/alltypes_null.json"}, + pilosaHosts: pilosaHost, + registryURL: registryHost, + kafkaHost: kafkaHost, + consumerConfigs: []ConsumerTestConfig{ + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes_null", + delete: false, + }, + }, + index: "alltypes_null", + queries: [][]string{ + { + //"Extract(All(), Rows(stringset_string), Rows(string_string), Rows(stringtq_string), Rows(stringset_bytes), Rows(string_bytes), Rows(stringsettq_bytes), Rows(stringset_stringarray), Rows(stringtq_stringarray), Rows(stringset_bytesarray), Rows(stringtq_bytesarray), Rows(idset_long), Rows(id_long), Rows(idtq_long), Rows(idset_int), Rows(id_int), Rows(idsettq_int), Rows(idset_longarray), Rows(idtq_longarray), Rows(idset_intarray), Rows(stringtq_stringarray), Rows(stringset_bytesarray), Rows(stringtq_bytesarray), Rows(idset_long), Rows(id_long), Rows(idtq_long), Rows(idset_int), Rows(id_int), Rows(idsettq_int), Rows(idset_longarray), Rows(idtq_longarray), Rows(idset_intarray), Rows(int_long), Rows(int_int), Rows(decimal_bytes), Rows(decimal_float), Rows(decimal_double), Rows(dateint_bytes_ts), Rows(bools), Rows(timestamp_bytes_ts), Rows(timestamp_bytes_int))", + "Count(All())", + "Count(Row(stringset_string='7EYSp'))", + "Count(Row(string_string='uirDR'))", + "Count(Row(stringtq_string='Qylqq'))", + "Count(Row(stringset_bytes='gL2Hg'))", + "Count(Row(string_bytes='BmvHF'))", + "Count(Row(stringsettq_bytes='798ka'))", + "Count(Intersect(Row(stringset_stringarray='vbbuf'), Row(stringset_stringarray='VQs7y'), Row(stringset_stringarray='9z4aw'), Row(stringset_stringarray='h1iqc'), Row(stringset_stringarray='aQQxr')))", + "Count(Intersect(Row(stringtq_stringarray='x5z8P'), Row(stringtq_stringarray='0UGJQ'), Row(stringtq_stringarray='58KIR'), Row(stringtq_stringarray='7EYSp'), Row(stringtq_stringarray='CKs1F')))", + "Count(Intersect(Row(stringset_bytesarray='u2Yr4'), Row(stringset_bytesarray='tvNOB'), Row(stringset_bytesarray='iYeOV'), Row(stringset_bytesarray='ZgkOB'), Row(stringset_bytesarray='RPGAm')))", + "Count(Intersect(Row(stringtq_bytesarray='BwqU2'), Row(stringtq_bytesarray='6iGIm'), Row(stringtq_bytesarray='fjQK2'), Row(stringtq_bytesarray='LBTEU'), Row(stringtq_bytesarray='C6xxn')))", + "Count(Row(idset_long=647))", + "Count(Row(id_long=792))", + "Count(Row(idtq_long=676))", + "Count(Row(idset_int=898))", + "Count(Row(id_int=63))", + "Count(Row(idsettq_int=890))", + "Count(Intersect(Row(idset_longarray=442), Row(idset_longarray=167), Row(idset_longarray=230), Row(idset_longarray=344), Row(idset_longarray=733)))", + "Count(Intersect(Row(idtq_longarray=385), Row(idtq_longarray=931), Row(idtq_longarray=157), Row(idtq_longarray=865), Row(idtq_longarray=394)))", + "Count(Intersect(Row(idset_intarray=442), Row(idset_intarray=614), Row(idset_intarray=394), Row(idset_intarray=284), Row(idset_intarray=344)))", + "Count(Row(int_long=584))", + "Count(Row(int_int=344))", + "Count(Row(decimal_bytes=1155.95))", + "Count(Row(decimal_float=3.23))", + "Count(Row(decimal_double=0.95))", + "Count(Row(dateint_bytes_ts=1676534039))", + "Count(Row(bools=bool_bool))", + "Count(Row(timestamp_bytes_ts='2023-02-16T07:53:59Z'))", + "Count(Row(timestamp_bytes_int=1676555639))", + }, + }, + expectedResults: [][]string{ + { + //"{\"results\":[{\"fields\":[{\"name\":\"stringset_string\",\"type\":\"[]string\"},{\"name\":\"string_string\",\"type\":\"string\"},{\"name\":\"stringtq_string\",\"type\":\"[]string\"},{\"name\":\"stringset_bytes\",\"type\":\"[]string\"},{\"name\":\"string_bytes\",\"type\":\"string\"},{\"name\":\"stringsettq_bytes\",\"type\":\"[]string\"},{\"name\":\"stringset_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringset_bytesarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_bytesarray\",\"type\":\"[]string\"},{\"name\":\"idset_long\",\"type\":\"[]uint64\"},{\"name\":\"id_long\",\"type\":\"uint64\"},{\"name\":\"idtq_long\",\"type\":\"[]uint64\"},{\"name\":\"idset_int\",\"type\":\"[]uint64\"},{\"name\":\"id_int\",\"type\":\"uint64\"},{\"name\":\"idsettq_int\",\"type\":\"[]uint64\"},{\"name\":\"idset_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idtq_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idset_intarray\",\"type\":\"[]uint64\"},{\"name\":\"stringtq_stringarray\",\"type\":\"[]string\"},{\"name\":\"stringset_bytesarray\",\"type\":\"[]string\"},{\"name\":\"stringtq_bytesarray\",\"type\":\"[]string\"},{\"name\":\"idset_long\",\"type\":\"[]uint64\"},{\"name\":\"id_long\",\"type\":\"uint64\"},{\"name\":\"idtq_long\",\"type\":\"[]uint64\"},{\"name\":\"idset_int\",\"type\":\"[]uint64\"},{\"name\":\"id_int\",\"type\":\"uint64\"},{\"name\":\"idsettq_int\",\"type\":\"[]uint64\"},{\"name\":\"idset_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idtq_longarray\",\"type\":\"[]uint64\"},{\"name\":\"idset_intarray\",\"type\":\"[]uint64\"}],\"columns\":[{\"column\":\"9z4aw|5ptDx|CKs1F\",\"rows\":[[\"7EYSp\"],\"uirDR\",[\"Qylqq\"],[\"gL2Hg\"],\"BmvHF\",[\"798ka\"],[\"9z4aw\",\"h1iqc\",\"aQQxr\",\"vbbuf\",\"VQs7y\"],[\"7EYSp\",\"CKs1F\",\"x5z8P\",\"0UGJQ\",\"58KIR\"],[\"u2Yr4\",\"tvNOB\",\"iYeOV\",\"ZgkOB\",\"RPGAm\"],[\"BwqU2\",\"6iGIm\",\"fjQK2\",\"LBTEU\",\"C6xxn\"],[647],792,[676],[898],63,[890],[167,230,344,442,733],[157,385,394,865,931],[284,344,394,442,614],[\"7EYSp\",\"CKs1F\",\"x5z8P\",\"0UGJQ\",\"58KIR\"],[\"u2Yr4\",\"tvNOB\",\"iYeOV\",\"ZgkOB\",\"RPGAm\"],[\"BwqU2\",\"6iGIm\",\"fjQK2\",\"LBTEU\",\"C6xxn\"],[647],792,[676],[898],63,[890],[167,230,344,442,733],[157,385,394,865,931],[284,344,394,442,614]]}]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + }, + }, + }, + { // confirm delete behavior when index is keyed + name: "all values can be deleted with keys", + pathsToAvroSchema: []string{ + "alltypes.json", + "alltypes_delete_fields.json", + "alltypes_delete_records.json", + "alltypes_delete_value.json", + }, + pathsToRecords: []string{ + "./testdata/records/alltypes.json", + "./testdata/records/alltypes_delete_fields.json", + "./testdata/records/alltypes_delete_records.json", + "./testdata/records/alltypes_delete_value.json", + }, + pilosaHosts: pilosaHost, + registryURL: registryHost, + kafkaHost: kafkaHost, + consumerConfigs: []ConsumerTestConfig{ + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes", + delete: false, + }, + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes_delete_fields", + delete: true, + }, + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes_delete_record", + delete: true, + }, + { + idType: "string", + keyFields: []string{"pk0", "pk1", "pk2"}, + topic: "alltypes_delete_values", + delete: true, + }, + }, + index: "alltypes_delete", + queries: [][]string{ + { + "Count(All())", + "Count(ConstRow(columns=['u2Yr4|sHaUv|x5z8P', 'DY2Ui|kUbdU|pjxqm']))", + "Count(Row(stringset_string='58KIR'))", + "Count(Row(string_string='8MGwy'))", + "Count(Row(stringtq_string='ivWWb'))", + "Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))", + "Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))", + "Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))", + "Row(string_bytes='vTwn4')", + "Row(stringsettq_bytes='798ka')", + "Row(stringsettq_bytes='798ka', from=\"2023-02-18\")", + "Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\")", + "Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))", + "Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")", + "Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))", + "Count(Row(idset_long=839))", + "Count(Row(id_long=809))", + "Count(Row(idtq_long=533))", + "Count(Row(idtq_long=533, from=\"2020-01-01\"))", + "Count(Row(idset_int=533))", + "Count(Row(id_int=168))", + "Count(Row(idsettq_int=113))", + "Row(idsettq_int=113, to=\"2024-01-01\")", + "Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))", + "Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))", + "Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))", + "Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))", + "Count(Row(int_long > 500))", + "Count(Row(int_int > 500))", + "Count(Row(decimal_bytes > 1000.00))", + "Count(Row(decimal_float > 3.05))", + "Count(Row(decimal_double > 4.11))", + "Count(Row(dateint_bytes_ts > 1675163490))", + "Count(Row(bools=bool_bool))", + "Count(Not(Row(bools=bool_bool)))", + "Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))", + "Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))", + }, + { + "Count(All())", + "Row(int_long=null)", + "Count(UnionRows(Rows(stringset_stringarray)))", + "Not(UnionRows(Rows(stringset_stringarray), Rows(string_string),Rows(stringset_bytes),Rows(string_bytes),Rows(stringset_stringarray),Rows(stringset_bytesarray),Rows(idset_long),Rows(id_long),Rows(idset_int),Rows(id_int),Rows(idset_longarray),Rows(idset_intarray)))", + "Row(int_int=null)", + "Row(decimal_bytes=null)", + "Row(decimal_float=null)", + "Count(Row(bools=bool_bool))", + "Count(Not(Row(bools-exists=bool_bool)))", + "Count(Row(dateint_bytes_ts=null))", + }, + { + "Count(All())", + }, + { + "Row(string_string=\"ZgkOB\")", + "Row(stringset_string =\"7EYSp\")", + "Count(Not(UnionRows(Rows(stringset_string))))", + "Intersect(Not(Intersect(Row(stringset_stringarray=\"u2Yr4\"),Row(stringset_stringarray=\"PYE8V\"), Row(stringset_stringarray=\"VBcyJ\"))), Intersect(Row(stringset_stringarray=\"Chgzr\"), Row(stringset_stringarray=\"DY2Ui\")))", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(idset_int)))))", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(id_int)))))", + "Row(int_int=969)", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(UnionRows(Rows(bools-exists)))))", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Row(decimal_double=null)))", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Not(Row(dateint_bytes_ts=null))))", + "Count(Intersect(ConstRow(columns=[\"u2Yr4|sHaUv|x5z8P\"]), Row(timestamp_bytes_int=null)))", + }, + }, + expectedResults: [][]string{ + { + "{\"results\":[10]}\n", + "{\"results\":[2]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[0]}\n", + "{\"results\":[1]}\n", + "{\"results\":[2]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"DY2Ui|kUbdU|pjxqm\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[{\"columns\":[]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"yg8hY|tvNOB|byHh9\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[4]}\n", + "{\"results\":[5]}\n", + "{\"results\":[4]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + "{\"results\":[8]}\n", + "{\"results\":[4]}\n", + "{\"results\":[6]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + }, + { + "{\"results\":[10]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[9]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"6TKzc|YKLk9|h1iqc\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"9z4aw|5ptDx|CKs1F\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"RKE3c|6TKzc|RKE3c\"]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"RKE3c|6TKzc|RKE3c\"]}]}\n", + "{\"results\":[3]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + }, + { + "{\"results\":[7]}\n", + }, + { + "{\"results\":[{\"columns\":[]}]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[],\"keys\":[\"u2Yr4|sHaUv|x5z8P\"]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + }, + }, + }, + { // confirm delete behavior when index is not keyed (note field level delete not supported with delete consumer on keys: false index) + name: "all values can be deleted with id keys", + pathsToAvroSchema: []string{ + "alltypes.json", + "alltypes_delete_records.json", + "alltypes_delete_value.json", + }, + pathsToRecords: []string{ + "./testdata/records/alltypes.json", + "./testdata/records/alltypes_delete_records_ids.json", + "./testdata/records/alltypes_delete_value_ids.json", + }, + pilosaHosts: pilosaHost, + registryURL: registryHost, + kafkaHost: kafkaHost, + consumerConfigs: []ConsumerTestConfig{ + { + idType: "generate", + topic: "alltypes", + delete: false, + }, + { + idType: "generate", + topic: "alltypes_delete_record_ids", + delete: true, + }, + { + idType: "generate", + topic: "alltypes_delete_values_ids", + delete: true, + }, + }, + index: "alltypes_delete_ids", + queries: [][]string{ + { + "Count(All())", + "Count(ConstRow(columns=[2, 3]))", + "Count(Row(stringset_string='58KIR'))", + "Count(Row(string_string='8MGwy'))", + "Count(Row(stringtq_string='ivWWb'))", + "Count(Row(stringtq_string='ivWWb', to=\"2023-02-03\"))", + "Count(Row(stringtq_string='ivWWb', from=\"2023-02-03\", to=\"2023-02-04\"))", + "Count(Union(Row(stringset_bytes='eNKWF'),Row(stringset_bytes='5ptDx')))", + "Count(Row(string_bytes='vTwn4'))", + "Count(Row(stringsettq_bytes='798ka'))", + "Count(Row(stringsettq_bytes='798ka', from=\"2023-02-18\"))", + "Count(Row(stringsettq_bytes='798ka', from=\"2023-02-16\", to=\"2023-02-18\"))", + "Intersect(Row(stringset_stringarray='u2Yr4'), Row(stringset_stringarray='PYE8V'), Row(stringset_stringarray='VBcyJ'), Row(stringset_stringarray='Chgzr'), Row(stringset_stringarray='DY2Ui'))", + "Row(stringtq_stringarray='oxjI0', from=\"2023-01-29\", to=\"2023-01-31\")", + "Intersect(Row(stringset_bytesarray='wNZ7o'), Row(stringset_bytesarray='OKNV2'),Row(stringset_bytesarray='F0uC4'),Row(stringset_bytesarray='VBcyJ'),Row(stringset_bytesarray='KMZnH'))", + "Count(Row(idset_long=839))", + "Count(Row(id_long=809))", + "Count(Row(idtq_long=533))", + "Count(Row(idtq_long=533, from=\"2020-01-01\"))", + "Count(Row(idset_int=533))", + "Count(Row(id_int=168))", + "Count(Row(idsettq_int=113))", + "Row(idsettq_int=113, to=\"2024-01-01\")", + "Count(Intersect(Row(idset_longarray=399),Row(idset_longarray=322), Row(idset_longarray=975), Row(idset_longarray=730), Row(idset_longarray=969)))", + "Count(Intersect(Row(idtq_longarray=172),Row(idtq_longarray=388), Row(idtq_longarray=731), Row(idtq_longarray=429), Row(idtq_longarray=730)))", + "Count(Intersect(Row(idtq_longarray=172, from=\"2022-01-01\"),Row(idtq_longarray=388, from=\"2022-01-01\"), Row(idtq_longarray=731, from=\"2022-01-01\"), Row(idtq_longarray=429, from=\"2022-01-01\"), Row(idtq_longarray=730, from=\"2022-01-01\")))", + "Count(Intersect(Row(idset_intarray=958),Row(idset_intarray=242), Row(idset_intarray=778), Row(idset_intarray=289), Row(idset_intarray=797)))", + "Count(Row(int_long > 500))", + "Count(Row(int_int > 500))", + "Count(Row(decimal_bytes > 1000.00))", + "Count(Row(decimal_float > 3.05))", + "Count(Row(decimal_double > 4.11))", + "Count(Row(dateint_bytes_ts > 1675163490))", + "Count(Row(bools=bool_bool))", + "Count(Not(Row(bools=bool_bool)))", + "Count(Row(timestamp_bytes_ts > \"2023-02-20T00:00:00Z\"))", + "Count(Row(timestamp_bytes_int > \"2023-02-20T00:00:00Z\"))", + }, + { + "Count(All())", + }, + { + "Row(string_string=\"ZgkOB\")", + "Row(stringset_string =\"7EYSp\")", + "Intersect(Not(Intersect(Row(stringset_stringarray=\"u2Yr4\"),Row(stringset_stringarray=\"PYE8V\"), Row(stringset_stringarray=\"VBcyJ\"))), Intersect(Row(stringset_stringarray=\"Chgzr\"), Row(stringset_stringarray=\"DY2Ui\")))", + "Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(idset_int)))))", + "Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(id_int)))))", + "Row(int_int=969)", + "Row(int_int=null)", + "Count(Intersect(ConstRow(columns=[10]), Not(UnionRows(Rows(bools-exists)))))", + "Count(Intersect(ConstRow(columns=[10]), Row(decimal_double=null)))", + "Count(Intersect(ConstRow(columns=[10]), Not(Row(dateint_bytes_ts=null))))", + "Count(Intersect(ConstRow(columns=[10]), Row(timestamp_bytes_int=null)))", + }, + }, + expectedResults: [][]string{ + { + "{\"results\":[10]}\n", + "{\"results\":[2]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[0]}\n", + "{\"results\":[1]}\n", + "{\"results\":[2]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[0]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[10]}]}\n", + "{\"results\":[{\"columns\":[6]}]}\n", + "{\"results\":[{\"columns\":[6]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[7]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[4]}\n", + "{\"results\":[5]}\n", + "{\"results\":[4]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + "{\"results\":[8]}\n", + "{\"results\":[4]}\n", + "{\"results\":[6]}\n", + "{\"results\":[3]}\n", + "{\"results\":[3]}\n", + }, + { + "{\"results\":[7]}\n", + }, + { + "{\"results\":[{\"columns\":[]}]}\n", + "{\"results\":[{\"columns\":[]}]}\n", + "{\"results\":[{\"columns\":[10]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[{\"columns\":[10]}]}\n", + "{\"results\":[{\"columns\":[8]}]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + "{\"results\":[1]}\n", + }, }, }, } for _, test := range tests { + // define some vars now := time.Now().UnixNano() index := fmt.Sprintf("%s_%d", test.index, now) - topic := fmt.Sprintf("%s_%d", test.topic, now) + //index := test.index - // read in records - var records []map[string]interface{} - var data map[string]interface{} - recordsFile, err := os.Open(test.pathToRecords) - if err != nil { - t.Errorf("opening records file") + // so topics of the same name end up being the same name after + // adding a timestamp to make sure they're unique + var topics map[string]string + for _, iteration := range test.consumerConfigs { + if topicName, ok := topics[iteration.topic]; ok { + iteration.topic = topicName + } else { + newTopicName := fmt.Sprintf("%s_%d", iteration.topic, now) + iteration.topic = newTopicName + + } } - defer recordsFile.Close() - s := bufio.NewScanner(recordsFile) - for s.Scan() { - err := json.Unmarshal(s.Bytes(), &data) + // iterate through tests (one per consumer config) + for i, iteration := range test.consumerConfigs { + // read in records + records, err := kafkaRecordFromFile(test.pathsToRecords[i]) if err != nil { - t.Errorf("unmarshal json: %s", err) + t.Errorf("%s", err) } - records = append(records, data) - data = make(map[string]interface{}) - } - // configure the consumer - consumer, err := NewMain() - if err != nil { - t.Fatalf("creating main %v", err) - } - configureTestFlags(consumer) - consumer.Index = index - consumer.Topics = []string{topic} - consumer.KafkaBootstrapServers = []string{test.kafkaHost} - consumer.SchemaRegistryURL = test.registryURL - switch test.idType { - case "id": - consumer.IDField = test.keyField - case "string": - consumer.PrimaryKeyFields = []string{test.keyField} - case "generate": - consumer.AutoGenerate = true - consumer.ExternalGenerate = true - default: - t.Errorf("incorrect idType supplied") - } - consumer.MaxMsgs = uint64(len(records)) + // load schema registry, create and produce to the topic + writeRecordsToKafka(t, test.pathsToAvroSchema[i], iteration.topic, test.registryURL, test.kafkaHost, records) - // load schema registry, create produce, topic and run consumer data - licodec := liDecodeTestSchema(t, test.pathToAvroSchema) - schemaID := postSchema(t, test.pathToAvroSchema, fmt.Sprintf("%s_id", topic), consumer.SchemaRegistryURL, nil) - p, err := confluent.NewProducer(&confluent.ConfigMap{ - "bootstrap.servers": kafkaHost, - }) - if err != nil { - t.Fatalf("Failed to create producer: %s", err) - } - defer p.Close() - tCreateTopic(t, topic, p) - tPutRecordsKafka(t, p, topic, schemaID, licodec, "akey", records...) - err = consumer.Run() - if err != nil { - t.Fatalf("running consumer: %v", err) - } - - // now run queries and confirm the data is as expected - client := consumer.PilosaClient() - for i, q := range test.queries { - status, body, err := client.HTTPRequest("POST", fmt.Sprintf("/index/%s/query", index), []byte(q), nil) + // configure the consumer + consumer, err := NewMain() if err != nil { - t.Fatalf("querying featurebase: status: %d, response: %s, error: %s", status, body, err) + t.Fatalf("creating main %v", err) } - if string(body[:]) != test.expectedResults[i] { - t.Fatalf("running query: %s... expected %s but got %s", q, test.expectedResults[i], body) + configureTestFlags(consumer) + consumer.Index = index + consumer.Topics = []string{iteration.topic} + consumer.KafkaBootstrapServers = []string{test.kafkaHost} + consumer.SchemaRegistryURL = test.registryURL + consumer.Delete = iteration.delete + switch iteration.idType { + case "id": + consumer.IDField = iteration.keyFields[0] + case "string": + consumer.PrimaryKeyFields = iteration.keyFields + case "generate": + consumer.AutoGenerate = true + consumer.ExternalGenerate = true + default: + t.Errorf("incorrect idType supplied") } + consumer.MaxMsgs = uint64(len(records)) + consumer.BatchSize = int(consumer.MaxMsgs) + pilosaHostsNew := strings.Split(test.pilosaHosts, ",") + consumer.PilosaHosts = pilosaHostsNew + //remove after flipping back to docker + host := strings.Split(pilosaHostsNew[0], ":") + consumer.PilosaGRPCHosts = []string{fmt.Sprintf("%s:20101", host[0])} + + // consumer records + err = consumer.Run() + if err != nil { + t.Fatalf("running consumer: %v", err) + } + + // now run queries and confirm the data is as expected + client := consumer.PilosaClient() + runTestQueries(t, index, test.queries[i], test.expectedResults[i], client) } } } +func runTestQueries(t *testing.T, index string, queries, expectedResults []string, client *pilosaclient.Client) { + for i, q := range queries { + status, body, err := client.HTTPRequest("POST", fmt.Sprintf("/index/%s/query", index), []byte(q), nil) + if err != nil { + t.Fatalf("querying featurebase: status: %d, response: %s, error: %s: on query: %s", status, body, err, q) + } + if string(body[:]) != expectedResults[i] { + t.Fatalf("running query %s against index %s. expected %s but got %s", q, index, expectedResults[i], body) + } + } +} + +func kafkaRecordFromFile(pathToRecords string) (records []map[string]interface{}, err error) { + var data map[string]interface{} + + recordsFile, err := os.Open(pathToRecords) + if err != nil { + return nil, fmt.Errorf("opening records file") + } + defer recordsFile.Close() + + s := bufio.NewScanner(recordsFile) + for s.Scan() { + err := json.Unmarshal(s.Bytes(), &data) + if err != nil { + return nil, fmt.Errorf("unmarshal json: %s", err) + } + records = append(records, data) + data = make(map[string]interface{}) + } + return records, nil +} + +func writeRecordsToKafka(t *testing.T, pathToAvroSchema, topic, schemaRegistryURL, kafkaHost string, records []map[string]interface{}) { + licodec := liDecodeTestSchema(t, pathToAvroSchema) + schemaID := postSchema(t, pathToAvroSchema, fmt.Sprintf("%s_id", topic), schemaRegistryURL, nil) + p, err := confluent.NewProducer(&confluent.ConfigMap{ + "bootstrap.servers": kafkaHost, + }) + if err != nil { + t.Fatalf("Failed to create producer: %s", err) + } + defer p.Close() + tCreateTopic(t, topic, p) + + tPutRecordsKafka(t, p, topic, schemaID, licodec, "akey", records...) +} + type sortableCRI []pilosaclient.CountResultItem func (s sortableCRI) Len() int { return len(s) } diff --git a/idk/kafka/source.go b/idk/kafka/source.go index 003c8d76f..2c1a848dd 100644 --- a/idk/kafka/source.go +++ b/idk/kafka/source.go @@ -109,7 +109,7 @@ func (s *Source) Record() (idk.Record, error) { return nil, idk.ErrFlush } - val, err := s.decodeAvroValueWithSchemaRegistry(rec.Record.Value) + val, avroSchema, err := s.decodeAvroValueWithSchemaRegistry(rec.Record.Value) if err != nil && err != idk.ErrSchemaChange { return nil, errors.Wrap(err, "decoding with schema registry") } @@ -124,12 +124,13 @@ func (s *Source) Record() (idk.Record, error) { defer s.mu.Unlock() s.spool = append(s.spool, msg.TopicPartition) return &Record{ - src: s, - topic: *msg.TopicPartition.Topic, - partition: int(msg.TopicPartition.Partition), - offset: int64(msg.TopicPartition.Offset), - idx: s.spoolBase + uint64(len(s.spool)), - data: data, + src: s, + topic: *msg.TopicPartition.Topic, + partition: int(msg.TopicPartition.Partition), + offset: int64(msg.TopicPartition.Offset), + idx: s.spoolBase + uint64(len(s.spool)), + data: data, + avroSchema: avroSchema, }, err } @@ -189,12 +190,13 @@ func (s *Source) toPDKRecord(vals map[string]interface{}) []interface{} { } type Record struct { - src *Source - topic string - partition int - offset int64 - idx uint64 - data []interface{} + src *Source + topic string + partition int + offset int64 + idx uint64 + data []interface{} + avroSchema avro.Schema } func (r *Record) StreamOffset() (string, uint64) { @@ -254,6 +256,10 @@ func (r *Record) Data() []interface{} { return r.data } +func (r *Record) Schema() interface{} { + return r.avroSchema +} + func (s *Source) CommitMessages(recs []confluent.TopicPartition) ([]confluent.TopicPartition, error) { return s.client.CommitOffsets(recs) } @@ -469,29 +475,29 @@ func (s *Source) Close() error { } // TODO change name -func (s *Source) decodeAvroValueWithSchemaRegistry(val []byte) (interface{}, error) { +func (s *Source) decodeAvroValueWithSchemaRegistry(val []byte) (interface{}, avro.Schema, error) { if len(val) < 6 || val[0] != 0 { - return nil, errors.Errorf("unexpected magic byte or length in avro kafka value, should be 0x00, but got %x", val) + return nil, nil, errors.Errorf("unexpected magic byte or length in avro kafka value, should be 0x00, but got %x", val) } id := int32(binary.BigEndian.Uint32(val[1:])) codec, err := s.getCodec(id) if err != nil { - return nil, errors.Wrap(err, "getting avro codec") + return nil, nil, errors.Wrap(err, "getting avro codec") } ret, err := avroDecode(codec, val[5:]) if err != nil { - return nil, errors.Wrap(err, "decoding avro record") + return nil, codec, errors.Wrap(err, "decoding avro record") } if id != s.lastSchemaID { s.lastSchema, err = avroToPDKSchema(codec) if err != nil { - return nil, errors.Wrap(err, "converting to FeatureBase schema") + return nil, codec, errors.Wrap(err, "converting to FeatureBase schema") } s.lastSchemaID = id - return ret, idk.ErrSchemaChange + return ret, codec, idk.ErrSchemaChange } - return ret, nil + return ret, codec, nil } // avroToPDKSchema converts a full avro schema to the much more @@ -564,13 +570,9 @@ func avroToPDKField(aField *avro.SchemaField) (idk.Field, error) { switch ft { case "decimal": - precision, _ := intProp(aField, "precision") - if precision > 18 || precision < 1 { - return nil, errors.Errorf("need precision for decimal in 1-18, but got:%d", precision) - } scale, err := intProp(aField, "scale") - if scale > precision || err == wrongType { - return nil, errors.Errorf("0<=scale<=precision, got:%d err:%v", scale, err) + if scale > 18 || err == wrongType { + return nil, errors.Errorf("0<=scale<=18, got:%d err:%v", scale, err) } return idk.DecimalField{ NameVal: aField.Name, @@ -697,7 +699,7 @@ func avroToPDKField(aField *avro.SchemaField) (idk.Field, error) { CacheConfig: cacheConfig, }, nil - case avro.Long: + case avro.Long, avro.Int: if ft, _ := stringProp(itemSchema, "fieldType"); ft == "decimal" { return nil, errors.New("arrays of decimal are not supported") } diff --git a/idk/kafka/source_test.go b/idk/kafka/source_test.go index e695cba11..d117b83e9 100644 --- a/idk/kafka/source_test.go +++ b/idk/kafka/source_test.go @@ -84,7 +84,7 @@ func TestAvroToPDKSchema(t *testing.T) { if err != nil { t.Fatalf("reading directory: %v", err) } - if len(files) != len(tests)+4 { // +4 because we aren't testing bigschema.json, the two delete ones, or the ID allocation one here. + if len(files) != len(tests)+9 { // +9 because we aren't testing bigschema.json, the five delete ones, alltypes. timeQuantums or the ID allocation one here. t.Errorf("have different number of schemas and tests: %d and %d\n%+v", len(files), len(tests), files) } diff --git a/idk/kafka/testdata/records/alltypes.json b/idk/kafka/testdata/records/alltypes.json new file mode 100644 index 000000000..a38149d38 --- /dev/null +++ b/idk/kafka/testdata/records/alltypes.json @@ -0,0 +1,10 @@ +{"pk0": "9z4aw", "pk1": "5ptDx", "pk2": "CKs1F", "stringset_string": {"string": "7EYSp"}, "string_string": {"string": "uirDR"}, "stringtq_string": {"string": "Qylqq"}, "stringset_bytes": {"bytes": "gL2Hg"}, "string_bytes": {"bytes": "BmvHF"}, "stringsettq_bytes": {"bytes": "798ka"}, "stringset_stringarray": {"array": ["vbbuf", "VQs7y", "9z4aw", "h1iqc", "aQQxr"]}, "stringtq_stringarray": {"array": ["x5z8P", "0UGJQ", "58KIR", "7EYSp", "CKs1F"]}, "stringset_bytesarray": {"array": ["u2Yr4", "tvNOB", "iYeOV", "ZgkOB", "RPGAm"]}, "stringtq_bytesarray": {"array": ["BwqU2", "6iGIm", "fjQK2", "LBTEU", "C6xxn"]}, "idset_long": {"long": 647}, "id_long": {"long": 792}, "idtq_long": {"long": 676}, "idset_int": {"int": 898}, "id_int": {"int": 63}, "idsettq_int": {"int": 890}, "idset_longarray": {"array": [442, 167, 230, 344, 733]}, "idtq_longarray": {"array": [385, 931, 157, 865, 394]}, "idset_intarray": {"array": [442, 614, 394, 284, 344]}, "int_long": {"long": 584}, "int_int": {"int": 344}, "decimal_bytes": {"bytes": "\u0001\u00cb"}, "decimal_float": {"float": 3.23}, "decimal_double": {"double": 0.95}, "dateint_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "recordtime_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "recordtime_bytes_int": {"bytes": "2023-02-16 07:53:59"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "timestamp_bytes_int": {"bytes": "1676555639"}} +{"pk0": "ASSAw", "pk1": "kauLy", "pk2": "oxjI0", "stringset_string": {"string": "iYeOV"}, "string_string": {"string": "LBTEU"}, "stringtq_string": {"string": "gDmCj"}, "stringset_bytes": {"bytes": "5ptDx"}, "string_bytes": {"bytes": "EyQoi"}, "stringsettq_bytes": {"bytes": "6TKzc"}, "stringset_stringarray": {"array": ["iYeOV", "XzEHj", "rrkYB", "v31XN", "uirDR"]}, "stringtq_stringarray": {"array": ["ARlcJ", "58KIR", "i0tva", "u2Yr4", "vhisL"]}, "stringset_bytesarray": {"array": ["X9jWC", "x5z8P", "PYE8V", "PYE8V", "vTwn4"]}, "stringtq_bytesarray": {"array": ["tvNOB", "VBcyJ", "d0U7s", "i0tva", "vbbuf"]}, "idset_long": {"long": 484}, "id_long": {"long": 23}, "idtq_long": {"long": 284}, "idset_int": {"int": 322}, "id_int": {"int": 320}, "idsettq_int": {"int": 175}, "idset_longarray": {"array": [792, 809, 168, 399, 639]}, "idtq_longarray": {"array": [100, 931, 584, 85, 388]}, "idset_intarray": {"array": [606, 293, 23, 358, 821]}, "int_long": {"long": 533}, "int_int": {"int": 884}, "decimal_bytes": {"bytes": "\u0001\u0029"}, "decimal_float": {"float": 4.32}, "decimal_double": {"double": 4.97}, "dateint_bytes_ts": {"bytes": "2023-02-22 14:32:23"}, "recordtime_bytes_ts": {"bytes": "2023-02-22 14:32:23"}, "recordtime_bytes_int": {"bytes": "2023-02-22 14:32:23"}, "bool_bool": {"boolean": true}, "timestamp_bytes_ts": {"bytes": "2023-02-22 14:32:23"}, "timestamp_bytes_int": {"bytes": "1677097943"}} +{"pk0": "BmvHF", "pk1": "798ka", "pk2": "6TKzc", "stringset_string": {"string": "X9jWC"}, "string_string": {"string": "I1gXJ"}, "stringtq_string": {"string": "C6xxn"}, "stringset_bytes": {"bytes": "gjWEI"}, "string_bytes": {"bytes": "thuky"}, "stringsettq_bytes": {"bytes": "6iGIm"}, "stringset_stringarray": {"array": ["tyP3m", "5ptDx", "TLaUE", "EyQoi", "Chgzr"]}, "stringtq_stringarray": {"array": ["oxjI0", "VBcyJ", "jYw4E", "aQQxr", "ivWWb"]}, "stringset_bytesarray": {"array": ["VQs7y", "h1iqc", "F0uC4", "d0U7s", "byHh9"]}, "stringtq_bytesarray": {"array": ["XzEHj", "tvNOB", "I1gXJ", "ivWWb", "CKs1F"]}, "idset_long": {"long": 166}, "id_long": {"long": 320}, "idtq_long": {"long": 792}, "idset_int": {"int": 232}, "id_int": {"int": 286}, "idsettq_int": {"int": 168}, "idset_longarray": {"array": [890, 975, 284, 289, 388]}, "idtq_longarray": {"array": [168, 635, 584, 614, 778]}, "idset_intarray": {"array": [865, 931, 614, 884, 322]}, "int_long": {"long": 857}, "int_int": {"int": 879}, "decimal_bytes": {"bytes": "\u0000\u00e6"}, "decimal_float": {"float": 2.84}, "decimal_double": {"double": 2.91}, "dateint_bytes_ts": {"bytes": "2023-01-31 11:11:30"}, "recordtime_bytes_ts": {"bytes": "2023-01-31 11:11:30"}, "recordtime_bytes_int": {"bytes": "2023-01-31 11:11:30"}, "bool_bool": {"boolean": true}, "timestamp_bytes_ts": {"bytes": "2023-01-31 11:11:30"}, "timestamp_bytes_int": {"bytes": "1675185090"}} +{"pk0": "tElMR", "pk1": "ARlcJ", "pk2": "n9HUP", "stringset_string": {"string": "58KIR"}, "string_string": {"string": "FW39I"}, "stringtq_string": {"string": "LBTEU"}, "stringset_bytes": {"bytes": "PNB4s"}, "string_bytes": {"bytes": "FW39I"}, "stringsettq_bytes": {"bytes": "d0U7s"}, "stringset_stringarray": {"array": ["X9jWC", "58KIR", "X9jWC", "6TKzc", "8MGwy"]}, "stringtq_stringarray": {"array": ["VQs7y", "MVNow", "Chgzr", "DDLN5", "ARlcJ"]}, "stringset_bytesarray": {"array": ["vhisL", "BmvHF", "eofzb", "TLaUE", "PNB4s"]}, "stringtq_bytesarray": {"array": ["TLaUE", "uirDR", "X9jWC", "eNKWF", "TLaUE"]}, "idset_long": {"long": 289}, "id_long": {"long": 695}, "idtq_long": {"long": 284}, "idset_int": {"int": 791}, "id_int": {"int": 821}, "idsettq_int": {"int": 733}, "idset_longarray": {"array": [2, 680, 958, 289, 389]}, "idtq_longarray": {"array": [296, 871, 453, 39, 387]}, "idset_intarray": {"array": [606, 890, 387, 102, 220]}, "int_long": {"long": 289}, "int_int": {"int": 2}, "decimal_bytes": {"bytes": "\u005f"}, "decimal_float": {"float": 2.65}, "decimal_double": {"double": 2.19}, "dateint_bytes_ts": {"bytes": "2023-02-20 17:04:21"}, "recordtime_bytes_ts": {"bytes": "2023-02-20 17:04:21"}, "recordtime_bytes_int": {"bytes": "2023-02-20 17:04:21"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-02-20 17:04:21"}, "timestamp_bytes_int": {"bytes": "1676934261"}} +{"pk0": "RKE3c", "pk1": "6TKzc", "pk2": "RKE3c", "stringset_string": {"string": "dF6kx"}, "string_string": {"string": "TLaUE"}, "stringtq_string": {"string": "m5d59"}, "stringset_bytes": {"bytes": "dxKKn"}, "string_bytes": {"bytes": "YdwQY"}, "stringsettq_bytes": {"bytes": "y2Y7b"}, "stringset_stringarray": {"array": ["5HIn2", "wNZ7o", "KdTtE", "x5z8P", "nVQrd"]}, "stringtq_stringarray": {"array": ["dF6kx", "pjxqm", "U123X", "n9HUP", "QwkWO"]}, "stringset_bytesarray": {"array": ["5HIn2", "tyP3m", "I6NST", "gjWEI", "Qylqq"]}, "stringtq_bytesarray": {"array": ["rrkYB", "WwTyQ", "gL2Hg", "0UGJQ", "LBTEU"]}, "idset_long": {"long": 39}, "id_long": {"long": 665}, "idtq_long": {"long": 821}, "idset_int": {"int": 113}, "id_int": {"int": 681}, "idsettq_int": {"int": 975}, "idset_longarray": {"array": [857, 63, 172, 220, 358]}, "idtq_longarray": {"array": [635, 931, 113, 969, 996]}, "idset_intarray": {"array": [220, 731, 647, 778, 665]}, "int_long": {"long": 582}, "int_int": {"int": 690}, "decimal_bytes": {"bytes": "\u0001\u00d1"}, "decimal_float": {"float": 1.36}, "decimal_double": {"double": 2.72}, "dateint_bytes_ts": {"bytes": "2023-02-16 20:09:13"}, "recordtime_bytes_ts": {"bytes": "2023-02-16 20:09:13"}, "recordtime_bytes_int": {"bytes": "2023-02-16 20:09:13"}, "bool_bool": {"boolean": true}, "timestamp_bytes_ts": {"bytes": "2023-02-16 20:09:13"}, "timestamp_bytes_int": {"bytes": "1676599753"}} +{"pk0": "yg8hY", "pk1": "tvNOB", "pk2": "byHh9", "stringset_string": {"string": "911oj"}, "string_string": {"string": "5HIn2"}, "stringtq_string": {"string": "gL2Hg"}, "stringset_bytes": {"bytes": "u2Yr4"}, "string_bytes": {"bytes": "qK5TE"}, "stringsettq_bytes": {"bytes": "MVNow"}, "stringset_stringarray": {"array": ["nVQrd", "fjQK2", "m5d59", "dxKKn", "d0U7s"]}, "stringtq_stringarray": {"array": ["oxjI0", "QwkWO", "rrkYB", "OKNV2", "XzEHj"]}, "stringset_bytesarray": {"array": ["wNZ7o", "OKNV2", "F0uC4", "VBcyJ", "KMZnH"]}, "stringtq_bytesarray": {"array": ["5ptDx", "tElMR", "DY2Ui", "byHh9", "Qylqq"]}, "idset_long": {"long": 839}, "id_long": {"long": 809}, "idtq_long": {"long": 533}, "idset_int": {"int": 533}, "id_int": {"int": 168}, "idsettq_int": {"int": 809}, "idset_longarray": {"array": [582, 629, 680, 63, 690]}, "idtq_longarray": {"array": [115, 257, 582, 242, 975]}, "idset_intarray": {"array": [969, 175, 172, 257, 115]}, "int_long": {"long": 433}, "int_int": {"int": 680}, "decimal_bytes": {"bytes": "\u0001\u00d6"}, "decimal_float": {"float": 1.27}, "decimal_double": {"double": 3.23}, "dateint_bytes_ts": {"bytes": "2023-01-30 06:56:05"}, "recordtime_bytes_ts": {"bytes": "2023-01-30 06:56:05"}, "recordtime_bytes_int": {"bytes": "2023-01-30 06:56:05"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-01-30 06:56:05"}, "timestamp_bytes_int": {"bytes": "1675083365"}} +{"pk0": "6TKzc", "pk1": "YKLk9", "pk2": "h1iqc", "stringset_string": {"string": "eofzb"}, "string_string": {"string": "n9HUP"}, "stringtq_string": {"string": "pjxqm"}, "stringset_bytes": {"bytes": "t5f7R"}, "string_bytes": {"bytes": "5HIn2"}, "stringsettq_bytes": {"bytes": "DDLN5"}, "stringset_stringarray": {"array": ["yg8hY", "xE5jX", "C6xxn", "BmvHF", "PYE8V"]}, "stringtq_stringarray": {"array": ["eNKWF", "7EYSp", "LBTEU", "jYw4E", "6TKzc"]}, "stringset_bytesarray": {"array": ["6TKzc", "vK0WD", "xE5jX", "jVVfZ", "pjxqm"]}, "stringtq_bytesarray": {"array": ["gL2Hg", "KMZnH", "4uK62", "6iGIm", "rHXM9"]}, "idset_long": {"long": 72}, "id_long": {"long": 387}, "idtq_long": {"long": 157}, "idset_int": {"int": 676}, "id_int": {"int": 797}, "idsettq_int": {"int": 113}, "idset_longarray": {"array": [399, 322, 975, 730, 969]}, "idtq_longarray": {"array": [172, 388, 731, 429, 730]}, "idset_intarray": {"array": [958, 242, 778, 289, 797]}, "int_long": {"long": 430}, "int_int": {"int": 23}, "decimal_bytes": {"bytes": "\u0001\u00d6"}, "decimal_float": {"float": 3.35}, "decimal_double": {"double": 0.78}, "dateint_bytes_ts": {"bytes": "2023-02-23 05:04:34"}, "recordtime_bytes_ts": {"bytes": "2023-02-23 05:04:34"}, "recordtime_bytes_int": {"bytes": "2023-02-23 05:04:34"}, "bool_bool": {"boolean": true}, "timestamp_bytes_ts": {"bytes": "2023-02-23 05:04:34"}, "timestamp_bytes_int": {"bytes": "1677150274"}} +{"pk0": "h1iqc", "pk1": "5ptDx", "pk2": "iYeOV", "stringset_string": {"string": "ASSAw"}, "string_string": {"string": "58KIR"}, "stringtq_string": {"string": "aQQxr"}, "stringset_bytes": {"bytes": "eNKWF"}, "string_bytes": {"bytes": "x5z8P"}, "stringsettq_bytes": {"bytes": "XzEHj"}, "stringset_stringarray": {"array": ["pjxqm", "6TKzc", "ZgkOB", "eofzb", "RKE3c"]}, "stringtq_stringarray": {"array": ["PYE8V", "ARlcJ", "798ka", "PYE8V", "qK5TE"]}, "stringset_bytesarray": {"array": ["BmvHF", "Qylqq", "5HIn2", "7EYSp", "yTeUQ"]}, "stringtq_bytesarray": {"array": ["gDmCj", "vK0WD", "4uK62", "v31XN", "TOWar"]}, "idset_long": {"long": 255}, "id_long": {"long": 647}, "idtq_long": {"long": 695}, "idset_int": {"int": 821}, "id_int": {"int": 389}, "idsettq_int": {"int": 694}, "idset_longarray": {"array": [606, 110, 320, 63, 344]}, "idtq_longarray": {"array": [958, 230, 751, 430, 113]}, "idset_intarray": {"array": [29, 289, 388, 257, 606]}, "int_long": {"long": 110}, "int_int": {"int": 148}, "decimal_bytes": {"bytes": "\u0001\u001c"}, "decimal_float": {"float": 0.4}, "decimal_double": {"double": 4.41}, "dateint_bytes_ts": {"bytes": "2023-02-19 08:52:56"}, "recordtime_bytes_ts": {"bytes": "2023-02-19 08:52:56"}, "recordtime_bytes_int": {"bytes": "2023-02-19 08:52:56"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-02-19 08:52:56"}, "timestamp_bytes_int": {"bytes": "1676818376"}} +{"pk0": "DY2Ui", "pk1": "kUbdU", "pk2": "pjxqm", "stringset_string": {"string": "tyP3m"}, "string_string": {"string": "8MGwy"}, "stringtq_string": {"string": "rHXM9"}, "stringset_bytes": {"bytes": "DDLN5"}, "string_bytes": {"bytes": "vTwn4"}, "stringsettq_bytes": {"bytes": "gDmCj"}, "stringset_stringarray": {"array": ["XzEHj", "8MGwy", "gjWEI", "xE5jX", "v31XN"]}, "stringtq_stringarray": {"array": ["RPGAm", "TOWar", "dxKKn", "F0uC4", "sHaUv"]}, "stringset_bytesarray": {"array": ["d0U7s", "u2Yr4", "d0U7s", "sDdtS", "y2Y7b"]}, "stringtq_bytesarray": {"array": ["n9HUP", "jYw4E", "fjQK2", "QwkWO", "n9HUP"]}, "idset_long": {"long": 984}, "id_long": {"long": 430}, "idtq_long": {"long": 695}, "idset_int": {"int": 931}, "id_int": {"int": 297}, "idsettq_int": {"int": 582}, "idset_longarray": {"array": [975, 733, 113, 751, 772]}, "idtq_longarray": {"array": [167, 242, 731, 585, 142]}, "idset_intarray": {"array": [297, 72, 694, 898, 384]}, "int_long": {"long": 63}, "int_int": {"int": 388}, "decimal_bytes": {"bytes": "\u007e"}, "decimal_float": {"float": 0.83}, "decimal_double": {"double": 4.23}, "dateint_bytes_ts": {"bytes": "2023-02-12 18:37:16"}, "recordtime_bytes_ts": {"bytes": "2023-02-12 18:37:16"}, "recordtime_bytes_int": {"bytes": "2023-02-12 18:37:16"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-02-12 18:37:16"}, "timestamp_bytes_int": {"bytes": "1676248636"}} +{"pk0": "u2Yr4", "pk1": "sHaUv", "pk2": "x5z8P", "stringset_string": {"string": "7EYSp"}, "string_string": {"string": "ZgkOB"}, "stringtq_string": {"string": "ivWWb"}, "stringset_bytes": {"bytes": "qK5TE"}, "string_bytes": {"bytes": "6iGIm"}, "stringsettq_bytes": {"bytes": "4uK62"}, "stringset_stringarray": {"array": ["u2Yr4", "PYE8V", "VBcyJ", "Chgzr", "DY2Ui"]}, "stringtq_stringarray": {"array": ["KdTtE", "I1gXJ", "ARlcJ", "ZgkOB", "wNZ7o"]}, "stringset_bytesarray": {"array": ["YdwQY", "kUbdU", "aQQxr", "KdTtE", "MVNow"]}, "stringtq_bytesarray": {"array": ["ARlcJ", "WwTyQ", "thuky", "v31XN", "798ka"]}, "idset_long": {"long": 148}, "id_long": {"long": 115}, "idtq_long": {"long": 100}, "idset_int": {"int": 890}, "id_int": {"int": 39}, "idsettq_int": {"int": 606}, "idset_longarray": {"array": [839, 63, 148, 984, 958]}, "idtq_longarray": {"array": [730, 320, 994, 167, 791]}, "idset_intarray": {"array": [731, 13, 167, 772, 629]}, "int_long": {"long": 13}, "int_int": {"int": 969}, "decimal_bytes": {"bytes": "\u0000\u009a"}, "decimal_float": {"float": 2.93}, "decimal_double": {"double": 2.29}, "dateint_bytes_ts": {"bytes": "2023-02-03 16:19:37"}, "recordtime_bytes_ts": {"bytes": "2023-02-03 16:19:37"}, "recordtime_bytes_int": {"bytes": "2023-02-03 16:19:37"}, "bool_bool": {"boolean": false}, "timestamp_bytes_ts": {"bytes": "2023-02-03 16:19:37"}, "timestamp_bytes_int": {"bytes": "1675462777"}} \ No newline at end of file diff --git a/idk/kafka/testdata/records/alltypes_delete_fields.json b/idk/kafka/testdata/records/alltypes_delete_fields.json new file mode 100644 index 000000000..5e67b60b8 --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_delete_fields.json @@ -0,0 +1,5 @@ +{"pk0": "9z4aw", "pk1": "5ptDx", "pk2": "CKs1F", "fields": ["int_long","int_int"]} +{"pk0": "6TKzc", "pk1": "YKLk9", "pk2": "h1iqc", "fields": ["stringset_string","string_string","stringset_bytes","string_bytes","stringset_stringarray","stringset_bytesarray","idset_long","id_long","idset_int","id_int","idset_longarray","idset_intarray"]} +{"pk0": "RKE3c", "pk1": "6TKzc", "pk2": "RKE3c", "fields": ["decimal_bytes","decimal_float","decimal_double"]} +{"pk0": "ASSAw", "pk1": "kauLy", "pk2": "oxjI0", "fields": ["bools","bools-exists"]} +{"pk0": "yg8hY", "pk1": "tvNOB", "pk2": "byHh9", "fields": ["dateint_bytes_ts"]} \ No newline at end of file diff --git a/idk/kafka/testdata/records/alltypes_delete_records.json b/idk/kafka/testdata/records/alltypes_delete_records.json new file mode 100644 index 000000000..dd685ee2d --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_delete_records.json @@ -0,0 +1,2 @@ +{"ids": {"null": null}, "keys": {"array": ["9z4aw|5ptDx|CKs1F", "ASSAw|kauLy|oxjI0"]}, "filter": {"null": null}} +{"ids": {"null": null}, "keys": {"null": null}, "filter": {"string": "Row(stringset_string='58KIR')"}} diff --git a/idk/kafka/testdata/records/alltypes_delete_records_ids.json b/idk/kafka/testdata/records/alltypes_delete_records_ids.json new file mode 100644 index 000000000..8358914c3 --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_delete_records_ids.json @@ -0,0 +1,2 @@ +{"ids": {"array": [1,2]}, "keys": {"null": null}, "filter": {"null": null}} +{"ids": {"null": null}, "keys": {"null": null}, "filter": {"string": "Row(stringset_string='58KIR')"}} diff --git a/idk/kafka/testdata/records/alltypes_delete_value.json b/idk/kafka/testdata/records/alltypes_delete_value.json new file mode 100644 index 000000000..4934554d6 --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_delete_value.json @@ -0,0 +1,2 @@ +{"_id": {"string": "u2Yr4|sHaUv|x5z8P"}, "stringset_string": {"null": null}, "string_string": {"string": "ZgkOB"}, "stringset_stringarray": {"array": ["u2Yr4", "PYE8V", "VBcyJ"]}, "idset_int": {"int": 890}, "id_int": {"int": 39}, "idset_intarray": {"array": [731, 13]}, "int_int": false, "bools": {"array": ["bool_bool"]}, "decimal_double": true, "dateint_bytes_ts": false, "timestamp_bytes_int": true} +{"_id": {"string": "h1iqc|5ptDx|iYeOV"}, "stringset_string": {"null": null}, "string_string": {"null": null}, "stringset_stringarray": {"null": null}, "idset_int": {"null": null}, "id_int": {"null": null}, "idset_intarray": {"null": null}, "int_int": true, "decimal_double": true, "bools": {"null": null}, "dateint_bytes_ts": true, "timestamp_bytes_int": true} \ No newline at end of file diff --git a/idk/kafka/testdata/records/alltypes_delete_value_ids.json b/idk/kafka/testdata/records/alltypes_delete_value_ids.json new file mode 100644 index 000000000..ef5f993cb --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_delete_value_ids.json @@ -0,0 +1,2 @@ +{"_id": {"int": 10}, "stringset_string": {"string": "7EYSp"}, "string_string": {"string": "ZgkOB"}, "stringset_stringarray": {"array": ["u2Yr4", "PYE8V", "VBcyJ"]}, "idset_int": {"int": 890}, "id_int": {"int": 39}, "idset_intarray": {"array": [731, 13]}, "int_int": false, "bools": {"array": ["bool_bool"]}, "decimal_double": true, "dateint_bytes_ts": false, "timestamp_bytes_int": true} +{"_id": {"int": 8}, "stringset_string": {"null": null}, "string_string": {"null": null}, "stringset_stringarray": {"null": null}, "idset_int": {"null": null}, "id_int": {"null": null}, "idset_intarray": {"null": null}, "int_int": true, "decimal_double": true, "bools": {"null": null}, "dateint_bytes_ts": true, "timestamp_bytes_int": true} \ No newline at end of file diff --git a/idk/kafka/testdata/records/alltypes_null.json b/idk/kafka/testdata/records/alltypes_null.json new file mode 100644 index 000000000..8ea415466 --- /dev/null +++ b/idk/kafka/testdata/records/alltypes_null.json @@ -0,0 +1,2 @@ +{"pk0": "9z4aw", "pk1": "5ptDx", "pk2": "CKs1F", "stringset_string": {"string": "7EYSp"}, "string_string": {"string": "uirDR"}, "stringtq_string": {"string": "Qylqq"}, "stringset_bytes": {"bytes": "gL2Hg"}, "string_bytes": {"bytes": "BmvHF"}, "stringsettq_bytes": {"bytes": "798ka"}, "stringset_stringarray": {"array": ["vbbuf", "VQs7y", "9z4aw", "h1iqc", "aQQxr"]}, "stringtq_stringarray": {"array": ["x5z8P", "0UGJQ", "58KIR", "7EYSp", "CKs1F"]}, "stringset_bytesarray": {"array": ["u2Yr4", "tvNOB", "iYeOV", "ZgkOB", "RPGAm"]}, "stringtq_bytesarray": {"array": ["BwqU2", "6iGIm", "fjQK2", "LBTEU", "C6xxn"]}, "idset_long": {"long": 647}, "id_long": {"long": 792}, "idtq_long": {"long": 676}, "idset_int": {"int": 898}, "id_int": {"int": 63}, "idsettq_int": {"int": 890}, "idset_longarray": {"array": [442, 167, 230, 344, 733]}, "idtq_longarray": {"array": [385, 931, 157, 865, 394]}, "idset_intarray": {"array": [442, 614, 394, 284, 344]}, "int_long": {"long": 584}, "int_int": {"int": 344}, "decimal_bytes": {"bytes": "\u0001\u00cb"}, "decimal_float": {"float": 3.23}, "decimal_double": {"double": 0.95}, "dateint_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "recordtime_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "recordtime_bytes_int": {"bytes": "2023-02-16 07:53:59"}, "bool_bool": {"boolean": true}, "timestamp_bytes_ts": {"bytes": "2023-02-16 07:53:59"}, "timestamp_bytes_int": {"bytes": "1676555639"}} +{"pk0": "9z4aw", "pk1": "5ptDx", "pk2": "CKs1F", "stringset_string": {"null": null}, "string_string": {"null": null}, "stringtq_string": {"null": null}, "stringset_bytes": {"null": null}, "string_bytes": {"null": null}, "stringsettq_bytes": {"null": null}, "stringset_stringarray": {"null": null}, "stringtq_stringarray": {"null": null}, "stringset_bytesarray": {"null": null}, "stringtq_bytesarray": {"null": null}, "idset_long": {"null": null}, "id_long": {"null": null}, "idtq_long": {"null": null}, "idset_int": {"null": null}, "id_int": {"null": null}, "idsettq_int": {"null": null}, "idset_longarray": {"null": null}, "idtq_longarray": {"null": null}, "idset_intarray": {"null": null}, "int_long": {"null": null}, "int_int": {"null": null}, "decimal_bytes": {"null": null}, "decimal_float": {"null": null}, "decimal_double": {"null": null}, "dateint_bytes_ts": {"null": null}, "recordtime_bytes_ts": {"null": null}, "recordtime_bytes_int": {"null": null}, "bool_bool": {"null": null}, "timestamp_bytes_ts": {"null": null}, "timestamp_bytes_int": {"null": null}} diff --git a/idk/kafka/testdata/schemas/alltypes.json b/idk/kafka/testdata/schemas/alltypes.json new file mode 100644 index 000000000..8d3f914bf --- /dev/null +++ b/idk/kafka/testdata/schemas/alltypes.json @@ -0,0 +1,41 @@ +{ + "namespace": "org.test", + "type": "record", + "name": "all_type_schema", + "doc": "All supported avro types and property variations", + "fields": [ + {"name": "pk0", "type": "string"}, + {"name": "pk1", "type": "string"}, + {"name": "pk2", "type": "string"}, + {"name": "stringset_string", "type": ["string", "null"], "mutex": false }, + {"name": "string_string", "type": ["string", "null"], "mutex": true }, + {"name": "stringtq_string","type": ["string", "null"], "quantum": "YMD"}, + {"name": "stringset_bytes", "type": ["bytes", "null"], "mutex": false}, + {"name": "string_bytes", "type": ["bytes", "null"] , "mutex": true }, + {"name": "stringsettq_bytes", "type": ["bytes", "null"], "quantum": "YMD"}, + {"name": "stringset_stringarray", "type": [{"type": "array", "items": "string"}, "null"]}, + {"name": "stringtq_stringarray", "type": [{"type": "array", "items": {"type": "string", "quantum": "YMD"}}, "null"]}, + {"name": "stringset_bytesarray", "type": [{"type": "array", "items": "string"}, "null"]}, + {"name": "stringtq_bytesarray", "type": [{"type": "array", "items": {"type": "bytes", "quantum": "YMD"}}, "null"]}, + {"name": "idset_long", "type": ["long", "null"], "mutex": false, "fieldType": "id"}, + {"name": "id_long", "type": ["long", "null"], "mutex": true, "fieldType": "id"}, + {"name": "idtq_long", "type": ["long", "null"], "quantum": "YMD", "fieldType": "id"}, + {"name": "idset_int", "type": ["int", "null"], "mutex": false, "fieldType": "id"}, + {"name": "id_int", "type": ["int", "null"], "mutex": true, "fieldType": "id"}, + {"name": "idsettq_int", "type": ["int", "null"], "quantum": "YMD", "fieldType": "id"}, + {"name": "idset_longarray", "type": [{"type": "array", "items": "long"}, "null"], "fieldType": "id"}, + {"name": "idtq_longarray", "type": [{"type": "array", "items": {"type": "long", "quantum": "YMD"}}, "null"], "fieldType": "id"}, + {"name": "idset_intarray", "type": [{"type": "array", "items": "int"}, "null"]}, + {"name": "int_long", "type": ["long", "null"], "fieldType": "int"}, + {"name": "int_int", "type": ["int", "null"], "fieldType": "int"}, + {"name": "decimal_bytes", "type": ["bytes", "null"], "fieldType": "decimal", "scale": 2}, + {"name": "decimal_float", "type": ["float", "null"], "fieldType": "decimal", "scale": 2}, + {"name": "decimal_double", "type": ["double", "null"], "fieldType": "decimal", "scale": 2}, + {"name": "dateint_bytes_ts", "type": ["bytes", "null"], "fieldType": "dateInt", "layout": "2006-01-02 15:04:05", "unit": "s", "epoch": "1970-01-01 00:00:00"}, + {"name": "recordtime_bytes_ts", "type": ["bytes", "null"], "fieldType": "recordTime", "layout": "2006-01-02 15:04:05", "unit": "s"}, + {"name": "recordtime_bytes_int", "type": ["bytes", "null"], "fieldType": "recordTime", "layout": "2006-01-02 15:04:05", "unit": "s"}, + {"name": "bool_bool", "type": ["boolean", "null"]}, + {"name": "timestamp_bytes_ts", "type": ["bytes", "null"], "fieldType": "timestamp", "layout": "2006-01-02 15:04:05", "epoch": "1970-01-01 00:00:00"}, + {"name": "timestamp_bytes_int", "type": ["bytes", "null"], "fieldType": "timestamp", "unit": "s", "layout": "2006-01-02 15:04:05", "epoch": "1970-01-01 00:00:00"} + ] +} diff --git a/idk/kafka/testdata/schemas/alltypes_delete_fields.json b/idk/kafka/testdata/schemas/alltypes_delete_fields.json new file mode 100644 index 000000000..f1ae8a234 --- /dev/null +++ b/idk/kafka/testdata/schemas/alltypes_delete_fields.json @@ -0,0 +1,12 @@ +{ + "namespace": "org.test", + "type": "record", + "name": "alltypes_delete_fields", + "delete": "fields", + "fields": [ + {"name": "pk0", "type": "string"}, + {"name": "pk1", "type": "string"}, + {"name": "pk2", "type": "string"}, + {"name": "fields", "type": {"type": "array", "items": "string"}} + ] +} diff --git a/idk/kafka/testdata/schemas/alltypes_delete_records.json b/idk/kafka/testdata/schemas/alltypes_delete_records.json new file mode 100644 index 000000000..04ed426f3 --- /dev/null +++ b/idk/kafka/testdata/schemas/alltypes_delete_records.json @@ -0,0 +1,12 @@ +{ + "namespace": "org.test", + "type": "record", + "name": "alltypes_delete_records", + "docs": "supply list of keys or a PQL filter", + "delete": "records", + "fields": [ + {"name": "ids", "type": [{"type": "array", "items": "int"}, "null"]}, + {"name": "keys", "type": [{"type": "array", "items": "string"}, "null"]}, + {"name": "filter", "type": ["string", "null"]} + ] +} \ No newline at end of file diff --git a/idk/kafka/testdata/schemas/alltypes_delete_value.json b/idk/kafka/testdata/schemas/alltypes_delete_value.json new file mode 100644 index 000000000..abd1ee5b9 --- /dev/null +++ b/idk/kafka/testdata/schemas/alltypes_delete_value.json @@ -0,0 +1,21 @@ +{ + "namespace": "org.test", + "type": "record", + "name": "delete_value_schema", + "doc": "All supported avro types and property variations", + "delete": "values", + "fields": [ + {"name": "_id", "type": ["string", "int"]}, + {"name": "stringset_string", "type": ["string", "null"]}, + {"name": "string_string", "type": ["string", "null"]}, + {"name": "stringset_stringarray", "type": [{"type": "array", "items": "string"}, "null"]}, + {"name": "idset_int", "type": ["int", "null"]}, + {"name": "id_int", "type": ["int", "null"]}, + {"name": "idset_intarray", "type": [{"type": "array", "items": "int"}, "null"]}, + {"name": "int_int", "type": "boolean"}, + {"name": "decimal_double", "type": "boolean"}, + {"name": "dateint_bytes_ts", "type": "boolean"}, + {"name": "bools", "type": [{"type": "array", "items": "string"}, "null"]}, + {"name": "timestamp_bytes_int", "type": "boolean"} + ] +} \ No newline at end of file diff --git a/idk/kafka_sasl/source.go b/idk/kafka_sasl/source.go index 32cfa0b8a..8a27c7305 100644 --- a/idk/kafka_sasl/source.go +++ b/idk/kafka_sasl/source.go @@ -149,6 +149,8 @@ func (r *Record) StreamOffset() (string, uint64) { var _ idk.OffsetStreamRecord = &Record{} +func (r *Record) Schema() interface{} { return nil } + func (r *Record) Commit(ctx context.Context) error { r.src.mu.Lock() defer r.src.mu.Unlock() diff --git a/idk/kafka_static/source.go b/idk/kafka_static/source.go index 6e3b02200..b8c871988 100644 --- a/idk/kafka_static/source.go +++ b/idk/kafka_static/source.go @@ -136,6 +136,8 @@ func (r *Record) StreamOffset() (string, uint64) { var _ idk.OffsetStreamRecord = &Record{} +func (r *Record) Schema() interface{} { return nil } + func (r *Record) Commit(ctx context.Context) error { idx, base := r.idx, r.src.spoolBase if idx < base { diff --git a/idk/kinesis/source.go b/idk/kinesis/source.go index 802800e4d..77301eecd 100644 --- a/idk/kinesis/source.go +++ b/idk/kinesis/source.go @@ -136,6 +136,8 @@ func (r *Record) StreamOffset() (string, uint64) { return fmt.Sprintf("%s:%s", r.src.StreamName, r.shardID), r.idx } +func (r *Record) Schema() interface{} { return nil } + var _ idk.OffsetStreamRecord = &Record{} func (r *Record) Commit(ctx context.Context) error { diff --git a/idk/sql/source.go b/idk/sql/source.go index 538fd52b9..81326df31 100644 --- a/idk/sql/source.go +++ b/idk/sql/source.go @@ -181,6 +181,10 @@ func (wr wikiRecord) Commit(ctx context.Context) error { return nil } +func (wr wikiRecord) Schema() interface{} { + return nil +} + func (wr wikiRecord) Data() []interface{} { return wr.record }