func (qr *driverRows) Close() error {
	if qr.err == sql.ErrNoRows || qr.err == io.EOF {
		return nil
	}
	qr.err = io.EOF
	hs := make(http.Header)
	if qr.stmt.user != "" {
		hs.Add(trinoUserHeader, qr.stmt.user)
	}
	req, err := qr.stmt.conn.newRequest("DELETE", qr.stmt.conn.baseURL+"/v1/query/"+url.PathEscape(qr.queryID), nil, hs)
	if err != nil {
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), DefaultCancelQueryTimeout)
	defer cancel()
	resp, err := qr.stmt.conn.roundTrip(ctx, req)
	if err != nil {
		qferr, ok := err.(*ErrQueryFailed)
		if ok && qferr.StatusCode == http.StatusNoContent {
			qr.nextURI = ""
			return nil
		}
		return err
	}
	resp.Body.Close()
	return qr.err
}
Close closes the rows iterator.
package trino
import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"database/sql"
	"database/sql/driver"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math"
	"net/http"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"
	"unicode"
	"gopkg.in/jcmturner/gokrb5.v6/client"
	"gopkg.in/jcmturner/gokrb5.v6/config"
	"gopkg.in/jcmturner/gokrb5.v6/keytab"
)
func init() {
	sql.Register("trino", &sqldriver{})
}
var (
	DefaultQueryTimeout = 60 * time.Second
	DefaultCancelQueryTimeout = 30 * time.Second
	ErrOperationNotSupported = errors.New("trino: operation not supported")
	ErrQueryCancelled = errors.New("trino: query cancelled")
	ErrUnsupportedHeader = errors.New("trino: server response contains an unsupported header")
)
const (
	preparedStatementHeader    = "X-Trino-Prepared-Statement"
	preparedStatementName      = "_trino_go"
	trinoUserHeader            = "X-Trino-User"
	trinoSourceHeader          = "X-Trino-Source"
	trinoCatalogHeader         = "X-Trino-Catalog"
	trinoSchemaHeader          = "X-Trino-Schema"
	trinoSessionHeader         = "X-Trino-Session"
	trinoSetCatalogHeader      = "X-Trino-Set-Catalog"
	trinoSetSchemaHeader       = "X-Trino-Set-Schema"
	trinoSetPathHeader         = "X-Trino-Set-Path"
	trinoSetSessionHeader      = "X-Trino-Set-Session"
	trinoClearSessionHeader    = "X-Trino-Clear-Session"
	trinoSetRoleHeader         = "X-Trino-Set-Role"
	trinoExtraCredentialHeader = "X-Trino-Extra-Credential"
	KerberosEnabledConfig    = "KerberosEnabled"
	kerberosKeytabPathConfig = "KerberosKeytabPath"
	kerberosPrincipalConfig  = "KerberosPrincipal"
	kerberosRealmConfig      = "KerberosRealm"
	kerberosConfigPathConfig = "KerberosConfigPath"
	SSLCertPathConfig        = "SSLCertPath"
)
var (
	responseToRequestHeaderMap = map[string]string{
		trinoSetSchemaHeader:  trinoSchemaHeader,
		trinoSetCatalogHeader: trinoCatalogHeader,
	}
	unsupportedResponseHeaders = []string{
		trinoSetPathHeader,
		trinoSetSessionHeader,
		trinoClearSessionHeader,
		trinoSetRoleHeader,
	}
)
type sqldriver struct{}
func (d *sqldriver) Open(name string) (driver.Conn, error) {
	return newConn(name)
}
var _ driver.Driver = &sqldriver{}
type Config struct {
	ServerURI          string             
	Source             string             
	Catalog            string             
	Schema             string             
	SessionProperties  map[string]string  
	ExtraCredentials   map[string]string  
	CustomClientName   string             
	KerberosEnabled    string             
	KerberosKeytabPath string             
	KerberosPrincipal  string             
	KerberosRealm      string             
	KerberosConfigPath string             
	SSLCertPath        string             
}
func (c *Config) FormatDSN() (string, error) {
	serverURL, err := url.Parse(c.ServerURI)
	if err != nil {
		return "", err
	}
	var sessionkv []string
	if c.SessionProperties != nil {
		for k, v := range c.SessionProperties {
			sessionkv = append(sessionkv, k+"="+v)
		}
	}
	var credkv []string
	if c.ExtraCredentials != nil {
		for k, v := range c.ExtraCredentials {
			credkv = append(credkv, k+"="+v)
		}
	}
	source := c.Source
	if source == "" {
		source = "trino-go-client"
	}
	query := make(url.Values)
	query.Add("source", source)
	KerberosEnabled, _ := strconv.ParseBool(c.KerberosEnabled)
	isSSL := serverURL.Scheme == "https"
	if isSSL && c.SSLCertPath != "" {
		query.Add(SSLCertPathConfig, c.SSLCertPath)
	}
	if KerberosEnabled {
		query.Add(KerberosEnabledConfig, "true")
		query.Add(kerberosKeytabPathConfig, c.KerberosKeytabPath)
		query.Add(kerberosPrincipalConfig, c.KerberosPrincipal)
		query.Add(kerberosRealmConfig, c.KerberosRealm)
		query.Add(kerberosConfigPathConfig, c.KerberosConfigPath)
		if !isSSL {
			return "", fmt.Errorf("trino: client configuration error, SSL must be enabled for secure env")
		}
	}
	for k, v := range map[string]string{
		"catalog":            c.Catalog,
		"schema":             c.Schema,
		"session_properties": strings.Join(sessionkv, ","),
		"extra_credentials":  strings.Join(credkv, ","),
		"custom_client":      c.CustomClientName,
	} {
		if v != "" {
			query[k] = []string{v}
		}
	}
	serverURL.RawQuery = query.Encode()
	return serverURL.String(), nil
}
type Conn struct {
	baseURL         string
	auth            *url.Userinfo
	httpClient      http.Client
	httpHeaders     http.Header
	kerberosClient  client.Client
	kerberosEnabled bool
}
var (
	_ driver.Conn               = &Conn{}
	_ driver.ConnPrepareContext = &Conn{}
)
func newConn(dsn string) (*Conn, error) {
	serverURL, err := url.Parse(dsn)
	if err != nil {
		return nil, fmt.Errorf("trino: malformed dsn: %v", err)
	}
	query := serverURL.Query()
	kerberosEnabled, _ := strconv.ParseBool(query.Get(KerberosEnabledConfig))
	var kerberosClient client.Client
	if kerberosEnabled {
		kt, err := keytab.Load(query.Get(kerberosKeytabPathConfig))
		if err != nil {
			return nil, fmt.Errorf("trino: Error loading Keytab: %v", err)
		}
		kerberosClient = client.NewClientWithKeytab(query.Get(kerberosPrincipalConfig), query.Get(kerberosRealmConfig), kt)
		conf, err := config.Load(query.Get(kerberosConfigPathConfig))
		if err != nil {
			return nil, fmt.Errorf("trino: Error loading krb config: %v", err)
		}
		kerberosClient.WithConfig(conf)
		loginErr := kerberosClient.Login()
		if loginErr != nil {
			return nil, fmt.Errorf("trino: Error login to KDC: %v", loginErr)
		}
	}
	var httpClient = http.DefaultClient
	if clientKey := query.Get("custom_client"); clientKey != "" {
		httpClient = getCustomClient(clientKey)
		if httpClient == nil {
			return nil, fmt.Errorf("trino: custom client not registered: %q", clientKey)
		}
	} else if certPath := query.Get(SSLCertPathConfig); certPath != "" && serverURL.Scheme == "https" {
		cert, err := ioutil.ReadFile(certPath)
		if err != nil {
			return nil, fmt.Errorf("trino: Error loading SSL Cert File: %v", err)
		}
		certPool := x509.NewCertPool()
		certPool.AppendCertsFromPEM(cert)
		httpClient = &http.Client{
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{
					RootCAs: certPool,
				},
			},
		}
	}
	c := &Conn{
		baseURL:         serverURL.Scheme + "://" + serverURL.Host,
		httpClient:      *httpClient,
		httpHeaders:     make(http.Header),
		kerberosClient:  kerberosClient,
		kerberosEnabled: kerberosEnabled,
	}
	var user string
	if serverURL.User != nil {
		user = serverURL.User.Username()
		pass, _ := serverURL.User.Password()
		if pass != "" && serverURL.Scheme == "https" {
			c.auth = serverURL.User
		}
	}
	for k, v := range map[string]string{
		trinoUserHeader:            user,
		trinoSourceHeader:          query.Get("source"),
		trinoCatalogHeader:         query.Get("catalog"),
		trinoSchemaHeader:          query.Get("schema"),
		trinoSessionHeader:         query.Get("session_properties"),
		trinoExtraCredentialHeader: query.Get("extra_credentials"),
	} {
		if v != "" {
			c.httpHeaders.Add(k, v)
		}
	}
	return c, nil
}
var customClientRegistry = struct {
	sync.RWMutex
	Index map[string]http.Client
}{
	Index: make(map[string]http.Client),
}
func RegisterCustomClient(key string, client *http.Client) error {
	if _, err := strconv.ParseBool(key); err == nil {
		return fmt.Errorf("trino: custom client key %q is reserved", key)
	}
	customClientRegistry.Lock()
	customClientRegistry.Index[key] = *client
	customClientRegistry.Unlock()
	return nil
}
func DeregisterCustomClient(key string) {
	customClientRegistry.Lock()
	delete(customClientRegistry.Index, key)
	customClientRegistry.Unlock()
}
func getCustomClient(key string) *http.Client {
	customClientRegistry.RLock()
	defer customClientRegistry.RUnlock()
	if client, ok := customClientRegistry.Index[key]; ok {
		return &client
	}
	return nil
}
func (c *Conn) Begin() (driver.Tx, error) {
	return nil, ErrOperationNotSupported
}
func (c *Conn) Prepare(query string) (driver.Stmt, error) {
	return nil, driver.ErrSkip
}
func (c *Conn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
	return &driverStmt{conn: c, query: query}, nil
}
func (c *Conn) Close() error {
	return nil
}
func (c *Conn) newRequest(method, url string, body io.Reader, hs http.Header) (*http.Request, error) {
	req, err := http.NewRequest(method, url, body)
	if err != nil {
		return nil, fmt.Errorf("trino: %v", err)
	}
	if c.kerberosEnabled {
		err = c.kerberosClient.SetSPNEGOHeader(req, "trino/"+req.URL.Hostname())
		if err != nil {
			return nil, fmt.Errorf("error setting client SPNEGO header: %v", err)
		}
	}
	for k, v := range c.httpHeaders {
		req.Header[k] = v
	}
	for k, v := range hs {
		req.Header[k] = v
	}
	if c.auth != nil {
		pass, _ := c.auth.Password()
		req.SetBasicAuth(c.auth.Username(), pass)
	}
	return req, nil
}
func (c *Conn) roundTrip(ctx context.Context, req *http.Request) (*http.Response, error) {
	delay := 100 * time.Millisecond
	const maxDelayBetweenRequests = float64(15 * time.Second)
	timer := time.NewTimer(0)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-timer.C:
			timeout := DefaultQueryTimeout
			if deadline, ok := ctx.Deadline(); ok {
				timeout = time.Until(deadline)
			}
			client := c.httpClient
			client.Timeout = timeout
			req.Cancel = ctx.Done()
			resp, err := client.Do(req)
			if err != nil {
				return nil, &ErrQueryFailed{Reason: err}
			}
			switch resp.StatusCode {
			case http.StatusOK:
				for src, dst := range responseToRequestHeaderMap {
					if v := resp.Header.Get(src); v != "" {
						c.httpHeaders.Set(dst, v)
					}
				}
				for _, name := range unsupportedResponseHeaders {
					if v := resp.Header.Get(name); v != "" {
						return nil, ErrUnsupportedHeader
					}
				}
				return resp, nil
			case http.StatusServiceUnavailable:
				resp.Body.Close()
				timer.Reset(delay)
				delay = time.Duration(math.Min(
					float64(delay)*math.Phi,
					maxDelayBetweenRequests,
				))
				continue
			default:
				return nil, newErrQueryFailedFromResponse(resp)
			}
		}
	}
}
type ErrQueryFailed struct {
	StatusCode int
	Reason     error
}
func (e *ErrQueryFailed) Error() string {
	return fmt.Sprintf("trino: query failed (%d %s): %q",
		e.StatusCode, http.StatusText(e.StatusCode), e.Reason)
}
func newErrQueryFailedFromResponse(resp *http.Response) *ErrQueryFailed {
	const maxBytes = 8 * 1024
	defer resp.Body.Close()
	qf := &ErrQueryFailed{StatusCode: resp.StatusCode}
	b, err := ioutil.ReadAll(io.LimitReader(resp.Body, maxBytes))
	if err != nil {
		qf.Reason = err
		return qf
	}
	reason := string(b)
	if resp.ContentLength > maxBytes {
		reason += "..."
	}
	qf.Reason = errors.New(reason)
	return qf
}
type driverStmt struct {
	conn  *Conn
	query string
	user  string
}
var (
	_ driver.Stmt             = &driverStmt{}
	_ driver.StmtQueryContext = &driverStmt{}
	_ driver.StmtExecContext  = &driverStmt{}
)
func (st *driverStmt) Close() error {
	return nil
}
func (st *driverStmt) NumInput() int {
	return -1
}
func (st *driverStmt) Exec(args []driver.Value) (driver.Result, error) {
	return nil, driver.ErrSkip
}
func (st *driverStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
	sr, err := st.exec(ctx, args)
	if err != nil {
		return nil, err
	}
	rows := &driverRows{
		ctx:          ctx,
		stmt:         st,
		queryID:      sr.ID,
		nextURI:      sr.NextURI,
		rowsAffected: sr.UpdateCount,
	}
	for err == nil {
		err = rows.fetch(true)
	}
	if err != nil && err != io.EOF {
		return nil, err
	}
	return rows, nil
}
type stmtResponse struct {
	ID          string    `json:"id"`
	InfoURI     string    `json:"infoUri"`
	NextURI     string    `json:"nextUri"`
	Stats       stmtStats `json:"stats"`
	Error       stmtError `json:"error"`
	UpdateType  string    `json:"updateType"`
	UpdateCount int64     `json:"updateCount"`
}
type stmtStats struct {
	State           string    `json:"state"`
	Scheduled       bool      `json:"scheduled"`
	Nodes           int       `json:"nodes"`
	TotalSplits     int       `json:"totalSplits"`
	QueuesSplits    int       `json:"queuedSplits"`
	RunningSplits   int       `json:"runningSplits"`
	CompletedSplits int       `json:"completedSplits"`
	UserTimeMillis  int       `json:"userTimeMillis"`
	CPUTimeMillis   int       `json:"cpuTimeMillis"`
	WallTimeMillis  int       `json:"wallTimeMillis"`
	ProcessedRows   int       `json:"processedRows"`
	ProcessedBytes  int       `json:"processedBytes"`
	RootStage       stmtStage `json:"rootStage"`
}
type stmtError struct {
	Message       string               `json:"message"`
	ErrorName     string               `json:"errorName"`
	ErrorCode     int                  `json:"errorCode"`
	ErrorLocation stmtErrorLocation    `json:"errorLocation"`
	FailureInfo   stmtErrorFailureInfo `json:"failureInfo"`
}
type stmtErrorLocation struct {
	LineNumber   int `json:"lineNumber"`
	ColumnNumber int `json:"columnNumber"`
}
type stmtErrorFailureInfo struct {
	Type string `json:"type"`
}
func (e stmtError) Error() string {
	return e.FailureInfo.Type + ": " + e.Message
}
type stmtStage struct {
	StageID         string      `json:"stageId"`
	State           string      `json:"state"`
	Done            bool        `json:"done"`
	Nodes           int         `json:"nodes"`
	TotalSplits     int         `json:"totalSplits"`
	QueuedSplits    int         `json:"queuedSplits"`
	RunningSplits   int         `json:"runningSplits"`
	CompletedSplits int         `json:"completedSplits"`
	UserTimeMillis  int         `json:"userTimeMillis"`
	CPUTimeMillis   int         `json:"cpuTimeMillis"`
	WallTimeMillis  int         `json:"wallTimeMillis"`
	ProcessedRows   int         `json:"processedRows"`
	ProcessedBytes  int         `json:"processedBytes"`
	SubStages       []stmtStage `json:"subStages"`
}
func (st *driverStmt) Query(args []driver.Value) (driver.Rows, error) {
	return nil, driver.ErrSkip
}
func (st *driverStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
	sr, err := st.exec(ctx, args)
	if err != nil {
		return nil, err
	}
	rows := &driverRows{
		ctx:     ctx,
		stmt:    st,
		queryID: sr.ID,
		nextURI: sr.NextURI,
	}
	if err = rows.fetch(false); err != nil {
		return nil, err
	}
	return rows, nil
}
func (st *driverStmt) exec(ctx context.Context, args []driver.NamedValue) (*stmtResponse, error) {
	query := st.query
	var hs http.Header
	if len(args) > 0 {
		hs = make(http.Header)
		var ss []string
		for _, arg := range args {
			s, err := Serial(arg.Value)
			if err != nil {
				return nil, err
			}
			if arg.Name == trinoUserHeader {
				st.user = arg.Value.(string)
				hs.Add(trinoUserHeader, st.user)
			} else {
				if hs.Get(preparedStatementHeader) == "" {
					hs.Add(preparedStatementHeader, preparedStatementName+"="+url.QueryEscape(st.query))
				}
				ss = append(ss, s)
			}
		}
		if len(ss) > 0 {
			query = "EXECUTE " + preparedStatementName + " USING " + strings.Join(ss, ", ")
		}
	}
	req, err := st.conn.newRequest("POST", st.conn.baseURL+"/v1/statement", strings.NewReader(query), hs)
	if err != nil {
		return nil, err
	}
	resp, err := st.conn.roundTrip(ctx, req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	var sr stmtResponse
	d := json.NewDecoder(resp.Body)
	d.UseNumber()
	err = d.Decode(&sr)
	if err != nil {
		return nil, fmt.Errorf("trino: %v", err)
	}
	return &sr, handleResponseError(resp.StatusCode, sr.Error)
}
type driverRows struct {
	ctx     context.Context
	stmt    *driverStmt
	queryID string
	nextURI string
	err          error
	rowindex     int
	columns      []string
	coltype      []*typeConverter
	data         []queryData
	rowsAffected int64
}
var _ driver.Rows = &driverRows{}
var _ driver.Result = &driverRows{}