be/src/format/arrow/arrow_row_batch.cpp

Source
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "format/arrow/arrow_row_batch.h"

#include <arrow/buffer.h>
#include <arrow/io/memory.h>
#include <arrow/ipc/writer.h>
#include <arrow/record_batch.h>
#include <arrow/result.h>
#include <arrow/status.h>
#include <arrow/type.h>
#include <arrow/type_fwd.h>
#include <arrow/util/key_value_metadata.h>
#include <glog/logging.h>
#include <stdint.h>

#include <algorithm>
#include <cstdlib>
#include <memory>
#include <utility>
#include <vector>

#include "core/block/block.h"
#include "core/data_type/data_type_agg_state.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_map.h"
#include "core/data_type/data_type_struct.h"
#include "core/data_type/define_primitive_type.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
#include "format/arrow/arrow_block_convertor.h"
#include "runtime/descriptors.h"

namespace doris {

Status convert_to_arrow_type(const DataTypePtr& origin_type,
                             std::shared_ptr<arrow::DataType>* result, const std::string& timezone,
                             bool datetime_naive) {
    auto type = get_serialized_type(origin_type);
    switch (type->get_primitive_type()) {
    case TYPE_NULL:
        *result = arrow::null();
        break;
    case TYPE_TINYINT:
        *result = arrow::int8();
        break;
    case TYPE_SMALLINT:
        *result = arrow::int16();
        break;
    case TYPE_INT:
        *result = arrow::int32();
        break;
    case TYPE_BIGINT:
        *result = arrow::int64();
        break;
    case TYPE_FLOAT:
        *result = arrow::float32();
        break;
    case TYPE_DOUBLE:
        *result = arrow::float64();
        break;
    case TYPE_TIMEV2:
        *result = arrow::float64();
        break;
    case TYPE_IPV4:
        // ipv4 is uint32, but parquet not uint32, it's will be convert to int64
        // so use int32 directly
        *result = arrow::int32();
        break;
    case TYPE_IPV6:
        *result = arrow::utf8();
        break;
    case TYPE_LARGEINT:
    case TYPE_VARCHAR:
    case TYPE_CHAR:
    case TYPE_DATE:
    case TYPE_DATETIME:
    case TYPE_STRING:
    case TYPE_JSONB:
        *result = arrow::utf8();
        break;
    case TYPE_DATEV2:
        *result = std::make_shared<arrow::Date32Type>();
        break;
    case TYPE_TIMESTAMP_NS:
        // TIMESTAMP_NS is stored as signed epoch nanoseconds, but its SQL type has no timezone.
        *result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::NANO);
        break;
    case TYPE_TIMESTAMPTZ:
    case TYPE_DATETIMEV2: {
        arrow::TimeUnit::type time_unit;
        if (type->get_scale() > 3) {
            time_unit = arrow::TimeUnit::MICRO;
        } else if (type->get_scale() > 0) {
            time_unit = arrow::TimeUnit::MILLI;
        } else {
            time_unit = arrow::TimeUnit::SECOND;
        }
        // Doris DATETIMEV2 represents a wall-clock value without a timezone. Arrow Flight
        // exposes it as a timezone-naive timestamp so clients do not interpret it as an instant.
        // This option only changes the DATETIMEV2 output schema. TIMESTAMPTZ remains timezone-aware,
        // and Arrow-to-Doris conversions are unaffected.
        if (type->get_primitive_type() == TYPE_DATETIMEV2 && datetime_naive) {
            *result = std::make_shared<arrow::TimestampType>(time_unit);
        } else {
            *result = std::make_shared<arrow::TimestampType>(time_unit, timezone);
        }
        break;
    }
    case TYPE_DECIMALV2:
    case TYPE_DECIMAL32:
    case TYPE_DECIMAL64:
    case TYPE_DECIMAL128I:
        *result = std::make_shared<arrow::Decimal128Type>(type->get_precision(), type->get_scale());
        break;
    case TYPE_DECIMAL256:
        *result = std::make_shared<arrow::Decimal256Type>(type->get_precision(), type->get_scale());
        break;
    case TYPE_BOOLEAN:
        *result = arrow::boolean();
        break;
    case TYPE_ARRAY: {
        const auto* type_arr = assert_cast<const DataTypeArray*>(remove_nullable(type).get());
        std::shared_ptr<arrow::DataType> item_type;
        RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone,
                                              datetime_naive));
        *result = std::make_shared<arrow::ListType>(item_type);
        break;
    }
    case TYPE_MAP: {
        const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(type).get());
        std::shared_ptr<arrow::DataType> key_type;
        std::shared_ptr<arrow::DataType> val_type;
        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone,
                                              datetime_naive));
        RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone,
                                              datetime_naive));
        *result = std::make_shared<arrow::MapType>(key_type, val_type);
        break;
    }
    case TYPE_STRUCT: {
        const auto* type_struct = assert_cast<const DataTypeStruct*>(remove_nullable(type).get());
        std::vector<std::shared_ptr<arrow::Field>> fields;
        for (size_t i = 0; i < type_struct->get_elements().size(); i++) {
            std::shared_ptr<arrow::DataType> field_type;
            RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type,
                                                  timezone, datetime_naive));
            fields.push_back(
                    std::make_shared<arrow::Field>(type_struct->get_element_name(i), field_type,
                                                   type_struct->get_element(i)->is_nullable()));
        }
        *result = std::make_shared<arrow::StructType>(fields);
        break;
    }
    case TYPE_VARIANT: {
        *result = arrow::utf8();
        break;
    }
    case TYPE_QUANTILE_STATE:
    case TYPE_BITMAP:
    case TYPE_HLL: {
        *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());
    }
    return Status::OK();
}

