go: Add multi-database support for cluster mode clients

- Add DatabaseId field to baseClientConfiguration for both standalone and cluster clients
- Implement WithDatabaseId() method for ClusterClientConfiguration
- Add database ID validation with proper error handling for negative values
- Refactor standalone client to use shared database ID logic from base configuration
- Add Select() and SelectWithOptions() methods for cluster clients with routing support
- Include comprehensive test coverage for database isolation, reconnection persistence, and error handling
- Add backward compatibility support - clients default to database 0 when no database_id specified
- Add server version compatibility checks (cluster multi-database requires Valkey 9.0+)
- Update documentation with warnings about SELECT command limitations and recommended configuration approach

This enables cluster mode clients to connect to specific databases at initialization time,
with the database selection persisting across reconnections, unlike the SELECT command
which resets on reconnection.

Signed-off-by: affonsov <67347924+affonsov@users.noreply.github.com>

Author: affonsov

go/config/config.go

@@ -92,6 +92,14 @@ func mapReadFrom(readFrom ReadFrom) protobuf.ReadFrom {
 	return protobuf.ReadFrom_Primary
 }
 
+// validateDatabaseId validates the database ID parameter.
+func validateDatabaseId(databaseId int) error {
+	if databaseId < 0 {
+		return errors.New("database_id must be non-negative")
+	}
+	return nil
+}
+
 type baseClientConfiguration struct {
 	addresses         []NodeAddress
 	useTLS            bool
@@ -102,6 +110,7 @@ type baseClientConfiguration struct {
 	clientAZ          string
 	reconnectStrategy *BackoffStrategy
 	lazyConnect       bool
+	DatabaseId        *int `json:"database_id,omitempty"`
 }
 
 func (config *baseClientConfiguration) toProtobuf() (*protobuf.ConnectionRequest, error) {
@@ -152,6 +161,13 @@ func (config *baseClientConfiguration) toProtobuf() (*protobuf.ConnectionRequest
 		request.LazyConnect = config.lazyConnect
 	}
 
+	if config.DatabaseId != nil {
+		if err := validateDatabaseId(*config.DatabaseId); err != nil {
+			return nil, err
+		}
+		request.DatabaseId = uint32(*config.DatabaseId)
+	}
+
 	return &request, nil
 }
 
@@ -214,7 +230,6 @@ func (strategy *BackoffStrategy) toProtobuf() *protobuf.ConnectionRetryStrategy
 // ClientConfiguration represents the configuration settings for a Standalone client.
 type ClientConfiguration struct {
 	baseClientConfiguration
-	databaseId         int
 	subscriptionConfig *StandaloneSubscriptionConfig
 	AdvancedClientConfiguration
 }
@@ -232,9 +247,6 @@ func (config *ClientConfiguration) ToProtobuf() (*protobuf.ConnectionRequest, er
 	}
 	request.ClusterModeEnabled = false
 
-	if config.databaseId != 0 {
-		request.DatabaseId = uint32(config.databaseId)
-	}
 	if config.subscriptionConfig != nil && len(config.subscriptionConfig.subscriptions) > 0 {
 		request.PubsubSubscriptions = config.subscriptionConfig.toProtobuf()
 	}
@@ -330,7 +342,7 @@ func (config *ClientConfiguration) WithReconnectStrategy(strategy *BackoffStrate
 
 // WithDatabaseId sets the index of the logical database to connect to.
 func (config *ClientConfiguration) WithDatabaseId(id int) *ClientConfiguration {
-	config.databaseId = id
+	config.DatabaseId = &id
 	return config
 }
 
@@ -481,6 +493,12 @@ func (config *ClusterClientConfiguration) WithReconnectStrategy(
 	return config
 }
 
+// WithDatabaseId sets the index of the logical database to connect to.
+func (config *ClusterClientConfiguration) WithDatabaseId(id int) *ClusterClientConfiguration {
+	config.DatabaseId = &id
+	return config
+}
+
 // WithAdvancedConfiguration sets the advanced configuration settings for the client.
 func (config *ClusterClientConfiguration) WithAdvancedConfiguration(
 	advancedConfig *AdvancedClusterClientConfiguration,

go/config/config_test.go

@@ -340,3 +340,122 @@ func TestConfig_LazyConnect(t *testing.T) {
 
 	assert.False(t, defaultClusterResult.LazyConnect)
 }
+
+func TestConfig_DatabaseId(t *testing.T) {
+	// Test standalone client with database ID
+	standaloneConfig := NewClientConfiguration().WithDatabaseId(5)
+	standaloneResult, err := standaloneConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert standalone config to protobuf: %v", err)
+	}
+	assert.Equal(t, uint32(5), standaloneResult.DatabaseId)
+
+	// Test cluster client with database ID
+	clusterConfig := NewClusterClientConfiguration().WithDatabaseId(3)
+	clusterResult, err := clusterConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert cluster config to protobuf: %v", err)
+	}
+	assert.Equal(t, uint32(3), clusterResult.DatabaseId)
+
+	// Test default behavior (no database ID set)
+	defaultStandaloneConfig := NewClientConfiguration()
+	defaultStandaloneResult, err := defaultStandaloneConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert default standalone config to protobuf: %v", err)
+	}
+	assert.Equal(t, uint32(0), defaultStandaloneResult.DatabaseId)
+
+	defaultClusterConfig := NewClusterClientConfiguration()
+	defaultClusterResult, err := defaultClusterConfig.ToProtobuf()
+	if err != nil {
+		t.Fatalf("Failed to convert default cluster config to protobuf: %v", err)
+	}
+	assert.Equal(t, uint32(0), defaultClusterResult.DatabaseId)
+}
+
+func TestConfig_DatabaseId_Validation(t *testing.T) {
+	// Test negative database ID validation for standalone
+	standaloneConfig := NewClientConfiguration().WithDatabaseId(-1)
+	_, err := standaloneConfig.ToProtobuf()
+	assert.EqualError(t, err, "database_id must be non-negative")
+
+	// Test negative database ID validation for cluster
+	clusterConfig := NewClusterClientConfiguration().WithDatabaseId(-1)
+	_, err = clusterConfig.ToProtobuf()
+	assert.EqualError(t, err, "database_id must be non-negative")
+
+	// Test valid database ID (0 should be valid)
+	validStandaloneConfig := NewClientConfiguration().WithDatabaseId(0)
+	_, err = validStandaloneConfig.ToProtobuf()
+	assert.NoError(t, err)
+
+	validClusterConfig := NewClusterClientConfiguration().WithDatabaseId(0)
+	_, err = validClusterConfig.ToProtobuf()
+	assert.NoError(t, err)
+}
+
+func TestConfig_DatabaseId_ExtendedValidation(t *testing.T) {
+	// Test various valid database IDs for both standalone and cluster
+	validDatabaseIds := []int{0, 1, 15, 50, 100, 999}
+
+	for _, dbId := range validDatabaseIds {
+		t.Run(fmt.Sprintf("ValidDatabaseId_%d", dbId), func(t *testing.T) {
+			// Test standalone configuration
+			standaloneConfig := NewClientConfiguration().WithDatabaseId(dbId)
+			standaloneResult, err := standaloneConfig.ToProtobuf()
+			assert.NoError(t, err)
+			assert.Equal(t, uint32(dbId), standaloneResult.DatabaseId)
+
+			// Test cluster configuration
+			clusterConfig := NewClusterClientConfiguration().WithDatabaseId(dbId)
+			clusterResult, err := clusterConfig.ToProtobuf()
+			assert.NoError(t, err)
+			assert.Equal(t, uint32(dbId), clusterResult.DatabaseId)
+		})
+	}
+
+	// Test invalid database IDs
+	invalidDatabaseIds := []int{-1, -10, -100}
+
+	for _, dbId := range invalidDatabaseIds {
+		t.Run(fmt.Sprintf("InvalidDatabaseId_%d", dbId), func(t *testing.T) {
+			// Test standalone configuration
+			standaloneConfig := NewClientConfiguration().WithDatabaseId(dbId)
+			_, err := standaloneConfig.ToProtobuf()
+			assert.EqualError(t, err, "database_id must be non-negative")
+
+			// Test cluster configuration
+			clusterConfig := NewClusterClientConfiguration().WithDatabaseId(dbId)
+			_, err = clusterConfig.ToProtobuf()
+			assert.EqualError(t, err, "database_id must be non-negative")
+		})
+	}
+}
+
+func TestConfig_DatabaseId_BaseConfiguration(t *testing.T) {
+	// Test that database_id is properly handled in base configuration for both client types
+
+	// Test standalone client inherits database_id from base configuration
+	standaloneConfig := NewClientConfiguration().WithDatabaseId(5)
+	standaloneResult, err := standaloneConfig.ToProtobuf()
+	assert.NoError(t, err)
+	assert.Equal(t, uint32(5), standaloneResult.DatabaseId)
+	assert.False(t, standaloneResult.ClusterModeEnabled)
+
+	// Test cluster client inherits database_id from base configuration
+	clusterConfig := NewClusterClientConfiguration().WithDatabaseId(3)
+	clusterResult, err := clusterConfig.ToProtobuf()
+	assert.NoError(t, err)
+	assert.Equal(t, uint32(3), clusterResult.DatabaseId)
+	assert.True(t, clusterResult.ClusterModeEnabled)
+
+	// Test that both configurations use the same base validation logic
+	standaloneConfigInvalid := NewClientConfiguration().WithDatabaseId(-1)
+	_, err = standaloneConfigInvalid.ToProtobuf()
+	assert.EqualError(t, err, "database_id must be non-negative")
+
+	clusterConfigInvalid := NewClusterClientConfiguration().WithDatabaseId(-1)
+	_, err = clusterConfigInvalid.ToProtobuf()
+	assert.EqualError(t, err, "database_id must be non-negative")
+}

