Fix the race condition in ExecuteAsync && correctly implement fast-path in Execute() (#13)

* Fix the race condition in ExecAsync:

- Never do more than one HTTP fetch in ExecAsync
- If it's a fast-path, multipart response, use the data that's already
  in the response, instead of fetching it again from the cloud.

NOTE:
- Exec is now correct, though not as performant as it could be,
    - b/c it currently always loops until relationalai.com reports the
      Transcation is COMPLETED
    - missing the fast-path optimization

* Add fast-path optimization to Execute().

We added a GotCompleteResult boolean to indicate whether or not a
TransactionAsyncResult is a fast-path response or not.

We think all the SDKs should follow this model.

Co-Authored-By: pete.vilter@gmail.com

* Fix type error in our ExecuteAsync code

* Add Execute() example file

Author: NHDaly

examples/execute/main.go

@@ -0,0 +1,71 @@
+// Copyright 2022 RelationalAI, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"io/ioutil"
+	"log"
+	"os"
+
+	"github.com/jessevdk/go-flags"
+	"github.com/relationalai/rai-sdk-go/rai"
+)
+
+type Options struct {
+	Database string `short:"d" long:"database" required:"true" description:"database name"`
+	Engine   string `short:"e" long:"engine" required:"true" descriptio:"engine name"`
+	Code     string `short:"c" long:"code" description:"rel source code"`
+	File     string `short:"f" long:"file" description:"rel source file"`
+	Readonly bool   `long:"readonly" description:"readonly query (default: false)"`
+	Profile  string `long:"profile" default:"default" description:"config profile"`
+}
+
+func getCode(opts *Options) (string, error) {
+	if opts.Code != "" {
+		return opts.Code, nil
+	}
+	bytes, err := ioutil.ReadFile(opts.File)
+	if err != nil {
+		return "", nil
+	}
+	return string(bytes), nil
+}
+
+func run(opts *Options) error {
+	client, err := rai.NewClientFromConfig(opts.Profile)
+	if err != nil {
+		return err
+	}
+	source, err := getCode(opts)
+	if err != nil {
+		return err
+	}
+	rsp, err := client.Execute(opts.Database, opts.Engine, source, nil, opts.Readonly)
+	if err != nil {
+		return err
+	}
+	rai.Print(rsp, 4)
+	return nil
+}
+
+func main() {
+	var opts Options
+	if _, err := flags.ParseArgs(&opts, os.Args); err != nil {
+		os.Exit(1)
+	}
+	if err := run(&opts); err != nil {
+		log.Fatal(err)
+	}
+}

rai/client.go

@@ -271,26 +271,41 @@ func unmarshal(rsp *http.Response, result interface{}) error {
 		return nil
 	}
 
