fill main select channel in chunks

go.mod

@@ -4,44 +4,42 @@ go 1.19
 
 require (
 	github.com/Ompluscator/dynamic-struct v1.3.0
-	github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20220908174654-449a477778fc
+	github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20221121180710-54cb9eaf018c
 	github.com/fatih/color v1.13.0
 	github.com/go-sql-driver/mysql v1.6.0
 	github.com/posener/cmd v1.3.4
-	github.com/vbauerster/mpb/v5 v5.4.0
+	github.com/vbauerster/mpb/v8 v8.0.2
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 require (
 	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect
-	github.com/cenkalti/backoff/v4 v4.1.3 // indirect
+	github.com/cenkalti/backoff/v4 v4.2.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/fatih/structs v1.1.0 // indirect
 	github.com/fatih/structtag v1.2.0 // indirect
 	github.com/go-redis/redis/v8 v8.11.5 // indirect
 	github.com/go-redsync/redsync/v4 v4.6.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/jinzhu/copier v0.3.5 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mattn/go-runewidth v0.0.13 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/posener/complete/v2 v2.0.1-alpha.13 // indirect
 	github.com/posener/formatter v1.0.0 // indirect
 	github.com/posener/script v1.1.5 // indirect
 	github.com/rivo/uniseg v0.4.2 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
-	github.com/vbauerster/mpb/v8 v8.0.2 // indirect
 	github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	go.uber.org/zap v1.23.0 // indirect
-	golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561 // indirect
-	golang.org/x/sync v0.0.0-20220907140024-f12130a52804 // indirect
+	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.0.0-20220909162455-aba9fc2a8ff2 // indirect
 )

go.sum

@@ -1,15 +1,14 @@
 github.com/Ompluscator/dynamic-struct v1.3.0 h1:Ozvw+T2UOmV/2J2/L7tPZxrypKen3CGSBwmWj8pwC4Q=
 github.com/Ompluscator/dynamic-struct v1.3.0/go.mod h1:01g22H1GC9IFcrpQ4JQBkzynp8RoT0wmUMx/OvXNnw8=
-github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20220908174654-449a477778fc h1:65MGz5nDTePXUkf30erKOKi8rUqk+9tFMoxGEhyMMXY=
-github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20220908174654-449a477778fc/go.mod h1:WDR25pLH1erLhCTlYS0aOzVMGgKuJWj22G+HRR8Hq/k=
-github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
+github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20221121180710-54cb9eaf018c h1:2PZt2AMS8XsmSAetlzYAujlhCdu4/dibWbnaaJ+ZWsw=
+github.com/StirlingMarketingGroup/cool-mysql v0.0.0-20221121180710-54cb9eaf018c/go.mod h1:R90FFhMAShSo8C2Fk9DN4/z6w0JWdDOXOWj7Tn1kcyU=
 github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow=
 github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
-github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8UtC4=
-github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
+github.com/cenkalti/backoff/v4 v4.2.0 h1:HN5dHm3WBOgndBH6E8V0q2jIYIR3s9yglV8k/+MN3u4=
+github.com/cenkalti/backoff/v4 v4.2.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
@@ -63,6 +62,8 @@ github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
 github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -86,7 +87,6 @@ github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Ky
 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
 github.com/mattn/go-isatty v0.0.16 h1:bq3VjFmv/sOjHtdEhmkEV4x1AJtvUvOJ2PFAZ5+peKQ=
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
-github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/mattn/go-runewidth v0.0.13 h1:lTGmDsbAYt5DmK6OnoV7EuIF1wEIFAcxld6ypU4OSgU=
 github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
@@ -110,7 +110,6 @@ github.com/onsi/gomega v1.20.0 h1:8W0cWlwFkflGPLltQvLRB7ZVD5HuP6ng320w2IS245Q=
 github.com/onsi/gomega v1.20.0/go.mod h1:DtrZpjmvpn2mPm4YWQa0/ALMDj9v4YxLgojwPeREyVo=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
-github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/posener/autogen v0.0.2/go.mod h1:23ND5WRzjjNM+lOMUvy4WudgDikSK3Sm0rmaXAfnIWo=
@@ -138,8 +137,6 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203 h1:QVqDTf3h2WHt08YuiTGPZLls0Wq99X9bWd0Q5ZSBesM=
 github.com/stvp/tempredis v0.0.0-20181119212430-b82af8480203/go.mod h1:oqN97ltKNihBbwlX8dLpwxCl3+HnXKV/R0e+sRLd9C8=
-github.com/vbauerster/mpb/v5 v5.4.0 h1:n8JPunifvQvh6P1D1HAl2Ur9YcmKT1tpoUuiea5mlmg=
-github.com/vbauerster/mpb/v5 v5.4.0/go.mod h1:fi4wVo7BVQ22QcvFObm+VwliQXlV1eBT8JDaKXR4JGI=
 github.com/vbauerster/mpb/v8 v8.0.2 h1:alVQG69Jg5+Ku9Hu1dakDx50uACEHnIzS7i356NQ/Vs=
 github.com/vbauerster/mpb/v8 v8.0.2/go.mod h1:Z9VJYIzXls7xZwirZjShGsi+14enzJhQfGyb/XZK0ZQ=
 github.com/vmihailenco/msgpack/v5 v5.3.5 h1:5gO0H1iULLWGhs2H5tbAHIZTV8/cYafcFOr9znI5mJU=
@@ -160,8 +157,6 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561 h1:MDc5xs78ZrZr3HMQugiXOAkSZtfTpbJLDr/lwfgO53E=
-golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561/go.mod h1:cyybsKvd6eL0RnXn6p/Grxp8F5bW7iYuBgsNCOHpMYE=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -180,8 +175,8 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20220907140024-f12130a52804 h1:0SH2R3f1b1VmIMG7BXbEZCBUu2dKmHschSmjqGUrW8A=
-golang.org/x/sync v0.0.0-20220907140024-f12130a52804/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -195,7 +190,6 @@ golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20201218084310-7d0127a74742/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

main.go

@@ -29,6 +29,8 @@ var (
 
 	tempTableSuffix = root.String("suffix", "_smgla_", "suffix of the temp table used for initial creation before the swap and drop")
 
+	verbose = root.Bool("v", false, "writes the full query log to stdout")
+
 	args = root.Args("connection", "connection, ex:\n"+
 		"smg-live-alter [flags] 'user:pass@(host)/dbname'\n\n"+
 		"see: https://github.com/go-sql-driver/mysql#dsn-data-source-name\n\n"+
@@ -64,6 +66,12 @@ func main() {
 		panic(err)
 	}
 
+	if *verbose {
+		db.Log = func(query string, params cool.Params, duration time.Duration, cacheHit bool) {
+			log.Println(query)
+		}
+	}
+
 	db.DisableUnusedColumnWarnings = true
 
 	alterQuery, err := promptText()
@@ -272,28 +280,68 @@ func main() {
 		panic(err)
 	}
 
-	newRowStruct, err := tableRowStruct(newColumns)
+	newRowStruct, pkIndexes, err := tableRowStruct(newColumns)
 	if err != nil {
 		panic(err)
 	}
 
 	// this gets the "type" of our struct from our dynamic struct
-	structType := reflect.Indirect(reflect.ValueOf(newRowStruct.Build().New())).Type()
+	structType := reflect.ValueOf(newRowStruct.Build().New()).Elem().Type()
 	// and then we make a channel with reflection for our new type of struct
 	chRef := reflect.MakeChan(reflect.ChanOf(reflect.BothDir, structType), *rowBufferSize)
 	ch := chRef.Interface()
 
-	// oh yeah, that's just one query. We don't actually have to chunk this selection
-	// because we're dealing with rows as they come in, instead of trying to select them
-	// all into memory or something first, which makes this code dramatically simpler
-	// and should work with tables of all sizes
+	prevIDs := make([]any, len(pkIndexes))
+
+	var exists bool
+	destFunc := reflect.MakeFunc(reflect.FuncOf([]reflect.Type{structType}, nil, false),
+		func(args []reflect.Value) (results []reflect.Value) {
+			chRef.Send(args[0])
+			exists = true
+
+			for i, field := range pkIndexes {
+				prevIDs[i] = args[0].Field(field).Interface()
+			}
+
+			return nil
+		})
+
 	go func() {
 		defer chRef.Close()
 
+		firstChunk := true
+
 		log.Println("selecting all the rows!")
-		err := db.Select(ch, "select /*+ MAX_EXECUTION_TIME(2147483647) */ "+selectColumns.String()+"from`"+tableName+"`", 0)
-		if err != nil {
-			panic(err)
+		for {
+			var where string
+			if !firstChunk {
+				where, _ = cool.InlineParams("where(@@pks)>(@@prevIDs)", cool.Params{
+					"pks":     cool.RawMySQL(quoteColumns(oldPrimaryColumns)),
+					"prevIDs": prevIDs,
+				})
+			}
+
+			exists = false
+			err := db.Select(destFunc.Interface(), "select /*+ MAX_EXECUTION_TIME(2147483647) */@@cols "+
+				"from @@table "+
+				"@@where "+
+				"order by @@pks "+
+				"limit @@limit ", 0, cool.Params{
+				"cols":  cool.RawMySQL(selectColumns.String()),
+				"table": cool.RawMySQL(fmt.Sprintf("`%s`", tableName)),
+				"where": cool.RawMySQL(where),
+				"pks":   cool.RawMySQL(quoteColumns(oldPrimaryColumns)),
+				"limit": *rowBufferSize,
+			})
+			if err != nil {
+				log.Fatalf("failed to execute main select: %v", err)
+			}
+
+			firstChunk = false
+
+			if !exists {
+				break
+			}
 		}
 	}()
 
@@ -317,6 +365,8 @@ func main() {
 	targetChunkTime := 500 * time.Millisecond
 	chunkStartTime := time.Now()
 
+	originalMaxInsertSize := db.MaxInsertSize.Get()
+
 	// start the import!
 	// Now this *does* have to be chunked because there's no way to stream
 	// rows to mysql, but cool mysql handles this for us, all it needs is the same
@@ -325,12 +375,24 @@ func main() {
 		chunkTime := time.Since(chunkStartTime)
 		if chunkTime > targetChunkTime {
 			db.MaxInsertSize.Set(int(float64(db.MaxInsertSize.Get()) * float64(targetChunkTime) / float64(chunkTime)))
+		} else {
+			current := db.MaxInsertSize.Get()
+			ratio := int(float64(db.MaxInsertSize.Get()) * float64(targetChunkTime) / float64(chunkTime))
+
+			addl := ratio - current
+			newMaxInsertSize := current + addl/10
+
+			if newMaxInsertSize <= originalMaxInsertSize {
+				// if the last chunk took too long, we drop the insert chunk size immediately,
+				// but if the chunk inserted faster than target time then increase the chunk size,
+				// but only by 10% of the difference, allowing for a steady increase
+				db.MaxInsertSize.Set(current + addl/10)
+			}
 		}
-
+		chunkStartTime = time.Now()
+	}).SetAfterRowExec(func(start time.Time) {
 		bar.Increment()
 		bar.DecoratorEwmaUpdate(time.Since(start))
-
-		chunkStartTime = time.Now()
 	}).Insert("insert ignore into`"+tempTableName+"`", ch)
 	if err != nil {
 		panic(err)

struct.go

@@ -10,11 +10,12 @@ import (
 	cool "github.com/StirlingMarketingGroup/cool-mysql"
 )
 
-func tableRowStruct(columns []*column) (dynamicstruct.Builder, error) {
+func tableRowStruct(columns []*column) (bld dynamicstruct.Builder, pkIndexes []int, err error) {
 	// this is our dynamic struct of the actual row, which will have
 	// properties added to it for each column in the following loop
 	rowStruct := dynamicstruct.NewStruct()
 
+	i := 0
 	for _, c := range columns {
 		// you can't insert into generated columns, and mysql will actually
 		// throw errors if you try and do this, so we simply skip them altogether
@@ -71,7 +72,7 @@ func tableRowStruct(columns []*column) (dynamicstruct.Builder, error) {
 			// our cool mysql literal is exactly what it sounds like;
 			// passed directly into the query with no escaping, which is know is
 			// safe here because a decimal from mysql can't contain breaking characters
-			v = new(cool.Literal)
+			v = new(cool.RawMySQL)
 		case "timestamp", "date", "datetime":
 			v = new(string)
 		case "binary", "varbinary", "blob", "tinyblob", "mediumblob", "longblob":
@@ -87,11 +88,15 @@ func tableRowStruct(columns []*column) (dynamicstruct.Builder, error) {
 		case "set":
 			v = new(any)
 		default:
-			return nil, fmt.Errorf("unknown mysql column of type %q", c.ColumnType)
+			return nil, nil, fmt.Errorf("unknown mysql column of type %q", c.ColumnType)
 		}
 
 		rowStruct.AddField(f, v, tag)
+		if c.PrimaryKey {
+			pkIndexes = append(pkIndexes, i)
+		}
+		i++
 	}
 
-	return rowStruct, nil
+	return rowStruct, pkIndexes, nil
 }