jberragan commented on code in PR #98:
URL:
https://github.com/apache/cassandra-analytics/pull/98#discussion_r1945290784
##########
cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java:
##########
@@ -264,6 +276,244 @@ public CompressionUtil compressionUtil()
return CompressionUtilImplementation.INSTANCE;
}
+ @Override
+ public long lastRepairTime(String keyspace, String table, SSTable ssTable)
throws IOException
+ {
+ Map<MetadataType, MetadataComponent> componentMap =
ReaderUtils.deserializeStatsMetadata(keyspace, table, ssTable,
EnumSet.of(MetadataType.STATS));
+ StatsMetadata statsMetadata = (StatsMetadata)
componentMap.get(MetadataType.STATS);
+ if (statsMetadata == null)
+ {
+ throw new IllegalStateException("Could not read StatsMetadata");
+ }
+ return statsMetadata.repairedAt;
+ }
+
+ @Override
+ public Pair<BigInteger, BigInteger> firstLastToken(SSTable ssTable,
Partitioner partitioner, int minIndexInterval, int maxIndexInterval) throws
IOException
+ {
+ IPartitioner iPartitioner = getPartitioner(partitioner);
+
+ // attempt Summary.db file first
+ Pair<BigInteger, BigInteger> firstAndLast = null;
+ try
+ {
+ SummaryDbUtils.Summary summary =
SummaryDbUtils.readSummary(ssTable, iPartitioner, minIndexInterval,
maxIndexInterval);
+ if (summary != null)
+ {
+ BigInteger first =
TokenUtils.tokenToBigInteger(summary.first().getToken());
+ BigInteger last =
TokenUtils.tokenToBigInteger(summary.last().getToken());
+ firstAndLast = Pair.of(first, last);
+ }
+ }
+ catch (IOException e)
+ {
+ // this can happen if the minIndexInterval and maxIndexInterval do
not match the expected serialized in the Summary.db file
+ LOGGER.warn("IOException reading Summary.db file", e);
+ }
+
+ if (firstAndLast == null)
+ {
+ // try Index.db file
+ LOGGER.warn("Failed to read Summary.db file, falling back to
Index.db file");
+ Pair<DecoratedKey, DecoratedKey> keys =
ReaderUtils.keysFromIndex(iPartitioner, ssTable);
+ Token first = keys.left.getToken();
+ Token last = keys.right.getToken();
+ if (first == null || last == null)
+ {
+ throw new RuntimeException("Unable to extract first and last
keys from Summary.db or Index.db");
+ }
+ firstAndLast = Pair.of(TokenUtils.tokenToBigInteger(first),
TokenUtils.tokenToBigInteger(last));
+ }
+
+ return firstAndLast;
+ }
+
+ @Override
+ public List<Boolean> overlaps(SSTable ssTable,
+ Partitioner partitioner,
+ int minIndexInterval,
+ int maxIndexInterval,
+ List<TokenRange> ranges) throws IOException
+ {
+ Pair<BigInteger, BigInteger> firstAndLastKeys =
firstLastToken(ssTable, partitioner, minIndexInterval, maxIndexInterval);
+ TokenRange sstableRange = TokenRange.closed(firstAndLastKeys.left,
firstAndLastKeys.right);
+ return ranges.stream()
+ .map(range -> range.isConnected(sstableRange))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public List<BigInteger> toTokens(Partitioner partitioner, @NotNull
List<ByteBuffer> partitionKeys)
+ {
+ IPartitioner iPartitioner = getPartitioner(partitioner);
+ return partitionKeys
+ .stream()
+ .map(key -> {
+ DecoratedKey decoratedKey = iPartitioner.decorateKey(key);
+ return
TokenUtils.tokenToBigInteger(decoratedKey.getToken());
+ })
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public List<ByteBuffer> encodePartitionKeys(Partitioner partitioner,
String keyspace, String createTableStmt, List<List<String>> keys)
+ {
+ CqlTable table = new SchemaBuilder(createTableStmt, keyspace,
ReplicationFactor.simple(1), partitioner).build();
+ return keys.stream().map(key -> buildPartitionKey(table,
key)).collect(Collectors.toList());
+ }
+
+ @Override
+ public List<Boolean> maybeContains(Partitioner partitioner,
+ String keyspace,
+ String table,
+ SSTable ssTable,
+ List<ByteBuffer> partitionKeys) throws
IOException
+ {
+ if (partitionKeys.isEmpty())
+ {
+ return List.of();
+ }
+ IPartitioner iPartitioner = getPartitioner(partitioner);
+ Descriptor descriptor = ReaderUtils.constructDescriptor(keyspace,
table, ssTable);
+ return maybeContains(descriptor, ssTable,
partitionKeys.stream().map(iPartitioner::decorateKey).collect(Collectors.toList()));
+ }
+
+ private List<Boolean> maybeContains(Descriptor descriptor, SSTable
ssTable, Collection<DecoratedKey> partitionKeys) throws IOException
+ {
+ if (partitionKeys.isEmpty())
+ {
+ return List.of();
+ }
+ BloomFilter filter = ReaderUtils.readFilter(ssTable, descriptor);
Review Comment:
`BloomFilter` is closable but it is because it inherits the Cassandra
reference counting code from `SharedCloseableImpl`, but we do not use this
reference counting code in Analytics . We have seen SIGSEGV errors calling
closable on `SharedCloseableImpl` instances which is why `BloomFilter` is not
closed anywhere in the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
