Repository: man-group/ArcticDB (language: python)

Issue:
Inconsistent column ordering with update and append on dynamic schema
### Describe the bug

When adding new columns to a dynamic schema with update, the new columns are added at the end.
When adding new columns with append, the new columns are added at the front.

### Steps/Code to Reproduce

```
lib = ac.create_library("test", library_options=LibraryOptions(dynamic_schema=True))

df_a = pd.DataFrame(data = {"a": [1, 2, 3]}, index=pd.date_range("2018-01-01", periods=3))
df_b = pd.DataFrame(data = {"b": [4, 5, 6]}, index=pd.date_range("2018-01-04", periods=3))

lib.write("symbol_append", df_a)
lib.append("symbol_append", df_b)

lib.write("symbol_update", df_a)
lib.update("symbol_update", df_b, upsert=True)

lib.read("symbol_append").data.columns
# Index(['b', 'a'], dtype='object')
lib.read("symbol_update").data.columns
# Index(['a', 'b'], dtype='object')
```

### Expected Results

Appending also adds new columns at the end.

### OS, Python Version and ArcticDB Version

Python 3.8

### Backend storage used

_No response_

### Additional Context

_No response_


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/pipeline/index_utils.cpp b/cpp/arcticdb/pipeline/index_utils.cpp
index 0dea1f51d4..6ea4656559 100644
--- a/cpp/arcticdb/pipeline/index_utils.cpp
+++ b/cpp/arcticdb/pipeline/index_utils.cpp
@@ -9,6 +9,8 @@
 #include <arcticdb/storage/store.hpp>
 #include <arcticdb/pipeline/index_writer.hpp>
 #include <arcticdb/pipeline/frame_utils.hpp>
+#include <arcticdb/version/version_utils.hpp>
+#include <arcticdb/entity/merge_descriptors.hpp>
 
 namespace arcticdb::pipelines::index {
 
@@ -77,4 +79,53 @@ std::pair<index::IndexSegmentReader, std::vector<SliceAndKey>> read_index_to_vec
     return {std::move(index_segment_reader), std::move(slice_and_keys)};
 }
 
+TimeseriesDescriptor get_merged_tsd(
+        int row_count,
+        bool dynamic_schema,
+        const TimeseriesDescriptor& existing_tsd,
+        const std::shared_ptr<pipelines::InputTensorFrame>& new_frame) {
+    auto existing_descriptor = existing_tsd.as_stream_descriptor();
+    auto merged_descriptor = existing_descriptor;
+    if (existing_tsd.proto().total_rows() == 0){
+        // If the existing dataframe is empty, we use the descriptor of the new_frame
+        merged_descriptor = new_frame->desc;
+    }
+    else if (dynamic_schema) {
+        // In case of dynamic schema
+        merged_descriptor = merge_descriptors(
+                existing_descriptor,
+                std::vector<std::shared_ptr<FieldCollection>>{new_frame->desc.fields_ptr()},
+                {}
+        );
+    } else {
+        // In case of static schema, we only promote empty types and fixed->dynamic strings
+        const auto &new_fields = new_frame->desc.fields();
+        for (size_t i = 0; i < new_fields.size(); ++i) {
+            const auto &new_type = new_fields.at(i).type();
+            TypeDescriptor &result_type = merged_descriptor.mutable_field(i).mutable_type();
+            // We allow promoting empty types
+            if (is_empty_type(result_type.data_type()) && !is_empty_type(new_type.data_type())) {
+                result_type = new_type;
+            }
+            // We allow promoting fixed strings to dynamic strings
+            else if (is_sequence_type(result_type.data_type()) &&
+                     is_sequence_type(new_type.data_type()) &&
+                     !is_dynamic_string_type(result_type.data_type()) &&
+                     is_dynamic_string_type(new_type.data_type())) {
+                result_type = new_type;
+            }
+        }
+    }
+    merged_descriptor.set_sorted(deduce_sorted(existing_descriptor.get_sorted(), new_frame->desc.get_sorted()));
+    return make_timeseries_descriptor(
+            row_count,
+            std::move(merged_descriptor),
+            std::move(new_frame->norm_meta),
+            std::move(new_frame->user_meta),
+            std::nullopt,
+            std::nullopt,
+            new_frame->bucketize_dynamic
+    );
+}
+
 } //namespace arcticdb::pipelines::index
