Merge 5f3afdc into c8c0211

core/src/main/java/org/apache/carbondata/core/constants/CarbonLoadOptionConstants.java

@@ -172,4 +172,17 @@ public final class CarbonLoadOptionConstants {
 
   public static final String CARBON_LOAD_SORT_MEMORY_SPILL_PERCENTAGE_DEFAULT = "0";
 
+
+  /**
+   * carbon binary decoder when writing string data to binary, like decode base64, Hex
+   */
+  @CarbonProperty
+  public static final String CARBON_OPTIONS_BINARY_DECODER = "carbon.binary.decoder";
+
+  public static final String CARBON_OPTIONS_BINARY_DECODER_DEFAULT = "";
+
+  public static final String CARBON_OPTIONS_BINARY_DECODER_BASE64 = "base64";
+
+  public static final String CARBON_OPTIONS_BINARY_DECODER_HEX = "hex";
+
 }

datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala

@@ -970,6 +970,65 @@ class MVCreateTestCase extends QueryTest with BeforeAndAfterAll {
     }
   }
 
+  test("test binary on mv") {
+    val querySQL = "select x19,x20,sum(x18) from all_table group by x19, x20"
+    val querySQL2 = "select x19,x20,sum(x18) from all_table where x20=cast('binary2' as binary ) group by x19, x20"
+
+    sql("drop datamap if exists all_table_mv")
+    sql("drop table if exists all_table")
+
+    sql(
+      """
+        | create table all_table(x1 bigint,x2 bigint,
+        | x3 string,x4 bigint,x5 bigint,x6 int,x7 string,x8 int, x9 int,x10 bigint,
+        | x11 bigint, x12 bigint,x13 bigint,x14 bigint,x15 bigint,x16 bigint,
+        | x17 bigint,x18 bigint,x19 bigint,x20 binary) stored by 'carbondata'""".stripMargin)
+    sql("insert into all_table select 1,1,null,1,1,1,null,1,1,1,1,1,1,1,1,1,1,1,1,'binary1'")
+    sql("insert into all_table select 1,1,null,1,1,1,null,1,1,1,1,1,1,1,1,1,1,12,2,'binary2'")
+    sql("insert into all_table select 1,1,null,1,1,1,null,1,1,1,1,1,1,1,1,1,1,1,2,'binary2'")
+
+    sql("create datamap all_table_mv on table all_table using 'mv' as " + querySQL)
+    sql("rebuild datamap all_table_mv")
+
+    var frame = sql(querySQL)
+    var analyzed = frame.queryExecution.analyzed
+    assert(verifyMVDataMap(analyzed, "all_table_mv"))
+    assert(2 == frame.collect().size)
+    frame.collect().foreach { each =>
+      if (1 == each.get(0)) {
+        assert("binary1".equals(new String(each.getAs[Array[Byte]](1))))
+        assert(1 == each.get(2))
+      } else if (2 == each.get(0)) {
+        assert("binary2".equals(new String(each.getAs[Array[Byte]](1))))
+        assert(13 == each.get(2))
+      } else {
+        assert(false)
+      }
+    }
+
+    frame = sql(querySQL2)
+    analyzed = frame.queryExecution.analyzed
+    assert(verifyMVDataMap(analyzed, "all_table_mv"))
+    assert(1 == frame.collect().size)
+    frame.collect().foreach { each =>
+      if (2 == each.get(0)) {
+        assert("binary2".equals(new String(each.getAs[Array[Byte]](1))))
+        assert(13 == each.get(2))
+      } else {
+        assert(false)
+      }
+    }
+
+    sql("drop table if exists all_table")
+  }
+
+  def verifyMVDataMap(logicalPlan: LogicalPlan, dataMapName: String): Boolean = {
+    val tables = logicalPlan collect {
+      case l: LogicalRelation => l.catalogTable.get
+    }
+    tables.exists(_.identifier.table.equalsIgnoreCase(dataMapName + "_table"))
+  }
+
   def drop(): Unit = {
     sql("drop table IF EXISTS fact_table1")
     sql("drop table IF EXISTS fact_table2")