Skip to content

Commit

Permalink
tests workaround
Browse files Browse the repository at this point in the history
  • Loading branch information
absorbb committed Jul 17, 2024
1 parent 262f143 commit ecdd20c
Show file tree
Hide file tree
Showing 2 changed files with 77 additions and 17 deletions.
16 changes: 12 additions & 4 deletions bulkerlib/implementations/sql/clickhouse.go
Original file line number Diff line number Diff line change
Expand Up @@ -301,14 +301,19 @@ func (ch *ClickHouse) OpenTx(ctx context.Context) (*TxSQLAdapter, error) {
origConfig := *ch.config
configCopy := origConfig
configCopy.Parameters = utils.MapCopy(ch.config.Parameters)
configCopy.Parameters["session_id"] = uuid.New()
sessionId := uuid.New()
configCopy.Parameters["session_id"] = sessionId
utils.MapPutIfAbsent(configCopy.Parameters, "session_timeout", "3600")
// create db pool just for one session because 'session_id' config parameter defines session
sessionDb, err := ch.dbConnectFunction(&configCopy)
if err != nil {
return nil, fmt.Errorf("failed to open session: %v", err)
}
db = sessionDb
c, err := sessionDb.Conn(ctx)
if err != nil {
return nil, fmt.Errorf("failed to open connection: %v", err)
}
db = NewConWithDB(sessionDb, c, sessionId)
} else {
var err error
db, err = ch.dataSource.Conn(ctx)
Expand Down Expand Up @@ -1136,8 +1141,11 @@ func (ch *ClickHouse) Ping(_ context.Context) error {
func chPing(db *sql.DB, httpMode bool) error {
if httpMode {
//keep select 1 and don't use Ping() because chproxy doesn't support /ping endpoint.
_, err := db.Exec("SELECT 1")
return err
//_, err := db.Exec("SELECT 1")
//return err

//not sure Ping is necessary in httpMode
return nil
} else {
return db.Ping()
}
Expand Down
78 changes: 65 additions & 13 deletions bulkerlib/implementations/sql/tx_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"github.com/jitsucom/bulker/jitsubase/errorj"
"github.com/jitsucom/bulker/jitsubase/logging"
"io"
"sync"
)

// TxWrapper is sql transaction wrapper. Used for handling and log errors with db type (postgres, mySQL, redshift or snowflake)
Expand Down Expand Up @@ -179,32 +180,83 @@ func (t *TxWrapper) Rollback() error {
}

type ConWithDB struct {
db *sql.DB
//con *sql.Conn
sessionId string
db *sql.DB
con *sql.Conn
}

func NewConWithDB(db *sql.DB) *ConWithDB {
return &ConWithDB{db: db}
var activeSession sync.Map

func NewConWithDB(db *sql.DB, con *sql.Conn, sessionId string) *ConWithDB {
return &ConWithDB{db: db, con: con, sessionId: sessionId}
}

func (c *ConWithDB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
return c.db.ExecContext(ctx, query, args...)
func (c *ConWithDB) ExecContext(ctx context.Context, query string, args ...any) (res sql.Result, err error) {
q, ok := activeSession.LoadOrStore(c.sessionId, query)
if !ok {
defer func() {
del := activeSession.CompareAndDelete(c.sessionId, query)
if !del {
logging.Errorf("session %s was REUSED CONCURRENTLY", c.sessionId)
}
}()
res, err = c.con.ExecContext(ctx, query, args...)
} else {
err = fmt.Errorf("session %s is CONCURRENTLY USED by: %s", c.sessionId, q.(string))
}
return
}

func (c *ConWithDB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {
return c.db.QueryContext(ctx, query, args...)
func (c *ConWithDB) QueryContext(ctx context.Context, query string, args ...any) (res *sql.Rows, err error) {
q, ok := activeSession.LoadOrStore(c.sessionId, query)
if !ok {
defer func() {
del := activeSession.CompareAndDelete(c.sessionId, query)
if !del {
logging.Fatalf("session %s was REUSED CONCURRENTLY", c.sessionId)
}
}()
res, err = c.con.QueryContext(ctx, query, args...)
} else {
err = fmt.Errorf("session %s is CONCURRENTLY USED by: %s", c.sessionId, q.(string))
}
return
}

func (c *ConWithDB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {
return c.db.QueryRowContext(ctx, query, args...)
func (c *ConWithDB) QueryRowContext(ctx context.Context, query string, args ...any) (row *sql.Row) {
q, ok := activeSession.LoadOrStore(c.sessionId, query)
if !ok {
defer func() {
del := activeSession.CompareAndDelete(c.sessionId, query)
if !del {
logging.Fatalf("session %s was REUSED CONCURRENTLY", c.sessionId)
}
}()
row = c.con.QueryRowContext(ctx, query, args...)
} else {
logging.Fatalf("session %s is CONCURRENTLY USED by: %s", c.sessionId, q.(string))
}
return
}

func (c *ConWithDB) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
return c.db.PrepareContext(ctx, query)
func (c *ConWithDB) PrepareContext(ctx context.Context, query string) (res *sql.Stmt, err error) {
q, ok := activeSession.LoadOrStore(c.sessionId, query)
if !ok {
defer func() {
del := activeSession.CompareAndDelete(c.sessionId, query)
if !del {
logging.Fatalf("session %s was REUSED CONCURRENTLY", c.sessionId)
}
}()
res, err = c.con.PrepareContext(ctx, query)
} else {
err = fmt.Errorf("session %s is CONCURRENTLY USED by: %s", c.sessionId, q.(string))
}
return
}

func (c *ConWithDB) Close() error {
//_ = c.con.Close()
_ = c.con.Close()
if c.db != nil {
return c.db.Close()
}
Expand Down

0 comments on commit ecdd20c

Please sign in to comment.