scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 16:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

use db in fragment allocation

Browse source 
This commit changes the fragment allocation algorithm in the cluster
to make use of the `DB` name. This allows each database to use a
different slice distribution.

Initially, the `frame` was going to be used for allocation, however,
this was problematic since queries can span multiple frames so it's
impossible to choose a single frame to use.
This commit is contained in:
Ben Johnson 2016-06-07 14:01:13 -06:00
parent 810108aa32
commit 2d22ae55ee
No known key found for this signature in database
GPG key ID: CBD06EAD6DFD9529
8 changed files with 50 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
client.go
View file

@ -90,14 +90,14 @@ func (c *Client) Schema() ([]*DBInfo, error) {
return rsp.DBs, nil
}
// SliceNodes returns a list of nodes that own a slice.
func (c *Client) SliceNodes(slice uint64) ([]*Node, error) {
// FragmentNodes returns a list of nodes that own a slice.
func (c *Client) FragmentNodes(slice uint64) ([]*Node, error) {
// Execute request against the host.
u := url.URL{
Scheme: "http",
Host: c.host,
Path: "/slices/nodes",
RawQuery: (url.Values{"slice": {strconv.FormatUint(slice, 10)}}).Encode(),
Path: "/fragment/nodes",
RawQuery: (url.Values{"db": {"d"}, "slice": {strconv.FormatUint(slice, 10)}}).Encode(),
}
resp, err := c.HTTPClient.Get(u.String())
if err != nil {
@ -182,7 +182,7 @@ func (c *Client) Import(db, frame string, slice uint64, bits []Bit) error {
}
// Retrieve a list of nodes that own the slice.
nodes, err := c.SliceNodes(slice)
nodes, err := c.FragmentNodes(slice)
if err != nil {
return fmt.Errorf("slice nodes: %s", err)
@ -327,7 +327,7 @@ func (c *Client) backupSliceTo(tw *tar.Writer, db, frame string, slice uint64) e
// This function tries slice owners until one succeeds.
func (c *Client) BackupSlice(db, frame string, slice uint64) (io.ReadCloser, error) {
// Retrieve a list of nodes that own the slice.
nodes, err := c.SliceNodes(slice)
nodes, err := c.FragmentNodes(slice)
if err != nil {
return nil, fmt.Errorf("slice nodes: %s", err)
}
@ -418,7 +418,7 @@ func (c *Client) RestoreFrom(r io.Reader, db, frame string) error {
// restoreSliceFrom restores a single slice to all owning nodes.
func (c *Client) restoreSliceFrom(buf []byte, db, frame string, slice uint64) error {
// Retrieve a list of nodes that own the slice.
nodes, err := c.SliceNodes(slice)
nodes, err := c.FragmentNodes(slice)
if err != nil {
return fmt.Errorf("slice nodes: %s", err)
}

15
cluster.go
View file

@ -64,24 +64,25 @@ func NewCluster() *Cluster {
}
// Partition returns the partition that a slice belongs to.
func (c *Cluster) Partition(slice uint64) int {
func (c *Cluster) Partition(db string, slice uint64) int {
var buf [8]byte
binary.BigEndian.PutUint64(buf[:], slice)
// Hash the bytes and mod by partition count.
h := fnv.New64a()
h.Write([]byte(db))
h.Write(buf[:])
return int(h.Sum64() % uint64(c.PartitionN))
}
// SliceNodes returns a list of nodes that own a slice.
func (c *Cluster) SliceNodes(slice uint64) []*Node {
return c.PartitionNodes(c.Partition(slice))
// FragmentNodes returns a list of nodes that own a fragment.
func (c *Cluster) FragmentNodes(db string, slice uint64) []*Node {
return c.PartitionNodes(c.Partition(db, slice))
}
// OwnsSlice returns true if a host owns slice.
func (c *Cluster) OwnsSlice(host string, slice uint64) bool {
return Nodes(c.SliceNodes(slice)).ContainsHost(host)
// OwnsFragment returns true if a host owns a fragment.
func (c *Cluster) OwnsFragment(host string, db string, slice uint64) bool {
return Nodes(c.FragmentNodes(db, slice)).ContainsHost(host)
}
// PartitionNodes returns a list of nodes that own a partition.

9
cluster_test.go
View file

@ -36,11 +36,11 @@ func TestCluster_Owners(t *testing.T) {
// Ensure the partitioner can assign a fragment to a partition.
func TestCluster_Partition(t *testing.T) {
if err := quick.Check(func(slice uint64, partitionN int) bool {
if err := quick.Check(func(db string, slice uint64, partitionN int) bool {
c := pilosa.NewCluster()
c.PartitionN = partitionN
partitionID := c.Partition(slice)
partitionID := c.Partition(db, slice)
if partitionID < 0 || partitionID > partitionN {
t.Errorf("partition out of range: slice=%d, p=%d, n=%d", slice, partitionID, partitionN)
}
@ -48,8 +48,9 @@ func TestCluster_Partition(t *testing.T) {
return true
}, &quick.Config{
Values: func(values []reflect.Value, rand *rand.Rand) {
values[0] = reflect.ValueOf(uint64(rand.Uint32()))
values[1] = reflect.ValueOf(rand.Intn(1000) + 1)
values[0], _ = quick.Value(reflect.TypeOf(""), rand)
values[1] = reflect.ValueOf(uint64(rand.Uint32()))
values[2] = reflect.ValueOf(rand.Intn(1000) + 1)
},
}); err != nil {
t.Fatal(err)

14
executor.go
View file

@ -102,7 +102,7 @@ func (e *Executor) executeCall(db string, c pql.Call, slices []uint64, opt *Exec
// executeBitmapCall executes a call that returns a bitmap.
func (e *Executor) executeBitmapCall(db string, c pql.BitmapCall, slices []uint64, opt *ExecOptions) (*Bitmap, error) {
other := NewBitmap()
for node, nodeSlices := range e.slicesByNode(slices) {
for node, nodeSlices := range e.slicesByNode(db, slices) {
// Execute locally if the hostname matches.
if node.Host == e.Host {
for _, slice := range nodeSlices {
@ -183,7 +183,7 @@ func (e *Executor) executeTopN(db string, c *pql.TopN, slices []uint64, opt *Exe
func (e *Executor) executeTopNSlices(db string, c *pql.TopN, slices []uint64, opt *ExecOptions) ([]Pair, error) {
var results []Pair
for node, nodeSlices := range e.slicesByNode(slices) {
for node, nodeSlices := range e.slicesByNode(db, slices) {
// Execute locally if the hostname matches.
if node.Host == e.Host {
for _, slice := range nodeSlices {
@ -334,7 +334,7 @@ func (e *Executor) executeUnionSlice(db string, c *pql.Union, slice uint64) (*Bi
// executeCount executes a count() call.
func (e *Executor) executeCount(db string, c *pql.Count, slices []uint64, opt *ExecOptions) (uint64, error) {
var n uint64
for node, nodeSlices := range e.slicesByNode(slices) {
for node, nodeSlices := range e.slicesByNode(db, slices) {
// Execute locally if the hostname matches.
if node.Host == e.Host {
for _, slice := range nodeSlices {
@ -367,7 +367,7 @@ func (e *Executor) executeProfile(db string, c *pql.Profile, opt *ExecOptions) (
func (e *Executor) executeClearBit(db string, c *pql.ClearBit, opt *ExecOptions) (bool, error) {
slice := c.ProfileID / SliceWidth
ret := false
for _, node := range e.Cluster.SliceNodes(slice) {
for _, node := range e.Cluster.FragmentNodes(db, slice) {
// Update locally if host matches.
if node.Host == e.Host {
f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice)
@ -399,7 +399,7 @@ func (e *Executor) executeSetBit(db string, c *pql.SetBit, opt *ExecOptions) (bo
slice := c.ProfileID / SliceWidth
ret := false
for _, node := range e.Cluster.SliceNodes(slice) {
for _, node := range e.Cluster.FragmentNodes(db, slice) {
// Update locally if host matches.
if node.Host == e.Host {
f, err := e.Index.CreateFragmentIfNotExists(db, c.Frame, slice)
@ -558,10 +558,10 @@ func (e *Executor) exec(node *Node, db string, q *pql.Query, slices []uint64, op
// slicesByNode returns a mapping of nodes to slices.
//
// NOTE: Currently the only primary node is used.
func (e *Executor) slicesByNode(slices []uint64) map[*Node][]uint64 {
func (e *Executor) slicesByNode(db string, slices []uint64) map[*Node][]uint64 {
m := make(map[*Node][]uint64)
for _, slice := range slices {
nodes := e.Cluster.SliceNodes(slice)
nodes := e.Cluster.FragmentNodes(db, slice)
node := nodes[0]
m[node] = append(m[node], slice)

4
fragment.go
View file

@ -1171,7 +1171,7 @@ type FragmentSyncer struct {
// then merges any blocks which have differences.
func (s *FragmentSyncer) SyncFragment() error {
// Determine replica set.
nodes := s.Cluster.SliceNodes(s.Fragment.Slice())
nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice())
// Create a set of blocks.
blockSets := make([][]FragmentBlock, 0, len(nodes))
@ -1247,7 +1247,7 @@ func (s *FragmentSyncer) syncBlock(id int) error {
// Read pairs from each remote block.
var pairSets []PairSet
var clients []*Client
for _, node := range s.Cluster.SliceNodes(f.Slice()) {
for _, node := range s.Cluster.FragmentNodes(f.DB(), f.Slice()) {
if s.Host == node.Host {
continue
}

27
handler.go
View file

@ -89,13 +89,6 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/slices/nodes":
switch r.Method {
case "GET":
h.handleGetSlicesNodes(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/slices/max":
switch r.Method {
case "GET":
@ -103,6 +96,13 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/fragment/nodes":
switch r.Method {
case "GET":
h.handleGetFragmentNodes(w, r)
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}
case "/fragment/data":
switch r.Method {
case "GET":
@ -392,7 +392,7 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
db, frame, slice := req.GetDB(), req.GetFrame(), req.GetSlice()
// Validate that this handler owns the slice.
if !h.Cluster.OwnsSlice(h.Host, slice) {
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
http.Error(w, "host does not own slice", http.StatusPreconditionFailed)
return
}
@ -425,9 +425,10 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) {
w.Write(buf)
}
// handleGetSlicesNodes handles /slices/nodes requests.
func (h *Handler) handleGetSlicesNodes(w http.ResponseWriter, r *http.Request) {
// handleGetFragmentNodes handles /fragment/nodes requests.
func (h *Handler) handleGetFragmentNodes(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
db := q.Get("db")
// Read slice parameter.
slice, err := strconv.ParseUint(q.Get("slice"), 10, 64)
@ -436,8 +437,8 @@ func (h *Handler) handleGetSlicesNodes(w http.ResponseWriter, r *http.Request) {
return
}
// Retrieve slice owner nodes.
nodes := h.Cluster.SliceNodes(slice)
// Retrieve fragment owner nodes.
nodes := h.Cluster.FragmentNodes(db, slice)
// Write to response.
if err := json.NewEncoder(w).Encode(nodes); err != nil {
@ -597,7 +598,7 @@ func (h *Handler) handlePostFrameRestore(w http.ResponseWriter, r *http.Request)
// Loop over each slice and import it if this node owns it.
for slice := uint64(0); slice <= sliceN; slice++ {
// Ignore this slice if we don't own it.
if !h.Cluster.OwnsSlice(h.Host, slice) {
if !h.Cluster.OwnsFragment(h.Host, db, slice) {
continue
}

12
handler_test.go
View file

@ -172,7 +172,7 @@ func TestHandler_Query_Bitmap_JSON(t *testing.T) {
h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d", strings.NewReader("Bitmap(100)")))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]}]}`+"\n" {
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,2097153]}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -204,7 +204,7 @@ func TestHandler_Query_Bitmap_Profiles_JSON(t *testing.T) {
h.ServeHTTP(w, MustNewHTTPRequest("POST", "/query?db=d&profiles=true", strings.NewReader("Bitmap(100)")))
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,65537]}],"profiles":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" {
} else if body := w.Body.String(); body != `{"results":[{"attrs":{"a":"b","c":1,"d":true},"bits":[1,3,66,2097153]}],"profiles":[{"id":3,"attrs":{"x":"y"}},{"id":66,"attrs":{"y":123,"z":false}}]}`+"\n" {
t.Fatalf("unexpected body: %s", body)
}
}
@ -473,18 +473,18 @@ func TestHandler_Version(t *testing.T) {
}
}
// Ensure the handler can return a list of nodes for a slice.
func TestHandler_Slices_Nodes(t *testing.T) {
// Ensure the handler can return a list of nodes for a fragment.
func TestHandler_Fragment_Nodes(t *testing.T) {
h := NewHandler()
h.Cluster = NewCluster(3)
h.Cluster.ReplicaN = 2
w := httptest.NewRecorder()
r := MustNewHTTPRequest("GET", "/slices/nodes?slice=0", nil)
r := MustNewHTTPRequest("GET", "/fragment/nodes?db=X&slice=0", nil)
h.ServeHTTP(w, r)
if w.Code != http.StatusOK {
t.Fatalf("unexpected status code: %d", w.Code)
} else if w.Body.String() != `[{"host":"host2"},{"host":"host0"}]`+"\n" {
} else if w.Body.String() != `[{"host":"host1"},{"host":"host2"}]`+"\n" {
t.Fatalf("unexpected body: %q", w.Body.String())
}
}

2
index.go
View file

@ -214,7 +214,7 @@ func (s *IndexSyncer) SyncIndex() error {
for _, fi := range di.Frames {
for slice := uint64(0); slice <= sliceN; slice++ {
// Ignore slices that this host doesn't own.
if !s.Cluster.OwnsSlice(s.Host, slice) {
if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) {
continue
}