Revert EndExecute

Author: kaidaguerre

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)
 	}
@@ -633,11 +633,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),
 	}
@@ -654,12 +655,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 ddata 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 ha ssame 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()