<commit_before>/*************************************************************************
 *
 * Copyright 2016 Realm Inc.
 *
 * Licensed 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.
 *
 **************************************************************************/

#include <new>
#include <algorithm>
#include <set>
#include <fstream>

#ifdef REALM_DEBUG
    #include <iostream>
    #include <iomanip>
#endif

#include <realm/util/file_mapper.hpp>
#include <realm/util/memory_stream.hpp>
#include <realm/util/miscellaneous.hpp>
#include <realm/util/thread.hpp>
#include <realm/impl/destroy_guard.hpp>
#include <realm/utilities.hpp>
#include <realm/exceptions.hpp>
#include <realm/column_linkbase.hpp>
#include <realm/column_backlink.hpp>
#include <realm/group_writer.hpp>
#include <realm/group.hpp>
#include <realm/replication.hpp>

using namespace realm;
using namespace realm::util;

namespace {

class Initialization {
public:
    Initialization()
    {
        realm::cpuid_init();
    }
};

Initialization initialization;

} // anonymous namespace


Group::Group():
    m_alloc(), // Throws
    m_top(m_alloc),
    m_tables(m_alloc),
    m_table_names(m_alloc),
    m_is_shared(false)
{
    init_array_parents();
    m_alloc.attach_empty(); // Throws
    Replication::HistoryType history_type = Replication::hist_None;
    int file_format_version = get_target_file_format_version_for_session(0, history_type);
    m_alloc.set_file_format_version(file_format_version);
    ref_type top_ref = 0; // Instantiate a new empty group
    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
}


int Group::get_file_format_version() const noexcept
{
    return m_alloc.get_file_format_version();
}


void Group::set_file_format_version(int file_format) noexcept
{
    m_alloc.set_file_format_version(file_format);
}


int Group::get_committed_file_format_version() const noexcept
{
    return m_alloc.get_committed_file_format_version();
}


int Group::get_target_file_format_version_for_session(int current_file_format_version,
                                                      int requested_history_type) noexcept
{
    Replication::HistoryType requested_history_type_2 =
        Replication::HistoryType(requested_history_type);

    // Note: This function is responsible for choosing the target file format
    // for a sessions. If it selects a file format that is different from
    // `current_file_format_version`, it will trigger a file format upgrade
    // process.

    // Note: Previously it was necessary to base the decision on the current
    // (original) file format and the selected type of history (of
    // changesets). At the present time, however (since the introduction of file
    // format version 5), all files need to use the latest file format.

    // Note: `current_file_format_version` may be zero at this time, which means
    // that the file format it is not yet decided (only possible for empty
    // Realms where top-ref is zero).

    static_cast<void>(current_file_format_version);
    static_cast<void>(requested_history_type_2);
    return Allocator::CURRENT_FILE_FORMAT_VERSION;
}


void Group::upgrade_file_format(int target_file_format_version)
{
    REALM_ASSERT(is_attached());

    // Be sure to revisit the following upgrade logic when a new file foprmat
    // version is introduced. The following assert attempt to help you not
    // forget it.
    REALM_ASSERT_EX(target_file_format_version == 5, target_file_format_version);

    int current_file_format_version = get_file_format_version();
    REALM_ASSERT(current_file_format_version < target_file_format_version);

    // SlabAlloc::validate_buffer() must ensure this. Be sure to revisit the
    // following upgrade logic when SlabAlloc::validate_buffer() is changed (or
    // vice versa).
    REALM_ASSERT_EX(current_file_format_version == 2 ||
                    current_file_format_version == 3 ||
                    current_file_format_version == 4, current_file_format_version);

    // Upgrade from 2 to 3
    if (current_file_format_version <= 2 && target_file_format_version >= 3) {
        for (size_t t = 0; t < m_tables.size(); t++) {
            TableRef table = get_table(t);
            table->upgrade_file_format();
        }
    }

    // Upgrade from 3 to 4
    if (current_file_format_version <= 3 && target_file_format_version >= 4) {
        // No-op
    }

    // Upgrade from 4 to 5 (datetime -> timestamp)
    if (current_file_format_version <= 4 && target_file_format_version >= 5) {
        for (size_t t = 0; t < m_tables.size(); t++) {
            TableRef table = get_table(t);
            table->upgrade_olddatetime();
        }
    }

    // NOTE: Additional future upgrade steps go here.

    set_file_format_version(target_file_format_version);
}


