Github user jackylk commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/1825#discussion_r171454525
--- Diff:
processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
---
@@ -386,29 +380,42 @@ protected void writeIndexFile() throws IOException,
CarbonDataWriterException {
.getIndexHeader(localCardinality, thriftColumnSchemaList,
model.getBucketId());
// get the block index info thrift
List<BlockIndex> blockIndexThrift =
CarbonMetadataUtil.getBlockIndexInfo(blockIndexInfoList);
- // 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);
+ String indexFileName;
+ if (enableDirectlyWriteData2Hdfs) {
+ String rawFileName = model.getCarbonDataDirectoryPath() +
File.separator + CarbonTablePath
+
.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),
+ model.getBucketId(), model.getTaskExtension(),
+ "" + model.getCarbonDataFileAttributes().getFactTimeStamp());
+ 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());
+ }
- String fileName = chosenTempLocation + File.separator + CarbonTablePath
-
.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),
- model.getBucketId(), model.getTaskExtension(),
- "" + model.getCarbonDataFileAttributes().getFactTimeStamp());
CarbonIndexFileWriter writer = new CarbonIndexFileWriter();
// open file
- writer.openThriftWriter(fileName);
+ writer.openThriftWriter(indexFileName);
// write the header first
writer.writeThrift(indexHeader);
// write the indexes
for (BlockIndex blockIndex : blockIndexThrift) {
writer.writeThrift(blockIndex);
}
writer.close();
- // copy from temp to actual store location
- CarbonUtil.copyCarbonDataFileToCarbonStorePath(fileName,
- model.getCarbonDataDirectoryPath(),
- fileSizeInBytes);
+ if (enableDirectlyWriteData2Hdfs) {
+ executorServiceSubmitList.add(executorService.submit(
+ new CopyThread(indexFileName, FileFactory.FileType.HDFS)));
--- End diff --
The name of CopyThread is confusing, is it copy or rename?
---
