[BugFix] Support inserting array type to hive table with CSV format (backport #67355) (#69089)

Signed-off-by: GavinMar <yangguansuo@starrocks.com>

Author: GavinMar

be/src/formats/csv/array_converter.cpp

@@ -27,14 +27,33 @@ Status ArrayConverter::write_string(OutputStream* os, const Column& column, size
 
     auto begin = offsets.get_data()[row_num];
     auto end = offsets.get_data()[row_num + 1];
+    char collection_delim = ',';
+    bool is_hive_format = false;
+
+    if (options.array_format_type == ArrayFormatType::kHive) {
+        collection_delim = HiveTextArrayReader::get_collection_delimiter(options.array_hive_collection_delimiter,
+                                                                         options.array_hive_mapkey_delimiter,
+                                                                         options.array_hive_nested_level);
+        is_hive_format = true;
+    } else {
+        RETURN_IF_ERROR(os->write('['));
+    }
 
-    RETURN_IF_ERROR(os->write('['));
     for (auto i = begin; i < end; i++) {
-        RETURN_IF_ERROR(_element_converter->write_quoted_string(os, elements, i, options));
+        if (is_hive_format) {
+            Options sub_options = options;
+            sub_options.array_hive_nested_level++;
+            // In Hive, we should use write_string() instead of write_quoted_string
+            RETURN_IF_ERROR(_element_converter->write_string(os, elements, i, sub_options));
+        } else {
+            RETURN_IF_ERROR(_element_converter->write_quoted_string(os, elements, i, options));
+        }
         if (i + 1 < end) {
-            RETURN_IF_ERROR(os->write(','));
+            RETURN_IF_ERROR(os->write(collection_delim));
         }
     }
+    RETURN_IF(is_hive_format, Status::OK());
+
     return os->write(']');
 }
 

be/src/formats/csv/csv_file_writer.cpp

@@ -19,6 +19,7 @@
 
 #include "common/http/content_type.h"
 #include "csv_escape.h"
+#include "exec/hdfs_scanner_text.h"
 #include "formats/utils.h"
 #include "output_stream_file.h"
 #include "runtime/current_thread.h"
@@ -55,6 +56,17 @@ Status CSVFileWriter::init() {
         }
         _column_converters.emplace_back(std::move(nullable_conv), csv::get_converter(type, false));
     }
+    _converter_options = std::make_shared<csv::Converter::Options>();
+    if (_writer_options->is_hive) {
+        _converter_options->is_hive = true;
+        _converter_options->array_format_type = csv::ArrayFormatType::kHive;
+        _converter_options->array_hive_collection_delimiter = _writer_options->collection_delim.empty()
+                                                                      ? DEFAULT_COLLECTION_DELIM.front()
+                                                                      : _writer_options->collection_delim.front();
+        _converter_options->array_hive_mapkey_delimiter = _writer_options->mapkey_delim.empty()
+                                                                  ? DEFAULT_MAPKEY_DELIM.front()
+                                                                  : _writer_options->mapkey_delim.front();
+    }
     return Status::OK();
 }
 
@@ -115,7 +127,7 @@ Status CSVFileWriter::write(Chunk* chunk) {
             } else {
                 converter = _column_converters[c].second.get();
             }
-            RETURN_IF_ERROR(converter->write_string(_output_stream.get(), *columns[c], r, {}));
+            RETURN_IF_ERROR(converter->write_string(_output_stream.get(), *columns[c], r, *_converter_options));
             if (c + 1 != columns.size()) {
                 RETURN_IF_ERROR(_output_stream->write(_writer_options->column_terminated_by));
             }
@@ -174,6 +186,15 @@ Status CSVFileWriterFactory::init() {
     if (_options.contains(CSVWriterOptions::INCLUDE_HEADER)) {
         _parsed_options->include_header = boost::iequals(_options[CSVWriterOptions::INCLUDE_HEADER], "true");
     }
+    if (_options.contains(CSVWriterOptions::COLLECTION_DELIM)) {
+        _parsed_options->collection_delim = _options[CSVWriterOptions::COLLECTION_DELIM];
+    }
+    if (_options.contains(CSVWriterOptions::MAPKEY_DELIM)) {
+        _parsed_options->mapkey_delim = _options[CSVWriterOptions::MAPKEY_DELIM];
+    }
+    if (_options.contains(CSVWriterOptions::IS_HIVE)) {
+        _parsed_options->is_hive = _options[CSVWriterOptions::IS_HIVE] == "true";
+    }
     return Status::OK();
 }
 

