From 767d5acc60f79151e9fbe991f1e987cf60454f97 Mon Sep 17 00:00:00 2001
From: Jiebao Xiao <xiaojiebao@xiaomi.com>
Date: Sat, 24 Sep 2022 22:50:41 +0800
Subject: [PATCH] backport patch #3728

---
 .../java/org/apache/iceberg/spark/Spark3Util.java  |  5 ++++-
 .../org/apache/iceberg/spark/SparkTestBase.java    |  6 +++++-
 .../org/apache/iceberg/spark/sql/TestSelect.java   | 14 ++++++++++++++
 3 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
index 2bfd0aaf8da7..45b2a87ef09a 100644
--- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
+++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
@@ -54,6 +54,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.io.BaseEncoding;
 import org.apache.iceberg.spark.SparkTableUtil.SparkPartition;
 import org.apache.iceberg.spark.source.SparkTable;
 import org.apache.iceberg.transforms.PartitionSpecVisitor;
@@ -61,6 +62,7 @@
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.SortOrderUtil;
 import org.apache.spark.sql.Dataset;
@@ -653,7 +655,8 @@ private static String sqlString(org.apache.iceberg.expressions.Literal<?> lit) {
       if (lit.value() instanceof String) {
         return "'" + lit.value() + "'";
       } else if (lit.value() instanceof ByteBuffer) {
-        throw new IllegalArgumentException("Cannot convert bytes to SQL literal: " + lit);
+        byte[] bytes = ByteBuffers.toByteArray((ByteBuffer) lit.value());
+        return "X'" + BaseEncoding.base16().encode(bytes) + "'";
       } else {
         return lit.value().toString();
       }
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/SparkTestBase.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/SparkTestBase.java
index 9db0d6d410ee..fa7f07f3313a 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/SparkTestBase.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/SparkTestBase.java
@@ -168,7 +168,11 @@ private void assertEquals(String context, Object[] expectedRow, Object[] actualR
       Object actualValue = actualRow[col];
       if (expectedValue != null && expectedValue.getClass().isArray()) {
         String newContext = String.format("%s (nested col %d)", context, col + 1);
-        assertEquals(newContext, (Object[]) expectedValue, (Object[]) actualValue);
+        if (expectedValue instanceof byte[]) {
+          Assert.assertArrayEquals(newContext, (byte[]) expectedValue, (byte[]) actualValue);
+        } else {
+          assertEquals(newContext, (Object[]) expectedValue, (Object[]) actualValue);
+        }
       } else if (expectedValue != ANY) {
         Assert.assertEquals(context + " contents should match", expectedValue, actualValue);
       }
diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
index f20ded4c7b2a..4723e69335a0 100644
--- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
+++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSelect.java
@@ -39,6 +39,7 @@
 public class TestSelect extends SparkCatalogTestBase {
   private int scanEventCount = 0;
   private ScanEvent lastScanEvent = null;
+  private String binaryTableName = tableName("binary_table");
 
   public TestSelect(String catalogName, String implementation, Map<String, String> config) {
     super(catalogName, implementation, config);
@@ -64,6 +65,7 @@ public void createTables() {
   @After
   public void removeTables() {
     sql("DROP TABLE IF EXISTS %s", tableName);
+    sql("DROP TABLE IF EXISTS %s", binaryTableName);
   }
 
   @Test
@@ -222,4 +224,16 @@ public void testSpecifySnapshotAndTimestamp() {
               .collectAsList();
         });
   }
+
+  @Test
+  public void testBinaryInFilter() {
+    sql("CREATE TABLE %s (id bigint, binary binary) USING iceberg", binaryTableName);
+    sql("INSERT INTO %s VALUES (1, X''), (2, X'1111'), (3, X'11')", binaryTableName);
+    List<Object[]> expected = ImmutableList.of(row(2L, new byte[] {0x11, 0x11}));
+
+    assertEquals(
+        "Should return all expected rows",
+        expected,
+        sql("SELECT id, binary FROM %s where binary > X'11'", binaryTableName));
+  }
 }