在ScalarDB中复制时PutIfNotExists未抛出错误

Question

我使用的是在Cassandra之上提供ACID功能的scalardb。我想对一个方法(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()

Answer 1

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