<commit_before>#ifndef TABLE_ROW_APPLIER_H
#define TABLE_ROW_APPLIER_H

#include "sql_functions.h"
#include "unique_key_clearer.h"

typedef map<ColumnValues, NullableRow> RowsByPrimaryKey;

template <typename DatabaseClient>
struct RowLoader {
	RowLoader(const Table &table, RowsByPrimaryKey &rows): table(table), rows(rows) {}

	void operator()(const typename DatabaseClient::RowType &database_row) {
		ColumnValues primary_key;
		primary_key.resize(table.primary_key_columns.size());
		for (size_t column = 0; column < table.primary_key_columns.size(); column++) {
			primary_key[column] = database_row.string_at(table.primary_key_columns[column]);
		}

		NullableRow &row(rows[primary_key]);
		row.resize(database_row.n_columns());
		for (size_t column = 0; column < database_row.n_columns(); column++) {
			if (!database_row.null_at(column)) {
				row[column] = database_row.string_at(column);
			}
		}
	}

	const Table &table;
	RowsByPrimaryKey &rows;
};

template <typename DatabaseClient>
struct TableRowApplier {
	TableRowApplier(DatabaseClient &client, const Table &table):
		client(client),
		table(table),
		primary_key_columns(columns_list(table.columns, table.primary_key_columns, client.quote_identifiers_with())),
		primary_key_clearer(client, table, table.primary_key_columns),
		insert_sql(client.replace_sql_prefix() + table.name + " VALUES\n(", ")"),
		rows_changed(0) {

		// if the client doesn't support REPLACE, we will need to delete rows with the PKs we want to
		// insert, and also clear later rows that have our unique key values in order to insert
		client.add_replace_clearers(unique_keys_clearers, table);
	}

	~TableRowApplier() {
		apply();
	}

	template <typename InputStream>
	size_t stream_from_input(Unpacker<InputStream> &input, const ColumnValues &matched_up_to_key, const ColumnValues &last_not_matching_key) {
		// we're being sent the range of rows > matched_up_to_key and <= last_not_matching_key; apply them to our end

		RowsByPrimaryKey existing_rows;

		if (last_not_matching_key.empty()) {
			// if the range is to the end of the table, clear all remaining rows at our end
			delete_range(matched_up_to_key, last_not_matching_key);
		} else {
			// otherwise, load our rows in the range so we can compare them
			RowLoader<DatabaseClient> row_loader(table, existing_rows);
			client.retrieve_rows(table, matched_up_to_key, last_not_matching_key, row_loader);
		}

		NullableRow row;
		size_t rows_in_range = 0;

		while (true) {
			// the rows command is unusual.  to avoid needing to know the number of results in advance,
			// instead of a single response object, there's one response object per row, terminated by
			// an empty row (which is not valid data, so is unambiguous).
			input >> row;
			if (row.size() == 0) break;
			rows_in_range++;

			if (last_not_matching_key.empty()) {
				// if we're inserting the range to the end of the table, we know we need to insert this row
				// since there can be no later rows, we don't need to clear unique keys these rows use
				add_to_insert(row);
				rows_changed++;
			} else {
				// otherwise, if we don't have this row or if our row is different, we need replace our row
				if (consider_replace(existing_rows, row)) {
					rows_changed++;
				}
			}
		}

		// clear any remaining rows the other end didn't have
		for (RowsByPrimaryKey::const_iterator it = existing_rows.begin(); it != existing_rows.end(); ++it) {
			add_to_primary_key_clearer(it->second);
		}
		rows_changed  += existing_rows.size();
		rows_in_range += existing_rows.size();

		return rows_in_range;
	}

	bool consider_replace(RowsByPrimaryKey &existing_rows, const NullableRow &row) {<commit_msg>update an outdated comment<commit_after>#ifndef TABLE_ROW_APPLIER_H
#define TABLE_ROW_APPLIER_H

#include "sql_functions.h"
#include "unique_key_clearer.h"

typedef map<ColumnValues, NullableRow> RowsByPrimaryKey;

template <typename DatabaseClient>
struct RowLoader {
	RowLoader(const Table &table, RowsByPrimaryKey &rows): table(table), rows(rows) {}

	void operator()(const typename DatabaseClient::RowType &database_row) {
		ColumnValues primary_key;
		primary_key.resize(table.primary_key_columns.size());
		for (size_t column = 0; column < table.primary_key_columns.size(); column++) {
			primary_key[column] = database_row.string_at(table.primary_key_columns[column]);
		}

		NullableRow &row(rows[primary_key]);
		row.resize(database_row.n_columns());
		for (size_t column = 0; column < database_row.n_columns(); column++) {
			if (!database_row.null_at(column)) {
				row[column] = database_row.string_at(column);
			}
		}
	}

	const Table &table;
	RowsByPrimaryKey &rows;
};

template <typename DatabaseClient>
struct TableRowApplier {
	TableRowApplier(DatabaseClient &client, const Table &table):
		client(client),
		table(table),
		primary_key_columns(columns_list(table.columns, table.primary_key_columns, client.quote_identifiers_with())),
		primary_key_clearer(client, table, table.primary_key_columns),
		insert_sql(client.replace_sql_prefix() + table.name + " VALUES\n(", ")"),
		rows_changed(0) {

		// if the client doesn't support REPLACE, we will need to delete rows with the PKs we want to
		// insert, and also clear later rows that have our unique key values in order to insert
		client.add_replace_clearers(unique_keys_clearers, table);
	}

	~TableRowApplier() {
		apply();
	}

	template <typename InputStream>
	size_t stream_from_input(Unpacker<InputStream> &input, const ColumnValues &matched_up_to_key, const ColumnValues &last_not_matching_key) {
		// we're being sent the range of rows > matched_up_to_key and <= last_not_matching_key; apply them to our end

		RowsByPrimaryKey existing_rows;

		if (last_not_matching_key.empty()) {
			// if the range is to the end of the table, clear all remaining rows at our end
			delete_range(matched_up_to_key, last_not_matching_key);
		} else {
			// otherwise, load our rows in the range so we can compare them
			RowLoader<DatabaseClient> row_loader(table, existing_rows);
			client.retrieve_rows(table, matched_up_to_key, last_not_matching_key, row_loader);
		}

		NullableRow row;
		size_t rows_in_range = 0;

		while (true) {
			// command responses are a series of arrays, terminated by an empty array.  this avoids having
			// to know the number of results in advance; an empty row is not valid, so it's unambiguous.
			input >> row;
			if (row.size() == 0) break;
			rows_in_range++;

			if (last_not_matching_key.empty()) {
				// if we're inserting the range to the end of the table, we know we need to insert this row
				// since there can be no later rows, we don't need to clear unique keys these rows use
				add_to_insert(row);
				rows_changed++;
			} else {
				// otherwise, if we don't have this row or if our row is different, we need replace our row
				if (consider_replace(existing_rows, row)) {
					rows_changed++;
				}
			}
		}

		// clear any remaining rows the other end didn't have
		for (RowsByPrimaryKey::const_iterator it = existing_rows.begin(); it != existing_rows.end(); ++it) {
			add_to_primary_key_clearer(it->second);
		}
		rows_changed  += existing_rows.size();
		rows_in_range += existing_rows.size();

		return rows_in_range;
	}

	bool consider_replace(RowsByPrimaryKey &existing_rows, const NullableRow &row) {
{'fraction_non_alphanumeric': 0.10039720586221065, 'fraction_numerical': 0.0013696753869332967, 'mean_word_length': 4.691348402182385, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1425351', 'n_tokens_mistral': 2562, 'n_tokens_neox': 2287, 'n_words': 668}