fix compare_at base function support nested types

be/src/vec/columns/column.h

@@ -453,6 +453,7 @@ class IColumn : public COW<IColumn> {
       * For example, if nan_direction_hint == -1 is used by descending sorting, NaNs will be at the end.
       *
       * For non Nullable and non floating point types, nan_direction_hint is ignored.
+      * For array/map/struct types, comparison means equal.
       */
     virtual int compare_at(size_t n, size_t m, const IColumn& rhs,
                            int nan_direction_hint) const = 0;

be/src/vec/columns/column_array.cpp

@@ -254,6 +254,27 @@ StringRef ColumnArray::serialize_value_into_arena(size_t n, Arena& arena,
     return res;
 }
 
+int ColumnArray::compare_at(size_t n, size_t m, const IColumn& rhs_, int nan_direction_hint) const {
+    // since column type is complex, we can't use this function
+    const auto& rhs = assert_cast<const ColumnArray&>(rhs_);
+
+    size_t lhs_size = size_at(n);
+    size_t rhs_size = rhs.size_at(m);
+    size_t min_size = std::min(lhs_size, rhs_size);
+    for (size_t i = 0; i < min_size; ++i) {
+        int res;
+        if (res = get_data().compare_at(offset_at(n) + i, rhs.offset_at(m) + i, *rhs.data.get(),
+                                        nan_direction_hint);
+            res) {
+            // if res != 0 , here is something different ,just return
+            return res;
+        }
+    }
+
+    // then we check size of array
+    return lhs_size < rhs_size ? -1 : (lhs_size == rhs_size ? 0 : 1);
+}
+
 const char* ColumnArray::deserialize_and_insert_from_arena(const char* pos) {
     size_t array_size = unaligned_load<size_t>(pos);
     pos += sizeof(array_size);

be/src/vec/columns/column_array.h

@@ -163,11 +163,8 @@ class ColumnArray final : public COWHelper<IColumn, ColumnArray> {
     //ColumnPtr index(const IColumn & indexes, size_t limit) const;
     template <typename Type>
     ColumnPtr index_impl(const PaddedPODArray<Type>& indexes, size_t limit) const;
-    [[noreturn]] int compare_at(size_t n, size_t m, const IColumn& rhs_,
-                                int nan_direction_hint) const override {
-        LOG(FATAL) << "compare_at not implemented";
-        __builtin_unreachable();
-    }
+    int compare_at(size_t n, size_t m, const IColumn& rhs_, int nan_direction_hint) const override;
+
     [[noreturn]] void get_permutation(bool reverse, size_t limit, int nan_direction_hint,
                                       Permutation& res) const override {
         LOG(FATAL) << "get_permutation not implemented";

be/src/vec/columns/column_map.cpp

@@ -217,7 +217,7 @@ StringRef ColumnMap::serialize_value_into_arena(size_t n, Arena& arena, char con
 
 const char* ColumnMap::deserialize_and_insert_from_arena(const char* pos) {
     size_t array_size = unaligned_load<size_t>(pos);
-    pos += 2 * sizeof(array_size);
+    pos += sizeof(array_size);
 
     for (size_t i = 0; i < array_size; ++i) {
         pos = get_keys().deserialize_and_insert_from_arena(pos);
@@ -231,6 +231,39 @@ const char* ColumnMap::deserialize_and_insert_from_arena(const char* pos) {
     return pos;
 }
 
+int ColumnMap::compare_at(size_t n, size_t m, const IColumn& rhs_, int nan_direction_hint) const {
+    const auto& rhs = assert_cast<const ColumnMap&>(rhs_);
+
+    size_t lhs_size = size_at(n);
+    size_t rhs_size = rhs.size_at(m);
+
+    size_t lhs_offset = offset_at(n);
+    size_t rhs_offset = rhs.offset_at(m);
+
+    size_t min_size = std::min(lhs_size, rhs_size);
+
+    int a = 0;
+    if (a) {
+        return 0;
+    }
+    for (size_t i = 0; i < min_size; ++i) {
+        // if any value in key not equal, just return
+        if (int res = get_keys().compare_at(lhs_offset + i, rhs_offset + i, rhs.get_keys(),
+                                            nan_direction_hint);
+            res) {
+            return res;
+        }
+        // // if any value in value not equal, just return
+        if (int res = get_values().compare_at(lhs_offset + i, rhs_offset + i, rhs.get_values(),
+                                              nan_direction_hint);
+            res) {
+            return res;
+        }
+    }
+
+    return lhs_size < rhs_size ? -1 : (lhs_size == rhs_size ? 0 : 1);
+}
+
 void ColumnMap::update_hash_with_value(size_t n, SipHash& hash) const {
     size_t kv_size = size_at(n);
     size_t offset = offset_at(n);

be/src/vec/columns/column_map.h

@@ -119,11 +119,8 @@ class ColumnMap final : public COWHelper<IColumn, ColumnMap> {
         return scatter_impl<ColumnMap>(num_columns, selector);
     }
 
-    [[noreturn]] int compare_at(size_t n, size_t m, const IColumn& rhs_,
-                                int nan_direction_hint) const override {
-        LOG(FATAL) << "compare_at not implemented";
-        __builtin_unreachable();
-    }
+    int compare_at(size_t n, size_t m, const IColumn& rhs_, int nan_direction_hint) const override;
+
     void get_permutation(bool reverse, size_t limit, int nan_direction_hint,
                          Permutation& res) const override {
         LOG(FATAL) << "get_permutation not implemented";

be/src/vec/columns/column_struct.cpp

@@ -185,6 +185,27 @@ const char* ColumnStruct::deserialize_and_insert_from_arena(const char* pos) {
     return pos;
 }
 
+int ColumnStruct::compare_at(size_t n, size_t m, const IColumn& rhs_,
+                             int nan_direction_hint) const {
+    const ColumnStruct& rhs = assert_cast<const ColumnStruct&>(rhs_);
+
+    const size_t lhs_tuple_size = columns.size();
+    const size_t rhs_tuple_size = rhs.tuple_size();
+
+    if (rhs_tuple_size < lhs_tuple_size) {
+        return -1;
+    } else if (rhs_tuple_size > lhs_tuple_size) {
+        return 1;
+    } else {
+        for (size_t i = 0; i < lhs_tuple_size; ++i) {
+            if (int res = columns[i]->compare_at(n, m, *rhs.columns[i], nan_direction_hint); res) {
+                return res;
+            }
+        }
+    }
+    return 0;
+}
+
 void ColumnStruct::update_hash_with_value(size_t n, SipHash& hash) const {
     for (const auto& column : columns) {
         column->update_hash_with_value(n, hash);

be/src/vec/columns/column_struct.h

@@ -158,11 +158,7 @@ class ColumnStruct final : public COWHelper<IColumn, ColumnStruct> {
     MutableColumns scatter(ColumnIndex num_columns, const Selector& selector) const override;
 
     // ColumnPtr index(const IColumn & indexes, size_t limit) const override;
-    [[noreturn]] int compare_at(size_t n, size_t m, const IColumn& rhs_,
-                                int nan_direction_hint) const override {
-        LOG(FATAL) << "compare_at not implemented";
-        __builtin_unreachable();
-    }
+    int compare_at(size_t n, size_t m, const IColumn& rhs_, int nan_direction_hint) const override;
 
     MutableColumnPtr get_shrinked_column() override;