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

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import javax.annotation.concurrent.NotThreadSafe;

import com.google.common.base.MoreObjects;

import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.index.sai.disk.ResettableByteBuffersIndexOutput;
import org.apache.cassandra.index.sai.disk.v1.SAICodecUtils;
import org.apache.cassandra.index.sai.utils.IndexEntry;
import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse;
import org.apache.lucene.store.ByteBuffersDataOutput;
import org.apache.lucene.store.DataOutput;
import org.apache.lucene.store.IndexOutput;
import org.apache.lucene.util.ArrayUtil;
import org.apache.lucene.util.BytesRef;
import org.apache.lucene.util.IntroSorter;
import org.apache.lucene.util.Sorter;
import org.apache.lucene.util.bkd.BKDWriter;

import static org.apache.cassandra.index.sai.postings.PostingList.END_OF_STREAM;

/**
 * This is a specialisation of the Lucene {@link BKDWriter} that only writes a single dimension
 * balanced tree.
 * <p>
 * Recursively builds a block balanced tree to assign all incoming points to smaller
 * and smaller rectangles (cells) until the number of points in a given
 * rectangle is &lt;= <code>maxPointsInLeafNode</code>.  The tree is
 * fully balanced, which means the leaf nodes will have between 50% and 100% of
 * the requested <code>maxPointsInLeafNode</code>.  Values that fall exactly
 * on a cell boundary may be in either cell.
 * <p>
 * Visual representation of the disk format:
 * <pre>
 *
 * +========+=======================================+==================+========+
 * | HEADER | LEAF BLOCK LIST                       | BALANCED TREE    | FOOTER |
 * +========+================+=====+================+==================+========+
 *          | LEAF BLOCK (0) | ... | LEAF BLOCK (N) | VALUES PER LEAF  |
 *          +----------------+-----+----------------+------------------|
 *          | ORDER INDEX    |                      | BYTES PER VALUE  |
 *          +----------------+                      +------------------+
 *          | PREFIX         |                      | NUMBER OF LEAVES |
 *          +----------------+                      +------------------+
 *          | VALUES         |                      | MINIMUM VALUE    |
 *          +----------------+                      +------------------+
 *                                                  | MAXIMUM VALUE    |
 *                                                  +------------------+
 *                                                  | TOTAL VALUES     |
 *                                                  +------------------+
 *                                                  | INDEX TREE       |
 *                                                  +--------+---------+
 *                                                  | LENGTH | BYTES   |
 *                                                  +--------+---------+
 *  </pre>
 *
 * <p>
 * <b>NOTE</b>: This can write at most Integer.MAX_VALUE * <code>maxPointsInLeafNode</code> total points.
 * <p>
 * @see BKDWriter
 */
@NotThreadSafe
public class BlockBalancedTreeWriter
{
    // Enable to check that values are added to the tree in correct order and within bounds
    public static final boolean DEBUG = CassandraRelevantProperties.SAI_TEST_BALANCED_TREE_DEBUG_ENABLED.getBoolean();

    // Default maximum number of point in each leaf block
    public static final int DEFAULT_MAX_POINTS_IN_LEAF_NODE = 1024;

    private final int bytesPerValue;
    private final int maxPointsInLeafNode;
    private final byte[] minPackedValue;
    private final byte[] maxPackedValue;
    private long valueCount;

    public BlockBalancedTreeWriter(int bytesPerValue, int maxPointsInLeafNode)
    {
        if (maxPointsInLeafNode <= 0)
            throw new IllegalArgumentException("maxPointsInLeafNode must be > 0; got " + maxPointsInLeafNode);
        if (maxPointsInLeafNode > ArrayUtil.MAX_ARRAY_LENGTH)
            throw new IllegalArgumentException("maxPointsInLeafNode must be <= ArrayUtil.MAX_ARRAY_LENGTH (= " +
                                               ArrayUtil.MAX_ARRAY_LENGTH + "); got " + maxPointsInLeafNode);

        this.maxPointsInLeafNode = maxPointsInLeafNode;
        this.bytesPerValue = bytesPerValue;

        minPackedValue = new byte[bytesPerValue];
        maxPackedValue = new byte[bytesPerValue];
    }

