adelapena commented on code in PR #2540: URL: https://github.com/apache/cassandra/pull/2540#discussion_r1294755066
########## test/unit/org/apache/cassandra/index/sai/disk/v1/keystore/KeyLookupTest.java: ########## @@ -0,0 +1,413 @@ +/* + * 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.index.sai.disk.v1.keystore; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; + +import org.apache.cassandra.db.Clustering; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.index.sai.SAITester; +import org.apache.cassandra.index.sai.disk.format.IndexComponent; +import org.apache.cassandra.index.sai.disk.format.IndexDescriptor; +import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter; +import org.apache.cassandra.index.sai.disk.v1.MetadataSource; +import org.apache.cassandra.index.sai.disk.v1.MetadataWriter; +import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesMeta; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesWriter; +import org.apache.cassandra.index.sai.utils.PrimaryKey; +import org.apache.cassandra.index.sai.utils.SAIRandomizedTester; +import org.apache.cassandra.io.util.FileHandle; +import org.apache.cassandra.utils.bytecomparable.ByteComparable; +import org.apache.cassandra.utils.bytecomparable.ByteSource; +import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse; +import org.apache.lucene.store.IndexInput; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class KeyLookupTest extends SAIRandomizedTester +{ + protected IndexDescriptor indexDescriptor; + + @Before + public void setup() throws Exception + { + indexDescriptor = newIndexDescriptor(); + } + + @Test + public void testFileValidation() throws Exception + { + List<PrimaryKey> primaryKeys = new ArrayList<>(); + + for (int x = 0; x < 11; x++) + { + ByteBuffer buffer = UTF8Type.instance.decompose(Integer.toString(x)); + DecoratedKey partitionKey = Murmur3Partitioner.instance.decorateKey(buffer); + PrimaryKey primaryKey = SAITester.TEST_FACTORY.create(partitionKey, Clustering.EMPTY); + primaryKeys.add(primaryKey); + } + + primaryKeys.sort(PrimaryKey::compareTo); + + try (MetadataWriter metadataWriter = new MetadataWriter(indexDescriptor.openPerSSTableOutput(IndexComponent.GROUP_META))) + { + IndexOutputWriter bytesWriter = indexDescriptor.openPerSSTableOutput(IndexComponent.PARTITION_KEY_BLOCKS); + NumericValuesWriter blockFPWriter = new NumericValuesWriter(indexDescriptor, IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, metadataWriter, true); + try (KeyStoreWriter writer = new KeyStoreWriter(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS), + metadataWriter, + bytesWriter, + blockFPWriter, + 4, + false)) + { + primaryKeys.forEach(primaryKey -> { + try + { + writer.add(primaryKey); + } + catch (IOException e) + { + e.printStackTrace(); + } + }); + } + } + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, false)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, false)); + } + + @Test + public void testLongPrefixesAndSuffixes() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + writeTerms(writer -> { + // The following writes a set of terms that cover the following conditions: + + // Start value 0 + byte[] bytes = new byte[20]; + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix > 15 + bytes = new byte[20]; + Arrays.fill(bytes, 16, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix == 15 + bytes = new byte[20]; + Arrays.fill(bytes, 15, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix < 15 + bytes = new byte[20]; + Arrays.fill(bytes, 14, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix > 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 4, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix == 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 5, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix < 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 6, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + + bytes = new byte[32]; + Arrays.fill(bytes, 0, 16, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix >= 15 && suffix >= 16 + bytes = new byte[32]; + Arrays.fill(bytes, 0, 32, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testNonUniqueTerms() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(5000); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointId() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointIdOutOfRange() throws Exception + { + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + withSortedTermsCursor(cursor -> { + assertThatThrownBy(() -> cursor.seekToPointId(-2)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, -2, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(Long.MAX_VALUE)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, Long.MAX_VALUE, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(4000)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, 4000, 4000)); + }); + } + + @Test + public void testSeekToTerm() throws Exception + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 4000; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, true); + + withSortedTermsCursor(cursor -> { + assertEquals(0L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)); + cursor.reset(); + assertEquals(160L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(160L)), 160L, 170L)); + cursor.reset(); + assertEquals(165L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(165L)), 160L, 170L)); + cursor.reset(); + assertEquals(175L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(175L)), 160L, 176L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 160L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 175L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(701)), 160L, 177L)); + cursor.reset(); + assertEquals(504L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(504L)), 200L, 600L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 0L, 1000L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 999L, 1000L)); + cursor.reset(); + assertEquals(999L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(999L)), 0L, 1000L)); + }); + } + + @Test + public void seekToTermOnNonPartitionedTest() throws Throwable + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 16; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, false); + + withSortedTermsCursor(cursor -> assertThatThrownBy(() -> cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)) + .isInstanceOf(AssertionError.class)); + } + + @Test + public void partitionedTermsMustBeInOrderInPartitions() throws Throwable + { + writeTerms(writer -> { + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(0))); + writer.add(ByteComparable.fixedLength(makeTerm(10))); + assertThatThrownBy(() -> writer.add(ByteComparable.fixedLength(makeTerm(9)))).isInstanceOf(IllegalArgumentException.class); + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(9))); + },true); + } + + private byte[] makeTerm(int value) + { + ByteBuffer buffer = Int32Type.instance.decompose(value); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + return ByteSourceInverse.readBytes(byteSource); + } + + private void doTestSortedTerms(List<byte[]> terms) throws Exception Review Comment: This could be named `doTestKeyLookup` ########## test/unit/org/apache/cassandra/index/sai/disk/v1/keystore/KeyLookupTest.java: ########## @@ -0,0 +1,413 @@ +/* + * 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.index.sai.disk.v1.keystore; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; + +import org.apache.cassandra.db.Clustering; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.index.sai.SAITester; +import org.apache.cassandra.index.sai.disk.format.IndexComponent; +import org.apache.cassandra.index.sai.disk.format.IndexDescriptor; +import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter; +import org.apache.cassandra.index.sai.disk.v1.MetadataSource; +import org.apache.cassandra.index.sai.disk.v1.MetadataWriter; +import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesMeta; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesWriter; +import org.apache.cassandra.index.sai.utils.PrimaryKey; +import org.apache.cassandra.index.sai.utils.SAIRandomizedTester; +import org.apache.cassandra.io.util.FileHandle; +import org.apache.cassandra.utils.bytecomparable.ByteComparable; +import org.apache.cassandra.utils.bytecomparable.ByteSource; +import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse; +import org.apache.lucene.store.IndexInput; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class KeyLookupTest extends SAIRandomizedTester +{ + protected IndexDescriptor indexDescriptor; + + @Before + public void setup() throws Exception + { + indexDescriptor = newIndexDescriptor(); + } + + @Test + public void testFileValidation() throws Exception + { + List<PrimaryKey> primaryKeys = new ArrayList<>(); + + for (int x = 0; x < 11; x++) + { + ByteBuffer buffer = UTF8Type.instance.decompose(Integer.toString(x)); + DecoratedKey partitionKey = Murmur3Partitioner.instance.decorateKey(buffer); + PrimaryKey primaryKey = SAITester.TEST_FACTORY.create(partitionKey, Clustering.EMPTY); + primaryKeys.add(primaryKey); + } + + primaryKeys.sort(PrimaryKey::compareTo); + + try (MetadataWriter metadataWriter = new MetadataWriter(indexDescriptor.openPerSSTableOutput(IndexComponent.GROUP_META))) + { + IndexOutputWriter bytesWriter = indexDescriptor.openPerSSTableOutput(IndexComponent.PARTITION_KEY_BLOCKS); + NumericValuesWriter blockFPWriter = new NumericValuesWriter(indexDescriptor, IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, metadataWriter, true); + try (KeyStoreWriter writer = new KeyStoreWriter(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS), + metadataWriter, + bytesWriter, + blockFPWriter, + 4, + false)) + { + primaryKeys.forEach(primaryKey -> { + try + { + writer.add(primaryKey); + } + catch (IOException e) + { + e.printStackTrace(); + } + }); + } + } + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, false)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, false)); + } + + @Test + public void testLongPrefixesAndSuffixes() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + writeTerms(writer -> { + // The following writes a set of terms that cover the following conditions: + + // Start value 0 + byte[] bytes = new byte[20]; + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix > 15 + bytes = new byte[20]; + Arrays.fill(bytes, 16, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix == 15 + bytes = new byte[20]; + Arrays.fill(bytes, 15, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix < 15 + bytes = new byte[20]; + Arrays.fill(bytes, 14, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix > 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 4, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix == 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 5, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix < 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 6, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + + bytes = new byte[32]; + Arrays.fill(bytes, 0, 16, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix >= 15 && suffix >= 16 + bytes = new byte[32]; + Arrays.fill(bytes, 0, 32, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testNonUniqueTerms() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(5000); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointId() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointIdOutOfRange() throws Exception + { + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + withSortedTermsCursor(cursor -> { + assertThatThrownBy(() -> cursor.seekToPointId(-2)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, -2, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(Long.MAX_VALUE)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, Long.MAX_VALUE, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(4000)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, 4000, 4000)); + }); + } + + @Test + public void testSeekToTerm() throws Exception + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 4000; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, true); + + withSortedTermsCursor(cursor -> { + assertEquals(0L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)); + cursor.reset(); + assertEquals(160L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(160L)), 160L, 170L)); + cursor.reset(); + assertEquals(165L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(165L)), 160L, 170L)); + cursor.reset(); + assertEquals(175L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(175L)), 160L, 176L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 160L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 175L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(701)), 160L, 177L)); + cursor.reset(); + assertEquals(504L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(504L)), 200L, 600L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 0L, 1000L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 999L, 1000L)); + cursor.reset(); + assertEquals(999L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(999L)), 0L, 1000L)); + }); + } + + @Test + public void seekToTermOnNonPartitionedTest() throws Throwable + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 16; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, false); + + withSortedTermsCursor(cursor -> assertThatThrownBy(() -> cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)) + .isInstanceOf(AssertionError.class)); + } + + @Test + public void partitionedTermsMustBeInOrderInPartitions() throws Throwable + { + writeTerms(writer -> { + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(0))); + writer.add(ByteComparable.fixedLength(makeTerm(10))); + assertThatThrownBy(() -> writer.add(ByteComparable.fixedLength(makeTerm(9)))).isInstanceOf(IllegalArgumentException.class); + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(9))); + },true); + } + + private byte[] makeTerm(int value) + { + ByteBuffer buffer = Int32Type.instance.decompose(value); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + return ByteSourceInverse.readBytes(byteSource); + } + + private void doTestSortedTerms(List<byte[]> terms) throws Exception + { + // iterate ascending + withSortedTermsCursor(cursor -> { + for (int x = 0; x < terms.size(); x++) + { + ByteComparable term = cursor.seekToPointId(x); + + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + + assertArrayEquals(terms.get(x), bytes); + } + }); + + // iterate ascending skipping blocks + withSortedTermsCursor(cursor -> { + for (int x = 0; x < terms.size(); x += 17) + { + ByteComparable term = cursor.seekToPointId(x); + + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + + assertArrayEquals(terms.get(x), bytes); + } + }); + + withSortedTermsCursor(cursor -> { + ByteComparable term = cursor.seekToPointId(7); + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + assertArrayEquals(terms.get(7), bytes); + + term = cursor.seekToPointId(7); + bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + assertArrayEquals(terms.get(7), bytes); + }); + } + + protected void writeTerms(ThrowingConsumer<KeyStoreWriter> testCode, boolean partitioned) throws IOException + { + try (MetadataWriter metadataWriter = new MetadataWriter(indexDescriptor.openPerSSTableOutput(IndexComponent.GROUP_META))) + { + IndexOutputWriter bytesWriter = indexDescriptor.openPerSSTableOutput(IndexComponent.PARTITION_KEY_BLOCKS); + NumericValuesWriter blockFPWriter = new NumericValuesWriter(indexDescriptor, IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, metadataWriter, true); + try (KeyStoreWriter writer = new KeyStoreWriter(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS), + metadataWriter, + bytesWriter, + blockFPWriter, + 4, + partitioned)) + { + testCode.accept(writer); + } + } + } + + @FunctionalInterface + public interface ThrowingConsumer<T> + { + void accept(T t) throws IOException; + } + + private void withSortedTermsReader(ThrowingConsumer<KeyLookup> testCode) throws IOException Review Comment: This could be named `withKeyLookup` ########## src/java/org/apache/cassandra/index/sai/disk/v1/keystore/KeyLookup.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.index.sai.disk.v1.keystore; + +import java.io.IOException; +import javax.annotation.Nonnull; +import javax.annotation.concurrent.NotThreadSafe; + +import com.google.common.annotations.VisibleForTesting; + +import org.apache.cassandra.index.sai.disk.io.IndexInputReader; +import org.apache.cassandra.index.sai.disk.v1.LongArray; +import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils; +import org.apache.cassandra.index.sai.disk.v1.bitpack.MonotonicBlockPackedReader; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesMeta; +import org.apache.cassandra.io.util.FileHandle; +import org.apache.cassandra.utils.FastByteOperations; +import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.bytecomparable.ByteComparable; +import org.apache.cassandra.utils.bytecomparable.ByteSource; +import org.apache.lucene.util.BytesRef; +import org.apache.lucene.util.BytesRefBuilder; + +/** + * Provides read access to an on-disk sequence of partition or clustering keys written by {@link KeyStoreWriter}. + * <p> + * Care has been taken to make this structure as efficient as possible. + * Reading keys does not require allocating data heap buffers per each read operation. + * Only one key at a time is loaded to memory. + * Low complexity algorithms are used – a lookup of the key by point id is constant time, + * and a lookup of the point id by the key is logarithmic. + * <p> + * Because the blocks are prefix compressed, random access applies only to the locating the whole block. + * In order to jump to a concrete key inside the block, the block keys are iterated from the block beginning. + * + * @see KeyStoreWriter + */ +@NotThreadSafe +public class KeyLookup +{ + public static final String INDEX_OUT_OF_BOUNDS = "The target point id [%d] cannot be less than 0 or greater than or equal to the key count [%d]"; + + private final FileHandle keysFileHandle; + private final KeyLookupMeta keyLookupMeta; + private final LongArray.Factory keyBlockOffsetsFactory; + + /** + * Creates a new reader based on its data components. + * <p> + * It does not own the components, so you must close them separately after you're done with the reader. + * @param keysFileHandle handle to the file with a sequence of prefix-compressed blocks + * each storing a fixed number of keys + * @param keysBlockOffsets handle to the file containing an encoded sequence of the file offsets pointing to the blocks + * @param keyLookupMeta metadata object created earlier by the writer + * @param keyBlockOffsetsMeta metadata object for the block offsets + */ + public KeyLookup(@Nonnull FileHandle keysFileHandle, + @Nonnull FileHandle keysBlockOffsets, + @Nonnull KeyLookupMeta keyLookupMeta, + @Nonnull NumericValuesMeta keyBlockOffsetsMeta) throws IOException + { + this.keysFileHandle = keysFileHandle; + this.keyLookupMeta = keyLookupMeta; + this.keyBlockOffsetsFactory = new MonotonicBlockPackedReader(keysBlockOffsets, keyBlockOffsetsMeta); + } + + /** + * Opens a cursor over the keys stored in the keys file. + * <p> + * This will read the first key into the key buffer and point to the first point in the keys file. + * <p> + * The cursor is to be used in a single thread. + * The cursor is valid as long this object hasn't been closed. + * You must close the cursor when you no longer need it. + */ + public @Nonnull Cursor openCursor() throws IOException + { + return new Cursor(keysFileHandle, keyBlockOffsetsFactory); + } + + /** + * Allows reading the keys from the keys file. + * Can quickly seek to a random key by point id. + * <p> + * This object is stateful and not thread safe. + * It maintains a position to the current key as well as a buffer that can hold one key. + */ + @NotThreadSafe + public class Cursor implements AutoCloseable + { + private final IndexInputReader keysInput; + private final int blockShift; + private final int blockMask; + private final boolean clustering; + private final long keysFilePointer; + private final LongArray blockOffsets; + + // The key the cursor currently points to. Initially empty. + private final BytesRef currentKey; + + // A temporary buffer used to hold the key at the start of the next block. + private final BytesRef nextBlockKey; + + // The point id the cursor currently points to. + private long currentPointId; + private long currentBlockIndex; + + Cursor(FileHandle keysFileHandle, LongArray.Factory blockOffsetsFactory) throws IOException + { + this.keysInput = IndexInputReader.create(keysFileHandle); + SAICodecUtils.validate(this.keysInput); + this.blockShift = this.keysInput.readVInt(); + this.blockMask = (1 << this.blockShift) - 1; + this.clustering = this.keysInput.readByte() == 1; + this.keysFilePointer = this.keysInput.getFilePointer(); + this.blockOffsets = new LongArray.DeferredLongArray(blockOffsetsFactory::open); + this.currentKey = new BytesRef(keyLookupMeta.maxKeyLength); + this.nextBlockKey = new BytesRef(keyLookupMeta.maxKeyLength); + keysInput.seek(keysFilePointer); + readKey(currentPointId, currentKey); + } + + /** + * Positions the cursor on the target point id and reads the keyat target to the current key buffer. Review Comment: ```suggestion * Positions the cursor on the target point id and reads the key at target to the current key buffer. ``` ########## test/unit/org/apache/cassandra/index/sai/disk/v1/keystore/KeyLookupTest.java: ########## @@ -0,0 +1,413 @@ +/* + * 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.index.sai.disk.v1.keystore; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.junit.Before; +import org.junit.Test; + +import org.apache.cassandra.db.Clustering; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.index.sai.SAITester; +import org.apache.cassandra.index.sai.disk.format.IndexComponent; +import org.apache.cassandra.index.sai.disk.format.IndexDescriptor; +import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter; +import org.apache.cassandra.index.sai.disk.v1.MetadataSource; +import org.apache.cassandra.index.sai.disk.v1.MetadataWriter; +import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesMeta; +import org.apache.cassandra.index.sai.disk.v1.bitpack.NumericValuesWriter; +import org.apache.cassandra.index.sai.utils.PrimaryKey; +import org.apache.cassandra.index.sai.utils.SAIRandomizedTester; +import org.apache.cassandra.io.util.FileHandle; +import org.apache.cassandra.utils.bytecomparable.ByteComparable; +import org.apache.cassandra.utils.bytecomparable.ByteSource; +import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse; +import org.apache.lucene.store.IndexInput; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class KeyLookupTest extends SAIRandomizedTester +{ + protected IndexDescriptor indexDescriptor; + + @Before + public void setup() throws Exception + { + indexDescriptor = newIndexDescriptor(); + } + + @Test + public void testFileValidation() throws Exception + { + List<PrimaryKey> primaryKeys = new ArrayList<>(); + + for (int x = 0; x < 11; x++) + { + ByteBuffer buffer = UTF8Type.instance.decompose(Integer.toString(x)); + DecoratedKey partitionKey = Murmur3Partitioner.instance.decorateKey(buffer); + PrimaryKey primaryKey = SAITester.TEST_FACTORY.create(partitionKey, Clustering.EMPTY); + primaryKeys.add(primaryKey); + } + + primaryKeys.sort(PrimaryKey::compareTo); + + try (MetadataWriter metadataWriter = new MetadataWriter(indexDescriptor.openPerSSTableOutput(IndexComponent.GROUP_META))) + { + IndexOutputWriter bytesWriter = indexDescriptor.openPerSSTableOutput(IndexComponent.PARTITION_KEY_BLOCKS); + NumericValuesWriter blockFPWriter = new NumericValuesWriter(indexDescriptor, IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, metadataWriter, true); + try (KeyStoreWriter writer = new KeyStoreWriter(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS), + metadataWriter, + bytesWriter, + blockFPWriter, + 4, + false)) + { + primaryKeys.forEach(primaryKey -> { + try + { + writer.add(primaryKey); + } + catch (IOException e) + { + e.printStackTrace(); + } + }); + } + } + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCKS, false)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, true)); + assertTrue(validateComponent(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, false)); + } + + @Test + public void testLongPrefixesAndSuffixes() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + writeTerms(writer -> { + // The following writes a set of terms that cover the following conditions: + + // Start value 0 + byte[] bytes = new byte[20]; + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix > 15 + bytes = new byte[20]; + Arrays.fill(bytes, 16, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix == 15 + bytes = new byte[20]; + Arrays.fill(bytes, 15, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix < 15 + bytes = new byte[20]; + Arrays.fill(bytes, 14, 20, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix > 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 4, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix == 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 5, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // suffix < 16 + bytes = new byte[20]; + Arrays.fill(bytes, 0, 6, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + + bytes = new byte[32]; + Arrays.fill(bytes, 0, 16, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + // prefix >= 15 && suffix >= 16 + bytes = new byte[32]; + Arrays.fill(bytes, 0, 32, (byte)1); + terms.add(bytes); + writer.add(ByteComparable.fixedLength(bytes)); + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testNonUniqueTerms() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(5000); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointId() throws Exception + { + List<byte[]> terms = new ArrayList<>(); + + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + terms.add(bytes); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + doTestSortedTerms(terms); + } + + @Test + public void testSeekToPointIdOutOfRange() throws Exception + { + writeTerms(writer -> { + for (int x = 0; x < 4000; x++) + { + ByteBuffer buffer = Int32Type.instance.decompose(x); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + byte[] bytes = ByteSourceInverse.readBytes(byteSource); + + writer.add(ByteComparable.fixedLength(bytes)); + } + }, false); + + withSortedTermsCursor(cursor -> { + assertThatThrownBy(() -> cursor.seekToPointId(-2)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, -2, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(Long.MAX_VALUE)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, Long.MAX_VALUE, 4000)); + assertThatThrownBy(() -> cursor.seekToPointId(4000)).isInstanceOf(IndexOutOfBoundsException.class) + .hasMessage(String.format(KeyLookup.INDEX_OUT_OF_BOUNDS, 4000, 4000)); + }); + } + + @Test + public void testSeekToTerm() throws Exception + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 4000; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, true); + + withSortedTermsCursor(cursor -> { + assertEquals(0L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)); + cursor.reset(); + assertEquals(160L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(160L)), 160L, 170L)); + cursor.reset(); + assertEquals(165L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(165L)), 160L, 170L)); + cursor.reset(); + assertEquals(175L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(175L)), 160L, 176L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 160L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(176L)), 175L, 177L)); + cursor.reset(); + assertEquals(176L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(701)), 160L, 177L)); + cursor.reset(); + assertEquals(504L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(504L)), 200L, 600L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 0L, 1000L)); + cursor.reset(); + assertEquals(-1L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(makeTerm(4000)), 999L, 1000L)); + cursor.reset(); + assertEquals(999L, cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(999L)), 0L, 1000L)); + }); + } + + @Test + public void seekToTermOnNonPartitionedTest() throws Throwable + { + Map<Long, byte[]> terms = new HashMap<>(); + + writeTerms(writer -> { + long pointId = 0; + for (int x = 0; x < 16; x += 4) + { + byte[] term = makeTerm(x); + terms.put(pointId++, term); + + writer.add(ByteComparable.fixedLength(term)); + } + }, false); + + withSortedTermsCursor(cursor -> assertThatThrownBy(() -> cursor.clusteredSeekToKey(ByteComparable.fixedLength(terms.get(0L)), 0L, 10L)) + .isInstanceOf(AssertionError.class)); + } + + @Test + public void partitionedTermsMustBeInOrderInPartitions() throws Throwable + { + writeTerms(writer -> { + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(0))); + writer.add(ByteComparable.fixedLength(makeTerm(10))); + assertThatThrownBy(() -> writer.add(ByteComparable.fixedLength(makeTerm(9)))).isInstanceOf(IllegalArgumentException.class); + writer.startPartition(); + writer.add(ByteComparable.fixedLength(makeTerm(9))); + },true); + } + + private byte[] makeTerm(int value) + { + ByteBuffer buffer = Int32Type.instance.decompose(value); + ByteSource byteSource = Int32Type.instance.asComparableBytes(buffer, ByteComparable.Version.OSS50); + return ByteSourceInverse.readBytes(byteSource); + } + + private void doTestSortedTerms(List<byte[]> terms) throws Exception + { + // iterate ascending + withSortedTermsCursor(cursor -> { + for (int x = 0; x < terms.size(); x++) + { + ByteComparable term = cursor.seekToPointId(x); + + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + + assertArrayEquals(terms.get(x), bytes); + } + }); + + // iterate ascending skipping blocks + withSortedTermsCursor(cursor -> { + for (int x = 0; x < terms.size(); x += 17) + { + ByteComparable term = cursor.seekToPointId(x); + + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + + assertArrayEquals(terms.get(x), bytes); + } + }); + + withSortedTermsCursor(cursor -> { + ByteComparable term = cursor.seekToPointId(7); + byte[] bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + assertArrayEquals(terms.get(7), bytes); + + term = cursor.seekToPointId(7); + bytes = ByteSourceInverse.readBytes(term.asComparableBytes(ByteComparable.Version.OSS50)); + assertArrayEquals(terms.get(7), bytes); + }); + } + + protected void writeTerms(ThrowingConsumer<KeyStoreWriter> testCode, boolean partitioned) throws IOException + { + try (MetadataWriter metadataWriter = new MetadataWriter(indexDescriptor.openPerSSTableOutput(IndexComponent.GROUP_META))) + { + IndexOutputWriter bytesWriter = indexDescriptor.openPerSSTableOutput(IndexComponent.PARTITION_KEY_BLOCKS); + NumericValuesWriter blockFPWriter = new NumericValuesWriter(indexDescriptor, IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, metadataWriter, true); + try (KeyStoreWriter writer = new KeyStoreWriter(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS), + metadataWriter, + bytesWriter, + blockFPWriter, + 4, + partitioned)) + { + testCode.accept(writer); + } + } + } + + @FunctionalInterface + public interface ThrowingConsumer<T> + { + void accept(T t) throws IOException; + } + + private void withSortedTermsReader(ThrowingConsumer<KeyLookup> testCode) throws IOException + { + MetadataSource metadataSource = MetadataSource.loadGroupMetadata(indexDescriptor); + NumericValuesMeta blockPointersMeta = new NumericValuesMeta(metadataSource.get(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS))); + KeyLookupMeta sortedTermsMeta = new KeyLookupMeta(metadataSource.get(indexDescriptor.componentName(IndexComponent.PARTITION_KEY_BLOCKS))); + try (FileHandle termsData = indexDescriptor.createPerSSTableFileHandle(IndexComponent.PARTITION_KEY_BLOCKS, null); + FileHandle blockOffsets = indexDescriptor.createPerSSTableFileHandle(IndexComponent.PARTITION_KEY_BLOCK_OFFSETS, null)) + { + KeyLookup reader = new KeyLookup(termsData, blockOffsets, sortedTermsMeta, blockPointersMeta); + testCode.accept(reader); + } + } + + private void withSortedTermsCursor(ThrowingConsumer<KeyLookup.Cursor> testCode) throws IOException Review Comment: This could be named `withKeyLookupCursor` -- 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]
