GH-51275: [C++][GLib] Allow the CSV Reader to accept multi-character delimiters

c_glib/arrow-glib/reader.cpp

@@ -934,6 +934,18 @@ garrow_csv_read_options_dispose(GObject *object)
   G_OBJECT_CLASS(garrow_csv_read_options_parent_class)->dispose(object);
 }
 
+static void
+garrow_csv_read_options_finalize(GObject *object)
+{
+  auto priv = GARROW_CSV_READ_OPTIONS_GET_PRIVATE(object);
+
+  priv->convert_options.~ConvertOptions();
+  priv->parse_options.~ParseOptions();
+  priv->read_options.~ReadOptions();
+
+  G_OBJECT_CLASS(garrow_csv_read_options_parent_class)->finalize(object);
+}
+
 static void
 garrow_csv_read_options_set_property(GObject *object,
                                      guint prop_id,
@@ -1052,6 +1064,9 @@ static void
 garrow_csv_read_options_init(GArrowCSVReadOptions *object)
 {
   auto priv = GARROW_CSV_READ_OPTIONS_GET_PRIVATE(object);
+  new (&priv->read_options) arrow::csv::ReadOptions;
+  new (&priv->parse_options) arrow::csv::ParseOptions;
+  new (&priv->convert_options) arrow::csv::ConvertOptions;
   priv->read_options = arrow::csv::ReadOptions::Defaults();
   priv->parse_options = arrow::csv::ParseOptions::Defaults();
   priv->convert_options = arrow::csv::ConvertOptions::Defaults();
@@ -1064,6 +1079,7 @@ garrow_csv_read_options_class_init(GArrowCSVReadOptionsClass *klass)
 
   auto gobject_class = G_OBJECT_CLASS(klass);
 
+  gobject_class->finalize = garrow_csv_read_options_finalize;
   gobject_class->dispose = garrow_csv_read_options_dispose;
   gobject_class->set_property = garrow_csv_read_options_set_property;
   gobject_class->get_property = garrow_csv_read_options_get_property;

cpp/src/arrow/csv/chunker.cc

@@ -20,6 +20,7 @@
 #include <algorithm>
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <string_view>
 #include <utility>
 
@@ -50,6 +51,9 @@ class Lexer {
       : options_(options), bulk_filter_(options_) {
     DCHECK_EQ(SpecializedOptions::quoting, options_.quoting);
     DCHECK_EQ(SpecializedOptions::escaping, options_.escaping);
+    if constexpr (SpecializedOptions::multi_delimiter) {
+      delimiter_matcher_.emplace(options_.delimiter_string);
+    }
   }
 
   void Reset() { state_ = FIELD_START; }
@@ -107,6 +111,9 @@ class Lexer {
     }
 
   FieldStart:
+    if constexpr (SpecializedOptions::multi_delimiter) {
+      delimiter_matcher_->Reset();
+    }
     if constexpr (!SpecializedOptions::quoting) {
       goto InField;
     } else {
@@ -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 FieldStart;
+        }
+      } else if (ARROW_PREDICT_FALSE(c == options_.delimiter)) {
+        goto FieldStart;
+      }
     }
     goto InField;
 
@@ -223,10 +252,6 @@ class Lexer {
       goto InField;
     }
 
-  FieldEnd:
-    // At the end of a field
-    goto FieldStart;
-
   LineEnd:
     state_ = FIELD_START;
     return data;
@@ -262,6 +287,8 @@ class Lexer {
   const ParseOptions& options_;
   const BulkFilterType bulk_filter_;
   State state_ = FIELD_START;
+
+  std::optional<internal::StreamingDelimiterMatcher> delimiter_matcher_;
 };
 
 // A BoundaryFinder implementation that assumes CSV cells can contain raw newlines,
@@ -380,23 +407,22 @@ std::unique_ptr<Chunker> MakeChunker(const ParseOptions& options) {
   if (!options.newlines_in_values) {
     delimiter = MakeNewlineBoundaryFinder();
   } else {
-    if (options.quoting) {
-      if (options.escaping) {
-        delimiter = std::make_shared<
-            LexingBoundaryFinder<internal::SpecializedOptions<true, true>>>(options);
-      } else {
-        delimiter = std::make_shared<
-            LexingBoundaryFinder<internal::SpecializedOptions<true, false>>>(options);
-      }
-    } else {
-      if (options.escaping) {
-        delimiter = std::make_shared<
-            LexingBoundaryFinder<internal::SpecializedOptions<false, true>>>(options);
-      } else {
-        delimiter = std::make_shared<
-            LexingBoundaryFinder<internal::SpecializedOptions<false, false>>>(options);
-      }
+    // Unlike the parser, the chunker only needs delimiters to identify field starts
+    // for quote recognition.  Without quoting, delimiters cannot affect line boundaries.
+    auto lexing_options = options;
+    if (lexing_options.delimiter_string.size() == 1) {
+      lexing_options.delimiter = lexing_options.delimiter_string.front();
     }
+    const bool multi_delimiter = options.quoting && options.delimiter_string.size() > 1;
+    delimiter = internal::DispatchBool(
+        [&]<bool Quoting, bool Escaping, bool MultiDelimiter>()
+            -> std::shared_ptr<BoundaryFinder> {
+          using SpecializedOptions =
+              internal::SpecializedOptions<Quoting, Escaping, MultiDelimiter>;
+          return std::make_shared<LexingBoundaryFinder<SpecializedOptions>>(
+              std::move(lexing_options));
+        },
+        options.quoting, options.escaping, multi_delimiter);
   }
   return std::make_unique<Chunker>(std::move(delimiter));
 }

cpp/src/arrow/csv/chunker_test.cc

@@ -193,6 +193,49 @@ TEST_P(BaseChunkerTest, QuotingNewline) {
   }
 }
 
+TEST_P(BaseChunkerTest, MultiDelimiter) {
+  if (!options_.newlines_in_values) {
+    GTEST_SKIP() << "Quoted newlines require newlines_in_values";
+  }
+  options_.delimiter_string = "||";
+  options_.escaping = true;
+  MakeChunker();
+
+  // The escape breaks the delimiter prefix, so the quote before "me" is literal.
+  auto partial = std::make_shared<Buffer>("na|\\x|\"me|");
+  auto block = std::make_shared<Buffer>("|mes|sage||\"hello\nworld\"||42\nnext||row\n");
+  std::shared_ptr<Buffer> completion;
+  std::shared_ptr<Buffer> rest;
+  ASSERT_OK(chunker_->ProcessWithPartial(partial, block, &completion, &rest));
+  ASSERT_EQ(completion->ToString(), "|mes|sage||\"hello\nworld\"||42\n");
+  ASSERT_EQ(rest->ToString(), "next||row\n");
+
+  partial = std::make_shared<Buffer>("name|");
+  block = std::make_shared<Buffer>("|message");
+  ASSERT_RAISES(Invalid,
+                chunker_->ProcessWithPartial(partial, block, &completion, &rest));
+
+  // Long ordinary fields make the bulk filter profitable.
+  for (size_t padding : {0, 256}) {
+    const std::string first_field = std::string(padding, 'a') + "a|b";
+    const std::string row = first_field + "||\"c\n d\"||e\n";
+    const std::string next_row = std::string(padding, 'n') + "next||row\n";
+    MakeChunker();
+    const auto lengths = {row.size(), next_row.size()};
+    AssertChunking(*chunker_, row + next_row, lengths);
+
+    for (size_t split :
+         {first_field.find('|') + 1, first_field.size() + 1, first_field.size() + 2}) {
+      ARROW_SCOPED_TRACE("padding = ", padding, ", split = ", split);
+      partial = Buffer::FromString(row.substr(0, split));
+      block = Buffer::FromString(row.substr(split) + next_row);
+      ASSERT_OK(chunker_->ProcessWithPartial(partial, block, &completion, &rest));
+      ASSERT_EQ(completion->ToString(), row.substr(split));
+      ASSERT_EQ(rest->ToString(), next_row);
+    }
+  }
+}
+
 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"});

cpp/src/arrow/csv/lexing_internal.h

@@ -19,22 +19,25 @@
 
 #include <cstdint>
 #include <cstring>
+#include <string>
 #include <string_view>
 #include <type_traits>
 #include <utility>
 
 #include "arrow/csv/options.h"
+#include "arrow/util/logging_internal.h"
 #include "arrow/util/simd.h"
 
 namespace arrow {
 namespace csv {
 namespace internal {
 
-template <bool Quoting, bool Escaping>
+template <bool Quoting, bool Escaping, bool MultiDelimiter>
 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,51 @@ decltype(auto) DispatchBool(Fn&& fn, Rest... rest)
   }
 }
 
+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 +135,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 +236,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 +261,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')) {}
 

cpp/src/arrow/csv/options.cc

@@ -23,8 +23,40 @@ 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");
+  }
+  if (quoting) {
+    // Keep structural characters out of delimiters when quoting is enabled.
+    const bool delimiter_contains_quote =
+        delimiter_string.empty() ? delimiter == quote_char
+                                 : delimiter_string.find(quote_char) != std::string::npos;
+    if (ARROW_PREDICT_FALSE(delimiter_contains_quote)) {
+      return Status::Invalid(
+          "ParseOptions: delimiter cannot contain 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");

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();