[CARBONDATA-4063] Refactor getBlockId and getShortBlcokId functions

Why is this PR needed?
Currently, getBlockId and getShortBlockId functions are too complex and unreadable, which can be more simpler and readable.

What changes were proposed in this PR?
Refactor the getBlockId and getShortBlockId functions without changing the blockid format and shortblockid format, There is no complatiability issue after the rafactoring.

Does this PR introduce any user interface change?
No

Is any new testcase added?
No

Author: marchpure

core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockIndex.java

@@ -63,7 +63,6 @@
 import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
 import org.apache.carbondata.core.util.BlockletIndexUtil;
 import org.apache.carbondata.core.util.ByteUtil;
-import org.apache.carbondata.core.util.CarbonUtil;
 import org.apache.carbondata.core.util.DataFileFooterConverter;
 import org.apache.carbondata.core.util.path.CarbonTablePath;
 
@@ -797,8 +796,8 @@ private boolean addBlockBasedOnMinMaxValue(FilterExecutor filterExecutor, byte[]
         // replaced with '#', to support multi level partitioning. For example, BlockId will be
         // look like `part1=1#part2=2/xxxxxxxxx`. During query also, blockId should be
         // replaced by '#' in place of '/', to match and prune data on SI table.
-        uniqueBlockPath = CarbonUtil
-            .getBlockId(carbonTable.getAbsoluteTableIdentifier(), filePath, "", true, false, true);
+        uniqueBlockPath = CarbonTablePath
+            .getBlockId(carbonTable.getAbsoluteTableIdentifier(), filePath, "", true);
       } else {
         uniqueBlockPath = filePath.substring(filePath.lastIndexOf("/Part") + 1);
       }

core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java

