Merge pull request #586 from v3io/development

dev

Author: dinal

backends/kv/writer.go

@@ -52,8 +52,7 @@ type Appender struct {
 
 const (
 	errorCodeString                        = "ErrorCode"
-	falseConditionOuterErrorCode           = "16777244"
-	falseConditionInnerErrorCode           = "16777245"
+	falseConditionErrorCode                = "16777244"
 	createNewItemOnlyExistingItemErrorCode = "369098809"
 
 	maximumAttributeNameLength = 256
@@ -571,9 +570,8 @@ func (a *Appender) updateItemWorker(doneChan chan<- struct{}) {
 func isFalseConditionError(err error) bool {
 	errString := err.Error()
 
-	if strings.Count(errString, errorCodeString) == 2 &&
-		strings.Contains(errString, falseConditionOuterErrorCode) &&
-		strings.Contains(errString, falseConditionInnerErrorCode) {
+	if strings.Count(errString, errorCodeString) == 1 &&
+		strings.Contains(errString, falseConditionErrorCode) {
 		return true
 	}
 

grpc/client.go

@@ -125,6 +125,7 @@ func (c *Client) Write(request *frames.WriteRequest) (frames.FrameAppender, erro
 		More:          request.HaveMore,
 		SaveMode:      request.SaveMode.String(),
 		PartitionKeys: request.PartitionKeys,
+		Condition:     request.Condition,
 	}
 
 	req := &pb.WriteRequest{

http/client.go

@@ -437,6 +437,7 @@ func pbWriteReq(req *frames.WriteRequest) (*pb.InitialWriteRequest, error) {
 		More:          req.HaveMore,
 		SaveMode:      req.SaveMode.String(),
 		PartitionKeys: req.PartitionKeys,
+		Condition:     req.Condition,
 	}
 
 	return msg, nil

test/kv_integration_test.go

@@ -1125,3 +1125,99 @@ func (kvSuite *KvTestSuite) TestWritePartitionedTableWithNullValues() {
 		kvSuite.Require().NoError(err, "item %v was not found", subPath)
 	}
 }
+
+func (kvSuite *KvTestSuite) TestUpdateExpressionWithCondition() {
+	table := fmt.Sprintf("kv_test_update_with_condition_%d", time.Now().UnixNano())
+
+	index := []string{"mike", "joe", "jim", "nil"}
+	icol, err := frames.NewSliceColumn("idx", index)
+	kvSuite.Require().NoError(err)
+
+	//boolCol
+	bools := []bool{true, true, false, false}
+	boolCol, err := frames.NewSliceColumn("bools", bools)
+	if err != nil {
+		kvSuite.Require().NoError(err)
+	}
+
+	numbers := []int{1, 1, 1, 1}
+	numbersCol, err := frames.NewSliceColumn("nums", numbers)
+	if err != nil {
+		kvSuite.Require().NoError(err)
+	}
+
+	columns := []frames.Column{numbersCol, boolCol}
+
+	frame, err := frames.NewFrame(columns, []frames.Column{icol}, nil)
+	kvSuite.Require().NoError(err)
+
+	kvSuite.T().Log("write: prepare")
+	wreq := &frames.WriteRequest{
+		Backend: kvSuite.backendName,
+		Table:   table,
+	}
+
+	appender, err := kvSuite.client.Write(wreq)
+	kvSuite.Require().NoError(err)
+
+	err = appender.Add(frame)
+	kvSuite.Require().NoError(err)
+
+	err = appender.WaitForComplete(3 * time.Second)
+	kvSuite.Require().NoError(err)
+
+	// Second Phase: Update existing rows with condition
+	kvSuite.T().Log("write: update with condition")
+	wreq = &frames.WriteRequest{
+		Backend:   kvSuite.backendName,
+		Table:     table,
+		SaveMode:  frames.UpdateItem,
+		Condition: "bools==True",
+	}
+
+	appender, err = kvSuite.client.Write(wreq)
+	kvSuite.Require().NoError(err)
+
+	newNumbers := []int{9, 9, 9, 9}
+	newNumbersCol, err := frames.NewSliceColumn("nums", newNumbers)
+	if err != nil {
+		kvSuite.Require().NoError(err)
+	}
+
+	newColumns := []frames.Column{newNumbersCol}
+
+	frame, err = frames.NewFrame(newColumns, []frames.Column{icol}, nil)
+	kvSuite.Require().NoError(err)
+
+	err = appender.Add(frame)
+	kvSuite.Require().NoError(err)
+
+	err = appender.WaitForComplete(3 * time.Second)
+	kvSuite.Require().NoError(err)
+
+	input := v3io.GetItemsInput{AttributeNames: []string{"__name", "nums", "bools"}}
+
+	iter, err := v3ioutils.NewAsyncItemsCursor(
+		kvSuite.v3ioContainer, &input, 1,
+		nil, kvSuite.internalLogger,
+		0, []string{table + "/"},
+		"", "")
+
+	for iter.Next() {
+		currentRow := iter.GetItem()
+		key, _ := currentRow.GetFieldString("__name")
+		if key == ".#schema" {
+			continue
+		}
+		num, _ := currentRow.GetFieldInt("nums")
+		b := currentRow.GetField("bools").(bool)
+
+		if b {
+			kvSuite.Require().Equal(9, num)
+		} else {
+			kvSuite.Require().Equal(1, num)
+		}
+	}
+
+	kvSuite.Require().NoError(iter.Err())
+}

test/kv_save_mode_integration_test.go

@@ -81,7 +81,7 @@ func (kvSuite *KvTestSuite) TestSaveModeOverwriteTableExists() {
 	err = appender.Add(frame2)
 	kvSuite.Require().NoError(err, "failed to write frame")
 
-	err = appender.WaitForComplete(time.Second)
+	err = appender.WaitForComplete(2 * time.Second)
 	kvSuite.Require().NoError(err)
 
 	rreq := &pb.ReadRequest{