PutIfNotExists在ScalarDB中不会因复制而引发错误

问题描述

我正在使用scalardb，它在Cassandra之上提供了ACID功能。我要对方法(add)进行单元测试，如果方法不是重复的，则应添加新行。

def add(transaction:distributedTransaction,answer:AnswerOfAPracticeQuestion,mutationCondition:MutationCondition = new PutIfNotExists()) = {
    val pAnswerKey = new Key(new TextValue("answered_by_user",answer.answeredBy.get.answerer_id.toString),new TextValue("question_id",answer.question_id.toString))
//I want to check if partition key exists,not clustering column. In future,I might allow multiple rows but for Now I don't need duplicates
//    val cAnswerKey = new Key(new TextValue("answer_id",answer.answer_id.toString))

    //logger.trace(s"created keys. ${pAnswerKey},${cAnswerKey}")
    val imageData = answer.image.map(imageList=>imageList).getorElse(List())
    logger.trace(s"will check in ${keyspaceName},${tablename}")
    val putAnswer: Put = new Put(pAnswerKey)
      .forNamespace(keyspaceName)
      .forTable(tablename)
      .withCondition(mutationCondition)
      .withValue...)

    logger.trace(s"putting answer ${putAnswer}")
    //checktest-add answer to respository
    transaction.put(putAnswer)
  }

在库的put方法中，我可以指定MutationCondition以避免重复输入。例如

val putAnswer: Put = new Put(pAnswerKey)
      .forNamespace(keyspaceName)
      .forTable(tablename)
      .withCondition(mutationCondition)
      .withValue(...)

我正在使用的MutationCondition的值为PutIfNotExists

但是当我在测试用例中使用重复的分区键调用put时，会出错。

preparing records Failed
com.scalar.db.exception.transaction.CommitException: preparing records Failed
Caused by: java.lang.IllegalArgumentException: The primary key is not properly specified.
    at com.scalar.db.storage.cassandra.Cassandra.throwIfNotMatched(Cassandra.java:201)
at com.scalar.db.storage.cassandra.Cassandra.checkIfPrimaryKeyExists(Cassandra.java:194)

问题1）是因为我没有使用clustering columns吗？

问题2）是否有一种方法可以使PutIfNotExists仅与partition key一起使用，还是我需要自己明确使用scan？

问题3）如果我也使用聚类列，则会得到异常com.datastax.driver.core.exceptions.InvalidQueryException: The column names contains duplicates。只是为了确认，这是否是重复的预期例外？

表架构为

CREATE TABLE codingjedi.answer_by_user_id_and_question_id (
answered_by_user text,question_id text,answer_id text,answer text,creation_month bigint,creation_year bigint,image text,notes text,PRIMARY KEY ((answered_by_user,question_id),answer_id)

测试用例是

"not add answer to respository if duplicate" in {

  val key1 = repoTestEnv.answerTestEnv.answerOfAPracticeQuestion.question_id
  val key2 = repoTestEnv.answerTestEnv.answerOfAPracticeQuestion.answeredBy.get.answerer_id
  val key3 = repoTestEnv.answerTestEnv.answerOfAPracticeQuestion.answer_id.get

  logger.trace(s"will use keys ${key1},${key2},${key3}")
  val insertStatement =
    s"""
       | INSERT INTO answer_by_user_id_and_question_id (question_id,answered_by_user,answer_id,answer,notes,image,creation_year,creation_month) VALUES
       | ('${key1}',| '${key2}',| '${key3}',| '{"answer":[{"filename":"some filename","answer":"some answer"}]}',| 'some notes',| '{"image":["some image data"]}',| ${repoTestEnv.year},| ${repoTestEnv.month})
    """.stripMargin

  repoTestEnv.executeStatements(CqlDataSet.ofStrings(insertStatement))


  //val keys = AnswerKeys(repoTestEnv.testEnv.mockHelperMethods.getUniqueID(),repoTestEnv.testEnv.mockHelperMethods.getUniqueID(),Some(repoTestEnv.testEnv.mockHelperMethods.getUniqueID()))

  val cassandraConnectionService = CassandraConnectionManagementService()
  //val (cassandraSession,cluster) = cassandraConnectionService.connectWithCassandra("cassandra://localhost:9042/codingjedi","codingJediCluster")
  val transactionService = cassandraConnectionService.connectWithCassandraWithTransactionSupport("localhost","9042","codingJediCluster"/*,dbUsername,dbPassword*/)
  //TodoM - pick the database and keyspace names from config file.
  //cassandraConnectionService.initKeySpace(cassandraSession.get,"codingjedi")
  logger.trace(s"created transaction service ${transactionService}")
  val distributedTransaction = transactionService.get.start()

  val repository = new AnswersTransactionRepository("codingjedi","answer_by_user_id_and_question_id")
  //logger.trace(s"will query using key ${keys}")

  val answer = repoTestEnv.answerTestEnv.answerOfAPracticeQuestion
  logger.trace(s"will add answer ${answer}")

  val imageData = answer.image.map(imageList=>imageList).getorElse(List())

  repository.add(distributedTransaction,repoTestEnv.answerTestEnv.answerOfAPracticeQuestion)
  distributedTransaction.commit()

解决方法

它正在检查是否存在重复的主键，而不是分区键。