More sort and finalize fixes

cpp/arcticdb/column_store/column.hpp

@@ -277,7 +277,8 @@ class Column {
         return TypedColumnIterator<TagType, const RawType>(*this, false);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(ssize_t row_offset, T val) {
         util::check(sizeof(T) == get_type_size(type_.data_type()), "Type mismatch in set_scalar, expected {}",
                     get_type_size(type_.data_type()));

cpp/arcticdb/column_store/memory_segment.hpp

@@ -74,13 +74,15 @@ class SegmentInMemory {
         impl_->end_row();
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(position_t idx, T val) {
         impl_->set_scalar(idx, val);
     }
 
-    template<class T, std::enable_if_t<std::is_same_v<std::decay_t<T>, std::string>, int> = 0>
-    void set_scalar(position_t idx, T val) {
+    template<typename T>
+    requires std::same_as<std::decay_t<T>, std::string>
+    void set_scalar(position_t idx, const T& val) {
         impl_->set_string(idx, val);
     }
 

cpp/arcticdb/entity/merge_descriptors.cpp

@@ -34,6 +34,7 @@ StreamDescriptor merge_descriptors(
                 merged_fields.emplace_back(idx.name());
                 merged_fields_map.try_emplace(idx.name(), TypeDescriptor{typename IndexType::TypeDescTag{}});
             });
+            index = default_index_type_from_descriptor(*default_index);
         } else {
             util::raise_rte("Descriptor has uninitialized index and no default supplied");
         }
@@ -71,7 +72,12 @@ StreamDescriptor merge_descriptors(
                         if(new_descriptor) {
                             merged_fields_map[field.name()] = *new_descriptor;
                         } else {
-                            util::raise_rte("No valid common type between {} and {} for column {}", existing_type_desc, type_desc, field.name());
+                            schema::raise<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                                "No valid common type between {} and {} for column {}",
+                                existing_type_desc,
+                                type_desc,
+                                field.name()
+                            );
                         }
                     }
                 } else {

cpp/arcticdb/pipeline/read_frame.cpp

@@ -601,7 +601,7 @@ struct ReduceColumnTask : async::BaseTask {
             } else {
                 column.default_initialize_rows(0, frame_.row_count(), false);
             }
-        } else {
+        } else if (column_data != slice_map_->columns_.end()) {
             if(dynamic_schema_) {
                 NullValueReducer null_reducer{column, context_, frame_, shared_data_, handler_data_};
                 for (const auto &row : column_data->second) {

cpp/arcticdb/processing/clause.cpp

@@ -13,6 +13,7 @@
 #include <arcticdb/processing/processing_unit.hpp>
 #include <arcticdb/column_store/string_pool.hpp>
 #include <arcticdb/util/offset_string.hpp>
+#include <arcticdb/stream/merge.hpp>
 
 #include <arcticdb/processing/clause.hpp>
 #include <arcticdb/pipeline/column_stats.hpp>
@@ -888,13 +889,12 @@ std::vector<EntityId> SortClause::process(std::vector<EntityId>&& entity_ids) co
     return push_entities(component_manager_, std::move(proc));
 }
 
-template<typename IndexType, typename DensityPolicy, typename QueueType, typename Comparator, typename StreamId>
+template<typename IndexType, typename DensityPolicy, typename QueueType>
 void merge_impl(
         std::shared_ptr<ComponentManager> component_manager,
         std::vector<std::vector<EntityId>>& ret,
         QueueType &input_streams,
         bool add_symbol_column,
-        StreamId stream_id,
         const RowRange& row_range,
         const ColRange& col_range,
         IndexType index,
@@ -908,19 +908,16 @@ void merge_impl(
     };
 
     using AggregatorType = stream::Aggregator<IndexType, stream::DynamicSchema, SegmentationPolicy, DensityPolicy>;
-    const auto& fields = stream_descriptor.fields();
-    FieldCollection new_fields{};
-    (void)new_fields.add(fields[0].ref());
-
-    auto index_desc = index_descriptor_from_range(stream_id, index, new_fields);
-    auto desc = StreamDescriptor{index_desc};
 
     AggregatorType agg{
-            stream::DynamicSchema{desc, index},
-            std::move(func), std::move(segmentation_policy), desc, std::nullopt
+        stream::DynamicSchema{stream_descriptor, index},
+        std::move(func),
+        std::move(segmentation_policy),
+        stream_descriptor,
+        std::nullopt
     };
 
-    stream::do_merge<IndexType, SegmentWrapper, AggregatorType, decltype(input_streams)>(
+    stream::do_merge<IndexType, AggregatorType, decltype(input_streams)>(
         input_streams, agg, add_symbol_column);
 }
 
@@ -977,15 +974,14 @@ std::optional<std::vector<std::vector<EntityId>>> MergeClause::repartition(std::
     std::vector<std::vector<EntityId>> ret;
     std::visit(
             [this, &ret, &input_streams, &comp=compare, stream_id=stream_id_, &row_range, &col_range](auto idx, auto density) {
-                merge_impl<decltype(idx), decltype(density), decltype(input_streams), decltype(comp), decltype(stream_id)>(component_manager_,
-                                                                                                      ret,
-                                                                                                      input_streams,
-                                                                                                      add_symbol_column_,
-                                                                                                      stream_id,
-                                                                                                      row_range,
-                                                                                                      col_range,
-                                                                                                      idx,
-                                                                                                      stream_descriptor_);
+                merge_impl<decltype(idx), decltype(density), decltype(input_streams)>(component_manager_,
+                                                                                      ret,
+                                                                                      input_streams,
+                                                                                      add_symbol_column_,
+                                                                                      row_range,
+                                                                                      col_range,
+                                                                                      idx,
+                                                                                      stream_descriptor_);
             }, index_, density_policy_);
     return ret;
 }

cpp/arcticdb/processing/clause.hpp

@@ -21,7 +21,6 @@
 #include <arcticdb/processing/grouper.hpp>
 #include <arcticdb/stream/aggregator.hpp>
 #include <arcticdb/util/movable_priority_queue.hpp>
-#include <arcticdb/stream/merge.hpp>
 #include <arcticdb/pipeline/index_utils.hpp>
 
 #include <folly/Poly.h>

cpp/arcticdb/processing/test/test_clause.cpp

@@ -373,6 +373,7 @@ TEST(Clause, Merge) {
 
     auto stream_id = StreamId("Merge");
     StreamDescriptor descriptor{};
+    descriptor.set_index(static_cast<IndexDescriptorImpl>(TimeseriesIndex::default_index()));
     descriptor.add_field(FieldRef{make_scalar_type(DataType::NANOSECONDS_UTC64),"time"});
     MergeClause merge_clause{TimeseriesIndex{"time"}, SparseColumnPolicy{}, stream_id, descriptor};
     merge_clause.set_component_manager(component_manager);

cpp/arcticdb/stream/aggregator.hpp

@@ -275,12 +275,14 @@ class Aggregator {
     void commit_impl(bool final);
 
 private:
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template <typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar(std::size_t pos, T val) {
         segment_.set_scalar(pos, val);
     }
 
-    template<class T, std::enable_if_t<std::is_integral_v<T> || std::is_floating_point_v<T>, int> = 0>
+    template<typename T>
+    requires std::integral<T> || std::floating_point<T>
     void set_scalar_by_name(std::string_view name, T val, DataType data_type) {
         position_t pos = schema_policy_.get_column_idx_by_name(segment_, name, make_scalar_type(data_type), segmenting_policy_.expected_row_size(), segment_.row_count());
         set_scalar(pos, val);

cpp/arcticdb/stream/append_map.cpp

@@ -212,14 +212,9 @@ folly::Future<arcticdb::entity::VariantKey> write_incomplete_frame(
     const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
     const std::shared_ptr<InputTensorFrame>& frame,
-    bool validate_index,
     std::optional<AtomKey>&& next_key)  {
     using namespace arcticdb::pipelines;
 
-    sorting::check<ErrorCode::E_UNSORTED_DATA>(
-            !validate_index || index_is_not_timeseries_or_is_sorted_ascending(*frame),
-            "When writing/appending staged data in parallel, input data must be sorted.");
-
     auto index_range = frame->index_range;
     auto segment = incomplete_segment_from_frame(frame, 0, std::move(next_key), false);
     return store->write(
@@ -234,10 +229,9 @@ folly::Future<arcticdb::entity::VariantKey> write_incomplete_frame(
 void write_parallel(
     const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
-    const std::shared_ptr<InputTensorFrame>& frame,
-    bool validate_index) {
+    const std::shared_ptr<InputTensorFrame>& frame) {
     // TODO: dynamic bucketize doesn't work with incompletes
-    (void)write_incomplete_frame(store, stream_id, frame, validate_index, std::nullopt).get();
+    (void)write_incomplete_frame(store, stream_id, frame, std::nullopt).get();
 }
 
 std::vector<SliceAndKey> get_incomplete(
@@ -376,8 +370,7 @@ AppendMapEntry entry_from_key(const std::shared_ptr<StreamSource>& store, const
 void append_incomplete(
         const std::shared_ptr<Store>& store,
         const StreamId& stream_id,
-        const std::shared_ptr<InputTensorFrame>& frame,
-        bool validate_index) {
+        const std::shared_ptr<InputTensorFrame>& frame) {
     using namespace arcticdb::proto::descriptors;
     using namespace arcticdb::stream;
     ARCTICDB_SAMPLE_DEFAULT(AppendIncomplete)
@@ -387,7 +380,7 @@ void append_incomplete(
     const auto num_rows = frame->num_rows;
     total_rows += num_rows;
     auto desc = frame->desc.clone();
-    auto new_key = write_incomplete_frame(store, stream_id, frame, validate_index, std::move(next_key)).get();
+    auto new_key = write_incomplete_frame(store, stream_id, frame, std::move(next_key)).get();
 
 
     ARCTICDB_DEBUG(log::version(), "Wrote incomplete frame for stream {}, {} rows, total rows {}", stream_id, num_rows, total_rows);

cpp/arcticdb/stream/append_map.hpp

@@ -53,8 +53,7 @@ void remove_incomplete_segments(
 void write_parallel(
     const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
-    const std::shared_ptr<pipelines::InputTensorFrame>& frame,
-    bool validate_index);
+    const std::shared_ptr<pipelines::InputTensorFrame>& frame);
 
 void write_head(
     const std::shared_ptr<Store>& store,
@@ -69,8 +68,7 @@ void append_incomplete_segment(
 void append_incomplete(
     const std::shared_ptr<Store>& store,
     const StreamId& stream_id,
-    const std::shared_ptr<pipelines::InputTensorFrame>& frame,
-    bool validate_index);
+    const std::shared_ptr<pipelines::InputTensorFrame>& frame);
 
 std::optional<int64_t> latest_incomplete_timestamp(
     const std::shared_ptr<Store>& store,

cpp/arcticdb/stream/index.cpp

@@ -69,22 +69,22 @@ void TimeseriesIndex::check(const FieldCollection& fields) const {
 
 IndexValue TimeseriesIndex::start_value_for_segment(const SegmentInMemory& segment) {
     if (segment.row_count() == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto first_ts = segment.template scalar_at<timestamp>(0, 0).value();
     return {first_ts};
 }
 
 IndexValue TimeseriesIndex::end_value_for_segment(const SegmentInMemory& segment) {
     auto row_count = segment.row_count();
     if (row_count == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto last_ts = segment.template scalar_at<timestamp>(row_count - 1, 0).value();
     return {last_ts};
 }
 
 IndexValue TimeseriesIndex::start_value_for_keys_segment(const SegmentInMemory& segment) {
     if (segment.row_count() == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto start_index_id = int(pipelines::index::Fields::start_index);
     auto first_ts = segment.template scalar_at<timestamp>(0, start_index_id).value();
     return {first_ts};
@@ -93,7 +93,7 @@ IndexValue TimeseriesIndex::start_value_for_keys_segment(const SegmentInMemory&
 IndexValue TimeseriesIndex::end_value_for_keys_segment(const SegmentInMemory& segment) {
     auto row_count = segment.row_count();
     if (row_count == 0)
-        return {NumericIndex{0}};
+        return {NumericIndex{min_index_value()}};
     auto end_index_id = int(pipelines::index::Fields::end_index);
     auto last_ts = segment.template scalar_at<timestamp>(row_count - 1, end_index_id).value();
     return {last_ts};

cpp/arcticdb/stream/index.hpp

@@ -52,6 +52,12 @@ class TimeseriesIndex : public BaseIndex<TimeseriesIndex> {
     static constexpr IndexDescriptorImpl::Type type() {
         return IndexDescriptorImpl::Type::TIMESTAMP;
     }
+
+    static constexpr timestamp min_index_value() {
+        // std::numeric_limits<timestamp>::min() is reserved for NaT
+        return std::numeric_limits<timestamp>::min() + 1;
+    }
+
     TimeseriesIndex(const std::string& name);
     static TimeseriesIndex default_index();
     void check(const FieldCollection& fields) const;
@@ -76,7 +82,7 @@ class TimeseriesIndex : public BaseIndex<TimeseriesIndex> {
 
   private:
     std::string name_;
-    timestamp ts_ = 0;
+    timestamp ts_ = min_index_value();
 };
 
 class TableIndex : public BaseIndex<TableIndex> {

cpp/arcticdb/stream/merge.hpp

@@ -7,19 +7,33 @@
 
 #pragma once
 
+#include <type_traits>
 #include <arcticdb/pipeline/index_utils.hpp>
+#include <arcticdb/util/constants.hpp>
+#include <arcticdb/entity/type_utils.hpp>
 
 namespace arcticdb::stream {
-template<typename IndexType, typename WrapperType, typename AggregatorType, typename QueueType>
+template<typename IndexType, typename AggregatorType, typename QueueType>
 void do_merge(
     QueueType& input_streams,
     AggregatorType& agg,
     bool add_symbol_column
     ) {
+    while (!input_streams.empty() && input_streams.top()->seg_.row_count() == 0) {
+        input_streams.pop_top();
+    }
+
+    // NaT is definied as std::numeric_limits<int64_t>::min(), if there are any NaT values they will be on the top of the queue
+    if (!input_streams.empty()) {
+        const auto& next = input_streams.top();
+        const auto index_value =
+            std::get<timestamp>(*pipelines::index::index_value_from_row(next->row(), IndexDescriptorImpl::Type::TIMESTAMP, 0));
+        sorting::check<ErrorCode::E_UNSORTED_DATA>(index_value != NaT, "NaT values are not allowed in the index");
+    }
+
     while (!input_streams.empty()) {
         auto next = input_streams.pop_top();
-
-        if (next->row().parent_->row_count() == 0) {
+        if (next->seg_.row_count() == 0) {
             continue;
         }
         agg.start_row(pipelines::index::index_value_from_row(next->row(), IndexDescriptorImpl::Type::TIMESTAMP, 0).value()) ([&next, add_symbol_column](auto &rb) {
@@ -30,8 +44,33 @@ void do_merge(
             std::advance(val, IndexType::field_count());
             for(; val != next->row().end(); ++val) {
                 val->visit_field([&rb] (const auto& opt_v, std::string_view name, const TypeDescriptor& type_desc) {
-                    if(opt_v)
-                        rb.set_scalar_by_name(name, opt_v.value(), type_desc.data_type());
+                    if (opt_v) {
+                        const StreamDescriptor& descriptor = rb.descriptor();
+                        const std::optional<size_t> field_idx = descriptor.find_field(name);
+                        if (!field_idx || type_desc == descriptor.field(*field_idx).type()) {
+                            rb.set_scalar_by_name(name, *opt_v, type_desc.data_type());
+                        } else {
+                            const auto common_type = has_valid_common_type(type_desc, descriptor.field(*field_idx).type());
+                            schema::check<ErrorCode::E_DESCRIPTOR_MISMATCH>(
+                                common_type,
+                                "No valid common type between staged segments for column {}. Mismatched types are {} "
+                                "and {}",
+                                name,
+                                type_desc,
+                                descriptor.field(*field_idx).type()
+                            );
+                            common_type->visit_tag([&](auto type_desc_tag) {
+                                using RawType = decltype(type_desc_tag)::DataTypeTag::raw_type;
+                                if constexpr (std::is_convertible_v<RawType, std::decay_t<decltype(*opt_v)>>) {
+                                    using RawType = decltype(type_desc_tag)::DataTypeTag::raw_type;
+                                    const auto cast_value = static_cast<RawType>(*opt_v);
+                                    rb.set_scalar_by_name(name, cast_value, type_desc.data_type());
+                                } else {
+                                    rb.set_scalar_by_name(name, *opt_v, type_desc.data_type());
+                                }
+                            });
+                        }
+                    }
                 });
             }
         });

cpp/arcticdb/stream/test/test_append_map.cpp

@@ -23,7 +23,7 @@ TEST(Append, Simple) {
     auto wrapper = get_test_simple_frame(stream_id, 10, 0);
     auto& frame = wrapper.frame_;
     auto desc = frame->desc.clone();
-    append_incomplete(store, stream_id, frame, true);
+    append_incomplete(store, stream_id, frame);
     pipelines::FilterRange range;
     auto pipeline_context = std::make_shared<PipelineContext>(desc);
     pipeline_context->selected_columns_ = util::BitSet(2);