    public long getValueCount()
    {
        return valueCount;
    }

    public int getBytesPerValue()
    {
        return bytesPerValue;
    }

    public int getMaxPointsInLeafNode()
    {
        return maxPointsInLeafNode;
    }

    /**
     * Write the sorted values from an {@link Iterator}.
     * <p>
     * @param treeOutput The {@link IndexOutput} to write the balanced tree to
     * @param iterator An {@link Iterator} of {@link IndexEntry}s containing the terms and postings, sorted in term order
     * @param callback The {@link Callback} used to record the leaf postings for each leaf
     *
     * @return The file pointer to the beginning of the balanced tree
     */
    public long write(IndexOutput treeOutput, Iterator<IndexEntry> iterator, final Callback callback) throws IOException
    {
        SAICodecUtils.writeHeader(treeOutput);

        LeafWriter leafWriter = new LeafWriter(treeOutput, callback);

        while (iterator.hasNext())
        {
            IndexEntry indexEntry = iterator.next();
            long segmentRowId;
            while ((segmentRowId = indexEntry.postingList.nextPosting()) != END_OF_STREAM)
                leafWriter.add(indexEntry.term, segmentRowId);
        }

        valueCount = leafWriter.finish();

        long treeFilePointer = valueCount == 0 ? -1 : treeOutput.getFilePointer();

        // There is only any point in writing the balanced tree if any values were added
        if (treeFilePointer >= 0)
            writeBalancedTree(treeOutput, maxPointsInLeafNode, leafWriter.leafBlockStartValues, leafWriter.leafBlockFilePointers);

        SAICodecUtils.writeFooter(treeOutput);

        return treeFilePointer;
    }

    private void writeBalancedTree(IndexOutput out, int countPerLeaf, List<byte[]> leafBlockStartValues, List<Long> leafBlockFilePointer) throws IOException
    {
        int numInnerNodes = leafBlockStartValues.size();
        byte[] splitValues = new byte[(1 + numInnerNodes) * bytesPerValue];
        int treeDepth = recurseBalanceTree(1, 0, numInnerNodes, 1, splitValues, leafBlockStartValues);
        long[] leafBlockFPs = leafBlockFilePointer.stream().mapToLong(l -> l).toArray();
        byte[] packedIndex = packIndex(leafBlockFPs, splitValues);

        out.writeVInt(countPerLeaf);
        out.writeVInt(bytesPerValue);

        out.writeVInt(leafBlockFPs.length);
        out.writeVInt(Math.min(treeDepth, leafBlockFPs.length));

        out.writeBytes(minPackedValue, 0, bytesPerValue);
        out.writeBytes(maxPackedValue, 0, bytesPerValue);

        out.writeVLong(valueCount);

        out.writeVInt(packedIndex.length);
        out.writeBytes(packedIndex, 0, packedIndex.length);
    }

    /**
     * This can, potentially, be removed in the future by CASSANDRA-18597
     */
    private int recurseBalanceTree(int nodeID, int offset, int count, int treeDepth, byte[] splitValues, List<byte[]> leafBlockStartValues)
    {
        if (count == 1)
        {
            treeDepth++;
            // Leaf index node
            System.arraycopy(leafBlockStartValues.get(offset), 0, splitValues, nodeID * bytesPerValue, bytesPerValue);
        }
        else if (count > 1)
        {
            treeDepth++;
            // Internal index node: binary partition of count
            int countAtLevel = 1;
            int totalCount = 0;
            while (true)
            {
                int countLeft = count - totalCount;
                if (countLeft <= countAtLevel)
                {
                    // This is the last level, possibly partially filled:
                    int lastLeftCount = Math.min(countAtLevel / 2, countLeft);
                    assert lastLeftCount >= 0;
                    int leftHalf = (totalCount - 1) / 2 + lastLeftCount;

                    int rootOffset = offset + leftHalf;

                    System.arraycopy(leafBlockStartValues.get(rootOffset), 0, splitValues, nodeID * bytesPerValue, bytesPerValue);

                    // TODO: we could optimize/specialize, when we know it's simply fully balanced binary tree
                    // under here, to save this while loop on each recursion

                    // Recurse left
                    int leftTreeDepth = recurseBalanceTree(2 * nodeID, offset, leftHalf, treeDepth, splitValues, leafBlockStartValues);

                    // Recurse right
                    int rightTreeDepth = recurseBalanceTree(2 * nodeID + 1, rootOffset + 1, count - leftHalf - 1, treeDepth, splitValues, leafBlockStartValues);
                    return Math.max(leftTreeDepth, rightTreeDepth);
                }
                totalCount += countAtLevel;
                countAtLevel *= 2;
            }
        }
        else
        {
            assert count == 0;
        }
        return treeDepth;
    }

