Merge pull request #92 from v3io/development

development --> Master 0.0.8

Author: gshatz

Diff for: pkg/partmgr/partmgr.go

@@ -41,19 +41,7 @@ func NewPartitionMngr(cfg *config.Schema, partPath string, cont *v3io.Container)
 		return nil, err
 	}
 	newMngr := &PartitionManager{cfg: cfg, path: partPath, cyclic: false, container: cont, currentPartitionInterval: currentPartitionInterval}
-	for _, part := range cfg.Partitions {
-		partPath := path.Join(newMngr.path, strconv.FormatInt(part.StartTime/1000, 10)) + "/"
-		newPart, err := NewDBPartition(newMngr, part.StartTime, partPath)
-		if err != nil {
-			return nil, err
-		}
-		newMngr.partitions = append(newMngr.partitions, newPart)
-		if newMngr.headPartition == nil {
-			newMngr.headPartition = newPart
-		} else if newMngr.headPartition.startTime < newPart.startTime {
-			newMngr.headPartition = newPart
-		}
-	}
+	newMngr.updatePartitions(cfg)
 	return newMngr, nil
 }
 
@@ -193,6 +181,41 @@ func (p *PartitionManager) updatePartitionInSchema(partition *DBPartition) error
 	return err
 }
 
+func (p *PartitionManager) ReadAndUpdateSchema() error {
+	fullPath := path.Join(p.path, config.SCHEMA_CONFIG)
+	resp, err := p.container.Sync.GetObject(&v3io.GetObjectInput{Path: fullPath})
+	if err != nil {
+		return errors.Wrap(err, "Failed to read schema at path: "+fullPath)
+	}
+
+	schema := config.Schema{}
+	err = json.Unmarshal(resp.Body(), &schema)
+	if err != nil {
+		return errors.Wrap(err, "Failed to Unmarshal schema at path: "+fullPath)
+	}
+	p.cfg = &schema
+	p.updatePartitions(&schema)
+	return nil
+}
+
+func (p *PartitionManager) updatePartitions(schema *config.Schema) error {
+	p.partitions = []*DBPartition{}
+	for _, part := range schema.Partitions {
+		partPath := path.Join(p.path, strconv.FormatInt(part.StartTime/1000, 10)) + "/"
+		newPart, err := NewDBPartition(p, part.StartTime, partPath)
+		if err != nil {
+			return err
+		}
+		p.partitions = append(p.partitions, newPart)
+		if p.headPartition == nil {
+			p.headPartition = newPart
+		} else if p.headPartition.startTime < newPart.startTime {
+			p.headPartition = newPart
+		}
+	}
+	return nil
+}
+
 func (p *PartitionManager) PartsForRange(mint, maxt int64) []*DBPartition {
 	var parts []*DBPartition
 	for _, part := range p.partitions {

Diff for: pkg/querier/querier.go

@@ -65,7 +65,10 @@ func (q *V3ioQuerier) selectQry(name, functions string, step int64, windows []in
 
 	filter = strings.Replace(filter, "__name__", "_name", -1)
 	q.logger.DebugWith("Select query", "func", functions, "step", step, "filter", filter, "window", windows)
-
+	err := q.partitionMngr.ReadAndUpdateSchema()
+	if err != nil {
+		return nullSeriesSet{}, err
+	}
 	parts := q.partitionMngr.PartsForRange(q.mint, q.maxt)
 	if len(parts) == 0 {
 		return nullSeriesSet{}, nil