content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.carbondata.processing.store;
  
  import java.io.ByteArrayOutputStream;
  import java.io.DataOutputStream;
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.Semaphore;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
  import org.apache.carbondata.core.datastore.compression.SnappyCompressor;
  import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
  import org.apache.carbondata.core.datastore.row.CarbonRow;
  import org.apache.carbondata.core.datastore.row.WriteStepRowUtil;
  import org.apache.carbondata.core.keygenerator.KeyGenException;
  import org.apache.carbondata.core.metadata.ColumnarFormatVersion;
  import org.apache.carbondata.core.metadata.datatype.DataType;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
  import org.apache.carbondata.core.util.CarbonProperties;
  import org.apache.carbondata.core.util.CarbonThreadFactory;
  import org.apache.carbondata.processing.datatypes.GenericDataType;
  import org.apache.carbondata.processing.store.writer.CarbonFactDataWriter;
  
  import org.apache.log4j.Logger;
  
  /**
   * Fact data handler class to handle the fact data
   */
  public class CarbonFactDataHandlerColumnar implements CarbonFactHandler {
  
    /**
     * LOGGER
     */
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(CarbonFactDataHandlerColumnar.class.getName());
  
    private CarbonFactDataHandlerModel model;
  
    /**
     * data writer
     */
    private CarbonFactDataWriter dataWriter;
  
    /**
     * total number of entries in blocklet
     */
    private int entryCount;
  
    /**
     * blocklet size (for V1 and V2) or page size (for V3). A Producer thread will start to process
     * once this size of input is reached
     */
    private int pageSize;
  
    private long processedDataCount;
    private ExecutorService producerExecutorService;
    private List<Future<Void>> producerExecutorServiceTaskList;
    private ExecutorService consumerExecutorService;
    private List<Future<Void>> consumerExecutorServiceTaskList;
    private List<CarbonRow> dataRows;
    private int[] noDictColumnPageSize;
    /**
     * semaphore which will used for managing node holder objects
     */
    private Semaphore semaphore;
    /**
     * counter that incremented for every job submitted to data writer thread
     */
    private int writerTaskSequenceCounter;
    /**
     * a private class that will hold the data for blocklets
     */
    private TablePageList tablePageList;
    /**
     * number of cores configured
     */
    private int numberOfCores;
    /**
     * integer that will be incremented for every new blocklet submitted to producer for processing
     * the data and decremented every time consumer fetches the blocklet for writing
     */
    private AtomicInteger blockletProcessingCount;
    /**
     * flag to check whether all blocklets have been finished writing
     */
    private boolean processingComplete;
  
    /**
     * current data format version
     */
    private ColumnarFormatVersion version;
  
    /*
    * cannot use the indexMap of model directly,
    * modifying map in model directly will create problem if accessed later,
    * Hence take a copy and work on it.
    * */
    private Map<Integer, GenericDataType> complexIndexMapCopy = null;
  
    /* configured page size in MB*/
    private int configuredPageSizeInBytes = 0;
  
    /**
     * CarbonFactDataHandler constructor
     */
    public CarbonFactDataHandlerColumnar(CarbonFactDataHandlerModel model) {
      this.model = model;
      initParameters(model);
      this.version = CarbonProperties.getInstance().getFormatVersion();
      StringBuffer noInvertedIdxCol = new StringBuffer();
      for (CarbonDimension cd : model.getSegmentProperties().getDimensions()) {
        if (!cd.isUseInvertedIndex()) {
          noInvertedIdxCol.append(cd.getColName()).append(",");
        }
      }
  
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Columns considered as NoInverted Index are " + noInvertedIdxCol.toString());
      }
      this.complexIndexMapCopy = new HashMap<>();
      for (Map.Entry<Integer, GenericDataType> entry: model.getComplexIndexMap().entrySet()) {
        this.complexIndexMapCopy.put(entry.getKey(), entry.getValue().deepCopy());
      }
      String pageSizeStrInBytes =
          model.getTableSpec().getCarbonTable().getTableInfo().getFactTable().getTableProperties()
              .get(CarbonCommonConstants.TABLE_PAGE_SIZE_INMB);
      if (pageSizeStrInBytes != null) {
        configuredPageSizeInBytes = Integer.parseInt(pageSizeStrInBytes) * 1024 * 1024;
      }
    }
  
    private void initParameters(CarbonFactDataHandlerModel model) {
      this.numberOfCores = model.getNumberOfCores();
      blockletProcessingCount = new AtomicInteger(0);
      producerExecutorService = Executors.newFixedThreadPool(model.getNumberOfCores(),
          new CarbonThreadFactory(
              String.format("ProducerPool:%s, range: %d",
                  model.getTableName(), model.getBucketId()), true));
      producerExecutorServiceTaskList =
          new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      LOGGER.debug("Initializing writer executors");
      consumerExecutorService = Executors.newFixedThreadPool(1, new CarbonThreadFactory(
          String.format("ConsumerPool:%s, range: %d",
                  model.getTableName(), model.getBucketId()), true));
      consumerExecutorServiceTaskList = new ArrayList<>(1);
      semaphore = new Semaphore(numberOfCores);
      tablePageList = new TablePageList();
  
      // Start the consumer which will take each blocklet/page in order and write to a file
      Consumer consumer = new Consumer(tablePageList);
      consumerExecutorServiceTaskList.add(consumerExecutorService.submit(consumer));
    }
  
    private void setComplexMapSurrogateIndex(int dimensionCount) {
      int surrIndex = 0;
      for (int i = 0; i < dimensionCount; i++) {
        GenericDataType complexDataType = model.getComplexIndexMap().get(i);
        if (complexDataType != null) {
          List<GenericDataType> primitiveTypes = new ArrayList<GenericDataType>();
          complexDataType.getAllPrimitiveChildren(primitiveTypes);
          for (GenericDataType eachPrimitive : primitiveTypes) {
            if (eachPrimitive.getIsColumnDictionary()) {
              eachPrimitive.setSurrogateIndex(surrIndex++);
            }
          }
        } else {
          surrIndex++;
        }
      }
    }
  
    /**
     * This method will be used to get and update the step properties which will
     * required to run this step
     *
     * @throws CarbonDataWriterException
     */
    @Override
    public void initialise() throws CarbonDataWriterException {
      setWritingConfiguration();
    }
  
    /**
     * below method will be used to add row to store
     *
     * @param row
     * @throws CarbonDataWriterException
     */
    @Override
    public void addDataToStore(CarbonRow row) throws CarbonDataWriterException {
      int totalComplexColumnDepth = setFlatCarbonRowForComplex(row);
      if (noDictColumnPageSize == null) {
        // initialization using first row.
        model.setNoDictAllComplexColumnDepth(totalComplexColumnDepth);
        if (model.getNoDictDataTypesList().size() + model.getNoDictAllComplexColumnDepth() > 0) {
          noDictColumnPageSize =
              new int[model.getNoDictDataTypesList().size() + model.getNoDictAllComplexColumnDepth()];
        }
      }
  
      dataRows.add(row);
      this.entryCount++;
      // if entry count reaches to leaf node size then we are ready to write
      // this to leaf node file and update the intermediate files
      if (this.entryCount == this.pageSize || needToCutThePage(row)) {
        try {
          semaphore.acquire();
  
          producerExecutorServiceTaskList.add(
              producerExecutorService.submit(
                  new Producer(tablePageList, dataRows, ++writerTaskSequenceCounter, false)
              )
          );
          blockletProcessingCount.incrementAndGet();
          // set the entry count to zero
          processedDataCount += entryCount;
  
          if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Total Number Of records added to store: " + processedDataCount);
          }
          dataRows = new ArrayList<>(this.pageSize);
          this.entryCount = 0;
          // re-init the complexIndexMap
          this.complexIndexMapCopy = new HashMap<>();
          for (Map.Entry<Integer, GenericDataType> entry : model.getComplexIndexMap().entrySet()) {
            this.complexIndexMapCopy.put(entry.getKey(), entry.getValue().deepCopy());
          }
          noDictColumnPageSize =
              new int[model.getNoDictDataTypesList().size() + model.getNoDictAllComplexColumnDepth()];
        } catch (InterruptedException e) {
          LOGGER.error(e.getMessage(), e);
          throw new CarbonDataWriterException(e);
        }
      }
    }
  
    /**
     * Check if column page can be added more rows after adding this row to page.
     * only few no-dictionary dimensions columns (string, varchar,
     * complex columns) can grow huge in size.
     *
     *
     * @param row carbonRow
     * @return false if next rows can be added to same page.
     * true if next rows cannot be added to same page
     */
    private boolean needToCutThePage(CarbonRow row) {
      List<DataType> noDictDataTypesList = model.getNoDictDataTypesList();
      int totalNoDictPageCount = noDictDataTypesList.size() + model.getNoDictAllComplexColumnDepth();
      if (totalNoDictPageCount > 0) {
        int currentElementLength;
        int bucketCounter = 0;
        if (configuredPageSizeInBytes == 0) {
          // no need to cut the page
          // use default value
          /*configuredPageSizeInBytes =
              CarbonCommonConstants.TABLE_PAGE_SIZE_INMB_DEFAULT * 1024 * 1024;*/
          return false;
        }
        Object[] nonDictArray = WriteStepRowUtil.getNoDictAndComplexDimension(row);
        for (int i = 0; i < noDictDataTypesList.size(); i++) {
          DataType columnType = noDictDataTypesList.get(i);
          if ((columnType == DataTypes.STRING) || (columnType == DataTypes.VARCHAR) || (columnType
              == DataTypes.BINARY)) {
            currentElementLength = ((byte[]) nonDictArray[i]).length;
            noDictColumnPageSize[bucketCounter] += currentElementLength;
            canSnappyHandleThisRow(noDictColumnPageSize[bucketCounter]);
            // If current page size is more than configured page size, cut the page here.
            if (noDictColumnPageSize[bucketCounter] + dataRows.size() * 4
                >= configuredPageSizeInBytes) {
              if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("cutting the page. Rows count in this page: " + dataRows.size());
              }
              // re-init for next page
              noDictColumnPageSize = new int[totalNoDictPageCount];
              return true;
            }
            bucketCounter++;
          } else if (columnType.isComplexType()) {
            // this is for depth of each complex column, model is having only total depth.
            GenericDataType genericDataType = complexIndexMapCopy
                .get(i - model.getNoDictionaryCount() +
                    model.getSegmentProperties().getNumberOfPrimitiveDimensions());
            int depth = genericDataType.getDepth();
            List<ArrayList<byte[]>> flatComplexColumnList = (List<ArrayList<byte[]>>) nonDictArray[i];
            for (int k = 0; k < depth; k++) {
              ArrayList<byte[]> children = flatComplexColumnList.get(k);
              // Add child element from inner list.
              int complexElementSize = 0;
              for (byte[] child : children) {
                complexElementSize += child.length;
              }
              noDictColumnPageSize[bucketCounter] += complexElementSize;
              canSnappyHandleThisRow(noDictColumnPageSize[bucketCounter]);
              // If current page size is more than configured page size, cut the page here.
              if (noDictColumnPageSize[bucketCounter] + dataRows.size() * 4
                  >= configuredPageSizeInBytes) {
                LOGGER.info("cutting the page. Rows count: " + dataRows.size());
                // re-init for next page
                noDictColumnPageSize = new int[totalNoDictPageCount];
                return true;
              }
              bucketCounter++;
            }
          }
        }
      }
      return false;
    }
  
    private int setFlatCarbonRowForComplex(CarbonRow row) {
      int noDictTotalComplexChildDepth = 0;
      Object[] noDictAndComplexDimension = WriteStepRowUtil.getNoDictAndComplexDimension(row);
      for (int i = 0; i < noDictAndComplexDimension.length; i++) {
        // complex types starts after no dictionary dimensions
        if (i >= model.getNoDictionaryCount() && (model.getTableSpec().getNoDictionaryDimensionSpec()
            .get(i).getSchemaDataType().isComplexType())) {
          // this is for depth of each complex column, model is having only total depth.
          GenericDataType genericDataType = complexIndexMapCopy.get(
              i - model.getNoDictionaryCount() +
                  model.getSegmentProperties().getNumberOfPrimitiveDimensions());
          int depth = genericDataType.getDepth();
          // initialize flatComplexColumnList
          List<ArrayList<byte[]>> flatComplexColumnList = new ArrayList<>(depth);
          for (int k = 0; k < depth; k++) {
            flatComplexColumnList.add(new ArrayList<byte[]>());
          }
          // flatten the complex byteArray as per depth
          try {
            ByteBuffer byteArrayInput = ByteBuffer.wrap((byte[])noDictAndComplexDimension[i]);
            ByteArrayOutputStream byteArrayOutput = new ByteArrayOutputStream();
            DataOutputStream dataOutputStream = new DataOutputStream(byteArrayOutput);
            genericDataType.parseComplexValue(byteArrayInput, dataOutputStream);
            genericDataType.getColumnarDataForComplexType(flatComplexColumnList,
                ByteBuffer.wrap(byteArrayOutput.toByteArray()));
            byteArrayOutput.close();
          } catch (IOException | KeyGenException e) {
            throw new CarbonDataWriterException("Problem in splitting and writing complex data", e);
          }
          noDictTotalComplexChildDepth += flatComplexColumnList.size();
          // update the complex column data with the flat data
          noDictAndComplexDimension[i] = flatComplexColumnList;
        }
      }
      return noDictTotalComplexChildDepth;
    }
  
    private void canSnappyHandleThisRow(int currentRowSize) {
      if (currentRowSize > SnappyCompressor.MAX_BYTE_TO_COMPRESS) {
        throw new RuntimeException(" page size: " + currentRowSize + " exceed snappy size: "
            + SnappyCompressor.MAX_BYTE_TO_COMPRESS + " Bytes. Snappy cannot compress it ");
      }
    }
  
    /**
     * generate the EncodedTablePage from the input rows (one page in case of V3 format)
     */
    private TablePage processDataRows(List<CarbonRow> dataRows)
        throws CarbonDataWriterException, IOException {
      if (dataRows.size() == 0) {
        return new TablePage(model, 0);
      }
      TablePage tablePage = new TablePage(model, dataRows.size());
      int rowId = 0;
  
      // convert row to columnar data
      for (CarbonRow row : dataRows) {
        tablePage.addRow(rowId++, row);
      }
  
      tablePage.encode();
  
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Number Of records processed: " + dataRows.size());
      }
      return tablePage;
    }
  
    /**
     * below method will be used to finish the data handler
     *
     * @throws CarbonDataWriterException
     */
    @Override
    public void finish() throws CarbonDataWriterException {
      // still some data is present in stores if entryCount is more
      // than 0
      if (null == dataWriter) {
        return;
      }
      if (producerExecutorService.isShutdown()) {
        return;
      }
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Started Finish Operation");
      }
      try {
        semaphore.acquire();
        producerExecutorServiceTaskList.add(producerExecutorService
            .submit(new Producer(tablePageList, dataRows, ++writerTaskSequenceCounter, true)));
        blockletProcessingCount.incrementAndGet();
        processedDataCount += entryCount;
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("Total Number Of records added to store: " + processedDataCount);
        }
        closeWriterExecutionService(producerExecutorService);
        processWriteTaskSubmitList(producerExecutorServiceTaskList);
        processingComplete = true;
      } catch (InterruptedException e) {
        LOGGER.error(e.getMessage(), e);
        throw new CarbonDataWriterException(e);
      }
    }
  
    /**
     * This method will close writer execution service and get the node holders and
     * add them to node holder list
     *
     * @param service the service to shutdown
     * @throws CarbonDataWriterException
     */
    private void closeWriterExecutionService(ExecutorService service)
        throws CarbonDataWriterException {
      try {
        service.shutdown();
        service.awaitTermination(1, TimeUnit.DAYS);
      } catch (InterruptedException e) {
        LOGGER.error(e.getMessage(), e);
        throw new CarbonDataWriterException(e);
      }
    }
  
    /**
     * This method will iterate through future task list and check if any exception
     * occurred during the thread execution
     *
     * @param taskList
     * @throws CarbonDataWriterException
     */
    private void processWriteTaskSubmitList(List<Future<Void>> taskList)
        throws CarbonDataWriterException {
      for (int i = 0; i < taskList.size(); i++) {
        try {
          taskList.get(i).get();
        } catch (InterruptedException | ExecutionException e) {
          LOGGER.error(e.getMessage(), e);
          throw new CarbonDataWriterException(e);
        }
      }
    }
  
    /**
     * below method will be used to close the handler
     */
    @Override
    public void closeHandler() throws CarbonDataWriterException {
      if (null != this.dataWriter) {
        // wait until all blocklets have been finished writing
        while (blockletProcessingCount.get() > 0) {
          try {
            Thread.sleep(50);
          } catch (InterruptedException e) {
            throw new CarbonDataWriterException(e);
          }
        }
        consumerExecutorService.shutdownNow();
        processWriteTaskSubmitList(consumerExecutorServiceTaskList);
        this.dataWriter.writeFooter();
        if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("All blocklets have been finished writing");
        }
        // close all the open stream for both the files
        this.dataWriter.closeWriter();
      }
      this.dataWriter = null;
    }
  
    /**
     * Below method will be to configure fact file writing configuration
     *
     * @throws CarbonDataWriterException
     */
    private void setWritingConfiguration() throws CarbonDataWriterException {
      // get blocklet size
      this.pageSize = Integer.parseInt(CarbonProperties.getInstance()
          .getProperty(CarbonCommonConstants.BLOCKLET_SIZE,
              CarbonCommonConstants.BLOCKLET_SIZE_DEFAULT_VAL));
      // support less than 32000 rows in one page, because we support super long string,
      // if it is long enough, a column page with 32000 rows will exceed 2GB
      if (version == ColumnarFormatVersion.V3) {
        this.pageSize =
            pageSize < CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT ?
                pageSize :
                CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT;
      }
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Number of rows per column page is configured as pageSize = " + pageSize);
      }
      dataRows = new ArrayList<>(this.pageSize);
      setComplexMapSurrogateIndex(model.getDictDimensionCount());
      this.dataWriter = getFactDataWriter();
      // initialize the channel;
      this.dataWriter.initializeWriter();
    }
  
    /**
     * Below method will be used to get the fact data writer instance
     *
     * @return data writer instance
     */
    private CarbonFactDataWriter getFactDataWriter() {
      return CarbonDataWriterFactory.getInstance().getFactDataWriter(version, model);
    }
  
    /**
     * This method will reset the block processing count
     */
    private void resetBlockletProcessingCount() {
      blockletProcessingCount.set(0);
    }
  
    /**
     * This class will hold the table page data
     */
    private final class TablePageList {
      /**
       * array of table page added by Producer and get by Consumer
       */
      private TablePage[] tablePages;
      /**
       * flag to check whether the producer has completed processing for holder
       * object which is required to be picked form an index
       */
      private AtomicBoolean available;
      /**
       * index from which data node holder object needs to be picked for writing
       */
      private int currentIndex;
  
      private TablePageList() {
        tablePages = new TablePage[numberOfCores];
        available = new AtomicBoolean(false);
      }
  
      /**
       * @return a node holder object
       * @throws InterruptedException if consumer thread is interrupted
       */
      public synchronized TablePage get() throws InterruptedException {
        TablePage tablePage = tablePages[currentIndex];
        // if node holder is null means producer thread processing the data which has to
        // be inserted at this current index has not completed yet
        if (null == tablePage && !processingComplete) {
          available.set(false);
        }
        while (!available.get()) {
          wait();
        }
        tablePage = tablePages[currentIndex];
        tablePages[currentIndex] = null;
        currentIndex++;
        // reset current index when it reaches length of node holder array
        if (currentIndex >= tablePages.length) {
          currentIndex = 0;
        }
        return tablePage;
      }
  
      /**
       * @param tablePage
       * @param index
       */
      public synchronized void put(TablePage tablePage, int index) {
        tablePages[index] = tablePage;
        // notify the consumer thread when index at which object is to be inserted
        // becomes equal to current index from where data has to be picked for writing
        if (index == currentIndex) {
          available.set(true);
          notifyAll();
        }
      }
    }
  
    /**
     * Producer which will process data equivalent to 1 blocklet size
     */
    private final class Producer implements Callable<Void> {
  
      private TablePageList tablePageList;
      private List<CarbonRow> dataRows;
      private int pageId;
      private boolean isLastPage;
  
      private Producer(TablePageList tablePageList, List<CarbonRow> dataRows,
          int pageId, boolean isLastPage) {
        this.tablePageList = tablePageList;
        this.dataRows = dataRows;
        this.pageId = pageId;
        this.isLastPage = isLastPage;
      }
  
      /**
       * Computes a result, or throws an exception if unable to do so.
       *
       * @return computed result
       */
      @Override
      public Void call() {
        try {
          TablePage tablePage = processDataRows(dataRows);
          dataRows = null;
          tablePage.setIsLastPage(isLastPage);
          // insert the object in array according to sequence number
          int indexInNodeHolderArray = (pageId - 1) % numberOfCores;
          tablePageList.put(tablePage, indexInNodeHolderArray);
          return null;
        } catch (Throwable throwable) {
          LOGGER.error("Error in producer", throwable);
          consumerExecutorService.shutdownNow();
          resetBlockletProcessingCount();
          throw new CarbonDataWriterException(throwable.getMessage(), throwable);
        }
      }
    }
  
    /**
     * Consumer class will get one blocklet data at a time and submit for writing
     */
    private final class Consumer implements Callable<Void> {
  
      private TablePageList tablePageList;
  
      private Consumer(TablePageList tablePageList) {
        this.tablePageList = tablePageList;
      }
  
      /**
       * Computes a result, or throws an exception if unable to do so.
       *
       * @return computed result
       */
      @Override
      public Void call() {
        while (!processingComplete || blockletProcessingCount.get() > 0) {
          TablePage tablePage = null;
          try {
            tablePage = tablePageList.get();
            if (null != tablePage) {
              dataWriter.writeTablePage(tablePage);
              tablePage.freeMemory();
            }
            blockletProcessingCount.decrementAndGet();
          } catch (Throwable throwable) {
            if (!processingComplete || blockletProcessingCount.get() > 0) {
              producerExecutorService.shutdownNow();
              resetBlockletProcessingCount();
              LOGGER.error("Problem while writing the carbon data file", throwable);
              throw new CarbonDataWriterException(throwable);
            }
          } finally {
            semaphore.release();
          }
        }
        return null;
      }
    }
  }
  