This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 4536166807d [fix](Export) Export data to csv file format with BOM
(#31223)
4536166807d is described below
commit 4536166807dbcceb2a0ff6dca10a31c75495d0b1
Author: Tiewei Fang <[email protected]>
AuthorDate: Thu Feb 22 17:58:45 2024 +0800
[fix](Export) Export data to csv file format with BOM (#31223)
back-port: #30533
---
be/src/vec/exec/format/csv/csv_reader.cpp | 35 ++-
be/src/vec/exec/format/csv/csv_reader.h | 5 +
be/src/vec/runtime/vfile_result_writer.cpp | 12 +
be/src/vec/runtime/vfile_result_writer.h | 2 +
be/src/vec/sink/vresult_sink.h | 3 +
.../Manipulation/EXPORT.md | 11 +
.../Manipulation/EXPORT.md | 2 +
.../java/org/apache/doris/analysis/ExportStmt.java | 9 +
.../org/apache/doris/analysis/OutFileClause.java | 8 +
.../main/java/org/apache/doris/load/ExportJob.java | 9 +
.../main/java/org/apache/doris/load/ExportMgr.java | 1 +
gensrc/thrift/DataSinks.thrift | 1 +
regression-test/data/export_p0/test_with_bom.out | 66 ++++++
.../suites/export_p0/test_with_bom.groovy | 241 +++++++++++++++++++++
14 files changed, 404 insertions(+), 1 deletion(-)
diff --git a/be/src/vec/exec/format/csv/csv_reader.cpp
b/be/src/vec/exec/format/csv/csv_reader.cpp
index fb3ee5c5be3..d9b166509f8 100644
--- a/be/src/vec/exec/format/csv/csv_reader.cpp
+++ b/be/src/vec/exec/format/csv/csv_reader.cpp
@@ -465,13 +465,24 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
size_t rows = 0;
bool success = false;
+ bool is_remove_bom = false;
if (_push_down_agg_type == TPushAggOp::type::COUNT) {
while (rows < batch_size && !_line_reader_eof) {
const uint8_t* ptr = nullptr;
size_t size = 0;
RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
+
+ // _skip_lines == 0 means this line is the actual data beginning
line for the entire file
+ // is_remove_bom means _remove_bom should only execute once
+ if (_skip_lines == 0 && !is_remove_bom) {
+ ptr = _remove_bom(ptr, size);
+ is_remove_bom = true;
+ }
+
+ // _skip_lines > 0 means we do not need to remove bom
if (_skip_lines > 0) {
_skip_lines--;
+ is_remove_bom = true;
continue;
}
if (size == 0) {
@@ -493,8 +504,18 @@ Status CsvReader::get_next_block(Block* block, size_t*
read_rows, bool* eof) {
const uint8_t* ptr = nullptr;
size_t size = 0;
RETURN_IF_ERROR(_line_reader->read_line(&ptr, &size,
&_line_reader_eof, _io_ctx));
+
+ // _skip_lines == 0 means this line is the actual data beginning
line for the entire file
+ // is_remove_bom means _remove_bom should only execute once
+ if (!is_remove_bom && _skip_lines == 0) {
+ ptr = _remove_bom(ptr, size);
+ is_remove_bom = true;
+ }
+
+ // _skip_lines > 0 means we do not remove bom
if (_skip_lines > 0) {
_skip_lines--;
+ is_remove_bom = true;
continue;
}
if (size == 0) {
@@ -540,10 +561,11 @@ Status
CsvReader::get_parsed_schema(std::vector<std::string>* col_names,
} else { // parse csv file with names
RETURN_IF_ERROR(_parse_col_names(col_names));
}
+
for (size_t j = 0; j < col_names->size(); ++j) {
col_types->emplace_back(TypeDescriptor::create_string_type());
}
- } else { // parse csv file without names and types
+ } else { // parse csv file with names and types
RETURN_IF_ERROR(_parse_col_names(col_names));
RETURN_IF_ERROR(_parse_col_types(col_names->size(), col_types));
}
@@ -925,6 +947,7 @@ Status CsvReader::_parse_col_nums(size_t* col_nums) {
if (!validate_utf8(const_cast<char*>(reinterpret_cast<const char*>(ptr)),
size)) {
return Status::InternalError<false>("Only support csv data in utf8
codec");
}
+ ptr = _remove_bom(ptr, size);
_split_line(Slice(ptr, size));
*col_nums = _split_values.size();
return Status::OK();
@@ -941,6 +964,7 @@ Status
CsvReader::_parse_col_names(std::vector<std::string>* col_names) {
if (!validate_utf8(const_cast<char*>(reinterpret_cast<const char*>(ptr)),
size)) {
return Status::InternalError<false>("Only support csv data in utf8
codec");
}
+ ptr = _remove_bom(ptr, size);
_split_line(Slice(ptr, size));
for (size_t idx = 0; idx < _split_values.size(); ++idx) {
col_names->emplace_back(_split_values[idx].to_string());
@@ -964,6 +988,15 @@ Status CsvReader::_parse_col_types(size_t col_nums,
std::vector<TypeDescriptor>*
return Status::OK();
}
+const uint8_t* CsvReader::_remove_bom(const uint8_t* ptr, size_t& size) {
+ if (size >= 3 && ptr[0] == 0xEF && ptr[1] == 0xBB && ptr[2] == 0xBF) {
+ LOG(INFO) << "remove bom";
+ size -= 3;
+ return ptr + 3;
+ }
+ return ptr;
+}
+
void CsvReader::close() {
if (_line_reader) {
_line_reader->close();
diff --git a/be/src/vec/exec/format/csv/csv_reader.h
b/be/src/vec/exec/format/csv/csv_reader.h
index 3a6d721beea..93d8c16a964 100644
--- a/be/src/vec/exec/format/csv/csv_reader.h
+++ b/be/src/vec/exec/format/csv/csv_reader.h
@@ -231,6 +231,11 @@ private:
// and the line is skipped as unqualified row, and the process should
continue.
Status _validate_line(const Slice& line, bool* success);
+ // If the CSV file is an UTF8 encoding with BOM,
+ // then remove the first 3 bytes at the beginning of this file
+ // and set size = size - 3.
+ const uint8_t* _remove_bom(const uint8_t* ptr, size_t& size);
+
RuntimeState* _state;
RuntimeProfile* _profile;
ScannerCounter* _counter;
diff --git a/be/src/vec/runtime/vfile_result_writer.cpp
b/be/src/vec/runtime/vfile_result_writer.cpp
index 080a351fa81..d6768fc7aff 100644
--- a/be/src/vec/runtime/vfile_result_writer.cpp
+++ b/be/src/vec/runtime/vfile_result_writer.cpp
@@ -69,6 +69,9 @@
#include "vec/sink/vresult_sink.h"
namespace doris::vectorized {
+
+static const unsigned char bom[] = {0xEF, 0xBB, 0xBF};
+
const size_t VFileResultWriter::OUTSTREAM_BUFFER_SIZE_BYTES = 1024 * 1024;
using doris::operator<<;
@@ -188,6 +191,7 @@ Status VFileResultWriter::_get_next_file_name(std::string*
file_name) {
ss << _file_opts->file_path << print_id(_fragment_instance_id) << "_" <<
(_file_idx++) << "."
<< suffix;
*file_name = ss.str();
+ _bom_sent = false;
_header_sent = false;
if (_storage_type == TStorageBackendType::LOCAL) {
// For local file writer, the file_path is a local dir.
@@ -430,6 +434,12 @@ std::string VFileResultWriter::gen_types() {
}
Status VFileResultWriter::write_csv_header() {
+ if (!_bom_sent && _file_opts->with_bom) {
+ RETURN_IF_ERROR(_file_writer_impl->append(reinterpret_cast<const
char*>(bom)));
+ _bom_sent = true;
+ _current_written_bytes += 3;
+ COUNTER_UPDATE(_written_data_bytes, 3);
+ }
if (!_header_sent && _header.size() > 0) {
std::string tmp_header = _header;
if (_header_type == BeConsts::CSV_WITH_NAMES_AND_TYPES) {
@@ -437,6 +447,8 @@ Status VFileResultWriter::write_csv_header() {
}
RETURN_IF_ERROR(_file_writer_impl->append(tmp_header));
_header_sent = true;
+ _current_written_bytes += tmp_header.length();
+ COUNTER_UPDATE(_written_data_bytes, tmp_header.length());
}
return Status::OK();
}
diff --git a/be/src/vec/runtime/vfile_result_writer.h
b/be/src/vec/runtime/vfile_result_writer.h
index 4ec425755d3..0693e4291f5 100644
--- a/be/src/vec/runtime/vfile_result_writer.h
+++ b/be/src/vec/runtime/vfile_result_writer.h
@@ -142,5 +142,7 @@ private:
RowDescriptor _output_row_descriptor;
// parquet/orc file writer
std::unique_ptr<VFileWriterWrapper> _vfile_writer;
+
+ bool _bom_sent = false;
};
} // namespace doris::vectorized
diff --git a/be/src/vec/sink/vresult_sink.h b/be/src/vec/sink/vresult_sink.h
index b144bd0d22b..d288dd45164 100644
--- a/be/src/vec/sink/vresult_sink.h
+++ b/be/src/vec/sink/vresult_sink.h
@@ -70,6 +70,8 @@ struct ResultFileOptions {
bool delete_existing_files = false;
std::string file_suffix;
+ //Bring BOM when exporting to CSV format
+ bool with_bom = false;
ResultFileOptions(const TResultFileSinkOptions& t_opt) {
file_path = t_opt.file_path;
@@ -81,6 +83,7 @@ struct ResultFileOptions {
delete_existing_files =
t_opt.__isset.delete_existing_files ?
t_opt.delete_existing_files : false;
file_suffix = t_opt.file_suffix;
+ with_bom = t_opt.with_bom;
is_local_file = true;
if (t_opt.__isset.broker_addresses) {
diff --git
a/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
b/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
index d7c10b7d059..347d5d5dd88 100644
---
a/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
+++
b/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
@@ -77,14 +77,25 @@ The bottom layer of the `Export` statement actually
executes the `select...outfi
The following parameters can be specified:
- `label`: This parameter is optional, specifies the label of the export
task. If this parameter is not specified, the system randomly assigns a label
to the export task.
+
- `column_separator`: Specifies the exported column separator, default is
`\t`, mulit-bytes is supported. This parameter is only used for `CSV` file
format.
+
- `line_delimiter`: Specifies the line delimiter for export, the default is
`\n`, mulit-bytes is supported. This parameter is only used for `CSV` file
format.
+
- `timeout`: The timeout period of the export job, the default is 2 hours,
the unit is seconds.
+
- `columns`: Specifies certain columns of the export job table
+
- `format`: Specifies the file format, support: parquet, orc, csv,
csv_with_names, csv_with_names_and_types.The default is csv format.
+
- `parallelism`: The concurrency degree of the `export` job, the default is
`1`. The export job will be divided into `select..outfile..` statements of the
number of `parallelism` to execute concurrently. (If the value of `parallelism`
is greater than the number of tablets in the table, the system will
automatically set `parallelism` to the number of tablets, that is, each
`select..outfile..` statement is responsible for one tablet)
+
- `delete_existing_files`: default `false`. If it is specified as true, you
will first delete all files specified in the directory specified by the
file_path, and then export the data to the directory.For example: "file_path" =
"/user/tmp", then delete all files and directory under "/user/"; "file_path" =
"/user/tmp/", then delete all files and directory under "/user/tmp/"
+
- `max_file_size`: it is the limit for the size of a single file in the
export job. If the result file exceeds this value, it will be split into
multiple files. The valid range for `max_file_size` is [5MB, 2GB], with a
default value of 1GB. (When exporting to the ORC file format, the actual size
of the split files will be multiples of 64MB, for example, if max_file_size is
specified as 5MB, the actual split size will be 64MB; if max_file_size is
specified as 65MB, the actual split size [...]
+
+ - `with_bom`: The default is false. If it is set to true, the exported file
is encoded in UTF8 with BOM (valid only for CSV-related file format).
+
- `timeout`: This is the timeout parameter of the export job, the default
timeout is 2 hours, and the unit is seconds.
> Note that to use the `delete_existing_files` parameter, you also need to
add the configuration `enable_delete_existing_files = true` to the fe.conf file
and restart the FE. Only then will the `delete_existing_files` parameter take
effect. Setting `delete_existing_files = true` is a dangerous operation and it
is recommended to only use it in a testing environment.
diff --git
a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
index e780c93084d..48eefbf7246 100644
---
a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
+++
b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/Manipulation/EXPORT.md
@@ -90,6 +90,8 @@ EXPORT
- `delete_existing_files`:
默认为false，若指定为true,则会先删除`export_path`所指定目录下的所有文件，然后导出数据到该目录下。例如："export_path" =
"/user/tmp", 则会删除"/user/"下所有文件及目录；"file_path" = "/user/tmp/",
则会删除"/user/tmp/"下所有文件及目录。
+ - `with_bom`: 默认为false，若指定为true，则导出的文件编码为带有BOM的UTF8编码（只对csv相关的文件格式生效）。
+
- `timeout`：导出作业的超时时间，默认为2小时，单位是秒。
>
注意：要使用delete_existing_files参数，还需要在fe.conf中添加配置`enable_delete_existing_files =
true`并重启fe，此时delete_existing_files才会生效。delete_existing_files = true
是一个危险的操作，建议只在测试环境中使用。
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
index 6eaa45cdfdd..3ad688f39a5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
@@ -80,6 +80,7 @@ public class ExportStmt extends StatementBase {
.add(PropertyAnalyzer.PROPERTIES_COLUMN_SEPARATOR)
.add(PropertyAnalyzer.PROPERTIES_LINE_DELIMITER)
.add("format")
+ .add(OutFileClause.PROP_WITH_BOM)
.build();
private TableName tblName;
@@ -104,6 +105,7 @@ public class ExportStmt extends StatementBase {
private String maxFileSize;
private String deleteExistingFiles;
+ private String withBom;
private SessionVariable sessionVariables;
private String qualifiedUser;
@@ -370,7 +372,10 @@ public class ExportStmt extends StatementBase {
// max_file_size
this.maxFileSize =
properties.getOrDefault(OutFileClause.PROP_MAX_FILE_SIZE, "");
+ // delete_existing_files
this.deleteExistingFiles =
properties.getOrDefault(OutFileClause.PROP_DELETE_EXISTING_FILES, "");
+ // with_bom
+ this.withBom = properties.getOrDefault(OutFileClause.PROP_WITH_BOM,
"false");
// timeout
String timeoutString =
properties.getOrDefault(LoadStmt.TIMEOUT_PROPERTY,
@@ -462,6 +467,10 @@ public class ExportStmt extends StatementBase {
return deleteExistingFiles;
}
+ public String getWithBom() {
+ return withBom;
+ }
+
public Integer getParallelNum() {
return parallelism;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
index f0121b001e6..f65d8957cc7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
@@ -140,6 +140,7 @@ public class OutFileClause {
private static final String PROP_SUCCESS_FILE_NAME = "success_file_name";
public static final String PROP_DELETE_EXISTING_FILES =
"delete_existing_files";
public static final String PROP_FILE_SUFFIX = "file_suffix";
+ public static final String PROP_WITH_BOM = "with_bom";
private static final String PARQUET_PROP_PREFIX = "parquet.";
private static final String SCHEMA = "schema";
@@ -159,6 +160,7 @@ public class OutFileClause {
private long maxFileSizeBytes = DEFAULT_MAX_FILE_SIZE_BYTES;
private boolean deleteExistingFiles = false;
private String fileSuffix = "";
+ private boolean withBom = false;
private BrokerDesc brokerDesc = null;
// True if result is written to local disk.
// If set to true, the brokerDesc must be null.
@@ -651,6 +653,11 @@ public class OutFileClause {
processedPropKeys.add(PROP_FILE_SUFFIX);
}
+ if (properties.containsKey(PROP_WITH_BOM)) {
+ withBom =
Boolean.valueOf(properties.get(PROP_WITH_BOM)).booleanValue();
+ processedPropKeys.add(PROP_WITH_BOM);
+ }
+
if (properties.containsKey(PROP_SUCCESS_FILE_NAME)) {
successFileName = properties.get(PROP_SUCCESS_FILE_NAME);
FeNameFormat.checkOutfileSuccessFileName("file name",
successFileName);
@@ -889,6 +896,7 @@ public class OutFileClause {
sinkOptions.setMaxFileSizeBytes(maxFileSizeBytes);
sinkOptions.setDeleteExistingFiles(deleteExistingFiles);
sinkOptions.setFileSuffix(fileSuffix);
+ sinkOptions.setWithBom(withBom);
if (brokerDesc != null) {
sinkOptions.setBrokerProperties(brokerDesc.getProperties());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
index 03f33bea30f..938a9d7711a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
@@ -151,6 +151,8 @@ public class ExportJob implements Writable {
private String outfileInfo;
@SerializedName("tabletsNum")
private Integer tabletsNum;
+ @SerializedName("withBom")
+ private String withBom;
// progress has two functions at EXPORTING stage:
// 1. when progress < 100, it indicates exporting
// 2. set progress = 100 ONLY when exporting progress is completely done
@@ -206,6 +208,7 @@ public class ExportJob implements Writable {
this.columnSeparator = "\t";
this.lineDelimiter = "\n";
this.columns = "";
+ this.withBom = "false";
}
public ExportJob(long jobId) {
@@ -237,6 +240,7 @@ public class ExportJob implements Writable {
this.maxFileSize = stmt.getMaxFileSize();
this.deleteExistingFiles = stmt.getDeleteExistingFiles();
this.partitionNames = stmt.getPartitions();
+ this.withBom = stmt.getWithBom();
this.exportTable =
db.getTableOrDdlException(stmt.getTblName().getTbl());
this.columns = stmt.getColumns();
@@ -390,6 +394,7 @@ public class ExportJob implements Writable {
if (!deleteExistingFiles.isEmpty()) {
outfileProperties.put(OutFileClause.PROP_DELETE_EXISTING_FILES,
deleteExistingFiles);
}
+ outfileProperties.put(OutFileClause.PROP_WITH_BOM, withBom);
// broker properties
// outfile clause's broker properties need 'broker.' prefix
@@ -576,6 +581,10 @@ public class ExportJob implements Writable {
return tableName;
}
+ public String getWithBom() {
+ return withBom;
+ }
+
public SessionVariable getSessionVariables() {
return sessionVariables;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
index 06960c68ff2..80ffc9b51ec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
@@ -420,6 +420,7 @@ public class ExportMgr extends MasterDaemon {
infoMap.put("broker", job.getBrokerDesc().getName());
infoMap.put("column_separator", job.getColumnSeparator());
infoMap.put("format", job.getFormat());
+ infoMap.put("with_bom", job.getWithBom());
infoMap.put("line_delimiter", job.getLineDelimiter());
infoMap.put("columns", job.getColumns());
infoMap.put("tablet_num", job.getTabletsNum());
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index b8826eaa1d2..1943295963c 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -125,6 +125,7 @@ struct TResultFileSinkOptions {
16: optional bool delete_existing_files;
17: optional string file_suffix;
+ 18: optional bool with_bom;
}
struct TMemoryScratchSink {
diff --git a/regression-test/data/export_p0/test_with_bom.out
b/regression-test/data/export_p0/test_with_bom.out
new file mode 100644
index 00000000000..de995bdce0a
--- /dev/null
+++ b/regression-test/data/export_p0/test_with_bom.out
@@ -0,0 +1,66 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !select_export1 --
+1 2017-10-01
+10 2017-10-01
+11 2017-10-01
+2 2017-10-01
+3 2017-10-01
+4 2017-10-01
+5 2017-10-01
+6 2017-10-01
+7 2017-10-01
+8 2017-10-01
+9 2017-10-01
+
+-- !select_load1 --
+1 2017-10-01
+10 2017-10-01
+11 2017-10-01
+2 2017-10-01
+3 2017-10-01
+4 2017-10-01
+5 2017-10-01
+6 2017-10-01
+7 2017-10-01
+8 2017-10-01
+9 2017-10-01
+
+-- !select_load1 --
+1 2017-10-01
+10 2017-10-01
+11 2017-10-01
+2 2017-10-01
+3 2017-10-01
+4 2017-10-01
+5 2017-10-01
+6 2017-10-01
+7 2017-10-01
+8 2017-10-01
+9 2017-10-01
+
+-- !select_load1 --
+1 2017-10-01
+10 2017-10-01
+11 2017-10-01
+2 2017-10-01
+3 2017-10-01
+4 2017-10-01
+5 2017-10-01
+6 2017-10-01
+7 2017-10-01
+8 2017-10-01
+9 2017-10-01
+
+-- !select_load1 --
+1 2017-10-01
+10 2017-10-01
+11 2017-10-01
+2 2017-10-01
+3 2017-10-01
+4 2017-10-01
+5 2017-10-01
+6 2017-10-01
+7 2017-10-01
+8 2017-10-01
+9 2017-10-01
+
diff --git a/regression-test/suites/export_p0/test_with_bom.groovy
b/regression-test/suites/export_p0/test_with_bom.groovy
new file mode 100644
index 00000000000..3ab404d3cf8
--- /dev/null
+++ b/regression-test/suites/export_p0/test_with_bom.groovy
@@ -0,0 +1,241 @@
+// 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.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.nio.file.Paths
+
+suite("test_with_bom", "p0") {
+ // open nereids
+ sql """ set enable_nereids_planner=true """
+ sql """ set enable_fallback_to_original_planner=false """
+
+
+ String ak = getS3AK()
+ String sk = getS3SK()
+ String s3_endpoint = getS3Endpoint()
+ String region = getS3Region()
+ String bucket = context.config.otherConfigs.get("s3BucketName");
+
+
+ def this_db = "regression_test_export_p0"
+ def table_export_name = "test_with_bom"
+ def table_load_name = "test_with_bom_load_back"
+ def outfile_path_prefix = """${bucket}/export/test_with_bom/exp_"""
+ def file_format = "csv"
+
+ // test csv with bom
+ sql """ DROP TABLE IF EXISTS ${table_export_name} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${table_export_name} (
+ `user_id` INT NOT NULL COMMENT "用户id",
+ `date` DATE NOT NULL COMMENT "数据灌入日期时间"
+ )
+ DISTRIBUTED BY HASH(user_id)
+ PROPERTIES("replication_num" = "1");
+ """
+ StringBuilder sb = new StringBuilder()
+ int i = 1
+ for (; i < 11; i ++) {
+ sb.append("""
+ (${i}, '2017-10-01'),
+ """)
+ }
+ sb.append("""
+ (${i}, '2017-10-01')
+ """)
+ sql """ INSERT INTO ${table_export_name} VALUES
+ ${sb.toString()}
+ """
+ def insert_res = sql "show last insert;"
+ logger.info("insert result: " + insert_res.toString())
+ order_qt_select_export1 """ SELECT * FROM ${table_export_name} t ORDER BY
user_id; """
+
+
+ def waiting_export = { export_label ->
+ while (true) {
+ def res = sql """ show export where label = "${export_label}";"""
+ logger.info("export state: " + res[0][2])
+ if (res[0][2] == "FINISHED") {
+ def json = parseJson(res[0][11])
+ assert json instanceof List
+ assertEquals("1", json.fileNumber[0][0])
+ log.info("outfile_path: ${json.url[0]}")
+ return json.url[0];
+ } else if (res[0][2] == "CANCELLED") {
+ throw new IllegalStateException("""export failed:
${res[0][10]}""")
+ } else {
+ sleep(5000)
+ }
+ }
+ }
+
+ def check_bytes = { except_bytes, export_label ->
+ def show_export_res = sql_return_maparray """ show export where label
= "${export_label}"; """
+ def export_job = show_export_res[0]
+ def outfile_json = parseJson(export_job.OutfileInfo)
+ assertEquals(except_bytes, outfile_json[0].fileSize)
+ }
+
+ // 1. exec export without bom
+ def uuid = UUID.randomUUID().toString()
+ def outFilePath = """${outfile_path_prefix}_${uuid}"""
+ def label = "label_${uuid}"
+ try {
+ // exec export
+ sql """
+ EXPORT TABLE ${table_export_name} TO "s3://${outFilePath}/"
+ PROPERTIES(
+ "label" = "${label}",
+ "format" = "${file_format}"
+ )
+ WITH S3(
+ "s3.endpoint" = "${s3_endpoint}",
+ "s3.region" = "${region}",
+ "s3.secret_key"="${sk}",
+ "s3.access_key" = "${ak}"
+ );
+ """
+ def outfile_url = waiting_export.call(label)
+
+ order_qt_select_load1 """ select * from s3(
+ "uri" =
"http://${s3_endpoint}${outfile_url.substring(4)}0.${file_format}",
+ "s3.access_key"= "${ak}",
+ "s3.secret_key" = "${sk}",
+ "format" = "csv",
+ "region" = "${region}"
+ ) ORDER BY c1;
+ """
+
+ // check outfile bytes
+ check_bytes("145bytes", label)
+ } finally {
+ }
+
+
+ // 2. exec export with bom
+ uuid = UUID.randomUUID().toString()
+ outFilePath = """${outfile_path_prefix}_${uuid}"""
+ label = "label_${uuid}"
+ try {
+ // exec export
+ sql """
+ EXPORT TABLE ${table_export_name} TO "s3://${outFilePath}/"
+ PROPERTIES(
+ "label" = "${label}",
+ "format" = "${file_format}",
+ "with_bom" = "true"
+ )
+ WITH S3(
+ "s3.endpoint" = "${s3_endpoint}",
+ "s3.region" = "${region}",
+ "s3.secret_key"="${sk}",
+ "s3.access_key" = "${ak}"
+ );
+ """
+ def outfile_url = waiting_export.call(label)
+
+ order_qt_select_load1 """ select * from s3(
+ "uri" =
"http://${s3_endpoint}${outfile_url.substring(4)}0.${file_format}",
+ "s3.access_key"= "${ak}",
+ "s3.secret_key" = "${sk}",
+ "format" = "csv",
+ "region" = "${region}"
+ ) ORDER BY c1;
+ """
+
+ // check outfile bytes
+ check_bytes("148bytes", label)
+ } finally {
+ }
+
+
+ // 3. test csv_with_names with bom
+ uuid = UUID.randomUUID().toString()
+ outFilePath = """${outfile_path_prefix}_${uuid}"""
+ label = "label_${uuid}"
+ try {
+ // exec export
+ sql """
+ EXPORT TABLE ${table_export_name} TO "s3://${outFilePath}/"
+ PROPERTIES(
+ "label" = "${label}",
+ "format" = "csv_with_names",
+ "with_bom" = "true"
+ )
+ WITH S3(
+ "s3.endpoint" = "${s3_endpoint}",
+ "s3.region" = "${region}",
+ "s3.secret_key"="${sk}",
+ "s3.access_key" = "${ak}"
+ );
+ """
+ def outfile_url = waiting_export.call(label)
+
+ order_qt_select_load1 """ select * from s3(
+ "uri" =
"http://${s3_endpoint}${outfile_url.substring(4)}0.${file_format}",
+ "s3.access_key"= "${ak}",
+ "s3.secret_key" = "${sk}",
+ "format" = "csv_with_names",
+ "region" = "${region}"
+ ) ORDER BY user_id;
+ """
+
+ // check outfile bytes
+ check_bytes("161bytes", label)
+ } finally {
+ }
+
+
+ // 4. test csv_with_names_and_types with bom
+ uuid = UUID.randomUUID().toString()
+ outFilePath = """${outfile_path_prefix}_${uuid}"""
+ label = "label_${uuid}"
+ try {
+ // exec export
+ sql """
+ EXPORT TABLE ${table_export_name} TO "s3://${outFilePath}/"
+ PROPERTIES(
+ "label" = "${label}",
+ "format" = "csv_with_names_and_types",
+ "with_bom" = "true"
+ )
+ WITH S3(
+ "s3.endpoint" = "${s3_endpoint}",
+ "s3.region" = "${region}",
+ "s3.secret_key"="${sk}",
+ "s3.access_key" = "${ak}"
+ );
+ """
+ def outfile_url = waiting_export.call(label)
+
+ order_qt_select_load1 """ select * from s3(
+ "uri" =
"http://${s3_endpoint}${outfile_url.substring(4)}0.${file_format}",
+ "s3.access_key"= "${ak}",
+ "s3.secret_key" = "${sk}",
+ "format" = "csv_with_names_and_types",
+ "region" = "${region}"
+ ) ORDER BY user_id;
+ """
+
+ // check outfile bytes
+ check_bytes("172bytes", label)
+ } finally {
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