\ No newline at end of file
diff --git a/cpp/arcticdb/pipeline/index_utils.hpp b/cpp/arcticdb/pipeline/index_utils.hpp
index a0b7f03f68..97dba5de9f 100644
--- a/cpp/arcticdb/pipeline/index_utils.hpp
+++ b/cpp/arcticdb/pipeline/index_utils.hpp
@@ -136,4 +136,12 @@ std::pair<index::IndexSegmentReader, std::vector<SliceAndKey>> read_index_to_vec
     const std::shared_ptr<Store>& store,
     const AtomKey& index_key);
 
+// Combines the stream descriptors of an existing index key and a new frame.
+// Can be used to get the metadata for [write_index] when updating or appending.
+TimeseriesDescriptor get_merged_tsd(
+    int row_count,
+    bool dynamic_schema,
+    const TimeseriesDescriptor& existing_tsd,
+    const std::shared_ptr<pipelines::InputTensorFrame>& new_frame);
+
 } //namespace arcticdb::pipelines::index
\ No newline at end of file
diff --git a/cpp/arcticdb/pipeline/write_frame.cpp b/cpp/arcticdb/pipeline/write_frame.cpp
index af689e9788..12be00d48a 100644
--- a/cpp/arcticdb/pipeline/write_frame.cpp
+++ b/cpp/arcticdb/pipeline/write_frame.cpp
@@ -247,35 +247,8 @@ folly::Future<entity::AtomKey> append_frame(
     .thenValue([dynamic_schema, slices_to_write=std::move(existing_slices), frame=frame, index_segment_reader=std::move(index_segment_reader), key=std::move(key), store](auto&& slice_and_keys_to_append) mutable {
         slices_to_write.insert(std::end(slices_to_write), std::make_move_iterator(std::begin(slice_and_keys_to_append)), std::make_move_iterator(std::end(slice_and_keys_to_append)));
         std::sort(std::begin(slices_to_write), std::end(slices_to_write));
-        if(dynamic_schema) {
-            auto merged_descriptor = merge_descriptors(
-                frame->desc,
-                std::vector<std::shared_ptr<FieldCollection>>{index_segment_reader.tsd().fields_ptr()},
-                {}
-            );
-            merged_descriptor.set_sorted(deduce_sorted(index_segment_reader.get_sorted(), frame->desc.get_sorted()));
-            auto tsd = make_timeseries_descriptor(
-                frame->num_rows + frame->offset,
-                std::move(merged_descriptor),
-                std::move(frame->norm_meta),
-                std::move(frame->user_meta),
-                std::nullopt,
-                std::nullopt,
-                frame->bucketize_dynamic
-            );
-            return index::write_index(stream::index_type_from_descriptor(frame->desc), std::move(tsd), std::move(slices_to_write), key, store);
-        } else {
-            const FieldCollection& new_fields{index_segment_reader.tsd().fields()};
-            for (size_t i = 0; i < new_fields.size(); ++i) {
-                const Field& new_field = new_fields.at(i);
-                TypeDescriptor& original_type = frame->desc.mutable_field(i).mutable_type();
-                if (is_empty_type(original_type.data_type()) && !is_empty_type(new_field.type().data_type())) {
-                    original_type = new_field.type();
-                }
-            }
-            frame->desc.set_sorted(deduce_sorted(index_segment_reader.get_sorted(), frame->desc.get_sorted()));
-            return index::write_index(frame, std::move(slices_to_write), key, store);
-        }
+        auto tsd = index::get_merged_tsd(frame->num_rows + frame->offset, dynamic_schema, index_segment_reader.tsd(), frame);
+        return index::write_index(stream::index_type_from_descriptor(tsd.as_stream_descriptor()), std::move(tsd), std::move(slices_to_write), key, store);
     });
 }
 
