[CARBONDATA-3336] Support configurable decode for loading binary data…

…, support base64 and Hex decode.

Support configurable decode for loading binary data, support base64 and Hex decode.
1. support configurable decode for loading
2. test datamap: mv, preaggregate, timeseries, bloomfilter, lucene
3. test datamap and configurable decode

Default non decoder for loading binary data, this PR support base64 and hex decoder

This closes #3188

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")