Implemented MinReplicationFactor. (#196)

Author: Lars Maier

cluster.go

@@ -203,6 +203,7 @@ type InventoryCollectionParameters struct {
 	Path                 string                 `json:"path,omitempty"`
 	PlanID               string                 `json:"planId,omitempty"`
 	ReplicationFactor    int                    `json:"replicationFactor,omitempty"`
+	MinReplicationFactor int                    `json:"minReplicationFactor,omitempty"`
 	ShardKeys            []string               `json:"shardKeys,omitempty"`
 	Shards               map[ShardID][]ServerID `json:"shards,omitempty"`
 	Status               CollectionStatus       `json:"status,omitempty"`

cluster_impl.go

@@ -249,6 +249,7 @@ type inventoryCollectionParametersInternal struct {
 	Path                 string                 `json:"path,omitempty"`
 	PlanID               string                 `json:"planId,omitempty"`
 	ReplicationFactor    replicationFactor      `json:"replicationFactor,omitempty"`
+	MinReplicationFactor int                    `json:"minReplicationFactor,omitempty"`
 	ShardKeys            []string               `json:"shardKeys,omitempty"`
 	Shards               map[ShardID][]ServerID `json:"shards,omitempty"`
 	Status               CollectionStatus       `json:"status,omitempty"`
@@ -277,6 +278,7 @@ func (p *InventoryCollectionParameters) asInternal() inventoryCollectionParamete
 		Path:                 p.Path,
 		PlanID:               p.PlanID,
 		ReplicationFactor:    replicationFactor(p.ReplicationFactor),
+		MinReplicationFactor: p.MinReplicationFactor,
 		ShardKeys:            p.ShardKeys,
 		Shards:               p.Shards,
 		Status:               p.Status,
@@ -310,6 +312,7 @@ func (p *inventoryCollectionParametersInternal) asExternal() InventoryCollection
 		Path:                 p.Path,
 		PlanID:               p.PlanID,
 		ReplicationFactor:    int(p.ReplicationFactor),
+		MinReplicationFactor: p.MinReplicationFactor,
 		ShardKeys:            p.ShardKeys,
 		Shards:               p.Shards,
 		Status:               p.Status,

collection.go

@@ -116,6 +116,10 @@ type CollectionProperties struct {
 	// ReplicationFactor contains how many copies of each shard are kept on different DBServers.
 	// Only available in cluster setup.
 	ReplicationFactor int `json:"replicationFactor,omitempty"`
+	// MinReplicationFactor contains how many copies must be available before a collection can be written.
+	// It is required that 1 <= MinReplicationFactor <= ReplicationFactor.
+	// Default is 1. Not available for satellite collections.
+	MinReplicationFactor int `json:"minReplicationFactor,omitempty"`
 	// SmartJoinAttribute
 	// See documentation for smart joins.
 	// This requires ArangoDB Enterprise Edition.
@@ -144,6 +148,8 @@ type SetCollectionPropertiesOptions struct {
 	// ReplicationFactor contains how many copies of each shard are kept on different DBServers.
 	// Only available in cluster setup.
 	ReplicationFactor int `json:"replicationFactor,omitempty"`
+	// MinReplicationFactor contains how many copies must be available before a collection can be written.
+	MinReplicationFactor int `json:"minReplicationFactor,omitempty"`
 }
 
 // CollectionStatus indicates the status of a collection.

collection_impl.go

@@ -274,40 +274,43 @@ type collectionPropertiesInternal struct {
 		Type          KeyGeneratorType `json:"type,omitempty"`
 		AllowUserKeys bool             `json:"allowUserKeys,omitempty"`
 	} `json:"keyOptions,omitempty"`
-	NumberOfShards     int               `json:"numberOfShards,omitempty"`
-	ShardKeys          []string          `json:"shardKeys,omitempty"`
-	ReplicationFactor  replicationFactor `json:"replicationFactor,omitempty"`
-	SmartJoinAttribute string            `json:"smartJoinAttribute,omitempty"`
-	ShardingStrategy   ShardingStrategy  `json:"shardingStrategy,omitempty"`
+	NumberOfShards       int               `json:"numberOfShards,omitempty"`
+	ShardKeys            []string          `json:"shardKeys,omitempty"`
+	ReplicationFactor    replicationFactor `json:"replicationFactor,omitempty"`
+	MinReplicationFactor int               `json:"minReplicationFactor,omitempty"`
+	SmartJoinAttribute   string            `json:"smartJoinAttribute,omitempty"`
+	ShardingStrategy     ShardingStrategy  `json:"shardingStrategy,omitempty"`
 }
 
 func (p *collectionPropertiesInternal) asExternal() CollectionProperties {
 	return CollectionProperties{
-		CollectionInfo:     p.CollectionInfo,
-		WaitForSync:        p.WaitForSync,
-		DoCompact:          p.DoCompact,
-		JournalSize:        p.JournalSize,
-		KeyOptions:         p.KeyOptions,
-		NumberOfShards:     p.NumberOfShards,
-		ShardKeys:          p.ShardKeys,
-		ReplicationFactor:  int(p.ReplicationFactor),
-		SmartJoinAttribute: p.SmartJoinAttribute,
-		ShardingStrategy:   p.ShardingStrategy,
+		CollectionInfo:       p.CollectionInfo,
+		WaitForSync:          p.WaitForSync,
+		DoCompact:            p.DoCompact,
+		JournalSize:          p.JournalSize,
+		KeyOptions:           p.KeyOptions,
+		NumberOfShards:       p.NumberOfShards,
+		ShardKeys:            p.ShardKeys,
+		ReplicationFactor:    int(p.ReplicationFactor),
+		MinReplicationFactor: p.MinReplicationFactor,
+		SmartJoinAttribute:   p.SmartJoinAttribute,
+		ShardingStrategy:     p.ShardingStrategy,
 	}
 }
 
 func (p *CollectionProperties) asInternal() collectionPropertiesInternal {
 	return collectionPropertiesInternal{
-		CollectionInfo:     p.CollectionInfo,
-		WaitForSync:        p.WaitForSync,
-		DoCompact:          p.DoCompact,
-		JournalSize:        p.JournalSize,
-		KeyOptions:         p.KeyOptions,
-		NumberOfShards:     p.NumberOfShards,
-		ShardKeys:          p.ShardKeys,
-		ReplicationFactor:  replicationFactor(p.ReplicationFactor),
-		SmartJoinAttribute: p.SmartJoinAttribute,
-		ShardingStrategy:   p.ShardingStrategy,
+		CollectionInfo:       p.CollectionInfo,
+		WaitForSync:          p.WaitForSync,
+		DoCompact:            p.DoCompact,
+		JournalSize:          p.JournalSize,
+		KeyOptions:           p.KeyOptions,
+		NumberOfShards:       p.NumberOfShards,
+		ShardKeys:            p.ShardKeys,
+		ReplicationFactor:    replicationFactor(p.ReplicationFactor),
+		MinReplicationFactor: p.MinReplicationFactor,
+		SmartJoinAttribute:   p.SmartJoinAttribute,
+		ShardingStrategy:     p.ShardingStrategy,
 	}
 }
 
@@ -320,6 +323,7 @@ func (p *CollectionProperties) fromInternal(i *collectionPropertiesInternal) {
 	p.NumberOfShards = i.NumberOfShards
 	p.ShardKeys = i.ShardKeys
 	p.ReplicationFactor = int(i.ReplicationFactor)
+	p.MinReplicationFactor = i.MinReplicationFactor
 	p.SmartJoinAttribute = i.SmartJoinAttribute
 	p.ShardingStrategy = i.ShardingStrategy
 }
@@ -341,23 +345,26 @@ func (p *CollectionProperties) UnmarshalJSON(d []byte) error {
 }
 
 type setCollectionPropertiesOptionsInternal struct {
-	WaitForSync       *bool             `json:"waitForSync,omitempty"`
-	JournalSize       int64             `json:"journalSize,omitempty"`
-	ReplicationFactor replicationFactor `json:"replicationFactor,omitempty"`
+	WaitForSync          *bool             `json:"waitForSync,omitempty"`
+	JournalSize          int64             `json:"journalSize,omitempty"`
+	ReplicationFactor    replicationFactor `json:"replicationFactor,omitempty"`
+	MinReplicationFactor int               `json:"minReplicationFactor,omitempty"`
 }
 
 func (p *SetCollectionPropertiesOptions) asInternal() setCollectionPropertiesOptionsInternal {
 	return setCollectionPropertiesOptionsInternal{
-		WaitForSync:       p.WaitForSync,
-		JournalSize:       p.JournalSize,
-		ReplicationFactor: replicationFactor(p.ReplicationFactor),
+		WaitForSync:          p.WaitForSync,
+		JournalSize:          p.JournalSize,
+		ReplicationFactor:    replicationFactor(p.ReplicationFactor),
+		MinReplicationFactor: p.MinReplicationFactor,
 	}
 }
 
 func (p *SetCollectionPropertiesOptions) fromInternal(i *setCollectionPropertiesOptionsInternal) {
 	p.WaitForSync = i.WaitForSync
 	p.JournalSize = i.JournalSize
 	p.ReplicationFactor = int(i.ReplicationFactor)
+	p.MinReplicationFactor = i.MinReplicationFactor
 }
 
 // MarshalJSON converts SetCollectionPropertiesOptions into json

database_collections.go

@@ -52,6 +52,10 @@ type CreateCollectionOptions struct {
 	// before the write operation is reported successful. If a server fails, this is detected automatically
 	// and one of the servers holding copies take over, usually without an error being reported.
 	ReplicationFactor int `json:"replicationFactor,omitempty"`
+	// MinReplicationFactor contains how many copies must be available before a collection can be written.
+	// It is required that 1 <= MinReplicationFactor <= ReplicationFactor.
+	// Default is 1. Not available for satellite collections.
+	MinReplicationFactor int `json:"minReplicationFactor,omitempty"`
 	// If true then the data is synchronized to disk before returning from a document create, update, replace or removal operation. (default: false)
 	WaitForSync bool `json:"waitForSync,omitempty"`
 	// Whether or not the collection will be compacted (default is true)

database_collections_impl.go

@@ -104,6 +104,7 @@ func (d *database) Collections(ctx context.Context) ([]Collection, error) {
 type createCollectionOptionsInternal struct {
 	JournalSize          int                   `json:"journalSize,omitempty"`
 	ReplicationFactor    replicationFactor     `json:"replicationFactor,omitempty"`
+	MinReplicationFactor int                   `json:"minReplicationFactor,omitempty"`
 	WaitForSync          bool                  `json:"waitForSync,omitempty"`
 	DoCompact            *bool                 `json:"doCompact,omitempty"`
 	IsVolatile           bool                  `json:"isVolatile,omitempty"`
@@ -174,6 +175,7 @@ func (d *database) CreateCollection(ctx context.Context, name string, options *C
 func (p *createCollectionOptionsInternal) fromExternal(i *CreateCollectionOptions) {
 	p.JournalSize = i.JournalSize
 	p.ReplicationFactor = replicationFactor(i.ReplicationFactor)
+	p.MinReplicationFactor = i.MinReplicationFactor
 	p.WaitForSync = i.WaitForSync
 	p.DoCompact = i.DoCompact
 	p.IsVolatile = i.IsVolatile

test/collection_test.go

@@ -552,3 +552,150 @@ func TestCollectionStatistics(t *testing.T) {
 		}
 	}
 }
+
+// TestCollectionMinReplFactCreate creates a collection with minReplicationFactor != 1
+func TestCollectionMinReplFactCreate(t *testing.T) {
+	c := createClientFromEnv(t, true)
+	skipBelowVersion(c, "3.5", t)
+	db := ensureDatabase(nil, c, "collection_min_repl_test", nil, t)
+	name := "test_min_repl_create"
+	minRepl := 2
+	options := driver.CreateCollectionOptions{
+		ReplicationFactor:    minRepl,
+		MinReplicationFactor: minRepl,
+	}
+	if _, err := db.CreateCollection(nil, name, &options); err != nil {
+		t.Fatalf("Failed to create collection '%s': %s", name, describe(err))
+	}
+
+	// Collection must exist now
+	if found, err := db.CollectionExists(nil, name); err != nil {
+		t.Errorf("CollectionExists('%s') failed: %s", name, describe(err))
+	} else if !found {
+		t.Errorf("CollectionExists('%s') return false, expected true", name)
+	}
+	// Check if the collection has a minReplicationFactor
+	if col, err := db.Collection(nil, name); err != nil {
+		t.Errorf("Collection('%s') failed: %s", name, describe(err))
+	} else {
+		if prop, err := col.Properties(nil); err != nil {
+			t.Errorf("Properties() failed: %s", describe(err))
+		} else {
+			if prop.MinReplicationFactor != minRepl {
+				t.Errorf("Collection does not have the correct min replication factor value, expected `%d`, found `%d`", minRepl, prop.MinReplicationFactor)
+			}
+		}
+	}
+}
+
+// TestCollectionMinReplFactInvalid creates a collection with minReplicationFactor > replicationFactor
+func TestCollectionMinReplFactInvalid(t *testing.T) {
+	c := createClientFromEnv(t, true)
+	skipBelowVersion(c, "3.5", t)
+	skipNoCluster(c, t)
+	db := ensureDatabase(nil, c, "collection_min_repl_test", nil, t)
+	name := "test_min_repl_create_invalid"
+	minRepl := 2
+	options := driver.CreateCollectionOptions{
+		ReplicationFactor:    minRepl,
+		MinReplicationFactor: minRepl + 1,
+	}
+	if _, err := db.CreateCollection(nil, name, &options); err == nil {
+		t.Fatalf("CreateCollection('%s') did not fail", name)
+	}
+	// Collection must not exist now
+	if found, err := db.CollectionExists(nil, name); err != nil {
+		t.Errorf("CollectionExists('%s') failed: %s", name, describe(err))
+	} else if found {
+		t.Errorf("Collection %s should not exist", name)
+	}
+}
+
+// TestCollectionMinReplFactClusterInv tests if minReplicationFactor is forwarded to ClusterInfo
+func TestCollectionMinReplFactClusterInv(t *testing.T) {
+	c := createClientFromEnv(t, true)
+	skipBelowVersion(c, "3.5", t)
+	skipNoCluster(c, t)
+	db := ensureDatabase(nil, c, "collection_min_repl_test", nil, t)
+	name := "test_min_repl_cluster_invent"
+	minRepl := 2
+	ensureCollection(nil, db, name, &driver.CreateCollectionOptions{
+		ReplicationFactor:    minRepl,
+		MinReplicationFactor: minRepl,
+	}, t)
+
+	cc, err := c.Cluster(nil)
+	if err != nil {
+		t.Fatalf("Failed to get Cluster: %s", describe(err))
+	}
+
+	inv, err := cc.DatabaseInventory(nil, db)
+	if err != nil {
+		t.Fatalf("Failed to get Database Inventory: %s", describe(err))
+	}
+
+	col, found := inv.CollectionByName(name)
+	if !found {
+		t.Fatalf("Failed to get find collection: %s", describe(err))
+	}
+
+	if col.Parameters.MinReplicationFactor != minRepl {
+		t.Errorf("Collection does not have the correct min replication factor value, expected `%d`, found `%d`", minRepl, col.Parameters.MinReplicationFactor)
+	}
+}
+
+// TestCollectionMinReplFactSetProp updates the minimal replication factor using SetProperties
+func TestCollectionMinReplFactSetProp(t *testing.T) {
+	c := createClientFromEnv(t, true)
+	skipBelowVersion(c, "3.5", t)
+	skipNoCluster(c, t)
+	db := ensureDatabase(nil, c, "collection_min_repl_test", nil, t)
+	name := "test_min_repl_set_prop"
+	minRepl := 2
+	col := ensureCollection(nil, db, name, &driver.CreateCollectionOptions{
+		ReplicationFactor:    minRepl,
+		MinReplicationFactor: minRepl,
+	}, t)
+
+	if err := col.SetProperties(nil, driver.SetCollectionPropertiesOptions{
+		MinReplicationFactor: 1,
+	}); err != nil {
+		t.Fatalf("Failed to update properties: %s", describe(err))
+	}
+
+	if prop, err := col.Properties(nil); err != nil {
+		t.Fatalf("Failed to get properties: %s", describe(err))
+	} else {
+		if prop.MinReplicationFactor != 1 {
+			t.Fatalf("MinReplicationFactor not updated, expected %d, found %d", 1, prop.MinReplicationFactor)
+		}
+	}
+}
+
+// TestCollectionMinReplFactSetPropInvalid updates the minimal replication factor to an invalid value using SetProperties
+func TestCollectionMinReplFactSetPropInvalid(t *testing.T) {
+	c := createClientFromEnv(t, true)
+	skipBelowVersion(c, "3.5", t)
+	skipNoCluster(c, t)
+	db := ensureDatabase(nil, c, "collection_min_repl_test", nil, t)
+	name := "test_min_repl_set_prop_inv"
+	minRepl := 2
+	col := ensureCollection(nil, db, name, &driver.CreateCollectionOptions{
+		ReplicationFactor:    minRepl,
+		MinReplicationFactor: minRepl,
+	}, t)
+
+	if err := col.SetProperties(nil, driver.SetCollectionPropertiesOptions{
+		MinReplicationFactor: minRepl + 1,
+	}); err == nil {
+		t.Errorf("SetProperties did not fail")
+	}
+
+	if prop, err := col.Properties(nil); err != nil {
+		t.Fatalf("Failed to get properties: %s", describe(err))
+	} else {
+		if prop.MinReplicationFactor != minRepl {
+			t.Fatalf("MinReplicationFactor not updated, expected %d, found %d", minRepl, prop.MinReplicationFactor)
+		}
+	}
+}