getBlockId code clean

Author: marchpure

core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java

@@ -446,11 +446,11 @@ private BlockExecutionInfo getBlockExecutionInfoForBlock(QueryModel queryModel,
             segmentProperties.getComplexDimensions(),
             blockExecutionInfo.getActualQueryMeasures().length,
             queryModel.getTable().getTableInfo().isTransactionalTable());
-    boolean isStandardTable = CarbonUtil.isStandardCarbonTable(queryModel.getTable());
+    boolean isFlatFolderTable = queryModel.getTable().isSupportFlatFolder();
     String blockId = CarbonUtil
         .getBlockId(queryModel.getAbsoluteTableIdentifier(), filePath, segment.getSegmentNo(),
             queryModel.getTable().getTableInfo().isTransactionalTable(),
-            isStandardTable, queryModel.getTable().isHivePartitionTable());
+            isFlatFolderTable, queryModel.getTable().isHivePartitionTable());
     blockExecutionInfo.setBlockId(CarbonTablePath.getShortBlockId(blockId));
     blockExecutionInfo.setDeleteDeltaFilePath(deleteDeltaFiles);
     blockExecutionInfo.setStartBlockletIndex(0);

core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java

@@ -2759,49 +2759,41 @@ public static String getBlockId(AbsoluteTableIdentifier identifier, String fileP
    * @return
    */
   public static String getBlockId(AbsoluteTableIdentifier identifier, String filePath,
-      String segmentId, boolean isTransactionalTable, boolean isStandardTable,
+      String segmentId, boolean isTransactionalTable, boolean isFlatFolderTable,
       boolean isPartitionTable) {
-    String blockId;
     String blockName = filePath.substring(filePath.lastIndexOf("/") + 1);
     String tablePath = identifier.getTablePath();
 
-    if (filePath.startsWith(tablePath)) {
-      if (!isTransactionalTable || isStandardTable) {
-        blockId = "Part0" + CarbonCommonConstants.FILE_SEPARATOR + "Segment_" + segmentId
-            + CarbonCommonConstants.FILE_SEPARATOR + blockName;
-      } else {
-        // This is the case with partition table.
-        String partitionDir;
-        int partLength = filePath.length() - blockName.length() - 1;
-        if (tablePath.length() + 1 < partLength) {
-          partitionDir =
-              filePath.substring(tablePath.length() + 1, partLength);
-        } else {
-          partitionDir = "";
-        }
-        if (isPartitionTable) {
-          blockId =
-              partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
-                  + CarbonCommonConstants.FILE_SEPARATOR + blockName;
-        } else {
-          // Replace / with # on partition director to support multi level partitioning. And access
-          // them all as a single entity.
-          if (partitionDir.isEmpty()) {
-            blockId = segmentId + CarbonCommonConstants.FILE_SEPARATOR + blockName;
-          } else {
-            blockId = partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
-                + CarbonCommonConstants.FILE_SEPARATOR + segmentId
-                + CarbonCommonConstants.FILE_SEPARATOR + blockName;
-          }
+    if (!filePath.startsWith(tablePath)) {
+      return filePath.substring(0, filePath.length() - blockName.length()).replace("/", "#")
+          + CarbonCommonConstants.FILE_SEPARATOR + segmentId
+          + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+    }
 
-        }
-      }
-    } else {
-      blockId = filePath.substring(0, filePath.length() - blockName.length()).replace("/", "#")
-          + CarbonCommonConstants.FILE_SEPARATOR + "Segment_" + segmentId
+    if (!isTransactionalTable || !isPartitionTable) {
+      return "0" + CarbonCommonConstants.FILE_SEPARATOR + segmentId
+              + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+    }
+
+    String partitionDir = getPartitionDir(tablePath, filePath, blockName);
+    if (isFlatFolderTable) {
+      return partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
+          + CarbonCommonConstants.FILE_SEPARATOR + segmentId
           + CarbonCommonConstants.FILE_SEPARATOR + blockName;
     }
-    return blockId;
+
+    return partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
+        + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+  }
+
+  public static String getPartitionDir(String tablePath, String filePath, String blockName) {
+    String partitionDir = "";
+    int partLength = filePath.length() - blockName.length() - 1;
+    if (tablePath.length() + 1 < partLength) {
+      partitionDir =
+         filePath.substring(tablePath.length() + 1, partLength);
+    }
+    return partitionDir;
   }
 
   /**