func (s *DatasourceInstance) HandleStreamQuery(ctx context.Context, q *query.Query) *query.ResultStream {
	start := time.Now()
	defer func() {
		end := time.Now()
		s.m.QueryTime.WithValues(q.Args.DB, q.Args.Collection, string(q.Type)).Observe(float64(end.Sub(start)))
	}()
	var result *query.ResultStream
	meta := s.GetActiveMeta()
	collection, err := meta.GetCollection(q.Args.DB, q.Args.ShardInstance, q.Args.Collection)
	if err != nil {
		return &query.ResultStream{Errors: []string{err.Error()}}
	}
	if q.Args.Fields != nil {
		for _, field := range q.Args.Fields {
			if !collection.IsValidProjection(field) {
				return &query.ResultStream{Errors: []string{"invalid projection field " + field}}
			}
		}
		for _, pkeyFieldName := range collection.PrimaryIndex.Fields {
			found := false
			for _, fieldListItem := range q.Args.Fields {
				if pkeyFieldName == fieldListItem {
					found = true
					break
				}
			}
			if !found {
				q.Args.Fields = append(q.Args.Fields, pkeyFieldName)
			}
		}
	}
	switch q.Type {
	case query.FilterStream:
		result = s.Store.FilterStream(ctx, q.Args)
	default:
		return &query.ResultStream{Errors: []string{"invalid stream query"}}
	}
	return result
}
TODO: push sort down to the datasource instance (needed to get the stream corect)
Handle all stream queries
package storagenode
import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/jacksontj/dataman/client"
	"github.com/jacksontj/dataman/datamantype"
	"github.com/jacksontj/dataman/metrics"
	"github.com/jacksontj/dataman/query"
	"github.com/jacksontj/dataman/storagenode/datasource"
	"github.com/jacksontj/dataman/storagenode/join"
	"github.com/jacksontj/dataman/storagenode/metadata"
)
func NewLocalDatasourceInstance(config *DatasourceInstanceConfig, meta *metadata.Meta) (*DatasourceInstance, error) {
	return NewDatasourceInstance(config, NewStaticMetadataStore(meta))
}
func NewDatasourceInstanceDefault(config *DatasourceInstanceConfig) (*DatasourceInstance, error) {
	metaStore, err := NewMetadataStore(config)
	if err != nil {
		return nil, err
	}
	return NewDatasourceInstance(config, metaStore)
}
func NewDatasourceInstance(config *DatasourceInstanceConfig, metaStore StorageMetadataStore) (*DatasourceInstance, error) {
	if config.Registry == nil {
		config.Registry = metrics.NewNamespaceRegistry("datasource_instance")
	}
	datasourceInstance := &DatasourceInstance{
		Config:    config,
		MetaStore: metaStore,
		syncChan:  make(chan chan error),
		registry:  config.Registry,
	}
	datasourceInstance.m = NewDatasourceInstanceMetrics(datasourceInstance.registry)
	datasourceInstance.MutableMetaStore, _ = datasourceInstance.MetaStore.(MutableStorageMetadataStore)
	go datasourceInstance.background()
	if err := datasourceInstance.Sync(); err != nil {
		return nil, err
	}
	var err error
	datasourceInstance.Store, err = config.GetStore(datasourceInstance.GetActiveMeta)
	if err != nil {
		return nil, err
	}
	if StoreSchema, ok := datasourceInstance.Store.(datasource.SchemaInterface); ok {
		datasourceInstance.StoreSchema = StoreSchema
	}
	datasourceInstance.client = &datamanclient.Client{NewDatasourceInstanceTransport(datasourceInstance)}
	return datasourceInstance, nil
}
type DatasourceInstance struct {
	Config           *DatasourceInstanceConfig
	MetaStore        StorageMetadataStore
	MutableMetaStore MutableStorageMetadataStore
	StoreSchema datasource.SchemaInterface
	Store       datasource.DataInterface
	meta atomic.Value
	activeMeta atomic.Value
	client *datamanclient.Client
	syncChan chan chan error
	schemaLock sync.Mutex
	registry metrics.Registry
	m        DatasourceInstanceMetrics
}
func (s *DatasourceInstance) Sync() error {
	errChan := make(chan error, 1)
	s.syncChan <- errChan
	return <-errChan
}
func (s *DatasourceInstance) GetActiveMeta() *metadata.Meta {
	return s.activeMeta.Load().(*metadata.Meta)
}
func (s *DatasourceInstance) GetMeta() *metadata.Meta {
	return s.meta.Load().(*metadata.Meta)
}
func (s *DatasourceInstance) background() {
	interval := time.Second  
	ticker := time.NewTicker(interval)
	for {
		select {
		case <-ticker.C:  
			s.RefreshMeta()
		case retChan := <-s.syncChan:  
			err := s.RefreshMeta()
			retChan <- err
			ticker = time.NewTicker(interval)
		}
	}
}
func (s *DatasourceInstance) RefreshMeta() error {
	s.schemaLock.Lock()
	defer s.schemaLock.Unlock()
	return s.refreshMeta()
}
func (s *DatasourceInstance) refreshMeta() (err error) {
	start := time.Now()
	defer func() {
		end := time.Now()
		if err == nil {
			s.m.MetaLastSync.WithValues("success").Observe(float64(end.Unix()))
			s.m.MetaLastDuration.WithValues("success").Observe(float64(end.Sub(start)))
		} else {
			s.m.MetaLastSync.WithValues("failure").Observe(float64(end.Unix()))
			s.m.MetaLastDuration.WithValues("failure").Observe(float64(end.Sub(start)))
		}
	}()
	ctx := context.Background()
	if meta, err := s.MetaStore.GetMeta(ctx); err == nil {
		s.meta.Store(meta)
	} else {
		return err
	}
	meta, err := s.MetaStore.GetMeta(ctx)
	if err != nil {
		return err
	}
	if !s.Config.SkipProvisionTrim {
		for key, database := range meta.Databases {
			if database.ProvisionState != metadata.Active {
				delete(meta.Databases, key)
			} else {
				for key, shardInstance := range database.ShardInstances {
					if shardInstance.ProvisionState != metadata.Active {
						delete(database.ShardInstances, key)
					} else {
						for key, collection := range shardInstance.Collections {
							if collection.ProvisionState != metadata.Active {
								delete(shardInstance.Collections, key)
							} else {
								for key, field := range collection.Fields {
									if field.ProvisionState != metadata.Active {
										delete(collection.Fields, key)
									}
								}
								for key, index := range collection.Indexes {
									if index.ProvisionState != metadata.Active {
										delete(collection.Indexes, key)
									}
								}
							}
						}
					}
				}
			}
		}
		for key, field := range meta.Fields {
			if field.ProvisionState != metadata.Active {
				delete(meta.Fields, key)
			}
		}
		for key, collection := range meta.Collections {
			if collection.ProvisionState != metadata.Active {
				delete(meta.Collections, key)
			}
		}
	}
	s.activeMeta.Store(meta)
	metadata.FieldTypeRegistry.Merge(meta.FieldTypeRegistry)
	return nil
}
func (s *DatasourceInstance) HandleQuery(ctx context.Context, q *query.Query) *query.Result {
	start := time.Now()
	defer func() {
		end := time.Now()
		s.m.QueryTime.WithValues(q.Args.DB, q.Args.Collection, string(q.Type)).Observe(float64(end.Sub(start)))
	}()
	var result *query.Result
	meta := s.GetActiveMeta()
	collection, err := meta.GetCollection(q.Args.DB, q.Args.ShardInstance, q.Args.Collection)
	if err != nil {
		return &query.Result{Errors: []string{err.Error()}}
	}
	if q.Args.Fields != nil {
		for _, field := range q.Args.Fields {
			if !collection.IsValidProjection(field) {
				return &query.Result{Errors: []string{"invalid projection field " + field}}
			}
		}
	}
	if q.Args.Join != nil {
		if meta.Databases[q.Args.DB].ShardInstances[q.Args.ShardInstance].Count != 1 {
			return &query.Result{Errors: []string{"Datasource-level joins only supported on collections with one shardInstance"}}
		}
	}
	var subRecordList []*join.WriteJoinSubrecord
	switch q.Type {
	case query.Set:
		if q.Args.Join != nil {
			if srecordMap, validationError, err := join.DoWriteJoins(ctx, s.client, q, meta, collection, q.Args.Join, q.Args.Record); validationError != nil {
				return &query.Result{ValidationError: validationError.Error()}
			} else if err != nil {
				return &query.Result{Errors: []string{err.Error()}}
			} else {
				subRecordList = srecordMap
			}
		}
		if insertValidationResultMap := collection.ValidateRecordInsert(q.Args.Record); !insertValidationResultMap.IsValid() {
			return &query.Result{ValidationError: insertValidationResultMap}
		}
	case query.Insert:
		if validationResultMap := collection.ValidateRecordInsert(q.Args.Record); !validationResultMap.IsValid() {
			return &query.Result{ValidationError: validationResultMap}
		}
	case query.InsertMany:
		for _, rec := range q.Args.Records {
			if validationResultMap := collection.ValidateRecordInsert(rec); !validationResultMap.IsValid() {
				return &query.Result{ValidationError: validationResultMap}
			}
		}
	case query.Update:
		if validationResultMap := collection.ValidateRecordUpdate(q.Args.Record); !validationResultMap.IsValid() {
			return &query.Result{ValidationError: validationResultMap}
		}
	}
	switch q.Type {
	case query.Get:
		result = s.Store.Get(ctx, q.Args)
		if q.Args.Join != nil {
			if err := join.DoReadJoins(ctx, s.client, q, meta, collection, q.Args.Join, result.Return); err != nil {
				result.Errors = []string{err.Error()}
				return result
			}
		}
	case query.Set:
		result = s.Store.Set(ctx, q.Args)
		for _, subRecord := range subRecordList {
			subRecord.Apply(result.Return[0])
		}
	case query.Insert:
		result = s.Store.Insert(ctx, q.Args)
	case query.InsertMany:
		result = s.Store.InsertMany(ctx, q.Args)
	case query.Update:
		result = s.Store.Update(ctx, q.Args)
	case query.Delete:
		result = s.Store.Delete(ctx, q.Args)
	case query.Filter:
		result = s.Store.Filter(ctx, q.Args)
		if q.Args.Join != nil {
			if err := join.DoReadJoins(ctx, s.client, q, meta, collection, q.Args.Join, result.Return); err != nil {
				result.Errors = []string{err.Error()}
				return result
			}
		}
	case query.Aggregate:
		result = s.Store.Aggregate(ctx, q.Args)
	default:
		return &query.Result{Errors: []string{"Unsupported query type " + string(q.Type)}}
	}
	if q.Args.Fields != nil {
		for _, pkeyFieldName := range collection.PrimaryIndex.Fields {
			found := false
			for _, fieldListItem := range q.Args.Fields {
				if pkeyFieldName == fieldListItem {
					found = true
					break
				}
			}
			if !found {
				q.Args.Fields = append(q.Args.Fields, pkeyFieldName)
			}
		}
	}
	return result
}