You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ku...@apache.org on 2019/07/22 11:00:36 UTC
[carbondata] branch master updated: [HOTFIX] Added taskid as UUID
while writing files in fileformat to avoid corrupting.
This is an automated email from the ASF dual-hosted git repository.
kunalkapoor pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new b6e5f69 [HOTFIX] Added taskid as UUID while writing files in fileformat to avoid corrupting.
b6e5f69 is described below
commit b6e5f6914fe64bac2d49dddd90aba2bb93c346a7
Author: ravipesala <ra...@gmail.com>
AuthorDate: Sat Jul 13 16:13:53 2019 +0530
[HOTFIX] Added taskid as UUID while writing files in fileformat to avoid corrupting.
Problem: In FIleFormat write, carbon is using task id as System.nanoTime()
when multiple tasks launched concurrently, there is a chance that two task
can have same id very rarely. Due to this, two spark task launched for one
insert will have same carbondata file name. so, when both tasks write to one file,
chances are more to corrupt the file. which leads in query failure
solution: use unique uuid task id instead of nano seconds.
This closes #3325
---
.../apache/carbondata/core/util/path/CarbonTablePath.java | 12 ++++++------
.../apache/carbondata/hadoop/api/CarbonInputFormat.java | 4 ++--
.../execution/datasources/SparkCarbonFileFormat.scala | 4 +++-
.../processing/partition/spliter/RowResultProcessor.java | 2 +-
.../processing/store/CarbonDataFileAttributes.java | 15 ++++++++++++---
.../processing/store/CarbonFactDataHandlerModel.java | 2 +-
.../apache/carbondata/sdk/file/CSVCarbonWriterTest.java | 4 ++--
.../carbondata/streaming/CarbonStreamRecordWriter.java | 2 +-
8 files changed, 28 insertions(+), 17 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
index 422a6dc..055cdee 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
@@ -305,13 +305,13 @@ public class CarbonTablePath {
* @param factUpdateTimeStamp unique identifier to identify an update
* @return gets data file name only with out path
*/
- public static String getCarbonDataFileName(Integer filePartNo, Long taskNo, int bucketNumber,
+ public static String getCarbonDataFileName(Integer filePartNo, String taskNo, int bucketNumber,
int batchNo, String factUpdateTimeStamp, String segmentNo) {
return DATA_PART_PREFIX + filePartNo + "-" + taskNo + BATCH_PREFIX + batchNo + "-"
+ bucketNumber + "-" + segmentNo + "-" + factUpdateTimeStamp + CARBON_DATA_EXT;
}
- public static String getShardName(Long taskNo, int bucketNumber, int batchNo,
+ public static String getShardName(String taskNo, int bucketNumber, int batchNo,
String factUpdateTimeStamp, String segmentNo) {
return taskNo + BATCH_PREFIX + batchNo + "-" + bucketNumber + "-" + segmentNo + "-"
+ factUpdateTimeStamp;
@@ -324,14 +324,14 @@ public class CarbonTablePath {
* @param factUpdatedTimeStamp time stamp
* @return filename
*/
- public static String getCarbonIndexFileName(long taskNo, int bucketNumber, int batchNo,
+ public static String getCarbonIndexFileName(String taskNo, int bucketNumber, int batchNo,
String factUpdatedTimeStamp, String segmentNo) {
return getShardName(taskNo, bucketNumber, batchNo, factUpdatedTimeStamp, segmentNo)
+ INDEX_FILE_EXT;
}
public static String getCarbonStreamIndexFileName() {
- return getCarbonIndexFileName(0, 0, 0, "0", "0");
+ return getCarbonIndexFileName("0", 0, 0, "0", "0");
}
public static String getCarbonStreamIndexFilePath(String segmentDir) {
@@ -516,8 +516,8 @@ public class CarbonTablePath {
/**
* Return the taskId part from taskNo(include taskId + batchNo)
*/
- public static long getTaskIdFromTaskNo(String taskNo) {
- return Long.parseLong(taskNo.split(BATCH_PREFIX)[0]);
+ public static String getTaskIdFromTaskNo(String taskNo) {
+ return taskNo.split(BATCH_PREFIX)[0];
}
/**
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
index 5359f13..ac9e11e 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonInputFormat.java
@@ -482,8 +482,8 @@ m filterExpression
// partition info first and then read data.
// For other normal query should use newest partitionIdList
if (partitionInfo != null && partitionInfo.getPartitionType() != PartitionType.NATIVE_HIVE) {
- long partitionId = CarbonTablePath.DataFileUtil
- .getTaskIdFromTaskNo(CarbonTablePath.DataFileUtil.getTaskNo(blocklet.getPath()));
+ long partitionId = Long.parseLong(CarbonTablePath.DataFileUtil
+ .getTaskIdFromTaskNo(CarbonTablePath.DataFileUtil.getTaskNo(blocklet.getPath())));
if (oldPartitionIdList != null) {
partitionIndex = oldPartitionIdList.indexOf((int) partitionId);
} else {
diff --git a/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala
index 185058c..25296a0 100644
--- a/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala
+++ b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.carbondata.execution.datasources
import java.net.URI
+import java.util.UUID
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
@@ -153,7 +154,8 @@ class SparkCarbonFileFormat extends FileFormat
path
}
context.getConfiguration.set("carbon.outputformat.writepath", updatedPath)
- context.getConfiguration.set("carbon.outputformat.taskno", System.nanoTime() + "")
+ context.getConfiguration.set("carbon.outputformat.taskno",
+ UUID.randomUUID().toString.replace("-", ""))
new CarbonOutputWriter(path, context, dataSchema.fields)
}
diff --git a/processing/src/main/java/org/apache/carbondata/processing/partition/spliter/RowResultProcessor.java b/processing/src/main/java/org/apache/carbondata/processing/partition/spliter/RowResultProcessor.java
index d7f23fb..75aadd1 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/partition/spliter/RowResultProcessor.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/partition/spliter/RowResultProcessor.java
@@ -57,7 +57,7 @@ public class RowResultProcessor {
CarbonFactDataHandlerModel.getCarbonFactDataHandlerModel(loadModel, carbonTable,
segProp, tableName, tempStoreLocation, carbonStoreLocation);
CarbonDataFileAttributes carbonDataFileAttributes =
- new CarbonDataFileAttributes(Long.parseLong(loadModel.getTaskNo()),
+ new CarbonDataFileAttributes(loadModel.getTaskNo(),
loadModel.getFactTimeStamp());
carbonFactDataHandlerModel.setCarbonDataFileAttributes(carbonDataFileAttributes);
carbonFactDataHandlerModel.setBucketId(bucketId);
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
index 7861b52..d43ed94 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonDataFileAttributes.java
@@ -25,7 +25,7 @@ public class CarbonDataFileAttributes {
/**
* task Id which is unique for each spark task
*/
- private long taskId;
+ private String taskId;
/**
* load start time
@@ -36,15 +36,24 @@ public class CarbonDataFileAttributes {
* @param taskId
* @param factTimeStamp
*/
- public CarbonDataFileAttributes(long taskId, long factTimeStamp) {
+ public CarbonDataFileAttributes(String taskId, long factTimeStamp) {
this.taskId = taskId;
this.factTimeStamp = factTimeStamp;
}
/**
+ * @param taskId
+ * @param factTimeStamp
+ */
+ public CarbonDataFileAttributes(long taskId, long factTimeStamp) {
+ this.taskId = String.valueOf(taskId);
+ this.factTimeStamp = factTimeStamp;
+ }
+
+ /**
* @return
*/
- public long getTaskId() {
+ public String getTaskId() {
return taskId;
}
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
index 726d906..39a83d9 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -260,7 +260,7 @@ public class CarbonFactDataHandlerModel {
}
}
CarbonDataFileAttributes carbonDataFileAttributes =
- new CarbonDataFileAttributes(Long.parseLong(configuration.getTaskNo()),
+ new CarbonDataFileAttributes(configuration.getTaskNo(),
(Long) configuration.getDataLoadProperty(DataLoadProcessorConstants.FACT_TIME_STAMP));
String carbonDataDirectoryPath = getCarbonDataFolderLocation(configuration);
diff --git a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CSVCarbonWriterTest.java b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CSVCarbonWriterTest.java
index 1452058..5f25d09 100644
--- a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CSVCarbonWriterTest.java
+++ b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CSVCarbonWriterTest.java
@@ -388,8 +388,8 @@ public class CSVCarbonWriterTest {
Assert.assertNotNull(dataFiles);
Assert.assertTrue(dataFiles.length > 0);
String taskNo = CarbonTablePath.DataFileUtil.getTaskNo(dataFiles[0].getName());
- long taskID = CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(taskNo);
- Assert.assertEquals("Task Id is not matched", taskID, 5);
+ String taskID = CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(taskNo);
+ Assert.assertEquals("Task Id is not matched", taskID, "5");
} catch (Exception e) {
e.printStackTrace();
Assert.fail(e.getMessage());
diff --git a/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java b/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
index 2a95935..5ef5ab9 100644
--- a/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
+++ b/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
@@ -139,7 +139,7 @@ public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
segmentDir = CarbonTablePath.getSegmentPath(
carbonTable.getAbsoluteTableIdentifier().getTablePath(), segmentId);
- fileName = CarbonTablePath.getCarbonDataFileName(0, taskNo, 0, 0, "0", segmentId);
+ fileName = CarbonTablePath.getCarbonDataFileName(0, taskNo + "", 0, 0, "0", segmentId);
// initialize metadata
isNoDictionaryDimensionColumn =