[PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg opened a new pull request, #3463:
URL: https://github.com/apache/cassandra/pull/3463

   (no comment)


-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838578961


##########
src/java/org/apache/cassandra/service/accord/txn/AbstractKeySorted.java:
##########
@@ -55,16 +54,39 @@ public AbstractKeySorted(List<T> items)
         items.toArray(arr);
         Arrays.sort(arr, this::compare);
         this.items = arr;
-        validateOrder();
-        this.itemKeys = extractItemKeys();
+        int duplicates = validateOrder();
+        this.itemKeys = extractItemKeys(duplicates);
     }
 
-    private Keys extractItemKeys()
+    private Keys extractItemKeys(int duplicates)
     {
-        Key[] keys = new Key[size()];
-        for (int i = 0 ; i < keys.length ; ++i)
-            keys[i] = getKey(items[i]);
-        return Keys.ofSorted(keys);
+        PartitionKey[] partitionKeys = new PartitionKey[items.length - duplicates];

Review Comment:
   Fixed by making `AbstractKeysSorted` extend `AbstractList` so that the transforming version of `Keys.of` can be used.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837923047


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -60,60 +64,58 @@
 
 public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
 {
-    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], null);
     private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
 
-    public static final String CAS_READ_NAME = "CAS_READ";
-    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
-
-    @Nonnull
-    private final Keys txnKeys;
-
     // Cassandra's consistency level used by Accord to safely read data written outside of Accord
     @Nullable
     private final ConsistencyLevel cassandraConsistencyLevel;
 
-    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel consistencyLevel)
     {
-        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+        items.sort(Comparator.comparing(TxnNamedRead::key));
+        return new TxnKeyRead(items, consistencyLevel);
     }
 
     public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
     {
         List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
         for (int i = 0; i < readCommands.size(); i++)
-            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
-        Keys keys = Keys.of(reads, TxnNamedRead::key);
-        return new TxnKeyRead(reads, keys, consistencyLevel);
+            reads.add(new TxnNamedRead(txnDataName(USER, i), readCommands.get(i)));
+        reads.sort(Comparator.comparing(TxnNamedRead::key));

Review Comment:
   does this dynamically allocate a lambda comparator? I'm never sure, and usually prefer to explicitly specify one so as to guarantee it is statically allocated



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833022022


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   It's C&P from `TxnKeyRead` which added it [here|https://github.com/apache/cassandra/commit/8deb6639f6477a5c61bcb7422f2801fd19b16331#diff-f48c552114deecaa1cf2678dd9085170b8a37da3d867ea734b06b97639cb3a90R207] and I don't recall why I thought it was necessary at the time.
   
   Maybe once simulator tests on Accord pass we can try removing that and see if it actually did anything?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833358414


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   Oh this is so weird. There are two sets of keys here. One that is `AbstractKeySorted` and another that is in `TxnKeyRead`. One of them is all the keys for the txn, the other is all the keys extracted from the actual reads. I don't get why we would want the read to include all the keys for the entire transaction? 
   
   Going to try eliminating that and see what happens...



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837902256


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;
+    private static final int TXN_DATA_NAME_INDEX_MASK = ~(~0 << TXN_DATA_NAME_INDEX_BITS);
+    public static final int TXN_DATA_NAME_INDEX_MAX = ((1 << TXN_DATA_NAME_INDEX_BITS) - 1);
+
+    public enum TxnDataNameKind
+    {
+        USER((byte) 0),
+        RETURNING((byte) 1),
+        AUTO_READ((byte) 2),
+        CAS_READ((byte) 3);
+
+        private final byte value;
+
+        TxnDataNameKind(byte value)
+        {
+            this.value = value;
+        }
+
+        public static TxnDataNameKind from(byte b)
+        {
+            switch (b)
+            {
+                case 0:
+                    return USER;
+                case 1:
+                    return RETURNING;
+                case 2:
+                    return AUTO_READ;
+                case 3:
+                    return CAS_READ;
+                default:
+                    throw new IllegalArgumentException("Unknown kind: " + b);
+            }
+        }
+    }
+
+    public static int txnDataName(TxnDataNameKind kind, int index)
+    {
+        checkArgument(index >= 0 && index <= TXN_DATA_NAME_INDEX_MAX);
+        int kindInt = (int)(((long)kind.value) << TXN_DATA_NAME_INDEX_BITS);
+        return kindInt | index;
+    }
+
+    public static int txnDataName(TxnDataNameKind kind)
+    {
+        return txnDataName(kind, 0);
+    }
+
+    public static TxnDataNameKind txnDataNameKind(int txnDataName)
+    {
+        int kind = txnDataName >> TXN_DATA_NAME_INDEX_BITS;

Review Comment:
   I know we don't go negative, but for bit fiddling I find it clearer to stick to unsigned shift right



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832598245


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -20,78 +20,59 @@
 
 import java.io.IOException;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
-import java.util.Set;
-
-import com.google.common.collect.Maps;
 
 import accord.api.Data;
 import org.apache.cassandra.db.EmptyIterators;
 import org.apache.cassandra.db.SinglePartitionReadCommand;
-import org.apache.cassandra.db.TypeSizes;
-import org.apache.cassandra.db.filter.ColumnFilter;
-import org.apache.cassandra.db.partitions.FilteredPartition;
 import org.apache.cassandra.db.partitions.PartitionIterators;
-import org.apache.cassandra.db.rows.DeserializationHelper;
-import org.apache.cassandra.db.rows.Row;
-import org.apache.cassandra.db.rows.UnfilteredRowIterator;
-import org.apache.cassandra.db.rows.UnfilteredRowIteratorSerializer;
-import org.apache.cassandra.db.rows.UnfilteredRowIterators;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.schema.TableId;
-import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.CollectionSerializers;
 import org.apache.cassandra.utils.NullableSerializer;
 import org.apache.cassandra.utils.ObjectSizes;
 
+import static accord.utils.Invariants.checkArgument;
 import static org.apache.cassandra.service.accord.txn.TxnResult.Kind.txn_data;
 
-public class TxnData extends TxnResult implements Data, Iterable<FilteredPartition>
+/**
+ * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
+ * when read results are returned to the coordinator to compute query results and writes.
+ */
+public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data

Review Comment:
   perhaps extend an open hash map like agrona for (hopefully) improved space efficiency? If we update TxnDataName to be represented by an integer this is even better.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833251881


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   > Can we do the switch from string to in there?
   
   Yep, absolutely
   
   > Keys I think only contains PartitionKey with a TableId not a TableMetadata.
   
   I'm fairly sure the deserialiser is looking up TableMetadata for each key. It's more the lookup that bothers me, it's actually quite expensive - though perhaps we should optimise it otherwise.
   



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838348951


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -60,60 +64,58 @@
 
 public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
 {
-    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], null);
     private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
 
-    public static final String CAS_READ_NAME = "CAS_READ";
-    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
-
-    @Nonnull
-    private final Keys txnKeys;
-
     // Cassandra's consistency level used by Accord to safely read data written outside of Accord
     @Nullable
     private final ConsistencyLevel cassandraConsistencyLevel;
 
-    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel consistencyLevel)
     {
-        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+        items.sort(Comparator.comparing(TxnNamedRead::key));
+        return new TxnKeyRead(items, consistencyLevel);
     }
 
     public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
     {
         List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
         for (int i = 0; i < readCommands.size(); i++)
-            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
-        Keys keys = Keys.of(reads, TxnNamedRead::key);
-        return new TxnKeyRead(reads, keys, consistencyLevel);
+            reads.add(new TxnNamedRead(txnDataName(USER, i), readCommands.get(i)));
+        reads.sort(Comparator.comparing(TxnNamedRead::key));

Review Comment:
   by static, I mean define it inline without capturing any arguments - in this case we're passing a statically allocated lambda to a dynamically allocated lambda. It's possible escape analysis prevents the allocation, but if we just define a statically allocated lambda it's all fine and basically just as neat.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832577607


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   If we're looking at these serialisers and classes, it would be nice also to remove the String[] from `TxnDataName` and use a simple integer before we ossify the serialization format, if you don't mind including it in scope?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833011536


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);

