diff --git a/clickhouse/CMakeLists.txt b/clickhouse/CMakeLists.txt
index 6664ee4f..e83c2964 100644
--- a/clickhouse/CMakeLists.txt
+++ b/clickhouse/CMakeLists.txt
@@ -76,7 +76,6 @@ SET ( clickhouse-cpp-lib-src
     columns/string.h
     columns/time.h
     columns/tuple.h
-    columns/utils.h
     columns/uuid.h
 
     types/bignum.h
@@ -257,7 +256,6 @@ INSTALL(FILES columns/map.h DESTINATION include/clickhouse/columns/)
 INSTALL(FILES columns/string.h DESTINATION include/clickhouse/columns/)
 INSTALL(FILES columns/time.h DESTINATION include/clickhouse/columns/)
 INSTALL(FILES columns/tuple.h DESTINATION include/clickhouse/columns/)
-INSTALL(FILES columns/utils.h DESTINATION include/clickhouse/columns/)
 INSTALL(FILES columns/uuid.h DESTINATION include/clickhouse/columns/)
 
 # types
diff --git a/clickhouse/columns/array.h b/clickhouse/columns/array.h
index f771e4af..54dc9484 100644
--- a/clickhouse/columns/array.h
+++ b/clickhouse/columns/array.h
@@ -2,7 +2,6 @@
 
 #include "column.h"
 #include "numeric.h"
-#include "utils.h"
 
 #include <memory>
 
@@ -128,26 +127,66 @@ class ColumnArrayT : public ColumnArray {
         : ColumnArrayT(std::make_shared<NestedColumnType>(std::forward<Args>(args)...))
     {}
 
-    /** Create a ColumnArrayT from a ColumnArray, without copying data and offsets, but by 'stealing' those from `col`.
+    /** Create a ColumnArrayT that SHARES the internals of `col` (nested data and
+     *  offsets) via shared_ptr, WITHOUT stealing or copying them.
      *
-     *  Ownership of column internals is transferred to returned object, original (argument) object
-     *  MUST NOT BE USED IN ANY WAY, it is only safe to dispose it.
+     *  The original `col` remains fully valid and usable. Both the original and the
+     *  returned wrapper reference the same underlying columns, so mutations through
+     *  one are visible through the other.
      *
-     *  Throws an exception if `col` is of wrong type, it is safe to use original col in this case.
-     *  This is a static method to make such conversion verbose.
+     *  The two-argument overloads are non-throwing: on a type mismatch they return
+     *  nullptr and, if `error` is non-null, assign a description to `*error`. The
+     *  single-argument overloads throw ValidationError on a type mismatch instead.
      */
-    static auto Wrap(ColumnArray&& col) {
-        auto nested_data = WrapColumn<NestedColumnType>(col.GetData());
+    static std::shared_ptr<ColumnArrayT<NestedColumnType>> Wrap(const ColumnArray& col, ValidationError* error) {
+        auto nested_data = WrapColumn<NestedColumnType>(col.data_, error);
+        if (!nested_data) {
+            return nullptr;
+        }
         return std::make_shared<ColumnArrayT<NestedColumnType>>(nested_data, col.offsets_);
     }
 
-    static auto Wrap(Column&& col) {
-        return Wrap(std::move(dynamic_cast<ColumnArray&&>(col)));
+    static std::shared_ptr<ColumnArrayT<NestedColumnType>> Wrap(const Column& col, ValidationError* error) {
+        if (auto* c = dynamic_cast<const ColumnArray*>(&col)) {
+            return Wrap(*c, error);
+        }
+        if (error) {
+            *error = ValidationError("Can't wrap column of type " + col.GetType().GetName() + " as Array");
+        }
+        return nullptr;
     }
 
     // Helper to simplify integration with other APIs
-    static auto Wrap(ColumnRef&& col) {
-        return Wrap(std::move(*col->AsStrict<ColumnArray>()));
+    static std::shared_ptr<ColumnArrayT<NestedColumnType>> Wrap(const ColumnRef& col, ValidationError* error) {
+        return Wrap(*col, error);
+    }
+
+    static auto Wrap(const ColumnArray& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    static auto Wrap(const Column& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    // Helper to simplify integration with other APIs
+    static auto Wrap(const ColumnRef& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
     }
 
     /// A single (row) value of the Array-column, i.e. readonly array of items.
diff --git a/clickhouse/columns/column.h b/clickhouse/columns/column.h
index 475df89a..ce2a9ec5 100644
--- a/clickhouse/columns/column.h
+++ b/clickhouse/columns/column.h
@@ -4,8 +4,10 @@
 #include "../columns/itemview.h"
 #include "../exceptions.h"
 
+#include <algorithm>
 #include <memory>
 #include <stdexcept>
+#include <vector>
 
 namespace clickhouse {
 
@@ -24,26 +26,25 @@ class Column : public std::enable_shared_from_this<Column> {
     virtual ~Column() {}
 
     /// Downcast pointer to the specific column's subtype.
+    ///
+    /// If T is a "wrappable" typed column (one exposing a static Wrap method, e.g.
+    /// ColumnArrayT/ColumnTupleT/ColumnMapT/ColumnNullableT/ColumnLowCardinalityT) and the
+    /// column is not already exactly T, this attempts to Wrap it as T (a storage-sharing
+    /// typed view). Returns nullptr when neither an exact cast nor a wrap is possible.
+    /// (Definitions are out-of-line below, after WrapColumn is declared.)
     template <typename T>
-    inline std::shared_ptr<T> As() {
-        return std::dynamic_pointer_cast<T>(shared_from_this());
-    }
+    inline std::shared_ptr<T> As();
 
-    /// Downcast pointer to the specific column's subtype.
+    /// Const overload. Unlike the non-const As(), this does NOT wrap: it only performs an
+    /// exact downcast and returns nullptr on mismatch (even for a wrappable T). Wrapping is
+    /// intentionally disabled here because it would synthesize a mutable, storage-sharing
+    /// view from a const column (requiring a const_cast), which is not const-correct.
     template <typename T>
-    inline std::shared_ptr<const T> As() const {
-        return std::dynamic_pointer_cast<const T>(shared_from_this());
-    }
+    inline std::shared_ptr<const T> As() const;
 
-    /// Downcast pointer to the specific column's subtype.
+    /// Like As(), but throws ValidationError instead of returning nullptr on failure.
     template <typename T>
-    inline std::shared_ptr<T> AsStrict() {
-        auto result = std::dynamic_pointer_cast<T>(shared_from_this());
-        if (!result) {
-            throw ValidationError("Can't cast from " + type_->GetName());
-        }
-        return result;
-    }
+    inline std::shared_ptr<T> AsStrict();
 
     /// Get type object of the column.
     inline TypeRef Type() const { return type_; }
@@ -104,4 +105,87 @@ class Column : public std::enable_shared_from_this<Column> {
     TypeRef type_;
 };
 
+template <typename T>
+std::vector<T> SliceVector(const std::vector<T>& vec, size_t begin, size_t len) {
+    std::vector<T> result;
+
+    if (begin < vec.size()) {
+        len = std::min(len, vec.size() - begin);
+        result.assign(vec.begin() + begin, vec.begin() + (begin + len));
+    }
+
+    return result;
+}
+
+template <typename T>
+struct HasWrapMethod {
+private:
+    static int detect(...);
+    template <typename U>
+    static decltype(U::Wrap(std::move(std::declval<ColumnRef>()))) detect(const U&);
+
+public:
+    static constexpr bool value = !std::is_same<int, decltype(detect(std::declval<T>()))>::value;
+};
+
+// Non-throwing: returns nullptr and (if `error` is non-null) fills `*error` when `column`
+// can't be wrapped as T.
+template <typename T>
+inline std::shared_ptr<T> WrapColumn(const ColumnRef& column, ValidationError* error) {
+    if constexpr (HasWrapMethod<T>::value) {
+        return T::Wrap(column, error);
+    } else {
+        auto result = column->template As<T>();
+        if (!result && error) {
+            *error = ValidationError("Can't wrap column of type " + column->GetType().GetName());
+        }
+        return result;
+    }
+}
+
+// Throwing convenience wrapper.
+template <typename T>
+inline std::shared_ptr<T> WrapColumn(const ColumnRef& column) {
+    ValidationError error;
+    auto result = WrapColumn<T>(column, &error);
+    if (!result) {
+        throw error;
+    }
+    return result;
+}
+
+template <typename T>
+inline std::shared_ptr<T> Column::As() {
+    if constexpr (HasWrapMethod<T>::value) {
+        if (auto exact = std::dynamic_pointer_cast<T>(shared_from_this())) {
+            return exact;
+        }
+        return WrapColumn<T>(shared_from_this(), nullptr);
+    } else {
+        return std::dynamic_pointer_cast<T>(shared_from_this());
+    }
+}
+
+template <typename T>
+inline std::shared_ptr<const T> Column::As() const {
+    // No wrapping for the const overload (see declaration): exact downcast only.
+    return std::dynamic_pointer_cast<const T>(shared_from_this());
+}
+
+template <typename T>
+inline std::shared_ptr<T> Column::AsStrict() {
+    if constexpr (HasWrapMethod<T>::value) {
+        if (auto exact = std::dynamic_pointer_cast<T>(shared_from_this())) {
+            return exact;
+        }
+        return WrapColumn<T>(shared_from_this());
+    } else {
+        auto result = std::dynamic_pointer_cast<T>(shared_from_this());
+        if (!result) {
+            throw ValidationError("Can't cast from " + type_->GetName());
+        }
+        return result;
+    }
+}
+
 }  // namespace clickhouse
diff --git a/clickhouse/columns/enum.cpp b/clickhouse/columns/enum.cpp
index 43fab893..1fa2cba5 100644
--- a/clickhouse/columns/enum.cpp
+++ b/clickhouse/columns/enum.cpp
@@ -1,5 +1,4 @@
 #include "enum.h"
-#include "utils.h"
 
 #include "../base/input.h"
 #include "../base/output.h"
diff --git a/clickhouse/columns/factory.cpp b/clickhouse/columns/factory.cpp
index a01304f8..f64109f2 100644
--- a/clickhouse/columns/factory.cpp
+++ b/clickhouse/columns/factory.cpp
@@ -233,13 +233,25 @@ static ColumnRef CreateColumnFromAst(const TypeAst& ast, CreateColumnByTypeSetti
                 }
             }
             else {
+                // Create the base ColumnLowCardinality (like Array/Nullable/Tuple/Map create their
+                // base types). Callers can obtain the strongly-typed ColumnLowCardinalityT<...> view
+                // on demand via Column::As<ColumnLowCardinalityT<...>>(), which wraps it.
                 switch (nested.code) {
-                    // TODO (nemkov): update this to maximize code reuse.
                     case Type::String:
-                        return std::make_shared<ColumnLowCardinalityT<ColumnString>>();
                     case Type::FixedString:
-                        return std::make_shared<ColumnLowCardinalityT<ColumnFixedString>>(GetASTChildElement(nested, 0).value);
+                        return std::make_shared<ColumnLowCardinality>(CreateColumnFromAst(nested, settings));
                     case Type::Nullable:
+                        // Nullable needs its own case (it can't reuse the generic CreateColumnFromAst
+                        // path above) for two reasons:
+                        //   1. Constructor overload: ColumnLowCardinality has a dedicated
+                        //      ColumnLowCardinality(shared_ptr<ColumnNullable>) ctor that seeds the
+                        //      special NULL item at dictionary index 0 (via AppendNullItem()). We must
+                        //      pass a statically-typed shared_ptr<ColumnNullable> so that overload is
+                        //      selected; passing a ColumnRef would statically bind to the generic
+                        //      ColumnLowCardinality(ColumnRef) ctor, which only appends the default
+                        //      item and would omit the null item, producing an incorrect nullable
+                        //      dictionary.
+                        //   2. It lets us construct the ColumnNullable with an explicit UInt8 null-map.
                         return std::make_shared<ColumnLowCardinality>(
                             std::make_shared<ColumnNullable>(
                                 CreateColumnFromAst(GetASTChildElement(nested, 0), settings),
diff --git a/clickhouse/columns/geo.cpp b/clickhouse/columns/geo.cpp
index fa987732..daf70664 100644
--- a/clickhouse/columns/geo.cpp
+++ b/clickhouse/columns/geo.cpp
@@ -1,7 +1,5 @@
 #include "geo.h"
 
-#include "utils.h"
-
 namespace {
 using namespace ::clickhouse;
 
diff --git a/clickhouse/columns/lowcardinality.cpp b/clickhouse/columns/lowcardinality.cpp
index e286c863..2a3bd6b0 100644
--- a/clickhouse/columns/lowcardinality.cpp
+++ b/clickhouse/columns/lowcardinality.cpp
@@ -159,6 +159,7 @@ ColumnLowCardinality::ColumnLowCardinality(ColumnRef dictionary_column)
     : Column(Type::CreateLowCardinality(dictionary_column->Type())),
       dictionary_column_(dictionary_column->CloneEmpty()), // safe way to get an column of the same type.
       index_column_(std::make_shared<ColumnUInt32>()),
+      unique_items_map_(std::make_shared<UniqueItems>()),
       index_type_code_(Type::UInt32)
 {
     Setup(dictionary_column);
@@ -168,6 +169,7 @@ ColumnLowCardinality::ColumnLowCardinality(std::shared_ptr<ColumnNullable> dicti
     : Column(Type::CreateLowCardinality(dictionary_column->Type())),
       dictionary_column_(dictionary_column->CloneEmpty()), // safe way to get an column of the same type.
       index_column_(std::make_shared<ColumnUInt32>()),
+      unique_items_map_(std::make_shared<UniqueItems>()),
       index_type_code_(Type::UInt32)
 {
     AppendNullItem();
@@ -381,7 +383,7 @@ bool ColumnLowCardinality::LoadBody(InputStream* input, size_t rows) {
 
         dictionary_column_->Swap(*new_dictionary);
         index_column_.swap(new_index);
-        unique_items_map_.swap(new_unique_items_map);
+        unique_items_map_->swap(new_unique_items_map);
         index_type_code_ = index_column_->Type()->GetCode();
 
         return true;
@@ -418,7 +420,7 @@ void ColumnLowCardinality::SaveBody(OutputStream* output) {
 void ColumnLowCardinality::Clear() {
     index_column_->Clear();
     dictionary_column_->Clear();
-    unique_items_map_.clear();
+    unique_items_map_->clear();
 
     if (auto columnNullable = dictionary_column_->As<ColumnNullable>()) {
         AppendNullItem();
@@ -457,7 +459,7 @@ void ColumnLowCardinality::Swap(Column& other) {
     dictionary_column_->Swap(*col.dictionary_column_);
 
     index_column_.swap(col.index_column_);
-    unique_items_map_.swap(col.unique_items_map_);
+    unique_items_map_->swap(*col.unique_items_map_);
     std::swap(index_type_code_, col.index_type_code_);
 }
 
@@ -480,7 +482,7 @@ void ColumnLowCardinality::AppendUnsafe(const ItemView & value) {
     const auto key = computeHashKey(value);
     const auto initial_index_size = index_column_->Size();
     // If the value is unique, then we are going to append it to a dictionary, hence new index is Size().
-    auto [iterator, is_new_item] = unique_items_map_.try_emplace(key, dictionary_column_->Size());
+    auto [iterator, is_new_item] = unique_items_map_->try_emplace(key, dictionary_column_->Size());
     try {
         // Order is important, adding to dictionary last, since it is much (MUCH!!!!) harder
         // to remove item from dictionary column than from index column
@@ -497,7 +499,7 @@ void ColumnLowCardinality::AppendUnsafe(const ItemView & value) {
         if (index_column_->Size() != initial_index_size)
             removeLastIndex();
         if (is_new_item)
-            unique_items_map_.erase(iterator);
+            unique_items_map_->erase(iterator);
 
         throw;
     }
@@ -507,13 +509,13 @@ void ColumnLowCardinality::AppendNullItem()
 {
     const auto null_item = GetNullItemForDictionary(dictionary_column_);
     AppendToDictionary(*dictionary_column_, null_item);
-    unique_items_map_.emplace(computeHashKey(null_item), 0);
+    unique_items_map_->emplace(computeHashKey(null_item), 0);
 }
 
 void ColumnLowCardinality::AppendDefaultItem()
 {
     const auto defaultItem = GetDefaultItemForDictionary(dictionary_column_);
-    unique_items_map_.emplace(computeHashKey(defaultItem), dictionary_column_->Size());
+    unique_items_map_->emplace(computeHashKey(defaultItem), dictionary_column_->Size());
     AppendToDictionary(*dictionary_column_, defaultItem);
 }
 
diff --git a/clickhouse/columns/lowcardinality.h b/clickhouse/columns/lowcardinality.h
index 33c339e6..0922defd 100644
--- a/clickhouse/columns/lowcardinality.h
+++ b/clickhouse/columns/lowcardinality.h
@@ -50,7 +50,15 @@ class ColumnLowCardinality : public Column {
     // so make sure to NOT change address of the dictionary object (with reset(), swap()) or with anything else.
     ColumnRef dictionary_column_;
     ColumnRef index_column_;
-    UniqueItems unique_items_map_;
+    // Shared so that a wrapped (ColumnLowCardinalityT::Wrap) column shares the same dedup map as its
+    // source, keeping dictionary/index/map coherent across both holders (same semantics as other columns).
+    std::shared_ptr<UniqueItems> unique_items_map_;
+
+protected:
+    // Shallow copy: shares dictionary_column_, index_column_ and unique_items_map_ (all shared_ptr),
+    // copies index_type_code_ and the base type. Used by ColumnLowCardinalityT::Wrap to create a
+    // non-destructive, storage-sharing view of `col`.
+    ColumnLowCardinality(const ColumnLowCardinality& col) = default;
 
 public:
     ColumnLowCardinality(ColumnLowCardinality&& col) = default;
@@ -136,6 +144,15 @@ class ColumnLowCardinalityT : public ColumnLowCardinality {
     {
     }
 
+    // Shares the internals of `col` (dictionary, index and dedup map) via shared_ptr, WITHOUT
+    // stealing or copying them. Used by Wrap to create a non-destructive, storage-sharing view.
+    explicit ColumnLowCardinalityT(const ColumnLowCardinality& col)
+        : ColumnLowCardinality(col)
+        ,  typed_dictionary_(dynamic_cast<DictionaryColumnType &>(*GetDictionary()))
+        ,  type_(GetTypeCode(typed_dictionary_))
+    {
+    }
+
     template <typename ...Args>
     explicit ColumnLowCardinalityT(Args &&... args)
         : ColumnLowCardinalityT(std::make_shared<DictionaryColumnType>(std::forward<Args>(args)...))
@@ -182,23 +199,74 @@ class ColumnLowCardinalityT : public ColumnLowCardinality {
         }
     }
 
-    /** Create a ColumnLowCardinalityT from a ColumnLowCardinality, without copying data and offsets, but by
-     * 'stealing' those from `col`.
+    /** Create a ColumnLowCardinalityT that SHARES the internals of `col` (dictionary, index and
+     *  dedup map) via shared_ptr, WITHOUT stealing or copying them.
      *
-     *  Ownership of column internals is transferred to returned object, original (argument) object
-     *  MUST NOT BE USED IN ANY WAY, it is only safe to dispose it.
+     *  The original `col` remains fully valid and usable. Both the original and the returned
+     *  wrapper reference the same underlying storage, so mutations through one are visible through
+     *  the other and remain coherent (the dedup map is shared as well).
      *
-     *  Throws an exception if `col` is of wrong type, it is safe to use original col in this case.
-     *  This is a static method to make such conversion verbose.
+     *  The two-argument overloads are non-throwing: on a type mismatch they return nullptr and,
+     *  if `error` is non-null, assign a description to `*error`. The single-argument overloads
+     *  throw ValidationError on a type mismatch instead.
      */
-    static auto Wrap(ColumnLowCardinality&& col) {
-        return std::make_shared<ColumnLowCardinalityT<WrappedColumnType>>(std::move(col));
+    static std::shared_ptr<ColumnLowCardinalityT<WrappedColumnType>> Wrap(const ColumnLowCardinality& col, ValidationError* error) {
+        // Strict (non-wrapping) check on purpose: the constructor binds typed_dictionary_ as a
+        // DictionaryColumnType& via a reference dynamic_cast, so the stored dictionary must be
+        // exactly DictionaryColumnType. Using the wrapping As<> here could pass for a base
+        // dictionary and then make that reference cast throw std::bad_cast.
+        if (!std::dynamic_pointer_cast<DictionaryColumnType>(col.dictionary_column_)) {
+            if (error) {
+                *error = ValidationError("Can't wrap LowCardinality column with dictionary of type "
+                                         + col.dictionary_column_->GetType().GetName());
+            }
+            return nullptr;
+        }
+        return std::make_shared<ColumnLowCardinalityT<WrappedColumnType>>(col);
     }
 
-    static auto Wrap(Column&& col) { return Wrap(std::move(dynamic_cast<ColumnLowCardinality&&>(col))); }
+    static std::shared_ptr<ColumnLowCardinalityT<WrappedColumnType>> Wrap(const Column& col, ValidationError* error) {
+        if (auto* c = dynamic_cast<const ColumnLowCardinality*>(&col)) {
+            return Wrap(*c, error);
+        }
+        if (error) {
+            *error = ValidationError("Can't wrap column of type " + col.GetType().GetName() + " as LowCardinality");
+        }
+        return nullptr;
+    }
 
     // Helper to simplify integration with other APIs
-    static auto Wrap(ColumnRef&& col) { return Wrap(std::move(*col->AsStrict<ColumnLowCardinality>())); }
+    static std::shared_ptr<ColumnLowCardinalityT<WrappedColumnType>> Wrap(const ColumnRef& col, ValidationError* error) {
+        return Wrap(*col, error);
+    }
+
+    static auto Wrap(const ColumnLowCardinality& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    static auto Wrap(const Column& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    // Helper to simplify integration with other APIs
+    static auto Wrap(const ColumnRef& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
     ColumnRef Slice(size_t begin, size_t size) const override {
         return Wrap(ColumnLowCardinality::Slice(begin, size));
diff --git a/clickhouse/columns/map.cpp b/clickhouse/columns/map.cpp
index 839b0668..a8d58967 100644
--- a/clickhouse/columns/map.cpp
+++ b/clickhouse/columns/map.cpp
@@ -3,7 +3,6 @@
 #include <stdexcept>
 
 #include "../exceptions.h"
-#include "utils.h"
 
 namespace {
 
diff --git a/clickhouse/columns/map.h b/clickhouse/columns/map.h
index 4d644802..ae3059d4 100644
--- a/clickhouse/columns/map.h
+++ b/clickhouse/columns/map.h
@@ -119,7 +119,9 @@ class ColumnMapT : public ColumnMap {
 
         inline auto At(const Key& key) const {
             auto it = Find(key);
-            if (it == end()) throw ValidationError("ColumnMap value key not found");
+            if (it == end()) {
+                throw ValidationError("ColumnMap value key not found");
+            }
             return (*it).second;
         }
 
@@ -240,15 +242,67 @@ class ColumnMapT : public ColumnMap {
         typed_data_->Append(Iterator{value.begin(), functor}, Iterator{value.end(), functor});
     }
 
-    static auto Wrap(ColumnMap&& col) {
-        auto data = ArrayColumnType::Wrap(std::move(col.data_));
-        return std::make_shared<ColumnMapT<K, V>>(std::move(data));
+    /** Create a ColumnMapT that SHARES the internals of `col` (its backing array of
+     *  key/value tuples) via shared_ptr, WITHOUT stealing or copying them.
+     *
+     *  The original `col` remains fully valid and usable. Both the original and the
+     *  returned wrapper reference the same underlying columns, so mutations through
+     *  one are visible through the other.
+     *
+     *  The two-argument overloads are non-throwing: on a type mismatch they return
+     *  nullptr and, if `error` is non-null, assign a description to `*error`. The
+     *  single-argument overloads throw ValidationError on a type mismatch instead.
+     */
+    static std::shared_ptr<ColumnMapT<K, V>> Wrap(const ColumnMap& col, ValidationError* error) {
+        auto data = ArrayColumnType::Wrap(*col.data_, error);
+        if (!data) {
+            return nullptr;
+        }
+        return std::make_shared<ColumnMapT<K, V>>(data);
     }
 
-    static auto Wrap(Column&& col) { return Wrap(std::move(dynamic_cast<ColumnMap&&>(col))); }
+    static std::shared_ptr<ColumnMapT<K, V>> Wrap(const Column& col, ValidationError* error) {
+        if (auto* c = dynamic_cast<const ColumnMap*>(&col)) {
+            return Wrap(*c, error);
+        }
+        if (error) {
+            *error = ValidationError("Can't wrap column of type " + col.GetType().GetName() + " as Map");
+        }
+        return nullptr;
+    }
 
     // Helper to simplify integration with other APIs
-    static auto Wrap(ColumnRef&& col) { return Wrap(std::move(*col->AsStrict<ColumnMap>())); }
+    static std::shared_ptr<ColumnMapT<K, V>> Wrap(const ColumnRef& col, ValidationError* error) {
+        return Wrap(*col, error);
+    }
+
+    static auto Wrap(const ColumnMap& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    static auto Wrap(const Column& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    // Helper to simplify integration with other APIs
+    static auto Wrap(const ColumnRef& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
 private:
     std::shared_ptr<ArrayColumnType> typed_data_;
diff --git a/clickhouse/columns/nullable.h b/clickhouse/columns/nullable.h
index 6b34552c..fa40bb98 100644
--- a/clickhouse/columns/nullable.h
+++ b/clickhouse/columns/nullable.h
@@ -108,25 +108,74 @@ class ColumnNullableT : public ColumnNullable {
         }
     }
 
-    /** Create a ColumnNullableT from a ColumnNullable, without copying data and offsets, but by
-     * 'stealing' those from `col`.
+    /** Create a ColumnNullableT that SHARES the internals of `col` (nested data and
+     *  null map) via shared_ptr, WITHOUT stealing or copying them.
      *
-     *  Ownership of column internals is transferred to returned object, original (argument) object
-     *  MUST NOT BE USED IN ANY WAY, it is only safe to dispose it.
+     *  The original `col` remains fully valid and usable. Both the original and the
+     *  returned wrapper reference the same underlying columns, so mutations through
+     *  one are visible through the other.
      *
-     *  Throws an exception if `col` is of wrong type, it is safe to use original col in this case.
-     *  This is a static method to make such conversion verbose.
+     *  The two-argument overloads are non-throwing: on a type mismatch they return
+     *  nullptr and, if `error` is non-null, assign a description to `*error`. The
+     *  single-argument overloads throw ValidationError on a type mismatch instead.
      */
-    static auto Wrap(ColumnNullable&& col) {
-        return std::make_shared<ColumnNullableT<NestedColumnType>>(
-            col.Nested()->AsStrict<NestedColumnType>(),
-            col.Nulls()->AsStrict<ColumnUInt8>()) ;
+    static std::shared_ptr<ColumnNullableT<NestedColumnType>> Wrap(const ColumnNullable& col, ValidationError* error) {
+        auto nested = WrapColumn<NestedColumnType>(col.Nested(), error);
+        if (!nested) {
+            return nullptr;
+        }
+        auto nulls = col.Nulls()->As<ColumnUInt8>();
+        if (!nulls) {
+            if (error) {
+                *error = ValidationError("Can't wrap Nullable column: unexpected null-map type");
+            }
+            return nullptr;
+        }
+        return std::make_shared<ColumnNullableT<NestedColumnType>>(nested, nulls);
+    }
+
+    static std::shared_ptr<ColumnNullableT<NestedColumnType>> Wrap(const Column& col, ValidationError* error) {
+        if (auto* c = dynamic_cast<const ColumnNullable*>(&col)) {
+            return Wrap(*c, error);
+        }
+        if (error) {
+            *error = ValidationError("Can't wrap column of type " + col.GetType().GetName() + " as Nullable");
+        }
+        return nullptr;
+    }
+
+    // Helper to simplify integration with other APIs
+    static std::shared_ptr<ColumnNullableT<NestedColumnType>> Wrap(const ColumnRef& col, ValidationError* error) {
+        return Wrap(*col, error);
     }
 
-    static auto Wrap(Column&& col) { return Wrap(std::move(dynamic_cast<ColumnNullable&&>(col))); }
+    static auto Wrap(const ColumnNullable& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    static auto Wrap(const Column& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
     // Helper to simplify integration with other APIs
-    static auto Wrap(ColumnRef&& col) { return Wrap(std::move(*col->AsStrict<ColumnNullable>())); }
+    static auto Wrap(const ColumnRef& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
     ColumnRef Slice(size_t begin, size_t size) const override {
         return Wrap(ColumnNullable::Slice(begin, size));
diff --git a/clickhouse/columns/numeric.cpp b/clickhouse/columns/numeric.cpp
index cc33d19d..2fdf393f 100644
--- a/clickhouse/columns/numeric.cpp
+++ b/clickhouse/columns/numeric.cpp
@@ -1,5 +1,4 @@
 #include "numeric.h"
-#include "utils.h"
 
 #include "../base/wire_format.h"
 
diff --git a/clickhouse/columns/string.cpp b/clickhouse/columns/string.cpp
index 50581eea..022aaa43 100644
--- a/clickhouse/columns/string.cpp
+++ b/clickhouse/columns/string.cpp
@@ -1,5 +1,4 @@
 #include "string.h"
-#include "utils.h"
 
 #include "../base/wire_format.h"
 
diff --git a/clickhouse/columns/tuple.h b/clickhouse/columns/tuple.h
index b6b0bbc7..eb270af1 100644
--- a/clickhouse/columns/tuple.h
+++ b/clickhouse/columns/tuple.h
@@ -1,8 +1,8 @@
 #pragma once
 
 #include "column.h"
-#include "utils.h"
 
+#include <tuple>
 #include <vector>
 
 namespace clickhouse {
@@ -98,27 +98,76 @@ class ColumnTupleT : public ColumnTuple {
         AppendTuple(std::move(value));
     }
 
-    /** Create a ColumnTupleT from a ColumnTuple, without copying data and offsets, but by
-     * 'stealing' those from `col`.
+    /** Create a ColumnTupleT that SHARES the internals of `col` (its element columns)
+     *  via shared_ptr, WITHOUT stealing or copying them.
      *
-     *  Ownership of column internals is transferred to returned object, original (argument) object
-     *  MUST NOT BE USED IN ANY WAY, it is only safe to dispose it.
+     *  The original `col` remains fully valid and usable. Both the original and the
+     *  returned wrapper reference the same underlying element columns, so mutations
+     *  through one are visible through the other.
      *
-     *  Throws an exception if `col` is of wrong type, it is safe to use original col in this case.
-     *  This is a static method to make such conversion verbose.
+     *  The two-argument overloads are non-throwing: on a type mismatch they return
+     *  nullptr and, if `error` is non-null, assign a description to `*error`. The
+     *  single-argument overloads throw ValidationError on a type mismatch instead.
      */
-    static auto Wrap(ColumnTuple&& col) {
+    static std::shared_ptr<ColumnTupleT<Columns...>> Wrap(const ColumnTuple& col, ValidationError* error) {
         if (col.TupleSize() != std::tuple_size_v<TupleOfColumns>) {
-            throw ValidationError("Can't wrap from " + col.GetType().GetName());
+            if (error) {
+                *error = ValidationError("Can't wrap from " + col.GetType().GetName());
+            }
+            return nullptr;
+        }
+        auto columns = TupleFromColumn(col, error);
+        const bool all_wrapped = std::apply(
+            [](const auto&... column) { return (... && static_cast<bool>(column)); }, columns);
+        if (!all_wrapped) {
+            return nullptr;
         }
         auto names = col.Type()->As<TupleType>()->GetItemNames();
-        return std::make_shared<ColumnTupleT<Columns...>>(VectorToTuple(std::move(col)), std::move(names));
+        return std::make_shared<ColumnTupleT<Columns...>>(std::move(columns), std::move(names));
     }
 
-    static auto Wrap(Column&& col) { return Wrap(std::move(dynamic_cast<ColumnTuple&&>(col))); }
+    static std::shared_ptr<ColumnTupleT<Columns...>> Wrap(const Column& col, ValidationError* error) {
+        if (auto* c = dynamic_cast<const ColumnTuple*>(&col)) {
+            return Wrap(*c, error);
+        }
+        if (error) {
+            *error = ValidationError("Can't wrap column of type " + col.GetType().GetName() + " as Tuple");
+        }
+        return nullptr;
+    }
+
+    // Helper to simplify integration with other APIs
+    static std::shared_ptr<ColumnTupleT<Columns...>> Wrap(const ColumnRef& col, ValidationError* error) {
+        return Wrap(*col, error);
+    }
+
+    static auto Wrap(const ColumnTuple& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
+
+    static auto Wrap(const Column& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
     // Helper to simplify integration with other APIs
-    static auto Wrap(ColumnRef&& col) { return Wrap(std::move(*col->AsStrict<ColumnTuple>())); }
+    static auto Wrap(const ColumnRef& col) {
+        ValidationError error;
+        auto result = Wrap(col, &error);
+        if (!result) {
+            throw error;
+        }
+        return result;
+    }
 
     ColumnRef Slice(size_t begin, size_t size) const override {
         return Wrap(ColumnTuple::Slice(begin, size));
@@ -159,6 +208,23 @@ class ColumnTupleT : public ColumnTuple {
         }
     }
 
+    // Builds a tuple of the element columns wrapped as their typed counterparts. Any element
+    // that can't be wrapped is left as a null shared_ptr (and `*error` is set if provided).
+    template <size_t column_index = std::tuple_size_v<TupleOfColumns>>
+    inline static auto TupleFromColumn([[maybe_unused]] const ColumnTuple& col,
+                                       [[maybe_unused]] ValidationError* error) {
+        static_assert(column_index <= std::tuple_size_v<TupleOfColumns>);
+        if constexpr (column_index == 0) {
+            return std::make_tuple();
+        } else {
+            using ColumnType =
+                typename std::tuple_element<column_index - 1, TupleOfColumns>::type::element_type;
+            auto column = WrapColumn<ColumnType>(col[column_index - 1], error);
+            return std::tuple_cat(TupleFromColumn<column_index - 1>(col, error),
+                                  std::make_tuple(std::move(column)));
+        }
+    }
+
     template <typename T, size_t column_index = std::tuple_size_v<TupleOfColumns>>
     inline static auto VectorToTuple([[maybe_unused]] T columns) {
         static_assert(column_index <= std::tuple_size_v<TupleOfColumns>);
diff --git a/clickhouse/columns/utils.h b/clickhouse/columns/utils.h
deleted file mode 100644
index 0fb8b99b..00000000
--- a/clickhouse/columns/utils.h
+++ /dev/null
@@ -1,41 +0,0 @@
-#pragma once
-
-#include <algorithm>
-#include <vector>
-#include <memory>
-
-namespace clickhouse {
-
-template <typename T>
-std::vector<T> SliceVector(const std::vector<T>& vec, size_t begin, size_t len) {
-    std::vector<T> result;
-
-    if (begin < vec.size()) {
-        len = std::min(len, vec.size() - begin);
-        result.assign(vec.begin() + begin, vec.begin() + (begin + len));
-    }
-
-    return result;
-}
-
-template <typename T>
-struct HasWrapMethod {
-private:
-    static int detect(...);
-    template <typename U>
-    static decltype(U::Wrap(std::move(std::declval<ColumnRef>()))) detect(const U&);
-
-public:
-    static constexpr bool value = !std::is_same<int, decltype(detect(std::declval<T>()))>::value;
-};
-
-template <typename T>
-inline std::shared_ptr<T> WrapColumn(ColumnRef&& column) {
-    if constexpr (HasWrapMethod<T>::value) {
-        return T::Wrap(std::move(column));
-    } else {
-        return column->template AsStrict<T>();
-    }
-}
-
-}
diff --git a/clickhouse/columns/uuid.cpp b/clickhouse/columns/uuid.cpp
index fbaff97d..85be389f 100644
--- a/clickhouse/columns/uuid.cpp
+++ b/clickhouse/columns/uuid.cpp
@@ -1,5 +1,4 @@
 #include "uuid.h"
-#include "utils.h"
 #include "../exceptions.h"
 
 #include <stdexcept>
diff --git a/clickhouse/exceptions.h b/clickhouse/exceptions.h
index d2cb639c..00375820 100644
--- a/clickhouse/exceptions.h
+++ b/clickhouse/exceptions.h
@@ -14,6 +14,11 @@ class Error : public std::runtime_error {
 // Caused by any user-related code, like invalid column types or arguments passed to any method.
 class ValidationError : public Error {
     using Error::Error;
+
+public:
+    // Convenience default constructor, useful for creating an empty error to pass as an
+    // output parameter (e.g. to Column*T::Wrap(col, &error)).
+    ValidationError() : Error(std::string()) {}
 };
 
 // Buffers+IO errors, failure to serialize/deserialize, checksum mismatches, etc.
diff --git a/clickhouse/types/types.cpp b/clickhouse/types/types.cpp
index 9b52255f..58318279 100644
--- a/clickhouse/types/types.cpp
+++ b/clickhouse/types/types.cpp
@@ -508,12 +508,18 @@ LowCardinalityType::~LowCardinalityType() {
 // Checks if `name` is a valid plain identifier (must not be quoted).
 // The condition for this is a match against `^[a-zA-Z_][0-9a-zA-Z_]*$`
 static bool IsPlainIdentifier(const std::string& name) {
-    if (name.empty()) return false;
+    if (name.empty()) {
+        return false;
+    }
     auto is_alpha_or_under = [](char c) { return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '_'; };
     auto is_alnum_or_under = [&is_alpha_or_under](char c) { return is_alpha_or_under(c) || (c >= '0' && c <= '9'); };
-    if (!is_alpha_or_under(name[0])) return false;
+    if (!is_alpha_or_under(name[0])) {
+        return false;
+    }
     for (size_t i = 1; i < name.size(); ++i)
-        if (!is_alnum_or_under(name[i])) return false;
+        if (!is_alnum_or_under(name[i])) {
+            return false;
+        }
     return true;
 }
 
diff --git a/ut/CreateColumnByType_ut.cpp b/ut/CreateColumnByType_ut.cpp
index 279a19cc..78225dd0 100644
--- a/ut/CreateColumnByType_ut.cpp
+++ b/ut/CreateColumnByType_ut.cpp
@@ -1,6 +1,7 @@
 #include <clickhouse/columns/bool.h>
 #include <clickhouse/columns/factory.h>
 #include <clickhouse/columns/date.h>
+#include <clickhouse/columns/lowcardinality.h>
 #include <clickhouse/columns/numeric.h>
 #include <clickhouse/columns/string.h>
 #include <clickhouse/columns/tuple.h>
@@ -45,6 +46,28 @@ TEST(CreateColumnByType, LowCardinalityAsWrappedColumn) {
     ASSERT_EQ(Type::FixedString, CreateColumnByType("LowCardinality(FixedString(10000))", create_column_settings)->As<ColumnFixedString>()->GetType().GetCode());
 }
 
+TEST(CreateColumnByType, LowCardinality) {
+    // In the default (non-wrapped) mode, LowCardinality(String)/LowCardinality(FixedString) map to
+    // the base ColumnLowCardinality (like Array/Nullable/Tuple/Map do), and the strongly-typed
+    // ColumnLowCardinalityT<...> view is obtained on demand via the wrapping As<>.
+    {
+        auto col = CreateColumnByType("LowCardinality(String)");
+        ASSERT_NE(nullptr, col);
+        EXPECT_EQ("LowCardinality(String)", col->GetType().GetName());
+        // Concrete type is the base ColumnLowCardinality, not ColumnLowCardinalityT<...>.
+        EXPECT_NE(nullptr, col->As<ColumnLowCardinality>());
+        // The wrapping As<> yields the strongly-typed view.
+        EXPECT_NE(nullptr, col->As<ColumnLowCardinalityT<ColumnString>>());
+    }
+    {
+        auto col = CreateColumnByType("LowCardinality(FixedString(10000))");
+        ASSERT_NE(nullptr, col);
+        EXPECT_EQ("LowCardinality(FixedString(10000))", col->GetType().GetName());
+        EXPECT_NE(nullptr, col->As<ColumnLowCardinality>());
+        EXPECT_NE(nullptr, col->As<ColumnLowCardinalityT<ColumnFixedString>>());
+    }
+}
+
 TEST(CreateColumnByType, DateTime) {
     ASSERT_NE(nullptr, CreateColumnByType("DateTime"));
     ASSERT_NE(nullptr, CreateColumnByType("DateTime('Europe/Moscow')"));
@@ -162,6 +185,8 @@ INSTANTIATE_TEST_SUITE_P(Parametrized, CreateColumnByTypeWithName, ::testing::Va
 
 INSTANTIATE_TEST_SUITE_P(Nested, CreateColumnByTypeWithName, ::testing::Values(
     "Nullable(FixedString(10000))",
+    "LowCardinality(String)",
+    "LowCardinality(FixedString(10000))",
     "Nullable(LowCardinality(FixedString(10000)))",
     "Array(Nullable(LowCardinality(FixedString(10000))))",
     "Array(Enum8('ONE' = 1, 'TWO' = 2))"
diff --git a/ut/client_ut.cpp b/ut/client_ut.cpp
index dff1400e..6dce77d1 100644
--- a/ut/client_ut.cpp
+++ b/ut/client_ut.cpp
@@ -1522,7 +1522,9 @@ TEST_P(ClientCase, InteractiveSelect_Basic) {
 
     std::vector<uint64_t> values;
     while (auto block = client_->NextBlock()) {
-        if (block->GetRowCount() == 0) continue;
+        if (block->GetRowCount() == 0) {
+            continue;
+        }
         auto col = block->At(0)->AsStrict<ColumnUInt64>();
         for (size_t i = 0; i < block->GetRowCount(); ++i) {
             values.push_back(col->At(i));
@@ -1561,7 +1563,9 @@ TEST_P(ClientCase, InteractiveSelect_MultipleBlocks) {
     size_t block_count = 0;
     std::vector<uint64_t> values;
     while (auto block = client_->NextBlock()) {
-        if (block->GetRowCount() == 0) continue;
+        if (block->GetRowCount() == 0) {
+            continue;
+        }
         EXPECT_LE(block->GetRowCount(), 2u);
         block_count++;
         auto col = block->At(0)->AsStrict<ColumnUInt64>();
@@ -1587,7 +1591,9 @@ TEST_P(ClientCase, InteractiveSelect_Cancel) {
     // Consume one block of data, skipping any blocks with 0 rows.
     size_t rows_before_cancel = 0;
     while (auto b = client_->NextBlock()) {
-        if (b->GetRowCount() == 0) continue;
+        if (b->GetRowCount() == 0) {
+            continue;
+        }
         rows_before_cancel = b->GetRowCount();
         break;
     }
diff --git a/ut/column_array_ut.cpp b/ut/column_array_ut.cpp
index 0fc28d4f..04e42ca5 100644
--- a/ut/column_array_ut.cpp
+++ b/ut/column_array_ut.cpp
@@ -337,6 +337,86 @@ TEST(ColumnArrayT, Wrap_UInt64_2D) {
     EXPECT_TRUE(CompareRecursive(values, array));
 }
 
+TEST(ColumnArrayT, Wrap_AcceptsLvalue) {
+    // Wrap no longer requires an rvalue: lvalues and const sources are accepted.
+
+    const std::vector<std::vector<uint64_t>> values = {
+        {1u, 2u},
+        {3u},
+        {}
+    };
+
+    auto arr = CreateArray<ColumnUInt64>(values);
+
+    // Lvalue ColumnRef, no std::move required and not consumed.
+    ColumnRef ref = arr;
+    auto w1 = ColumnArrayT<ColumnUInt64>::Wrap(ref);
+    EXPECT_TRUE(CompareRecursive(values, *w1));
+    EXPECT_NE(ref, nullptr);
+
+    // Const lvalue concrete column.
+    const ColumnArray& cref = *arr;
+    auto w2 = ColumnArrayT<ColumnUInt64>::Wrap(cref);
+    EXPECT_TRUE(CompareRecursive(values, *w2));
+
+    // Non-const lvalue concrete column.
+    auto w3 = ColumnArrayT<ColumnUInt64>::Wrap(*arr);
+    EXPECT_TRUE(CompareRecursive(values, *w3));
+}
+
+TEST(ColumnArrayT, Wrap_DoesNotStealSource_UInt64) {
+    // Wrap shares storage with the source ColumnArray and leaves its contents intact.
+
+    const std::vector<std::vector<uint64_t>> values = {
+        {1u, 2u, 3u},
+        {4u, 5u, 6u, 7u, 8u, 9u},
+        {0u},
+        {},
+        {13, 14}
+    };
+
+    auto original = CreateArray<ColumnUInt64>(values);
+    // Keep an independent handle to the same underlying ColumnArray.
+    auto keep = original;
+    auto wrapped_array = ColumnArrayT<ColumnUInt64>::Wrap(std::move(original));
+
+    // Wrapper sees the same data.
+    EXPECT_TRUE(CompareRecursive(values, *wrapped_array));
+
+    // Source array contents are left intact (not stolen from).
+    ASSERT_NE(keep, nullptr);
+    EXPECT_EQ(keep->Size(), values.size());
+
+    // Storage is shared: appending a row through the source is visible via the wrapper.
+    keep->AppendAsColumn(std::make_shared<ColumnUInt64>(std::vector<uint64_t>{42, 43}));
+    EXPECT_EQ(wrapped_array->Size(), values.size() + 1);
+    EXPECT_EQ(wrapped_array->At(values.size()).At(0), 42u);
+    EXPECT_EQ(wrapped_array->At(values.size()).At(1), 43u);
+}
+
+TEST(ColumnArrayT, Wrap_DoesNotStealSource_UInt64_2D) {
+    // Wrap shares all nesting layers with the source.
+
+    const std::vector<std::vector<std::vector<uint64_t>>> values = {
+        {{1u, 2u}, {3u}},
+        {{4u}, {5u, 6u, 7u}, {8u, 9u}, {}},
+        {{0u}},
+        {{}},
+        {{13}, {14, 15}}
+    };
+
+    auto original = Create2DArray<ColumnUInt64>(values);
+    auto keep = original;
+    auto wrapped_array = ColumnArrayT<ColumnArrayT<ColumnUInt64>>::Wrap(std::move(original));
+
+    EXPECT_TRUE(CompareRecursive(values, *wrapped_array));
+
+    // Source array contents are left intact (not stolen from).
+    ASSERT_NE(keep, nullptr);
+    EXPECT_EQ(keep->Size(), values.size());
+    EXPECT_TRUE(CompareRecursive(values, *ColumnArrayT<ColumnArrayT<ColumnUInt64>>::Wrap(std::move(keep))));
+}
+
 TEST(ColumnArrayT, Bool) {
     // Check inserting\reading back data from clickhouse::ColumnArrayT<ColumnBool>
 
diff --git a/ut/columns_ut.cpp b/ut/columns_ut.cpp
index d79debbe..3f7ca5e1 100644
--- a/ut/columns_ut.cpp
+++ b/ut/columns_ut.cpp
@@ -1088,6 +1088,69 @@ TEST(ColumnsCase, ColumnLowCardinalityString_Append_and_Read) {
     }
 }
 
+TEST(ColumnsCase, ColumnLowCardinalityT_Wrap_DoesNotStealSource) {
+    // Populate via the typed column (the only ergonomic per-value insert path), then Wrap it
+    // through an untyped ColumnRef handle, as with a column received from a query.
+    auto source = std::make_shared<ColumnLowCardinalityT<ColumnString>>();
+    source->Append("a");
+    source->Append("b");
+    source->Append("a");
+
+    ColumnRef untyped = source;
+    auto wrapped = ColumnLowCardinalityT<ColumnString>::Wrap(untyped);
+
+    // Wrapper reads the same data.
+    ASSERT_EQ(wrapped->Size(), 3u);
+    EXPECT_EQ(wrapped->At(0), "a");
+    EXPECT_EQ(wrapped->At(1), "b");
+    EXPECT_EQ(wrapped->At(2), "a");
+
+    // Source (and the untyped handle) are left intact after Wrap (non-stealing).
+    EXPECT_NE(untyped, nullptr);
+    ASSERT_EQ(source->Size(), 3u);
+    EXPECT_EQ(source->At(0), "a");
+    EXPECT_EQ(source->At(2), "a");
+
+    // Storage (dictionary + index + dedup map) is shared and stays coherent:
+    // a new unique value appended via the source, and a repeat appended via the wrapper.
+    const auto dict_before = source->GetDictionarySize();
+    source->Append("c");     // new unique -> dictionary grows, visible via the wrapper
+    wrapped->Append("a");    // repeat -> deduped against the shared map, no dictionary growth
+
+    EXPECT_EQ(source->Size(), 5u);
+    EXPECT_EQ(wrapped->Size(), 5u);
+    EXPECT_EQ(wrapped->At(3), "c");
+    EXPECT_EQ(wrapped->At(4), "a");
+    EXPECT_EQ(source->At(4), "a");
+    // "c" added exactly one dictionary entry; "a" added none (shared dedup map).
+    EXPECT_EQ(source->GetDictionarySize(), dict_before + 1);
+    EXPECT_EQ(wrapped->GetDictionarySize(), dict_before + 1);
+}
+
+TEST(ColumnsCase, ColumnLowCardinalityT_Wrap_AcceptsLvalue) {
+    auto source = std::make_shared<ColumnLowCardinalityT<ColumnString>>();
+    source->Append("x");
+    source->Append("y");
+
+    using LC = ColumnLowCardinalityT<ColumnString>;
+
+    // Non-const lvalue (untyped base reference), no std::move required.
+    ColumnLowCardinality& base = *source;
+    auto w1 = LC::Wrap(base);
+    EXPECT_EQ(w1->At(0), "x");
+
+    // Const lvalue.
+    const ColumnLowCardinality& cref = *source;
+    auto w2 = LC::Wrap(cref);
+    EXPECT_EQ(w2->At(1), "y");
+
+    // Lvalue ColumnRef, no std::move required and not consumed.
+    ColumnRef ref = source;
+    auto w3 = LC::Wrap(ref);
+    EXPECT_EQ(w3->Size(), 2u);
+    EXPECT_NE(ref, nullptr);
+}
+
 TEST(ColumnsCase, ColumnLowCardinalityString_Clear_and_Append) {
     const size_t items_count = 11;
     ColumnLowCardinalityT<ColumnString> col;
@@ -1254,6 +1317,61 @@ TEST(ColumnsCase, ColumnTupleT) {
     EXPECT_EQ(val, col.At(0));
 }
 
+TEST(ColumnsCase, ColumnNullableT_Wrap_DoesNotStealSource) {
+    auto nested = std::make_shared<ColumnUInt64>();
+    auto nulls = std::make_shared<ColumnUInt8>();
+    ColumnNullable col(nested, nulls);
+
+    col.Append(false);
+    nested->Append(1);
+    col.Append(true);
+    nested->Append(0);
+
+    using TestNullable = ColumnNullableT<ColumnUInt64>;
+    auto wrapped = TestNullable::Wrap(std::move(col));
+
+    // Wrapper sees the same data.
+    EXPECT_EQ(wrapped->Size(), 2u);
+    EXPECT_EQ(wrapped->At(0), std::optional<uint64_t>(1));
+    EXPECT_EQ(wrapped->At(1), std::optional<uint64_t>{});
+
+    // Source column is left intact after Wrap (non-stealing).
+    EXPECT_EQ(col.Size(), 2u);
+    EXPECT_FALSE(col.IsNull(0));
+    EXPECT_TRUE(col.IsNull(1));
+
+    // Storage is shared: appending through the original is visible via the wrapper.
+    col.Append(false);
+    nested->Append(42);
+    EXPECT_EQ(wrapped->Size(), 3u);
+    EXPECT_EQ(wrapped->At(2), std::optional<uint64_t>(42));
+}
+
+TEST(ColumnsCase, ColumnNullableT_Wrap_AcceptsLvalue) {
+    auto nested = std::make_shared<ColumnUInt64>();
+    auto nulls = std::make_shared<ColumnUInt8>();
+    ColumnNullable col(nested, nulls);
+    col.Append(false);
+    nested->Append(7);
+
+    using TestNullable = ColumnNullableT<ColumnUInt64>;
+
+    // Non-const lvalue concrete column, no std::move required.
+    auto w1 = TestNullable::Wrap(col);
+    EXPECT_EQ(w1->At(0), std::optional<uint64_t>(7));
+
+    // Const lvalue concrete column.
+    const ColumnNullable& cref = col;
+    auto w2 = TestNullable::Wrap(cref);
+    EXPECT_EQ(w2->At(0), std::optional<uint64_t>(7));
+
+    // Lvalue ColumnRef, no std::move required and not consumed.
+    ColumnRef ref = std::make_shared<ColumnNullable>(nested, nulls);
+    auto w3 = TestNullable::Wrap(ref);
+    EXPECT_EQ(w3->At(0), std::optional<uint64_t>(7));
+    EXPECT_NE(ref, nullptr);
+}
+
 TEST(ColumnsCase, ColumnTupleT_Wrap) {
     ColumnTuple col ({
             std::make_shared<ColumnUInt64>(),
@@ -1275,6 +1393,84 @@ TEST(ColumnsCase, ColumnTupleT_Wrap) {
     EXPECT_EQ(val, wrapped_col->At(0));
 }
 
+TEST(ColumnsCase, ColumnTupleT_Wrap_DoesNotStealSource) {
+    ColumnTuple col ({
+            std::make_shared<ColumnUInt64>(),
+            std::make_shared<ColumnString>(),
+            std::make_shared<ColumnFixedString>(3)
+        }
+    );
+
+    const auto val = std::make_tuple(1, "a", "bcd");
+
+    col[0]->AsStrict<ColumnUInt64>()->Append(std::get<0>(val));
+    col[1]->AsStrict<ColumnString>()->Append(std::get<1>(val));
+    col[2]->AsStrict<ColumnFixedString>()->Append(std::get<2>(val));
+
+    using TestTuple = ColumnTupleT<ColumnUInt64, ColumnString, ColumnFixedString>;
+    auto wrapped = TestTuple::Wrap(std::move(col));
+
+    // Wrapper sees the same data.
+    EXPECT_EQ(wrapped->Size(), 1u);
+    EXPECT_EQ(val, wrapped->At(0));
+
+    // Source column is left intact after Wrap (non-stealing).
+    EXPECT_EQ(col.TupleSize(), 3u);
+    EXPECT_EQ(col.Size(), 1u);
+
+    // Storage is shared: appending through the original element columns is visible via the wrapper.
+    col[0]->AsStrict<ColumnUInt64>()->Append(2);
+    col[1]->AsStrict<ColumnString>()->Append("xy");
+    col[2]->AsStrict<ColumnFixedString>()->Append("zzz");
+    EXPECT_EQ(wrapped->Size(), 2u);
+    EXPECT_EQ(std::make_tuple(2, "xy", "zzz"), wrapped->At(1));
+}
+
+TEST(ColumnsCase, ColumnTupleT_Wrap_DoesNotStealSource_PreservesNames) {
+    ColumnTuple base(
+        {std::make_shared<ColumnUInt64>(), std::make_shared<ColumnString>()},
+        {"id", "name"}
+    );
+
+    using TestTuple = ColumnTupleT<ColumnUInt64, ColumnString>;
+    auto wrapped = TestTuple::Wrap(std::move(base));
+    EXPECT_EQ(wrapped->Type()->GetName(), "Tuple(id UInt64, name String)");
+
+    // Source remains usable after Wrap (non-stealing).
+    EXPECT_EQ(base.Type()->GetName(), "Tuple(id UInt64, name String)");
+    EXPECT_EQ(base.TupleSize(), 2u);
+}
+
+TEST(ColumnsCase, ColumnTupleT_Wrap_AcceptsLvalue) {
+    ColumnTuple col({
+        std::make_shared<ColumnUInt64>(),
+        std::make_shared<ColumnString>()
+    });
+    col[0]->AsStrict<ColumnUInt64>()->Append(1);
+    col[1]->AsStrict<ColumnString>()->Append("a");
+
+    using TestTuple = ColumnTupleT<ColumnUInt64, ColumnString>;
+
+    // Non-const lvalue concrete column, no std::move required.
+    auto w1 = TestTuple::Wrap(col);
+    EXPECT_EQ(w1->At(0), std::make_tuple(uint64_t(1), std::string_view("a")));
+
+    // Const lvalue concrete column.
+    const ColumnTuple& cref = col;
+    auto w2 = TestTuple::Wrap(cref);
+    EXPECT_EQ(w2->At(0), std::make_tuple(uint64_t(1), std::string_view("a")));
+
+    // Lvalue ColumnRef, no std::move required and not consumed.
+    ColumnRef ref = std::make_shared<ColumnTuple>(std::vector<ColumnRef>{col[0], col[1]});
+    auto w3 = TestTuple::Wrap(ref);
+    EXPECT_EQ(w3->At(0), std::make_tuple(uint64_t(1), std::string_view("a")));
+    EXPECT_NE(ref, nullptr);
+
+    // Source column left intact.
+    EXPECT_EQ(col.TupleSize(), 2u);
+    EXPECT_EQ(col.Size(), 1u);
+}
+
 TEST(ColumnsCase, ColumnTupleT_Empty) {
     using TestTuple = ColumnTupleT<>;
 
@@ -1381,3 +1577,179 @@ TEST(ColumnsCase, ColumnMapT_Wrap) {
     EXPECT_EQ("123", map_view.At(1));
     EXPECT_EQ("abc", map_view.At(2));
 }
+
+TEST(ColumnsCase, ColumnMapT_Wrap_AcceptsLvalue) {
+    auto tupls = std::make_shared<ColumnTuple>(std::vector<ColumnRef>{
+            std::make_shared<ColumnUInt64>(),
+            std::make_shared<ColumnString>()});
+
+    auto data = std::make_shared<ColumnArray>(tupls);
+
+    auto val = tupls->CloneEmpty()->As<ColumnTuple>();
+    (*val)[0]->AsStrict<ColumnUInt64>()->Append(1);
+    (*val)[1]->AsStrict<ColumnString>()->Append("123");
+    data->AppendAsColumn(val);
+
+    ColumnMap col{data};
+
+    using TestMap = ColumnMapT<ColumnUInt64, ColumnString>;
+
+    // Non-const lvalue concrete column, no std::move required.
+    auto w1 = TestMap::Wrap(col);
+    EXPECT_EQ("123", w1->At(0).At(1));
+
+    // Const lvalue concrete column.
+    const ColumnMap& cref = col;
+    auto w2 = TestMap::Wrap(cref);
+    EXPECT_EQ("123", w2->At(0).At(1));
+
+    // Lvalue ColumnRef, no std::move required and not consumed.
+    ColumnRef ref = std::make_shared<ColumnMap>(data);
+    auto w3 = TestMap::Wrap(ref);
+    EXPECT_EQ("123", w3->At(0).At(1));
+    EXPECT_NE(ref, nullptr);
+
+    // Source column left intact.
+    EXPECT_EQ(col.Size(), 1u);
+}
+
+TEST(ColumnsCase, ColumnMapT_Wrap_DoesNotStealSource) {
+    auto tupls = std::make_shared<ColumnTuple>(std::vector<ColumnRef>{
+            std::make_shared<ColumnUInt64>(),
+            std::make_shared<ColumnString>()});
+
+    auto data = std::make_shared<ColumnArray>(tupls);
+
+    auto val = tupls->CloneEmpty()->As<ColumnTuple>();
+
+    (*val)[0]->AsStrict<ColumnUInt64>()->Append(1);
+    (*val)[1]->AsStrict<ColumnString>()->Append("123");
+
+    (*val)[0]->AsStrict<ColumnUInt64>()->Append(2);
+    (*val)[1]->AsStrict<ColumnString>()->Append("abc");
+
+    data->AppendAsColumn(val);
+
+    ColumnMap col{data};
+
+    using TestMap = ColumnMapT<ColumnUInt64, ColumnString>;
+    auto wrapped_col = TestMap::Wrap(std::move(col));
+
+    // Wrapper sees the same data.
+    auto map_view = wrapped_col->At(0);
+    EXPECT_THROW(map_view.At(0), ValidationError);
+    EXPECT_EQ("123", map_view.At(1));
+    EXPECT_EQ("abc", map_view.At(2));
+
+    // Source column is left intact after Wrap (non-stealing).
+    EXPECT_EQ(col.Size(), 1u);
+
+    // Storage is shared: appending a row through the original is visible via the wrapper.
+    auto val2 = tupls->CloneEmpty()->As<ColumnTuple>();
+    (*val2)[0]->AsStrict<ColumnUInt64>()->Append(7);
+    (*val2)[1]->AsStrict<ColumnString>()->Append("xyz");
+    data->AppendAsColumn(val2);
+
+    EXPECT_EQ(wrapped_col->Size(), 2u);
+    EXPECT_EQ("xyz", wrapped_col->At(1).At(7));
+}
+
+// --- Wrap error-reporting overloads ---------------------------------------------------------
+// The two-argument Wrap(col, ValidationError*) returns nullptr (never throws) on a type
+// mismatch; the single-argument Wrap(col) throws ValidationError on the same mismatch.
+
+TEST(ColumnsCase, ColumnArrayT_Wrap_TypeMismatch) {
+    using TestArray = ColumnArrayT<ColumnUInt64>;
+
+    // Right kind (Array), wrong element type (String instead of UInt64).
+    ColumnRef bad_element = std::make_shared<ColumnArray>(std::make_shared<ColumnString>());
+    // Wrong kind entirely.
+    ColumnRef not_array = std::make_shared<ColumnUInt64>();
+
+    ValidationError error;
+    EXPECT_NO_THROW({
+        EXPECT_EQ(TestArray::Wrap(bad_element, &error), nullptr);
+    });
+    EXPECT_FALSE(std::string_view(error.what()).empty());
+
+    // Passing nullptr for the error is allowed and still non-throwing.
+    EXPECT_NO_THROW({
+        EXPECT_EQ(TestArray::Wrap(not_array, nullptr), nullptr);
+    });
+
+    // Single-argument overload throws on the same mismatches.
+    EXPECT_THROW(TestArray::Wrap(bad_element), ValidationError);
+    EXPECT_THROW(TestArray::Wrap(not_array), ValidationError);
+
+    // Sanity: a matching column wraps fine through both overloads.
+    ColumnRef good = std::make_shared<ColumnArray>(std::make_shared<ColumnUInt64>());
+    EXPECT_NE(TestArray::Wrap(good, nullptr), nullptr);
+    EXPECT_NE(TestArray::Wrap(good), nullptr);
+}
+
+TEST(ColumnsCase, ColumnNullableT_Wrap_TypeMismatch) {
+    using TestNullable = ColumnNullableT<ColumnUInt64>;
+
+    // Nullable of the wrong nested type.
+    ColumnRef bad_nested = std::make_shared<ColumnNullable>(
+        std::make_shared<ColumnString>(), std::make_shared<ColumnUInt8>());
+    ColumnRef not_nullable = std::make_shared<ColumnUInt64>();
+
+    ValidationError error;
+    EXPECT_EQ(TestNullable::Wrap(bad_nested, &error), nullptr);
+    EXPECT_FALSE(std::string_view(error.what()).empty());
+    EXPECT_EQ(TestNullable::Wrap(not_nullable, nullptr), nullptr);
+
+    EXPECT_THROW(TestNullable::Wrap(bad_nested), ValidationError);
+    EXPECT_THROW(TestNullable::Wrap(not_nullable), ValidationError);
+}
+
+TEST(ColumnsCase, ColumnTupleT_Wrap_TypeMismatch) {
+    using TestTuple = ColumnTupleT<ColumnUInt64, ColumnString>;
+
+    // Correct arity, wrong element type.
+    ColumnRef bad_element = std::make_shared<ColumnTuple>(std::vector<ColumnRef>{
+        std::make_shared<ColumnUInt64>(), std::make_shared<ColumnUInt64>()});
+    // Wrong arity.
+    ColumnRef bad_arity = std::make_shared<ColumnTuple>(std::vector<ColumnRef>{
+        std::make_shared<ColumnUInt64>()});
+    // Wrong kind.
+    ColumnRef not_tuple = std::make_shared<ColumnUInt64>();
+
+    ValidationError error;
+    EXPECT_EQ(TestTuple::Wrap(bad_element, &error), nullptr);
+    EXPECT_FALSE(std::string_view(error.what()).empty());
+    EXPECT_EQ(TestTuple::Wrap(bad_arity, nullptr), nullptr);
+    EXPECT_EQ(TestTuple::Wrap(not_tuple, nullptr), nullptr);
+
+    EXPECT_THROW(TestTuple::Wrap(bad_element), ValidationError);
+    EXPECT_THROW(TestTuple::Wrap(bad_arity), ValidationError);
+    EXPECT_THROW(TestTuple::Wrap(not_tuple), ValidationError);
+}
+
+TEST(ColumnsCase, ColumnMapT_Wrap_TypeMismatch) {
+    using TestMap = ColumnMapT<ColumnUInt64, ColumnString>;
+    ColumnRef not_map = std::make_shared<ColumnUInt64>();
+
+    ValidationError error;
+    EXPECT_EQ(TestMap::Wrap(not_map, &error), nullptr);
+    EXPECT_FALSE(std::string_view(error.what()).empty());
+    EXPECT_EQ(TestMap::Wrap(not_map, nullptr), nullptr);
+    EXPECT_THROW(TestMap::Wrap(not_map), ValidationError);
+}
+
+TEST(ColumnsCase, ColumnLowCardinalityT_Wrap_TypeMismatch) {
+    using TestLC = ColumnLowCardinalityT<ColumnString>;
+
+    // LowCardinality with the wrong (but valid) dictionary type.
+    ColumnRef bad_dict = std::make_shared<ColumnLowCardinality>(std::make_shared<ColumnFixedString>(4));
+    ColumnRef not_lc = std::make_shared<ColumnUInt64>();
+
+    ValidationError error;
+    EXPECT_EQ(TestLC::Wrap(bad_dict, &error), nullptr);
+    EXPECT_FALSE(std::string_view(error.what()).empty());
+    EXPECT_EQ(TestLC::Wrap(not_lc, nullptr), nullptr);
+
+    EXPECT_THROW(TestLC::Wrap(bad_dict), ValidationError);
+    EXPECT_THROW(TestLC::Wrap(not_lc), ValidationError);
+}