//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package sql

import (
	"errors"

	"github.com/bcicen/jstream"
	"github.com/minio/simdjson-go"
)

var (
	errKeyLookup                  = errors.New("Cannot look up key in non-object value")
	errIndexLookup                = errors.New("Cannot look up array index in non-array value")

		}
		return nil, errParquetParsingError(err)
	}

	kvs := jstream.KVS{}
	for _, col := range pr.r.Columns() {

		var value interface{}
		if v, ok := nextRow[col.FlatName()]; ok {
			value, err = convertFromAnnotation(col.Element(), v)
			if err != nil {
				return nil, errParquetParsingError(err)
			}
		}
		kvs = append(kvs, jstream.KV{Key: col.FlatName(), Value: value})
	}

	// Reuse destination if we can.

		return nil, err
	}
	if cap(dst.KVS) < len(elems.Elements) {
		dst.KVS = make(jstream.KVS, 0, len(elems.Elements))
	}
	for _, elem := range elems.Elements {
		v, err := sql.IterToValue(elem.Iter)
		if err != nil {
			v, err = elem.Iter.Interface()
			if err != nil {
				panic(err)
			}
		}
		dst.KVS = append(dst.KVS, jstream.KV{
			Key:   elem.Name,
			Value: v,
		})
	}
	return dst, nil
}

    })
```

Sample call:
```go
    // Get a connection to the remote host
    conn := manager.Connection(host).Subroute("asubroute")
	
    payload := []byte("request")
    stream, err := conn.NewStream(ctx, grid.HandlerDiskInfo, payload)
	if err != nil {
        return err
    }
    // Read results from the stream
    err = stream.Results(func(result []byte) error {

	w.Flush()
	return w.Error()
}

// WriteJSON - encodes to JSON data.
func (r *Record) WriteJSON(writer io.Writer) error {
	var kvs jstream.KVS = make([]jstream.KV, 0, len(r.columnNames))
	for i, cn := range r.columnNames {
		if i < len(r.csvRecord) {
			kvs = append(kvs, jstream.KV{Key: cn, Value: r.csvRecord[i]})
		}
	}
	return json.NewEncoder(writer).Encode(kvs)
}

package sql

import (
	"bytes"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"testing"

	"github.com/alecthomas/participle"
	"github.com/bcicen/jstream"
)

func getJSONStructs(b []byte) ([]interface{}, error) {
	dec := jstream.NewDecoder(bytes.NewBuffer(b), 0).ObjectAsKVS()
	var result []interface{}
	for parsedVal := range dec.Stream() {
		result = append(result, parsedVal.Value)
	}

	if false {
		sr.SkipEncrypted(true)
		return sr.DebugStream(os.Stdout)
	}
	extracted := false
	for {
		stream, err := sr.NextStream()
		if err != nil {
			if err == io.EOF {
				if extracted {
					return nil
				}
				return errors.New("no data found on stream")
			}
			if errors.Is(err, estream.ErrNoKey) {
				if stream.Name == "inspect.zip" {
					return errors.New("incorrect private key")
				}

					continue
				}
				dst, ok := r.kvDstPool.Get().([]jstream.KVS)
				if !ok {
					dst = make([]jstream.KVS, 0, 1000)
				}

				d := jstream.NewDecoder(bytes.NewBuffer(in.input), 0).ObjectAsKVS()
				stream := d.Stream()
				all := dst[:0]
				for mv := range stream {
					var kvs jstream.KVS
					if mv.ValueType == jstream.Object {
						// This is a JSON object type (that preserves key

	<-gotCall
	remote.debugMsg(debugKillInbound)
	local.debugMsg(debugKillInbound)
	<-gotResp

	// Must reconnect
	errFatal(remoteConn.WaitForConnect(context.Background()))

	stream, err := remoteConn.NewStream(context.Background(), handlerTest2, []byte(testPayload))
	errFatal(err)
	go func() {
		for resp := range stream.responses {
			t.Log("Resp:", resp, err)
		}
		gotResp <- struct{}{}
	}()

	<-gotCall

		}
		return nil, io.EOF
	}

	var kvs jstream.KVS
	if v.ValueType == jstream.Object {
		// This is a JSON object type (that preserves key
		// order)
		kvs = v.Value.(jstream.KVS)
	} else {
		// To be AWS S3 compatible Select for JSON needs to
		// output non-object JSON as single column value
		// i.e. a map with `_1` as key and value as the
		// non-object.
		kvs = jstream.KVS{jstream.KV{Key: "_1", Value: v.Value}}
	}