processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
/**
   * This API will write the load level metadata for the loadmanagement module inorder to
   * manage the load and query execution management smoothly.
   *
   * @param newMetaEntry
   * @param loadModel
   * @param uuid
   * @return boolean which determines whether status update is done or not.
   * @throws IOException
   */
This API will write the load level metadata for the loadmanagement module inorder to
manage the load and query execution management smoothly.

@param newMetaEntry
@param loadModel
@param uuid
@return boolean which determines whether status update is done or not.
This API will write the load level metadata for the loadmanagement module inorder to manage the load and query execution management smoothly . @param newMetaEntry @param loadModel @param uuid @return boolean which determines whether status update is done or not .
public static boolean recordNewLoadMetadata(LoadMetadataDetails newMetaEntry,
      CarbonLoadModel loadModel, boolean loadStartEntry, boolean insertOverwrite, String uuid)
      throws IOException {
    boolean status = false;
    AbsoluteTableIdentifier absoluteTableIdentifier =
        loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
    CarbonTablePath carbonTablePath = CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier);
    String metadataPath = carbonTablePath.getMetadataDirectoryPath();
    FileType fileType = FileFactory.getFileType(metadataPath);
    if (!FileFactory.isFileExist(metadataPath, fileType)) {
      FileFactory.mkdirs(metadataPath, fileType);
    }
    String tableStatusPath;
    if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isChildDataMap() && !uuid.isEmpty()) {
      tableStatusPath = carbonTablePath.getTableStatusFilePathWithUUID(uuid);
    } else {
      tableStatusPath = carbonTablePath.getTableStatusFilePath();
    }
    SegmentStatusManager segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier);
    ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
    int retryCount = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
            CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT);
    int maxTimeout = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
            CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT);
    try {
      if (carbonLock.lockWithRetries(retryCount, maxTimeout)) {
        LOGGER.info(
            "Acquired lock for table" + loadModel.getDatabaseName() + "." + loadModel.getTableName()
                + " for table status updation");
        LoadMetadataDetails[] listOfLoadFolderDetailsArray =
            SegmentStatusManager.readLoadMetadata(carbonTablePath.getMetadataDirectoryPath());
        List<LoadMetadataDetails> listOfLoadFolderDetails =
            new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
        List<CarbonFile> staleFolders = new ArrayList<>();
        Collections.addAll(listOfLoadFolderDetails, listOfLoadFolderDetailsArray);
        if (loadStartEntry) {
          String segmentId =
              String.valueOf(SegmentStatusManager.createNewSegmentId(listOfLoadFolderDetailsArray));
          loadModel.setLoadMetadataDetails(listOfLoadFolderDetails);
          if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isChildDataMap() && !loadModel
              .getSegmentId().isEmpty()) {
            newMetaEntry.setLoadName(loadModel.getSegmentId());
          } else {
            newMetaEntry.setLoadName(segmentId);
            loadModel.setSegmentId(segmentId);
          }
          for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
            if (entry.getSegmentStatus() == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS
                && SegmentStatusManager.isLoadInProgress(
                    absoluteTableIdentifier, entry.getLoadName())) {
              throw new RuntimeException("Already insert overwrite is in progress");
            } else if (newMetaEntry.getSegmentStatus() == SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS
                && entry.getSegmentStatus() == SegmentStatus.INSERT_IN_PROGRESS
                && SegmentStatusManager.isLoadInProgress(
                    absoluteTableIdentifier, entry.getLoadName())) {
              throw new RuntimeException("Already insert into or load is in progress");
            }
          }
          listOfLoadFolderDetails.add(newMetaEntry);
        } else {
          newMetaEntry.setLoadName(String.valueOf(loadModel.getSegmentId()));
          int indexToOverwriteNewMetaEntry = 0;
          for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
            if (entry.getLoadName().equals(newMetaEntry.getLoadName())
                && entry.getLoadStartTime() == newMetaEntry.getLoadStartTime()) {
              break;
            }
            indexToOverwriteNewMetaEntry++;
          }
          if (insertOverwrite) {
            for (LoadMetadataDetails entry : listOfLoadFolderDetails) {
              if (entry.getSegmentStatus() != SegmentStatus.INSERT_OVERWRITE_IN_PROGRESS) {
                entry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
                addToStaleFolders(carbonTablePath, staleFolders, entry);
              }
            }
          }
          listOfLoadFolderDetails.set(indexToOverwriteNewMetaEntry, newMetaEntry);
        }
        if (newMetaEntry.getSegmentStatus() == SegmentStatus.MARKED_FOR_DELETE) {
          addToStaleFolders(carbonTablePath, staleFolders, newMetaEntry);
        }
        SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetails
            .toArray(new LoadMetadataDetails[listOfLoadFolderDetails.size()]));
        for (CarbonFile staleFolder : staleFolders) {
          try {
            CarbonUtil.deleteFoldersAndFiles(staleFolder);
          } catch (IOException | InterruptedException e) {
            LOGGER.error("Failed to delete stale folder: " + e.getMessage());
          }
        }
        status = true;
      } else {
        LOGGER.error("Not able to acquire the lock for Table status updation for table " + loadModel
            .getDatabaseName() + "." + loadModel.getTableName());
      };
    } finally {
      if (carbonLock.unlock()) {
        LOGGER.info(
            "Table unlocked successfully after table status updation" + loadModel.getDatabaseName()
                + "." + loadModel.getTableName());
      } else {
        LOGGER.error(
            "Unable to unlock Table lock for table" + loadModel.getDatabaseName() + "." + loadModel
                .getTableName() + " during table status updation");
      }
    }
    return status;
  }
