Column info changes

client/gosqldriver/rows.go

@@ -19,6 +19,7 @@ package gosqldriver
 
 import (
 	"database/sql/driver"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -35,11 +36,14 @@ type rows struct {
 	hera           *heraConnection
 	vals           []driver.Value
 	cols           int
+	colInfo        []string
 	currentRow     int
 	fetchChunkSize []byte
 	completed      bool
 }
 
+var columnInfo = []string{"Name", "Type", "Width", "Precision", "Scale"}
+
 // TODO: fetch chunk size
 func newRows(hera *heraConnection, cols int, fetchChunkSize []byte) (*rows, error) {
 	rs := &rows{hera: hera, cols: cols, currentRow: 0, fetchChunkSize: fetchChunkSize}
@@ -51,24 +55,65 @@ func newRows(hera *heraConnection, cols int, fetchChunkSize []byte) (*rows, erro
 }
 
 func (r *rows) fetchResults() error {
+	var localVals []driver.Value
+	var err error
+	var ns *netstring.Netstring
+outer:
 	for {
-		ns, err := r.hera.getResponse()
+		ns, err = r.hera.getResponse()
 		if err != nil {
-			return err
+			break outer
 		}
 		switch ns.Cmd {
 		case common.RcValue:
-			r.vals = append(r.vals, ns.Payload)
+			localVals = append(localVals, ns.Payload)
 		case common.RcOK:
-			return nil
+			break outer
 		case common.RcNoMoreData:
+			r.completed = true
+			break outer
+		}
+	}
+	//process localVals if length > 0
+	if len(localVals) > 0 {
+		if logger.GetLogger().V(logger.Verbose) {
+			logger.GetLogger().Log(logger.Verbose, r.hera.id, "Rows: cols = ", r.cols, ", numValues =", len(localVals))
+		}
+		if len(localVals) >= (r.cols*5 + 1) {
 			if logger.GetLogger().V(logger.Verbose) {
-				logger.GetLogger().Log(logger.Verbose, r.hera.id, "Rows: cols = ", r.cols, ", numValues =", len(r.vals))
+				logger.GetLogger().Log(logger.Verbose, r.hera.id, "Column info present in data")
 			}
-			r.completed = true
-			return nil
+			r.vals = localVals[r.cols*5+1:]
+			//Create JSON string for column info
+			localVals = localVals[1 : r.cols*5+1]
+			for index := 0; index < len(localVals); {
+				columnData := localVals[index : index+5] //Each column has column info of size 5
+				columnInfoMap := make(map[string]string)
+				for index2 := 0; index2 < len(columnData); index2++ {
+					switch valType := columnData[index2].(type) {
+					case string:
+						columnInfoMap[columnInfo[index2]] = valType
+					case []byte:
+						columnInfoMap[columnInfo[index2]] = string(valType)
+					default:
+						if logger.GetLogger().V(logger.Verbose) {
+							logger.GetLogger().Log(logger.Verbose, r.hera.id, "Invalid data: ", valType)
+						}
+						columnInfoMap[columnInfo[index2]] = "UNDEFINED"
+					}
+				}
+				columnInfoStr, _ := json.Marshal(columnInfoMap)
+				r.colInfo = append(r.colInfo, string(columnInfoStr))
+				if logger.GetLogger().V(logger.Verbose) {
+					logger.GetLogger().Log(logger.Verbose, r.hera.id, "Column info :", string(columnInfoStr))
+				}
+				index = index + 5
+			}
+		} else {
+			r.vals = localVals
 		}
 	}
+	return err
 }
 
 // Columns returns the names of the columns. The number of
@@ -77,6 +122,9 @@ func (r *rows) fetchResults() error {
 // string should be returned for that entry.
 func (r *rows) Columns() []string {
 	// TODO using hera column names command
+	if len(r.colInfo) > 0 {
+		return r.colInfo
+	}
 	return make([]string, r.cols)
 }
 

client/gosqldriver/statement.go

@@ -252,7 +252,7 @@ func (st *stmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driv
 	return res, nil
 }
 
-// Implements driver.Stmt.
+// Query Implements driver.Stmt.
 // Query executes a query that may return rows, such as a SELECT.
 func (st *stmt) Query(args []driver.Value) (driver.Rows, error) {
 	sk := 0
@@ -264,7 +264,7 @@ func (st *stmt) Query(args []driver.Value) (driver.Rows, error) {
 		crid = 1
 	}
 	binds := len(args)
-	nss := make([]*netstring.Netstring, crid /*CmdClientCorrelationID*/ +1 /*CmdPrepare*/ +2*binds /* CmdBindName and BindValue */ +sk /*CmdShardKey*/ +1 /*CmdExecute*/ +1 /* CmdFetch */)
+	nss := make([]*netstring.Netstring, crid /*CmdClientCorrelationID*/ +1 /*CmdPrepare*/ +2*binds /* CmdBindName and BindValue */ +sk /*CmdShardKey*/ +1 /*CmdExecute*/ +1 /*CmdColsInfo*/ +1 /* CmdFetch */)
 	idx := 0
 	if crid == 1 {
 		nss[0] = st.hera.corrID
@@ -299,6 +299,8 @@ func (st *stmt) Query(args []driver.Value) (driver.Rows, error) {
 	}
 	nss[idx] = netstring.NewNetstringFrom(common.CmdExecute, nil)
 	idx++
+	nss[idx] = netstring.NewNetstringFrom(common.CmdColsInfo, nil)
+	idx++
 	nss[idx] = netstring.NewNetstringFrom(common.CmdFetch, st.fetchChunkSize)
 	cmd := netstring.NewNetstringEmbedded(nss)
 	err := st.hera.execNs(cmd)
@@ -354,7 +356,7 @@ Loop:
 	return newRows(st.hera, cols, st.fetchChunkSize)
 }
 
-// Implements driver.StmtQueryContextx
+// QueryContext Implements driver.StmtQueryContextx
 // QueryContext executes a query that may return rows, such as a SELECT
 func (st *stmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
 	// TODO: refactor Query/QueryContext to reuse code
@@ -368,7 +370,7 @@ func (st *stmt) QueryContext(ctx context.Context, args []driver.NamedValue) (dri
 		crid = 1
 	}
 	binds := len(args)
-	nss := make([]*netstring.Netstring, crid /*ClientCalCorrelationID*/ +1 /*CmdPrepare*/ +2*binds /* CmdBindName and BindValue */ +sk /*ShardKey*/ +1 /*Execute*/ +1 /* Fetch */)
+	nss := make([]*netstring.Netstring, crid /*ClientCalCorrelationID*/ +1 /*CmdPrepare*/ +2*binds /* CmdBindName and BindValue */ +sk /*ShardKey*/ +1 /*Execute*/ +1 /*CmdColsInfo*/ +1 /* Fetch */)
 	idx := 0
 	if crid == 1 {
 		nss[0] = st.hera.corrID
@@ -407,6 +409,8 @@ func (st *stmt) QueryContext(ctx context.Context, args []driver.NamedValue) (dri
 	}
 	nss[idx] = netstring.NewNetstringFrom(common.CmdExecute, nil)
 	idx++
+	nss[idx] = netstring.NewNetstringFrom(common.CmdColsInfo, nil)
+	idx++
 	nss[idx] = netstring.NewNetstringFrom(common.CmdFetch, st.fetchChunkSize)
 	cmd := netstring.NewNetstringEmbedded(nss)
 	err := st.hera.execNs(cmd)

go.mod

@@ -1,10 +1,15 @@
 module github.com/paypal/hera
 
-go 1.12
+go 1.18
 
 require (
 	github.com/go-sql-driver/mysql v1.4.1
 	github.com/godror/godror v0.26.3
 	github.com/lib/pq v1.10.3
+)
+
+require (
+	github.com/go-logfmt/logfmt v0.5.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
+	google.golang.org/protobuf v1.27.1 // indirect
 )

go.sum

@@ -1,35 +1,32 @@
-github.com/go-goracle/goracle v2.1.14+incompatible h1:uw4p4j6b6/ustNb/Yqd2rXIrBEJCGPh9ZCNFl00I9E0=
-github.com/go-goracle/goracle v2.1.14+incompatible/go.mod h1:+EZ+XI0XYe/t96DSQbwEA8wvYxlyJftRDsjUhw/EVI8=
-github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
-github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
+github.com/go-logfmt/logfmt v0.5.0 h1:TrB8swr/68K7m9CcGut2g3UOihhbcbiMAYiuTXdEih4=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
 github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA=
 github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
-github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/godror/godror v0.26.3 h1:V+z+Q/OBGgmmYzuAwyJzpcn4LsPF4Ev0xHAea68V00c=
 github.com/godror/godror v0.26.3/go.mod h1:1QCn6oXh3r+IlB3DLE8V6qkHXLSHd18a3Hw7szQ9/3Y=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
-github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/lib/pq v1.10.3 h1:v9QZf2Sn6AmjXtQeFpdoq/eaNtYP6IN+7lcrygsIAtg=
 github.com/lib/pq v1.10.3 h1:v9QZf2Sn6AmjXtQeFpdoq/eaNtYP6IN+7lcrygsIAtg=
 github.com/lib/pq v1.10.3/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
-github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
-github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/lib/pq v1.10.3/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
 golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
-golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e h1:vcxGaoTs7kV8m5Np9uUNQin4BrLOthgV7252N8V+FwY=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.6.7 h1:FZR1q0exgwxzPzp/aF+VccGrSfxfPpkBqjIIEq3ru6c=
 google.golang.org/appengine v1.6.7/go.mod h1:8WjMMxjGQR8xUklV/ARdw2HLXBOI7O7uCIDZVag1xfc=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-gopkg.in/goracle.v2 v2.12.3 h1:LnNrOqNF9xj6suiuenVcOTd00S9/4aneyHarj84EaT4=
-gopkg.in/goracle.v2 v2.12.3/go.mod h1:QhfGFGWSfZKBnBWnAkjd5GreYK1tan9ikkA7BMirttE=

tests/unittest/coordinator_rqid/main_test.go

@@ -52,10 +52,10 @@ func before() error {
 	if tableName == "" {
 		tableName = "jdbc_hera_test"
 	}
-        if strings.HasPrefix(os.Getenv("TWO_TASK"), "tcp") { // mysql
+	if strings.HasPrefix(os.Getenv("TWO_TASK"), "tcp") { // mysql
 		// with testutil.RunDML, extra log line throws off test
 		testutil.DBDirect("create table jdbc_hera_test ( ID BIGINT, INT_VAL BIGINT, STR_VAL VARCHAR(500))", os.Getenv("MYSQL_IP"), "heratestdb", testutil.MySQL)
-        }
+	}
 	return nil
 }
 
@@ -75,32 +75,33 @@ func TestCoordinatorRqId(t *testing.T) {
 	db.SetMaxIdleConns(0)
 	defer db.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	// cleanup and insert one row in the table
 	conn, err := db.Conn(ctx)
 	if err != nil {
-		t.Fatalf("Error getting connection %s\n", err.Error())
+		t.Errorf("Error getting connection %s\n", err.Error())
 	}
 	tx, _ := conn.BeginTx(ctx, nil)
 	stmt, _ := tx.PrepareContext(ctx, "/*TestCoordinatorRqId*/delete from "+tableName)
 	_, err = stmt.Exec()
 	if err != nil {
-		t.Fatalf("Error preparing test (delete table) %s\n", err.Error())
+		t.Errorf("Error preparing test (delete table) %s\n", err.Error())
 	}
 	stmt, _ = tx.PrepareContext(ctx, "/*TestCoordinatorRqId*/insert into "+tableName+" (id, int_val, str_val) VALUES(?, ?, ?)")
 	_, err = stmt.Exec(1, time.Now().Unix(), "val 1")
 	if err != nil {
-		t.Fatalf("Error preparing test (create row in table) %s\n", err.Error())
+		t.Errorf("Error preparing test (create row in table) %s\n", err.Error())
 	}
 	err = tx.Commit()
 	if err != nil {
-		t.Fatalf("Error commit %s\n", err.Error())
+		t.Errorf("Error commit %s\n", err.Error())
 	}
 
+	conn, err = db.Conn(ctx)
 	stmt, _ = conn.PrepareContext(ctx, "/*TestCoordinatorRqId*/Select id, int_val from "+tableName+" where id=?")
 	rows, _ := stmt.Query(1)
 	if !rows.Next() {
-		t.Fatalf("Expected 1 row")
+		t.Errorf("Expected 1 row")
 	}
 
 	rows.Close()
@@ -121,10 +122,10 @@ func TestCoordinatorRqId(t *testing.T) {
 		t.Fatalf("Expected 'wrqId: 13 ): EOR code: 0 , rqId:  13'")
 	}
 
-	out, err = testutil.BashCmd("grep 'wrqId: 19 ): EOR code: 0 , rqId:  19' hera.log | wc -l")
+	out, err = testutil.BashCmd("grep 'wrqId: 21 ): EOR code: 0 , rqId:  21' hera.log | wc -l")
 	if (err != nil) || (len(out) == 0) || (out[0] != '1') {
 		err = nil
-		t.Fatalf("Expected 'wrqId: 19 ): EOR code: 0 , rqId:  19'")
+		t.Fatalf("Expected 'wrqId: 21 ): EOR code: 0 , rqId:  21'")
 	}
 
 	logger.GetLogger().Log(logger.Debug, "TestCoordinatorRqId done  -------------------------------------------------------------")