be/src/formats/csv/csv_file_writer.h

@@ -24,10 +24,16 @@ struct CSVWriterOptions : FileWriterOptions {
     std::string column_terminated_by = ",";
     std::string line_terminated_by = "\n";
     bool include_header = false;
+    std::string collection_delim = ",";
+    std::string mapkey_delim = ",";
+    bool is_hive = false;
 
     inline static std::string COLUMN_TERMINATED_BY = "column_terminated_by";
     inline static std::string LINE_TERMINATED_BY = "line_terminated_by";
     inline static std::string INCLUDE_HEADER = "include_header";
+    inline static std::string COLLECTION_DELIM = "collection_delim";
+    inline static std::string MAPKEY_DELIM = "mapkey_delim";
+    inline static std::string IS_HIVE = "is_hive";
 };
 
 // The primary purpose of this class is to support hive + csv. Use with caution in other cases.
@@ -59,6 +65,7 @@ class CSVFileWriter final : public FileWriter {
     std::vector<std::unique_ptr<ColumnEvaluator>> _column_evaluators;
     std::shared_ptr<CSVWriterOptions> _writer_options;
     const std::function<void()> _rollback_action;
+    std::shared_ptr<csv::Converter::Options> _converter_options;
 
     int64_t _num_rows = 0;
     bool _header_written = false;

be/src/runtime/hive_table_sink.cpp

@@ -82,6 +82,9 @@ Status HiveTableSink::decompose_to_pipeline(pipeline::OpFactories prev_operators
         DCHECK(boost::iequals(t_hive_sink.file_format, formats::TEXTFILE));
         sink_ctx->options[formats::CSVWriterOptions::COLUMN_TERMINATED_BY] = DEFAULT_FIELD_DELIM;
         sink_ctx->options[formats::CSVWriterOptions::LINE_TERMINATED_BY] = DEFAULT_LINE_DELIM;
+        sink_ctx->options[formats::CSVWriterOptions::COLLECTION_DELIM] = DEFAULT_COLLECTION_DELIM;
+        sink_ctx->options[formats::CSVWriterOptions::MAPKEY_DELIM] = DEFAULT_MAPKEY_DELIM;
+        sink_ctx->options[formats::CSVWriterOptions::IS_HIVE] = "true";
 
         // use customized value if specified
         if (t_hive_sink.text_file_desc.__isset.field_delim) {
@@ -90,6 +93,13 @@ Status HiveTableSink::decompose_to_pipeline(pipeline::OpFactories prev_operators
         if (t_hive_sink.text_file_desc.__isset.line_delim) {
             sink_ctx->options[formats::CSVWriterOptions::LINE_TERMINATED_BY] = t_hive_sink.text_file_desc.line_delim;
         }
+        if (t_hive_sink.text_file_desc.__isset.collection_delim) {
+            sink_ctx->options[formats::CSVWriterOptions::COLLECTION_DELIM] =
+                    t_hive_sink.text_file_desc.collection_delim;
+        }
+        if (t_hive_sink.text_file_desc.__isset.mapkey_delim) {
+            sink_ctx->options[formats::CSVWriterOptions::MAPKEY_DELIM] = t_hive_sink.text_file_desc.mapkey_delim;
+        }
     }
     sink_ctx->fragment_context = fragment_ctx;
 

be/test/formats/csv/csv_file_writer_test.cpp

@@ -569,6 +569,74 @@ TEST_F(CSVFileWriterTest, TestWriteArrayWithNull) {
     ASSERT_EQ(content, expect);
 }
 
+TEST_F(CSVFileWriterTest, TestWriteHiveArray) {
+    // type_descs: ARRAY<INT>
+    std::vector<TypeDescriptor> type_descs;
+    auto type_int = TypeDescriptor::from_logical_type(TYPE_INT);
+    auto type_int_array = TypeDescriptor::from_logical_type(TYPE_ARRAY);
+    type_int_array.children.push_back(type_int);
+    type_descs.push_back(type_int_array);
+
+    auto column_names = _make_type_names(type_descs);
+    auto maybe_output_file = _fs.new_writable_file(_file_path);
+    EXPECT_OK(maybe_output_file.status());
+    auto output_file = std::move(maybe_output_file.value());
+    auto output_stream = std::make_unique<csv::OutputStreamFile>(std::move(output_file), 1024);
+    auto column_evaluators = ColumnSlotIdEvaluator::from_types(type_descs);
+    auto writer_options = std::make_shared<formats::CSVWriterOptions>();
+    writer_options->is_hive = true;
+    writer_options->collection_delim = '\002';
+    auto writer =
+            std::make_unique<formats::CSVFileWriter>(_file_path, std::move(output_stream), column_names, type_descs,
+                                                     std::move(column_evaluators), writer_options, []() {});
+    ASSERT_OK(writer->init());
+
+    auto chunk = std::make_shared<Chunk>();
+    {
+        // Create ARRAY<INT> column with data: [1,2,3], [4,5], [], [6]
+        auto elements_data = Int32Column::create();
+        auto offsets = UInt32Column::create();
+        auto null_column = UInt8Column::create();
+
+        // row 0: [1,2,3]
+        elements_data->append(1);
+        elements_data->append(2);
+        elements_data->append(3);
+        offsets->append(0);
+
+        // row 1: [4,5]
+        elements_data->append(4);
+        elements_data->append(5);
+        offsets->append(3);
+
+        // row 2: [] (empty array)
+        offsets->append(5);
+
+        // row 3: [6]
+        elements_data->append(6);
+        offsets->append(5);
+        offsets->append(6);
+
+        auto array_column =
+                ArrayColumn::create(NullableColumn::create(elements_data, UInt8Column::create(6, 0)), offsets);
+        null_column->append_numbers(std::vector<uint8_t>{0, 0, 0, 0}.data(), 4);
+        auto nullable_column = NullableColumn::create(std::move(array_column), std::move(null_column));
+        chunk->append_column(std::move(nullable_column), chunk->num_columns());
+    }
+
+    // write chunk
+    ASSERT_OK(writer->write(chunk.get()));
+    auto result = writer->commit();
+    ASSERT_OK(result.io_status);
+    ASSERT_EQ(result.file_statistics.record_count, 4);
+
+    // verify correctness
+    std::string content;
+    ASSERT_OK(_fs.read_file(_file_path, &content));
+    std::string expect = "1\0022\0023\n4\0025\n\n6\n";
+    ASSERT_EQ(content, expect);
+}
+
 TEST_F(CSVFileWriterTest, TestWriteMap) {
     // type_descs
     std::vector<TypeDescriptor> type_descs;

test/sql/test_hive/R/test_write_csv_array

@@ -0,0 +1,35 @@
+-- name: test_write_csv_array
+create external catalog hive_sink_test_${uuid0} PROPERTIES ("type"="hive", "hive.metastore.uris"="${hive_metastore_uris}");
+-- result:
+-- !result
+set catalog hive_sink_test_${uuid0};
+-- result:
+-- !result
+create database hive_db_${uuid0};
+-- result:
+-- !result
+use hive_db_${uuid0};
+-- result:
+-- !result
+create table t1(c1 string, c2 array<integer>) properties("file_format"="textfile");
+-- result:
+-- !result
+insert into t1(c1,c2) values('hello', [123456,0,-128,0]);
+-- result:
+-- !result
+select * from t1;
+-- result:
+hello	[123456,0,-128,0]
+-- !result
+drop table t1 force;
+-- result:
+-- !result
+drop database hive_db_${uuid0};
+-- result:
+-- !result
+drop catalog hive_sink_test_${uuid0};
+-- result:
+-- !result
+set catalog default_catalog;
+-- result:
+-- !result

test/sql/test_hive/T/test_write_csv_array

@@ -0,0 +1,16 @@
+-- name: test_write_csv_array
+
+create external catalog hive_sink_test_${uuid0} PROPERTIES ("type"="hive", "hive.metastore.uris"="${hive_metastore_uris}");
+set catalog hive_sink_test_${uuid0};
+create database hive_db_${uuid0};
+use hive_db_${uuid0};
+
+create table t1(c1 string, c2 array<integer>) properties("file_format"="textfile");
+insert into t1(c1,c2) values('hello', [123456,0,-128,0]);
+
+select * from t1;
+drop table t1 force;
+
+drop database hive_db_${uuid0};
+drop catalog hive_sink_test_${uuid0};
+set catalog default_catalog;