public static boolean recordNewLoadMetadata ( LoadMetadataDetails newMetaEntry , CarbonLoadModel loadModel , boolean loadStartEntry , boolean insertOverwrite , String uuid ) throws IOException { boolean status = false ; AbsoluteTableIdentifier absoluteTableIdentifier = loadModel . getCarbonDataLoadSchema ( ) . getCarbonTable ( ) . getAbsoluteTableIdentifier ( ) ; CarbonTablePath carbonTablePath = CarbonStorePath . getCarbonTablePath ( absoluteTableIdentifier ) ; String metadataPath = carbonTablePath . getMetadataDirectoryPath ( ) ; FileType fileType = FileFactory . getFileType ( metadataPath ) ; if ( ! FileFactory . isFileExist ( metadataPath , fileType ) ) { FileFactory . mkdirs ( metadataPath , fileType ) ; } String tableStatusPath ; if ( loadModel . getCarbonDataLoadSchema ( ) . getCarbonTable ( ) . isChildDataMap ( ) && ! uuid . isEmpty ( ) ) { tableStatusPath = carbonTablePath . getTableStatusFilePathWithUUID ( uuid ) ; } else { tableStatusPath = carbonTablePath . getTableStatusFilePath ( ) ; } SegmentStatusManager segmentStatusManager = new SegmentStatusManager ( absoluteTableIdentifier ) ; ICarbonLock carbonLock = segmentStatusManager . getTableStatusLock ( ) ; int retryCount = CarbonLockUtil . getLockProperty ( CarbonCommonConstants . NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK , CarbonCommonConstants . NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT ) ; int maxTimeout = CarbonLockUtil . getLockProperty ( CarbonCommonConstants . MAX_TIMEOUT_FOR_CONCURRENT_LOCK , CarbonCommonConstants . MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT ) ; try { if ( carbonLock . lockWithRetries ( retryCount , maxTimeout ) ) { LOGGER . info ( "Acquired lock for table" + loadModel . getDatabaseName ( ) + "." + loadModel . getTableName ( ) + " for table status updation" ) ; LoadMetadataDetails [ ] listOfLoadFolderDetailsArray = SegmentStatusManager . readLoadMetadata ( carbonTablePath . getMetadataDirectoryPath ( ) ) ; List < LoadMetadataDetails > listOfLoadFolderDetails = new ArrayList < > ( CarbonCommonConstants . DEFAULT_COLLECTION_SIZE ) ; List < CarbonFile > staleFolders = new ArrayList < > ( ) ; Collections . addAll ( listOfLoadFolderDetails , listOfLoadFolderDetailsArray ) ; if ( loadStartEntry ) { String segmentId = String . valueOf ( SegmentStatusManager . createNewSegmentId ( listOfLoadFolderDetailsArray ) ) ; loadModel . setLoadMetadataDetails ( listOfLoadFolderDetails ) ; if ( loadModel . getCarbonDataLoadSchema ( ) . getCarbonTable ( ) . isChildDataMap ( ) && ! loadModel . getSegmentId ( ) . isEmpty ( ) ) { newMetaEntry . setLoadName ( loadModel . getSegmentId ( ) ) ; } else { newMetaEntry . setLoadName ( segmentId ) ; loadModel . setSegmentId ( segmentId ) ; } for ( LoadMetadataDetails entry : listOfLoadFolderDetails ) { if ( entry . getSegmentStatus ( ) == SegmentStatus . INSERT_OVERWRITE_IN_PROGRESS && SegmentStatusManager . isLoadInProgress ( absoluteTableIdentifier , entry . getLoadName ( ) ) ) { throw new RuntimeException ( "Already insert overwrite is in progress" ) ; } else if ( newMetaEntry . getSegmentStatus ( ) == SegmentStatus . INSERT_OVERWRITE_IN_PROGRESS && entry . getSegmentStatus ( ) == SegmentStatus . INSERT_IN_PROGRESS && SegmentStatusManager . isLoadInProgress ( absoluteTableIdentifier , entry . getLoadName ( ) ) ) { throw new RuntimeException ( "Already insert into or load is in progress" ) ; } } listOfLoadFolderDetails . add ( newMetaEntry ) ; } else { newMetaEntry . setLoadName ( String . valueOf ( loadModel . getSegmentId ( ) ) ) ; int indexToOverwriteNewMetaEntry = 0 ; for ( LoadMetadataDetails entry : listOfLoadFolderDetails ) { if ( entry . getLoadName ( ) . equals ( newMetaEntry . getLoadName ( ) ) && entry . getLoadStartTime ( ) == newMetaEntry . getLoadStartTime ( ) ) { break ; } indexToOverwriteNewMetaEntry ++ ; } if ( insertOverwrite ) { for ( LoadMetadataDetails entry : listOfLoadFolderDetails ) { if ( entry . getSegmentStatus ( ) != SegmentStatus . INSERT_OVERWRITE_IN_PROGRESS ) { entry . setSegmentStatus ( SegmentStatus . MARKED_FOR_DELETE ) ; addToStaleFolders ( carbonTablePath , staleFolders , entry ) ; } } } listOfLoadFolderDetails . set ( indexToOverwriteNewMetaEntry , newMetaEntry ) ; } if ( newMetaEntry . getSegmentStatus ( ) == SegmentStatus . MARKED_FOR_DELETE ) { addToStaleFolders ( carbonTablePath , staleFolders , newMetaEntry ) ; } SegmentStatusManager . writeLoadDetailsIntoFile ( tableStatusPath , listOfLoadFolderDetails . toArray ( new LoadMetadataDetails [ listOfLoadFolderDetails . size ( ) ] ) ) ; for ( CarbonFile staleFolder : staleFolders ) { try { CarbonUtil . deleteFoldersAndFiles ( staleFolder ) ; } catch ( IOException | InterruptedException e ) { LOGGER . error ( "Failed to delete stale folder: " + e . getMessage ( ) ) ; } } status = true ; } else { LOGGER . error ( "Not able to acquire the lock for Table status updation for table " + loadModel . getDatabaseName ( ) + "." + loadModel . getTableName ( ) ) ; } ; } finally { if ( carbonLock . unlock ( ) ) { LOGGER . info ( "Table unlocked successfully after table status updation" + loadModel . getDatabaseName ( ) + "." + loadModel . getTableName ( ) ) ; } else { LOGGER . error ( "Unable to unlock Table lock for table" + loadModel . getDatabaseName ( ) + "." + loadModel . getTableName ( ) + " during table status updation" ) ; } } return status ; }
This API will write the load level metadata for the loadmanagement module inorder to
 manage the load and query execution management smoothly.
This API will write the load level metadata for the loadmanagement module inorder to manage the load and query execution management smoothly .
// create a new segment Id if load has just begun else add the already generated Id // Segment id would be provided in case this is compaction flow for aggregate data map. // If that is true then used the segment id as the load name. // Exception should be thrown if: // 1. If insert overwrite is in progress and any other load or insert operation // is triggered // 2. If load or insert into operation is in progress and insert overwrite operation // is triggered // existing entry needs to be overwritten as the entry will exist with some // intermediate status // For insert overwrite, we will delete the old segment folder immediately // So collect the old segments here // when no records are inserted then newSegmentEntry will be SegmentStatus.MARKED_FOR_DELETE // so empty segment folder should be deleted // Delete all old stale segment folders // try block is inside for loop because even if there is failure in deletion of 1 stale // folder still remaining stale folders should be deleted