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

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import javax.annotation.Nonnull;

import com.google.common.base.Function;

import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DeletionPurger;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.ComplexCellMergeIterator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.RowMergeIterator;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.memory.Cloner;

/**
 * Storage engine representation of a row.
 *
 * A row mainly contains the following informations:
 *   1) Its {@code Clustering}, which holds the values for the clustering columns identifying the row.
 *   2) Its row level informations: the primary key liveness infos and the row deletion (see
 *      {@link #primaryKeyLivenessInfo()} and {@link #deletion()} for more details).
 *   3) Data for the columns it contains, or in other words, it's a (sorted) collection of
 *      {@code ColumnData}.
 *
 * Also note that as for every other storage engine object, a {@code Row} object cannot shadow
 * it's own data. For instance, a {@code Row} cannot contains a cell that is deleted by its own
 * row deletion.
 */
public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
{
    /**
     * The clustering values for this row.
     */
    @Override
    public Clustering<?> clustering();

    /**
     * An in-natural-order collection of the columns for which data (incl. simple tombstones)
     * is present in this row.
     */
    public Collection<ColumnMetadata> columns();


    /**
     * The number of columns for which data (incl. simple tombstones) is present in this row.
     */
    public int columnCount();

    /**
     * The row deletion.
     *
     * This correspond to the last row deletion done on this row.
     *
     * @return the row deletion.
     */
    public Deletion deletion();

    /**
     * Liveness information for the primary key columns of this row.
     * <p>
     * As a row is uniquely identified by its primary key, all its primary key columns
     * share the same {@code LivenessInfo}. This liveness information is what allows us
     * to distinguish between a dead row (it has no live cells and its primary key liveness
     * info is empty) and a live row but where all non PK columns are null (it has no
     * live cells, but its primary key liveness is not empty). Please note that the liveness
     * info (including it's eventually ttl/local deletion time) only apply to the primary key
     * columns and has no impact on the row content.
     * <p>
     * Note in particular that a row may have live cells but no PK liveness info, because the
     * primary key liveness informations are only set on {@code INSERT} (which makes sense
     * in itself, see #6782) but live cells can be added through {@code UPDATE} even if the row
     * wasn't pre-existing (which users are encouraged not to do, but we can't validate).
     */
    public LivenessInfo primaryKeyLivenessInfo();

    /**
     * Whether the row correspond to a static row or not.
     *
     * @return whether the row correspond to a static row or not.
     */
    public boolean isStatic();

    /**
     * Whether the row has no information whatsoever. This means no PK liveness info, no row
     * deletion, no cells and no complex deletion info.
     *
     * @return {@code true} if the row has no data, {@code false} otherwise.
     */
    public boolean isEmpty();

    /**
     * Whether the row has some live information (i.e. it's not just deletion informations).
     * 
     * @param nowInSec the current time to decide what is deleted and what isn't
     * @param enforceStrictLiveness whether the row should be purged if there is no PK liveness info,
     *                              normally retrieved from {@link TableMetadata#enforceStrictLiveness()}
     * @return true if there is some live information
     */
    public boolean hasLiveData(long nowInSec, boolean enforceStrictLiveness);

    /**
     * Returns a cell for a simple column.
     *
     * @param c the simple column for which to fetch the cell.
     * @return the corresponding cell or {@code null} if the row has no such cell.
     */
    public Cell<?> getCell(ColumnMetadata c);

    /**
     * Return a cell for a given complex column and cell path.
     *
     * @param c the complex column for which to fetch the cell.
     * @param path the cell path for which to fetch the cell.
     * @return the corresponding cell or {@code null} if the row has no such cell.
     */
    public Cell<?> getCell(ColumnMetadata c, CellPath path);

    /**
     * The data for a complex column.
     * <p>
     * The returned object groups all the cells for the column, as well as it's complex deletion (if relevant).
     *
     * @param c the complex column for which to return the complex data.
     * @return the data for {@code c} or {@code null} if the row has no data for this column.
     */
    public ComplexColumnData getComplexColumnData(ColumnMetadata c);

    /**
     * Returns the {@link ColumnData} for the specified column.
     *
     * @param c the column for which to fetch the data.
     * @return the data for the column or {@code null} if the row has no data for this column.
     */
    public ColumnData getColumnData(ColumnMetadata c);

    /**
     * An iterable over the cells of this row.
     * <p>
     * The iterable guarantees that cells are returned in order of {@link Cell#comparator}.
     *
     * @return an iterable over the cells of this row.
     */
    public Iterable<Cell<?>> cells();

    /**
     * A collection of the ColumnData representation of this row, for columns with some data (possibly not live) present
     * <p>
     * The data is returned in column order.
     *
     * @return a Collection of the non-empty ColumnData for this row.
     */
    public Collection<ColumnData> columnData();

    /**
     * An iterable over the cells of this row that return cells in "legacy order".
     * <p>
     * In 3.0+, columns are sorted so that all simple columns are before all complex columns. Previously
     * however, the cells where just sorted by the column name. This iterator return cells in that
     * legacy order. It's only ever meaningful for backward/thrift compatibility code.
     *
     * @param metadata the table this is a row of.
     * @param reversed if cells should returned in reverse order.
     * @return an iterable over the cells of this row in "legacy order".
     */
    public Iterable<Cell<?>> cellsInLegacyOrder(TableMetadata metadata, boolean reversed);

    /**
     * Whether the row stores any (non-live) complex deletion for any complex column.
     */
    public boolean hasComplexDeletion();

    /**
     * Whether the row stores any (non-RT) data for any complex column.
     */
    boolean hasComplex();

    /**
     * Whether the row has any deletion info (row deletion, cell tombstone, expired cell or complex deletion).
     *
     * @param nowInSec the current time in seconds to decid if a cell is expired.
     */
    public boolean hasDeletion(long nowInSec);

    /**
     * The smallest local deletion time of all the data in this row (row deletion, primary key liveness, cells and
     * complex deletions), or {@link Cell#MAX_DELETION_TIME} if the row has no deletion nor expiring data.
     * <p>
     * Unlike {@link #hasDeletion(long)}, this value is independent of the current time. In particular, a value of
     * {@link Cell#MAX_DELETION_TIME} guarantees the row carries neither tombstones nor expiring data.
     */
    public long minLocalDeletionTime();

    /**
     * An iterator to efficiently search data for a given column.
     *
     * @return a search iterator for the cells of this row.
     */
    public SearchIterator<ColumnMetadata, ColumnData> searchIterator();

    /**
     * Returns a copy of this row that:
     *   1) only includes the data for the column included by {@code filter}.
     *   2) doesn't include any data that belongs to a dropped column (recorded in {@code metadata}).
     */
    public Row filter(ColumnFilter filter, TableMetadata metadata);

    /**
     * Returns a copy of this row that:
     *   1) only includes the data for the column included by {@code filter}.
     *   2) doesn't include any data that belongs to a dropped column (recorded in {@code metadata}).
     *   3) doesn't include any data that is shadowed/deleted by {@code activeDeletion}.
     *   4) uses {@code activeDeletion} as row deletion iff {@code setActiveDeletionToRow} and {@code activeDeletion} supersedes the row deletion.
     */
    public Row filter(ColumnFilter filter, DeletionTime activeDeletion, boolean setActiveDeletionToRow, TableMetadata metadata);

    /**
     * Requires that {@code function} returns either {@code null} or {@code ColumnData} for the same column.
     *
     * Returns a copy of this row that:
     *   1) {@code function} has been applied to the members of
     *   2) doesn't include any {@code null} results of {@code function}
     *   3) has precisely the provided {@code LivenessInfo} and {@code Deletion}
     */
    public Row transformAndFilter(LivenessInfo info, Deletion deletion, Function<ColumnData, ColumnData> function);

    /**
     * Requires that {@code function} returns either {@code null} or {@code ColumnData} for the same column.
     *
     * Returns a copy of this row that:
     *   1) {@code function} has been applied to the members of
     *   2) doesn't include any {@code null} results of {@code function}
     */
    public Row transformAndFilter(Function<ColumnData, ColumnData> function);

    public Row clone(Cloner cloner);

    /**
     * Returns a copy of this row without any deletion info that should be purged according to {@code purger}.
     *
     * @param purger the {@code DeletionPurger} to use to decide what can be purged.
     * @param nowInSec the current time to decide what is deleted and what isn't (in the case of expired cells).
     * @param enforceStrictLiveness whether the row should be purged if there is no PK liveness info,
     *                              normally retrieved from {@link TableMetadata#enforceStrictLiveness()}
     *
     *        When enforceStrictLiveness is set, rows with empty PK liveness info
     *        and no row deletion are purged.
     *
     *        Currently this is only used by views with normal base column as PK column
     *        so updates to other base columns do not make the row live when the PK column
     *        is not live. See CASSANDRA-11500.
     *
     * @return this row but without any deletion info purged by {@code purger}. If the purged row is empty, returns
     *         {@code null}.
     */
    public Row purge(DeletionPurger purger, long nowInSec, boolean enforceStrictLiveness);

    /**
     * Returns a copy of this row which only include the data queried by {@code filter}, excluding anything _fetched_ for
     * internal reasons but not queried by the user (see {@link ColumnFilter} for details).
     *
     * @param filter the {@code ColumnFilter} to use when deciding what is user queried. This should be the filter
     * that was used when querying the row on which this method is called.
     * @return the row but with all data that wasn't queried by the user skipped.
     */
    public Row withOnlyQueriedData(ColumnFilter filter);

    /*
     * Returns a copy of this row without any data with a timestamp older than the one provided
     */
    public Row purgeDataOlderThan(long timestamp, boolean enforceStrictLiveness);

    /**
     * Returns a copy of this row where all counter cells have they "local" shard marked for clearing.
     */
    public Row markCounterLocalToBeCleared();

    /**
     * Returns a copy of this row where all live timestamp have been replaced by {@code newTimestamp} and every deletion
     * timestamp by {@code newTimestamp - 1}.
     *
     * @param newTimestamp the timestamp to use for all live data in the returned row.
     *
     * @see Commit for why we need this.
     */
    public Row updateAllTimestamp(long newTimestamp);

    public Row updateTimesAndPathsForAccord(@Nonnull Function<Cell, CellPath> cellToMaybeNewListPath, long newTimestamp, long newLocalDeletionTime);

    /**
     * Returns a copy of this row with the new deletion as row deletion if it is more recent
     * than the current row deletion.
     * <p>
     * WARNING: this method <b>does not</b> check that nothing in the row is shadowed by the provided
     * deletion and if that is the case, the created row will be <b>invalid</b>. It is thus up to the
     * caller to verify that this is not the case and the only reasonable use case of this is probably
     * when the row and the deletion comes from the same {@code UnfilteredRowIterator} since that gives
     * use this guarantee.
     */
    public Row withRowDeletion(DeletionTime deletion);

    public int dataSize();

    public long unsharedHeapSizeExcludingData();

    public String toString(TableMetadata metadata, boolean fullDetails);
    public long unsharedHeapSize();

    /**
     * Apply a function to every column in a row
     */
    public void apply(Consumer<ColumnData> function);

    /**
     * Apply a function to every column in a row
     */
    public <A> void apply(BiConsumer<A, ColumnData> function, A arg);

    /**
     * Apply an accumulation funtion to every column in a row
     */

    public long accumulate(LongAccumulator<ColumnData> accumulator, long initialValue);

    public long accumulate(LongAccumulator<ColumnData> accumulator, Comparator<ColumnData> comparator, ColumnData from, long initialValue);

    public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, long initialValue);

    public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, Comparator<ColumnData> comparator, ColumnData from, long initialValue);

    /**
     * A row deletion/tombstone.
     * <p>
     * A row deletion mostly consists of the time of said deletion, but there is 2 variants: shadowable
     * and regular row deletion.
     * <p>
     * A shadowable row deletion only exists if the row has no timestamp. In other words, the deletion is only
     * valid as long as no newer insert is done (thus setting a row timestamp; note that if the row timestamp set
     * is lower than the deletion, it is shadowed (and thus ignored) as usual).
     * <p>
     * That is, if a row has a shadowable deletion with timestamp A and an update is made to that row with a
     * timestamp B such that {@code B > A} (and that update sets the row timestamp), then the shadowable deletion is 'shadowed'
     * by that update. A concrete consequence is that if said update has cells with timestamp lower than A, then those
     * cells are preserved(since the deletion is removed), and this is contrary to a normal (regular) deletion where the
     * deletion is preserved and such cells are removed.
     * <p>
     * Currently, the only use of shadowable row deletions is Materialized Views, see CASSANDRA-10261.
     */
    public static class Deletion
    {
        public static final Deletion LIVE = new Deletion(DeletionTime.LIVE, false);
        private static final long EMPTY_SIZE = ObjectSizes.measure(DeletionTime.build(0, 0));

        private final DeletionTime time;
        private final boolean isShadowable;

        public Deletion(DeletionTime time, boolean isShadowable)
        {
            assert !time.isLive() || !isShadowable;
            this.time = time;
            this.isShadowable = isShadowable;
        }

        public static Deletion regular(DeletionTime time)
        {
            return time.isLive() ? LIVE : new Deletion(time, false);
        }

        /** @deprecated See CAASSANDRA-10261 */
        @Deprecated(since = "4.0")
        public static Deletion shadowable(DeletionTime time)
        {
            return new Deletion(time, true);
        }

        /**
         * The time of the row deletion.
         *
         * @return the time of the row deletion.
         */
        public DeletionTime time()
        {
            return time;
        }

        /**
         * Whether the deletion is a shadowable one or not.
         *
         * @return whether the deletion is a shadowable one. Note that if {@code isLive()}, then this is
         * guarantee to return {@code false}.
         */
        public boolean isShadowable()
        {
            return isShadowable;
        }

        /**
         * Wether the deletion is live or not, that is if its an actual deletion or not.
         *
         * @return {@code true} if this represents no deletion of the row, {@code false} if that's an actual
         * deletion.
         */
        public boolean isLive()
        {
            return time().isLive();
        }

        public boolean supersedes(DeletionTime that)
        {
            return time.supersedes(that);
        }

        public boolean supersedes(Deletion that)
        {
            return time.supersedes(that.time);
        }

        public boolean isShadowedBy(LivenessInfo primaryKeyLivenessInfo)
        {
            return isShadowable && primaryKeyLivenessInfo.timestamp() > time.markedForDeleteAt();
        }

        public boolean deletes(LivenessInfo info)
        {
            return time.deletes(info);
        }

        public boolean deletes(Cell<?> cell)
        {
            return time.deletes(cell);
        }

        public void digest(Digest digest)
        {
            time.digest(digest);
            digest.updateWithBoolean(isShadowable);
        }

        public int dataSize()
        {
            return time.dataSize() + 1;
        }

        @Override
        public boolean equals(Object o)
        {
            if(!(o instanceof Deletion))
                return false;
            Deletion that = (Deletion)o;
            return this.time.equals(that.time) && this.isShadowable == that.isShadowable;
        }

        public long unsharedHeapSize()
        {
            if(this == LIVE)
                return 0;

            return EMPTY_SIZE + time().unsharedHeapSize();
        }

        @Override
        public final int hashCode()
        {
            return Objects.hash(time, isShadowable);
        }

        @Override
        public String toString()
        {
            return String.format("%s%s", time, isShadowable ? "(shadowable)" : "");
        }
    }

    /**
     * Interface for building rows.
     * <p>
     * The builder of a row should always abid to the following rules:
     *   1) {@link #newRow} is always called as the first thing for the row.
     *   2) {@link #addPrimaryKeyLivenessInfo} and {@link #addRowDeletion}, if called, are called before
     *      any {@link #addCell}/{@link #addComplexDeletion} call.
     *   3) {@link #build} is called to construct the new row. The builder can then be reused.
     *
     * There is 2 variants of a builder: sorted and unsorted ones. A sorted builder expects user to abid to the
     * following additional rules:
     *   4) Calls to {@link #addCell}/{@link #addComplexDeletion} are done in strictly increasing column order.
     *      In other words, all calls to these methods for a give column {@code c} are done after any call for
     *      any column before {@code c} and before any call for any column after {@code c}.
     *   5) Calls to {@link #addCell} are further done in strictly increasing cell order (the one defined by
     *      {@link Cell#comparator}. That is, for a give column, cells are passed in {@code CellPath} order.
     *   6) No shadowed data should be added. Concretely, this means that if a a row deletion is added, it doesn't
     *      deletes the row timestamp or any cell added later, and similarly no cell added is deleted by the complex
     *      deletion of the column this is a cell of.
     *
     * An unsorted builder will not expect those last rules however: {@link #addCell} and {@link #addComplexDeletion}
     * can be done in any order. And in particular unsorted builder allows multiple calls for the same column/cell. In
     * that latter case, the result will follow the usual reconciliation rules (so equal cells are reconciled with
     * {@link Cells#reconcile} and the "biggest" of multiple complex deletion for the same column wins).
     */
    public interface Builder
    {
        /**
         * Creates a copy of this {@code Builder}.
         * @return a copy of this {@code Builder}
         */
        public Builder copy();

        /**
         * Whether the builder is a sorted one or not.
         *
         * @return if the builder requires calls to be done in sorted order or not (see above).
         */
        public boolean isSorted();

        /**
         * Prepares the builder to build a new row of clustering {@code clustering}.
         * <p>
         * This should always be the first call for a given row.
         *
         * @param clustering the clustering for the new row.
         */
        public void newRow(Clustering<?> clustering);

        /**
         * The clustering for the row that is currently being built.
         *
         * @return the clustering for the row that is currently being built, or {@code null} if {@link #newRow} hasn't
         * yet been called.
         */
        public Clustering<?> clustering();

        /**
         * Adds the liveness information for the partition key columns of this row.
         *
         * This call is optional (skipping it is equivalent to calling {@code addPartitionKeyLivenessInfo(LivenessInfo.NONE)}).
         *
         * @param info the liveness information for the partition key columns of the built row.
         */
        public void addPrimaryKeyLivenessInfo(LivenessInfo info);

        /**
         * Adds the deletion information for this row.
         *
         * This call is optional and can be skipped if the row is not deleted.
         *
         * @param deletion the row deletion time, or {@code Deletion.LIVE} if the row isn't deleted.
         */
        public void addRowDeletion(Deletion deletion);

        /**
         * Adds a cell to this builder.
         *
         * @param cell the cell to add.
         */
        public void addCell(Cell<?> cell);

        /**
         * Adds a complex deletion.
         *
         * @param column the column for which to add the {@code complexDeletion}.
         * @param complexDeletion the complex deletion time to add.
         */
        public void addComplexDeletion(ColumnMetadata column, DeletionTime complexDeletion);

        /**
         * Builds and return built row.
         *
         * @return the last row built by this builder.
         */
        public Row build();
    }

    /**
     * Row builder interface geared towards human.
     * <p>
     * Where the {@link Builder} deals with building rows efficiently from internal objects ({@code Cell}, {@code
     * LivenessInfo}, ...), the {@code SimpleBuilder} is geared towards building rows from string column name and
     * 'native' values (string for text, ints for numbers, et...). In particular, it is meant to be convenient, not
     * efficient, and should be used only in place where performance is not of the utmost importance (it is used to
     * build schema mutation for instance).
     * <p>
     * Also note that contrarily to {@link Builder}, the {@code SimpleBuilder} API has no {@code newRow()} method: it is
     * expected that the clustering of the row built is provided by the constructor of the builder.
     */
    public interface SimpleBuilder
    {
        /**
         * Sets the timestamp to use for the following additions.
         * <p>
         * Note that the for non-compact tables, this method must be called before any column addition for this
         * timestamp to be used for the row {@code LivenessInfo}.
         *
         * @param timestamp the timestamp to use for following additions. If that timestamp hasn't been set, the current
         * time in microseconds will be used.
         * @return this builder.
         */
        public SimpleBuilder timestamp(long timestamp);

        /**
         * Sets the ttl to use for the following additions.
         * <p>
         * Note that the for non-compact tables, this method must be called before any column addition for this
         * ttl to be used for the row {@code LivenessInfo}.
         *
         * @param ttl the ttl to use for following additions. If that ttl hasn't been set, no ttl will be used.
         * @return this builder.
         */
        public SimpleBuilder ttl(int ttl);

        /**
         * Adds a value to a given column.
         *
         * @param columnName the name of the column for which to add a new value.
         * @param value the value to add, which must be of the proper type for {@code columnName}. This can be {@code
         * null} in which case the this is equivalent to {@code delete(columnName)}.
         * @return this builder.
         */
        public SimpleBuilder add(String columnName, Object value);

        /**
         * Appends new values to a given non-frozen collection column.
         * <p>
         * This method is similar to {@code add()} but the collection elements added through this method are "appended"
         * to any pre-exising elements. In other words, this is like {@code add()} except that it doesn't delete the
         * previous value of the collection. This can only be called on non-frozen collection columns.
         * <p>
         * Note that this method can be used in replacement of {@code add()} if you know that there can't be any
         * pre-existing value for that column, in which case this is slightly less expensive as it avoid the collection
         * tombstone inherent to {@code add()}.
         *
         * @param columnName the name of the column for which to add a new value, which must be a non-frozen collection.
         * @param value the value to add, which must be of the proper type for {@code columnName} (in other words, it
         * <b>must</b> be a collection).
         * @return this builder.
         *
         * @throws IllegalArgumentException if columnName is not a non-frozen collection column.
         */
        public SimpleBuilder appendAll(String columnName, Object value);

        /**
         * Deletes the whole row.
         * <p>
         * If called, this is generally the only method called on the builder (outside of {@code timestamp()}.
         *
         * @return this builder.
         */
        public SimpleBuilder delete();

        /**
         * Deletes the whole row with a timestamp that is just before the new data's timestamp, to make sure no expired
         * data remains on the row.
         *
         * @return this builder.
         */
        public SimpleBuilder deletePrevious();

        /**
         * Removes the value for a given column (creating a tombstone).
         *
         * @param columnName the name of the column to delete.
         * @return this builder.
         */
        public SimpleBuilder delete(String columnName);

        /**
         * Don't include any primary key {@code LivenessInfo} in the built row.
         *
         * @return this builder.
         */
        public SimpleBuilder noPrimaryKeyLivenessInfo();

        /**
         * Returns the built row.
         *
         * @return the built row.
         */
        public Row build();
    }

    /**
     * Utility class to help merging rows from multiple inputs (UnfilteredRowIterators).
     */
    public static class Merger
    {
        private final Row[] rows;
        private final List<Iterator<ColumnData>> columnDataIterators;

        private Clustering<?> clustering;
        private int rowsToMerge;
        private int lastRowSet = -1;

        private static final ColumnData[] EMPTY_DATA_BUFFER = new ColumnData[0];

        private ColumnData[] dataBuffer = EMPTY_DATA_BUFFER;
        private int dataBufferSize;
        private final ColumnDataReducer columnDataReducer;

        public Merger(int size, boolean hasComplex)
        {
            this.rows = new Row[size];
            this.columnDataIterators = new ArrayList<>(size);
            this.columnDataReducer = new ColumnDataReducer(size, hasComplex);
        }

        public void clear()
        {
            Arrays.fill(dataBuffer, 0, dataBufferSize, null);
            dataBufferSize = 0;
            Arrays.fill(rows, null);
            columnDataIterators.clear();
            rowsToMerge = 0;
            lastRowSet = -1;
        }

        public void add(int i, Row row)
        {
            clustering = row.clustering();
            rows[i] = row;
            ++rowsToMerge;
            lastRowSet = i;
        }

        public Row merge(DeletionTime activeDeletion)
        {
            // If for this clustering we have only one row version and have no activeDeletion (i.e. nothing to filter out),
            // then we can just return that single row
            if (rowsToMerge == 1 && activeDeletion.isLive())
            {
                Row row = rows[lastRowSet];
                assert row != null;
                return row;
            }

            LivenessInfo rowInfo = LivenessInfo.EMPTY;
            Deletion rowDeletion = Deletion.LIVE;
            int columnsCountEstimation = 0;
            // Track the smallest local deletion time across all inputs: if none of them carries any deletion or
            // expiring data (i.e. this stays at MAX_DELETION_TIME), the merged row can't either, so we can hand the
            // value to BTreeRow.create() below and skip the full btree scan it would otherwise do to recompute it.
            long minDeletionTime = Cell.MAX_DELETION_TIME;
            for (Row row : rows)
            {
                if (row == null)
                    continue;

                if (row.primaryKeyLivenessInfo().supersedes(rowInfo))
                    rowInfo = row.primaryKeyLivenessInfo();
                if (row.deletion().supersedes(rowDeletion))
                    rowDeletion = row.deletion();

                minDeletionTime = Math.min(minDeletionTime, row.minLocalDeletionTime());

                columnDataIterators.add(row.iterator());
                columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
            }

            if (rowDeletion.isShadowedBy(rowInfo))
                rowDeletion = Deletion.LIVE;

            if (rowDeletion.supersedes(activeDeletion))
                activeDeletion = rowDeletion.time();
            else
                rowDeletion = Deletion.LIVE;

            if (activeDeletion.deletes(rowInfo))
                rowInfo = LivenessInfo.EMPTY;

            // try to estimate and set a potential target capacity
            if (dataBuffer.length < columnsCountEstimation)
                dataBuffer = new ColumnData[columnsCountEstimation];

            columnDataReducer.setActiveDeletion(activeDeletion);
            Iterator<ColumnData> merged = RowMergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
            while (merged.hasNext())
            {
                ColumnData data = merged.next();
                if (data != null)
                {
                    ensureDataBufferCapacity();
                    dataBuffer[dataBufferSize++] = data;
                }
            }

            // Because some data might have been shadowed by the 'activeDeletion', we could have an empty row
            if (rowInfo.isEmpty() && rowDeletion.isLive() && dataBufferSize == 0)
                return null;

            try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer))
            {
                Object[] tree = BTree.build(it, dataBufferSize, UpdateFunction.noOp());
                // If none of the merged rows had any deletion or expiring data, neither does the result, so we can
                // pass the already-known min local deletion time and avoid rescanning the whole btree to recompute it.
                return minDeletionTime == Cell.MAX_DELETION_TIME
                       ? BTreeRow.create(clustering, rowInfo, rowDeletion, tree, Cell.MAX_DELETION_TIME)
                       : BTreeRow.create(clustering, rowInfo, rowDeletion, tree);
            }
        }

        private void ensureDataBufferCapacity()
        {
            if (dataBufferSize == dataBuffer.length)
                // increase capacity by 50%, use 4 as a default capacity
                dataBuffer = Arrays.copyOf(dataBuffer, Math.max(dataBuffer.length + (dataBuffer.length >> 1), 4));
        }

        public Clustering<?> mergedClustering()
        {
            return clustering;
        }

        public Row[] mergedRows()
        {
            return rows;
        }

        private static class ColumnDataReducer extends RowMergeIterator.Reducer<ColumnData, ColumnData>
        {
            private ColumnMetadata column;
            private final ColumnData[] versions;
            private int versionsSize;

            private DeletionTime activeDeletion;

            private final ComplexColumnData.Builder complexBuilder;
            private final List<Iterator<Cell<?>>> complexCells;
            private final CellReducer cellReducer;

            public ColumnDataReducer(int size, boolean hasComplex)
            {
                this.versions = new ColumnData[size];
                this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null;
                this.complexCells = hasComplex ? new ArrayList<>(size) : null;
                this.cellReducer = new CellReducer();
            }

            public void setActiveDeletion(DeletionTime activeDeletion)
            {
                this.activeDeletion = activeDeletion;
            }

            public void reduce(int idx, ColumnData data)
            {
                if (useColumnMetadata(data.column()))
                    column = data.column();

                versions[versionsSize++] = data;
            }

            /**
             * Determines whether {@code dataColumn} should replace the currently selected column metadata,
             * i.e. whether no column has been selected yet or {@code dataColumn} is a newer version.
             * @param dataColumn the candidate {@code ColumnMetadata} to evaluate.
             * @return {@code true} if {@code dataColumn} should be used, {@code false} otherwise.
             */
            private boolean useColumnMetadata(ColumnMetadata dataColumn)
            {
                ColumnMetadata currentColumn = column;
                if (currentColumn == null)
                    return true;
                if (currentColumn == dataColumn)
                    return false;

                return ColumnMetadataVersionComparator.INSTANCE.compare(currentColumn, dataColumn) < 0;
            }

            protected ColumnData getReduced()
            {
                if (column.isSimple())
                {
                    Cell<?> merged = null;
                    for (int i = 0; i < versionsSize; i++)
                    {
                        Cell<?> cell = (Cell<?>) versions[i];
                        if (!activeDeletion.deletes(cell))
                            merged = merged == null ? cell : Cells.reconcile(merged, cell);
                    }
                    return merged;
                }
                else
                {
                    complexBuilder.newColumn(column);
                    complexCells.clear();
                    DeletionTime complexDeletion = DeletionTime.LIVE;
                    for (int i = 0; i < versionsSize; i++)
                    {
                        ColumnData data = versions[i];
                        ComplexColumnData cd = (ComplexColumnData)data;
                        if (cd.complexDeletion().supersedes(complexDeletion))
                            complexDeletion = cd.complexDeletion();
                        complexCells.add(cd.iterator());
                    }

                    if (complexDeletion.supersedes(activeDeletion))
                    {
                        cellReducer.setActiveDeletion(complexDeletion);
                        complexBuilder.addComplexDeletion(complexDeletion);
                    }
                    else
                    {
                        cellReducer.setActiveDeletion(activeDeletion);
                    }

                    Iterator<Cell<?>> cells = ComplexCellMergeIterator.get(complexCells, Cell.comparator, cellReducer);
                    while (cells.hasNext())
                    {
                        Cell<?> merged = cells.next();
                        if (merged != null)
                            complexBuilder.addCell(merged);
                    }
                    return complexBuilder.build();
                }
            }

            protected void onKeyChange()
            {
                column = null;
                Arrays.fill(versions, 0, versionsSize, null);
                versionsSize = 0;
            }
        }

        private static class CellReducer extends ComplexCellMergeIterator.Reducer<Cell<?>, Cell<?>>
        {
            private DeletionTime activeDeletion;
            private Cell<?> merged;

            public void setActiveDeletion(DeletionTime activeDeletion)
            {
                this.activeDeletion = activeDeletion;
                onKeyChange();
            }

            public void reduce(int idx, Cell<?> cell)
            {
                if (!activeDeletion.deletes(cell))
                    merged = merged == null ? cell : Cells.reconcile(merged, cell);
            }

            protected Cell<?> getReduced()
            {
                return merged;
            }

            protected void onKeyChange()
            {
                merged = null;
            }
        }
    }
}
