support native format

Author: eldenmoon

be/src/service/internal_service.cpp

@@ -122,6 +122,7 @@
 #include "vec/exec/format/csv/csv_reader.h"
 #include "vec/exec/format/generic_reader.h"
 #include "vec/exec/format/json/new_json_reader.h"
+#include "vec/exec/format/native/native_reader.h"
 #include "vec/exec/format/orc/vorc_reader.h"
 #include "vec/exec/format/parquet/vparquet_reader.h"
 #include "vec/exec/format/text/text_reader.h"
@@ -856,6 +857,14 @@ void PInternalService::fetch_table_schema(google::protobuf::RpcController* contr
             reader = vectorized::OrcReader::create_unique(params, range, "", &io_ctx);
             break;
         }
+        case TFileFormatType::FORMAT_NATIVE: {
+            // Doris Native binary format reader for schema probing in S3/FILE TVF.
+            // Use a reasonable default batch size; it only affects internal buffering.
+            size_t batch_size = 4096;
+            reader = vectorized::NativeReader::create_unique(profile.get(), params, range,
+                                                             batch_size, &io_ctx, nullptr);
+            break;
+        }
         case TFileFormatType::FORMAT_JSON: {
             reader = vectorized::NewJsonReader::create_unique(profile.get(), params, range,
                                                               file_slots, &io_ctx);

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

@@ -89,6 +89,8 @@ class IColumn;
 class Arena;
 class IDataType;
 struct CastParameters;
+class BufferWritable;
+class ReadBuffer;
 
 class DataTypeSerDe;
 using DataTypeSerDeSPtr = std::shared_ptr<DataTypeSerDe>;
@@ -458,6 +460,30 @@ class DataTypeSerDe {
                                           int64_t start, int64_t end,
                                           const cctz::time_zone& ctz) const = 0;
 
+    // Doris Native binary columnar format serializer and deserializer.
+    // These APIs are designed to be the low-level building blocks for
+    // NativeReader / VNativeTransformer.
+    //
+    // Contract:
+    //  - write_column_to_native writes [offset, offset + limit) rows of `column`
+    //    into `buf` in a compact binary form, including null map if necessary.
+    //  - read_column_from_native reads exactly `rows` values from `buf` and
+    //    appends them to `column`.
+    //
+    // Default implementation returns NotSupported to allow incremental adoption
+    // by each concrete SerDe.
+    virtual Status write_column_to_native(const IColumn& /*column*/, size_t /*offset*/,
+                                          size_t /*limit*/, BufferWritable& /*buf*/) const {
+        return Status::NotSupported("write_column_to_native is not implemented for type {}",
+                                    get_name());
+    }
+
+    virtual Status read_column_from_native(IColumn& /*column*/, ReadBuffer& /*buf*/,
+                                           size_t /*rows*/) const {
+        return Status::NotSupported("read_column_from_native is not implemented for type {}",
+                                    get_name());
+    }
+
     // ORC serializer
     virtual Status write_column_to_orc(const std::string& timezone, const IColumn& column,
                                        const NullMap* null_map,

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

@@ -17,6 +17,8 @@
 
 #include "data_type_variant_serde.h"
 
+#include <arrow/array/builder_binary.h>
+
 #include <cstdint>
 #include <string>
 
@@ -145,21 +147,34 @@ Status DataTypeVariantSerDe::write_column_to_arrow(const IColumn& column, const
                                                    int64_t start, int64_t end,
                                                    const cctz::time_zone& ctz) const {
     const auto* var = check_and_get_column<ColumnVariant>(column);
-    auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
-    for (size_t i = start; i < end; ++i) {
-        if (null_map && (*null_map)[i]) {
-            RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column.get_name(),
-                                             array_builder->type()->name()));
-        } else {
-            std::string serialized_value;
-            var->serialize_one_row_to_string(i, &serialized_value);
-            RETURN_IF_ERROR(
-                    checkArrowStatus(builder.Append(serialized_value.data(),
-                                                    static_cast<int>(serialized_value.size())),
-                                     column.get_name(), array_builder->type()->name()));
+
+    auto write_impl = [&](auto& builder) -> Status {
+        for (size_t i = start; i < end; ++i) {
+            if (null_map && (*null_map)[i]) {
+                RETURN_IF_ERROR(checkArrowStatus(builder.AppendNull(), column.get_name(),
+                                                 array_builder->type()->name()));
+            } else {
+                std::string serialized_value;
+                var->serialize_one_row_to_string(i, &serialized_value);
+                RETURN_IF_ERROR(
+                        checkArrowStatus(builder.Append(serialized_value.data(),
+                                                        static_cast<int>(serialized_value.size())),
+                                         column.get_name(), array_builder->type()->name()));
+            }
         }
+        return Status::OK();
+    };
+
+    if (array_builder->type()->id() == arrow::Type::LARGE_STRING) {
+        auto& builder = assert_cast<arrow::LargeStringBuilder&>(*array_builder);
+        return write_impl(builder);
+    } else if (array_builder->type()->id() == arrow::Type::STRING) {
+        auto& builder = assert_cast<arrow::StringBuilder&>(*array_builder);
+        return write_impl(builder);
     }
-    return Status::OK();
+
+    return Status::InvalidArgument("Unsupported arrow type for variant column: {}",
+                                   array_builder->type()->name());
 }
 
 void DataTypeVariantSerDe::to_string(const IColumn& column, size_t row_num,