// Helper function to create an Arrow Field with type metadata if applicable, such as IP types
std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
        const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
        bool is_nullable, PrimitiveType primitive_type) {
    if (primitive_type == PrimitiveType::TYPE_IPV4) {
        auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV4"});
        return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
    } else if (primitive_type == PrimitiveType::TYPE_IPV6) {
        auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV6"});
        return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
    } else if (primitive_type == PrimitiveType::TYPE_LARGEINT) {
        auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"LARGEINT"});
        return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
    } else {
        return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable);
    }
}

Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Schema>* result,
                                   const std::string& timezone, bool datetime_naive) {
    std::vector<std::shared_ptr<arrow::Field>> fields;
    for (const auto& type_and_name : block) {
        std::shared_ptr<arrow::DataType> arrow_type;
        RETURN_IF_ERROR(
                convert_to_arrow_type(type_and_name.type, &arrow_type, timezone, datetime_naive));
        auto field = create_arrow_field_with_metadata(type_and_name.name, arrow_type,
                                                      type_and_name.type->is_nullable(),
                                                      type_and_name.type->get_primitive_type());
        fields.push_back(field);
    }
    *result = arrow::schema(std::move(fields));
    return Status::OK();
}

Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
                                       std::shared_ptr<arrow::Schema>* result,
                                       const std::string& timezone, bool datetime_naive) {
    std::vector<std::shared_ptr<arrow::Field>> fields;
    for (int i = 0; i < output_vexpr_ctxs.size(); i++) {
        std::shared_ptr<arrow::DataType> arrow_type;
        auto root_expr = output_vexpr_ctxs.at(i)->root();
        RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone,
                                              datetime_naive));
        auto field_name = root_expr->is_slot_ref() && !root_expr->expr_label().empty()
                                  ? root_expr->expr_label()
                                  : fmt::format("{}_{}", root_expr->data_type()->get_name(), i);
        auto field =
                create_arrow_field_with_metadata(field_name, arrow_type, root_expr->is_nullable(),
                                                 root_expr->data_type()->get_primitive_type());
        fields.push_back(field);
    }
    *result = arrow::schema(std::move(fields));
    return Status::OK();
}

Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::string* result) {
    // create sink memory buffer outputstream with the computed capacity
    int64_t capacity;
    arrow::Status a_st = arrow::ipc::GetRecordBatchSize(record_batch, &capacity);
    if (!a_st.ok()) {
        return Status::InternalError("GetRecordBatchSize failure, reason: {}", a_st.ToString());
    }
    auto sink_res = arrow::io::BufferOutputStream::Create(capacity, arrow::default_memory_pool());
    if (!sink_res.ok()) {
        return Status::InternalError("create BufferOutputStream failure, reason: {}",
                                     sink_res.status().ToString());
    }
    std::shared_ptr<arrow::io::BufferOutputStream> sink = sink_res.ValueOrDie();
    // create RecordBatch Writer
    auto res = arrow::ipc::MakeStreamWriter(sink.get(), record_batch.schema());
    if (!res.ok()) {
        return Status::InternalError("open RecordBatchStreamWriter failure, reason: {}",
                                     res.status().ToString());
    }
    // write RecordBatch to memory buffer outputstream
    std::shared_ptr<arrow::ipc::RecordBatchWriter> record_batch_writer = res.ValueOrDie();
    a_st = record_batch_writer->WriteRecordBatch(record_batch);
    if (!a_st.ok()) {
        return Status::InternalError("write record batch failure, reason: {}", a_st.ToString());
    }
    a_st = record_batch_writer->Close();
    if (!a_st.ok()) {
        return Status::InternalError("Close failed, reason: {}", a_st.ToString());
    }
    auto finish_res = sink->Finish();
    if (!finish_res.ok()) {
        return Status::InternalError("allocate result buffer failure, reason: {}",
                                     finish_res.status().ToString());
    }
    *result = finish_res.ValueOrDie()->ToString();
    // close the sink
    a_st = sink->Close();
    if (!a_st.ok()) {
        return Status::InternalError("Close failed, reason: {}", a_st.ToString());
    }
    return Status::OK();
}