void Group::open(const std::string& file_path, const char* encryption_key, OpenMode mode)
{
    if (is_attached() || m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    SlabAlloc::Config cfg;
    cfg.read_only = mode == mode_ReadOnly;
    cfg.no_create = mode == mode_ReadWriteNoCreate;
    cfg.encryption_key = encryption_key;
    ref_type top_ref = m_alloc.attach_file(file_path, cfg); // Throws
    SlabAlloc::DetachGuard dg(m_alloc);

    // Select file format if it is still undecided.
    int current_file_format_version = get_file_format_version();
    Replication::HistoryType history_type = Replication::hist_None;
    int target_file_format_version =
        get_target_file_format_version_for_session(current_file_format_version, history_type);
    if (current_file_format_version == 0) {
        set_file_format_version(target_file_format_version);
    }
    else {
        // From a technical point of view, we could upgrade the Realm file
        // format in memory here, but since upgrading can be expensive, it is
        // currently disallowed by the SlabAlloc::validate_buffer().
        REALM_ASSERT(target_file_format_version == current_file_format_version);
    }

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
    dg.release(); // Do not detach after all
}


void Group::open(BinaryData buffer, bool take_ownership)
{
    REALM_ASSERT(buffer.data());

    if (is_attached() || m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    ref_type top_ref = m_alloc.attach_buffer(buffer.data(), buffer.size()); // Throws
    SlabAlloc::DetachGuard dg(m_alloc);

    // Select file format if it is still undecided.
    int current_file_format_version = get_file_format_version();
    Replication::HistoryType history_type = Replication::hist_None;
    int target_file_format_version =
        get_target_file_format_version_for_session(current_file_format_version, history_type);
    if (current_file_format_version == 0) {
        set_file_format_version(target_file_format_version);
    }
    else {
        // From a technical point of view, we could upgrade the Realm file
        // format in memory here, but since upgrading can be expensive, it is
        // currently disallowed by the SlabAlloc::validate_buffer().
        REALM_ASSERT(target_file_format_version == current_file_format_version);
    }

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
    dg.release(); // Do not detach after all

    if (take_ownership)
        m_alloc.own_buffer();
}


Group::~Group() noexcept
{
    // If this group accessor is detached at this point in time, it is either
    // because it is SharedGroup::m_group (m_is_shared), or it is a free-stading
    // group accessor that was never successfully opened.
    if (!m_top.is_attached())
        return;

    // Free-standing group accessor

    detach_table_accessors();

    // Just allow the allocator to release all memory in one chunk without
    // having to traverse the entire tree first
    m_alloc.detach();
}


void Group::remap(size_t new_file_size)
{
    size_t old_baseline = m_alloc.get_baseline();
    if (new_file_size > old_baseline)
        m_alloc.remap(new_file_size); // Throws
}


void Group::remap_and_update_refs(ref_type new_top_ref, size_t new_file_size)
{
    size_t old_baseline = m_alloc.get_baseline();

    if (new_file_size > old_baseline) {
        m_alloc.remap(new_file_size); // Throws
    }

    m_alloc.invalidate_cache();
    update_refs(new_top_ref, old_baseline);
}


void Group::attach(ref_type top_ref, bool create_group_when_missing)
{
    REALM_ASSERT(!m_top.is_attached());

    // If this function throws, it must leave the group accesor in a the
    // unattached state.

    m_tables.detach();
    m_table_names.detach();

    if (top_ref != 0) {
        m_top.init_from_ref(top_ref);
        size_t top_size = m_top.size();
        static_cast<void>(top_size);

        if (top_size < 8) {
            REALM_ASSERT_11(top_size, ==, 3, ||, top_size, ==, 5, ||, top_size, ==, 7);
        }
        else {
            REALM_ASSERT_3(top_size, ==, 9);
        }

        m_table_names.init_from_parent();
        m_tables.init_from_parent();

        // The 3rd slot in m_top is
        // `RefOrTagged::make_tagged(logical_file_size)`, and the logical file
        // size must never exceed actual file size.
        REALM_ASSERT_3(m_top.get_as_ref_or_tagged(2).get_as_int(), <=, m_alloc.get_baseline());
    }
    else if (create_group_when_missing) {
        create_empty_group(); // Throws
    }

    m_attached = true;
}


void Group::detach() noexcept
{
    detach_table_accessors();
    m_table_accessors.clear();

    m_table_names.detach();
    m_tables.detach();
    m_top.detach();

    m_attached = false;
}


void Group::attach_shared(ref_type new_top_ref, size_t new_file_size, bool writable)
{
    REALM_ASSERT_3(new_top_ref, <, new_file_size);
    REALM_ASSERT(!is_attached());

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    // Update memory mapping if database file has grown
    if (new_file_size > m_alloc.get_baseline())
        m_alloc.remap(new_file_size); // Throws

    // When `new_top_ref` is null, ask attach() to create a new node structure
    // for an empty group, but only during the initiation of write
    // transactions. When the transaction being initiated is a read transaction,
    // we instead have to leave array accessors m_top, m_tables, and
    // m_table_names in their detached state, as there are no underlying array
    // nodes to attached them to. In the case of write transactions, the nodes
    // have to be created, as they have to be ready for being modified.
    bool create_group_when_missing = writable;
    attach(new_top_ref, create_group_when_missing); // Throws
}


void Group::detach_table_accessors() noexcept
{
    for (const auto& table_accessor : m_table_accessors) {
        if (Table* t = table_accessor) {
            typedef _impl::TableFriend tf;
            tf::detach(*t);
            tf::unbind_ptr(*t);
        }
    }
}


void Group::create_empty_group()
{
    m_top.create(Array::type_HasRefs); // Throws
    _impl::DeepArrayDestroyGuard dg_top(&m_top);
    {
        m_table_names.create(); // Throws
        _impl::DestroyGuard<ArrayString> dg(&m_table_names);
        m_top.add(m_table_names.get_ref()); // Throws
        dg.release();
    }
    {
        m_tables.create(Array::type_HasRefs); // Throws
        _impl::DestroyGuard<ArrayInteger> dg(&m_tables);
        m_top.add(m_tables.get_ref()); // Throws
        dg.release();
    }
    size_t initial_logical_file_size = sizeof (SlabAlloc::Header);
    m_top.add(RefOrTagged::make_tagged(initial_logical_file_size)); // Throws
    dg_top.release();
}


Table* Group::do_get_table(size_t table_ndx, DescMatcher desc_matcher)
{
    REALM_ASSERT(m_table_accessors.empty() || m_table_accessors.size() == m_tables.size());

    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);

    if (m_table_accessors.empty())
        m_table_accessors.resize(m_tables.size()); // Throws

    // Get table accessor from cache if it exists, else create
    Table* table = m_table_accessors[table_ndx];
    if (!table)
        table = create_table_accessor(table_ndx); // Throws

    if (desc_matcher) {
        typedef _impl::TableFriend tf;
        if (desc_matcher && !(*desc_matcher)(tf::get_spec(*table)))
            throw DescriptorMismatch();
    }

    return table;
}


Table* Group::do_get_table(StringData name, DescMatcher desc_matcher)
{
    if (!m_table_names.is_attached())
        return 0;
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        return 0;

    Table* table = do_get_table(table_ndx, desc_matcher); // Throws
    return table;
}


Table* Group::do_insert_table(size_t table_ndx, StringData name, DescSetter desc_setter,
                              bool require_unique_name)
{
    if (require_unique_name && has_table(name))
        throw TableNameInUse();
    return do_insert_table(table_ndx, name, desc_setter); // Throws
}


Table* Group::do_insert_table(size_t table_ndx, StringData name, DescSetter desc_setter)
{
    if (table_ndx > m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    create_and_insert_table(table_ndx, name); // Throws
    Table* table = do_get_table(table_ndx, nullptr); // Throws
    if (desc_setter)
        (*desc_setter)(*table); // Throws
    return table;
}

Table* Group::do_get_or_insert_table(size_t table_ndx, StringData name, DescMatcher desc_matcher,
                                     DescSetter desc_setter, bool* was_added)
{
    Table* table;
    size_t existing_table_ndx = m_table_names.find_first(name);
    if (existing_table_ndx == not_found) {
        table = do_insert_table(table_ndx, name, desc_setter); // Throws
        if (was_added)
            *was_added = true;
    }
    else {
        table = do_get_table(existing_table_ndx, desc_matcher); // Throws
        if (was_added)
            *was_added = false;
    }
    return table;
}


Table* Group::do_get_or_add_table(StringData name, DescMatcher desc_matcher,
                                  DescSetter desc_setter, bool* was_added)
{
    REALM_ASSERT(m_table_names.is_attached());
    Table* table;
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found) {
        table = do_insert_table(m_tables.size(), name, desc_setter); // Throws
    }
    else {
        table = do_get_table(table_ndx, desc_matcher); // Throws
    }
    if (was_added)
        *was_added = (table_ndx == not_found);
    return table;
}


void Group::create_and_insert_table(size_t table_ndx, StringData name)
{
    if (REALM_UNLIKELY(name.size() > max_table_name_length))
        throw LogicError(LogicError::table_name_too_long);

    using namespace _impl;
    typedef TableFriend tf;
    ref_type ref = tf::create_empty_table(m_alloc); // Throws
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    size_t prior_num_tables = m_tables.size();
    m_tables.insert(table_ndx, ref); // Throws
    m_table_names.insert(table_ndx, name); // Throws

    // Need slot for table accessor
    if (!m_table_accessors.empty()) {
        m_table_accessors.insert(m_table_accessors.begin() + table_ndx, nullptr); // Throws
    }

    update_table_indices([&](size_t old_table_ndx) {
        if (old_table_ndx >= table_ndx) {
            return old_table_ndx + 1;
        }
        return old_table_ndx;
    }); // Throws

    if (Replication* repl = m_alloc.get_replication())
        repl->insert_group_level_table(table_ndx, prior_num_tables, name); // Throws
}


Table* Group::create_table_accessor(size_t table_ndx)
{
    REALM_ASSERT(m_table_accessors.empty() || table_ndx < m_table_accessors.size());

    if (m_table_accessors.empty())
        m_table_accessors.resize(m_tables.size()); // Throws

    // Whenever a table has a link column, the column accessor must be set up to
    // refer to the target table accessor, so the target table accessor needs to
    // be created too, if it does not already exist. This, of course, applies
    // recursively, and it applies to the opposide direction of links too (from
    // target side to origin side). This means that whenever we create a table
    // accessor, we actually need to create the entire cluster of table
    // accessors, that is reachable in zero or more steps along links, or
    // backwards along links.
    //
    // To be able to do this, and to handle the cases where the link
    // relathionship graph contains cycles, each table accessor need to be
    // created in the following steps:
    //
    //  1) Create table accessor, but skip creation of column accessors
    //  2) Register incomplete table accessor in group accessor
    //  3) Mark table accessor
    //  4) Create column accessors
    //  5) Unmark table accessor
    //
    // The marking ensures that the establsihment of the connection between link
    // and backlink column accessors is postponed until both column accessors
    // are created. Infinite recursion due to cycles is prevented by the early
    // registration in the group accessor of inclomplete table accessors.

    typedef _impl::TableFriend tf;
    ref_type ref = m_tables.get_as_ref(table_ndx);
    Table* table = tf::create_incomplete_accessor(m_alloc, ref, this, table_ndx); // Throws

    // The new accessor cannot be leaked, because no exceptions can be thrown
    // before it becomes referenced from `m_column_accessors`.

    // Increase reference count from 0 to 1 to make the group accessor keep
    // the table accessor alive. This extra reference count will be revoked
    // during destruction of the group accessor.
    tf::bind_ptr(*table);

    tf::mark(*table);
    m_table_accessors[table_ndx] = table;
    tf::complete_accessor(*table); // Throws
    tf::unmark(*table);
    return table;
}


void Group::remove_table(StringData name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        throw NoSuchTable();
    remove_table(table_ndx); // Throws
}


void Group::remove_table(size_t table_ndx)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    TableRef table = get_table(table_ndx);

    // In principle we could remove a table even if it is the target of link
    // columns of other tables, however, to do that, we would have to
    // automatically remove the "offending" link columns from those other
    // tables. Such a behaviour is deemed too obscure, and we shall therefore
    // require that a removed table does not contain foreigh origin backlink
    // columns.
    typedef _impl::TableFriend tf;
    if (tf::is_cross_table_link_target(*table))
        throw CrossTableLinkTarget();

    // There is no easy way for Group::TransactAdvancer to handle removal of
    // tables that contain foreign target table link columns, because that
    // involves removal of the corresponding backlink columns. For that reason,
    // we start by removing all columns, which will generate individual
    // replication instructions for each column removal with sufficient
    // information for Group::TransactAdvancer to handle them.
    size_t n = table->get_column_count();
    for (size_t i = n; i > 0; --i)
        table->remove_column(i - 1);

    int64_t ref_64 = m_tables.get(table_ndx);
    REALM_ASSERT(!int_cast_has_overflow<ref_type>(ref_64));
    ref_type ref = ref_type(ref_64);

    // Remove table and move all successive tables
    m_tables.erase(table_ndx); // Throws
    m_table_names.erase(table_ndx); // Throws
    m_table_accessors.erase(m_table_accessors.begin() + table_ndx);

    tf::detach(*table);
    tf::unbind_ptr(*table);

    // Unless the removed table is the last, update all indices of tables after
    // the removed table.
    bool last_table_removed = table_ndx == m_tables.size();
    if (!last_table_removed) {
        update_table_indices([&](size_t old_table_ndx) {
            REALM_ASSERT(old_table_ndx != table_ndx); // We should not see links to the removed table
            if (old_table_ndx > table_ndx) {
                return old_table_ndx - 1;
            }
            return old_table_ndx;
        }); // Throws
    }

    // Destroy underlying node structure
    Array::destroy_deep(ref, m_alloc);

    size_t prior_num_tables = m_tables.size() + 1;
    if (Replication* repl = m_alloc.get_replication())
        repl->erase_group_level_table(table_ndx, prior_num_tables); // Throws
}


