[feature](catalog) support varbinary type mapping in hive/iceberg/paimon table (apache#57821)

Problem Summary:
support varbinary type in hive/iceberg/paimon table, could mapping
varbinary type into doris directly, not of use string type, could use
catalog properties enable.mapping.varbinary control it, and default is
false.
and TVF function, eg HDFS also have param could control, and default is
false.

1. when parquet file column type is tparquet::Type::BYTE_ARRAY and no
logicalType and converted_type，read it to column_varbianry directly. so
both physical convert and logical convert are consistent.

if tparquet::Type::BYTE_ARRAY and have set logicalType, eg String, so
those will be reading as column_string, and if the table column create
as binary column, so VarBinaryConverter used convert column_string to
column_varbinary.

2. when orc file column is binary type, also mapping to varbinary type
directly, and could reuse StringVectorBatch.

3. add cast between string and varbinary type.

4. mapping UUID to binary type instead of string  in iceberg .

5. change the bool safe_cast_string(**const char\* startptr, size_t
buffer_size**, xxx) signature to safe_cast_string(**const StringRef&
str_ref**, xxx).

6. add **const** to read_date_text_impl function.

7. add some test with paimon catalog test varbinary, will add more case
for hive/iceberg and update doc.
```
mysql> show create table binary_demo3;
+--------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Table        | Create Table                                                                                                                                                                                                                                                                                                                                                                                     |
+--------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| binary_demo3 | CREATE TABLE `binary_demo3` (
  `id` int NULL,
  `record_name` char(10) NULL,
  `vrecord_name` text NULL,
  `bin` varbinary(10) NULL,
  `varbin` varbinary(2147483647) NULL
) ENGINE=PAIMON_EXTERNAL_TABLE
LOCATION 'file:/mnt/disk2/zhangsida/test_paimon/demo.db/binary_demo3'
PROPERTIES (
  "path" = "file:/mnt/disk2/zhangsida/test_paimon/demo.db/binary_demo3",
  "primary-key" = "id"
); |
+--------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row in set (0.00 sec)

mysql> select *, length(record_name),length(vrecord_name),length(bin),length(varbin) from binary_demo3;
+------+-------------+--------------+------------------------+----------------+---------------------+----------------------+-------------+----------------+
| id   | record_name | vrecord_name | bin                    | varbin         | length(record_name) | length(vrecord_name) | length(bin) | length(varbin) |
+------+-------------+--------------+------------------------+----------------+---------------------+----------------------+-------------+----------------+
|    1 | AAAA        | AAAA         | 0xAAAA0000000000000000 | 0xAAAA         |                  10 |                    4 |          10 |              2 |
|    2 | 6161        | 6161         | 0x61610000000000000000 | 0x6161         |                  10 |                    4 |          10 |              2 |
|    3 | NULL        | NULL         | NULL                   | NULL           |                NULL |                 NULL |        NULL |           NULL |
+------+-------------+--------------+------------------------+----------------+---------------------+----------------------+-------------+----------------+

```

support varbinary type mapping in hive/iceberg/paimon table

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(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(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(StringRef* keys, const size_t num_rows) override;
+
+    void serialize(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);
     }