<commit_before>/* Conversion of Postgres server-side structures into the wire protocol, which
 * is emitted by the output plugin and consumed by the client. */

#include "protocol_server.h"
#include "io_util.h"
#include "oid2avro.h"

#include <stdarg.h>
#include <string.h>
#include "access/heapam.h"
#include "lib/stringinfo.h"
#include "utils/lsyscache.h"

int extract_tuple_key(schema_cache_entry *entry, Relation rel, TupleDesc tupdesc, HeapTuple tuple, bytea **key_out);
int update_frame_with_table_schema(avro_value_t *frame_val, schema_cache_entry *entry);
int update_frame_with_insert_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *new_bin);
int update_frame_with_update_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *old_bin, bytea *new_bin);
int update_frame_with_delete_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *old_bin);
int schema_cache_lookup(schema_cache_t cache, Relation rel, schema_cache_entry **entry_out);
schema_cache_entry *schema_cache_entry_new(schema_cache_t cache);
void schema_cache_entry_update(schema_cache_entry *entry, Relation rel);
void schema_cache_entry_decrefs(schema_cache_entry *entry);
uint64 fnv_hash(uint64 base, char *str, int len);
uint64 fnv_format(uint64 base, char *fmt, ...) __attribute__ ((format (printf, 2, 3)));
uint64 schema_hash_for_relation(Relation rel);
void tupdesc_debug_info(StringInfo msg, TupleDesc tupdesc);

/* http://www.isthe.com/chongo/tech/comp/fnv/index.html#FNV-param */
#define FNV_HASH_BASE UINT64CONST(0xcbf29ce484222325)
#define FNV_HASH_PRIME UINT64CONST(0x100000001b3)
#define FNV_HASH_BUFSIZE 256

/* Populates a wire protocol message for a "begin transaction" event. */
int update_frame_with_begin_txn(avro_value_t *frame_val, ReorderBufferTXN *txn) {
    int err = 0;
    avro_value_t msg_val, union_val, record_val, xid_val;

    check(err, avro_value_get_by_index(frame_val, 0, &msg_val, NULL));
    check(err, avro_value_append(&msg_val, &union_val, NULL));
    check(err, avro_value_set_branch(&union_val, PROTOCOL_MSG_BEGIN_TXN, &record_val));
    check(err, avro_value_get_by_index(&record_val, 0, &xid_val, NULL));
    check(err, avro_value_set_long(&xid_val, txn->xid));
    return err;
}

/* Populates a wire protocol message for a "commit transaction" event. */
int update_frame_with_commit_txn(avro_value_t *frame_val, ReorderBufferTXN *txn,
        XLogRecPtr commit_lsn) {
    int err = 0;
    avro_value_t msg_val, union_val, record_val, xid_val, lsn_val;

    check(err, avro_value_get_by_index(frame_val, 0, &msg_val, NULL));
    check(err, avro_value_append(&msg_val, &union_val, NULL));
    check(err, avro_value_set_branch(&union_val, PROTOCOL_MSG_COMMIT_TXN, &record_val));
    check(err, avro_value_get_by_index(&record_val, 0, &xid_val, NULL));
    check(err, avro_value_get_by_index(&record_val, 1, &lsn_val, NULL));
    check(err, avro_value_set_long(&xid_val, txn->xid));
    check(err, avro_value_set_long(&lsn_val, commit_lsn));
    return err;
}

/* If we're using a primary key/replica identity index for a given table, this
 * function extracts that index' columns from a row tuple, and encodes the values
 * as an Avro string using the table's key schema. */
int extract_tuple_key(schema_cache_entry *entry, Relation rel, TupleDesc tupdesc, HeapTuple tuple, bytea **key_out) {
    int err = 0;
    if (entry->key_schema) {
        check(err, avro_value_reset(&entry->key_value));

        Relation index_rel = table_key_index(rel);
        err = tuple_to_avro_key(&entry->key_value, tupdesc, tuple, rel, index_rel->rd_index);
        relation_close(index_rel, AccessShareLock);

        if(err){
          return err;
        }

        check(err, try_writing(key_out, &write_avro_binary, &entry->key_value));
    }
    return err;
}

/* Updates the given frame value with a tuple inserted into a table. The table
 * schema is automatically included in the frame if it's not in the cache. This
 * function is used both during snapshot and during stream replication.
 *
 * The TupleDesc parameter is not redundant. During stream replication, it is just
 * RelationGetDescr(rel), but during snapshot it is taken from the result set.
 * The difference is that the result set tuple has dropped (logically invisible)
 * columns omitted. */