void Group::rename_table(StringData name, StringData new_name, bool require_unique_name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        throw NoSuchTable();
    rename_table(table_ndx, new_name, require_unique_name); // Throws
}


void Group::rename_table(size_t table_ndx, StringData new_name, bool require_unique_name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    if (require_unique_name && has_table(new_name))
        throw TableNameInUse();
    m_table_names.set(table_ndx, new_name);
    if (Replication* repl = m_alloc.get_replication())
        repl->rename_group_level_table(table_ndx, new_name); // Throws
}


void Group::move_table(size_t from_table_ndx, size_t to_table_ndx)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    REALM_ASSERT_EX(m_table_accessors.empty() || m_table_accessors.size() == m_tables.size(),
                    m_table_accessors.size(), m_tables.size());
    if (from_table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    if (to_table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);

    if (from_table_ndx == to_table_ndx)
        return;

    // Tables between from_table_ndx and to_table_ndx change their indices, so
    // link columns have to be adjusted (similar to remove_table).

    // Build a map of all table indices that are going to change:
    std::map<size_t, size_t> moves; // from -> to
    moves[from_table_ndx] = to_table_ndx;
    if (from_table_ndx < to_table_ndx) {
        // Move up:
        for (size_t i = from_table_ndx + 1; i <= to_table_ndx; ++i) {
            moves[i] = i - 1;
        }
    }
    else { // from_table_ndx > to_table_ndx
        // Move down:
        for (size_t i = to_table_ndx; i < from_table_ndx; ++i) {
            moves[i] = i + 1;
        }
    }

    // Move entries in internal data structures.
    m_tables.move_rotate(from_table_ndx, to_table_ndx);
    m_table_names.move_rotate(from_table_ndx, to_table_ndx);

    // Move accessors.
    if (!m_table_accessors.empty()) {
        using iter = decltype(m_table_accessors.begin());
        iter first, new_first, last;
        if (from_table_ndx < to_table_ndx) {
            // Rotate left.
            first     = m_table_accessors.begin() + from_table_ndx;
            new_first = first + 1;
            last      = m_table_accessors.begin() + to_table_ndx + 1;
        }
        else { // from_table_ndx > to_table_ndx
            // Rotate right.
            first     = m_table_accessors.begin() + to_table_ndx;
            new_first = m_table_accessors.begin() + from_table_ndx;
            last      = new_first + 1;
        }
        std::rotate(first, new_first, last);
    }

    update_table_indices([&](size_t old_table_ndx) {
        auto it = moves.find(old_table_ndx);
        if (it != moves.end()) {
            return it->second;
        }
        return old_table_ndx;
    });

    if (Replication* repl = m_alloc.get_replication())
        repl->move_group_level_table(from_table_ndx, to_table_ndx); // Throws
}


class Group::DefaultTableWriter: public Group::TableWriter {
public:
    DefaultTableWriter(const Group& group):
        m_group(group)
    {
    }
    ref_type write_names(_impl::OutputStream& out) override
    {
        bool deep = true; // Deep
        bool only_if_modified = false; // Always
        return m_group.m_table_names.write(out, deep, only_if_modified); // Throws
    }
    ref_type write_tables(_impl::OutputStream& out) override
    {
        bool deep = true; // Deep
        bool only_if_modified = false; // Always
        return m_group.m_tables.write(out, deep, only_if_modified); // Throws
    }
private:
    const Group& m_group;
};

void Group::write(std::ostream& out, bool pad) const
{
    write(out, pad, 0);
}

void Group::write(std::ostream& out, bool pad_for_encryption, uint_fast64_t version_number) const
{
    REALM_ASSERT(is_attached());
    DefaultTableWriter table_writer(*this);
    bool no_top_array = !m_top.is_attached();
    write(out, m_alloc, table_writer, no_top_array, pad_for_encryption, version_number); // Throws
}

void Group::write(const std::string& path, const char* encryption_key) const
{
    write(path, encryption_key, 0);
}

void Group::write(const std::string& path, const char* encryption_key, uint_fast64_t version_number) const
{
    File file;
    int flags = 0;
    file.open(path, File::access_ReadWrite, File::create_Must, flags);
    write(file, encryption_key, version_number);
}

void Group::write(File& file, const char* encryption_key, uint_fast64_t version_number) const
{
    REALM_ASSERT(file.get_size() == 0);

    file.set_encryption_key(encryption_key);
    File::Streambuf streambuf(&file);
    std::ostream out(&streambuf);
    write(out, encryption_key != 0, version_number);
}

BinaryData Group::write_to_mem() const
{
    REALM_ASSERT(is_attached());

    // Get max possible size of buffer
    //
    // FIXME: This size could potentially be vastly bigger that what
    // is actually needed.
    size_t max_size = m_alloc.get_total_size();

    char* buffer = static_cast<char*>(malloc(max_size)); // Throws
    if (!buffer)
        throw std::bad_alloc();
    try {
        MemoryOutputStream out; // Throws
        out.set_buffer(buffer, buffer + max_size);
        write(out); // Throws
        size_t buffer_size = out.size();
        return BinaryData(buffer, buffer_size);
    }
    catch (...) {
        free(buffer);
        throw;
    }
}


void Group::write(std::ostream& out, const Allocator& alloc, TableWriter& table_writer,
                  bool no_top_array, bool pad_for_encryption, uint_fast64_t version_number)
{
    _impl::OutputStream out_2(out);

    // Write the file header
    SlabAlloc::Header streaming_header;
    int file_format_version = (no_top_array ? 0 : alloc.get_file_format_version());
    SlabAlloc::init_streaming_header(&streaming_header, file_format_version);
    out_2.write(reinterpret_cast<const char*>(&streaming_header), sizeof streaming_header);

    ref_type top_ref = 0;
    size_t final_file_size = sizeof streaming_header;
    if (no_top_array) {
        // Accept version number 1 as that number is (unfortunately) also used
        // to denote the empty initial state of a Realm file.
        REALM_ASSERT(version_number == 0 || version_number == 1);
    }
    else {
        // Because we need to include the total logical file size in the
        // top-array, we have to start by writing everything except the
        // top-array, and then finally compute and write a correct version of
        // the top-array. The free-space information of the group will only be
        // included if a non-zero version number is given as parameter,
        // indicating that versioning info is to be saved. This is used from
        // SharedGroup to compact the database by writing only the live data
        // into a separate file.
        ref_type names_ref  = table_writer.write_names(out_2); // Throws
        ref_type tables_ref = table_writer.write_tables(out_2); // Throws
        SlabAlloc new_alloc;
        new_alloc.attach_empty(); // Throws
        new_alloc.set_file_format_version(alloc.get_file_format_version());
        Array top(new_alloc);
        top.create(Array::type_HasRefs); // Throws
        _impl::ShallowArrayDestroyGuard dg_top(&top);
        // FIXME: We really need an alternative to Array::truncate() that is able to expand.
        int_fast64_t value_1 = from_ref(names_ref);
        int_fast64_t value_2 = from_ref(tables_ref);
        top.add(value_1); // Throws
        top.add(value_2); // Throws
        top.add(0); // Throws

        int top_size = 3;
        if (version_number) {
            Array free_list(new_alloc);
            Array size_list(new_alloc);
            Array version_list(new_alloc);
            free_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_1(&free_list);
            size_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_2(&size_list);
            version_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_3(&version_list);
            bool deep = true; // Deep
            bool only_if_modified = false; // Always
            ref_type free_list_ref = free_list.write(out_2, deep, only_if_modified);
            ref_type size_list_ref = size_list.write(out_2, deep, only_if_modified);
            ref_type version_list_ref = version_list.write(out_2, deep, only_if_modified);
            top.add(RefOrTagged::make_ref(free_list_ref)); // Throws
            top.add(RefOrTagged::make_ref(size_list_ref)); // Throws
            top.add(RefOrTagged::make_ref(version_list_ref)); // Throws
            top.add(RefOrTagged::make_tagged(version_number)); // Throws
            top_size = 7;
        }
        top_ref = out_2.get_ref_of_next_array();

        // Produce a preliminary version of the top array whose
        // representation is guaranteed to be able to hold the final file
        // size
        size_t max_top_byte_size = Array::get_max_byte_size(top_size);
        size_t max_final_file_size = size_t(top_ref) + max_top_byte_size;
        top.ensure_minimum_width(RefOrTagged::make_tagged(max_final_file_size)); // Throws

        // Finalize the top array by adding the projected final file size
        // to it
        size_t top_byte_size = top.get_byte_size();
        final_file_size = size_t(top_ref) + top_byte_size;
        top.set(2, RefOrTagged::make_tagged(final_file_size)); // Throws

        // Write the top array
        bool deep = false; // Shallow
        bool only_if_modified = false; // Always
        top.write(out_2, deep, only_if_modified); // Throws
        REALM_ASSERT_3(size_t(out_2.get_ref_of_next_array()), ==, final_file_size);

        dg_top.reset(nullptr); // Destroy now
    }

    // encryption will pad the file to a multiple of the page, so ensure the
    // footer is aligned to the end of a page
    if (pad_for_encryption) {
#if REALM_ENABLE_ENCRYPTION
        size_t unrounded_size = final_file_size + sizeof(SlabAlloc::StreamingFooter);
        size_t rounded_size = round_up_to_page_size(unrounded_size);
        if (rounded_size != unrounded_size) {
            std::unique_ptr<char[]> buffer(new char[rounded_size - unrounded_size]());
            out_2.write(buffer.get(), rounded_size - unrounded_size);
        }
#endif
    }

    // Write streaming footer
    SlabAlloc::StreamingFooter footer;
    footer.m_top_ref = top_ref;
    footer.m_magic_cookie = SlabAlloc::footer_magic_cookie;
    out_2.write(reinterpret_cast<const char*>(&footer), sizeof footer);
}


