You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by aj...@apache.org on 2020/02/24 08:55:43 UTC
[carbondata] branch master updated: [CARBONDATA-3688] Add
compressor name in data file name
This is an automated email from the ASF dual-hosted git repository.
ajantha 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 66a9d3e [CARBONDATA-3688] Add compressor name in data file name
66a9d3e is described below
commit 66a9d3ea3642dc57674ceeaa3a37e483d06749df
Author: Jacky Li <ja...@qq.com>
AuthorDate: Sat Feb 22 20:27:07 2020 +0800
[CARBONDATA-3688] Add compressor name in data file name
Why is this PR needed?
Currently there is no easy way to tell what compressor is used given one carbondata file unless to read the FileFooter in the file.
What changes were proposed in this PR?
Added compressor name in the carbondata file name, now the file name pattern is
partNo-taskNo-batchNo-bucketNo-segmentNo-timestamp.compressorName.carbondata
Does this PR introduce any user interface change?
No
Is any new testcase added?
Yes
This closes #3606
---
.../indexstore/blockletindex/BlockDataMap.java | 7 +-
.../indexstore/blockletindex/BlockletDataMap.java | 4 +-
.../LatestFilesReadCommittedScope.java | 12 +--
.../statusmanager/SegmentUpdateStatusManager.java | 15 ++-
.../carbondata/core/util/CarbonProperties.java | 4 +
.../apache/carbondata/core/util/CarbonUtil.java | 9 +-
.../carbondata/core/util/path/CarbonTablePath.java | 117 +++++++++++++--------
.../hadoop/api/CarbonFileInputFormat.java | 5 +-
.../hadoop/api/CarbonTableOutputFormat.java | 2 +-
.../describeTable/TestDescribeTable.scala | 9 --
.../table/CarbonDescribeFormattedCommand.scala | 5 -
.../store/writer/AbstractFactDataWriter.java | 3 +-
.../carbondata/sdk/file/CSVCarbonWriterTest.java | 25 +++++
.../org/apache/carbondata/sdk/file/ImageTest.java | 2 +-
.../streaming/CarbonStreamRecordWriter.java | 10 +-
15 files changed, 135 insertions(+), 94 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
index ca061cd..acccf8a 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockDataMap.java
@@ -396,7 +396,8 @@ public class BlockDataMap extends CoarseGrainDataMap
row.setInt((int) fileFooter.getNumberOfRows(), ordinal++);
// add file name
byte[] filePathBytes =
- getFileNameFromPath(filePath).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
+ CarbonTablePath.getCarbonDataFileName(filePath)
+ .getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
row.setByteArray(filePathBytes, ordinal++);
// add version number
row.setShort(fileFooter.getVersionId().number(), ordinal++);
@@ -432,10 +433,6 @@ public class BlockDataMap extends CoarseGrainDataMap
row.setRow(minMaxFlagRow, ordinal);
}
- protected String getFileNameFromPath(String filePath) {
- return CarbonTablePath.getCarbonDataFileName(filePath);
- }
-
protected String getFilePath() {
if (isFilePathStored) {
return getTableTaskInfo(SUMMARY_INDEX_PATH);
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
index 29262fb..9c21748 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/BlockletDataMap.java
@@ -41,6 +41,7 @@ import org.apache.carbondata.core.metadata.blocklet.index.BlockletMinMaxIndex;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.util.BlockletDataMapUtil;
+import org.apache.carbondata.core.util.path.CarbonTablePath;
/**
* Datamap implementation for blocklet.
@@ -168,7 +169,8 @@ public class BlockletDataMap extends BlockDataMap implements Serializable {
row.setInt(blockletInfo.getNumberOfRows(), ordinal++);
// add file name
byte[] filePathBytes =
- getFileNameFromPath(filePath).getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
+ CarbonTablePath.getCarbonDataFileName(filePath)
+ .getBytes(CarbonCommonConstants.DEFAULT_CHARSET_CLASS);
row.setByteArray(filePathBytes, ordinal++);
// add version number
row.setShort(fileFooter.getVersionId().number(), ordinal++);
diff --git a/core/src/main/java/org/apache/carbondata/core/readcommitter/LatestFilesReadCommittedScope.java b/core/src/main/java/org/apache/carbondata/core/readcommitter/LatestFilesReadCommittedScope.java
index bfdbf4c..6975620 100644
--- a/core/src/main/java/org/apache/carbondata/core/readcommitter/LatestFilesReadCommittedScope.java
+++ b/core/src/main/java/org/apache/carbondata/core/readcommitter/LatestFilesReadCommittedScope.java
@@ -213,19 +213,19 @@ public class LatestFilesReadCommittedScope implements ReadCommittedScope {
// Get Segment Name from the IndexFile.
String indexFilePath =
FileFactory.getUpdatedFilePath(carbonIndexFiles[i].getAbsolutePath());
- String segId = getSegmentID(carbonIndexFiles[i].getName(), indexFilePath);
+ String timestamp = getSegmentID(carbonIndexFiles[i].getName(), indexFilePath);
// TODO. During Partition table handling, place Segment File Name.
List<String> indexList;
SegmentRefreshInfo segmentRefreshInfo;
- if (indexFileStore.get(segId) == null) {
+ if (indexFileStore.get(timestamp) == null) {
indexList = new ArrayList<>(1);
segmentRefreshInfo =
new SegmentRefreshInfo(carbonIndexFiles[i].getLastModifiedTime(), 0);
- segmentTimestampUpdaterMap.put(segId, segmentRefreshInfo);
+ segmentTimestampUpdaterMap.put(timestamp, segmentRefreshInfo);
} else {
// Entry is already present.
- indexList = indexFileStore.get(segId);
- segmentRefreshInfo = segmentTimestampUpdaterMap.get(segId);
+ indexList = indexFileStore.get(timestamp);
+ segmentRefreshInfo = segmentTimestampUpdaterMap.get(timestamp);
}
indexList.add(indexFilePath);
if (segmentRefreshInfo.getSegmentUpdatedTimestamp() < carbonIndexFiles[i]
@@ -233,7 +233,7 @@ public class LatestFilesReadCommittedScope implements ReadCommittedScope {
segmentRefreshInfo
.setSegmentUpdatedTimestamp(carbonIndexFiles[i].getLastModifiedTime());
}
- indexFileStore.put(segId, indexList);
+ indexFileStore.put(timestamp, indexList);
segmentRefreshInfo.setCountOfFileInSegment(indexList.size());
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
index 5dbc5ba..2d5500b 100644
--- a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentUpdateStatusManager.java
@@ -405,8 +405,8 @@ public class SegmentUpdateStatusManager {
String firstPart = deltaFilePathName.substring(0, deltaFilePathName.lastIndexOf('.'));
String blockName =
firstPart.substring(0, firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN));
- long timestamp = Long.parseLong(firstPart
- .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1, firstPart.length()));
+ long timestamp =
+ Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(deltaFilePathName));
// It compares whether this delta file belongs to this block or not. And also checks that
// corresponding delta file is valid or not by considering its load start and end time with
// the file timestamp.
@@ -450,8 +450,8 @@ public class SegmentUpdateStatusManager {
&& pathName.getSize() > 0) {
String firstPart = fileName.substring(0, fileName.indexOf('.'));
String blkName = firstPart.substring(0, firstPart.lastIndexOf("-"));
- long timestamp = Long.parseLong(
- firstPart.substring(firstPart.lastIndexOf("-") + 1, firstPart.length()));
+ long timestamp =
+ Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(fileName));
if (blockName.equals(blkName) && (Long.compare(timestamp, deltaEndTimeStamp) <= 0)
&& (Long.compare(timestamp, deltaStartTimestamp) >= 0)) {
return true;
@@ -503,11 +503,8 @@ public class SegmentUpdateStatusManager {
String fileName = eachFile.getName();
if (fileName.endsWith(fileExtension)) {
- String firstPart = fileName.substring(0, fileName.lastIndexOf('.'));
-
- long timestamp = Long.parseLong(firstPart
- .substring(firstPart.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
- firstPart.length()));
+ long timestamp =
+ Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(fileName));
if (excludeOriginalFact) {
if (Long.compare(factTimeStampFinal, timestamp) == 0) {
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
index fc69b43..c6908bb 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonProperties.java
@@ -1962,4 +1962,8 @@ public final class CarbonProperties {
return badRecordHandling.equalsIgnoreCase("true");
}
}
+
+ public String getDefaultCompressor() {
+ return getProperty(CarbonCommonConstants.COMPRESSOR, CarbonCommonConstants.DEFAULT_COMPRESSOR);
+ }
}
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
index d28520a..4949f18 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java
@@ -1586,17 +1586,16 @@ public final class CarbonUtil {
UpdateVO invalidBlockVOForSegmentId, SegmentUpdateStatusManager updateStatusMngr) {
if (!updateStatusMngr.isBlockValid(segmentId,
- CarbonTablePath.getCarbonDataFileName(filePath) + CarbonTablePath
- .getCarbonDataExtension())) {
+ CarbonTablePath.getCarbonDataFileName(filePath) +
+ CarbonTablePath.getCarbonDataExtension())) {
return true;
}
// in case of compaction over si table the factTimeStamp will be null for the
// main table's compacted segments in that case no need to validate the block
if (null != invalidBlockVOForSegmentId &&
null != invalidBlockVOForSegmentId.getFactTimestamp()) {
- Long blockTimeStamp = Long.parseLong(filePath
- .substring(filePath.lastIndexOf('-') + 1,
- filePath.lastIndexOf('.')));
+ long blockTimeStamp =
+ Long.parseLong(CarbonTablePath.DataFileUtil.getTimeStampFromFileName(filePath));
if ((blockTimeStamp > invalidBlockVOForSegmentId.getFactTimestamp() && (
invalidBlockVOForSegmentId.getUpdateDeltaStartTimestamp() != null
&& blockTimeStamp < invalidBlockVOForSegmentId.getUpdateDeltaStartTimestamp()))) {
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 18108f9..c02bdac 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
@@ -18,6 +18,7 @@
package org.apache.carbondata.core.util.path;
import java.io.File;
+import java.util.Objects;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
@@ -26,6 +27,9 @@ import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.locks.LockUsage;
import org.apache.carbondata.core.metadata.ColumnarFormatVersion;
+import static org.apache.carbondata.core.constants.CarbonCommonConstants.DASH;
+import static org.apache.carbondata.core.constants.CarbonCommonConstants.POINT;
+
import org.apache.hadoop.conf.Configuration;
/**
@@ -35,8 +39,6 @@ public class CarbonTablePath {
private static final String METADATA_DIR = "Metadata";
private static final String DICTIONARY_EXT = ".dict";
- private static final String DICTIONARY_META_EXT = ".dictmeta";
- private static final String SORT_INDEX_EXT = ".sortindex";
public static final String SCHEMA_FILE = "schema";
private static final String FACT_DIR = "Fact";
private static final String SEGMENT_PREFIX = "Segment_";
@@ -230,7 +232,7 @@ public class CarbonTablePath {
if (bucketNumber.equals("-1")) {
return file.getName().startsWith(taskId) && file.getName().endsWith(INDEX_FILE_EXT);
}
- return file.getName().startsWith(taskId + "-" + bucketNumber) && file.getName()
+ return file.getName().startsWith(taskId + DASH + bucketNumber) && file.getName()
.endsWith(INDEX_FILE_EXT);
}
});
@@ -270,10 +272,21 @@ public class CarbonTablePath {
private static String getCarbonIndexFileName(String taskNo, int bucketNumber,
String factUpdatedtimeStamp, String segmentNo) {
if (bucketNumber == -1) {
- return taskNo + "-" + segmentNo + "-" + factUpdatedtimeStamp + INDEX_FILE_EXT;
+ return new StringBuilder()
+ .append(taskNo).append(DASH)
+ .append(segmentNo).append(DASH)
+ .append(factUpdatedtimeStamp)
+ .append(INDEX_FILE_EXT)
+ .toString();
+ } else {
+ return new StringBuilder()
+ .append(taskNo).append(DASH)
+ .append(bucketNumber).append(DASH)
+ .append(segmentNo).append(DASH)
+ .append(factUpdatedtimeStamp)
+ .append(INDEX_FILE_EXT)
+ .toString();
}
- return taskNo + "-" + bucketNumber + "-" + segmentNo + "-" + factUpdatedtimeStamp
- + INDEX_FILE_EXT;
}
/**
@@ -285,22 +298,38 @@ public class CarbonTablePath {
}
/**
- * Gets data file name only with out path
- *
- * @param filePartNo data file part number
- * @param taskNo task identifier
- * @param factUpdateTimeStamp unique identifier to identify an update
- * @return gets data file name only with out path
+ * Gets data file name only, without parent path
*/
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;
+ int batchNo, String factUpdateTimeStamp, String segmentNo, String compressor) {
+ Objects.requireNonNull(filePartNo);
+ Objects.requireNonNull(taskNo);
+ Objects.requireNonNull(factUpdateTimeStamp);
+ Objects.requireNonNull(compressor);
+
+ // Start from CarbonData 2.0, the data file name patten is:
+ // partNo-taskNo-batchNo-bucketNo-segmentNo-timestamp.compressor.carbondata
+ // For example:
+ // part-0-0_batchno0-0-0-1580982686749.zstd.carbondata
+ //
+ // If the compressor name is missing, the file is compressed by snappy, which is
+ // the default compressor in CarbonData 1.x
+
+ return new StringBuilder()
+ .append(DATA_PART_PREFIX)
+ .append(filePartNo).append(DASH)
+ .append(taskNo).append(BATCH_PREFIX)
+ .append(batchNo).append(DASH)
+ .append(bucketNumber).append(DASH)
+ .append(segmentNo).append(DASH)
+ .append(factUpdateTimeStamp).append(POINT)
+ .append(compressor).append(CARBON_DATA_EXT)
+ .toString();
}
public static String getShardName(String taskNo, int bucketNumber, int batchNo,
String factUpdateTimeStamp, String segmentNo) {
- return taskNo + BATCH_PREFIX + batchNo + "-" + bucketNumber + "-" + segmentNo + "-"
+ return taskNo + BATCH_PREFIX + batchNo + DASH + bucketNumber + DASH + segmentNo + DASH
+ factUpdateTimeStamp;
}
@@ -409,11 +438,10 @@ public class CarbonTablePath {
* @return
*/
public static Boolean compareCarbonFileTimeStamp(String fileName, Long timestamp) {
- int lastIndexOfHyphen = fileName.lastIndexOf("-");
- int lastIndexOfDot = fileName.lastIndexOf(".");
- if (lastIndexOfHyphen > 0 && lastIndexOfDot > 0) {
- return fileName.substring(fileName.lastIndexOf("-") + 1, fileName.lastIndexOf("."))
- .equals(timestamp.toString());
+ int startIndex = fileName.lastIndexOf(DASH);
+ int endIndex = fileName.indexOf(CarbonCommonConstants.POINT, startIndex);
+ if (startIndex > 0 && endIndex > 0) {
+ return fileName.substring(startIndex + 1, endIndex).equals(timestamp.toString());
}
return false;
}
@@ -422,8 +450,7 @@ public class CarbonTablePath {
* Return the timestamp present in the delete delta file.
*/
public static String getTimeStampFromDeleteDeltaFile(String fileName) {
- return fileName.substring(fileName.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1,
- fileName.lastIndexOf("."));
+ return DataFileUtil.getTimeStampFromFileName(fileName);
}
/**
@@ -441,12 +468,12 @@ public class CarbonTablePath {
// Get the file name from path
String fileName = getFileName(carbonFilePath);
// + 1 for size of "-"
- int firstDashPos = fileName.indexOf("-");
- int secondDash = fileName.indexOf("-", firstDashPos + 1);
- int startIndex = fileName.indexOf("-", secondDash + 1) + 1;
- int endIndex = fileName.indexOf("-", startIndex);
+ int firstDashPos = fileName.indexOf(DASH);
+ int secondDash = fileName.indexOf(DASH, firstDashPos + 1);
+ int startIndex = fileName.indexOf(DASH, secondDash + 1) + 1;
+ int endIndex = fileName.indexOf(DASH, startIndex);
// to support backward compatibility
- if (startIndex == -1 || endIndex == -1) {
+ if (endIndex == -1) {
return "-1";
}
return fileName.substring(startIndex, endIndex);
@@ -459,8 +486,8 @@ public class CarbonTablePath {
// Get the file name from path
String fileName = getFileName(carbonDataFileName);
// + 1 for size of "-"
- int startIndex = fileName.indexOf("-") + 1;
- int endIndex = fileName.indexOf("-", startIndex);
+ int startIndex = fileName.indexOf(DASH) + 1;
+ int endIndex = fileName.indexOf(DASH, startIndex);
return fileName.substring(startIndex, endIndex);
}
@@ -471,9 +498,9 @@ public class CarbonTablePath {
// Get the file name from path
String fileName = getFileName(carbonDataFileName);
// + 1 for size of "-"
- int firstDashPos = fileName.indexOf("-");
- int startIndex = fileName.indexOf("-", firstDashPos + 1) + 1;
- int endIndex = fileName.indexOf("-", startIndex);
+ int firstDashPos = fileName.indexOf(DASH);
+ int startIndex = fileName.indexOf(DASH, firstDashPos + 1) + 1;
+ int endIndex = fileName.indexOf(DASH, startIndex);
return fileName.substring(startIndex, endIndex);
}
@@ -491,13 +518,13 @@ public class CarbonTablePath {
// Get the file name from path
String fileName = getFileName(carbonDataFileName);
// + 1 for size of "-"
- int firstDashPos = fileName.indexOf("-");
- int startIndex1 = fileName.indexOf("-", firstDashPos + 1) + 1;
- int endIndex1 = fileName.indexOf("-", startIndex1);
- int startIndex = fileName.indexOf("-", endIndex1 + 1);
+ int firstDashPos = fileName.indexOf(DASH);
+ int startIndex1 = fileName.indexOf(DASH, firstDashPos + 1) + 1;
+ int endIndex1 = fileName.indexOf(DASH, startIndex1);
+ int startIndex = fileName.indexOf(DASH, endIndex1 + 1);
if (startIndex > -1) {
startIndex += 1;
- int endIndex = fileName.indexOf("-", startIndex);
+ int endIndex = fileName.indexOf(DASH, startIndex);
if (endIndex == -1) {
return null;
}
@@ -546,7 +573,7 @@ public class CarbonTablePath {
CarbonCommonConstants.FILE_SEPARATOR, endIndex - 1) + 1;
String segmentDirStr = dataFileAbsolutePath.substring(startIndex, endIndex);
//identify id in segment_<id>
- String[] segmentDirSplits = segmentDirStr.split("_");
+ String[] segmentDirSplits = segmentDirStr.split(CarbonCommonConstants.UNDERSCORE);
try {
if (segmentDirSplits.length == 2) {
return segmentDirSplits[1];
@@ -559,7 +586,7 @@ public class CarbonTablePath {
}
/**
- * Return the carbondata file name
+ * Return the carbondata file name without extension name
*/
public static String getCarbonDataFileName(String carbonDataFilePath) {
return carbonDataFilePath.substring(
@@ -659,15 +686,15 @@ public class CarbonTablePath {
String segmentNoStr = DataFileUtil.getSegmentNo(actualBlockName);
StringBuilder shardName = new StringBuilder();
if (null != segmentNoStr) {
- shardName.append(DataFileUtil.getTaskNo(actualBlockName)).append("-");
- shardName.append(DataFileUtil.getBucketNo(actualBlockName)).append("-");
- shardName.append(segmentNoStr).append("-");
+ shardName.append(DataFileUtil.getTaskNo(actualBlockName)).append(DASH);
+ shardName.append(DataFileUtil.getBucketNo(actualBlockName)).append(DASH);
+ shardName.append(segmentNoStr).append(DASH);
shardName.append(DataFileUtil.getTimeStampFromFileName(actualBlockName));
return shardName.toString();
} else {
// data before version 1.4 does not have SegmentNo in carbondata filename
- shardName.append(DataFileUtil.getTaskNo(actualBlockName)).append("-");
- shardName.append(DataFileUtil.getBucketNo(actualBlockName)).append("-");
+ shardName.append(DataFileUtil.getTaskNo(actualBlockName)).append(DASH);
+ shardName.append(DataFileUtil.getBucketNo(actualBlockName)).append(DASH);
shardName.append(DataFileUtil.getTimeStampFromFileName(actualBlockName));
return shardName.toString();
}
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
index 6ab91cb..002ab7b 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonFileInputFormat.java
@@ -146,8 +146,9 @@ public class CarbonFileInputFormat<T> extends CarbonInputFormat<T> implements Se
seg = new Segment(load.getLoadName(), null, readCommittedScope);
if (fileLists != null) {
for (int i = 0; i < fileLists.size(); i++) {
- if (fileLists.get(i).toString().endsWith(seg.getSegmentNo()
- + CarbonTablePath.CARBON_DATA_EXT)) {
+ String timestamp =
+ CarbonTablePath.DataFileUtil.getTimeStampFromFileName(fileLists.get(i).toString());
+ if (timestamp.equals(seg.getSegmentNo())) {
externalTableSegments.add(seg);
break;
}
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableOutputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableOutputFormat.java
index d42b627..2f6b4c7 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableOutputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableOutputFormat.java
@@ -474,7 +474,7 @@ public class CarbonTableOutputFormat extends FileOutputFormat<NullWritable, Obje
future.get();
} catch (ExecutionException e) {
LOG.error("Error while loading data", e);
- throw new InterruptedException(e.getMessage());
+ throw new RuntimeException(e);
} finally {
executorService.shutdownNow();
dataLoadExecutor.close();
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/describeTable/TestDescribeTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/describeTable/TestDescribeTable.scala
index dee2295..18fc953 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/describeTable/TestDescribeTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/describeTable/TestDescribeTable.scala
@@ -68,15 +68,6 @@ class TestDescribeTable extends QueryTest with BeforeAndAfterAll {
assert(descPar.exists(_.toString().contains("Partition Parameters:")))
}
- test(testName = "Compressor Type update from carbon properties") {
- sql("drop table if exists b")
- sql(sqlText = "create table b(a int,b string) STORED AS carbondata")
- CarbonProperties.getInstance().addProperty(CarbonCommonConstants.COMPRESSOR, "gzip")
- val result = sql(sqlText = "desc formatted b").collect()
- assert(result.filter(row => row.getString(0).contains("Data File Compressor")).head.getString
- (1).equalsIgnoreCase("gzip"))
- }
-
override def afterAll: Unit = {
sql("DROP TABLE Desc1")
sql("DROP TABLE Desc2")
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDescribeFormattedCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDescribeFormattedCommand.scala
index 5f54b32..ce240c7 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDescribeFormattedCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDescribeFormattedCommand.scala
@@ -141,11 +141,6 @@ private[sql] case class CarbonDescribeFormattedCommand(
Strings.formatSize(
tblProps.getOrElse(CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB,
CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB_DEFAULT).toFloat), ""),
- ("Data File Compressor ", tblProps
- .getOrElse(CarbonCommonConstants.COMPRESSOR,
- CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.COMPRESSOR,
- CarbonCommonConstants.DEFAULT_COMPRESSOR)), ""),
//////////////////////////////////////////////////////////////////////////////
// Index Information
//////////////////////////////////////////////////////////////////////////////
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
index ad54728..17c47bd 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/writer/AbstractFactDataWriter.java
@@ -308,7 +308,8 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
this.carbonDataFileName = CarbonTablePath
.getCarbonDataFileName(fileCount, model.getCarbonDataFileAttributes().getTaskId(),
model.getBucketId(), model.getTaskExtension(),
- "" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());
+ "" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId(),
+ model.getColumnCompressor());
this.carbonDataFileStorePath = model.getCarbonDataDirectoryPath() + File.separator
+ carbonDataFileName;
try {
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 533480c..f693ef5 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
@@ -89,6 +89,31 @@ public class CSVCarbonWriterTest {
FileUtils.deleteDirectory(new File(path));
}
+ // [CARBONDATA-3688]: compressor name is added in data file name
+ @Test
+ public void testFileName() throws IOException {
+ String path = "./testWriteFiles";
+ FileUtils.deleteDirectory(new File(path));
+
+ Field[] fields = new Field[2];
+ fields[0] = new Field("name", DataTypes.STRING);
+ fields[1] = new Field("age", DataTypes.INT);
+
+ TestUtil.writeFilesAndVerify(new Schema(fields), path);
+
+ File[] dataFiles = new File(path).listFiles(new FileFilter() {
+ @Override
+ public boolean accept(File pathname) {
+ return pathname.getName().endsWith(
+ CarbonCommonConstants.DEFAULT_COMPRESSOR + CarbonCommonConstants.FACT_FILE_EXT);
+ }
+ });
+
+ Assert.assertTrue(dataFiles.length > 0);
+
+ FileUtils.deleteDirectory(new File(path));
+ }
+
@Test
public void testWriteFilesJsonSchema() throws IOException {
String path = "./testWriteFilesJsonSchema";
diff --git a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/ImageTest.java b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/ImageTest.java
index 3a64377..631f9ab 100644
--- a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/ImageTest.java
+++ b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/ImageTest.java
@@ -1066,7 +1066,7 @@ public class ImageTest extends TestCase {
.builder()
.withFileLists(fileLists)
.getSplits(true);
- Assert.assertTrue(5 == splits.length);
+ Assert.assertEquals(5, splits.length);
for (int j = 0; j < splits.length; j++) {
ArrowCarbonReader.builder(splits[j]).build();
}
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 22bc1ee..3209e8d 100644
--- a/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
+++ b/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
@@ -26,7 +26,6 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.datastore.row.CarbonRow;
@@ -38,6 +37,7 @@ import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
import org.apache.carbondata.core.reader.CarbonHeaderReader;
import org.apache.carbondata.core.util.ByteUtil;
import org.apache.carbondata.core.util.CarbonMetadataUtil;
+import org.apache.carbondata.core.util.CarbonProperties;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.core.util.path.CarbonTablePath;
@@ -139,7 +139,9 @@ 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, CarbonProperties.getInstance().getDefaultCompressor());
// initialize metadata
isNoDictionaryDimensionColumn =
@@ -178,7 +180,7 @@ public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
if (header.isSetCompressor_name()) {
compressorName = header.getCompressor_name();
} else {
- compressorName = CompressorFactory.NativeSupportedCompressor.SNAPPY.getName();
+ compressorName = CarbonProperties.getInstance().getDefaultCompressor();
}
} else {
// IF the file is not existed, use the create api
@@ -186,7 +188,7 @@ public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
compressorName = carbonTable.getTableInfo().getFactTable().getTableProperties().get(
CarbonCommonConstants.COMPRESSOR);
if (null == compressorName) {
- compressorName = CompressorFactory.getInstance().getCompressor().getName();
+ compressorName = CarbonProperties.getInstance().getDefaultCompressor();
}
writeFileHeader();
}