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

Issue:
Add C++ support for None type
**Is your feature request related to a problem? Please describe.**
In order to complete #540 we would need to change the C++ to support a special undefined type that can be changed into any other type in a subsequent data key. See also the discussion in #240.

**Describe the solution you'd like**
Add a none type in the C++ layer

**Describe alternatives you've considered**

**Additional context**



Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/CMakeLists.txt b/cpp/arcticdb/CMakeLists.txt
index c9b1fa95cb..2bf3bb06ca 100644
--- a/cpp/arcticdb/CMakeLists.txt
+++ b/cpp/arcticdb/CMakeLists.txt
@@ -634,6 +634,8 @@ pybind11_add_module(arcticdb_ext MODULE
         python/arctic_version.hpp
         python/reader.hpp
         python/adapt_read_dataframe.hpp
+        python/python_handlers.hpp
+        python/python_handlers.cpp
         )
 
 set (additional_link_flags "")
diff --git a/cpp/arcticdb/async/async_store.cpp b/cpp/arcticdb/async/async_store.cpp
index f47f85453b..771d601ecb 100644
--- a/cpp/arcticdb/async/async_store.cpp
+++ b/cpp/arcticdb/async/async_store.cpp
@@ -29,4 +29,4 @@ std::pair<entity::VariantKey, std::optional<Segment>> lookup_match_in_dedup_map(
         return std::make_pair(de_dup_key.value(), std::nullopt);
     }
 }
-}
\ No newline at end of file
+}
diff --git a/cpp/arcticdb/codec/codec.cpp b/cpp/arcticdb/codec/codec.cpp
index 900d6b218b..012e5f0b8e 100644
--- a/cpp/arcticdb/codec/codec.cpp
+++ b/cpp/arcticdb/codec/codec.cpp
@@ -30,8 +30,13 @@ std::pair<size_t, size_t> ColumnEncoder::max_compressed_size(
 
         while (auto block = column_data.next<TDT>()) {
             const auto nbytes = block.value().nbytes();
-            util::check(nbytes, "Zero-sized block");
-            uncompressed_bytes += nbytes;
+            if constexpr(!is_empty_type(TDT::DataTypeTag::data_type)) {
+                util::check(nbytes > 0, "Zero-sized block");
+                uncompressed_bytes += nbytes;
+            }
+            // For the empty type the column will contain 0 size of user data however the encoder might need add some
+            // encoder specific data to the buffer, thus the uncompressed size will be 0 but the max_compressed_bytes
+            // might be non-zero.
             max_compressed_bytes += Encoder::max_compressed_size(codec_opts, block.value());
         }
 
diff --git a/cpp/arcticdb/codec/codec.hpp b/cpp/arcticdb/codec/codec.hpp
index fafbb85ce4..51b18a8b91 100644
--- a/cpp/arcticdb/codec/codec.hpp
+++ b/cpp/arcticdb/codec/codec.hpp
@@ -86,7 +86,9 @@ struct ColumnEncoder {
             ARCTICDB_TRACE(log::codec(), "Column data has {} blocks", column_data.num_blocks());
 
             while (auto block = column_data.next<TDT>()) {
-                util::check(block.value().nbytes(), "Zero-sized block");
+                if constexpr(!is_empty_type(TDT::DataTypeTag::data_type)) {
+                    util::check(block.value().nbytes() > 0, "Zero-sized block");
+                }
                 Encoder::encode(codec_opts, block.value(), field, out, pos);
             }
         });
diff --git a/cpp/arcticdb/column_store/column_utils.hpp b/cpp/arcticdb/column_store/column_utils.hpp
index f91e81f18e..cadf95a0a3 100644
--- a/cpp/arcticdb/column_store/column_utils.hpp
+++ b/cpp/arcticdb/column_store/column_utils.hpp
@@ -22,17 +22,14 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
             using TypeTag = std::decay_t<decltype(tag)>;
             constexpr auto data_type = TypeTag::DataTypeTag::data_type;
             std::string dtype;