    // Packs the two arrays, representing a balanced binary tree, into a compact byte[] structure.
    private byte[] packIndex(long[] leafBlockFPs, byte[] splitValues) throws IOException
    {
        int numLeaves = leafBlockFPs.length;

        // Possibly rotate the leaf block FPs, if the index is not a fully balanced binary tree (only happens
        // if it was created by TreeWriter).  In this case the leaf nodes may straddle the two bottom
        // levels of the binary tree:
        if (numLeaves > 1)
        {
            int levelCount = 2;
            while (true)
            {
                if (numLeaves >= levelCount && numLeaves <= 2 * levelCount)
                {
                    int lastLevel = 2 * (numLeaves - levelCount);
                    assert lastLevel >= 0;
                    if (lastLevel != 0)
                    {
                        // Last level is partially filled, so we must rotate the leaf FPs to match.  We do this here, after loading
                        // at read-time, so that we can still delta code them on disk at write:
                        long[] newLeafBlockFPs = new long[numLeaves];
                        System.arraycopy(leafBlockFPs, lastLevel, newLeafBlockFPs, 0, leafBlockFPs.length - lastLevel);
                        System.arraycopy(leafBlockFPs, 0, newLeafBlockFPs, leafBlockFPs.length - lastLevel, lastLevel);
                        leafBlockFPs = newLeafBlockFPs;
                    }
                    break;
                }

                levelCount *= 2;
            }
        }

        // Reused while packing the index
        try (ResettableByteBuffersIndexOutput writeBuffer = new ResettableByteBuffersIndexOutput("PackedIndex"))
        {
            // This is the "file" we append the byte[] to:
            List<byte[]> blocks = new ArrayList<>();
            byte[] lastSplitValue = new byte[bytesPerValue];
            int totalSize = recursePackIndex(writeBuffer, leafBlockFPs, splitValues, 0, blocks, 1, lastSplitValue, false);
            // Compact the byte[] blocks into single byte index:
            byte[] index = new byte[totalSize];
            int upto = 0;
            for (byte[] block : blocks)
            {
                System.arraycopy(block, 0, index, upto, block.length);
                upto += block.length;
            }
            assert upto == totalSize;

            return index;
        }
    }

