Added tests for GetItem;

Fixed size checks in ItemView
Fixed GetItem type/value for ColumnDate, ColumnDecimal, ColumnIPv4, ColumnIPv6, ColumnUUID

Author: Enmk

clickhouse/columns/date.cpp

@@ -58,7 +58,7 @@ void ColumnDate::Swap(Column& other) {
 }
 
 ItemView ColumnDate::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView(Type::Date, data_->GetItem(index));
 }
 
 
@@ -128,7 +128,7 @@ void ColumnDateTime::Swap(Column& other) {
 }
 
 ItemView ColumnDateTime::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView(Type::DateTime, data_->GetItem(index));
 }
 
 ColumnDateTime64::ColumnDateTime64(size_t precision)
@@ -186,7 +186,7 @@ size_t ColumnDateTime64::Size() const {
 }
 
 ItemView ColumnDateTime64::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView(Type::DateTime64, data_->GetItem(index));
 }
 
 void ColumnDateTime64::Swap(Column& other) {

clickhouse/columns/decimal.cpp

@@ -229,7 +229,7 @@ void ColumnDecimal::Swap(Column& other) {
 }
 
 ItemView ColumnDecimal::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView{GetType().GetCode(), data_->GetItem(index)};
 }
 
 size_t ColumnDecimal::GetScale() const

clickhouse/columns/ip4.cpp

@@ -97,7 +97,7 @@ void ColumnIPv4::Swap(Column& other) {
 }
 
 ItemView ColumnIPv4::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView(Type::IPv4, data_->GetItem(index));
 }
 
 }

clickhouse/columns/ip6.cpp

@@ -97,7 +97,7 @@ void ColumnIPv6::Swap(Column& other) {
 }
 
 ItemView ColumnIPv6::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    return ItemView{Type::IPv6, data_->GetItem(index)};
 }
 
 }

clickhouse/columns/itemview.cpp

@@ -35,13 +35,13 @@ void ItemView::ValidateData(Type::Code type, DataType data) {
         case Type::Code::UInt64:
         case Type::Code::Float64:
         case Type::Code::DateTime64:
-        case Type::Code::IPv6:
         case Type::Code::Decimal64:
             expected_size = 8;
             break;
 
         case Type::Code::String:
         case Type::Code::FixedString:
+        case Type::Code::Decimal:
             // value can be of any size
             return;
 
@@ -51,9 +51,9 @@ void ItemView::ValidateData(Type::Code type, DataType data) {
         case Type::Code::LowCardinality:
             throw UnimplementedError("Unsupported type in ItemView: " + std::to_string(static_cast<int>(type)));
 
+        case Type::Code::IPv6:
         case Type::Code::UUID:
         case Type::Code::Int128:
-        case Type::Code::Decimal:
         case Type::Code::Decimal128:
             expected_size = 16;
             break;

clickhouse/columns/itemview.h

@@ -5,6 +5,7 @@
 
 #include <string_view>
 #include <stdexcept>
+#include <type_traits>
 
 namespace clickhouse {
 
@@ -43,8 +44,15 @@ struct ItemView {
         ValidateData(type, data);
     }
 
+    ItemView(Type::Code type, ItemView other)
+        : type(type),
+          data(other.data)
+    {
+        ValidateData(type, data);
+    }
+
     explicit ItemView()
-        : ItemView(Type::Void, {})
+        : ItemView(Type::Void, std::string_view{})
     {}
 
     template <typename T>
@@ -53,11 +61,12 @@ struct ItemView {
     {}
 
     template <typename T>
-    T get() const {
-        if constexpr (std::is_same_v<std::string_view, T> || std::is_same_v<std::string, T>) {
+    auto get() const {
+        using ValueType = std::remove_cv_t<std::decay_t<T>>;
+        if constexpr (std::is_same_v<std::string_view, ValueType> || std::is_same_v<std::string, ValueType>) {
             return data;
-        } else if constexpr (std::is_fundamental_v<T> || std::is_same_v<Int128, T>) {
-            if (sizeof(T) == data.size()) {
+        } else if constexpr (std::is_fundamental_v<ValueType> || std::is_same_v<Int128, ValueType>) {
+            if (sizeof(ValueType) == data.size()) {
                 return *reinterpret_cast<const T*>(data.data());
             } else {
                 throw AssertionError("Incompatitable value type and size.");

clickhouse/columns/uuid.cpp

@@ -70,7 +70,11 @@ void ColumnUUID::Swap(Column& other) {
 }
 
 ItemView ColumnUUID::GetItem(size_t index) const {
-    return data_->GetItem(index);
+    // We know that ColumnUInt64 stores it's data in continius memory region,
+    // and that every 2 values from data represent 1 UUID value.
+    const auto data_item_view = data_->GetItem(index * 2);
+
+    return ItemView{Type::UUID, std::string_view{data_item_view.data.data(), data_item_view.data.size() * 2}};
 }
 
 }

ut/Column_ut.cpp

@@ -150,6 +150,41 @@ TYPED_TEST(GenericColumnTest, Append) {
     EXPECT_TRUE(CompareRecursive(values, *column));
 }
 
+// To make some value types compatitable with Column::GetItem()
+template <typename ColumnType, typename ValueType>
+inline auto convertValueForGetItem(ValueType&& t) {
+    using T = std::remove_cv_t<std::decay_t<ValueType>>;
+
+    if constexpr (std::is_same_v<T, clickhouse::UInt128>
+            || std::is_same_v<T, clickhouse::Int128>) {
+        return std::string_view{reinterpret_cast<const char*>(&t), sizeof(T)};
+    } else if constexpr (std::is_same_v<T, in_addr>) {
+        return htonl(t.s_addr);
+    } else if constexpr (std::is_same_v<T, in6_addr>) {
+        return std::string_view(reinterpret_cast<const char*>(t.s6_addr), 16);
+    } else if constexpr (std::is_same_v<ColumnType, ColumnDate>) {
+        return static_cast<uint16_t>(t / std::time_t(86400));
+    } else if constexpr (std::is_same_v<ColumnType, ColumnDateTime>) {
+        return static_cast<uint32_t>(t);
+    } else {
+        return t;
+    }
+}
+
+TYPED_TEST(GenericColumnTest, GetItem) {
+    auto [column, values] = this->MakeColumnWithValues(100);
+
+    ASSERT_EQ(values.size(), column->Size());
+    ASSERT_EQ(column->GetItem(0).type, column->GetType().GetCode());
+
+    for (size_t i = 0; i < values.size(); ++i) {
+        const auto v = convertValueForGetItem<typename TestFixture::ColumnType>(values[i]);
+        const ItemView item = column->GetItem(i);
+
+        ASSERT_TRUE(CompareRecursive(item.get<decltype(v)>(), v));
+    }
+}
+
 TYPED_TEST(GenericColumnTest, Slice) {
     auto [column, values] = this->MakeColumnWithValues(100);