void Group::commit()
{
    if (!is_attached())
        throw LogicError(LogicError::detached_accessor);
    if (m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    GroupWriter out(*this); // Throws

    // Recursively write all changed arrays to the database file. We
    // postpone the commit until we are sure that no exceptions can be
    // thrown.
    ref_type top_ref = out.write_group(); // Throws

    // Since the group is persisiting in single-thread (unshared)
    // mode we have to make sure that the group stays valid after
    // commit

    // Mark all managed space (beyond the attached file) as free.
    reset_free_space_tracking(); // Throws

    size_t old_baseline = m_alloc.get_baseline();

    // Remap file if it has grown
    size_t new_file_size = out.get_file_size();
    if (new_file_size > old_baseline) {
        m_alloc.remap(new_file_size); // Throws
    }

    out.commit(top_ref); // Throws

    // Recursively update refs in all active tables (columns, arrays..)
    update_refs(top_ref, old_baseline);
}


void Group::update_refs(ref_type top_ref, size_t old_baseline) noexcept
{
    // After Group::commit() we will always have free space tracking
    // info.
    REALM_ASSERT_3(m_top.size(), >=, 5);

    // Array nodes that are part of the previous version of the
    // database will not be overwritten by Group::commit(). This is
    // necessary for robustness in the face of abrupt termination of
    // the process. It also means that we can be sure that an array
    // remains unchanged across a commit if the new ref is equal to
    // the old ref and the ref is below the previous baseline.

    if (top_ref < old_baseline && m_top.get_ref() == top_ref)
        return;

    m_top.init_from_ref(top_ref);

    // Now we can update it's child arrays
    m_table_names.update_from_parent(old_baseline);

    // If m_tables has not been modfied we don't
    // need to update attached table accessors
    if (!m_tables.update_from_parent(old_baseline))
        return;

    // Update all attached table accessors including those attached to
    // subtables.
    for (const auto& table_accessor : m_table_accessors) {
        typedef _impl::TableFriend tf;
        if (Table* table = table_accessor)
            tf::update_from_parent(*table, old_baseline);
    }
}


bool Group::operator==(const Group& g) const
{
    size_t n = size();
    if (n != g.size())
        return false;
    for (size_t i = 0; i < n; ++i) {
        const StringData& table_name_1 = get_table_name(i); // Throws
        const StringData& table_name_2 = g.get_table_name(i); // Throws
        if (table_name_1 != table_name_2)
            return false;

        ConstTableRef table_1 = get_table(i); // Throws
        ConstTableRef table_2 = g.get_table(i); // Throws
        if (*table_1 != *table_2)
            return false;
    }
    return true;
}


void Group::to_string(std::ostream& out) const
{
    // Calculate widths
    size_t index_width = 4;
    size_t name_width = 10;
    size_t rows_width = 6;
    size_t count = size();
    for (size_t i = 0; i < count; ++i) {
        StringData name = get_table_name(i);
        if (name_width < name.size())
            name_width = name.size();

        ConstTableRef table = get_table(name);
        size_t row_count = table->size();
        if (rows_width < row_count) { // FIXME: should be the number of digits in row_count: floor(log10(row_count+1))
            rows_width = row_count;
        }
    }


    // Print header
    out << std::setw(int(index_width + 1)) << std::left << " ";
    out << std::setw(int(name_width + 1))  << std::left << "tables";
    out << std::setw(int(rows_width))    << std::left << "rows"    << std::endl;

    // Print tables
    for (size_t i = 0; i < count; ++i) {
        StringData name = get_table_name(i);
        ConstTableRef table = get_table(name);
        size_t row_count = table->size();

        out << std::setw(int(index_width)) << std::right << i                 << " ";
        out << std::setw(int(name_width))  << std::left  << std::string(name) << " ";
        out << std::setw(int(rows_width))  << std::left  << row_count         << std::endl;
    }
}


void Group::mark_all_table_accessors() noexcept
{
    size_t num_tables = m_table_accessors.size();
    for (size_t table_ndx = 0; table_ndx != num_tables; ++table_ndx) {
        if (Table* table = m_table_accessors[table_ndx]) {
            typedef _impl::TableFriend tf;
            tf::recursive_mark(*table); // Also all subtable accessors
        }
    }
}


namespace {

class MarkDirtyUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::mark(table);
    }

    void update_parent(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::mark(table);
    }

    size_t m_col_ndx;
    DataType m_type;
};


class InsertColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    InsertColumnUpdater(size_t col_ndx):
        m_col_ndx(col_ndx)
    {
    }

    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::adj_insert_column(table, m_col_ndx); // Throws
        tf::mark_link_target_tables(table, m_col_ndx + 1);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx;
};


class EraseColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    EraseColumnUpdater(size_t col_ndx):
        m_col_ndx(col_ndx)
    {
    }

    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::adj_erase_column(table, m_col_ndx);
        tf::mark_link_target_tables(table, m_col_ndx);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx;
};


class MoveColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    MoveColumnUpdater(size_t col_ndx_1, size_t col_ndx_2):
        m_col_ndx_1(col_ndx_1), m_col_ndx_2(col_ndx_2)
    {
    }

    void update(Table& table) override
    {
        using tf = _impl::TableFriend;
        tf::adj_move_column(table, m_col_ndx_1, m_col_ndx_2);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx_1;
    size_t m_col_ndx_2;
};

} // anonymous namespace