    /**
     * lastSplitValue is the split value previously seen; we use this to prefix-code the split byte[] on each
     * inner node
     */
    private int recursePackIndex(ResettableByteBuffersIndexOutput writeBuffer, long[] leafBlockFPs, byte[] splitValues,
                                 long minBlockFP, List<byte[]> blocks, int nodeID, byte[] lastSplitValue, boolean isLeft) throws IOException
    {
        if (nodeID >= leafBlockFPs.length)
        {
            int leafID = nodeID - leafBlockFPs.length;

            // In the unbalanced case it's possible the left most node only has one child:
            if (leafID < leafBlockFPs.length)
            {
                long delta = leafBlockFPs[leafID] - minBlockFP;
                if (isLeft)
                {
                    assert delta == 0;
                    return 0;
                }
                else
                {
                    assert nodeID == 1 || delta > 0 : "nodeID=" + nodeID;
                    writeBuffer.writeVLong(delta);
                    return appendBlock(writeBuffer, blocks);
                }
            }
            else
            {
                throw new IllegalStateException("Unbalanced tree");
            }
        }
        else
        {
            long leftBlockFP;
            if (!isLeft)
            {
                leftBlockFP = getLeftMostLeafBlockFP(leafBlockFPs, nodeID);
                long delta = leftBlockFP - minBlockFP;
                assert nodeID == 1 || delta > 0;
                writeBuffer.writeVLong(delta);
            }
            else
            {
                // The left tree's left most leaf block FP is always the minimal FP:
                leftBlockFP = minBlockFP;
            }

            int address = nodeID * bytesPerValue;

            // find common prefix with last split value in this dim:
            int prefix = 0;
            for (; prefix < bytesPerValue; prefix++)
            {
                if (splitValues[address + prefix] != lastSplitValue[prefix])
                {
                    break;
                }
            }

            int firstDiffByteDelta;
            if (prefix < bytesPerValue)
            {
                firstDiffByteDelta = (splitValues[address + prefix] & 0xFF) - (lastSplitValue[prefix] & 0xFF);
                // If this is left then we need to negate the delta
                if (isLeft)
                    firstDiffByteDelta = -firstDiffByteDelta;
                assert firstDiffByteDelta > 0;
            }
            else
            {
                firstDiffByteDelta = 0;
            }

            // pack the prefix and delta first diff byte into a single vInt:
            int code = (firstDiffByteDelta * (1 + bytesPerValue) + prefix);

            writeBuffer.writeVInt(code);

            // write the split value, prefix coded vs. our parent's split value:
            int suffix = bytesPerValue - prefix;
            byte[] savSplitValue = new byte[suffix];
            if (suffix > 1)
            {
                writeBuffer.writeBytes(splitValues, address + prefix + 1, suffix - 1);
            }

            byte[] cmp = lastSplitValue.clone();

            System.arraycopy(lastSplitValue, prefix, savSplitValue, 0, suffix);

            // copy our split value into lastSplitValue for our children to prefix-code against
            System.arraycopy(splitValues, address + prefix, lastSplitValue, prefix, suffix);

            int numBytes = appendBlock(writeBuffer, blocks);

            // placeholder for left-tree numBytes; we need this so that at search time if we only need to recurse into
            // the right subtree we can quickly seek to its starting point
            int idxSav = blocks.size();
            blocks.add(null);

            int leftNumBytes = recursePackIndex(writeBuffer, leafBlockFPs, splitValues, leftBlockFP, blocks, 2 * nodeID, lastSplitValue, true);

            if (nodeID * 2 < leafBlockFPs.length)
            {
                writeBuffer.writeVInt(leftNumBytes);
            }
            else
            {
                assert leftNumBytes == 0 : "leftNumBytes=" + leftNumBytes;
            }
            int numBytes2 = Math.toIntExact(writeBuffer.getFilePointer());
            byte[] bytes2 = writeBuffer.toArrayCopy();
            writeBuffer.reset();
            // replace our placeholder:
            blocks.set(idxSav, bytes2);

            int rightNumBytes = recursePackIndex(writeBuffer, leafBlockFPs, splitValues, leftBlockFP, blocks, 2 * nodeID + 1, lastSplitValue, false);

            // restore lastSplitValue to what caller originally passed us:
            System.arraycopy(savSplitValue, 0, lastSplitValue, prefix, suffix);

            assert Arrays.equals(lastSplitValue, cmp);

            return numBytes + numBytes2 + leftNumBytes + rightNumBytes;
        }
    }

    /** Appends the current contents of writeBuffer as another block on the growing in-memory file */
    private int appendBlock(ResettableByteBuffersIndexOutput writeBuffer, List<byte[]> blocks)
    {
        int pos = Math.toIntExact(writeBuffer.getFilePointer());
        byte[] bytes = writeBuffer.toArrayCopy();
        writeBuffer.reset();
        blocks.add(bytes);
        return pos;
    }

