Use an INSERT rather than an UPSERT when only touching partition/clus…

…tering columns

This is protection against a nasty Cassandra bug where UPSERTing a row
with all the non-partition/clustering column keys as 'null' causes the
entire row to be deleted. This is undesirable behaviour and we should
rather use an INSERT to retain the original data

https://issues.apache.org/jira/browse/CASSANDRA-11805

table.go

@@ -3,6 +3,7 @@ package gocassa
 import (
 	"bytes"
 	"fmt"
+	"reflect"
 	"strconv"
 	"strings"
 
@@ -124,6 +125,25 @@ func transformFields(m map[string]interface{}) {
 	}
 }
 
+// allFieldValuesAreNullable checks if all the fields passed in contain
+// items that will eventually be marked in Cassandra as null (so we can
+// better predict if we should do an INSERT or an UPSERT). This is to
+// protect against https://issues.apache.org/jira/browse/CASSANDRA-11805
+func allFieldValuesAreNullable(fields map[string]interface{}) bool {
+	for _, value := range fields {
+		rv := reflect.ValueOf(value)
+		switch rv.Kind() {
+		case reflect.Array, reflect.Slice, reflect.Map:
+			if rv.Len() > 0 {
+				return false
+			}
+		default:
+			return false
+		}
+	}
+	return true
+}
+
 // INSERT INTO Hollywood.NerdMovies (user_uuid, fan)
 //   VALUES ('cfd66ccc-d857-4e90-b1e5-df98a3d40cd6', 'johndoe')
 //
@@ -161,10 +181,8 @@ func (t t) Set(i interface{}) Op {
 	}
 	ks := append(t.info.keys.PartitionKeys, t.info.keys.ClusteringColumns...)
 	updFields := removeFields(m, ks)
-	if len(updFields) == 0 {
-		return newWriteOp(t.keySpace.qe, filter{
-			t: t,
-		}, insertOpType, m)
+	if len(updFields) == 0 || allFieldValuesAreNullable(updFields) {
+		return newWriteOp(t.keySpace.qe, filter{t: t}, insertOpType, m)
 	}
 	transformFields(updFields)
 	rels := relations(t.info.keys, m)

table_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/gocql/gocql"
+	"github.com/stretchr/testify/assert"
 )
 
 func createIf(cs TableChanger, tes *testing.T) {
@@ -351,3 +352,34 @@ func TestExecuteWithConsistency(t *testing.T) {
 		t.Fatal(fmt.Sprint("Expected consistency:", cons, "got:", resultOpts.Consistency))
 	}
 }
+
+func TestAllFieldValuesAreNullable(t *testing.T) {
+	// all collection types defined are nullable
+	assert.True(t, allFieldValuesAreNullable(map[string]interface{}{
+		"field1": []byte{},
+		"field2": map[string]string{},
+		"field3": [0]int{},
+	}))
+
+	// not nullable due to populated byte array
+	assert.False(t, allFieldValuesAreNullable(map[string]interface{}{
+		"field1": []byte{0x00},
+		"field2": map[string]string{},
+		"field3": [0]int{},
+	}))
+
+	// not nullable due to string
+	assert.False(t, allFieldValuesAreNullable(map[string]interface{}{
+		"field1": []byte{},
+		"field4": "",
+	}))
+
+	// not nullable due to int
+	assert.False(t, allFieldValuesAreNullable(map[string]interface{}{
+		"field2": map[string]string{},
+		"field5": 0,
+	}))
+
+	// the empty field list is nullable
+	assert.True(t, allFieldValuesAreNullable(map[string]interface{}{}))
+}