Review Comment:
   Oops yeah when I pushed the branch after rebasing I didn't include a fixup patch.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833252163


##########
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java:
##########
@@ -277,19 +276,81 @@ private AsyncChain<Data> readChains()
                     try (PartitionIterator iterator = StorageProxy.readRegular(group, consistencyLevel, this, requestTime))
                     {
                         if (iterator.hasNext())
+                        {
+                            try (RowIterator partition = iterator.next())
+                            {
+                                TxnDataKeyValue value = new TxnDataKeyValue(partition);
+                                if (value.hasRows() || command.selectsFullPartition())
+                                    result.put(fragment.txnDataName(), value);
+                            }
+                        }
+                    }
+                    return result;
+                }));
+            });
+        });
+        return results;
+    }
+
+    private List<AsyncChain<Data>> rangeReadChains(int nowInSeconds, Dispatcher.RequestTime requestTime)
+    {
+        TxnKeyRead read = (TxnKeyRead) txn.read();
+        Seekables<?, ?> keys = txn.read().keys();
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        keys.forEach(key -> {
+            read.forEachWithKey((PartitionKey) key, fragment -> {

Review Comment:
   Sounds good



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838348951


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -60,60 +64,58 @@
 
 public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
 {
-    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], null);
     private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
 
-    public static final String CAS_READ_NAME = "CAS_READ";
-    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
-
-    @Nonnull
-    private final Keys txnKeys;
-
     // Cassandra's consistency level used by Accord to safely read data written outside of Accord
     @Nullable
     private final ConsistencyLevel cassandraConsistencyLevel;
 
-    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel consistencyLevel)
     {
-        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+        items.sort(Comparator.comparing(TxnNamedRead::key));
+        return new TxnKeyRead(items, consistencyLevel);
     }
 
     public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
     {
         List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
         for (int i = 0; i < readCommands.size(); i++)
-            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
-        Keys keys = Keys.of(reads, TxnNamedRead::key);
-        return new TxnKeyRead(reads, keys, consistencyLevel);
+            reads.add(new TxnNamedRead(txnDataName(USER, i), readCommands.get(i)));
+        reads.sort(Comparator.comparing(TxnNamedRead::key));

Review Comment:
   by static, I mean define it inline without capturing any arguments - in this case we're passing a statically allocated lambda to a dynamically allocated lambda. It's possibly escape analysis prevents the allocation, but if we just define a statically allocated lambda it's all fine and basically just as neat.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832574794


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   can we pass the Key through to the deserialiser, so it doesn't need to find the `TableMetadata` again?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833345334


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   Where do you see that? I looked at the `PartitionKey` deserializer and it does `TableId.deserialize(in).intern();`, `TableId.deserialize` and the constructor don't do anything special.
   
   Intern does
   ```
       public TableId intern()
       {
           TableId interned = internCache.putIfAbsent(this, this);
           return interned == null ? this : interned;
       }
   ```
   
   I think it would actually easier to construct the `Keys` after deserializing the reads which should contain all the Keys that are in `Keys` in the first place. That is how we construct the `Keys` in the first place.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838345982


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -60,60 +64,58 @@
 
 public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
 {
-    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], null);
     private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
 
-    public static final String CAS_READ_NAME = "CAS_READ";
-    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
-
-    @Nonnull
-    private final Keys txnKeys;
-
     // Cassandra's consistency level used by Accord to safely read data written outside of Accord
     @Nullable
     private final ConsistencyLevel cassandraConsistencyLevel;
 
-    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    private TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel cassandraConsistencyLevel)
     {
         super(items);
+        checkNotNull(items, "items is null");
         checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
-        this.txnKeys = txnKeys;
         this.cassandraConsistencyLevel = cassandraConsistencyLevel;
     }
 
-    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nullable ConsistencyLevel consistencyLevel)
     {
-        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+        items.sort(Comparator.comparing(TxnNamedRead::key));
+        return new TxnKeyRead(items, consistencyLevel);
     }
 
     public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
     {
         List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
         for (int i = 0; i < readCommands.size(); i++)
-            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
-        Keys keys = Keys.of(reads, TxnNamedRead::key);
-        return new TxnKeyRead(reads, keys, consistencyLevel);
+            reads.add(new TxnNamedRead(txnDataName(USER, i), readCommands.get(i)));
+        reads.sort(Comparator.comparing(TxnNamedRead::key));

Review Comment:
   I can make it static. I don't know if allocates.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837881736


##########
src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java:
##########
@@ -244,11 +247,11 @@ List<TxnNamedRead> createNamedReads(NamedSelect namedSelect, QueryOptions option
 
         List<TxnNamedRead> list = new ArrayList<>(selectQuery.queries.size());
         for (int i = 0; i < selectQuery.queries.size(); i++)
-            list.add(new TxnNamedRead(TxnDataName.returning(i), selectQuery.queries.get(i)));
+            list.add(new TxnNamedRead(txnDataName(RETURNING, i), selectQuery.queries.get(i)));
         return list;
     }
 
-    private List<TxnNamedRead> createNamedReads(QueryOptions options, ClientState state, Map<TxnDataName, NamedSelect> autoReads, Consumer<Key> keyConsumer)
+    private List<TxnNamedRead> createNamedReads(QueryOptions options, ClientState state, Map<Integer, NamedSelect> autoReads, Consumer<Key> keyConsumer)

Review Comment:
   nit: maybe `Int2ObjectHashMap`? Not super important



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838565234


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;

Review Comment:
   This didn't need to count very high so I gave a few extra bits to support more kinds of data names in the future, but it's really not that necessary or helpful.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837892294


##########
src/java/org/apache/cassandra/service/accord/txn/AbstractKeySorted.java:
##########
@@ -55,16 +54,39 @@ public AbstractKeySorted(List<T> items)
         items.toArray(arr);
         Arrays.sort(arr, this::compare);
         this.items = arr;
-        validateOrder();
-        this.itemKeys = extractItemKeys();
+        int duplicates = validateOrder();
+        this.itemKeys = extractItemKeys(duplicates);
     }
 
-    private Keys extractItemKeys()
+    private Keys extractItemKeys(int duplicates)
     {
-        Key[] keys = new Key[size()];
-        for (int i = 0 ; i < keys.length ; ++i)
-            keys[i] = getKey(items[i]);
-        return Keys.ofSorted(keys);
+        PartitionKey[] partitionKeys = new PartitionKey[items.length - duplicates];

Review Comment:
   given the duplicate case is likely to be uncommon, we could simplify this and just call `Keys.ofSorted`? I don't have a super strong opinion though, this is totally fine, but it is a shame to have dedicated de-dup code. We could also introduce a version of `Keys.ofSorted()` that accepts a size, the getter function and the input array to transform?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832669746


##########
src/java/org/apache/cassandra/service/StorageProxy.java:
##########
@@ -2141,28 +2157,96 @@ private static PartitionIterator readWithConsensus(SinglePartitionReadCommand.Gr
         return lastResult.serialReadResult;
     }
 
-    private static ConsensusAttemptResult readWithAccord(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    private static ConsistencyLevel consistencyLevelForAccordRead(ClusterMetadata cm, SinglePartitionReadCommand.Group group, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        // Null means no specific consistency behavior is required from Accord, it's functionally similar to
+        // reading at ONE if you are reading data that wasn't written via Accord
+        if (consistencyLevel == null)
+            return null;
+
+        TableId tableId = group.queries.get(0).metadata().id;
+        TableParams tableParams = getTableMetadata(cm, tableId).params;
+        TransactionalMode mode = tableParams.transactionalMode;
+        TransactionalMigrationFromMode migrationFromMode = tableParams.transactionalMigrationFrom;
+        for (SinglePartitionReadCommand command : group.queries)
+        {
+            // readCLForStrategy should return either null or the supplied consistency level
+            // in which case we will read everything at that CL since Accord doesn't support per table
+            // read consistency
+            ConsistencyLevel commitCL = mode.readCLForStrategy(migrationFromMode, consistencyLevel, cm, tableId, command.partitionKey().getToken());
+            if (commitCL != null)
+                return commitCL;
+        }
+        return null;
+    }
+
+
+    public static AsyncTxnResult readWithAccord(ClusterMetadata cm, PartitionRangeReadCommand command, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    {
+        if (consistencyLevel != null && !IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS.contains(consistencyLevel))
+            throw new InvalidRequestException(consistencyLevel + " is not supported by Accord");
+        TableMetadata tableMetadata = getTableMetadata(cm, command.metadata().id);
+        TableParams tableParams = tableMetadata.params;
+        Range<Token> readRange = new Range<>(command.dataRange().startKey().getToken(), command.dataRange().stopKey().getToken());
+        consistencyLevel = tableParams.transactionalMode.readCLForStrategy(tableParams.transactionalMigrationFrom, consistencyLevel, cm, tableMetadata.id, readRange);
+        TxnRead read = new TxnRangeRead(command, consistencyLevel);
+        Txn.Kind kind = Txn.Kind.Read;

Review Comment:
   Shouldn't this be an `EphemeralRead`? We only want to offer per-key linearisable read semantics I thought? This is also much safer, as we do not recover ephemeral reads, so a large range read will simply timeout.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833009942


##########
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java:
##########
@@ -277,19 +276,81 @@ private AsyncChain<Data> readChains()
                     try (PartitionIterator iterator = StorageProxy.readRegular(group, consistencyLevel, this, requestTime))
                     {
                         if (iterator.hasNext())
+                        {
+                            try (RowIterator partition = iterator.next())
+                            {
+                                TxnDataKeyValue value = new TxnDataKeyValue(partition);
+                                if (value.hasRows() || command.selectsFullPartition())
+                                    result.put(fragment.txnDataName(), value);
+                            }
+                        }
+                    }
+                    return result;
+                }));
+            });
+        });
+        return results;
+    }
+
+    private List<AsyncChain<Data>> rangeReadChains(int nowInSeconds, Dispatcher.RequestTime requestTime)
+    {
+        TxnKeyRead read = (TxnKeyRead) txn.read();
+        Seekables<?, ?> keys = txn.read().keys();
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        keys.forEach(key -> {
+            read.forEachWithKey((PartitionKey) key, fragment -> {

Review Comment:
   There is no live migration in this so yeah it's not reachable (and doesn't work) in `mixed` and `full` because the tests only cover the fully migrated case which will never use `AccordInteropExecution`.
   
   I'll remove the interop execution changes and put them in the live migration patch instead.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg merged PR #3463:
URL: https://github.com/apache/cassandra/pull/3463


-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833029293


##########
src/java/org/apache/cassandra/service/StorageProxy.java:
##########
@@ -2141,28 +2157,96 @@ private static PartitionIterator readWithConsensus(SinglePartitionReadCommand.Gr
         return lastResult.serialReadResult;
     }
 
-    private static ConsensusAttemptResult readWithAccord(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    private static ConsistencyLevel consistencyLevelForAccordRead(ClusterMetadata cm, SinglePartitionReadCommand.Group group, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        // Null means no specific consistency behavior is required from Accord, it's functionally similar to
+        // reading at ONE if you are reading data that wasn't written via Accord
+        if (consistencyLevel == null)
+            return null;
+
+        TableId tableId = group.queries.get(0).metadata().id;
+        TableParams tableParams = getTableMetadata(cm, tableId).params;
+        TransactionalMode mode = tableParams.transactionalMode;
+        TransactionalMigrationFromMode migrationFromMode = tableParams.transactionalMigrationFrom;
+        for (SinglePartitionReadCommand command : group.queries)
+        {
+            // readCLForStrategy should return either null or the supplied consistency level
+            // in which case we will read everything at that CL since Accord doesn't support per table
+            // read consistency
+            ConsistencyLevel commitCL = mode.readCLForStrategy(migrationFromMode, consistencyLevel, cm, tableId, command.partitionKey().getToken());
+            if (commitCL != null)
+                return commitCL;
+        }
+        return null;
+    }
+
+
+    public static AsyncTxnResult readWithAccord(ClusterMetadata cm, PartitionRangeReadCommand command, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    {
+        if (consistencyLevel != null && !IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS.contains(consistencyLevel))
+            throw new InvalidRequestException(consistencyLevel + " is not supported by Accord");
+        TableMetadata tableMetadata = getTableMetadata(cm, command.metadata().id);
+        TableParams tableParams = tableMetadata.params;
+        Range<Token> readRange = new Range<>(command.dataRange().startKey().getToken(), command.dataRange().stopKey().getToken());
+        consistencyLevel = tableParams.transactionalMode.readCLForStrategy(tableParams.transactionalMigrationFrom, consistencyLevel, cm, tableMetadata.id, readRange);
+        TxnRead read = new TxnRangeRead(command, consistencyLevel);
+        Txn.Kind kind = Txn.Kind.Read;

Review Comment:
   Ah, ephemeral reads work for range reads? I mistakenly thought they were limited to single key reads, but now I realize that it's just because of per key linearizability.
   
   I'll add them for range reads.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837898317


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;
+    private static final int TXN_DATA_NAME_INDEX_MASK = ~(~0 << TXN_DATA_NAME_INDEX_BITS);

Review Comment:
   I've never seen this construction before, I like it. I usually do `(0xffffffff >>> (32 - TXN_DATA_NAME_INDEX_BITS))` but this is also neat, particularly the `~0`.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837906146


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;

Review Comment:
   Also, not important, but if we use the lower bits for kind we can encode as a `vint` which I would expect to normally be a single byte.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833002782


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   I have a follow up patch that will do a bit more and fix some issues with how the ranges are passed to `TxnRangeRead` as part of making it live migratable. Can we do the switch from string to in there?
   
   When you mean pass keys through so it doesn't need to find `TableMetadata` what do you mean? Keys I think only contains `PartitionKey` with a `TableId` not a `TableMetadata`.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832599399


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -20,78 +20,59 @@
 
 import java.io.IOException;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
-import java.util.Set;
-
-import com.google.common.collect.Maps;
 
 import accord.api.Data;
 import org.apache.cassandra.db.EmptyIterators;
 import org.apache.cassandra.db.SinglePartitionReadCommand;
-import org.apache.cassandra.db.TypeSizes;
-import org.apache.cassandra.db.filter.ColumnFilter;
-import org.apache.cassandra.db.partitions.FilteredPartition;
 import org.apache.cassandra.db.partitions.PartitionIterators;
-import org.apache.cassandra.db.rows.DeserializationHelper;
-import org.apache.cassandra.db.rows.Row;
-import org.apache.cassandra.db.rows.UnfilteredRowIterator;
-import org.apache.cassandra.db.rows.UnfilteredRowIteratorSerializer;
-import org.apache.cassandra.db.rows.UnfilteredRowIterators;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
-import org.apache.cassandra.schema.Schema;
-import org.apache.cassandra.schema.TableId;
-import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.CollectionSerializers;
 import org.apache.cassandra.utils.NullableSerializer;
 import org.apache.cassandra.utils.ObjectSizes;
 
+import static accord.utils.Invariants.checkArgument;
 import static org.apache.cassandra.service.accord.txn.TxnResult.Kind.txn_data;
 
-public class TxnData extends TxnResult implements Data, Iterable<FilteredPartition>
+/**
+ * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
+ * when read results are returned to the coordinator to compute query results and writes.
+ */
+public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data

Review Comment:
   Or even could probably be a simple array in this case



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832623609


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   I think this is unnecessary?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838567402


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;
+    private static final int TXN_DATA_NAME_INDEX_MASK = ~(~0 << TXN_DATA_NAME_INDEX_BITS);
+    public static final int TXN_DATA_NAME_INDEX_MAX = ((1 << TXN_DATA_NAME_INDEX_BITS) - 1);
+
+    public enum TxnDataNameKind
+    {
+        USER((byte) 0),
+        RETURNING((byte) 1),
+        AUTO_READ((byte) 2),
+        CAS_READ((byte) 3);
+
+        private final byte value;
+
+        TxnDataNameKind(byte value)
+        {
+            this.value = value;
+        }
+
+        public static TxnDataNameKind from(byte b)
+        {
+            switch (b)
+            {
+                case 0:
+                    return USER;
+                case 1:
+                    return RETURNING;
+                case 2:
+                    return AUTO_READ;
+                case 3:
+                    return CAS_READ;
+                default:
+                    throw new IllegalArgumentException("Unknown kind: " + b);
+            }
+        }
+    }
+
+    public static int txnDataName(TxnDataNameKind kind, int index)
+    {
+        checkArgument(index >= 0 && index <= TXN_DATA_NAME_INDEX_MAX);
+        int kindInt = (int)(((long)kind.value) << TXN_DATA_NAME_INDEX_BITS);
+        return kindInt | index;
+    }
+
+    public static int txnDataName(TxnDataNameKind kind)
+    {
+        return txnDataName(kind, 0);
+    }
+
+    public static TxnDataNameKind txnDataNameKind(int txnDataName)
+    {
+        int kind = txnDataName >> TXN_DATA_NAME_INDEX_BITS;

Review Comment:
   Oh that's a genuine mistake/bug. I forgot which shift was which. We don't want signed shift because it would result in the wrong value in kind if it extended a 1.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838559737


##########
src/java/org/apache/cassandra/service/accord/txn/AbstractKeySorted.java:
##########
@@ -55,16 +54,39 @@ public AbstractKeySorted(List<T> items)
         items.toArray(arr);
         Arrays.sort(arr, this::compare);
         this.items = arr;
-        validateOrder();
-        this.itemKeys = extractItemKeys();
+        int duplicates = validateOrder();
+        this.itemKeys = extractItemKeys(duplicates);
     }
 
-    private Keys extractItemKeys()
+    private Keys extractItemKeys(int duplicates)
     {
-        Key[] keys = new Key[size()];
-        for (int i = 0 ; i < keys.length ; ++i)
-            keys[i] = getKey(items[i]);
-        return Keys.ofSorted(keys);
+        PartitionKey[] partitionKeys = new PartitionKey[items.length - duplicates];

Review Comment:
   I was just trying to reuse the work done to ensure that is sorted. The simpler approach is to just remove the sorting check and rely on the one in `Keys.ofSorted`



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833083886


##########
src/java/org/apache/cassandra/service/StorageProxy.java:
##########
@@ -2141,28 +2157,96 @@ private static PartitionIterator readWithConsensus(SinglePartitionReadCommand.Gr
         return lastResult.serialReadResult;
     }
 
-    private static ConsensusAttemptResult readWithAccord(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    private static ConsistencyLevel consistencyLevelForAccordRead(ClusterMetadata cm, SinglePartitionReadCommand.Group group, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        // Null means no specific consistency behavior is required from Accord, it's functionally similar to
+        // reading at ONE if you are reading data that wasn't written via Accord
+        if (consistencyLevel == null)
+            return null;
+
+        TableId tableId = group.queries.get(0).metadata().id;
+        TableParams tableParams = getTableMetadata(cm, tableId).params;
+        TransactionalMode mode = tableParams.transactionalMode;
+        TransactionalMigrationFromMode migrationFromMode = tableParams.transactionalMigrationFrom;
+        for (SinglePartitionReadCommand command : group.queries)
+        {
+            // readCLForStrategy should return either null or the supplied consistency level
+            // in which case we will read everything at that CL since Accord doesn't support per table
+            // read consistency
+            ConsistencyLevel commitCL = mode.readCLForStrategy(migrationFromMode, consistencyLevel, cm, tableId, command.partitionKey().getToken());
+            if (commitCL != null)
+                return commitCL;
+        }
+        return null;
+    }
+
+
+    public static AsyncTxnResult readWithAccord(ClusterMetadata cm, PartitionRangeReadCommand command, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime)
+    {
+        if (consistencyLevel != null && !IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS.contains(consistencyLevel))
+            throw new InvalidRequestException(consistencyLevel + " is not supported by Accord");
+        TableMetadata tableMetadata = getTableMetadata(cm, command.metadata().id);
+        TableParams tableParams = tableMetadata.params;
+        Range<Token> readRange = new Range<>(command.dataRange().startKey().getToken(), command.dataRange().stopKey().getToken());
+        consistencyLevel = tableParams.transactionalMode.readCLForStrategy(tableParams.transactionalMigrationFrom, consistencyLevel, cm, tableMetadata.id, readRange);
+        TxnRead read = new TxnRangeRead(command, consistencyLevel);
+        Txn.Kind kind = Txn.Kind.Read;

Review Comment:
   I thought the plan was to *only* support range reads as ephemeral (or, at least, that we intended to offer no better semantics so we can slice up the token range for eg paging)



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833022022


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   It's C&P from `TxnKeyRead` which added it [here](https://github.com/apache/cassandra/commit/8deb6639f6477a5c61bcb7422f2801fd19b16331#diff-f48c552114deecaa1cf2678dd9085170b8a37da3d867ea734b06b97639cb3a90R207) and I don't recall why I thought it was necessary at the time.
   
   Maybe once simulator tests on Accord pass we can try removing that and see if it actually did anything?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833259681


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   It's necessary if you use synchronised and need the simulator to be aware of it, and to manage the locking (this is necessary if you want to simulate thread interleaving, or if other simulated actions might deadlock if the simulator doesn't know a monitor is being held). It shouldn't be necessary for any transitive monitor access (i.e. that within method calls). They need to be annotated themselves.
   
   So I don't think it is necessary in this case.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832593769


##########
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java:
##########
@@ -277,19 +276,81 @@ private AsyncChain<Data> readChains()
                     try (PartitionIterator iterator = StorageProxy.readRegular(group, consistencyLevel, this, requestTime))
                     {
                         if (iterator.hasNext())
+                        {
+                            try (RowIterator partition = iterator.next())
+                            {
+                                TxnDataKeyValue value = new TxnDataKeyValue(partition);
+                                if (value.hasRows() || command.selectsFullPartition())
+                                    result.put(fragment.txnDataName(), value);
+                            }
+                        }
+                    }
+                    return result;
+                }));
+            });
+        });
+        return results;
+    }
+
+    private List<AsyncChain<Data>> rangeReadChains(int nowInSeconds, Dispatcher.RequestTime requestTime)
+    {
+        TxnKeyRead read = (TxnKeyRead) txn.read();
+        Seekables<?, ?> keys = txn.read().keys();
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        keys.forEach(key -> {
+            read.forEachWithKey((PartitionKey) key, fragment -> {

Review Comment:
   I'm struggling to see how this can work - surely this shouldn't be a partition key here?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833372345


##########
src/java/org/apache/cassandra/service/accord/txn/TxnKeyRead.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.collect.ImmutableList;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.Keys;
+import accord.primitives.Participants;
+import accord.primitives.Ranges;
+import accord.primitives.Seekable;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.service.accord.api.PartitionKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnKeyRead extends AbstractKeySorted<TxnNamedRead> implements TxnRead
+{
+    public static final TxnKeyRead EMPTY = new TxnKeyRead(new TxnNamedRead[0], Keys.EMPTY, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String CAS_READ_NAME = "CAS_READ";
+    public static final TxnDataName CAS_READ = new TxnDataName(TxnDataName.Kind.CAS_READ, CAS_READ_NAME);
+
+    @Nonnull
+    private final Keys txnKeys;
+
+    // Cassandra's consistency level used by Accord to safely read data written outside of Accord
+    @Nullable
+    private final ConsistencyLevel cassandraConsistencyLevel;
+
+    public TxnKeyRead(@Nonnull TxnNamedRead[] items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public TxnKeyRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(items);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.txnKeys = txnKeys;
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+    }
+
+    public static TxnKeyRead createTxnRead(@Nonnull List<TxnNamedRead> items, @Nonnull Keys txnKeys, @Nullable ConsistencyLevel consistencyLevel)
+    {
+        return new TxnKeyRead(items, txnKeys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createSerialRead(List<SinglePartitionReadCommand> readCommands, ConsistencyLevel consistencyLevel)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(readCommands.size());
+        for (int i = 0; i < readCommands.size(); i++)
+            reads.add(new TxnNamedRead(TxnDataName.user(String.valueOf(i)), readCommands.get(i)));
+        Keys keys = Keys.of(reads, TxnNamedRead::key);
+        return new TxnKeyRead(reads, keys, consistencyLevel);
+    }
+
+    public static TxnKeyRead createCasRead(SinglePartitionReadCommand readCommand, ConsistencyLevel consistencyLevel)
+    {
+        TxnNamedRead read = new TxnNamedRead(CAS_READ, readCommand);
+        return new TxnKeyRead(ImmutableList.of(read), Keys.of(read.key()), consistencyLevel);
+    }
+
+    // A read that declares it will read from keys but doesn't actually read any data so dependent transactions will
+    // still be applied first
+    public static TxnKeyRead createNoOpRead(Keys keys)
+    {
+        return new TxnKeyRead(ImmutableList.of(), keys, null);
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        for (TxnNamedRead read : items)
+            size += read.estimatedSizeOnHeap();
+        return size;
+    }
+
+    @Override
+    int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
+    {
+        return left.txnDataName().compareTo(right.txnDataName());
+    }
+
+    @Override
+    PartitionKey getKey(TxnNamedRead read)
+    {
+        return read.key();
+    }
+
+    @Override
+    TxnNamedRead[] newArray(int size)
+    {
+        return new TxnNamedRead[size];
+    }
+
+    @Override
+    public Keys keys()
+    {
+        return txnKeys;
+    }
+
+    @Override
+    public ConsistencyLevel cassandraConsistencyLevel()
+    {
+        return cassandraConsistencyLevel;
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return intersecting(itemKeys.slice(ranges));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        return intersecting(itemKeys.intersecting(participants));
+    }
+
+    private Read intersecting(Keys select)
+    {
+        Keys keys = itemKeys.intersecting(select);
+        List<TxnNamedRead> reads = new ArrayList<>(keys.size());
+
+        for (TxnNamedRead read : items)
+            if (keys.contains(read.key()))
+                reads.add(read);
+
+        return createTxnRead(reads, txnKeys.intersecting(select), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public Read merge(Read read)
+    {
+        List<TxnNamedRead> reads = new ArrayList<>(items.length);
+        Collections.addAll(reads, items);
+
+        for (TxnNamedRead namedRead : (TxnKeyRead) read)
+            if (!reads.contains(namedRead))
+                reads.add(namedRead);
+
+        return createTxnRead(reads, txnKeys.with((Keys)read.keys()), cassandraConsistencyLevel);
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
+    {
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        forEachWithKey((PartitionKey) key, read -> results.add(read.read(cassandraConsistencyLevel, executeAt)));
+
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        if (results.size() == 1)
+            return results.get(0);
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.key;
+    }
+
+    @Simulate(with = MONITORS)
+    public static final TxnReadSerializer<TxnKeyRead> serializer = new TxnReadSerializer<TxnKeyRead>()
+    {
+        @Override
+        public void serialize(TxnKeyRead read, DataOutputPlus out, int version) throws IOException
+        {
+            KeySerializers.keys.serialize(read.txnKeys, out, version);
+            serializeArray(read.items, out, version, TxnNamedRead.serializer);
+            serializeNullable(read.cassandraConsistencyLevel, out, version, consistencyLevelSerializer);
+        }
+
+        @Override
+        public TxnKeyRead deserialize(DataInputPlus in, int version) throws IOException
+        {
+            Keys keys = KeySerializers.keys.deserialize(in, version);
+            TxnNamedRead[] items = deserializeArray(in, version, TxnNamedRead.serializer, TxnNamedRead[]::new);

Review Comment:
   No, we should definitely be slicing the keys same as we slice the data. Agreed we can infer keys from the data fragments, although in principle they might be duplicated so could be more efficient in rare cases to encode the keys then encode an index into the keys for the values - but not likely to be important enough to matter 



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837906146


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;

Review Comment:
   Also, not important, but if we use the lower bits for kind we can encode as a `vint` (but, perhaps we don't really want to)



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1833260366


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   (It's also fairly harmless, but a bit confusing if you know what it means)



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1834756715


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);
+    }
+
+    private TxnRangeRead(@Nonnull ByteBuffer commandBytes, @Nonnull ConsistencyLevel cassandraConsistencyLevel, @Nonnull Ranges covering)
+    {
+        super(commandBytes);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        this.covering = covering;
+    }
+
+    @Override
+    public Seekables<?, ?> keys()
+    {
+        return covering;
+    }
+
+    @Override
+    public AsyncChain<Data> read(Seekable range, SafeCommandStore commandStore, Timestamp executeAt, DataStore store)
+    {
+        // It's fine for our nowInSeconds to lag slightly our insertion timestamp, as to the user
+        // this simply looks like the transaction witnessed TTL'd data and the data then expired
+        // immediately after the transaction executed, and this simplifies things a great deal
+        int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
+
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        PartitionRangeReadCommand command = (PartitionRangeReadCommand)get();
+        Ranges intersecting = covering.slice(Ranges.of(range.asRange()), Slice.Minimal);
+        for (Range subRange : intersecting)
+            results.add(performLocalRead(command, subRange, nowInSeconds));
+        if (results.isEmpty())
+            // Result type must match everywhere
+            return AsyncChains.success(new TxnData());
+
+        return AsyncChains.reduce(results, Data::merge);
+    }
+
+    private AsyncChain<Data> performLocalRead(PartitionRangeReadCommand command, Range r, int nowInSeconds)
+    {
+        Callable<Data> readCallable = () ->
+        {
+            AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
+            PartitionPosition startPP = bounds.left;
+            PartitionPosition endPP = bounds.right;
+            TokenKey startTokenKey = new TokenKey(command.metadata().id, startPP.getToken());
+            Token subRangeStartToken = ((AccordRoutingKey)r.start()).asTokenKey().token();
+            Token subRangeEndToken = ((AccordRoutingKey)r.end()).asTokenKey().token();
+            /*
+             * The way ranges/bounds work for range queries is that the beginning and ending bounds from the command
+             * could be tokens (and min/max key bounds) or actual keys depending on the bounds of the top level query we
+             * are running and where we are in paging. We need to preserve whatever is in the command in case it is a
+             * key and not a token, or it's a token but might be a min/max key bound.
+             *
+             * Then Accord will further subdivide the range in the command so need to inject additional bounds in the middle
+             * that match the range ownership of Accord.
+             *
+             * The command still contains the original bound and then the Accord range passed in determines what subset of
+             * that bound we want. We have to make sure to use the bounds from the command if it is the start or end instead
+             * of a key bound created from the Accord range since it could be a real key or min/max bound.
+             *
+             * When we are dealing with a bound created by Accord's further subdivision we use a maxKeyBound (exclusive)
+             * for both beginning and end because Bounds is left and right inclusive while Range is only left inclusive.
+             * We only use TokenRange with Accord which matches the left/right inclusivity of Cassandra's Range.
+             *
+             * That means the Range we get from Accord overlaps the previous Range on the left which when converted to a Bound
+             * would potentially read the same Token twice. So the left needs to be a maxKeyBound to exclude the data that isn't
+             * owned here and to avoid potentially reading the same data twice. The right bound also needs to be a maxKeyBound since Range
+             * is right inclusive so every partition we find needs to be < the right bound.
+             */
+            boolean isFirstSubrange = startPP.getToken().equals(subRangeStartToken);
+            PartitionPosition subRangeStartPP = isFirstSubrange ? startPP : subRangeStartToken.maxKeyBound();
+            PartitionPosition subRangeEndPP = endPP.getToken().equals(subRangeEndToken) ? endPP : subRangeEndToken.maxKeyBound();
+            // Need to preserve the fact it is a bounds for paging to work, a range is not left inclusive and will not start from where we left off
+            AbstractBounds<PartitionPosition> subRange = isFirstSubrange ? bounds.withNewRight(subRangeEndPP) : new org.apache.cassandra.dht.Range(subRangeStartPP, subRangeEndPP);
+            PartitionRangeReadCommand read = command.forSubRangeWithNowInSeconds(nowInSeconds, subRange, startTokenKey.equals(r.start()));
+
+            try (ReadExecutionController controller = read.executionController();
+                 UnfilteredPartitionIterator partition = read.executeLocally(controller);
+                 PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, read.nowInSec()))
+            {
+                TxnData result = new TxnData();
+                TxnDataRangeValue value = new TxnDataRangeValue();
+                while(iterator.hasNext())
+                {
+                    try (RowIterator rows = iterator.next())
+                    {
+                        FilteredPartition filtered = FilteredPartition.create(rows);
+                        if (filtered.hasRows() || read.selectsFullPartition())
+                        {
+                            value.add(filtered);
+                        }
+                    }
+                }
+                result.put(RANGE_READ, value);
+                return result;
+            }
+        };
+
+        return AsyncChains.ofCallable(Stage.READ.executor(), readCallable, (callable, receiver) ->
+                                                                           new DebuggableTask.RunnableDebuggableTask()
+                                                                           {
+                                                                               private final long approxCreationTimeNanos = MonotonicClock.Global.approxTime.now();
+                                                                               private volatile long approxStartTimeNanos;
+
+                                                                               @Override
+                                                                               public void run()
+                                                                               {
+                                                                                   approxStartTimeNanos = MonotonicClock.Global.approxTime.now();
+
+                                                                                   try
+                                                                                   {
+                                                                                       Data call = callable.call();
+                                                                                       receiver.accept(call, null);
+                                                                                   }
+                                                                                   catch (Throwable t)
+                                                                                   {
+                                                                                       logger.debug("AsyncChain Callable threw an Exception", t);
+                                                                                       receiver.accept(null, t);
+                                                                                   }
+                                                                               }
+
+                                                                               @Override
+                                                                               public long creationTimeNanos()
+                                                                               {
+                                                                                   return approxCreationTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public long startTimeNanos()
+                                                                               {
+                                                                                   return approxStartTimeNanos;
+                                                                               }
+
+                                                                               @Override
+                                                                               public String description()
+                                                                               {
+                                                                                   return command.toCQLString();
+                                                                               }
+                                                                           }
+        );
+    }
+
+    @Override
+    public Read slice(Ranges ranges)
+    {
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.slice(ranges, Slice.Minimal));
+    }
+
+    @Override
+    public Read intersecting(Participants<?> participants)
+    {
+        checkArgument(participants.domain() == Domain.Range, "Can only intersect with ranges");
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, covering.intersecting(participants, Slice.Minimal));
+    }
+
+    @Override
+    public Read merge(Read other)
+    {
+        Ranges newCovering = ((TxnRangeRead)other).covering.union(UnionMode.MERGE_OVERLAPPING, covering);
+        return new TxnRangeRead(bytes(), cassandraConsistencyLevel, newCovering);
+    }
+
+    @Override
+    public Kind kind()
+    {
+        return Kind.range;
+    }
+
+    @Override
+    public long estimatedSizeOnHeap()
+    {
+        long size = EMPTY_SIZE;
+        size += ByteBufferUtil.estimatedSizeOnHeap(bytes());
+        size += AccordObjectSizes.ranges(covering);
+        return size;
+    }
+
+    @Simulate(with = MONITORS)

Review Comment:
   OK will remove them.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832674931


##########
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java:
##########
@@ -277,19 +276,81 @@ private AsyncChain<Data> readChains()
                     try (PartitionIterator iterator = StorageProxy.readRegular(group, consistencyLevel, this, requestTime))
                     {
                         if (iterator.hasNext())
+                        {
+                            try (RowIterator partition = iterator.next())
+                            {
+                                TxnDataKeyValue value = new TxnDataKeyValue(partition);
+                                if (value.hasRows() || command.selectsFullPartition())
+                                    result.put(fragment.txnDataName(), value);
+                            }
+                        }
+                    }
+                    return result;
+                }));
+            });
+        });
+        return results;
+    }
+
+    private List<AsyncChain<Data>> rangeReadChains(int nowInSeconds, Dispatcher.RequestTime requestTime)
+    {
+        TxnKeyRead read = (TxnKeyRead) txn.read();
+        Seekables<?, ?> keys = txn.read().keys();
+        List<AsyncChain<Data>> results = new ArrayList<>();
+        keys.forEach(key -> {
+            read.forEachWithKey((PartitionKey) key, fragment -> {

Review Comment:
   That is, shouldn't `txn.read()` be a `TxnRangeRead`?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1832621929


##########
src/java/org/apache/cassandra/service/accord/txn/TxnRangeRead.java:
##########
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service.accord.txn;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.api.Data;
+import accord.api.DataStore;
+import accord.api.Read;
+import accord.local.SafeCommandStore;
+import accord.primitives.AbstractRanges.UnionMode;
+import accord.primitives.Participants;
+import accord.primitives.Range;
+import accord.primitives.Ranges;
+import accord.primitives.Routable.Domain;
+import accord.primitives.Routables.Slice;
+import accord.primitives.Seekable;
+import accord.primitives.Seekables;
+import accord.primitives.Timestamp;
+import accord.utils.async.AsyncChain;
+import accord.utils.async.AsyncChains;
+import org.apache.cassandra.concurrent.DebuggableTask;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.PartitionPosition;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.RowIterator;
+import org.apache.cassandra.dht.AbstractBounds;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordObjectSizes;
+import org.apache.cassandra.service.accord.TokenRange;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.SentinelKey;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.accord.serializers.KeySerializers;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.MonotonicClock;
+import org.apache.cassandra.utils.ObjectSizes;
+import org.apache.cassandra.utils.Simulate;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.cassandra.service.accord.AccordSerializers.consistencyLevelSerializer;
+import static org.apache.cassandra.service.accord.IAccordService.SUPPORTED_READ_CONSISTENCY_LEVELS;
+import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
+import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
+import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
+import static org.apache.cassandra.utils.Simulate.With.MONITORS;
+
+public class TxnRangeRead extends AbstractSerialized<ReadCommand> implements TxnRead
+{
+    private static final Logger logger = LoggerFactory.getLogger(TxnRangeRead.class);
+
+    public static final TxnRangeRead EMPTY = new TxnRangeRead(null, null, null);
+    private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY);
+
+    public static final String RANGE_READ_NAME = "RANGE_READ";
+    public static final TxnDataName RANGE_READ = TxnDataName.user(RANGE_READ_NAME);
+
+    @Nonnull
+    private final ConsistencyLevel cassandraConsistencyLevel;
+    @Nonnull
+    private final Ranges covering;
+
+    public TxnRangeRead(@Nonnull PartitionRangeReadCommand command, @Nonnull ConsistencyLevel cassandraConsistencyLevel)
+    {
+        super(command);
+        checkArgument(cassandraConsistencyLevel == null || SUPPORTED_READ_CONSISTENCY_LEVELS.contains(cassandraConsistencyLevel), "Unsupported consistency level for read");
+        this.cassandraConsistencyLevel = cassandraConsistencyLevel;
+        TableId tableId = command.metadata().id;
+
+        AbstractBounds<PartitionPosition> range = command.dataRange().keyRange();
+
+        // Read commands can contain a mix of different kinds of bounds to facilitate paging
+        // and we need to communicate that to Accord as its own ranges. We will use
+        // TokenKey and Sentinel key and stick exclusively with left exclusive/right inclusive
+        // ranges rather add more types of ranges to the mix
+        boolean inclusiveLeft = range.inclusiveLeft();
+        Token startToken = range.left.getToken();
+        AccordRoutingKey startAccordRoutingKey;
+        if (startToken.isMinimum() && inclusiveLeft)
+            startAccordRoutingKey = SentinelKey.min(tableId);
+        else
+            startAccordRoutingKey = new TokenKey(tableId, startToken);
+
+        boolean inclusiveRight = range.inclusiveRight();
+        Token stopToken = range.right.getToken();
+        AccordRoutingKey stopAccordRoutingKey;
+        if (inclusiveRight)
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken);
+        else
+            stopAccordRoutingKey = new TokenKey(tableId, stopToken.decreaseSlightly());
+
+        covering = Ranges.of(new TokenRange(startAccordRoutingKey, stopAccordRoutingKey));
+        logger.info("Ariel Creating txn range read of " + command);

Review Comment:
   this probably needs to be removed :) 



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837898317


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;
+    private static final int TXN_DATA_NAME_INDEX_MASK = ~(~0 << TXN_DATA_NAME_INDEX_BITS);

Review Comment:
   I've never seen this construction before, I like it. I usually do `(0xffffffff >>> (32 - TXN_DATA_NAME_INDEX_BITS))` (or sometimes `(1<<bits)-1`) but this is also neat, particularly the `~0`.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837901043


##########
src/java/org/apache/cassandra/service/accord/txn/TxnData.java:
##########
@@ -40,18 +41,77 @@
  * Fairly generic holder for result values for Accord txns as well as data exchange during Accord txn execution
  * when read results are returned to the coordinator to compute query results and writes.
  */
-public class TxnData extends HashMap<TxnDataName, TxnDataValue> implements TxnResult, Data
+public class TxnData extends Int2ObjectHashMap<TxnDataValue> implements TxnResult, Data
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnData());
 
+    private static final int TXN_DATA_NAME_INDEX_BITS = 32 - 6;

Review Comment:
   why 6?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837895307


##########
src/java/org/apache/cassandra/service/accord/txn/AbstractKeySorted.java:
##########
@@ -55,16 +54,39 @@ public AbstractKeySorted(List<T> items)
         items.toArray(arr);
         Arrays.sort(arr, this::compare);
         this.items = arr;
-        validateOrder();
-        this.itemKeys = extractItemKeys();
+        int duplicates = validateOrder();
+        this.itemKeys = extractItemKeys(duplicates);
     }
 
-    private Keys extractItemKeys()
+    private Keys extractItemKeys(int duplicates)
     {
-        Key[] keys = new Key[size()];
-        for (int i = 0 ; i < keys.length ; ++i)
-            keys[i] = getKey(items[i]);
-        return Keys.ofSorted(keys);
+        PartitionKey[] partitionKeys = new PartitionKey[items.length - duplicates];

Review Comment:
   Alternatively, we can avoid additional key comparisons by using `ArrayBuffers.cachedKeys` to accumulate the keys. I think it would still be nice to offer this as a static builder in either `Keys` or `SortedArrays` (or in both, with the former using the latter to build the array) so that it can be reused. WDYT?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["belliottsmith (via GitHub)" <gi...@apache.org>]

belliottsmith commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1837886028


##########
src/java/org/apache/cassandra/cql3/transactions/RowDataReference.java:
##########
@@ -250,15 +251,15 @@ private void resolveFinished()
             isResolved = true;
         }
 
-        public void resolveReference(Map<TxnDataName, ReferenceSource> sources)
+        public void resolveReference(Map<Integer, ReferenceSource> sources, Map<String, Integer> nameToTxnDataName, int userReadIndex)
         {
             if (isResolved)
                 return;
 
-            // root level name
-            tupleName = TxnDataName.user(tuple.toString());
+            // root level name, use the one that was already created if it exists otherwise generate a new one
+            tupleName = nameToTxnDataName.getOrDefault(tuple.toString(), userReadIndex);
             ReferenceSource source = sources.get(tupleName);
-            checkNotNull(source, CANNOT_FIND_TUPLE_MESSAGE, tupleName.name());
+            checkNotNull(source, CANNOT_FIND_TUPLE_MESSAGE, tuple.toString());

Review Comment:
   nit: maybe introduce an accessor that makes clear we're just extracting the existing prepared `text` field?



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org

Re: [PR] CASSANDRA-19437 non-serial reads and range reads without live migration support [cassandra]

["aweisberg (via GitHub)" <gi...@apache.org>]

aweisberg commented on code in PR #3463:
URL: https://github.com/apache/cassandra/pull/3463#discussion_r1838578961


##########
src/java/org/apache/cassandra/service/accord/txn/AbstractKeySorted.java:
##########
@@ -55,16 +54,39 @@ public AbstractKeySorted(List<T> items)
         items.toArray(arr);
         Arrays.sort(arr, this::compare);
         this.items = arr;
-        validateOrder();
-        this.itemKeys = extractItemKeys();
+        int duplicates = validateOrder();
+        this.itemKeys = extractItemKeys(duplicates);
     }
 
-    private Keys extractItemKeys()
+    private Keys extractItemKeys(int duplicates)
     {
-        Key[] keys = new Key[size()];
-        for (int i = 0 ; i < keys.length ; ++i)
-            keys[i] = getKey(items[i]);
-        return Keys.ofSorted(keys);
+        PartitionKey[] partitionKeys = new PartitionKey[items.length - duplicates];

Review Comment:
   Fixed by making `AbstractKeysSorted` extend `AbstractList` so that the transforming version of `Keys.of` can be used.



-- 
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: pr-unsubscribe@cassandra.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org