From 31adcd23be1354c18c45ea2abb689362190915c8 Mon Sep 17 00:00:00 2001 From: Zehua Zou Date: Thu, 10 Sep 2026 21:07:59 +0800 Subject: [PATCH 1/2] allow the CSV Reader to accept multi-character delimiters --- cpp/src/arrow/csv/chunker.cc | 97 ++++++++++++++++++++--------- cpp/src/arrow/csv/chunker_test.cc | 21 +++++++ cpp/src/arrow/csv/lexing_internal.h | 63 +++++++++++++++++-- cpp/src/arrow/csv/options.cc | 35 ++++++++++- cpp/src/arrow/csv/options.h | 2 + cpp/src/arrow/csv/parser.cc | 75 +++++++++++++++++----- cpp/src/arrow/csv/parser_test.cc | 32 ++++++++++ cpp/src/arrow/csv/reader_test.cc | 22 +++++++ cpp/src/arrow/dataset/file_csv.cc | 5 ++ 9 files changed, 304 insertions(+), 48 deletions(-) diff --git a/cpp/src/arrow/csv/chunker.cc b/cpp/src/arrow/csv/chunker.cc index 55019e9dab33..602e68ae61a5 100644 --- a/cpp/src/arrow/csv/chunker.cc +++ b/cpp/src/arrow/csv/chunker.cc @@ -47,12 +47,19 @@ class Lexer { }; explicit Lexer(const ParseOptions& options) - : options_(options), bulk_filter_(options_) { + : options_(options), + bulk_filter_(options_), + delimiter_matcher_(InitDelimiterMatcher(options_)) { DCHECK_EQ(SpecializedOptions::quoting, options_.quoting); DCHECK_EQ(SpecializedOptions::escaping, options_.escaping); } - void Reset() { state_ = FIELD_START; } + void Reset() { + state_ = FIELD_START; + if constexpr (SpecializedOptions::multi_delimiter) { + delimiter_matcher_.Reset(); + } + } // Decide whether it's worth using a bulk filter over the given data area bool ShouldUseBulkFilter(const char* data, const char* data_end) { @@ -126,13 +133,27 @@ class Lexer { InField: // Inside a non-quoted part of a field - if (UseBulkFilter) { - const char* bulk_end = RunBulkFilter(data, data_end); - if (ARROW_PREDICT_FALSE(bulk_end == nullptr)) { - state_ = IN_FIELD; - goto AbortLine; + if constexpr (UseBulkFilter) { + if constexpr (SpecializedOptions::multi_delimiter) { + if (!delimiter_matcher_.has_partial_match()) { + const char* bulk_end = RunBulkFilter(data, data_end); + if (ARROW_PREDICT_FALSE(bulk_end == nullptr)) { + state_ = IN_FIELD; + goto AbortLine; + } + data = bulk_end; + } else if (ARROW_PREDICT_FALSE(data == data_end)) { + state_ = IN_FIELD; + goto AbortLine; + } + } else { + const char* bulk_end = RunBulkFilter(data, data_end); + if (ARROW_PREDICT_FALSE(bulk_end == nullptr)) { + state_ = IN_FIELD; + goto AbortLine; + } + data = bulk_end; } - data = bulk_end; } else { if (ARROW_PREDICT_FALSE(data == data_end)) { state_ = IN_FIELD; @@ -141,6 +162,9 @@ class Lexer { } c = *data++; if (SpecializedOptions::escaping && ARROW_PREDICT_FALSE(c == options_.escape_char)) { + if constexpr (SpecializedOptions::multi_delimiter) { + delimiter_matcher_.Reset(); + } if (ARROW_PREDICT_FALSE(data == data_end)) { state_ = AT_ESCAPE; goto AbortLine; @@ -157,9 +181,14 @@ class Lexer { if (ARROW_PREDICT_FALSE(c == '\n')) { goto LineEnd; } - // treat delimiter as a normal token if quoting is disabled - if (ARROW_PREDICT_FALSE(SpecializedOptions::quoting && c == options_.delimiter)) { - goto FieldEnd; + if constexpr (SpecializedOptions::quoting) { + if constexpr (SpecializedOptions::multi_delimiter) { + if (ARROW_PREDICT_FALSE(delimiter_matcher_.Consume(c))) { + goto FieldEnd; + } + } else if (ARROW_PREDICT_FALSE(c == options_.delimiter)) { + goto FieldEnd; + } } goto InField; @@ -229,6 +258,9 @@ class Lexer { LineEnd: state_ = FIELD_START; + if constexpr (SpecializedOptions::multi_delimiter) { + delimiter_matcher_.Reset(); + } return data; AbortLine: @@ -262,6 +294,21 @@ class Lexer { const ParseOptions& options_; const BulkFilterType bulk_filter_; State state_ = FIELD_START; + + private: + struct Empty {}; + + static auto InitDelimiterMatcher(const ParseOptions& options) { + if constexpr (SpecializedOptions::multi_delimiter) { + return internal::StreamingDelimiterMatcher(internal::GetDelimiter(options)); + } else { + return Empty{}; + } + } + + [[no_unique_address]] + std::conditional_t delimiter_matcher_; }; // A BoundaryFinder implementation that assumes CSV cells can contain raw newlines, @@ -380,23 +427,17 @@ std::unique_ptr MakeChunker(const ParseOptions& options) { if (!options.newlines_in_values) { delimiter = MakeNewlineBoundaryFinder(); } else { - if (options.quoting) { - if (options.escaping) { - delimiter = std::make_shared< - LexingBoundaryFinder>>(options); - } else { - delimiter = std::make_shared< - LexingBoundaryFinder>>(options); - } - } else { - if (options.escaping) { - delimiter = std::make_shared< - LexingBoundaryFinder>>(options); - } else { - delimiter = std::make_shared< - LexingBoundaryFinder>>(options); - } - } + // Unlike the parser, the chunker only needs delimiters to identify field starts + // for quote recognition. Without quoting, delimiters cannot affect line boundaries. + const bool multi_delimiter = options.quoting && !options.delimiter_string.empty(); + delimiter = internal::DispatchBool( + [&]() + -> std::shared_ptr { + using SpecializedOptions = + internal::SpecializedOptions; + return std::make_shared>(options); + }, + options.quoting, options.escaping, multi_delimiter); } return std::make_unique(std::move(delimiter)); } diff --git a/cpp/src/arrow/csv/chunker_test.cc b/cpp/src/arrow/csv/chunker_test.cc index 40aace314b26..36ff219ea366 100644 --- a/cpp/src/arrow/csv/chunker_test.cc +++ b/cpp/src/arrow/csv/chunker_test.cc @@ -193,6 +193,27 @@ TEST_P(BaseChunkerTest, QuotingNewline) { } } +TEST_P(BaseChunkerTest, MultiDelimiter) { + if (!options_.newlines_in_values) { + return; + } + options_.delimiter_string = "||"; + MakeChunker(); + + auto partial = std::make_shared("name|"); + auto block = std::make_shared("|message||\"hello\nworld\"||42\nnext||row\n"); + std::shared_ptr completion; + std::shared_ptr rest; + ASSERT_OK(chunker_->ProcessWithPartial(partial, block, &completion, &rest)); + ASSERT_EQ(completion->ToString(), "|message||\"hello\nworld\"||42\n"); + ASSERT_EQ(rest->ToString(), "next||row\n"); + + partial = std::make_shared("name|"); + block = std::make_shared("|message"); + ASSERT_RAISES(Invalid, + chunker_->ProcessWithPartial(partial, block, &completion, &rest)); +} + TEST_P(BaseChunkerTest, QuotingUnbalanced) { // Quote introduces a quoted field that doesn't end auto csv = MakeCSVData({"a,b\n", "1,\",3,,5\n", "c,d\n"}); diff --git a/cpp/src/arrow/csv/lexing_internal.h b/cpp/src/arrow/csv/lexing_internal.h index d5ca120eb58a..13e9fc3b1ab3 100644 --- a/cpp/src/arrow/csv/lexing_internal.h +++ b/cpp/src/arrow/csv/lexing_internal.h @@ -19,22 +19,25 @@ #include #include +#include #include #include #include #include "arrow/csv/options.h" +#include "arrow/util/logging_internal.h" #include "arrow/util/simd.h" namespace arrow { namespace csv { namespace internal { -template +template class SpecializedOptions { public: static constexpr bool quoting = Quoting; static constexpr bool escaping = Escaping; + static constexpr bool multi_delimiter = MultiDelimiter; }; /// Convert runtime boolean options into template arguments for a callable. @@ -61,6 +64,58 @@ decltype(auto) DispatchBool(Fn&& fn, Rest... rest) } } +inline std::string_view GetDelimiter(const ParseOptions& options) { + if (!options.delimiter_string.empty()) { + return options.delimiter_string; + } + return {&options.delimiter, 1}; +} + +inline char GetDelimiterFirstByte(const ParseOptions& options) { + if (!options.delimiter_string.empty()) { + return options.delimiter_string.front(); + } + return options.delimiter; +} + +class StreamingDelimiterMatcher { + public: + explicit StreamingDelimiterMatcher(std::string_view delimiter) : delimiter_(delimiter) { + DCHECK(!delimiter_.empty()); + } + + bool Consume(char c) { + if (c == delimiter_[matched_]) { + matched_++; + if (matched_ == delimiter_.size()) { + matched_ = 0; + return true; + } + return false; + } + + const std::string_view delimiter(delimiter_); + for (size_t length = matched_; length > 0; --length) { + if (c == delimiter[length - 1] && + delimiter.substr(0, length - 1) == + delimiter.substr(matched_ - (length - 1), length - 1)) { + matched_ = length; + return false; + } + } + matched_ = 0; + return false; + } + + void Reset() { matched_ = 0; } + + bool has_partial_match() const { return matched_ != 0; } + + private: + const std::string delimiter_; + size_t matched_ = 0; +}; + // // Bulk filters for packed character matching. // These filters allow checking multiple CSV bytes at once for specific @@ -87,7 +142,7 @@ class BaseBloomFilter { auto add_char = [&](char c) { filter |= CharFilter(c); }; add_char('\n'); add_char('\r'); - add_char(options.delimiter); + add_char(GetDelimiterFirstByte(options)); if (options.escaping) { add_char(options.escape_char); } @@ -188,7 +243,7 @@ class SSE42Filter { // Make a SIMD word of the characters we want to match const char cr = '\r'; const char lf = '\n'; - const char delim = options.delimiter; + const char delim = GetDelimiterFirstByte(options); const char quote = SpecializedOptions::quoting ? options.quote_char : cr; const char escape = SpecializedOptions::escaping ? options.escape_char : cr; @@ -213,7 +268,7 @@ class NeonFilter { using WordType = uint8x8_t; explicit NeonFilter(const ParseOptions& options) - : delim_(vdup_n_u8(options.delimiter)), + : delim_(vdup_n_u8(GetDelimiterFirstByte(options))), quote_(vdup_n_u8(SpecializedOptions::quoting ? options.quote_char : '\n')), escape_(vdup_n_u8(SpecializedOptions::escaping ? options.escape_char : '\n')) {} diff --git a/cpp/src/arrow/csv/options.cc b/cpp/src/arrow/csv/options.cc index 365b5646b66c..143a46def172 100644 --- a/cpp/src/arrow/csv/options.cc +++ b/cpp/src/arrow/csv/options.cc @@ -23,8 +23,39 @@ namespace csv { ParseOptions ParseOptions::Defaults() { return ParseOptions(); } Status ParseOptions::Validate() const { - if (ARROW_PREDICT_FALSE(delimiter == '\n' || delimiter == '\r')) { - return Status::Invalid("ParseOptions: delimiter cannot be \\r or \\n"); + // The chunker handles escapes before delimiter matching, so allowing the escape + // character in a delimiter could make it disagree with the parser. + if (escaping) { + if (delimiter_string.empty()) { + if (delimiter == escape_char) { + return Status::Invalid( + "ParseOptions: delimiter cannot be the escape character when escaping is " + "enabled"); + } + } else if (delimiter_string.find(escape_char) != std::string::npos) { + return Status::Invalid( + "ParseOptions: delimiter_string cannot contain the escape character when " + "escaping is enabled"); + } + } + // Line endings delimit records and therefore cannot be part of a field delimiter. + if (delimiter_string.empty()) { + if (ARROW_PREDICT_FALSE(delimiter == '\n' || delimiter == '\r')) { + return Status::Invalid("ParseOptions: delimiter cannot be \\r or \\n"); + } + } else if (ARROW_PREDICT_FALSE(delimiter_string.find_first_of("\r\n") != + std::string::npos)) { + return Status::Invalid("ParseOptions: delimiter_string cannot contain \\r or \\n"); + } + // The chunker handles quotes before delimiter matching at field boundaries, so + // allowing the quote character at the start of a delimiter could make it disagree + // with the parser. + const char delimiter_first_byte = + delimiter_string.empty() ? delimiter : delimiter_string.front(); + if (ARROW_PREDICT_FALSE(quoting && delimiter_first_byte == quote_char)) { + return Status::Invalid( + "ParseOptions: delimiter cannot start with the quote character when quoting is " + "enabled"); } if (ARROW_PREDICT_FALSE(quoting && (quote_char == '\n' || quote_char == '\r'))) { return Status::Invalid("ParseOptions: quote_char cannot be \\r or \\n"); diff --git a/cpp/src/arrow/csv/options.h b/cpp/src/arrow/csv/options.h index 41d0b63dacbb..4379ed11e26d 100644 --- a/cpp/src/arrow/csv/options.h +++ b/cpp/src/arrow/csv/options.h @@ -65,6 +65,8 @@ struct ARROW_EXPORT ParseOptions { bool pad_short_rows = false; /// Whether rows with more columns than expected should ignore the extra columns. bool ignore_extra_columns = false; + /// Multi-character field delimiter. If non-empty, overrides `delimiter`. + std::string delimiter_string; /// Create parsing options with default values static ParseOptions Defaults(); diff --git a/cpp/src/arrow/csv/parser.cc b/cpp/src/arrow/csv/parser.cc index 20f984ffeb66..e94aa6668ab8 100644 --- a/cpp/src/arrow/csv/parser.cc +++ b/cpp/src/arrow/csv/parser.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include @@ -59,6 +60,22 @@ Status MismatchingColumns(const InvalidRow& row) { inline bool IsControlChar(uint8_t c) { return c < ' '; } +enum class DelimiterMatch { NoMatch, Match, Incomplete }; + +DelimiterMatch MatchDelimiter(std::string_view delimiter, const char* data, + const char* data_end, bool is_final) { + DCHECK(!delimiter.empty()); + const auto available = static_cast(data_end - data); + const auto compared = std::min(available, delimiter.size()); + if (std::memcmp(data, delimiter.data(), compared) != 0) { + return DelimiterMatch::NoMatch; + } + if (available < delimiter.size()) { + return is_final ? DelimiterMatch::NoMatch : DelimiterMatch::Incomplete; + } + return DelimiterMatch::Match; +} + template constexpr bool ShouldWrite([[maybe_unused]] bool ignoring_extra_field) { if constexpr (IgnoreExtraColumns) { @@ -341,16 +358,31 @@ class BlockParserImpl { FieldStart: // At the start of a field - if (*data == options_.delimiter) { - // Empty cells are very common in some files, shortcut them - StartField(false /* quoted */); - FinishField(); - ++data; - ++num_cols; - if (ARROW_PREDICT_FALSE(data == data_end)) { - goto AbortLine; + { + size_t delimiter_len = 0; + if constexpr (SpecializedOptions::multi_delimiter) { + const auto delimiter = internal::GetDelimiter(options_); + const auto delimiter_match = MatchDelimiter(delimiter, data, data_end, is_final); + if (ARROW_PREDICT_FALSE(delimiter_match == DelimiterMatch::Incomplete)) { + goto AbortLine; + } + if (ARROW_PREDICT_FALSE(delimiter_match == DelimiterMatch::Match)) { + delimiter_len = delimiter.size(); + } + } else if (*data == options_.delimiter) { + delimiter_len = 1; + } + if (ARROW_PREDICT_FALSE(delimiter_len > 0)) { + // Empty cells are very common in some files, shortcut them + StartField(false /* quoted */); + FinishField(); + data += delimiter_len; + ++num_cols; + if (ARROW_PREDICT_FALSE(data == data_end)) { + goto AbortLine; + } + goto FieldStart; } - goto FieldStart; } // Quoting is only recognized at start of field @@ -382,8 +414,9 @@ class BlockParserImpl { } } - c = *data++; + c = *data; if (SpecializedOptions::escaping && ARROW_PREDICT_FALSE(c == options_.escape_char)) { + ++data; if (ARROW_PREDICT_FALSE(data == data_end)) { goto AbortLine; } @@ -393,9 +426,21 @@ class BlockParserImpl { } goto InField; } - if (ARROW_PREDICT_FALSE(c == options_.delimiter)) { + if constexpr (SpecializedOptions::multi_delimiter) { + const auto delimiter = internal::GetDelimiter(options_); + const auto delimiter_match = MatchDelimiter(delimiter, data, data_end, is_final); + if (ARROW_PREDICT_FALSE(delimiter_match == DelimiterMatch::Incomplete)) { + goto AbortLine; + } + if (ARROW_PREDICT_FALSE(delimiter_match == DelimiterMatch::Match)) { + data += delimiter.size(); + goto FieldEnd; + } + } else if (ARROW_PREDICT_FALSE(c == options_.delimiter)) { + ++data; goto FieldEnd; } + ++data; if (ARROW_PREDICT_FALSE(IsControlChar(c))) { if (c == '\r') { // In the middle of a newline separator? @@ -727,12 +772,14 @@ class BlockParserImpl { Status Parse(const std::vector& data, bool is_final, uint32_t* out_size) { return internal::DispatchBool( - [&]() { - using SpecializedOptions = internal::SpecializedOptions; + [&]() { + using SpecializedOptions = + internal::SpecializedOptions; return ParseSpecialized(data, is_final, out_size); }, - options_.quoting, options_.escaping, options_.ignore_extra_columns); + options_.quoting, options_.escaping, options_.ignore_extra_columns, + !options_.delimiter_string.empty()); } protected: diff --git a/cpp/src/arrow/csv/parser_test.cc b/cpp/src/arrow/csv/parser_test.cc index bd09f9911831..107eb2defd6b 100644 --- a/cpp/src/arrow/csv/parser_test.cc +++ b/cpp/src/arrow/csv/parser_test.cc @@ -268,6 +268,38 @@ TEST(BlockParser, Basics) { } } +TEST(BlockParser, MultiDelimiter) { + auto options = ParseOptions::Defaults(); + options.delimiter_string = "||"; + + BlockParser parser(options); + AssertParseFinal(parser, + Views({"name||message||score\n", "alice||\"hello||world\"||42"})); + AssertColumnsEq(parser, + {{"name", "alice"}, {"message", "hello||world"}, {"score", "42"}}, + {{false, false}, {false, true}, {false, false}}); +} + +TEST(BlockParser, DelimiterPrefix) { + auto options = ParseOptions::Defaults(); + options.delimiter_string = "||"; + + BlockParser parser(options, /*num_cols=*/2); + AssertParsePartial(parser, "a||b|", 0); + AssertParseFinal(parser, "a||b|"); + AssertColumnsEq(parser, {{"a"}, {"b|"}}); +} + +TEST(BlockParser, SingleCharacterDelimiterString) { + auto options = ParseOptions::Defaults(); + options.delimiter = ','; + options.delimiter_string = "|"; + + BlockParser parser(options); + AssertParseFinal(parser, "a|b\n"); + AssertColumnsEq(parser, {{"a"}, {"b"}}); +} + TEST(BlockParser, PadShortRows) { auto options = ParseOptions::Defaults(); options.pad_short_rows = true; diff --git a/cpp/src/arrow/csv/reader_test.cc b/cpp/src/arrow/csv/reader_test.cc index 009bbd5fc25a..ec1893e2dac5 100644 --- a/cpp/src/arrow/csv/reader_test.cc +++ b/cpp/src/arrow/csv/reader_test.cc @@ -620,6 +620,28 @@ TEST(ReaderTests, DefaultColumnTypeAllStringsNoHeader) { ASSERT_TRUE(table->Equals(*expected_table)); } +TEST(ReaderTests, MultiDelimiter) { + auto input = + std::make_shared(std::make_shared("a||b||c\n1||2||3\n")); + auto read_options = ReadOptions::Defaults(); + read_options.block_size = 8; + auto parse_options = ParseOptions::Defaults(); + parse_options.delimiter_string = "||"; + auto convert_options = ConvertOptions::Defaults(); + convert_options.default_column_type = utf8(); + + ASSERT_OK_AND_ASSIGN(auto reader, + TableReader::Make(io::default_io_context(), input, read_options, + parse_options, convert_options)); + ASSERT_OK_AND_ASSIGN(auto table, reader->Read()); + + auto expected_schema = + schema({field("a", utf8()), field("b", utf8()), field("c", utf8())}); + auto expected_table = + TableFromJSON(expected_schema, {R"([{"a":"1", "b":"2", "c":"3"}])"}); + ASSERT_TRUE(table->Equals(*expected_table)); +} + TEST(ReaderTests, ShortRows) { auto input = std::make_shared(std::make_shared("a,b,c\n1,2\n3,,")); diff --git a/cpp/src/arrow/dataset/file_csv.cc b/cpp/src/arrow/dataset/file_csv.cc index 3b9e8d6ca204..2ccd6c47ffea 100644 --- a/cpp/src/arrow/dataset/file_csv.cc +++ b/cpp/src/arrow/dataset/file_csv.cc @@ -373,6 +373,7 @@ bool CsvFileFormat::Equals(const FileFormat& format) const { checked_cast(format).parse_options; return parse_options.delimiter == other_parse_options.delimiter && + parse_options.delimiter_string == other_parse_options.delimiter_string && parse_options.quoting == other_parse_options.quoting && parse_options.quote_char == other_parse_options.quote_char && parse_options.double_quote == other_parse_options.double_quote && @@ -499,6 +500,10 @@ Result> CsvFileFormat::MakeWriter( if (!Equals(*options->format())) { return Status::TypeError("Mismatching format/write options."); } + if (!parse_options.delimiter_string.empty()) { + return Status::NotImplemented( + "Writing CSV files with delimiter_string is not supported"); + } auto csv_options = checked_pointer_cast(options); ARROW_ASSIGN_OR_RAISE( auto writer, csv::MakeCSVWriter(destination, schema, *csv_options->write_options)); From a161efff34c436a89e2120628a67d88f113cf301 Mon Sep 17 00:00:00 2001 From: Zehua Zou Date: Thu, 10 Sep 2026 21:49:27 +0800 Subject: [PATCH 2/2] fix ci --- cpp/src/arrow/csv/chunker.cc | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/cpp/src/arrow/csv/chunker.cc b/cpp/src/arrow/csv/chunker.cc index 602e68ae61a5..85613e8cd2e5 100644 --- a/cpp/src/arrow/csv/chunker.cc +++ b/cpp/src/arrow/csv/chunker.cc @@ -184,10 +184,10 @@ class Lexer { if constexpr (SpecializedOptions::quoting) { if constexpr (SpecializedOptions::multi_delimiter) { if (ARROW_PREDICT_FALSE(delimiter_matcher_.Consume(c))) { - goto FieldEnd; + goto FieldStart; } } else if (ARROW_PREDICT_FALSE(c == options_.delimiter)) { - goto FieldEnd; + goto FieldStart; } } goto InField; @@ -252,10 +252,6 @@ class Lexer { goto InField; } - FieldEnd: - // At the end of a field - goto FieldStart; - LineEnd: state_ = FIELD_START; if constexpr (SpecializedOptions::multi_delimiter) {