[GLUTEN-12597][CORE] Migrate WriteRel to Substrait 0.98 (relocate bucket_spec; add create_mode/advanced_extension)

docs/developers/SubstraitModifications.md

@@ -11,10 +11,16 @@ Since it is still under active development, there are some lacking representatio
 operations. At the same time, some existing representations need to be modified a bit to satisfy the needs of computing.
 
 
-In Gluten, the base version of Substrait is `v0.23.0`. This page records all the Gluten changes to Substrait proto
+In Gluten, the base version of Substrait is `v0.23.0`, with some messages since rebased onto later upstream
+versions (noted per entry below). This page records all the Gluten changes to Substrait proto
 files for reference. It is preferred to upstream these changes to Substrait, but for those cannot be upstreamed,
 alternatives like `AdvancedExtension` could be considered.
 
+**Numbering convention for local additions.** Gluten-local fields are numbered from `1000` up, outside the range
+upstream allocates, so that a future upstream field addition cannot collide with them. This convention exists
+because the older practice of grafting local fields onto the next free low number caused collisions: Gluten's
+`bucket_spec` was originally grafted at `WriteRel` field 7, which upstream `v0.98.0` later assigned to `common`.
+
 ## Modifications to algebra.proto
 
 * Added `JsonReadOptions` and `TextReadOptions` in `FileOrFiles`([#1584](https://github.com/apache/gluten/pull/1584)).
@@ -30,6 +36,14 @@ changed `Unbounded` in `WindowFunction` into `Unbounded_Preceding` and `Unbounde
 * Added `ref` field in window bound `Preceding` and `Following` ([#5626](https://github.com/apache/gluten/pull/5626)).
 * Added `BucketSpec` field in `WriteRel`([#8386](https://github.com/apache/gluten/pull/8386))
 * Added `StreamKafka` in `ReadRel`([#8321](https://github.com/apache/gluten/pull/8321))
+* Rebased the `WriteRel` body onto upstream `v0.98.0`: field 7 is now `common`, with `create_mode` (8) and
+`advanced_extension` (9) added, and the `OutputMode` value `OUTPUT_MODE_MODIFIED_TUPLES` renamed to
+`OUTPUT_MODE_MODIFIED_RECORDS`. Gluten's `BucketSpec` field moved off field 7 to 1000. The enclosing
+`Rel.write` oneof tag is left at 18 for now; reconciling the whole `Rel` oneof to upstream's numbers is a
+separate change. Note that Gluten attaches its writer configuration to `named_table.advanced_extension`, not
+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.
 
 ## Modifications to type.proto
 

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

@@ -645,42 +645,64 @@ message WriteRel {
   // The type of operation to perform
   WriteOp op = 4;
 
-  // The relation that determines the tuples to add/remove/modify
+  // The relation that determines the records to add/remove/modify
   // the schema must match with table_schema. Default values must be explicitly stated
   // in a ProjectRel at the top of the input. The match must also
   // occur in case of DELETE to ensure multi-engine plans are unequivocal.
   Rel input = 5;
 
+  CreateMode create_mode = 8; // Used with CTAS to determine what to do if the table already exists
+
   // Output mode determines what is the output of executing this rel
   OutputMode output = 6;
-
-  // The bucket spec for the writer.
-  BucketSpec bucket_spec = 7;
+  // Note: tag 7 previously held Gluten's local `bucket_spec` (now 1000). Both `BucketSpec` and
+  // `RelCommon` are length-delimited, so a JAR and a native library built from opposite sides of
+  // that move mis-read tag 7 silently instead of failing; they must be rebuilt together.
+  RelCommon common = 7;
 
   enum WriteOp {
     WRITE_OP_UNSPECIFIED = 0;
-    // The insert of new tuples in a table
+    // The insert of new records in a table
     WRITE_OP_INSERT = 1;
-    // The removal of tuples from a table
+    // The removal of records from a table
     WRITE_OP_DELETE = 2;
-    // The modification of existing tuples within a table
+    // The modification of existing records within a table
     WRITE_OP_UPDATE = 3;
-    // The Creation of a new table, and the insert of new tuples in the table
+    // The Creation of a new table, and the insert of new records in the table
     WRITE_OP_CTAS = 4;
   }
 
+  enum CreateMode {
+    CREATE_MODE_UNSPECIFIED = 0;
+    CREATE_MODE_APPEND_IF_EXISTS = 1; // Append the data to the table if it already exists
+    CREATE_MODE_REPLACE_IF_EXISTS = 2; // Replace the table if it already exists ("OR REPLACE")
+    CREATE_MODE_IGNORE_IF_EXISTS = 3; // Ignore the request if the table already exists ("IF NOT EXISTS")
+    CREATE_MODE_ERROR_IF_EXISTS = 4; // Throw an error if the table already exists (default behavior)
+  }
+
   enum OutputMode {
     OUTPUT_MODE_UNSPECIFIED = 0;
-    // return no tuples at all
+    // return no records at all
     OUTPUT_MODE_NO_OUTPUT = 1;
-    // this mode makes the operator return all the tuple INSERTED/DELETED/UPDATED by the operator.
+    // this mode makes the operator return all the record INSERTED/DELETED/UPDATED by the operator.
     // The operator returns the AFTER-image of any change. This can be further manipulated by operators upstreams
-    // (e.g., retunring the typical "count of modified tuples").
+    // (e.g., retunring the typical "count of modified records").
     // For scenarios in which the BEFORE image is required, the user must implement a spool (via references to
     // subplans in the body of the Rel input) and return those with anounter PlanRel.relations.
-    OUTPUT_MODE_MODIFIED_TUPLES = 2;
+    OUTPUT_MODE_MODIFIED_RECORDS = 2;
   }
 
+  // Note: Gluten's writer carries its configuration in `named_table.advanced_extension`, not in
+  // this field, and nothing on the native side reads this one.
+  substrait.extensions.AdvancedExtension advanced_extension = 9;
+
+  // Gluten addition (not in upstream Substrait): the bucket spec for the writer, used for
+  // Hive-compatible bucketed writes. This started at field 7, the next free number when Gluten's
+  // Substrait base was v0.23.0, and upstream 0.98 then claimed 7 for `common`. Numbering the
+  // graft from 1000 up keeps it clear of the range upstream allocates from (WriteRel currently
+  // ends at 9), so a future upstream field addition cannot collide with it again.
+  BucketSpec bucket_spec = 1000;
+
   // A container for bucketing information.
   message BucketSpec {
     int32 num_buckets = 1;

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

@@ -0,0 +1,104 @@
+/*
+ * 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.substrait.`type`.TypeBuilder
+import org.apache.gluten.substrait.SubstraitContext
+
+import com.google.protobuf.Descriptors.Descriptor
+import io.substrait.proto.WriteRel
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Collections
+
+/**
+ * Pins the wire tags of the vendored `WriteRel` after its rebase onto upstream Substrait v0.98.0. A
+ * round trip through the generated classes cannot catch a renumber or an enum-value rename, because
+ * producer and consumer share one schema, so these assert on the descriptors instead. See
+ * docs/developers/SubstraitModifications.md for the numbering convention.
+ */
+class WriteRelProtoSuite extends AnyFunSuite {
+
+  private def assertFieldNumbers(descriptor: Descriptor, expected: (String, Int)*): Unit =
+    expected.foreach {
+      case (name, number) =>
+        val field = descriptor.findFieldByName(name)
+        assert(field != null, s"${descriptor.getName} has no field named $name")
+        assert(field.getNumber === number, s"${descriptor.getName} field $name changed its number")
+    }
+
+  test("makeWriteRel targets a named table with an explicit schema and bucket spec") {
+    val context = new SubstraitContext
+    val bucketSpec = WriteRel.BucketSpec
+      .newBuilder()
+      .setNumBuckets(4)
+      .addBucketColumnNames("c0")
+      .build()
+    val rel = RelBuilder.makeWriteRel(
+      null,
+      Collections.singletonList(TypeBuilder.makeI32(false)),
+      Collections.singletonList("c0"),
+      Collections.emptyList(),
+      null,
+      bucketSpec,
+      context,
+      0L)
+    val writeRel = rel.toProtobuf.getWrite
+
+    assert(writeRel.hasNamedTable)
+    assert(writeRel.hasTableSchema)
+    assert(!writeRel.hasInput)
+    assert(writeRel.hasBucketSpec)
+    assert(writeRel.getBucketSpec.getNumBuckets === 4)
+    assert(writeRel.getBucketSpec.getBucketColumnNamesList.contains("c0"))
+    // The producer never sets these 0.98 fields; they must stay at their proto defaults.
+    assert(writeRel.getOp === WriteRel.WriteOp.WRITE_OP_UNSPECIFIED)
+    assert(writeRel.getOutput === WriteRel.OutputMode.OUTPUT_MODE_UNSPECIFIED)
+    assert(!writeRel.hasCommon)
+  }
+
+  test("WriteRel field numbers and OutputMode value match upstream Substrait v0.98.0") {
+    assertFieldNumbers(
+      WriteRel.getDescriptor,
+      "named_table" -> 1,
+      "extension_table" -> 2,
+      "table_schema" -> 3,
+      "op" -> 4,
+      "input" -> 5,
+      "output" -> 6,
+      "common" -> 7,
+      "create_mode" -> 8,
+      "advanced_extension" -> 9,
+      // Gluten-local graft, relocated off upstream's field 7 to the 1000+ range.
+      "bucket_spec" -> 1000
+    )
+    assertFieldNumbers(
+      WriteRel.BucketSpec.getDescriptor,
+      "num_buckets" -> 1,
+      "bucket_column_names" -> 2,
+      "sort_column_names" -> 3)
+
+    // Upstream renamed the modified-rows value; tag 2 must carry the new name and not the old one.
+    val modifiedRecords = WriteRel.OutputMode.getDescriptor.findValueByNumber(2)
+    assert(modifiedRecords != null, "WriteRel.OutputMode has no value numbered 2")
+    assert(modifiedRecords.getName === "OUTPUT_MODE_MODIFIED_RECORDS")
+    assert(
+      WriteRel.OutputMode.getDescriptor.findValueByName("OUTPUT_MODE_MODIFIED_TUPLES") === null,
+      "the pre-0.98 OUTPUT_MODE_MODIFIED_TUPLES name must be gone"
+    )
+  }
+}