Repository: openfga/openfga

Issue:
Datastore.Write pagination for deletes, inserts & changelog same as select for update
In order to support larger inputs for `Write` command we need to update statement builders for Delete tuple, Insert tuple and Insert Changelog statements under Datastore.Write implementation(s)

> Yes, that's correct. We should probably page everything.
> 
> For now we're limiting to 40 items per write, so this won't be applicable, but if we want to increase the num writes - that would make sense.
> 
> Can we create an issue & do it in a future PR?

_Originally posted by @cikasfm in https://github.com/openfga/openfga/pull/2663#discussion_r2325668849_
            

Discussion:

Solution patch:
diff --git a/pkg/storage/sqlcommon/sqlcommon.go b/pkg/storage/sqlcommon/sqlcommon.go
index a508f2cf..c35f1c3d 100644
--- a/pkg/storage/sqlcommon/sqlcommon.go
+++ b/pkg/storage/sqlcommon/sqlcommon.go
@@ -578,6 +578,34 @@ func buildRowConstructorIN(keys []tupleLockKey) (string, []interface{}) {
 	return sb.String(), args
 }
 
+// selectExistingRowsForWrite selects existing rows for the given keys and locks them FOR UPDATE.
+// The existing rows are added to the existing map.
+func selectExistingRowsForWrite(ctx context.Context, dbInfo *DBInfo, store string, keys []tupleLockKey, txn *sql.Tx, existing map[string]*openfgav1.Tuple) error {
+	inExpr, args := buildRowConstructorIN(keys)
+
+	selectBuilder := dbInfo.stbl.
+		Select(SQLIteratorColumns()...).
+		From("tuple").
+		Where(sq.Eq{"store": store}).
+		// Row-constructor IN on full composite key for precise point locks.
+		Where(sq.Expr("(object_type, object_id, relation, _user, user_type) IN "+inExpr, args...)).
+		Suffix("FOR UPDATE").
+		RunWith(txn) // make sure to run in the same transaction
+
+	iter := NewSQLTupleIterator(selectBuilder, dbInfo.HandleSQLError)
+	defer iter.Stop()
+
+	items, _, err := iter.ToArray(ctx, storage.PaginationOptions{PageSize: len(keys)})
+
+	if err != nil {
+		return err
+	}
+	for _, tuple := range items {
+		existing[tupleUtils.TupleKeyToString(tuple.GetKey())] = tuple
+	}
+	return nil
+}
+
 // Write provides the common method for writing to database across sql storage.
 func Write(
 	ctx context.Context,
@@ -598,72 +626,33 @@ func Write(
 	// 2. Compile a SELECT … FOR UPDATE statement to read the tuples for writes and lock tuples for deletes
 	// Build a deduped, sorted list of keys to lock.
 	lockKeys := makeTupleLockKeys(deletes, writes)
-	if len(lockKeys) == 0 {
+	total := len(lockKeys)
+	if total == 0 {
 		// Nothing to do.
 		return nil
 	}
 
-	existing := make(map[string]*openfgav1.Tuple, len(lockKeys))
+	existing := make(map[string]*openfgav1.Tuple, total)
 
 	// 3. If list compiled in step 2 is not empty, execute SELECT … FOR UPDATE statement
 
-	// Efficient, deterministic locking: probe existing rows using a row-constructor IN on the composite key.
-	// Batch to keep SQL size reasonable.
-	const lockBatchSize = 128
-
-	for i := 0; i < len(lockKeys); i += lockBatchSize {
-		j := i + lockBatchSize
-		if j > len(lockKeys) {
-			j = len(lockKeys)
+	for start := 0; start < total; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > total {
+			end = total
 		}
-		keys := lockKeys[i:j]
-
-		inExpr, args := buildRowConstructorIN(keys)
-
-		selectBuilder := dbInfo.stbl.
-			Select(SQLIteratorColumns()...).
-			From("tuple").
-			Where(sq.Eq{"store": store}).
-			// Row-constructor IN on full composite key for precise point locks.
-			Where(sq.Expr("(object_type, object_id, relation, _user, user_type) IN "+inExpr, args...)).
-			Suffix("FOR UPDATE").
-			RunWith(txn) // make sure to run in the same transaction
-
-		iter := NewSQLTupleIterator(selectBuilder, dbInfo.HandleSQLError)
-		defer iter.Stop()
+		keys := lockKeys[start:end]
 
-		items, _, err := iter.ToArray(ctx, storage.PaginationOptions{PageSize: len(keys)})
-
-		if err != nil {
+		if err = selectExistingRowsForWrite(ctx, dbInfo, store, keys, txn, existing); err != nil {
 			return err
 		}
-		for _, tuple := range items {
-			existing[tupleUtils.TupleKeyToString(tuple.GetKey())] = tuple
-		}
 	}
 
-	changelogBuilder := dbInfo.stbl.
-		Insert("changelog").
-		Columns(
-			"store",
-			"object_type",
-			"object_id",
-			"relation",
-			"_user",
-			"condition_name",
-			"condition_context",
-			"operation",
-			"ulid",
-			"inserted_at",
-		)
-
-	deleteBuilder := dbInfo.stbl.Delete("tuple").Where(sq.Eq{"store": store})
+	changeLogItems := make([][]interface{}, 0, len(deletes)+len(writes))
 
 	// ensures increasingly unique values within a single thread
 	entropy := ulid.DefaultEntropy()
 
-	var deleteCount int64
-
 	deleteConditions := sq.Or{}
 
 	// 4. For deletes
@@ -692,8 +681,6 @@ func Write(
 			}
 		}
 
-		deleteCount++
-
 		id := ulid.MustNew(ulid.Timestamp(now), entropy).String()
 		objectType, objectID := tupleUtils.SplitObject(tk.GetObject())
 
@@ -705,7 +692,7 @@ func Write(
 			"user_type":   tupleUtils.GetUserTypeFromUser(tk.GetUser()),
 		})
 
-		changelogBuilder = changelogBuilder.Values(
+		changeLogItems = append(changeLogItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -716,25 +703,10 @@ func Write(
 			openfgav1.TupleOperation_TUPLE_OPERATION_DELETE,
 			id,
 			sq.Expr("NOW()"),
-		)
-	}
-
-	insertBuilder := dbInfo.stbl.
-		Insert("tuple").
-		Columns(
-			"store",
-			"object_type",
-			"object_id",
-			"relation",
-			"_user",
-			"user_type",
-			"condition_name",
-			"condition_context",
-			"ulid",
-			"inserted_at",
-		)
-
-	insertCount := 0
+		})
+	}
+
+	writeItems := make([][]interface{}, 0, len(writes))
 
 	// 5. For writes
 	// a. If on_duplicate: error ( default behavior )
@@ -776,8 +748,7 @@ func Write(
 			return err
 		}
 
-		insertCount++
-		insertBuilder = insertBuilder.Values(
+		writeItems = append(writeItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -788,8 +759,9 @@ func Write(
 			conditionContext,
 			id,
 			sq.Expr("NOW()"),
-		)
-		changelogBuilder = changelogBuilder.Values(
+		})
+
+		changeLogItems = append(changeLogItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -800,12 +772,19 @@ func Write(
 			openfgav1.TupleOperation_TUPLE_OPERATION_WRITE,
 			id,
 			sq.Expr("NOW()"),
-		)
+		})
 	}
 
