[iceberg] branch master updated: Spark 3.4: Cherry pick case sensitivity backport (#7324)

[ru...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

russellspitzer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/master by this push:
     new 1f0e493fbb Spark 3.4: Cherry pick case sensitivity backport (#7324)
1f0e493fbb is described below

commit 1f0e493fbbdacea736262ff797fda33618786c5b
Author: Karuppayya <ka...@gmail.com>
AuthorDate: Tue Apr 25 13:22:05 2023 -0700

    Spark 3.4: Cherry pick case sensitivity backport (#7324)
---
 .../v2/SetWriteDistributionAndOrderingExec.scala   |  3 ++-
 .../TestSetWriteDistributionAndOrdering.java       | 26 ++++++++++++++++++++++
 2 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/SetWriteDistributionAndOrderingExec.scala b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/SetWriteDistributionAndOrderingExec.scala
index 386485b10b..feecc02350 100644
--- a/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/SetWriteDistributionAndOrderingExec.scala
+++ b/spark/v3.4/spark-extensions/src/main/scala/org/apache/spark/sql/execution/datasources/v2/SetWriteDistributionAndOrderingExec.scala
@@ -24,6 +24,7 @@ import org.apache.iceberg.NullOrder
 import org.apache.iceberg.SortDirection
 import org.apache.iceberg.TableProperties.WRITE_DISTRIBUTION_MODE
 import org.apache.iceberg.expressions.Term
+import org.apache.iceberg.spark.SparkUtil
 import org.apache.iceberg.spark.source.SparkTable
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.Attribute
@@ -46,7 +47,7 @@ case class SetWriteDistributionAndOrderingExec(
       case iceberg: SparkTable =>
         val txn = iceberg.table.newTransaction()
 
-        val orderBuilder = txn.replaceSortOrder()
+        val orderBuilder = txn.replaceSortOrder().caseSensitive(SparkUtil.caseSensitive(session))
         sortOrder.foreach {
           case (term, SortDirection.ASC, nullOrder) =>
             orderBuilder.asc(term, nullOrder)
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSetWriteDistributionAndOrdering.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSetWriteDistributionAndOrdering.java
index e7e5280679..caa9752004 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSetWriteDistributionAndOrdering.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSetWriteDistributionAndOrdering.java
@@ -25,6 +25,9 @@ import org.apache.iceberg.NullOrder;
 import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.spark.sql.internal.SQLConf;
+import org.assertj.core.api.Assertions;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
@@ -64,6 +67,29 @@ public class TestSetWriteDistributionAndOrdering extends SparkExtensionsTestBase
     Assert.assertEquals("Should have expected order", expected, table.sortOrder());
   }
 
+  @Test
+  public void testSetWriteOrderWithCaseSensitiveColumnNames() {
+    sql(
+        "CREATE TABLE %s (Id bigint NOT NULL, Category string, ts timestamp, data string) USING iceberg",
+        tableName);
+    Table table = validationCatalog.loadTable(tableIdent);
+    Assert.assertTrue("Table should start unsorted", table.sortOrder().isUnsorted());
+    sql("SET %s=true", SQLConf.CASE_SENSITIVE().key());
+    Assertions.assertThatThrownBy(
+            () -> {
+              sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
+            })
+        .isInstanceOf(ValidationException.class)
+        .hasMessageContaining("Cannot find field 'category' in struct");
+
+    sql("SET %s=false", SQLConf.CASE_SENSITIVE().key());
+    sql("ALTER TABLE %s WRITE ORDERED BY category, id", tableName);
+    table = validationCatalog.loadTable(tableIdent);
+    SortOrder expected =
+        SortOrder.builderFor(table.schema()).withOrderId(1).asc("Category").asc("Id").build();
+    Assert.assertEquals("Should have expected order", expected, table.sortOrder());
+  }
+
   @Test
   public void testSetWriteOrderByColumnWithDirection() {
     sql(