ExecuteBatchCAS not applied, MapExecuteBatchCAS is applied

[danthegoodman1]

Using the latest version of scylla on docker, I have the following code:

	b := tx.session.NewBatch(gocql.UnloggedBatch)

	for key, val := range tx.pendingWrites {
		// Write rowLock and data to primary key
		lock := rowLock{
			PrimaryLockKey: tx.primaryLockKey,
			StartTs:        tx.readTime.UnixNano(),
			TimeoutTs:      tx.readTime.Add(time.Second * 5).UnixNano(),
		}

		encodedLock, err := lock.Encode()
		if err != nil {
			return fmt.Errorf("error in rowLock.Encode: %w", err)
		}

		// Insert the lock
		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("insert into \"%s\" (key, ts, col, val) values (?, 0, 'l', ?) if not exists", tx.table),
			Args: []any{key, []byte(encodedLock)},
		})

		// Insert the data record
		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("insert into \"%s\" (key, ts, col, val) values (?, ?, 'd', ?) if not exists", tx.table),
			Args: []any{key, tx.readTime.UnixNano(), val},
		})
	}

	// Non-map version was always having applied: false
	applied, _, err := tx.session.MapExecuteBatchCAS(b, make(map[string]interface{}))
	if err != nil {
		return fmt.Errorf("error in MapExecuteBatchCAS: %w", err)
	}

	if !applied {
		return fmt.Errorf("%w: prewrite not applied (confict)", &TxnAborted{})
	}

When using ExecuteBatchCAS(b), the applied return would always be false, but when using MapExecuteBatchCAS(b, make(map[string]interface{})) (despite not actaully binding to anything), applied would be true.

Why is this? My understand is that it si a difference of binding only.

[testisnullus]

@danthegoodman1 Hello, I am investigating this issue. For now, I can verify that ExecuteBatchCAS() and MapExecuteBatchCAS() functions are working correctly.

When using MapExecuteBatchCAS(), the function involves passing a map which is intended to hold any values returned by the operation if the batch is not applied. The typical usage of MapExecuteBatchCAS() involves passing a map (even the empty one, like make(map[string]interface{})) that will store column values from the row that prevented the batch from being applied successfully.

Also, If the data state matches the condition (like IF NOT EXISTS in this particular case), the MapExecuteBatchCAS() operation can return applied = true, in case the data did not need any modification or was modified as expected. But, if the data state does not match, you cas see applied = false, indicating the batch was not applied because conditions weren't met.

[danthegoodman1]

@testisnullus I'm not sure you've understood the issue, or rather I was likely not clear enough. The issue is that ExecuteBatchCAS is returning applied false, when it should be true (there were no issues).

[testisnullus]

@danthegoodman1 I am testing on the Cassandra 4.1.4 version. The ExecuteBatchCAS() function works without any issues for me. I've tried to execute it with some data that is not in the table to bypass the IF NOT EXISTS statement provided in the CQL query:

		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("INSERT INTO \"%s\" (name, study_date, sleep_time_hours) VALUES (?, ?, ?) IF NOT EXISTS", tx.table),
			Args: []any{key, studyDate, sleepTimeHours32},
		})

		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("UPDATE \"%s\" SET sleep_time_hours=? WHERE name=? AND study_date=?", tx.table),
			Args: []any{sleepTimeHours32, key, "2023-04-30"},
		})

I've called the ExecuteBatchCAS() func the same way:

	applied, _, err := tx.session.ExecuteBatchCAS(b)
	if err != nil {
		return fmt.Errorf("error in ExecuteBatchCAS: %w", err)
	}

If I println the applied variable here, it appears true in this case.

Could you please provide more information about your issues with the ExecuteBatchCAS() function? Do you have any error messages related to this operation?

[danthegoodman1]

No errors, I noticed purely that it would never apply (checking the DB shows that it indeed did not apply), but if I simply swapped it with the Map- version it would apply.

[danthegoodman1]

Here is the keyspace and table I used

func createKeyspace(s *gocql.Session) error {
	if err := s.Query("CREATE KEYSPACE if not exists test WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor' : 1}").Exec(); err != nil {
		return fmt.Errorf("error creating keyspace: %w", err)
	}

	return nil
}

func createTable(s *gocql.Session) error {
	if err := s.Query("create table if not exists testkv (key text, col text, ts bigint, val blob, primary key (key, col, ts)) with clustering order by (col desc, ts desc)").Exec(); err != nil {
		return fmt.Errorf("error creating table: %w", err)
	}

	return nil
}

[testisnullus]

@danthegoodman1 Hello, I've tested the ExecuteBatchCAS() with your keyspace and table examples.

The table looks like this after batch operation:

ra@cqlsh> SELECT * FROM test.testkv;

 key  | col | ts                  | val
------+-----+---------------------+--------------------------
 key1 |   l |                   0 | 0x656e636f6465644c6f636b
 key1 |   d | 1714566727267344355 |           0x76616c756531