-            py::array::ShapeContainer shapes{0};
-            ssize_t esize = get_type_size(data_type);
+            constexpr ssize_t esize = is_sequence_type(data_type) && is_fixed_string_type(data_type) ? 1 : get_type_size(data_type);
             if constexpr (is_sequence_type(data_type)) {
-                if(is_fixed_string_type(data_type)) {
+                if constexpr (is_fixed_string_type(data_type)) {
                     dtype = data_type == DataType::ASCII_FIXED64 ? "<S0" : "<U0";
-                    esize = 1;
-                }
-                else {
+                } else {
                     dtype = "O";
                 }
-            } else {
+            } else if constexpr(is_numeric_type(data_type) || is_bool_type(data_type)) {
                 constexpr auto dim = TypeTag::DimensionTag::value;
                 util::check(dim == Dimension::Dim0, "Only scalars supported, {}", data_type);
                 if constexpr (data_type == DataType::NANOSECONDS_UTC64) {
@@ -47,34 +44,35 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
                 } else {
                     dtype = fmt::format("{}{:d}", get_dtype_specifier(data_type), esize);
                 }
+            } else if constexpr (is_empty_type(data_type)) {
+                dtype= "O";
+            } else {
+                static_assert(!sizeof(data_type), "Unhandled data type");
             }
-            py::array::StridesContainer strides{esize};
-            return py::array{py::dtype{dtype}, std::move(shapes), std::move(strides)};
+            return py::array{py::dtype{dtype}, py::array::ShapeContainer{0}, py::array::StridesContainer{esize}};
         });
     }
     return visit_field(frame.field(col_pos), [&, frame=frame, col_pos=col_pos] (auto &&tag) {
         using TypeTag = std::decay_t<decltype(tag)>;
-        constexpr auto dt = TypeTag::DataTypeTag::data_type;
-        auto &buffer = frame.column(col_pos).data().buffer();
+        constexpr auto data_type = TypeTag::DataTypeTag::data_type;
+        const auto &buffer = frame.column(col_pos).data().buffer();
         std::string dtype;
-        std::vector<shape_t> shapes;
-        shapes.push_back(frame.row_count());
-        ssize_t esize = get_type_size(dt);
-        if constexpr (is_sequence_type(dt)) {
-            if (is_fixed_string_type(dt)) {
+        ssize_t esize = get_type_size(data_type);
+        if constexpr (is_sequence_type(data_type)) {
+            if (is_fixed_string_type(data_type)) {
                 esize = buffer.bytes() / frame.row_count();
                 auto char_count = esize;
-                if (dt == DataType::UTF_FIXED64) {
+                if (data_type == DataType::UTF_FIXED64) {
                     char_count /= UNICODE_WIDTH;
                 }
-                dtype = fmt::format((dt == DataType::ASCII_FIXED64 ? "<S{:d}" : "<U{:d}"), char_count);
+                dtype = fmt::format((data_type == DataType::ASCII_FIXED64 ? "<S{:d}" : "<U{:d}"), char_count);
             } else {
                 dtype = "O";
             }
-        } else {
+        } else if constexpr(is_numeric_type(data_type) || is_bool_type(data_type)) {
             constexpr auto dim = TypeTag::DimensionTag::value;
             util::check(dim == Dimension::Dim0, "Only scalars supported, {}", frame.field(col_pos));
-            if constexpr (dt == DataType::NANOSECONDS_UTC64) {
+            if constexpr (data_type == DataType::NANOSECONDS_UTC64) {
                 // NOTE: this is safe as of Pandas < 2.0 because `datetime64` _always_ has been using nanosecond resolution,
                 // i.e. Pandas < 2.0 _always_ provides `datetime64[ns]` and ignores any other resolution.
                 // Yet, this has changed in Pandas 2.0 and other resolution can be used,
@@ -84,15 +82,17 @@ inline py::array array_at(const SegmentInMemory& frame, std::size_t col_pos, py:
                 // rely uniquely on the resolution-less 'M' specifier if it this doable.
                 dtype = "datetime64[ns]";
             } else {
-                dtype = fmt::format("{}{:d}", get_dtype_specifier(dt), esize);
+                dtype = fmt::format("{}{:d}", get_dtype_specifier(data_type), esize);
             }
+        } else if constexpr (is_empty_type(data_type)) {
+            dtype = "O";
+        } else {
+            static_assert(!sizeof(data_type), "Unhandled data type");
         }
-        std::vector<stride_t> strides;
-        strides.push_back(esize);
         // Note how base is passed to the array to register the data owner.
         // It's especially important to keep the frame data object alive for as long as the array is alive
         // so that regular python ref counting logic handles the liveness
-        return py::array(py::dtype{dtype}, std::move(shapes), std::move(strides), buffer.data(), anchor);
+        return py::array(py::dtype{dtype}, {frame.row_count()}, {esize}, buffer.data(), anchor);
     });
 }
 
diff --git a/cpp/arcticdb/entity/type_utils.hpp b/cpp/arcticdb/entity/type_utils.hpp
index d8c51ee37f..a702962c08 100644
--- a/cpp/arcticdb/entity/type_utils.hpp
+++ b/cpp/arcticdb/entity/type_utils.hpp
@@ -34,6 +34,16 @@ inline std::optional<entity::TypeDescriptor> has_valid_type_promotion(entity::Ty
     if (source == target)
         return target;
 
+    // Empty type is coercible to any type
+    if(is_empty_type(source.data_type())) {
+        return target;
+    }
+
+    // Nothing is coercible to the empty type.
+    if(is_empty_type(target.data_type())) {
+        return std::nullopt;
+    }
+
     auto source_type = source.data_type();
     auto target_type = target.data_type();
     auto source_size = slice_bit_size(source_type);
diff --git a/cpp/arcticdb/entity/types-inl.hpp b/cpp/arcticdb/entity/types-inl.hpp
index b9cd969b77..06ab5925e6 100644
--- a/cpp/arcticdb/entity/types-inl.hpp
+++ b/cpp/arcticdb/entity/types-inl.hpp
@@ -38,6 +38,7 @@ auto visit_dim(DataType dt, Callable &&c) {
         DT_CASE(ASCII_DYNAMIC64)
         DT_CASE(UTF_FIXED64)
         DT_CASE(UTF_DYNAMIC64)
+        DT_CASE(EMPTYVAL)
 #undef DT_CASE
     default: util::raise_rte("Invalid dtype '{}' in visit dim", datatype_to_str(dt));
     }
@@ -64,6 +65,7 @@ auto visit_type(DataType dt, Callable &&c) {
         DT_CASE(ASCII_DYNAMIC64)
         DT_CASE(UTF_FIXED64)
         DT_CASE(UTF_DYNAMIC64)
+        DT_CASE(EMPTYVAL)
 #undef DT_CASE
     default: util::raise_rte("Invalid dtype '{}' in visit type", datatype_to_str(dt));
     }
diff --git a/cpp/arcticdb/entity/types.cpp b/cpp/arcticdb/entity/types.cpp
index e6f149ab52..5683d0ca20 100644
--- a/cpp/arcticdb/entity/types.cpp
+++ b/cpp/arcticdb/entity/types.cpp
@@ -35,6 +35,7 @@ std::string_view datatype_to_str(const DataType dt) {
         TO_STR(ASCII_DYNAMIC64)
         TO_STR(UTF_FIXED64)
         TO_STR(UTF_DYNAMIC64)
+        TO_STR(EMPTYVAL)
         //    TO_STR(UTF8_STRING)
 //     TO_STR(BYTES)
         //    TO_STR(PICKLE)
diff --git a/cpp/arcticdb/entity/types.hpp b/cpp/arcticdb/entity/types.hpp
index b2a3ca45bc..aaceb12d0c 100644
--- a/cpp/arcticdb/entity/types.hpp
+++ b/cpp/arcticdb/entity/types.hpp
@@ -115,6 +115,9 @@ enum class ValueType : uint8_t {
 
     UTF_DYNAMIC = 11,
     ASCII_DYNAMIC = 12,
+    /// Used to represent null types. Each type can be converted to Empty and Empty can be converted to each type.
+    EMPTY = 13,
+    COUNT // Not a real value type, should not be added to proto descriptor. Used to count the number of items in the enum
 };
 
 // Sequence types are composed of more than one element
@@ -152,12 +155,17 @@ constexpr bool is_utf_type(ValueType v) {
     return v == ValueType::UTF8_FIXED || v == ValueType::UTF_DYNAMIC;
 }
 
+constexpr bool is_empty_type(ValueType v) {
+    return v == ValueType::EMPTY;
+}
+
 enum class SizeBits : uint8_t {
     UNKNOWN_SIZE_BITS = 0,
     S8 = 1,
     S16 = 2,
     S32 = 3,
     S64 = 4,
+    COUNT
 };
 
 constexpr SizeBits get_size_bits(uint8_t size) {
@@ -196,6 +204,7 @@ enum class DataType : uint8_t {
     UTF_FIXED64 = detail::combine_val_bits(ValueType::UTF8_FIXED, SizeBits::S64),
     UTF_DYNAMIC64 = detail::combine_val_bits(ValueType::UTF_DYNAMIC, SizeBits::S64),
     BYTES_DYNAMIC64 = detail::combine_val_bits(ValueType::BYTES, SizeBits::S64),
+    EMPTYVAL = detail::combine_val_bits(ValueType::EMPTY, SizeBits::S64),
 #undef DT_COMBINE
     UNKNOWN = 0,
 };
@@ -282,6 +291,10 @@ constexpr bool is_utf_type(DataType v){
     return is_utf_type(slice_value_type(v));
 }
 
+constexpr bool is_empty_type(DataType v){
+    return is_empty_type(slice_value_type(v));
+}
+
 static_assert(slice_value_type((DataType::UINT16)) == ValueType(1));
 static_assert(get_type_size(DataType::UINT32) == 4);
 static_assert(get_type_size(DataType::UINT64) == 8);
@@ -371,6 +384,7 @@ DATA_TYPE_TAG(ASCII_FIXED64, std::uint64_t)
 DATA_TYPE_TAG(ASCII_DYNAMIC64, std::uint64_t)
 DATA_TYPE_TAG(UTF_FIXED64, std::uint64_t)
 DATA_TYPE_TAG(UTF_DYNAMIC64, std::uint64_t)
+DATA_TYPE_TAG(EMPTYVAL, std::uint64_t)
 #undef DATA_TYPE_TAG
 
 enum class Dimension : uint8_t {
diff --git a/cpp/arcticdb/pipeline/frame_utils.hpp b/cpp/arcticdb/pipeline/frame_utils.hpp
index 073dada9b9..0ae859b194 100644
--- a/cpp/arcticdb/pipeline/frame_utils.hpp
+++ b/cpp/arcticdb/pipeline/frame_utils.hpp
@@ -133,23 +133,20 @@ void aggregator_set_data(
                 for (size_t s = 0; s < rows_to_write; ++s, ++ptr_data) {
                     if (*ptr_data == none.ptr()) {
                         agg.set_no_string_at(col, s, not_a_string());
-                    }
-                    else if(is_py_nan(*ptr_data)){
+                    } else if(is_py_nan(*ptr_data)){
                         agg.set_no_string_at(col, s, nan_placeholder());
-                    }
-                    else {
+                    } else {
                         if constexpr (is_utf_type(slice_value_type(dt))) {
                             auto wrapper= convert::py_unicode_to_buffer(*ptr_data);
                             agg.set_string_at(col, s, wrapper.buffer_, wrapper.length_);
-                        }
-                        else {
+                        } else {
                             auto wrapper = convert::pystring_to_buffer(*ptr_data);
                             agg.set_string_at(col, s, wrapper.buffer_, wrapper.length_);
                         }
                     }
                 }
             }
-        } else {
+        } else if constexpr (is_numeric_type(dt) || is_bool_type(dt)) {
             auto ptr = tensor.template ptr_cast<RawType>(row);
             if (sparsify_floats) {
                 if constexpr (is_floating_point_type(dt)) {
@@ -172,6 +169,8 @@ void aggregator_set_data(
                     agg.set_array(col, t);
                 }
             }
+        }  else if constexpr (!is_empty_type(dt)) {
+            static_assert(!sizeof(dt), "Unknown data type");
         }
     });
 }
@@ -202,4 +201,4 @@ size_t get_slice_rowcounts(
 std::pair<size_t, size_t> offset_and_row_count(
     const std::shared_ptr<pipelines::PipelineContext>& context);
 
-} //namespace arcticdb
\ No newline at end of file
+} //namespace arcticdb
diff --git a/cpp/arcticdb/pipeline/read_frame.cpp b/cpp/arcticdb/pipeline/read_frame.cpp
index 327fabfec0..b2bf926e02 100644
--- a/cpp/arcticdb/pipeline/read_frame.cpp
+++ b/cpp/arcticdb/pipeline/read_frame.cpp
@@ -293,6 +293,18 @@ void advance_skipped_cols(
     }
 }
 
+template<typename IteratorType>
+bool remaining_fields_empty(IteratorType it, const PipelineContextRow& context) {
+    while(it.has_next()) {
+        const StreamDescriptor& stream_desc = context.descriptor();
+        const Field& field = stream_desc.fields(it.source_field_pos());
+        if(!is_empty_type(field.type().data_type())) {
+            return false;
+        }
+    }
+    return true;
+}
+
 void decode_into_frame_static(
     SegmentInMemory &frame,
     PipelineContextRow &context,
@@ -338,7 +350,7 @@ void decode_into_frame_static(
                         it.dest_col(),
                         m.frame_field_descriptor_.name());
 
-            util::check(data != end, "Reached end of input block with {} fields to decode", it.remaining_fields());
+            util::check(data != end || remaining_fields_empty(it, context), "Reached end of input block with {} fields to decode", it.remaining_fields());
             decode_or_expand(data, buffer.data() + m.offset_bytes_, encoded_field, m.source_type_desc_,  m.dest_bytes_, buffers);
             ARCTICDB_TRACE(log::codec(), "Decoded column {} to position {}", field_name, data - begin);
 
@@ -1121,7 +1133,6 @@ folly::Future<std::vector<VariantKey>> fetch_data(
                     decode_into_frame_dynamic(frame, row, std::move(key_seg.segment()), buffers);
                 else
                     decode_into_frame_static(frame, row, std::move(key_seg.segment()), buffers);
-
                 return std::get<AtomKey>(key_seg.variant_key());
             });
         }
diff --git a/cpp/arcticdb/processing/aggregation.hpp b/cpp/arcticdb/processing/aggregation.hpp
index d3fda01f6d..cda96ed03b 100644
--- a/cpp/arcticdb/processing/aggregation.hpp
+++ b/cpp/arcticdb/processing/aggregation.hpp
@@ -48,6 +48,11 @@ struct OutputType<DataTypeTag<DataType::NANOSECONDS_UTC64>, void> {
     using type = ScalarTagType<DataTypeTag<DataType::NANOSECONDS_UTC64>>;
 };
 
+template<>
+struct OutputType<DataTypeTag<DataType::EMPTYVAL>, void> {
+    using type = ScalarTagType<DataTypeTag<DataType::EMPTYVAL>>;
+};
+
 void add_data_type_impl(DataType data_type, std::optional<DataType>& current_data_type);
 
 struct MinMaxAggregatorData {
diff --git a/cpp/arcticdb/python/python_handlers.cpp b/cpp/arcticdb/python/python_handlers.cpp
new file mode 100644
index 0000000000..1fe50db08c
--- /dev/null
+++ b/cpp/arcticdb/python/python_handlers.cpp
@@ -0,0 +1,30 @@
+/* Copyright 2023 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+#include <python/python_handlers.hpp>
+#include <python/gil_lock.hpp>
+
+namespace arcticdb {
+    void EmptyHandler::handle_type(
+        const uint8_t*&,
+        uint8_t* dest,
+        const VariantField&,
+        const entity::TypeDescriptor&,
+        size_t dest_bytes,
+        std::shared_ptr<BufferHolder>
+    ) {
+        ARCTICDB_SAMPLE(HandleEmpty, 0);
+        util::check(dest != nullptr, "Got null destination pointer");
+        const size_t num_rows = dest_bytes / get_type_size(DataType::EMPTYVAL);
+        static_assert(get_type_size(DataType::EMPTYVAL) == sizeof(PyObject*));
+        const PyObject** ptr_dest = reinterpret_cast<const PyObject**>(dest);
+        py::none none = {};
+        for(auto row = 0u; row < num_rows; ++row) {
+            none.inc_ref();
+            *ptr_dest++ = none.ptr();
+        }
+    }
+}
diff --git a/cpp/arcticdb/python/python_handlers.hpp b/cpp/arcticdb/python/python_handlers.hpp
new file mode 100644
index 0000000000..c12876570c
--- /dev/null
+++ b/cpp/arcticdb/python/python_handlers.hpp
@@ -0,0 +1,27 @@
+/* Copyright 2023 Man Group Operations Limited
+ *
+ * Use of this software is governed by the Business Source License 1.1 included in the file licenses/BSL.txt.
+ *
+ * As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
+ */
+#pragma once
+
+#include <entity/types.hpp>
+#include <util/type_handler.hpp>
+
+namespace arcticdb {
+    namespace py = pybind11;
+    struct EmptyHandler {
+
+        EmptyHandler() = default;
+
+        /// @see arcticdb::ITypeHandler
+        void handle_type(
+            const uint8_t*& data,
+            uint8_t* dest,
+            const VariantField& encoded_field,
+            const entity::TypeDescriptor& type_descriptor,
+            size_t dest_bytes,
+            std::shared_ptr<BufferHolder> buffers);
+    };
+} //namespace arcticdb
diff --git a/cpp/arcticdb/python/python_module.cpp b/cpp/arcticdb/python/python_module.cpp
index 35e3ccfe58..ecdf09ed08 100644
--- a/cpp/arcticdb/python/python_module.cpp
+++ b/cpp/arcticdb/python/python_module.cpp
@@ -25,6 +25,8 @@
 #include <arcticdb/util/global_lifetimes.hpp>
 #include <arcticdb/util/configs_map.hpp>
 #include <arcticdb/util/error_code.hpp>
+#include <arcticdb/util/type_handler.hpp>
+#include <arcticdb/python/python_handlers.hpp>
 
 #include <pybind11/pybind11.h>
 #include <folly/system/ThreadName.h>
@@ -254,6 +256,12 @@ void register_metrics(py::module && m){
     });
 }
 
+/// Register handling of non-trivial types. For more information @see arcticdb::TypeHandlerRegistry and
+/// @see arcticdb::ITypeHandler
+void register_type_handlers() {
+    arcticdb::TypeHandlerRegistry::instance()->register_handler(arcticdb::DataType::EMPTYVAL, arcticdb::EmptyHandler());
+}
+
 PYBIND11_MODULE(arcticdb_ext, m) {
     m.doc() = R"pbdoc(
         ArcticDB Extension plugin
@@ -299,6 +307,7 @@ PYBIND11_MODULE(arcticdb_ext, m) {
     register_log(m.def_submodule("log"));
     register_instrumentation(m.def_submodule("instrumentation"));
     register_metrics(m.def_submodule("metrics"));
+    register_type_handlers();
 
     auto cleanup_callback = []() {
         using namespace arcticdb;
@@ -316,4 +325,3 @@ PYBIND11_MODULE(arcticdb_ext, m) {
     m.attr("__version__") = "dev";
 #endif
 }
-
diff --git a/cpp/arcticdb/python/python_to_tensor_frame.cpp b/cpp/arcticdb/python/python_to_tensor_frame.cpp
index 9b55611bd2..bb885a924d 100644
--- a/cpp/arcticdb/python/python_to_tensor_frame.cpp
+++ b/cpp/arcticdb/python/python_to_tensor_frame.cpp
@@ -68,19 +68,43 @@ NativeTensor obj_to_tensor(PyObject *ptr) {
             auto none = py::none{};
             auto obj = reinterpret_cast<PyObject **>(arr->data);
             bool empty = false;
+            bool all_nans = false;
             PyObject *sample = *obj;
+            // Arctic allows both None and NaN to represent a string with no value. We have 3 options:
+            // * In case all values are None we can mark this column segment as EmptyType and avoid allocating storage
+            //      memory for it
+            // * In case all values are NaN we can't sample the value to check if it's UTF or ASCII (since NaN is not
+            //      UTF nor ASCII). In that case we choose to save it as UTF
+            // * In case there is at least one actual string we can sample it and decide the type of the column segment
+            //      based on it
+            // Note: ValueType::ASCII_DYNAMIC was used when Python 2 was supported. It is no longer supported, and
+            //  we're not expected to enter that branch.
             if (sample == none.ptr() || is_py_nan(sample)) {
-                // Iterate till we find the first non null element, the frontend ensures there is at least one.
+                empty = true;
+                all_nans = true;
                 util::check(c_style, "Non contiguous columns with first element as None not supported yet.");
-                auto casted_col =
-                    std::find_if(obj, obj + size, [&none](auto val) { return val != none.ptr() && !is_py_nan(val); });
-                empty = casted_col == obj + size;
-                sample = *casted_col;
+                PyObject** current_object = obj;
+                while(current_object < obj + size) {
+                    if(*current_object == none.ptr()) {
+                        all_nans = false;
+                    } else if(is_py_nan(*current_object)) {
+                        empty = false;
+                    } else {
+                        all_nans = false;
+                        empty = false;
+                        break;
+                    }
+                    ++current_object;
+                }
+                sample = *current_object;
             }
-            if (empty || is_unicode(sample)) {
+            if (empty) {
+                val_type = ValueType::EMPTY;
+            } else if(all_nans || is_unicode(sample)){
                 val_type = ValueType::UTF_DYNAMIC;
-            } else if (PYBIND11_BYTES_CHECK(sample))
+            } else if (PYBIND11_BYTES_CHECK(sample)) {
                 val_type = ValueType::ASCII_DYNAMIC;
+            }
         }
     }
 
@@ -189,4 +213,4 @@ InputTensorFrame py_none_to_frame() {
     return res;
 }
 
-} // namespace arcticdb::convert
\ No newline at end of file
+} // namespace arcticdb::convert
diff --git a/cpp/arcticdb/util/buffer.hpp b/cpp/arcticdb/util/buffer.hpp
index 7de5a642e1..1c2a617c3c 100644
--- a/cpp/arcticdb/util/buffer.hpp
+++ b/cpp/arcticdb/util/buffer.hpp
@@ -149,7 +149,7 @@ struct Buffer : public BaseBuffer<Buffer, true> {
     }
 
     template<typename T>
-    T *ptr_cast(size_t bytes_offset, size_t required_bytes) {
+    [[nodiscard]] T *ptr_cast(size_t bytes_offset, size_t required_bytes) {
         check_invariants();
         if (bytes_offset  + required_bytes > bytes()) {
             std::string err = fmt::format("Cursor overflow in reallocating buffer ptr_cast, cannot read {} bytes from a buffer of size {} with cursor "
@@ -172,9 +172,10 @@ struct Buffer : public BaseBuffer<Buffer, true> {
     }
 
     inline void ensure(size_t bytes) {
-        if(bytes > available())
-            resize(bytes);
-        else {
+        const size_t total_size = bytes + preamble_bytes_;
+        if(total_size > capacity_) {
+            resize(total_size);
+        } else {
             ARCTICDB_TRACE(log::version(), "Buffer {} has sufficient bytes for {}, ptr {} data {}, capacity {}",
                                 uintptr_t(this), bytes, uintptr_t(ptr_), uintptr_t(data_), capacity_, body_bytes_);
         }
@@ -197,11 +198,11 @@ struct Buffer : public BaseBuffer<Buffer, true> {
         return {ptr_, body_bytes_};
     }
 
-    uint8_t &operator[](size_t bytes_offset) {
+    [[nodiscard]] uint8_t &operator[](size_t bytes_offset) {
         return ptr_[bytes_offset];
     }
 
-    const uint8_t &operator[](size_t bytes_offset) const {
+    [[nodiscard]] const uint8_t &operator[](size_t bytes_offset) const {
         return ptr_[bytes_offset];
     }
 
@@ -213,36 +214,34 @@ struct Buffer : public BaseBuffer<Buffer, true> {
         return preamble_bytes_;
     }
 
-    uint8_t* preamble() {
+    [[nodiscard]] uint8_t* preamble() {
         return data_;
     }
     
-    size_t available() const {
+    [[nodiscard]] size_t available() const {
         return capacity_ >= preamble_bytes_ ? capacity_ - preamble_bytes_ : 0;
     }
 
   private:
-    inline void resize(size_t bytes) {
-        if (bytes == 0) {
-            reset();
+    inline void resize(size_t alloc_bytes) {
+        const size_t bytes = alloc_bytes - preamble_bytes_;
+        util::check(alloc_bytes >= preamble_bytes_, "The requested size of a resizes call is less than the preamble bytes");
+        auto [mem_ptr, ts] = ptr_ ?
+                             Allocator::realloc(std::make_pair(data_, ts_), alloc_bytes)
+                                  :
+                             Allocator::aligned_alloc(alloc_bytes);
+
+        ARCTICDB_TRACE(log::codec(), "Allocating {} bytes ({} + {} bytes preamble)", alloc_bytes, bytes, preamble_bytes_);
+        if (mem_ptr) {
+            data_ = mem_ptr;
+            ptr_ = data_ + preamble_bytes_;
+            ts_ = ts;
+            body_bytes_ = bytes;
+            capacity_ = body_bytes_ + preamble_bytes_;
+            ARCTICDB_TRACE(log::version(), "Buffer {} did realloc for {}, ptr {} data {}, capacity {}",
+                                uintptr_t(this), bytes,  uintptr_t(ptr_), uintptr_t(data_), capacity_, body_bytes_);
         } else {
-            auto alloc_bytes = bytes + preamble_bytes_;
-            auto [mem_ptr, ts] = ptr_ ?
-                                 Allocator::realloc(std::make_pair(data_, ts_), alloc_bytes)
-                                      :
-                                 Allocator::aligned_alloc(alloc_bytes);
-
-            ARCTICDB_TRACE(log::codec(), "Allocating {} bytes ({} + {} bytes preamble)", alloc_bytes, bytes, preamble_bytes_);
-            if (mem_ptr) {
-                data_ = mem_ptr;
-                ptr_ = data_ + preamble_bytes_;
-                ts_ = ts;
-                body_bytes_ = bytes;
-                capacity_ = body_bytes_ + preamble_bytes_;
-                ARCTICDB_TRACE(log::version(), "Buffer {} did realloc for {}, ptr {} data {}, capacity {}",
-                                    uintptr_t(this), bytes,  uintptr_t(ptr_), uintptr_t(data_), capacity_, body_bytes_);
-            } else
-                throw std::bad_alloc();
+            throw std::bad_alloc();
         }
         check_invariants();
     }
diff --git a/cpp/arcticdb/util/type_handler.hpp b/cpp/arcticdb/util/type_handler.hpp
index f0a838617e..d2bcc60318 100644
--- a/cpp/arcticdb/util/type_handler.hpp
+++ b/cpp/arcticdb/util/type_handler.hpp
@@ -24,14 +24,19 @@ struct ITypeHandler {
     template<class Base>
     struct Interface : Base {
 
+        /// Handle decoding of a non-trivial type to Python object
+        /// @param[in] source Data to be decoded to Python objects
+        /// @param[out] dest Memory where the resulting Python objects are stored
+        /// @param[in] dest_bytes Size of dest in bytes
         void handle_type(
-            const uint8_t*& data,
+            const uint8_t*& source,
             uint8_t* dest,
             const VariantField& encoded_field_info,
             const entity::TypeDescriptor& type_descriptor,
             size_t dest_bytes,
-            std::shared_ptr<BufferHolder> buffers
-            ) { folly::poly_call<0>(*this, data, dest, encoded_field_info, type_descriptor, dest_bytes, buffers); }
+            std::shared_ptr<BufferHolder> buffers) {
+            folly::poly_call<0>(*this, source, dest, encoded_field_info, type_descriptor, dest_bytes, buffers);
+        }
     };
 
     template<class T>
@@ -40,6 +45,8 @@ struct ITypeHandler {
 
 using TypeHandler = folly::Poly<ITypeHandler>;
 
+/// Some types cannot be trivially converted from storage to Python types .This singleton holds a set of type erased
+/// handlers (implementing the handle_type function) which handle the parsing from storage to python.
 class TypeHandlerRegistry {
 public:
     static std::shared_ptr<TypeHandlerRegistry> instance_;
diff --git a/cpp/arcticdb/version/schema_checks.hpp b/cpp/arcticdb/version/schema_checks.hpp
index 2a58458cd2..0e01b176c6 100644
--- a/cpp/arcticdb/version/schema_checks.hpp
+++ b/cpp/arcticdb/version/schema_checks.hpp
@@ -90,4 +90,4 @@ inline void fix_descriptor_mismatch_or_throw(
             operation);
     }
 }
-} // namespace arcticc
\ No newline at end of file
+} // namespace arcticdb
diff --git a/cpp/proto/arcticc/pb2/descriptors.proto b/cpp/proto/arcticc/pb2/descriptors.proto
index b089eb9666..bf6efb2263 100644
--- a/cpp/proto/arcticc/pb2/descriptors.proto
+++ b/cpp/proto/arcticc/pb2/descriptors.proto
@@ -30,6 +30,7 @@ message TypeDescriptor {
         PICKLE = 10; // BYTES + pickle specific encoding
         DYNAMIC_STRING = 11; // offset reference to deduplicated string pool
         DYNAMIC_UTF8 = 12;
+        EMPTY = 13;
     }
 
     enum SizeBits {


Test patch:
diff --git a/cpp/arcticdb/processing/test/test_has_valid_type_promotion.cpp b/cpp/arcticdb/processing/test/test_has_valid_type_promotion.cpp
index f4fd9f7da3..2bfa1dafea 100644
--- a/cpp/arcticdb/processing/test/test_has_valid_type_promotion.cpp
+++ b/cpp/arcticdb/processing/test/test_has_valid_type_promotion.cpp
@@ -121,3 +121,30 @@ TEST(HasValidTypePromotion, FloatFloat) {
     EXPECT_FALSE(has_valid_type_promotion(float64, float32));
 }
 
+TEST(HasValidTypePromotion, EmptyToEverything) {
+    using namespace arcticdb;
+    using namespace arcticdb::entity;
+    TypeDescriptor source(ValueType::EMPTY, SizeBits::S64, Dimension::Dim0);
+    for(int value_type = int(ValueType::UNKNOWN_VALUE_TYPE); value_type < int(ValueType::COUNT); ++value_type) {
+        for(int size_bits = int(SizeBits::UNKNOWN_SIZE_BITS); size_bits < int(SizeBits::COUNT); ++size_bits) {
+            const TypeDescriptor target(ValueType(value_type), SizeBits(size_bits), Dimension::Dim0);
+            ASSERT_EQ(has_valid_type_promotion(source, target), target);
+        }
+    }
+}
+
+TEST(HasValidTypePromotion, EverythingToEmpty) {
+    using namespace arcticdb;
+    using namespace arcticdb::entity;
+    const TypeDescriptor target(ValueType::EMPTY, SizeBits::S64, Dimension::Dim0);
+    for(int value_type = int(ValueType::UNKNOWN_VALUE_TYPE); value_type < int(ValueType::COUNT); ++value_type) {
+        for(int size_bits = int(SizeBits::UNKNOWN_SIZE_BITS); size_bits < int(SizeBits::COUNT); ++size_bits) {
+            const TypeDescriptor source(ValueType(value_type), SizeBits(size_bits), Dimension::Dim0);
+            if(!is_empty_type(source.data_type())) {
+                ASSERT_FALSE(has_valid_type_promotion(source, target).has_value());
+            } else {
+                ASSERT_EQ(has_valid_type_promotion(source, target), target);
+            }
+        }
+    }
+}