diff --git a/be/src/format/file_reader/new_plain_text_line_reader.cpp b/be/src/format/file_reader/new_plain_text_line_reader.cpp index c0396a39e84c78..25a7a0a7ac2476 100644 --- a/be/src/format/file_reader/new_plain_text_line_reader.cpp +++ b/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); diff --git a/be/src/format/file_reader/new_plain_text_line_reader.h b/be/src/format/file_reader/new_plain_text_line_reader.h index 7262b203e02556..ed7f80493b02d4 100644 --- a/be/src/format/file_reader/new_plain_text_line_reader.h +++ b/be/src/format/file_reader/new_plain_text_line_reader.h @@ -156,10 +156,12 @@ class EncloseCsvLineReaderCtx final : public BaseTextLineReaderContext + +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 diff --git a/be/src/format/text/text_reader.cpp b/be/src/format/text/text_reader.cpp index 23501f94cd6122..693fde2d9a871b 100644 --- a/be/src/format/text/text_reader.cpp +++ b/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; } diff --git a/be/src/format_v2/delimited_text/csv_reader.cpp b/be/src/format_v2/delimited_text/csv_reader.cpp index 711146a9880479..bb6c55d30847b2 100644 --- a/be/src/format_v2/delimited_text/csv_reader.cpp +++ b/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( _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( + _enclose_reader_ctx = std::make_shared( _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 diff --git a/be/src/format_v2/delimited_text/csv_reader.h b/be/src/format_v2/delimited_text/csv_reader.h index e5d1ce25a74f40..dd1d77d5a34b9a 100644 --- a/be/src/format_v2/delimited_text/csv_reader.h +++ b/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,6 +62,7 @@ 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; @@ -68,6 +70,7 @@ class CsvReader final : public ::doris::format::DelimitedTextReader { bool _trim_tailing_spaces = false; bool _empty_field_as_null = false; bool _keep_cr = false; + std::shared_ptr _enclose_reader_ctx; }; } // namespace doris::format::csv diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.cpp b/be/src/format_v2/delimited_text/delimited_text_reader.cpp index b26e0d4445615f..f19d12c75714b9 100644 --- a/be/src/format_v2/delimited_text/delimited_text_reader.cpp +++ b/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(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(ptr[0]) == 0xEF && static_cast(ptr[1]) == 0xBB && static_cast(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; } diff --git a/be/src/format_v2/delimited_text/delimited_text_reader.h b/be/src/format_v2/delimited_text/delimited_text_reader.h index 06cb93dd7f7b65..daea3bc90942d8 100644 --- a/be/src/format_v2/delimited_text/delimited_text_reader.h +++ b/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 diff --git a/be/src/format_v2/delimited_text/text_reader.cpp b/be/src/format_v2/delimited_text/text_reader.cpp index 930052a14f1229..6dea7a07a4c6fa 100644 --- a/be/src/format_v2/delimited_text/text_reader.cpp +++ b/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; } diff --git a/be/test/format/text/hive_text_field_splitter_test.cpp b/be/test/format/text/hive_text_field_splitter_test.cpp index 6887ff6d2db4a1..d52078aead0dff 100644 --- a/be/test/format/text/hive_text_field_splitter_test.cpp +++ b/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 \ No newline at end of file +} // namespace doris diff --git a/be/test/format_v2/delimited_text/csv_reader_test.cpp b/be/test/format_v2/delimited_text/csv_reader_test.cpp index d5eafec7142329..0ce99630c7af8b 100644 --- a/be/test/format_v2/delimited_text/csv_reader_test.cpp +++ b/be/test/format_v2/delimited_text/csv_reader_test.cpp @@ -24,6 +24,7 @@ #include #include +#include "common/config.h" #include "common/consts.h" #include "common/object_pool.h" #include "core/assert_cast.h" @@ -45,6 +46,8 @@ #include "runtime/runtime_profile.h" #include "testutil/desc_tbl_builder.h" #include "testutil/mock/mock_runtime_state.h" +#include "util/debug_points.h" +#include "util/defer_op.h" namespace doris::format::csv { namespace { @@ -780,6 +783,187 @@ TEST_F(CsvV2ReaderTest, EnclosedFieldKeepsSeparatorInsideStringValue) { EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "alice,team"); } +// Bare quotes and escapes outside enclosed fields stay in NORMAL state in +// EncloseCsvLineReaderCtx. Field slicing must reuse those exact separator positions. +TEST_F(CsvV2ReaderTest, EncloseFieldSplittingMatchesLineReaderStateMachine) { + const auto quoted_path = (_test_dir / "enclose_state.csv").string(); + std::ofstream output(quoted_path, std::ios::binary); + output << "id,name,score\n"; + output << "1,ab\"cd,20\n"; + output << "2,C:\\dir\\,30,40\n"; + output.close(); + + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); + auto reader = create_reader(quoted_path, &_params, _slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(1)), + LocalColumnIndex::top_level(LocalColumnId(2))}; + request->local_positions.emplace(LocalColumnId(1), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(2), LocalIndex(1)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {1, 2}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 2); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "ab\"cd"); + EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 0), 20); + EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 1), 30); +} + +// OpenCSV permits escape and enclose to use the same character. The shared line-reader state +// machine handles doubled quotes without treating every quote as a generic escape. +TEST_F(CsvV2ReaderTest, MatchingEscapeAndEncloseStillSplitQuotedFields) { + const auto quoted_path = (_test_dir / "matching_escape_enclose.csv").string(); + std::ofstream output(quoted_path, std::ios::binary); + output << "id,name,score\n"; + output << "\"1\",\"alice\",\"10\"\n"; + output.close(); + + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('"'); + auto reader = create_reader(quoted_path, &_params, _slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0)), + LocalColumnIndex::top_level(LocalColumnId(1)), + LocalColumnIndex::top_level(LocalColumnId(2))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(1), LocalIndex(1)); + request->local_positions.emplace(LocalColumnId(2), LocalIndex(2)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0, 1, 2}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0), 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(1).column, 0), "alice"); + EXPECT_EQ(nullable_int_at(*block.get_by_position(2).column, 0), 10); +} + +// OpenCSV custom quote characters only remove that configured enclosure. Literal double quotes in +// a field must remain when the configured enclosure is a single quote. +TEST_F(CsvV2ReaderTest, CustomEnclosePreservesLiteralDoubleQuotes) { + const auto quoted_path = (_test_dir / "custom_enclose.csv").string(); + std::ofstream output(quoted_path, std::ios::binary); + output << R"("Project Manager")" << '\n'; + output.close(); + + auto value_slot = make_test_slot(&_pool, 10, 0, + make_nullable(std::make_shared()), "value"); + std::vector slots {value_slot}; + _params.__set_column_idxs({0}); + _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_column_separator("\t"); + _params.file_attributes.text_params.__set_enclose('\''); + _params.file_attributes.text_params.__set_escape('|'); + auto reader = create_reader(quoted_path, &_params, slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), R"("Project Manager")"); +} + +// A column separator that overlaps the line delimiter must not be recorded because CsvReader +// splits a Slice that excludes the line delimiter bytes. +TEST_F(CsvV2ReaderTest, ColumnSeparatorOverlappingLineDelimiterStaysOutsideReturnedLine) { + const auto overlap_path = (_test_dir / "overlapping_delimiters.csv").string(); + std::ofstream output(overlap_path, std::ios::binary); + output << "value|\n"; + output.close(); + + auto value_slot = make_test_slot(&_pool, 10, 0, + make_nullable(std::make_shared()), "value"); + std::vector slots {value_slot}; + _params.__set_column_idxs({0}); + _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_column_separator("|\n"); + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); + auto reader = create_reader(overlap_path, &_params, slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "value|"); +} + +// When a logical row is refilled, partial-separator backtracking must not cross the end of a +// separator that was already accepted from the previous buffer contents. +TEST_F(CsvV2ReaderTest, MultiCharacterSeparatorAcrossOutputBufferRefill) { + const auto refill_path = (_test_dir / "separator_refill.csv").string(); + std::ofstream output(refill_path, std::ios::binary); + output << std::string(28, 'a') << "|||||b\n"; + output.close(); + + auto first_slot = make_test_slot(&_pool, 10, 0, + make_nullable(std::make_shared()), "first"); + auto second_slot = make_test_slot(&_pool, 11, 1, + make_nullable(std::make_shared()), "second"); + std::vector slots {first_slot, second_slot}; + _params.__set_column_idxs({0, 1}); + _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_column_separator("|||"); + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); + + const bool enable_debug_points = config::enable_debug_points; + config::enable_debug_points = true; + DebugPoints::instance()->add_with_params("NewPlainTextLineReader.shrink_output_buf", + {{"output_buf_size", "32"}}); + Defer restore_debug_point([enable_debug_points]() { + DebugPoints::instance()->remove("NewPlainTextLineReader.shrink_output_buf"); + config::enable_debug_points = enable_debug_points; + }); + + auto reader = create_reader(refill_path, &_params, slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0)), + LocalColumnIndex::top_level(LocalColumnId(1))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(1), LocalIndex(1)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0, 1}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), std::string(28, 'a')); + EXPECT_EQ(nullable_string_at(*block.get_by_position(1).column, 0), "||b"); +} + // Scenario: when the CSV row has fewer fields than the FE-provided file slot list, v2 fills the // missing requested field with NULL instead of failing or shifting later columns. TEST_F(CsvV2ReaderTest, MissingRequestedFieldUsesNullFormat) { @@ -845,6 +1029,8 @@ TEST_F(CsvV2ReaderTest, BomIsRemovedFromFirstDataLineWithoutHeader) { output.close(); _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); auto reader = create_reader(bom_path, &_params, _slots, &_state, &_profile); std::vector schema; ASSERT_TRUE(reader->get_schema(&schema).ok()); @@ -862,6 +1048,83 @@ TEST_F(CsvV2ReaderTest, BomIsRemovedFromFirstDataLineWithoutHeader) { EXPECT_EQ(nullable_int_at(*block.get_by_position(0).column, 0), 5); } +// The enclosed line reader sees bytes before DelimitedTextReader removes the BOM. It must skip the +// BOM structurally so the quote immediately after it still starts an enclosed first field and the +// comma inside that field is not recorded as a column separator. +TEST_F(CsvV2ReaderTest, BomBeforeQuotedFirstFieldWithComma) { + const auto bom_path = (_test_dir / "bom_quoted_comma.csv").string(); + std::ofstream output(bom_path, std::ios::binary); + output.write("\xEF\xBB\xBF", 3); + output << "\"alice,team\",20\n"; + output.close(); + + auto string_slot = make_test_slot(&_pool, 10, 0, + make_nullable(std::make_shared()), "name"); + auto score_slot = make_test_slot(&_pool, 11, 1, + make_nullable(std::make_shared()), "score"); + std::vector slots {string_slot, score_slot}; + _params.__set_column_idxs({0, 1}); + _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); + auto reader = create_reader(bom_path, &_params, slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0)), + LocalColumnIndex::top_level(LocalColumnId(1))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(1), LocalIndex(1)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0, 1}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "alice,team"); + EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 0), 20); +} + +// The same ordering matters before logical-row detection: a newline inside the first enclosed +// field belongs to the field, even when the opening quote follows a file BOM. +TEST_F(CsvV2ReaderTest, BomBeforeQuotedFirstFieldWithNewline) { + const auto bom_path = (_test_dir / "bom_quoted_newline.csv").string(); + std::ofstream output(bom_path, std::ios::binary); + output.write("\xEF\xBB\xBF", 3); + output << "\"alice\nteam\",20\n"; + output.close(); + + auto string_slot = make_test_slot(&_pool, 10, 0, + make_nullable(std::make_shared()), "name"); + auto score_slot = make_test_slot(&_pool, 11, 1, + make_nullable(std::make_shared()), "score"); + std::vector slots {string_slot, score_slot}; + _params.__set_column_idxs({0, 1}); + _params.file_attributes.__isset.header_type = false; + _params.file_attributes.text_params.__set_enclose('"'); + _params.file_attributes.text_params.__set_escape('\\'); + auto reader = create_reader(bom_path, &_params, slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(0)), + LocalColumnIndex::top_level(LocalColumnId(1))}; + request->local_positions.emplace(LocalColumnId(0), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(1), LocalIndex(1)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {0, 1}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "alice\nteam"); + EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 0), 20); +} + // Scenario: when FE does not set header_type, CSV v2 must honor skip_lines exactly as the old // reader does. TEST_F(CsvV2ReaderTest, SkipLinesUsedWhenHeaderTypeUnset) { @@ -951,6 +1214,37 @@ TEST_F(CsvV2ReaderTest, NullFormatAndEmptyFieldAsNullProduceNullableValues) { EXPECT_TRUE(is_null_at(*block.get_by_position(1).column, 0)); } +// trim_double_quotes preserves quoted null literals as strings while an unquoted marker remains +// NULL, matching the V1 CSV reader's converted-from-string behavior. +TEST_F(CsvV2ReaderTest, QuotedNullFormatIsAStringLiteral) { + const auto null_path = (_test_dir / "quoted_null_format.csv").string(); + std::ofstream output(null_path, std::ios::binary); + output << "id,name,score\n"; + output << "1,\"\\N\",10\n"; + output << "2,\\N,20\n"; + output.close(); + + _params.file_attributes.__set_trim_double_quotes(true); + _params.file_attributes.text_params.__set_null_format("\\N"); + auto reader = create_reader(null_path, &_params, _slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(1))}; + request->local_positions.emplace(LocalColumnId(1), LocalIndex(0)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {1}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 2); + EXPECT_FALSE(is_null_at(*block.get_by_position(0).column, 0)); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), "\\N"); + EXPECT_TRUE(is_null_at(*block.get_by_position(0).column, 1)); +} + // Scenario: OpenCSV keeps an empty field as an empty string when empty_field_as_null is false, // even if FE passes an empty null_format. This differs from Hive text serde, where an empty // serialization.null.format is a real NULL marker. diff --git a/be/test/format_v2/delimited_text/text_reader_test.cpp b/be/test/format_v2/delimited_text/text_reader_test.cpp index b6402cab5d86d6..f5f7309e490ff1 100644 --- a/be/test/format_v2/delimited_text/text_reader_test.cpp +++ b/be/test/format_v2/delimited_text/text_reader_test.cpp @@ -486,6 +486,35 @@ TEST_F(TextV2ReaderTest, EscapedSeparatorStaysInsideStringField) { EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 0), 10); } +// Hive LazySimpleSerDe treats an even-length escape run as escaped escape characters, so the +// following delimiter remains structural. V1 and V2 must both split the row at that delimiter. +TEST_F(TextV2ReaderTest, DoubleEscapeBeforeSeparatorStillSplitsField) { + const auto escaped_path = (_test_dir / "double_escaped.text").string(); + std::ofstream output(escaped_path, std::ios::binary); + output << R"(1|alice\\|10)" << '\n'; + output.close(); + + _params.file_attributes.text_params.__set_column_separator("|"); + auto reader = create_reader(escaped_path, &_params, _slots, &_state, &_profile); + std::vector schema; + ASSERT_TRUE(reader->get_schema(&schema).ok()); + + auto request = std::make_shared(); + request->non_predicate_columns = {LocalColumnIndex::top_level(LocalColumnId(1)), + LocalColumnIndex::top_level(LocalColumnId(2))}; + request->local_positions.emplace(LocalColumnId(1), LocalIndex(0)); + request->local_positions.emplace(LocalColumnId(2), LocalIndex(1)); + ASSERT_TRUE(reader->open(request).ok()); + + auto block = make_block(schema, {1, 2}); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(nullable_string_at(*block.get_by_position(0).column, 0), R"(alice\)"); + EXPECT_EQ(nullable_int_at(*block.get_by_position(1).column, 0), 10); +} + // Scenario: Hive text supports multi-character field separators. V2 must not split on partial // matches and must still honor FileScanRequest output positions. TEST_F(TextV2ReaderTest, MultiCharacterSeparatorReadsRequestedColumns) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java index f936eccd59ed8b..3ea39987d021f7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java @@ -613,7 +613,7 @@ protected TFileAttributes getFileAttributes() throws UserException { textParams.setNullFormat(""); fileAttributes.setTextParams(textParams); fileAttributes.setHeaderType(""); - if (textParams.isSetEnclose()) { + if (shouldTrimDoubleQuotes(textParams)) { fileAttributes.setTrimDoubleQuotes(true); } fileAttributes.setEnableTextValidateUtf8( @@ -670,6 +670,10 @@ protected TFileAttributes getFileAttributes() throws UserException { return fileAttributes; } + static boolean shouldTrimDoubleQuotes(TFileTextScanRangeParams textParams) { + return textParams.isSetEnclose() && textParams.getEnclose() == (byte) '"'; + } + @Override protected TFileCompressType getFileCompressType(FileSplit fileSplit) throws UserException { TFileCompressType compressType = super.getFileCompressType(fileSplit); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/source/HiveScanNodeTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/source/HiveScanNodeTest.java index e0c7b0464e0511..db7a8abe024004 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/source/HiveScanNodeTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/hive/source/HiveScanNodeTest.java @@ -28,6 +28,7 @@ import org.apache.doris.planner.ScanContext; import org.apache.doris.qe.SessionVariable; import org.apache.doris.thrift.TFileScanRangeParams; +import org.apache.doris.thrift.TFileTextScanRangeParams; import com.google.common.collect.ImmutableMap; import org.junit.Assert; @@ -127,6 +128,16 @@ public void testMarkTransactionalHiveScanParams() { scanParams.getTableFormatParams().getTableFormatType()); } + @Test + public void testTrimDoubleQuotesOnlyForDoubleQuoteEnclose() { + TFileTextScanRangeParams textParams = new TFileTextScanRangeParams(); + textParams.setEnclose((byte) '"'); + Assert.assertTrue(HiveScanNode.shouldTrimDoubleQuotes(textParams)); + + textParams.setEnclose((byte) '\''); + Assert.assertFalse(HiveScanNode.shouldTrimDoubleQuotes(textParams)); + } + private HiveScanNode createHiveScanNode() { return createHiveScanNode(false); } diff --git a/regression-test/data/external_table_p0/tvf/csv_enclose_state.csv b/regression-test/data/external_table_p0/tvf/csv_enclose_state.csv new file mode 100644 index 00000000000000..f6259106d54027 --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/csv_enclose_state.csv @@ -0,0 +1,3 @@ +id,name,score,extra +1,ab"cd,20,x +2,C:\dir\,30,40,y diff --git a/regression-test/data/external_table_p0/tvf/csv_matching_escape_enclose.csv b/regression-test/data/external_table_p0/tvf/csv_matching_escape_enclose.csv new file mode 100644 index 00000000000000..adbafedf68ee96 --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/csv_matching_escape_enclose.csv @@ -0,0 +1,2 @@ +id,name,score +"1","alice","10" diff --git a/regression-test/data/external_table_p0/tvf/csv_quoted_null.csv b/regression-test/data/external_table_p0/tvf/csv_quoted_null.csv new file mode 100644 index 00000000000000..83c42bb7104fb5 --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/csv_quoted_null.csv @@ -0,0 +1,3 @@ +id,name,score +1,"\N",10 +2,\N,20 diff --git a/regression-test/data/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.out b/regression-test/data/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.out new file mode 100644 index 00000000000000..b3b0f227aca48b --- /dev/null +++ b/regression-test/data/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.out @@ -0,0 +1,12 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !enclose_state -- +1 20 x +2 30 40 + +-- !matching_escape_enclose -- +1 alice 10 + +-- !quoted_null -- +1 5C4E false +2 \N true + diff --git a/regression-test/suites/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.groovy b/regression-test/suites/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.groovy new file mode 100644 index 00000000000000..331faa8ddb3ff6 --- /dev/null +++ b/regression-test/suites/external_table_p0/tvf/test_local_tvf_csv_enclose_consistency.groovy @@ -0,0 +1,73 @@ +// 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. + +suite("test_local_tvf_csv_enclose_consistency", "p0,external") { + def backends = sql "show backends" + assertTrue(backends.size() > 0) + def beId = backends[0][0] + def dataPath = context.config.dataPath + "/external_table_p0/tvf" + def filePath = "/" + + def files = [ + "csv_enclose_state.csv", + "csv_matching_escape_enclose.csv", + "csv_quoted_null.csv" + ] + for (def backend : backends) { + for (def file : files) { + scpFiles("root", backend[1], "${dataPath}/${file}", filePath, false) + } + } + + sql "set enable_file_scanner_v2 = true" + + order_qt_enclose_state """ + select id, score, extra + from local( + "file_path" = "${filePath}/csv_enclose_state.csv", + "backend_id" = "${beId}", + "format" = "csv_with_names", + "column_separator" = ",", + "enclose" = "\\\"", + "escape" = "\\\\") + order by id + """ + + order_qt_matching_escape_enclose """ + select id, name, score + from local( + "file_path" = "${filePath}/csv_matching_escape_enclose.csv", + "backend_id" = "${beId}", + "format" = "csv_with_names", + "column_separator" = ",", + "enclose" = "\\\"", + "escape" = "\\\"") + order by id + """ + + order_qt_quoted_null """ + select id, hex(name), name is null + from local( + "file_path" = "${filePath}/csv_quoted_null.csv", + "backend_id" = "${beId}", + "format" = "csv_with_names", + "column_separator" = ",", + "trim_double_quotes" = "true", + "null_format" = "\\\\N") + order by id + """ +}