go/glide_client.go

@@ -210,6 +210,20 @@ func (client *Client) ConfigGet(ctx context.Context, args []string) (map[string]
 
 // Select changes the currently selected database.
 //
+// WARNING: This command is NOT RECOMMENDED for production use.
+// Upon reconnection, the client will revert to the database_id specified
+// in the client configuration (default: 0), NOT the database selected
+// via this command.
+//
+// RECOMMENDED APPROACH: Use the database_id parameter in client
+// configuration instead:
+//
+//	config := &config.ClientConfiguration{
+//		Addresses: []config.NodeAddress{{Host: "localhost", Port: 6379}},
+//		DatabaseId: &databaseId, // Recommended: persists across reconnections
+//	}
+//	client, err := NewClient(config)
+//
 // See [valkey.io] for details.
 //
 // Parameters:

go/glide_cluster_client.go

@@ -335,6 +335,87 @@ func (client *ClusterClient) CustomCommandWithRoute(ctx context.Context,
 	return models.CreateClusterValue[any](data), nil
 }
 
+// Select changes the currently selected database on cluster nodes.
+// The command will be routed to all primary nodes by default.
+//
+// WARNING: This command is NOT RECOMMENDED for production use.
+// Upon reconnection, nodes will revert to the database_id specified
+// in the client configuration (default: 0), NOT the database selected
+// via this command.
+//
+// RECOMMENDED APPROACH: Use the database_id parameter in client
+// configuration instead:
+//
+//	config := &config.ClusterClientConfiguration{
+//		Addresses: []config.NodeAddress{{Host: "localhost", Port: 6379}},
+//		DatabaseId: &databaseId, // Recommended: persists across reconnections
+//	}
+//	client, err := NewClusterClient(config)
+//
+// CLUSTER BEHAVIOR: This command routes to all nodes by default
+// to maintain consistency across the cluster.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	ctx - The context for controlling the command execution.
+//	index - The index of the database to select.
+//
+// Return value:
+//
+//	A simple `"OK"` response.
+//
+// [valkey.io]: https://valkey.io/commands/select/
+func (client *ClusterClient) Select(ctx context.Context, index int64) (string, error) {
+	result, err := client.executeCommand(ctx, C.Select, []string{utils.IntToString(index)})
+	if err != nil {
+		return models.DefaultStringResponse, err
+	}
+
+	return handleOkResponse(result)
+}
+
+// Select changes the currently selected database on cluster nodes.
+//
+// WARNING: This command is NOT RECOMMENDED for production use.
+// Upon reconnection, nodes will revert to the database_id specified
+// in the client configuration (default: 0), NOT the database selected
+// via this command.
+//
+// RECOMMENDED APPROACH: Use the database_id parameter in client
+// configuration instead.
+//
+// CLUSTER BEHAVIOR: This command routes to all nodes by default
+// to maintain consistency across the cluster.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	ctx - The context for controlling the command execution.
+//	index - The index of the database to select.
+//	routeOption - Specifies the routing configuration for the command. The client will route the
+//	              command to the nodes defined by routeOption.Route. Defaults to all primary nodes.
+//
+// Return value:
+//
+//	A simple `"OK"` response.
+//
+// [valkey.io]: https://valkey.io/commands/select/
+func (client *ClusterClient) SelectWithOptions(
+	ctx context.Context,
+	index int64,
+	routeOption options.RouteOption,
+) (string, error) {
+	result, err := client.executeCommandWithRoute(ctx, C.Select, []string{utils.IntToString(index)}, routeOption.Route)
+	if err != nil {
+		return models.DefaultStringResponse, err
+	}
+
+	return handleOkResponse(result)
+}
+
 // Pings the server.
 // The command will be routed to all primary nodes.
 //

go/integTest/batch_test.go

@@ -2385,6 +2385,12 @@ func CreateServerManagementTests(batch *pipeline.ClusterBatch, isAtomic bool, se
 	testData = append(testData, CommandTestData{ExpectedResponse: "OK", TestName: "ConfigResetStat()"})
 	// ConfigRewrite skipped, because depends on config
 
+	// SELECT command is only available in Valkey 9+ for cluster mode
+	if serverVer >= "9.0.0" {
+		batch.Select(1)
+		testData = append(testData, CommandTestData{ExpectedResponse: "OK", TestName: "Select(1)"})
+	}
+
 	return BatchTestData{CommandTestData: testData, TestName: "Server Management commands"}
 }