Update field_diff logic to handle array field types (#537)

Co-authored-by: Laurentiu Cristofor <[email protected]>

production/db/core/src/payload_diff.cpp

@@ -44,12 +44,9 @@ void compute_payload_diff(
     for (auto field_view : catalog_core_t::list_fields(type_record_id))
     {
         field_position_t field_position = field_view.position();
-        payload_types::data_holder_t data_holder1 = payload_types::get_field_value(
-            type_id, payload1, schema.data(), schema.size(), field_position);
-        payload_types::data_holder_t data_holder2 = payload_types::get_field_value(
-            type_id, payload2, schema.data(), schema.size(), field_position);
 
-        if (data_holder1.compare(data_holder2) != 0)
+        if (!payload_types::are_field_values_equal(
+                type_id, payload1, payload2, schema.data(), schema.size(), field_position))
         {
             changed_fields->push_back(field_position);
         }

production/db/inc/payload_types/field_access.hpp

@@ -11,6 +11,7 @@
 #include "flatbuffers/reflection.h"
 
 #include "gaia/exception.hpp"
+
 #include "data_holder.hpp"
 #include "type_cache.hpp"
 
@@ -100,6 +101,16 @@ bool verify_data_schema(
     size_t serialized_data_size,
     const uint8_t* binary_schema);
 
+// Returns true if the field values are equal and false if they are different.
+// This function handles both scalar fields and array fields.
+bool are_field_values_equal(
+    gaia::common::gaia_type_t type_id,
+    const uint8_t* first_serialized_data,
+    const uint8_t* second_serialized_data,
+    const uint8_t* binary_schema,
+    size_t binary_schema_size,
+    gaia::common::field_position_t field_position);
+
 // Get the field value of a table record payload.
 data_holder_t get_field_value(
     gaia::common::gaia_type_t type_id,
@@ -110,7 +121,7 @@ data_holder_t get_field_value(
 
 // Set the scalar field value of a table record payload.
 //
-// This function only works for scalar fields (integers and floating point numbers).
+// This function only works for scalar fields of numeric types (integers and floating point numbers).
 bool set_field_value(
     gaia::common::gaia_type_t type_id,
     uint8_t* serialized_data,
@@ -185,7 +196,7 @@ data_holder_t get_field_array_element(
 //
 // An exception will be thrown if the index is out of bounds.
 //
-// This function only works for scalar fields (integers and floating point numbers).
+// This function only works for scalar fields of numeric types (integers and floating point numbers).
 void set_field_array_element(
     gaia::common::gaia_type_t type_id,
     uint8_t* serialized_data,

production/db/payload_types/src/field_access.cpp

@@ -206,6 +206,82 @@ void get_table_field_array_information(
     }
 }
 
+bool are_field_values_equal(
+    gaia_type_t type_id,
+    const uint8_t* first_serialized_data,
+    const uint8_t* second_serialized_data,
+    const uint8_t* binary_schema,
+    size_t binary_schema_size,
+    field_position_t field_position)
+{
+    const flatbuffers::Table* first_root_table = nullptr;
+    auto_type_information_t auto_type_information;
+    type_information_t local_type_information;
+    const reflection::Field* field = nullptr;
+
+    get_table_field_information(
+        type_id, first_serialized_data, binary_schema, binary_schema_size, field_position,
+        first_root_table, auto_type_information, local_type_information, field);
+
+    const flatbuffers::Table* second_root_table = flatbuffers::GetAnyRoot(second_serialized_data);
+    if (second_root_table == nullptr)
+    {
+        throw invalid_serialized_data();
+    }
+
+    // Compare field values according to their type.
+    if (flatbuffers::IsInteger(field->type()->base_type()))
+    {
+        int64_t first_value = flatbuffers::GetAnyFieldI(*first_root_table, *field);
+        int64_t second_value = flatbuffers::GetAnyFieldI(*second_root_table, *field);
+
+        return first_value == second_value;
+    }
+    else if (flatbuffers::IsFloat(field->type()->base_type()))
+    {
+        double first_value = flatbuffers::GetAnyFieldF(*first_root_table, *field);
+        double second_value = flatbuffers::GetAnyFieldF(*second_root_table, *field);
+
+        return first_value == second_value;
+    }
+    else if (field->type()->base_type() == reflection::String)
+    {
+        const flatbuffers::String* first_value = flatbuffers::GetFieldS(*first_root_table, *field);
+        const flatbuffers::String* second_value = flatbuffers::GetFieldS(*second_root_table, *field);
+
+        if (first_value == nullptr || second_value == nullptr)
+        {
+            return first_value == second_value;
+        }
+        else
+        {
+            return strcmp(first_value->c_str(), second_value->c_str()) == 0;
+        }
+    }
+    else if (field->type()->base_type() == reflection::Vector)
+    {
+        flatbuffers::VectorOfAny* first_value = flatbuffers::GetFieldAnyV(*first_root_table, *field);
+        flatbuffers::VectorOfAny* second_value = flatbuffers::GetFieldAnyV(*second_root_table, *field);
+
+        if (first_value == nullptr || second_value == nullptr)
+        {
+            return first_value == second_value;
+        }
+        else if (first_value->size() != second_value->size())
+        {
+            return false;
+        }
+        else
+        {
+            return memcmp(first_value->Data(), second_value->Data(), first_value->size()) == 0;
+        }
+    }
+    else
+    {
+        throw unhandled_field_type(field->type()->base_type());
+    }
+}
+
 // The access method for scalar fields and strings.
 data_holder_t get_field_value(
     gaia_type_t type_id,

production/db/payload_types/tests/test_field_access.cpp

@@ -284,6 +284,31 @@ void get_fields_data(
             ASSERT_TRUE(credit_amount.hold.float_value <= c_last_yearly_top_credit_amounts[i] + 1);
         }
     }
+
+    // A few quick equality checks.
+    ASSERT_TRUE(are_field_values_equal(
+        c_type_id,
+        data_loader.get_data(),
+        data_loader.get_data(),
+        pass_schema ? schema_loader.get_data() : nullptr,
+        pass_schema ? schema_loader.get_data_length() : 0,
+        field::last_name));
+
+    ASSERT_TRUE(are_field_values_equal(
+        c_type_id,
+        data_loader.get_data(),
+        data_loader.get_data(),
+        pass_schema ? schema_loader.get_data() : nullptr,
+        pass_schema ? schema_loader.get_data_length() : 0,
+        field::age));
+
+    ASSERT_TRUE(are_field_values_equal(
+        c_type_id,
+        data_loader.get_data(),
+        data_loader.get_data(),
+        pass_schema ? schema_loader.get_data() : nullptr,
+        pass_schema ? schema_loader.get_data_length() : 0,
+        field::known_aliases));
 }
 
 void process_flatbuffers_data(bool access_fields = false)
@@ -538,6 +563,23 @@ void update_flatbuffers_data()
         field::last_yearly_top_credit_amounts,
         c_new_count_credit_amounts);
 
+    // A couple of quick (in)equality checks.
+    ASSERT_FALSE(are_field_values_equal(
+        c_type_id,
+        data_loader.get_data(),
+        serialization.data(),
+        schema_loader.get_data(),
+        schema_loader.get_data_length(),
+        field::last_name));
+
+    ASSERT_FALSE(are_field_values_equal(
+        c_type_id,
+        data_loader.get_data(),
+        serialization.data(),
+        schema_loader.get_data(),
+        schema_loader.get_data_length(),
+        field::known_aliases));
+
     // Write out the final serialization.
     ofstream file;
     file.open(c_updated_bin_filename, ios::binary | ios::out | ios::trunc);