@@ -822,9 +822,7 @@ public static void createBlockDetailsMap(BlockMappingVO blockMappingVO,
       SegmentUpdateDetails detail = segmentUpdateStatusManager.getDetailsForABlock(key);
 
       if (null != detail) {
-
         alreadyDeletedCount = Long.parseLong(detail.getDeletedRowsInBlock());
-
       }
 
       RowCountDetailsVO rowCountDetailsVO =
@@ -845,16 +843,7 @@ public static void createBlockDetailsMap(BlockMappingVO blockMappingVO,
    */
   public static String getSegmentBlockNameKey(String segID, String blockName,
       boolean isPartitionTable) {
-    String blockNameWithOutPartAndBatchNo = blockName
-        .substring(blockName.indexOf(CarbonCommonConstants.HYPHEN) + 1,
-            blockName.lastIndexOf(CarbonTablePath.getCarbonDataExtension()))
-        .replace(CarbonTablePath.BATCH_PREFIX, CarbonCommonConstants.UNDERSCORE);
-    // to remove compressor name
-    int index = blockNameWithOutPartAndBatchNo.lastIndexOf(CarbonCommonConstants.POINT);
-    if (index != -1) {
-      blockNameWithOutPartAndBatchNo = blockNameWithOutPartAndBatchNo
-          .replace(blockNameWithOutPartAndBatchNo.substring(index), "");
-    }
+    String blockNameWithOutPartAndBatchNo = CarbonTablePath.getShortBlockId(blockName);
     if (isPartitionTable) {
       return blockNameWithOutPartAndBatchNo;
     } else {

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

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

core/src/main/java/org/apache/carbondata/core/scan/expression/conditional/ImplicitExpression.java

@@ -17,7 +17,6 @@
 
 package org.apache.carbondata.core.scan.expression.conditional;
 
-import java.io.File;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -60,7 +59,8 @@ public ImplicitExpression(Map<String, Set<Integer>> blockIdToBlockletIdMapping)
   }
 
   private void addBlockEntry(String blockletPath) {
-    String blockId = blockletPath.substring(0, blockletPath.lastIndexOf(File.separator));
+    String blockId = blockletPath.substring(0,
+        blockletPath.lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR));
     // Check if blockId contains old tuple id format, and convert it to compatible format.
     // Example for non-partition table:
     //    Old tuple id format: 0/0/0-0_batchno0-0-0-1599806689305.snappy
@@ -71,9 +71,9 @@ private void addBlockEntry(String blockletPath) {
     if (blockId.contains(CarbonTablePath.BATCH_PREFIX)) {
       blockId = CarbonTablePath.getShortBlockId(blockId);
       // In case of non-partition table, remove index of part prefix (Part0) from blockId.
-      if (!blockId.substring(0, blockId.indexOf(File.separator))
+      if (!blockId.substring(0, blockId.indexOf(CarbonCommonConstants.FILE_SEPARATOR))
           .contains(CarbonCommonConstants.EQUALS)) {
-        blockId = blockId.substring(blockId.indexOf(File.separator) + 1);
+        blockId = blockId.substring(blockId.indexOf(CarbonCommonConstants.FILE_SEPARATOR) + 1);
       }
     }
     Set<Integer> blockletIds = blockIdToBlockletIdMapping.get(blockId);
@@ -82,7 +82,8 @@ private void addBlockEntry(String blockletPath) {
       blockIdToBlockletIdMapping.put(blockId, blockletIds);
     }
     blockletIds.add(
-        Integer.parseInt(blockletPath.substring(blockletPath.lastIndexOf(File.separator) + 1)));
+        Integer.parseInt(blockletPath.substring(
+            blockletPath.lastIndexOf(CarbonCommonConstants.FILE_SEPARATOR) + 1)));
   }
 
   @Override

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

@@ -2737,72 +2737,6 @@ private static long getMaxOfBlockAndFileSize(long blockSize, long fileSize) {
     return maxSize;
   }
 
-  /**
-   * Generate the blockId as per the block path
-   *
-   * @param identifier
-   * @param filePath
-   * @param segmentId
-   * @param isTransactionalTable
-   * @param isStandardTable
-   * @return
-   */
-  public static String getBlockId(AbsoluteTableIdentifier identifier, String filePath,
-      String segmentId, boolean isTransactionalTable, boolean isStandardTable) {
-    return getBlockId(identifier, filePath, segmentId, isTransactionalTable, isStandardTable,
-        false);
-  }
-
-  /**
-   * Generate the blockId as per the block path
-   *
-   * @return
-   */
-  public static String getBlockId(AbsoluteTableIdentifier identifier, String filePath,
-      String segmentId, boolean isTransactionalTable, boolean isStandardTable,
-      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;
-          }
-
-        }
-      }
-    } else {
-      blockId = filePath.substring(0, filePath.length() - blockName.length()).replace("/", "#")
-          + CarbonCommonConstants.FILE_SEPARATOR + "Segment_" + segmentId
-          + CarbonCommonConstants.FILE_SEPARATOR + blockName;
-    }
-    return blockId;
-  }
 
   /**
    * sets the local dictionary columns to wrapper schema, if the table property

core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java

@@ -25,6 +25,7 @@
 import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.locks.LockUsage;
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
 import org.apache.carbondata.core.metadata.ColumnarFormatVersion;
 
 import static org.apache.carbondata.core.constants.CarbonCommonConstants.DASH;
@@ -215,38 +216,6 @@ public static String getTableStatusFilePathWithUUID(String tablePath, String uui
     }
   }
 
-  /**
-   * Below method will be used to get the index file present in the segment folder
-   * based on task id
-   *
-   * @param taskId      task id of the file
-   * @param segmentId   segment number
-   * @return full qualified carbon index path
-   */
-  private static String getCarbonIndexFilePath(final String tablePath, final String taskId,
-      final String segmentId, final String bucketNumber) {
-    String segmentDir = getSegmentPath(tablePath, segmentId);
-    CarbonFile carbonFile =
-        FileFactory.getCarbonFile(segmentDir);
-
-    CarbonFile[] files = carbonFile.listFiles(new CarbonFileFilter() {
-      @Override
-      public boolean accept(CarbonFile file) {
-        if (bucketNumber.equals("-1")) {
-          return file.getName().startsWith(taskId) && file.getName().endsWith(INDEX_FILE_EXT);
-        }
-        return file.getName().startsWith(taskId + DASH + bucketNumber) && file.getName()
-            .endsWith(INDEX_FILE_EXT);
-      }
-    });
-    if (files.length > 0) {
-      return files[0].getAbsolutePath();
-    } else {
-      throw new RuntimeException("Missing Carbon index file for Segment[" + segmentId + "], "
-          + "taskId[" + taskId + "]");
-    }
-  }
-
   /**
    * Below method will be used to get the carbon index file path
    * @param taskId
@@ -635,42 +604,86 @@ public static String getCarbonMergeIndexExtension() {
     return MERGE_INDEX_FILE_EXT;
   }
 
+  /**
+   * Generate the blockId as per the block path
+   *
+   * @param identifier
+   * @param filePath
+   * @param segmentId
+   * @param isPartitionTable
+   * @return blockid, which is the identify of a block
+   */
+  public static String getBlockId(AbsoluteTableIdentifier identifier, String filePath,
+      String segmentId, boolean isPartitionTable) {
+    String blockName = filePath.substring(filePath.lastIndexOf(
+        CarbonCommonConstants.FILE_SEPARATOR) + 1);
+    String tablePath = identifier.getTablePath();
+
+    String partitionDir = "";
+    // 1. For block of Added Segments, The BlockId consistsof
+    //    <partitionPath><segmentId><blockName>
+    if (!filePath.startsWith(tablePath)) {
+      partitionDir = getPartitionDir(tablePath, filePath, blockName);
+      return partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
+          + CarbonCommonConstants.FILE_SEPARATOR + segmentId
+          + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+    }
+
+    // 2. For block of partitiontable, The BlockId consistsof
+    //    <partitionPath><blockName>
+    if (isPartitionTable) {
+      partitionDir = getPartitionDir(tablePath, filePath, blockName);
+      return partitionDir.replace(CarbonCommonConstants.FILE_SEPARATOR, "#")
+          + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+    }
+
+    // 3. For nonpartitiontable, The BlockId consistsof
+    //    <segmentId><blockName>
+    return segmentId + CarbonCommonConstants.FILE_SEPARATOR + blockName;
+  }
+
   /**
    * This method will remove strings in path and return short block id
    *
    * @param blockId
    * @return shortBlockId
    */
   public static String getShortBlockId(String blockId) {
-    String blockIdWithCompressorName =
-        blockId.replace(PARTITION_PREFIX + "0" + CarbonCommonConstants.FILE_SEPARATOR, "")
-            .replace(SEGMENT_PREFIX, "").replace(BATCH_PREFIX, CarbonCommonConstants.UNDERSCORE)
-            .replace(DATA_PART_PREFIX, "").replace(CARBON_DATA_EXT, "");
-    // to remove compressor name
-    if (!blockId.equalsIgnoreCase(blockIdWithCompressorName)) {
-      int index = blockIdWithCompressorName.lastIndexOf(POINT);
-      int fileSeperatorIndex = blockIdWithCompressorName.lastIndexOf(File.separator);
-      if (index != -1) {
-        String modifiedBlockId;
-        if (index > fileSeperatorIndex) {
-          // Default case when path ends with compressor name.
-          // Example: 0/0-0_0-0-0-1600789595862.snappy
-          modifiedBlockId =
-              blockIdWithCompressorName.replace(blockIdWithCompressorName.substring(index), "");
-        } else {
-          // in case of CACHE_LEVEL = BLOCKLET, blockId path contains both block id and blocklet id
-          // so check for next file seperator and remove compressor name.
-          // Example: 0/0-0_0-0-0-1600789595862.snappy/0
-          modifiedBlockId = blockIdWithCompressorName
-              .replace(blockIdWithCompressorName.substring(index, fileSeperatorIndex), "");
-        }
-        return modifiedBlockId;
-      } else {
-        return blockIdWithCompressorName;
-      }
-    } else {
-      return blockIdWithCompressorName;
+    // 1. If the blockid is already shortblockid, return directly
+    int suffixIndex = blockId.lastIndexOf(CARBON_DATA_EXT);
+    if (suffixIndex < 0) {
+      return blockId;
+    }
+    // 2. get the filepath. in the type of part=a/part-0-0_batchno0-0-0-1597409791503.snappy
+    String filePath = blockId.substring(0, suffixIndex);
+
+    // 3. get the compressor name, in the type of '.snappy'/'.zstd'/'.gzip'
+    String compressorName = filePath.substring(filePath.lastIndexOf(POINT));
+
+    // 4. get rid of 'Part0/' 'Segment_' '_batchno' 'part-' '.carbondata' and compressorname
+    return blockId.replace(compressorName, "")
+        .replace(PARTITION_PREFIX + "0" + CarbonCommonConstants.FILE_SEPARATOR, "")
+        .replace(SEGMENT_PREFIX, "")
+        .replace(BATCH_PREFIX, CarbonCommonConstants.UNDERSCORE)
+        .replace(DATA_PART_PREFIX, "")
+        .replace(CARBON_DATA_EXT, "");
+  }
+
+  /**
+   * get the partition path in the block path
+   *
+   * @param tablePath
+   * @param filePath
+   * @param blockName
+   * @return blockid, which is the identify of a block
+   */
+  public static String getPartitionDir(String tablePath, String filePath, String blockName) {
+    // The filepath is consist with <tablePath><partitionPath><blockName>
+    // The partitionPath is the string truncated between tablePath and blockName
+    if (!filePath.startsWith(tablePath)) {
+      return filePath.substring(0, filePath.length() - blockName.length());
     }
+    return filePath.substring(tablePath.length() + 1, filePath.length() - blockName.length() - 1);
   }
 
   /**

integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala

@@ -288,21 +288,17 @@ class DeleteCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
 
     val carbonDataFilename = new File(carbonTable.getTablePath + "/Fact/Part0/Segment_0/")
       .listFiles().filter(fn => fn.getName.endsWith(".carbondata"))
-    val blockId = CarbonUtil.getBlockId(carbonTable.getAbsoluteTableIdentifier,
+    val blockId = CarbonTablePath.getBlockId(carbonTable.getAbsoluteTableIdentifier,
       carbonDataFilename(0).getAbsolutePath,
-      "0",
-      carbonTable.isTransactionalTable,
-      CarbonUtil.isStandardCarbonTable(carbonTable))
+      "0", false)
 
-    assert(blockId.startsWith("Part0/Segment_0/part-0-0_batchno0-0-0-"))
+    assert(blockId.startsWith("0/part-0-0_batchno0-0-0-"))
     val carbonDataFilename_part = new File(carbonTable_part.getTablePath + "/c3=aa").listFiles()
       .filter(fn => fn.getName.endsWith(".carbondata"))
-    val blockId_part = CarbonUtil.getBlockId(carbonTable.getAbsoluteTableIdentifier,
+    val blockId_part = CarbonTablePath.getBlockId(carbonTable.getAbsoluteTableIdentifier,
       carbonDataFilename_part(0).getAbsolutePath,
-      "0",
-      carbonTable.isTransactionalTable,
-      CarbonUtil.isStandardCarbonTable(carbonTable))
-    assert(blockId_part.startsWith("Part0/Segment_0/part-0-100100000100001_batchno0-0-0-"))
+      "0", false)
+    assert(blockId_part.startsWith("0/part-0-100100000100001_batchno0-0-0-"))
     val tableBlockPath = CarbonUpdateUtil
       .getTableBlockPath(listOfTupleId(0),
         carbonTable.getTablePath,

integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala

@@ -1024,7 +1024,7 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
   }
 
   test("check data after update with row.filter pushdown as false") {
-    sql("""drop table if exists iud.dest33_flat""")
+    sql("""drop table if exists iud.dest33_part""")
     sql(
       """create table iud.dest33_part (c1 int,c2 string, c3 short) STORED AS carbondata"""
         .stripMargin)