branch-4.0: [feature](catalog) support varbinary type mapping in hive/iceberg/paimon table (#57821) (#58482)

### What problem does this PR solve?
Problem Summary:
cherry-pick from (#57821)

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [ ] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: zhangstar333

be/src/util/arrow/row_batch.cpp

@@ -161,6 +161,10 @@ Status convert_to_arrow_type(const vectorized::DataTypePtr& origin_type,
         *result = arrow::binary();
         break;
     }
+    case TYPE_VARBINARY: {
+        *result = arrow::binary();
+        break;
+    }
     default:
         return Status::InvalidArgument("Unknown primitive type({}) convert to Arrow type",
                                        type->get_name());

be/src/vec/columns/column_varbinary.cpp

@@ -20,6 +20,7 @@
 #include <glog/logging.h>
 
 #include <cstddef>
+#include <cstdint>
 
 #include "runtime/primitive_type.h"
 #include "vec/columns/column.h"
@@ -162,5 +163,64 @@ void ColumnVarbinary::replace_column_data(const IColumn& rhs, size_t row, size_t
     _data[self_row] = doris::StringView(dst, val.size());
 }
 
+size_t ColumnVarbinary::get_max_row_byte_size() const {
+    uint32_t max_size = 0;
+    size_t num_rows = size();
+    for (size_t i = 0; i < num_rows; ++i) {
+        max_size = std::max(max_size, _data[i].size());
+    }
+    return max_size + sizeof(uint32_t);
+}
+
+void ColumnVarbinary::serialize_vec(StringRef* keys, size_t num_rows) const {
+    for (size_t i = 0; i < num_rows; ++i) {
+        keys[i].size += serialize_impl(const_cast<char*>(keys[i].data + keys[i].size), i);
+    }
+}
+
+void ColumnVarbinary::deserialize_vec(StringRef* keys, const size_t num_rows) {
+    for (size_t i = 0; i != num_rows; ++i) {
+        auto sz = deserialize_impl(keys[i].data);
+        keys[i].data += sz;
+        keys[i].size -= sz;
+    }
+}
+
+template <bool positive>
+struct ColumnVarbinary::less {
+    const ColumnVarbinary& parent;
+    explicit less(const ColumnVarbinary& parent_) : parent(parent_) {}
+    bool operator()(size_t lhs, size_t rhs) const {
+        int res = parent._data[lhs].compare(parent._data[rhs]);
+        return positive ? (res < 0) : (res > 0);
+    }
+};
+
+void ColumnVarbinary::get_permutation(bool reverse, size_t limit, int /*nan_direction_hint*/,
+                                      IColumn::Permutation& res) const {
+    size_t s = _data.size();
+    res.resize(s);
+    for (size_t i = 0; i < s; ++i) {
+        res[i] = i;
+    }
+
+    if (reverse) {
+        pdqsort(res.begin(), res.end(), less<false>(*this));
+    } else {
+        pdqsort(res.begin(), res.end(), less<true>(*this));
+    }
+}
+
+void ColumnVarbinary::insert_many_strings(const StringRef* strings, size_t num) {
+    for (size_t i = 0; i < num; i++) {
+        insert_data(strings[i].data, strings[i].size);
+    }
+}
+
+void ColumnVarbinary::insert_many_strings_overflow(const StringRef* strings, size_t num,
+                                                   size_t max_length) {
+    insert_many_strings(strings, num);
+}
+
 #include "common/compile_check_end.h"
 } // namespace doris::vectorized

be/src/vec/columns/column_varbinary.h

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <glog/logging.h>
+#include <pdqsort.h>
 
 #include <cstddef>
 
@@ -34,8 +35,11 @@ class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
 private:
     using Self = ColumnVarbinary;
     friend class COWHelper<IColumn, ColumnVarbinary>;
+    template <bool positive>
+    struct less;
 
 public:
+    using value_type = typename PrimitiveTypeTraits<TYPE_VARBINARY>::ColumnItemType;
     using Container = PaddedPODArray<doris::StringView>;
     ColumnVarbinary() = default;
     ColumnVarbinary(const size_t n) : _data(n) {}
@@ -105,10 +109,19 @@ class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
 
     int compare_at(size_t n, size_t m, const IColumn& rhs_,
                    int /*nan_direction_hint*/) const override {
-        const ColumnVarbinary& rhs = assert_cast<const ColumnVarbinary&>(rhs_);
+        const auto& rhs = assert_cast<const ColumnVarbinary&>(rhs_);
         return this->_data[n].compare(rhs.get_data()[m]);
     }
 
+    void get_permutation(bool reverse, size_t limit, int /*nan_direction_hint*/,
+                         IColumn::Permutation& res) const override;
+
+    size_t get_max_row_byte_size() const override;
+
+    void deserialize_vec(StringRef* keys, const size_t num_rows) override;
+
+    void serialize_vec(StringRef* keys, const size_t num_rows) const override;
+
     void pop_back(size_t n) override { resize(size() - n); }
 
     StringRef serialize_value_into_arena(size_t n, Arena& arena,
@@ -167,6 +180,11 @@ class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
         return _data[row].size() + sizeof(uint32_t);
     }
 
+    void insert_many_strings(const StringRef* strings, size_t num) override;
+
+    void insert_many_strings_overflow(const StringRef* strings, size_t num,
+                                      size_t max_length) override;
+
 private:
     Container _data;
     Arena _arena;

be/src/vec/data_types/serde/data_type_varbinary_serde.cpp

@@ -65,6 +65,71 @@ Status DataTypeVarbinarySerDe::_write_column_to_mysql(const IColumn& column,
     return Status::OK();
 }
 
+Status DataTypeVarbinarySerDe::write_column_to_arrow(const IColumn& column, const NullMap* null_map,
+                                                     arrow::ArrayBuilder* array_builder,
+                                                     int64_t start, int64_t end,
+                                                     const cctz::time_zone& ctz) const {
+    auto lambda_function = [&](auto& builder) -> Status {
+        const auto& varbinary_column_data = assert_cast<const ColumnVarbinary&>(column).get_data();
+        for (size_t i = start; i < end; ++i) {
+            if (null_map && (*null_map)[i]) {
+                RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column.get_name(),
+                                                 builder.type()->name()));
+                continue;
+            }
+            auto string_view = varbinary_column_data[i];
+            RETURN_IF_ERROR(checkArrowStatus(builder.Append(string_view.data(), string_view.size()),
+                                             column.get_name(), builder.type()->name()));
+        }
+        return Status::OK();
+    };
+    if (array_builder->type()->id() == arrow::Type::BINARY) {
+        auto& builder = assert_cast<arrow::BinaryBuilder&>(*array_builder);
+        return lambda_function(builder);
+    } else if (array_builder->type()->id() == arrow::Type::STRING) {
+        auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
+        return lambda_function(builder);
+    } else {
+        return Status::InvalidArgument("Unsupported arrow type for varbinary column: {}",
+                                       array_builder->type()->name());
+    }
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::write_column_to_orc(const std::string& timezone,
+                                                   const IColumn& column, const NullMap* null_map,
+                                                   orc::ColumnVectorBatch* orc_col_batch,
+                                                   int64_t start, int64_t end,
+                                                   vectorized::Arena& arena) const {
+    auto* cur_batch = dynamic_cast<orc::StringVectorBatch*>(orc_col_batch);
+    const auto& varbinary_column_data = assert_cast<const ColumnVarbinary&>(column).get_data();
+
+    for (auto row_id = start; row_id < end; row_id++) {
+        cur_batch->data[row_id] = const_cast<char*>(varbinary_column_data[row_id].data());
+        cur_batch->length[row_id] = varbinary_column_data[row_id].size();
+    }
+
+    cur_batch->numElements = end - start;
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::serialize_one_cell_to_json(const IColumn& column, int64_t row_num,
+                                                          BufferWritable& bw,
+                                                          FormatOptions& options) const {
+    auto result = check_column_const_set_readability(column, row_num);
+    ColumnPtr ptr = result.first;
+    row_num = result.second;
+    const auto& value = assert_cast<const ColumnVarbinary&>(*ptr).get_data_at(row_num);
+    bw.write(value.data, value.size);
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::deserialize_one_cell_from_json(IColumn& column, Slice& slice,
+                                                              const FormatOptions& options) const {
+    assert_cast<ColumnVarbinary&>(column).insert_data(slice.data, slice.size);
+    return Status::OK();
+}
+
 void DataTypeVarbinarySerDe::to_string(const IColumn& column, size_t row_num,
                                        BufferWritable& bw) const {
     const auto value = assert_cast<const ColumnVarbinary&>(column).get_data_at(row_num);

be/src/vec/data_types/serde/data_type_varbinary_serde.h

@@ -23,8 +23,10 @@
 
 #include "common/status.h"
 #include "data_type_serde.h"
+#include "data_type_string_serde.h"
 #include "vec/columns/column.h"
 #include "vec/columns/column_const.h"
+#include "vec/columns/column_varbinary.h"
 #include "vec/common/arena.h"
 #include "vec/data_types/serde/data_type_nullable_serde.h"
 
@@ -39,19 +41,15 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
     std::string get_name() const override { return "Varbinary"; }
 
     Status serialize_one_cell_to_json(const IColumn& column, int64_t row_num, BufferWritable& bw,
-                                      FormatOptions& options) const override {
-        return Status::NotSupported("serialize_one_cell_to_json with type " + column.get_name());
-    }
+                                      FormatOptions& options) const override;
 
     Status serialize_column_to_json(const IColumn& column, int64_t start_idx, int64_t end_idx,
                                     BufferWritable& bw, FormatOptions& options) const override {
         return Status::NotSupported("serialize_column_to_json with type " + column.get_name());
     }
+
     Status deserialize_one_cell_from_json(IColumn& column, Slice& slice,
-                                          const FormatOptions& options) const override {
-        return Status::NotSupported("deserialize_one_cell_from_text with type " +
-                                    column.get_name());
-    }
+                                          const FormatOptions& options) const override;
 
     Status deserialize_column_from_json_vector(IColumn& column, std::vector<Slice>& slices,
                                                uint64_t* num_deserialized,
@@ -64,6 +62,7 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
                               int64_t end) const override {
         return Status::NotSupported("write_column_to_pb with type " + column.get_name());
     }
+
     Status read_column_from_pb(IColumn& column, const PValues& arg) const override {
         return Status::NotSupported("read_column_from_pb with type " + column.get_name());
     }
@@ -75,9 +74,8 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
 
     Status write_column_to_arrow(const IColumn& column, const NullMap* null_map,
                                  arrow::ArrayBuilder* array_builder, int64_t start, int64_t end,
-                                 const cctz::time_zone& ctz) const override {
-        return Status::NotSupported("write_column_to_arrow with type " + column.get_name());
-    }
+                                 const cctz::time_zone& ctz) const override;
+
     Status read_column_from_arrow(IColumn& column, const arrow::Array* arrow_array, int64_t start,
                                   int64_t end, const cctz::time_zone& ctz) const override {
         return Status::Error(ErrorCode::NOT_IMPLEMENTED_ERROR,
@@ -94,11 +92,7 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
 
     Status write_column_to_orc(const std::string& timezone, const IColumn& column,
                                const NullMap* null_map, orc::ColumnVectorBatch* orc_col_batch,
-                               int64_t start, int64_t end,
-                               vectorized::Arena& arena) const override {
-        return Status::Error(ErrorCode::NOT_IMPLEMENTED_ERROR,
-                             "write_column_to_orc with type " + column.get_name());
-    }
+                               int64_t start, int64_t end, vectorized::Arena& arena) const override;
 
     void to_string(const IColumn& column, size_t row_num, BufferWritable& bw) const override;
 

be/src/vec/exec/format/column_type_convert.cpp

@@ -18,13 +18,11 @@
 #include "vec/exec/format/column_type_convert.h"
 
 #include "common/cast_set.h"
+#include "runtime/define_primitive_type.h"
 
 namespace doris::vectorized::converter {
 #include "common/compile_check_begin.h"
 
-const std::set<std::string> SafeCastString<TYPE_BOOLEAN>::FALSE_VALUES = {"false", "off", "no", "0",
-                                                                          ""};
-
 #define FOR_LOGICAL_INTEGER_TYPES(M) \
     M(TYPE_TINYINT)                  \
     M(TYPE_SMALLINT)                 \
@@ -225,6 +223,8 @@ static std::unique_ptr<ColumnTypeConverter> _to_string_converter(const DataTypeP
         default:
             return std::make_unique<UnsupportedConverter>(src_type, dst_type);
         }
+    } else if (is_varbinary(src_primitive_type)) {
+        return std::make_unique<VarBinaryConverter<TYPE_STRING, TYPE_STRING>>();
     }
     return std::make_unique<UnsupportedConverter>(src_type, dst_type);
 }
@@ -240,6 +240,9 @@ static std::unique_ptr<ColumnTypeConverter> _from_string_converter(const DataTyp
                 remove_nullable(dst_type));
         FOR_ALL_LOGICAL_TYPES(DISPATCH)
 #undef DISPATCH
+    case TYPE_VARBINARY: {
+        return std::make_unique<VarBinaryConverter<TYPE_STRING, TYPE_VARBINARY>>();
+    }
     default:
         return std::make_unique<UnsupportedConverter>(src_type, dst_type);
     }