diff --git a/clickhouse/CMakeLists.txt b/clickhouse/CMakeLists.txt
index e2afb3ae..6a851241 100644
--- a/clickhouse/CMakeLists.txt
+++ b/clickhouse/CMakeLists.txt
@@ -7,6 +7,7 @@ SET ( clickhouse-cpp-lib-src
     base/wire_format.cpp
 
     columns/array.cpp
+    columns/column.cpp
     columns/date.cpp
     columns/decimal.cpp
     columns/enum.cpp
diff --git a/clickhouse/columns/array.cpp b/clickhouse/columns/array.cpp
index c71684d1..983000dd 100644
--- a/clickhouse/columns/array.cpp
+++ b/clickhouse/columns/array.cpp
@@ -54,22 +54,34 @@ void ColumnArray::Append(ColumnRef column) {
     }
 }
 
-bool ColumnArray::Load(InputStream* input, size_t rows) {
+bool ColumnArray::LoadPrefix(InputStream* input, size_t rows) {
     if (!rows) {
         return true;
     }
-    if (!offsets_->Load(input, rows)) {
+
+    return data_->LoadPrefix(input, rows);
+}
+
+bool ColumnArray::LoadBody(InputStream* input, size_t rows) {
+    if (!rows) {
+        return true;
+    }
+    if (!offsets_->LoadBody(input, rows)) {
         return false;
     }
-    if (!data_->Load(input, (*offsets_)[rows - 1])) {
+    if (!data_->LoadBody(input, (*offsets_)[rows - 1])) {
         return false;
     }
     return true;
 }
 
-void ColumnArray::Save(OutputStream* output) {
-    offsets_->Save(output);
-    data_->Save(output);
+void ColumnArray::SavePrefix(OutputStream* output) {
+    data_->SavePrefix(output);
+}
+
+void ColumnArray::SaveBody(OutputStream* output) {
+    offsets_->SaveBody(output);
+    data_->SaveBody(output);
 }
 
 void ColumnArray::Clear() {
@@ -92,6 +104,7 @@ void ColumnArray::OffsetsIncrease(size_t n) {
 }
 
 size_t ColumnArray::GetOffset(size_t n) const {
+
     return (n == 0) ? 0 : (*offsets_)[n - 1];
 }
 
diff --git a/clickhouse/columns/array.h b/clickhouse/columns/array.h
index e96e70c4..cfe3eb4b 100644
--- a/clickhouse/columns/array.h
+++ b/clickhouse/columns/array.h
@@ -24,11 +24,17 @@ class ColumnArray : public Column {
     /// Appends content of given column to the end of current one.
     void Append(ColumnRef column) override;
 
+    /// Loads column prefix from input stream.
+    bool LoadPrefix(InputStream* input, size_t rows) override;
+
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
+
+    /// Saves column prefix to output stream.
+    void SavePrefix(OutputStream* output) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/column.cpp b/clickhouse/columns/column.cpp
new file mode 100644
index 00000000..7f881d7c
--- /dev/null
+++ b/clickhouse/columns/column.cpp
@@ -0,0 +1,24 @@
+#include "column.h"
+
+namespace clickhouse {
+
+bool Column::LoadPrefix(InputStream*, size_t) {
+    /// does nothing by default
+    return true;
+}
+
+bool Column::Load(InputStream* input, size_t rows) {
+    return LoadPrefix(input, rows) && LoadBody(input, rows);
+}
+
+void Column::SavePrefix(OutputStream*) {
+    /// does nothing by default
+}
+
+/// Saves column data to output stream.
+void Column::Save(OutputStream* output) {
+    SavePrefix(output);
+    SaveBody(output);
+}
+
+}
diff --git a/clickhouse/columns/column.h b/clickhouse/columns/column.h
index 6db5b39a..8a89134d 100644
--- a/clickhouse/columns/column.h
+++ b/clickhouse/columns/column.h
@@ -42,11 +42,27 @@ class Column : public std::enable_shared_from_this<Column> {
     /// Appends content of given column to the end of current one.
     virtual void Append(ColumnRef column) = 0;
 
+    /// Template method to load column data from input stream. It'll call LoadPrefix and LoadBody.
+    /// Should be called only once from the client. Derived classes should not call it.
+    bool Load(InputStream* input, size_t rows);
+
+    /// Loads column prefix from input stream.
+    virtual bool LoadPrefix(InputStream* input, size_t rows);
+
     /// Loads column data from input stream.
-    virtual bool Load(InputStream* input, size_t rows) = 0;
+    virtual bool LoadBody(InputStream* input, size_t rows) = 0;
+
+    /// Saves column prefix to output stream. Column types with prefixes must implement it.
+    virtual void SavePrefix(OutputStream* output);
+
+    /// Saves column body to output stream.
+    virtual void SaveBody(OutputStream* output) = 0;
 
-    /// Saves column data to output stream.
-    virtual void Save(OutputStream* output) = 0;
+    /// Template method to save to output stream. It'll call SavePrefix and SaveBody respectively
+    /// Should be called only once from the client. Derived classes should not call it.
+    /// Save is split in Prefix and Body because some data types require prefixes and specific serialization order.
+    /// For instance, Array(LowCardinality(X)) requires LowCardinality.key_version bytes to come before Array.offsets
+    void Save(OutputStream* output);
 
     /// Clear column data .
     virtual void Clear() = 0;
diff --git a/clickhouse/columns/date.cpp b/clickhouse/columns/date.cpp
index 78a86be1..6e3c93d4 100644
--- a/clickhouse/columns/date.cpp
+++ b/clickhouse/columns/date.cpp
@@ -27,12 +27,12 @@ void ColumnDate::Append(ColumnRef column) {
     }
 }
 
-bool ColumnDate::Load(InputStream* input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnDate::LoadBody(InputStream* input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnDate::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnDate::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 size_t ColumnDate::Size() const {
@@ -89,12 +89,12 @@ void ColumnDateTime::Append(ColumnRef column) {
     }
 }
 
-bool ColumnDateTime::Load(InputStream* input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnDateTime::LoadBody(InputStream* input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnDateTime::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnDateTime::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 size_t ColumnDateTime::Size() const {
@@ -162,12 +162,12 @@ void ColumnDateTime64::Append(ColumnRef column) {
     }
 }
 
-bool ColumnDateTime64::Load(InputStream* input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnDateTime64::LoadBody(InputStream* input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnDateTime64::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnDateTime64::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 void ColumnDateTime64::Clear() {
diff --git a/clickhouse/columns/date.h b/clickhouse/columns/date.h
index 62ca4e05..d7841489 100644
--- a/clickhouse/columns/date.h
+++ b/clickhouse/columns/date.h
@@ -26,10 +26,10 @@ class ColumnDate : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
@@ -70,13 +70,13 @@ class ColumnDateTime : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Clear column data .
     void Clear() override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Returns count of rows in the column.
     size_t Size() const override;
@@ -118,13 +118,13 @@ class ColumnDateTime64 : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Clear column data .
     void Clear() override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Returns count of rows in the column.
     size_t Size() const override;
diff --git a/clickhouse/columns/decimal.cpp b/clickhouse/columns/decimal.cpp
index 8a4186a0..ad77a6c1 100644
--- a/clickhouse/columns/decimal.cpp
+++ b/clickhouse/columns/decimal.cpp
@@ -197,12 +197,12 @@ void ColumnDecimal::Append(ColumnRef column) {
     }
 }
 
-bool ColumnDecimal::Load(InputStream * input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnDecimal::LoadBody(InputStream * input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnDecimal::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnDecimal::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 void ColumnDecimal::Clear() {
diff --git a/clickhouse/columns/decimal.h b/clickhouse/columns/decimal.h
index b28699ae..825cf5e6 100644
--- a/clickhouse/columns/decimal.h
+++ b/clickhouse/columns/decimal.h
@@ -21,8 +21,8 @@ class ColumnDecimal : public Column {
 
 public:
     void Append(ColumnRef column) override;
-    bool Load(InputStream* input, size_t rows) override;
-    void Save(OutputStream* output) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
+    void SaveBody(OutputStream* output) override;
     void Clear() override;
     size_t Size() const override;
     ColumnRef Slice(size_t begin, size_t len) const override;
diff --git a/clickhouse/columns/enum.cpp b/clickhouse/columns/enum.cpp
index fc07e629..1192e479 100644
--- a/clickhouse/columns/enum.cpp
+++ b/clickhouse/columns/enum.cpp
@@ -74,13 +74,13 @@ void ColumnEnum<T>::Append(ColumnRef column) {
 }
 
 template <typename T>
-bool ColumnEnum<T>::Load(InputStream* input, size_t rows) {
+bool ColumnEnum<T>::LoadBody(InputStream* input, size_t rows) {
     data_.resize(rows);
     return WireFormat::ReadBytes(*input, data_.data(), data_.size() * sizeof(T));
 }
 
 template <typename T>
-void ColumnEnum<T>::Save(OutputStream* output) {
+void ColumnEnum<T>::SaveBody(OutputStream* output) {
     WireFormat::WriteBytes(*output, data_.data(), data_.size() * sizeof(T));
 }
 
diff --git a/clickhouse/columns/enum.h b/clickhouse/columns/enum.h
index 34c672f6..892a818a 100644
--- a/clickhouse/columns/enum.h
+++ b/clickhouse/columns/enum.h
@@ -33,11 +33,11 @@ class ColumnEnum : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
-    
+    void SaveBody(OutputStream* output) override;
+
     /// Clear column data .
     void Clear() override;
 
diff --git a/clickhouse/columns/ip4.cpp b/clickhouse/columns/ip4.cpp
index b2327e86..82b22b04 100644
--- a/clickhouse/columns/ip4.cpp
+++ b/clickhouse/columns/ip4.cpp
@@ -71,12 +71,12 @@ void ColumnIPv4::Append(ColumnRef column) {
     }
 }
 
-bool ColumnIPv4::Load(InputStream * input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnIPv4::LoadBody(InputStream * input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnIPv4::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnIPv4::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 size_t ColumnIPv4::Size() const {
diff --git a/clickhouse/columns/ip4.h b/clickhouse/columns/ip4.h
index 0b1d41ab..713e9954 100644
--- a/clickhouse/columns/ip4.h
+++ b/clickhouse/columns/ip4.h
@@ -41,10 +41,10 @@ class ColumnIPv4 : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/ip6.cpp b/clickhouse/columns/ip6.cpp
index 1d9f14db..e92b23db 100644
--- a/clickhouse/columns/ip6.cpp
+++ b/clickhouse/columns/ip6.cpp
@@ -71,12 +71,12 @@ void ColumnIPv6::Append(ColumnRef column) {
     }
 }
 
-bool ColumnIPv6::Load(InputStream* input, size_t rows) {
-    return data_->Load(input, rows);
+bool ColumnIPv6::LoadBody(InputStream* input, size_t rows) {
+    return data_->LoadBody(input, rows);
 }
 
-void ColumnIPv6::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnIPv6::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 size_t ColumnIPv6::Size() const {
diff --git a/clickhouse/columns/ip6.h b/clickhouse/columns/ip6.h
index e523912a..e654dff3 100644
--- a/clickhouse/columns/ip6.h
+++ b/clickhouse/columns/ip6.h
@@ -39,10 +39,10 @@ class ColumnIPv6 : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/lowcardinality.cpp b/clickhouse/columns/lowcardinality.cpp
index 694e3eda..7e499131 100644
--- a/clickhouse/columns/lowcardinality.cpp
+++ b/clickhouse/columns/lowcardinality.cpp
@@ -200,15 +200,6 @@ auto Load(ColumnRef new_dictionary_column, InputStream& input, size_t rows) {
     // As for now those features are not used in client-server protocol and minimal implementation suffices,
     // however some day they may.
 
-    // prefix
-    uint64_t key_version;
-    if (!WireFormat::ReadFixed(input, &key_version))
-        throw ProtocolError("Failed to read key serialization version.");
-
-    if (key_version != KeySerializationVersion::SharedDictionariesWithAdditionalKeys)
-        throw ProtocolError("Invalid key serialization version value.");
-
-    // body
     uint64_t index_serialization_type;
     if (!WireFormat::ReadFixed(input, &index_serialization_type))
         throw ProtocolError("Failed to read index serializaton type.");
@@ -224,7 +215,7 @@ auto Load(ColumnRef new_dictionary_column, InputStream& input, size_t rows) {
     if (!WireFormat::ReadFixed(input, &number_of_keys))
         throw ProtocolError("Failed to read number of rows in dictionary column.");
 
-    if (!new_dictionary_column->Load(&input, number_of_keys))
+    if (!new_dictionary_column->LoadBody(&input, number_of_keys))
         throw ProtocolError("Failed to read values of dictionary column.");
 
     uint64_t number_of_rows;
@@ -234,7 +225,7 @@ auto Load(ColumnRef new_dictionary_column, InputStream& input, size_t rows) {
     if (number_of_rows != rows)
         throw AssertionError("LowCardinality column must be read in full.");
 
-    new_index_column->Load(&input, number_of_rows);
+    new_index_column->LoadBody(&input, number_of_rows);
 
     ColumnLowCardinality::UniqueItems new_unique_items_map;
     for (size_t i = 0; i < new_dictionary_column->Size(); ++i) {
@@ -250,7 +241,19 @@ auto Load(ColumnRef new_dictionary_column, InputStream& input, size_t rows) {
 
 }
 
-bool ColumnLowCardinality::Load(InputStream* input, size_t rows) {
+bool ColumnLowCardinality::LoadPrefix(InputStream* input, size_t) {
+    uint64_t key_version;
+
+    if (!WireFormat::ReadFixed(*input, &key_version))
+        throw ProtocolError("Failed to read key serialization version.");
+
+    if (key_version != KeySerializationVersion::SharedDictionariesWithAdditionalKeys)
+        throw ProtocolError("Invalid key serialization version value.");
+
+    return true;
+}
+
+bool ColumnLowCardinality::LoadBody(InputStream* input, size_t rows) {
     try {
         auto [new_dictionary, new_index, new_unique_items_map] = ::Load(dictionary_column_->Slice(0, 0), *input, rows);
 
@@ -264,25 +267,22 @@ bool ColumnLowCardinality::Load(InputStream* input, size_t rows) {
     }
 }
 
-void ColumnLowCardinality::Save(OutputStream* output) {
-    // prefix
-    const uint64_t version = static_cast<uint64_t>(KeySerializationVersion::SharedDictionariesWithAdditionalKeys);
+void ColumnLowCardinality::SavePrefix(OutputStream* output) {
+    const auto version = static_cast<uint64_t>(KeySerializationVersion::SharedDictionariesWithAdditionalKeys);
     WireFormat::WriteFixed(*output, version);
+}
 
-    // body
+void ColumnLowCardinality::SaveBody(OutputStream* output) {
     const uint64_t index_serialization_type = indexTypeFromIndexColumn(*index_column_) | IndexFlag::HasAdditionalKeysBit;
     WireFormat::WriteFixed(*output, index_serialization_type);
 
     const uint64_t number_of_keys = dictionary_column_->Size();
     WireFormat::WriteFixed(*output, number_of_keys);
-    dictionary_column_->Save(output);
+    dictionary_column_->SaveBody(output);
 
     const uint64_t number_of_rows = index_column_->Size();
     WireFormat::WriteFixed(*output, number_of_rows);
-    index_column_->Save(output);
-
-    // suffix
-    // NOP
+    index_column_->SaveBody(output);
 }
 
 void ColumnLowCardinality::Clear() {
diff --git a/clickhouse/columns/lowcardinality.h b/clickhouse/columns/lowcardinality.h
index 6d834428..7b0944af 100644
--- a/clickhouse/columns/lowcardinality.h
+++ b/clickhouse/columns/lowcardinality.h
@@ -54,11 +54,16 @@ class ColumnLowCardinality : public Column {
     /// Appends another LowCardinality column to the end of this one, updating dictionary.
     void Append(ColumnRef /*column*/) override;
 
+    bool LoadPrefix(InputStream* input, size_t rows) override;
+
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
+
+    /// Saves column prefix to output stream.
+    void SavePrefix(OutputStream* output) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data.
     void Clear() override;
diff --git a/clickhouse/columns/lowcardinalityadaptor.h b/clickhouse/columns/lowcardinalityadaptor.h
index 70261a51..a8b1295f 100644
--- a/clickhouse/columns/lowcardinalityadaptor.h
+++ b/clickhouse/columns/lowcardinalityadaptor.h
@@ -27,12 +27,19 @@ class LowCardinalitySerializationAdaptor : public AdaptedColumnType
 public:
     using AdaptedColumnType::AdaptedColumnType;
 
+    bool LoadPrefix(InputStream* input, size_t rows) override {
+        auto new_data_column = this->Slice(0, 0)->template As<AdaptedColumnType>();
+        ColumnLowCardinalityT<AdaptedColumnType> low_cardinality_col(new_data_column);
+
+        return low_cardinality_col.LoadPrefix(input, rows);
+    }
+
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override {
+    bool LoadBody(InputStream* input, size_t rows) override {
         auto new_data_column = this->Slice(0, 0)->template As<AdaptedColumnType>();
 
         ColumnLowCardinalityT<AdaptedColumnType> low_cardinality_col(new_data_column);
-        if (!low_cardinality_col.Load(input, rows))
+        if (!low_cardinality_col.LoadBody(input, rows))
             return false;
 
         // It safe to reuse `flat_data_column` later since ColumnLowCardinalityT makes a deep copy, but still check just in case.
@@ -46,8 +53,8 @@ class LowCardinalitySerializationAdaptor : public AdaptedColumnType
     }
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override {
-        ColumnLowCardinalityT<AdaptedColumnType>(this->template As<AdaptedColumnType>()).Save(output);
+    void SaveBody(OutputStream* output) override {
+        ColumnLowCardinalityT<AdaptedColumnType>(this->template As<AdaptedColumnType>()).SaveBody(output);
     }
 };
 
diff --git a/clickhouse/columns/nothing.h b/clickhouse/columns/nothing.h
index f3a1e5e2..822e41af 100644
--- a/clickhouse/columns/nothing.h
+++ b/clickhouse/columns/nothing.h
@@ -51,16 +51,20 @@ class ColumnNothing : public Column {
 	}
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override {
+    bool LoadBody(InputStream* input, size_t rows) override {
 		input->Skip(rows);
 		size_ += rows;
 		return true;
 	}
+        /// Saves column prefix to output stream.
+    void SavePrefix(OutputStream*) override {
+        throw std::runtime_error("method Save is not supported for Nothing column");
+    }
 
     /// Saves column data to output stream.
-    void Save(OutputStream*) override {
-        throw UnimplementedError("method Save is not supported for Nothing column");
-	}
+    void SaveBody(OutputStream*) override {
+        throw UnimplementedError("method SaveBody is not supported for Nothing column");
+    }
 
     /// Clear column data .
     void Clear() override { size_ = 0; }
diff --git a/clickhouse/columns/nullable.cpp b/clickhouse/columns/nullable.cpp
index 65336b31..9150e154 100644
--- a/clickhouse/columns/nullable.cpp
+++ b/clickhouse/columns/nullable.cpp
@@ -50,19 +50,27 @@ void ColumnNullable::Clear() {
     nulls_->Clear();
 }
 
-bool ColumnNullable::Load(InputStream* input, size_t rows) {
-    if (!nulls_->Load(input, rows)) {
+bool ColumnNullable::LoadPrefix(InputStream* input, size_t rows) {
+    return nested_->LoadPrefix(input, rows);
+}
+
+bool ColumnNullable::LoadBody(InputStream* input, size_t rows) {
+    if (!nulls_->LoadBody(input, rows)) {
         return false;
     }
-    if (!nested_->Load(input, rows)) {
+    if (!nested_->LoadBody(input, rows)) {
         return false;
     }
     return true;
 }
 
-void ColumnNullable::Save(OutputStream* output) {
-    nulls_->Save(output);
-    nested_->Save(output);
+void ColumnNullable::SavePrefix(OutputStream* output) {
+    nested_->SavePrefix(output);
+}
+
+void ColumnNullable::SaveBody(OutputStream* output) {
+    nulls_->SaveBody(output);
+    nested_->SaveBody(output);
 }
 
 size_t ColumnNullable::Size() const {
diff --git a/clickhouse/columns/nullable.h b/clickhouse/columns/nullable.h
index 8cde2781..72d31f18 100644
--- a/clickhouse/columns/nullable.h
+++ b/clickhouse/columns/nullable.h
@@ -28,11 +28,17 @@ class ColumnNullable : public Column {
     /// Appends content of given column to the end of current one.
     void Append(ColumnRef column) override;
 
+    /// Loads column prefix from input stream.
+    bool LoadPrefix(InputStream* input, size_t rows) override;
+
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
+
+    /// Saves column prefix to output stream.
+    void SavePrefix(OutputStream* output) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/numeric.cpp b/clickhouse/columns/numeric.cpp
index 479d1e79..d1e66fc9 100644
--- a/clickhouse/columns/numeric.cpp
+++ b/clickhouse/columns/numeric.cpp
@@ -61,14 +61,14 @@ void ColumnVector<T>::Append(ColumnRef column) {
 }
 
 template <typename T>
-bool ColumnVector<T>::Load(InputStream* input, size_t rows) {
+bool ColumnVector<T>::LoadBody(InputStream* input, size_t rows) {
     data_.resize(rows);
 
     return WireFormat::ReadBytes(*input, data_.data(), data_.size() * sizeof(T));
 }
 
 template <typename T>
-void ColumnVector<T>::Save(OutputStream* output) {
+void ColumnVector<T>::SaveBody(OutputStream* output) {
     WireFormat::WriteBytes(*output, data_.data(), data_.size() * sizeof(T));
 }
 
diff --git a/clickhouse/columns/numeric.h b/clickhouse/columns/numeric.h
index 65b21300..e13e58f1 100644
--- a/clickhouse/columns/numeric.h
+++ b/clickhouse/columns/numeric.h
@@ -35,10 +35,10 @@ class ColumnVector : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/string.cpp b/clickhouse/columns/string.cpp
index 612f1a4c..f43d1c3d 100644
--- a/clickhouse/columns/string.cpp
+++ b/clickhouse/columns/string.cpp
@@ -77,7 +77,7 @@ void ColumnFixedString::Append(ColumnRef column) {
     }
 }
 
-bool ColumnFixedString::Load(InputStream * input, size_t rows) {
+bool ColumnFixedString::LoadBody(InputStream * input, size_t rows) {
     data_.resize(string_size_ * rows);
     if (!WireFormat::ReadBytes(*input, &data_[0], data_.size())) {
         return false;
@@ -86,7 +86,7 @@ bool ColumnFixedString::Load(InputStream * input, size_t rows) {
     return true;
 }
 
-void ColumnFixedString::Save(OutputStream* output) {
+void ColumnFixedString::SaveBody(OutputStream* output) {
     WireFormat::WriteBytes(*output, data_.data(), data_.size());
 }
 
@@ -220,7 +220,7 @@ void ColumnString::Append(ColumnRef column) {
     }
 }
 
-bool ColumnString::Load(InputStream* input, size_t rows) {
+bool ColumnString::LoadBody(InputStream* input, size_t rows) {
     items_.clear();
     blocks_.clear();
 
@@ -245,7 +245,7 @@ bool ColumnString::Load(InputStream* input, size_t rows) {
     return true;
 }
 
-void ColumnString::Save(OutputStream* output) {
+void ColumnString::SaveBody(OutputStream* output) {
     for (const auto & item : items_) {
         WireFormat::WriteString(*output, item);
     }
diff --git a/clickhouse/columns/string.h b/clickhouse/columns/string.h
index d1cec652..013511e1 100644
--- a/clickhouse/columns/string.h
+++ b/clickhouse/columns/string.h
@@ -43,10 +43,10 @@ class ColumnFixedString : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
@@ -95,10 +95,10 @@ class ColumnString : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/tuple.cpp b/clickhouse/columns/tuple.cpp
index 88201f36..2b55bae9 100644
--- a/clickhouse/columns/tuple.cpp
+++ b/clickhouse/columns/tuple.cpp
@@ -44,9 +44,9 @@ ColumnRef ColumnTuple::Slice(size_t begin, size_t len) const {
     return std::make_shared<ColumnTuple>(sliced_columns);
 }
 
-bool ColumnTuple::Load(InputStream* input, size_t rows) {
+bool ColumnTuple::LoadPrefix(InputStream* input, size_t rows) {
     for (auto ci = columns_.begin(); ci != columns_.end(); ++ci) {
-        if (!(*ci)->Load(input, rows)) {
+        if (!(*ci)->LoadPrefix(input, rows)) {
             return false;
         }
     }
@@ -54,9 +54,25 @@ bool ColumnTuple::Load(InputStream* input, size_t rows) {
     return true;
 }
 
-void ColumnTuple::Save(OutputStream* output) {
+bool ColumnTuple::LoadBody(InputStream* input, size_t rows) {
     for (auto ci = columns_.begin(); ci != columns_.end(); ++ci) {
-        (*ci)->Save(output);
+        if (!(*ci)->LoadBody(input, rows)) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+void ColumnTuple::SavePrefix(OutputStream* output) {
+    for (auto & column : columns_) {
+        column->SavePrefix(output);
+    }
+}
+
+void ColumnTuple::SaveBody(OutputStream* output) {
+    for (auto & column : columns_) {
+        column->SaveBody(output);
     }
 }
 
diff --git a/clickhouse/columns/tuple.h b/clickhouse/columns/tuple.h
index b1ac784a..c87943fd 100644
--- a/clickhouse/columns/tuple.h
+++ b/clickhouse/columns/tuple.h
@@ -24,11 +24,17 @@ class ColumnTuple : public Column {
     /// Appends content of given column to the end of current one.
     void Append(ColumnRef column) override;
 
+    /// Loads column prefix from input stream.
+    bool LoadPrefix(InputStream* input, size_t rows) override;
+
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
+
+    /// Saves column prefix to output stream.
+    void SavePrefix(OutputStream* output) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
 
     /// Clear column data .
     void Clear() override;
diff --git a/clickhouse/columns/uuid.cpp b/clickhouse/columns/uuid.cpp
index b64bb6b5..6bb6cd8c 100644
--- a/clickhouse/columns/uuid.cpp
+++ b/clickhouse/columns/uuid.cpp
@@ -44,12 +44,12 @@ void ColumnUUID::Append(ColumnRef column) {
     }
 }
 
-bool ColumnUUID::Load(InputStream* input, size_t rows) {
-    return data_->Load(input, rows * 2);
+bool ColumnUUID::LoadBody(InputStream* input, size_t rows) {
+    return data_->LoadBody(input, rows * 2);
 }
 
-void ColumnUUID::Save(OutputStream* output) {
-    data_->Save(output);
+void ColumnUUID::SaveBody(OutputStream* output) {
+    data_->SaveBody(output);
 }
 
 size_t ColumnUUID::Size() const {
diff --git a/clickhouse/columns/uuid.h b/clickhouse/columns/uuid.h
index 40319459..c1c6e78d 100644
--- a/clickhouse/columns/uuid.h
+++ b/clickhouse/columns/uuid.h
@@ -30,10 +30,10 @@ class ColumnUUID : public Column {
     void Append(ColumnRef column) override;
 
     /// Loads column data from input stream.
-    bool Load(InputStream* input, size_t rows) override;
+    bool LoadBody(InputStream* input, size_t rows) override;
 
     /// Saves column data to output stream.
-    void Save(OutputStream* output) override;
+    void SaveBody(OutputStream* output) override;
     
     /// Clear column data .
     void Clear() override;
diff --git a/ut/CMakeLists.txt b/ut/CMakeLists.txt
index baa72b91..50c1dd03 100644
--- a/ut/CMakeLists.txt
+++ b/ut/CMakeLists.txt
@@ -15,7 +15,7 @@ SET ( clickhouse-cpp-ut-src
     utils.cpp
     readonly_client_test.cpp
     connection_failed_client_test.cpp
-)
+    array_of_low_cardinality_tests.cpp)
 
 IF (WITH_OPENSSL)
     LIST (APPEND clickhouse-cpp-ut-src ssl_ut.cpp)
diff --git a/ut/array_of_low_cardinality_tests.cpp b/ut/array_of_low_cardinality_tests.cpp
new file mode 100644
index 00000000..e022c218
--- /dev/null
+++ b/ut/array_of_low_cardinality_tests.cpp
@@ -0,0 +1,99 @@
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <iterator>
+#include <vector>
+
+#include <clickhouse/columns/array.h>
+#include <clickhouse/columns/string.h>
+#include <clickhouse/columns/lowcardinality.h>
+#include "clickhouse/block.h"
+#include "clickhouse/client.h"
+#include "utils.h"
+#include "clickhouse/base/buffer.h"
+#include "clickhouse/base/output.h"
+
+namespace
+{
+using namespace clickhouse;
+}
+
+std::shared_ptr<ColumnArray> buildTestColumn(const std::vector<std::vector<std::string>>& rows) {
+    auto arrayColumn = std::make_shared<ColumnArray>(std::make_shared<ColumnLowCardinalityT<ColumnString>>());
+
+    for (const auto& row : rows) {
+        auto column = std::make_shared<ColumnLowCardinalityT<ColumnString>>();
+
+        for (const auto& string : row) {
+            column->Append(string);
+        }
+
+        arrayColumn->AppendAsColumn(column);
+    }
+
+    return arrayColumn;
+}
+
+TEST(ArrayOfLowCardinality, Serialization) {
+    const auto inputColumn = buildTestColumn({
+        { "aa", "bb" },
+        { "cc" }
+    });
+
+    // The serialization data was extracted from a successful insert.
+    // When compared to what Clickhouse/NativeWriter does for the same fields, the only differences are the index type and indexes.
+    // Since we are setting a different index type in clickhouse-cpp, it's expected to have different indexes.
+    const std::vector<uint8_t> expectedSerialization {
+        0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00,
+        0x00, 0x00, 0x02, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0x61, 0x61,
+        0x02, 0x62, 0x62, 0x02, 0x63, 0x63, 0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00,
+        0x03, 0x00, 0x00, 0x00
+    };
+
+    Buffer buf;
+
+    BufferOutput output(&buf);
+    inputColumn->Save(&output);
+    output.Flush();
+
+    ASSERT_EQ(expectedSerialization, buf);
+}
+
+TEST(ArrayOfLowCardinality, InsertAndQuery) {
+
+    const auto localHostEndpoint = ClientOptions()
+                                       .SetHost(           getEnvOrDefault("CLICKHOUSE_HOST",     "localhost"))
+                                       .SetPort(   getEnvOrDefault<size_t>("CLICKHOUSE_PORT",     "9000"))
+                                       .SetUser(           getEnvOrDefault("CLICKHOUSE_USER",     "default"))
+                                       .SetPassword(       getEnvOrDefault("CLICKHOUSE_PASSWORD", ""))
+                                       .SetDefaultDatabase(getEnvOrDefault("CLICKHOUSE_DB",       "default"));
+
+    Client client(ClientOptions(localHostEndpoint)
+                      .SetPingBeforeQuery(true));
+
+    const auto testData = std::vector<std::vector<std::string>> {
+        { "aa", "bb" },
+        { "cc" }
+    };
+
+    auto column = buildTestColumn(testData);
+
+    Block block;
+    block.AppendColumn("arr", column);
+
+    client.Execute("DROP TEMPORARY TABLE IF EXISTS array_lc");
+    client.Execute("CREATE TEMPORARY TABLE IF NOT EXISTS array_lc (arr Array(LowCardinality(String))) ENGINE = Memory");
+    client.Insert("array_lc", block);
+
+    client.Select("SELECT * FROM array_lc", [&](const Block& bl) {
+          for (size_t c = 0; c < bl.GetRowCount(); ++c) {
+              auto col = bl[0]->As<ColumnArray>()->GetAsColumn(c);
+              for (size_t i = 0; i < col->Size(); ++i) {
+                  auto stringColumn = col->As<ColumnString>();
+                  const auto string = stringColumn->At(i);
+
+                  ASSERT_EQ(testData[c][i], string);
+              }
+          }
+      }
+    );
+}