Merge pull request #1 from gigapi/feat/layers

data layers support

Author: akvlad

.gitignore

@@ -1 +1,2 @@
 /.idea/
+_testdata

json_db_index.go

@@ -9,58 +9,83 @@ import (
 )
 
 type jsonDBIndex struct {
-	root string
+	layers []jsonLayer
 }
 
-/*func NewJSONDBIndex(root string) DBIndex {
+func NewJSONDBIndex(layers []Layer) DBIndex {
+	jLayers := make([]jsonLayer, len(layers))
+	for i, layer := range layers {
+		jLayers[i] = layer2JsonLayer(layer)
+	}
 	return &jsonDBIndex{
-		root: root,
+		layers: jLayers,
 	}
-}*/
+}
 
 func (j *jsonDBIndex) Databases() ([]string, error) {
-	ents, err := os.ReadDir(j.root)
-	if err != nil {
-		return nil, err
-	}
-	var res []string
-	for _, ent := range ents {
-		if ent.IsDir() {
-			res = append(res, ent.Name())
+	res := map[string]bool{}
+	for _, l := range j.layers {
+		ents, err := os.ReadDir(path.Join(l.Path))
+		if errors.Is(err, os.ErrNotExist) {
+			continue
+		}
+		if err != nil {
+			return nil, err
 		}
+		for _, ent := range ents {
+			if ent.IsDir() {
+				res[ent.Name()] = true
+			}
+		}
+	}
+	_res := make([]string, 0, len(res))
+	for k := range res {
+		_res = append(_res, k)
 	}
-	return res, nil
+	return _res, nil
 }
 
 func (j *jsonDBIndex) Tables(database string) ([]string, error) {
-	ents, err := os.ReadDir(path.Join(j.root, database))
-	if errors.Is(err, os.ErrNotExist) {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-	var res []string
-	for _, ent := range ents {
-		if ent.IsDir() {
-			res = append(res, ent.Name())
+	res := map[string]bool{}
+	for _, l := range j.layers {
+		ents, err := os.ReadDir(path.Join(l.Path, database))
+		if errors.Is(err, os.ErrNotExist) {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		for _, ent := range ents {
+			if ent.IsDir() {
+				res[ent.Name()] = true
+			}
 		}
 	}
-	return res, nil
+	_res := make([]string, 0, len(res))
+	for k := range res {
+		_res = append(_res, k)
+	}
+	return _res, nil
 }
 
-func (j *jsonDBIndex) Paths(database string, table string) []string {
-	root := path.Join(j.root, database, table)
-	var res []string
-	filepath.Walk(path.Join(j.root, database, table), func(path string, info os.FileInfo, err error) error {
-		if !info.IsDir() {
+func (j *jsonDBIndex) Paths(database string, table string) ([]string, error) {
+	res := map[string]bool{}
+	for _, l := range j.layers {
+		root := path.Join(l.Path, database, table)
+		filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
+			if !info.IsDir() {
+				return nil
+			}
+			if strings.HasPrefix(info.Name(), "hour=") {
+				res[path[len(root)+1:]] = true
+				return filepath.SkipDir
+			}
 			return nil
-		}
-		if strings.HasPrefix(info.Name(), "hour=") {
-			res = append(res, path[len(root):])
-			return filepath.SkipDir
-		}
-		return nil
-	})
-	return res
+		})
+	}
+	_res := make([]string, 0, len(res))
+	for k := range res {
+		_res = append(_res, k)
+	}
+	return _res, nil
 }

json_drop_planner.go

@@ -0,0 +1,37 @@
+package metadata
+
+import "path"
+
+func (J *JSONIndex) GetDropQueue(writerId string, layer string) (DropPlan, error) {
+	parts := J.parts[layer]
+	if parts == nil {
+		return DropPlan{}, nil
+	}
+	for _, idx := range parts {
+		p, err := idx.GetDropPlanner().GetDropQueue(writerId, layer)
+		if err != nil {
+			return DropPlan{}, err
+		}
+		if p.Path != "" {
+			return p, nil
+		}
+	}
+	return DropPlan{}, nil
+}
+
+func (J *JSONIndex) GetDropPlanner() TableDropPlanner {
+	return J
+}
+
+func (J *JSONIndex) RmFromDropQueue(plan DropPlan) Promise[int32] {
+	l := J.parts[plan.Layer]
+	if l == nil {
+		return Fulfilled(nil, int32(0))
+	}
+	dir := path.Dir(plan.Path)
+	part := l[dir]
+	if part != nil {
+		return part.RmFromDropQueue(plan)
+	}
+	return Fulfilled(nil, int32(0))
+}