// In general, this class cannot assume more than minimal accessor consistency
// (See AccessorConsistencyLevels., it can however assume that replication
// instruction arguments are meaningfull with respect to the current state of
// the accessor hierarchy. For example, a column index argument of `i` is known
// to refer to the `i`'th entry of Table::m_cols.
//
// FIXME: There is currently no checking on valid instruction arguments such as
// column index within bounds. Consider whether we can trust the contents of the
// transaction log enough to skip these checks.
class Group::TransactAdvancer {
public:
    TransactAdvancer(Group& group, bool& schema_changed):
        m_group(group),<commit_msg>Mark linked tables for refresh in column move operation when advancing transactions.<commit_after>/*************************************************************************
 *
 * Copyright 2016 Realm Inc.
 *
 * Licensed 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.
 *
 **************************************************************************/

#include <new>
#include <algorithm>
#include <set>
#include <fstream>

#ifdef REALM_DEBUG
    #include <iostream>
    #include <iomanip>
#endif

#include <realm/util/file_mapper.hpp>
#include <realm/util/memory_stream.hpp>
#include <realm/util/miscellaneous.hpp>
#include <realm/util/thread.hpp>
#include <realm/impl/destroy_guard.hpp>
#include <realm/utilities.hpp>
#include <realm/exceptions.hpp>
#include <realm/column_linkbase.hpp>
#include <realm/column_backlink.hpp>
#include <realm/group_writer.hpp>
#include <realm/group.hpp>
#include <realm/replication.hpp>

using namespace realm;
using namespace realm::util;

namespace {

class Initialization {
public:
    Initialization()
    {
        realm::cpuid_init();
    }
};

Initialization initialization;

} // anonymous namespace


Group::Group():
    m_alloc(), // Throws
    m_top(m_alloc),
    m_tables(m_alloc),
    m_table_names(m_alloc),
    m_is_shared(false)
{
    init_array_parents();
    m_alloc.attach_empty(); // Throws
    Replication::HistoryType history_type = Replication::hist_None;
    int file_format_version = get_target_file_format_version_for_session(0, history_type);
    m_alloc.set_file_format_version(file_format_version);
    ref_type top_ref = 0; // Instantiate a new empty group
    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
}


int Group::get_file_format_version() const noexcept
{
    return m_alloc.get_file_format_version();
}


void Group::set_file_format_version(int file_format) noexcept
{
    m_alloc.set_file_format_version(file_format);
}


int Group::get_committed_file_format_version() const noexcept
{
    return m_alloc.get_committed_file_format_version();
}


int Group::get_target_file_format_version_for_session(int current_file_format_version,
                                                      int requested_history_type) noexcept
{
    Replication::HistoryType requested_history_type_2 =
        Replication::HistoryType(requested_history_type);

    // Note: This function is responsible for choosing the target file format
    // for a sessions. If it selects a file format that is different from
    // `current_file_format_version`, it will trigger a file format upgrade
    // process.

    // Note: Previously it was necessary to base the decision on the current
    // (original) file format and the selected type of history (of
    // changesets). At the present time, however (since the introduction of file
    // format version 5), all files need to use the latest file format.

    // Note: `current_file_format_version` may be zero at this time, which means
    // that the file format it is not yet decided (only possible for empty
    // Realms where top-ref is zero).

    static_cast<void>(current_file_format_version);
    static_cast<void>(requested_history_type_2);
    return Allocator::CURRENT_FILE_FORMAT_VERSION;
}


void Group::upgrade_file_format(int target_file_format_version)
{
    REALM_ASSERT(is_attached());

    // Be sure to revisit the following upgrade logic when a new file foprmat
    // version is introduced. The following assert attempt to help you not
    // forget it.
    REALM_ASSERT_EX(target_file_format_version == 5, target_file_format_version);

    int current_file_format_version = get_file_format_version();
    REALM_ASSERT(current_file_format_version < target_file_format_version);

    // SlabAlloc::validate_buffer() must ensure this. Be sure to revisit the
    // following upgrade logic when SlabAlloc::validate_buffer() is changed (or
    // vice versa).
    REALM_ASSERT_EX(current_file_format_version == 2 ||
                    current_file_format_version == 3 ||
                    current_file_format_version == 4, current_file_format_version);

    // Upgrade from 2 to 3
    if (current_file_format_version <= 2 && target_file_format_version >= 3) {
        for (size_t t = 0; t < m_tables.size(); t++) {
            TableRef table = get_table(t);
            table->upgrade_file_format();
        }
    }

    // Upgrade from 3 to 4
    if (current_file_format_version <= 3 && target_file_format_version >= 4) {
        // No-op
    }

    // Upgrade from 4 to 5 (datetime -> timestamp)
    if (current_file_format_version <= 4 && target_file_format_version >= 5) {
        for (size_t t = 0; t < m_tables.size(); t++) {
            TableRef table = get_table(t);
            table->upgrade_olddatetime();
        }
    }

    // NOTE: Additional future upgrade steps go here.

    set_file_format_version(target_file_format_version);
}


void Group::open(const std::string& file_path, const char* encryption_key, OpenMode mode)
{
    if (is_attached() || m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    SlabAlloc::Config cfg;
    cfg.read_only = mode == mode_ReadOnly;
    cfg.no_create = mode == mode_ReadWriteNoCreate;
    cfg.encryption_key = encryption_key;
    ref_type top_ref = m_alloc.attach_file(file_path, cfg); // Throws
    SlabAlloc::DetachGuard dg(m_alloc);

    // Select file format if it is still undecided.
    int current_file_format_version = get_file_format_version();
    Replication::HistoryType history_type = Replication::hist_None;
    int target_file_format_version =
        get_target_file_format_version_for_session(current_file_format_version, history_type);
    if (current_file_format_version == 0) {
        set_file_format_version(target_file_format_version);
    }
    else {
        // From a technical point of view, we could upgrade the Realm file
        // format in memory here, but since upgrading can be expensive, it is
        // currently disallowed by the SlabAlloc::validate_buffer().
        REALM_ASSERT(target_file_format_version == current_file_format_version);
    }

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
    dg.release(); // Do not detach after all
}


void Group::open(BinaryData buffer, bool take_ownership)
{
    REALM_ASSERT(buffer.data());

    if (is_attached() || m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    ref_type top_ref = m_alloc.attach_buffer(buffer.data(), buffer.size()); // Throws
    SlabAlloc::DetachGuard dg(m_alloc);

    // Select file format if it is still undecided.
    int current_file_format_version = get_file_format_version();
    Replication::HistoryType history_type = Replication::hist_None;
    int target_file_format_version =
        get_target_file_format_version_for_session(current_file_format_version, history_type);
    if (current_file_format_version == 0) {
        set_file_format_version(target_file_format_version);
    }
    else {
        // From a technical point of view, we could upgrade the Realm file
        // format in memory here, but since upgrading can be expensive, it is
        // currently disallowed by the SlabAlloc::validate_buffer().
        REALM_ASSERT(target_file_format_version == current_file_format_version);
    }

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    bool create_group_when_missing = true;
    attach(top_ref, create_group_when_missing); // Throws
    dg.release(); // Do not detach after all

    if (take_ownership)
        m_alloc.own_buffer();
}


Group::~Group() noexcept
{
    // If this group accessor is detached at this point in time, it is either
    // because it is SharedGroup::m_group (m_is_shared), or it is a free-stading
    // group accessor that was never successfully opened.
    if (!m_top.is_attached())
        return;

    // Free-standing group accessor

    detach_table_accessors();

    // Just allow the allocator to release all memory in one chunk without
    // having to traverse the entire tree first
    m_alloc.detach();
}


void Group::remap(size_t new_file_size)
{
    size_t old_baseline = m_alloc.get_baseline();
    if (new_file_size > old_baseline)
        m_alloc.remap(new_file_size); // Throws
}


void Group::remap_and_update_refs(ref_type new_top_ref, size_t new_file_size)
{
    size_t old_baseline = m_alloc.get_baseline();

    if (new_file_size > old_baseline) {
        m_alloc.remap(new_file_size); // Throws
    }

    m_alloc.invalidate_cache();
    update_refs(new_top_ref, old_baseline);
}


void Group::attach(ref_type top_ref, bool create_group_when_missing)
{
    REALM_ASSERT(!m_top.is_attached());

    // If this function throws, it must leave the group accesor in a the
    // unattached state.

    m_tables.detach();
    m_table_names.detach();

    if (top_ref != 0) {
        m_top.init_from_ref(top_ref);
        size_t top_size = m_top.size();
        static_cast<void>(top_size);

        if (top_size < 8) {
            REALM_ASSERT_11(top_size, ==, 3, ||, top_size, ==, 5, ||, top_size, ==, 7);
        }
        else {
            REALM_ASSERT_3(top_size, ==, 9);
        }

        m_table_names.init_from_parent();
        m_tables.init_from_parent();

        // The 3rd slot in m_top is
        // `RefOrTagged::make_tagged(logical_file_size)`, and the logical file
        // size must never exceed actual file size.
        REALM_ASSERT_3(m_top.get_as_ref_or_tagged(2).get_as_int(), <=, m_alloc.get_baseline());
    }
    else if (create_group_when_missing) {
        create_empty_group(); // Throws
    }

    m_attached = true;
}


void Group::detach() noexcept
{
    detach_table_accessors();
    m_table_accessors.clear();

    m_table_names.detach();
    m_tables.detach();
    m_top.detach();

    m_attached = false;
}


void Group::attach_shared(ref_type new_top_ref, size_t new_file_size, bool writable)
{
    REALM_ASSERT_3(new_top_ref, <, new_file_size);
    REALM_ASSERT(!is_attached());

    // Make all dynamically allocated memory (space beyond the attached file) as
    // available free-space.
    reset_free_space_tracking(); // Throws

    // Update memory mapping if database file has grown
    if (new_file_size > m_alloc.get_baseline())
        m_alloc.remap(new_file_size); // Throws

    // When `new_top_ref` is null, ask attach() to create a new node structure
    // for an empty group, but only during the initiation of write
    // transactions. When the transaction being initiated is a read transaction,
    // we instead have to leave array accessors m_top, m_tables, and
    // m_table_names in their detached state, as there are no underlying array
    // nodes to attached them to. In the case of write transactions, the nodes
    // have to be created, as they have to be ready for being modified.
    bool create_group_when_missing = writable;
    attach(new_top_ref, create_group_when_missing); // Throws
}


void Group::detach_table_accessors() noexcept
{
    for (const auto& table_accessor : m_table_accessors) {
        if (Table* t = table_accessor) {
            typedef _impl::TableFriend tf;
            tf::detach(*t);
            tf::unbind_ptr(*t);
        }
    }
}


void Group::create_empty_group()
{
    m_top.create(Array::type_HasRefs); // Throws
    _impl::DeepArrayDestroyGuard dg_top(&m_top);
    {
        m_table_names.create(); // Throws
        _impl::DestroyGuard<ArrayString> dg(&m_table_names);
        m_top.add(m_table_names.get_ref()); // Throws
        dg.release();
    }
    {
        m_tables.create(Array::type_HasRefs); // Throws
        _impl::DestroyGuard<ArrayInteger> dg(&m_tables);
        m_top.add(m_tables.get_ref()); // Throws
        dg.release();
    }
    size_t initial_logical_file_size = sizeof (SlabAlloc::Header);
    m_top.add(RefOrTagged::make_tagged(initial_logical_file_size)); // Throws
    dg_top.release();
}


Table* Group::do_get_table(size_t table_ndx, DescMatcher desc_matcher)
{
    REALM_ASSERT(m_table_accessors.empty() || m_table_accessors.size() == m_tables.size());

    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);

    if (m_table_accessors.empty())
        m_table_accessors.resize(m_tables.size()); // Throws

    // Get table accessor from cache if it exists, else create
    Table* table = m_table_accessors[table_ndx];
    if (!table)
        table = create_table_accessor(table_ndx); // Throws

    if (desc_matcher) {
        typedef _impl::TableFriend tf;
        if (desc_matcher && !(*desc_matcher)(tf::get_spec(*table)))
            throw DescriptorMismatch();
    }

    return table;
}


Table* Group::do_get_table(StringData name, DescMatcher desc_matcher)
{
    if (!m_table_names.is_attached())
        return 0;
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        return 0;

    Table* table = do_get_table(table_ndx, desc_matcher); // Throws
    return table;
}


Table* Group::do_insert_table(size_t table_ndx, StringData name, DescSetter desc_setter,
                              bool require_unique_name)
{
    if (require_unique_name && has_table(name))
        throw TableNameInUse();
    return do_insert_table(table_ndx, name, desc_setter); // Throws
}


Table* Group::do_insert_table(size_t table_ndx, StringData name, DescSetter desc_setter)
{
    if (table_ndx > m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    create_and_insert_table(table_ndx, name); // Throws
    Table* table = do_get_table(table_ndx, nullptr); // Throws
    if (desc_setter)
        (*desc_setter)(*table); // Throws
    return table;
}

Table* Group::do_get_or_insert_table(size_t table_ndx, StringData name, DescMatcher desc_matcher,
                                     DescSetter desc_setter, bool* was_added)
{
    Table* table;
    size_t existing_table_ndx = m_table_names.find_first(name);
    if (existing_table_ndx == not_found) {
        table = do_insert_table(table_ndx, name, desc_setter); // Throws
        if (was_added)
            *was_added = true;
    }
    else {
        table = do_get_table(existing_table_ndx, desc_matcher); // Throws
        if (was_added)
            *was_added = false;
    }
    return table;
}


Table* Group::do_get_or_add_table(StringData name, DescMatcher desc_matcher,
                                  DescSetter desc_setter, bool* was_added)
{
    REALM_ASSERT(m_table_names.is_attached());
    Table* table;
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found) {
        table = do_insert_table(m_tables.size(), name, desc_setter); // Throws
    }
    else {
        table = do_get_table(table_ndx, desc_matcher); // Throws
    }
    if (was_added)
        *was_added = (table_ndx == not_found);
    return table;
}


void Group::create_and_insert_table(size_t table_ndx, StringData name)
{
    if (REALM_UNLIKELY(name.size() > max_table_name_length))
        throw LogicError(LogicError::table_name_too_long);

    using namespace _impl;
    typedef TableFriend tf;
    ref_type ref = tf::create_empty_table(m_alloc); // Throws
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    size_t prior_num_tables = m_tables.size();
    m_tables.insert(table_ndx, ref); // Throws
    m_table_names.insert(table_ndx, name); // Throws

    // Need slot for table accessor
    if (!m_table_accessors.empty()) {
        m_table_accessors.insert(m_table_accessors.begin() + table_ndx, nullptr); // Throws
    }

    update_table_indices([&](size_t old_table_ndx) {
        if (old_table_ndx >= table_ndx) {
            return old_table_ndx + 1;
        }
        return old_table_ndx;
    }); // Throws

    if (Replication* repl = m_alloc.get_replication())
        repl->insert_group_level_table(table_ndx, prior_num_tables, name); // Throws
}


Table* Group::create_table_accessor(size_t table_ndx)
{
    REALM_ASSERT(m_table_accessors.empty() || table_ndx < m_table_accessors.size());

    if (m_table_accessors.empty())
        m_table_accessors.resize(m_tables.size()); // Throws

    // Whenever a table has a link column, the column accessor must be set up to
    // refer to the target table accessor, so the target table accessor needs to
    // be created too, if it does not already exist. This, of course, applies
    // recursively, and it applies to the opposide direction of links too (from
    // target side to origin side). This means that whenever we create a table
    // accessor, we actually need to create the entire cluster of table
    // accessors, that is reachable in zero or more steps along links, or
    // backwards along links.
    //
    // To be able to do this, and to handle the cases where the link
    // relathionship graph contains cycles, each table accessor need to be
    // created in the following steps:
    //
    //  1) Create table accessor, but skip creation of column accessors
    //  2) Register incomplete table accessor in group accessor
    //  3) Mark table accessor
    //  4) Create column accessors
    //  5) Unmark table accessor
    //
    // The marking ensures that the establsihment of the connection between link
    // and backlink column accessors is postponed until both column accessors
    // are created. Infinite recursion due to cycles is prevented by the early
    // registration in the group accessor of inclomplete table accessors.

    typedef _impl::TableFriend tf;
    ref_type ref = m_tables.get_as_ref(table_ndx);
    Table* table = tf::create_incomplete_accessor(m_alloc, ref, this, table_ndx); // Throws

    // The new accessor cannot be leaked, because no exceptions can be thrown
    // before it becomes referenced from `m_column_accessors`.

    // Increase reference count from 0 to 1 to make the group accessor keep
    // the table accessor alive. This extra reference count will be revoked
    // during destruction of the group accessor.
    tf::bind_ptr(*table);

    tf::mark(*table);
    m_table_accessors[table_ndx] = table;
    tf::complete_accessor(*table); // Throws
    tf::unmark(*table);
    return table;
}


void Group::remove_table(StringData name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        throw NoSuchTable();
    remove_table(table_ndx); // Throws
}


void Group::remove_table(size_t table_ndx)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    TableRef table = get_table(table_ndx);

    // In principle we could remove a table even if it is the target of link
    // columns of other tables, however, to do that, we would have to
    // automatically remove the "offending" link columns from those other
    // tables. Such a behaviour is deemed too obscure, and we shall therefore
    // require that a removed table does not contain foreigh origin backlink
    // columns.
    typedef _impl::TableFriend tf;
    if (tf::is_cross_table_link_target(*table))
        throw CrossTableLinkTarget();

    // There is no easy way for Group::TransactAdvancer to handle removal of
    // tables that contain foreign target table link columns, because that
    // involves removal of the corresponding backlink columns. For that reason,
    // we start by removing all columns, which will generate individual
    // replication instructions for each column removal with sufficient
    // information for Group::TransactAdvancer to handle them.
    size_t n = table->get_column_count();
    for (size_t i = n; i > 0; --i)
        table->remove_column(i - 1);

    int64_t ref_64 = m_tables.get(table_ndx);
    REALM_ASSERT(!int_cast_has_overflow<ref_type>(ref_64));
    ref_type ref = ref_type(ref_64);

    // Remove table and move all successive tables
    m_tables.erase(table_ndx); // Throws
    m_table_names.erase(table_ndx); // Throws
    m_table_accessors.erase(m_table_accessors.begin() + table_ndx);

    tf::detach(*table);
    tf::unbind_ptr(*table);

    // Unless the removed table is the last, update all indices of tables after
    // the removed table.
    bool last_table_removed = table_ndx == m_tables.size();
    if (!last_table_removed) {
        update_table_indices([&](size_t old_table_ndx) {
            REALM_ASSERT(old_table_ndx != table_ndx); // We should not see links to the removed table
            if (old_table_ndx > table_ndx) {
                return old_table_ndx - 1;
            }
            return old_table_ndx;
        }); // Throws
    }

    // Destroy underlying node structure
    Array::destroy_deep(ref, m_alloc);

    size_t prior_num_tables = m_tables.size() + 1;
    if (Replication* repl = m_alloc.get_replication())
        repl->erase_group_level_table(table_ndx, prior_num_tables); // Throws
}


void Group::rename_table(StringData name, StringData new_name, bool require_unique_name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    size_t table_ndx = m_table_names.find_first(name);
    if (table_ndx == not_found)
        throw NoSuchTable();
    rename_table(table_ndx, new_name, require_unique_name); // Throws
}


void Group::rename_table(size_t table_ndx, StringData new_name, bool require_unique_name)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    if (table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    if (require_unique_name && has_table(new_name))
        throw TableNameInUse();
    m_table_names.set(table_ndx, new_name);
    if (Replication* repl = m_alloc.get_replication())
        repl->rename_group_level_table(table_ndx, new_name); // Throws
}


void Group::move_table(size_t from_table_ndx, size_t to_table_ndx)
{
    if (REALM_UNLIKELY(!is_attached()))
        throw LogicError(LogicError::detached_accessor);
    REALM_ASSERT_3(m_tables.size(), ==, m_table_names.size());
    REALM_ASSERT_EX(m_table_accessors.empty() || m_table_accessors.size() == m_tables.size(),
                    m_table_accessors.size(), m_tables.size());
    if (from_table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);
    if (to_table_ndx >= m_tables.size())
        throw LogicError(LogicError::table_index_out_of_range);

    if (from_table_ndx == to_table_ndx)
        return;

    // Tables between from_table_ndx and to_table_ndx change their indices, so
    // link columns have to be adjusted (similar to remove_table).

    // Build a map of all table indices that are going to change:
    std::map<size_t, size_t> moves; // from -> to
    moves[from_table_ndx] = to_table_ndx;
    if (from_table_ndx < to_table_ndx) {
        // Move up:
        for (size_t i = from_table_ndx + 1; i <= to_table_ndx; ++i) {
            moves[i] = i - 1;
        }
    }
    else { // from_table_ndx > to_table_ndx
        // Move down:
        for (size_t i = to_table_ndx; i < from_table_ndx; ++i) {
            moves[i] = i + 1;
        }
    }

    // Move entries in internal data structures.
    m_tables.move_rotate(from_table_ndx, to_table_ndx);
    m_table_names.move_rotate(from_table_ndx, to_table_ndx);

    // Move accessors.
    if (!m_table_accessors.empty()) {
        using iter = decltype(m_table_accessors.begin());
        iter first, new_first, last;
        if (from_table_ndx < to_table_ndx) {
            // Rotate left.
            first     = m_table_accessors.begin() + from_table_ndx;
            new_first = first + 1;
            last      = m_table_accessors.begin() + to_table_ndx + 1;
        }
        else { // from_table_ndx > to_table_ndx
            // Rotate right.
            first     = m_table_accessors.begin() + to_table_ndx;
            new_first = m_table_accessors.begin() + from_table_ndx;
            last      = new_first + 1;
        }
        std::rotate(first, new_first, last);
    }

    update_table_indices([&](size_t old_table_ndx) {
        auto it = moves.find(old_table_ndx);
        if (it != moves.end()) {
            return it->second;
        }
        return old_table_ndx;
    });

    if (Replication* repl = m_alloc.get_replication())
        repl->move_group_level_table(from_table_ndx, to_table_ndx); // Throws
}


class Group::DefaultTableWriter: public Group::TableWriter {
public:
    DefaultTableWriter(const Group& group):
        m_group(group)
    {
    }
    ref_type write_names(_impl::OutputStream& out) override
    {
        bool deep = true; // Deep
        bool only_if_modified = false; // Always
        return m_group.m_table_names.write(out, deep, only_if_modified); // Throws
    }
    ref_type write_tables(_impl::OutputStream& out) override
    {
        bool deep = true; // Deep
        bool only_if_modified = false; // Always
        return m_group.m_tables.write(out, deep, only_if_modified); // Throws
    }
private:
    const Group& m_group;
};

void Group::write(std::ostream& out, bool pad) const
{
    write(out, pad, 0);
}

void Group::write(std::ostream& out, bool pad_for_encryption, uint_fast64_t version_number) const
{
    REALM_ASSERT(is_attached());
    DefaultTableWriter table_writer(*this);
    bool no_top_array = !m_top.is_attached();
    write(out, m_alloc, table_writer, no_top_array, pad_for_encryption, version_number); // Throws
}

void Group::write(const std::string& path, const char* encryption_key) const
{
    write(path, encryption_key, 0);
}

void Group::write(const std::string& path, const char* encryption_key, uint_fast64_t version_number) const
{
    File file;
    int flags = 0;
    file.open(path, File::access_ReadWrite, File::create_Must, flags);
    write(file, encryption_key, version_number);
}

void Group::write(File& file, const char* encryption_key, uint_fast64_t version_number) const
{
    REALM_ASSERT(file.get_size() == 0);

    file.set_encryption_key(encryption_key);
    File::Streambuf streambuf(&file);
    std::ostream out(&streambuf);
    write(out, encryption_key != 0, version_number);
}

BinaryData Group::write_to_mem() const
{
    REALM_ASSERT(is_attached());

    // Get max possible size of buffer
    //
    // FIXME: This size could potentially be vastly bigger that what
    // is actually needed.
    size_t max_size = m_alloc.get_total_size();

    char* buffer = static_cast<char*>(malloc(max_size)); // Throws
    if (!buffer)
        throw std::bad_alloc();
    try {
        MemoryOutputStream out; // Throws
        out.set_buffer(buffer, buffer + max_size);
        write(out); // Throws
        size_t buffer_size = out.size();
        return BinaryData(buffer, buffer_size);
    }
    catch (...) {
        free(buffer);
        throw;
    }
}


void Group::write(std::ostream& out, const Allocator& alloc, TableWriter& table_writer,
                  bool no_top_array, bool pad_for_encryption, uint_fast64_t version_number)
{
    _impl::OutputStream out_2(out);

    // Write the file header
    SlabAlloc::Header streaming_header;
    int file_format_version = (no_top_array ? 0 : alloc.get_file_format_version());
    SlabAlloc::init_streaming_header(&streaming_header, file_format_version);
    out_2.write(reinterpret_cast<const char*>(&streaming_header), sizeof streaming_header);

    ref_type top_ref = 0;
    size_t final_file_size = sizeof streaming_header;
    if (no_top_array) {
        // Accept version number 1 as that number is (unfortunately) also used
        // to denote the empty initial state of a Realm file.
        REALM_ASSERT(version_number == 0 || version_number == 1);
    }
    else {
        // Because we need to include the total logical file size in the
        // top-array, we have to start by writing everything except the
        // top-array, and then finally compute and write a correct version of
        // the top-array. The free-space information of the group will only be
        // included if a non-zero version number is given as parameter,
        // indicating that versioning info is to be saved. This is used from
        // SharedGroup to compact the database by writing only the live data
        // into a separate file.
        ref_type names_ref  = table_writer.write_names(out_2); // Throws
        ref_type tables_ref = table_writer.write_tables(out_2); // Throws
        SlabAlloc new_alloc;
        new_alloc.attach_empty(); // Throws
        new_alloc.set_file_format_version(alloc.get_file_format_version());
        Array top(new_alloc);
        top.create(Array::type_HasRefs); // Throws
        _impl::ShallowArrayDestroyGuard dg_top(&top);
        // FIXME: We really need an alternative to Array::truncate() that is able to expand.
        int_fast64_t value_1 = from_ref(names_ref);
        int_fast64_t value_2 = from_ref(tables_ref);
        top.add(value_1); // Throws
        top.add(value_2); // Throws
        top.add(0); // Throws

        int top_size = 3;
        if (version_number) {
            Array free_list(new_alloc);
            Array size_list(new_alloc);
            Array version_list(new_alloc);
            free_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_1(&free_list);
            size_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_2(&size_list);
            version_list.create(Array::type_Normal); // Throws
            _impl::DeepArrayDestroyGuard dg_3(&version_list);
            bool deep = true; // Deep
            bool only_if_modified = false; // Always
            ref_type free_list_ref = free_list.write(out_2, deep, only_if_modified);
            ref_type size_list_ref = size_list.write(out_2, deep, only_if_modified);
            ref_type version_list_ref = version_list.write(out_2, deep, only_if_modified);
            top.add(RefOrTagged::make_ref(free_list_ref)); // Throws
            top.add(RefOrTagged::make_ref(size_list_ref)); // Throws
            top.add(RefOrTagged::make_ref(version_list_ref)); // Throws
            top.add(RefOrTagged::make_tagged(version_number)); // Throws
            top_size = 7;
        }
        top_ref = out_2.get_ref_of_next_array();

        // Produce a preliminary version of the top array whose
        // representation is guaranteed to be able to hold the final file
        // size
        size_t max_top_byte_size = Array::get_max_byte_size(top_size);
        size_t max_final_file_size = size_t(top_ref) + max_top_byte_size;
        top.ensure_minimum_width(RefOrTagged::make_tagged(max_final_file_size)); // Throws

        // Finalize the top array by adding the projected final file size
        // to it
        size_t top_byte_size = top.get_byte_size();
        final_file_size = size_t(top_ref) + top_byte_size;
        top.set(2, RefOrTagged::make_tagged(final_file_size)); // Throws

        // Write the top array
        bool deep = false; // Shallow
        bool only_if_modified = false; // Always
        top.write(out_2, deep, only_if_modified); // Throws
        REALM_ASSERT_3(size_t(out_2.get_ref_of_next_array()), ==, final_file_size);

        dg_top.reset(nullptr); // Destroy now
    }

    // encryption will pad the file to a multiple of the page, so ensure the
    // footer is aligned to the end of a page
    if (pad_for_encryption) {
#if REALM_ENABLE_ENCRYPTION
        size_t unrounded_size = final_file_size + sizeof(SlabAlloc::StreamingFooter);
        size_t rounded_size = round_up_to_page_size(unrounded_size);
        if (rounded_size != unrounded_size) {
            std::unique_ptr<char[]> buffer(new char[rounded_size - unrounded_size]());
            out_2.write(buffer.get(), rounded_size - unrounded_size);
        }
#endif
    }

    // Write streaming footer
    SlabAlloc::StreamingFooter footer;
    footer.m_top_ref = top_ref;
    footer.m_magic_cookie = SlabAlloc::footer_magic_cookie;
    out_2.write(reinterpret_cast<const char*>(&footer), sizeof footer);
}


void Group::commit()
{
    if (!is_attached())
        throw LogicError(LogicError::detached_accessor);
    if (m_is_shared)
        throw LogicError(LogicError::wrong_group_state);

    GroupWriter out(*this); // Throws

    // Recursively write all changed arrays to the database file. We
    // postpone the commit until we are sure that no exceptions can be
    // thrown.
    ref_type top_ref = out.write_group(); // Throws

    // Since the group is persisiting in single-thread (unshared)
    // mode we have to make sure that the group stays valid after
    // commit

    // Mark all managed space (beyond the attached file) as free.
    reset_free_space_tracking(); // Throws

    size_t old_baseline = m_alloc.get_baseline();

    // Remap file if it has grown
    size_t new_file_size = out.get_file_size();
    if (new_file_size > old_baseline) {
        m_alloc.remap(new_file_size); // Throws
    }

    out.commit(top_ref); // Throws

    // Recursively update refs in all active tables (columns, arrays..)
    update_refs(top_ref, old_baseline);
}


void Group::update_refs(ref_type top_ref, size_t old_baseline) noexcept
{
    // After Group::commit() we will always have free space tracking
    // info.
    REALM_ASSERT_3(m_top.size(), >=, 5);

    // Array nodes that are part of the previous version of the
    // database will not be overwritten by Group::commit(). This is
    // necessary for robustness in the face of abrupt termination of
    // the process. It also means that we can be sure that an array
    // remains unchanged across a commit if the new ref is equal to
    // the old ref and the ref is below the previous baseline.

    if (top_ref < old_baseline && m_top.get_ref() == top_ref)
        return;

    m_top.init_from_ref(top_ref);

    // Now we can update it's child arrays
    m_table_names.update_from_parent(old_baseline);

    // If m_tables has not been modfied we don't
    // need to update attached table accessors
    if (!m_tables.update_from_parent(old_baseline))
        return;

    // Update all attached table accessors including those attached to
    // subtables.
    for (const auto& table_accessor : m_table_accessors) {
        typedef _impl::TableFriend tf;
        if (Table* table = table_accessor)
            tf::update_from_parent(*table, old_baseline);
    }
}


bool Group::operator==(const Group& g) const
{
    size_t n = size();
    if (n != g.size())
        return false;
    for (size_t i = 0; i < n; ++i) {
        const StringData& table_name_1 = get_table_name(i); // Throws
        const StringData& table_name_2 = g.get_table_name(i); // Throws
        if (table_name_1 != table_name_2)
            return false;

        ConstTableRef table_1 = get_table(i); // Throws
        ConstTableRef table_2 = g.get_table(i); // Throws
        if (*table_1 != *table_2)
            return false;
    }
    return true;
}


void Group::to_string(std::ostream& out) const
{
    // Calculate widths
    size_t index_width = 4;
    size_t name_width = 10;
    size_t rows_width = 6;
    size_t count = size();
    for (size_t i = 0; i < count; ++i) {
        StringData name = get_table_name(i);
        if (name_width < name.size())
            name_width = name.size();

        ConstTableRef table = get_table(name);
        size_t row_count = table->size();
        if (rows_width < row_count) { // FIXME: should be the number of digits in row_count: floor(log10(row_count+1))
            rows_width = row_count;
        }
    }


    // Print header
    out << std::setw(int(index_width + 1)) << std::left << " ";
    out << std::setw(int(name_width + 1))  << std::left << "tables";
    out << std::setw(int(rows_width))    << std::left << "rows"    << std::endl;

    // Print tables
    for (size_t i = 0; i < count; ++i) {
        StringData name = get_table_name(i);
        ConstTableRef table = get_table(name);
        size_t row_count = table->size();

        out << std::setw(int(index_width)) << std::right << i                 << " ";
        out << std::setw(int(name_width))  << std::left  << std::string(name) << " ";
        out << std::setw(int(rows_width))  << std::left  << row_count         << std::endl;
    }
}


void Group::mark_all_table_accessors() noexcept
{
    size_t num_tables = m_table_accessors.size();
    for (size_t table_ndx = 0; table_ndx != num_tables; ++table_ndx) {
        if (Table* table = m_table_accessors[table_ndx]) {
            typedef _impl::TableFriend tf;
            tf::recursive_mark(*table); // Also all subtable accessors
        }
    }
}


namespace {

class MarkDirtyUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::mark(table);
    }

