func (s *sqlDB) createTableSchema(cols []string, colTypes []*sql.ColumnType) insertSchema {
	var (
		colNameHolder = make([]string, len(cols))
		colValHolder  = make([]string, len(cols))
	)
	for i := range cols {
		colNameHolder[i] = fmt.Sprintf(`"%s"`, cols[i])
		if s.opt.DBType == dbTypePostgres {
			colValHolder[i] = fmt.Sprintf("$%d", i+1)
		} else {
			colValHolder[i] = "?"
		}
	}
	var (
		fields   = make([]string, len(cols))
		typ      = ""
		unlogged = ""
	)
	for i := 0; i < len(cols); i++ {
		typ = colTypes[i].DatabaseTypeName()
		switch colTypes[i].DatabaseTypeName() {
		case "INT2", "INT4", "INT8",  
			"TINYINT", "SMALLINT", "INT", "MEDIUMINT", "BIGINT":  
			typ = "BIGINT"
		case "FLOAT4", "FLOAT8",  
			"DECIMAL", "FLOAT", "DOUBLE", "NUMERIC":  
			typ = "DECIMAL"
		case "TIMESTAMP",  
			"DATETIME":  
			typ = "TIMESTAMP"
		case "DATE":  
			typ = "DATE"
		case "BOOLEAN":  
			typ = "BOOLEAN"
		case "JSON", "JSONB":  
			if s.opt.DBType != dbTypePostgres {
				typ = "TEXT"
			}
		case "_INT4":  
			typ = "_INT4"
		case "_INT8":  
			typ = "_INT8"
		case "_TEXT":  
			typ = "_TEXT"
		default:
			typ = "TEXT"
		}
		if nullable, ok := colTypes[i].Nullable(); ok && !nullable {
			typ += " NOT NULL"
		}
		fields[i] = fmt.Sprintf(`"%s" %s`, cols[i], typ)
	}
	if s.opt.DBType == dbTypePostgres && s.opt.UnloggedTables {
		unlogged = "UNLOGGED"
	}
	return insertSchema{
		dropTable:   `DROP TABLE IF EXISTS "%s";`,
		createTable: fmt.Sprintf(`CREATE %s TABLE IF NOT EXISTS "%%s" (%s);`, unlogged, strings.Join(fields, ",")),
		insertRow: fmt.Sprintf(`INSERT INTO "%%s" (%s) VALUES (%s)`, strings.Join(colNameHolder, ","),
			strings.Join(colValHolder, ",")),
	}
}
createTableSchema takes an SQL query results, gets its column names and types,
and generates a sqlDB CREATE TABLE() schema for the results.
package backends
import (
	"database/sql"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"
)
const (
	dbTypePostgres = "postgres"
	dbTypeMysql    = "mysql"
)
type Opt struct {
	DBType         string
	ResultsTable   string
	UnloggedTables bool
}
type sqlDB struct {
	db     *sql.DB
	opt    Opt
	logger *log.Logger
	resTableSchemas map[string]insertSchema
	schemaMutex     sync.RWMutex
}
type sqlDBWriter struct {
	jobID       string
	taskName    string
	colsWritten bool
	cols        []string
	rows        [][]byte
	tx          *sql.Tx
	tbl         string
	backend *sqlDB
}
type insertSchema struct {
	dropTable   string
	createTable string
	insertRow   string
}
func NewSQLBackend(db *sql.DB, opt Opt, l *log.Logger) (ResultBackend, error) {
	s := sqlDB{
		db:              db,
		opt:             opt,
		resTableSchemas: make(map[string]insertSchema),
		schemaMutex:     sync.RWMutex{},
		logger:          l,
	}
	if opt.ResultsTable != "" {
		s.opt.ResultsTable = opt.ResultsTable
	} else {
		s.opt.ResultsTable = "results_%s"
	}
	return &s, nil
}
func (s *sqlDB) NewResultSet(jobID, taskName string, ttl time.Duration) (ResultSet, error) {
	tx, err := s.db.Begin()
	if err != nil {
		return nil, err
	}
	return &sqlDBWriter{
		jobID:    jobID,
		taskName: taskName,
		backend:  s,
		tbl:      fmt.Sprintf(s.opt.ResultsTable, jobID),
		tx:       tx,
	}, nil
}
func (w *sqlDBWriter) RegisterColTypes(cols []string, colTypes []*sql.ColumnType) error {
	if w.IsColTypesRegistered() {
		return errors.New("column types are already registered")
	}
	w.cols = make([]string, len(cols))
	copy(w.cols, cols)
	var (
		colNameHolder = make([]string, len(cols))
		colValHolder  = make([]string, len(cols))
	)
	for i := range w.cols {
		colNameHolder[i] = fmt.Sprintf(`"%s"`, w.cols[i])
		if w.backend.opt.DBType == dbTypePostgres {
			colValHolder[i] = fmt.Sprintf("$%d", i+1)
		} else {
			colValHolder[i] = "?"
		}
	}
	ins := fmt.Sprintf(`INSERT INTO "%%s" (%s) `, strings.Join(colNameHolder, ","))
	ins += fmt.Sprintf("VALUES (%s)", strings.Join(colValHolder, ","))
	w.backend.schemaMutex.Lock()
	w.backend.resTableSchemas[w.taskName] = w.backend.createTableSchema(cols, colTypes)
	w.backend.schemaMutex.Unlock()
	return nil
}
func (w *sqlDBWriter) IsColTypesRegistered() bool {
	w.backend.schemaMutex.RLock()
	_, ok := w.backend.resTableSchemas[w.taskName]
	w.backend.schemaMutex.RUnlock()
	return ok
}
func (w *sqlDBWriter) WriteCols(cols []string) error {
	if w.colsWritten {
		return fmt.Errorf("columns for '%s' are already written", w.taskName)
	}
	w.backend.schemaMutex.RLock()
	rSchema, ok := w.backend.resTableSchemas[w.taskName]
	w.backend.schemaMutex.RUnlock()
	if !ok {
		return fmt.Errorf("column types for '%s' have not been registered", w.taskName)
	}
	tx, err := w.backend.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, err := tx.Exec(fmt.Sprintf(rSchema.dropTable, w.tbl)); err != nil {
		return err
	}
	if _, err := tx.Exec(fmt.Sprintf(rSchema.createTable, w.tbl)); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	return err
}
func (w *sqlDBWriter) WriteRow(row []interface{}) error {
	w.backend.schemaMutex.RLock()
	rSchema, ok := w.backend.resTableSchemas[w.taskName]
	w.backend.schemaMutex.RUnlock()
	if !ok {
		return fmt.Errorf("column types for '%s' have not been registered", w.taskName)
	}
	_, err := w.tx.Exec(fmt.Sprintf(rSchema.insertRow, w.tbl), row...)
	return err
}
func (w *sqlDBWriter) Flush() error {
	err := w.tx.Commit()
	if err != nil {
		return err
	}
	return nil
}
func (w *sqlDBWriter) Close() error {
	if w.tx != nil {
		return w.tx.Rollback()
	}
	return nil
}