Use Go 1.19

 Acc support for EndExecute to ensure data is cached if Postgres terminates a scan early (e.g. for non-pushed down limit)

Author: kaidaguerre

fdw.go

@@ -333,7 +333,6 @@ func goFdwEndForeignScan(node *C.ForeignScanState) {
 	if s != nil && pluginHub != nil {
 		log.Printf("[TRACE] goFdwEndForeignScan, iterator: %p", s.Iter)
 		pluginHub.EndScan(s.Iter, int64(s.State.limit))
-
 	}
 	ClearExecState(node.fdw_state)
 	node.fdw_state = nil

fdw/Makefile

@@ -33,7 +33,7 @@ PG_CFLAGS = -I${SERVER_LIB} -I${INTERNAL_LIB} -g
 include $(PGXS)
 
 go: ../fdw.go
-	go build -o steampipe_postgres_fdw.a -buildmode=c-archive ../*.go
+	go1.19rc1 build -o steampipe_postgres_fdw.a -buildmode=c-archive ../*.go
 
 inst:
 	mkdir -p ../build-${PLATFORM}

go.mod

@@ -1,9 +1,10 @@
 module github.com/turbot/steampipe-postgres-fdw
 
-go 1.18
+go 1.19
 
-//replace github.com/turbot/steampipe-plugin-sdk/v4 => /Users/kai/Dev/github/turbot/steampipe-plugin-sdk
-//replace github.com/turbot/steampipe => /Users/kai/Dev/github/turbot/steampipe
+replace github.com/turbot/steampipe-plugin-sdk/v4 => /Users/kai/Dev/github/turbot/steampipe-plugin-sdk
+
+replace github.com/turbot/steampipe => /Users/kai/Dev/github/turbot/steampipe
 
 require (
 	github.com/dgraph-io/ristretto v0.1.0 // indirect
@@ -112,7 +113,7 @@ require (
 	github.com/sethvargo/go-retry v0.1.0 // indirect
 	github.com/shiena/ansicolor v0.0.0-20200904210342-c7312218db18 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
-	github.com/sirupsen/logrus v1.8.1 // indirect
+	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/spf13/afero v1.8.2 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
@@ -145,7 +146,7 @@ require (
 	golang.org/x/net v0.0.0-20220520000938-2e3eb7b945c2 // indirect
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20220720214146-176da50484ac // indirect

go.sum

@@ -867,8 +867,8 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88=
 github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
-github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=
-github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
+github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
+github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/assertions v1.13.0 h1:Dx1kYM01xsSqKPno3aqLnrwac2LetPvN23diwyr69Qs=
 github.com/smartystreets/goconvey v0.0.0-20180222194500-ef6db91d284a/go.mod h1:XDJAKZRPZ1CvBcN2aX5YOUTYGHki24fSF0Iv48Ibg0s=
@@ -935,10 +935,6 @@ github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1
 github.com/tombuildsstuff/giovanni v0.15.1/go.mod h1:0TZugJPEtqzPlMpuJHYfXY6Dq2uLPrXf98D2XQSxNbA=
 github.com/turbot/go-kit v0.4.0 h1:EdD7Bf2EGAjvHRGQxRiWpDawzZSk3T+eghqbj74qiSc=
 github.com/turbot/go-kit v0.4.0/go.mod h1:SBdPRngbEfYubiR81iAVtO43oPkg1+ASr+XxvgbH7/k=
-github.com/turbot/steampipe v1.7.0-rc.0.0.20220722125407-74cf746365a4 h1:GwuhRfCpzalNNhqbAbhx1WASNHfTfZhbEhBAyWB7N/w=
-github.com/turbot/steampipe v1.7.0-rc.0.0.20220722125407-74cf746365a4/go.mod h1:IOmpgdlP0mhgftCUgqFbtG6rPdOAxtyRSz68tMuPo/s=
-github.com/turbot/steampipe-plugin-sdk/v4 v4.0.0-alpha.2 h1:Lq+b9zbrYEHRwQ2lH+hJCfEyVLNRGrhPTKJpOnslInw=
-github.com/turbot/steampipe-plugin-sdk/v4 v4.0.0-alpha.2/go.mod h1:v6bYQWg9z+3IhTpMG7kldX3b774T4eFRxU5llMam9X8=
 github.com/ugorji/go v0.0.0-20180813092308-00b869d2f4a5/go.mod h1:hnLbHMwcvSihnDhEfx2/BzKp2xb0Y+ErdfYcrs9tkJQ=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
@@ -1260,8 +1256,8 @@ golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a h1:dGzPydgVsqGcTRVwiLJ1jVbufYwmzD3LfVPLKsKg+0k=
-golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20201210144234-2321bbc49cbf/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

hub/hub.go

@@ -248,8 +248,8 @@ func (h *Hub) Abort() {
 	for _, iter := range h.runningIterators {
 		// read the scan metadata from the iterator and add to our stack
 		h.AddScanMetadata(iter)
-		// close the iterator
-		iter.Close()
+		// abort the iterator
+		iter.Abort()
 		// remove it from the saved list of iterators
 		h.RemoveIterator(iter)
 	}
@@ -633,12 +633,11 @@ func (h *Hub) startScan(iterator *scanIterator, queryContext *proto.QueryContext
 
 	table := iterator.table
 	connectionPlugin := iterator.connectionPlugin
-	callId := grpc.BuildCallId()
 
 	req := &proto.ExecuteRequest{
 		Table:                 table,
 		QueryContext:          queryContext,
-		CallId:                callId,
+		CallId:                iterator.callId,
 		TraceContext:          grpc.CreateCarrierFromContext(traceCtx.Ctx),
 		ExecuteConnectionData: make(map[string]*proto.ExecuteConnectionData),
 	}
@@ -655,12 +654,12 @@ func (h *Hub) startScan(iterator *scanIterator, queryContext *proto.QueryContext
 		req.ExecuteConnectionData[connectionName] = data
 	}
 
-	log.Printf("[INFO] StartScan for table: %s, callId %s, cache enabled: %v, iterator %p", table, callId, req.CacheEnabled, iterator)
+	log.Printf("[INFO] StartScan for table: %s, callId %s, cache enabled: %v, iterator %p", table, iterator.callId, req.CacheEnabled, iterator)
 	stream, ctx, cancel, err := connectionPlugin.PluginClient.Execute(req)
 	// format GRPC errors and ignore not implemented errors for backwards compatibility
 	err = grpc.HandleGrpcError(err, connectionPlugin.PluginName, "Execute")
 	if err != nil {
-		log.Printf("[WARN] startScan: plugin Execute function callId: %s returned error: %v\n", callId, err)
+		log.Printf("[WARN] startScan: plugin Execute function callId: %s returned error: %v\n", iterator.callId, err)
 		iterator.setError(err)
 		return err
 	}

hub/in_memory_iterator.go

@@ -60,6 +60,11 @@ func (i *inMemoryIterator) Close() {
 	i.status = QueryStatusReady
 }
 
+func (i *inMemoryIterator) Abort() {
+	log.Printf("[TRACE] inMemoryIterator Abort() (%p)", i)
+	i.Close()
+}
+
 func (i *inMemoryIterator) CanIterate() bool {
 	switch i.status {
 	case QueryStatusError, QueryStatusComplete:

hub/iterator.go

@@ -9,8 +9,10 @@ type Iterator interface {
 	ConnectionName() string
 	// Next returns next row. Nil slice means there is no more rows to scan.
 	Next() (map[string]interface{}, error)
-	// Close stops an iteration and frees any resources.
+	// Close stops an iteration, ensures the plugin writes outstanding ddata to the cache and frees any resources.
 	Close()
+	// Abort stops an iteration,and frees any resources.
+	Abort()
 	Status() queryStatus
 	Error() error
 	CanIterate() bool

hub/legacy_group_iterator.go

@@ -176,6 +176,10 @@ func (i *legacyGroupIterator) Close() {
 	i.traceCtx.Span.End()
 }
 
+func (i *legacyGroupIterator) Abort() {
+	// for legacy iterator just call close
+	i.Close()
+}
 func (i *legacyGroupIterator) CanIterate() bool {
 	switch i.Status() {
 	case QueryStatusError, QueryStatusComplete:

hub/legacy_scan_iterator.go

@@ -151,7 +151,11 @@ func (i *legacyScanIterator) Close() {
 	}
 
 	i.closeSpan()
+}
 
+func (i *legacyScanIterator) Abort() {
+	// for legacy iterator just call Close which ha ssame behaviour as abort
+	i.Close()
 }
 
 // CanIterate returns true if this iterator has results available to iterate

hub/scan_iterator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/turbot/steampipe-plugin-sdk/v4/grpc"
 	"log"
 	"time"
 
@@ -44,8 +45,8 @@ type scanIterator struct {
 	connectionPlugin   *steampipeconfig.ConnectionPlugin
 	cancel             context.CancelFunc
 	traceCtx           *telemetry.TraceCtx
-
-	startTime time.Time
+	startTime          time.Time
+	callId             string
 }
 
 func newScanIterator(hub *Hub, connectionPlugin *steampipeconfig.ConnectionPlugin, connectionName, table string, connectionLimitMap map[string]int64, qualMap map[string]*proto.Quals, columns []string, traceCtx *telemetry.TraceCtx) *scanIterator {
@@ -62,6 +63,7 @@ func newScanIterator(hub *Hub, connectionPlugin *steampipeconfig.ConnectionPlugi
 		connectionPlugin:   connectionPlugin,
 		traceCtx:           traceCtx,
 		startTime:          time.Now(),
+		callId:             grpc.BuildCallId(),
 	}
 }
 
@@ -148,6 +150,21 @@ func (i *scanIterator) Start(stream proto.WrapperPlugin_ExecuteClient, ctx conte
 }
 
 func (i *scanIterator) Close() {
+	// first, send a message to our plugin terminating execution
+	// (to ensure that after the context is cancelled, data is written to the cache)
+	req := &proto.EndExecuteRequest{
+		CallId: i.callId,
+	}
+	err := i.connectionPlugin.PluginClient.EndExecute(req)
+	if err != nil {
+		log.Printf("[WARN] EndExecute failed: %s", err.Error())
+	}
+
+	// now call abort to shut down the iterator and cancel the context
+	i.Abort()
+}
+
+func (i *scanIterator) Abort() {
 	// call the context cancellation function
 	i.cancel()