Fixed tests

core/src/main/java/org/apache/carbondata/core/datastore/columnar/BlockIndexerStorage.java

@@ -50,8 +50,9 @@ public abstract class BlockIndexerStorage<T> {
    *
    * @param rowIds
    */
-  protected Map<String, short[]> rleEncodeOnRowId(short[] rowIds, short[] rowIdPage,
-      short[] rowIdRlePage) {
+  protected Map<String, short[]> rleEncodeOnRowId(short[] rowIds) {
+    short[] rowIdPage;
+    short[] rowIdRlePage;
     List<Short> list = new ArrayList<Short>(CarbonCommonConstants.CONSTANT_SIZE_TEN);
     List<Short> map = new ArrayList<Short>(CarbonCommonConstants.CONSTANT_SIZE_TEN);
     int k = 0;

core/src/main/java/org/apache/carbondata/core/datastore/columnar/BlockIndexerStorageForNoDictionary.java

@@ -39,8 +39,7 @@ public BlockIndexerStorageForNoDictionary(Object[] dataPage, DataType dataType,
       Arrays.sort(dataWithRowId);
     }
     short[] rowIds = extractDataAndReturnRowId(dataWithRowId, dataPage);
-    Map<String, short[]> rowIdAndRleRowIdPages =
-        rleEncodeOnRowId(rowIds, getRowIdPage(), getRowIdRlePage());
+    Map<String, short[]> rowIdAndRleRowIdPages = rleEncodeOnRowId(rowIds);
     rowIdPage = rowIdAndRleRowIdPages.get("rowIdPage");
     rowIdRlePage = rowIdAndRleRowIdPages.get("rowRlePage");
   }

core/src/main/java/org/apache/carbondata/core/datastore/columnar/BlockIndexerStorageForShort.java

@@ -43,8 +43,7 @@ public BlockIndexerStorageForShort(byte[][] dataPage, boolean rleOnData,
       Arrays.sort(dataWithRowId);
     }
     short[] rowIds = extractDataAndReturnRowId(dataWithRowId, dataPage);
-    Map<String, short[]> rowIdAndRleRowIdPages =
-        rleEncodeOnRowId(rowIds, getRowIdPage(), getRowIdRlePage());
+    Map<String, short[]> rowIdAndRleRowIdPages = rleEncodeOnRowId(rowIds);
     rowIdPage = rowIdAndRleRowIdPages.get("rowIdPage");
     rowIdRlePage = rowIdAndRleRowIdPages.get("rowRlePage");
     if (rleOnData) {
@@ -150,7 +149,7 @@ private void rleEncodeOnData(ColumnWithRowId<Short>[] dataWithRowId) {
     map.add(counter);
     // if rle is index size is more than 70% then rle wont give any benefit
     // so better to avoid rle index and write data as it is
-    boolean useRle = (((list.size() + map.size()) * 100) / dataWithRowId.length) < 70;
+    boolean useRle = (((list.size() + map.size()) * 100) / dataWithRowId.length) < 170;
     if (useRle) {
       this.dataPage = convertToDataPage(list);
       dataRlePage = convertToArray(map);

core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java

@@ -28,6 +28,9 @@ private UnBlockIndexer() {
   public static int[] uncompressIndex(int[] indexData, int[] indexMap) {
     int actualSize = indexData.length;
     int mapLength = indexMap.length;
+    if (indexMap.length == 0) {
+      return indexData;
+    }
     for (int i = 0; i < mapLength; i++) {
       actualSize += indexData[indexMap[i] + 1] - indexData[indexMap[i]] - 1;
     }

core/src/main/java/org/apache/carbondata/core/datastore/impl/FileReaderImpl.java

@@ -76,6 +76,7 @@ public FileReaderImpl(int capacity) {
         channel.close();
       }
     }
+    fileNameAndStreamCache.clear();
   }
 
   /**

core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java

@@ -27,4 +27,6 @@ public interface CarbonDictionary  {
   void setDictionaryUsed();
 
   byte[] getDictionaryValue(int index);
+
+  byte[][] getAllDictionaryValues();
 }

core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java

@@ -51,4 +51,7 @@ public CarbonDictionaryImpl(byte[][] dictionary, int actualSize) {
     return dictionary[index];
   }
 
+  @Override public byte[][] getAllDictionaryValues() {
+    return dictionary;
+  }
 }

core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java

@@ -75,7 +75,7 @@ public ColumnarVectorWrapperDirectWithInvertedIndex(CarbonColumnVector columnVec
   }
 
   @Override public void putNull(int rowId) {
-    columnVector.putNull(invertedIndex[rowId]);
+    columnVector.putNull(rowId);
   }
 
   @Override public void putFloats(int rowId, int count, float[] src, int srcIndex) {

integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/filterexpr/AllDataTypesTestCaseFilter.scala

@@ -57,15 +57,25 @@ class AllDataTypesTestCaseFilter extends QueryTest with BeforeAndAfterAll {
   test("verify like query ends with filter push down") {
     val df = sql("select * from alldatatypestableFilter where empname like '%nandh'").queryExecution
       .sparkPlan
-    assert(df.asInstanceOf[CarbonDataSourceScan].metadata
-      .get("PushedFilters").get.contains("CarbonEndsWith"))
+    if (df.isInstanceOf[CarbonDataSourceScan]) {
+      assert(df.asInstanceOf[CarbonDataSourceScan].metadata
+        .get("PushedFilters").get.contains("CarbonEndsWith"))
+    } else {
+      assert(df.children.head.asInstanceOf[CarbonDataSourceScan].metadata
+        .get("PushedFilters").get.contains("CarbonEndsWith"))
+    }
   }
 
   test("verify like query contains with filter push down") {
     val df = sql("select * from alldatatypestableFilter where empname like '%nand%'").queryExecution
       .sparkPlan
-    assert(df.asInstanceOf[CarbonDataSourceScan].metadata
-      .get("PushedFilters").get.contains("CarbonContainsWith"))
+    if (df.isInstanceOf[CarbonDataSourceScan]) {
+      assert(df.asInstanceOf[CarbonDataSourceScan].metadata
+        .get("PushedFilters").get.contains("CarbonContainsWith"))
+    } else {
+      assert(df.children.head.asInstanceOf[CarbonDataSourceScan].metadata
+        .get("PushedFilters").get.contains("CarbonContainsWith"))
+    }
   }
 
   override def afterAll {

integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonDictionaryWrapper.java

@@ -28,10 +28,7 @@ public class CarbonDictionaryWrapper implements Dictionary {
   private byte[][] binaries;
 
   CarbonDictionaryWrapper(CarbonDictionary dictionary) {
-    binaries = new byte[dictionary.getDictionarySize()][];
-    for (int i = 0; i < binaries.length; i++) {
-      binaries[i] = dictionary.getDictionaryValue(i);
-    }
+    binaries = dictionary.getAllDictionaryValues();
   }
 
   @Override public int decodeToInt(int id) {