Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Using correct CQL Statement for scylla read calls #167

Merged
merged 9 commits into from
Feb 3, 2025
Merged

fix: Using correct CQL Statement for scylla read calls #167

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
bb3d736
adding logs to view key batching calculations
Jan 30, 2025
ca4f9d1
fix: Logging
Jan 30, 2025
6a92af6
changing cache logic in BatchedKeysOnlineRead
Jan 31, 2025
2dd8dda
modified BatchedKeysOnlineRead
Jan 31, 2025
07471d0
modified BatchedKeysOnlineRead
Jan 31, 2025
6e72718
fix: Logging
Jan 31, 2025
b6c993f
fix: naming
Jan 31, 2025
6bc335f
fix: Remove logs
vanitabhagwat Feb 3, 2025
4f0b9f5
fix: Remove logs
vanitabhagwat Feb 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 9 additions & 12 deletions go/internal/feast/onlinestore/cassandraonlinestore.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,7 +257,6 @@ func (c *CassandraOnlineStore) getMultiKeyCQLStatement(tableName string, feature
for i := 0; i < nkeys; i++ {
keyPlaceholders[i] = "?"
}

return fmt.Sprintf(
`SELECT "entity_key", "feature_name", "event_ts", "value" FROM %s WHERE "entity_key" IN (%s) AND "feature_name" IN (%s)`,
tableName,
Expand Down Expand Up @@ -424,7 +423,6 @@ func (c *CassandraOnlineStore) BatchedKeysOnlineRead(ctx context.Context, entity
if len(uniqueNames) != 1 {
return nil, fmt.Errorf("rejecting OnlineRead as more than 1 feature view was tried to be read at once")
}

serializedEntityKeys, serializedEntityKeyToIndex, err := c.buildCassandraEntityKeys(entityKeys)

if err != nil {
Expand All @@ -449,7 +447,6 @@ func (c *CassandraOnlineStore) BatchedKeysOnlineRead(ctx context.Context, entity
nKeys := len(serializedEntityKeys)
batchSize := c.keyBatchSize
nBatches := int(math.Ceil(float64(nKeys) / float64(batchSize)))

batches := make([][]any, nBatches)
nAssigned := 0
for i := 0; i < nBatches; i++ {
Expand All @@ -465,18 +462,18 @@ func (c *CassandraOnlineStore) BatchedKeysOnlineRead(ctx context.Context, entity
waitGroup.Add(nBatches)

errorsChannel := make(chan error, nBatches)
var currentBatchLength int
var prevBatchLength int
var cqlStatement string
for _, batch := range batches {
go func(keyBatch []any) {
currentBatchLength = len(batch)
if currentBatchLength != prevBatchLength {
cqlStatement = c.getMultiKeyCQLStatement(tableName, featureNames, currentBatchLength)
prevBatchLength = currentBatchLength
}
go func(keyBatch []any, statement string) {
defer waitGroup.Done()

// this caches the previous batch query if it had the same number of keys
if len(keyBatch) != prevBatchLength {
cqlStatement = c.getMultiKeyCQLStatement(tableName, featureNames, len(keyBatch))
}

iter := c.session.Query(cqlStatement, keyBatch...).WithContext(ctx).Iter()
iter := c.session.Query(statement, keyBatch...).WithContext(ctx).Iter()

scanner := iter.Scanner()
var entityKey string
Expand Down Expand Up @@ -539,7 +536,7 @@ func (c *CassandraOnlineStore) BatchedKeysOnlineRead(ctx context.Context, entity
results[serializedEntityKeyToIndex[keyString]][featureNamesToIdx[featName]] = featureData
}
}
}(batch)
}(batch, cqlStatement)
}
// wait until all concurrent single-key queries are done
waitGroup.Wait()
Expand Down
Loading