Merge pull request #560 from v3io/development

dev -> master

Author: dinal

backends/kv/backend.go

@@ -41,6 +41,7 @@ type Backend struct {
 	inactivityTimeout  time.Duration
 	framesConfig       *frames.Config
 	v3ioContext        v3io.Context
+	maxRecordsInfer    int
 }
 
 // NewBackend returns a new NoSQL (key/value) backend
@@ -52,6 +53,7 @@ func NewBackend(logger logger.Logger, v3ioContext v3io.Context, config *frames.B
 		framesConfig:       framesConfig,
 		v3ioContext:        v3ioContext,
 		inactivityTimeout:  0,
+		maxRecordsInfer:    config.MaxRecordsInferSchema,
 	}
 	return &newBackend, nil
 }

backends/kv/inferschema.go

@@ -50,18 +50,17 @@ func (b *Backend) inferSchema(request *frames.ExecRequest) error {
 	if val, ok := request.Proto.Args["key"]; ok {
 		keyField = val.GetSval()
 	}
-	maxrec := 10
 
 	input := v3io.GetItemsInput{Path: table, Filter: "", AttributeNames: []string{"*"}}
 	b.logger.DebugWith("GetItems for schema", "input", input)
-	iter, err := v3ioutils.NewAsyncItemsCursor(container, &input, b.numWorkers, []string{}, b.logger, 0, []string{table}, "", "")
+	iter, err := v3ioutils.NewAsyncItemsCursor(container, &input, b.numWorkers, []string{}, b.logger, b.maxRecordsInfer, []string{table}, "", "")
 	if err != nil {
 		return err
 	}
 
 	var rowSet []map[string]interface{}
 
-	for rowNum := 0; rowNum < maxrec && iter.Next(); rowNum++ {
+	for rowNum := 0; iter.Next(); rowNum++ {
 		row := iter.GetFields()
 		rowSet = append(rowSet, row)
 	}

backends/kv/reader.go

@@ -23,6 +23,7 @@ package kv
 import (
 	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/pkg/errors"
 	"github.com/v3io/frames"
@@ -338,6 +339,16 @@ func init() {
 	}
 }
 
+func filterPartitions(dirs []v3io.CommonPrefix) []v3io.CommonPrefix {
+	tmp := dirs[:0]
+	for _, p := range dirs {
+		if strings.Contains(p.Prefix, "=") {
+			tmp = append(tmp, p)
+		}
+	}
+	return tmp
+}
+
 func (kv *Backend) getPartitions(path string, container v3io.Container) ([]string, error) {
 	var partitions []string
 	var done bool
@@ -352,6 +363,7 @@ func (kv *Backend) getPartitions(path string, container v3io.Container) ([]strin
 			res.Release() // Releasing underlying fasthttp response
 		}
 		out := res.Output.(*v3io.GetContainerContentsOutput)
+		out.CommonPrefixes = filterPartitions(out.CommonPrefixes)
 		if len(out.CommonPrefixes) > 0 {
 			for _, partition := range out.CommonPrefixes {
 				parts, err := kv.getPartitions(partition.Prefix, container)

config.go

@@ -142,6 +142,7 @@ type BackendConfig struct {
 	V3ioGoRequestChanLength int    `json:"v3ioGoRequestChanLength"`
 	MaxConnections          int    `json:"maxConnections"`
 	DialTimeoutSeconds      int    `json:"dialTimeoutSeconds"`
+	MaxRecordsInferSchema   int    `json:"maxRecordsInferSchema"`
 
 	// backend specific options
 	Options map[string]interface{} `json:"options"`
@@ -232,4 +233,8 @@ func initBackendDefaults(cfg *BackendConfig, framesConfig *Config) {
 	if cfg.V3ioGoRequestChanLength == 0 {
 		cfg.V3ioGoRequestChanLength = cfg.V3ioGoWorkers * 256
 	}
+
+	if cfg.MaxRecordsInferSchema == 0 {
+		cfg.MaxRecordsInferSchema = 10
+	}
 }

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/pkg/errors v0.8.1
 	github.com/stretchr/testify v1.4.0
 	github.com/v3io/v3io-go v0.1.9
-	github.com/v3io/v3io-tsdb v0.11.3
+	github.com/v3io/v3io-tsdb v0.11.4
 	github.com/valyala/fasthttp v1.2.0
 	github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2
 	golang.org/x/net v0.0.0-20190311183353-d8887717615a

go.sum

@@ -71,6 +71,7 @@ github.com/v3io/v3io-go v0.1.9/go.mod h1:5poBlcjZG5TiexRTYI44PE6tHzZz5Z60w+iS899
 github.com/v3io/v3io-tsdb v0.11.2 h1:knT+IwB+Bkd1aRPTIKltBnYrue1z+2EXmESJLkCexe8=
 github.com/v3io/v3io-tsdb v0.11.2/go.mod h1:l+WxedjLmjy/TThj2vhSW/OnpRw8C7dOntIXHmM/a7I=
 github.com/v3io/v3io-tsdb v0.11.3/go.mod h1:kp586KxTfROIGwb/nzNxwDbX2Wterxro+HbiZHmK548=
+github.com/v3io/v3io-tsdb v0.11.4/go.mod h1:kp586KxTfROIGwb/nzNxwDbX2Wterxro+HbiZHmK548=
 github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v1.2.0 h1:dzZJf2IuMiclVjdw0kkT+f9u4YdrapbNyGAN47E/qnk=