Revert "[SPARK-33593][SQL] Vector reader got incorrect data with bina…

…ry partition value"

This reverts commit 4de3028.

sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java

@@ -54,8 +54,6 @@ public static void populate(WritableColumnVector col, InternalRow row, int field
     } else {
       if (t == DataTypes.BooleanType) {
         col.putBooleans(0, capacity, row.getBoolean(fieldIdx));
-      } else if (t == DataTypes.BinaryType) {
-        col.putByteArray(0, row.getBinary(fieldIdx));
       } else if (t == DataTypes.ByteType) {
         col.putBytes(0, capacity, row.getByte(fieldIdx));
       } else if (t == DataTypes.ShortType) {
@@ -96,9 +94,6 @@ public static void populate(WritableColumnVector col, InternalRow row, int field
         col.putInts(0, capacity, row.getInt(fieldIdx));
       } else if (t instanceof TimestampType) {
         col.putLongs(0, capacity, row.getLong(fieldIdx));
-      } else {
-        throw new RuntimeException(String.format("DataType %s is not supported" +
-            " in column vectorized reader.", t.sql()));
       }
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -3114,32 +3114,6 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       }
     }
   }
-
-  test("SPARK-33593: Vector reader got incorrect data with binary partition value") {
-    Seq("false", "true").foreach(value => {
-      withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> value) {
-        withTable("t1") {
-          sql(
-            """CREATE TABLE t1(name STRING, id BINARY, part BINARY)
-              |USING PARQUET PARTITIONED BY (part)""".stripMargin)
-          sql("INSERT INTO t1 PARTITION(part = 'Spark SQL') VALUES('a', X'537061726B2053514C')")
-          checkAnswer(sql("SELECT name, cast(id as string), cast(part as string) FROM t1"),
-            Row("a", "Spark SQL", "Spark SQL"))
-        }
-      }
-
-      withSQLConf(SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> value) {
-        withTable("t2") {
-          sql(
-            """CREATE TABLE t2(name STRING, id BINARY, part BINARY)
-              |USING ORC PARTITIONED BY (part)""".stripMargin)
-          sql("INSERT INTO t2 PARTITION(part = 'Spark SQL') VALUES('a', X'537061726B2053514C')")
-          checkAnswer(sql("SELECT name, cast(id as string), cast(part as string) FROM t2"),
-            Row("a", "Spark SQL", "Spark SQL"))
-        }
-      }
-    })
-  }
 }
 
 case class Foo(bar: Option[String])

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala

@@ -734,15 +734,14 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext {
       Seq(1).toDF().repartition(1).write.parquet(dir.getCanonicalPath)
 
       val dataTypes =
-        Seq(StringType, BooleanType, ByteType, BinaryType, ShortType, IntegerType, LongType,
+        Seq(StringType, BooleanType, ByteType, ShortType, IntegerType, LongType,
           FloatType, DoubleType, DecimalType(25, 5), DateType, TimestampType)
 
       val constantValues =
         Seq(
           UTF8String.fromString("a string"),
           true,
           1.toByte,
-          "Spark SQL".getBytes,
           2.toShort,
           3,
           Long.MaxValue,
@@ -770,11 +769,7 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext {
           // in order to use get(...) method which is not implemented in `ColumnarBatch`.
           val actual = row.copy().get(1, dt)
           val expected = v
-          if (dt.isInstanceOf[BinaryType]) {
-            assert(actual.asInstanceOf[Array[Byte]] sameElements expected.asInstanceOf[Array[Byte]])
-          } else {
-            assert(actual == expected)
-          }
+          assert(actual == expected)
         } finally {
           vectorizedReader.close()
         }