You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "dcapwell (via GitHub)" <gi...@apache.org> on 2024/12/13 21:29:32 UTC
[PR] First dumb of logic to get route index running on the journal [cassandra]
dcapwell opened a new pull request, #3743:
URL: https://github.com/apache/cassandra/pull/3743
(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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917097314
##########
src/java/org/apache/cassandra/service/accord/AccordService.java:
##########
@@ -1249,14 +1251,17 @@ public CompactionInfo getCompactionInfo()
Int2ObjectHashMap<RedundantBefore> redundantBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<DurableBefore> durableBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<RangesForEpoch> ranges = new Int2ObjectHashMap<>();
- AsyncChains.getBlockingAndRethrow(node.commandStores().forEach(safeStore -> {
- synchronized (redundantBefores)
- {
- redundantBefores.put(safeStore.commandStore().id(), safeStore.redundantBefore());
- ranges.put(safeStore.commandStore().id(), safeStore.ranges());
- durableBefores.put(safeStore.commandStore().id(), safeStore.durableBefore());
- }
- }));
+ if (node.commandStores().all().length > 0)
Review Comment:
this is no longer a true statement (about the test), but its a failure mode.... if you had accord then migrate away you still have the accord tables... if compaction runs then we fail without this check.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917519897
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -108,42 +111,57 @@ public enum RegisterStatus
private volatile boolean initBuildStarted = false;
private volatile RegisterStatus registerStatus = RegisterStatus.PENDING;
- public RouteIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ public ParticipantsJournalIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
{
- if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
- if (!AccordKeyspace.COMMANDS.equals(baseCfs.name))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
-
- TableMetadata tableMetadata = baseCfs.metadata();
- Pair<ColumnMetadata, IndexTarget.Type> target = TargetParser.parse(tableMetadata, indexMetadata);
- if (!AccordKeyspace.CommandsColumns.participants.name.equals(target.left.name))
- throw new IllegalArgumentException("Attempted to index the wrong column; needed " + AccordKeyspace.CommandsColumns.participants.name + " but given " + target.left.name);
-
- if (target.right != IndexTarget.Type.VALUES)
- throw new IllegalArgumentException("Attempted to index " + AccordKeyspace.CommandsColumns.participants.name + " with index type " + target.right + "; only " + IndexTarget.Type.VALUES + " is supported");
+ validateTargets(baseCfs, indexMetadata);
this.baseCfs = baseCfs;
+ // type is only IndexTarget.Type.VALUES
+ this.record = AccordKeyspace.JournalColumns.record;
Review Comment:
done
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917524843
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -407,7 +416,7 @@ public Searcher searcherFor(ReadCommand command)
return null;
}
}
- else if (e.column() == AccordKeyspace.CommandsColumns.store_id && e.operator() == Operator.EQ)
+ else if (e.column() == AccordJournalTable.SyntheticColumn.store_id.metadata && e.operator() == Operator.EQ)
Review Comment:
CQL won't hit this logic so that feels safe
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913837113
##########
src/java/org/apache/cassandra/service/accord/IJournal.java:
##########
@@ -37,4 +41,12 @@ default SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId,
}
Persister<DurableBefore, DurableBefore> durableBeforePersister();
+
+ RangeSearcher rangeSearcher();
+
+ interface RangeSearcher
Review Comment:
due to recursive interface logic I made this top level.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "iamaleksey (via GitHub)" <gi...@apache.org>.
iamaleksey commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1920076395
##########
src/java/org/apache/cassandra/journal/SegmentCompactor.java:
##########
@@ -30,5 +31,7 @@ static <K, V> SegmentCompactor<K, V> noop()
return (SegmentCompactor<K, V>) NOOP;
}
+ //TODO (review): org.apache.cassandra.journal.Compactor.run documents that this is nullable but no implementation returns null.... If we want to keep nullable I put it in the interface so you have to check, but if we want to remove I can go clean that assumption up.
Review Comment:
Doesn't matter much - can adjust when another implementation comes around.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1920459731
##########
src/java/org/apache/cassandra/journal/SegmentCompactor.java:
##########
@@ -30,5 +31,7 @@ static <K, V> SegmentCompactor<K, V> noop()
return (SegmentCompactor<K, V>) NOOP;
}
+ //TODO (review): org.apache.cassandra.journal.Compactor.run documents that this is nullable but no implementation returns null.... If we want to keep nullable I put it in the interface so you have to check, but if we want to remove I can go clean that assumption up.
Review Comment:
thanks, ill remove all references to `null` for this code path
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884691630
##########
src/java/org/apache/cassandra/service/accord/AccordService.java:
##########
@@ -248,6 +248,8 @@ public static IVerbHandler<? extends Reply> responseHandlerOrNoop()
public synchronized static void startup(NodeId tcmId)
{
+ if (instance != null)
Review Comment:
safety check, I hit an issue and saw we didn't detect this so just added to be safer.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913502378
##########
src/java/org/apache/cassandra/service/accord/IJournal.java:
##########
@@ -37,4 +41,12 @@ default SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId,
}
Persister<DurableBefore, DurableBefore> durableBeforePersister();
+
+ RangeSearcher rangeSearcher();
+
+ interface RangeSearcher
Review Comment:
range search is getting rewritten? what 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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "ifesdjeen (via GitHub)" <gi...@apache.org>.
ifesdjeen commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1916715105
##########
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java:
##########
Review Comment:
Since the index is called participant index, should we also call this test participant index test, or maybe should we rename test? Consistency will help discoverability.
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -108,42 +111,57 @@ public enum RegisterStatus
private volatile boolean initBuildStarted = false;
private volatile RegisterStatus registerStatus = RegisterStatus.PENDING;
- public RouteIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ public ParticipantsJournalIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
{
- if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
- if (!AccordKeyspace.COMMANDS.equals(baseCfs.name))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
-
- TableMetadata tableMetadata = baseCfs.metadata();
- Pair<ColumnMetadata, IndexTarget.Type> target = TargetParser.parse(tableMetadata, indexMetadata);
- if (!AccordKeyspace.CommandsColumns.participants.name.equals(target.left.name))
- throw new IllegalArgumentException("Attempted to index the wrong column; needed " + AccordKeyspace.CommandsColumns.participants.name + " but given " + target.left.name);
-
- if (target.right != IndexTarget.Type.VALUES)
- throw new IllegalArgumentException("Attempted to index " + AccordKeyspace.CommandsColumns.participants.name + " with index type " + target.right + "; only " + IndexTarget.Type.VALUES + " is supported");
+ validateTargets(baseCfs, indexMetadata);
this.baseCfs = baseCfs;
+ // type is only IndexTarget.Type.VALUES
+ this.record = AccordKeyspace.JournalColumns.record;
Review Comment:
nit: maybe use them straight from journal columns? why have them as separate fields here
##########
src/java/org/apache/cassandra/service/accord/AccordCache.java:
##########
@@ -1190,7 +1190,7 @@ public Runnable save(AccordCommandStore commandStore, TxnId txnId, @Nullable Com
return null;
}
- return commandStore.appendToKeyspace(txnId, value);
+ return null;
Review Comment:
Do I understand it correctly though that since the command is already in the log, we basically already cached it?
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -108,42 +111,57 @@ public enum RegisterStatus
private volatile boolean initBuildStarted = false;
private volatile RegisterStatus registerStatus = RegisterStatus.PENDING;
- public RouteIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ public ParticipantsJournalIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
{
- if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
- if (!AccordKeyspace.COMMANDS.equals(baseCfs.name))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
-
- TableMetadata tableMetadata = baseCfs.metadata();
- Pair<ColumnMetadata, IndexTarget.Type> target = TargetParser.parse(tableMetadata, indexMetadata);
- if (!AccordKeyspace.CommandsColumns.participants.name.equals(target.left.name))
- throw new IllegalArgumentException("Attempted to index the wrong column; needed " + AccordKeyspace.CommandsColumns.participants.name + " but given " + target.left.name);
-
- if (target.right != IndexTarget.Type.VALUES)
- throw new IllegalArgumentException("Attempted to index " + AccordKeyspace.CommandsColumns.participants.name + " with index type " + target.right + "; only " + IndexTarget.Type.VALUES + " is supported");
+ validateTargets(baseCfs, indexMetadata);
this.baseCfs = baseCfs;
+ // type is only IndexTarget.Type.VALUES
+ this.record = AccordKeyspace.JournalColumns.record;
+ this.user_version = AccordKeyspace.JournalColumns.user_version;
this.indexMetadata = indexMetadata;
+
this.memtableIndexManager = new RouteMemtableIndexManager(this);
this.sstableManager = new RouteSSTableManager();
this.indexMetrics = new IndexMetrics(this);
- this.column = target.left;
Tracker tracker = baseCfs.getTracker();
tracker.subscribe(this);
}
- public ColumnMetadata column()
+ public static boolean allowed(JournalKey id)
+ {
+ return id.type == JournalKey.Type.COMMAND_DIFF && allowed(id.id);
+ }
+
+ public static boolean allowed(TxnId id)
{
- return column;
+ return id.domain().isRange();
+ }
+
+ private static void validateTargets(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ {
+ // this contains 2 columns....
+ if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ if (!AccordKeyspace.JOURNAL.equals(baseCfs.name))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ Set<String> columns = Splitter.on(',').trimResults().omitEmptyStrings().splitToStream(indexMetadata.options.get("target")).collect(Collectors.toSet());
+ Set<String> expected = Set.of("record", "user_version");
+ if (!expected.equals(columns))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table, and on the record/user_value columns; given " + baseCfs.metadata() + " and columns " + columns);
}
public IndexMetrics indexMetrics()
{
return indexMetrics;
}
+ public RegisterStatus registerStatus()
+ {
+ return registerStatus;
+ }
+
Review Comment:
Could not add a comment directly to L201, so placing here. It looks like `starting` is an unused boolean; and I think we do not need `forceBlockingFlush` anymore.
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -407,7 +416,7 @@ public Searcher searcherFor(ReadCommand command)
return null;
}
}
- else if (e.column() == AccordKeyspace.CommandsColumns.store_id && e.operator() == Operator.EQ)
+ else if (e.column() == AccordJournalTable.SyntheticColumn.store_id.metadata && e.operator() == Operator.EQ)
Review Comment:
Shouldw e maybe forbid anything except `EQ` for the store?
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
Review Comment:
In this class, in `#getInitializationTask`, there is `forceBlockingFlush` task. Journal tables do not have memtables, so I would either remove this code, or replace it with the logic that would assert that memtables are empty.
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
Review Comment:
(unresolved because of the pending comments)
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ // wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
Review Comment:
I understand; I am just saying that this is OK as a dev solution, but we need to actually solve this problem. Would you mind to convert this into `TODO (now)` if you are not going to include the fix to this patch?
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -494,6 +503,50 @@ NavigableSet<ByteBuffer> search(int storeId,
};
}
+ @Override
+ public void handleNotification(INotification notification, Object sender)
+ {
+ // unfortunately, we can only check the type of notification via instanceof :(
+ if (notification instanceof SSTableAddedNotification)
+ {
+ SSTableAddedNotification notice = (SSTableAddedNotification) notification;
+ sstableManager.onSSTableChanged(Collections.emptySet(), notice.added);
+ }
+ else if (notification instanceof SSTableListChangedNotification)
+ {
+ SSTableListChangedNotification notice = (SSTableListChangedNotification) notification;
+ sstableManager.onSSTableChanged(notice.removed, notice.added);
+ }
+ else if (notification instanceof MemtableRenewedNotification)
Review Comment:
I _think_ we should never get any of these. Should we try throwing on memtable renewed and discarded?
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
Review Comment:
What would generally happen if the index is non-queriable? We probably need to throw, since if there's a chance we can return empty resultset, it can be a severe correctness violation.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "ifesdjeen (via GitHub)" <gi...@apache.org>.
ifesdjeen commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1910450616
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ // wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
Review Comment:
I think we need to investigate the race conditions rather than try to patch it here: if after waiting for `isSwitched` we still have lingering references, we best check which ones are there, and let them be scheduled appropriately.
##########
src/java/org/apache/cassandra/service/accord/IJournal.java:
##########
@@ -37,4 +41,12 @@ default SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId,
}
Persister<DurableBefore, DurableBefore> durableBeforePersister();
+
+ RangeSearcher rangeSearcher();
+
+ interface RangeSearcher
Review Comment:
This probably will go during rebase, since we do not have a separate interface in the integration anymore. You can just pull this to the top level on rebase, or make it as a part of one of the searcher subclasses. Either way works for me.
##########
src/java/org/apache/cassandra/service/accord/ParticipantsInMemoryIndex.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.local.StoreParticipants;
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.ParticipantsJournalIndex;
+import org.apache.cassandra.index.accord.RouteIndexFormat;
+import org.apache.cassandra.journal.Journal;
+import org.apache.cassandra.journal.RecordPointer;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class ParticipantsInMemoryIndex<K extends JournalKey, V> implements Journal.Listener<K, V>, IJournal.RangeSearcher
+{
+ private final Long2ObjectHashMap<ParticipantsInMemorySegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ @Override
+ public void onWrite(K id, Journal.Writer writer, Set<Integer> hosts, RecordPointer pointer)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ SavedCommand.Writer saveCommandWriter = (SavedCommand.Writer) writer;
+ if (!saveCommandWriter.hasField(SavedCommand.Fields.PARTICIPANTS))
+ return;
+ StoreParticipants participants = saveCommandWriter.after().participants();
+ Route<?> route = participants.route();
+ if (route != null)
+ update(pointer.segment, id.commandStoreId, id.id, route);
+ }
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, ParticipantsInMemorySegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public void update(long segment, K id, ByteBuffer buffer, int userVersion)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ var participants = RouteIndexFormat.extract(id.id, buffer, userVersion).participants();
+ if (participants == null || participants.route() == null)
+ return;
+ update(segment, id.commandStoreId, id.id, participants.route());
+ }
+
+ @Override
+ public synchronized void onCompact(Collection<StaticSegment<K, V>> oldSegments, Collection<StaticSegment<K, V>> compactedSegments)
+ {
+ oldSegments.forEach(s -> segmentIndexes.remove(s.id()));
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey key)
+ {
+ return search(storeId, key.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(key));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] key)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey start, AccordRoutingKey end)
+ {
+ return search(storeId, start.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(start), OrderedRouteSerializer.serializeRoutingKeyNoTable(end));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] start, byte[] end)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public synchronized void truncateForTesting()
+ {
+ segmentIndexes.clear();
+ }
+
+ @Override
+ public void intersects(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ var result = search(commandStoreId, range.start(), range.end());
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ @Override
+ public void intersects(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ var result = search(commandStoreId, key);
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ private void consume(Iterator<TxnId> it, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ while (it.hasNext())
+ {
+ TxnId next = it.next();
+ if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0)
+ forEach.accept(next);
+ }
+ }
+
+ private static class ParticipantsInMemorySegmentIndex
+ {
+ private final Int2ObjectHashMap<ParticipantsInMemoryStoreIndex> storeIndexes = new Int2ObjectHashMap<>();
+
+ private ParticipantsInMemorySegmentIndex(long segment)
+ {
+ }
+
+ public void add(int commandStoreId, TxnId id, Route<?> route)
+ {
+ storeIndexes.computeIfAbsent(commandStoreId, ParticipantsInMemoryStoreIndex::new).add(id, route);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ ParticipantsInMemoryStoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, start, end, fn);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ ParticipantsInMemoryStoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, key, fn);
+ }
+ }
+
+ private static class ParticipantsInMemoryStoreIndex
Review Comment:
Since we are already scoped in `ParticipantsInMemoryIndex`, might be best to just call this `StoreIndex`
##########
src/java/org/apache/cassandra/service/accord/ParticipantsInMemoryIndex.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.local.StoreParticipants;
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.ParticipantsJournalIndex;
+import org.apache.cassandra.index.accord.RouteIndexFormat;
+import org.apache.cassandra.journal.Journal;
+import org.apache.cassandra.journal.RecordPointer;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class ParticipantsInMemoryIndex<K extends JournalKey, V> implements Journal.Listener<K, V>, IJournal.RangeSearcher
+{
+ private final Long2ObjectHashMap<ParticipantsInMemorySegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ @Override
+ public void onWrite(K id, Journal.Writer writer, Set<Integer> hosts, RecordPointer pointer)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ SavedCommand.Writer saveCommandWriter = (SavedCommand.Writer) writer;
+ if (!saveCommandWriter.hasField(SavedCommand.Fields.PARTICIPANTS))
+ return;
+ StoreParticipants participants = saveCommandWriter.after().participants();
+ Route<?> route = participants.route();
+ if (route != null)
+ update(pointer.segment, id.commandStoreId, id.id, route);
+ }
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, ParticipantsInMemorySegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public void update(long segment, K id, ByteBuffer buffer, int userVersion)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ var participants = RouteIndexFormat.extract(id.id, buffer, userVersion).participants();
+ if (participants == null || participants.route() == null)
+ return;
+ update(segment, id.commandStoreId, id.id, participants.route());
+ }
+
+ @Override
+ public synchronized void onCompact(Collection<StaticSegment<K, V>> oldSegments, Collection<StaticSegment<K, V>> compactedSegments)
+ {
+ oldSegments.forEach(s -> segmentIndexes.remove(s.id()));
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey key)
+ {
+ return search(storeId, key.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(key));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] key)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey start, AccordRoutingKey end)
+ {
+ return search(storeId, start.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(start), OrderedRouteSerializer.serializeRoutingKeyNoTable(end));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] start, byte[] end)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public synchronized void truncateForTesting()
+ {
+ segmentIndexes.clear();
+ }
+
+ @Override
+ public void intersects(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ var result = search(commandStoreId, range.start(), range.end());
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ @Override
+ public void intersects(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ var result = search(commandStoreId, key);
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ private void consume(Iterator<TxnId> it, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ while (it.hasNext())
+ {
+ TxnId next = it.next();
+ if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0)
+ forEach.accept(next);
+ }
+ }
+
+ private static class ParticipantsInMemorySegmentIndex
+ {
+ private final Int2ObjectHashMap<ParticipantsInMemoryStoreIndex> storeIndexes = new Int2ObjectHashMap<>();
+
+ private ParticipantsInMemorySegmentIndex(long segment)
+ {
+ }
+
+ public void add(int commandStoreId, TxnId id, Route<?> route)
+ {
+ storeIndexes.computeIfAbsent(commandStoreId, ParticipantsInMemoryStoreIndex::new).add(id, route);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ ParticipantsInMemoryStoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, start, end, fn);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ ParticipantsInMemoryStoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, key, fn);
+ }
+ }
+
+ private static class ParticipantsInMemoryStoreIndex
+ {
+ private final Map<TableId, ParticipantsInMemoryTableIndex> tableIndex = new HashMap<>();
+
+ private ParticipantsInMemoryStoreIndex(int commandStoreId)
+ {
+ }
+
+ public void add(TxnId id, Route<?> route)
+ {
+ for (var keyOrRange : route)
+ add(id, keyOrRange);
+ }
+
+ private void add(TxnId id, Unseekable keyOrRange)
+ {
+ if (keyOrRange.domain() != Routable.Domain.Range)
+ throw new IllegalArgumentException("Unexpected domain: " + keyOrRange.domain());
+ TokenRange ts = (TokenRange) keyOrRange;
+ TableId tableId = ts.table();
+ tableIndex.computeIfAbsent(tableId, ParticipantsInMemoryTableIndex::new).add(id, ts);
+ }
+
+ public void search(TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ var index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(start, end, fn);
+ }
+
+ public void search(TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ var index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(key, fn);
+ }
+ }
+
+ private static class ParticipantsInMemoryTableIndex
Review Comment:
Same as above, just `TableIndex`
##########
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java:
##########
@@ -216,11 +207,12 @@ public void test()
{
cfs().disableAutoCompaction(); // let the test control compaction
//TODO (coverage): include with the ability to mark ranges as durable for compaction cleanup
- AccordService.unsafeSetNoop(); // disable accord service since compaction touches it. It would be nice to include this for cleanup support....
- stateful().withExamples(50).check(commands(() -> State::new, i -> cfs())
- .destroySut(sut -> sut.truncateBlocking())
- .add(FLUSH)
+ stateful().withExamples(10).withSteps(500).check(commands(() -> State::new, Sut::new)
+ .destroyState(State::close)
+ .destroySut(Sut::close)
+ .addIf(State::mayFlush, FLUSH)
Review Comment:
May I suggest a slight change in semantics for testing a bit?
```
diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
index 6807269c95..a4fc38a708 100644
--- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
@@ -160,7 +160,6 @@ public class RouteIndexTest extends CQLTester.InMemory
private Command<State, Sut, ?> insert(RandomSource rs, State state)
{
-// int storeId = rs.nextInt(0, state.numStores);
Domain domain = state.domainGen.next(rs);
TxnId txnId = state.nextTxnId(domain);
Route<?> route = createRoute(state, rs, domain, rs.nextInt(1, 20));
@@ -210,8 +209,9 @@ public class RouteIndexTest extends CQLTester.InMemory
stateful().withExamples(10).withSteps(500).check(commands(() -> State::new, Sut::new)
.destroyState(State::close)
.destroySut(Sut::close)
- .addIf(State::mayFlush, FLUSH)
- .add(COMPACT)
+ .add(CLOSE)
+ .add(FLUSH)
+ .add(PURGE)
.add(RESTART)
.add(this::insert)
.add(RouteIndexTest::rangeSearch)
@@ -378,7 +378,7 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
- private static final CassandraCommand FLUSH = new CassandraCommand("Flush")
+ private static final CassandraCommand CLOSE = new CassandraCommand("Close")
{
@Override
public void runUnit(Sut sut)
@@ -387,24 +387,24 @@ public class RouteIndexTest extends CQLTester.InMemory
}
};
- private static final CassandraCommand COMPACT = new CassandraCommand("Compact")
+ private static final CassandraCommand FLUSH = new CassandraCommand("Flush")
{
@Override
public void runUnit(Sut sut)
{
sut.journal.get().runCompactorForTesting();
- try
- {
- sut.cfs.enableAutoCompaction();
- FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(sut.cfs));
- }
- finally
- {
- sut.cfs.disableAutoCompaction();
- }
}
};
+ private static final CassandraCommand PURGE = new CassandraCommand("Purge")
+ {
+ @Override
+ public void runUnit(Sut sut)
+ {
+ sut.journal.get().purge(sut.stores.get());
+ }
+ };
+
private static final UnitCommand<State, Sut> RESTART = new UnitCommand<State, Sut>()
{
@Override
@@ -544,10 +544,12 @@ public class RouteIndexTest extends CQLTester.InMemory
{
private final ColumnFamilyStore cfs;
private final Supplier<AccordJournal> journal;
+ private final Supplier<CommandStores> stores;
public Sut(State state)
{
cfs = cfs();
+ this.stores = () -> state.accordService.node().commandStores();
this.journal = () -> state.accordService.journal();
```
In short, the idea here is that Journal's flow is a bit different: we first _flush_ multiple segments into one SSTable, then we _compact_ multiple SSTables together (which would be great to have as a separate event), and then we _purge_ SSTables (i.e. compact them and clean up / throw away keys that state machine tells us we can throw away).
##########
src/java/org/apache/cassandra/service/accord/ParticipantsInMemoryIndex.java:
##########
@@ -0,0 +1,286 @@
+/*
+ * 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;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.local.StoreParticipants;
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.ParticipantsJournalIndex;
+import org.apache.cassandra.index.accord.RouteIndexFormat;
+import org.apache.cassandra.journal.Journal;
+import org.apache.cassandra.journal.RecordPointer;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class ParticipantsInMemoryIndex<K extends JournalKey, V> implements Journal.Listener<K, V>, IJournal.RangeSearcher
+{
+ private final Long2ObjectHashMap<ParticipantsInMemorySegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ @Override
+ public void onWrite(K id, Journal.Writer writer, Set<Integer> hosts, RecordPointer pointer)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ SavedCommand.Writer saveCommandWriter = (SavedCommand.Writer) writer;
+ if (!saveCommandWriter.hasField(SavedCommand.Fields.PARTICIPANTS))
+ return;
+ StoreParticipants participants = saveCommandWriter.after().participants();
+ Route<?> route = participants.route();
+ if (route != null)
+ update(pointer.segment, id.commandStoreId, id.id, route);
+ }
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, ParticipantsInMemorySegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public void update(long segment, K id, ByteBuffer buffer, int userVersion)
+ {
+ if (!ParticipantsJournalIndex.allowed(id))
+ return;
+ var participants = RouteIndexFormat.extract(id.id, buffer, userVersion).participants();
Review Comment:
nit: var
##########
src/java/org/apache/cassandra/service/accord/ParticipantsInMemoryIndex.java:
##########
Review Comment:
There are a few `var`s in this file; would you mind to ask Idea to replace them with full types?
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
Review Comment:
There is a TODO saying `flesh this stuff out...` in the middle of this class. Does this still apply? It is unlikely we will return to this class, or at least this is not on the roadmap; maybe best to check this out now or at least describe what fleshing out means?
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -124,6 +126,36 @@ public class Journal<K, V> implements Shutdownable
final OpOrder readOrder = new OpOrder();
+ public interface Listener<K, V>
+ {
+ void onWrite(K id, Writer writer, Set<Integer> hosts, RecordPointer pointer);
Review Comment:
We already have a `flush` notification, which is probably the only thing we need to have in a Journal API. Moreover, this functionality probably belongs only to `AccordJournal`, and not to Journal. One of the options would be to update in-memory index on flush (in `saveCommand`).
As regards compaction, I think the best integration point is `AccordSegmentCompactor`, since we have both old and new sets there, and it is accord-only, so we are not modifying the Journal API.
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
Review Comment:
Is there any reason not to use `discarding.isEmpty()`?
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917524125
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -108,42 +111,57 @@ public enum RegisterStatus
private volatile boolean initBuildStarted = false;
private volatile RegisterStatus registerStatus = RegisterStatus.PENDING;
- public RouteIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ public ParticipantsJournalIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
{
- if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
- if (!AccordKeyspace.COMMANDS.equals(baseCfs.name))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
-
- TableMetadata tableMetadata = baseCfs.metadata();
- Pair<ColumnMetadata, IndexTarget.Type> target = TargetParser.parse(tableMetadata, indexMetadata);
- if (!AccordKeyspace.CommandsColumns.participants.name.equals(target.left.name))
- throw new IllegalArgumentException("Attempted to index the wrong column; needed " + AccordKeyspace.CommandsColumns.participants.name + " but given " + target.left.name);
-
- if (target.right != IndexTarget.Type.VALUES)
- throw new IllegalArgumentException("Attempted to index " + AccordKeyspace.CommandsColumns.participants.name + " with index type " + target.right + "; only " + IndexTarget.Type.VALUES + " is supported");
+ validateTargets(baseCfs, indexMetadata);
this.baseCfs = baseCfs;
+ // type is only IndexTarget.Type.VALUES
+ this.record = AccordKeyspace.JournalColumns.record;
+ this.user_version = AccordKeyspace.JournalColumns.user_version;
this.indexMetadata = indexMetadata;
+
this.memtableIndexManager = new RouteMemtableIndexManager(this);
this.sstableManager = new RouteSSTableManager();
this.indexMetrics = new IndexMetrics(this);
- this.column = target.left;
Tracker tracker = baseCfs.getTracker();
tracker.subscribe(this);
}
- public ColumnMetadata column()
+ public static boolean allowed(JournalKey id)
+ {
+ return id.type == JournalKey.Type.COMMAND_DIFF && allowed(id.id);
+ }
+
+ public static boolean allowed(TxnId id)
{
- return column;
+ return id.domain().isRange();
+ }
+
+ private static void validateTargets(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ {
+ // this contains 2 columns....
+ if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ if (!AccordKeyspace.JOURNAL.equals(baseCfs.name))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ Set<String> columns = Splitter.on(',').trimResults().omitEmptyStrings().splitToStream(indexMetadata.options.get("target")).collect(Collectors.toSet());
+ Set<String> expected = Set.of("record", "user_version");
+ if (!expected.equals(columns))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table, and on the record/user_value columns; given " + baseCfs.metadata() + " and columns " + columns);
}
public IndexMetrics indexMetrics()
{
return indexMetrics;
}
+ public RegisterStatus registerStatus()
+ {
+ return registerStatus;
+ }
+
Review Comment:
> and I think we do not need forceBlockingFlush anymore.
personally rather keep it... as i try to bake in the assumption memtable doesn't change into this index things start to break... this is part of the correctness of table 2i so still prefer to keep.
We do not have any logic right now that blocks an insert/update/delete into the table... so why bake this into the index itself which could actually handle those?
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913854253
##########
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java:
##########
@@ -216,11 +207,12 @@ public void test()
{
cfs().disableAutoCompaction(); // let the test control compaction
//TODO (coverage): include with the ability to mark ranges as durable for compaction cleanup
- AccordService.unsafeSetNoop(); // disable accord service since compaction touches it. It would be nice to include this for cleanup support....
- stateful().withExamples(50).check(commands(() -> State::new, i -> cfs())
- .destroySut(sut -> sut.truncateBlocking())
- .add(FLUSH)
+ stateful().withExamples(10).withSteps(500).check(commands(() -> State::new, Sut::new)
+ .destroyState(State::close)
+ .destroySut(Sut::close)
+ .addIf(State::mayFlush, FLUSH)
Review Comment:
done
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917166128
##########
src/java/org/apache/cassandra/service/accord/AccordCache.java:
##########
@@ -1190,7 +1190,7 @@ public Runnable save(AccordCommandStore commandStore, TxnId txnId, @Nullable Com
return null;
}
- return commandStore.appendToKeyspace(txnId, value);
+ return null;
Review Comment:
yep
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1887727147
##########
src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java:
##########
@@ -67,6 +67,8 @@ public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment
KeyOrderReader<JournalKey> reader = segment.keyOrderReader();
if (reader.advance())
readers.add(reader);
+ else
+ reader.close();
Review Comment:
another leak detected in the test
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884692070
##########
src/java/org/apache/cassandra/service/accord/AccordService.java:
##########
@@ -1249,14 +1251,17 @@ public CompactionInfo getCompactionInfo()
Int2ObjectHashMap<RedundantBefore> redundantBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<DurableBefore> durableBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<RangesForEpoch> ranges = new Int2ObjectHashMap<>();
- AsyncChains.getBlockingAndRethrow(node.commandStores().forEach(safeStore -> {
- synchronized (redundantBefores)
- {
- redundantBefores.put(safeStore.commandStore().id(), safeStore.redundantBefore());
- ranges.put(safeStore.commandStore().id(), safeStore.ranges());
- durableBefores.put(safeStore.commandStore().id(), safeStore.durableBefore());
- }
- }));
+ if (node.commandStores().all().length > 0)
Review Comment:
in my tests we are testing journal but we don't have any tables marked for accord, so we had 0 stores... compaction tests were also impacted as we could have data but due to us dropping accord we no longer have stores
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "ifesdjeen (via GitHub)" <gi...@apache.org>.
ifesdjeen commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1918072293
##########
src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.impl.CommandChange.Field;
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.RouteJournalIndex;
+import org.apache.cassandra.journal.Journal;
+import org.apache.cassandra.journal.RecordPointer;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class RouteInMemoryIndex<K extends JournalKey, V> implements RangeSearcher
+{
+ private final Long2ObjectHashMap<SegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ public boolean isSupported(JournalKey id)
+ {
+ return RouteJournalIndex.allowed(id);
+ }
+
+ public void onWrite(JournalKey id, Journal.Writer writer, RecordPointer pointer)
+ {
+ if (!RouteJournalIndex.allowed(id))
+ return;
+ AccordJournal.Writer saveCommandWriter = (AccordJournal.Writer) writer;
+ if (!saveCommandWriter.hasField(Field.PARTICIPANTS))
+ return;
+ Route<?> route = saveCommandWriter.after.participants().route();
+ if (route != null)
+ update(pointer.segment, id.commandStoreId, id.id, route);
+ }
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!RouteJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, SegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public synchronized void onCompact(Collection<StaticSegment<JournalKey, V>> oldSegments)
+ {
+ // As of this writing compact in accord journal takes StaticSegments, writes them to a SSTable, and pushes to a table;
+ // it then stops managing those segments... for this reason compactedSegments is normally empty and none of the
+ // oldSegments are expected to be tracked anymore, so this index should remove the reference (there is normal table 2i to pick up the job)
+ oldSegments.forEach(s -> segmentIndexes.remove(s.id()));
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey key)
+ {
+ return search(storeId, key.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(key));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] key)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey start, AccordRoutingKey end)
+ {
+ return search(storeId, start.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(start), OrderedRouteSerializer.serializeRoutingKeyNoTable(end));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] start, byte[] end)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public synchronized void truncateForTesting()
+ {
+ segmentIndexes.clear();
+ }
+
+ @Override
+ public void intersects(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ NavigableMap<IndexRange, Set<TxnId>> result = search(commandStoreId, range.start(), range.end());
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ @Override
+ public void intersects(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ NavigableMap<IndexRange, Set<TxnId>> result = search(commandStoreId, key);
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ private void consume(Iterator<TxnId> it, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ while (it.hasNext())
+ {
+ TxnId next = it.next();
+ if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0)
+ forEach.accept(next);
+ }
+ }
+
+ private static class SegmentIndex
+ {
+ private final Int2ObjectHashMap<StoreIndex> storeIndexes = new Int2ObjectHashMap<>();
+
+ private SegmentIndex(long segment)
+ {
+ }
+
+ public void add(int commandStoreId, TxnId id, Route<?> route)
+ {
+ storeIndexes.computeIfAbsent(commandStoreId, StoreIndex::new).add(id, route);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, start, end, fn);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, key, fn);
+ }
+ }
+
+ private static class StoreIndex
+ {
+ private final Map<TableId, TableIndex> tableIndex = new HashMap<>();
+
+ private StoreIndex(int commandStoreId)
Review Comment:
nit (can be aloso done on commit): store id is unused; but might be useful for debugging.
##########
src/java/org/apache/cassandra/journal/SegmentCompactor.java:
##########
@@ -30,5 +31,7 @@ static <K, V> SegmentCompactor<K, V> noop()
return (SegmentCompactor<K, V>) NOOP;
}
+ //TODO (review): org.apache.cassandra.journal.Compactor.run documents that this is nullable but no implementation returns null.... If we want to keep nullable I put it in the interface so you have to check, but if we want to remove I can go clean that assumption up.
Review Comment:
I think at some point we were returning null, which was helping to make distinction between a no-op compaction (i.e. that should leave all segments ad they were) and compaction that replaces segment collection with an empty list. Right now we always return an empty list. I suspect that other _potential_ implementations may have different opinions. Maybe @iamaleksey can comment?
##########
src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java:
##########
@@ -0,0 +1,279 @@
+/*
+ * 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;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.impl.CommandChange.Field;
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.RouteJournalIndex;
+import org.apache.cassandra.journal.Journal;
+import org.apache.cassandra.journal.RecordPointer;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class RouteInMemoryIndex<K extends JournalKey, V> implements RangeSearcher
+{
+ private final Long2ObjectHashMap<SegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ public boolean isSupported(JournalKey id)
+ {
+ return RouteJournalIndex.allowed(id);
+ }
+
+ public void onWrite(JournalKey id, Journal.Writer writer, RecordPointer pointer)
+ {
+ if (!RouteJournalIndex.allowed(id))
+ return;
+ AccordJournal.Writer saveCommandWriter = (AccordJournal.Writer) writer;
+ if (!saveCommandWriter.hasField(Field.PARTICIPANTS))
+ return;
+ Route<?> route = saveCommandWriter.after.participants().route();
+ if (route != null)
+ update(pointer.segment, id.commandStoreId, id.id, route);
+ }
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!RouteJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, SegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public synchronized void onCompact(Collection<StaticSegment<JournalKey, V>> oldSegments)
+ {
+ // As of this writing compact in accord journal takes StaticSegments, writes them to a SSTable, and pushes to a table;
+ // it then stops managing those segments... for this reason compactedSegments is normally empty and none of the
+ // oldSegments are expected to be tracked anymore, so this index should remove the reference (there is normal table 2i to pick up the job)
+ oldSegments.forEach(s -> segmentIndexes.remove(s.id()));
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey key)
+ {
+ return search(storeId, key.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(key));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] key)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public NavigableMap<IndexRange, Set<TxnId>> search(int storeId, AccordRoutingKey start, AccordRoutingKey end)
+ {
+ return search(storeId, start.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(start), OrderedRouteSerializer.serializeRoutingKeyNoTable(end));
+ }
+
+ private synchronized NavigableMap<IndexRange, Set<TxnId>> search(int storeId, TableId tableId, byte[] start, byte[] end)
+ {
+ TreeMap<IndexRange, Set<TxnId>> matches = new TreeMap<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, e -> matches.computeIfAbsent(e.getKey(), i -> new HashSet<>()).add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableMap() : matches;
+ }
+
+ public synchronized void truncateForTesting()
+ {
+ segmentIndexes.clear();
+ }
+
+ @Override
+ public void intersects(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ NavigableMap<IndexRange, Set<TxnId>> result = search(commandStoreId, range.start(), range.end());
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ @Override
+ public void intersects(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ NavigableMap<IndexRange, Set<TxnId>> result = search(commandStoreId, key);
+ TreeSet<TxnId> matches = new TreeSet<>();
+ result.values().forEach(s -> matches.addAll(s));
+ consume(matches.iterator(), minTxnId, maxTxnId, forEach);
+ }
+
+ private void consume(Iterator<TxnId> it, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ {
+ while (it.hasNext())
+ {
+ TxnId next = it.next();
+ if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0)
+ forEach.accept(next);
+ }
+ }
+
+ private static class SegmentIndex
+ {
+ private final Int2ObjectHashMap<StoreIndex> storeIndexes = new Int2ObjectHashMap<>();
+
+ private SegmentIndex(long segment)
+ {
+ }
+
+ public void add(int commandStoreId, TxnId id, Route<?> route)
+ {
+ storeIndexes.computeIfAbsent(commandStoreId, StoreIndex::new).add(id, route);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, start, end, fn);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, key, fn);
+ }
+ }
+
+ private static class StoreIndex
+ {
+ private final Map<TableId, TableIndex> tableIndex = new HashMap<>();
+
+ private StoreIndex(int commandStoreId)
+ {
+ }
+
+ public void add(TxnId id, Route<?> route)
+ {
+ for (Unseekable keyOrRange : route)
+ add(id, keyOrRange);
+ }
+
+ private void add(TxnId id, Unseekable keyOrRange)
+ {
+ if (keyOrRange.domain() != Routable.Domain.Range)
+ throw new IllegalArgumentException("Unexpected domain: " + keyOrRange.domain());
+ TokenRange ts = (TokenRange) keyOrRange;
+ TableId tableId = ts.table();
+ tableIndex.computeIfAbsent(tableId, TableIndex::new).add(id, ts);
+ }
+
+ public void search(TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ TableIndex index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(start, end, fn);
+ }
+
+ public void search(TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ TableIndex index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(key, fn);
+ }
+ }
+
+ private static class TableIndex
+ {
+ private final RangeTree<byte[], IndexRange, TxnId> index = createRangeTree();
+
+ private TableIndex(TableId tableId)
Review Comment:
nit: same as above with table id: unused, but might be useful as a 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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913822259
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -403,7 +418,7 @@ public void purge(CommandStores commandStores)
public void replay(CommandStores commandStores)
{
journal.closeCurrentSegmentForTestingIfNonEmpty();
- try (AccordJournalTable.KeyOrderIterator<JournalKey> iter = journalTable.readAll())
+ try (AccordJournalTable.KeyOrderIterator<JournalKey> iter = journalTable.readAll(AccordJournalTable.ReadAllIntent.Reply))
Review Comment:
this is less critical after the rebase as the `checkAll` function got removed... when we are reading for validation we don't want to populate the index... I left this here to be explicit about this assumption but I can remove.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1887726662
##########
test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java:
##########
@@ -796,18 +832,30 @@ else if (mode == MapMode.READ_WRITE)
long pos = position;
try
{
- while (local.hasRemaining())
+ // the channel could be closed... so always create a new channel to avoid this problem
+ try (FileChannel channel = provider().newFileChannel(path, Set.of(StandardOpenOption.WRITE)))
Review Comment:
I am finding cases where the journal file is deleted *before* we finish writing to it... which is fine from a FS point of view... so to better handle that without failing due to the channel being closed, I moved to always opening a new channel for the write... if the file doesn't exist we just no-op (which is the behavior on real FS anyways
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917522278
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
@@ -494,6 +503,50 @@ NavigableSet<ByteBuffer> search(int storeId,
};
}
+ @Override
+ public void handleNotification(INotification notification, Object sender)
+ {
+ // unfortunately, we can only check the type of notification via instanceof :(
+ if (notification instanceof SSTableAddedNotification)
+ {
+ SSTableAddedNotification notice = (SSTableAddedNotification) notification;
+ sstableManager.onSSTableChanged(Collections.emptySet(), notice.added);
+ }
+ else if (notification instanceof SSTableListChangedNotification)
+ {
+ SSTableListChangedNotification notice = (SSTableListChangedNotification) notification;
+ sstableManager.onSSTableChanged(notice.removed, notice.added);
+ }
+ else if (notification instanceof MemtableRenewedNotification)
Review Comment:
i tried that from your patch and this breaks `TRUNCATE`.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917521139
##########
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java:
##########
Review Comment:
we index routes, and we query by routes... so I feel the name still works. `participant` is a impl detail, so renamed the index to make this clear
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917517216
##########
src/java/org/apache/cassandra/index/accord/ParticipantsJournalIndex.java:
##########
Review Comment:
> In this class, in #getInitializationTask, there is forceBlockingFlush task. Journal tables do not have memtables, so I would either remove this code, or replace it with the logic that would assert that memtables are empty.
So, memtable and 2i requires these stuff, so when i started asserting that memtables are never touched some tests started to fail... stuff like `TRUNCATE` break... i rather not try to block at this layer as it causes issues.
> What would generally happen if the index is non-queriable? We probably need to throw, since if there's a chance we can return empty resultset, it can be a severe correctness violation.
on startup this won't be an issue, this would only be an issue if the user does `nodetool rebuildindex system_accord.journal`. If the index isn't up we kinda just broke the clusters ability to function...
For now ill add a check that the index is up
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913836667
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -124,6 +126,36 @@ public class Journal<K, V> implements Shutdownable
final OpOrder readOrder = new OpOrder();
+ public interface Listener<K, V>
+ {
+ void onWrite(K id, Writer writer, Set<Integer> hosts, RecordPointer pointer);
Review Comment:
moved to `AccordJournal`
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884691443
##########
src/java/org/apache/cassandra/service/accord/AccordSegmentCompactor.java:
##########
@@ -126,6 +126,7 @@ public Collection<StaticSegment<JournalKey, V>> compact(Collection<StaticSegment
while ((advanced = reader.advance()) && reader.key().equals(key));
if (advanced) readers.offer(reader); // there is more to this reader, but not with this key
+ else reader.close();
Review Comment:
this leak detection was caught by `org.apache.cassandra.index.accord.RouteIndexTest#test`. The fs would run out of space and when you looked it had very few files and only ~40mb of data, yet 4g was allocated! The root cause was that each time we open a reader we create a new channel and didn't close it, so we could never purge segments.
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884689629
##########
src/java/org/apache/cassandra/service/accord/AccordCache.java:
##########
@@ -1190,7 +1190,7 @@ public Runnable save(AccordCommandStore commandStore, TxnId txnId, @Nullable Com
return null;
}
- return commandStore.appendToKeyspace(txnId, value);
+ return null;
Review Comment:
to lower the diff I didn't want to rewrite the whole cache logic... `null` is the only thing possible to return now
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884694931
##########
test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java:
##########
@@ -54,7 +54,7 @@ public void partitionerToToken()
@Test
public void partitionerKeys()
{
- qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners()))
+ qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners().map(CassandraGenerators::simplify)))
Review Comment:
flakey test during my first CI run. `LocalPartitioner` can have very complex type, which leads to tokens too large causing the test to be flakey
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell closed pull request #3743: First dumb of logic to get route index running on the journal
URL: https://github.com/apache/cassandra/pull/3743
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#issuecomment-2599395346
merged https://github.com/apache/cassandra/commit/d834c5d6340b27521d938f18adafe9f6ebb89d87
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913816945
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
Review Comment:
logically no, but I think I was going with what I could validate better... ill use `isEmpty` and if things pop up we can fix
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884694660
##########
test/unit/org/apache/cassandra/dht/IPartitionerTest.java:
##########
@@ -93,7 +94,9 @@ private boolean isTestType(Class<? extends IPartitioner> klass)
@Test
public void byteCompareSerde()
{
- qt().forAll(AccordGenerators.fromQT(CassandraGenerators.token())).check(token -> {
+ // make sure to use simplify as local partitioner can have a type that could generate data too large causing this test to be flakey
+ Gen<Token> qt = CassandraGenerators.partitioners().flatMap(p -> CassandraGenerators.token(CassandraGenerators.simplify(p)));
Review Comment:
flakey test during my first CI run. `LocalPartitioner` can have very complex type, which leads to tokens too large causing the test to be flakey
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1887724012
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -939,6 +985,8 @@ private StaticSegmentIterator()
StaticSegment.KeyOrderReader<K> reader = staticSegment.keyOrderReader();
if (reader.advance())
this.readers.add(reader);
+ else
+ reader.close();
Review Comment:
file leak found in testing
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -962,6 +1010,8 @@ public void readAllForKey(K key, RecordConsumer<K> reader)
reader.accept(next.descriptor.timestamp, next.offset, next.key(), next.record(), next.hosts(), next.descriptor.userVersion);
if (next.advance())
readers.add(next);
+ else
+ next.close();
Review Comment:
file leak found in testing
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1887728530
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
Review Comment:
I still don't feel this is safe... but its been stable for me...
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884692674
##########
src/java/org/apache/cassandra/service/accord/IndexRange.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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;
+
+import org.apache.cassandra.cache.IMeasurableMemory;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.ObjectSizes;
+
+public class IndexRange implements Comparable<IndexRange>, IMeasurableMemory
Review Comment:
this was just extracted from the previous impl and put to the top level. With journal and tables both touching this, it was easier to make top level rather than an inner class
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1884693062
##########
test/unit/org/apache/cassandra/concurrent/ForwardingExecutorFactory.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.concurrent;
+
+public class ForwardingExecutorFactory implements ExecutorFactory
Review Comment:
used by `RouteIndexTest` to make the close/release threads immediately run rather than async, this avoided race condition issues with validation
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1913820477
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ // wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
Review Comment:
its been awhile but we do stuff like release/close file references in different threads... this pattern of constantly pushing things cross threads was what I remember caused this...
--
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] First dumb of logic to get route index running on the journal [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3743:
URL: https://github.com/apache/cassandra/pull/3743#discussion_r1917511632
##########
src/java/org/apache/cassandra/journal/Journal.java:
##########
@@ -906,8 +941,19 @@ private String maybeAddDiskSpaceContext(String message)
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (discarding.index.size() > 0) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ // wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
Review Comment:
done
--
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