int update_frame_with_insert(avro_value_t *frame_val, schema_cache_t cache, Relation rel, TupleDesc tupdesc, HeapTuple newtuple) {
    int err = 0;
    schema_cache_entry *entry;
    bytea *key_bin = NULL, *new_bin = NULL;

    int changed = schema_cache_lookup(cache, rel, &entry);
    if (changed) {
        check(err, update_frame_with_table_schema(frame_val, entry));
    }

    check(err, extract_tuple_key(entry, rel, tupdesc, newtuple, &key_bin));<commit_msg>Tweak whitespace to match conventions<commit_after>/* Conversion of Postgres server-side structures into the wire protocol, which
 * is emitted by the output plugin and consumed by the client. */

#include "protocol_server.h"
#include "io_util.h"
#include "oid2avro.h"

#include <stdarg.h>
#include <string.h>
#include "access/heapam.h"
#include "lib/stringinfo.h"
#include "utils/lsyscache.h"

int extract_tuple_key(schema_cache_entry *entry, Relation rel, TupleDesc tupdesc, HeapTuple tuple, bytea **key_out);
int update_frame_with_table_schema(avro_value_t *frame_val, schema_cache_entry *entry);
int update_frame_with_insert_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *new_bin);
int update_frame_with_update_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *old_bin, bytea *new_bin);
int update_frame_with_delete_raw(avro_value_t *frame_val, Oid relid, bytea *key_bin, bytea *old_bin);
int schema_cache_lookup(schema_cache_t cache, Relation rel, schema_cache_entry **entry_out);
schema_cache_entry *schema_cache_entry_new(schema_cache_t cache);
void schema_cache_entry_update(schema_cache_entry *entry, Relation rel);
void schema_cache_entry_decrefs(schema_cache_entry *entry);
uint64 fnv_hash(uint64 base, char *str, int len);
uint64 fnv_format(uint64 base, char *fmt, ...) __attribute__ ((format (printf, 2, 3)));
uint64 schema_hash_for_relation(Relation rel);
void tupdesc_debug_info(StringInfo msg, TupleDesc tupdesc);

/* http://www.isthe.com/chongo/tech/comp/fnv/index.html#FNV-param */
#define FNV_HASH_BASE UINT64CONST(0xcbf29ce484222325)
#define FNV_HASH_PRIME UINT64CONST(0x100000001b3)
#define FNV_HASH_BUFSIZE 256

/* Populates a wire protocol message for a "begin transaction" event. */
int update_frame_with_begin_txn(avro_value_t *frame_val, ReorderBufferTXN *txn) {
    int err = 0;
    avro_value_t msg_val, union_val, record_val, xid_val;

    check(err, avro_value_get_by_index(frame_val, 0, &msg_val, NULL));
    check(err, avro_value_append(&msg_val, &union_val, NULL));
    check(err, avro_value_set_branch(&union_val, PROTOCOL_MSG_BEGIN_TXN, &record_val));
    check(err, avro_value_get_by_index(&record_val, 0, &xid_val, NULL));
    check(err, avro_value_set_long(&xid_val, txn->xid));
    return err;
}

/* Populates a wire protocol message for a "commit transaction" event. */
int update_frame_with_commit_txn(avro_value_t *frame_val, ReorderBufferTXN *txn,
        XLogRecPtr commit_lsn) {
    int err = 0;
    avro_value_t msg_val, union_val, record_val, xid_val, lsn_val;

    check(err, avro_value_get_by_index(frame_val, 0, &msg_val, NULL));
    check(err, avro_value_append(&msg_val, &union_val, NULL));
    check(err, avro_value_set_branch(&union_val, PROTOCOL_MSG_COMMIT_TXN, &record_val));
    check(err, avro_value_get_by_index(&record_val, 0, &xid_val, NULL));
    check(err, avro_value_get_by_index(&record_val, 1, &lsn_val, NULL));
    check(err, avro_value_set_long(&xid_val, txn->xid));
    check(err, avro_value_set_long(&lsn_val, commit_lsn));
    return err;
}

/* If we're using a primary key/replica identity index for a given table, this
 * function extracts that index' columns from a row tuple, and encodes the values
 * as an Avro string using the table's key schema. */
int extract_tuple_key(schema_cache_entry *entry, Relation rel, TupleDesc tupdesc, HeapTuple tuple, bytea **key_out) {
    int err = 0;
    if (entry->key_schema) {
        check(err, avro_value_reset(&entry->key_value));

        Relation index_rel = table_key_index(rel);
        err = tuple_to_avro_key(&entry->key_value, tupdesc, tuple, rel, index_rel->rd_index);
        relation_close(index_rel, AccessShareLock);

        if (err) {
            return err;
        }

        check(err, try_writing(key_out, &write_avro_binary, &entry->key_value));
    }
    return err;
}

/* Updates the given frame value with a tuple inserted into a table. The table
 * schema is automatically included in the frame if it's not in the cache. This
 * function is used both during snapshot and during stream replication.
 *
 * The TupleDesc parameter is not redundant. During stream replication, it is just
 * RelationGetDescr(rel), but during snapshot it is taken from the result set.
 * The difference is that the result set tuple has dropped (logically invisible)
 * columns omitted. */
int update_frame_with_insert(avro_value_t *frame_val, schema_cache_t cache, Relation rel, TupleDesc tupdesc, HeapTuple newtuple) {
    int err = 0;
    schema_cache_entry *entry;
    bytea *key_bin = NULL, *new_bin = NULL;

    int changed = schema_cache_lookup(cache, rel, &entry);
    if (changed) {
        check(err, update_frame_with_table_schema(frame_val, entry));
    }

    check(err, extract_tuple_key(entry, rel, tupdesc, newtuple, &key_bin));
{'fraction_non_alphanumeric': 0.10069081013188193, 'fraction_numerical': 0.01088549298723048, 'mean_word_length': 5.374249499666444, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, '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': '2091495', 'n_tokens_mistral': 3441, 'n_tokens_neox': 3333, 'n_words': 852}