10172

Build # 10172

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/042e1e8bbda46367370851d5ad6b15c3b50950e7">042e1e8bb<a href="https://github.com/apache/carbondata/commit/042e1e8bbda46367370851d5ad6b15c3b50950e7">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/ed5b2d33c594a0172eeedb0499e7736aae6d4bc9">ed5b2d33c</a>">ed5b2d33c</a><a href="https://github.com/apache/carbondata/commit/042e1e8bbda46367370851d5ad6b15c3b50950e7"> into bd752e9d5">bd752e9d5</a>

Run Details

23 of 23 new or added lines in 1 file covered. (100.0%)

61528 of 77832 relevant lines covered (79.05%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.25

/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java

/*
 * 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.writer;

import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.nio.channels.Channels;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
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.TimeUnit;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonLoadOptionConstants;
import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.keygenerator.mdkey.NumberCompressor;
import org.apache.carbondata.core.metadata.converter.SchemaConverter;
import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
import org.apache.carbondata.core.metadata.index.BlockIndexInfo;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.util.CarbonMetadataUtil;
import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonThreadFactory;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
import org.apache.carbondata.core.writer.CarbonIndexFileWriter;
import org.apache.carbondata.format.BlockIndex;
import org.apache.carbondata.format.BlockletInfo3;
import org.apache.carbondata.format.IndexHeader;
import org.apache.carbondata.processing.datamap.DataMapWriterListener;
import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;

import org.apache.log4j.Logger;

public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(AbstractFactDataWriter.class.getName());

  /**
   * file channel to write
   */
  protected WritableByteChannel fileChannel;
  protected long currentOffsetInFile;
  /**
   * The path of CarbonData file to write in hdfs/s3
   */
  private String carbonDataFileStorePath;
  /**
   * The temp path of carbonData file used on executor
   */
  private String carbonDataFileTempPath;

  /**
   * The name of carbonData file (blockId)
   */
  protected String carbonDataFileName;

  /**
   * The sequence number of blocklet inside one block
   */
  protected int blockletId = 0;

  /**
   * The sequence number of page inside one blocklet
   */
  protected int pageId = 0;

  /**
   * Local cardinality for the segment
   */
  protected int[] localCardinality;
  /**
   * thrift column schema
   */
  protected List<org.apache.carbondata.format.ColumnSchema> thriftColumnSchemaList;
  protected NumberCompressor numberCompressor;
  protected CarbonFactDataHandlerModel model;
  protected List<List<Long>> dataChunksOffsets;
  protected List<List<Short>> dataChunksLength;
  /**
   * data file size;
   */
  protected long fileSizeInBytes;
  /**
   * file count will be used to give sequence number to the data file
   */
  private int fileCount;
  /**
   * executorService
   */
  private ExecutorService executorService;
  /**
   * executorService
   */
  private List<Future<Void>> executorServiceSubmitList;
  /**
   * data block size for one carbon data file
   */
  private long blockSizeThreshold;
  /**
   * file size at any given point
   */
  private long currentFileSize;

  protected DataOutputStream fileOutputStream;

  protected List<BlockIndexInfo> blockIndexInfoList;

  /**
   * list of metadata for V3 format
   */
  protected List<BlockletInfo3> blockletMetadata;

  /**
   * list of blocklet index
   */
  protected List<org.apache.carbondata.format.BlockletIndex> blockletIndex;

  /**
   * listener to write data map
   */
  protected DataMapWriterListener listener;
  /**
   * Whether directly write fact data to store path
   */
  private boolean enableDirectlyWriteDataToStorePath = false;

  protected ExecutorService fallbackExecutorService;

  public AbstractFactDataWriter(CarbonFactDataHandlerModel model) {
    this.model = model;
    blockIndexInfoList = new ArrayList<>();
    // get max file size;
    CarbonProperties propInstance = CarbonProperties.getInstance();
    // if blocksize=2048, then 2048*1024*1024 will beyond the range of Int
    this.fileSizeInBytes =
        (long) this.model.getBlockSizeInMB() * CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR
            * CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR;

    // size reserved in one file for writing block meta data. It will be in percentage
    int spaceReservedForBlockMetaSize = Integer.parseInt(propInstance
        .getProperty(CarbonCommonConstants.CARBON_BLOCK_META_RESERVED_SPACE,
            CarbonCommonConstants.CARBON_BLOCK_META_RESERVED_SPACE_DEFAULT));
    this.blockSizeThreshold =
        fileSizeInBytes - (fileSizeInBytes * spaceReservedForBlockMetaSize) / 100;
    LOGGER
        .info("Total file size: " + fileSizeInBytes + " and dataBlock Size: " + blockSizeThreshold);

    // whether to directly write fact data to HDFS
    String directlyWriteData2Hdfs = propInstance
        .getProperty(CarbonLoadOptionConstants.ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH,
            CarbonLoadOptionConstants.ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH_DEFAULT);
    this.enableDirectlyWriteDataToStorePath = "TRUE".equalsIgnoreCase(directlyWriteData2Hdfs);

    if (enableDirectlyWriteDataToStorePath) {
      LOGGER.info("Carbondata will directly write fact data to store path.");
    } else {
      LOGGER.info("Carbondata will write temporary fact data to local disk.");
    }

    this.executorService = Executors.newFixedThreadPool(1,
        new CarbonThreadFactory("CompleteHDFSBackendPool:" + this.model.getTableName()));
    executorServiceSubmitList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    // in case of compaction we will pass the cardinality.
    this.localCardinality = this.model.getColCardinality();

    List<Integer> cardinalityList = new ArrayList<Integer>();
    thriftColumnSchemaList = getColumnSchemaListAndCardinality(cardinalityList, localCardinality,
        this.model.getWrapperColumnSchema());
    this.numberCompressor = new NumberCompressor(Integer.parseInt(CarbonProperties.getInstance()
        .getProperty(CarbonCommonConstants.BLOCKLET_SIZE,
            CarbonCommonConstants.BLOCKLET_SIZE_DEFAULT_VAL)));
    this.dataChunksOffsets = new ArrayList<>();
    this.dataChunksLength = new ArrayList<>();
    blockletMetadata = new ArrayList<BlockletInfo3>();
    blockletIndex = new ArrayList<>();
    listener = this.model.getDataMapWriterlistener();
    if (model.getColumnLocalDictGenMap().size() > 0) {
      int numberOfCores = 1;
      if (model.getNumberOfCores() > 1) {
        numberOfCores = model.getNumberOfCores() / 2;
      }
      fallbackExecutorService = Executors.newFixedThreadPool(numberOfCores, new CarbonThreadFactory(
          "FallbackPool:" + model.getTableName() + ", range: " + model.getBucketId()));
    }
  }

  /**
   * This method will be used to update the file channel with new file if exceeding block size
   * threshold, new file will be created once existing file reached the file size limit This
   * method will first check whether existing file size is exceeded the file
   * size limit if yes then write the leaf metadata to file then set the
   * current file size to 0 close the existing file channel get the new file
   * name and get the channel for new file
   *
   * @param blockletSizeToBeAdded data size of one block
   * @throws CarbonDataWriterException if any problem
   */
  protected void createNewFileIfReachThreshold(long blockletSizeToBeAdded)
      throws CarbonDataWriterException {
    if ((currentFileSize + blockletSizeToBeAdded) >= blockSizeThreshold && currentFileSize != 0) {
      // set the current file size to zero
      String activeFile =
          enableDirectlyWriteDataToStorePath ? carbonDataFileStorePath : carbonDataFileTempPath;
      LOGGER.info("Writing data to file as max file size reached for file: "
          + activeFile + ". Data block size: " + currentFileSize);
      // write meta data to end of the existing file
      writeFooterToFile();
      this.dataChunksOffsets = new ArrayList<>();
      this.dataChunksLength = new ArrayList<>();
      this.blockletMetadata = new ArrayList<>();
      this.blockletIndex = new ArrayList<>();
      commitCurrentFile(false);
      // initialize the new channel
      this.currentFileSize = 0;
      initializeWriter();
    }
    currentFileSize += blockletSizeToBeAdded;
  }

  private void notifyDataMapBlockStart() {
    if (listener != null) {
      try {
        listener.onBlockStart(carbonDataFileName);
      } catch (IOException e) {
        throw new CarbonDataWriterException("Problem while writing datamap", e);
      }
    }
  }

  private void notifyDataMapBlockEnd() {
    if (listener != null) {
      try {
        listener.onBlockEnd(carbonDataFileName);
      } catch (IOException e) {
        throw new CarbonDataWriterException("Problem while writing datamap", e);
      }
    }
  }

  /**
   * Finish writing current file. It will flush stream, copy and rename temp file to final file
   * @param copyInCurrentThread set to false if want to do data copy in a new thread
   */
  protected void commitCurrentFile(boolean copyInCurrentThread) {
    notifyDataMapBlockEnd();
    CarbonUtil.closeStreams(this.fileOutputStream, this.fileChannel);
    if (!enableDirectlyWriteDataToStorePath) {
      try {
        if (currentFileSize == 0) {
          handleEmptyDataFile(carbonDataFileTempPath);
        } else {
          if (copyInCurrentThread) {
            CarbonUtil.copyCarbonDataFileToCarbonStorePath(carbonDataFileTempPath,
                model.getCarbonDataDirectoryPath(), fileSizeInBytes);
            FileFactory.deleteFile(carbonDataFileTempPath,
                FileFactory.getFileType(carbonDataFileTempPath));
          } else {
            executorServiceSubmitList
                .add(executorService.submit(new CompleteHdfsBackendThread(carbonDataFileTempPath)));
          }
        }
      } catch (IOException e) {
        LOGGER.error(e);
      }
    } else {
      if (currentFileSize == 0) {
        try {
          handleEmptyDataFile(carbonDataFileStorePath);
        } catch (IOException e) {
          LOGGER.error(e);
        }
      }
    }
  }

  private void handleEmptyDataFile(String filePath) throws IOException {
    FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath));
    if (blockIndexInfoList.size() > 0 && blockIndexInfoList.get(blockIndexInfoList.size() - 1)
        .getFileName().equals(carbonDataFileName)) {
      // no need add this entry in index file
      blockIndexInfoList.remove(blockIndexInfoList.size() - 1);
      // TODO: currently there is no implementation for notifyDataMapBlockEnd(),
      // hence no impact, once implementation is done. Need to undo it in this case.
    }
  }

  /**
   * This method will be used to initialize the channel
   *
   * @throws CarbonDataWriterException
   */
  public void initializeWriter() throws CarbonDataWriterException {
    this.carbonDataFileName = CarbonTablePath
        .getCarbonDataFileName(fileCount, model.getCarbonDataFileAttributes().getTaskId(),
            model.getBucketId(), model.getTaskExtension(),
            "" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());
    this.carbonDataFileStorePath = model.getCarbonDataDirectoryPath() + File.separator
        + carbonDataFileName;
    try {
      if (enableDirectlyWriteDataToStorePath) {
        // the block size will be twice the block_size specified by user to make sure that
        // one carbondata file only consists exactly one HDFS block.
        fileOutputStream = FileFactory
            .getDataOutputStream(carbonDataFileStorePath, FileFactory.FileType.HDFS,
                CarbonCommonConstants.BYTEBUFFER_SIZE, fileSizeInBytes * 2);
      } else {
        //each time we initialize writer, we choose a local temp location randomly
        String[] tempFileLocations = model.getStoreLocation();
        String chosenTempLocation =
            tempFileLocations[new Random().nextInt(tempFileLocations.length)];
        LOGGER.info("Randomly choose factdata temp location: " + chosenTempLocation);
        carbonDataFileTempPath = chosenTempLocation + File.separator + carbonDataFileName;
        fileOutputStream = FileFactory.getDataOutputStream(carbonDataFileTempPath,
            FileFactory.FileType.LOCAL, CarbonCommonConstants.BYTEBUFFER_SIZE, true);
      }

      this.fileCount++;
      // open channel for new data file
      this.fileChannel = Channels.newChannel(fileOutputStream);
      this.currentOffsetInFile = 0;
    } catch (IOException ex) {
      throw new CarbonDataWriterException(
          "Problem while getting the channel for fact data file", ex);
    }
    notifyDataMapBlockStart();
  }

  /**
   * This method will write metadata at the end of file file format in thrift format
   */
  protected abstract void writeFooterToFile() throws CarbonDataWriterException;

  /**
   * Below method will be used to fill the vlock info details
   *
   * @param numberOfRows    number of rows in file
   * @param carbonDataFileName The name of carbonData file
   * @param currentPosition current offset
   */
  protected abstract void fillBlockIndexInfoDetails(long numberOfRows, String carbonDataFileName,
      long currentPosition);

  public static List<org.apache.carbondata.format.ColumnSchema> getColumnSchemaListAndCardinality(
      List<Integer> cardinality, int[] dictionaryColumnCardinality,
      List<ColumnSchema> wrapperColumnSchemaList) {
    List<org.apache.carbondata.format.ColumnSchema> columnSchemaList =
        new ArrayList<org.apache.carbondata.format.ColumnSchema>(
            CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
    int counter = 0;
    for (int i = 0; i < wrapperColumnSchemaList.size(); i++) {
      columnSchemaList
          .add(schemaConverter.fromWrapperToExternalColumnSchema(wrapperColumnSchemaList.get(i)));
      if (CarbonUtil.hasEncoding(wrapperColumnSchemaList.get(i).getEncodingList(),
          org.apache.carbondata.core.metadata.encoder.Encoding.DICTIONARY)) {
        cardinality.add(dictionaryColumnCardinality[counter]);
        counter++;
      } else if (!wrapperColumnSchemaList.get(i).isDimensionColumn()) {
        continue;
      } else {
        cardinality.add(-1);
      }
    }
    return columnSchemaList;
  }

  /**
   * Below method will be used to write the idex file
   *
   * @throws IOException               throws io exception if any problem while writing
   * @throws CarbonDataWriterException data writing
   */
  protected void writeIndexFile() throws IOException, CarbonDataWriterException {
    if (blockIndexInfoList.size() == 0) {
      // no need to write index file, if data file is not there.
      return;
    }
    // get the header
    IndexHeader indexHeader = CarbonMetadataUtil
        .getIndexHeader(localCardinality, thriftColumnSchemaList, model.getBucketId(),
            model.getSchemaUpdatedTimeStamp());
    // get the block index info thrift
    List<BlockIndex> blockIndexThrift = CarbonMetadataUtil.getBlockIndexInfo(blockIndexInfoList);
    String indexFileName;
    if (enableDirectlyWriteDataToStorePath) {
      String rawFileName = model.getCarbonDataDirectoryPath() + CarbonCommonConstants.FILE_SEPARATOR
          + CarbonTablePath.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),
          model.getBucketId(), model.getTaskExtension(),
          "" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());
      indexFileName = FileFactory.getUpdatedFilePath(rawFileName, FileFactory.FileType.HDFS);
    } else {
      // randomly choose a temp location for index file
      String[] tempLocations = model.getStoreLocation();
      String chosenTempLocation = tempLocations[new Random().nextInt(tempLocations.length)];
      LOGGER.info("Randomly choose index file location: " + chosenTempLocation);
      indexFileName = chosenTempLocation + File.separator + CarbonTablePath
          .getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),
              model.getBucketId(), model.getTaskExtension(),
              "" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());
    }

    CarbonIndexFileWriter writer = new CarbonIndexFileWriter();
    // open file
    writer.openThriftWriter(indexFileName);
    // write the header first
    writer.writeThrift(indexHeader);
    // write the indexes
    for (BlockIndex blockIndex : blockIndexThrift) {
      writer.writeThrift(blockIndex);
    }
    writer.close();
    if (!enableDirectlyWriteDataToStorePath) {
      CarbonUtil
          .copyCarbonDataFileToCarbonStorePath(indexFileName, model.getCarbonDataDirectoryPath(),
              fileSizeInBytes);
      FileFactory.deleteFile(indexFileName, FileFactory.getFileType(indexFileName));
    }
  }

  /**
   * This method will close the executor service which is used for copying carbon
   * data files to carbon store path
   *
   * @throws CarbonDataWriterException
   */
  protected void closeExecutorService() throws CarbonDataWriterException {
    CarbonDataWriterException exception = null;
    try {
      listener.finish();
      listener = null;
    } catch (IOException e) {
      exception = new CarbonDataWriterException(e);
    }
    try {
      executorService.shutdown();
      executorService.awaitTermination(2, TimeUnit.HOURS);
      for (int i = 0; i < executorServiceSubmitList.size(); i++) {
        executorServiceSubmitList.get(i).get();
      }
    } catch (InterruptedException | ExecutionException e) {
      if (null == exception) {
        exception = new CarbonDataWriterException(e);
      }
    }
    if (null != fallbackExecutorService) {
      fallbackExecutorService.shutdownNow();
    }
    if (exception != null) {
      throw exception;
    }
  }



  /**
   * This method will complete hdfs backend storage for this file.
   * It may copy the carbon data file from local store location to carbon store location,
   * it may also complete the remaining replications for the existing hdfs file.
   */
  private final class CompleteHdfsBackendThread implements Callable<Void> {

    /**
     * complete path along with file name which needs to be copied to
     * carbon store path
     */
    private String fileName;

    private CompleteHdfsBackendThread(String fileName) {
      this.fileName = fileName;
    }

    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    @Override
    public Void call() throws Exception {
      CarbonUtil.copyCarbonDataFileToCarbonStorePath(fileName, model.getCarbonDataDirectoryPath(),
          fileSizeInBytes);
      FileFactory.deleteFile(fileName, FileFactory.getFileType(fileName));
      return null;
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.processing.store.writer;
19
20	import java.io.DataOutputStream;
21	import java.io.File;
22	import java.io.IOException;
23	import java.nio.channels.Channels;
24	import java.nio.channels.WritableByteChannel;
25	import java.util.ArrayList;
26	import java.util.List;
27	import java.util.Random;
28	import java.util.concurrent.Callable;
29	import java.util.concurrent.ExecutionException;
30	import java.util.concurrent.ExecutorService;
31	import java.util.concurrent.Executors;
32	import java.util.concurrent.Future;
33	import java.util.concurrent.TimeUnit;
34
35	import org.apache.carbondata.common.logging.LogServiceFactory;
36	import org.apache.carbondata.core.constants.CarbonCommonConstants;
37	import org.apache.carbondata.core.constants.CarbonLoadOptionConstants;
38	import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
39	import org.apache.carbondata.core.datastore.impl.FileFactory;
40	import org.apache.carbondata.core.keygenerator.mdkey.NumberCompressor;
41	import org.apache.carbondata.core.metadata.converter.SchemaConverter;
42	import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
43	import org.apache.carbondata.core.metadata.index.BlockIndexInfo;
44	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
45	import org.apache.carbondata.core.util.CarbonMetadataUtil;
46	import org.apache.carbondata.core.util.CarbonProperties;
47	import org.apache.carbondata.core.util.CarbonThreadFactory;
48	import org.apache.carbondata.core.util.CarbonUtil;
49	import org.apache.carbondata.core.util.path.CarbonTablePath;
50	import org.apache.carbondata.core.writer.CarbonIndexFileWriter;
51	import org.apache.carbondata.format.BlockIndex;
52	import org.apache.carbondata.format.BlockletInfo3;
53	import org.apache.carbondata.format.IndexHeader;
54	import org.apache.carbondata.processing.datamap.DataMapWriterListener;
55	import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
56
57	import org.apache.log4j.Logger;
58
59	public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
60
61	private static final Logger LOGGER =	1✔
62	LogServiceFactory.getLogService(AbstractFactDataWriter.class.getName());	1✔
63
64	/**
65	* file channel to write
66	*/
67	protected WritableByteChannel fileChannel;
68	protected long currentOffsetInFile;
69	/**
70	* The path of CarbonData file to write in hdfs/s3
71	*/
72	private String carbonDataFileStorePath;
73	/**
74	* The temp path of carbonData file used on executor
75	*/
76	private String carbonDataFileTempPath;
77
78	/**
79	* The name of carbonData file (blockId)
80	*/
81	protected String carbonDataFileName;
82
83	/**
84	* The sequence number of blocklet inside one block
85	*/
86	protected int blockletId = 0;	1✔
87
88	/**
89	* The sequence number of page inside one blocklet
90	*/
91	protected int pageId = 0;	1✔
92
93	/**
94	* Local cardinality for the segment
95	*/
96	protected int[] localCardinality;
97	/**
98	* thrift column schema
99	*/
100	protected List<org.apache.carbondata.format.ColumnSchema> thriftColumnSchemaList;
101	protected NumberCompressor numberCompressor;
102	protected CarbonFactDataHandlerModel model;
103	protected List<List<Long>> dataChunksOffsets;
104	protected List<List<Short>> dataChunksLength;
105	/**
106	* data file size;
107	*/
108	protected long fileSizeInBytes;
109	/**
110	* file count will be used to give sequence number to the data file
111	*/
112	private int fileCount;
113	/**
114	* executorService
115	*/
116	private ExecutorService executorService;
117	/**
118	* executorService
119	*/
120	private List<Future<Void>> executorServiceSubmitList;
121	/**
122	* data block size for one carbon data file
123	*/
124	private long blockSizeThreshold;
125	/**
126	* file size at any given point
127	*/
128	private long currentFileSize;
129
130	protected DataOutputStream fileOutputStream;
131
132	protected List<BlockIndexInfo> blockIndexInfoList;
133
134	/**
135	* list of metadata for V3 format
136	*/
137	protected List<BlockletInfo3> blockletMetadata;
138
139	/**
140	* list of blocklet index
141	*/
142	protected List<org.apache.carbondata.format.BlockletIndex> blockletIndex;
143
144	/**
145	* listener to write data map
146	*/
147	protected DataMapWriterListener listener;
148	/**
149	* Whether directly write fact data to store path
150	*/
151	private boolean enableDirectlyWriteDataToStorePath = false;	1✔
152
153	protected ExecutorService fallbackExecutorService;
154
155	public AbstractFactDataWriter(CarbonFactDataHandlerModel model) {	1✔
156	this.model = model;	1✔
157	blockIndexInfoList = new ArrayList<>();	1✔
158	// get max file size;
159	CarbonProperties propInstance = CarbonProperties.getInstance();	1✔
160	// if blocksize=2048, then 204810241024 will beyond the range of Int
161	this.fileSizeInBytes =	1✔
162	(long) this.model.getBlockSizeInMB() * CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR	1✔
163	* CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR;
164
165	// size reserved in one file for writing block meta data. It will be in percentage
166	int spaceReservedForBlockMetaSize = Integer.parseInt(propInstance	1✔
167	.getProperty(CarbonCommonConstants.CARBON_BLOCK_META_RESERVED_SPACE,	1✔
168	CarbonCommonConstants.CARBON_BLOCK_META_RESERVED_SPACE_DEFAULT));
169	this.blockSizeThreshold =	1✔
170	fileSizeInBytes - (fileSizeInBytes * spaceReservedForBlockMetaSize) / 100;
171	LOGGER	1✔
172	.info("Total file size: " + fileSizeInBytes + " and dataBlock Size: " + blockSizeThreshold);	1✔
173
174	// whether to directly write fact data to HDFS
175	String directlyWriteData2Hdfs = propInstance	1✔
176	.getProperty(CarbonLoadOptionConstants.ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH,	1✔
177	CarbonLoadOptionConstants.ENABLE_CARBON_LOAD_DIRECT_WRITE_TO_STORE_PATH_DEFAULT);
178	this.enableDirectlyWriteDataToStorePath = "TRUE".equalsIgnoreCase(directlyWriteData2Hdfs);	1✔
179
180	if (enableDirectlyWriteDataToStorePath) {	1✔
181	LOGGER.info("Carbondata will directly write fact data to store path.");	1✔
182	} else {
183	LOGGER.info("Carbondata will write temporary fact data to local disk.");	1✔
184	}
185
186	this.executorService = Executors.newFixedThreadPool(1,	1✔
187	new CarbonThreadFactory("CompleteHDFSBackendPool:" + this.model.getTableName()));	1✔
188	executorServiceSubmitList = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);	1✔
189	// in case of compaction we will pass the cardinality.
190	this.localCardinality = this.model.getColCardinality();	1✔
191
192	List<Integer> cardinalityList = new ArrayList<Integer>();	1✔
193	thriftColumnSchemaList = getColumnSchemaListAndCardinality(cardinalityList, localCardinality,	1✔
194	this.model.getWrapperColumnSchema());	1✔
195	this.numberCompressor = new NumberCompressor(Integer.parseInt(CarbonProperties.getInstance()	1✔
196	.getProperty(CarbonCommonConstants.BLOCKLET_SIZE,	1✔
197	CarbonCommonConstants.BLOCKLET_SIZE_DEFAULT_VAL)));
198	this.dataChunksOffsets = new ArrayList<>();	1✔
199	this.dataChunksLength = new ArrayList<>();	1✔
200	blockletMetadata = new ArrayList<BlockletInfo3>();	1✔
201	blockletIndex = new ArrayList<>();	1✔
202	listener = this.model.getDataMapWriterlistener();	1✔
203	if (model.getColumnLocalDictGenMap().size() > 0) {	1✔
204	int numberOfCores = 1;	1✔
205	if (model.getNumberOfCores() > 1) {	1✔
206	numberOfCores = model.getNumberOfCores() / 2;	1✔
207	}
208	fallbackExecutorService = Executors.newFixedThreadPool(numberOfCores, new CarbonThreadFactory(	1✔
209	"FallbackPool:" + model.getTableName() + ", range: " + model.getBucketId()));	1✔
210	}
211	}	1✔
212
213	/**
214	* This method will be used to update the file channel with new file if exceeding block size
215	* threshold, new file will be created once existing file reached the file size limit This
216	* method will first check whether existing file size is exceeded the file
217	* size limit if yes then write the leaf metadata to file then set the
218	* current file size to 0 close the existing file channel get the new file
219	* name and get the channel for new file
220	*
221	* @param blockletSizeToBeAdded data size of one block
222	* @throws CarbonDataWriterException if any problem
223	*/
224	protected void createNewFileIfReachThreshold(long blockletSizeToBeAdded)
225	throws CarbonDataWriterException {
226	if ((currentFileSize + blockletSizeToBeAdded) >= blockSizeThreshold && currentFileSize != 0) {	1✔
227	// set the current file size to zero
228	String activeFile =	1✔
229	enableDirectlyWriteDataToStorePath ? carbonDataFileStorePath : carbonDataFileTempPath;
230	LOGGER.info("Writing data to file as max file size reached for file: "	1✔
231	+ activeFile + ". Data block size: " + currentFileSize);
232	// write meta data to end of the existing file
233	writeFooterToFile();	1✔
234	this.dataChunksOffsets = new ArrayList<>();	1✔
235	this.dataChunksLength = new ArrayList<>();	1✔
236	this.blockletMetadata = new ArrayList<>();	1✔
237	this.blockletIndex = new ArrayList<>();	1✔
238	commitCurrentFile(false);	1✔
239	// initialize the new channel
240	this.currentFileSize = 0;	1✔
241	initializeWriter();	1✔
242	}
243	currentFileSize += blockletSizeToBeAdded;	1✔
244	}	1✔
245
246	private void notifyDataMapBlockStart() {
247	if (listener != null) {	1✔
248	try {
249	listener.onBlockStart(carbonDataFileName);	1✔
250	} catch (IOException e) {	×
251	throw new CarbonDataWriterException("Problem while writing datamap", e);	×
252	}	1✔
253	}
254	}	1✔
255
256	private void notifyDataMapBlockEnd() {
257	if (listener != null) {	1✔
258	try {
259	listener.onBlockEnd(carbonDataFileName);	1✔
260	} catch (IOException e) {	×
261	throw new CarbonDataWriterException("Problem while writing datamap", e);	×
262	}	1✔
263	}
264	}	1✔
265
266	/**
267	* Finish writing current file. It will flush stream, copy and rename temp file to final file
268	* @param copyInCurrentThread set to false if want to do data copy in a new thread
269	*/
270	protected void commitCurrentFile(boolean copyInCurrentThread) {
271	notifyDataMapBlockEnd();	1✔
272	CarbonUtil.closeStreams(this.fileOutputStream, this.fileChannel);	1✔
273	if (!enableDirectlyWriteDataToStorePath) {	1✔
274	try {
275	if (currentFileSize == 0) {	1✔
276	handleEmptyDataFile(carbonDataFileTempPath);	1✔
277	} else {
278	if (copyInCurrentThread) {	1✔
279	CarbonUtil.copyCarbonDataFileToCarbonStorePath(carbonDataFileTempPath,	1✔
280	model.getCarbonDataDirectoryPath(), fileSizeInBytes);	1✔
281	FileFactory.deleteFile(carbonDataFileTempPath,	1✔
282	FileFactory.getFileType(carbonDataFileTempPath));	1✔
283	} else {
284	executorServiceSubmitList	1✔
285	.add(executorService.submit(new CompleteHdfsBackendThread(carbonDataFileTempPath)));	1✔
286	}
287	}
288	} catch (IOException e) {	×
289	LOGGER.error(e);	×
290	}	1✔
291	} else {
292	if (currentFileSize == 0) {	1✔
293	try {
294	handleEmptyDataFile(carbonDataFileStorePath);	×
295	} catch (IOException e) {	×
296	LOGGER.error(e);	×
297	}	×
298	}
299	}
300	}	1✔
301
302	private void handleEmptyDataFile(String filePath) throws IOException {
303	FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath));	1✔
304	if (blockIndexInfoList.size() > 0 && blockIndexInfoList.get(blockIndexInfoList.size() - 1)	1✔
305	.getFileName().equals(carbonDataFileName)) {	×
306	// no need add this entry in index file
307	blockIndexInfoList.remove(blockIndexInfoList.size() - 1);	×
308	// TODO: currently there is no implementation for notifyDataMapBlockEnd(),
309	// hence no impact, once implementation is done. Need to undo it in this case.
310	}
311	}	1✔
312
313	/**
314	* This method will be used to initialize the channel
315	*
316	* @throws CarbonDataWriterException
317	*/
318	public void initializeWriter() throws CarbonDataWriterException {
319	this.carbonDataFileName = CarbonTablePath	1✔
320	.getCarbonDataFileName(fileCount, model.getCarbonDataFileAttributes().getTaskId(),	1✔
321	model.getBucketId(), model.getTaskExtension(),	1✔
322	"" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());	1✔
323	this.carbonDataFileStorePath = model.getCarbonDataDirectoryPath() + File.separator	1✔
324	+ carbonDataFileName;
325	try {
326	if (enableDirectlyWriteDataToStorePath) {	1✔
327	// the block size will be twice the block_size specified by user to make sure that
328	// one carbondata file only consists exactly one HDFS block.
329	fileOutputStream = FileFactory	1✔
330	.getDataOutputStream(carbonDataFileStorePath, FileFactory.FileType.HDFS,	1✔
331	CarbonCommonConstants.BYTEBUFFER_SIZE, fileSizeInBytes * 2);
332	} else {
333	//each time we initialize writer, we choose a local temp location randomly
334	String[] tempFileLocations = model.getStoreLocation();	1✔
335	String chosenTempLocation =	1✔
336	tempFileLocations[new Random().nextInt(tempFileLocations.length)];	1✔
337	LOGGER.info("Randomly choose factdata temp location: " + chosenTempLocation);	1✔
338	carbonDataFileTempPath = chosenTempLocation + File.separator + carbonDataFileName;	1✔
339	fileOutputStream = FileFactory.getDataOutputStream(carbonDataFileTempPath,	1✔
340	FileFactory.FileType.LOCAL, CarbonCommonConstants.BYTEBUFFER_SIZE, true);
341	}
342
343	this.fileCount++;	1✔
344	// open channel for new data file
345	this.fileChannel = Channels.newChannel(fileOutputStream);	1✔
346	this.currentOffsetInFile = 0;	1✔
347	} catch (IOException ex) {	×
348	throw new CarbonDataWriterException(	×
349	"Problem while getting the channel for fact data file", ex);
350	}	1✔
351	notifyDataMapBlockStart();	1✔
352	}	1✔
353
354	/**
355	* This method will write metadata at the end of file file format in thrift format
356	*/
357	protected abstract void writeFooterToFile() throws CarbonDataWriterException;
358
359	/**
360	* Below method will be used to fill the vlock info details
361	*
362	* @param numberOfRows number of rows in file
363	* @param carbonDataFileName The name of carbonData file
364	* @param currentPosition current offset
365	*/
366	protected abstract void fillBlockIndexInfoDetails(long numberOfRows, String carbonDataFileName,
367	long currentPosition);
368
369	public static List<org.apache.carbondata.format.ColumnSchema> getColumnSchemaListAndCardinality(
370	List<Integer> cardinality, int[] dictionaryColumnCardinality,
371	List<ColumnSchema> wrapperColumnSchemaList) {
372	List<org.apache.carbondata.format.ColumnSchema> columnSchemaList =	1✔
373	new ArrayList<org.apache.carbondata.format.ColumnSchema>(
374	CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
375	SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();	1✔
376	int counter = 0;	1✔
377	for (int i = 0; i < wrapperColumnSchemaList.size(); i++) {	1✔
378	columnSchemaList	1✔
379	.add(schemaConverter.fromWrapperToExternalColumnSchema(wrapperColumnSchemaList.get(i)));	1✔
380	if (CarbonUtil.hasEncoding(wrapperColumnSchemaList.get(i).getEncodingList(),	1✔
381	org.apache.carbondata.core.metadata.encoder.Encoding.DICTIONARY)) {
382	cardinality.add(dictionaryColumnCardinality[counter]);	1✔
383	counter++;	1✔
384	} else if (!wrapperColumnSchemaList.get(i).isDimensionColumn()) {	1✔
385	continue;	1✔
386	} else {
387	cardinality.add(-1);	1✔
388	}
389	}
390	return columnSchemaList;	1✔
391	}
392
393	/**
394	* Below method will be used to write the idex file
395	*
396	* @throws IOException throws io exception if any problem while writing
397	* @throws CarbonDataWriterException data writing
398	*/
399	protected void writeIndexFile() throws IOException, CarbonDataWriterException {
400	if (blockIndexInfoList.size() == 0) {	1✔
401	// no need to write index file, if data file is not there.
402	return;	1✔
403	}
404	// get the header
405	IndexHeader indexHeader = CarbonMetadataUtil	1✔
406	.getIndexHeader(localCardinality, thriftColumnSchemaList, model.getBucketId(),	1✔
407	model.getSchemaUpdatedTimeStamp());	1✔
408	// get the block index info thrift
409	List<BlockIndex> blockIndexThrift = CarbonMetadataUtil.getBlockIndexInfo(blockIndexInfoList);	1✔
410	String indexFileName;
411	if (enableDirectlyWriteDataToStorePath) {	1✔
412	String rawFileName = model.getCarbonDataDirectoryPath() + CarbonCommonConstants.FILE_SEPARATOR	1✔
413	+ CarbonTablePath.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),	1✔
414	model.getBucketId(), model.getTaskExtension(),	1✔
415	"" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());	1✔
416	indexFileName = FileFactory.getUpdatedFilePath(rawFileName, FileFactory.FileType.HDFS);	1✔
417	} else {	1✔
418	// randomly choose a temp location for index file
419	String[] tempLocations = model.getStoreLocation();	1✔
420	String chosenTempLocation = tempLocations[new Random().nextInt(tempLocations.length)];	1✔
421	LOGGER.info("Randomly choose index file location: " + chosenTempLocation);	1✔
422	indexFileName = chosenTempLocation + File.separator + CarbonTablePath	1✔
423	.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),	1✔
424	model.getBucketId(), model.getTaskExtension(),	1✔
425	"" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());	1✔
426	}
427
428	CarbonIndexFileWriter writer = new CarbonIndexFileWriter();	1✔
429	// open file
430	writer.openThriftWriter(indexFileName);	1✔
431	// write the header first
432	writer.writeThrift(indexHeader);	1✔
433	// write the indexes
434	for (BlockIndex blockIndex : blockIndexThrift) {	1✔
435	writer.writeThrift(blockIndex);	1✔
436	}	1✔
437	writer.close();	1✔
438	if (!enableDirectlyWriteDataToStorePath) {	1✔
439	CarbonUtil	1✔
440	.copyCarbonDataFileToCarbonStorePath(indexFileName, model.getCarbonDataDirectoryPath(),	1✔
441	fileSizeInBytes);
442	FileFactory.deleteFile(indexFileName, FileFactory.getFileType(indexFileName));	1✔
443	}
444	}	1✔
445
446	/**
447	* This method will close the executor service which is used for copying carbon
448	* data files to carbon store path
449	*
450	* @throws CarbonDataWriterException
451	*/
452	protected void closeExecutorService() throws CarbonDataWriterException {
453	CarbonDataWriterException exception = null;	1✔
454	try {
455	listener.finish();	1✔
456	listener = null;	1✔
457	} catch (IOException e) {	×
458	exception = new CarbonDataWriterException(e);	×
459	}	1✔
460	try {
461	executorService.shutdown();	1✔
462	executorService.awaitTermination(2, TimeUnit.HOURS);	1✔
463	for (int i = 0; i < executorServiceSubmitList.size(); i++) {	1✔
464	executorServiceSubmitList.get(i).get();	1✔
465	}
466	} catch (InterruptedException \| ExecutionException e) {	×
467	if (null == exception) {	×
468	exception = new CarbonDataWriterException(e);	×
469	}
470	}	1✔
471	if (null != fallbackExecutorService) {	1✔
472	fallbackExecutorService.shutdownNow();	1✔
473	}
474	if (exception != null) {	1✔
475	throw exception;	×
476	}
477	}	1✔
478
479
480
481	/**
482	* This method will complete hdfs backend storage for this file.
483	* It may copy the carbon data file from local store location to carbon store location,
484	* it may also complete the remaining replications for the existing hdfs file.
485	*/
486	private final class CompleteHdfsBackendThread implements Callable<Void> {
487
488	/**
489	* complete path along with file name which needs to be copied to
490	* carbon store path
491	*/
492	private String fileName;
493
494	private CompleteHdfsBackendThread(String fileName) {	1✔
495	this.fileName = fileName;	1✔
496	}	1✔
497
498	/**
499	* Computes a result, or throws an exception if unable to do so.
500	*
501	* @return computed result
502	* @throws Exception if unable to compute a result
503	*/
504	@Override
505	public Void call() throws Exception {
506	CarbonUtil.copyCarbonDataFileToCarbonStorePath(fileName, model.getCarbonDataDirectoryPath(),	1✔
507	fileSizeInBytes);
508	FileFactory.deleteFile(fileName, FileFactory.getFileType(fileName));	1✔
509	return null;	1✔
510	}
511	}
512	}

apache / carbondata / 10172

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous