fix: Add support for Read key batching for Redis. Also default read batch size to 100 for Redis/Valkey (#304)

Co-authored-by: Bhargav Dodla <bdodla@expediagroup.com>

Author: EXPEbdodla

go/internal/feast/onlinestore/egvalkeyonlinestore.go

@@ -171,7 +171,7 @@ func NewValkeyOnlineStore(project string, config *registry.RepoConfig, onlineSto
 	// Parse read batch size
 	var readBatchSize float64
 	if readBatchSizeJsonValue, ok := onlineStoreConfig["read_batch_size"]; !ok {
-		readBatchSize = -1.0 // Default to -1 (no batching)
+		readBatchSize = 100.0 // Default to 100 Keys Per Batch
 	} else if readBatchSize, ok = readBatchSizeJsonValue.(float64); !ok {
 		return nil, fmt.Errorf("failed to convert read_batch_size: %+v", readBatchSizeJsonValue)
 	}

go/internal/feast/onlinestore/redisonlinestore.go

@@ -53,6 +53,9 @@ type RedisOnlineStore struct {
 	clusterClient *redis.ClusterClient
 
 	config *registry.RepoConfig
+
+	// Number of keys to read in a batch
+	ReadBatchSize int
 }
 
 func NewRedisOnlineStore(project string, config *registry.RepoConfig, onlineStoreConfig map[string]interface{}) (*RedisOnlineStore, error) {
@@ -111,6 +114,19 @@ func NewRedisOnlineStore(project string, config *registry.RepoConfig, onlineStor
 		}
 	}
 
+	// Parse read batch size
+	var readBatchSize float64
+	if readBatchSizeJsonValue, ok := onlineStoreConfig["read_batch_size"]; !ok {
+		readBatchSize = 100.0 // Default to 100 Keys Per Batch
+	} else if readBatchSize, ok = readBatchSizeJsonValue.(float64); !ok {
+		return nil, fmt.Errorf("failed to convert read_batch_size: %+v", readBatchSizeJsonValue)
+	}
+	store.ReadBatchSize = int(readBatchSize)
+
+	if store.ReadBatchSize >= 1 {
+		log.Info().Msgf("Reads will be done in key batches of size: %d", store.ReadBatchSize)
+	}
+
 	// Metrics are not showing up when the service name is set to DD_SERVICE
 	redisTraceServiceName := os.Getenv("DD_SERVICE") + "-redis"
 	if redisTraceServiceName == "" {

sdk/python/feast/infra/online_stores/eg_valkey.py

@@ -87,8 +87,8 @@ class EGValkeyOnlineStoreConfig(FeastConfigBaseModel):
     full_scan_for_deletion: Optional[bool] = True
     """(Optional) whether to scan for deletion of features"""
 
-    read_batch_size: Optional[int] = -1
-    """(Optional) number of keys to read in a single batch for online read requests. Anything <= 1 means no batching."""
+    read_batch_size: Optional[int] = 100
+    """(Optional) number of keys to read in a single batch for online read requests. Anything < 1 means no batching."""
 
 
 class EGValkeyOnlineStore(OnlineStore):

sdk/python/feast/infra/online_stores/redis.py

@@ -79,6 +79,9 @@ class RedisOnlineStoreConfig(FeastConfigBaseModel):
     full_scan_for_deletion: Optional[bool] = True
     """(Optional) whether to scan for deletion of features"""
 
+    read_batch_size: Optional[int] = 100
+    """(Optional) number of keys to read in a single batch for online read requests. Anything < 1 means no batching."""
+
 
 class RedisOnlineStore(OnlineStore):
     """