diff --git a/cpp/arcticdb/version/version_core.cpp b/cpp/arcticdb/version/version_core.cpp
index a575ed9ca5..8d3a71cded 100644
--- a/cpp/arcticdb/version/version_core.cpp
+++ b/cpp/arcticdb/version/version_core.cpp
@@ -333,7 +333,6 @@ VersionedItem update_impl(
     sorted_data_check_update(*frame, index_segment_reader);
     bool bucketize_dynamic = index_segment_reader.bucketize_dynamic();
     (void)check_and_mark_slices(index_segment_reader, dynamic_schema, false, std::nullopt, bucketize_dynamic);
-    auto combined_sorting_info = deduce_sorted(frame->desc.get_sorted(), index_segment_reader.get_sorted());
     fix_descriptor_mismatch_or_throw(UPDATE, dynamic_schema, index_segment_reader, *frame);
 
     std::vector<FilterQuery<index::IndexSegmentReader>> queries =
@@ -394,16 +393,8 @@ VersionedItem update_impl(
                 unaffected_keys.size(), new_keys_size, affected_keys.size(), flattened_slice_and_keys.size());
 
     std::sort(std::begin(flattened_slice_and_keys), std::end(flattened_slice_and_keys));
-    auto existing_desc = index_segment_reader.tsd().as_stream_descriptor();
-    auto desc = merge_descriptors(existing_desc, std::vector<std::shared_ptr<FieldCollection>>{ frame->desc.fields_ptr() }, {});
-    desc.set_sorted(combined_sorting_info);
-    auto time_series = make_timeseries_descriptor(row_count, std::move(desc), std::move(frame->norm_meta), std::move(frame->user_meta), std::nullopt, std::nullopt, bucketize_dynamic);
-    auto index = index_type_from_descriptor(time_series.as_stream_descriptor());
-
-    auto version_key_fut = util::variant_match(index, [&time_series, &flattened_slice_and_keys, &stream_id, &update_info, &store] (auto idx) {
-        using IndexType = decltype(idx);
-        return index::write_index<IndexType>(std::move(time_series), std::move(flattened_slice_and_keys), IndexPartialKey{stream_id, update_info.next_version_id_}, store);
-    });
+    auto tsd = index::get_merged_tsd(row_count, dynamic_schema, index_segment_reader.tsd(), frame);
+    auto version_key_fut = index::write_index(stream::index_type_from_descriptor(tsd.as_stream_descriptor()), std::move(tsd), std::move(flattened_slice_and_keys), IndexPartialKey{stream_id, update_info.next_version_id_}, store);
     auto version_key = std::move(version_key_fut).get();
     auto versioned_item = VersionedItem(to_atom(std::move(version_key)));
     ARCTICDB_DEBUG(log::version(), "updated stream_id: {} , version_id: {}", stream_id, update_info.next_version_id_);


Test patch:
diff --git a/python/tests/unit/arcticdb/version_store/test_append.py b/python/tests/unit/arcticdb/version_store/test_append.py
index b24ff49406..b1aaf988a2 100644
--- a/python/tests/unit/arcticdb/version_store/test_append.py
+++ b/python/tests/unit/arcticdb/version_store/test_append.py
@@ -60,6 +60,20 @@ def test_append_string_of_different_sizes(lmdb_version_store):
     assert_frame_equal(vit.data, expected)
 
 
+def test_append_dynamic_schema_add_column(lmdb_version_store_dynamic_schema):
+    symbol = "symbol"
+    lib = lmdb_version_store_dynamic_schema
+    df_1 = pd.DataFrame(data={"a": [1.0, 2.0]}, index=pd.date_range("2018-01-01", periods=2))
+    df_2 = pd.DataFrame(data={"b": [3.0, 4.0]}, index=pd.date_range("2018-01-03", periods=2))
+
+    lib.write(symbol, df_1)
+    lib.append(symbol, df_2)
+
+    expected_df = pd.concat([df_1, df_2])
+    result_df = lib.read(symbol).data
+    assert_frame_equal(result_df, expected_df)
+
+
 def test_append_snapshot_delete(lmdb_version_store):
     symbol = "test_append_snapshot_delete"
     if sys.platform == "win32":
diff --git a/python/tests/unit/arcticdb/version_store/test_column_type_changes.py b/python/tests/unit/arcticdb/version_store/test_column_type_changes.py
index 1d7051787b..76711b53c0 100644
--- a/python/tests/unit/arcticdb/version_store/test_column_type_changes.py
+++ b/python/tests/unit/arcticdb/version_store/test_column_type_changes.py
@@ -43,18 +43,8 @@ def test_changing_numeric_type(version_store_factory, dynamic_schema):
         assert_frame_equal(expected_update, received_update)
 
 
-@pytest.mark.parametrize("dynamic_schema, dynamic_strings_first", [
-    (True, True),
-    (True, False),
-    pytest.param(False,
-                 True,
-                 marks=pytest.mark.xfail(
-                     reason="""Issue with appending/updating a dynamic string column with fixed-width strings
-                     https://github.com/man-group/ArcticDB/issues/1204"""
-                 )
-                 ),
-    (False, False),
-])
+@pytest.mark.parametrize("dynamic_schema", [True, False])
+@pytest.mark.parametrize("dynamic_strings_first", [True, False])
 def test_changing_string_type(version_store_factory, dynamic_schema, dynamic_strings_first):
     lib = version_store_factory(dynamic_strings=True, dynamic_schema=dynamic_schema)
     sym_append = "test_changing_string_type_append"
diff --git a/python/tests/unit/arcticdb/version_store/test_empty_writes.py b/python/tests/unit/arcticdb/version_store/test_empty_writes.py
index 0d01201542..67bb439fbe 100644
--- a/python/tests/unit/arcticdb/version_store/test_empty_writes.py
+++ b/python/tests/unit/arcticdb/version_store/test_empty_writes.py
@@ -146,19 +146,34 @@ def test_empty_series(lmdb_version_store_dynamic_schema, sym):
     assert_series_equal(lmdb_version_store_dynamic_schema.read(sym).data, ser, check_index_type=False)
 
 
-@pytest.mark.parametrize("dtype", ["int64", "float64"])
-def test_append_empty_series(lmdb_version_store_dynamic_schema, sym, dtype):
-    ser = pd.Series([])
+@pytest.mark.parametrize("dtype, existing_empty, append_empty", [
+    ("int64", True, False),
+    ("float64", True, False),
+    ("float64", False, True),
+    ("float64", True, True),
+])
+def test_append_empty_series(lmdb_version_store_dynamic_schema, sym, dtype, existing_empty, append_empty):
+    empty_ser = pd.Series([])
+    non_empty_ser = pd.Series([1, 2, 3], dtype=dtype)
+    if existing_empty:
+        ser = empty_ser
+    else:
+        ser = non_empty_ser
     lmdb_version_store_dynamic_schema.write(sym, ser)
     assert not lmdb_version_store_dynamic_schema.is_symbol_pickled(sym)
 
     # ArcticDB stores empty columns under a dedicated `EMPTYVAL` type, so the types are not going to match with pandas
-    # until the first append.
-    assert_series_equal(lmdb_version_store_dynamic_schema.read(sym).data, ser, check_index_type=False)
+    # if the series is empty.
+    assert_series_equal(lmdb_version_store_dynamic_schema.read(sym).data, ser, check_index_type=(len(ser) > 0))
 
-    new_ser = pd.Series([1, 2, 3], dtype=dtype)
+    if append_empty:
+        new_ser = empty_ser
+    else:
+        new_ser = non_empty_ser
     lmdb_version_store_dynamic_schema.append(sym, new_ser)
-    assert_series_equal(lmdb_version_store_dynamic_schema.read(sym).data, new_ser)
+
+    result_ser = pd.concat([ser, new_ser])
+    assert_series_equal(lmdb_version_store_dynamic_schema.read(sym).data, result_ser, check_index_type=(len(result_ser) > 0))
 
 
 def test_entirely_empty_column(lmdb_version_store):