-	if deleteCount > 0 {
-		res, err := deleteBuilder.
-			Where(deleteConditions).
+	for start, totalDeletes := 0, len(deleteConditions); start < totalDeletes; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalDeletes {
+			end = totalDeletes
+		}
+
+		deleteConditionsBatch := deleteConditions[start:end]
+
+		res, err := dbInfo.stbl.Delete("tuple").Where(sq.Eq{"store": store}).
+			Where(deleteConditionsBatch).
 			RunWith(txn). // Part of a txn.
 			ExecContext(ctx)
 		if err != nil {
@@ -817,15 +796,42 @@ func Write(
 			return dbInfo.HandleSQLError(err)
 		}
 
-		if rowsAffected != deleteCount {
+		if rowsAffected != int64(len(deleteConditionsBatch)) {
 			// If we deleted fewer rows than planned (after read before write), means we hit a race condition - someone else deleted the same row(s).
 			return storage.ErrWriteConflictOnDelete
 		}
 	}
 
-	if insertCount > 0 {
-		_, err = insertBuilder.RunWith(txn). // Part of a txn.
-							ExecContext(ctx)
+	for start, totalWrites := 0, len(writeItems); start < totalWrites; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalWrites {
+			end = totalWrites
+		}
+
+		writesBatch := writeItems[start:end]
+
+		insertBuilder := dbInfo.stbl.
+			Insert("tuple").
+			Columns(
+				"store",
+				"object_type",
+				"object_id",
+				"relation",
+				"_user",
+				"user_type",
+				"condition_name",
+				"condition_context",
+				"ulid",
+				"inserted_at",
+			)
+
+		for _, item := range writesBatch {
+			insertBuilder = insertBuilder.Values(item...)
+		}
+
+		_, err = insertBuilder.
+			RunWith(txn). // Part of a txn.
+			ExecContext(ctx)
 		if err != nil {
 			dberr := dbInfo.HandleSQLError(err)
 			if errors.Is(dberr, storage.ErrCollision) {
@@ -837,8 +843,34 @@ func Write(
 	}
 
 	// 6. Execute INSERT changelog statements
-	if deleteCount > 0 || insertCount > 0 {
-		_, err := changelogBuilder.RunWith(txn).ExecContext(ctx) // Part of a txn.
+	for start, totalItems := 0, len(changeLogItems); start < totalItems; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalItems {
+			end = totalItems
+		}
+
+		changeLogBatch := changeLogItems[start:end]
+
+		changelogBuilder := dbInfo.stbl.
+			Insert("changelog").
+			Columns(
+				"store",
+				"object_type",
+				"object_id",
+				"relation",
+				"_user",
+				"condition_name",
+				"condition_context",
+				"operation",
+				"ulid",
+				"inserted_at",
+			)
+
+		for _, item := range changeLogBatch {
+			changelogBuilder = changelogBuilder.Values(item...)
+		}
+
+		_, err = changelogBuilder.RunWith(txn).ExecContext(ctx) // Part of a txn.
 		if err != nil {
 			return dbInfo.HandleSQLError(err)
 		}
diff --git a/pkg/storage/sqlite/sqlite.go b/pkg/storage/sqlite/sqlite.go
index 64916697..2a939894 100644
--- a/pkg/storage/sqlite/sqlite.go
+++ b/pkg/storage/sqlite/sqlite.go
@@ -338,6 +338,33 @@ func buildRowConstructorIN(keys []tupleLockKey) (string, []interface{}) {
 	return sb.String(), args
 }
 
+// selectExistingRowsForWrite selects existing rows for the given keys and locks them FOR UPDATE.
+// The existing rows are added to the existing map.
+func (s *Datastore) selectExistingRowsForWrite(ctx context.Context, store string, keys []tupleLockKey, txn *sql.Tx, existing map[string]*openfgav1.Tuple) error {
+	inExpr, args := buildRowConstructorIN(keys)
+
+	selectBuilder := s.stbl.
+		Select(tupleColumns...).
+		Where(sq.Eq{"store": store}).
+		From("tuple").
+		// Row-constructor IN on full composite key for precise point locks.
+		Where(sq.Expr("(object_type, object_id, relation, user_object_type, user_object_id, user_relation, user_type) IN "+inExpr, args...)).
+		RunWith(txn) // make sure to run in the same transaction
+
+	iter := NewSQLTupleIterator(selectBuilder, HandleSQLError)
+	defer iter.Stop()
+
+	items, _, err := iter.ToArray(ctx, storage.PaginationOptions{PageSize: len(keys)})
+
+	if err != nil {
+		return err
+	}
+	for _, tuple := range items {
+		existing[tupleUtils.TupleKeyToString(tuple.GetKey())] = tuple
+	}
+	return nil
+}
+
 // Write provides the common method for writing to database across sql storage.
 func (s *Datastore) write(
 	ctx context.Context,
@@ -364,73 +391,33 @@ func (s *Datastore) write(
 	// 2. Compile a SELECT … FOR UPDATE statement to read the tuples for writes and lock tuples for deletes
 	// Build a deduped, sorted list of keys to lock.
 	lockKeys := makeTupleLockKeys(deletes, writes)
-	if len(lockKeys) == 0 {
+	total := len(lockKeys)
+	if total == 0 {
 		// Nothing to do.
 		return nil
 	}
 
-	existing := make(map[string]*openfgav1.Tuple, len(lockKeys))
+	existing := make(map[string]*openfgav1.Tuple, total)
 
 	// 3. If list compiled in step 2 is not empty, execute SELECT … FOR UPDATE statement
 
-	// Efficient, deterministic locking: probe existing rows using a row-constructor IN on the composite key.
-	// Batch to keep SQL size reasonable.
-	const lockBatchSize = 128
-
-	for i := 0; i < len(lockKeys); i += lockBatchSize {
-		j := i + lockBatchSize
-		if j > len(lockKeys) {
-			j = len(lockKeys)
+	for start := 0; start < total; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > total {
+			end = total
 		}
-		keys := lockKeys[i:j]
-
-		inExpr, args := buildRowConstructorIN(keys)
-
-		selectBuilder := s.stbl.
-			Select(tupleColumns...).
-			Where(sq.Eq{"store": store}).
-			From("tuple").
-			// Row-constructor IN on full composite key for precise point locks.
-			Where(sq.Expr("(object_type, object_id, relation, user_object_type, user_object_id, user_relation, user_type) IN "+inExpr, args...)).
-			RunWith(txn) // make sure to run in the same transaction
-
-		iter := NewSQLTupleIterator(selectBuilder, HandleSQLError)
-		defer iter.Stop()
+		keys := lockKeys[start:end]
 
-		items, _, err := iter.ToArray(ctx, storage.PaginationOptions{PageSize: len(keys)})
-
-		if err != nil {
+		if err = s.selectExistingRowsForWrite(ctx, store, keys, txn, existing); err != nil {
 			return err
 		}
-		for _, tuple := range items {
-			existing[tupleUtils.TupleKeyToString(tuple.GetKey())] = tuple
-		}
 	}
 
-	changelogBuilder := s.stbl.
-		Insert("changelog").
-		Columns(
-			"store",
-			"object_type",
-			"object_id",
-			"relation",
-			"user_object_type",
-			"user_object_id",
-			"user_relation",
-			"condition_name",
-			"condition_context",
-			"operation",
-			"ulid",
-			"inserted_at",
-		)
-
-	deleteBuilder := s.stbl.Delete("tuple").Where(sq.Eq{"store": store})
+	changeLogItems := make([][]interface{}, 0, len(deletes)+len(writes))
 
 	// ensures increasingly unique values within a single thread
 	entropy := ulid.DefaultEntropy()
 
-	var deleteCount int64
-
 	deleteConditions := sq.Or{}
 
 	// 4. For deletes
@@ -459,8 +446,6 @@ func (s *Datastore) write(
 			}
 		}
 
-		deleteCount++
-
 		id := ulid.MustNew(ulid.Timestamp(now), entropy).String()
 		objectType, objectID := tupleUtils.SplitObject(tk.GetObject())
 		userObjectType, userObjectID, userRelation := tupleUtils.ToUserParts(tk.GetUser())
@@ -475,7 +460,7 @@ func (s *Datastore) write(
 			"user_type":        tupleUtils.GetUserTypeFromUser(tk.GetUser()),
 		})
 
-		changelogBuilder = changelogBuilder.Values(
+		changeLogItems = append(changeLogItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -488,27 +473,10 @@ func (s *Datastore) write(
 			openfgav1.TupleOperation_TUPLE_OPERATION_DELETE,
 			id,
 			sq.Expr("datetime('subsec')"),
-		)
+		})
 	}
 
-	insertBuilder := s.stbl.
-		Insert("tuple").
-		Columns(
-			"store",
-			"object_type",
-			"object_id",
-			"relation",
-			"user_object_type",
-			"user_object_id",
-			"user_relation",
-			"user_type",
-			"condition_name",
-			"condition_context",
-			"ulid",
-			"inserted_at",
-		)
-
-	insertCount := 0
+	writeItems := make([][]interface{}, 0, len(writes))
 
 	// 5. For writes
 	// a. If on_duplicate: error ( default behavior )
@@ -551,8 +519,7 @@ func (s *Datastore) write(
 			return err
 		}
 
-		insertCount++
-		insertBuilder = insertBuilder.Values(
+		writeItems = append(writeItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -565,8 +532,9 @@ func (s *Datastore) write(
 			conditionContext,
 			id,
 			sq.Expr("datetime('subsec')"),
-		)
-		changelogBuilder = changelogBuilder.Values(
+		})
+
+		changeLogItems = append(changeLogItems, []interface{}{
 			store,
 			objectType,
 			objectID,
@@ -579,37 +547,68 @@ func (s *Datastore) write(
 			openfgav1.TupleOperation_TUPLE_OPERATION_WRITE,
 			id,
 			sq.Expr("datetime('subsec')"),
-		)
+		})
 	}
 
-	if deleteCount > 0 {
-		var res sql.Result
+	for start, totalDeletes := 0, len(deleteConditions); start < totalDeletes; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalDeletes {
+			end = totalDeletes
+		}
 
-		err = busyRetry(func() error {
-			res, err = deleteBuilder.
-				Where(deleteConditions).
-				RunWith(txn). // Part of a txn.
-				ExecContext(ctx)
-			return err
-		})
+		deleteConditionsBatch := deleteConditions[start:end]
+
+		res, err := s.stbl.Delete("tuple").Where(sq.Eq{"store": store}).
+			Where(deleteConditionsBatch).
+			RunWith(txn). // Part of a txn.
+			ExecContext(ctx)
+		if err != nil {
+			return HandleSQLError(err)
+		}
 
 		rowsAffected, err := res.RowsAffected()
 		if err != nil {
 			return HandleSQLError(err)
 		}
 
-		if rowsAffected != deleteCount {
+		if rowsAffected != int64(len(deleteConditionsBatch)) {
 			// If we deleted fewer rows than planned (after read before write), means we hit a race condition - someone else deleted the same row(s).
 			return storage.ErrWriteConflictOnDelete
 		}
 	}
 
-	if insertCount > 0 {
-		err = busyRetry(func() error {
-			_, err = insertBuilder.RunWith(txn). // Part of a txn.
-								ExecContext(ctx)
-			return err
-		})
+	for start, totalWrites := 0, len(writeItems); start < totalWrites; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalWrites {
+			end = totalWrites
+		}
+
+		writesBatch := writeItems[start:end]
+
+		insertBuilder := s.stbl.
+			Insert("tuple").
+			Columns(
+				"store",
+				"object_type",
+				"object_id",
+				"relation",
+				"user_object_type",
+				"user_object_id",
+				"user_relation",
+				"user_type",
+				"condition_name",
+				"condition_context",
+				"ulid",
+				"inserted_at",
+			)
+
+		for _, item := range writesBatch {
+			insertBuilder = insertBuilder.Values(item...)
+		}
+
+		_, err = insertBuilder.
+			RunWith(txn). // Part of a txn.
+			ExecContext(ctx)
 		if err != nil {
 			dberr := HandleSQLError(err)
 			if errors.Is(dberr, storage.ErrCollision) {
@@ -621,11 +620,36 @@ func (s *Datastore) write(
 	}
 
 	// 6. Execute INSERT changelog statements
-	if deleteCount > 0 || insertCount > 0 {
-		err := busyRetry(func() error {
-			_, err := changelogBuilder.RunWith(txn).ExecContext(ctx) // Part of a txn.
-			return err
-		})
+	for start, totalItems := 0, len(changeLogItems); start < totalItems; start += storage.DefaultMaxTuplesPerWrite {
+		end := start + storage.DefaultMaxTuplesPerWrite
+		if end > totalItems {
+			end = totalItems
+		}
+
+		changeLogBatch := changeLogItems[start:end]
+
+		changelogBuilder := s.stbl.
+			Insert("changelog").
+			Columns(
+				"store",
+				"object_type",
+				"object_id",
+				"relation",
+				"user_object_type",
+				"user_object_id",
+				"user_relation",
+				"condition_name",
+				"condition_context",
+				"operation",
+				"ulid",
+				"inserted_at",
+			)
+
+		for _, item := range changeLogBatch {
+			changelogBuilder = changelogBuilder.Values(item...)
+		}
+
+		_, err = changelogBuilder.RunWith(txn).ExecContext(ctx) // Part of a txn.
 		if err != nil {
 			return HandleSQLError(err)
 		}



Test patch:
diff --git a/pkg/storage/mysql/mysql_test.go b/pkg/storage/mysql/mysql_test.go
index 6d494790..94b2e2fe 100644
--- a/pkg/storage/mysql/mysql_test.go
+++ b/pkg/storage/mysql/mysql_test.go
@@ -34,6 +34,15 @@ func TestMySQLDatastore(t *testing.T) {
 	defer ds.Close()
 
 	test.RunAllTests(t, ds)
+
+	// Run tests with a custom large max_tuples_per_write value.
+	dsCustom, err := New(uri, sqlcommon.NewConfig(
+		sqlcommon.WithMaxTuplesPerWrite(5000),
+	))
+	require.NoError(t, err)
+	defer dsCustom.Close()
+
+	t.Run("WriteTuplesWithMaxTuplesPerWrite", test.WriteTuplesWithMaxTuplesPerWrite(dsCustom, context.Background()))
 }
 
 func TestMySQLDatastoreAfterCloseIsNotReady(t *testing.T) {
diff --git a/pkg/storage/postgres/postgres_test.go b/pkg/storage/postgres/postgres_test.go
index 239196e0..f6cbe33e 100644
--- a/pkg/storage/postgres/postgres_test.go
+++ b/pkg/storage/postgres/postgres_test.go
@@ -31,6 +31,15 @@ func TestPostgresDatastore(t *testing.T) {
 	require.NoError(t, err)
 	defer ds.Close()
 	test.RunAllTests(t, ds)
+
+	// Run tests with a custom large max_tuples_per_write value.
+	dsCustom, err := New(uri, sqlcommon.NewConfig(
+		sqlcommon.WithMaxTuplesPerWrite(5000),
+	))
+	require.NoError(t, err)
+	defer dsCustom.Close()
+
+	t.Run("WriteTuplesWithMaxTuplesPerWrite", test.WriteTuplesWithMaxTuplesPerWrite(dsCustom, context.Background()))
 }
 
 func TestPostgresDatastoreStartup(t *testing.T) {
diff --git a/pkg/storage/sqlite/sqlite_test.go b/pkg/storage/sqlite/sqlite_test.go
index d69ac655..3ab0e789 100644
--- a/pkg/storage/sqlite/sqlite_test.go
+++ b/pkg/storage/sqlite/sqlite_test.go
@@ -25,6 +25,15 @@ func TestSQLiteDatastore(t *testing.T) {
 	require.NoError(t, err)
 	defer ds.Close()
 	test.RunAllTests(t, ds)
+
+	// Run tests with a custom large max_tuples_per_write value.
+	dsCustom, err := New(uri, sqlcommon.NewConfig(
+		sqlcommon.WithMaxTuplesPerWrite(5000),
+	))
+	require.NoError(t, err)
+	defer dsCustom.Close()
+
+	t.Run("WriteTuplesWithMaxTuplesPerWrite", test.WriteTuplesWithMaxTuplesPerWrite(dsCustom, context.Background()))
 }
 
 func TestSQLiteDatastoreAfterCloseIsNotReady(t *testing.T) {
diff --git a/pkg/storage/test/tuples.go b/pkg/storage/test/tuples.go
index d1b5a678..365679e8 100644
--- a/pkg/storage/test/tuples.go
+++ b/pkg/storage/test/tuples.go
@@ -843,43 +843,7 @@ func TupleWritingAndReadingTest(t *testing.T, datastore storage.OpenFGADatastore
 		assert.Equalf(t, tk2.String(), tk2fromDB.GetKey().String(), "expected %s, got %s", tk2.String(), tk2fromDB.String())
 	})
 
-	t.Run("write_and_delete_many_tuples", func(t *testing.T) {
-		numTuples := 40
-
-		storeID := ulid.Make().String()
-		// Write a lot of tuples to both stores to ensure that the delete logic is not
-		// affected by the number of tuples in the database.
-		tuplesToWrite := make([]*openfgav1.TupleKey, numTuples)
-		for i := 0; i < numTuples; i++ {
-			tuplesToWrite[i] = &openfgav1.TupleKey{
-				Object:   fmt.Sprintf("doc:readme%d", i),
-				Relation: "owner",
-				User:     fmt.Sprintf("user:%d", i),
-			}
-		}
-		require.NoError(t, datastore.Write(ctx, storeID, nil, tuplesToWrite))
-		require.NoError(t, datastore.Write(ctx, storeID, nil, tuplesToWrite, storage.WithOnDuplicateInsert(storage.OnDuplicateInsertIgnore)))
-
-		tk1, err := datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[1], storage.ReadUserTupleOptions{})
-		require.NoError(t, err)
-		require.Equal(t, tk1.GetKey().String(), tuplesToWrite[1].String())
-		tkLast, err := datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[len(tuplesToWrite)-1], storage.ReadUserTupleOptions{})
-		require.NoError(t, err)
-		require.Equal(t, tkLast.GetKey().String(), tuplesToWrite[len(tuplesToWrite)-1].String())
-
-		tuplesToDelete := make([]*openfgav1.TupleKeyWithoutCondition, numTuples)
-		for i := 0; i < numTuples; i++ {
-			tuplesToDelete[i] = tuple.TupleKeyToTupleKeyWithoutCondition(tuplesToWrite[i])
-		}
-		require.NoError(t, datastore.Write(ctx, storeID, tuplesToDelete, nil))
-		require.NoError(t, datastore.Write(ctx, storeID, tuplesToDelete, nil, storage.WithOnMissingDelete(storage.OnMissingDeleteIgnore)))
-
-		// Ensure all tuple deleted
-		_, err = datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[1], storage.ReadUserTupleOptions{})
-		require.ErrorIs(t, err, storage.ErrNotFound)
-		_, err = datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[len(tuplesToWrite)-1], storage.ReadUserTupleOptions{})
-		require.ErrorIs(t, err, storage.ErrNotFound)
-	})
+	t.Run("write_and_delete_many_tuples", WriteTuplesWithMaxTuplesPerWrite(datastore, ctx))
 
 	t.Run("deleting_a_tuple_which_exists_succeeds", func(t *testing.T) {
 		storeID := ulid.Make().String()
@@ -1820,6 +1784,46 @@ func TupleWritingAndReadingTest(t *testing.T, datastore storage.OpenFGADatastore
 	})
 }
 
+func WriteTuplesWithMaxTuplesPerWrite(datastore storage.OpenFGADatastore, ctx context.Context) func(t *testing.T) {
+	return func(t *testing.T) {
+		numTuples := datastore.MaxTuplesPerWrite()
+
+		storeID := ulid.Make().String()
+		// Write a lot of tuples to both stores to ensure that the delete logic is not
+		// affected by the number of tuples in the database.
+		tuplesToWrite := make([]*openfgav1.TupleKey, numTuples)
+		for i := 0; i < numTuples; i++ {
+			tuplesToWrite[i] = &openfgav1.TupleKey{
+				Object:   fmt.Sprintf("doc:readme%d", i),
+				Relation: "owner",
+				User:     fmt.Sprintf("user:%d", i),
+			}
+		}
+		require.NoError(t, datastore.Write(ctx, storeID, nil, tuplesToWrite))
+		require.NoError(t, datastore.Write(ctx, storeID, nil, tuplesToWrite, storage.WithOnDuplicateInsert(storage.OnDuplicateInsertIgnore)))
+
+		tk1, err := datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[1], storage.ReadUserTupleOptions{})
+		require.NoError(t, err)
+		require.Equal(t, tk1.GetKey().String(), tuplesToWrite[1].String())
+		tkLast, err := datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[len(tuplesToWrite)-1], storage.ReadUserTupleOptions{})
+		require.NoError(t, err)
+		require.Equal(t, tkLast.GetKey().String(), tuplesToWrite[len(tuplesToWrite)-1].String())
+
+		tuplesToDelete := make([]*openfgav1.TupleKeyWithoutCondition, numTuples)
+		for i := 0; i < numTuples; i++ {
+			tuplesToDelete[i] = tuple.TupleKeyToTupleKeyWithoutCondition(tuplesToWrite[i])
+		}
+		require.NoError(t, datastore.Write(ctx, storeID, tuplesToDelete, nil))
+		require.NoError(t, datastore.Write(ctx, storeID, tuplesToDelete, nil, storage.WithOnMissingDelete(storage.OnMissingDeleteIgnore)))
+
+		// Ensure all tuple deleted
+		_, err = datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[1], storage.ReadUserTupleOptions{})
+		require.ErrorIs(t, err, storage.ErrNotFound)
+		_, err = datastore.ReadUserTuple(ctx, storeID, tuplesToWrite[len(tuplesToWrite)-1], storage.ReadUserTupleOptions{})
+		require.ErrorIs(t, err, storage.ErrNotFound)
+	}
+}
+
 func ReadStartingWithUserTest(t *testing.T, datastore storage.OpenFGADatastore) {
 	ctx := context.Background()