+	dstValues := reflect.ValueOf(result).Elem()
+
 	switch data.(type) {
 
-	case []byte:
-		err := json.Unmarshal(data.([]byte), &result)
-		if err != nil {
+	case []byte: // Got back a JSON response
+
+		if dstValues.Type() == reflect.TypeOf(TransactionAsyncResult{}) {
+			// But the user asked for the whole TransactionResult struct,
+			// so we need to parse the JSON TransactionResponse, and fill it in
+			// the TransactionResult.
+			var txnResult TransactionAsyncResult
+			err := json.Unmarshal(data.([]byte), &txnResult.Transaction)
+			if err != nil {
+				return err
+			}
+			txnResult.GotCompleteResult = false
+
+			// Set it into result
+			srcValues := reflect.ValueOf(txnResult)
+
+			dstValues.Set(srcValues)
 			return err
 		}
 
-	case []TransactionAsyncFile:
-		dstValues := reflect.ValueOf(result).Elem()
-		if dstValues.Type() == reflect.TypeOf(TransactionAsyncResponse{}) {
-			rsp, err := readTransactionAsyncFiles(data.([]TransactionAsyncFile))
-			srcValues := reflect.ValueOf(rsp.Transaction)
-			dstValues.Set(srcValues)
+		// If they asked for just a regular JSON object, unmarshal it.
+		err := json.Unmarshal(data.([]byte), &result)
+		if err != nil {
 			return err
 		}
 
+	case []TransactionAsyncFile: // Multipart response
 		if dstValues.Type() == reflect.TypeOf(TransactionAsyncResult{}) {
 			rsp, err := readTransactionAsyncFiles(data.([]TransactionAsyncFile))
-			srcValues := reflect.ValueOf(rsp)
+			rsp.GotCompleteResult = true
+			srcValues := reflect.ValueOf(*rsp)
 			dstValues.Set(srcValues)
 			return err
 		}
@@ -489,7 +504,7 @@ func readTransactionAsyncFiles(files []TransactionAsyncFile) (*TransactionAsyncR
 		return nil, err
 	}
 
-	return &TransactionAsyncResult{txn, results, metadata, problems}, nil
+	return &TransactionAsyncResult{true, txn, results, metadata, problems}, nil
 }
 
 type HTTPError struct {
@@ -1194,40 +1209,10 @@ func (c *Client) ExecuteAsync(
 		Source:   source,
 		Readonly: readonly,
 	}
-	var txn TransactionAsyncResponse
+	txnResult := &TransactionAsyncResult{}
 	data := tx.payload(inputs)
-	err := c.Post(PathTransactions, tx.QueryArgs(), data, &txn)
-	if txn.State == "CREATED" {
-		return &TransactionAsyncResult{
-			Transaction: txn,
-			Results:     make([]ArrowRelation, 0),
-			Metadata:    make([]TransactionAsyncMetadataResponse, 0),
-			Problems:    make([]interface{}, 0),
-		}, err
-	}
-
-	results, err := c.GetTransactionResults(txn.ID)
-	if err != nil {
-		return nil, err
-	}
-
-	metadata, err := c.GetTransactionMetadata(txn.ID)
-	if err != nil {
-		return nil, err
-	}
-
-	problems, err := c.GetTransactionProblems(txn.ID)
-	if err != nil {
-		return nil, err
-	}
-
-	return &TransactionAsyncResult{
-		Transaction: txn,
-		Results:     results,
-		Metadata:    metadata,
-		Problems:    problems,
-	}, nil
-
+	err := c.Post(PathTransactions, tx.QueryArgs(), data, txnResult)
+	return txnResult, err
 }
 
 func (c *Client) Execute(
@@ -1240,6 +1225,13 @@ func (c *Client) Execute(
 		return nil, err
 	}
 
+	// Fast-path optimization
+	if rsp.GotCompleteResult {
+		return rsp, err
+	}
+
+	// Slow-path
+
 	id := rsp.Transaction.ID
 	count := 0
 	for {
@@ -1264,7 +1256,7 @@ func (c *Client) Execute(
 	metadata, _ := c.GetTransactionMetadata(id)
 	problems, _ := c.GetTransactionProblems(id)
 
-	return &TransactionAsyncResult{txn.Transaction, results, metadata, problems}, nil
+	return &TransactionAsyncResult{true, txn.Transaction, results, metadata, problems}, nil
 }
 
 func (c *Client) GetTransactions() (*TransactionAsyncMultipleResponses, error) {

rai/models.go

@@ -301,8 +301,10 @@ type Source struct {
 }
 
 type TransactionAsyncResult struct {
-	Transaction TransactionAsyncResponse
-	Results     []ArrowRelation
-	Metadata    []TransactionAsyncMetadataResponse
-	Problems    []interface{}
+	// If !GotCompleteResult, keep polling until Transaction reaches terminal State.
+	GotCompleteResult bool
+	Transaction       TransactionAsyncResponse
+	Results           []ArrowRelation
+	Metadata          []TransactionAsyncMetadataResponse
+	Problems          []interface{}
 }