    void update_parent(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::mark(table);
    }

    size_t m_col_ndx;
    DataType m_type;
};


class InsertColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    InsertColumnUpdater(size_t col_ndx):
        m_col_ndx(col_ndx)
    {
    }

    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::adj_insert_column(table, m_col_ndx); // Throws
        tf::mark_link_target_tables(table, m_col_ndx + 1);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx;
};


class EraseColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    EraseColumnUpdater(size_t col_ndx):
        m_col_ndx(col_ndx)
    {
    }

    void update(Table& table) override
    {
        typedef _impl::TableFriend tf;
        tf::adj_erase_column(table, m_col_ndx);
        tf::mark_link_target_tables(table, m_col_ndx);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx;
};


class MoveColumnUpdater: public _impl::TableFriend::AccessorUpdater {
public:
    MoveColumnUpdater(size_t col_ndx_1, size_t col_ndx_2):
        m_col_ndx_1(col_ndx_1), m_col_ndx_2(col_ndx_2)
    {
    }

    void update(Table& table) override
    {
        using tf = _impl::TableFriend;
        tf::adj_move_column(table, m_col_ndx_1, m_col_ndx_2);
        size_t min_ndx = std::min(m_col_ndx_1, m_col_ndx_2);
        tf::mark_link_target_tables(table, min_ndx);
    }

    void update_parent(Table&) override
    {
    }

private:
    size_t m_col_ndx_1;
    size_t m_col_ndx_2;
};

} // anonymous namespace


// In general, this class cannot assume more than minimal accessor consistency
// (See AccessorConsistencyLevels., it can however assume that replication
// instruction arguments are meaningfull with respect to the current state of
// the accessor hierarchy. For example, a column index argument of `i` is known
// to refer to the `i`'th entry of Table::m_cols.
//
// FIXME: There is currently no checking on valid instruction arguments such as
// column index within bounds. Consider whether we can trust the contents of the
// transaction log enough to skip these checks.
class Group::TransactAdvancer {
public:
    TransactAdvancer(Group& group, bool& schema_changed):
        m_group(group),
{'fraction_non_alphanumeric': 0.10390096121479597, 'fraction_numerical': 0.003964543516475297, 'mean_word_length': 3.1071172555043525, 'pattern_counts': {'":': 0, '<': 189, '<?xml version=': 0, '>': 127, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5131869', 'n_tokens_mistral': 25916, 'n_tokens_neox': 25220, 'n_words': 6354}