Fix convert int to float

vasil-pashov · vasil-pashov · commit 4fa6bb9c1a99 · 2025-03-12T16:10:23.000+02:00
diff --git a/cpp/arcticdb/entity/merge_descriptors.cpp b/cpp/arcticdb/entity/merge_descriptors.cpp
@@ -11,11 +11,15 @@
 #include <arcticdb/entity/type_utils.hpp>
 
 namespace arcticdb {
+/// @param convert_int_to_float This will convert all integer types (both signed and unsigned) to FLOAT64 without
+///   performing any type checks. This is available only via the V1 Library API and is used by tick collectors. It can
+///   be true only if descriptors are merged during a compact_incomplete call. Otherwise, it must be false.
 StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
     std::span<const std::shared_ptr<FieldCollection>> entries,
-    const std::unordered_set<std::string_view> &filtered_set,
-    const std::optional<IndexDescriptorImpl>& default_index) {
+    const std::unordered_set<std::string_view>& filtered_set,
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float) {
     using namespace arcticdb::stream;
     std::vector<std::string_view> merged_fields;
     std::unordered_map<std::string_view, TypeDescriptor> merged_fields_map;
@@ -55,10 +59,10 @@ StreamDescriptor merge_descriptors(
             );
         }
 
-        for (size_t idx = has_index ? 1u : 0u; idx < static_cast<size_t>(fields->size()); ++idx) {
+        for (size_t idx = has_index ? 1u : 0u; idx < fields->size(); ++idx) {
             const auto& field = fields->at(idx);
             const auto& type_desc = field.type();
-            if (filtered_set.empty() || (filtered_set.find(field.name()) != filtered_set.end())) {
+            if (!filtered_set.contains(field.name())) {
                 if(auto existing = merged_fields_map.find(field.name()); existing != merged_fields_map.end()) {
                     auto existing_type_desc = existing->second;
                     if(existing_type_desc != type_desc) {
@@ -68,17 +72,22 @@ StreamDescriptor merge_descriptors(
                                 "New type descriptor                     : {}",
                                 field.name(), existing_type_desc, type_desc
                         );
-                        auto new_descriptor = has_valid_common_type(existing_type_desc, type_desc);
-                        if(new_descriptor) {
-                            merged_fields_map[field.name()] = *new_descriptor;
+                        if (convert_int_to_float && is_integer_type(existing_type_desc.data_type()) && is_integer_type(type_desc.data_type())) {
+                            merged_fields_map[field.name()] = TypeDescriptor{DataType::FLOAT64, existing_type_desc.dimension()};
                         } else {
-                            schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
-                                "No valid common type between {} and {} for column {}",
-                                existing_type_desc,
-                                type_desc,
-                                field.name()
-                            );
+                            auto new_descriptor = has_valid_common_type(existing_type_desc, type_desc);
+                            if(new_descriptor) {
+                                merged_fields_map[field.name()] = *new_descriptor;
+                            } else {
+                                schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                                    "No valid common type between {} and {} for column {}",
+                                    existing_type_desc,
+                                    type_desc,
+                                    field.name()
+                                );
+                            }
                         }
+
                     }
                 } else {
                     merged_fields.emplace_back(field.name());
@@ -98,46 +107,50 @@ StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
     const std::vector<std::shared_ptr<FieldCollection>> &entries,
     const std::optional<std::vector<std::string>> &filtered_columns,
-    const std::optional<IndexDescriptorImpl>& default_index) {
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float) {
     std::unordered_set<std::string_view> filtered_set = filtered_columns.has_value()
         ? std::unordered_set<std::string_view>(filtered_columns->begin(), filtered_columns->end())
         : std::unordered_set<std::string_view>{};
-    return merge_descriptors(original, entries, filtered_set, default_index);
+    return merge_descriptors(original, entries, filtered_set, default_index, convert_int_to_float);
 }
 
 StreamDescriptor merge_descriptors(
     const StreamDescriptor& original,
     std::span<const std::shared_ptr<FieldCollection>> entries,
     const std::optional<std::vector<std::string>>& filtered_columns,
-    const std::optional<IndexDescriptorImpl>& default_index) {
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float) {
     std::unordered_set<std::string_view> filtered_set = filtered_columns.has_value()
         ? std::unordered_set<std::string_view>(filtered_columns->begin(), filtered_columns->end())
         : std::unordered_set<std::string_view>{};
-    return merge_descriptors(original, entries, filtered_set, default_index);
+    return merge_descriptors(original, entries, filtered_set, default_index, convert_int_to_float);
 }
 
 StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
     const std::optional<std::vector<std::string>> &filtered_columns,
-    const std::optional<IndexDescriptorImpl>& default_index) {
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float) {
     std::vector<std::shared_ptr<FieldCollection>> fields;
     for (const auto &entry : entries) {
         fields.push_back(std::make_shared<FieldCollection>(entry.slice_.desc()->fields().clone()));
     }
-    return merge_descriptors(original, fields, filtered_columns, default_index);
+    return merge_descriptors(original, fields, filtered_columns, default_index, convert_int_to_float);
 }
 
 StreamDescriptor merge_descriptors(
     const std::shared_ptr<Store>& store,
     const StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
     const std::unordered_set<std::string_view> &filtered_set,
-    const std::optional<IndexDescriptorImpl>& default_index) {
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float) {
     std::vector<std::shared_ptr<FieldCollection>> fields;
     for (const auto &entry : entries) {
         fields.push_back(std::make_shared<FieldCollection>(entry.segment(store).descriptor().fields().clone()));
     }
-    return merge_descriptors(original, fields, filtered_set, default_index);
+    return merge_descriptors(original, fields, filtered_set, default_index, convert_int_to_float);
 }
 }
diff --git a/cpp/arcticdb/entity/merge_descriptors.hpp b/cpp/arcticdb/entity/merge_descriptors.hpp
@@ -14,30 +14,35 @@ StreamDescriptor merge_descriptors(
     const StreamDescriptor &original,
     std::span<const std::shared_ptr<FieldCollection>> entries,
     const std::unordered_set<std::string_view> &filtered_set,
-    const std::optional<IndexDescriptorImpl>& default_index);
+    const std::optional<IndexDescriptorImpl>& default_index,
+    bool convert_int_to_float=false);
 
 entity::StreamDescriptor merge_descriptors(
     const entity::StreamDescriptor &original,
     const std::vector<std::shared_ptr<FieldCollection>> &entries,
     const std::optional<std::vector<std::string>> &filtered_columns,
-    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
+    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt,
+    bool convert_int_to_float=false);
 
 entity::StreamDescriptor merge_descriptors(
     const entity::StreamDescriptor& original,
     std::span<const std::shared_ptr<FieldCollection>> entries,
     const std::optional<std::vector<std::string>>& filtered_columns,
-    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
+    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt,
+    bool convert_int_to_float=false);
 
 entity::StreamDescriptor merge_descriptors(
     const entity::StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
     const std::optional<std::vector<std::string>> &filtered_columns,
-    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
+    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt,
+    bool convert_int_to_float=false);
 
 entity::StreamDescriptor merge_descriptors(
     const std::shared_ptr<Store>& store,
     const entity::StreamDescriptor &original,
     const std::vector<pipelines::SliceAndKey> &entries,
     const std::unordered_set<std::string_view> &filtered_set,
-    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt);
+    const std::optional<entity::IndexDescriptorImpl>& default_index = std::nullopt,
+    bool convert_int_to_float=false);
 }
