aweisberg commented on code in PR #3694:
URL: https://github.com/apache/cassandra/pull/3694#discussion_r1868310901
##########
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java:
##########
@@ -290,6 +291,66 @@ private AsyncChain<Data> readChains()
}));
});
});
+ return results;
+ }
+
+ private List<AsyncChain<Data>> rangeReadChains(long nowInSeconds,
Dispatcher.RequestTime requestTime)
+ {
+ TxnRangeRead read = (TxnRangeRead) txn.read();
+ Seekables<?, ?> keys = txn.read().keys();
+ List<AsyncChain<Data>> results = new ArrayList<>();
+ keys.forEach(key -> {
+ TokenRange range = (TokenRange)key;
+ PartitionRangeReadCommand command = read.commandForSubrange(range,
nowInSeconds);
+
+ // This should only rarely occur when coordinators start a
transaction in a migrating range
+ // because they haven't yet updated their cluster metadata.
+ // It would be harmless to do the read, because it will be
rejected in `TxnQuery` anyways,
+ // but it's faster to skip the read
+ // TODO (required): To make migration work we need to validate
that the range is all on Accord
+ // if any part isn't we should reject the read
+// TableMigrationState tms =
ConsensusTableMigration.getTableMigrationState(command.metadata().id);
+// AccordClientRequestMetrics metrics = txn.kind().isWrite() ?
accordWriteMetrics : accordReadMetrics;
+// if
(ConsensusRequestRouter.instance.isKeyInMigratingOrMigratedRangeFromAccord(command.metadata(),
tms, command.partitionKey()))
+// {
+// metrics.migrationSkippedReads.mark();
+//
results.add(AsyncChains.success(TxnData.emptyPartition(fragment.txnDataName(),
command)));
+// return;
+// }
+
+
results.add(AsyncChains.ofCallable(Stage.ACCORD_MIGRATION.executor(), () -> {
+ TxnData result = new TxnData();
+ try (PartitionIterator iterator =
StorageProxy.getRangeSlice(command, consistencyLevel, this, requestTime))
+ {
+ TxnDataRangeValue value = new TxnDataRangeValue();
+ while (iterator.hasNext())
+ {
+ try (RowIterator partition = iterator.next())
+ {
+ FilteredPartition filtered =
FilteredPartition.create(partition);
+ if (filtered.hasRows() ||
command.selectsFullPartition())
+ value.add(filtered);
+ }
+ }
+ result.put(TxnData.txnDataName(TxnDataNameKind.USER),
value);
Review Comment:
Interop execution is not limited to non-`TransactionStatement` SQL. The
purpose of interop execution is to allow reading at quorum (and read repair)
and synchronous commit (at a specified consistency level) which also applies to
transaction statement in modes like `mixed_reads` and during migration.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
