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

kafka delete functionality

Browse source
This commit is contained in:
jacob 2023-03-06 09:03:35 -06:00 • committed by Jacob Brinlee
parent 33d05e6267
commit 549566b6c2
1 changed files with 10 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
idk/ingest.go
View file

@ -1131,7 +1131,7 @@ func (m *Main) runDeleter(limitCounter *msgCounter) error {
client := m.PilosaClient()
index := m.index
// Pull records one by one from kafka
// Pull records one by one from source (e.g. kafka)
for ; !limitCounter.IsDone(); rec, err = source.Record() {
if err == ErrFlush {
continue
@ -1489,15 +1489,16 @@ func (m *Main) runDeleter(limitCounter *msgCounter) error {
}
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)
}
} else {
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)
_, 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'")