    private long getLeftMostLeafBlockFP(long[] leafBlockFPs, int nodeID)
    {
        // TODO: can we do this cheaper, e.g. a closed form solution instead of while loop?  Or
        // change the recursion while packing the index to return this left-most leaf block FP
        // from each recursion instead?
        //
        // Still, the overall cost here is minor: this method's cost is O(log(N)), and while writing
        // we call it O(N) times (N = number of leaf blocks)
        while (nodeID < leafBlockFPs.length)
        {
            nodeID *= 2;
        }
        int leafID = nodeID - leafBlockFPs.length;
        long result = leafBlockFPs[leafID];
        if (result < 0)
        {
            throw new AssertionError(result + " for leaf " + leafID);
        }
        return result;
    }

    interface Callback
    {
        void writeLeafPostings(RowIDAndIndex[] leafPostings, int offset, int count);
    }

    static class RowIDAndIndex
    {
        public int valueOrderIndex;
        public long rowID;

        @Override
        public String toString()
        {
            return MoreObjects.toStringHelper(this)
                              .add("valueOrderIndex", valueOrderIndex)
                              .add("rowID", rowID)
                              .toString();
        }
    }

    /**
     * Responsible for writing the leaf blocks at the beginning of the balanced tree index.
     */
    private class LeafWriter
    {
        private final IndexOutput treeOutput;
        private final List<Long> leafBlockFilePointers = new ArrayList<>();
        private final List<byte[]> leafBlockStartValues = new ArrayList<>();
        private final byte[] leafValues = new byte[maxPointsInLeafNode * bytesPerValue];
        private final long[] leafRowIDs = new long[maxPointsInLeafNode];
        private final RowIDAndIndex[] rowIDAndIndexes = new RowIDAndIndex[maxPointsInLeafNode];
        private final int[] orderIndex = new int[maxPointsInLeafNode];
        private final Callback callback;
        private final ByteBuffersDataOutput leafOrderIndexOutput = new ByteBuffersDataOutput(2 * 1024);
        private final ByteBuffersDataOutput leafBlockOutput = new ByteBuffersDataOutput(32 * 1024);
        private final byte[] packedValue = new byte[bytesPerValue];
        private final byte[] lastPackedValue = new byte[bytesPerValue];

        private long valueCount;
        private int leafValueCount;
        private long lastRowID;

        LeafWriter(IndexOutput treeOutput, Callback callback)
        {
            assert callback != null : "Callback cannot be null in TreeWriter";

            this.treeOutput = treeOutput;
            this.callback = callback;

            for (int x = 0; x < rowIDAndIndexes.length; x++)
            {
                rowIDAndIndexes[x] = new RowIDAndIndex();
            }
        }

        /**
         * Adds a value and row ID to the current leaf block. If the leaf block is full after the addition
         * the current leaf block is written to disk.
         */
        void add(ByteComparable value, long rowID) throws IOException
        {
            ByteSourceInverse.copyBytes(value.asComparableBytes(ByteComparable.Version.OSS50), packedValue);

            if (DEBUG)
                valueInOrder(valueCount + leafValueCount, lastPackedValue, packedValue, 0, rowID, lastRowID);

            System.arraycopy(packedValue, 0, leafValues, leafValueCount * bytesPerValue, bytesPerValue);
            leafRowIDs[leafValueCount] = rowID;
            leafValueCount++;

            if (leafValueCount == maxPointsInLeafNode)
            {
                // We write a block once we hit exactly the max count
                writeLeafBlock();
                leafValueCount = 0;
            }

            if (DEBUG)
                if ((lastRowID = rowID) < 0)
                    throw new AssertionError("row id must be >= 0; got " + rowID);
        }

        /**
         * Write a leaf block if we have unwritten values and return the total number of values added
         */
        public long finish() throws IOException
        {
            if (leafValueCount > 0)
                writeLeafBlock();

            return valueCount;
        }

