natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to lucene-core 9.7.0 

Notes on the upgrade path:
- RamIndexOutput is replaced with ResettableByteBuffersIndexOutput, an extension of ByteBuffersIndexOutput, which was the closest thing to a replacement of RamIndexOutput.
- Lucene exposes the code we needed from DirectReaders more or less directly in DirectReader now, so the old copied code has been deleted.
- Lucene changed its data files to be little endian, but to keep its compatibility story simple it retained BE for the header and footer ints. That's the cause of the changes in SAICodecUtils.
- We could gain a bit of performance making our own code natively little endian but that is too big of a change for this patch.

patch by Jonathan Ellis; reviewed by Andrés de la Peña, Caleb Rackliffe, and Mike Adamson for CASSANDRA-18494
This commit is contained in:
Jonathan Ellis 2023-06-26 14:50:01 -05:00 committed by Caleb Rackliffe
parent 661a1d5bb6
commit ebea2ba6ad
22 changed files with 302 additions and 461 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.build/parent-pom-template.xml
View File

@ -1053,7 +1053,7 @@
<dependency>
<groupId>org.apache.lucene</groupId>
<artifactId>lucene-core</artifactId>
<version>7.5.0</version>
<version>9.7.0</version>
</dependency>
<dependency>
<groupId>com.carrotsearch.randomizedtesting</groupId>

153
src/java/org/apache/cassandra/index/sai/disk/ResettableByteBuffersIndexOutput.java Normal file
View File

@ -0,0 +1,153 @@
/*
* 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;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import org.apache.lucene.store.ByteBuffersDataOutput;
import org.apache.lucene.store.ByteBuffersIndexOutput;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.store.IndexOutput;
/***
* A wrapper around {@link ByteBuffersIndexOutput} that adds several methods that interact
* with the underlying delegate.
*/
public class ResettableByteBuffersIndexOutput extends IndexOutput
{
private final ByteBuffersIndexOutput bbio;
private final ByteBuffersDataOutput delegate;
public ResettableByteBuffersIndexOutput(String name)
{
//TODO CASSANDRA-18280 to investigate the initial size allocation
this(128, name);
}
public ResettableByteBuffersIndexOutput(int expectedSize, String name)
{
super("", name);
delegate = new ByteBuffersDataOutput(expectedSize);
bbio = new ByteBuffersIndexOutput(delegate, "", name + "-bb");
}
public void copyTo(IndexOutput out) throws IOException
{
delegate.copyTo(out);
}
public int intSize() {
return Math.toIntExact(bbio.getFilePointer());
}
public byte[] toArrayCopy() {
return delegate.toArrayCopy();
}
public void reset()
{
delegate.reset();
}
@Override
public String toString()
{
return "Resettable" + bbio.toString();
}
@Override
public void close() throws IOException
{
bbio.close();
}
@Override
public long getFilePointer()
{
return bbio.getFilePointer();
}
@Override
public long getChecksum() throws IOException
{
return bbio.getChecksum();
}
@Override
public void writeByte(byte b) throws IOException
{
bbio.writeByte(b);
}
@Override
public void writeBytes(byte[] b, int offset, int length) throws IOException
{
bbio.writeBytes(b, offset, length);
}
@Override
public void writeBytes(byte[] b, int length) throws IOException
{
bbio.writeBytes(b, length);
}
@Override
public void writeInt(int i) throws IOException
{
bbio.writeInt(i);
}
@Override
public void writeShort(short i) throws IOException
{
bbio.writeShort(i);
}
@Override
public void writeLong(long i) throws IOException
{
bbio.writeLong(i);
}
@Override
public void writeString(String s) throws IOException
{
bbio.writeString(s);
}
@Override
public void copyBytes(DataInput input, long numBytes) throws IOException
{
bbio.copyBytes(input, numBytes);
}
@Override
public void writeMapOfStrings(Map<String, String> map) throws IOException
{
bbio.writeMapOfStrings(map);
}
@Override
public void writeSetOfStrings(Set<String> set) throws IOException
{
bbio.writeSetOfStrings(set);
}
}

35
src/java/org/apache/cassandra/index/sai/disk/io/IndexInputReader.java
View File

