func (dc *DirectConnection) writeComQuery(sql string) error {
	dc.conn.SetSequence(0)
	data := dc.conn.StartEphemeralPacket(len(sql) + 1)
	data[0] = mysql.ComQuery
	copy(data[1:], sql)
	if err := dc.writeEphemeralPacket(); err != nil {
		return err
	}
	return nil
}
writeComQuery send ComQuery request use EphemeralBuffer
package backend
import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"net"
	"strings"
	"github.com/nooncall/shazam/log"
	"github.com/nooncall/shazam/mysql"
	"github.com/nooncall/shazam/util/sync2"
)
type DirectConnection struct {
	conn *mysql.Conn
	addr     string
	user     string
	password string
	db       string
	capability uint32
	sessionVariables *mysql.SessionVariables
	status uint16
	collation mysql.CollationID
	charset   string
	salt      []byte
	defaultCollation mysql.CollationID
	defaultCharset   string
	pkgErr error
	closed sync2.AtomicBool
}
func NewDirectConnection(addr string, user string, password string, db string, charset string, collationID mysql.CollationID) (*DirectConnection, error) {
	dc := &DirectConnection{
		addr:             addr,
		user:             user,
		password:         password,
		db:               db,
		charset:          charset,
		collation:        collationID,
		defaultCharset:   charset,
		defaultCollation: collationID,
		closed:           sync2.NewAtomicBool(false),
		sessionVariables: mysql.NewSessionVariables(),
	}
	err := dc.connect()
	return dc, err
}
func (dc *DirectConnection) connect() error {
	if dc.conn != nil {
		dc.conn.Close()
	}
	typ := "tcp"
	if strings.Contains(dc.addr, "/") {
		typ = "unix"
	}
	netConn, err := net.Dial(typ, dc.addr)
	if err != nil {
		return err
	}
	tcpConn := netConn.(*net.TCPConn)
	tcpConn.SetNoDelay(true)
	tcpConn.SetKeepAlive(true)
	dc.conn = mysql.NewConn(tcpConn)
	if err := dc.readInitialHandshake(); err != nil {
		dc.conn.Close()
		return err
	}
	if err := dc.writeHandshakeResponse41(); err != nil {
		dc.conn.Close()
		return err
	}
	response, err := dc.readPacket()
	if err != nil {
		dc.conn.Close()
		return err
	}
	switch response[0] {
	case mysql.OKHeader:
	default:
		return errors.New("dc connection handshake failed with mysql")
	}
	if !dc.IsAutoCommit() {
		if _, err := dc.exec("set autocommit = 1"); err != nil {
			dc.conn.Close()
			return err
		}
	}
	return nil
}
func (dc *DirectConnection) Close() {
	if dc.conn != nil {
		dc.conn.Close()
	}
	dc.conn = nil
	dc.salt = nil
	dc.pkgErr = nil
	dc.closed.Set(true)
	return
}
func (dc *DirectConnection) IsClosed() bool {
	return dc.closed.Get()
}
func (dc *DirectConnection) readPacket() ([]byte, error) {
	data, err := dc.conn.ReadPacket()
	dc.pkgErr = err
	return data, err
}
func (dc *DirectConnection) writePacket(data []byte) error {
	err := dc.conn.WritePacket(data)
	if err != nil && strings.Contains(err.Error(), "broken pipe") {
		for i := 0; i < 3; i++ {
			dc.Close()
			e := dc.connect()
			if e == nil {  
				break
			}
		}
	}
	return err
}
func (dc *DirectConnection) writeEphemeralPacket() error {
	err := dc.conn.WriteEphemeralPacket()
	if err != nil && strings.Contains(err.Error(), "broken pipe") {
		for i := 0; i < 3; i++ {
			dc.Close()
			e := dc.connect()
			if e == nil {  
				break
			}
		}
	}
	return err
}
func (dc *DirectConnection) readInitialHandshake() error {
	data, err := dc.readPacket()
	if err != nil {
		return err
	}
	if data[0] == mysql.ErrHeader {
		return errors.New("read initial handshake error")
	}
	if data[0] < mysql.MinProtocolVersion {
		return fmt.Errorf("invalid protocol version %d, must >= 10", data[0])
	}
	pos := 1 + bytes.IndexByte(data[1:], 0x00) + 1 + 4
	dc.salt = append(dc.salt, data[pos:pos+8]...)
	pos += 8 + 1
	dc.capability = uint32(binary.LittleEndian.Uint16(data[pos : pos+2]))
	pos += 2
	if len(data) > pos {
		pos++
		dc.status = binary.LittleEndian.Uint16(data[pos : pos+2])
		pos += 2
		dc.capability = uint32(binary.LittleEndian.Uint16(data[pos:pos+2]))<<16 | dc.capability
		pos += 2
		pos += 10 + 1
		dc.salt = append(dc.salt, data[pos:pos+12]...)
	}
	return nil
}
func (dc *DirectConnection) writeHandshakeResponse41() error {
	capability := mysql.ClientProtocol41 | mysql.ClientSecureConnection |
		mysql.ClientLongPassword | mysql.ClientTransactions | mysql.ClientLongFlag
	capability &= dc.capability
	auth := mysql.CalcPassword(dc.salt, []byte(dc.password))
	length := 4 +  
		4 +  
		1 +  
		23 +  
		mysql.LenNullString(dc.user) +  
		1 +
		len(auth)
	if len(dc.db) > 0 {
		capability |= mysql.ClientConnectWithDB
		length += mysql.LenNullString(dc.db)
	}
	dc.capability = capability
	data := make([]byte, length, length)
	pos := 0
	pos = mysql.WriteUint32(data, pos, capability)
	pos = mysql.WriteZeroes(data, pos, 4)
	pos = mysql.WriteByte(data, pos, byte(dc.collation))
	pos = mysql.WriteZeroes(data, pos, 23)
	pos = mysql.WriteNullString(data, pos, dc.user)
	data[pos] = byte(len(auth))
	pos++
	pos += copy(data[pos:], auth)
	if len(dc.db) > 0 {
		pos = mysql.WriteNullString(data, pos, dc.db)
	}
	if err := dc.writePacket(data); err != nil {
		return err
	}
	return nil
}
func (dc *DirectConnection) writeComInitDB(db string) error {
	dc.conn.SetSequence(0)
	data := make([]byte, len(db)+1, len(db)+1)
	data[0] = mysql.ComInitDB
	copy(data[1:], db)
	if err := dc.writePacket(data); err != nil {
		return err
	}
	return nil
}