[GLUTEN-12597][CORE] Migrate ReadRel text read options to Substrait 0.98 (TextReadOptions -> DelimiterSeparatedTextReadOptions)

cpp-ch/local-engine/Storages/SubstraitSource/ExcelTextFormatFile.cpp

@@ -94,13 +94,16 @@ DB::FormatSettings ExcelTextFormatFile::createFormatSettings() const
     format_settings.csv.delimiter = *delimiter.data();
 
     if (file_info.start() == 0)
-        format_settings.csv.skip_first_lines = file_info.text().header();
+        format_settings.csv.skip_first_lines = file_info.text().header_lines_to_skip();
 
     if (delimiter == "\t" || delimiter == " ")
         format_settings.csv.allow_whitespace_or_tab_as_delimiter = true;
 
-    if (!file_info.text().null_value().empty())
-        format_settings.csv.null_representation = file_info.text().null_value();
+    /// `value_treated_as_null` is an `optional` field, so presence -- not emptiness -- says whether
+    /// the producer asked for a null representation at all. An explicitly set empty string means
+    /// "an empty field is NULL", which is not the same as the field being absent.
+    if (file_info.text().has_value_treated_as_null())
+        format_settings.csv.null_representation = file_info.text().value_treated_as_null();
 
     bool empty_as_null = true;
     if (context->getSettingsRef().has(EXCEL_EMPTY_AS_NULL))

cpp-ch/local-engine/tests/gtest_parser.cpp

