diff --git a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
index 19139e0f47c..32381e7fc65 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/DataTypeUtil.java
@@ -327,6 +327,11 @@ public static Object getDataBasedOnDataType(String data, DataType actualDataType
           return null;
         }
         return Float.parseFloat(data);
+      } else if (actualDataType == DataTypes.BYTE) {
+        if (data.isEmpty()) {
+          return null;
+        }
+        return Byte.parseByte(data);
       } else if (actualDataType == DataTypes.DOUBLE) {
         if (data.isEmpty()) {
           return null;
diff --git a/integration/presto/src/main/prestosql/org/apache/carbondata/presto/PrestoFilterUtil.java b/integration/presto/src/main/prestosql/org/apache/carbondata/presto/PrestoFilterUtil.java
index e5449f05169..874ca118e00 100755
--- a/integration/presto/src/main/prestosql/org/apache/carbondata/presto/PrestoFilterUtil.java
+++ b/integration/presto/src/main/prestosql/org/apache/carbondata/presto/PrestoFilterUtil.java
@@ -78,6 +78,8 @@ private static DataType spi2CarbondataTypeMapper(HiveColumnHandle columnHandle)
     HiveType colType = columnHandle.getHiveType();
     if (colType.equals(HiveType.HIVE_BOOLEAN)) {
       return DataTypes.BOOLEAN;
+    } else if (colType.equals(HiveType.HIVE_BINARY)) {
+      return DataTypes.BINARY;
     } else if (colType.equals(HiveType.HIVE_SHORT)) {
       return DataTypes.SHORT;
     } else if (colType.equals(HiveType.HIVE_INT)) {
@@ -88,6 +90,8 @@ private static DataType spi2CarbondataTypeMapper(HiveColumnHandle columnHandle)
       return DataTypes.DOUBLE;
     } else if (colType.equals(HiveType.HIVE_STRING)) {
       return DataTypes.STRING;
+    } else if (colType.equals(HiveType.HIVE_BYTE)) {
+      return DataTypes.BYTE;
     } else if (colType.equals(HiveType.HIVE_DATE)) {
       return DataTypes.DATE;
     } else if (colType.equals(HiveType.HIVE_TIMESTAMP)) {
@@ -284,7 +288,7 @@ private static Object convertDataByType(Object rawData, HiveType type) {
       return Integer.valueOf(rawData.toString());
     } else if (type.equals(HiveType.HIVE_LONG)) {
       return rawData;
-    } else if (type.equals(HiveType.HIVE_STRING)) {
+    } else if (type.equals(HiveType.HIVE_STRING) || type.equals(HiveType.HIVE_BINARY)) {
       if (rawData instanceof Slice) {
         return ((Slice) rawData).toStringUtf8();
       } else {
diff --git a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
index 611f33d24a6..9d575358c93 100644
--- a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
+++ b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
@@ -229,6 +229,37 @@ class PrestoTestNonTransactionalTableFiles extends FunSuiteLike with BeforeAndAf
     }
   }
 
+  def buildOnlyBinary(rows: Int, sortColumns: Array[String], path : String): Any = {
+    val fields: Array[Field] = new Array[Field](2)
+    fields(0) = new Field("id", DataTypes.INT)
+    fields(1) = new Field("bin", DataTypes.BINARY)
+    try {
+      val binaryValue1 = String.valueOf("abc")
+      val binaryValue2 = String.valueOf("hello")
+      val binaryValue3 = String.valueOf("abc")
+    val builder = CarbonWriter.builder()
+    val writer =
+      builder.outputPath(path)
+        .uniqueIdentifier(System.currentTimeMillis()).withBlockSize(2).sortBy(sortColumns)
+        .withCsvInput(new Schema(fields)).writtenBy("PrestoTestNonTransactionalCarbonTable").build()
+      writer
+        .write(Array[String](
+          String.valueOf(1),
+          binaryValue1))
+      writer
+        .write(Array[String](
+          String.valueOf(2),
+          binaryValue2))
+      writer
+        .write(Array[String](
+          String.valueOf(3),
+          binaryValue3))
+    writer.close()
+    } catch {
+      case ex: Throwable => throw new RuntimeException(ex)
+    }
+  }
+
   def cleanTestData(): Unit = {
     FileUtils.deleteDirectory(new File(writerPath))
   }
@@ -656,4 +687,23 @@ class PrestoTestNonTransactionalTableFiles extends FunSuiteLike with BeforeAndAf
     FileUtils.deleteDirectory(new File(writerPathComplex))
   }
 
+  test("filter on binary datatype") {
+    val writerPathBinary = storePath + "/sdk_output/files7"
+    FileUtils.deleteDirectory(new File(writerPathBinary))
+    prestoServer.execute("drop table if exists sdk_output.files7")
+    prestoServer
+      .execute(
+        "create table sdk_output.files7(id int, bin varbinary) with" +
+        "(format='CARBON') ")
+    buildOnlyBinary(3, null, writerPathBinary)
+    val actualResult: List[Map[String, Any]] = prestoServer
+      .executeQuery("select * from files7 where bin = cast('abc' as varbinary) ")
+    assert(actualResult.size == 2)
+    assert(actualResult.head("bin").asInstanceOf[Array[Byte]](0) == 97)
+    assert(actualResult.head("bin").asInstanceOf[Array[Byte]](1) == 98)
+    assert(actualResult.head("bin").asInstanceOf[Array[Byte]](2) == 99)
+
+    FileUtils.deleteDirectory(new File(writerPathBinary))
+  }
+
 }