ClickHouse Replicates a Column With an Index Instead of Copying It

TL;DR: A hash join fans out: one left row can match ten right rows, and the right-hand columns have to appear ten times in the output. ColumnReplicated refuses to copy them. It keeps the original column and stores, per output row, the index of the source row it should read — one UInt8/UInt16/UInt32/UInt64 per row instead of the whole value. The index column (ColumnIndex) even widens its own integer type on demand, and is shared between all the columns that were replicated together.

A join is a row multiplier. users JOIN orders ON users.id = orders.user_id produces one output row per matching pair, and the right-hand table’s columns have to be repeated in the output — once per match.

The naive implementation is column->replicate(offsets), and it is exactly what you would expect: a loop that copies the value of row i of the right table into every output slot that matched it. For a 10x fan-out, that is ten times the memory of the right table, and for a 1000x fan-out it is a thousand times.

But the right table’s columns are usually not read at all. Most queries join, then aggregate, then throw away the joined columns. The copies are pure waste — and worse, they are copies of data that is already in memory somewhere else.

ClickHouse’s answer is to make the repetition implicit: store the source column once, and store the mapping as an index.

The class

cpp
// src/Columns/ColumnReplicated.h:12
/** Column for replicated representation.
 *  It stores original column and indexes in this column.
 *  Used to perform lazy column replication.
 */

The state is two columns and a cache:

cpp
// src/Columns/ColumnReplicated.h:200
private:
    WrappedPtr nested_column;
    ColumnIndex indexes;

    /// Unique id taken from static global_id_counter field at creation.
    /// It's used as the key in the insertion cache.
    UInt64 id;

plus an insertion_cache — a map from the source column’s id to a map of source_index -> inserted_index — that remembers which values have already been copied, so a merge does not copy the same row ten times.

Every accessor is a double indirection — read the index, then read the source:

cpp
// src/Columns/ColumnReplicated.cpp:123
std::string_view ColumnReplicated::getDataAt(size_t n) const
{
    return nested_column->getDataAt(indexes.getIndexAt(n));
}

Every accessor is the same double indirection: read the index, then read the source (isDefaultAt, isNullAt, getUInt, get — all of them).

The column’s own size is the size of the index, not of the data (src/Columns/ColumnReplicated.h:65), and the type is the source column’s type, so a ColumnReplicated over a Nullable(String) is a Nullable(String) as far as the rest of the engine is concerned. The one thing that is not inherited is contiguity — a replicated column is by definition not a flat array of values:

cpp
// src/Columns/ColumnReplicated.h:180
    bool isFixedAndContiguous() const override { return false; }

From offsets to indexes

The join produces, for each left row, the number of right rows it matched. That is an offsets array: offsets[i] right rows for left row i. The replicated representation wants the opposite: for each output row, which right row it came from.

The conversion is a run-length expansion:

cpp
// src/Columns/ColumnReplicated.cpp:749
template <typename T>
ColumnPtr convertOffsetsToIndexesImpl(const IColumn::Offsets & offsets)
{
    auto result = ColumnVector<T>::create();
    auto & data = result->getData();
    data.reserve_exact(offsets.back());
    for (size_t i = 0; i != offsets.size(); ++i)
        data.resize_fill(data.size() + offsets[i] - offsets[i - 1], static_cast<T>(i));
    return result;
}

offsets = [2, 0, 3] becomes indexes = [0, 1, 0, 1, 2] — left row 0 matched right rows 0 and 1, left row 1 matched nothing, left row 2 matched right rows 0, 1 and 2.

And the index’s integer type is chosen to be exactly as wide as it needs to be:

cpp
// src/Columns/ColumnReplicated.cpp:762
ColumnPtr convertOffsetsToIndexes(const IColumn::Offsets & offsets)
{
    size_t max_index = offsets.size();
    if (max_index <= std::numeric_limits<UInt8>::max())
        return convertOffsetsToIndexesImpl<UInt8>(offsets);
    if (max_index <= std::numeric_limits<UInt16>::max())
        return convertOffsetsToIndexesImpl<UInt16>(offsets);
    if (max_index <= std::numeric_limits<UInt32>::max())
        return convertOffsetsToIndexesImpl<UInt32>(offsets);
    return convertOffsetsToIndexesImpl<UInt64>(offsets);
}

So a join of two 100-row tables stores its replication index in one byte per row.

`ColumnIndex`: a column that widens itself

ColumnIndex is not a Column — it is a small wrapper around one:

cpp
// src/Columns/ColumnIndex.h:11
/// Wrapper around ColumnVector to store indexes.
/// Supports dynamic change of underlying index type (UInt8 -> UInt16 -> UInt32 -> UInt64).
class ColumnIndex
{

The widening is the interesting part. An index column starts life as UInt8 and becomes UInt16, UInt32, UInt64 when the values no longer fit:

cpp
// src/Columns/ColumnIndex.cpp:135
void ColumnIndex::expandType()
{
    auto expand = [&](auto type)
    {
        using CurIndexType = decltype(type);
        constexpr auto next_size = NumberTraits::nextSize(sizeof(CurIndexType));
        if (next_size == sizeof(CurIndexType))
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Can't expand indexes type for ColumnLowCardinality from type: {}",
                            demangle(typeid(CurIndexType).name()));

        using NewIndexType = typename NumberTraits::Construct<false, false, next_size>::Type;
        convertIndexes<NewIndexType>();
    };

    callForType(std::move(expand), size_of_type);
}

and the insert path checks the bound before every value:

cpp
// src/Columns/ColumnIndex.cpp:198
void ColumnIndex::insertIndex(size_t index)
{
    while (index > getMaxIndexForCurrentType())
        expandType();

    auto insert = [&]<typename CurIndexType>(CurIndexType /*type_value*/)
    {
        /// `size_of_type` tracks the type of `indexes` and `expandType` keeps the two in sync, so
        /// this is `static_cast` in release. `getIndexesData` would run a `typeid_cast` instead, and
        /// this is the per-value path.
        assert_cast<ColumnVector<CurIndexType> *>(indexes.get())->getData().push_back(static_cast<CurIndexType>(index));
    };

    callForType(std::move(insert), size_of_type);

The point is that the reader never pays for width it does not use. A ColumnReplicated over a 10-row source has a one-byte-per-row index; over a 100 000-row source it has two. The type is discovered from the data on deserialization (getSizeOfIndexType sniffs UInt8/UInt16/UInt32/ UInt64 from the column’s actual type), and the width is carried on the wire in the Native format (see below).

Where the replication happens

The join is the main producer. HashJoinResult::appendRightColumns decides per column whether to copy or to index:

cpp
// src/Interpreters/HashJoin/HashJoinResult.cpp:79
static void replicateColumnLazily(ColumnPtr & column, const IColumn::Offsets & offsets, ColumnPtr & indexes, bool force_lazy_replication)
{
    if ((force_lazy_replication && !column->isConst()) || isLazyReplicationUseful(column))
    {
        if (!indexes)
            indexes = convertOffsetsToIndexes(offsets);
        column = ColumnReplicated::create(column, indexes);
    }
    else
    {
        column = column->replicate(offsets);
    }
}

The predicate that decides whether the trick is worth it:

cpp
// src/Columns/ColumnReplicated.cpp:774
bool isLazyReplicationUseful(const ColumnPtr & column)
{
    return !column->isConst() && !column->isReplicated() && !column->lowCardinality() && (!column->isFixedAndContiguous() || column->sizeOfValueIfFixed() > 8);
}

Read that as four rules:

Not a Const — a constant has no rows to index.
Not already replicated — no point nesting.
Not LowCardinality — its dictionary is already shared; the wrapper would only add an indirection.
Either not fixed-and-contiguous, or wider than 8 bytes — if a value fits in 8 bytes, the copy is cheaper than the index lookup.

The setting that turns the whole thing on:

cpp
// src/Core/Settings.cpp:8157
    DECLARE(Bool, enable_lazy_columns_replication, true, R"(
Enables lazy columns replication in `JOIN`, `ARRAY JOIN` and lambda captures of higher-order functions (e.g. `arrayMap`), it allows to avoid unnecessary copy of the same rows multiple times in memory.
)", 0, \

It is on by default since 25.11.

The same machinery serves ARRAY JOIN, where the “right” side is the array being flattened (replicateColumnsLazily, src/Interpreters/ExpressionActions.cpp:823).

The shared index

Here is the part that makes this more than a per-column trick. When a join replicates six right-hand columns, all six get the same index column — the same ColumnPtr, shared. And a transformation applied to the replicated columns (a sort, a filter, a permutation) must be applied to the index once, not six times.

cpp
// src/Columns/ColumnReplicated.cpp:794
        auto & col = columns[pos];
        if (col->isReplicated())
        {
            const auto & replicated_col = typeid_cast<const ColumnReplicated &>(*col);
            const auto & src_index = replicated_col.getIndexesColumn();
            auto [it, inserted] = transformed_indexes.try_emplace(src_index.get(), nullptr);
            if (inserted)
                it->second = index_transform(src_index);
            col = ColumnReplicated::create(replicated_col.getNestedColumn(), it->second);
        }
        else
            non_replicated_transform(col);

The map is keyed by the pointer of the index column. Two replicated columns that share an index column share the transformed index too. A sort of the join output permutes the index once and reuses it for every replicated column.

cpp
// src/Interpreters/sortBlock.cpp:457
    transformColumnsWithSharedIndex(columns, [&](const ColumnPtr & col) { return col->permute(permutation, permutation.size()); });

Functions compute on the deduplicated rows

The second payoff is bigger than the memory saving. When a function takes a replicated column as an argument, it does not have to compute the function once per output row — it computes it once per source row, and re-wraps the result with the same index:

cpp
// src/Functions/IFunction.cpp:689
        /// If we have only constants and replicated columns with the same indexes
        /// we can execute function on nested columns and create replicated column
        /// from the result using common indexes.

The check is that all the replicated arguments share the same index column:

cpp
// src/Functions/IFunction.cpp:699
            if (const auto * column_replicated = typeid_cast<const ColumnReplicated *>(argument.column.get()))
            {
                nested_columns.push_back(column_replicated->getNestedColumn());
                if (!common_replicated_indexes)
                    common_replicated_indexes = column_replicated->getIndexesColumn();
                else if (common_replicated_indexes != column_replicated->getIndexesColumn())
                {
                    common_replicated_indexes.reset();
                    break;
                }
            }

and the function then runs on the nested (deduplicated) columns, at nested_column_size rows:

cpp
// src/Functions/IFunction.cpp:746
        auto result = executeWithoutReplicatedColumns(arguments_without_replicated, result_type, nested_column_size, dry_run);

        if (isLazyReplicationUseful(result))
            return ColumnReplicated::create(result, common_replicated_indexes);

        return result->index(*common_replicated_indexes, 0);

So SELECT f(right.x) FROM ... JOIN ... computes f once per distinct right row, not once per output row. With a 1000x fan-out that is a 1000x reduction in the work.

There is a subtlety that the code handles explicitly: if the function can throw, it must not be evaluated on rows that no output row references:

cpp
// src/Functions/IFunction.cpp:726
        /// In case the function might throw an exception replicated columns must be compacted
        // to avoid throwing on unused rows in the nested data.
        if (canThrow(argument_types))
        {
            ColumnIndex column_index(common_replicated_indexes);
            auto res = column_index.buildCompactIndexedColumns(nested_columns);
            common_replicated_indexes = std::move(res.compact_indexes);
            nested_columns = std::move(res.compact_indexed_columns);
        }

Materialisation: when to give up

A replicated column is a lazy representation. Somewhere it has to become real data — when the consumer genuinely needs a flat column, when the replication does not pay, or when the nested data is not smaller than the output.

The size check:

cpp
// src/Columns/ColumnReplicated.cpp:852
ColumnPtr convertToFullColumnIfReplicationNotUseful(const ColumnPtr & column, bool with_size_check)
{
    if (!column->isReplicated())
        return column;

    const auto & replicated = typeid_cast<const ColumnReplicated &>(*column);
    /// Materialize if the type doesn't benefit from lazy replication, or if the referenced nested data isn't smaller than output.
    if (!isLazyReplicationUseful(replicated.getNestedColumn())
        || (with_size_check && replicated.getNestedColumn()->size() >= replicated.size() && getReferencedRowsBound(replicated) >= replicated.size()))
        return replicated.convertToFullColumnIfReplicated();

    return column;
}

The second condition is subtle: replication only saves memory if the number of distinct source rows that the index refers to is smaller than the number of output rows. getReferencedRowsBound estimates that bound from the extremes of the index and from the number of runs of equal indexes.

The materialisation itself has a fast path for arrays. The generic path (nested_column->index(...)) would build a UInt64 index per element of the array; the array path copies each array row once:

cpp
// src/Columns/ColumnReplicated.cpp:186
ColumnPtr ColumnReplicated::convertToFullColumnIfReplicated() const
{
    if (const auto * src_array = typeid_cast<const ColumnArray *>(nested_column.get()))
    {
        /// Per-array materialization
        size_t src_rows_count = src_array->size();
        size_t src_elements = src_array->getOffsets().back();
        if (src_rows_count != 0 && src_elements >= src_rows_count * ELEMENTS_PER_ROW_THRESHOLD)
            return convertToFullColumnArray(*src_array, *indexes.getIndexes());
    }
    /// Per-element materialization
    return nested_column->index(*indexes.getIndexes(), 0);
}

The threshold is spelled out in the source: “Break-even is around 8 elements per row” (src/Columns/ColumnReplicated.cpp:131).

Compaction: dropping the rows nobody points at

After a filter, the index may no longer reference some of the nested rows. buildCompactIndexedColumns finds the unreferenced rows, filters them out of the nested data and renumbers the index:

cpp
// src/Columns/ColumnIndex.cpp:421
ColumnIndex::CompactIndexedColumnsResult ColumnIndex::buildCompactIndexedColumns(const Columns & indexed_columns) const
{
    if (indexed_columns.empty())
        return {getIndexes(), indexed_columns};

    size_t indexed_data_size = indexed_columns[0]->size();
    for (size_t i = 1; i < indexed_columns.size(); ++i)
        chassert(indexed_columns[i]->size() == indexed_data_size);

    /// First, create a filter for indexed data to filter out all unused rows.
    auto filter_opt = buildUsedRowsFilter(indexed_data_size);
    if (!filter_opt.has_value())
        return {getIndexes(), indexed_columns};

    IColumn::Filter & filter = filter_opt.value();

    /// Second, adjust indexes.
    ColumnIndex compact_column_index(IColumn::mutate(getIndexes()));
    size_t result_size_hint = compact_column_index.compactIndexes(filter, indexed_data_size);

    Columns filtered_columns;
    filtered_columns.reserve(indexed_columns.size());
    for (const auto & column : indexed_columns)
        filtered_columns.push_back(column->filter(filter, result_size_hint));

    return {compact_column_index.getIndexes(), filtered_columns};
}

The driver (compactReplicatedColumns, src/Columns/ColumnReplicated.cpp:866) groups the columns by their shared index, so the compaction is done once per index, not once per column.

The insertion cache

When two replicated columns are merged (a sort, a merge-sort), the values would be copied once per output row. The insertion_cache remembers which source rows have already been copied, keyed by the source column’s id:

cpp
// src/Columns/ColumnReplicated.cpp:249
            /// Use insertion_cache to avoid copying of values from source nested column if
            /// we already inserted them earlier and can use indexes of already inserted values.
            auto & indexes_match = insertion_cache[src_replicated->id];

            auto insert = [&](size_t, size_t src_index)
            {
                auto it = indexes_match.find(src_index);
                if (it == indexes_match.end())
                {
                    nested_column->insertFrom(*src_replicated->nested_column, src_index);
                    it = indexes_match.emplace(src_index, nested_column->size() - 1).first;
                }

                indexes.insertIndex(it->second);
            };

A value that appears in ten output rows is copied once and referenced ten times.

On the wire

The Native format has a dedicated serialization for this column, because a replicated column has to travel as two things: the index and the data.

cpp
// src/DataTypes/Serializations/SerializationReplicated.cpp:159
    /// We write ColumnReplicated data in the following format:
    /// - number of rows in column
    /// - size of indexes type
    /// - indexes data
    /// - number of rows in the nested column
    /// - data of nested column

The index width travels with the data, so the reader knows whether to expect one byte or eight per row:

cpp
// src/DataTypes/Serializations/SerializationReplicated.cpp:166
    writeVarUInt(UInt64(limit), *indexes_stream);
    auto size_of_indexes_type = column_replicated.getIndexes().getSizeOfIndexType();
    writeBinaryLittleEndian(UInt8(size_of_indexes_type), *indexes_stream);

and the nested data goes into its own stream, with its own row count (src/DataTypes/Serializations/SerializationReplicated.cpp:188).

So the replication survives the trip to another server: the receiver gets the same index and the same data, and the memory saving is kept.

Trade-offs

What it buys: the right-hand columns of a join are stored once, the fan-out costs one byte (or two, or four) per output row instead of a full copy, and functions over replicated columns are evaluated once per distinct source row rather than once per output row.

What it costs:

An indirection on every read. getDataAt(n) is nested->getDataAt(index[n]). For a column that is read exactly once, in order, the copy was cheaper.
It is not fixed-and-contiguous. Code that assumes a flat array has to materialise the column first.
It only pays when the column is wide. The sizeOfValueIfFixed() > 8 rule means a UInt32 column is copied eagerly even with a 1000x fan-out — 4 bytes of copy versus 1 byte of index plus an indirection is a wash.
The index is not free either. It is 1–8 bytes per output row, and it has to be permuted, filtered and compacted along with everything else.
It changes the shape of the output. Anything that inspects the column type — isFixedAndContiguous, the JIT’s getColumnData — has to handle ColumnReplicated, or force materialisation first.

The escape hatch

sql
SELECT * FROM a JOIN b USING (id)
SETTINGS enable_lazy_columns_replication = 0;

Wrapping up

The one thing to remember: ColumnReplicated is a materialised view of the mapping, not of the data. The join’s fan-out becomes an index column, and the index column’s integer width is chosen to fit.

The part people get wrong: the trick does not just save memory — it changes where the work happens. A function over a replicated argument runs on the deduplicated rows, so a 1000x fan-out costs 1/1000 of the function calls. That is often a bigger win than the RAM.

What to go do about it: look at the join’s right-hand side. If the query is SELECT count() FROM ... JOIN ..., the joined columns are probably never read — but they are copied anyway unless lazy replication is on. It is on by default since 25.11, so if you are on an older server, that is your first thing to change.