[CARBONDATA-1731,CARBONDATA-1728] Update,Delete fails incorrectly with error

[QueryExecution] Fetch BlockletId in Executor. Currently the blockletId are not propagated to the excutrs properly. Due to this reason the Implicit columns i.e. tupleId formed in executers can wrongly form duplicate tupleId. For e.g. a Block having two blocklets i.e. 0 and 1, in executors each tasks picks each blocklets. As the ID is not propagated each executor will term the Blocklets as ID 0. Solution is to propagate the Blocklet IDs to the executor from driver

This closes apache#1719

Author: anubhav100

core/src/main/java/org/apache/carbondata/core/datastore/DataRefNode.java

@@ -51,6 +51,12 @@ public interface DataRefNode {
    */
   long nodeNumber();
 
+  /**
+   * Method is used for retreiving the BlockletId.
+   * @return the blockletid related to the data block.
+   */
+  String blockletId();
+
   /**
    * This method will be used to get the max value of all the columns this can
    * be used in case of filter query

core/src/main/java/org/apache/carbondata/core/datastore/block/TableBlockInfo.java

@@ -64,6 +64,11 @@ public class TableBlockInfo implements Distributable, Serializable {
    */
   private String segmentId;
 
+  /**
+   * id of the Blocklet.
+   */
+  private String blockletId;
+
   private String[] locations;
 
   private ColumnarFormatVersion version;
@@ -76,7 +81,7 @@ public class TableBlockInfo implements Distributable, Serializable {
    * map of block location and storage id
    */
   private Map<String, String> blockStorageIdMap =
-          new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+      new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
 
   /**
    * delete delta files path for this block
@@ -85,9 +90,11 @@ public class TableBlockInfo implements Distributable, Serializable {
 
   private BlockletDetailInfo detailInfo;
 
-  public TableBlockInfo(String filePath, long blockOffset, String segmentId, String[] locations,
-      long blockLength, ColumnarFormatVersion version, String[] deletedDeltaFilePath) {
+  public TableBlockInfo(String filePath, long blockOffset, String segmentId,
+      String[] locations, long blockLength, ColumnarFormatVersion version,
+      String[] deletedDeltaFilePath) {
     this.filePath = FileFactory.getUpdatedFilePath(filePath);
+    this.blockletId = "0";
     this.blockOffset = blockOffset;
     this.segmentId = segmentId;
     this.locations = locations;
@@ -113,10 +120,29 @@ public TableBlockInfo() {
   public TableBlockInfo(String filePath, long blockOffset, String segmentId, String[] locations,
       long blockLength, BlockletInfos blockletInfos, ColumnarFormatVersion version,
       String[] deletedDeltaFilePath) {
-    this(filePath, blockOffset, segmentId, locations, blockLength, version, deletedDeltaFilePath);
+    this(filePath, blockOffset, segmentId, locations, blockLength, version,
+        deletedDeltaFilePath);
     this.blockletInfos = blockletInfos;
   }
 
+  /**
+   * constructor to initialize the TableBlockInfo with blockletIds
+   *
+   * @param filePath
+   * @param blockOffset
+   * @param segmentId
+   * @param locations
+   * @param blockLength
+   * @param blockletInfos
+   */
+  public TableBlockInfo(String filePath, String blockletId, long blockOffset, String segmentId,
+      String[] locations, long blockLength, BlockletInfos blockletInfos,
+      ColumnarFormatVersion version, String[] deletedDeltaFilePath) {
+    this(filePath, blockOffset, segmentId, locations, blockLength, blockletInfos, version,
+        deletedDeltaFilePath);
+    this.blockletId = blockletId;
+  }
+
   /**
    * constructor to initialize the TableBlockInfo with blockStorageIdMap
    *
@@ -129,11 +155,12 @@ public TableBlockInfo(String filePath, long blockOffset, String segmentId, Strin
    * @param version
    * @param blockStorageIdMap
    */
-  public TableBlockInfo(String filePath, long blockOffset, String segmentId, String[] locations,
-      long blockLength, BlockletInfos blockletInfos, ColumnarFormatVersion version,
-      Map<String, String> blockStorageIdMap, String[] deletedDeltaFilePath) {
-    this(filePath, blockOffset, segmentId, locations, blockLength, blockletInfos, version,
-        deletedDeltaFilePath);
+  public TableBlockInfo(String filePath, String blockletId, long blockOffset, String segmentId,
+      String[] locations, long blockLength, BlockletInfos blockletInfos,
+      ColumnarFormatVersion version, Map<String, String> blockStorageIdMap,
+      String[] deletedDeltaFilePath) {
+    this(filePath, blockletId, blockOffset, segmentId, locations, blockLength, blockletInfos,
+        version, deletedDeltaFilePath);
     this.blockStorageIdMap = blockStorageIdMap;
   }
 
@@ -356,4 +383,12 @@ public BlockletDetailInfo getDetailInfo() {
   public void setDetailInfo(BlockletDetailInfo detailInfo) {
     this.detailInfo = detailInfo;
   }
+
+  public String getBlockletId() {
+    return blockletId;
+  }
+
+  public void setBlockletId(String blockletId) {
+    this.blockletId = blockletId;
+  }
 }

core/src/main/java/org/apache/carbondata/core/datastore/impl/btree/BTreeNonLeafNode.java

@@ -135,6 +135,10 @@ public BTreeNonLeafNode() {
     throw new UnsupportedOperationException("Unsupported operation");
   }
 
+  @Override public String blockletId() {
+    throw new UnsupportedOperationException("Unsupported operation");
+  }
+
   /**
    * This method will be used to get the max value of all the columns this can
    * be used in case of filter query

core/src/main/java/org/apache/carbondata/core/datastore/impl/btree/BlockBTreeLeafNode.java

@@ -59,6 +59,14 @@ public TableBlockInfo getTableBlockInfo() {
     return blockInfo.getTableBlockInfo();
   }
 
+  /**
+   * Below method is suppose to return the Blocklet ID.
+   * @return
+   */
+  @Override public String blockletId() {
+    return blockInfo.getTableBlockInfo().getDetailInfo().getBlockletId().toString();
+  }
+
   /**
    * number of pages in blocklet
    * @return

core/src/main/java/org/apache/carbondata/core/datastore/impl/btree/BlockletBTreeLeafNode.java

@@ -98,6 +98,10 @@ public BlockletBTreeLeafNode(BTreeBuilderInfo builderInfos, int leafIndex, long
     }
   }
 
+  @Override public String blockletId() {
+    return "0";
+  }
+
   /**
    * Below method will be used to get the dimension chunks
    *

core/src/main/java/org/apache/carbondata/core/indexstore/BlockletDetailInfo.java

@@ -36,6 +36,8 @@ public class BlockletDetailInfo implements Serializable, Writable {
 
   private short versionNumber;
 
+  private short blockletId;
+
   private int[] dimLens;
 
   private long schemaUpdatedTimeStamp;
@@ -94,6 +96,7 @@ public void setSchemaUpdatedTimeStamp(long schemaUpdatedTimeStamp) {
     out.writeInt(rowCount);
     out.writeShort(pagesCount);
     out.writeShort(versionNumber);
+    out.writeShort(blockletId);
     out.writeShort(dimLens.length);
     for (int i = 0; i < dimLens.length; i++) {
       out.writeInt(dimLens[i]);
@@ -106,6 +109,7 @@ public void setSchemaUpdatedTimeStamp(long schemaUpdatedTimeStamp) {
     rowCount = in.readInt();
     pagesCount = in.readShort();
     versionNumber = in.readShort();
+    blockletId = in.readShort();
     dimLens = new int[in.readShort()];
     for (int i = 0; i < dimLens.length; i++) {
       dimLens[i] = in.readInt();
@@ -114,4 +118,12 @@ public void setSchemaUpdatedTimeStamp(long schemaUpdatedTimeStamp) {
     blockletInfo = new BlockletInfo();
     blockletInfo.readFields(in);
   }
+
+  public Short getBlockletId() {
+    return blockletId;
+  }
+
+  public void setBlockletId(Short blockletId) {
+    this.blockletId = blockletId;
+  }
 }

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

@@ -599,6 +599,7 @@ private ExtendedBlocklet createBlocklet(DataMapRow row, int blockletId) {
     detailInfo.setRowCount(row.getInt(ROW_COUNT_INDEX));
     detailInfo.setPagesCount(row.getShort(PAGE_COUNT_INDEX));
     detailInfo.setVersionNumber(row.getShort(VERSION_INDEX));
+    detailInfo.setBlockletId((short) blockletId);
     detailInfo.setDimLens(columnCardinality);
     detailInfo.setSchemaUpdatedTimeStamp(row.getLong(SCHEMA_UPADATED_TIME_INDEX));
     BlockletInfo blockletInfo = new BlockletInfo();

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

@@ -53,6 +53,7 @@ public BlockletDataRefNodeWrapper(List<TableBlockInfo> blockInfos, int index,
       BlockletDetailInfo detailInfo = blockInfo.getDetailInfo();
       detailInfo.getBlockletInfo().setNumberOfRows(detailInfo.getRowCount());
       detailInfo.getBlockletInfo().setNumberOfPages(detailInfo.getPagesCount());
+      detailInfo.setBlockletId(blockInfo.getDetailInfo().getBlockletId());
       int[] pageRowCount = new int[detailInfo.getPagesCount()];
       int numberOfPagesCompletelyFilled = detailInfo.getRowCount()
           / CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT;
@@ -86,6 +87,10 @@ public BlockletDataRefNodeWrapper(List<TableBlockInfo> blockInfos, int index,
     return index;
   }
 
+  @Override public String blockletId() {
+    return blockInfos.get(index).getDetailInfo().getBlockletId().toString();
+  }
+
   @Override public byte[][] getColumnsMaxValue() {
     return null;
   }

core/src/main/java/org/apache/carbondata/core/scan/scanner/AbstractBlockletScanner.java

@@ -76,7 +76,7 @@ public AbstractBlockletScanner(BlockExecutionInfo tableBlockExecutionInfos) {
         totalPagesScanned.getCount() + blocksChunkHolder.getDataBlock().numberOfPages());
     scannedResult.setBlockletId(
         blockExecutionInfo.getBlockId() + CarbonCommonConstants.FILE_SEPARATOR + blocksChunkHolder
-            .getDataBlock().nodeNumber());
+            .getDataBlock().blockletId());
     DimensionRawColumnChunk[] dimensionRawColumnChunks =
         blocksChunkHolder.getDimensionRawDataChunk();
     DimensionColumnDataChunk[][] dimensionColumnDataChunks =

core/src/main/java/org/apache/carbondata/core/scan/scanner/impl/FilterScanner.java

@@ -175,7 +175,7 @@ private AbstractScannedResult fillScannedResult(BlocksChunkHolder blocksChunkHol
     AbstractScannedResult scannedResult = new FilterQueryScannedResult(blockExecutionInfo);
     scannedResult.setBlockletId(
         blockExecutionInfo.getBlockId() + CarbonCommonConstants.FILE_SEPARATOR + blocksChunkHolder
-            .getDataBlock().nodeNumber());
+            .getDataBlock().blockletId());
     // valid scanned blocklet
     QueryStatistic validScannedBlockletStatistic = queryStatisticsModel.getStatisticsTypeAndObjMap()
         .get(QueryStatisticsConstants.VALID_SCAN_BLOCKLET_NUM);

hadoop/src/main/java/org/apache/carbondata/hadoop/CarbonInputFormat.java

@@ -409,7 +409,7 @@ private List<InputSplit> getSplitsInternal(JobContext job) throws IOException {
       if (segmentId.equals(CarbonCommonConstants.INVALID_SEGMENT_ID)) {
         continue;
       }
-      carbonSplits.add(CarbonInputSplit.from(segmentId, fileSplit,
+      carbonSplits.add(CarbonInputSplit.from(segmentId, "0", fileSplit,
           ColumnarFormatVersion.valueOf(
               CarbonCommonConstants.CARBON_DATA_FILE_DEFAULT_VERSION)));
     }
@@ -452,9 +452,9 @@ private List<InputSplit> getSplits(JobContext job, FilterResolverIntf filterReso
     Boolean  isIUDTable = false;
 
     AbsoluteTableIdentifier absoluteTableIdentifier =
-            getOrCreateCarbonTable(job.getConfiguration()).getAbsoluteTableIdentifier();
+        getOrCreateCarbonTable(job.getConfiguration()).getAbsoluteTableIdentifier();
     SegmentUpdateStatusManager updateStatusManager =
-            new SegmentUpdateStatusManager(absoluteTableIdentifier);
+        new SegmentUpdateStatusManager(absoluteTableIdentifier);
 
     isIUDTable = (updateStatusManager.getUpdateStatusDetails().length != 0);
 
@@ -476,22 +476,23 @@ private List<InputSplit> getSplits(JobContext job, FilterResolverIntf filterReso
           // In case IUD is not performed in this table avoid searching for
           // invalidated blocks.
           if (CarbonUtil
-                  .isInvalidTableBlock(tableBlockInfo.getSegmentId(), tableBlockInfo.getFilePath(),
-                          invalidBlockVOForSegmentId, updateStatusManager)) {
+              .isInvalidTableBlock(tableBlockInfo.getSegmentId(), tableBlockInfo.getFilePath(),
+                  invalidBlockVOForSegmentId, updateStatusManager)) {
             continue;
           }
           // When iud is done then only get delete delta files for a block
           try {
             deleteDeltaFilePath =
-                    updateStatusManager.getDeleteDeltaFilePath(tableBlockInfo.getFilePath());
+                updateStatusManager.getDeleteDeltaFilePath(tableBlockInfo.getFilePath());
           } catch (Exception e) {
             throw new IOException(e);
           }
         }
-        result.add(new CarbonInputSplit(segmentNo, new Path(tableBlockInfo.getFilePath()),
-            tableBlockInfo.getBlockOffset(), tableBlockInfo.getBlockLength(),
-            tableBlockInfo.getLocations(), tableBlockInfo.getBlockletInfos().getNoOfBlockLets(),
-            tableBlockInfo.getVersion(), deleteDeltaFilePath));
+        result.add(new CarbonInputSplit(segmentNo, tableBlockInfo.getBlockletId(),
+            new Path(tableBlockInfo.getFilePath()), tableBlockInfo.getBlockOffset(),
+            tableBlockInfo.getBlockLength(), tableBlockInfo.getLocations(),
+            tableBlockInfo.getBlockletInfos().getNoOfBlockLets(), tableBlockInfo.getVersion(),
+            deleteDeltaFilePath));
       }
     }
     return result;
@@ -583,7 +584,7 @@ private List<TableBlockInfo> getTableBlockInfo(JobContext job,
       Set<SegmentTaskIndexStore.TaskBucketHolder> taskKeys, UpdateVO updateDetails,
       SegmentUpdateStatusManager updateStatusManager,
       String segmentId, Set<SegmentTaskIndexStore.TaskBucketHolder> validTaskKeys)
-    throws IOException {
+      throws IOException {
     List<TableBlockInfo> tableBlockInfoList = new ArrayList<TableBlockInfo>();
 
     // get file location of all files of given segment
@@ -603,7 +604,8 @@ private List<TableBlockInfo> getTableBlockInfo(JobContext job,
         BlockletInfos blockletInfos = new BlockletInfos(carbonInputSplit.getNumberOfBlocklets(), 0,
             carbonInputSplit.getNumberOfBlocklets());
         tableBlockInfoList.add(
-            new TableBlockInfo(carbonInputSplit.getPath().toString(), carbonInputSplit.getStart(),
+            new TableBlockInfo(carbonInputSplit.getPath().toString(),
+                carbonInputSplit.getBlockletId(), carbonInputSplit.getStart(),
                 tableSegmentUniqueIdentifier.getSegmentId(), carbonInputSplit.getLocations(),
                 carbonInputSplit.getLength(), blockletInfos, carbonInputSplit.getVersion(),
                 carbonInputSplit.getBlockStorageIdMap(), carbonInputSplit.getDeleteDeltaFiles()));
@@ -613,9 +615,9 @@ private List<TableBlockInfo> getTableBlockInfo(JobContext job,
   }
 
   private boolean isValidBlockBasedOnUpdateDetails(
-          Set<SegmentTaskIndexStore.TaskBucketHolder> taskKeys, CarbonInputSplit carbonInputSplit,
-          UpdateVO updateDetails, SegmentUpdateStatusManager updateStatusManager, String segmentId,
-          Set<SegmentTaskIndexStore.TaskBucketHolder> validTaskKeys) {
+      Set<SegmentTaskIndexStore.TaskBucketHolder> taskKeys, CarbonInputSplit carbonInputSplit,
+      UpdateVO updateDetails, SegmentUpdateStatusManager updateStatusManager, String segmentId,
+      Set<SegmentTaskIndexStore.TaskBucketHolder> validTaskKeys) {
     String taskID = null;
     if (null != carbonInputSplit) {
       if (!updateStatusManager.isBlockValid(segmentId, carbonInputSplit.getPath().getName())) {
@@ -741,7 +743,7 @@ public BlockMappingVO getBlockRowCount(JobContext job,
                 updateStatusManager);
         for (Map.Entry<SegmentTaskIndexStore.TaskBucketHolder, AbstractIndex> taskMap :
             taskAbstractIndexMap
-            .entrySet()) {
+                .entrySet()) {
           AbstractIndex taskAbstractIndex = taskMap.getValue();
           countOfBlocksInSeg += new BlockLevelTraverser()
               .getBlockRowMapping(taskAbstractIndex, blockRowCountMapping, eachValidSeg,