diff --git a/cpp/arcticdb/stream/incompletes.cpp b/cpp/arcticdb/stream/incompletes.cpp
@@ -430,11 +430,9 @@ std::vector<SliceAndKey> get_incomplete(
                             util::raise_rte("Only timestamp based ranges supported for filtering.");
                         },
                         [&entries](const IndexRange &index_range) {
-                            entries.erase(
-                                std::remove_if(std::begin(entries), std::end(entries), [&](const auto &entry) {
-                                    return !intersects(index_range, entry.slice_and_key_.key().index_range());
-                                }),
-                                std::end(entries));
+                            std::erase_if(entries, [&](const auto &entry) {
+                                return !intersects(index_range, entry.slice_and_key_.key().index_range());
+                            });
                         },
                         [](const auto &) {
                             // Don't know what to do with this index
diff --git a/cpp/arcticdb/stream/segment_aggregator.hpp b/cpp/arcticdb/stream/segment_aggregator.hpp
@@ -33,7 +33,7 @@ inline void convert_column_types(SegmentInMemory& segment) {
 }
 
 template<class Index, class Schema, class SegmentingPolicy = RowCountSegmentPolicy, class DensityPolicy = DenseColumnPolicy>
-    class SegmentAggregator : public Aggregator<Index, Schema, SegmentingPolicy, DensityPolicy> {
+class SegmentAggregator : public Aggregator<Index, Schema, SegmentingPolicy, DensityPolicy> {
 public:
     using AggregatorType = Aggregator<Index, Schema, SegmentingPolicy, DensityPolicy>;
     using SliceCallBack = folly::Function<void(pipelines::FrameSlice&&)>;
@@ -49,6 +49,11 @@ template<class Index, class Schema, class SegmentingPolicy = RowCountSegmentPoli
 
     void add_segment(SegmentInMemory&& seg, const pipelines::FrameSlice& slice, bool convert_int_to_float) {
         auto segment = std::move(seg);
+        // Very specific use-case, you probably don't want this. This is applied by design even to static schema. It is
+        // part of an old API that is still used in some tick collectors.
+        if(convert_int_to_float) {
+            convert_column_types(segment);
+        }
         if constexpr (std::is_same_v<Schema, FixedSchema>) {
             if (stream_descriptor_.has_value()) {
                 schema::check<ErrorCode::E_DESCRIPTOR_MISMATCH>(
@@ -60,9 +65,6 @@ template<class Index, class Schema, class SegmentingPolicy = RowCountSegmentPoli
         }
         segment.reset_timeseries_descriptor();
         AggregatorType::stats().update_many(segment.row_count(), segment.num_bytes());
-        //TODO very specific use-case, you probably don't want this
-        if(convert_int_to_float)
-            convert_column_types(segment);
 
         ARCTICDB_DEBUG(log::version(), "Adding segment with descriptor {} uncompressed_bytes {}", segment.descriptor(), segment.descriptor().uncompressed_bytes());
         segments_.push_back(segment);
diff --git a/cpp/arcticdb/version/schema_checks.cpp b/cpp/arcticdb/version/schema_checks.cpp
@@ -96,9 +96,15 @@ bool index_names_match(
     return true;
 }
 
+/// @param convert_int_to_float If this is true it will consider all pairs of integer types (both signed and unsigned)
+///   as identical. If a field in df_in_store_descriptor is FLOAT64 and the corresponding field in new_df_descriptor
+///   is of any integer type they will be considered identical. Note that this makes the function unsymmetrical. If a
+///   field in new_df_descriptor is FLOAT64 and the corresponding field in df_in_store_descriptor is of integer type
+///   the types won't be considered identical. This is supposed to be used only from compact_incomplete.B
 bool columns_match(
     const StreamDescriptor& df_in_store_descriptor,
-    const StreamDescriptor& new_df_descriptor
+    const StreamDescriptor& new_df_descriptor,
+    const bool convert_int_to_float
 ) {
     const int index_field_size =
         df_in_store_descriptor.index().type() == IndexDescriptor::Type::EMPTY ? new_df_descriptor.index().field_count() : 0;
@@ -117,8 +123,17 @@ bool columns_match(
         const TypeDescriptor& right_type = new_df_descriptor.fields(i + index_field_size).type();
 
         if (!trivially_compatible_types(left_type, right_type) &&
-            !(is_empty_type(left_type.data_type()) || is_empty_type(right_type.data_type())))
-            return false;
+            !(is_empty_type(left_type.data_type()) || is_empty_type(right_type.data_type()))) {
+            if (convert_int_to_float) {
+                const bool both_are_int = is_integer_type(left_type.data_type()) && is_integer_type(right_type.data_type());
+                if (!(both_are_int || (left_type.data_type() == DataType::FLOAT64 && is_integer_type(right_type.data_type())))) {
+                    return false;
+                }
+            } else {
+                return false;
+            }
+        }
+
     }
     return true;
 }
diff --git a/cpp/arcticdb/version/schema_checks.hpp b/cpp/arcticdb/version/schema_checks.hpp
@@ -39,7 +39,8 @@ bool index_names_match(
 
 bool columns_match(
     const StreamDescriptor& df_in_store_descriptor,
-    const StreamDescriptor& new_df_descriptor
+    const StreamDescriptor& new_df_descriptor,
+    const bool convert_int_to_float=false
 );
 
 void fix_descriptor_mismatch_or_throw(
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
diff --git a/cpp/arcticdb/version/version_core.hpp b/cpp/arcticdb/version/version_core.hpp
diff --git a/python/tests/unit/arcticdb/version_store/test_parallel.py b/python/tests/unit/arcticdb/version_store/test_parallel.py
