-
Notifications
You must be signed in to change notification settings - Fork 0
/
conn.go
4480 lines (3953 loc) · 116 KB
/
conn.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package nzgo
import (
"bufio"
"context"
"crypto/md5"
"crypto/sha256"
"database/sql"
"database/sql/driver"
b64 "encoding/base64"
"encoding/binary"
"errors"
"fmt"
"io"
"math"
"net"
"os"
"os/user"
"path"
"path/filepath"
"runtime"
"strconv"
"strings"
"time"
"unicode"
"unsafe"
"github.com/IBM/nzgo/v12/oid"
)
// Common error types
var (
ErrNotSupported = errors.New("pq: Unsupported command")
ErrInFailedTransaction = errors.New("pq: Could not complete operation in a failed transaction")
ErrSSLNotSupported = errors.New("pq: SSL is not enabled on the server")
ErrSSLKeyHasWorldPermissions = errors.New("pq: Private key file has group or world access. Permissions should be u=rw (0600) or less")
ErrCouldNotDetectUsername = errors.New("pq: Could not detect default username. Please provide one explicitly")
errUnexpectedReady = errors.New("unexpected ReadyForQuery")
errNoRowsAffected = errors.New("no RowsAffected available after the empty statement")
errNoLastInsertID = errors.New("no LastInsertId available after the empty statement")
)
/* NPS handshake version negotiation packet structure */
type HsVersion struct {
opcode int
version int
}
type DbosTupleDesc struct {
version int /* CTable.m_version */
nullsAllowed int /* CTable.nullsAllowed */
sizeWord int /* CTable.m_sizeWord */
sizeWordSize int /* CTable.m_sizeWordSize */
numFixedFields int /* CTable.m_numFixedFields */
numVaryingFields int /* CTable.m_numVaryingFields */
fixedFieldsSize int /* CTable.m_fixedFieldsSize */
maxRecordSize int /* CTable.m_maxRecordSize */
numFields int /* CTable.m_numFields */
field_type []int /* field_t.type */
field_size []int /* field_t.size */
field_trueSize []int /* field_t.trueSize */
field_offset []int /* field_t.offset */
field_physField []int /* field_t.physField */
field_logField []int /* field_t.logField */
field_nullAllowed []int /* field_t.nullAllowed */
field_fixedSize []int /* field_t.fixedSize */
field_springField []int /* field_t.springField */
DateStyle int
EuroDates int
DBcharset int
EnableTime24 int
}
type DATE_STRUCT struct {
year int
month int
day int
}
type TIME_STRUCT struct {
hour uint16
minute uint16
second uint16
}
type timeStamp struct {
tm_year int
tm_mon int
tm_mday int
tm_hour int
tm_min int
tm_sec int
}
type Interval struct {
time int64 /* all time units other than months and years */ // NZ - was double
month int /* months and years, after time for alignment */
}
type TimeTzADT struct {
time int64 // all time units other than months and years
zone int // numeric time zone, in seconds
}
type TIMESTAMP_STRUCT struct {
year int
month int
day int
hour int
minute int
second int
fraction int
}
// External table stuff (copied from nde/client/exttable.h)
const (
EXTAB_SOCK_DATA = 1 + iota // block of records
EXTAB_SOCK_ERROR // error message
EXTAB_SOCK_DONE // normal wrap-up
EXTAB_SOCK_FLUSH // Flush the current buffer/data
)
const (
PGRES_EMPTY_QUERY = 0 + iota
PGRES_COMMAND_OK /* a query command that doesn't return */
/* anything was executed properly by the backend */
PGRES_TUPLES_OK /* a query command that returns tuples */
/* was executed properly by the backend */
PGRES_FIELDS_OK /* field information from a query was successful */
PGRES_END_TUPLES /* all is ok till here; all after this is error */
PGRES_NONFATAL_ERROR
PGRES_FATAL_ERROR
PGRES_BAD_RESPONSE /* an unexpected response was recv'd from the backend */
PGRES_INTERNAL_ERROR /* memory allocation error in driver */
)
const (
NzTypeRecAddr = 1 + iota // !NOTE-bmz need to add this to all switch stmts
NzTypeDouble
NzTypeInt
NzTypeFloat
NzTypeMoney
NzTypeDate
NzTypeNumeric
NzTypeTime
NzTypeTimestamp
NzTypeInterval
NzTypeTimeTz
NzTypeBool
NzTypeInt1
NzTypeBinary
NzTypeChar
NzTypeVarChar
NzDEPR_Text // OBSOLETE 3.0: BLAST Era Large 'text' Object, (Postgres 'text' datatype overload, too)
NzTypeUnknown // corresponds to PG UNKNOWNOID data type - an untyped string literal
NzTypeInt2
NzTypeInt8
NzTypeVarFixedChar
NzTypeGeometry
NzTypeVarBinary
NzDEPR_Blob // OBSOLETE 3.0: BLAST Era Large 'binary' Object
NzTypeNChar
NzTypeNVarChar
NzDEPR_NText // OBSOLETE 3.0: BLAST Era Large 'nchar text' Object
_ // skip 28
_ // skip 29
NzTypeJson // 30
NzTypeJsonb
NzTypeJsonpath
NzTypeLastEntry // KEEP THIS ENTRY LAST - used internally to size an array
)
const (
CONN_NOT_CONNECTED = 0 + iota /* Connection has not been established */
CONN_CONNECTED /* Connection is up and has been established */
CONN_EXECUTING /* the connection is currently executing a statement */
CONN_FETCHING /* the connection is currently executing a select */
CONN_CANCELLED /* the connection is currently cancelling a statement */
)
/* const to datatype string mapping to use in logger */
var dataType = map[int]string{
NzTypeChar: "NzTypeChar",
NzTypeVarChar: "NzTypeVarChar",
NzTypeVarFixedChar: "NzTypeVarFixedChar",
NzTypeGeometry: "NzTypeGeometry",
NzTypeVarBinary: "NzTypeVarBinary",
NzTypeNChar: "NzTypeNChar",
NzTypeNVarChar: "NzTypeNVarChar",
NzTypeJson: "NzTypeJson",
NzTypeJsonb: "NzTypeJsonb",
NzTypeJsonpath: "NzTypeJsonpath",
}
const (
CP_VERSION_1 = 1 + iota
CP_VERSION_2
CP_VERSION_3
CP_VERSION_4
CP_VERSION_5
CP_VERSION_6
)
/* Client type */
const (
NPS_CLIENT = 0 + iota
IPS_CLIENT
)
type HSV2Msg struct {
/* all message have a packet length (int) prepended
* the opcode len is included in the size.
*/
opcode int
payload string
}
/* Authentication types */
const (
AUTH_REQ_OK = 0 + iota
AUTH_REQ_KRB4
AUTH_REQ_KRB5
AUTH_REQ_PASSWORD
AUTH_REQ_CRYPT
AUTH_REQ_MD5
AUTH_REQ_SHA256
)
/*
* This is used by the postmaster and clients in their handshake.
* This indicates type of information being exchanged between NPS and driver.
*/
const (
HSV2_INVALID_OPCODE = 0 + iota
HSV2_CLIENT_BEGIN
HSV2_DB
HSV2_USER
HSV2_OPTIONS
HSV2_TTY
HSV2_REMOTE_PID
HSV2_PRIOR_PID
HSV2_CLIENT_TYPE
HSV2_PROTOCOL
HSV2_HOSTCASE
HSV2_SSL_NEGOTIATE
HSV2_SSL_CONNECT
HSV2_APPNAME
HSV2_CLIENT_OS
HSV2_CLIENT_HOST_NAME
HSV2_CLIENT_OS_USER
HSV2_64BIT_VARLENA_ENABLED
)
const (
HSV2_CLIENT_DONE = 1000 + iota
HSV2_SERVER_BEGIN
HSV2_PWD
HSV2_SERVER_DONE = 2000
)
const (
PG_PROTOCOL_3 = 3 + iota
PG_PROTOCOL_4
PG_PROTOCOL_5
)
//Client Type
const (
NPSCLIENT_TYPE_GOLANG = 12
)
// Driver is the Postgres database driver.
type Driver struct{}
// Open opens a new connection to the database. name is a connection string.
// Most users should only use it through database/sql package from the standard
// library.
func (d *Driver) Open(name string) (c driver.Conn, err error) {
c, err = Open(name)
return c, err
}
func init() {
sql.Register("nzgo", &Driver{})
}
type parameterStatus struct {
// server version in the same format as server_version_num, or 0 if
// unavailable
serverVersion int
// the current location based on the TimeZone value of the session, if
// available
currentLocation *time.Location
}
type transactionStatus byte
const (
txnStatusIdle transactionStatus = 'I'
txnStatusIdleInTransaction transactionStatus = 'T'
txnStatusInFailedTransaction transactionStatus = 'E'
)
func (s transactionStatus) String() string {
switch s {
case txnStatusIdle:
return "idle"
case txnStatusIdleInTransaction:
return "idle in transaction"
case txnStatusInFailedTransaction:
return "in a failed transaction"
default:
msg := fmt.Sprintf("unknown transactionStatus %d", s)
return msg
}
msg := "not reached"
elog.Debugln(msg)
return msg
}
// Dialer is the dialer interface. It can be used to obtain more control over
// how pq creates network connections.
type Dialer interface {
Dial(network, address string) (net.Conn, error)
DialTimeout(network, address string, timeout time.Duration) (net.Conn, error)
}
type DialerContext interface {
DialContext(ctx context.Context, network, address string) (net.Conn, error)
}
type defaultDialer struct {
d net.Dialer
}
func (d defaultDialer) Dial(network, address string) (net.Conn, error) {
return d.d.Dial(network, address)
}
func (d defaultDialer) DialTimeout(network, address string, timeout time.Duration) (net.Conn, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
return d.DialContext(ctx, network, address)
}
func (d defaultDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
return d.d.DialContext(ctx, network, address)
}
type conn struct {
c net.Conn
buf *bufio.Reader
namei int
scratch [2048]byte
txnStatus transactionStatus
txnFinish func()
// Save connection arguments to use during CancelRequest.
dialer Dialer
opts values
// Cancellation key data for use with CancelRequest messages.
processID int
secretKey int
parameterStatus parameterStatus
saveMessageType byte
saveMessageBuffer []byte
// If true, this connection is bad and all public-facing functions should
// return ErrBadConn.
bad bool
// If set, this connection should never use the binary format when
// receiving query results from prepared statements. Only provided for
// debugging.
disablePreparedBinaryResult bool
// Whether to always send []byte parameters over as binary. Enables single
// round-trip mode for non-prepared Query calls.
binaryParameters bool
// If true this connection is in the middle of a COPY
inCopy bool
//netezza specific
hsVersion int
protocol1 int
protocol2 int
commandNumber int
status int
guardium_clientHostName string
guardium_clientOSUser string
guardium_applName string
guardium_clientOS string
}
// Handle driver-side settings in parsed connection string.
func (cn *conn) handleDriverSettings(o values) (err error) {
boolSetting := func(key string, val *bool) error {
if value, ok := o[key]; ok {
if value == "yes" {
*val = true
} else if value == "no" {
*val = false
} else {
return elog.Fatalf(chopPath(funName()), "unrecognized value %q for %s", value, key)
}
}
return nil
}
err = boolSetting("disable_prepared_binary_result", &cn.disablePreparedBinaryResult)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
return boolSetting("binary_parameters", &cn.binaryParameters)
}
func (cn *conn) handlePgpass(o values) {
// if a password was supplied, do not process .pgpass
if _, ok := o["password"]; ok {
return
}
filename := os.Getenv("PGPASSFILE")
if filename == "" {
// XXX this code doesn't work on Windows where the default filename is
// XXX %APPDATA%\postgresql\pgpass.conf
// Prefer $HOME over user.Current due to glibc bug: golang.org/issue/13470
userHome := os.Getenv("HOME")
if userHome == "" {
user, err := user.Current()
if err != nil {
return
}
userHome = user.HomeDir
}
filename = filepath.Join(userHome, ".pgpass")
}
fileinfo, err := os.Stat(filename)
if err != nil {
return
}
mode := fileinfo.Mode()
if mode&(0x77) != 0 {
// XXX should warn about incorrect .pgpass permissions as psql does
return
}
file, err := os.Open(filename)
if err != nil {
return
}
defer file.Close()
scanner := bufio.NewScanner(io.Reader(file))
hostname := o["host"]
ntw, _ := network(o)
port := o["port"]
db := o["dbname"]
username := o["user"]
// From: https://github.com/tg/pgpass/blob/master/reader.go
getFields := func(s string) []string {
fs := make([]string, 0, 5)
f := make([]rune, 0, len(s))
var esc bool
for _, c := range s {
switch {
case esc:
f = append(f, c)
esc = false
case c == '\\':
esc = true
case c == ':':
fs = append(fs, string(f))
f = f[:0]
default:
f = append(f, c)
}
}
return append(fs, string(f))
}
for scanner.Scan() {
line := scanner.Text()
if len(line) == 0 || line[0] == '#' {
continue
}
split := getFields(line)
if len(split) != 5 {
continue
}
if (split[0] == "*" || split[0] == hostname || (split[0] == "localhost" && (hostname == "" || ntw == "unix"))) && (split[1] == "*" || split[1] == port) && (split[2] == "*" || split[2] == db) && (split[3] == "*" || split[3] == username) {
o["password"] = split[4]
return
}
}
}
func (cn *conn) writeBuf(b byte) *writeBuf {
cn.scratch[0] = b
return &writeBuf{
buf: cn.scratch[:4],
pos: 0,
}
}
// Open opens a new connection to the database. dsn is a connection string.
// Most users should only use it through database/sql package from the standard
// library.
func Open(dsn string) (_ driver.Conn, err error) {
return DialOpen(defaultDialer{}, dsn)
}
// DialOpen opens a new connection to the database using a dialer.
func DialOpen(d Dialer, dsn string) (_ driver.Conn, err error) {
c, err := NewConnector(dsn)
if err != nil {
return nil, err
}
c.dialer = d
return c.open(context.Background())
}
func (c *Connector) open(ctx context.Context) (cn *conn, err error) {
o := c.opts
cn = &conn{
opts: o,
dialer: c.dialer,
}
err = cn.handleDriverSettings(o)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return nil, err
}
cn.handlePgpass(o)
cn.c, err = dial(ctx, c.dialer, o)
if err != nil {
if cn.c != nil {
cn.c.Close()
}
elog.Fatalf(chopPath(funName()), err.Error())
return nil, err
}
cn.buf = bufio.NewReader(cn.c)
err = cn.startup(o)
if err != nil {
return nil, err
}
// reset the deadline, in case one was set (see dial)
if timeout, ok := o["connect_timeout"]; ok && timeout != "0" {
err = cn.c.SetDeadline(time.Time{})
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return cn, err
}
}
return cn, nil
}
func dial(ctx context.Context, d Dialer, o values) (net.Conn, error) {
network, address := network(o)
// SSL is not necessary or supported over UNIX domain sockets
if network == "unix" {
o["sslmode"] = "disable"
}
elog.Debugln("Network ", network)
elog.Debugln("Address ", address)
// Zero or not specified means wait indefinitely.
if timeout, ok := o["connect_timeout"]; ok && timeout != "0" {
seconds, err := strconv.ParseInt(timeout, 10, 0)
if err != nil {
return nil, elog.Fatalf(chopPath(funName()), "invalid value for parameter connect_timeout: %s", err)
}
duration := time.Duration(seconds) * time.Second
// connect_timeout should apply to the entire connection establishment
// procedure, so we both use a timeout for the TCP connection
// establishment and set a deadline for doing the initial handshake.
// The deadline is then reset after startup() is done.
deadline := time.Now().Add(duration)
var conn net.Conn
if dctx, ok := d.(DialerContext); ok {
ctx, cancel := context.WithTimeout(ctx, duration)
defer cancel()
conn, err = dctx.DialContext(ctx, network, address)
} else {
conn, err = d.DialTimeout(network, address, duration)
}
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return nil, err
}
err = conn.SetDeadline(deadline)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return nil, err
}
return conn, nil
}
if dctx, ok := d.(DialerContext); ok {
return dctx.DialContext(ctx, network, address)
}
return d.Dial(network, address)
}
func network(o values) (string, string) {
host := o["host"]
if strings.HasPrefix(host, "/") {
sockPath := path.Join(host, ".s.PGSQL."+o["port"])
return "unix", sockPath
}
return "tcp", net.JoinHostPort(host, o["port"])
}
type values map[string]string
// scanner implements a tokenizer for libpq-style option strings.
type scanner struct {
s []rune
i int
}
// newScanner returns a new scanner initialized with the option string s.
func newScanner(s string) *scanner {
return &scanner{[]rune(s), 0}
}
// Next returns the next rune.
// It returns 0, false if the end of the text has been reached.
func (s *scanner) Next() (rune, bool) {
if s.i >= len(s.s) {
return 0, false
}
r := s.s[s.i]
s.i++
return r, true
}
// SkipSpaces returns the next non-whitespace rune.
// It returns 0, false if the end of the text has been reached.
func (s *scanner) SkipSpaces() (rune, bool) {
r, ok := s.Next()
for unicode.IsSpace(r) && ok {
r, ok = s.Next()
}
return r, ok
}
// parseOpts parses the options from name and adds them to the values.
//
// The parsing code is based on conninfo_parse from libpq's fe-connect.c
func parseOpts(name string, o values) error {
s := newScanner(name)
for {
var (
keyRunes, valRunes []rune
r rune
ok bool
)
if r, ok = s.SkipSpaces(); !ok {
break
}
// Scan the key
for !unicode.IsSpace(r) && r != '=' {
keyRunes = append(keyRunes, r)
if r, ok = s.Next(); !ok {
break
}
}
// Skip any whitespace if we're not at the = yet
if r != '=' {
r, ok = s.SkipSpaces()
}
// The current character should be =
if r != '=' || !ok {
return elog.Fatalf(chopPath(funName()), `missing "=" after %q in connection info string"`, string(keyRunes))
}
// Skip any whitespace after the =
if r, ok = s.SkipSpaces(); !ok {
// If we reach the end here, the last value is just an empty string as per libpq.
o[string(keyRunes)] = ""
break
}
if r != '\'' {
for !unicode.IsSpace(r) {
if r == '\\' {
if r, ok = s.Next(); !ok {
return elog.Fatalf(chopPath(funName()), `missing character after backslash`)
}
}
valRunes = append(valRunes, r)
if r, ok = s.Next(); !ok {
break
}
}
} else {
quote:
for {
if r, ok = s.Next(); !ok {
return elog.Fatalf(chopPath(funName()), `unterminated quoted string literal in connection string`)
}
switch r {
case '\'':
break quote
case '\\':
r, _ = s.Next()
fallthrough
default:
valRunes = append(valRunes, r)
}
}
}
o[string(keyRunes)] = string(valRunes)
}
err := elog.Initialize(o["logLevel"], o["logPath"], o["additionalLogFile"])
if err != nil {
return err
}
return nil
}
func (cn *conn) isInTransaction() bool {
return cn.txnStatus == txnStatusIdleInTransaction ||
cn.txnStatus == txnStatusInFailedTransaction
}
func (cn *conn) checkIsInTransaction(intxn bool) error {
if cn.isInTransaction() != intxn {
cn.bad = true
return elog.Fatalf(chopPath(funName()), "unexpected transaction status %v", cn.txnStatus)
}
return nil
}
func (cn *conn) Begin() (_ driver.Tx, err error) {
return cn.begin("")
}
func (cn *conn) begin(mode string) (_ driver.Tx, err error) {
if cn.bad {
return nil, driver.ErrBadConn
}
defer cn.errRecover(&err)
err = cn.checkIsInTransaction(false)
_, commandTag, err := cn.simpleExec("BEGIN")
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return nil, err
}
cn.txnStatus = txnStatusIdleInTransaction
if commandTag != "BEGIN" {
cn.bad = true
return nil, elog.Fatalf(chopPath(funName()), "unexpected command tag %s", commandTag)
}
if cn.txnStatus != txnStatusIdleInTransaction {
cn.bad = true
return nil, elog.Fatalf(chopPath(funName()), "unexpected transaction status %v", cn.txnStatus)
}
return cn, nil
}
func (cn *conn) closeTxn() {
if finish := cn.txnFinish; finish != nil {
finish()
}
}
func (cn *conn) Commit() (err error) {
defer cn.closeTxn()
if cn.bad {
return driver.ErrBadConn
}
defer cn.errRecover(&err)
err = cn.checkIsInTransaction(true)
// We don't want the client to think that everything is okay if it tries
// to commit a failed transaction. However, no matter what we return,
// database/sql will release this connection back into the free connection
// pool so we have to abort the current transaction here. Note that you
// would get the same behaviour if you issued a COMMIT in a failed
// transaction, so it's also the least surprising thing to do here.
if cn.txnStatus == txnStatusInFailedTransaction {
if err := cn.Rollback(); err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
return ErrInFailedTransaction
}
_, commandTag, err := cn.simpleExec("COMMIT")
if err != nil {
if cn.isInTransaction() {
cn.bad = true
}
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
cn.txnStatus = txnStatusIdle
if commandTag != "COMMIT" {
cn.bad = true
return elog.Fatalf(chopPath(funName()), "unexpected command tag %s", commandTag)
}
err = cn.checkIsInTransaction(false)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
return nil
}
func (cn *conn) Rollback() (err error) {
defer cn.closeTxn()
if cn.bad {
return driver.ErrBadConn
}
defer cn.errRecover(&err)
err = cn.checkIsInTransaction(true)
_, commandTag, err := cn.simpleExec("ROLLBACK")
if err != nil {
if cn.isInTransaction() {
cn.bad = true
}
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
cn.txnStatus = txnStatusIdle
if commandTag != "ROLLBACK" {
return elog.Fatalf(chopPath(funName()), "unexpected command tag %s", commandTag)
}
err = cn.checkIsInTransaction(false)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return err
}
return nil
}
func (cn *conn) gname() string {
cn.namei++
return strconv.FormatInt(int64(cn.namei), 10)
}
func (cn *conn) simpleExec(query string) (res driver.Result, commandTag string, err error) {
var fname string
var filename readBuf
var fh *os.File
if cn.status == CONN_EXECUTING || cn.status == CONN_FETCHING {
cn.status = CONN_CONNECTED
cn.Sock_clear_socket()
} else if cn.status == CONN_CANCELLED {
// Control will reach here only when the query was really huge and
// even after Cancel request sent, it took too long to cancel and
// Conn_clear_sock returned as data was not yet available
cn.Sock_clear_socket()
}
elog.Infoln("Processing query:", query)
var buffer *writeBuf
if cn.commandNumber != -1 {
cn.commandNumber++
buffer = &writeBuf{
buf: []byte{'P', '\x00', '\x00', '\x00', byte(cn.commandNumber)},
pos: 1,
}
if cn.commandNumber > 100000 {
cn.commandNumber = 1
}
}
buffer.string(query)
elog.Debugln(chopPath(funName()), "Buffer sent to nps: ", buffer.buf)
_, err = cn.c.Write(buffer.buf)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return emptyRows, commandTag, err
}
cn.status = CONN_EXECUTING
for {
var response byte
response, err = cn.recvSingleByte()
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return emptyRows, commandTag, err
}
elog.Debugf(chopPath(funName()), "Backend response %c \n", response)
cn.recv_n_bytes(4)
switch response {
case 'C':
length, _ := cn.recv_n_bytes(4)
responseBuf, _ := cn.recv_n_bytes(int(length.int32()))
res, commandTag, err = cn.parseComplete(responseBuf.string())
case 'Z': /* Backend is ready for new query (6.4) */
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return res, commandTag, err
}
return res, commandTag, nil
case 'E':
length, _ := cn.recv_n_bytes(4)
responseBuf, err := cn.recv_n_bytes(int(length.int32()))
errorString := responseBuf.string()
err = errors.New(errorString)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return res, commandTag, err
}
return res, commandTag, nil
case 'I':
res = emptyRows
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return res, commandTag, err
}
return res, commandTag, nil
case 'N':
length, _ := cn.recv_n_bytes(4)
responseBuf, _ := cn.recv_n_bytes(int(length.int32()))
elog.Infoln(funName(), responseBuf.string())
case 'l':
err := cn.xferTable()
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return nil, commandTag, err
}
break
case 'x': /* handle Ext Tbl parser abort */
cn.recv_n_bytes(4)
errorString := fmt.Sprintf("Error operation cancel")
err = errors.New(errorString)
if err != nil {
elog.Fatalf(chopPath(funName()), err.Error())
return res, commandTag, err
}
return res, commandTag, nil
break
case 'e':
length, _ := cn.recv_n_bytes(4)
logDir, _ := cn.recv_n_bytes(int(length.int32()))
char, _ := cn.recvSingleByte()
for char != 0 {
filename = append(filename, char)
char, _ = cn.recvSingleByte()
}
filename = append(filename, '\x00') /* null terminate it */
logType, _ := cn.recv_n_bytes(4)
if !(cn.getFileFromBE(logDir.string(), filename.string(), logType.int32())) {
elog.Debugln(chopPath(funName()), "Error in writing file received from BE")
}
break
case 'u': /* unload - initialize application protocol */
// in ODBC, the first 10 bytes are utilized to populate clientVersion, formatType and bufSize
// these are not needed in go lang, hence ignoring 10 bytes
cn.recv_n_bytes(10)
/* Next 16 bytes are Reserved Bytes for future extension*/
cn.recv_n_bytes(16)
/* Get the filename (specified in dataobject)*/
fileSpecSize, _ := cn.recv_n_bytes(4)
fname, _ := cn.recv_n_bytes(fileSpecSize.int32())
fname = append(fname, '\x00') /* null terminate it */
fh, err = os.OpenFile(fname.string(), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0644)
if err != nil { // file open failed
// Report error to the client
elog.Fatalf(chopPath(funName()), err.Error())
return emptyRows, commandTag, err
} else {
// file open successfully, send status back to datawriter
elog.Debugln(chopPath(funName()), "Successfully opened file: ", fh.Name())
buf := []byte{'\x00', '\x00', '\x00', '\x00'}
cn.c.Write(buf)
}