/*
 * 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.db;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

import com.google.common.primitives.Ints;
import org.junit.Assert;
import org.junit.Test;

import org.apache.cassandra.Util;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.columniterator.AbstractSSTableIterator;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
import org.apache.cassandra.db.rows.AbstractUnfilteredRowIterator;
import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.BufferCell;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredSerializer;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.sstable.IndexInfo;
import org.apache.cassandra.io.sstable.format.SSTableFlushObserver;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.serializers.LongSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;

import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertTrue;

public class RowIndexEntryTest extends CQLTester
{
    private static final List<AbstractType<?>> clusterTypes = Collections.singletonList(LongType.instance);
    private static final ClusteringComparator comp = new ClusteringComparator(clusterTypes);

    private static final byte[] dummy_100k = new byte[100000];

    private static Clustering cn(long l)
    {
        return Util.clustering(comp, l);
    }

    @Test
    public void testC11206AgainstPreviousArray() throws Exception
    {
        DatabaseDescriptor.setColumnIndexCacheSize(99999);
        testC11206AgainstPrevious();
    }

    @Test
    public void testC11206AgainstPreviousShallow() throws Exception
    {
        DatabaseDescriptor.setColumnIndexCacheSize(0);
        testC11206AgainstPrevious();
    }

    private static void testC11206AgainstPrevious() throws Exception
    {
        // partition without IndexInfo
        try (DoubleSerializer doubleSerializer = new DoubleSerializer())
        {
            doubleSerializer.build(null, partitionKey(42L),
                                   Collections.singletonList(cn(42)),
                                   0L);
            assertEquals(doubleSerializer.rieOldSerialized, doubleSerializer.rieNewSerialized);
        }

        // partition with multiple IndexInfo
        try (DoubleSerializer doubleSerializer = new DoubleSerializer())
        {
            doubleSerializer.build(null, partitionKey(42L),
                                   Arrays.asList(cn(42), cn(43), cn(44)),
                                   0L);
            assertEquals(doubleSerializer.rieOldSerialized, doubleSerializer.rieNewSerialized);
        }

        // partition with multiple IndexInfo
        try (DoubleSerializer doubleSerializer = new DoubleSerializer())
        {
            doubleSerializer.build(null, partitionKey(42L),
                                   Arrays.asList(cn(42), cn(43), cn(44), cn(45), cn(46), cn(47), cn(48), cn(49), cn(50), cn(51)),
                                   0L);
            assertEquals(doubleSerializer.rieOldSerialized, doubleSerializer.rieNewSerialized);
        }
    }

    private static DecoratedKey partitionKey(long l)
    {
        ByteBuffer key = LongSerializer.instance.serialize(l);
        Token token = Murmur3Partitioner.instance.getToken(key);
        return new BufferDecoratedKey(token, key);
    }

    private static class DoubleSerializer implements AutoCloseable
    {
        CFMetaData cfMeta = CFMetaData.compile("CREATE TABLE pipe.dev_null (pk bigint, ck bigint, val text, PRIMARY KEY(pk, ck))", "foo");
        Version version = BigFormat.latestVersion;

        DeletionTime deletionInfo = new DeletionTime(FBUtilities.timestampMicros(), FBUtilities.nowInSeconds());
        LivenessInfo primaryKeyLivenessInfo = LivenessInfo.EMPTY;
        Row.Deletion deletion = Row.Deletion.LIVE;

        SerializationHeader header = new SerializationHeader(true, cfMeta, cfMeta.partitionColumns(), EncodingStats.NO_STATS);

        // create C-11206 + old serializer instances
        RowIndexEntry.IndexSerializer rieSerializer = new RowIndexEntry.Serializer(cfMeta, version, header);
        Pre_C_11206_RowIndexEntry.Serializer oldSerializer = new Pre_C_11206_RowIndexEntry.Serializer(cfMeta, version, header);

        @SuppressWarnings({ "resource", "IOResourceOpenedButNotSafelyClosed" })
        final DataOutputBuffer rieOutput = new DataOutputBuffer(1024);
        @SuppressWarnings({ "resource", "IOResourceOpenedButNotSafelyClosed" })
        final DataOutputBuffer oldOutput = new DataOutputBuffer(1024);

        final SequentialWriter dataWriterNew;
        final SequentialWriter dataWriterOld;
        final org.apache.cassandra.db.ColumnIndex columnIndex;

        RowIndexEntry rieNew;
        ByteBuffer rieNewSerialized;
        Pre_C_11206_RowIndexEntry rieOld;
        ByteBuffer rieOldSerialized;

        DoubleSerializer() throws IOException
        {
            SequentialWriterOption option = SequentialWriterOption.newBuilder().bufferSize(1024).build();
            File f = File.createTempFile("RowIndexEntryTest-", "db");
            dataWriterNew = new SequentialWriter(f, option);
            columnIndex = new org.apache.cassandra.db.ColumnIndex(header, dataWriterNew, version, Collections.emptyList(),
                                                                  rieSerializer.indexInfoSerializer());

            f = File.createTempFile("RowIndexEntryTest-", "db");
            dataWriterOld = new SequentialWriter(f, option);
        }

        public void close() throws Exception
        {
            dataWriterNew.close();
            dataWriterOld.close();
        }

        void build(Row staticRow, DecoratedKey partitionKey,
                   Collection<Clustering> clusterings, long startPosition) throws IOException
        {

            Iterator<Clustering> clusteringIter = clusterings.iterator();
            columnIndex.buildRowIndex(makeRowIter(staticRow, partitionKey, clusteringIter, dataWriterNew));
            rieNew = RowIndexEntry.create(startPosition, 0L,
                                          deletionInfo, columnIndex.headerLength, columnIndex.columnIndexCount,
                                          columnIndex.indexInfoSerializedSize(),
                                          columnIndex.indexSamples(), columnIndex.offsets(),
                                          rieSerializer.indexInfoSerializer());
            rieSerializer.serialize(rieNew, rieOutput, columnIndex.buffer());
            rieNewSerialized = rieOutput.buffer().duplicate();

            Iterator<Clustering> clusteringIter2 = clusterings.iterator();
            ColumnIndex columnIndex = RowIndexEntryTest.ColumnIndex.writeAndBuildIndex(makeRowIter(staticRow, partitionKey, clusteringIter2, dataWriterOld),
                                                                                       dataWriterOld, header, Collections.emptySet(), BigFormat.latestVersion);
            rieOld = Pre_C_11206_RowIndexEntry.create(startPosition, deletionInfo, columnIndex);
            oldSerializer.serialize(rieOld, oldOutput);
            rieOldSerialized = oldOutput.buffer().duplicate();
        }

        private AbstractUnfilteredRowIterator makeRowIter(Row staticRow, DecoratedKey partitionKey,
                                                          Iterator<Clustering> clusteringIter, SequentialWriter dataWriter)
        {
            return new AbstractUnfilteredRowIterator(cfMeta, partitionKey, deletionInfo, cfMeta.partitionColumns(),
                                                     staticRow, false, new EncodingStats(0, 0, 0))
            {
                protected Unfiltered computeNext()
                {
                    if (!clusteringIter.hasNext())
                        return endOfData();
                    try
                    {
                        // write some fake bytes to the data file to force writing the IndexInfo object
                        dataWriter.write(dummy_100k);
                    }
                    catch (IOException e)
                    {
                        throw new RuntimeException(e);
                    }
                    return buildRow(clusteringIter.next());
                }
            };
        }

        private Unfiltered buildRow(Clustering clustering)
        {
            BTree.Builder<ColumnData> builder = BTree.builder(ColumnData.comparator);
            builder.add(BufferCell.live(cfMeta.partitionColumns().iterator().next(),
                                        1L,
                                        ByteBuffer.allocate(0)));
            return BTreeRow.create(clustering, primaryKeyLivenessInfo, deletion, builder.build());
        }
    }

    /**
     * Pre C-11206 code.
     */
    static final class ColumnIndex
    {
        final long partitionHeaderLength;
        final List<IndexInfo> columnsIndex;

        private static final ColumnIndex EMPTY = new ColumnIndex(-1, Collections.emptyList());

        private ColumnIndex(long partitionHeaderLength, List<IndexInfo> columnsIndex)
        {
            assert columnsIndex != null;

            this.partitionHeaderLength = partitionHeaderLength;
            this.columnsIndex = columnsIndex;
        }

        static ColumnIndex writeAndBuildIndex(UnfilteredRowIterator iterator,
                                              SequentialWriter output,
                                              SerializationHeader header,
                                              Collection<SSTableFlushObserver> observers,
                                              Version version) throws IOException
        {
            assert !iterator.isEmpty() && version.storeRows();

            Builder builder = new Builder(iterator, output, header, observers, version.correspondingMessagingVersion());
            return builder.build();
        }

        public static ColumnIndex nothing()
        {
            return EMPTY;
        }

        /**
         * Help to create an index for a column family based on size of columns,
         * and write said columns to disk.
         */
        private static class Builder
        {
            private final UnfilteredRowIterator iterator;
            private final SequentialWriter writer;
            private final SerializationHeader header;
            private final int version;

            private final List<IndexInfo> columnsIndex = new ArrayList<>();
            private final long initialPosition;
            private long headerLength = -1;

            private long startPosition = -1;

            private int written;
            private long previousRowStart;

            private ClusteringPrefix firstClustering;
            private ClusteringPrefix lastClustering;

            private DeletionTime openMarker;

            private final Collection<SSTableFlushObserver> observers;

            Builder(UnfilteredRowIterator iterator,
                           SequentialWriter writer,
                           SerializationHeader header,
                           Collection<SSTableFlushObserver> observers,
                           int version)
            {
                this.iterator = iterator;
                this.writer = writer;
                this.header = header;
                this.version = version;
                this.observers = observers == null ? Collections.emptyList() : observers;
                this.initialPosition = writer.position();
            }

            private void writePartitionHeader(UnfilteredRowIterator iterator) throws IOException
            {
                ByteBufferUtil.writeWithShortLength(iterator.partitionKey().getKey(), writer);
                DeletionTime.serializer.serialize(iterator.partitionLevelDeletion(), writer);
                if (header.hasStatic())
                    UnfilteredSerializer.serializer.serializeStaticRow(iterator.staticRow(), header, writer, version);
            }

            public ColumnIndex build() throws IOException
            {
                writePartitionHeader(iterator);
                this.headerLength = writer.position() - initialPosition;

                while (iterator.hasNext())
                    add(iterator.next());

                return close();
            }

            private long currentPosition()
            {
                return writer.position() - initialPosition;
            }

            private void addIndexBlock()
            {
                IndexInfo cIndexInfo = new IndexInfo(firstClustering,
                                                     lastClustering,
                                                     startPosition,
                                                                             currentPosition() - startPosition,
                                                     openMarker);
                columnsIndex.add(cIndexInfo);
                firstClustering = null;
            }

            private void add(Unfiltered unfiltered) throws IOException
            {
                long pos = currentPosition();

                if (firstClustering == null)
                {
                    // Beginning of an index block. Remember the start and position
                    firstClustering = unfiltered.clustering();
                    startPosition = pos;
                }

                UnfilteredSerializer.serializer.serialize(unfiltered, header, writer, pos - previousRowStart, version);

                // notify observers about each new row
                if (!observers.isEmpty())
                    observers.forEach((o) -> o.nextUnfilteredCluster(unfiltered));

                lastClustering = unfiltered.clustering();
                previousRowStart = pos;
                ++written;

                if (unfiltered.kind() == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER)
                {
                    RangeTombstoneMarker marker = (RangeTombstoneMarker)unfiltered;
                    openMarker = marker.isOpen(false) ? marker.openDeletionTime(false) : null;
                }

                // if we hit the column index size that we have to index after, go ahead and index it.
                if (currentPosition() - startPosition >= DatabaseDescriptor.getColumnIndexSize())
                    addIndexBlock();

            }

            private ColumnIndex close() throws IOException
            {
                UnfilteredSerializer.serializer.writeEndOfPartition(writer);

                // It's possible we add no rows, just a top level deletion
                if (written == 0)
                    return RowIndexEntryTest.ColumnIndex.EMPTY;

                // the last column may have fallen on an index boundary already.  if not, index it explicitly.
                if (firstClustering != null)
                    addIndexBlock();

                // we should always have at least one computed index block, but we only write it out if there is more than that.
                assert !columnsIndex.isEmpty() && headerLength >= 0;
                return new ColumnIndex(headerLength, columnsIndex);
            }
        }
    }

    @Test
    public void testSerializedSize() throws Throwable
    {
        String tableName = createTable("CREATE TABLE %s (a int, b text, c int, PRIMARY KEY(a, b))");
        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(tableName);

        Pre_C_11206_RowIndexEntry simple = new Pre_C_11206_RowIndexEntry(123);

        DataOutputBuffer buffer = new DataOutputBuffer();
        SerializationHeader header = new SerializationHeader(true, cfs.metadata, cfs.metadata.partitionColumns(), EncodingStats.NO_STATS);
        Pre_C_11206_RowIndexEntry.Serializer serializer = new Pre_C_11206_RowIndexEntry.Serializer(cfs.metadata, BigFormat.latestVersion, header);

        serializer.serialize(simple, buffer);

        assertEquals(buffer.getLength(), serializer.serializedSize(simple));

        // write enough rows to ensure we get a few column index entries
        for (int i = 0; i <= DatabaseDescriptor.getColumnIndexSize() / 4; i++)
            execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 0, String.valueOf(i), i);

        ImmutableBTreePartition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs).build());

        File tempFile = File.createTempFile("row_index_entry_test", null);
        tempFile.deleteOnExit();
        SequentialWriter writer = new SequentialWriter(tempFile);
        ColumnIndex columnIndex = RowIndexEntryTest.ColumnIndex.writeAndBuildIndex(partition.unfilteredIterator(), writer, header, Collections.emptySet(), BigFormat.latestVersion);
        Pre_C_11206_RowIndexEntry withIndex = Pre_C_11206_RowIndexEntry.create(0xdeadbeef, DeletionTime.LIVE, columnIndex);
        IndexInfo.Serializer indexSerializer = cfs.metadata.serializers().indexInfoSerializer(BigFormat.latestVersion, header);

        // sanity check
        assertTrue(columnIndex.columnsIndex.size() >= 3);

        buffer = new DataOutputBuffer();
        serializer.serialize(withIndex, buffer);
        assertEquals(buffer.getLength(), serializer.serializedSize(withIndex));

        // serialization check

        ByteBuffer bb = buffer.buffer();
        DataInputBuffer input = new DataInputBuffer(bb, false);
        serializationCheck(withIndex, indexSerializer, bb, input);

        // test with an output stream that doesn't support a file-pointer
        buffer = new DataOutputBuffer()
        {
            public boolean hasPosition()
            {
                return false;
            }

            public long position()
            {
                throw new UnsupportedOperationException();
            }
        };
        serializer.serialize(withIndex, buffer);
        bb = buffer.buffer();
        input = new DataInputBuffer(bb, false);
        serializationCheck(withIndex, indexSerializer, bb, input);

        //

        bb = buffer.buffer();
        input = new DataInputBuffer(bb, false);
        Pre_C_11206_RowIndexEntry.Serializer.skip(input, BigFormat.latestVersion);
        Assert.assertEquals(0, bb.remaining());
    }

    private static void serializationCheck(Pre_C_11206_RowIndexEntry withIndex, IndexInfo.Serializer indexSerializer, ByteBuffer bb, DataInputBuffer input) throws IOException
    {
        Assert.assertEquals(0xdeadbeef, input.readUnsignedVInt());
        Assert.assertEquals(withIndex.promotedSize(indexSerializer), input.readUnsignedVInt());

        Assert.assertEquals(withIndex.headerLength(), input.readUnsignedVInt());
        Assert.assertEquals(withIndex.deletionTime(), DeletionTime.serializer.deserialize(input));
        Assert.assertEquals(withIndex.columnsIndex().size(), input.readUnsignedVInt());

        int offset = bb.position();
        int[] offsets = new int[withIndex.columnsIndex().size()];
        for (int i = 0; i < withIndex.columnsIndex().size(); i++)
        {
            int pos = bb.position();
            offsets[i] = pos - offset;
            IndexInfo info = indexSerializer.deserialize(input);
            int end = bb.position();

            Assert.assertEquals(indexSerializer.serializedSize(info), end - pos);

            Assert.assertEquals(withIndex.columnsIndex().get(i).offset, info.offset);
            Assert.assertEquals(withIndex.columnsIndex().get(i).width, info.width);
            Assert.assertEquals(withIndex.columnsIndex().get(i).endOpenMarker, info.endOpenMarker);
            Assert.assertEquals(withIndex.columnsIndex().get(i).firstName, info.firstName);
            Assert.assertEquals(withIndex.columnsIndex().get(i).lastName, info.lastName);
        }

        for (int i = 0; i < withIndex.columnsIndex().size(); i++)
            Assert.assertEquals(offsets[i], input.readInt());

        Assert.assertEquals(0, bb.remaining());
    }

    static class Pre_C_11206_RowIndexEntry implements IMeasurableMemory
    {
        private static final long EMPTY_SIZE = ObjectSizes.measure(new Pre_C_11206_RowIndexEntry(0));

        public final long position;

        Pre_C_11206_RowIndexEntry(long position)
        {
            this.position = position;
        }

        protected int promotedSize(IndexInfo.Serializer idxSerializer)
        {
            return 0;
        }

        public static Pre_C_11206_RowIndexEntry create(long position, DeletionTime deletionTime, ColumnIndex index)
        {
            assert index != null;
            assert deletionTime != null;

            // we only consider the columns summary when determining whether to create an IndexedEntry,
            // since if there are insufficient columns to be worth indexing we're going to seek to
            // the beginning of the row anyway, so we might as well read the tombstone there as well.
            if (index.columnsIndex.size() > 1)
                return new Pre_C_11206_RowIndexEntry.IndexedEntry(position, deletionTime, index.partitionHeaderLength, index.columnsIndex);
            else
                return new Pre_C_11206_RowIndexEntry(position);
        }

        /**
         * @return true if this index entry contains the row-level tombstone and column summary.  Otherwise,
         * caller should fetch these from the row header.
         */
        public boolean isIndexed()
        {
            return !columnsIndex().isEmpty();
        }

        public DeletionTime deletionTime()
        {
            throw new UnsupportedOperationException();
        }

        /**
         * The length of the row header (partition key, partition deletion and static row).
         * This value is only provided for indexed entries and this method will throw
         * {@code UnsupportedOperationException} if {@code !isIndexed()}.
         */
        public long headerLength()
        {
            throw new UnsupportedOperationException();
        }

        public List<IndexInfo> columnsIndex()
        {
            return Collections.emptyList();
        }

        public long unsharedHeapSize()
        {
            return EMPTY_SIZE;
        }

        public static class Serializer
        {
            private final IndexInfo.Serializer idxSerializer;
            private final Version version;

            Serializer(CFMetaData metadata, Version version, SerializationHeader header)
            {
                this.idxSerializer = metadata.serializers().indexInfoSerializer(version, header);
                this.version = version;
            }

            public void serialize(Pre_C_11206_RowIndexEntry rie, DataOutputPlus out) throws IOException
            {
                assert version.storeRows() : "We read old index files but we should never write them";

                out.writeUnsignedVInt(rie.position);
                out.writeUnsignedVInt(rie.promotedSize(idxSerializer));

                if (rie.isIndexed())
                {
                    out.writeUnsignedVInt(rie.headerLength());
                    DeletionTime.serializer.serialize(rie.deletionTime(), out);
                    out.writeUnsignedVInt(rie.columnsIndex().size());

                    // Calculate and write the offsets to the IndexInfo objects.

                    int[] offsets = new int[rie.columnsIndex().size()];

                    if (out.hasPosition())
                    {
                        // Out is usually a SequentialWriter, so using the file-pointer is fine to generate the offsets.
                        // A DataOutputBuffer also works.
                        long start = out.position();
                        int i = 0;
                        for (IndexInfo info : rie.columnsIndex())
                        {
                            offsets[i] = i == 0 ? 0 : (int)(out.position() - start);
                            i++;
                            idxSerializer.serialize(info, out);
                        }
                    }
                    else
                    {
                        // Not sure this branch will ever be needed, but if it is called, it has to calculate the
                        // serialized sizes instead of simply using the file-pointer.
                        int i = 0;
                        int offset = 0;
                        for (IndexInfo info : rie.columnsIndex())
                        {
                            offsets[i++] = offset;
                            idxSerializer.serialize(info, out);
                            offset += idxSerializer.serializedSize(info);
                        }
                    }

                    for (int off : offsets)
                        out.writeInt(off);
                }
            }

            public Pre_C_11206_RowIndexEntry deserialize(DataInputPlus in) throws IOException
            {
                if (!version.storeRows())
                {
                    long position = in.readLong();

                    int size = in.readInt();
                    if (size > 0)
                    {
                        DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);

                        int entries = in.readInt();
                        List<IndexInfo> columnsIndex = new ArrayList<>(entries);

                        long headerLength = 0L;
                        for (int i = 0; i < entries; i++)
                        {
                            IndexInfo info = idxSerializer.deserialize(in);
                            columnsIndex.add(info);
                            if (i == 0)
                                headerLength = info.offset;
                        }

                        return new Pre_C_11206_RowIndexEntry.IndexedEntry(position, deletionTime, headerLength, columnsIndex);
                    }
                    else
                    {
                        return new Pre_C_11206_RowIndexEntry(position);
                    }
                }

                long position = in.readUnsignedVInt();

                int size = (int)in.readUnsignedVInt();
                if (size > 0)
                {
                    long headerLength = in.readUnsignedVInt();
                    DeletionTime deletionTime = DeletionTime.serializer.deserialize(in);
                    int entries = (int)in.readUnsignedVInt();
                    List<IndexInfo> columnsIndex = new ArrayList<>(entries);
                    for (int i = 0; i < entries; i++)
                        columnsIndex.add(idxSerializer.deserialize(in));

                    in.skipBytesFully(entries * TypeSizes.sizeof(0));

                    return new Pre_C_11206_RowIndexEntry.IndexedEntry(position, deletionTime, headerLength, columnsIndex);
                }
                else
                {
                    return new Pre_C_11206_RowIndexEntry(position);
                }
            }

            // Reads only the data 'position' of the index entry and returns it. Note that this left 'in' in the middle
            // of reading an entry, so this is only useful if you know what you are doing and in most case 'deserialize'
            // should be used instead.
            static long readPosition(DataInputPlus in, Version version) throws IOException
            {
                return version.storeRows() ? in.readUnsignedVInt() : in.readLong();
            }

            public static void skip(DataInputPlus in, Version version) throws IOException
            {
                readPosition(in, version);
                skipPromotedIndex(in, version);
            }

            private static void skipPromotedIndex(DataInputPlus in, Version version) throws IOException
            {
                int size = version.storeRows() ? (int)in.readUnsignedVInt() : in.readInt();
                if (size <= 0)
                    return;

                in.skipBytesFully(size);
            }

            public int serializedSize(Pre_C_11206_RowIndexEntry rie)
            {
                assert version.storeRows() : "We read old index files but we should never write them";

                int indexedSize = 0;
                if (rie.isIndexed())
                {
                    List<IndexInfo> index = rie.columnsIndex();

                    indexedSize += TypeSizes.sizeofUnsignedVInt(rie.headerLength());
                    indexedSize += DeletionTime.serializer.serializedSize(rie.deletionTime());
                    indexedSize += TypeSizes.sizeofUnsignedVInt(index.size());

                    for (IndexInfo info : index)
                        indexedSize += idxSerializer.serializedSize(info);

                    indexedSize += index.size() * TypeSizes.sizeof(0);
                }

                return TypeSizes.sizeofUnsignedVInt(rie.position) + TypeSizes.sizeofUnsignedVInt(indexedSize) + indexedSize;
            }
        }

        /**
         * An entry in the row index for a row whose columns are indexed.
         */
        private static final class IndexedEntry extends Pre_C_11206_RowIndexEntry
        {
            private final DeletionTime deletionTime;

            // The offset in the file when the index entry end
            private final long headerLength;
            private final List<IndexInfo> columnsIndex;
            private static final long BASE_SIZE =
            ObjectSizes.measure(new IndexedEntry(0, DeletionTime.LIVE, 0, Arrays.asList(null, null)))
            + ObjectSizes.measure(new ArrayList<>(1));

            private IndexedEntry(long position, DeletionTime deletionTime, long headerLength, List<IndexInfo> columnsIndex)
            {
                super(position);
                assert deletionTime != null;
                assert columnsIndex != null && columnsIndex.size() > 1;
                this.deletionTime = deletionTime;
                this.headerLength = headerLength;
                this.columnsIndex = columnsIndex;
            }

            @Override
            public DeletionTime deletionTime()
            {
                return deletionTime;
            }

            @Override
            public long headerLength()
            {
                return headerLength;
            }

            @Override
            public List<IndexInfo> columnsIndex()
            {
                return columnsIndex;
            }

            @Override
            protected int promotedSize(IndexInfo.Serializer idxSerializer)
            {
                long size = TypeSizes.sizeofUnsignedVInt(headerLength)
                            + DeletionTime.serializer.serializedSize(deletionTime)
                            + TypeSizes.sizeofUnsignedVInt(columnsIndex.size()); // number of entries
                for (IndexInfo info : columnsIndex)
                    size += idxSerializer.serializedSize(info);

                size += columnsIndex.size() * TypeSizes.sizeof(0);

                return Ints.checkedCast(size);
            }

            @Override
            public long unsharedHeapSize()
            {
                long entrySize = 0;
                for (IndexInfo idx : columnsIndex)
                    entrySize += idx.unsharedHeapSize();

                return BASE_SIZE
                       + entrySize
                       + deletionTime.unsharedHeapSize()
                       + ObjectSizes.sizeOfReferenceArray(columnsIndex.size());
            }
        }
    }

    @Test
    public void testIndexFor() throws IOException
    {
        DeletionTime deletionInfo = new DeletionTime(FBUtilities.timestampMicros(), FBUtilities.nowInSeconds());

        List<IndexInfo> indexes = new ArrayList<>();
        indexes.add(new IndexInfo(cn(0L), cn(5L), 0, 0, deletionInfo));
        indexes.add(new IndexInfo(cn(10L), cn(15L), 0, 0, deletionInfo));
        indexes.add(new IndexInfo(cn(20L), cn(25L), 0, 0, deletionInfo));

        RowIndexEntry rie = new RowIndexEntry(0L)
        {
            public IndexInfoRetriever openWithIndex(FileHandle indexFile)
            {
                return new IndexInfoRetriever()
                {
                    public IndexInfo columnsIndex(int index)
                    {
                        return indexes.get(index);
                    }

                    public void close()
                    {
                    }
                };
            }

            public int columnsIndexCount()
            {
                return indexes.size();
            }
        };
        
        AbstractSSTableIterator.IndexState indexState = new AbstractSSTableIterator.IndexState(
            null, comp, rie, false, null                                                                                              
        );
        
        assertEquals(0, indexState.indexFor(cn(-1L), -1));
        assertEquals(0, indexState.indexFor(cn(5L), -1));
        assertEquals(1, indexState.indexFor(cn(12L), -1));
        assertEquals(2, indexState.indexFor(cn(17L), -1));
        assertEquals(3, indexState.indexFor(cn(100L), -1));
        assertEquals(3, indexState.indexFor(cn(100L), 0));
        assertEquals(3, indexState.indexFor(cn(100L), 1));
        assertEquals(3, indexState.indexFor(cn(100L), 2));
        assertEquals(3, indexState.indexFor(cn(100L), 3));

        indexState = new AbstractSSTableIterator.IndexState(
            null, comp, rie, true, null
        );

        assertEquals(-1, indexState.indexFor(cn(-1L), -1));
        assertEquals(0, indexState.indexFor(cn(5L), 3));
        assertEquals(0, indexState.indexFor(cn(5L), 2));
        assertEquals(1, indexState.indexFor(cn(17L), 3));
        assertEquals(2, indexState.indexFor(cn(100L), 3));
        assertEquals(2, indexState.indexFor(cn(100L), 4));
        assertEquals(1, indexState.indexFor(cn(12L), 3));
        assertEquals(1, indexState.indexFor(cn(12L), 2));
        assertEquals(1, indexState.indexFor(cn(100L), 1));
        assertEquals(2, indexState.indexFor(cn(100L), 2));
    }
}