Status serialize_arrow_schema(std::shared_ptr<arrow::Schema>* schema, std::string* result) {
    auto make_empty_result = arrow::RecordBatch::MakeEmpty(*schema);
    if (!make_empty_result.ok()) {
        return Status::InternalError("serialize_arrow_schema failed, reason: {}",
                                     make_empty_result.status().ToString());
    }
    auto batch = make_empty_result.ValueOrDie();
    return serialize_record_batch(*batch, result);
}

} // namespace doris

Coverage Report

Created: 2026-09-09 00:32

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "format/arrow/arrow_row_batch.h"
19
20		#include <arrow/buffer.h>
21		#include <arrow/io/memory.h>
22		#include <arrow/ipc/writer.h>
23		#include <arrow/record_batch.h>
24		#include <arrow/result.h>
25		#include <arrow/status.h>
26		#include <arrow/type.h>
27		#include <arrow/type_fwd.h>
28		#include <arrow/util/key_value_metadata.h>
29		#include <glog/logging.h>
30		#include <stdint.h>
31
32		#include <algorithm>
33		#include <cstdlib>
34		#include <memory>
35		#include <utility>
36		#include <vector>
37
38		#include "core/block/block.h"
39		#include "core/data_type/data_type_agg_state.h"
40		#include "core/data_type/data_type_array.h"
41		#include "core/data_type/data_type_map.h"
42		#include "core/data_type/data_type_struct.h"
43		#include "core/data_type/define_primitive_type.h"
44		#include "exprs/vexpr.h"
45		#include "exprs/vexpr_context.h"
46		#include "format/arrow/arrow_block_convertor.h"
47		#include "runtime/descriptors.h"
48
49		namespace doris {
50
51		Status convert_to_arrow_type(const DataTypePtr& origin_type,
52		std::shared_ptr<arrow::DataType>* result, const std::string& timezone,
53	586	bool datetime_naive) {
54	586	auto type = get_serialized_type(origin_type);
55	586	switch (type->get_primitive_type()) {
56	0	case TYPE_NULL:
57	0	*result = arrow::null();
58	0	break;
59	9	case TYPE_TINYINT:
60	9	*result = arrow::int8();
61	9	break;
62	9	case TYPE_SMALLINT:
63	9	*result = arrow::int16();
64	9	break;
65	38	case TYPE_INT:
66	38	*result = arrow::int32();
67	38	break;
68	9	case TYPE_BIGINT:
69	9	*result = arrow::int64();
70	9	break;
71	9	case TYPE_FLOAT:
72	9	*result = arrow::float32();
73	9	break;
74	22	case TYPE_DOUBLE:
75	22	*result = arrow::float64();
76	22	break;
77	0	case TYPE_TIMEV2:
78	0	*result = arrow::float64();
79	0	break;
80	25	case TYPE_IPV4:
81		// ipv4 is uint32, but parquet not uint32, it's will be convert to int64
82		// so use int32 directly
83	25	*result = arrow::int32();
84	25	break;
85	20	case TYPE_IPV6:
86	20	*result = arrow::utf8();
87	20	break;
88	23	case TYPE_LARGEINT:
89	23	case TYPE_VARCHAR:
90	23	case TYPE_CHAR:
91	40	case TYPE_DATE:
92	53	case TYPE_DATETIME:
93	102	case TYPE_STRING:
94	102	case TYPE_JSONB:
95	102	*result = arrow::utf8();
96	102	break;
97	13	case TYPE_DATEV2:
98	13	*result = std::make_shared<arrow::Date32Type>();
99	13	break;
100	6	case TYPE_TIMESTAMP_NS:
101		// TIMESTAMP_NS is stored as signed epoch nanoseconds, but its SQL type has no timezone.
102	6	*result = std::make_shared<arrow::TimestampType>(arrow::TimeUnit::NANO);
103	6	break;
104	1	case TYPE_TIMESTAMPTZ:
105	22	case TYPE_DATETIMEV2: {
106	22	arrow::TimeUnit::type time_unit;
107	22	if (type->get_scale() > 3) {
108	4	time_unit = arrow::TimeUnit::MICRO;
109	18	} else if (type->get_scale() > 0) {
110	4	time_unit = arrow::TimeUnit::MILLI;
111	14	} else {
112	14	time_unit = arrow::TimeUnit::SECOND;
113	14	}
114		// Doris DATETIMEV2 represents a wall-clock value without a timezone. Arrow Flight
115		// exposes it as a timezone-naive timestamp so clients do not interpret it as an instant.
116		// This option only changes the DATETIMEV2 output schema. TIMESTAMPTZ remains timezone-aware,
117		// and Arrow-to-Doris conversions are unaffected.
118	22	if (type->get_primitive_type() == TYPE_DATETIMEV2 && datetime_naive) {
119	2	*result = std::make_shared<arrow::TimestampType>(time_unit);
120	20	} else {
121	20	*result = std::make_shared<arrow::TimestampType>(time_unit, timezone);
122	20	}
123	22	break;
124	1	}
125	5	case TYPE_DECIMALV2:
126	27	case TYPE_DECIMAL32:
127	43	case TYPE_DECIMAL64:
128	55	case TYPE_DECIMAL128I:
129	55	*result = std::make_shared<arrow::Decimal128Type>(type->get_precision(), type->get_scale());
130	55	break;
131	16	case TYPE_DECIMAL256:
132	16	*result = std::make_shared<arrow::Decimal256Type>(type->get_precision(), type->get_scale());
133	16	break;
134	9	case TYPE_BOOLEAN:
135	9	*result = arrow::boolean();
136	9	break;
137	103	case TYPE_ARRAY: {
138	103	const auto* type_arr = assert_cast<const DataTypeArray*>(remove_nullable(type).get());
139	103	std::shared_ptr<arrow::DataType> item_type;
140	103	RETURN_IF_ERROR(convert_to_arrow_type(type_arr->get_nested_type(), &item_type, timezone,
141	103	datetime_naive));
142	103	*result = std::make_shared<arrow::ListType>(item_type);
143	103	break;
144	103	}
145	63	case TYPE_MAP: {
146	63	const auto* type_map = assert_cast<const DataTypeMap*>(remove_nullable(type).get());
147	63	std::shared_ptr<arrow::DataType> key_type;
148	63	std::shared_ptr<arrow::DataType> val_type;
149	63	RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_key_type(), &key_type, timezone,
150	63	datetime_naive));
151	63	RETURN_IF_ERROR(convert_to_arrow_type(type_map->get_value_type(), &val_type, timezone,
152	63	datetime_naive));
153	63	*result = std::make_shared<arrow::MapType>(key_type, val_type);
154	63	break;
155	63	}
156	54	case TYPE_STRUCT: {
157	54	const auto* type_struct = assert_cast<const DataTypeStruct*>(remove_nullable(type).get());
158	54	std::vector<std::shared_ptr<arrow::Field>> fields;
159	178	for (size_t i = 0; i < type_struct->get_elements().size(); i++) {
160	124	std::shared_ptr<arrow::DataType> field_type;
161	124	RETURN_IF_ERROR(convert_to_arrow_type(type_struct->get_element(i), &field_type,
162	124	timezone, datetime_naive));
163	124	fields.push_back(
164	124	std::make_shared<arrow::Field>(type_struct->get_element_name(i), field_type,
165	124	type_struct->get_element(i)->is_nullable()));
166	124	}
167	54	*result = std::make_shared<arrow::StructType>(fields);
168	54	break;
169	54	}
170	0	case TYPE_VARIANT: {
171	0	*result = arrow::utf8();
172	0	break;
173	54	}
174	0	case TYPE_QUANTILE_STATE:
175	1	case TYPE_BITMAP:
176	2	case TYPE_HLL: {
177	2	*result = arrow::binary();
178	2	break;
179	1	}
180	0	case TYPE_VARBINARY: {
181	0	*result = arrow::binary();
182	0	break;
183	1	}
184	0	default:
185	0	return Status::InvalidArgument("Unknown primitive type({}) convert to Arrow type",
186	0	type->get_name());
187	586	}
188	586	return Status::OK();
189	586	}
190
191		// Helper function to create an Arrow Field with type metadata if applicable, such as IP types
192		std::shared_ptr<arrow::Field> create_arrow_field_with_metadata(
193		const std::string& field_name, const std::shared_ptr<arrow::DataType>& arrow_type,
194	212	bool is_nullable, PrimitiveType primitive_type) {
195	212	if (primitive_type == PrimitiveType::TYPE_IPV4) {
196	4	auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV4"});
197	4	return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
198	208	} else if (primitive_type == PrimitiveType::TYPE_IPV6) {
199	4	auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"IPV6"});
200	4	return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
201	204	} else if (primitive_type == PrimitiveType::TYPE_LARGEINT) {
202	4	auto metadata = arrow::KeyValueMetadata::Make({"doris_type"}, {"LARGEINT"});
203	4	return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable, metadata);
204	200	} else {
205	200	return std::make_shared<arrow::Field>(field_name, arrow_type, is_nullable);
206	200	}
207	212	}
208
209		Status get_arrow_schema_from_block(const Block& block, std::shared_ptr<arrow::Schema>* result,
210	31	const std::string& timezone, bool datetime_naive) {
211	31	std::vector<std::shared_ptr<arrow::Field>> fields;
212	212	for (const auto& type_and_name : block) {
213	212	std::shared_ptr<arrow::DataType> arrow_type;
214	212	RETURN_IF_ERROR(
215	212	convert_to_arrow_type(type_and_name.type, &arrow_type, timezone, datetime_naive));
216	212	auto field = create_arrow_field_with_metadata(type_and_name.name, arrow_type,
217	212	type_and_name.type->is_nullable(),
218	212	type_and_name.type->get_primitive_type());
219	212	fields.push_back(field);
220	212	}
221	31	*result = arrow::schema(std::move(fields));
222	31	return Status::OK();
223	31	}
224
225		Status get_arrow_schema_from_expr_ctxs(const VExprContextSPtrs& output_vexpr_ctxs,
226		std::shared_ptr<arrow::Schema>* result,
227	0	const std::string& timezone, bool datetime_naive) {
228	0	std::vector<std::shared_ptr<arrow::Field>> fields;
229	0	for (int i = 0; i < output_vexpr_ctxs.size(); i++) {
230	0	std::shared_ptr<arrow::DataType> arrow_type;
231	0	auto root_expr = output_vexpr_ctxs.at(i)->root();
232	0	RETURN_IF_ERROR(convert_to_arrow_type(root_expr->data_type(), &arrow_type, timezone,
233	0	datetime_naive));
234	0	auto field_name = root_expr->is_slot_ref() && !root_expr->expr_label().empty()
235	0	? root_expr->expr_label()
236	0	: fmt::format("{}_{}", root_expr->data_type()->get_name(), i);
237	0	auto field =
238	0	create_arrow_field_with_metadata(field_name, arrow_type, root_expr->is_nullable(),
239	0	root_expr->data_type()->get_primitive_type());
240	0	fields.push_back(field);
241	0	}
242	0	*result = arrow::schema(std::move(fields));
243	0	return Status::OK();
244	0	}
245
246	0	Status serialize_record_batch(const arrow::RecordBatch& record_batch, std::string* result) {
247		// create sink memory buffer outputstream with the computed capacity
248	0	int64_t capacity;
249	0	arrow::Status a_st = arrow::ipc::GetRecordBatchSize(record_batch, &capacity);
250	0	if (!a_st.ok()) {
251	0	return Status::InternalError("GetRecordBatchSize failure, reason: {}", a_st.ToString());
252	0	}
253	0	auto sink_res = arrow::io::BufferOutputStream::Create(capacity, arrow::default_memory_pool());
254	0	if (!sink_res.ok()) {
255	0	return Status::InternalError("create BufferOutputStream failure, reason: {}",
256	0	sink_res.status().ToString());
257	0	}
258	0	std::shared_ptr<arrow::io::BufferOutputStream> sink = sink_res.ValueOrDie();
259		// create RecordBatch Writer
260	0	auto res = arrow::ipc::MakeStreamWriter(sink.get(), record_batch.schema());
261	0	if (!res.ok()) {
262	0	return Status::InternalError("open RecordBatchStreamWriter failure, reason: {}",
263	0	res.status().ToString());
264	0	}
265		// write RecordBatch to memory buffer outputstream
266	0	std::shared_ptr<arrow::ipc::RecordBatchWriter> record_batch_writer = res.ValueOrDie();
267	0	a_st = record_batch_writer->WriteRecordBatch(record_batch);
268	0	if (!a_st.ok()) {
269	0	return Status::InternalError("write record batch failure, reason: {}", a_st.ToString());
270	0	}
271	0	a_st = record_batch_writer->Close();
272	0	if (!a_st.ok()) {
273	0	return Status::InternalError("Close failed, reason: {}", a_st.ToString());
274	0	}
275	0	auto finish_res = sink->Finish();
276	0	if (!finish_res.ok()) {
277	0	return Status::InternalError("allocate result buffer failure, reason: {}",
278	0	finish_res.status().ToString());
279	0	}
280	0	*result = finish_res.ValueOrDie()->ToString();
281		// close the sink
282	0	a_st = sink->Close();
283	0	if (!a_st.ok()) {
284	0	return Status::InternalError("Close failed, reason: {}", a_st.ToString());
285	0	}
286	0	return Status::OK();
287	0	}
288
289	0	Status serialize_arrow_schema(std::shared_ptr<arrow::Schema>* schema, std::string* result) {
290	0	auto make_empty_result = arrow::RecordBatch::MakeEmpty(*schema);
291	0	if (!make_empty_result.ok()) {
292	0	return Status::InternalError("serialize_arrow_schema failed, reason: {}",
293	0	make_empty_result.status().ToString());
294	0	}
295	0	auto batch = make_empty_result.ValueOrDie();
296	0	return serialize_record_batch(*batch, result);
297	0	}
298
299		} // namespace doris