func (s *service) Query(request *QueryRequest) *QueryResponse {
	var response = &QueryResponse{
		BaseResponse: NewBaseOkResponse(),
		Records:      make([]map[string]interface{}, 0),
		Validation:   &assertly.Validation{},
	}
	if !validateDatastores(s.registry, response.BaseResponse, request.Datastore) {
		return response
	}
	manager := s.registry.Get(request.Datastore)
	macroEvaluator := assertly.NewDefaultMacroEvaluator()
	context := toolbox.NewContext()
	state := s.getContextState(context)
	SQL, err := macroEvaluator.Expand(context, request.SQL)
	if err != nil {
		response.SetError(err)
		return response
	}
	if state != nil {
		SQL = state.Expand(toolbox.AsString(SQL))
	}
	err = manager.ReadAll(&response.Records, toolbox.AsString(SQL), nil, nil)
	if err != nil {
		response.SetError(err)
		return response
	}
	if len(request.Expect) > 0 {
		response.Validation, err = assertly.Assert(request.Expect, response.Records, assertly.NewDataPath("sql"))
		response.SetError(err)
	}
	return response
}
Query returns query from database
package dsunit
import (
	"fmt"
	"github.com/pkg/errors"
	"github.com/viant/assertly"
	"github.com/viant/dsc"
	"github.com/viant/dsunit/script"
	"github.com/viant/toolbox"
	"github.com/viant/toolbox/data"
	"github.com/viant/toolbox/storage"
	"github.com/viant/toolbox/url"
	"io"
	"strings"
	"sync"
	"time"
)
var batchSize = 200
var SubstitutionMapKey = (*data.Map)(nil)
type Service interface {
	Registry() dsc.ManagerRegistry
	Register(request *RegisterRequest) *RegisterResponse
	Recreate(request *RecreateRequest) *RecreateResponse
	RunSQL(request *RunSQLRequest) *RunSQLResponse
	RunScript(request *RunScriptRequest) *RunSQLResponse
	AddTableMapping(request *MappingRequest) *MappingResponse
	Init(request *InitRequest) *InitResponse
	Prepare(request *PrepareRequest) *PrepareResponse
	Expect(request *ExpectRequest) *ExpectResponse
	Query(request *QueryRequest) *QueryResponse
	Sequence(request *SequenceRequest) *SequenceResponse
	Freeze(request *FreezeRequest) *FreezeResponse
	Dump(request *DumpRequest) *DumpResponse
	Compare(request *CompareRequest) *CompareResponse
	CheckSchema(request *CheckSchemaRequest) *CheckSchemaResponse
	Ping(request *PingRequest) *PingResponse
	SetContext(context toolbox.Context)
}
type service struct {
	registry        dsc.ManagerRegistry
	mapper          *Mapper
	context         toolbox.Context
	adminDatastores map[string]string
}
func (s *service) Registry() dsc.ManagerRegistry {
	return s.registry
}
func (s *service) Register(request *RegisterRequest) *RegisterResponse {
	var response = &RegisterResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	var err = request.Init()
	if err == nil {
		err = request.Validate()
	}
	if err != nil {
		response.SetError(err)
		return response
	}
	config, err := expandDscConfig(request.Config, request.Datastore)
	if err != nil {
		response.SetError(err)
		return response
	}
	manager, err := dsc.NewManagerFactory().Create(config)
	if err == nil {
		s.registry.Register(request.Datastore, manager)
		if len(request.Tables) > 0 {
			for _, table := range request.Tables {
				_ = manager.TableDescriptorRegistry().Register(table)
			}
		}
	}
	if err != nil {
		response.SetError(err)
	}
	if request.Ping {
		request.PingRequest.Datastore = request.Datastore
		pingResponse := s.Ping(&request.PingRequest)
		response.SetError(pingResponse.Error())
	}
	return response
}
func (s *service) Recreate(request *RecreateRequest) *RecreateResponse {
	var response = &RecreateResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	if request.AdminDatastore == "" {
		request.AdminDatastore = request.Datastore
	}
	err := RecreateDatastore(request.AdminDatastore, request.Datastore, s.registry)
	response.SetError(err)
	return response
}
func (s *service) expandSQLIfNeeded(request *RunSQLRequest, manager dsc.Manager) []string {
	if !request.Expand {
		return request.SQL
	}
	context := s.newContext(manager)
	state := s.getContextState(context)
	if state == nil {
		return request.SQL
	}
	result := make([]string, 0)
	for _, SQL := range request.SQL {
		result = append(result, state.ExpandAsText(SQL))
	}
	return result
}
func (s *service) RunSQL(request *RunSQLRequest) *RunSQLResponse {
	var response = &RunSQLResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	if !validateDatastores(s.registry, response.BaseResponse, request.Datastore) {
		return response
	}
	manager := s.registry.Get(request.Datastore)
	var SQL = s.expandSQLIfNeeded(request, manager)
	results, err := manager.ExecuteAll(SQL)
	if err != nil {
		response.SetError(err)
		return response
	}
	for _, result := range results {
		count, err := result.RowsAffected()
		if err != nil {
			response.SetError(err)
			return response
		}
		response.RowsAffected += int(count)
	}
	return response
}
func (s *service) RunScript(request *RunScriptRequest) *RunSQLResponse {
	var response = &RunSQLResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	if len(request.Scripts) == 0 {
		return response
	}
	var SQL = []string{}
	var err error
	var storageService storage.Service
	var storageObject storage.Object
	for _, resource := range request.Scripts {
		err = resource.Init()
		if err != nil {
			break
		}
		var reader io.ReadCloser
		if storageService, err = storage.NewServiceForURL(resource.URL, resource.Credentials); err == nil {
			if storageObject, err = storageService.StorageObject(resource.URL); err == nil {
				if reader, err = storageService.Download(storageObject); err == nil {
					defer reader.Close()
					SQL = append(SQL, script.ParseWithReader(reader)...)
				}
			}
		}
		if err != nil {
			break
		}
	}
	if err != nil {
		response.SetError(err)
		return response
	}
	return s.RunSQL(&RunSQLRequest{
		Expand:    request.Expand,
		Datastore: request.Datastore,
		SQL:       SQL,
	})
}
func (s *service) AddTableMapping(request *MappingRequest) *MappingResponse {
	var response = &MappingResponse{
		BaseResponse: NewBaseOkResponse(),
		Tables:       make([]string, 0),
	}
	err := request.Init()
	if err == nil {
		err = request.Validate()
	}
	if err != nil {
		response.SetError(err)
		return response
	}
	if len(request.Mappings) == 0 {
		return response
	}
	for _, mapping := range request.Mappings {
		s.mapper.Add(mapping)
		response.Tables = append(response.Tables, mapping.Tables()...)
	}
	return response
}
func (s *service) Init(request *InitRequest) *InitResponse {
	var response = &InitResponse{BaseResponse: NewBaseOkResponse()}
	err := request.Init()
	if err == nil {
		err = request.Validate()
	}
	if err != nil {
		response.SetError(err)
		return response
	}
	registerRequest := request.RegisterRequest
	if registerRequest.Datastore == "" {
		registerRequest.Datastore = request.Datastore
	}
	registerRequests := []*RegisterRequest{registerRequest, request.Admin}
	for _, registerRequest := range registerRequests {
		if registerRequest == nil {
			continue
		}
		serviceResponse := s.Register(registerRequest)
		if serviceResponse.Status != StatusOk {
			response.BaseResponse = serviceResponse.BaseResponse
			return response
		}
	}
	var adminDatastore = registerRequest.Datastore
	if request.Admin != nil {
		adminDatastore = request.Admin.Datastore
	}
	s.adminDatastores[request.Datastore] = adminDatastore
	if request.Recreate {
		serviceResponse := s.Recreate(NewRecreateRequest(registerRequest.Datastore, adminDatastore))
		if serviceResponse.Status != StatusOk {
			response.BaseResponse = serviceResponse.BaseResponse
			return response
		}
	} else {
		err := s.createDbIfDoesNotExists(registerRequest.Datastore, adminDatastore)
		if err != nil {
			response.SetError(err)
			return response
		}
	}
	if request.RunScriptRequest != nil && len(request.Scripts) > 0 {
		if request.RunScriptRequest.Datastore == "" {
			request.RunScriptRequest.Datastore = request.Datastore
		}
		serviceResponse := s.RunScript(request.RunScriptRequest)
		if serviceResponse.Status != StatusOk {
			response.BaseResponse = serviceResponse.BaseResponse
			return response
		}
	}
	if request.MappingRequest != nil && len(request.Mappings) > 0 {
		serviceResponse := s.AddTableMapping(request.MappingRequest)
		if serviceResponse.Status != StatusOk {
			response.BaseResponse = serviceResponse.BaseResponse
			return response
		}
		response.Tables = serviceResponse.Tables
	}
	return response
}
func (s *service) getContextState(context toolbox.Context) *data.Map {
	if !context.Contains(SubstitutionMapKey) {
		return nil
	}
	var state = context.GetOptional(SubstitutionMapKey).(*data.Map)
	return state
}
func (s *service) newContext(manager dsc.Manager) toolbox.Context {
	context := toolbox.NewContext()
	if s.context != nil {
		context = s.context.Clone()
	}
	dialect := dsc.GetDatastoreDialect(manager.Config().DriverName)
	_ = context.Replace((*dsc.Manager)(nil), &manager)
	_ = context.Replace((*dsc.DatastoreDialect)(nil), &dialect)
	return context
}
func (s *service) deleteDatasetIfNeeded(datastore string, dataset *Dataset, table *dsc.TableDescriptor, response *PrepareResponse, context toolbox.Context, manager dsc.Manager, connection dsc.Connection) (err error) {
	if dataset.Records.ShouldDeleteAll() {
		sqlResult, err := manager.ExecuteOnConnection(connection, fmt.Sprintf("DELETE FROM %s", table.Table), nil)
		if err != nil {
			return err
		}
		deleted, _ := sqlResult.RowsAffected()
		response.Modification[dataset.Table].Deleted = int(deleted)
		_ = connection.Commit()
		_ = connection.Begin()
		_, err = s.disableForeignKeyCheck(datastore, connection, true)
	}
	return err
}
func (s *service) getTableDescriptor(dataset *Dataset, manager dsc.Manager, context toolbox.Context) (*dsc.TableDescriptor, error) {
	macroEvaluator := assertly.NewDefaultMacroEvaluator()
	expandedTable, err := macroEvaluator.Expand(context, dataset.Table)
	if err != nil {
		return nil, err
	}
	var state = s.getContextState(context)
	tableName := state.ExpandAsText(toolbox.AsString(expandedTable))
	table := manager.TableDescriptorRegistry().Get(tableName)
	if table == nil {
		table = &dsc.TableDescriptor{Table: tableName}
		_ = manager.TableDescriptorRegistry().Register(table)
	}
	var autoincrement = dataset.Records.Autoincrement()
	var uniqueKeys = dataset.Records.UniqueKeys()
	var fromQuery, fromQueryAlias = dataset.Records.FromQuery()
	if !table.Autoincrement {
		table.Autoincrement = autoincrement
	}
	if fromQuery != "" {
		state := s.getContextState(context)
		if state != nil {
			fromQuery = state.ExpandAsText(fromQuery)
		}
	}
	table.FromQuery = fromQuery
	table.FromQueryAlias = fromQueryAlias
	if len(table.PkColumns) == 0 {
		table.PkColumns = uniqueKeys
	} else if len(uniqueKeys) == 0 {
		if len(dataset.Records) > 0 {
			if len(dataset.Records[0]) == 0 {
				dataset.Records =
					append([]map[string]interface{}{{
						assertly.IndexByDirective: table.PkColumns,
					},
					}, dataset.Records...)
			} else {
				dataset.Records[0][assertly.IndexByDirective] = table.PkColumns
			}
		}
	}
	var columns = dataset.Records.Columns()
	if len(columns) > 0 {
		table.Columns = columns
	}
	return table, nil
}
func (s *service) populate(datastore string, dataset *Dataset, response *PrepareResponse, context toolbox.Context, manager dsc.Manager, connection dsc.Connection) (err error) {
	if s.mapper.Has(dataset.Table) {
		datasets := s.mapper.Map(dataset)
		for _, dataset := range datasets {
			if err = s.populate(datastore, dataset, response, context, manager, connection); err != nil {
				return err
			}
		}
		return
	}
	if len(response.Modification) == 0 {
		response.Modification = make(map[string]*ModificationInfo)
	}
	response.Modification[dataset.Table] = &ModificationInfo{Subject: dataset.Table, Method: "persist"}
	var modification = response.Modification[dataset.Table]
	var table *dsc.TableDescriptor
	if table, err = s.getTableDescriptor(dataset, manager, context); err != nil {
		return err
	}
	if err = s.deleteDatasetIfNeeded(datastore, dataset, table, response, context, manager, connection); err != nil {
		return err
	}
	_ = context.Replace((*Dataset)(nil), dataset)
	_ = context.Replace((*dsc.TableDescriptor)(nil), table)
	var records []interface{}
	expandDataIfNeeded(context, dataset.Records)
	if records, err = dataset.Records.Expand(context, false); err != nil {
		return err
	}
	var dmlBuilder = newDatasetDmlProvider(dsc.NewDmlBuilder(table))
	if len(table.PkColumns) == 0 {  
		modification.Method = "load"
		modification.Added, err = manager.PersistData(connection, records, table.Table, nil, insertSQLProvider(dmlBuilder))  
		return err
	}
	modification.Added, modification.Modified, err = manager.PersistAllOnConnection(connection, &records, table.Table, dmlBuilder)
	return err
}
func (s *service) prepare(request *PrepareRequest, response *PrepareResponse, manager dsc.Manager, connection dsc.Connection) {
	err := connection.Begin()
	if err != nil {
		response.SetError(err)
	}
	context := s.newContext(manager)
	for _, dataset := range request.Datasets {
		err = s.populate(request.Datastore, dataset, response, context, manager, connection)
		if err != nil {
			break
		}
	}
	if err == nil {
		err = connection.Commit()
	} else {
		_ = connection.Rollback()
	}
	if err != nil {
		response.SetError(err)
	}
}
func (s *service) Prepare(request *PrepareRequest) *PrepareResponse {
	var response = &PrepareResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	err := s.prepareWithRequest(request, response)
	if err != nil {
		response.SetError(err)
		return response
	}
	return response
}
func (s *service) prepareWithRequest(request *PrepareRequest, response *PrepareResponse) error {
	err := request.Init()
	if err == nil {
		err = request.Validate()
	}
	if err != nil {
		return err
	}
	if !validateDatastores(s.registry, response.BaseResponse, request.Datastore) {
		return nil
	}
	var connection dsc.Connection
	manager := s.registry.Get(request.Datastore)
	if err = request.Load(); err == nil {
		if len(request.Datasets) == 0 {
			return fmt.Errorf("no dataset: %v/%v", request.URL, request.Prefix+"*"+request.Postfix)
		}
		connection, err = manager.ConnectionProvider().Get()
	}
	if err != nil {
		return err
	}
	defer connection.Close()
	adminConnection, err := s.disableForeignKeyCheck(request.Datastore, connection, false)
	if err != nil {
		return err
	}
	defer s.enableForeignKeyCheck(request.Datastore, adminConnection)
	s.prepare(request, response, manager, connection)
	return nil
}
func (s *service) enableForeignKeyCheck(datastore string, connection dsc.Connection) error {
	dialect := GetDatastoreDialect(datastore, s.registry)
	manager := s.registry.Get(datastore)
	adminManager := manager
	var err error
	if !dialect.IsKeyCheckSwitchSessionLevel() {
		adminManager, err = s.getAdminManager(datastore)
		if err != nil {
			return err
		}
	}
	if manager != adminManager {
		defer connection.Close()
	}
	err = dialect.EnableForeignKeyCheck(adminManager, connection)
	return err
}
func (s *service) disableForeignKeyCheck(datastore string, connection dsc.Connection, closeIfAdmin bool) (dsc.Connection, error) {
	dialect := GetDatastoreDialect(datastore, s.registry)
	manager := s.registry.Get(datastore)
	var err error
	adminManager := manager
	if !dialect.IsKeyCheckSwitchSessionLevel() {
		adminManager, err = s.getAdminManager(datastore)
		if err != nil {
			return nil, err
		}
	}
	isAdmin := manager != adminManager
	if isAdmin {
		connection, err = adminManager.ConnectionProvider().Get()
	}
	err = dialect.DisableForeignKeyCheck(adminManager, connection)
	if isAdmin && closeIfAdmin {
		if err == nil {
			err = connection.Close()
		}
	}
	return connection, err
}
func (s *service) expect(policy int, dataset *Dataset, response *ExpectResponse, context toolbox.Context, manager dsc.Manager) (err error) {
	if s.mapper.Has(dataset.Table) {
		datasets := s.mapper.Map(dataset)
		for _, dataset := range datasets {
			if err = s.expect(policy, dataset, response, context, manager); err != nil {
				return err
			}
		}
		return err
	}
	var table *dsc.TableDescriptor
	if table, err = s.getTableDescriptor(dataset, manager, context); err != nil {
		return err
	}
	_ = context.Replace((*Dataset)(nil), dataset)
	_ = context.Replace((*dsc.TableDescriptor)(nil), table)
	expandDataIfNeeded(context, dataset.Records)
	expectedRecords, err := dataset.Records.Expand(context, true)
	if err != nil {
		return err
	}
	expected := dataset.Records
	var columns = dataset.Records.Columns()
	dialect := dsc.GetDatastoreDialect(manager.Config().DriverName)
	datastore, _ := dialect.GetCurrentDatastore(manager)
	var sqlColumns []dsc.Column
	if table.FromQuery == "" {
		sqlColumns, _ = dialect.GetColumns(manager, datastore, table.Table)
	}
	var mapper = newDatasetRowMapper(columns, sqlColumns)
	var parametrizedSQL *dsc.ParametrizedSQL
	sqlBuilder := dsc.NewQueryBuilder(table, "")
	var actual = make([]interface{}, 0)
	var validation = &DatasetValidation{
		Dataset: dataset.Table,
	}
	if policy == FullTableDatasetCheckPolicy || len(table.PkColumns) == 0 {  
		parametrizedSQL = sqlBuilder.BuildQueryAll(columns)
		if err = manager.ReadAll(&actual, parametrizedSQL.SQL, parametrizedSQL.Values, mapper); err != nil {
			return err
		}
	} else {
		pkValues := buildBatchedPkValues(expected, table.PkColumns)
		for _, parametrizedSQL = range sqlBuilder.BuildBatchedQueryOnPk(columns, pkValues, batchSize) {
			var batched = make([]interface{}, 0)
			err := manager.ReadAll(&batched, parametrizedSQL.SQL, parametrizedSQL.Values, mapper)
			if err != nil {
				return err
			}
			actual = append(actual, batched...)
		}
	}
	validation.Expected = expectedRecords
	validation.Actual = actual
	validation.Validation, err = assertly.Assert(expectedRecords, actual, assertly.NewDataPath(table.Table))
	if err == nil {
		if policy == FullTableDatasetCheckPolicy {
			expectedRecords = removeDirectiveRecord(expectedRecords)
			if len(actual) != len(expectedRecords) {
				validation.Validation.AddFailure(assertly.NewFailure("", "count", assertly.EqualViolation, len(expectedRecords), len(actual)))
			}
		}
		response.Validation = append(response.Validation, validation)
		response.FailedCount += validation.Validation.FailedCount
		response.PassedCount += validation.Validation.PassedCount
		response.Message += "\n" + dataset.Table + "\n" + validation.Report()
		if validation.HasFailure() {
			response.Status = "failed"
		} else {
			response.Status = "ok"
		}
	}
	return err
}
func (s *service) Expect(request *ExpectRequest) *ExpectResponse {
	var response = &ExpectResponse{
		BaseResponse: NewBaseOkResponse(),
	}
	err := request.Init()
	if err == nil {
		err = request.Validate()
	}
	if err != nil {
		response.SetError(err)
		return response
	}
	if !validateDatastores(s.registry, response.BaseResponse, request.Datastore) {
		return response
	}
	manager := s.registry.Get(request.Datastore)
	context := s.newContext(manager)
	if err = request.Load(); err == nil {
		if len(request.Datasets) == 0 {
			response.SetError(fmt.Errorf("no dataset: %v/%v", request.URL, request.Prefix+"*"+request.Postfix))
			return response
		}
		for _, dataset := range request.Datasets {
			if err = s.expect(request.CheckPolicy, dataset, response, context, manager); err != nil {
				break
			}
		}
	}
	response.SetError(err)
	return response
}