[fix](be) Align CSV V2 enclosed field parsing

be/src/format/file_reader/new_plain_text_line_reader.cpp

@@ -46,11 +46,31 @@
 
 namespace doris {
 const uint8_t* EncloseCsvLineReaderCtx::read_line_impl(const uint8_t* start, const size_t length) {
+ if (_skip_utf8_bom && !_first_record_prefix_checked && _idx == 0) {
+ constexpr uint8_t UTF8_BOM[] = {0xEF, 0xBB, 0xBF};
+ constexpr size_t UTF8_BOM_SIZE = sizeof(UTF8_BOM);
+ const size_t prefix_size = std::min(length, UTF8_BOM_SIZE);
+ if (std::memcmp(start, UTF8_BOM, prefix_size) != 0) {
+ _first_record_prefix_checked = true;
+ } else if (length < UTF8_BOM_SIZE) {
+ // The input buffer can end inside the BOM. Wait for the remaining prefix bytes instead
+ // of feeding a partial BOM into START, which would permanently select NORMAL state.
+ return nullptr;
+ } else {
+ // Keep offsets relative to the original buffer. CsvReader removes these three bytes
+ // from the returned line and shifts separator positions by the same amount.
+ _idx = UTF8_BOM_SIZE;
+ _first_record_prefix_checked = true;
+ }
+ }
  // Avoid part bytes of the multi-char column separator have already been parsed,
  // causing parse column separator error.
  if (_state.curr_state == ReaderState::NORMAL ||
  _state.curr_state == ReaderState::MATCH_ENCLOSE) {
- _idx -= std::min(_column_sep_len - 1, _idx);
+ const size_t last_column_sep_end =
+ _column_sep_positions.empty() ? 0 : _column_sep_positions.back() + _column_sep_len;
+ DORIS_CHECK_LE(last_column_sep_end, _idx);
+ _idx -= std::min(_column_sep_len - 1, _idx - last_column_sep_end);
  }
  _total_len = length;
  size_t bound = update_reading_bound(start);
@@ -134,7 +154,8 @@ void EncloseCsvLineReaderCtx::_on_start(const uint8_t* start, size_t& len) {
 
 void EncloseCsvLineReaderCtx::_on_normal(const uint8_t* start, size_t& len) {
  const uint8_t* curr_start = start + _idx;
- size_t curr_len = len - _idx;
+ const size_t field_search_bound = _result == nullptr ? len : _result - start;
+ const size_t curr_len = field_search_bound - _idx;
  const uint8_t* col_sep_pos =
  find_col_sep_func(curr_start, curr_len, _column_sep.c_str(), _column_sep_len);
 
@@ -190,7 +211,8 @@ void EncloseCsvLineReaderCtx::_on_pre_match_enclose(const uint8_t* start, size_t
 
 void EncloseCsvLineReaderCtx::_on_match_enclose(const uint8_t* start, size_t& len) {
  const uint8_t* curr_start = start + _idx;
- size_t curr_len = len - _idx;
+ const size_t field_search_bound = _result == nullptr ? len : _result - start;
+ const size_t curr_len = field_search_bound - _idx;
  const uint8_t* delim_pos =
  find_col_sep_func(curr_start, curr_len, _column_sep.c_str(), _column_sep_len);
 

be/src/format/file_reader/new_plain_text_line_reader.h

@@ -156,10 +156,12 @@ class EncloseCsvLineReaderCtx final : public BaseTextLineReaderContext<EncloseCs
  explicit EncloseCsvLineReaderCtx(const std::string& line_delimiter_,
  const size_t line_delimiter_len_, std::string column_sep_,
  const size_t column_sep_len_, size_t col_sep_num,
- const char enclose, const char escape, const bool keep_cr_)
+ const char enclose, const char escape, const bool keep_cr_,
+ const bool skip_utf8_bom_ = false)
  : BaseTextLineReaderContext(line_delimiter_, line_delimiter_len_, keep_cr_),
  _enclose(enclose),
  _escape(escape),
+ _skip_utf8_bom(skip_utf8_bom_),
  _column_sep_len(column_sep_len_),
  _column_sep(std::move(column_sep_)) {
  if (column_sep_len_ == 1) {
@@ -210,6 +212,11 @@ class EncloseCsvLineReaderCtx final : public BaseTextLineReaderContext<EncloseCs
  ReaderStateWrapper _state;
  const char _enclose;
  const char _escape;
+ // Only V2 readers starting at file offset zero enable this. The line reader must recognize the
+ // BOM before deciding whether the first field is enclosed; the format reader later removes the
+ // same bytes from the returned Slice and adjusts the recorded separator positions.
+ const bool _skip_utf8_bom;
+ bool _first_record_prefix_checked = false;
  const uint8_t* _result = nullptr;
 
  size_t _total_len;

be/src/format/hive_text_util.h

@@ -0,0 +1,33 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+
+namespace doris {
+
+inline bool is_hive_text_separator_escaped(const char* data, size_t separator_pos,
+ char escape_char) {
+ size_t escape_count = 0;
+ while (separator_pos > escape_count && data[separator_pos - escape_count - 1] == escape_char) {
+ ++escape_count;
+ }
+ return escape_count % 2 == 1;
+}
+
+} // namespace doris

be/src/format/text/text_reader.cpp

@@ -34,6 +34,7 @@
 #include "exec/scan/scanner.h"
 #include "format/csv/csv_reader.h"
 #include "format/file_reader/new_plain_text_line_reader.h"
+#include "format/hive_text_util.h"
 #include "format/line_reader.h"
 #include "io/file_factory.h"
 #include "io/fs/buffered_reader.h"
@@ -60,7 +61,7 @@ void HiveTextFieldSplitter::_split_field_single_char(const Slice& line,
  for (size_t i = 0; i < size; ++i) {
  if (data[i] == _value_sep[0]) {
  // hive will escape the field separator in string
- if (_escape_char != 0 && i > 0 && data[i - 1] == _escape_char) {
+ if (_escape_char != 0 && is_hive_text_separator_escaped(data, i, _escape_char)) {
  continue;
  }
  process_value_func(data, value_start, i - value_start, _trimming_char, splitted_values);
@@ -98,7 +99,7 @@ void HiveTextFieldSplitter::_split_field_multi_char(const Slice& line,
  }
  if (j == (int)_value_sep_len - 1) {
  size_t curpos = i - _value_sep_len + 1;
- if (_escape_char != 0 && curpos > 0 && data[curpos - 1] == _escape_char) {
+ if (_escape_char != 0 && is_hive_text_separator_escaped(data, curpos, _escape_char)) {
  j = next[j];
  continue;
  }

be/src/format_v2/delimited_text/csv_reader.cpp

@@ -134,14 +134,13 @@ Status CsvReader::_create_line_reader() {
  text_line_reader_ctx = std::make_shared<PlainTextLineReaderCtx>(
  _line_delimiter, _line_delimiter.size(), _keep_cr);
  } else {
- // The enclosed-line context finds logical records that may span physical newlines.
- // Field slicing still happens in `_split_line()` because the v2 scan request may ask
- // for CSV ordinals in a different order from the physical file.
  const size_t col_sep_num =
  _source_file_slot_descs.size() > 1 ? _source_file_slot_descs.size() - 1 : 0;
-text_line_reader_ctx = std::make_shared<EncloseCsvLineReaderCtx>(
+_enclose_reader_ctx = std::make_shared<EncloseCsvLineReaderCtx>(
  _line_delimiter, _line_delimiter.size(), _value_separator,
- _value_separator.size(), col_sep_num, _enclose, _escape, _keep_cr);
+ _value_separator.size(), col_sep_num, _enclose, _escape, _keep_cr,
+ _start_offset == 0);
+ text_line_reader_ctx = _enclose_reader_ctx;
  }
  _line_reader = NewPlainTextLineReader::create_unique(
  _profile, _file_reader, _decompressor.get(), std::move(text_line_reader_ctx), _size,
@@ -174,77 +173,40 @@ void CsvReader::_split_line(const Slice& line) {
  return;
  }
 
- // The text line reader is responsible for split boundaries and multi-line quoted fields.
- // Field slicing still happens here because FileScannerV2 asks columns by file-local id, so we
- // must be able to materialize only the requested CSV ordinals without building a row object.
- // Example: for `1,"a,b",10` and column separator `,`, this loop returns three slices:
- // `1`, `a,b`, and `10`; the comma inside quotes does not create an extra field.
- bool in_quote = false;
- bool escaped = false;
- size_t start = 0;
- size_t i = 0;
- while (i < line.size) {
- const char ch = line.data[i];
- if (_enclose != 0) {
- if (escaped) {
- escaped = false;
- ++i;
- continue;
- }
- if (_escape != 0 && ch == _escape) {
- escaped = true;
- ++i;
- continue;
- }
- if (ch == _enclose) {
- if (in_quote && i + 1 < line.size && line.data[i + 1] == _enclose) {
- i += 2;
- continue;
- }
- in_quote = !in_quote;
- ++i;
- continue;
- }
+ const auto append_value = [&](size_t value_start, size_t value_len) {
+ while (_trim_tailing_spaces && value_len > 0 &&
+ line.data[value_start + value_len - 1] == ' ') {
+ --value_len;
  }
- if (!in_quote && starts_with_at(line, i, _value_separator)) {
- size_t value_start = start;
- size_t value_len = i - start;
- while (_trim_tailing_spaces && value_len > 0 &&
- line.data[value_start + value_len - 1] == ' ') {
- --value_len;
- }
- if (_trim_double_quotes && value_len > 1 && line.data[value_start] == '"' &&
- line.data[value_start + value_len - 1] == '"') {
- ++value_start;
- value_len -= 2;
- } else if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
- line.data[value_start + value_len - 1] == _enclose) {
- ++value_start;
- value_len -= 2;
- }
- _split_values.emplace_back(line.data + value_start, value_len);
- i += _value_separator.size();
- start = i;
- continue;
+ if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
+ line.data[value_start + value_len - 1] == _enclose) {
+ ++value_start;
+ value_len -= 2;
  }
-++i;
- }
+_split_values.emplace_back(line.data + value_start, value_len);
+ };
 
- size_t value_start = start;
- size_t value_len = line.size - start;
- while (_trim_tailing_spaces && value_len > 0 && line.data[value_start + value_len - 1] == ' ') {
- --value_len;
- }
- if (_trim_double_quotes && value_len > 1 && line.data[value_start] == '"' &&
- line.data[value_start + value_len - 1] == '"') {
- ++value_start;
- value_len -= 2;
- } else if (_enclose != 0 && value_len > 1 && line.data[value_start] == _enclose &&
- line.data[value_start + value_len - 1] == _enclose) {
- ++value_start;
- value_len -= 2;
+ size_t value_start = 0;
+ if (_enclose_reader_ctx != nullptr) {
+ for (const size_t separator_position : _enclose_reader_ctx->column_sep_positions()) {
+ DORIS_CHECK_LE(value_start, separator_position);
+ DORIS_CHECK_LE(separator_position, line.size);
+ append_value(value_start, separator_position - value_start);
+ value_start = separator_position + _value_separator.size();
+ }
+ } else {
+ for (size_t i = 0; i < line.size;) {
+ if (starts_with_at(line, i, _value_separator)) {
+ append_value(value_start, i - value_start);
+ i += _value_separator.size();
+ value_start = i;
+ } else {
+ ++i;
+ }
+ }
  }
- _split_values.emplace_back(line.data + value_start, value_len);
+ DORIS_CHECK_LE(value_start, line.size);
+ append_value(value_start, line.size - value_start);
 }
 
 Status CsvReader::_deserialize_one_cell(const RequestedColumn& column, IColumn* output,
@@ -261,7 +223,8 @@ Status CsvReader::_deserialize_one_cell(const RequestedColumn& column, IColumn*
  // CSV keeps empty-field handling separate from null_format matching. An empty
  // null_format must not turn every empty CSV field into NULL unless FE explicitly sets
  // empty_field_as_null; OpenCSV-compatible tables expect empty fields to stay empty strings.
- if (_options.null_len > 0 && value.size == _options.null_len &&
+ const bool quoted = _options.converted_from_string && value.trim_double_quotes();
+ if (!quoted && _options.null_len > 0 && value.size == _options.null_len &&
  std::memcmp(value.data, _options.null_format, value.size) == 0) {
  null_column.insert_data(nullptr, 0);
  return Status::OK();
@@ -292,4 +255,10 @@ bool CsvReader::_can_split() const {
  _file_format_type == TFileFormatType::FORMAT_CSV_PLAIN);
 }
 
+void CsvReader::_on_bom_removed(size_t bom_size) {
+ if (_enclose_reader_ctx != nullptr) {
+ _enclose_reader_ctx->adjust_column_sep_positions(bom_size);
+ }
+}
+
 } // namespace doris::format::csv

be/src/format_v2/delimited_text/csv_reader.h

@@ -25,6 +25,7 @@
 #include "util/slice.h"
 
 namespace doris {
+class EncloseCsvLineReaderCtx;
 class SlotDescriptor;
 } // namespace doris
 
@@ -61,13 +62,15 @@ class CsvReader final : public ::doris::format::DelimitedTextReader {
  Slice value) override;
  Slice _normalize_value(Slice value) const override;
  bool _can_split() const override;
+ void _on_bom_removed(size_t bom_size) override;
 
  TFileFormatType::type _file_format_type = TFileFormatType::FORMAT_CSV_PLAIN;
  char _enclose = 0;
  bool _trim_double_quotes = false;
  bool _trim_tailing_spaces = false;
  bool _empty_field_as_null = false;
  bool _keep_cr = false;
+ std::shared_ptr<EncloseCsvLineReaderCtx> _enclose_reader_ctx;
 };
 
 } // namespace doris::format::csv

be/src/format_v2/delimited_text/delimited_text_reader.cpp

@@ -624,6 +624,10 @@ bool DelimitedTextReader::_can_split() const {
  return _file_compress_type == TFileCompressType::PLAIN;
 }
 
+void DelimitedTextReader::_on_bom_removed(size_t bom_size) {
+ (void)bom_size;
+}
+
 Status DelimitedTextReader::_append_null(IColumn* output) {
  DORIS_CHECK(output != nullptr);
  auto* nullable = assert_cast<ColumnNullable*>(output);
@@ -635,8 +639,10 @@ const uint8_t* DelimitedTextReader::_remove_bom(const uint8_t* ptr, size_t* size
  DORIS_CHECK(size != nullptr);
  if (ptr != nullptr && *size >= 3 && static_cast<uint8_t>(ptr[0]) == 0xEF &&
  static_cast<uint8_t>(ptr[1]) == 0xBB && static_cast<uint8_t>(ptr[2]) == 0xBF) {
- *size -= 3;
- return ptr + 3;
+ constexpr size_t BOM_SIZE = 3;
+ *size -= BOM_SIZE;
+ _on_bom_removed(BOM_SIZE);
+ return ptr + BOM_SIZE;
  }
  return ptr;
 }

be/src/format_v2/delimited_text/delimited_text_reader.h

@@ -123,6 +123,8 @@ class DelimitedTextReader : public FileReader {
  // Whether this file can start at a non-zero split offset. Compressed delimited files cannot be
  // split because the decompressor needs the stream from the beginning.
  virtual bool _can_split() const;
+ // Let formats adjust parser metadata that was computed before the common reader removed a BOM.
+ virtual void _on_bom_removed(size_t bom_size);
 
  Status _append_null(IColumn* output);
  // Match the generic nullable serde semantics exactly: a field is NULL when its raw slice is

be/src/format_v2/delimited_text/text_reader.cpp

@@ -25,6 +25,7 @@
 #include "core/data_type/data_type_string.h"
 #include "core/data_type_serde/data_type_string_serde.h"
 #include "format/file_reader/new_plain_text_line_reader.h"
+#include "format/hive_text_util.h"
 #include "runtime/descriptors.h"
 #include "util/decompressor.h"
 
@@ -104,7 +105,7 @@ void TextReader::_split_line_single_char(const Slice& line) {
  if (line.data[i] == _value_separator[0]) {
  // Hive text lets a string escape the field separator. The backslash remains in the
  // field slice so deserialize_one_cell_from_hive_text() can unescape the final value.
- if (_escape != 0 && i > 0 && line.data[i - 1] == _escape) {
+ if (_escape != 0 && is_hive_text_separator_escaped(line.data, i, _escape)) {
  continue;
  }
  _split_values.emplace_back(line.data + value_start, i - value_start);
@@ -119,7 +120,7 @@ void TextReader::_split_line_multi_char(const Slice& line) {
  size_t i = 0;
  while (i < line.size) {
  if (starts_with_at(line, i, _value_separator)) {
- if (_escape != 0 && i > 0 && line.data[i - 1] == _escape) {
+ if (_escape != 0 && is_hive_text_separator_escaped(line.data, i, _escape)) {
  ++i;
  continue;
  }

be/test/format/text/hive_text_field_splitter_test.cpp

@@ -81,7 +81,9 @@ TEST_F(HiveTextFieldSplitterTest, OverlappingPatterns) {
 // Test escape character functionality
 TEST_F(HiveTextFieldSplitterTest, EscapeCharacter) {
  verify_field_split("a\\,b,c", ",", {"a\\,b", "c"}, '\\');
+ verify_field_split(R"(a\\,b)", ",", {R"(a\\)", "b"}, '\\');
  verify_field_split("a\\||b||c", "||", {"a\\||b", "c"}, '\\');
+ verify_field_split(R"(a\\||b)", "||", {R"(a\\)", "b"}, '\\');
  verify_field_split("field1\\|+|field2|+|field3", "|+|", {"field1\\|+|field2", "field3"}, '\\');
 }
 
@@ -94,4 +96,4 @@ TEST_F(HiveTextFieldSplitterTest, RealWorldScenarios) {
  verify_field_split("a|+||+|c", "|+|", {"a", "", "c"});
 }
 
-} // namespace doris
+} // namespace doris