Revert EndExecute function code - this would lead to incomplete cached results

Author: kaidaguerre

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/hashicorp/go-version v1.6.0 // indirect
 	github.com/turbot/go-kit v0.4.0
 	github.com/turbot/steampipe v0.16.0-rc.7
-	github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.1
+	github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.2
 	go.opentelemetry.io/otel v1.9.0
 	google.golang.org/protobuf v1.28.1
 )

go.sum

@@ -920,8 +920,8 @@ 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 v0.16.0-rc.7 h1:yzx2IMR8xFDZpsAMEPIota336lwpgzIJXW0U6pK3KeY=
 github.com/turbot/steampipe v0.16.0-rc.7/go.mod h1:iRCE1vgFaMgf1imtWQ7bi6VSdvc1aLs+xy5my2J1E6I=
-github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.1 h1:3rLG09QC94NiiMCzXjhbECvOUgSYmcKYZDko9fCXBKM=
-github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.1/go.mod h1:W6QjUrkeWyDgfnAt1qzckI7YEEY4x8WOevxKsVSWLM4=
+github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.2 h1:dr680ALqQ0MbGhOcALx1tXS2Q2ujpIxrwFFvFsSI250=
+github.com/turbot/steampipe-plugin-sdk/v4 v4.0.1-rc.2/go.mod h1:W6QjUrkeWyDgfnAt1qzckI7YEEY4x8WOevxKsVSWLM4=
 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/ulikunitz/xz v0.5.8 h1:ERv8V6GKqVi23rgu5cj9pVfVzJbOqAY2Ntl88O6c2nQ=

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)
-		// abort the iterator
-		iter.Abort()
+		// close the iterator
+		iter.Close()
 		// remove it from the saved list of iterators
 		h.RemoveIterator(iter)
 	}
@@ -641,11 +641,12 @@ 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:                iterator.callId,
+		CallId:                callId,
 		TraceContext:          grpc.CreateCarrierFromContext(traceCtx.Ctx),
 		ExecuteConnectionData: make(map[string]*proto.ExecuteConnectionData),
 	}
@@ -662,12 +663,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, iterator.callId, req.CacheEnabled, iterator)
+	log.Printf("[INFO] StartScan for table: %s, callId %s, cache enabled: %v, iterator %p", table, 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", iterator.callId, err)
+		log.Printf("[WARN] startScan: plugin Execute function callId: %s returned error: %v\n", callId, err)
 		iterator.setError(err)
 		return err
 	}

hub/in_memory_iterator.go

@@ -60,11 +60,6 @@ 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,10 +9,8 @@ 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, ensures the plugin writes outstanding data to the cache and frees any resources.
+	// Close stops an iteration 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,10 +176,6 @@ 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,11 +151,7 @@ func (i *legacyScanIterator) Close() {
 	}
 
 	i.closeSpan()
-}
 
-func (i *legacyScanIterator) Abort() {
-	// for legacy iterator just call Close which has same behaviour as abort
-	i.Close()
 }
 
 // CanIterate returns true if this iterator has results available to iterate

hub/scan_iterator.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/turbot/steampipe-plugin-sdk/v4/grpc"
 	"log"
 	"time"
 
@@ -45,8 +44,8 @@ type scanIterator struct {
 	connectionPlugin   *steampipeconfig.ConnectionPlugin
 	cancel             context.CancelFunc
 	traceCtx           *telemetry.TraceCtx
-	startTime          time.Time
-	callId             string
+
+	startTime time.Time
 }
 
 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 {
@@ -63,7 +62,6 @@ func newScanIterator(hub *Hub, connectionPlugin *steampipeconfig.ConnectionPlugi
 		connectionPlugin:   connectionPlugin,
 		traceCtx:           traceCtx,
 		startTime:          time.Now(),
-		callId:             grpc.BuildCallId(),
 	}
 }
 
@@ -150,21 +148,6 @@ 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()