@ -33,6 +33,11 @@ import org.apache.lucene.store.IndexInput;
*/
public class IndexInputReader extends IndexInput
{
/**
* the byte order of `input`'s native readX operations doesn't matter,
* because we only use `readFully` and `readByte` methods. IndexInput calls these
* (via DataInput) with methods that enforce LittleEndian-ness.
*/
private final RandomAccessReader input;
private final Runnable doOnClose;
@ -72,36 +77,6 @@ public class IndexInputReader extends IndexInput
input.readFully(bytes, off, len);
}
/**
* Using {@link RandomAccessReader#readShort()} directly is faster than {@link DataInput#readShort()} which calls
* {@link DataInput#readByte()} one by one
*/
@Override
public short readShort() throws IOException
{
return input.readShort();
}
/**
* Using {@link RandomAccessReader#readInt()} directly is faster than {@link DataInput#readInt()} which
* calls {@link DataInput#readByte()} one by one
*/
@Override
public int readInt() throws IOException
{
return input.readInt();
}
/**
* Using {@link RandomAccessReader#readLong()} directly is faster than {@link DataInput#readLong()} which
* calls {@link DataInput#readByte()} one by one
*/
@Override
public long readLong() throws IOException
{
return input.readLong();
}
@Override
public void close()
{

5
src/java/org/apache/cassandra/index/sai/disk/io/IndexOutputWriter.java
View File

@ -37,6 +37,11 @@ public class IndexOutputWriter extends IndexOutput
{
private static final Logger logger = LoggerFactory.getLogger(IndexOutputWriter.class);
/**
* the byte order of `out`'s native writeX operations doesn't matter,
* because we only use `write(byte[])` and `writeByte` methods. IndexOutput calls these
* (via DataOutput) with methods that enforce LittleEndian-ness.
*/
private final SequentialWriter out;
private boolean closed;

86
src/java/org/apache/cassandra/index/sai/disk/io/RAMIndexOutput.java
View File

@ -1,86 +0,0 @@
/*
* 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.io;
import java.io.IOException;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.lucene.store.GrowableByteArrayDataOutput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.util.BytesRef;
/**
* A replacement for {@link org.apache.lucene.store.RAMOutputStream}
* that doesn't use monitor locks.
*/
@NotThreadSafe
public class RAMIndexOutput extends IndexOutput
{
private final GrowableByteArrayDataOutput out;
public RAMIndexOutput(String name)
{
super("", name);
//TODO CASSANDRA-18280 to investigate the initial size allocation
out = new GrowableByteArrayDataOutput(128);
}
@Override
public long getChecksum()
{
throw new UnsupportedOperationException();
}
@Override
public long getFilePointer()
{
return out.getPosition();
}
@Override
public void writeByte(byte b)
{
out.writeByte(b);
}
@Override
public void writeBytes(byte[] bytes, int offset, int len)
{
out.writeBytes(bytes, offset, len);
}
public void writeTo(IndexOutput externalOut) throws IOException
{
externalOut.writeBytes(out.getBytes(), 0, out.getPosition());
}
public BytesRef getBytes()
{
return new BytesRef(out.getBytes(), 0, out.getPosition());
}
public void reset()
{
out.reset();
}
@Override
public void close()
{}
}

230
src/java/org/apache/cassandra/index/sai/disk/v1/DirectReaders.java
View File

@ -15,236 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.index.sai.disk.v1;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.function.Supplier;
import com.google.common.collect.Sets;
import org.apache.lucene.store.RandomAccessInput;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.IndexInput;
public class DirectReaders
{
public static final Set<Integer> SUPPORTED_BITS_PER_VALUE = Sets.newHashSet(0, 1, 2, 4, 8, 12, 16, 20, 24, 28, 32, 40, 48, 56, 64);
public static final String SUPPORTED_BITS_PER_VALUE_STRING = SUPPORTED_BITS_PER_VALUE.stream()
.map(i -> Integer.toString(i))
.collect(Collectors.joining(", "));
public interface Reader
public static void checkBitsPerValue(int valuesBitsPerValue, IndexInput input, Supplier<String> source) throws CorruptIndexException
{
/**
* Returns the indexed delta value from the block starting at offset.
* This is implemented for each of the supported bits-per-value where
* each implementation is responsible for decoding the bits.
*/
long get(RandomAccessInput in, long offset, long index);
}
public static Reader getReaderForBitsPerValue(byte bitsPerValue)
{
switch (bitsPerValue)
if (valuesBitsPerValue > 64)
{
case 0:
return READER_0;
case 1:
return READER_1;
case 2:
return READER_2;
case 4:
return READER_4;
case 8:
return READER_8;
case 12:
return READER_12;
case 16:
return READER_16;
case 20:
return READER_20;
case 24:
return READER_24;
case 28:
return READER_28;
case 32:
return READER_32;
case 40:
return READER_40;
case 48:
return READER_48;
case 56:
return READER_56;
case 64:
return READER_64;
default:
throw new IllegalArgumentException("unsupported bitsPerValue: " + bitsPerValue);
String message = String.format("%s is corrupted: Bits per value for block offsets must be no more than 64 and is %d", source.get(), valuesBitsPerValue);
throw new CorruptIndexException(message, input);
}
}
private static final Reader READER_0 = (in, offset, index) -> 0;
private static final Reader READER_1 = (in, offset, index) -> {
try
{
int shift = 7 - (int) (index & 7);
return (in.readByte(offset + (index >>> 3)) >>> shift) & 0x1;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_2 = (in, offset, index) -> {
try
{
int shift = (3 - (int) (index & 3)) << 1;
return (in.readByte(offset + (index >>> 2)) >>> shift) & 0x3;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_4 = (in, offset, index) -> {
try
{
int shift = (int) ((index + 1) & 1) << 2;
return (in.readByte(offset + (index >>> 1)) >>> shift) & 0xF;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_8 = (in, offset, index) -> {
try
{
return in.readByte(offset + index) & 0xFF;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_12 = (in, offset, index) -> {
try
{
long o = (index * 12) >>> 3;
int shift = (int) ((index + 1) & 1) << 2;
return (in.readShort(offset + o) >>> shift) & 0xFFF;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_16 = (in, offset, index) -> {
try
{
return in.readShort(offset + (index << 1)) & 0xFFFF;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_20 = (in, offset, index) -> {
try
{
long o = (index * 20) >>> 3;
int v = in.readInt(offset + o) >>> 8;
int shift = (int) ((index + 1) & 1) << 2;
return (v >>> shift) & 0xFFFFF;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_24 = (in, offset, index) -> {
try
{
return in.readInt(offset + index * 3) >>> 8;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_28 = (in, offset, index) -> {
try
{
long o = (index * 28) >>> 3;
int shift = (int) ((index + 1) & 1) << 2;
return (in.readInt(offset + o) >>> shift) & 0xFFFFFFFL;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_32 = (in, offset, index) -> {
try
{
return in.readInt(offset + (index << 2)) & 0xFFFFFFFFL;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_40 = (in, offset, index) -> {
try
{
return in.readLong(offset + index * 5) >>> 24;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_48 = (in, offset, index) -> {
try
{
return in.readLong(offset + index * 6) >>> 16;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_56 = (in, offset, index) -> {
try
{
return in.readLong(offset + index * 7) >>> 8;
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
private static final Reader READER_64 = (in, offset, index) -> {
try
{
return in.readLong(offset + (index << 3));
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
};
}

7
src/java/org/apache/cassandra/index/sai/disk/v1/MetadataSource.java
View File

@ -26,7 +26,8 @@ import org.apache.cassandra.index.sai.IndexContext;
import org.apache.cassandra.index.sai.disk.format.IndexComponent;
import org.apache.cassandra.index.sai.disk.format.IndexDescriptor;
import org.apache.lucene.store.BufferedChecksumIndexInput;
import org.apache.lucene.store.ByteArrayIndexInput;
import org.apache.lucene.store.ByteArrayDataInput;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.util.BytesRef;
@ -81,7 +82,7 @@ public class MetadataSource
return new MetadataSource(components);
}
public IndexInput get(String name)
public DataInput get(String name)
{
BytesRef bytes = components.get(name);
@ -91,6 +92,6 @@ public class MetadataSource
name, components.keySet()));
}
return new ByteArrayIndexInput(name, bytes.bytes);
return new ByteArrayDataInput(bytes.bytes);
}
}

6
src/java/org/apache/cassandra/index/sai/disk/v1/MetadataWriter.java
View File

@ -23,7 +23,7 @@ import java.util.HashMap;
import java.util.Map;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.index.sai.disk.io.RAMIndexOutput;
import org.apache.cassandra.index.sai.disk.ResettableByteBuffersIndexOutput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.util.BytesRef;
@ -43,7 +43,7 @@ public class MetadataWriter implements Closeable
return new Builder(name);
}
public class Builder extends RAMIndexOutput implements Closeable
public class Builder extends ResettableByteBuffersIndexOutput implements Closeable
{
private Builder(String name)
{
@ -53,7 +53,7 @@ public class MetadataWriter implements Closeable
@Override
public void close()
{
map.put(getName(), getBytes());
map.put(getName(), new BytesRef(toArrayCopy(), 0, intSize()));
}
}

24
src/java/org/apache/cassandra/index/sai/disk/v1/SAICodecUtils.java
View File

@ -30,27 +30,35 @@ import org.apache.lucene.store.IndexOutput;
import static org.apache.lucene.codecs.CodecUtil.CODEC_MAGIC;
import static org.apache.lucene.codecs.CodecUtil.FOOTER_MAGIC;
import static org.apache.lucene.codecs.CodecUtil.readBEInt;
import static org.apache.lucene.codecs.CodecUtil.readBELong;
import static org.apache.lucene.codecs.CodecUtil.writeBEInt;
import static org.apache.lucene.codecs.CodecUtil.writeBELong;
public class SAICodecUtils
{
// Lucene switched from big-endian to little-endian file format, but retained
// big-endian values in CodecUtils header and footer for compatibility.
// We follow their lead and use explicitly big-endian values here.
public static final String FOOTER_POINTER = "footerPointer";
public static void writeHeader(IndexOutput out) throws IOException
{
out.writeInt(CODEC_MAGIC);
writeBEInt(out, CODEC_MAGIC);
out.writeString(Version.LATEST.toString());
}
public static void writeFooter(IndexOutput out) throws IOException
{
out.writeInt(FOOTER_MAGIC);
out.writeInt(0);
writeBEInt(out, FOOTER_MAGIC);
writeBEInt(out, 0);
writeCRC(out);
}
public static void checkHeader(DataInput in) throws IOException
{
final int actualMagic = in.readInt();
final int actualMagic = readBEInt(in);
if (actualMagic != CODEC_MAGIC)
{
throw new CorruptIndexException("codec header mismatch: actual header=" + actualMagic + " vs expected header=" + CODEC_MAGIC, in);
@ -210,14 +218,14 @@ public class SAICodecUtils
}
}
final int magic = in.readInt();
final int magic = readBEInt(in);
if (magic != FOOTER_MAGIC)
{
throw new CorruptIndexException("codec footer mismatch (file truncated?): actual footer=" + magic + " vs expected footer=" + FOOTER_MAGIC, in);
}
final int algorithmID = in.readInt();
final int algorithmID = readBEInt(in);
if (algorithmID != 0)
{
@ -239,7 +247,7 @@ public class SAICodecUtils
{
throw new IllegalStateException("Illegal CRC-32 checksum: " + value + " (resource=" + output + ')');
}
output.writeLong(value);
writeBELong(output, value);
}
/**
@ -249,7 +257,7 @@ public class SAICodecUtils
*/
private static long readCRC(IndexInput input) throws IOException
{
long value = input.readLong();
long value = readBELong(input);
if ((value & 0xFFFFFFFF00000000L) != 0)
{
throw new CorruptIndexException("Illegal CRC-32 checksum: " + value, input);

9
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/AbstractBlockPackedReader.java
View File

@ -20,9 +20,10 @@ package org.apache.cassandra.index.sai.disk.v1.bitpack;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.cassandra.index.sai.disk.io.SeekingRandomAccessInput;
import org.apache.cassandra.index.sai.disk.v1.DirectReaders;
import org.apache.cassandra.index.sai.disk.v1.LongArray;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.util.LongValues;
import org.apache.lucene.util.packed.DirectReader;
@NotThreadSafe
public abstract class AbstractBlockPackedReader implements LongArray
@ -54,8 +55,10 @@ public abstract class AbstractBlockPackedReader implements LongArray
int blockIndex = (int) (valueIndex >>> blockShift);
int inBlockIndex = (int) (valueIndex & blockMask);
DirectReaders.Reader subReader = DirectReaders.getReaderForBitsPerValue(blockBitsPerValue[blockIndex]);
return delta(blockIndex, inBlockIndex) + subReader.get(input, blockOffsetAt(blockIndex), inBlockIndex);
byte bitsPerValue = blockBitsPerValue[blockIndex];
final LongValues subReader = bitsPerValue == 0 ? LongValues.ZEROES
: DirectReader.getInstance(input, bitsPerValue, blockOffsetAt(blockIndex));
return delta(blockIndex, inBlockIndex) + subReader.get(inBlockIndex);
}
@Override

8
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/AbstractBlockPackedWriter.java
View File

@ -19,7 +19,7 @@ package org.apache.cassandra.index.sai.disk.v1.bitpack;
import java.io.IOException;
import org.apache.cassandra.index.sai.disk.io.RAMIndexOutput;
import org.apache.cassandra.index.sai.disk.ResettableByteBuffersIndexOutput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.util.packed.DirectWriter;
@ -39,7 +39,7 @@ public abstract class AbstractBlockPackedWriter
// This collects metadata specific to the block packed writer being used during the
// writing of the block packed data. This cached metadata is then written to the end
// of the data file when the block packed writer is finished.
protected final RAMIndexOutput blockMetaWriter;
protected final ResettableByteBuffersIndexOutput blockMetaWriter;
protected int blockIndex;
protected boolean finished;
@ -48,7 +48,7 @@ public abstract class AbstractBlockPackedWriter
{
checkBlockSize(blockSize, MIN_BLOCK_SIZE, MAX_BLOCK_SIZE);
this.indexOutput = indexOutput;
this.blockMetaWriter = new RAMIndexOutput("BlockPackedMeta");
this.blockMetaWriter = new ResettableByteBuffersIndexOutput(blockSize, "BlockPackedMeta");
blockValues = new long[blockSize];
}
@ -80,7 +80,7 @@ public abstract class AbstractBlockPackedWriter
flush();
}
final long fp = indexOutput.getFilePointer();
blockMetaWriter.writeTo(indexOutput);
blockMetaWriter.copyTo(indexOutput);
finished = true;
return fp;
}

8
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/BlockPackedReader.java
View File

@ -26,7 +26,6 @@ import org.apache.cassandra.index.sai.disk.v1.LongArray;
import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.IndexInput;
import static org.apache.cassandra.index.sai.disk.v1.SAICodecUtils.checkBlockSize;
@ -70,11 +69,8 @@ public class BlockPackedReader implements LongArray.Factory
{
final int token = in.readByte() & 0xFF;
final int bitsPerValue = token >>> BlockPackedWriter.BPV_SHIFT;
if (!DirectReaders.SUPPORTED_BITS_PER_VALUE.contains(bitsPerValue))
{
throw new CorruptIndexException(String.format("Block %d is corrupted. Bits per value is %d. Supported values are %s.",
i, bitsPerValue, DirectReaders.SUPPORTED_BITS_PER_VALUE_STRING), in);
}
int blockIndex = i;
DirectReaders.checkBitsPerValue(bitsPerValue, in, () -> String.format("Block %d", blockIndex));
if ((token & BlockPackedWriter.MIN_VALUE_EQUALS_0) == 0)
{
long val = zigZagDecode(1L + readVLong(in));

7
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/MonotonicBlockPackedReader.java
View File

@ -26,7 +26,6 @@ import org.apache.cassandra.index.sai.disk.v1.LongArray;
import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.util.packed.PackedInts;
import org.apache.lucene.util.packed.PackedLongValues;
@ -71,11 +70,7 @@ public class MonotonicBlockPackedReader implements LongArray.Factory
minValuesBuilder.add(in.readZLong());
averages[i] = Float.intBitsToFloat(in.readInt());
final int bitsPerValue = in.readVInt();
if (!DirectReaders.SUPPORTED_BITS_PER_VALUE.contains(bitsPerValue))
{
throw new CorruptIndexException(String.format("Block %d is corrupted. Bits per value is %d. Supported values are %s.",
i, bitsPerValue, DirectReaders.SUPPORTED_BITS_PER_VALUE_STRING), in);
}
DirectReaders.checkBitsPerValue(bitsPerValue, in, () -> "Postings list header");
blockBitsPerValue[i] = (byte) bitsPerValue;
// when bitsPerValue is 0, block offset won't be used
blockOffsetsBuilder.add(bitsPerValue == 0 ? -1 : in.readVLong());

4
src/java/org/apache/cassandra/index/sai/disk/v1/bitpack/NumericValuesMeta.java
View File

@ -19,7 +19,7 @@ package org.apache.cassandra.index.sai.disk.v1.bitpack;
import java.io.IOException;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.store.IndexOutput;
public class NumericValuesMeta
@ -28,7 +28,7 @@ public class NumericValuesMeta
public final int blockSize;
public final long blockMetaOffset;
public NumericValuesMeta(IndexInput input) throws IOException
public NumericValuesMeta(DataInput input) throws IOException
{
valueCount = input.readLong();
blockSize = input.readInt();

74
src/java/org/apache/cassandra/index/sai/disk/v1/postings/PostingsReader.java
View File

@ -23,16 +23,20 @@ import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.index.sai.disk.v1.DirectReaders;
import org.apache.cassandra.index.sai.postings.OrdinalPostingList;
import org.apache.cassandra.index.sai.postings.PostingList;
import org.apache.cassandra.index.sai.disk.v1.DirectReaders;
import org.apache.cassandra.index.sai.disk.v1.LongArray;
import org.apache.cassandra.index.sai.metrics.QueryEventListener;
import org.apache.cassandra.index.sai.disk.io.SeekingRandomAccessInput;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.RandomAccessInput;
import org.apache.lucene.util.LongValues;
import org.apache.lucene.util.packed.DirectReader;
/**
@ -44,6 +48,8 @@ import org.apache.lucene.store.RandomAccessInput;
@NotThreadSafe
public class PostingsReader implements OrdinalPostingList
{
private static final Logger logger = LoggerFactory.getLogger(PostingsReader.class);
private final IndexInput input;
private final SeekingRandomAccessInput seekingInput;
private final QueryEventListener.PostingListEventListener listener;
@ -57,7 +63,7 @@ public class PostingsReader implements OrdinalPostingList
private long actualPosting;
private long currentPosition;
private DirectReaders.Reader currentFoRValues;
private LongValues currentFoRValues;
private long postingsDecoded = 0;
@VisibleForTesting
@ -68,6 +74,7 @@ public class PostingsReader implements OrdinalPostingList
public PostingsReader(IndexInput input, BlocksSummary summary, QueryEventListener.PostingListEventListener listener) throws IOException
{
logger.debug("Opening postings reader for {}", input);
this.input = input;
this.seekingInput = new SeekingRandomAccessInput(input);
this.listener = listener;
@ -104,23 +111,15 @@ public class PostingsReader implements OrdinalPostingList
long maxBlockValuesOffset = input.getFilePointer() + maxBlockValuesLength;
byte offsetBitsPerValue = input.readByte();
if (!DirectReaders.SUPPORTED_BITS_PER_VALUE.contains((int)offsetBitsPerValue))
{
throw new CorruptIndexException(
String.format("Postings list header is corrupted: Bits per value for block offsets is %s. Supported values are %s.",
offsetBitsPerValue, DirectReaders.SUPPORTED_BITS_PER_VALUE_STRING), input);
}
this.offsets = new LongArrayReader(randomAccessInput, DirectReaders.getReaderForBitsPerValue(offsetBitsPerValue), input.getFilePointer(), numBlocks);
DirectReaders.checkBitsPerValue(offsetBitsPerValue, input, () -> "Postings list header");
LongValues lvOffsets = offsetBitsPerValue == 0 ? LongValues.ZEROES : DirectReader.getInstance(randomAccessInput, offsetBitsPerValue, input.getFilePointer());
this.offsets = new LongArrayReader(lvOffsets, numBlocks);
input.seek(maxBlockValuesOffset);
byte valuesBitsPerValue = input.readByte();
if (!DirectReaders.SUPPORTED_BITS_PER_VALUE.contains((int)valuesBitsPerValue))
{
throw new CorruptIndexException(
String.format("Postings list header is corrupted: Bits per value for value samples is %s. Supported values are %s.",
valuesBitsPerValue, DirectReaders.SUPPORTED_BITS_PER_VALUE_STRING), input);
}
this.maxValues = new LongArrayReader(randomAccessInput, DirectReaders.getReaderForBitsPerValue(valuesBitsPerValue), input.getFilePointer(), numBlocks);
DirectReaders.checkBitsPerValue(valuesBitsPerValue, input, () -> "Postings list header");
LongValues lvValues = valuesBitsPerValue == 0 ? LongValues.ZEROES : DirectReader.getInstance(randomAccessInput, valuesBitsPerValue, input.getFilePointer());
this.maxValues = new LongArrayReader(lvValues, numBlocks);
}
void close()
@ -130,23 +129,19 @@ public class PostingsReader implements OrdinalPostingList
private static class LongArrayReader implements LongArray
{
private final RandomAccessInput input;
private final DirectReaders.Reader reader;
private final long offset;
private final LongValues reader;
private final int length;
private LongArrayReader(RandomAccessInput input, DirectReaders.Reader reader, long offset, int length)
private LongArrayReader(LongValues reader, int length)
{
this.input = input;
this.reader = reader;
this.offset = offset;
this.length = length;
}
@Override
public long get(long idx)
{
return reader.get(input, offset, idx);
return reader.get(idx);
}
@Override
@ -311,17 +306,9 @@ public class PostingsReader implements OrdinalPostingList
private int nextFoRValue()
{
// currentFoRValues is null when the all the values in the block are the same
if (currentFoRValues == null)
{
return 0;
}
else
{
long id = currentFoRValues.get(seekingInput, currentPosition, postingIndex);
postingsDecoded++;
return Math.toIntExact(id);
}
long id = currentFoRValues.get(postingIndex);
postingsDecoded++;
return Math.toIntExact(id);
}
private void advanceOnePosition(long nextPosting)
@ -334,7 +321,11 @@ public class PostingsReader implements OrdinalPostingList
private void reBuffer() throws IOException
{
long pointer = summary.offsets.get(blockIndex);
if (pointer < 4)
{
// the first 4 bytes must be CODEC_MAGIC
throw new CorruptIndexException(String.format("Invalid block offset %d for postings block idx %d", pointer, blockIndex), input);
}
input.seek(pointer);
long left = summary.numPostings - totalPostingsRead;
@ -357,10 +348,15 @@ public class PostingsReader implements OrdinalPostingList
if (bitsPerValue == 0)
{
// currentFORValues is null when the all the values in the block are the same
currentFoRValues = null;
// If bitsPerValue is 0 then all the values in the block are the same
currentFoRValues = LongValues.ZEROES;
return;
}
currentFoRValues = DirectReaders.getReaderForBitsPerValue(bitsPerValue);
else if (bitsPerValue > 64)
{
throw new CorruptIndexException(
String.format("Postings list #%s block is corrupted. Bits per value should be no more than 64 and is %d.", blockIndex, bitsPerValue), input);
}
currentFoRValues = DirectReader.getInstance(seekingInput, bitsPerValue, currentPosition);
}
}

25
src/java/org/apache/cassandra/index/sai/disk/v1/postings/PostingsWriter.java
View File

@ -23,22 +23,23 @@ import java.io.IOException;
import javax.annotation.concurrent.NotThreadSafe;
import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.agrona.collections.LongArrayList;
import org.apache.cassandra.index.sai.IndexContext;
import org.apache.cassandra.index.sai.disk.v1.DirectReaders;
import org.apache.cassandra.index.sai.postings.PostingList;
import org.apache.cassandra.index.sai.disk.ResettableByteBuffersIndexOutput;
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.RAMIndexOutput;
import org.apache.cassandra.index.sai.disk.io.IndexOutputWriter;
import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils;
import org.apache.cassandra.index.sai.postings.PostingList;
import org.apache.lucene.store.DataOutput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.util.packed.DirectWriter;
import static com.google.common.base.Preconditions.checkArgument;
import static java.lang.Math.max;
import static org.apache.lucene.codecs.lucene50.Lucene50PostingsFormat.BLOCK_SIZE;
/**
* Encodes, compresses and writes postings lists to disk.
@ -86,6 +87,11 @@ import static org.apache.lucene.codecs.lucene50.Lucene50PostingsFormat.BLOCK_SIZ
@NotThreadSafe
public class PostingsWriter implements Closeable
{
protected static final Logger logger = LoggerFactory.getLogger(PostingsWriter.class);
// import static org.apache.lucene.codecs.lucene50.Lucene50PostingsFormat.BLOCK_SIZE;
private final static int BLOCK_SIZE = 128;
private static final String POSTINGS_MUST_BE_SORTED_ERROR_MSG = "Postings must be sorted ascending, got [%s] after [%s]";
private final IndexOutput dataOutput;
@ -93,7 +99,7 @@ public class PostingsWriter implements Closeable
private final long[] deltaBuffer;
private final LongArrayList blockOffsets = new LongArrayList();
private final LongArrayList blockMaximumPostings = new LongArrayList();
private final RAMIndexOutput inMemoryOutput = new RAMIndexOutput("blockOffsets");
private final ResettableByteBuffersIndexOutput inMemoryOutput = new ResettableByteBuffersIndexOutput("blockOffsets");
private final long startOffset;
@ -116,6 +122,8 @@ public class PostingsWriter implements Closeable
private PostingsWriter(IndexOutput dataOutput, int blockSize) throws IOException
{
assert dataOutput instanceof IndexOutputWriter;
logger.debug("Creating postings writer for output {}", dataOutput);
this.blockSize = blockSize;
this.dataOutput = dataOutput;
startOffset = dataOutput.getFilePointer();
@ -132,7 +140,7 @@ public class PostingsWriter implements Closeable
}
/**
* @return file pointer where index structure begins
* @return file pointer where index structure begins (before header)
*/
public long getStartOffset()
{
@ -259,7 +267,7 @@ public class PostingsWriter implements Closeable
writeSortedFoRBlock(blockOffsets, inMemoryOutput);
dataOutput.writeVLong(inMemoryOutput.getFilePointer());
inMemoryOutput.writeTo(dataOutput);
inMemoryOutput.copyTo(dataOutput);
writeSortedFoRBlock(blockMaximumPostings, dataOutput);
}
@ -267,9 +275,6 @@ public class PostingsWriter implements Closeable
{
final int bitsPerValue = maxDelta == 0 ? 0 : DirectWriter.unsignedBitsRequired(maxDelta);
assert DirectReaders.SUPPORTED_BITS_PER_VALUE.contains(bitsPerValue) :
"Unsupported bits per value of " + bitsPerValue + " bits. Supported values are: " + DirectReaders.SUPPORTED_BITS_PER_VALUE_STRING;
// If we have a first posting, indicating that this is the first block in the posting list
// then write it prior to the deltas.
if (firstPosting != Long.MIN_VALUE)

12
src/java/org/apache/cassandra/index/sai/disk/v1/segment/SegmentMetadata.java
View File

@ -36,7 +36,7 @@ import org.apache.cassandra.index.sai.disk.v1.MetadataSource;
import org.apache.cassandra.index.sai.disk.v1.MetadataWriter;
import org.apache.cassandra.index.sai.utils.PrimaryKey;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.store.IndexOutput;
/**
@ -112,7 +112,7 @@ public class SegmentMetadata
this.componentMetadatas = componentMetadatas;
}
private SegmentMetadata(IndexInput input, PrimaryKey.Factory primaryKeyFactory) throws IOException
private SegmentMetadata(DataInput input, PrimaryKey.Factory primaryKeyFactory) throws IOException
{
this.rowIdOffset = input.readLong();
this.numRows = input.readLong();
@ -128,7 +128,7 @@ public class SegmentMetadata
@SuppressWarnings({"resource", "RedundantSuppression"})
public static List<SegmentMetadata> load(MetadataSource source, PrimaryKey.Factory primaryKeyFactory) throws IOException
{
IndexInput input = source.get(NAME);
DataInput input = source.get(NAME);
int segmentCount = input.readVInt();
@ -179,7 +179,7 @@ public class SegmentMetadata
'}';
}
private static ByteBuffer readBytes(IndexInput input) throws IOException
private static ByteBuffer readBytes(DataInput input) throws IOException
{
int len = input.readInt();
byte[] bytes = new byte[len];
@ -210,7 +210,7 @@ public class SegmentMetadata
{
private final Map<IndexComponent, ComponentMetadata> metas = new EnumMap<>(IndexComponent.class);
ComponentMetadataMap(IndexInput input) throws IOException
ComponentMetadataMap(DataInput input) throws IOException
{
int size = input.readInt();
@ -312,7 +312,7 @@ public class SegmentMetadata
this.attributes = attributes;
}
ComponentMetadata(IndexInput input) throws IOException
ComponentMetadata(DataInput input) throws IOException
{
this.root = input.readLong();
this.offset = input.readLong();

4
src/java/org/apache/cassandra/index/sai/disk/v1/sortedterms/SortedTermsMeta.java
View File

@ -20,7 +20,7 @@ package org.apache.cassandra.index.sai.disk.v1.sortedterms;
import java.io.IOException;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.store.IndexOutput;
/**
@ -32,7 +32,7 @@ public class SortedTermsMeta
public final long termCount;
public final int maxTermLength;
public SortedTermsMeta(IndexInput input) throws IOException
public SortedTermsMeta(DataInput input) throws IOException
{
this.trieFilePointer = input.readLong();
this.termCount = input.readLong();

12
src/java/org/apache/cassandra/index/sai/memory/ByteSliceReader.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.index.sai.memory;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.lucene.store.DataInput;
import org.apache.lucene.util.BitUtil;
import org.apache.lucene.util.ByteBlockPool;
/**
@ -88,10 +89,7 @@ final class ByteSliceReader extends DataInput
public void nextSlice()
{
// Skip to our next slice
final int nextIndex = ((buffer[limit] & 0xff) << 24) +
((buffer[1 + limit] & 0xff) << 16) +
((buffer[2 + limit] & 0xff) << 8) +
(buffer[3 + limit] & 0xff);
final int nextIndex = (int) BitUtil.VH_LE_INT.get(buffer, limit);
level = ByteBlockPool.NEXT_LEVEL_ARRAY[level];
final int newSize = ByteBlockPool.LEVEL_SIZE_ARRAY[level];
@ -116,6 +114,12 @@ final class ByteSliceReader extends DataInput
}
}
@Override
public void skipBytes(long l)
{
throw new UnsupportedOperationException("skipBytes is not supported by ByteSliceReader");
}
@Override
public void readBytes(byte[] b, int offset, int len)
{

6
test/unit/org/apache/cassandra/index/sai/disk/v1/MetadataTest.java
View File

@ -39,10 +39,9 @@ import org.apache.cassandra.index.sai.utils.SAIRandomizedTester;
import org.apache.cassandra.io.util.File;
import org.apache.lucene.codecs.CodecUtil;
import org.apache.lucene.index.CorruptIndexException;
import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.DataInput;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@ -87,10 +86,9 @@ public class MetadataTest extends SAIRandomizedTester
for (Map.Entry<String, byte[]> entry : data.entrySet())
{
final IndexInput input = reader.get(entry.getKey());
final DataInput input = reader.get(entry.getKey());
assertNotNull(input);
final byte[] expectedBytes = entry.getValue();
assertEquals(expectedBytes.length, input.length());
final byte[] actualBytes = new byte[expectedBytes.length];
input.readBytes(actualBytes, 0, expectedBytes.length);
assertArrayEquals(expectedBytes, actualBytes);

44
test/unit/org/apache/cassandra/index/sai/disk/v1/SAICodecUtilsTest.java
View File

@ -33,6 +33,8 @@ import org.mockito.Mockito;
import static org.apache.lucene.codecs.CodecUtil.CODEC_MAGIC;
import static org.apache.lucene.codecs.CodecUtil.FOOTER_MAGIC;
import static org.apache.lucene.codecs.CodecUtil.writeBEInt;
import static org.apache.lucene.codecs.CodecUtil.writeBELong;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.when;
@ -66,7 +68,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
{
writer.writeInt(1234);
writeBEInt(writer, 1234);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
@ -82,7 +84,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
{
try (IndexOutputWriter writer = IndexFileUtils.instance.openOutput(file))
{
writer.writeInt(CODEC_MAGIC);
writeBEInt(writer, CODEC_MAGIC);
writer.writeString("zz");
}
@ -149,7 +151,7 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
SAICodecUtils.writeFooter(writer);
writer.writeInt(1234);
writeBEInt(writer, 1234);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
@ -174,9 +176,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(1234);
writer.writeInt(0);
writer.writeLong(writer.getChecksum());
writeBEInt(writer, 1234);
writeBEInt(writer, 0);
writeBELong(writer, writer.getChecksum());
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
@ -201,9 +203,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(FOOTER_MAGIC);
writer.writeInt(1);
writer.writeLong(writer.getChecksum());
writeBEInt(writer, FOOTER_MAGIC);
writeBEInt(writer, 1);
writeBELong(writer, writer.getChecksum());
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
@ -228,9 +230,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(FOOTER_MAGIC);
writer.writeInt(0);
writer.writeLong(0);
writeBEInt(writer, FOOTER_MAGIC);
writeBEInt(writer, 0);
writeBELong(writer, 0);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
@ -255,9 +257,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(FOOTER_MAGIC);
writer.writeInt(0);
writer.writeLong(0xFFFFFFFF00000000L);
writeBEInt(writer, FOOTER_MAGIC);
writeBEInt(writer, 0);
writeBELong(writer, 0xFFFFFFFF00000000L);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file);
@ -298,9 +300,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(FOOTER_MAGIC);
writer.writeInt(0);
writer.writeLong(0);
writeBEInt(writer, FOOTER_MAGIC);
writeBEInt(writer, 0);
writeBELong(writer, 0);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))
@ -324,9 +326,9 @@ public class SAICodecUtilsTest extends SAIRandomizedTester
for (int value = 0; value < numBytes; value++)
writer.writeByte(getRandom().nextByte());
writer.writeInt(FOOTER_MAGIC);
writer.writeInt(0);
writer.writeLong(0xFFFFFFFF00000000L);
writeBEInt(writer, FOOTER_MAGIC);
writeBEInt(writer, 0);
writeBELong(writer, 0xFFFFFFFF00000000L);
}
try (IndexInput input = IndexFileUtils.instance.openBlockingInput(file))

2
test/unit/org/apache/cassandra/index/sai/disk/v1/TermsReaderTest.java
View File

@ -138,7 +138,7 @@ public class TermsReaderTest extends SAIRandomizedTester
{
final long expectedRowID = expectedPostingList.get(i);
long result = actualPostingList.nextPosting();
assertEquals(expectedRowID, result);
assertEquals(String.format("row %d mismatch of %d in enum %d", i, expectedPostingList.size(), termsEnum.indexOf(pair)), expectedRowID, result);
}
long lastResult = actualPostingList.nextPosting();