(2 rows)

I can share the full code snippet with you also:

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/gocql/gocql"
)

type Tx struct {
	session       *gocql.Session
	pendingWrites map[string][]byte
	readTime      time.Time
	table         string
}

type TxnAborted struct{}

func (e *TxnAborted) Error() string {
	return "Transaction aborted due to write conflict"
}

func createKeyspace(s *gocql.Session) error {
	if err := s.Query("CREATE KEYSPACE if not exists test WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor' : 1}").Exec(); err != nil {
		return fmt.Errorf("error creating keyspace: %w", err)
	}

	return nil
}

func createTable(s *gocql.Session) error {
	if err := s.Query("create table if not exists testkv (key text, col text, ts bigint, val blob, primary key (key, col, ts)) with clustering order by (col desc, ts desc)").Exec(); err != nil {
		return fmt.Errorf("error creating table: %w", err)
	}

	return nil
}

func (tx *Tx) ExecuteBatchTransaction() error {
	b := tx.session.NewBatch(gocql.UnloggedBatch)

	for key, val := range tx.pendingWrites {
		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("insert into \"%s\" (key, ts, col, val) values (?, 0, 'l', ?) if not exists", tx.table),
			Args: []any{key, []byte("encodedLock")},
		})

		// Insert the data record
		b.Entries = append(b.Entries, gocql.BatchEntry{
			Stmt: fmt.Sprintf("insert into \"%s\" (key, ts, col, val) values (?, ?, 'd', ?) if not exists", tx.table),
			Args: []any{key, tx.readTime.UnixNano(), val},
		})
	}

	applied, _, err := tx.session.ExecuteBatchCAS(b)
	if err != nil {
		return fmt.Errorf("error in ExecuteBatchCAS: %w", err)
	}

	if !applied {
		return fmt.Errorf("%w: prewrite not applied (conflict)", &TxnAborted{})
	}

	fmt.Println(applied)

	return nil
}

func main() {
	// connect to the cluster
	cluster := gocql.NewCluster("host1", "host2", "host3")
	cluster.Consistency = gocql.Quorum
	cluster.ProtoVersion = 4
	cluster.ConnectTimeout = time.Second * 10
	cluster.Keyspace = "test"
	cluster.Authenticator = gocql.PasswordAuthenticator{Username: "name", Password: "pass", AllowedAuthenticators: []string{"com.instaclustr.cassandra.auth.InstaclustrPasswordAuthenticator"}}
	session, err := cluster.CreateSession()
	if err != nil {
		log.Println(err)
		return
	}
	defer session.Close()

	err = createKeyspace(session)
	if err != nil {
		fmt.Println("Keyspace failed:", err)
	}

	err = createTable(session)
	if err != nil {
		fmt.Println("Table failed:", err)
	}

	tx := Tx{
		session:       session,
		pendingWrites: map[string][]byte{"key1": []byte("value1")},
		readTime:      time.Now(),
		table:         "testkv",
	}

	err = tx.ExecuteBatchTransaction()
	if err != nil {
		fmt.Println("Transaction failed:", err)
	}
}

The applied variable is also true in this case.

[testisnullus]

Just tested on the latest ScyllaDB image (scylladb/scylla:latest), and it also works well without any issues.

What version of GoCQL and Scylla do you use now?

[danthegoodman1]

Scylla is still latest, gocql was version v1.6.0

[danthegoodman1]

I was also quite confused considering I changed nothing on the query and simply swapped out ExecuteBatchCAS for MapExecuteBatchCAS, no other changes.

[testisnullus]

@danthegoodman1 I cannot reproduce the issue for now :(
Does the issue affect executing queries for you or can you easily use the MapExecuteBatchCAS() function for your operations? I will be tracking this issue and will try to test the ExecuteBatchCAS() more closely to identify this problem's root cause.

[danthegoodman1]

I had just swapped out for the map one with no issues.

…

On Mon, May 6, 2024 at 4:20 AM Danylo Savchenko ***@***.***> wrote: @danthegoodman1 <https://github.com/danthegoodman1> I cannot reproduce the issue for now :( Does the issue affect executing queries for you or can you easily use the MapExecuteBatchCAS() function for your operations? I will be tracking this issue and will try to test the ExecuteBatchCAS() more closely to identify this problem's root cause. — Reply to this email directly, view it on GitHub <#1746 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AEH3JLEDM2EDBGFGBXZCTK3ZA44MHAVCNFSM6AAAAABF3LN6KCVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMZDAOJVGQZTEOJSGI> . You are receiving this because you were mentioned.Message ID: ***@***.***>

[theoribeiro]

Interestingly, I'm also having the same issue happen to me. What's even more strange is that even though ExecuteBatchCAS() says it wasn't applied returning false (but no errors), the batch is applied successfully.
MapExecuteBatchCAS() works as expected!