weaviate

1
//                           _       _
2
// __      _____  __ ___   ___  __ _| |_ ___
3
// \ \ /\ / / _ \/ _` \ \ / / |/ _` | __/ _ \
4
//  \ V  V /  __/ (_| |\ V /| | (_| | ||  __/
5
//   \_/\_/ \___|\__,_| \_/ |_|\__,_|\__\___|
6
//
7
//  Copyright © 2016 - 2024 Weaviate B.V. All rights reserved.
8
//
9
//  CONTACT: hello@weaviate.io
10
//
11

12
package store
13

14
import (
15
	"context"
16

17
	"github.com/cenkalti/backoff/v4"
18
	"github.com/weaviate/weaviate/cluster/utils"
19
	"github.com/weaviate/weaviate/entities/models"
20
	"github.com/weaviate/weaviate/usecases/sharding"
21
)
22

23
// retrySchema is used for retrying schema queries. It is a thin wrapper around
24
// the original schema, separating retry logic from the actual operation.
25
// Retry may be needed due to eventual consistency issues where
26
// updates might take some time to arrive at the follower.
27
type retrySchema struct {
28
	schema          *schema
29
	versionedSchema versionedSchema
30
}
31

32
func (rs retrySchema) ClassInfo(class string) (ci ClassInfo) {
33
	res, _ := rs.ClassInfoWithVersion(context.TODO(), class, 0)
34
	return res
35
}
36

37
// ClassEqual returns the name of an existing class with a similar name, and "" otherwise
38
// strings.EqualFold is used to compare classes
39
func (rs retrySchema) ClassEqual(name string) string {
40
	return rs.schema.ClassEqual(name)
41
}
42

43
func (rs retrySchema) MultiTenancy(class string) models.MultiTenancyConfig {
44
	res, _ := rs.MultiTenancyWithVersion(context.TODO(), class, 0)
45
	return res
46
}
47

48
// Read performs a read operation `reader` on the specified class and sharding state
49
func (rs retrySchema) Read(class string, reader func(*models.Class, *sharding.State) error) error {
50
	return rs.retry(func(s *schema) error {
51
		return s.Read(class, reader)
52
	})
53
}
54

55
// ReadOnlyClass returns a shallow copy of a class.
56
// The copy is read-only and should not be modified.
57
func (rs retrySchema) ReadOnlyClass(class string) (cls *models.Class) {
58
	res, _ := rs.ReadOnlyClassWithVersion(context.TODO(), class, 0)
59
	return res
60
}
61

62
func (rs retrySchema) metaClass(class string) (meta *metaClass) {
63
	rs.retry(func(s *schema) error {
64
		if meta = s.metaClass(class); meta == nil {
65
			return errClassNotFound
66
		}
67
		return nil
68
	})
69
	return
70
}
71

72
// ReadOnlySchema returns a read only schema
73
// Changing the schema outside this package might lead to undefined behavior.
74
//
75
// it creates a shallow copy of existing classes
76
//
77
// This function assumes that class attributes are being overwritten.
78
// The properties attribute is the only one that might vary in size;
79
// therefore, we perform a shallow copy of the existing properties.
80
// This implementation assumes that individual properties are overwritten rather than partially updated
81
func (rs retrySchema) ReadOnlySchema() models.Schema {
82
	return rs.schema.ReadOnlySchema()
83
}
84

85
// ShardOwner returns the node owner of the specified shard
86
func (rs retrySchema) ShardOwner(class, shard string) (owner string, err error) {
87
	res, err := rs.ShardOwnerWithVersion(context.TODO(), class, shard, 0)
88
	return res, err
89
}
90

91
// ShardFromUUID returns shard name of the provided uuid
92
func (rs retrySchema) ShardFromUUID(class string, uuid []byte) (shard string) {
93
	res, _ := rs.ShardFromUUIDWithVersion(context.TODO(), class, uuid, 0)
94
	return res
95
}
96

97
// ShardReplicas returns the replica nodes of a shard
98
func (rs retrySchema) ShardReplicas(class, shard string) (nodes []string, err error) {
99
	res, err := rs.ShardReplicasWithVersion(context.TODO(), class, shard, 0)
100
	return res, err
101
}
102

103
// TenantsShards returns shard name for the provided tenant and its activity status
104
func (rs retrySchema) TenantsShards(class string, tenants ...string) (map[string]string, error) {
105
	return rs.TenantsShardsWithVersion(context.TODO(), 0, class, tenants...)
106
}
107

108
func (rs retrySchema) CopyShardingState(class string) (ss *sharding.State) {
109
	res, _ := rs.CopyShardingStateWithVersion(context.TODO(), class, 0)
110
	return res
111
}
112

113
func (rs retrySchema) GetShardsStatus(class, tenant string) (models.ShardStatusList, error) {
114
	return rs.schema.GetShardsStatus(class, tenant)
115
}
116

117
func (rs retrySchema) Len() int { return rs.schema.len() }
118

119
func (rs retrySchema) retry(f func(*schema) error) error {
120
	return backoff.Retry(func() error {
121
		return f(rs.schema)
122
	}, utils.NewBackoff())
123
}
124