        private void writeLeafBlock() throws IOException
        {
            assert leafValueCount != 0;
            if (valueCount == 0)
            {
                System.arraycopy(leafValues, 0, minPackedValue, 0, bytesPerValue);
            }
            System.arraycopy(leafValues, (leafValueCount - 1) * bytesPerValue, maxPackedValue, 0, bytesPerValue);

            valueCount += leafValueCount;

            if (leafBlockFilePointers.size() > 0)
            {
                // Save the first (minimum) value in each leaf block except the first, to build the split value index in the end:
                leafBlockStartValues.add(ArrayUtil.copyOfSubArray(leafValues, 0, bytesPerValue));
            }
            leafBlockFilePointers.add(treeOutput.getFilePointer());
            checkMaxLeafNodeCount(leafBlockFilePointers.size());

            // Find the common prefix between the first and last values in the block
            int commonPrefixLength = bytesPerValue;
            int offset = (leafValueCount - 1) * bytesPerValue;
            for (int j = 0; j < bytesPerValue; j++)
            {
                if (leafValues[j] != leafValues[offset + j])
                {
                    commonPrefixLength = j;
                    break;
                }
            }

            treeOutput.writeVInt(leafValueCount);

            for (int x = 0; x < leafValueCount; x++)
            {
                rowIDAndIndexes[x].valueOrderIndex = x;
                rowIDAndIndexes[x].rowID = leafRowIDs[x];
            }

            final Sorter sorter = new IntroSorter()
            {
                RowIDAndIndex pivot;

                @Override
                protected void swap(int i, int j)
                {
                    RowIDAndIndex o = rowIDAndIndexes[i];
                    rowIDAndIndexes[i] = rowIDAndIndexes[j];
                    rowIDAndIndexes[j] = o;
                }

                @Override
                protected void setPivot(int i)
                {
                    pivot = rowIDAndIndexes[i];
                }

                @Override
                protected int comparePivot(int j)
                {
                    return Long.compare(pivot.rowID, rowIDAndIndexes[j].rowID);
                }
            };

            sorter.sort(0, leafValueCount);

            // write the leaf order index: leaf rowID -> orig index
            leafOrderIndexOutput.reset();

            // iterate in row ID order to get the row ID index for the given value order index
            // place into an array to be written as packed ints
            for (int x = 0; x < leafValueCount; x++)
                orderIndex[rowIDAndIndexes[x].valueOrderIndex] = x;

            LeafOrderMap.write(orderIndex, leafValueCount, maxPointsInLeafNode - 1, leafOrderIndexOutput);

            treeOutput.writeVInt((int) leafOrderIndexOutput.size());
            leafOrderIndexOutput.copyTo(treeOutput);

            callback.writeLeafPostings(rowIDAndIndexes, 0, leafValueCount);

            // Write the common prefix for the leaf block
            writeCommonPrefix(treeOutput, commonPrefixLength);

            // Write the run length encoded packed values for the leaf block
            leafBlockOutput.reset();

            if (DEBUG)
                valuesInOrderAndBounds(leafValueCount,
                                       ArrayUtil.copyOfSubArray(leafValues, 0, bytesPerValue),
                                       ArrayUtil.copyOfSubArray(leafValues, (leafValueCount - 1) * bytesPerValue, leafValueCount * bytesPerValue),
                                       leafRowIDs);

            writeLeafBlockPackedValues(leafBlockOutput, commonPrefixLength, leafValueCount);

            leafBlockOutput.copyTo(treeOutput);
        }

        private void checkMaxLeafNodeCount(int numLeaves)
        {
            if (bytesPerValue * (long) numLeaves > ArrayUtil.MAX_ARRAY_LENGTH)
            {
                throw new IllegalStateException("too many nodes; increase maxPointsInLeafNode (currently " + maxPointsInLeafNode + ") and reindex");
            }
        }

        private void writeCommonPrefix(DataOutput treeOutput, int commonPrefixLength) throws IOException
        {
            treeOutput.writeVInt(commonPrefixLength);
            if (commonPrefixLength > 0)
                treeOutput.writeBytes(leafValues, 0, commonPrefixLength);
        }

