scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

adding kafka delete functionality

Browse source
This commit is contained in:
jacob 2023-03-05 23:03:52 -06:00 • committed by Jacob Brinlee
parent c4b0e1e1fb
commit aad32f1dbd
26 changed files with 1293 additions and 304 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
idk/api/source.go
View file

@ -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
)

8
idk/cmd/molecula-consumer-github/main.go
View file

@ -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 {

2
idk/csv/source.go
View file

@ -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()

2
idk/datagen/cmd.go
View file

@ -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

4
idk/idallocator_test.go
View file

@ -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)
}

553
idk/ingest.go
View file

@ -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) {

4
idk/ingest_test.go
View file

@ -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) {

37
idk/interfaces.go
View file

@ -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:

804
idk/kafka/cmd_test.go
View file

@ -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) }

56
idk/kafka/source.go
View file

@ -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")
}

2
idk/kafka/source_test.go
View file

@ -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)
}

10
idk/kafka/testdata/records/alltypes.json vendored Normal file
View file

@ -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"}}

5
idk/kafka/testdata/records/alltypes_delete_fields.json vendored Normal file
View file

@ -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"]}

2
idk/kafka/testdata/records/alltypes_delete_records.json vendored Normal file
View file

@ -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')"}}

2
idk/kafka/testdata/records/alltypes_delete_records_ids.json vendored Normal file
View file

@ -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')"}}

2
idk/kafka/testdata/records/alltypes_delete_value.json vendored Normal file
View file

@ -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}

2
idk/kafka/testdata/records/alltypes_delete_value_ids.json vendored Normal file
View file

@ -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}

2
idk/kafka/testdata/records/alltypes_null.json vendored Normal file
View file

@ -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}}

41
idk/kafka/testdata/schemas/alltypes.json vendored Normal file
View file

@ -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"}
]
}

12
idk/kafka/testdata/schemas/alltypes_delete_fields.json vendored Normal file
View file

@ -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"}}
]
}

12
idk/kafka/testdata/schemas/alltypes_delete_records.json vendored Normal file
View file

@ -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"]}
]
}

21
idk/kafka/testdata/schemas/alltypes_delete_value.json vendored Normal file
View file

@ -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"}
]
}

2
idk/kafka_sasl/source.go
View file

@ -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()

2
idk/kafka_static/source.go
View file

@ -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 {

2
idk/kinesis/source.go
View file

@ -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 {

4
idk/sql/source.go
View file

@ -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
}