@@ -38,7 +38,7 @@ INCBIN(_readcsv_plan, SOURCE_DIR "/utils/extern-local-engine/tests/json/read_stu
 TEST(LocalExecutor, ReadCSV)
 {
     constexpr std::string_view split_template
-        = R"({"items":[{"uriFile":"{replace_local_files}","length":"56","text":{"fieldDelimiter":",","maxBlockSize":"8192","header":"1"},"schema":{"names":["id","name","language"],"struct":{"types":[{"string":{"nullability":"NULLABILITY_NULLABLE"}},{"string":{"nullability":"NULLABILITY_NULLABLE"}},{"string":{"nullability":"NULLABILITY_NULLABLE"}}]}},"metadataColumns":[{}]}]})";
+        = R"({"items":[{"uriFile":"{replace_local_files}","length":"56","text":{"fieldDelimiter":",","maxBlockSize":"8192","headerLinesToSkip":"1"},"schema":{"names":["id","name","language"],"struct":{"types":[{"string":{"nullability":"NULLABILITY_NULLABLE"}},{"string":{"nullability":"NULLABILITY_NULLABLE"}},{"string":{"nullability":"NULLABILITY_NULLABLE"}}]}},"metadataColumns":[{}]}]})";
     const std::string split = replaceLocalFilesWildcards(
         split_template, GLUTEN_SOURCE_URI("/backends-velox/src/test/resources/datasource/csv/student_option_schema.csv"));
     auto plan = local_engine::JsonStringToMessage<substrait::Plan>(EMBEDDED_PLAN(_readcsv_plan));

docs/developers/SubstraitModifications.md

@@ -24,6 +24,9 @@ because the older practice of grafting local fields onto the next free low numbe
 ## Modifications to algebra.proto
 
 * Added `JsonReadOptions` and `TextReadOptions` in `FileOrFiles`([#1584](https://github.com/apache/gluten/pull/1584)).
+`TextReadOptions` has since been rebased onto upstream's `DelimiterSeparatedTextReadOptions`; the two knobs
+this PR added that are still Gluten-local, `max_block_size` and `empty_as_default`, live at 1000/1001. See the
+rebase entry below.
 * Changed join type `JOIN_TYPE_SEMI` to `JOIN_TYPE_LEFT_SEMI` and `JOIN_TYPE_RIGHT_SEMI`([#408](https://github.com/apache/gluten/pull/408)).
 * Added `WindowRel`, added `column_name` and `window_type` in `WindowFunction`,
 changed `Unbounded` in `WindowFunction` into `Unbounded_Preceding` and `Unbounded_Following`, and added WindowType([#485](https://github.com/apache/gluten/pull/485)).
@@ -44,6 +47,18 @@ separate change. Note that Gluten attaches its writer configuration to `named_ta
 to the new top-level `WriteRel.advanced_extension`, which no Gluten code reads. `WriteRel.common` uses
 Gluten's pre-0.98 `RelCommon` copy (missing `rel_anchor`, `Hint.alias`, `Hint.output_names` and the
 saved/loaded computation messages), so it cannot carry a full 0.98 `common` payload.
+* Rebased the text read options in `FileOrFiles` onto upstream `v0.98.0`: `TextReadOptions` is now
+`DelimiterSeparatedTextReadOptions`, adopting upstream's fields verbatim (`field_delimiter` 1, `max_line_size`
+2, `quote` 3, `header_lines_to_skip` 4, `escape` 5, `value_treated_as_null` 6). `header_lines_to_skip` and
+`value_treated_as_null` are the old `header` (was 5) and `null_value` (was 7) renamed, and the deprecated
+`schema` field is dropped. Gluten's two local knobs are relocated to the 1000 range: `max_block_size` to 1000
+(rows per output block -- distinct from upstream's byte-oriented `max_line_size`, so the two are kept as
+separate fields) and `empty_as_default` to 1001. The `file_format` oneof tag stays at 14, which is also
+upstream's number for it. Note that the reused tags kept their old wire types, so a JAR and a native library
+built from opposite sides of this change mis-read each other silently rather than failing -- in particular old
+tag 6 (`escape`, which Gluten's Hive text path always sets) now parses as `value_treated_as_null` -- and the two must be
+rebuilt together. `JsonReadOptions` keeps its Gluten-local fork until upstream adds equivalent JSON read
+options.
 
 ## Modifications to type.proto
 

gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java

@@ -259,17 +259,24 @@ public ReadRel.LocalFiles toProtobuf() {
           String header = fileReadProperties.getOrDefault("header", "0");
           String escape = fileReadProperties.getOrDefault("escape", "");
           String nullValue = fileReadProperties.getOrDefault("nullValue", "");
-          ReadRel.LocalFiles.FileOrFiles.TextReadOptions textReadOptions =
-              ReadRel.LocalFiles.FileOrFiles.TextReadOptions.newBuilder()
-                  .setFieldDelimiter(field_delimiter)
-                  .setQuote(quote)
-                  .setHeader(Long.parseLong(header))
-                  .setEscape(escape)
-                  .setNullValue(nullValue)
-                  .setMaxBlockSize(GlutenConfig.get().textInputMaxBlockSize())
-                  .setEmptyAsDefault(GlutenConfig.get().textIputEmptyAsDefault())
-                  .build();
-          fileBuilder.setText(textReadOptions);
+          GlutenConfig textConfig = GlutenConfig.get();
+          ReadRel.LocalFiles.FileOrFiles.DelimiterSeparatedTextReadOptions.Builder
+              textReadOptionsBuilder =
+                  ReadRel.LocalFiles.FileOrFiles.DelimiterSeparatedTextReadOptions.newBuilder()
+                      .setFieldDelimiter(field_delimiter)
+                      .setQuote(quote)
+                      .setHeaderLinesToSkip(Long.parseLong(header))
+                      .setEscape(escape)
+                      .setMaxBlockSize(textConfig.textInputMaxBlockSize())
+                      .setEmptyAsDefault(textConfig.textIputEmptyAsDefault());
+          // `value_treated_as_null` is an `optional` field: setting it marks the value as present
+          // even when it is the empty string, which upstream defines as "the empty string is NULL
+          // and the file is entirely nullable strings". Leave it unset when no nullValue was given
+          // so that the default stays "disabled".
+          if (!nullValue.isEmpty()) {
+            textReadOptionsBuilder.setValueTreatedAsNull(nullValue);
+          }
+          fileBuilder.setText(textReadOptionsBuilder.build());
           break;
         case JsonReadFormat:
           ReadRel.LocalFiles.FileOrFiles.JsonReadOptions jsonReadOptions =

gluten-substrait/src/main/resources/substrait/proto/substrait/algebra.proto

@@ -153,15 +153,47 @@ message ReadRel {
       message ArrowReadOptions {}
       message OrcReadOptions {}
       message DwrfReadOptions {}
-      message TextReadOptions {
+      message DelimiterSeparatedTextReadOptions {
+        // Delimiter separated files may be compressed.  The reader should
+        // autodetect this and decompress as needed.
+
+        // The character(s) used to separate fields.  Common values are comma,
+        // tab, and pipe.  Multiple characters are allowed.
         string field_delimiter = 1;
-        uint64 max_block_size = 2;
-        NamedStruct schema = 3 [deprecated=true];
-        string quote = 4;
-        uint64 header = 5;
-        string escape = 6;
-        string null_value = 7;
-        bool empty_as_default = 8;
+        // The maximum number of bytes to read from a single line.  If a line
+        // exceeds this limit the resulting behavior is undefined.
+        uint64 max_line_size = 2;
+        // The character(s) used to quote strings.  Common values are single
+        // and double quotation marks.
+        string quote = 3;
+        // The number of lines to skip at the beginning of the file.
+        uint64 header_lines_to_skip = 4;
+        // The character used to escape characters in strings.  Backslash is
+        // a common value.  Note that a double quote mark can also be used as an
+        // escape character but the external quotes should be removed first.
+        string escape = 5;
+        // If this value is encountered (including empty string), the resulting
+        // value is null instead.  Leave unset to disable.  If this value is
+        // provided, the effective schema of this file is comprised entirely of
+        // nullable strings.  If not provided, the effective schema is instead
+        // made up of non-nullable strings.
+        optional string value_treated_as_null = 6;
+
+        // Note: this message was renumbered when it was rebased onto upstream v0.98.0, and the
+        // reused tags kept their old wire types, so a JAR and a native library built from
+        // opposite sides of that move mis-read each other SILENTLY instead of failing; they must
+        // be rebuilt together. Tag 2 was Gluten's `max_block_size` (now 1000) and is upstream's
+        // `max_line_size`, both varint; tag 6 was `escape` and is now `value_treated_as_null`,
+        // both length-delimited -- and Gluten's Hive text path always sets `escape`, so a
+        // stale producer turns every escape character into a null representation.
+
+        // Gluten-local knobs, numbered from 1000 per the convention documented
+        // in docs/developers/SubstraitModifications.md.
+        // Rows per output block for the native reader -- a batching knob,
+        // distinct from upstream's byte-oriented max_line_size above.
+        uint64 max_block_size = 1000;
+        // Treat an empty unquoted field as the column default value.
+        bool empty_as_default = 1001;
       }
       message JsonReadOptions {
         uint64 max_block_size = 1;
@@ -219,7 +251,7 @@ message ReadRel {
         OrcReadOptions orc = 11;
         google.protobuf.Any extension = 12;
         DwrfReadOptions dwrf = 13;
-        TextReadOptions text = 14;
+        DelimiterSeparatedTextReadOptions text = 14;
         JsonReadOptions json = 15;
         IcebergReadOptions iceberg = 16;
         DeltaReadOptions delta = 22;

gluten-substrait/src/test/scala/org/apache/gluten/substrait/rel/DelimiterSeparatedTextReadOptionsProtoSuite.scala

@@ -0,0 +1,177 @@
+/*
+ * 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.
+ */
+package org.apache.gluten.substrait.rel
+
+import org.apache.gluten.config.GlutenConfig
+
+import com.google.protobuf.Descriptors.{Descriptor, FieldDescriptor}
+import io.substrait.proto.ReadRel
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.{Collections, Map => JMap}
+
+/**
+ * Pins the wire tags of the vendored text read options after their rebase onto upstream Substrait
+ * v0.98.0, where `TextReadOptions` became `DelimiterSeparatedTextReadOptions`. Producer and
+ * consumer share one schema, so a renumber or a field rename round-trips cleanly through the
+ * generated classes and cannot be caught by exercising them; these assert on the descriptors
+ * instead.
+ *
+ * The critical guard is that Gluten's `max_block_size` (rows per output block) and upstream's
+ * `max_line_size` (bytes per line) stay two distinct fields. The fork put `max_block_size` on tag
+ * 2, exactly where 0.98 puts `max_line_size`; conflating them would compile but silently change
+ * meaning, so `max_block_size` is relocated to the 1000 graft range and both are pinned.
+ *
+ * The producer tests at the end cover what descriptors cannot: `value_treated_as_null` is
+ * `optional` in 0.98, so setting it -- even to the empty string -- declares "this value is NULL"
+ * and, per upstream, makes the whole file nullable strings. `LocalFilesNode` must therefore leave
+ * it unset when the reader supplied no `nullValue`.
+ */
+class DelimiterSeparatedTextReadOptionsProtoSuite extends AnyFunSuite {
+
+  private def fileOrFiles: Descriptor =
+    ReadRel.LocalFiles.FileOrFiles.getDescriptor
+
+  private def textOptions: Descriptor =
+    ReadRel.LocalFiles.FileOrFiles.DelimiterSeparatedTextReadOptions.getDescriptor
+
+  private def field(name: String, descriptor: Descriptor): FieldDescriptor = {
+    val f = descriptor.findFieldByName(name)
+    assert(f != null, s"${descriptor.getName} has no field named $name")
+    f
+  }
+
+  /** Builds a text split through the real producer so the emitted options can be inspected. */
+  private def producedTextOptions(
+      properties: JMap[String, String]
+  ): ReadRel.LocalFiles.FileOrFiles.DelimiterSeparatedTextReadOptions = {
+    val node = LocalFilesBuilder.makeLocalFiles(
+      Integer.valueOf(0),
+      Collections.singletonList("file:///tmp/students.csv"),
+      Collections.singletonList(java.lang.Long.valueOf(0L)),
+      Collections.singletonList(java.lang.Long.valueOf(56L)),
+      Collections.emptyList[java.lang.Long](),
+      Collections.emptyList[java.lang.Long](),
+      Collections.singletonList[JMap[String, String]](Collections.emptyMap[String, String]()),
+      Collections.emptyList[JMap[String, String]](),
+      LocalFilesNode.ReadFileFormat.TextReadFormat,
+      Collections.emptyList[String](),
+      properties,
+      Collections.emptyList[JMap[String, Object]]()
+    )
+    node.toProtobuf.getItems(0).getText
+  }
+
+  test("the message was renamed to DelimiterSeparatedTextReadOptions") {
+    assert(
+      fileOrFiles.findNestedTypeByName("DelimiterSeparatedTextReadOptions") != null,
+      "FileOrFiles must declare DelimiterSeparatedTextReadOptions")
+    assert(
+      fileOrFiles.findNestedTypeByName("TextReadOptions") === null,
+      "the pre-0.98 TextReadOptions message name must be gone")
+  }
+
+  test("upstream v0.98.0 fields carry their verbatim tags") {
+    assert(field("field_delimiter", textOptions).getNumber === 1)
+    assert(field("max_line_size", textOptions).getNumber === 2)
+    assert(field("quote", textOptions).getNumber === 3)
+    assert(field("header_lines_to_skip", textOptions).getNumber === 4)
+    assert(field("escape", textOptions).getNumber === 5)
+    assert(field("value_treated_as_null", textOptions).getNumber === 6)
+    // Upstream declares value_treated_as_null as `optional`, so it tracks presence.
+    assert(
+      field("value_treated_as_null", textOptions).hasPresence,
+      "value_treated_as_null must stay an optional (presence-tracking) field")
+  }
+
+  test("Gluten-local knobs live in the 1000 graft range, distinct from max_line_size") {
+    // The trap: the fork had max_block_size (rows per block) on tag 2, which 0.98 gives to
+    // max_line_size (bytes per line). Conflating them compiles but silently changes meaning.
+    assert(
+      field("max_block_size", textOptions).getNumber === 1000,
+      "max_block_size (rows per block) must be grafted at 1000, NOT conflated with max_line_size")
+    assert(field("empty_as_default", textOptions).getNumber === 1001)
+  }
+
+  test("the renamed and dropped fork field names are gone") {
+    assert(
+      textOptions.findFieldByName("header") === null,
+      "header was renamed to header_lines_to_skip")
+    assert(
+      textOptions.findFieldByName("null_value") === null,
+      "null_value was renamed to value_treated_as_null")
+    assert(
+      textOptions.findFieldByName("schema") === null,
+      "the deprecated schema field was dropped")
+  }
+
+  test("the file_format oneof still exposes text on tag 14 with the renamed message type") {
+    val text = field("text", fileOrFiles)
+    assert(text.getNumber === 14, "FileOrFiles.text changed its number")
+    assert(
+      text.getContainingOneof != null && text.getContainingOneof.getName === "file_format",
+      "FileOrFiles.text must stay in the file_format oneof")
+    assert(
+      text.getMessageType.getName === "DelimiterSeparatedTextReadOptions",
+      "FileOrFiles.text must hold DelimiterSeparatedTextReadOptions")
+  }
+
+  test("LocalFilesNode maps the CSV read properties onto the renamed fields") {
+    val properties = new java.util.HashMap[String, String]()
+    properties.put("field_delimiter", ",")
+    properties.put("quote", "\"")
+    properties.put("header", "2")
+    properties.put("escape", "\\")
+    val opts = producedTextOptions(properties)
+
+    assert(opts.getFieldDelimiter === ",")
+    assert(opts.getQuote === "\"")
+    assert(opts.getHeaderLinesToSkip === 2L)
+    assert(opts.getEscape === "\\")
+    // Gluten always drives this from its own config, never from the read properties. Compare
+    // against the config rather than a fixed value so the assertion cannot depend on whatever
+    // SQLConf happens to be active when the suite runs.
+    assert(
+      opts.getMaxBlockSize === GlutenConfig.get.textInputMaxBlockSize,
+      "max_block_size must come from GlutenConfig")
+    assert(
+      !opts.getEmptyAsDefault,
+      "empty_as_default must come from GlutenConfig, whose default is false")
+    // The mirror image, and the regression this suite exists to prevent: the row count must never
+    // be wired into upstream's byte-oriented max_line_size, which now occupies the tag the fork
+    // used to keep max_block_size on. Gluten produces no value for it at all.
+    assert(opts.getMaxLineSize === 0L, "Gluten must leave upstream's max_line_size unset")
+  }
+
+  test("LocalFilesNode leaves value_treated_as_null unset when no nullValue was supplied") {
+    val opts = producedTextOptions(new java.util.HashMap[String, String]())
+    // Setting the field -- even to "" -- means "this value is NULL and the file is entirely
+    // nullable strings" (see the field comment in algebra.proto). Absent must stay absent.
+    assert(
+      !opts.hasValueTreatedAsNull,
+      "value_treated_as_null must stay absent when the reader gave no nullValue")
+  }
+
+  test("LocalFilesNode sets value_treated_as_null when a nullValue was supplied") {
+    val properties = new java.util.HashMap[String, String]()
+    properties.put("nullValue", "NULL")
+    val opts = producedTextOptions(properties)
+
+    assert(opts.hasValueTreatedAsNull)
+    assert(opts.getValueTreatedAsNull === "NULL")
+  }
+}