        private void writeLeafBlockPackedValues(DataOutput out, int commonPrefixLength, int count) throws IOException
        {
            // If all the values are the same (e.g. the common prefix length == bytes per value) then we don't
            // need to write anything. Otherwise, we run length compress the values to disk.
            if (commonPrefixLength != bytesPerValue)
            {
                int compressedByteOffset = commonPrefixLength;
                commonPrefixLength++;
                for (int i = 0; i < count; )
                {
                    // do run-length compression on the byte at compressedByteOffset
                    int runLen = runLen(i, Math.min(i + 0xff, count), compressedByteOffset);
                    assert runLen <= 0xff;
                    byte prefixByte = leafValues[i * bytesPerValue + compressedByteOffset];
                    out.writeByte(prefixByte);
                    out.writeByte((byte) runLen);
                    writeLeafBlockPackedValuesRange(out, commonPrefixLength, i, i + runLen);
                    i += runLen;
                    assert i <= count;
                }
            }
        }

        private void writeLeafBlockPackedValuesRange(DataOutput out, int commonPrefixLength, int start, int end) throws IOException
        {
            for (int i = start; i < end; ++i)
            {
                out.writeBytes(leafValues, i * bytesPerValue + commonPrefixLength, bytesPerValue - commonPrefixLength);
            }
        }

        private int runLen(int start, int end, int byteOffset)
        {
            byte b = leafValues[start * bytesPerValue + byteOffset];
            for (int i = start + 1; i < end; ++i)
            {
                byte b2 = leafValues[i * bytesPerValue + byteOffset];
                assert Byte.toUnsignedInt(b2) >= Byte.toUnsignedInt(b);
                if (b != b2)
                {
                    return i - start;
                }
            }
            return end - start;
        }

        // The following 3 methods are only used when DEBUG is true:

        private void valueInBounds(byte[] packedValues, int packedValueOffset, byte[] minPackedValue, byte[] maxPackedValue)
        {
            if (ByteArrayUtil.compareUnsigned(packedValues,
                                              packedValueOffset,
                                              minPackedValue,
                                              0,
                                              bytesPerValue) < 0)
            {
                throw new AssertionError("value=" + new BytesRef(packedValues, packedValueOffset, bytesPerValue) +
                                         " is < minPackedValue=" + new BytesRef(minPackedValue));
            }

            if (ByteArrayUtil.compareUnsigned(packedValues,
                                              packedValueOffset,
                                              maxPackedValue, 0,
                                              bytesPerValue) > 0)
            {
                throw new AssertionError("value=" + new BytesRef(packedValues, packedValueOffset, bytesPerValue) +
                                         " is > maxPackedValue=" + new BytesRef(maxPackedValue));
            }
        }

        private void valuesInOrderAndBounds(int count, byte[] minPackedValue, byte[] maxPackedValue, long[] rowIds)
        {
            byte[] lastPackedValue = new byte[bytesPerValue];
            long lastRowId = -1;
            for (int i = 0; i < count; i++)
            {
                valueInOrder(i, lastPackedValue, leafValues, i * bytesPerValue, rowIds[i], lastRowId);
                lastRowId = rowIds[i];

                // Make sure this value does in fact fall within this leaf cell:
                valueInBounds(leafValues, i * bytesPerValue, minPackedValue, maxPackedValue);
            }
        }

        private void valueInOrder(long ord, byte[] lastPackedValue, byte[] packedValues, int packedValueOffset, long rowId, long lastRowId)
        {
            if (ord > 0)
            {
                int cmp = ByteArrayUtil.compareUnsigned(lastPackedValue, 0, packedValues, packedValueOffset, bytesPerValue);
                if (cmp > 0)
                {
                    throw new AssertionError("values out of order: last value=" + new BytesRef(lastPackedValue) +
                                             " current value=" + new BytesRef(packedValues, packedValueOffset, bytesPerValue) +
                                             " ord=" + ord);
                }
                if (cmp == 0 && rowId < lastRowId)
                {
                    throw new AssertionError("row IDs out of order: last rowID=" + lastRowId + " current rowID=" + rowId + " ord=" + ord);
                }
            }
            System.arraycopy(packedValues, packedValueOffset, lastPackedValue, 0, bytesPerValue);
        }
    }
}
