You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2019/11/29 09:15:06 UTC
[carbondata] branch master updated: [CARBONDATA-2863] Refactored
CarbonFile interface
This is an automated email from the ASF dual-hosted git repository.
jackylk 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 83174b7 [CARBONDATA-2863] Refactored CarbonFile interface
83174b7 is described below
commit 83174b781645429abe587f6d51a0fbc3c5436eb1
Author: kunal642 <ku...@gmail.com>
AuthorDate: Thu Aug 16 10:15:44 2018 +0530
[CARBONDATA-2863] Refactored CarbonFile interface
Refactor the CarbonFileInterface and the implementations to make the code clean, reduce calls to file system (like listStatus API) and throw proper exceptions.
This closes #2465
---
.../cache/dictionary/AbstractDictionaryCache.java | 3 +-
.../cache/dictionary/ManageDictionaryAndBTree.java | 6 +-
.../carbondata/core/datamap/TableDataMap.java | 2 +-
.../core/datamap/dev/BlockletSerializer.java | 4 +-
.../carbondata/core/datamap/dev/DataMapWriter.java | 2 +-
.../status/DiskBasedDataMapStatusProvider.java | 3 +-
.../filesystem/AbstractDFSCarbonFile.java | 434 +++++++++------------
.../datastore/filesystem/AlluxioCarbonFile.java | 112 +-----
.../core/datastore/filesystem/CarbonFile.java | 65 +--
.../core/datastore/filesystem/HDFSCarbonFile.java | 38 +-
.../core/datastore/filesystem/LocalCarbonFile.java | 218 ++++-------
.../core/datastore/filesystem/S3CarbonFile.java | 56 +--
.../datastore/filesystem/ViewFSCarbonFile.java | 26 +-
.../datastore/impl/DefaultFileTypeProvider.java | 4 +-
.../core/datastore/impl/FileFactory.java | 142 +++----
.../CarbonFileException.java} | 20 +-
.../fileoperations/AtomicFileOperationFactory.java | 2 +-
.../fileoperations/AtomicFileOperationS3Impl.java | 5 +-
.../fileoperations/AtomicFileOperationsImpl.java | 20 +-
.../core/indexstore/ExtendedBlockletWrapper.java | 5 +-
.../blockletindex/SegmentIndexFileStore.java | 2 +-
.../carbondata/core/locks/AbstractCarbonLock.java | 2 +-
.../apache/carbondata/core/locks/HdfsFileLock.java | 7 +-
.../carbondata/core/locks/LocalFileLock.java | 4 +-
.../apache/carbondata/core/locks/S3FileLock.java | 7 +-
.../carbondata/core/metadata/SegmentFileStore.java | 22 +-
.../table/DiskBasedDMSchemaStorageProvider.java | 17 +-
.../carbondata/core/mutate/CarbonUpdateUtil.java | 9 +-
.../reader/CarbonDeleteDeltaFileReaderImpl.java | 11 +-
.../core/reader/CarbonDeleteFilesDataReader.java | 3 +-
.../carbondata/core/reader/ThriftReader.java | 3 +-
.../CarbonDictionarySortIndexReaderImpl.java | 3 +-
.../core/statusmanager/SegmentStatusManager.java | 10 +-
.../statusmanager/SegmentUpdateStatusManager.java | 6 +-
.../carbondata/core/stream/StreamPruner.java | 3 +-
.../carbondata/core/util/BlockletDataMapUtil.java | 2 +-
.../apache/carbondata/core/util/CarbonUtil.java | 44 +--
.../carbondata/core/util/DeleteLoadFolders.java | 4 +-
.../carbondata/core/util/path/CarbonTablePath.java | 2 +-
.../core/writer/CarbonDeleteDeltaWriterImpl.java | 10 +-
.../core/writer/CarbonDictionaryWriterImpl.java | 5 +-
.../carbondata/core/writer/ThriftWriter.java | 3 +-
.../CarbonDictionarySortIndexWriterImpl.java | 2 +-
.../dictionary/AbstractDictionaryCacheTest.java | 3 +-
.../store/impl/FileFactoryImplUnitTest.java | 34 +-
.../filesystem/store/impl/TestFileProvider.java | 4 +-
.../core/datastore/CompressdFileTest.java | 2 +-
.../filesystem/AlluxioCarbonFileTest.java | 23 +-
.../datastore/filesystem/HDFSCarbonFileTest.java | 31 +-
.../datastore/filesystem/LocalCarbonFileTest.java | 12 +-
.../datastore/filesystem/ViewFsCarbonFileTest.java | 35 +-
.../CarbonDictionarySortIndexReaderImplTest.java | 3 +-
.../carbondata/core/util/CarbonUtilTest.java | 4 +-
.../core/util/DataFileFooterConverterTest.java | 2 +-
.../writer/CarbonDictionaryWriterImplTest.java | 8 +-
.../datamap/bloom/AbstractBloomDataMapWriter.java | 7 +-
.../bloom/BloomCoarseGrainDataMapFactory.java | 3 +-
.../datamap/bloom/BloomIndexFileStore.java | 24 +-
.../datamap/examples/MinMaxDataWriter.java | 4 +-
.../datamap/examples/MinMaxIndexDataMap.java | 2 +-
.../datamap/lucene/LuceneDataMapFactoryBase.java | 6 +-
.../apache/carbondata/examples/HiveExample.scala | 4 +-
.../examples/util/AllDictionaryUtil.scala | 8 +-
.../hadoop/api/CarbonFileInputFormat.java | 2 +-
.../carbondata/hadoop/testutil/StoreCreator.java | 5 +-
.../PrestoAllDataTypeLocalDictTest.scala | 5 +-
.../integrationtest/PrestoAllDataTypeTest.scala | 4 +-
.../PrestoTestNonTransactionalTableFiles.scala | 4 +-
.../presto/util/CarbonDataStoreCreator.scala | 6 +-
...teTableUsingSparkCarbonFileFormatTestCase.scala | 2 +-
.../testsuite/dataload/TestLoadDataGeneral.scala | 3 -
.../blockprune/BlockPruneQueryTestCase.scala | 8 +-
.../CarbonCustomBlockDistributionTest.scala | 8 +-
...bonFileInputFormatWithExternalCarbonTable.scala | 4 +-
.../TestNonTransactionalCarbonTable.scala | 11 +-
.../testsuite/datamap/CGDataMapTestCase.scala | 4 +-
.../testsuite/datamap/FGDataMapTestCase.scala | 6 +-
.../SubqueryWithFilterAndSortTestCase.scala | 6 +-
.../ValueCompressionDataTypeTestCase.scala | 7 +-
.../iud/HorizontalCompactionTestCase.scala | 2 +-
.../TestDataLoadingForPartitionTable.scala | 2 +-
.../org/apache/carbondata/api/CarbonStore.scala | 13 +-
.../carbondata/spark/DictionaryDetailHelper.scala | 3 +-
.../spark/rdd/AlterTableAddColumnRDD.scala | 5 +-
.../spark/rdd/CarbonGlobalDictionaryRDD.scala | 6 +-
.../carbondata/spark/rdd/StreamHandoffRDD.scala | 5 +-
.../apache/carbondata/spark/util/CommonUtil.scala | 7 +-
.../spark/util/GlobalDictionaryUtil.scala | 5 +-
.../carbondata/streaming/StreamSinkFactory.scala | 9 +-
.../streaming/CarbonAppendableStreamSink.scala | 6 +-
.../spark/sql/test/ResourceRegisterAndCopier.scala | 13 +-
.../apache/spark/sql/test/TestQueryExecutor.scala | 10 +-
.../scala/org/apache/spark/util/FileUtils.scala | 3 +-
...TestCreateTableUsingSparkCarbonFileFormat.scala | 3 +-
.../datamap/IndexDataMapRebuildRDD.scala | 2 +-
.../carbondata/indexserver/DataMapJobs.scala | 4 +-
.../spark/rdd/CarbonDataRDDFactory.scala | 2 +-
.../command/management/CarbonAddLoadCommand.scala | 2 +-
.../command/management/CarbonLoadDataCommand.scala | 10 +-
.../management/RefreshCarbonTableCommand.scala | 4 +-
.../command/mutation/DeleteExecution.scala | 3 +-
.../command/table/CarbonDropTableCommand.scala | 6 +-
.../spark/sql/hive/CarbonFileMetastore.scala | 10 +-
.../org/apache/spark/sql/hive/CarbonRelation.scala | 6 +-
.../sql/parser/CarbonSparkSqlParserUtil.scala | 2 +-
.../org/apache/spark/util/AlterTableUtil.scala | 3 +-
.../partition/TestAlterPartitionTable.scala | 2 +-
.../spark/util/AllDictionaryTestCase.scala | 5 +-
.../util/ExternalColumnDictionaryTestCase.scala | 5 +-
.../carbondata/BadRecordPathLoadOptionTest.scala | 3 +-
.../processing/loading/BadRecordsLogger.java | 19 +-
.../loading/TableProcessingOperations.java | 5 +-
.../loading/sort/unsafe/UnsafeSortDataRows.java | 2 +-
.../holder/UnsafeSortTempFileChunkHolder.java | 2 +-
.../UnsafeInMemoryIntermediateDataMerger.java | 3 +-
.../merger/UnsafeIntermediateFileMerger.java | 4 +-
.../processing/merger/CarbonCompactionUtil.java | 20 +-
.../processing/merger/CarbonDataMergerUtil.java | 7 +-
.../sort/sortdata/IntermediateFileMerger.java | 2 +-
.../processing/sort/sortdata/SortDataRows.java | 2 +-
.../sort/sortdata/SortTempFileChunkHolder.java | 2 +-
.../store/writer/AbstractFactDataWriter.java | 20 +-
.../processing/util/CarbonBadRecordUtil.java | 5 +-
.../processing/util/CarbonLoaderUtil.java | 18 +-
.../carbondata/lcm/locks/LocalFileLockTest.java | 6 +-
.../carbondata/sdk/file/CarbonSchemaReader.java | 2 +-
.../carbondata/store/MetaCachedCarbonStore.java | 2 +-
.../carbondata/sdk/file/CarbonReaderTest.java | 8 +-
.../streaming/CarbonStreamRecordWriter.java | 7 +-
.../streaming/segment/StreamSegment.java | 40 +-
130 files changed, 773 insertions(+), 1216 deletions(-)
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
index 23b98a6..7eef6b7 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCache.java
@@ -144,8 +144,7 @@ public abstract class AbstractDictionaryCache<K extends DictionaryColumnUniqueId
private CarbonFile getDictionaryMetaCarbonFile(
DictionaryColumnUniqueIdentifier dictionaryColumnUniqueIdentifier) throws IOException {
String dictionaryFilePath = dictionaryColumnUniqueIdentifier.getDictionaryFilePath();
- FileFactory.FileType fileType = FileFactory.getFileType(dictionaryFilePath);
- CarbonFile dictFile = FileFactory.getCarbonFile(dictionaryFilePath, fileType);
+ CarbonFile dictFile = FileFactory.getCarbonFile(dictionaryFilePath);
// When rename table triggered parallely with select query, dictionary files may not exist
if (!dictFile.exists()) {
throw new IOException("Dictionary file does not exist: " + dictionaryFilePath);
diff --git a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ManageDictionaryAndBTree.java b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ManageDictionaryAndBTree.java
index 571e737..98ae012 100644
--- a/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ManageDictionaryAndBTree.java
+++ b/core/src/main/java/org/apache/carbondata/core/cache/dictionary/ManageDictionaryAndBTree.java
@@ -58,8 +58,7 @@ public class ManageDictionaryAndBTree {
public static void deleteDictionaryFileAndCache(final ColumnSchema columnSchema,
AbsoluteTableIdentifier identifier) {
String metadataDirectoryPath = CarbonTablePath.getMetadataPath(identifier.getTablePath());
- CarbonFile metadataDir = FileFactory
- .getCarbonFile(metadataDirectoryPath, FileFactory.getFileType(metadataDirectoryPath));
+ CarbonFile metadataDir = FileFactory.getCarbonFile(metadataDirectoryPath);
if (metadataDir.exists()) {
// sort index file is created with dictionary size appended to it. So all the files
// with a given column ID need to be listed
@@ -76,8 +75,7 @@ public class ManageDictionaryAndBTree {
// try catch is inside for loop because even if one deletion fails, other files
// still need to be deleted
try {
- FileFactory.deleteFile(file.getCanonicalPath(),
- FileFactory.getFileType(file.getCanonicalPath()));
+ FileFactory.deleteFile(file.getCanonicalPath());
} catch (IOException e) {
LOGGER.error("Failed to delete dictionary or sortIndex file for column "
+ columnSchema.getColumnName() + "with column ID "
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
index 9182c70..5f097e3 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/TableDataMap.java
@@ -399,7 +399,7 @@ public final class TableDataMap extends OperationEventListener {
identifier.getTablePath() + CarbonCommonConstants.FILE_SEPARATOR + dataMapSchema
.getDataMapName();
if (dataMapFactory.getDataMapLevel() == DataMapLevel.FG) {
- FileFactory.mkdirs(writePath, FileFactory.getFileType(writePath));
+ FileFactory.mkdirs(writePath);
}
for (Blocklet blocklet : blocklets) {
ExtendedBlocklet detailedBlocklet = blockletDetailsFetcher
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/BlockletSerializer.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/BlockletSerializer.java
index 7c8e63e..0d82275 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/BlockletSerializer.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/BlockletSerializer.java
@@ -41,7 +41,7 @@ public class BlockletSerializer {
public void serializeBlocklet(FineGrainBlocklet grainBlocklet, String writePath)
throws IOException {
DataOutputStream dataOutputStream =
- FileFactory.getDataOutputStream(writePath, FileFactory.getFileType(writePath));
+ FileFactory.getDataOutputStream(writePath);
try {
grainBlocklet.write(dataOutputStream);
} finally {
@@ -57,7 +57,7 @@ public class BlockletSerializer {
*/
public FineGrainBlocklet deserializeBlocklet(String writePath) throws IOException {
DataInputStream inputStream =
- FileFactory.getDataInputStream(writePath, FileFactory.getFileType(writePath));
+ FileFactory.getDataInputStream(writePath);
FineGrainBlocklet blocklet = new FineGrainBlocklet();
try {
blocklet.readFields(inputStream);
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
index cd0d86e..ec69b19 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/dev/DataMapWriter.java
@@ -120,7 +120,7 @@ public abstract class DataMapWriter {
String segmentPath = CarbonTablePath.getSegmentPath(tablePath, segmentId);
if (carbonFilePath.length() > 0) {
carbonFilePath = segmentPath + carbonFilePath;
- FileFactory.mkdirs(carbonFilePath, FileFactory.getFileType(carbonFilePath));
+ FileFactory.mkdirs(carbonFilePath);
} else {
carbonFilePath = segmentPath;
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datamap/status/DiskBasedDataMapStatusProvider.java b/core/src/main/java/org/apache/carbondata/core/datamap/status/DiskBasedDataMapStatusProvider.java
index 723010e..bb1b188 100644
--- a/core/src/main/java/org/apache/carbondata/core/datamap/status/DiskBasedDataMapStatusProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/datamap/status/DiskBasedDataMapStatusProvider.java
@@ -72,8 +72,7 @@ public class DiskBasedDataMapStatusProvider implements DataMapStatusStorageProvi
if (!FileFactory.isFileExist(statusPath)) {
return new DataMapStatusDetail[0];
}
- dataInputStream =
- FileFactory.getDataInputStream(statusPath, FileFactory.getFileType(statusPath));
+ dataInputStream = FileFactory.getDataInputStream(statusPath);
inStream = new InputStreamReader(dataInputStream,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
buffReader = new BufferedReader(inStream);
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
index f3362d3..51c4337 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java
@@ -30,6 +30,7 @@ import java.util.List;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.impl.FileFactory;
+import org.apache.carbondata.core.exception.CarbonFileException;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.hadoop.conf.Configuration;
@@ -58,85 +59,82 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
*/
private static final Logger LOGGER =
LogServiceFactory.getLogService(AbstractDFSCarbonFile.class.getName());
- protected FileStatus fileStatus;
- public FileSystem fs;
+ protected FileSystem fileSystem;
protected Configuration hadoopConf;
+ protected Path path;
- public AbstractDFSCarbonFile(String filePath) {
+ AbstractDFSCarbonFile(String filePath) {
this(filePath, FileFactory.getConfiguration());
}
- public AbstractDFSCarbonFile(String filePath, Configuration hadoopConf) {
- this.hadoopConf = hadoopConf;
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- try {
- fs = path.getFileSystem(this.hadoopConf);
- fileStatus = fs.getFileStatus(path);
- } catch (IOException e) {
- LOGGER.debug("Exception occurred:" + e.getMessage(), e);
- }
+ AbstractDFSCarbonFile(String filePath, Configuration hadoopConf) {
+ this(new Path(filePath), hadoopConf);
}
- public AbstractDFSCarbonFile(Path path) {
+ AbstractDFSCarbonFile(Path path) {
this(path, FileFactory.getConfiguration());
}
- public AbstractDFSCarbonFile(Path path, Configuration hadoopConf) {
+ AbstractDFSCarbonFile(Path path, Configuration hadoopConf) {
this.hadoopConf = hadoopConf;
- FileSystem fs;
try {
- fs = path.getFileSystem(this.hadoopConf);
- fileStatus = fs.getFileStatus(path);
+ Path newPath = new Path(path.toString().replace("\\", "/"));
+ fileSystem = newPath.getFileSystem(this.hadoopConf);
+ this.path = newPath;
} catch (IOException e) {
- LOGGER.debug("Exception occurred:" + e.getMessage(), e);
+ throw new CarbonFileException("Error while getting File System: ", e);
}
}
- public AbstractDFSCarbonFile(FileStatus fileStatus) {
- this.hadoopConf = FileFactory.getConfiguration();
- this.fileStatus = fileStatus;
+ AbstractDFSCarbonFile(FileStatus fileStatus) {
+ this(fileStatus.getPath());
}
@Override
public boolean createNewFile() {
- Path path = fileStatus.getPath();
- FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- return fs.createNewFile(path);
+ return fileSystem.createNewFile(path);
} catch (IOException e) {
- return false;
+ throw new CarbonFileException("Unable to create file: " + path.toString(), e);
}
}
@Override
public String getAbsolutePath() {
- return fileStatus.getPath().toString();
+ try {
+ return fileSystem.getFileStatus(path).getPath().toString();
+ } catch (IOException e) {
+ throw new CarbonFileException("Unable to get file status: ", e);
+ }
}
@Override
public String getName() {
- return fileStatus.getPath().getName();
+ return path.getName();
}
@Override
public boolean isDirectory() {
- return fileStatus.isDirectory();
+ try {
+ return fileSystem.getFileStatus(path).isDirectory();
+ } catch (IOException e) {
+ throw new CarbonFileException("Unable to get file status: ", e);
+ }
+ }
+
+ @Override
+ public CarbonFile getParentFile() {
+ Path parentPath = path.getParent();
+ return parentPath == null ? null : FileFactory.getCarbonFile(parentPath.toString());
}
@Override
public boolean exists() {
- FileSystem fs;
try {
- if (null != fileStatus) {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- return fs.exists(fileStatus.getPath());
- }
+ return fileSystem.exists(path);
} catch (IOException e) {
- LOGGER.error("Exception occurred:" + e.getMessage(), e);
+ throw new CarbonFileException(e);
}
- return false;
}
@Override
@@ -151,54 +149,47 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
@Override
public long getSize() {
- return fileStatus.getLen();
+ try {
+ return fileSystem.getFileStatus(path).getLen();
+ } catch (IOException e) {
+ throw new CarbonFileException("Unable to get file status: ", e);
+ }
}
- public boolean renameTo(String changeToName) {
- FileSystem fs;
+ @Override
+ public boolean renameTo(String changetoName) {
try {
- if (null != fileStatus) {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- return fs.rename(fileStatus.getPath(), new Path(changeToName));
- }
+ return fileSystem.rename(path, new Path(changetoName));
} catch (IOException e) {
- LOGGER.error("Exception occurred:" + e.getMessage(), e);
- return false;
+ throw new CarbonFileException("Failed to rename file: ", e);
}
- return false;
}
public boolean delete() {
- FileSystem fs;
try {
- if (null != fileStatus) {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- return fs.delete(fileStatus.getPath(), true);
- }
+ return fileSystem.delete(path, true);
} catch (IOException e) {
- LOGGER.error("Exception occurred:" + e.getMessage(), e);
- return false;
+ throw new CarbonFileException("Failed to delete file:", e);
}
- return false;
}
@Override
public long getLastModifiedTime() {
- return fileStatus.getModificationTime();
+ try {
+ return fileSystem.getFileStatus(path).getModificationTime();
+ } catch (IOException e) {
+ throw new CarbonFileException("Unable to get file status: ", e);
+ }
}
@Override
public boolean setLastModifiedTime(long timestamp) {
- FileSystem fs;
try {
- if (null != fileStatus) {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- fs.setTimes(fileStatus.getPath(), timestamp, timestamp);
- }
+ fileSystem.setTimes(path, timestamp, timestamp);
+ return true;
} catch (IOException e) {
- return false;
+ throw new CarbonFileException("Error while setting modified time: ", e);
}
- return true;
}
/**
@@ -215,22 +206,21 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
(int) validDataEndOffset;
// temporary file name
String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
try {
CarbonFile tempFile;
// delete temporary file if it already exists at a given path
- if (FileFactory.isFileExist(tempWriteFilePath, fileType)) {
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
+ if (FileFactory.isFileExist(tempWriteFilePath)) {
+ tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
tempFile.delete();
}
// create new temporary file
- FileFactory.createNewFile(tempWriteFilePath, fileType);
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
+ FileFactory.createNewFile(tempWriteFilePath);
+ tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
byte[] buff = new byte[bufferSize];
- dataInputStream = FileFactory.getDataInputStream(fileName, fileType);
+ dataInputStream = FileFactory.getDataInputStream(fileName);
// read the data
int read = dataInputStream.read(buff, 0, buff.length);
- dataOutputStream = FileFactory.getDataOutputStream(tempWriteFilePath, fileType);
+ dataOutputStream = FileFactory.getDataOutputStream(tempWriteFilePath);
dataOutputStream.write(buff, 0, read);
long remaining = validDataEndOffset - read;
// anytime we should not cross the offset to be read
@@ -273,71 +263,52 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, boolean append) throws IOException {
- Path pt = new Path(path);
- FileSystem fileSystem = pt.getFileSystem(FileFactory.getConfiguration());
+ public DataOutputStream getDataOutputStream(int bufferSize, boolean append)
+ throws IOException {
FSDataOutputStream stream = null;
if (append) {
// append to a file only if file already exists else file not found
// exception will be thrown by hdfs
- if (CarbonUtil.isFileExists(path)) {
- stream = fileSystem.append(pt, bufferSize);
+ if (CarbonUtil.isFileExists(path.toString())) {
+ stream = fileSystem.append(path, bufferSize);
} else {
- stream = fileSystem.create(pt, true, bufferSize);
+ stream = fileSystem.create(path, true, bufferSize);
}
} else {
- stream = fileSystem.create(pt, true, bufferSize);
+ stream = fileSystem.create(path, true, bufferSize);
}
return stream;
}
@Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, Configuration hadoopConf) throws IOException {
- return getDataInputStream(path, fileType, bufferSize,
- CarbonUtil.inferCompressorFromFileName(path), hadoopConf);
- }
-
- @Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, String compressor) throws IOException {
- return getDataInputStream(path, fileType, bufferSize, FileFactory.getConfiguration());
+ public DataInputStream getDataInputStream(int bufferSize) throws IOException {
+ return getDataInputStream(bufferSize,
+ CarbonUtil.inferCompressorFromFileName(path.toString()));
}
/**
* return the datainputStream which is seek to the offset of file
*
- * @param path
- * @param fileType
- * @param bufferSize
- * @param offset
* @return DataInputStream
* @throws IOException
*/
@Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long offset) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(FileFactory.getConfiguration());
- FSDataInputStream stream = fs.open(pt, bufferSize);
+ public DataInputStream getDataInputStream(int bufferSize, long offset)
+ throws IOException {
+ FSDataInputStream stream = fileSystem.open(path, bufferSize);
stream.seek(offset);
return new DataInputStream(new BufferedInputStream(stream));
}
- private DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, String compressor, Configuration configuration) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
+ @Override
+ public DataInputStream getDataInputStream(int bufferSize, String compressor)
+ throws IOException {
InputStream inputStream;
- FileSystem fs = pt.getFileSystem(configuration);
if (bufferSize <= 0) {
- inputStream = fs.open(pt);
+ inputStream = fileSystem.open(path);
} else {
- inputStream = fs.open(pt, bufferSize);
+ inputStream = fileSystem.open(path, bufferSize);
}
-
String codecName = getCodecNameFromCompressor(compressor);
if (!codecName.isEmpty()) {
CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
@@ -352,6 +323,7 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
/**
* get codec name from user specified compressor name
+ *
* @param compressorName user specified compressor name
* @return name of codec
* @throws IOException
@@ -373,160 +345,109 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType)
- throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(FileFactory.getConfiguration());
- return fs.create(pt, true);
+ public DataOutputStream getDataOutputStream() throws IOException {
+ return fileSystem.create(path, true);
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long blockSize) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- short replication = pt.getFileSystem(FileFactory.getConfiguration()).getDefaultReplication(pt);
- return getDataOutputStream(path, fileType, bufferSize, blockSize, replication);
+ public DataOutputStream getDataOutputStream(int bufferSize, long blockSize)
+ throws IOException {
+ short replication = fileSystem.getDefaultReplication(path);
+ return getDataOutputStream(bufferSize, blockSize, replication);
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long blockSize, short replication) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(FileFactory.getConfiguration());
- return fs.create(pt, true, bufferSize, replication, blockSize);
+ public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
+ throws IOException {
+ return fileSystem.create(path, true, bufferSize, replication, blockSize);
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, String compressor) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
+ public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
+ throws IOException {
OutputStream outputStream;
if (bufferSize <= 0) {
- outputStream = fs.create(pt);
+ outputStream = fileSystem.create(path);
} else {
- outputStream = fs.create(pt, true, bufferSize);
+ outputStream = fileSystem.create(path, true, bufferSize);
}
-
String codecName = getCodecNameFromCompressor(compressor);
if (!codecName.isEmpty()) {
CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
outputStream = codec.createOutputStream(outputStream);
}
-
return new DataOutputStream(new BufferedOutputStream(outputStream));
}
@Override
- public boolean isFileExist(String filePath, boolean performFileCheck)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
+ public boolean isFileExist(boolean performFileCheck) throws IOException {
if (performFileCheck) {
- return fs.exists(path) && fs.isFile(path);
+ return fileSystem.exists(path) && fileSystem.isFile(path);
} else {
- return fs.exists(path);
+ return fileSystem.exists(path);
}
}
@Override
- public boolean isFileExist(String filePath)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.exists(path);
- }
-
- @Override
- public boolean createNewFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.createNewFile(path);
+ public boolean isFileExist() throws IOException {
+ return isFileExist(false);
}
@Override
- public boolean createNewFile(String filePath, FileFactory.FileType fileType, boolean doAs,
- FsPermission permission) throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- if (fs.exists(path)) {
+ public boolean createNewFile(FsPermission permission) throws IOException {
+ if (fileSystem.exists(path)) {
return false;
} else {
if (permission == null) {
- permission = FsPermission.getFileDefault().applyUMask(FsPermission.getUMask(fs.getConf()));
+ permission =
+ FsPermission.getFileDefault().applyUMask(FsPermission.getUMask(fileSystem.getConf()));
}
// Pass the permissions duringg file creation itself
- fs.create(path, permission, false, fs.getConf().getInt("io.file.buffer.size", 4096),
- fs.getDefaultReplication(path), fs.getDefaultBlockSize(path), null).close();
+ fileSystem
+ .create(path, permission, false, fileSystem.getConf().getInt("io.file.buffer.size", 4096),
+ fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
+ .close();
return true;
}
}
@Override
- public boolean deleteFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.delete(path, true);
+ public boolean deleteFile() throws IOException {
+ return fileSystem.delete(path, true);
}
@Override
- public boolean mkdirs(String filePath)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.mkdirs(path);
+ public boolean mkdirs() throws IOException {
+ return fileSystem.mkdirs(path);
}
@Override
- public DataOutputStream getDataOutputStreamUsingAppend(String path, FileFactory.FileType fileType)
- throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(FileFactory.getConfiguration());
- return fs.append(pt);
+ public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
+ return fileSystem.append(path);
}
@Override
- public boolean createNewLockFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- if (fs.exists(path)) {
+ public boolean createNewLockFile() throws IOException {
+ if (fileSystem.exists(path)) {
return false;
} else {
// Pass the permissions duringg file creation itself
- fs.create(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL), false,
- fs.getConf().getInt("io.file.buffer.size", 4096), fs.getDefaultReplication(path),
- fs.getDefaultBlockSize(path), null).close();
+ fileSystem.create(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL), false,
+ fileSystem.getConf().getInt("io.file.buffer.size", 4096),
+ fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
+ .close();
// haddop masks the permission accoding to configured permission, so need to set permission
// forcefully
- fs.setPermission(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
+ fileSystem.setPermission(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
return true;
}
}
@Override
public CarbonFile[] listFiles() {
- FileStatus[] listStatus = null;
+ FileStatus[] listStatus;
try {
- if (null != fileStatus && fileStatus.isDirectory()) {
- Path path = fileStatus.getPath();
- listStatus = path.getFileSystem(hadoopConf).listStatus(path);
- } else {
- return new CarbonFile[0];
- }
+ listStatus = fileSystem.listStatus(path);
} catch (IOException e) {
LOGGER.error("Exception occured: " + e.getMessage(), e);
return new CarbonFile[0];
@@ -534,18 +455,40 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
return getFiles(listStatus);
}
+ /**
+ * Get the CarbonFiles from filestatus array
+ */
+ protected abstract CarbonFile[] getFiles(FileStatus[] listStatus);
+
@Override
public List<CarbonFile> listFiles(Boolean recursive) throws IOException {
- RemoteIterator<LocatedFileStatus> listStatus = null;
- if (null != fileStatus && fileStatus.isDirectory()) {
- Path path = fileStatus.getPath();
- listStatus = fs.listFiles(path, recursive);
- } else {
- return new ArrayList<CarbonFile>();
- }
+ RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(path, recursive);
return getFiles(listStatus);
}
+ @Override
+ public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) throws IOException {
+ List<FileStatus> listStatus = new ArrayList<>();
+ RemoteIterator<LocatedFileStatus> iter = fileSystem.listLocatedStatus(path);
+ while (iter.hasNext()) {
+ LocatedFileStatus fileStatus = iter.next();
+ if (pathFilter.accept(fileStatus.getPath()) && fileStatus.getLen() > 0) {
+ listStatus.add(fileStatus);
+ }
+ }
+ return getFiles(listStatus.toArray(new FileStatus[listStatus.size()]));
+ }
+
+ protected List<CarbonFile> getFiles(RemoteIterator<LocatedFileStatus> listStatus)
+ throws IOException {
+ List<CarbonFile> carbonFiles = new ArrayList<>();
+ while (listStatus.hasNext()) {
+ Path filePath = listStatus.next().getPath();
+ carbonFiles.add(FileFactory.getCarbonFile(filePath.toString()));
+ }
+ return carbonFiles;
+ }
+
/**
* Method used to list files recursively and apply file filter on the result.
*
@@ -554,8 +497,10 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter)
throws IOException {
List<CarbonFile> carbonFiles = new ArrayList<>();
+ FileStatus fileStatus = fileSystem.getFileStatus(path);
if (null != fileStatus && fileStatus.isDirectory()) {
- RemoteIterator<LocatedFileStatus> listStatus = fs.listFiles(fileStatus.getPath(), recursive);
+ RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(fileStatus.getPath(),
+ recursive);
while (listStatus.hasNext()) {
LocatedFileStatus locatedFileStatus = listStatus.next();
CarbonFile carbonFile = FileFactory.getCarbonFile(locatedFileStatus.getPath().toString());
@@ -568,68 +513,49 @@ public abstract class AbstractDFSCarbonFile implements CarbonFile {
}
@Override
- public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) throws IOException {
- if (null != fileStatus && fileStatus.isDirectory()) {
- List<FileStatus> listStatus = new ArrayList<>();
- Path path = fileStatus.getPath();
- RemoteIterator<LocatedFileStatus> iter = fs.listLocatedStatus(path);
- while (iter.hasNext()) {
- LocatedFileStatus fileStatus = iter.next();
- if (pathFilter.accept(fileStatus.getPath()) && fileStatus.getLen() > 0) {
- listStatus.add(fileStatus);
- }
- }
- return getFiles(listStatus.toArray(new FileStatus[listStatus.size()]));
- }
- return new CarbonFile[0];
- }
-
- /**
- * Get the CarbonFiles from filestatus array
- */
- protected abstract CarbonFile[] getFiles(FileStatus[] listStatus);
-
- protected List<CarbonFile> getFiles(RemoteIterator<LocatedFileStatus> listStatus)
- throws IOException {
- List<CarbonFile> carbonFiles = new ArrayList<>();
- while (listStatus.hasNext()) {
- Path filePath = listStatus.next().getPath();
- carbonFiles.add(FileFactory.getCarbonFile(filePath.toString()));
- }
- return carbonFiles;
- }
-
- @Override
public String[] getLocations() throws IOException {
BlockLocation[] blkLocations;
+ FileStatus fileStatus = fileSystem.getFileStatus(path);
if (fileStatus instanceof LocatedFileStatus) {
- blkLocations = ((LocatedFileStatus)fileStatus).getBlockLocations();
+ blkLocations = ((LocatedFileStatus) fileStatus).getBlockLocations();
} else {
- FileSystem fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
- blkLocations = fs.getFileBlockLocations(fileStatus.getPath(), 0L, fileStatus.getLen());
+ blkLocations =
+ fileSystem.getFileBlockLocations(fileStatus.getPath(), 0L, fileStatus.getLen());
}
-
return blkLocations[0].getHosts();
}
@Override
- public boolean setReplication(String filePath, short replication) throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.setReplication(path, replication);
+ public boolean setReplication(short replication) throws IOException {
+ return fileSystem.setReplication(path, replication);
+ }
+
+ @Override
+ public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
+ CarbonFile[] files = listFiles();
+ if (files != null && files.length >= 1) {
+ List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
+ for (int i = 0; i < files.length; i++) {
+ if (fileFilter.accept(files[i])) {
+ fileList.add(files[i]);
+ }
+ }
+ if (fileList.size() >= 1) {
+ return fileList.toArray(new CarbonFile[fileList.size()]);
+ } else {
+ return new CarbonFile[0];
+ }
+ }
+ return files;
}
@Override
- public short getDefaultReplication(String filePath) throws IOException {
- filePath = filePath.replace("\\", "/");
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(FileFactory.getConfiguration());
- return fs.getDefaultReplication(path);
+ public short getDefaultReplication() {
+ return fileSystem.getDefaultReplication(path);
}
@Override
- public long getLength() {
- return fileStatus.getLen();
+ public long getLength() throws IOException {
+ return fileSystem.getFileStatus(path).getLen();
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFile.java
index affb469..03ccd92 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFile.java
@@ -17,24 +17,13 @@
package org.apache.carbondata.core.datastore.filesystem;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
-import org.apache.carbondata.core.util.CarbonUtil;
-
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.log4j.Logger;
public class AlluxioCarbonFile extends HDFSCarbonFile {
@@ -56,10 +45,6 @@ public class AlluxioCarbonFile extends HDFSCarbonFile {
super(fileStatus);
}
- /**
- * @param listStatus
- * @return
- */
@Override
protected CarbonFile[] getFiles(FileStatus[] listStatus) {
if (listStatus == null) {
@@ -73,65 +58,11 @@ public class AlluxioCarbonFile extends HDFSCarbonFile {
}
@Override
- public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- CarbonFile[] files = listFiles();
- if (files != null && files.length >= 1) {
- List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
- }
- }
- if (fileList.size() >= 1) {
- return fileList.toArray(new CarbonFile[fileList.size()]);
- } else {
- return new CarbonFile[0];
- }
- }
- return files;
- }
-
- @Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new AlluxioCarbonFile(parent);
- }
-
- /**
- * <p>RenameForce of the fileName for the AlluxioFileSystem Implementation.
- * Involves by opening a {@link FSDataInputStream} from the existing path and copy
- * bytes to {@link FSDataOutputStream}.
- * </p>
- * <p>Close the output and input streams only after the files have been written
- * Also check for the existence of the changed path and then delete the previous Path.
- * The No of Bytes that can be read is controlled by {@literal io.file.buffer.size},
- * where the default value is 4096.</p>
- * @param changeToName
- * @return
- */
- @Override
- public boolean renameForce(String changeToName) {
- FileSystem fs = null;
- FSDataOutputStream fsdos = null;
- FSDataInputStream fsdis = null;
+ public boolean renameForce(String changetoName) {
try {
- Path actualPath = fileStatus.getPath();
- Path changedPath = new Path(changeToName);
- fs = actualPath.getFileSystem(hadoopConf);
- fsdos = fs.create(changedPath, true);
- fsdis = fs.open(actualPath);
- if (null != fsdis && null != fsdos) {
- try {
- IOUtils.copyBytes(fsdis, fsdos, hadoopConf, true);
- if (fs.exists(changedPath)) {
- fs.delete(actualPath, true);
- }
- // Reassigning fileStatus to the changedPath.
- fileStatus = fs.getFileStatus(changedPath);
- } catch (IOException e) {
- LOGGER.error("Exception occured: " + e.getMessage());
- return false;
- }
+ if (fileSystem instanceof DistributedFileSystem) {
+ ((DistributedFileSystem) fileSystem).rename(path, new Path(changetoName),
+ org.apache.hadoop.fs.Options.Rename.OVERWRITE);
return true;
}
return false;
@@ -141,37 +72,4 @@ public class AlluxioCarbonFile extends HDFSCarbonFile {
}
}
- @Override
- public DataOutputStream getDataOutputStreamUsingAppend(String path, FileFactory.FileType fileType)
- throws IOException {
- return getDataOutputStream(path, fileType, CarbonCommonConstants.BYTEBUFFER_SIZE, true);
- }
-
- @Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, boolean append) throws IOException {
- path = path.replace("\\", "/");
- Path pt = new Path(path);
- FileSystem fileSystem = pt.getFileSystem(FileFactory.getConfiguration());
- FSDataOutputStream stream;
- if (append) {
- if (CarbonUtil.isFileExists(path)) {
- DataInputStream dataInputStream = fileSystem.open(pt);
- int count = dataInputStream.available();
- // create buffer
- byte[] byteStreamBuffer = new byte[count];
- int bytesRead = dataInputStream.read(byteStreamBuffer);
- dataInputStream.close();
- stream = fileSystem.create(pt, true, bufferSize);
- if (bytesRead > 0) {
- stream.write(byteStreamBuffer, 0, bytesRead);
- }
- } else {
- stream = fileSystem.create(pt, true, bufferSize);
- }
- } else {
- stream = fileSystem.create(pt, true, bufferSize);
- }
- return stream;
- }
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/CarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/CarbonFile.java
index f1c4123..00d37e7 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/CarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/CarbonFile.java
@@ -22,9 +22,6 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.util.List;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
-
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.permission.FsPermission;
@@ -42,7 +39,6 @@ public interface CarbonFile {
/**
* It returns list of files with location details.
- * @return
*/
CarbonFile[] locationAwareListFiles(PathFilter pathFilter) throws IOException;
@@ -84,80 +80,60 @@ public interface CarbonFile {
*
* @param fileTimeStamp time to be compared with latest timestamp of file
* @param endOffset file length to be compared with current length of file
- * @return
*/
boolean isFileModified(long fileTimeStamp, long endOffset);
- DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType, int bufferSize,
- boolean append) throws IOException;
+ DataOutputStream getDataOutputStream(int bufferSize, boolean append) throws IOException;
- DataInputStream getDataInputStream(String path, FileFactory.FileType fileType, int bufferSize,
- Configuration configuration) throws IOException;
+ DataInputStream getDataInputStream(int bufferSize) throws
+ IOException;
/**
* get data input stream
- * @param path
- * @param fileType
- * @param bufferSize
* @param compressor name of compressor to write this file
* @return dataInputStream
* @throws IOException
*/
- DataInputStream getDataInputStream(String path, FileFactory.FileType fileType, int bufferSize,
- String compressor) throws IOException;
+ DataInputStream getDataInputStream(int bufferSize, String compressor) throws IOException;
- DataInputStream getDataInputStream(String path, FileFactory.FileType fileType, int bufferSize,
- long offset) throws IOException;
+ DataInputStream getDataInputStream(int bufferSize, long offset) throws IOException;
- DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType)
- throws IOException;
+ DataOutputStream getDataOutputStream() throws IOException;
- DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType, int bufferSize,
- long blockSize) throws IOException;
+ DataOutputStream getDataOutputStream(int bufferSize, long blockSize) throws IOException;
/**
* get data output stream
- * @param path file path
- * @param fileType file type
* @param bufferSize write buffer size
* @param blockSize block size
* @param replication replication for this file
* @return data output stream
* @throws IOException if error occurs
*/
- DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType, int bufferSize,
+ DataOutputStream getDataOutputStream(int bufferSize,
long blockSize, short replication) throws IOException;
/**
* get data output stream
- * @param path
- * @param fileType
- * @param bufferSize
* @param compressor name of compressor to write this file
* @return DataOutputStream
* @throws IOException
*/
- DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType, int bufferSize,
- String compressor) throws IOException;
+ DataOutputStream getDataOutputStream(int bufferSize, String compressor) throws IOException;
- boolean isFileExist(String filePath, boolean performFileCheck)
- throws IOException;
+ boolean isFileExist(boolean performFileCheck) throws IOException;
- boolean isFileExist(String filePath) throws IOException;
+ boolean isFileExist() throws IOException;
- boolean createNewFile(String filePath, FileFactory.FileType fileType) throws IOException;
+ boolean createNewFile(final FsPermission permission) throws IOException;
- boolean createNewFile(String filePath, FileFactory.FileType fileType, boolean doAs,
- final FsPermission permission) throws IOException;
+ boolean deleteFile() throws IOException;
- boolean deleteFile(String filePath, FileFactory.FileType fileType) throws IOException;
+ boolean mkdirs() throws IOException;
- boolean mkdirs(String filePath) throws IOException;
+ DataOutputStream getDataOutputStreamUsingAppend() throws IOException;
- DataOutputStream getDataOutputStreamUsingAppend(String path, FileFactory.FileType fileType)
- throws IOException;
-
- boolean createNewLockFile(String filePath, FileFactory.FileType fileType) throws IOException;
+ boolean createNewLockFile() throws IOException;
/**
* Returns locations of the file
@@ -168,24 +144,21 @@ public interface CarbonFile {
/**
* set the replication factor for this file
*
- * @param filePath file path
* @param replication replication
* @return true, if success; false, if failed
* @throws IOException if error occurs
*/
- boolean setReplication(String filePath, short replication) throws IOException;
+ boolean setReplication(short replication) throws IOException;
/**
* get the default replication for this file
- * @param filePath file path
* @return replication factor
- * @throws IOException if error occurs
*/
- short getDefaultReplication(String filePath) throws IOException;
+ short getDefaultReplication();
/**
* Get the length of this file, in bytes.
* @return the length of this file, in bytes.
*/
- long getLength();
+ long getLength() throws IOException;
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
index 306d8f6..ac508c3 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFile.java
@@ -18,14 +18,11 @@
package org.apache.carbondata.core.datastore.filesystem;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.log4j.Logger;
@@ -74,41 +71,14 @@ public class HDFSCarbonFile extends AbstractDFSCarbonFile {
}
@Override
- public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- CarbonFile[] files = listFiles();
- if (files != null && files.length >= 1) {
- List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
- }
- }
- if (fileList.size() >= 1) {
- return fileList.toArray(new CarbonFile[fileList.size()]);
- } else {
- return new CarbonFile[0];
- }
- }
- return files;
- }
-
- @Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new HDFSCarbonFile(parent, hadoopConf);
- }
-
- @Override
- public boolean renameForce(String changeToName) {
- FileSystem fs;
+ public boolean renameForce(String changetoName) {
try {
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- if (fs instanceof DistributedFileSystem) {
- ((DistributedFileSystem) fs).rename(fileStatus.getPath(), new Path(changeToName),
+ if (fileSystem instanceof DistributedFileSystem) {
+ ((DistributedFileSystem) fileSystem).rename(path, new Path(changetoName),
org.apache.hadoop.fs.Options.Rename.OVERWRITE);
return true;
} else {
- return fs.rename(fileStatus.getPath(), new Path(changeToName));
+ return fileSystem.rename(path, new Path(changetoName));
}
} catch (IOException e) {
LOGGER.error("Exception occured: " + e.getMessage(), e);
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFile.java
index 3a6a687..1bb8e03 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFile.java
@@ -22,7 +22,6 @@ import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
-import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
@@ -48,7 +47,6 @@ import net.jpountz.lz4.LZ4BlockOutputStream;
import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.permission.FsPermission;
@@ -59,15 +57,22 @@ import org.xerial.snappy.SnappyOutputStream;
public class LocalCarbonFile implements CarbonFile {
private static final Logger LOGGER =
LogServiceFactory.getLogService(LocalCarbonFile.class.getName());
+
private File file;
+ private String absoluteFilePath;
+
public LocalCarbonFile(String filePath) {
Path pathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
- file = new File(pathWithoutSchemeAndAuthority.toString());
+ filePath = pathWithoutSchemeAndAuthority.toString().replace("\\", "/");
+ absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
+ file = new File(absoluteFilePath);
}
- public LocalCarbonFile(File file) {
+ LocalCarbonFile(File file) {
this.file = file;
+ String filePath = file.getAbsolutePath().replace("\\", "/");
+ absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
}
@Override
@@ -77,28 +82,17 @@ public class LocalCarbonFile implements CarbonFile {
@Override
public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- if (!file.isDirectory()) {
+ if (!isDirectory()) {
return new CarbonFile[0];
}
-
- File[] files = file.listFiles(new FileFilter() {
-
- @Override
- public boolean accept(File pathname) {
- return fileFilter.accept(new LocalCarbonFile(pathname));
- }
- });
-
+ File[] files = file.listFiles(pathname -> fileFilter.accept(new LocalCarbonFile(pathname)));
if (files == null) {
return new CarbonFile[0];
}
-
CarbonFile[] carbonFiles = new CarbonFile[files.length];
-
for (int i = 0; i < carbonFiles.length; i++) {
carbonFiles[i] = new LocalCarbonFile(files[i]);
}
-
return carbonFiles;
}
@@ -114,10 +108,7 @@ public class LocalCarbonFile implements CarbonFile {
@Override
public boolean exists() {
- if (file != null) {
- return file.exists();
- }
- return false;
+ return file.exists();
}
@Override
@@ -145,14 +136,14 @@ public class LocalCarbonFile implements CarbonFile {
return file.length();
}
- public boolean renameTo(String changeToName) {
- changeToName = FileFactory.getUpdatedFilePath(changeToName, FileFactory.FileType.LOCAL);
- return file.renameTo(new File(changeToName));
+ public boolean renameTo(String changetoName) {
+ changetoName = FileFactory.getUpdatedFilePath(changetoName);
+ return file.renameTo(new File(changetoName));
}
public boolean delete() {
try {
- return deleteFile(file.getAbsolutePath(), FileFactory.getFileType(file.getAbsolutePath()));
+ return deleteFile();
} catch (IOException e) {
LOGGER.error("Exception occurred:" + e.getMessage(), e);
return false;
@@ -161,7 +152,6 @@ public class LocalCarbonFile implements CarbonFile {
@Override
public CarbonFile[] listFiles() {
-
if (!file.isDirectory()) {
return new CarbonFile[0];
}
@@ -173,21 +163,16 @@ public class LocalCarbonFile implements CarbonFile {
for (int i = 0; i < carbonFiles.length; i++) {
carbonFiles[i] = new LocalCarbonFile(files[i]);
}
-
return carbonFiles;
-
}
@Override
public List<CarbonFile> listFiles(Boolean recursive) {
- if (!file.isDirectory()) {
- return new ArrayList<CarbonFile>();
+ if (!isDirectory()) {
+ return new ArrayList<>();
}
Collection<File> fileCollection = FileUtils.listFiles(file, null, true);
- if (fileCollection == null) {
- return new ArrayList<CarbonFile>();
- }
- List<CarbonFile> carbonFiles = new ArrayList<CarbonFile>();
+ List<CarbonFile> carbonFiles = new ArrayList<>();
for (File file : fileCollection) {
carbonFiles.add(new LocalCarbonFile(file));
}
@@ -198,13 +183,13 @@ public class LocalCarbonFile implements CarbonFile {
public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter)
throws IOException {
if (!file.isDirectory()) {
- return new ArrayList<CarbonFile>();
+ return new ArrayList<>();
}
Collection<File> fileCollection = FileUtils.listFiles(file, null, recursive);
- if (fileCollection == null) {
- return new ArrayList<CarbonFile>();
+ if (fileCollection.isEmpty()) {
+ return new ArrayList<>();
}
- List<CarbonFile> carbonFiles = new ArrayList<CarbonFile>();
+ List<CarbonFile> carbonFiles = new ArrayList<>();
for (File file : fileCollection) {
CarbonFile carbonFile = new LocalCarbonFile(file);
if (fileFilter.accept(carbonFile)) {
@@ -243,17 +228,16 @@ public class LocalCarbonFile implements CarbonFile {
boolean fileTruncatedSuccessfully = false;
// temporary file name
String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
try {
- CarbonFile tempFile = null;
+ CarbonFile tempFile;
// delete temporary file if it already exists at a given path
- if (FileFactory.isFileExist(tempWriteFilePath, fileType)) {
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
+ if (isFileExist()) {
+ tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
tempFile.delete();
}
// create new temporary file
- FileFactory.createNewFile(tempWriteFilePath, fileType);
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
+ FileFactory.createNewFile(tempWriteFilePath);
+ tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
source = new FileInputStream(fileName).getChannel();
destination = new FileOutputStream(tempWriteFilePath).getChannel();
long read = destination.transferFrom(source, 0, validDataEndOffset);
@@ -282,7 +266,7 @@ public class LocalCarbonFile implements CarbonFile {
*
* @param fileTimeStamp time to be compared with latest timestamp of file
* @param endOffset file length to be compared with current length of file
- * @return
+ * @return true if the file is modified otherwise return false.
*/
@Override
public boolean isFileModified(long fileTimeStamp, long endOffset) {
@@ -305,38 +289,34 @@ public class LocalCarbonFile implements CarbonFile {
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, boolean append) throws FileNotFoundException {
- path = FileFactory.getUpdatedFilePath(path, FileFactory.FileType.LOCAL);
+ public DataOutputStream getDataOutputStream(int bufferSize, boolean append) throws
+ FileNotFoundException {
return new DataOutputStream(
- new BufferedOutputStream(new FileOutputStream(path, append), bufferSize));
+ new BufferedOutputStream(new FileOutputStream(absoluteFilePath, append), bufferSize));
}
@Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, Configuration configuration) throws IOException {
- return getDataInputStream(path, fileType, bufferSize,
- CarbonUtil.inferCompressorFromFileName(path));
+ public DataInputStream getDataInputStream(int bufferSize) throws IOException {
+ return getDataInputStream(bufferSize,
+ CarbonUtil.inferCompressorFromFileName(file.getAbsolutePath()));
}
@Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, String compressor) throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, fileType);
+ public DataInputStream getDataInputStream(int bufferSize, String compressor)
+ throws IOException {
InputStream inputStream;
if (compressor.isEmpty()) {
- inputStream = new FileInputStream(path);
+ inputStream = new FileInputStream(absoluteFilePath);
} else if ("GZIP".equalsIgnoreCase(compressor)) {
- inputStream = new GZIPInputStream(new FileInputStream(path));
+ inputStream = new GZIPInputStream(new FileInputStream(absoluteFilePath));
} else if ("BZIP2".equalsIgnoreCase(compressor)) {
- inputStream = new BZip2CompressorInputStream(new FileInputStream(path));
+ inputStream = new BZip2CompressorInputStream(new FileInputStream(absoluteFilePath));
} else if ("SNAPPY".equalsIgnoreCase(compressor)) {
- inputStream = new SnappyInputStream(new FileInputStream(path));
+ inputStream = new SnappyInputStream(new FileInputStream(absoluteFilePath));
} else if ("LZ4".equalsIgnoreCase(compressor)) {
- inputStream = new LZ4BlockInputStream(new FileInputStream(path));
+ inputStream = new LZ4BlockInputStream(new FileInputStream(absoluteFilePath));
} else if ("ZSTD".equalsIgnoreCase(compressor)) {
- inputStream = new ZstdInputStream(new FileInputStream(path));
+ inputStream = new ZstdInputStream(new FileInputStream(absoluteFilePath));
} else {
throw new IOException("Unsupported compressor: " + compressor);
}
@@ -351,19 +331,15 @@ public class LocalCarbonFile implements CarbonFile {
/**
* return the datainputStream which is seek to the offset of file
*
- * @param path
- * @param fileType
* @param bufferSize
* @param offset
* @return DataInputStream
* @throws IOException
*/
@Override
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long offset) throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, fileType);
- FileInputStream fis = new FileInputStream(path);
+ public DataInputStream getDataInputStream(int bufferSize, long offset) throws
+ IOException {
+ FileInputStream fis = new FileInputStream(absoluteFilePath);
long actualSkipSize = 0;
long skipSize = offset;
try {
@@ -379,47 +355,41 @@ public class LocalCarbonFile implements CarbonFile {
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType)
+ public DataOutputStream getDataOutputStream()
throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, FileFactory.FileType.LOCAL);
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path)));
+ return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(absoluteFilePath)));
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long blockSize) throws IOException {
- return getDataOutputStream(path, fileType, bufferSize, blockSize, (short) 1);
+ public DataOutputStream getDataOutputStream(int bufferSize, long blockSize) throws IOException {
+ return getDataOutputStream(bufferSize, blockSize, (short) 1);
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, long blockSize, short replication) throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, fileType);
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path), bufferSize));
+ public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
+ throws IOException {
+ return new DataOutputStream(
+ new BufferedOutputStream(new FileOutputStream(absoluteFilePath), bufferSize));
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, String compressor) throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, fileType);
+ public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
+ throws IOException {
OutputStream outputStream;
if (compressor.isEmpty()) {
- outputStream = new FileOutputStream(path);
+ outputStream = new FileOutputStream(absoluteFilePath);
} else if ("GZIP".equalsIgnoreCase(compressor)) {
- outputStream = new GZIPOutputStream(new FileOutputStream(path));
+ outputStream = new GZIPOutputStream(new FileOutputStream(absoluteFilePath));
} else if ("BZIP2".equalsIgnoreCase(compressor)) {
- outputStream = new BZip2CompressorOutputStream(new FileOutputStream(path));
+ outputStream = new BZip2CompressorOutputStream(new FileOutputStream(absoluteFilePath));
} else if ("SNAPPY".equalsIgnoreCase(compressor)) {
- outputStream = new SnappyOutputStream(new FileOutputStream(path));
+ outputStream = new SnappyOutputStream(new FileOutputStream(absoluteFilePath));
} else if ("LZ4".equalsIgnoreCase(compressor)) {
- outputStream = new LZ4BlockOutputStream(new FileOutputStream(path));
+ outputStream = new LZ4BlockOutputStream(new FileOutputStream(absoluteFilePath));
} else if ("ZSTD".equalsIgnoreCase(compressor)) {
// compression level 1 is cost-effective for sort temp file
// which is not used for storage
- outputStream = new ZstdOutputStream(new FileOutputStream(path), 1);
+ outputStream = new ZstdOutputStream(new FileOutputStream(absoluteFilePath), 1);
} else {
throw new IOException("Unsupported compressor: " + compressor);
}
@@ -432,78 +402,42 @@ public class LocalCarbonFile implements CarbonFile {
}
@Override
- public boolean isFileExist(String filePath, boolean performFileCheck)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath);
- File defaultFile = new File(filePath);
-
+ public boolean isFileExist(boolean performFileCheck) throws IOException {
if (performFileCheck) {
- return defaultFile.exists() && defaultFile.isFile();
+ return file.exists() && file.isFile();
} else {
- return defaultFile.exists();
+ return file.exists();
}
}
@Override
- public boolean isFileExist(String filePath)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath);
- File defaultFile = new File(filePath);
- return defaultFile.exists();
- }
-
- @Override
- public boolean createNewFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath, fileType);
- File file = new File(filePath);
- return file.createNewFile();
+ public boolean isFileExist() throws IOException {
+ return file.exists();
}
@Override
- public boolean createNewFile(String filePath, FileFactory.FileType fileType, boolean doAs,
- final FsPermission permission) throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath, fileType);
- File file = new File(filePath);
+ public boolean createNewFile(final FsPermission permission) throws IOException {
return file.createNewFile();
}
@Override
- public boolean deleteFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath, fileType);
- File file = new File(filePath);
+ public boolean deleteFile() throws IOException {
return FileFactory.deleteAllFilesOfDir(file);
}
@Override
- public boolean mkdirs(String filePath)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath);
- File file = new File(filePath);
+ public boolean mkdirs() throws IOException {
return file.mkdirs();
}
@Override
- public DataOutputStream getDataOutputStreamUsingAppend(String path, FileFactory.FileType fileType)
- throws IOException {
- path = path.replace("\\", "/");
- path = FileFactory.getUpdatedFilePath(path, fileType);
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path, true)));
+ public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
+ return new DataOutputStream(
+ new BufferedOutputStream(new FileOutputStream(absoluteFilePath, true)));
}
@Override
- public boolean createNewLockFile(String filePath, FileFactory.FileType fileType)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- filePath = FileFactory.getUpdatedFilePath(filePath, fileType);
- File file = new File(filePath);
+ public boolean createNewLockFile() throws IOException {
return file.createNewFile();
}
@@ -518,13 +452,13 @@ public class LocalCarbonFile implements CarbonFile {
}
@Override
- public boolean setReplication(String filePath, short replication) throws IOException {
+ public boolean setReplication(short replication) throws IOException {
// local carbon file does not need replication
return true;
}
@Override
- public short getDefaultReplication(String filePath) throws IOException {
+ public short getDefaultReplication() {
return 1;
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/S3CarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/S3CarbonFile.java
index 76c26f4..c0e9aba 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/S3CarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/S3CarbonFile.java
@@ -29,7 +29,6 @@ import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
@@ -42,12 +41,12 @@ public class S3CarbonFile extends HDFSCarbonFile {
super(filePath);
}
- public S3CarbonFile(String filePath, Configuration hadoopConf) {
- super(filePath, hadoopConf);
+ public S3CarbonFile(FileStatus fileStatus) {
+ super(fileStatus);
}
- public S3CarbonFile(Path path, Configuration hadoopConf) {
- super(path, hadoopConf);
+ public S3CarbonFile(String filePath, Configuration hadoopConf) {
+ super(filePath, hadoopConf);
}
/**
@@ -60,22 +59,19 @@ public class S3CarbonFile extends HDFSCarbonFile {
Refer CARBONDATA-2670 for tracking this.
*/
@Override
- public boolean renameForce(String changeToName) {
- FileSystem fs;
+ public boolean renameForce(String changetoName) {
try {
- deleteFile(changeToName, FileFactory.getFileType(changeToName));
- fs = fileStatus.getPath().getFileSystem(hadoopConf);
- return fs.rename(fileStatus.getPath(), new Path(changeToName));
+ // check if any file with the new name exists and delete it.
+ CarbonFile newCarbonFile = FileFactory.getCarbonFile(changetoName);
+ newCarbonFile.delete();
+ // rename the old file to the new name.
+ return fileSystem.rename(path, new Path(changetoName));
} catch (IOException e) {
LOGGER.error("Exception occured: " + e.getMessage(), e);
return false;
}
}
- /**
- * @param listStatus
- * @return
- */
@Override
protected CarbonFile[] getFiles(FileStatus[] listStatus) {
if (listStatus == null) {
@@ -83,50 +79,40 @@ public class S3CarbonFile extends HDFSCarbonFile {
}
CarbonFile[] files = new CarbonFile[listStatus.length];
for (int i = 0; i < files.length; i++) {
- files[i] = new HDFSCarbonFile(listStatus[i]);
+ files[i] = new S3CarbonFile(listStatus[i]);
}
return files;
}
@Override
- public DataOutputStream getDataOutputStreamUsingAppend(String path, FileFactory.FileType fileType)
+ public DataOutputStream getDataOutputStreamUsingAppend()
throws IOException {
- return getDataOutputStream(path, fileType, CarbonCommonConstants.BYTEBUFFER_SIZE, true);
+ return getDataOutputStream(CarbonCommonConstants.BYTEBUFFER_SIZE, true);
}
@Override
- public DataOutputStream getDataOutputStream(String path, FileFactory.FileType fileType,
- int bufferSize, boolean append) throws IOException {
- Path pt = new Path(path);
- FileSystem fileSystem = pt.getFileSystem(FileFactory.getConfiguration());
+ public DataOutputStream getDataOutputStream(int bufferSize, boolean append) throws
+ IOException {
FSDataOutputStream stream;
if (append) {
// append to a file only if file already exists else file not found
// exception will be thrown by hdfs
- if (CarbonUtil.isFileExists(path)) {
- DataInputStream dataInputStream = fileSystem.open(pt);
+ if (CarbonUtil.isFileExists(getAbsolutePath())) {
+ DataInputStream dataInputStream = fileSystem.open(path);
int count = dataInputStream.available();
// create buffer
byte[] byteStreamBuffer = new byte[count];
int bytesRead = dataInputStream.read(byteStreamBuffer);
dataInputStream.close();
- stream = fileSystem.create(pt, true, bufferSize);
- if (bytesRead > 0) {
- stream.write(byteStreamBuffer, 0, bytesRead);
- }
+ stream = fileSystem.create(path, true, bufferSize);
+ stream.write(byteStreamBuffer, 0, bytesRead);
} else {
- stream = fileSystem.create(pt, true, bufferSize);
+ stream = fileSystem.create(path, true, bufferSize);
}
} else {
- stream = fileSystem.create(pt, true, bufferSize);
+ stream = fileSystem.create(path, true, bufferSize);
}
return stream;
}
- @Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new S3CarbonFile(parent, hadoopConf);
- }
-
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/ViewFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/ViewFSCarbonFile.java
index ae7ab84..70a86d8 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/ViewFSCarbonFile.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/filesystem/ViewFSCarbonFile.java
@@ -22,10 +22,8 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.viewfs.ViewFileSystem;
import org.apache.log4j.Logger;
@@ -41,10 +39,6 @@ public class ViewFSCarbonFile extends AbstractDFSCarbonFile {
super(filePath);
}
- public ViewFSCarbonFile(Path path) {
- super(path);
- }
-
public ViewFSCarbonFile(FileStatus fileStatus) {
super(fileStatus);
}
@@ -70,9 +64,9 @@ public class ViewFSCarbonFile extends AbstractDFSCarbonFile {
CarbonFile[] files = listFiles();
if (files != null && files.length >= 1) {
List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
+ for (CarbonFile file : files) {
+ if (fileFilter.accept(file)) {
+ fileList.add(file);
}
}
if (fileList.size() >= 1) {
@@ -85,19 +79,11 @@ public class ViewFSCarbonFile extends AbstractDFSCarbonFile {
}
@Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new ViewFSCarbonFile(parent);
- }
-
- @Override
public boolean renameForce(String changeToName) {
- FileSystem fs;
try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
- if (fs instanceof ViewFileSystem) {
- fs.delete(new Path(changeToName), true);
- fs.rename(fileStatus.getPath(), new Path(changeToName));
+ if (fileSystem instanceof ViewFileSystem) {
+ fileSystem.delete(new Path(changeToName), true);
+ fileSystem.rename(path, new Path(changeToName));
return true;
} else {
return false;
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/impl/DefaultFileTypeProvider.java b/core/src/main/java/org/apache/carbondata/core/datastore/impl/DefaultFileTypeProvider.java
index d2dd6a4..ae51f00 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/impl/DefaultFileTypeProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/impl/DefaultFileTypeProvider.java
@@ -101,7 +101,7 @@ public class DefaultFileTypeProvider implements FileTypeInterface {
FileFactory.FileType fileType = FileFactory.getFileType(path);
switch (fileType) {
case LOCAL:
- return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path, fileType));
+ return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path));
case HDFS:
return new HDFSCarbonFile(path, conf);
case S3:
@@ -111,7 +111,7 @@ public class DefaultFileTypeProvider implements FileTypeInterface {
case VIEWFS:
return new ViewFSCarbonFile(path);
default:
- return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path, fileType));
+ return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path));
}
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
index 95b3d41..711996f 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java
@@ -46,7 +46,8 @@ public final class FileFactory {
*/
private static final Logger LOGGER =
LogServiceFactory.getLogService(FileFactory.class.getName());
- private static Configuration configuration = null;
+
+ private static Configuration configuration;
static {
configuration = new Configuration();
@@ -152,108 +153,98 @@ public final class FileFactory {
return fileFileTypeInterface.getCarbonFile(path, getConfiguration());
}
- public static CarbonFile getCarbonFile(String path, FileType fileType) {
- //TODO ignoring this fileType now to avoid refactoring. Remove the unused argument later.
- return fileFileTypeInterface.getCarbonFile(path, getConfiguration());
- }
-
public static CarbonFile getCarbonFile(String path,
Configuration hadoopConf) {
return fileFileTypeInterface.getCarbonFile(path, hadoopConf);
}
- public static DataInputStream getDataInputStream(String path, FileType fileType)
+ public static DataInputStream getDataInputStream(String path)
throws IOException {
- return getDataInputStream(path, fileType, -1);
+ return getDataInputStream(path, -1);
}
- public static DataInputStream getDataInputStream(String path, FileType fileType,
+ public static DataInputStream getDataInputStream(String path,
Configuration configuration) throws IOException {
- return getDataInputStream(path, fileType, -1, configuration);
+ return getDataInputStream(path, -1, configuration);
}
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize)
+ public static DataInputStream getDataInputStream(String path, int bufferSize)
throws IOException {
- return getDataInputStream(path, fileType, bufferSize, getConfiguration());
+ return getDataInputStream(path, bufferSize, getConfiguration());
}
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
+ public static DataInputStream getDataInputStream(String path, int bufferSize,
Configuration configuration) throws IOException {
- return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, configuration);
+ return getCarbonFile(path, configuration).getDataInputStream(bufferSize);
}
/**
* get data input stream
* @param path
- * @param fileType
* @param bufferSize
* @param compressorName name of compressor to read this file
* @return data input stream
* @throws IOException
*/
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
+ public static DataInputStream getDataInputStream(String path, int bufferSize,
String compressorName) throws IOException {
- return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, compressorName);
+ return getCarbonFile(path).getDataInputStream(bufferSize, compressorName);
}
/**
* return the datainputStream which is seek to the offset of file
*
* @param path
- * @param fileType
* @param bufferSize
* @param offset
* @return DataInputStream
* @throws IOException
*/
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
+ public static DataInputStream getDataInputStream(String path, int bufferSize,
long offset) throws IOException {
- return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, offset);
+ return getCarbonFile(path).getDataInputStream(bufferSize, offset);
}
- public static DataOutputStream getDataOutputStream(String path, FileType fileType)
+ public static DataOutputStream getDataOutputStream(String path)
throws IOException {
- return getCarbonFile(path).getDataOutputStream(path, fileType);
+ return getCarbonFile(path).getDataOutputStream();
}
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
+ public static DataOutputStream getDataOutputStream(String path, int bufferSize,
boolean append) throws IOException {
- return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, append);
+ return getCarbonFile(path).getDataOutputStream(bufferSize, append);
}
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
+ public static DataOutputStream getDataOutputStream(String path, int bufferSize,
long blockSize) throws IOException {
- return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize);
+ return getCarbonFile(path).getDataOutputStream(bufferSize, blockSize);
}
/**
* get data output stream
* @param path file path
- * @param fileType file type
* @param bufferSize write buffer size
* @param blockSize block size
* @param replication replication
* @return data output stream
* @throws IOException if error occurs
*/
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
+ public static DataOutputStream getDataOutputStream(String path, int bufferSize,
long blockSize, short replication) throws IOException {
- return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize,
- replication);
+ return getCarbonFile(path).getDataOutputStream(bufferSize, blockSize, replication);
}
/**
* get data out put stream
* @param path
- * @param fileType
* @param bufferSize
* @param compressorName name of compressor to write this file
* @return data out put stream
* @throws IOException
*/
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
+ public static DataOutputStream getDataOutputStream(String path, int bufferSize,
String compressorName) throws IOException {
- return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, compressorName);
+ return getCarbonFile(path).getDataOutputStream(bufferSize, compressorName);
}
/**
@@ -265,17 +256,7 @@ public final class FileFactory {
*/
public static boolean isFileExist(String filePath, boolean performFileCheck)
throws IOException {
- return getCarbonFile(filePath).isFileExist(filePath, performFileCheck);
- }
-
- /**
- * This method checks the given path exists or not.
- *
- * @param filePath - Path
- * @param fileType - FileType Local/HDFS
- */
- public static boolean isFileExist(String filePath, FileType fileType) throws IOException {
- return getCarbonFile(filePath).isFileExist(filePath);
+ return getCarbonFile(filePath).isFileExist(performFileCheck);
}
/**
@@ -284,23 +265,21 @@ public final class FileFactory {
* @param filePath - Path
*/
public static boolean isFileExist(String filePath) throws IOException {
- return isFileExist(filePath, getFileType(filePath));
+ return getCarbonFile(filePath).isFileExist();
}
- public static boolean createNewFile(String filePath, FileType fileType) throws IOException {
- return createNewFile(filePath, fileType, true, null);
+ public static boolean createNewFile(String filePath) throws IOException {
+ return createNewFile(filePath, null);
}
public static boolean createNewFile(
String filePath,
- FileType fileType,
- boolean doAs,
final FsPermission permission) throws IOException {
- return getCarbonFile(filePath).createNewFile(filePath, fileType, doAs, permission);
+ return getCarbonFile(filePath).createNewFile(permission);
}
- public static boolean deleteFile(String filePath, FileType fileType) throws IOException {
- return getCarbonFile(filePath).deleteFile(filePath, fileType);
+ public static boolean deleteFile(String filePath) throws IOException {
+ return getCarbonFile(filePath).deleteFile();
}
public static boolean deleteAllFilesOfDir(File path) {
@@ -334,40 +313,38 @@ public final class FileFactory {
return path.delete();
}
- public static boolean mkdirs(String filePath, FileType fileType) throws IOException {
- return getCarbonFile(filePath).mkdirs(filePath);
+ public static boolean mkdirs(String filePath) throws IOException {
+ return getCarbonFile(filePath).mkdirs();
}
public static boolean mkdirs(String filePath, Configuration configuration) throws IOException {
- return getCarbonFile(filePath, configuration).mkdirs(filePath);
+ return getCarbonFile(filePath, configuration).mkdirs();
}
/**
* for getting the dataoutput stream using the hdfs filesystem append API.
*
* @param path
- * @param fileType
* @return
* @throws IOException
*/
- public static DataOutputStream getDataOutputStreamUsingAppend(String path, FileType fileType)
+ public static DataOutputStream getDataOutputStreamUsingAppend(String path)
throws IOException {
- return getCarbonFile(path).getDataOutputStreamUsingAppend(path, fileType);
+ return getCarbonFile(path).getDataOutputStreamUsingAppend();
}
/**
* this method will truncate the file to the new size.
* @param path
- * @param fileType
* @param newSize
* @throws IOException
*/
- public static void truncateFile(String path, FileType fileType, long newSize) throws IOException {
+ public static void truncateFile(String path, long newSize) throws IOException {
path = path.replace("\\", "/");
FileChannel fileChannel = null;
- switch (fileType) {
+ switch (getFileType(path)) {
case LOCAL:
- path = getUpdatedFilePath(path, fileType);
+ path = getUpdatedFilePath(path);
fileChannel = new FileOutputStream(path, true).getChannel();
try {
fileChannel.truncate(newSize);
@@ -393,7 +370,7 @@ public final class FileFactory {
} catch (NoSuchMethodException e) {
LOGGER.error("the version of hadoop is below 2.7, there is no 'truncate'"
+ " method in FileSystem, It needs to use 'CarbonFile.truncate'.");
- CarbonFile carbonFile = FileFactory.getCarbonFile(path, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(path);
carbonFile.truncate(path, newSize);
} catch (Exception e) {
LOGGER.error("Other exception occurred while truncating the file " + e.getMessage(), e);
@@ -417,12 +394,11 @@ public final class FileFactory {
* then in case of abrupt shutdown then the stream to that file will be closed.
*
* @param filePath
- * @param fileType
* @return
* @throws IOException
*/
- public static boolean createNewLockFile(String filePath, FileType fileType) throws IOException {
- return getCarbonFile(filePath).createNewLockFile(filePath, fileType);
+ public static boolean createNewLockFile(String filePath) throws IOException {
+ return getCarbonFile(filePath).createNewLockFile();
}
public enum FileType {
@@ -459,11 +435,10 @@ public final class FileFactory {
* it removes the file:/ from the path
*
* @param filePath
- * @param fileType
* @return updated file path without url for local
*/
- public static String getUpdatedFilePath(String filePath, FileType fileType) {
- switch (fileType) {
+ public static String getUpdatedFilePath(String filePath) {
+ switch (getFileType(filePath)) {
case HDFS:
case VIEWFS:
case S3:
@@ -493,19 +468,6 @@ public final class FileFactory {
}
/**
- * below method will be used to update the file path
- * for local type
- * it removes the file:/ from the path
- *
- * @param filePath
- * @return updated file path without url for local
- */
- public static String getUpdatedFilePath(String filePath) {
- FileType fileType = getFileType(filePath);
- return getUpdatedFilePath(filePath, fileType);
- }
-
- /**
* It computes size of directory
*
* @param filePath
@@ -525,7 +487,7 @@ public final class FileFactory {
return fs.getContentSummary(path).getLength();
case LOCAL:
default:
- filePath = getUpdatedFilePath(filePath, fileType);
+ filePath = getUpdatedFilePath(filePath);
File file = new File(filePath);
return FileUtils.sizeOfDirectory(file);
}
@@ -575,7 +537,7 @@ public final class FileFactory {
return;
case LOCAL:
default:
- directoryPath = FileFactory.getUpdatedFilePath(directoryPath, fileType);
+ directoryPath = FileFactory.getUpdatedFilePath(directoryPath);
File file = new File(directoryPath);
if (!file.mkdirs()) {
LOGGER.error(" Failed to create directory path " + directoryPath);
@@ -634,26 +596,22 @@ public final class FileFactory {
* set the file replication
*
* @param path file path
- * @param fileType file type
* @param replication replication
* @return true, if success; false, if failed
* @throws IOException if error occurs
*/
- public static boolean setReplication(String path, FileFactory.FileType fileType,
- short replication) throws IOException {
- return getCarbonFile(path, fileType).setReplication(path, replication);
+ public static boolean setReplication(String path, short replication) throws IOException {
+ return getCarbonFile(path).setReplication(replication);
}
/**
* get the default replication
*
* @param path file path
- * @param fileType file type
* @return replication
* @throws IOException if error occurs
*/
- public static short getDefaultReplication(String path, FileFactory.FileType fileType)
- throws IOException {
- return getCarbonFile(path, fileType).getDefaultReplication(path);
+ public static short getDefaultReplication(String path) {
+ return getCarbonFile(path).getDefaultReplication();
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java b/core/src/main/java/org/apache/carbondata/core/exception/CarbonFileException.java
similarity index 60%
copy from core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java
copy to core/src/main/java/org/apache/carbondata/core/exception/CarbonFileException.java
index d9de031..c4b1ad0 100644
--- a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/exception/CarbonFileException.java
@@ -15,18 +15,18 @@
* limitations under the License.
*/
-package org.apache.carbondata.core.fileoperations;
+package org.apache.carbondata.core.exception;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
+/**
+ * Thrown when any file related operation fails.
+ */
+public class CarbonFileException extends RuntimeException {
-public class AtomicFileOperationFactory {
+ public CarbonFileException(String message, Exception e) {
+ super(message, e);
+ }
- public static AtomicFileOperations getAtomicFileOperations(String filePath) {
- FileFactory.FileType fileType = FileFactory.getFileType(filePath);
- if (fileType == FileFactory.FileType.S3 || fileType == FileFactory.FileType.ALLUXIO) {
- return new AtomicFileOperationS3Impl(filePath);
- } else {
- return new AtomicFileOperationsImpl(filePath, fileType);
- }
+ public CarbonFileException(Exception e) {
+ super(e);
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java
index d9de031..ca6352e 100644
--- a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationFactory.java
@@ -26,7 +26,7 @@ public class AtomicFileOperationFactory {
if (fileType == FileFactory.FileType.S3 || fileType == FileFactory.FileType.ALLUXIO) {
return new AtomicFileOperationS3Impl(filePath);
} else {
- return new AtomicFileOperationsImpl(filePath, fileType);
+ return new AtomicFileOperationsImpl(filePath);
}
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationS3Impl.java b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationS3Impl.java
index f3939a1..43f30f2 100644
--- a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationS3Impl.java
+++ b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationS3Impl.java
@@ -47,7 +47,7 @@ class AtomicFileOperationS3Impl implements AtomicFileOperations {
@Override
public DataInputStream openForRead() throws IOException {
- return FileFactory.getDataInputStream(filePath, FileFactory.getFileType(filePath));
+ return FileFactory.getDataInputStream(filePath);
}
@Override
@@ -60,8 +60,7 @@ class AtomicFileOperationS3Impl implements AtomicFileOperations {
@Override
public DataOutputStream openForWrite(FileWriteOperation operation) throws IOException {
filePath = filePath.replace("\\", "/");
- FileFactory.FileType fileType = FileFactory.getFileType(filePath);
- dataOutStream = FileFactory.getDataOutputStream(filePath, fileType);
+ dataOutStream = FileFactory.getDataOutputStream(filePath);
return dataOutStream;
}
diff --git a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationsImpl.java b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationsImpl.java
index 16e4517..b668bdb 100644
--- a/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationsImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/fileoperations/AtomicFileOperationsImpl.java
@@ -25,7 +25,6 @@ import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.impl.FileFactory;
-import org.apache.carbondata.core.datastore.impl.FileFactory.FileType;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.log4j.Logger;
@@ -39,22 +38,18 @@ class AtomicFileOperationsImpl implements AtomicFileOperations {
LogServiceFactory.getLogService(AtomicFileOperationsImpl.class.getName());
private String filePath;
- private FileType fileType;
-
private String tempWriteFilePath;
private DataOutputStream dataOutStream;
private boolean setFailed;
- AtomicFileOperationsImpl(String filePath, FileType fileType) {
+ AtomicFileOperationsImpl(String filePath) {
this.filePath = filePath;
-
- this.fileType = fileType;
}
@Override
public DataInputStream openForRead() throws IOException {
- return FileFactory.getDataInputStream(filePath, fileType);
+ return FileFactory.getDataInputStream(filePath);
}
@Override
@@ -64,13 +59,13 @@ class AtomicFileOperationsImpl implements AtomicFileOperations {
tempWriteFilePath = filePath + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
- if (FileFactory.isFileExist(tempWriteFilePath, fileType)) {
- FileFactory.getCarbonFile(tempWriteFilePath, fileType).delete();
+ if (FileFactory.isFileExist(tempWriteFilePath)) {
+ FileFactory.getCarbonFile(tempWriteFilePath).delete();
}
- FileFactory.createNewFile(tempWriteFilePath, fileType);
+ FileFactory.createNewFile(tempWriteFilePath);
- dataOutStream = FileFactory.getDataOutputStream(tempWriteFilePath, fileType);
+ dataOutStream = FileFactory.getDataOutputStream(tempWriteFilePath);
return dataOutStream;
@@ -78,10 +73,9 @@ class AtomicFileOperationsImpl implements AtomicFileOperations {
@Override
public void close() throws IOException {
-
if (null != dataOutStream) {
CarbonUtil.closeStream(dataOutStream);
- CarbonFile tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
+ CarbonFile tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
if (!this.setFailed) {
if (!tempFile.renameForce(filePath)) {
throw new IOException(
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlockletWrapper.java b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlockletWrapper.java
index 6093e9a..ad4f804 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlockletWrapper.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/ExtendedBlockletWrapper.java
@@ -82,13 +82,12 @@ public class ExtendedBlockletWrapper implements Writable, Serializable {
try {
final CarbonFile carbonFile = FileFactory.getCarbonFile(folderPath);
boolean isFolderExists = true;
- if (!carbonFile.isFileExist(folderPath)) {
+ if (!carbonFile.isFileExist()) {
LOGGER.warn("Folder:" + folderPath + "doesn't exists, data will be send through netwrok");
isFolderExists = false;
}
if (isFolderExists) {
stream = FileFactory.getDataOutputStream(folderPath + "/" + fileName,
- FileFactory.getFileType(folderPath),
BUFFER_SIZE, BLOCK_SIZE, (short) 1);
writeBlockletToStream(stream, bytes, uniqueLocations, extendedBlockletList);
this.dataSize = stream.size();
@@ -182,7 +181,7 @@ public class ExtendedBlockletWrapper implements Writable, Serializable {
final String folderPath = CarbonUtil.getIndexServerTempPath(tablePath, queryId);
String fileName = new String(bytes, CarbonCommonConstants.DEFAULT_CHARSET);
stream = FileFactory
- .getDataInputStream(folderPath + "/" + fileName, FileFactory.getFileType(folderPath));
+ .getDataInputStream(folderPath + "/" + fileName);
data = new byte[dataSize];
stream.readFully(data);
} finally {
diff --git a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
index 9cf026b..73a001c 100644
--- a/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/indexstore/blockletindex/SegmentIndexFileStore.java
@@ -300,7 +300,7 @@ public class SegmentIndexFileStore {
public void readIndexFile(CarbonFile indexFile) throws IOException {
String indexFilePath = indexFile.getCanonicalPath();
DataInputStream dataInputStream = FileFactory
- .getDataInputStream(indexFilePath, FileFactory.getFileType(indexFilePath), configuration);
+ .getDataInputStream(indexFilePath, configuration);
byte[] bytes = new byte[(int) indexFile.getSize()];
try {
dataInputStream.readFully(bytes);
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/AbstractCarbonLock.java b/core/src/main/java/org/apache/carbondata/core/locks/AbstractCarbonLock.java
index 37f77b2..661cea0 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/AbstractCarbonLock.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/AbstractCarbonLock.java
@@ -94,7 +94,7 @@ public abstract class AbstractCarbonLock implements ICarbonLock {
public boolean releaseLockManually(String lockFile) {
try {
- return FileFactory.deleteFile(lockFile, FileFactory.getFileType(lockFile));
+ return FileFactory.deleteFile(lockFile);
} catch (Exception e) {
return false;
}
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/HdfsFileLock.java b/core/src/main/java/org/apache/carbondata/core/locks/HdfsFileLock.java
index 65cfa0f..ca1343e 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/HdfsFileLock.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/HdfsFileLock.java
@@ -61,13 +61,12 @@ public class HdfsFileLock extends AbstractCarbonLock {
try {
if (null != this.lockFileDir &&
!FileFactory.isFileExist(lockFileDir)) {
- FileFactory.mkdirs(lockFileDir, FileFactory.getFileType(lockFileDir));
+ FileFactory.mkdirs(lockFileDir);
}
if (!FileFactory.isFileExist(lockFilePath)) {
- FileFactory.createNewLockFile(lockFilePath, FileFactory.getFileType(lockFilePath));
+ FileFactory.createNewLockFile(lockFilePath);
}
- dataOutputStream = FileFactory.getDataOutputStreamUsingAppend(lockFilePath,
- FileFactory.getFileType(lockFilePath));
+ dataOutputStream = FileFactory.getDataOutputStreamUsingAppend(lockFilePath);
return true;
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/LocalFileLock.java b/core/src/main/java/org/apache/carbondata/core/locks/LocalFileLock.java
index 1a5c713..5f2b4c4 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/LocalFileLock.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/LocalFileLock.java
@@ -76,10 +76,10 @@ public class LocalFileLock extends AbstractCarbonLock {
public boolean lock() {
try {
if (!FileFactory.isFileExist(lockFileDir)) {
- FileFactory.mkdirs(lockFileDir, FileFactory.getFileType(lockFileDir));
+ FileFactory.mkdirs(lockFileDir);
}
if (!FileFactory.isFileExist(lockFilePath)) {
- FileFactory.createNewLockFile(lockFilePath, FileFactory.getFileType(lockFilePath));
+ FileFactory.createNewLockFile(lockFilePath);
}
channel = FileChannel.open(Paths.get(lockFilePath), StandardOpenOption.WRITE,
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/S3FileLock.java b/core/src/main/java/org/apache/carbondata/core/locks/S3FileLock.java
index eab34ad..1b8688c 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/S3FileLock.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/S3FileLock.java
@@ -86,14 +86,13 @@ public class S3FileLock extends AbstractCarbonLock {
public boolean lock() {
try {
if (!FileFactory.isFileExist(lockFileDir)) {
- FileFactory.mkdirs(lockFileDir, FileFactory.getFileType(lockFileDir));
+ FileFactory.mkdirs(lockFileDir);
}
if (!FileFactory.isFileExist(lockFilePath)) {
- FileFactory.createNewLockFile(lockFilePath, FileFactory.getFileType(lockFilePath));
+ FileFactory.createNewLockFile(lockFilePath);
}
dataOutputStream =
- FileFactory.getDataOutputStreamUsingAppend(lockFilePath,
- FileFactory.getFileType(lockFilePath));
+ FileFactory.getDataOutputStreamUsingAppend(lockFilePath);
return true;
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
index 8418711..0fe92e1 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
@@ -95,7 +95,7 @@ public class SegmentFileStore {
String writePath = CarbonTablePath.getSegmentFilesLocation(tablePath) + "/" + tempFolderLoc;
CarbonFile carbonFile = FileFactory.getCarbonFile(writePath);
if (!carbonFile.exists()) {
- carbonFile.mkdirs(writePath);
+ carbonFile.mkdirs();
}
CarbonFile tempFolder = null;
if (isMergeIndexFlow) {
@@ -233,7 +233,7 @@ public class SegmentFileStore {
String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
if (!carbonFile.exists()) {
- carbonFile.mkdirs(segmentFileFolder);
+ carbonFile.mkdirs();
}
// write segment info to new file.
writeSegmentFile(segmentFile,
@@ -279,7 +279,7 @@ public class SegmentFileStore {
String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
if (!carbonFile.exists()) {
- carbonFile.mkdirs(segmentFileFolder);
+ carbonFile.mkdirs();
}
// write segment info to new file.
writeSegmentFile(segmentFile,
@@ -345,7 +345,7 @@ public class SegmentFileStore {
String segmentFileFolder = CarbonTablePath.getSegmentFilesLocation(tablePath);
CarbonFile carbonFile = FileFactory.getCarbonFile(segmentFileFolder);
if (!carbonFile.exists()) {
- carbonFile.mkdirs(segmentFileFolder);
+ carbonFile.mkdirs();
}
String segmentFileName = genSegmentFileName(segmentId, UUID) + CarbonTablePath.SEGMENT_EXT;
// write segment info to new file.
@@ -625,7 +625,7 @@ public class SegmentFileStore {
AtomicFileOperationFactory.getAtomicFileOperations(segmentFilePath);
try {
- if (!FileFactory.isFileExist(segmentFilePath, FileFactory.getFileType(segmentFilePath))) {
+ if (!FileFactory.isFileExist(segmentFilePath)) {
return null;
}
dataInputStream = fileOperation.openForRead();
@@ -971,10 +971,10 @@ public class SegmentFileStore {
}
if (toBeDeletedIndexFiles.size() > 0) {
for (String dataFile : toBeDeletedIndexFiles) {
- FileFactory.deleteFile(dataFile, FileFactory.getFileType(dataFile));
+ FileFactory.deleteFile(dataFile);
}
for (String dataFile : toBeDeletedDataFiles) {
- FileFactory.deleteFile(dataFile, FileFactory.getFileType(dataFile));
+ FileFactory.deleteFile(dataFile);
}
}
}
@@ -996,21 +996,21 @@ public class SegmentFileStore {
FileFactory.getConfiguration());
Map<String, List<String>> indexFilesMap = fileStore.getIndexFilesMap();
for (Map.Entry<String, List<String>> entry : indexFilesMap.entrySet()) {
- FileFactory.deleteFile(entry.getKey(), FileFactory.getFileType(entry.getKey()));
+ FileFactory.deleteFile(entry.getKey());
for (String file : entry.getValue()) {
String[] deltaFilePaths =
updateStatusManager.getDeleteDeltaFilePath(file, segment.getSegmentNo());
for (String deltaFilePath : deltaFilePaths) {
- FileFactory.deleteFile(deltaFilePath, FileFactory.getFileType(deltaFilePath));
+ FileFactory.deleteFile(deltaFilePath);
}
- FileFactory.deleteFile(file, FileFactory.getFileType(file));
+ FileFactory.deleteFile(file);
}
}
deletePhysicalPartition(partitionSpecs, indexFilesMap, indexOrMergeFiles, tablePath);
String segmentFilePath =
CarbonTablePath.getSegmentFilePath(tablePath, segment.getSegmentFileName());
// Deletes the physical segment file
- FileFactory.deleteFile(segmentFilePath, FileFactory.getFileType(segmentFilePath));
+ FileFactory.deleteFile(segmentFilePath);
}
/**
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
index eb866aa..d0e5a42 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DiskBasedDMSchemaStorageProvider.java
@@ -67,17 +67,16 @@ public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStoragePro
DataOutputStream dataOutputStream = null;
Gson gsonObjectToWrite = new Gson();
String schemaPath = getSchemaPath(storePath, dataMapSchema.getDataMapName());
- FileFactory.FileType fileType = FileFactory.getFileType(schemaPath);
- if (FileFactory.isFileExist(schemaPath, fileType)) {
+ if (FileFactory.isFileExist(schemaPath)) {
throw new IOException(
"DataMap with name " + dataMapSchema.getDataMapName() + " already exists in storage");
}
// write the datamap shema in json format.
try {
- FileFactory.mkdirs(storePath, fileType);
- FileFactory.createNewFile(schemaPath, fileType);
+ FileFactory.mkdirs(storePath);
+ FileFactory.createNewFile(schemaPath);
dataOutputStream =
- FileFactory.getDataOutputStream(schemaPath, fileType);
+ FileFactory.getDataOutputStream(schemaPath);
brWriter = new BufferedWriter(new OutputStreamWriter(dataOutputStream,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
@@ -153,7 +152,7 @@ public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStoragePro
String absolutePath = file.getAbsolutePath();
dataInputStream =
FileFactory.getDataInputStream(
- absolutePath, FileFactory.getFileType(absolutePath));
+ absolutePath);
inStream = new InputStreamReader(dataInputStream,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
buffReader = new BufferedReader(inStream);
@@ -169,7 +168,7 @@ public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStoragePro
public void dropSchema(String dataMapName)
throws IOException {
String schemaPath = getSchemaPath(storePath, dataMapName);
- if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
+ if (!FileFactory.isFileExist(schemaPath)) {
throw new IOException("DataMap with name " + dataMapName + " does not exists in storage");
}
Iterator<DataMapSchema> iterator = dataMapSchemas.iterator();
@@ -180,7 +179,7 @@ public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStoragePro
}
}
touchMDTFile();
- if (!FileFactory.deleteFile(schemaPath, FileFactory.getFileType(schemaPath))) {
+ if (!FileFactory.deleteFile(schemaPath)) {
throw new IOException("DataMap with name " + dataMapName + " cannot be deleted");
}
}
@@ -209,8 +208,6 @@ public class DiskBasedDMSchemaStorageProvider implements DataMapSchemaStoragePro
if (!FileFactory.isFileExist(mdtFilePath)) {
FileFactory.createNewFile(
mdtFilePath,
- FileFactory.getFileType(mdtFilePath),
- true,
new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
}
long lastModifiedTime = System.currentTimeMillis();
diff --git a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
index 8155825..90c3e39 100644
--- a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
@@ -316,7 +316,7 @@ public class CarbonUpdateUtil {
AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
String partitionDir = CarbonTablePath.getPartitionDir(identifier.getTablePath());
CarbonFile file =
- FileFactory.getCarbonFile(partitionDir, FileFactory.getFileType(partitionDir));
+ FileFactory.getCarbonFile(partitionDir);
if (!file.exists()) {
return;
}
@@ -415,7 +415,7 @@ public class CarbonUpdateUtil {
String segmentDirPath = CarbonTablePath.getSegmentPath(tablePath, segment.getSegmentNo());
// scan all the carbondata files and get the latest task ID.
CarbonFile segmentDir =
- FileFactory.getCarbonFile(segmentDirPath, FileFactory.getFileType(segmentDirPath));
+ FileFactory.getCarbonFile(segmentDirPath);
CarbonFile[] carbonDataFiles = segmentDir.listFiles(new CarbonFileFilter() {
@Override
public boolean accept(CarbonFile file) {
@@ -483,7 +483,7 @@ public class CarbonUpdateUtil {
String segmentPath = CarbonTablePath.getSegmentPath(
table.getAbsoluteTableIdentifier().getTablePath(), segment.getLoadName());
CarbonFile segDir =
- FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+ FileFactory.getCarbonFile(segmentPath);
CarbonFile[] allSegmentFiles = segDir.listFiles();
// scan through the segment and find the carbondatafiles and index files.
@@ -594,8 +594,7 @@ public class CarbonUpdateUtil {
String tablePath = table.getAbsoluteTableIdentifier().getTablePath();
CarbonFile metaFolder = FileFactory.getCarbonFile(
- CarbonTablePath.getMetadataPath(tablePath),
- FileFactory.getFileType(CarbonTablePath.getMetadataPath(tablePath)));
+ CarbonTablePath.getMetadataPath(tablePath));
CarbonFile[] invalidUpdateStatusFiles = metaFolder.listFiles(new CarbonFileFilter() {
@Override
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteDeltaFileReaderImpl.java b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteDeltaFileReaderImpl.java
index f3f55a0..a23a388 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteDeltaFileReaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteDeltaFileReaderImpl.java
@@ -39,8 +39,6 @@ public class CarbonDeleteDeltaFileReaderImpl implements CarbonDeleteDeltaFileRea
private String filePath;
- private FileFactory.FileType fileType;
-
private DataInputStream dataInputStream = null;
private InputStreamReader inputStream = null;
@@ -49,12 +47,9 @@ public class CarbonDeleteDeltaFileReaderImpl implements CarbonDeleteDeltaFileRea
/**
* @param filePath
- * @param fileType
*/
- public CarbonDeleteDeltaFileReaderImpl(String filePath, FileFactory.FileType fileType) {
+ public CarbonDeleteDeltaFileReaderImpl(String filePath) {
this.filePath = filePath;
-
- this.fileType = fileType;
}
/**
@@ -71,7 +66,7 @@ public class CarbonDeleteDeltaFileReaderImpl implements CarbonDeleteDeltaFileRea
// Configure Buffer based on our requirement
char[] buffer = new char[DEFAULT_BUFFER_SIZE];
StringWriter sw = new StringWriter();
- dataInputStream = FileFactory.getDataInputStream(filePath, fileType);
+ dataInputStream = FileFactory.getDataInputStream(filePath);
inputStream = new InputStreamReader(dataInputStream,
CarbonCommonConstants.DEFAULT_CHARSET);
int n = 0;
@@ -96,7 +91,7 @@ public class CarbonDeleteDeltaFileReaderImpl implements CarbonDeleteDeltaFileRea
AtomicFileOperations fileOperation =
AtomicFileOperationFactory.getAtomicFileOperations(filePath);
try {
- if (!FileFactory.isFileExist(filePath, FileFactory.getFileType(filePath))) {
+ if (!FileFactory.isFileExist(filePath)) {
return new DeleteDeltaBlockDetails("");
}
dataInputStream = fileOperation.openForRead();
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteFilesDataReader.java b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteFilesDataReader.java
index 9b7a6ed..331bb6c 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteFilesDataReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/CarbonDeleteFilesDataReader.java
@@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastore.impl.FileFactory;
import org.apache.carbondata.core.mutate.DeleteDeltaBlockDetails;
import org.apache.carbondata.core.mutate.DeleteDeltaBlockletDetails;
import org.apache.carbondata.core.mutate.DeleteDeltaVo;
@@ -207,7 +206,7 @@ public class CarbonDeleteFilesDataReader {
@Override
public DeleteDeltaBlockDetails call() throws IOException {
CarbonDeleteDeltaFileReaderImpl deltaFileReader =
- new CarbonDeleteDeltaFileReaderImpl(deltaFile, FileFactory.getFileType(deltaFile));
+ new CarbonDeleteDeltaFileReaderImpl(deltaFile);
return deltaFileReader.readJson();
}
}
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java b/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
index f5ecda6..d087323 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/ThriftReader.java
@@ -97,8 +97,7 @@ public class ThriftReader {
*/
public void open() throws IOException {
Configuration conf = configuration != null ? configuration : FileFactory.getConfiguration();
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
- dataInputStream = FileFactory.getDataInputStream(fileName, fileType, conf);
+ dataInputStream = FileFactory.getDataInputStream(fileName, conf);
binaryIn = new TCompactProtocol(new TIOStreamTransport(dataInputStream));
}
diff --git a/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java b/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
index 0edad87..fe57013 100644
--- a/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImpl.java
@@ -140,8 +140,7 @@ public class CarbonDictionarySortIndexReaderImpl implements CarbonDictionarySort
getChunkMetaObjectForLastSegmentEntry();
long dictOffset = chunkMetaObjectForLastSegmentEntry.getEnd_offset();
this.sortIndexFilePath = dictionaryColumnUniqueIdentifier.getSortIndexFilePath(dictOffset);
- if (!FileFactory
- .isFileExist(this.sortIndexFilePath, FileFactory.getFileType(this.sortIndexFilePath))) {
+ if (!FileFactory.isFileExist(this.sortIndexFilePath)) {
this.sortIndexFilePath = dictionaryColumnUniqueIdentifier.getSortIndexFilePath();
}
} catch (IOException e) {
diff --git a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
index 7699575..bc52082 100755
--- a/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
+++ b/core/src/main/java/org/apache/carbondata/core/statusmanager/SegmentStatusManager.java
@@ -100,10 +100,10 @@ public class SegmentStatusManager {
public static long getTableStatusLastModifiedTime(AbsoluteTableIdentifier identifier)
throws IOException {
String tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath());
- if (!FileFactory.isFileExist(tableStatusPath, FileFactory.getFileType(tableStatusPath))) {
+ if (!FileFactory.isFileExist(tableStatusPath)) {
return 0L;
} else {
- return FileFactory.getCarbonFile(tableStatusPath, FileFactory.getFileType(tableStatusPath))
+ return FileFactory.getCarbonFile(tableStatusPath)
.getLastModifiedTime();
}
}
@@ -264,7 +264,7 @@ public class SegmentStatusManager {
AtomicFileOperations fileOperation =
AtomicFileOperationFactory.getAtomicFileOperations(tableStatusPath);
- if (!FileFactory.isFileExist(tableStatusPath, FileFactory.getFileType(tableStatusPath))) {
+ if (!FileFactory.isFileExist(tableStatusPath)) {
return new LoadMetadataDetails[0];
}
@@ -388,7 +388,7 @@ public class SegmentStatusManager {
String dataLoadLocation = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath());
LoadMetadataDetails[] listOfLoadFolderDetailsArray = null;
- if (!FileFactory.isFileExist(dataLoadLocation, FileFactory.getFileType(dataLoadLocation))) {
+ if (!FileFactory.isFileExist(dataLoadLocation)) {
// log error.
LOG.error("Load metadata file is not present.");
return loadIds;
@@ -467,7 +467,7 @@ public class SegmentStatusManager {
String dataLoadLocation = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath());
LoadMetadataDetails[] listOfLoadFolderDetailsArray = null;
- if (!FileFactory.isFileExist(dataLoadLocation, FileFactory.getFileType(dataLoadLocation))) {
+ if (!FileFactory.isFileExist(dataLoadLocation)) {
// Table status file is not present, maybe table is empty, ignore this operation
LOG.warn("Trying to update table metadata file which is not present.");
return invalidLoadTimestamps;
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 fd0df29..4cefea2 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
@@ -198,7 +198,7 @@ public class SegmentUpdateStatusManager {
String segmentPath = CarbonTablePath.getSegmentPath(
identifier.getTablePath(), segmentId);
CarbonFile segDir =
- FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+ FileFactory.getCarbonFile(segmentPath);
for (LoadMetadataDetails eachSeg : segmentDetails) {
if (eachSeg.getLoadName().equalsIgnoreCase(segmentId)) {
// if the segment is found then take the start and end time stamp.
@@ -406,7 +406,7 @@ public class SegmentUpdateStatusManager {
String segmentPath = CarbonTablePath.getSegmentPath(
identifier.getTablePath(), segmentId.getSegmentNo());
CarbonFile segDir =
- FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+ FileFactory.getCarbonFile(segmentPath);
for (SegmentUpdateDetails block : updateDetails) {
if ((block.getBlockName().equalsIgnoreCase(blockName)) &&
(block.getSegmentName().equalsIgnoreCase(segmentId.getSegmentNo()))
@@ -661,7 +661,7 @@ public class SegmentUpdateStatusManager {
try {
if (!FileFactory
- .isFileExist(tableUpdateStatusPath, FileFactory.getFileType(tableUpdateStatusPath))) {
+ .isFileExist(tableUpdateStatusPath)) {
return new SegmentUpdateDetails[0];
}
dataInputStream = fileOperation.openForRead();
diff --git a/core/src/main/java/org/apache/carbondata/core/stream/StreamPruner.java b/core/src/main/java/org/apache/carbondata/core/stream/StreamPruner.java
index 3bb2d78..c92a8a1 100644
--- a/core/src/main/java/org/apache/carbondata/core/stream/StreamPruner.java
+++ b/core/src/main/java/org/apache/carbondata/core/stream/StreamPruner.java
@@ -117,8 +117,7 @@ public class StreamPruner {
String segmentDir = CarbonTablePath.getSegmentPath(
carbonTable.getAbsoluteTableIdentifier().getTablePath(), segment.getSegmentNo());
String indexFile = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);
- FileFactory.FileType fileType = FileFactory.getFileType(indexFile);
- if (FileFactory.isFileExist(indexFile, fileType)) {
+ if (FileFactory.isFileExist(indexFile)) {
CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
indexReader.openThriftReader(indexFile);
try {
diff --git a/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java b/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java
index f22462b..eb064ed 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/BlockletDataMapUtil.java
@@ -186,7 +186,7 @@ public class BlockletDataMapUtil {
if (!FileFactory.isFileExist(carbonDataFile)) {
return null;
}
- CarbonFile carbonFile = FileFactory.getCarbonFile(carbonDataFile, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(carbonDataFile);
return new BlockMetaInfo(new String[] { "localhost" }, carbonFile.getSize());
default:
return fileNameToMetaInfoMapping.get(carbonDataFile);
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 4d5de6b..05ac316 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
@@ -712,8 +712,7 @@ public final class CarbonUtil {
*/
public static boolean isFileExists(String fileName) {
try {
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
- if (FileFactory.isFileExist(fileName, fileType)) {
+ if (FileFactory.isFileExist(fileName)) {
return true;
}
} catch (IOException e) {
@@ -728,11 +727,10 @@ public final class CarbonUtil {
public static boolean checkAndCreateFolder(String path) {
boolean created = false;
try {
- FileFactory.FileType fileType = FileFactory.getFileType(path);
- if (FileFactory.isFileExist(path, fileType)) {
+ if (FileFactory.isFileExist(path)) {
created = true;
} else {
- created = FileFactory.mkdirs(path, fileType);
+ created = FileFactory.mkdirs(path);
}
} catch (IOException e) {
LOGGER.error(e.getMessage(), e);
@@ -747,8 +745,7 @@ public final class CarbonUtil {
public static boolean checkAndCreateFolderWithPermission(String path) {
boolean created = false;
try {
- FileFactory.FileType fileType = FileFactory.getFileType(path);
- if (FileFactory.isFileExist(path, fileType)) {
+ if (FileFactory.isFileExist(path)) {
created = true;
} else {
FileFactory.createDirectoryAndSetPermission(path,
@@ -765,8 +762,7 @@ public final class CarbonUtil {
* This method will return the size of a given file
*/
public static long getFileSize(String filePath) {
- FileFactory.FileType fileType = FileFactory.getFileType(filePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(filePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(filePath);
return carbonFile.getSize();
}
@@ -1017,7 +1013,7 @@ public final class CarbonUtil {
.getTimeStampFromFileName(tableBlockInfoList.get(0).getFilePath()),
tableBlockInfoList.get(0).getVersion());
CarbonFile carbonFile = FileFactory
- .getCarbonFile(carbonIndexFilePath, FileFactory.getFileType(carbonIndexFilePath));
+ .getCarbonFile(carbonIndexFilePath);
// in case of carbonIndex file whole file is meta only so reading complete file.
return carbonFile.getSize();
}
@@ -1268,7 +1264,7 @@ public final class CarbonUtil {
try {
fileReader =
- FileFactory.getDataInputStream(csvFilePath, FileFactory.getFileType(csvFilePath));
+ FileFactory.getDataInputStream(csvFilePath);
bufferedReader = new BufferedReader(new InputStreamReader(fileReader,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
readLine = bufferedReader.readLine();
@@ -1287,7 +1283,7 @@ public final class CarbonUtil {
try {
fileReader = FileFactory.getDataInputStream(
- csvFilePath, FileFactory.getFileType(csvFilePath), -1, hadoopConf);
+ csvFilePath, -1, hadoopConf);
bufferedReader = new BufferedReader(new InputStreamReader(fileReader,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
readLine = bufferedReader.readLine();
@@ -2352,9 +2348,8 @@ public final class CarbonUtil {
public static void dropDatabaseDirectory(String databasePath)
throws IOException, InterruptedException {
- FileFactory.FileType fileType = FileFactory.getFileType(databasePath);
- if (FileFactory.isFileExist(databasePath, fileType)) {
- CarbonFile dbPath = FileFactory.getCarbonFile(databasePath, fileType);
+ if (FileFactory.isFileExist(databasePath)) {
+ CarbonFile dbPath = FileFactory.getCarbonFile(databasePath);
CarbonUtil.deleteFoldersAndFiles(dbPath);
}
}
@@ -2535,10 +2530,9 @@ public final class CarbonUtil {
}
String tableStatusPath =
CarbonTablePath.getTableStatusFilePath(identifier.getTablePath());
- if (FileFactory.isFileExist(tableStatusPath, FileFactory.getFileType(tableStatusPath))) {
+ if (FileFactory.isFileExist(tableStatusPath)) {
lastUpdateTime =
- FileFactory.getCarbonFile(tableStatusPath, FileFactory.getFileType(tableStatusPath))
- .getLastModifiedTime();
+ FileFactory.getCarbonFile(tableStatusPath).getLastModifiedTime();
}
if (!FileFactory.isFileExist(metadataPath)) {
dataSize = FileFactory.getDirectorySize(carbonTable.getTablePath());
@@ -2598,7 +2592,7 @@ public final class CarbonUtil {
break;
case LOCAL:
default:
- segmentPath = FileFactory.getUpdatedFilePath(segmentPath, fileType);
+ segmentPath = FileFactory.getUpdatedFilePath(segmentPath);
File file = new File(segmentPath);
File[] segmentFiles = file.listFiles();
if (null != segmentFiles) {
@@ -2827,10 +2821,10 @@ public final class CarbonUtil {
+ " (bytes");
}
dataOutputStream = FileFactory
- .getDataOutputStream(carbonStoreFilePath, FileFactory.getFileType(carbonStoreFilePath),
+ .getDataOutputStream(carbonStoreFilePath,
bufferSize, blockSize);
dataInputStream = FileFactory
- .getDataInputStream(localFilePath, FileFactory.getFileType(localFilePath), bufferSize);
+ .getDataInputStream(localFilePath, bufferSize);
IOUtils.copyBytes(dataInputStream, dataOutputStream, bufferSize);
} finally {
CarbonUtil.closeStream(dataInputStream);
@@ -3173,8 +3167,7 @@ public final class CarbonUtil {
// if the carbontable is support flat folder
if (supportFlatFolder) {
segmentPath = carbonTable.getTablePath();
- FileFactory.FileType fileType = FileFactory.getFileType(segmentPath);
- if (FileFactory.isFileExist(segmentPath, fileType)) {
+ if (FileFactory.isFileExist(segmentPath)) {
fileStore.readAllIIndexOfSegment(segmentPath);
Map<String, byte[]> carbonIndexMap = fileStore.getCarbonIndexMap();
if (carbonIndexMap.size() == 0) {
@@ -3204,8 +3197,7 @@ public final class CarbonUtil {
// get the carbon index file header from a valid segment
for (Segment segment : validSegments) {
segmentPath = carbonTable.getSegmentPath(segment.getSegmentNo());
- FileFactory.FileType fileType = FileFactory.getFileType(segmentPath);
- if (FileFactory.isFileExist(segmentPath, fileType)) {
+ if (FileFactory.isFileExist(segmentPath)) {
fileStore.readAllIIndexOfSegment(segmentPath);
Map<String, byte[]> carbonIndexMap = fileStore.getCarbonIndexMap();
if (carbonIndexMap.size() == 0) {
@@ -3382,7 +3374,7 @@ public final class CarbonUtil {
throws IOException {
final String path = getIndexServerTempPath(tablePath, queryId);
CarbonFile file = FileFactory.getCarbonFile(path);
- if (!file.mkdirs(path)) {
+ if (!file.mkdirs()) {
LOGGER.info("Unable to create table directory for index server");
return null;
} else {
diff --git a/core/src/main/java/org/apache/carbondata/core/util/DeleteLoadFolders.java b/core/src/main/java/org/apache/carbondata/core/util/DeleteLoadFolders.java
index fa63a10..4287c42 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/DeleteLoadFolders.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/DeleteLoadFolders.java
@@ -122,8 +122,8 @@ public final class DeleteLoadFolders {
} else {
String path = getSegmentPath(carbonTable.getAbsoluteTableIdentifier(), oneLoad);
boolean status = false;
- if (FileFactory.isFileExist(path, FileFactory.getFileType(path))) {
- CarbonFile file = FileFactory.getCarbonFile(path, FileFactory.getFileType(path));
+ if (FileFactory.isFileExist(path)) {
+ CarbonFile file = FileFactory.getCarbonFile(path);
CarbonFile[] filesToBeDeleted = file.listFiles(new CarbonFileFilter() {
@Override
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 6d1a4ff..b72589d 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
@@ -239,7 +239,7 @@ public class CarbonTablePath {
final String segmentId, final String bucketNumber) {
String segmentDir = getSegmentPath(tablePath, segmentId);
CarbonFile carbonFile =
- FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir));
+ FileFactory.getCarbonFile(segmentDir);
CarbonFile[] files = carbonFile.listFiles(new CarbonFileFilter() {
@Override
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDeleteDeltaWriterImpl.java b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDeleteDeltaWriterImpl.java
index 497cc26..f4b6631 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDeleteDeltaWriterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDeleteDeltaWriterImpl.java
@@ -43,17 +43,13 @@ public class CarbonDeleteDeltaWriterImpl implements CarbonDeleteDeltaWriter {
private String filePath;
- private FileFactory.FileType fileType;
-
private DataOutputStream dataOutStream = null;
/**
* @param filePath
- * @param fileType
*/
- public CarbonDeleteDeltaWriterImpl(String filePath, FileFactory.FileType fileType) {
+ public CarbonDeleteDeltaWriterImpl(String filePath) {
this.filePath = filePath;
- this.fileType = fileType;
}
@@ -66,8 +62,8 @@ public class CarbonDeleteDeltaWriterImpl implements CarbonDeleteDeltaWriter {
public void write(DeleteDeltaBlockDetails deleteDeltaBlockDetails) throws IOException {
BufferedWriter brWriter = null;
try {
- FileFactory.createNewFile(filePath, fileType);
- dataOutStream = FileFactory.getDataOutputStream(filePath, fileType);
+ FileFactory.createNewFile(filePath);
+ dataOutStream = FileFactory.getDataOutputStream(filePath);
Gson gsonObjectToWrite = new Gson();
brWriter = new BufferedWriter(new OutputStreamWriter(dataOutStream,
CarbonCommonConstants.DEFAULT_CHARSET));
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
index a9069e7..a687fd3 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImpl.java
@@ -235,7 +235,7 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
this.chunk_start_offset = CarbonUtil.getFileSize(this.dictionaryFilePath);
validateDictionaryFileOffsetWithLastSegmentEntryOffset();
} else if (dictFileExists) {
- FileFactory.getCarbonFile(dictionaryFilePath, FileFactory.getFileType(dictionaryFilePath))
+ FileFactory.getCarbonFile(dictionaryFilePath)
.delete();
}
openThriftWriter(this.dictionaryFilePath);
@@ -282,8 +282,7 @@ public class CarbonDictionaryWriterImpl implements CarbonDictionaryWriter {
LOGGER.info("some inconsistency in dictionary file for column "
+ this.dictionaryColumnUniqueIdentifier.getColumnIdentifier());
// truncate the dictionary data till chunk meta end offset
- FileFactory.FileType fileType = FileFactory.getFileType(this.dictionaryFilePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(this.dictionaryFilePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(this.dictionaryFilePath);
boolean truncateSuccess = carbonFile
.truncate(this.dictionaryFilePath, chunkMetaObjectForLastSegmentEntry.getEnd_offset());
if (!truncateSuccess) {
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/ThriftWriter.java b/core/src/main/java/org/apache/carbondata/core/writer/ThriftWriter.java
index 93cd238..5d0e4b5 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/ThriftWriter.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/ThriftWriter.java
@@ -79,8 +79,7 @@ public class ThriftWriter {
* Open the file for writing.
*/
public void open() throws IOException {
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
- dataOutputStream = FileFactory.getDataOutputStream(fileName, fileType, bufferSize, append);
+ dataOutputStream = FileFactory.getDataOutputStream(fileName, bufferSize, append);
binaryOut = new TCompactProtocol(new TIOStreamTransport(dataOutputStream));
}
diff --git a/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java b/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
index de27bab..4824a00 100644
--- a/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/writer/sortindex/CarbonDictionarySortIndexWriterImpl.java
@@ -148,7 +148,7 @@ public class CarbonDictionarySortIndexWriterImpl implements CarbonDictionarySort
* @param dictPath
*/
protected void cleanUpOldSortIndex(String dictPath) {
- CarbonFile dictFile = FileFactory.getCarbonFile(dictPath, FileFactory.getFileType(dictPath));
+ CarbonFile dictFile = FileFactory.getCarbonFile(dictPath);
CarbonFile[] files = CarbonTablePath.getSortIndexFiles(dictFile.getParentFile(),
dictionaryColumnUniqueIdentifier.getColumnIdentifier().getColumnId());
int maxTime;
diff --git a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
index 8008a94..5263faf 100644
--- a/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/cache/dictionary/AbstractDictionaryCacheTest.java
@@ -115,8 +115,7 @@ public class AbstractDictionaryCacheTest {
* this method will delete the store path
*/
protected void deleteStorePath() {
- FileFactory.FileType fileType = FileFactory.getFileType(this.carbonStorePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(this.carbonStorePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(this.carbonStorePath);
deleteRecursiveSilent(carbonFile);
}
diff --git a/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/FileFactoryImplUnitTest.java b/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/FileFactoryImplUnitTest.java
index 804dc94..cd685d9 100644
--- a/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/FileFactoryImplUnitTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/FileFactoryImplUnitTest.java
@@ -70,11 +70,11 @@ public class FileFactoryImplUnitTest {
}
@Test public void testFileExistsForVIEWFSType() throws IOException {
- FileFactory.isFileExist("fakefilePath", FileFactory.FileType.VIEWFS);
+ FileFactory.isFileExist("fakefilePath");
}
@Test public void testFileExistsForDefaultType() throws IOException {
- FileFactory.isFileExist("fakefilePath", FileFactory.FileType.LOCAL);
+ FileFactory.isFileExist("fakefilePath");
}
@Test public void testFileExistsForDefaultTypeWithPerformFileCheck() throws IOException {
@@ -95,25 +95,25 @@ public class FileFactoryImplUnitTest {
@Test public void testCreateNewFileWithDefaultFileType() throws IOException {
cleanUp();
- assertTrue(FileFactory.createNewFile(filePath, FileFactory.FileType.LOCAL));
+ assertTrue(FileFactory.createNewFile(filePath));
}
@Test public void testCreateNewLockFileWithDefaultFileType() throws IOException {
cleanUp();
- assertTrue(FileFactory.createNewLockFile(filePath, FileFactory.FileType.LOCAL));
+ assertTrue(FileFactory.createNewLockFile(filePath));
}
@Test public void testCreateNewLockFileWithViewFsFileType() throws IOException {
cleanUp();
- assertTrue(FileFactory.createNewLockFile(filePath, FileFactory.FileType.VIEWFS));
+ assertTrue(FileFactory.createNewLockFile(filePath));
}
@Test public void testCreateNewLockFileWithViewFsFileTypeWhenFileExists() throws IOException {
- assertFalse(FileFactory.createNewLockFile(filePath, FileFactory.FileType.VIEWFS));
+ assertFalse(FileFactory.createNewLockFile(filePath));
}
@Test public void testCreateNewFileWithDefaultFileTypeWhenFileExists() throws IOException {
- assertFalse(FileFactory.createNewFile(filePath, FileFactory.FileType.LOCAL));
+ assertFalse(FileFactory.createNewFile(filePath));
}
@Test public void testCreateNewFileWithVIEWFSFileType() throws IOException {
@@ -121,11 +121,11 @@ public class FileFactoryImplUnitTest {
if (file.exists()) {
file.delete();
}
- assertTrue(FileFactory.createNewFile(filePath, FileFactory.FileType.VIEWFS));
+ assertTrue(FileFactory.createNewFile(filePath));
}
@Test public void testCreateNewFileWithVIEWFSFileTypeWhenFileExists() throws IOException {
- assertFalse(FileFactory.createNewFile(filePath, FileFactory.FileType.VIEWFS));
+ assertFalse(FileFactory.createNewFile(filePath));
}
@Test public void testMkDirWithVIEWFSFileType() throws IOException {
@@ -137,13 +137,13 @@ public class FileFactoryImplUnitTest {
}
};
tearDown();
- assertTrue(FileFactory.mkdirs(filePath, FileFactory.FileType.VIEWFS));
+ assertTrue(FileFactory.mkdirs(filePath));
}
@Test public void testGetDataOutputStreamUsingAppendeForException() throws IOException {
DataOutputStream outputStream = null;
try {
- outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath, FileFactory.FileType.VIEWFS);
+ outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath);
} catch (Exception exception) {
assertEquals("Not supported", exception.getMessage());
} finally {
@@ -154,20 +154,20 @@ public class FileFactoryImplUnitTest {
}
@Test public void getDataOutputStreamForVIEWFSType() throws IOException {
- DataOutputStream outputStream = FileFactory.getDataOutputStream(filePath, FileFactory.FileType.VIEWFS);
+ DataOutputStream outputStream = FileFactory.getDataOutputStream(filePath);
assertNotNull(outputStream);
outputStream.close();
}
@Test public void getDataOutputStreamForLocalType() throws IOException {
- DataOutputStream outputStream = FileFactory.getDataOutputStream(filePath, FileFactory.FileType.LOCAL);
+ DataOutputStream outputStream = FileFactory.getDataOutputStream(filePath);
assertNotNull(outputStream);
outputStream.close();
}
@Test public void testGetCarbonFile() throws IOException {
- FileFactory.getDataOutputStream(filePath, FileFactory.FileType.VIEWFS);
- assertNotNull(FileFactory.getCarbonFile(filePath, FileFactory.FileType.HDFS));
+ FileFactory.getDataOutputStream(filePath);
+ assertNotNull(FileFactory.getCarbonFile(filePath));
}
@Test public void testTruncateFile() {
@@ -187,7 +187,6 @@ public class FileFactoryImplUnitTest {
// truncate file to 4000 bytes
FileFactory.truncateFile(
path,
- FileFactory.getFileType(path),
4000);
file = FileFactory.getCarbonFile(path);
assertEquals(file.getSize(), 4000L);
@@ -204,7 +203,7 @@ public class FileFactoryImplUnitTest {
}
if (path != null) {
try {
- FileFactory.deleteFile(path, FileFactory.getFileType(path));
+ FileFactory.deleteFile(path);
} catch (IOException e) {
e.printStackTrace();
}
@@ -253,4 +252,3 @@ public class FileFactoryImplUnitTest {
}
}
-
diff --git a/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/TestFileProvider.java b/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/TestFileProvider.java
index ce71cbf..cf643d6 100644
--- a/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/TestFileProvider.java
+++ b/core/src/test/java/org/apache/carbondata/core/carbon/datastorage/filesystem/store/impl/TestFileProvider.java
@@ -40,7 +40,7 @@ public class TestFileProvider implements FileTypeInterface {
FileFactory.FileType fileType = FileFactory.getFileType(path);
switch (fileType) {
case LOCAL:
- return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path, fileType));
+ return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path));
case HDFS:
return new HDFSCarbonFile(path, configuration);
case S3:
@@ -50,7 +50,7 @@ public class TestFileProvider implements FileTypeInterface {
case VIEWFS:
return new ViewFSCarbonFile(path);
default:
- return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path, fileType));
+ return new LocalCarbonFile(FileFactory.getUpdatedFilePath(path));
}
}
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/CompressdFileTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/CompressdFileTest.java
index 6b228e0..a21b646 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/CompressdFileTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/CompressdFileTest.java
@@ -98,7 +98,7 @@ public class CompressdFileTest
try {
fileReader =
- FileFactory.getDataInputStream(path, FileType.HDFS);
+ FileFactory.getDataInputStream(path);
bufferedReader = new BufferedReader(new InputStreamReader(fileReader,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
readLine = bufferedReader.readLine();
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFileTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFileTest.java
index 58cf58e..7c8f942 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFileTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/AlluxioCarbonFileTest.java
@@ -25,9 +25,11 @@ import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.util.Progressable;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -89,16 +91,29 @@ public class AlluxioCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
- throw new IOException();
+ return new DistributedFileSystem();
}
};
+ new MockUp<DistributedFileSystem>() {
+ @Mock
+ public void rename(Path src, Path dst, final Options.Rename... options) throws IOException {
+ throw new IOException();
+ }
+ };
alluxioCarbonFile = new AlluxioCarbonFile(fileStatus);
alluxioCarbonFile.renameForce(fileName);
}
@Test
public void testListFilesWithOutDirectoryPermission() {
+ new MockUp<LocalFileSystem>() {
+ @Mock
+ public FileStatus[] listStatus(Path p) throws IOException {
+ return null;
+ }
+
+ };
alluxioCarbonFile = new AlluxioCarbonFile(fileStatusWithOutDirectoryPermission);
assertArrayEquals(alluxioCarbonFile.listFiles(), new CarbonFile[0]);
}
@@ -174,11 +189,11 @@ public class AlluxioCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
- throw new IOException();
+ return new DistributedFileSystem();
}
};
- new MockUp<DummyAlluxioFileSystem>() {
+ new MockUp<DistributedFileSystem>() {
@Mock
public FileStatus[] listStatus(Path var1) throws IOException {
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFileTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFileTest.java
index 8adefb3..268b814 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFileTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/HDFSCarbonFileTest.java
@@ -29,6 +29,7 @@ import mockit.MockUp;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Options;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
@@ -41,6 +42,7 @@ import org.junit.Test;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -100,12 +102,18 @@ public class HDFSCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
- throw new IOException();
+ return new DistributedFileSystem();
}
};
+ new MockUp<DistributedFileSystem>() {
+ @Mock
+ public void rename(Path src, Path dst, final Options.Rename... options) throws IOException {
+ throw new IOException();
+ }
+ };
hdfsCarbonFile = new HDFSCarbonFile(fileStatus);
- hdfsCarbonFile.renameForce(fileName);
+ assertFalse(hdfsCarbonFile.renameForce(fileName));
}
@Test
@@ -134,6 +142,13 @@ public class HDFSCarbonFileTest {
}
};
+ new MockUp<LocalFileSystem>() {
+ @Mock
+ public FileStatus[] listStatus(Path p) throws IOException {
+ return null;
+ }
+
+ };
assertArrayEquals(hdfsCarbonFile.listFiles(), new CarbonFile[0]);
}
@@ -200,7 +215,7 @@ public class HDFSCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
- throw new IOException();
+ return new DistributedFileSystem();
}
};
@@ -298,10 +313,10 @@ public class HDFSCarbonFileTest {
}
};
- new MockUp<FileStatus>() {
+ new MockUp<DistributedFileSystem>() {
@Mock
- public Path getPath() {
- return new Path(fileName);
+ public FileStatus getFileStatus(Path f) throws IOException {
+ return new FileStatus();
}
};
@@ -362,7 +377,6 @@ public class HDFSCarbonFileTest {
@Test
public void testForNonDisributedSystem() {
- new HDFSCarbonFile(fileStatus);
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
@@ -376,7 +390,8 @@ public class HDFSCarbonFileTest {
return true;
}
};
- assertEquals(hdfsCarbonFile.renameForce(fileName), true);
+
+ assertEquals(new HDFSCarbonFile(fileStatus).renameForce(fileName), true);
}
@Test
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFileTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFileTest.java
index c60c3fa..316cca1 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFileTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFileTest.java
@@ -162,7 +162,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public boolean isFileExist(String filePath, FileFactory.FileType fileType) throws IOException {
+ public boolean isFileExist(String filePath) throws IOException {
{
return true;
}
@@ -176,7 +176,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public boolean createNewFile(String filePath, FileFactory.FileType fileType) throws IOException {
+ public boolean createNewFile(String filePath) throws IOException {
{
return true;
}
@@ -184,7 +184,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public CarbonFile getCarbonFile(String path, FileFactory.FileType fileType) {
+ public CarbonFile getCarbonFile(String path) {
{
return new LocalCarbonFile(path);
}
@@ -231,7 +231,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public boolean isFileExist(String filePath, FileFactory.FileType fileType) throws IOException {
+ public boolean isFileExist(String filePath) throws IOException {
{
return true;
}
@@ -239,7 +239,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public CarbonFile getCarbonFile(String path, FileFactory.FileType fileType) {
+ public CarbonFile getCarbonFile(String path) {
{
return new LocalCarbonFile(path);
}
@@ -253,7 +253,7 @@ public class LocalCarbonFileTest {
};
new MockUp<FileFactory>() {
@Mock
- public boolean createNewFile(String filePath, FileFactory.FileType fileType) throws IOException {
+ public boolean createNewFile(String filePath) throws IOException {
{
throw new IOException();
}
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/ViewFsCarbonFileTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/ViewFsCarbonFileTest.java
index 82d1501..7320c0c 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/ViewFsCarbonFileTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/filesystem/ViewFsCarbonFileTest.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.viewfs.ViewFileSystem;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.web.WebHdfsFileSystem;
+import org.apache.hadoop.security.AccessControlException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -84,22 +85,45 @@ public class ViewFsCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
+ return new ViewFileSystem();
+ }
+ };
+ new MockUp<ViewFileSystem>() {
+ @Mock
+ public boolean delete(Path f, boolean recursive) throws AccessControlException, FileNotFoundException, IOException {
throw new IOException();
}
-
};
viewFSCarbonFile = new ViewFSCarbonFile(fileStatus);
- viewFSCarbonFile.renameForce(fileName);
+ assert(!viewFSCarbonFile.renameForce(fileName));
}
@Test
public void testListFilesWithOutDirectoryPermission() {
+ new MockUp<Path>() {
+ @Mock public FileSystem getFileSystem(Configuration conf) throws IOException {
+ return new ViewFileSystem();
+ }
+ };
+ new MockUp<ViewFileSystem>() {
+ @Mock
+ public FileStatus[] listStatus(Path f)
+ throws AccessControlException, FileNotFoundException, IOException {
+ return null;
+ }
+ };
viewFSCarbonFile = new ViewFSCarbonFile(fileStatusWithOutDirectoryPermission);
assertArrayEquals(viewFSCarbonFile.listFiles(), new CarbonFile[0]);
}
@Test
public void testConstructorWithFilePath() {
+ new MockUp<Path>() {
+ @Mock
+ public FileSystem getFileSystem(Configuration conf) throws IOException {
+ return new ViewFileSystem();
+ }
+ };
viewFSCarbonFile = new ViewFSCarbonFile(file.getAbsolutePath());
assertTrue(viewFSCarbonFile instanceof ViewFSCarbonFile);
}
@@ -137,7 +161,6 @@ public class ViewFsCarbonFileTest {
@Test
public void testListDirectory() {
- viewFSCarbonFile = new ViewFSCarbonFile(fileStatus);
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
@@ -153,7 +176,7 @@ public class ViewFsCarbonFileTest {
}
};
-
+ viewFSCarbonFile = new ViewFSCarbonFile(fileStatus);
assertTrue(viewFSCarbonFile.listFiles().length == 1);
}
@@ -171,7 +194,7 @@ public class ViewFsCarbonFileTest {
new MockUp<Path>() {
@Mock
public FileSystem getFileSystem(Configuration conf) throws IOException {
- throw new IOException();
+ return new ViewFileSystem();
}
};
@@ -304,7 +327,7 @@ public class ViewFsCarbonFileTest {
}
};
-
+ viewFSCarbonFile = new ViewFSCarbonFile(fileStatus);
assertTrue(viewFSCarbonFile.renameForce(fileName));
}
diff --git a/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java b/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
index 7983a2d..a28986d 100644
--- a/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/reader/sortindex/CarbonDictionarySortIndexReaderImplTest.java
@@ -120,8 +120,7 @@ public class CarbonDictionarySortIndexReaderImplTest {
* this method will delete the store path
*/
private void deleteStorePath() {
- FileFactory.FileType fileType = FileFactory.getFileType(this.storePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(this.storePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(this.storePath);
deleteRecursiveSilent(carbonFile);
}
diff --git a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
index 32568bd..c03e32d 100644
--- a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
@@ -656,7 +656,7 @@ public class CarbonUtilTest {
public void testToReadHeaderWithFileNotFoundException() throws IOException {
new MockUp<FileFactory>() {
@SuppressWarnings("unused") @Mock
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType)
+ public DataInputStream getDataInputStream(String path)
throws FileNotFoundException {
throw new FileNotFoundException();
}
@@ -669,7 +669,7 @@ public class CarbonUtilTest {
public void testToReadHeaderWithIOException() throws IOException {
new MockUp<FileFactory>() {
@SuppressWarnings("unused") @Mock
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType)
+ public DataInputStream getDataInputStream(String path)
throws IOException {
throw new IOException();
}
diff --git a/core/src/test/java/org/apache/carbondata/core/util/DataFileFooterConverterTest.java b/core/src/test/java/org/apache/carbondata/core/util/DataFileFooterConverterTest.java
index 2705b63..a60491e 100644
--- a/core/src/test/java/org/apache/carbondata/core/util/DataFileFooterConverterTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/util/DataFileFooterConverterTest.java
@@ -143,7 +143,7 @@ public class DataFileFooterConverterTest {
final DataInputStream dataInputStream = new DataInputStream(byteArrayInputStream);
new MockUp<FileFactory>() {
@SuppressWarnings("unused") @Mock
- public DataInputStream getDataInputStream(String path, FileFactory.FileType fileType,
+ public DataInputStream getDataInputStream(String path,
int bufferSize, Configuration configuration) {
return dataInputStream;
}
diff --git a/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java b/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
index 9f2dc8e..b8f3984 100644
--- a/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/writer/CarbonDictionaryWriterImplTest.java
@@ -478,8 +478,7 @@ public class CarbonDictionaryWriterImplTest {
* this method will delete the store path
*/
private void deleteStorePath() {
- FileFactory.FileType fileType = FileFactory.getFileType(this.tablePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(this.tablePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(this.tablePath);
deleteRecursiveSilent(carbonFile);
}
@@ -528,9 +527,8 @@ public class CarbonDictionaryWriterImplTest {
*/
private void initDictionaryDirPaths() throws IOException {
String dictionaryLocation = CarbonTablePath.getMetadataPath(tablePath);
- FileFactory.FileType fileType = FileFactory.getFileType(dictionaryLocation);
- if(!FileFactory.isFileExist(dictionaryLocation, fileType)) {
- FileFactory.mkdirs(dictionaryLocation, fileType);
+ if(!FileFactory.isFileExist(dictionaryLocation)) {
+ FileFactory.mkdirs(dictionaryLocation);
}
this.dictionaryFilePath = CarbonTablePath.getDictionaryFilePath(tablePath, columnIdentifier.getColumnId());
this.dictionaryMetaFilePath = CarbonTablePath.getDictionaryMetaFilePath(tablePath, columnIdentifier.getColumnId());
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/AbstractBloomDataMapWriter.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/AbstractBloomDataMapWriter.java
index ffb781a..6f03ba6 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/AbstractBloomDataMapWriter.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/AbstractBloomDataMapWriter.java
@@ -155,7 +155,7 @@ public abstract class AbstractBloomDataMapWriter extends DataMapWriter {
private void initDataMapFile() throws IOException {
if (!FileFactory.isFileExist(dataMapPath)) {
- if (!FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))) {
+ if (!FileFactory.mkdirs(dataMapPath)) {
throw new IOException("Failed to create directory " + dataMapPath);
}
}
@@ -164,9 +164,8 @@ public abstract class AbstractBloomDataMapWriter extends DataMapWriter {
indexColumns.get(indexColId).getColName());
DataOutputStream dataOutStream = null;
try {
- FileFactory.createNewFile(dmFile, FileFactory.getFileType(dmFile));
- dataOutStream = FileFactory.getDataOutputStream(dmFile,
- FileFactory.getFileType(dmFile));
+ FileFactory.createNewFile(dmFile);
+ dataOutStream = FileFactory.getDataOutputStream(dmFile);
} catch (IOException e) {
CarbonUtil.closeStreams(dataOutStream);
throw new IOException(e);
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
index a0c7e30..e3902b3 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomCoarseGrainDataMapFactory.java
@@ -359,8 +359,7 @@ public class BloomCoarseGrainDataMapFactory extends DataMapFactory<CoarseGrainDa
String datamapPath = CarbonTablePath
.getDataMapStorePath(getCarbonTable().getTablePath(), segmentId, dataMapName);
if (FileFactory.isFileExist(datamapPath)) {
- CarbonFile file = FileFactory.getCarbonFile(datamapPath,
- FileFactory.getFileType(datamapPath));
+ CarbonFile file = FileFactory.getCarbonFile(datamapPath);
CarbonUtil.deleteFoldersAndFilesSilent(file);
}
clear(segmentId);
diff --git a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomIndexFileStore.java b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomIndexFileStore.java
index 3505d66..5d436d8 100644
--- a/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomIndexFileStore.java
+++ b/datamap/bloom/src/main/java/org/apache/carbondata/datamap/bloom/BloomIndexFileStore.java
@@ -64,8 +64,7 @@ public class BloomIndexFileStore {
// Step 1. check current folders
// get all shard paths of old store
- CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString,
- FileFactory.getFileType(dmSegmentPathString));
+ CarbonFile segmentPath = FileFactory.getCarbonFile(dmSegmentPathString);
CarbonFile[] shardPaths = segmentPath.listFiles(new CarbonFileFilter() {
@Override
public boolean accept(CarbonFile file) {
@@ -81,15 +80,14 @@ public class BloomIndexFileStore {
try {
// delete mergeShard folder if exists
if (FileFactory.isFileExist(mergeShardPath)) {
- FileFactory.deleteFile(mergeShardPath, FileFactory.getFileType(mergeShardPath));
+ FileFactory.deleteFile(mergeShardPath);
}
// create flag file before creating mergeShard folder
if (!FileFactory.isFileExist(mergeInprogressFile)) {
- FileFactory.createNewFile(
- mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));
+ FileFactory.createNewFile(mergeInprogressFile);
}
// create mergeShard output folder
- if (!FileFactory.mkdirs(mergeShardPath, FileFactory.getFileType(mergeShardPath))) {
+ if (!FileFactory.mkdirs(mergeShardPath)) {
throw new RuntimeException("Failed to create directory " + mergeShardPath);
}
} catch (IOException e) {
@@ -105,13 +103,12 @@ public class BloomIndexFileStore {
DataInputStream dataInputStream = null;
DataOutputStream dataOutputStream = null;
try {
- FileFactory.createNewFile(mergeIndexFile, FileFactory.getFileType(mergeIndexFile));
+ FileFactory.createNewFile(mergeIndexFile);
dataOutputStream = FileFactory.getDataOutputStream(
- mergeIndexFile, FileFactory.getFileType(mergeIndexFile));
+ mergeIndexFile);
for (CarbonFile shardPath : shardPaths) {
String bloomIndexFile = getBloomIndexFile(shardPath.getCanonicalPath(), indexCol);
- dataInputStream = FileFactory.getDataInputStream(
- bloomIndexFile, FileFactory.getFileType(bloomIndexFile));
+ dataInputStream = FileFactory.getDataInputStream(bloomIndexFile);
byte[] fileData = new byte[(int) FileFactory.getCarbonFile(bloomIndexFile).getSize()];
dataInputStream.readFully(fileData);
byte[] shardName = shardPath.getName().getBytes(Charset.forName("UTF-8"));
@@ -135,7 +132,7 @@ public class BloomIndexFileStore {
// Step 4. delete flag file and mergeShard can be used
try {
- FileFactory.deleteFile(mergeInprogressFile, FileFactory.getFileType(mergeInprogressFile));
+ FileFactory.deleteFile(mergeInprogressFile);
} catch (IOException e) {
LOGGER.error("Error occurs while deleting file " + mergeInprogressFile, e);
throw new RuntimeException("Error occurs while deleting file " + mergeInprogressFile);
@@ -167,7 +164,7 @@ public class BloomIndexFileStore {
List<CarbonBloomFilter> bloomFilters = new ArrayList<>();
try {
String indexFile = getBloomIndexFile(shardPath, colName);
- dataInStream = FileFactory.getDataInputStream(indexFile, FileFactory.getFileType(indexFile));
+ dataInStream = FileFactory.getDataInputStream(indexFile);
while (dataInStream.available() > 0) {
CarbonBloomFilter bloomFilter = new CarbonBloomFilter();
bloomFilter.readFields(dataInStream);
@@ -194,8 +191,7 @@ public class BloomIndexFileStore {
DataInputStream mergeIndexInStream = null;
List<CarbonBloomFilter> bloomFilters = new ArrayList<>();
try {
- mergeIndexInStream = FileFactory.getDataInputStream(
- mergeIndexFile, FileFactory.getFileType(mergeIndexFile));
+ mergeIndexInStream = FileFactory.getDataInputStream(mergeIndexFile);
while (mergeIndexInStream.available() > 0) {
// read shard name
int shardNameByteLength = mergeIndexInStream.readInt();
diff --git a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
index 457f856..b354bba 100644
--- a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
+++ b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxDataWriter.java
@@ -244,8 +244,8 @@ public class MinMaxDataWriter extends DataMapWriter {
BufferedWriter brWriter = null;
DataOutputStream dataOutStream = null;
try {
- FileFactory.createNewFile(filePath, FileFactory.getFileType(filePath));
- dataOutStream = FileFactory.getDataOutputStream(filePath, FileFactory.getFileType(filePath));
+ FileFactory.createNewFile(filePath);
+ dataOutStream = FileFactory.getDataOutputStream(filePath);
Gson gsonObjectToWrite = new Gson();
brWriter = new BufferedWriter(new OutputStreamWriter(dataOutStream, "UTF-8"));
String minmaxIndexData = gsonObjectToWrite.toJson(minMaxIndexBlockDetails);
diff --git a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMap.java b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMap.java
index b5a6fbb..02aa94b 100644
--- a/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMap.java
+++ b/datamap/examples/src/minmaxdatamap/main/java/org/apache/carbondata/datamap/examples/MinMaxIndexDataMap.java
@@ -99,7 +99,7 @@ public class MinMaxIndexDataMap extends CoarseGrainDataMap {
AtomicFileOperationFactory.getAtomicFileOperations(filePath);
try {
- if (!FileFactory.isFileExist(filePath, FileFactory.getFileType(filePath))) {
+ if (!FileFactory.isFileExist(filePath) {
return null;
}
dataInputStream = fileOperation.openForRead();
diff --git a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
index 091ea31..52dfa09 100644
--- a/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
+++ b/datamap/lucene/src/main/java/org/apache/carbondata/datamap/lucene/LuceneDataMapFactoryBase.java
@@ -269,8 +269,7 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> extends DataMapFactor
String datamapPath = CarbonTablePath
.getDataMapStorePath(tableIdentifier.getTablePath(), segmentId, dataMapName);
if (FileFactory.isFileExist(datamapPath)) {
- CarbonFile file = FileFactory.getCarbonFile(datamapPath,
- FileFactory.getFileType(datamapPath));
+ CarbonFile file = FileFactory.getCarbonFile(datamapPath);
CarbonUtil.deleteFoldersAndFilesSilent(file);
}
} catch (InterruptedException ex) {
@@ -316,8 +315,7 @@ abstract class LuceneDataMapFactoryBase<T extends DataMap> extends DataMapFactor
List<CarbonFile> indexFiles;
String dmPath = CarbonTablePath.getDataMapStorePath(tablePath, segmentId,
dataMap.getDataMapSchema().getDataMapName());
- FileFactory.FileType fileType = FileFactory.getFileType(dmPath);
- final CarbonFile dirPath = FileFactory.getCarbonFile(dmPath, fileType);
+ final CarbonFile dirPath = FileFactory.getCarbonFile(dmPath);
indexFiles = Arrays.asList(dirPath.listFiles(new CarbonFileFilter() {
@Override
public boolean accept(CarbonFile file) {
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/HiveExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/HiveExample.scala
index 241ee8c..e946fe2 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/HiveExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/HiveExample.scala
@@ -115,10 +115,10 @@ object HiveExample {
val dbLockPath = FileFactory.getUpdatedFilePath(s"$metaStoreLoc/db.lck")
val dbexLockPath = FileFactory.getUpdatedFilePath(s"$metaStoreLoc/dbex.lck")
if(FileFactory.isFileExist(dbLockPath)) {
- FileFactory.deleteFile(dbLockPath, FileFactory.getFileType(dbLockPath))
+ FileFactory.deleteFile(dbLockPath)
}
if(FileFactory.isFileExist(dbexLockPath)) {
- FileFactory.deleteFile(dbexLockPath, FileFactory.getFileType(dbexLockPath))
+ FileFactory.deleteFile(dbexLockPath)
}
}
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/util/AllDictionaryUtil.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/util/AllDictionaryUtil.scala
index d5308c3..5ce4c98d 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/util/AllDictionaryUtil.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/util/AllDictionaryUtil.scala
@@ -70,8 +70,7 @@ object AllDictionaryUtil {
def cleanDictionary(outputPath: String): Unit = {
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (file.exists()) {
file.delete()
}
@@ -84,12 +83,11 @@ object AllDictionaryUtil {
def saveToFile(contents: Array[String], outputPath: String): Unit = {
var writer: DataOutputStream = null
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (!file.exists()) {
file.createNewFile()
}
- writer = FileFactory.getDataOutputStream(outputPath, fileType)
+ writer = FileFactory.getDataOutputStream(outputPath)
for (content <- contents) {
writer.writeBytes(content + "\n")
}
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 2438890..0fecb75 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
@@ -78,7 +78,7 @@ public class CarbonFileInputFormat<T> extends CarbonInputFormat<T> implements Se
} else {
String schemaPath = CarbonTablePath
.getSchemaFilePath(getAbsoluteTableIdentifier(configuration).getTablePath());
- if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
+ if (!FileFactory.isFileExist(schemaPath)) {
TableInfo tableInfoInfer =
SchemaReader.inferSchema(getAbsoluteTableIdentifier(configuration), true);
localCarbonTable = CarbonTable.buildFromTableInfo(tableInfoInfer);
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/testutil/StoreCreator.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/testutil/StoreCreator.java
index 1340be1..064f6bb 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/testutil/StoreCreator.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/testutil/StoreCreator.java
@@ -345,9 +345,8 @@ public class StoreCreator {
thriftTableInfo.getFact_table().getSchema_evolution().getSchema_evolution_history()
.add(schemaEvolutionEntry);
- FileFactory.FileType fileType = FileFactory.getFileType(schemaMetadataPath);
- if (!FileFactory.isFileExist(schemaMetadataPath, fileType)) {
- FileFactory.mkdirs(schemaMetadataPath, fileType);
+ if (!FileFactory.isFileExist(schemaMetadataPath)) {
+ FileFactory.mkdirs(schemaMetadataPath);
}
ThriftWriter thriftWriter = new ThriftWriter(schemaFilePath, false);
diff --git a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeLocalDictTest.scala b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeLocalDictTest.scala
index 2735969..e9bde45 100644
--- a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeLocalDictTest.scala
+++ b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeLocalDictTest.scala
@@ -26,7 +26,6 @@ import org.scalatest.{BeforeAndAfterAll, FunSuiteLike}
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.datastore.impl.FileFactory.FileType
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
import org.apache.carbondata.presto.server.PrestoServer
@@ -288,14 +287,14 @@ class PrestoAllDataTypeLocalDictTest extends FunSuiteLike with BeforeAndAfterAll
}
private def cleanUp(): Unit = {
- FileFactory.deleteFile(s"$storePath/Fact", FileType.LOCAL)
+ FileFactory.deleteFile(s"$storePath/Fact")
FileFactory
.createDirectoryAndSetPermission(s"$storePath/_system",
new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
FileFactory
.createDirectoryAndSetPermission(s"$storePath/.DS_Store",
new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
- FileFactory.createNewFile(s"$storePath/testdb/.DS_STORE",FileType.LOCAL)
+ FileFactory.createNewFile(s"$storePath/testdb/.DS_STORE")
}
}
\ No newline at end of file
diff --git a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeTest.scala b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeTest.scala
index 205469c..966791e 100644
--- a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeTest.scala
+++ b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoAllDataTypeTest.scala
@@ -530,14 +530,14 @@ class PrestoAllDataTypeTest extends FunSuiteLike with BeforeAndAfterAll {
}
private def cleanUp(): Unit = {
- FileFactory.deleteFile(s"$storePath/Fact", FileType.LOCAL)
+ FileFactory.deleteFile(s"$storePath/Fact")
FileFactory
.createDirectoryAndSetPermission(s"$storePath/_system",
new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
FileFactory
.createDirectoryAndSetPermission(s"$storePath/.DS_Store",
new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL))
- FileFactory.createNewFile(s"$storePath/testdb/.DS_STORE",FileType.LOCAL)
+ FileFactory.createNewFile(s"$storePath/testdb/.DS_STORE")
}
test("test the OR operator on same column"){
diff --git a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
index 97691d6..af79aa3 100644
--- a/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
+++ b/integration/presto/src/test/scala/org/apache/carbondata/presto/integrationtest/PrestoTestNonTransactionalTableFiles.scala
@@ -212,9 +212,7 @@ class PrestoTestNonTransactionalTableFiles extends FunSuiteLike with BeforeAndAf
}
def deleteFile(path: String, extension: String): Unit = {
- val file: CarbonFile = FileFactory
- .getCarbonFile(path, FileFactory.getFileType(path))
-
+ val file: CarbonFile = FileFactory.getCarbonFile(path)
for (eachDir <- file.listFiles) {
if (!eachDir.isDirectory) {
if (eachDir.getName.endsWith(extension)) {
diff --git a/integration/presto/src/test/scala/org/apache/carbondata/presto/util/CarbonDataStoreCreator.scala b/integration/presto/src/test/scala/org/apache/carbondata/presto/util/CarbonDataStoreCreator.scala
index 639e178..8a48c89 100644
--- a/integration/presto/src/test/scala/org/apache/carbondata/presto/util/CarbonDataStoreCreator.scala
+++ b/integration/presto/src/test/scala/org/apache/carbondata/presto/util/CarbonDataStoreCreator.scala
@@ -318,10 +318,8 @@ object CarbonDataStoreCreator {
tableInfo.getLastUpdatedTime)
thriftTableInfo.getFact_table.getSchema_evolution.getSchema_evolution_history
.add(schemaEvolutionEntry)
- val fileType: FileFactory.FileType =
- FileFactory.getFileType(schemaMetadataPath)
- if (!FileFactory.isFileExist(schemaMetadataPath, fileType)) {
- FileFactory.mkdirs(schemaMetadataPath, fileType)
+ if (!FileFactory.isFileExist(schemaMetadataPath)) {
+ FileFactory.mkdirs(schemaMetadataPath)
}
val thriftWriter: ThriftWriter = new ThriftWriter(schemaFilePath, false)
thriftWriter.open()
diff --git a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/datasource/CreateTableUsingSparkCarbonFileFormatTestCase.scala b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/datasource/CreateTableUsingSparkCarbonFileFormatTestCase.scala
index ecf9ff4..c1ed9c0 100644
--- a/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/datasource/CreateTableUsingSparkCarbonFileFormatTestCase.scala
+++ b/integration/spark-common-cluster-test/src/test/scala/org/apache/carbondata/cluster/sdv/generated/datasource/CreateTableUsingSparkCarbonFileFormatTestCase.scala
@@ -84,7 +84,7 @@ class CreateTableUsingSparkCarbonFileFormatTestCase extends FunSuite with Before
def deleteIndexFile(path: String, extension: String): Unit = {
val file: CarbonFile = FileFactory
- .getCarbonFile(path, FileFactory.getFileType(path))
+ .getCarbonFile(path)
for (eachDir <- file.listFiles) {
if (!eachDir.isDirectory) {
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala
index 51e7f68..c15917d 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala
@@ -50,9 +50,6 @@ class TestLoadDataGeneral extends QueryTest with BeforeAndAfterEach {
val carbonTable = CarbonMetadata.getInstance().getCarbonTable(databaseName, tableName)
val partitionPath =
CarbonTablePath.getPartitionDir(carbonTable.getAbsoluteTableIdentifier.getTablePath)
- val fileType: FileFactory.FileType = FileFactory.getFileType(partitionPath)
- val carbonFile = FileFactory.getCarbonFile(partitionPath, fileType)
- val segments: ArrayBuffer[String] = ArrayBuffer()
val segment = Segment.getSegment(segmentId, carbonTable.getAbsoluteTableIdentifier.getTablePath)
segment != null
}
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/BlockPruneQueryTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/BlockPruneQueryTestCase.scala
index d56b465..72b50f5 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/BlockPruneQueryTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/BlockPruneQueryTestCase.scala
@@ -36,12 +36,11 @@ class BlockPruneQueryTestCase extends QueryTest with BeforeAndAfterAll {
testData(2) = "c"
var writer: DataOutputStream = null
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (!file.exists()) {
file.createNewFile()
}
- writer = FileFactory.getDataOutputStream(outputPath, fileType)
+ writer = FileFactory.getDataOutputStream(outputPath)
for (i <- 0 to 2) {
for (j <- 0 to 240000) {
writer.writeBytes(testData(i) + "," + j + "\n")
@@ -95,8 +94,7 @@ class BlockPruneQueryTestCase extends QueryTest with BeforeAndAfterAll {
override def afterAll {
// delete the temp data file
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (file.exists()) {
file.delete()
}
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/CarbonCustomBlockDistributionTest.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/CarbonCustomBlockDistributionTest.scala
index 91e4219..ba33a5f 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/CarbonCustomBlockDistributionTest.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/blockprune/CarbonCustomBlockDistributionTest.scala
@@ -40,12 +40,11 @@ class CarbonCustomBlockDistributionTest extends QueryTest with BeforeAndAfterAll
testData(2) = "c"
var writer: DataOutputStream = null
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (!file.exists()) {
file.createNewFile()
}
- writer = FileFactory.getDataOutputStream(outputPath, fileType)
+ writer = FileFactory.getDataOutputStream(outputPath)
for (i <- 0 to 2) {
for (j <- 0 to 240000) {
writer.writeBytes(testData(i) + "," + j + "\n")
@@ -100,8 +99,7 @@ class CarbonCustomBlockDistributionTest extends QueryTest with BeforeAndAfterAll
// delete the temp data file
CarbonProperties.getInstance().addProperty("carbon.custom.distribution","false")
try {
- val fileType = FileFactory.getFileType(outputPath)
- val file = FileFactory.getCarbonFile(outputPath, fileType)
+ val file = FileFactory.getCarbonFile(outputPath)
if (file.exists()) {
file.delete()
}
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCarbonFileInputFormatWithExternalCarbonTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCarbonFileInputFormatWithExternalCarbonTable.scala
index 553fa25..95676d0 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCarbonFileInputFormatWithExternalCarbonTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestCarbonFileInputFormatWithExternalCarbonTable.scala
@@ -75,9 +75,7 @@ class TestCarbonFileInputFormatWithExternalCarbonTable extends QueryTest with Be
}
def deleteIndexFile(path: String, extension: String) : Unit = {
- val file: CarbonFile = FileFactory
- .getCarbonFile(path, FileFactory.getFileType(path))
-
+ val file: CarbonFile = FileFactory.getCarbonFile(path)
for (eachDir <- file.listFiles) {
if (!eachDir.isDirectory) {
if (eachDir.getName.endsWith(extension)) {
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
index 274cca9..557e748 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/createTable/TestNonTransactionalCarbonTable.scala
@@ -238,8 +238,7 @@ class TestNonTransactionalCarbonTable extends QueryTest with BeforeAndAfterAll {
}
def deleteFile(path: String, extension: String): Unit = {
- val file: CarbonFile = FileFactory
- .getCarbonFile(path, FileFactory.getFileType(path))
+ val file: CarbonFile = FileFactory.getCarbonFile(path)
for (eachDir <- file.listFiles) {
if (!eachDir.isDirectory) {
@@ -747,11 +746,11 @@ class TestNonTransactionalCarbonTable extends QueryTest with BeforeAndAfterAll {
val path1 = writerPath + "/0testdir"
val path2 = writerPath + "/testdir"
- FileFactory.getCarbonFile(path1, FileFactory.getFileType(path1))
- FileFactory.mkdirs(path1, FileFactory.getFileType(path1))
+ FileFactory.getCarbonFile(path1)
+ FileFactory.mkdirs(path1)
- FileFactory.getCarbonFile(path2, FileFactory.getFileType(path2))
- FileFactory.mkdirs(path2, FileFactory.getFileType(path2))
+ FileFactory.getCarbonFile(path2)
+ FileFactory.mkdirs(path2)
sql("DROP TABLE IF EXISTS sdkOutputTable")
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
index 97aaa23..66312d0 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/CGDataMapTestCase.scala
@@ -329,10 +329,10 @@ class CGDataMapWriter(
* class.
*/
override def finish(): Unit = {
- FileFactory.mkdirs(dataMapPath, FileFactory.getFileType(dataMapPath))
+ FileFactory.mkdirs(dataMapPath)
val file = dataMapPath + "/testcg.datamap"
val stream: DataOutputStream = FileFactory
- .getDataOutputStream(file, FileFactory.getFileType(file))
+ .getDataOutputStream(file)
val out = new ByteOutputStream()
val outStream = new ObjectOutputStream(out)
outStream.writeObject(maxMin)
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
index a49b233..1748840 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/FGDataMapTestCase.scala
@@ -296,9 +296,9 @@ class FGDataMapWriter(carbonTable: CarbonTable,
this.taskName = shardName
if (stream == null) {
val path = fgwritepath.substring(0, fgwritepath.lastIndexOf("/"))
- FileFactory.mkdirs(path, FileFactory.getFileType(path))
+ FileFactory.mkdirs(path)
stream = FileFactory
- .getDataOutputStream(fgwritepath, FileFactory.getFileType(fgwritepath))
+ .getDataOutputStream(fgwritepath)
}
}
@@ -415,7 +415,7 @@ class FGDataMapWriter(carbonTable: CarbonTable,
* class.
*/
override def finish(): Unit = {
- FileFactory.mkdirs(fgwritepath, FileFactory.getFileType(fgwritepath))
+ FileFactory.mkdirs(fgwritepath)
val out = new ByteOutputStream()
val outStream = new ObjectOutputStream(out)
outStream.writeObject(maxMin)
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SubqueryWithFilterAndSortTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SubqueryWithFilterAndSortTestCase.scala
index bb5182f..ab27de5 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SubqueryWithFilterAndSortTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/SubqueryWithFilterAndSortTestCase.scala
@@ -27,7 +27,7 @@ class SubqueryWithFilterAndSortTestCase extends QueryTest with BeforeAndAfterAll
val tempFilePath = s"$resourcesPath/temp/subqueryfilterwithsort.csv"
override def beforeAll {
- FileFactory.mkdirs(tempDirPath,FileType.LOCAL)
+ FileFactory.mkdirs(tempDirPath)
sql("drop table if exists subqueryfilterwithsort")
sql("drop table if exists subqueryfilterwithsort_hive")
sql("CREATE TABLE subqueryfilterwithsort (name String, id int) STORED BY 'org.apache.carbondata.format'")
@@ -59,12 +59,12 @@ class SubqueryWithFilterAndSortTestCase extends QueryTest with BeforeAndAfterAll
}
def writedata(filePath: String, data: String) = {
- val dis = FileFactory.getDataOutputStream(filePath, FileFactory.getFileType(filePath))
+ val dis = FileFactory.getDataOutputStream(filePath)
dis.writeBytes(data.toString())
dis.close()
}
def deleteFile(filePath: String) {
- val file = FileFactory.getCarbonFile(filePath, FileFactory.getFileType(filePath))
+ val file = FileFactory.getCarbonFile(filePath)
file.delete()
}
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ValueCompressionDataTypeTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ValueCompressionDataTypeTestCase.scala
index 2136efd..e496667 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ValueCompressionDataTypeTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/detailquery/ValueCompressionDataTypeTestCase.scala
@@ -25,7 +25,7 @@ class ValueCompressionDataTypeTestCase extends QueryTest with BeforeAndAfterAll
val tempDirPath = s"$resourcesPath/tempdir"
override def beforeAll {
- FileFactory.mkdirs(tempDirPath,FileType.LOCAL)
+ FileFactory.mkdirs(tempDirPath)
}
test("ActualDataType:double,ChangedDatatype:Short,CompressionType:NonDecimalMaxMin") {
@@ -121,12 +121,13 @@ class ValueCompressionDataTypeTestCase extends QueryTest with BeforeAndAfterAll
}
def writeData(filePath: String, data: String) = {
- val dis = FileFactory.getDataOutputStream(filePath, FileFactory.getFileType(filePath))
+ val dis = FileFactory.getDataOutputStream(filePath)
dis.writeBytes(data.toString())
dis.close()
}
+
def deleteFile(filePath: String) {
- val file = FileFactory.getCarbonFile(filePath, FileFactory.getFileType(filePath))
+ val file = FileFactory.getCarbonFile(filePath)
file.delete()
}
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
index 2852501..a14dc77 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/HorizontalCompactionTestCase.scala
@@ -413,7 +413,7 @@ class HorizontalCompactionTestCase extends QueryTest with BeforeAndAfterAll {
val identifier = carbonTable.getAbsoluteTableIdentifier()
val dataFilesDir = CarbonTablePath.getSegmentPath(identifier.getTablePath, "0")
val carbonFile =
- FileFactory.getCarbonFile(dataFilesDir, FileFactory.getFileType(dataFilesDir))
+ FileFactory.getCarbonFile(dataFilesDir)
var updateDeltaFiles = getDeltaFiles(carbonFile, CarbonCommonConstants.UPDATE_INDEX_FILE_EXT)
var deletaDeltaFiles = getDeltaFiles(carbonFile, CarbonCommonConstants.DELETE_DELTA_FILE_EXT)
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
index f4c725e..723d969 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
@@ -70,7 +70,7 @@ class TestDataLoadingForPartitionTable extends QueryTest with BeforeAndAfterAll
val segmentDir = carbonTable.getSegmentPath(segmentId)
val dataFiles = if (FileFactory.isFileExist(segmentDir)) {
- val carbonFile = FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir))
+ val carbonFile = FileFactory.getCarbonFile(segmentDir)
carbonFile.listFiles(new CarbonFileFilter() {
override def accept(file: CarbonFile): Boolean = {
return file.getName.endsWith(".carbondata")
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
index cac0b11..3705e12 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/api/CarbonStore.scala
@@ -114,7 +114,7 @@ object CarbonStore {
val indexFile = FileFactory.getCarbonFile(indexPath)
if (indexFile.exists()) {
val indices =
- StreamSegment.readIndexFile(indexPath, FileFactory.getFileType(indexPath))
+ StreamSegment.readIndexFile(indexPath)
(indices.asScala.map(_.getFile_size).sum, indexFile.getSize)
} else {
(-1L, -1L)
@@ -193,8 +193,7 @@ object CarbonStore {
// in case of force clean the lock is not required
if (forceTableClean) {
FileFactory.deleteAllCarbonFilesOfDir(
- FileFactory.getCarbonFile(absoluteTableIdentifier.getTablePath,
- FileFactory.getFileType(absoluteTableIdentifier.getTablePath)))
+ FileFactory.getCarbonFile(absoluteTableIdentifier.getTablePath))
} else {
carbonCleanFilesLock =
CarbonLockUtil
@@ -236,8 +235,7 @@ object CarbonStore {
val loadMetadataDetails = SegmentStatusManager
.readLoadMetadata(carbonTable.getMetadataPath)
- val fileType = FileFactory.getFileType(carbonTable.getTablePath)
- val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath, fileType)
+ val carbonFile = FileFactory.getCarbonFile(carbonTable.getTablePath)
// list all files from table path
val listOfDefaultPartFilesIterator = carbonFile.listFiles(true)
@@ -252,9 +250,8 @@ object CarbonStore {
val partitionLocation = partitionSpec.getLocation
// For partition folder outside the tablePath
if (!partitionLocation.toString.startsWith(carbonTable.getTablePath)) {
- val fileType = FileFactory.getFileType(partitionLocation.toString)
val partitionCarbonFile = FileFactory
- .getCarbonFile(partitionLocation.toString, fileType)
+ .getCarbonFile(partitionLocation.toString)
// list all files from partitionLoacation
val listOfExternalPartFilesIterator = partitionCarbonFile.listFiles(true)
// delete all files of @loadStartTime from externalPath
@@ -279,7 +276,7 @@ object CarbonStore {
val fileName = carbonFile.getName
if (CarbonTablePath.DataFileUtil.compareCarbonFileTimeStamp(fileName, timestamp)) {
// delete the file
- FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath))
+ FileFactory.deleteFile(filePath)
}
}
}
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala
index dbf47ab..25a1903 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/DictionaryDetailHelper.scala
@@ -32,9 +32,8 @@ class DictionaryDetailHelper extends DictionaryDetailService {
val dictFileExists = new Array[Boolean](primDimensions.length)
val columnIdentifier = new Array[ColumnIdentifier](primDimensions.length)
- val fileType = FileFactory.getFileType(dictfolderPath)
// Metadata folder
- val metadataDirectory = FileFactory.getCarbonFile(dictfolderPath, fileType)
+ val metadataDirectory = FileFactory.getCarbonFile(dictfolderPath)
// need list all dictionary file paths with exists flag
val carbonFiles = metadataDirectory.listFiles(new CarbonFileFilter {
@Override def accept(pathname: CarbonFile): Boolean = {
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
index d5cce27..bb2dc91 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableAddColumnRDD.scala
@@ -80,9 +80,8 @@ class AlterTableAddColumnRDD[K, V](
CarbonProperties.getInstance.addProperty(CarbonCommonConstants.LOCK_TYPE, lockType)
// Create table and metadata folders if not exist
val metadataDirectoryPath = CarbonTablePath.getMetadataPath(identifier.getTablePath)
- val fileType = FileFactory.getFileType(metadataDirectoryPath)
- if (!FileFactory.isFileExist(metadataDirectoryPath, fileType)) {
- FileFactory.mkdirs(metadataDirectoryPath, fileType)
+ if (!FileFactory.isFileExist(metadataDirectoryPath)) {
+ FileFactory.mkdirs(metadataDirectoryPath)
}
GlobalDictionaryUtil.loadDefaultDictionaryValueForNewColumn(
columnSchema,
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
index adfdaad..8168386 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala
@@ -392,8 +392,7 @@ class CarbonGlobalDictionaryGenerateRDD(
} is locked for updation. Please try after some time")
}
val t2 = System.currentTimeMillis
- val fileType = FileFactory.getFileType(model.dictFilePaths(split.index))
- val isDictFileExists = FileFactory.isFileExist(model.dictFilePaths(split.index), fileType)
+ val isDictFileExists = FileFactory.isFileExist(model.dictFilePaths(split.index))
dictionaryForDistinctValueLookUp = if (isDictFileExists) {
CarbonLoaderUtil.getDictionary(model.table,
model.columnIdentifier(split.index),
@@ -532,8 +531,7 @@ class CarbonColumnDictGenerateRDD(
var inputStream: DataInputStream = null
var colDictData: java.util.Iterator[Array[String]] = null
try {
- inputStream = FileFactory.getDataInputStream(preDefDictFilePath,
- FileFactory.getFileType(preDefDictFilePath))
+ inputStream = FileFactory.getDataInputStream(preDefDictFilePath)
csvReader = new CSVReader(new InputStreamReader(inputStream, Charset.defaultCharset),
carbonLoadModel.getCsvDelimiter.charAt(0))
// read the column data to list iterator
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
index 794e7d2..72b7c3e 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
@@ -379,9 +379,8 @@ object StreamHandoffRDD {
val metaDataFilepath = loadModel.getCarbonDataLoadSchema.getCarbonTable.getMetadataPath
val identifier = loadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier
val metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath)
- val fileType = FileFactory.getFileType(metadataPath)
- if (!FileFactory.isFileExist(metadataPath, fileType)) {
- FileFactory.mkdirs(metadataPath, fileType)
+ if (!FileFactory.isFileExist(metadataPath)) {
+ FileFactory.mkdirs(metadataPath)
}
val tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath)
val segmentStatusManager = new SegmentStatusManager(identifier)
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index f05689d..b16c434 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -707,9 +707,8 @@ object CommonUtil {
return
}
try {
- val fileType = FileFactory.getFileType(databaseLocation)
- if (FileFactory.isFileExist(databaseLocation, fileType)) {
- val file = FileFactory.getCarbonFile(databaseLocation, fileType)
+ if (FileFactory.isFileExist(databaseLocation)) {
+ val file = FileFactory.getCarbonFile(databaseLocation)
if (file.isDirectory) {
val tableFolders = file.listFiles()
tableFolders.foreach { tableFolder =>
@@ -719,7 +718,7 @@ object CommonUtil {
val tableUniqueName = dbName + "_" + tableFolder.getName
val tableStatusFile =
CarbonTablePath.getTableStatusFilePath(tablePath)
- if (FileFactory.isFileExist(tableStatusFile, fileType)) {
+ if (FileFactory.isFileExist(tableStatusFile)) {
try {
val carbonTable = CarbonMetadata.getInstance
.getCarbonTable(tableUniqueName)
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index fb337cd..6a338a5 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -575,10 +575,9 @@ object GlobalDictionaryUtil {
* @return (isNonempty, isDirectory)
*/
private def validateAllDictionaryPath(allDictionaryPath: String): Boolean = {
- val fileType = FileFactory.getFileType(allDictionaryPath)
val filePath = new Path(allDictionaryPath)
- val file = FileFactory.getCarbonFile(filePath.toString, fileType)
- val parentFile = FileFactory.getCarbonFile(filePath.getParent.toString, fileType)
+ val file = FileFactory.getCarbonFile(filePath.toString)
+ val parentFile = FileFactory.getCarbonFile(filePath.getParent.toString)
// filepath regex, look like "/path/*.dictionary"
if (filePath.getName.startsWith("*")) {
val dictExt = filePath.getName.substring(1)
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
index 2596e8a..0f25703 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/streaming/StreamSinkFactory.scala
@@ -166,18 +166,17 @@ object StreamSinkFactory {
private def getStreamSegmentId(carbonTable: CarbonTable): String = {
val segmentId = StreamSegment.open(carbonTable)
val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)
- val fileType = FileFactory.getFileType(segmentDir)
val metadataPath = CarbonTablePath.getMetadataPath(carbonTable.getTablePath)
- if (!FileFactory.isFileExist(metadataPath, fileType)) {
+ if (!FileFactory.isFileExist(metadataPath)) {
// Create table directory path, in case of enabling hive metastore first load may not have
// table folder created.
- FileFactory.mkdirs(metadataPath, fileType)
+ FileFactory.mkdirs(metadataPath)
}
- if (FileFactory.isFileExist(segmentDir, fileType)) {
+ if (FileFactory.isFileExist(segmentDir)) {
// recover fault
StreamSegment.recoverSegmentIfRequired(segmentDir)
} else {
- FileFactory.mkdirs(segmentDir, fileType)
+ FileFactory.mkdirs(segmentDir)
}
segmentId
}
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
index ca187e8..6384bf0 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/streaming/CarbonAppendableStreamSink.scala
@@ -179,13 +179,12 @@ class CarbonAppendableStreamSink(
val segmentId = StreamSegment.open(carbonTable)
if (segmentId.equals(currentSegmentId)) {
val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, currentSegmentId)
- val fileType = FileFactory.getFileType(segmentDir)
if (segmentMaxSize <= StreamSegment.size(segmentDir)) {
val newSegmentId = StreamSegment.close(carbonTable, currentSegmentId)
currentSegmentId = newSegmentId
val newSegmentDir =
CarbonTablePath.getSegmentPath(carbonTable.getTablePath, currentSegmentId)
- FileFactory.mkdirs(newSegmentDir, fileType)
+ FileFactory.mkdirs(newSegmentDir)
// trigger hand off operation
if (enableAutoHandoff) {
@@ -200,8 +199,7 @@ class CarbonAppendableStreamSink(
currentSegmentId = segmentId
val newSegmentDir =
CarbonTablePath.getSegmentPath(carbonTable.getTablePath, currentSegmentId)
- val fileType = FileFactory.getFileType(newSegmentDir)
- FileFactory.mkdirs(newSegmentDir, fileType)
+ FileFactory.mkdirs(newSegmentDir)
}
}
}
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/ResourceRegisterAndCopier.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/ResourceRegisterAndCopier.scala
index ff9b8c4..fee1bbe 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/ResourceRegisterAndCopier.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/ResourceRegisterAndCopier.scala
@@ -43,8 +43,7 @@ object ResourceRegisterAndCopier {
def copyResourcesifNotExists(hdfsPath: String,
resourcePath: String,
dataFilesPath: String): Unit = {
- val fileType = FileFactory.getFileType(hdfsPath)
- val file = FileFactory.getCarbonFile(hdfsPath, fileType)
+ val file = FileFactory.getCarbonFile(hdfsPath)
if (!file.exists()) {
sys.error(s"""Provided path $hdfsPath does not exist""")
}
@@ -57,7 +56,7 @@ object ResourceRegisterAndCopier {
val resources = readDataFiles(dataFilesPath)
resources.foreach { file =>
val hdfsDataPath = hdfsPath + "/" + file
- val rsFile = FileFactory.getCarbonFile(hdfsDataPath, fileType)
+ val rsFile = FileFactory.getCarbonFile(hdfsDataPath)
val target = resourcePath + "/" + file
if (!rsFile.exists()) {
if (file.lastIndexOf("/") > -1) {
@@ -125,11 +124,9 @@ object ResourceRegisterAndCopier {
def copyLocalFile(dst: String,
src: String): Unit = {
LOGGER.info(s"Copying file : $src to $dst")
- if (FileFactory.isFileExist(src, FileFactory.getFileType(src))) {
- val dataOutputStream = FileFactory.getDataOutputStream(dst,
- FileFactory.getFileType(dst))
- val dataInputStream = FileFactory.getDataInputStream(src,
- FileFactory.getFileType(src))
+ if (FileFactory.isFileExist(src)) {
+ val dataOutputStream = FileFactory.getDataOutputStream(dst)
+ val dataInputStream = FileFactory.getDataInputStream(src)
IOUtils.copyBytes(dataInputStream, dataOutputStream, 8 * 1024)
CarbonUtil.closeStream(dataInputStream)
CarbonUtil.closeStream(dataOutputStream)
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
index 0af832b..f70d7cb 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/TestQueryExecutor.scala
@@ -105,8 +105,7 @@ object TestQueryExecutor {
val storeLocation = if (hdfsUrl.startsWith("hdfs://")) {
CarbonProperties.getInstance().addProperty(CarbonCommonConstants.LOCK_TYPE,
CarbonCommonConstants.CARBON_LOCK_TYPE_HDFS)
- val carbonFile = FileFactory.
- getCarbonFile(s"$hdfsUrl/store", FileFactory.getFileType(s"$hdfsUrl/store"))
+ val carbonFile = FileFactory.getCarbonFile(s"$hdfsUrl/store")
FileFactory.deleteAllCarbonFilesOfDir(carbonFile)
s"$hdfsUrl/store_" + System.nanoTime()
} else {
@@ -115,8 +114,7 @@ object TestQueryExecutor {
s"$target/store"
}
val warehouse = if (hdfsUrl.startsWith("hdfs://")) {
- val carbonFile = FileFactory.
- getCarbonFile(s"$hdfsUrl/warehouse", FileFactory.getFileType(s"$hdfsUrl/warehouse"))
+ val carbonFile = FileFactory.getCarbonFile(s"$hdfsUrl/warehouse")
FileFactory.deleteAllCarbonFilesOfDir(carbonFile)
s"$hdfsUrl/warehouse_" + System.nanoTime()
} else {
@@ -137,7 +135,7 @@ object TestQueryExecutor {
val hiveresultpath = if (hdfsUrl.startsWith("hdfs://")) {
val p = s"$hdfsUrl/hiveresultpath"
- FileFactory.mkdirs(p, FileFactory.getFileType(p))
+ FileFactory.mkdirs(p)
p
} else {
val p = s"$target/hiveresultpath"
@@ -189,6 +187,6 @@ object TestQueryExecutor {
}
private def createDirectory(badStoreLocation: String) = {
- FileFactory.mkdirs(badStoreLocation, FileFactory.getFileType(badStoreLocation))
+ FileFactory.mkdirs(badStoreLocation)
}
}
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
index 3c7fbbb..de825c0 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/FileUtils.scala
@@ -107,8 +107,7 @@ object FileUtils {
def createDatabaseDirectory(dbName: String, storePath: String, sparkContext: SparkContext) {
val databasePath: String = storePath + File.separator + dbName.toLowerCase
- val fileType = FileFactory.getFileType(databasePath)
- FileFactory.mkdirs(databasePath, fileType)
+ FileFactory.mkdirs(databasePath)
val operationContext = new OperationContext
val createDatabasePostExecutionEvent = new CreateDatabasePostExecutionEvent(dbName,
databasePath, sparkContext)
diff --git a/integration/spark-datasource/src/test/scala/org/apache/spark/sql/carbondata/datasource/TestCreateTableUsingSparkCarbonFileFormat.scala b/integration/spark-datasource/src/test/scala/org/apache/spark/sql/carbondata/datasource/TestCreateTableUsingSparkCarbonFileFormat.scala
index 7814878..5958d39 100644
--- a/integration/spark-datasource/src/test/scala/org/apache/spark/sql/carbondata/datasource/TestCreateTableUsingSparkCarbonFileFormat.scala
+++ b/integration/spark-datasource/src/test/scala/org/apache/spark/sql/carbondata/datasource/TestCreateTableUsingSparkCarbonFileFormat.scala
@@ -96,8 +96,7 @@ class TestCreateTableUsingSparkCarbonFileFormat extends FunSuite with BeforeAndA
}
def deleteIndexFile(path: String, extension: String) : Unit = {
- val file: CarbonFile = FileFactory
- .getCarbonFile(path, FileFactory.getFileType(path))
+ val file: CarbonFile = FileFactory.getCarbonFile(path)
for (eachDir <- file.listFiles) {
if (!eachDir.isDirectory) {
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
index 5ee5982..e339b34 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/datamap/IndexDataMapRebuildRDD.scala
@@ -96,7 +96,7 @@ object IndexDataMapRebuildRDD {
}.filter(p => !FileFactory.isFileExist(p._2)).toMap
segments2DmStorePath.foreach { case (_, dmPath) =>
- if (!FileFactory.mkdirs(dmPath, FileFactory.getFileType(dmPath))) {
+ if (!FileFactory.mkdirs(dmPath)) {
throw new IOException(
s"Failed to create directory $dmPath for rebuilding datamap ${ schema.getDataMapName }")
}
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DataMapJobs.scala b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DataMapJobs.scala
index 49ef8bb..3474795 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DataMapJobs.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/indexserver/DataMapJobs.scala
@@ -26,7 +26,6 @@ import org.apache.spark.util.SizeEstimator
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datamap.{AbstractDataMapJob, DistributableDataMapFormat}
-import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.indexstore.ExtendedBlocklet
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.scan.expression.BinaryExpression
@@ -71,8 +70,7 @@ class DistributedDataMapJob extends AbstractDataMapJob {
val tmpPath = CarbonUtil
.getIndexServerTempPath(dataMapFormat.getCarbonTable.getTablePath,
dataMapFormat.getQueryId)
- if (null != splitFolderPath &&
- !splitFolderPath.deleteFile(tmpPath, FileFactory.getFileType(tmpPath))) {
+ if (null != splitFolderPath && !splitFolderPath.deleteFile()) {
LOGGER.error("Problem while deleting the temp directory:" + tmpPath)
}
}
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 6bb7e76..57973da 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -578,7 +578,7 @@ object CarbonDataRDDFactory {
// delete corresponding segment file from metadata
val segmentFile = CarbonTablePath.getSegmentFilesLocation(carbonLoadModel.getTablePath) +
File.separator + segmentFileName
- FileFactory.deleteFile(segmentFile, FileFactory.getFileType(segmentFile))
+ FileFactory.deleteFile(segmentFile)
clearDataMapFiles(carbonTable, carbonLoadModel.getSegmentId)
}
LOGGER.info("********clean up done**********")
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAddLoadCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAddLoadCommand.scala
index 176c969..7a40566 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAddLoadCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAddLoadCommand.scala
@@ -326,7 +326,7 @@ case class CarbonAddLoadCommand(
// delete corresponding segment file from metadata
val segmentFile = CarbonTablePath.getSegmentFilesLocation(carbonTable.getTablePath) +
File.separator + segment.getSegmentFileName
- FileFactory.deleteFile(segmentFile, FileFactory.getFileType(segmentFile))
+ FileFactory.deleteFile(segmentFile)
clearDataMapFiles(carbonTable, model.getSegmentId)
LOGGER.info("********clean up done**********")
LOGGER.error("Data load failed due to failure in table status updation.")
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index 2227094..bb691f7 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -334,9 +334,8 @@ case class CarbonLoadDataCommand(
// Create table and metadata folders if not exist
if (carbonLoadModel.isCarbonTransactionalTable) {
val metadataDirectoryPath = CarbonTablePath.getMetadataPath(table.getTablePath)
- val fileType = FileFactory.getFileType(metadataDirectoryPath)
- if (!FileFactory.isFileExist(metadataDirectoryPath, fileType)) {
- FileFactory.mkdirs(metadataDirectoryPath, fileType)
+ if (!FileFactory.isFileExist(metadataDirectoryPath)) {
+ FileFactory.mkdirs(metadataDirectoryPath)
}
} else {
carbonLoadModel.setSegmentId(System.currentTimeMillis().toString)
@@ -401,9 +400,8 @@ case class CarbonLoadDataCommand(
val partitionLocation = CarbonProperties.getStorePath + "/partition/" +
table.getDatabaseName + "/" +
table.getTableName + "/"
- val fileType = FileFactory.getFileType(partitionLocation)
- if (FileFactory.isFileExist(partitionLocation, fileType)) {
- val file = FileFactory.getCarbonFile(partitionLocation, fileType)
+ if (FileFactory.isFileExist(partitionLocation)) {
+ val file = FileFactory.getCarbonFile(partitionLocation)
CarbonUtil.deleteFoldersAndFiles(file)
}
}
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/RefreshCarbonTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/RefreshCarbonTableCommand.scala
index cebf606..6dca2d8 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/RefreshCarbonTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/RefreshCarbonTableCommand.scala
@@ -73,7 +73,7 @@ case class RefreshCarbonTableCommand(
val schemaFilePath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath)
// if schema file does not exist then the table will either non carbon table or stale
// carbon table
- if (FileFactory.isFileExist(schemaFilePath, FileFactory.getFileType(schemaFilePath))) {
+ if (FileFactory.isFileExist(schemaFilePath)) {
// read TableInfo
val tableInfo = SchemaReader.getTableInfo(identifier)
// refresh the column schema in case of store before V3
@@ -196,7 +196,7 @@ case class RefreshCarbonTableCommand(
val tablePath = CarbonEnv.getTablePath(Some(dbName), tableName)(sparkSession)
val schemaFilePath = CarbonTablePath.getSchemaFilePath(tablePath)
try {
- fileExist = FileFactory.isFileExist(schemaFilePath, FileFactory.getFileType(schemaFilePath))
+ fileExist = FileFactory.isFileExist(schemaFilePath)
} catch {
case e: Exception =>
fileExist = false
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
index 625f594..ecb74dc 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mutation/DeleteExecution.scala
@@ -269,8 +269,7 @@ object DeleteExecution {
CarbonCommonConstants.FACT_FILE_EXT)
val deleteDeletaPath = CarbonUpdateUtil
.getDeleteDeltaFilePath(blockPath, blockName, timestamp)
- val carbonDeleteWriter = new CarbonDeleteDeltaWriterImpl(deleteDeletaPath,
- FileFactory.getFileType(deleteDeletaPath))
+ val carbonDeleteWriter = new CarbonDeleteDeltaWriterImpl(deleteDeletaPath)
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
index 6b80bbe..751f52a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/table/CarbonDropTableCommand.scala
@@ -227,12 +227,10 @@ case class CarbonDropTableCommand(
ManageDictionaryAndBTree.clearBTreeAndDictionaryLRUCache(carbonTable)
// delete the table folder
val tablePath = carbonTable.getTablePath
- val fileType = FileFactory.getFileType(tablePath)
-
// delete table data only if it is not external table
- if (FileFactory.isFileExist(tablePath, fileType) &&
+ if (FileFactory.isFileExist(tablePath) &&
!(carbonTable.isExternalTable || carbonTable.isFileLevelFormat)) {
- val file = FileFactory.getCarbonFile(tablePath, fileType)
+ val file = FileFactory.getCarbonFile(tablePath)
CarbonUtil.deleteFoldersAndFilesSilent(file)
}
// Delete lock directory if external lock path is specified.
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
index 900b69c..6c9c6cd 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
@@ -288,8 +288,7 @@ class CarbonFileMetastore extends CarbonMetaStore {
Some(wrapperTableInfo)
} else {
val tableMetadataFile = CarbonTablePath.getSchemaFilePath(tablePath)
- val fileType = FileFactory.getFileType(tableMetadataFile)
- if (FileFactory.isFileExist(tableMetadataFile, fileType)) {
+ if (FileFactory.isFileExist(tableMetadataFile)) {
val tableInfo: TableInfo = CarbonUtil.readSchemaFile(tableMetadataFile)
val wrapperTableInfo =
schemaConverter.fromExternalToWrapperTableInfo(tableInfo, dbName, tableName, tablePath)
@@ -414,8 +413,7 @@ class CarbonFileMetastore extends CarbonMetaStore {
thriftTableInfo: TableInfo): String = {
val schemaFilePath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath)
val schemaMetadataPath = CarbonTablePath.getFolderContainingFile(schemaFilePath)
- val fileType = FileFactory.getFileType(schemaMetadataPath)
- if (!FileFactory.isFileExist(schemaMetadataPath, fileType)) {
+ if (!FileFactory.isFileExist(schemaMetadataPath)) {
val isDirCreated = FileFactory
.mkdirs(schemaMetadataPath, SparkSession.getActiveSession.get.sessionState.newHadoopConf())
if (!isDirCreated) {
@@ -463,8 +461,7 @@ class CarbonFileMetastore extends CarbonMetaStore {
try {
val tablePath = lookupRelation(tableIdentifier)(sparkSession)
.asInstanceOf[CarbonRelation].carbonTable.getTablePath
- val fileType = FileFactory.getFileType(tablePath)
- FileFactory.isFileExist(tablePath, fileType)
+ FileFactory.isFileExist(tablePath)
} catch {
case _: Exception =>
false
@@ -507,7 +504,6 @@ class CarbonFileMetastore extends CarbonMetaStore {
CarbonFileMetastore.updateTableSchemaModifiedTime(tableUniqueId, timeStamp)
}
-
override def isSchemaRefreshed(absoluteTableIdentifier: AbsoluteTableIdentifier,
sparkSession: SparkSession): Boolean = {
val localTimeStamp = Option(tableModifiedTimeStore.get(absoluteTableIdentifier
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
index 7adc025..cc22d50 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonRelation.scala
@@ -168,8 +168,7 @@ case class CarbonRelation(
def sizeInBytes: Long = {
if (carbonTable.isExternalTable) {
val tablePath = carbonTable.getTablePath
- val fileType = FileFactory.getFileType(tablePath)
- if (FileFactory.isFileExist(tablePath, fileType)) {
+ if (FileFactory.isFileExist(tablePath)) {
sizeInBytesLocalValue = FileFactory.getDirectorySize(tablePath)
}
} else {
@@ -182,8 +181,7 @@ case class CarbonRelation(
sizeInBytesLocalValue = 0L
} else {
val tablePath = carbonTable.getTablePath
- val fileType = FileFactory.getFileType(tablePath)
- if (FileFactory.isFileExist(tablePath, fileType)) {
+ if (FileFactory.isFileExist(tablePath)) {
// get the valid segments
val segments = allSegments.getValidSegments.asScala
var size = 0L
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParserUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParserUtil.scala
index 4d85e88..b720a19 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParserUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParserUtil.scala
@@ -157,7 +157,7 @@ object CarbonSparkSqlParserUtil {
tableIdentifier.table.toLowerCase())
val table = try {
val schemaPath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath)
- if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
+ if (!FileFactory.isFileExist(schemaPath)) {
if (provider.equalsIgnoreCase("'carbonfile'")) {
SchemaReader.inferSchema(identifier, true)
} else {
diff --git a/integration/spark2/src/main/scala/org/apache/spark/util/AlterTableUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/util/AlterTableUtil.scala
index 6add24a..f21e4db 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/util/AlterTableUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/util/AlterTableUtil.scala
@@ -236,8 +236,7 @@ object AlterTableUtil {
val database = oldCarbonTable.getDatabaseName
val newCarbonTableIdentifier = new CarbonTableIdentifier(database, newTableName, tableId)
val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
- val fileType = FileFactory.getFileType(tablePath)
- if (FileFactory.isFileExist(tablePath, fileType)) {
+ if (FileFactory.isFileExist(tablePath)) {
val tableInfo = metastore.getThriftTableInfo(oldCarbonTable)
val evolutionEntryList = tableInfo.fact_table.schema_evolution.schema_evolution_history
val updatedTime = evolutionEntryList.get(evolutionEntryList.size() - 1).time_stamp
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
index 5bdf3b4c..bb04d0a 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
@@ -864,7 +864,7 @@ class TestAlterPartitionTable extends QueryTest with BeforeAndAfterAll {
def getDataFiles(carbonTable: CarbonTable, segmentId: String): Array[String] = {
val segment = Segment.getSegment(segmentId, carbonTable.getTablePath)
val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)
- val carbonFile = FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir))
+ val carbonFile = FileFactory.getCarbonFile(segmentDir)
val dataFiles = carbonFile.listFiles(new CarbonFileFilter() {
override def accept(file: CarbonFile): Boolean = {
return file.getName.endsWith(".carbondata")
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
index c7f080d..1bbb1b3 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/AllDictionaryTestCase.scala
@@ -67,9 +67,8 @@ class AllDictionaryTestCase extends Spark2QueryTest with BeforeAndAfterAll {
LoadOption.getCsvHeaderColumns(carbonLoadModel, FileFactory.getConfiguration))
// Create table and metadata folders if not exist
val metadataDirectoryPath = CarbonTablePath.getMetadataPath(table.getTablePath)
- val fileType = FileFactory.getFileType(metadataDirectoryPath)
- if (!FileFactory.isFileExist(metadataDirectoryPath, fileType)) {
- FileFactory.mkdirs(metadataDirectoryPath, fileType)
+ if (!FileFactory.isFileExist(metadataDirectoryPath)) {
+ FileFactory.mkdirs(metadataDirectoryPath)
}
import scala.collection.JavaConverters._
val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
index e4dca67..41aa8ed 100644
--- a/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/util/ExternalColumnDictionaryTestCase.scala
@@ -182,9 +182,8 @@ class ExternalColumnDictionaryTestCase extends Spark2QueryTest with BeforeAndAft
carbonLoadModel.setMaxColumns("100")
// Create table and metadata folders if not exist
val metadataDirectoryPath = CarbonTablePath.getMetadataPath(table.getTablePath)
- val fileType = FileFactory.getFileType(metadataDirectoryPath)
- if (!FileFactory.isFileExist(metadataDirectoryPath, fileType)) {
- FileFactory.mkdirs(metadataDirectoryPath, fileType)
+ if (!FileFactory.isFileExist(metadataDirectoryPath)) {
+ FileFactory.mkdirs(metadataDirectoryPath)
}
import scala.collection.JavaConverters._
val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/BadRecordPathLoadOptionTest.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/BadRecordPathLoadOptionTest.scala
index 1cec6ec..21a17de 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/BadRecordPathLoadOptionTest.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/BadRecordPathLoadOptionTest.scala
@@ -62,8 +62,7 @@ class BadRecordPathLoadOptionTest extends Spark2QueryTest with BeforeAndAfterAll
val badStorePath =
CarbonEnv.getCarbonTable(Some("default"), "salestest")(sqlContext.sparkSession).getTableInfo
.getFactTable.getTableProperties.get("bad_record_path") + "/0/0"
- val carbonFile: CarbonFile = FileFactory
- .getCarbonFile(badStorePath, FileFactory.getFileType(badStorePath))
+ val carbonFile: CarbonFile = FileFactory.getCarbonFile(badStorePath)
var exists: Boolean = carbonFile.exists()
if (exists) {
val listFiles: Array[CarbonFile] = carbonFile.listFiles(new CarbonFileFilter {
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/BadRecordsLogger.java b/processing/src/main/java/org/apache/carbondata/processing/loading/BadRecordsLogger.java
index e0dcd26..6a939bf 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/BadRecordsLogger.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/BadRecordsLogger.java
@@ -30,7 +30,6 @@ import java.util.concurrent.ConcurrentHashMap;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.impl.FileFactory;
-import org.apache.carbondata.core.datastore.impl.FileFactory.FileType;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException;
@@ -177,16 +176,15 @@ public class BadRecordsLogger {
}
try {
if (null == bufferedWriter) {
- FileType fileType = FileFactory.getFileType(storePath);
- if (!FileFactory.isFileExist(this.storePath, fileType)) {
+ if (!FileFactory.isFileExist(this.storePath)) {
// create the folders if not exist
- FileFactory.mkdirs(this.storePath, fileType);
+ FileFactory.mkdirs(this.storePath);
// create the files
- FileFactory.createNewFile(logFilePath, fileType);
+ FileFactory.createNewFile(logFilePath);
}
- outStream = FileFactory.getDataOutputStream(logFilePath, fileType);
+ outStream = FileFactory.getDataOutputStream(logFilePath);
bufferedWriter = new BufferedWriter(new OutputStreamWriter(outStream,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
@@ -217,16 +215,15 @@ public class BadRecordsLogger {
}
try {
if (null == bufferedCSVWriter) {
- FileType fileType = FileFactory.getFileType(storePath);
- if (!FileFactory.isFileExist(this.storePath, fileType)) {
+ if (!FileFactory.isFileExist(this.storePath)) {
// create the folders if not exist
- FileFactory.mkdirs(this.storePath, fileType);
+ FileFactory.mkdirs(this.storePath);
// create the files
- FileFactory.createNewFile(csvFilePath, fileType);
+ FileFactory.createNewFile(csvFilePath);
}
- outCSVStream = FileFactory.getDataOutputStream(csvFilePath, fileType);
+ outCSVStream = FileFactory.getDataOutputStream(csvFilePath);
bufferedCSVWriter = new BufferedWriter(new OutputStreamWriter(outCSVStream,
Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)));
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java b/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
index 523a70a..a45e45b 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/TableProcessingOperations.java
@@ -58,9 +58,8 @@ public class TableProcessingOperations {
//delete folder which metadata no exist in tablestatus
String partitionPath = CarbonTablePath.getPartitionDir(carbonTable.getTablePath());
- FileFactory.FileType fileType = FileFactory.getFileType(partitionPath);
- if (FileFactory.isFileExist(partitionPath, fileType)) {
- CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath, fileType);
+ if (FileFactory.isFileExist(partitionPath)) {
+ CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath);
CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() {
@Override
public boolean accept(CarbonFile path) {
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/UnsafeSortDataRows.java b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/UnsafeSortDataRows.java
index e647381..3248bae 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/UnsafeSortDataRows.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/UnsafeSortDataRows.java
@@ -299,7 +299,7 @@ public class UnsafeSortDataRows {
DataOutputStream stream = null;
try {
// open stream
- stream = FileFactory.getDataOutputStream(file.getPath(), FileFactory.FileType.LOCAL,
+ stream = FileFactory.getDataOutputStream(file.getPath(),
parameters.getFileWriteBufferSize(), parameters.getSortTempCompressorName());
int actualSize = rowPage.getBuffer().getActualSize();
// write number of entries to the file
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/holder/UnsafeSortTempFileChunkHolder.java b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/holder/UnsafeSortTempFileChunkHolder.java
index 5664e86..7d51cd5 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/holder/UnsafeSortTempFileChunkHolder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/holder/UnsafeSortTempFileChunkHolder.java
@@ -144,7 +144,7 @@ public class UnsafeSortTempFileChunkHolder implements SortTempChunkHolder {
private void initialise() {
try {
- stream = FileFactory.getDataInputStream(tempFile.getPath(), FileFactory.FileType.LOCAL,
+ stream = FileFactory.getDataInputStream(tempFile.getPath(),
readBufferSize, compressorName);
this.entryCount = stream.readInt();
LOGGER.info("Processing unsafe mode file rows with size : " + entryCount);
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeInMemoryIntermediateDataMerger.java b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeInMemoryIntermediateDataMerger.java
index 8a4503b..c9dcba5 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeInMemoryIntermediateDataMerger.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeInMemoryIntermediateDataMerger.java
@@ -251,8 +251,7 @@ public class UnsafeInMemoryIntermediateDataMerger implements Callable<Void> {
+ sortParameters.getRangeId() + '_' + System.nanoTime()
+ CarbonCommonConstants.SORT_TEMP_FILE_EXT);
outputStream = FileFactory.getDataOutputStream(outputFile.getPath(),
- FileFactory.FileType.LOCAL, sortParameters.getFileWriteBufferSize(),
- sortParameters.getSortTempCompressorName());
+ sortParameters.getFileWriteBufferSize(), sortParameters.getSortTempCompressorName());
outputStream.writeInt(totalSize);
}
diff --git a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateFileMerger.java b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateFileMerger.java
index 9af15aa..2764acc 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateFileMerger.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/loading/sort/unsafe/merger/UnsafeIntermediateFileMerger.java
@@ -138,8 +138,8 @@ public class UnsafeIntermediateFileMerger implements Callable<Void> {
*/
private void initialize() throws CarbonSortKeyAndGroupByException {
try {
- stream = FileFactory.getDataOutputStream(outPutFile.getPath(), FileFactory.FileType.LOCAL,
- writeBufferSize, compressorName);
+ stream =
+ FileFactory.getDataOutputStream(outPutFile.getPath(), writeBufferSize, compressorName);
this.stream.writeInt(this.totalNumberOfRecords);
} catch (FileNotFoundException e) {
throw new CarbonSortKeyAndGroupByException("Problem while getting the file", e);
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionUtil.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionUtil.java
index d42b46a..98122e1 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonCompactionUtil.java
@@ -200,10 +200,8 @@ public class CarbonCompactionUtil {
String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
+ CarbonCommonConstants.majorCompactionRequiredFile;
try {
- if (FileFactory.isFileExist(minorCompactionStatusFile,
- FileFactory.getFileType(minorCompactionStatusFile)) || FileFactory
- .isFileExist(majorCompactionStatusFile,
- FileFactory.getFileType(majorCompactionStatusFile))) {
+ if (FileFactory.isFileExist(minorCompactionStatusFile) || FileFactory
+ .isFileExist(majorCompactionStatusFile)) {
return true;
}
} catch (IOException e) {
@@ -224,12 +222,10 @@ public class CarbonCompactionUtil {
String majorCompactionStatusFile = metaFolderPath + CarbonCommonConstants.FILE_SEPARATOR
+ CarbonCommonConstants.majorCompactionRequiredFile;
try {
- if (FileFactory.isFileExist(minorCompactionStatusFile,
- FileFactory.getFileType(minorCompactionStatusFile))) {
+ if (FileFactory.isFileExist(minorCompactionStatusFile)) {
return CompactionType.MINOR;
}
- if (FileFactory.isFileExist(majorCompactionStatusFile,
- FileFactory.getFileType(majorCompactionStatusFile))) {
+ if (FileFactory.isFileExist(majorCompactionStatusFile)) {
return CompactionType.MAJOR;
}
@@ -257,9 +253,9 @@ public class CarbonCompactionUtil {
}
try {
if (FileFactory
- .isFileExist(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))) {
+ .isFileExist(compactionRequiredFile)) {
if (FileFactory
- .getCarbonFile(compactionRequiredFile, FileFactory.getFileType(compactionRequiredFile))
+ .getCarbonFile(compactionRequiredFile)
.delete()) {
LOGGER.info("Deleted the compaction request file " + compactionRequiredFile);
return true;
@@ -292,8 +288,8 @@ public class CarbonCompactionUtil {
+ CarbonCommonConstants.majorCompactionRequiredFile;
}
try {
- if (!FileFactory.isFileExist(statusFile, FileFactory.getFileType(statusFile))) {
- if (FileFactory.createNewFile(statusFile, FileFactory.getFileType(statusFile))) {
+ if (!FileFactory.isFileExist(statusFile)) {
+ if (FileFactory.createNewFile(statusFile)) {
LOGGER.info("successfully created a compaction required file - " + statusFile);
return true;
} else {
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
index ca34c1d..163526a 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CarbonDataMergerUtil.java
@@ -736,7 +736,7 @@ public final class CarbonDataMergerUtil {
private static long getSizeOfSegment(String tablePath, String segId) {
String loadPath = CarbonTablePath.getSegmentPath(tablePath, segId);
CarbonFile segmentFolder =
- FileFactory.getCarbonFile(loadPath, FileFactory.getFileType(loadPath));
+ FileFactory.getCarbonFile(loadPath);
return getSizeOfFactFileInLoad(segmentFolder);
}
@@ -1099,7 +1099,7 @@ public final class CarbonDataMergerUtil {
String segmentPath = CarbonTablePath.getSegmentPath(
identifier.getTablePath(), seg.getSegmentNo());
CarbonFile segDir =
- FileFactory.getCarbonFile(segmentPath, FileFactory.getFileType(segmentPath));
+ FileFactory.getCarbonFile(segmentPath);
CarbonFile[] allSegmentFiles = segDir.listFiles();
updateDeltaFiles = segmentUpdateStatusManager
@@ -1296,8 +1296,7 @@ public final class CarbonDataMergerUtil {
throw new IOException();
}
CarbonDeleteDeltaWriterImpl carbonDeleteWriter =
- new CarbonDeleteDeltaWriterImpl(fullBlockFilePath,
- FileFactory.getFileType(fullBlockFilePath));
+ new CarbonDeleteDeltaWriterImpl(fullBlockFilePath);
try {
carbonDeleteWriter.write(deleteDeltaBlockDetails);
} catch (IOException e) {
diff --git a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/IntermediateFileMerger.java b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/IntermediateFileMerger.java
index 9376635..d601fe5 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/IntermediateFileMerger.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/IntermediateFileMerger.java
@@ -134,7 +134,7 @@ public class IntermediateFileMerger implements Callable<Void> {
*/
private void initialize() throws CarbonSortKeyAndGroupByException {
try {
- stream = FileFactory.getDataOutputStream(outPutFile.getPath(), FileFactory.FileType.LOCAL,
+ stream = FileFactory.getDataOutputStream(outPutFile.getPath(),
writeBufferSize, compressorName);
this.stream.writeInt(this.totalNumberOfRecords);
} catch (FileNotFoundException e) {
diff --git a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortDataRows.java b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortDataRows.java
index bc45f5e..b9ead2d 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortDataRows.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortDataRows.java
@@ -237,7 +237,7 @@ public class SortDataRows {
DataOutputStream stream = null;
try {
// open stream
- stream = FileFactory.getDataOutputStream(file.getPath(), FileFactory.FileType.LOCAL,
+ stream = FileFactory.getDataOutputStream(file.getPath(),
parameters.getFileWriteBufferSize(), parameters.getSortTempCompressorName());
// write number of entries to the file
stream.writeInt(entryCountLocal);
diff --git a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortTempFileChunkHolder.java b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortTempFileChunkHolder.java
index b627d8b..652652d 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortTempFileChunkHolder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/sort/sortdata/SortTempFileChunkHolder.java
@@ -158,7 +158,7 @@ public class SortTempFileChunkHolder implements Comparable<SortTempFileChunkHold
private void initialise() throws CarbonSortKeyAndGroupByException {
try {
- stream = FileFactory.getDataInputStream(tempFile.getPath(), FileFactory.FileType.LOCAL,
+ stream = FileFactory.getDataInputStream(tempFile.getPath(),
readBufferSize, compressorName);
this.entryCount = stream.readInt();
if (prefetch) {
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 6ccc75f..66e9e16 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
@@ -282,8 +282,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
if (copyInCurrentThread) {
CarbonUtil.copyCarbonDataFileToCarbonStorePath(carbonDataFileTempPath,
model.getCarbonDataDirectoryPath(), fileSizeInBytes);
- FileFactory.deleteFile(carbonDataFileTempPath,
- FileFactory.getFileType(carbonDataFileTempPath));
+ FileFactory.deleteFile(carbonDataFileTempPath);
} else {
executorServiceSubmitList
.add(executorService.submit(new CompleteHdfsBackendThread(carbonDataFileTempPath)));
@@ -304,7 +303,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
}
private void handleEmptyDataFile(String filePath) throws IOException {
- FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath));
+ FileFactory.deleteFile(filePath);
if (blockIndexInfoList.size() > 0 && blockIndexInfoList.get(blockIndexInfoList.size() - 1)
.getFileName().equals(carbonDataFileName)) {
// no need add this entry in index file
@@ -331,8 +330,8 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
// 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);
+ .getDataOutputStream(carbonDataFileStorePath, CarbonCommonConstants.BYTEBUFFER_SIZE,
+ fileSizeInBytes * 2);
} else {
//each time we initialize writer, we choose a local temp location randomly
String[] tempFileLocations = model.getStoreLocation();
@@ -340,8 +339,9 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
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);
+ fileOutputStream = FileFactory
+ .getDataOutputStream(carbonDataFileTempPath, CarbonCommonConstants.BYTEBUFFER_SIZE,
+ true);
}
this.fileCount++;
@@ -419,7 +419,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
+ CarbonTablePath.getCarbonIndexFileName(model.getCarbonDataFileAttributes().getTaskId(),
model.getBucketId(), model.getTaskExtension(),
"" + model.getCarbonDataFileAttributes().getFactTimeStamp(), model.getSegmentId());
- indexFileName = FileFactory.getUpdatedFilePath(rawFileName, FileFactory.FileType.HDFS);
+ indexFileName = FileFactory.getUpdatedFilePath(rawFileName);
} else {
// randomly choose a temp location for index file
String[] tempLocations = model.getStoreLocation();
@@ -445,7 +445,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
CarbonUtil
.copyCarbonDataFileToCarbonStorePath(indexFileName, model.getCarbonDataDirectoryPath(),
fileSizeInBytes);
- FileFactory.deleteFile(indexFileName, FileFactory.getFileType(indexFileName));
+ FileFactory.deleteFile(indexFileName);
}
}
@@ -509,7 +509,7 @@ public abstract class AbstractFactDataWriter implements CarbonFactDataWriter {
public Void call() throws Exception {
CarbonUtil.copyCarbonDataFileToCarbonStorePath(fileName, model.getCarbonDataDirectoryPath(),
fileSizeInBytes);
- FileFactory.deleteFile(fileName, FileFactory.getFileType(fileName));
+ FileFactory.deleteFile(fileName);
return null;
}
}
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java
index 194d6ec..5e9ed7d 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonBadRecordUtil.java
@@ -77,15 +77,14 @@ public class CarbonBadRecordUtil {
}
badLogStoreLocation = badLogStoreLocation + File.separator + storeLocation;
- FileFactory.FileType fileType = FileFactory.getFileType(badLogStoreLocation);
try {
- if (!FileFactory.isFileExist(badLogStoreLocation, fileType)) {
+ if (!FileFactory.isFileExist(badLogStoreLocation)) {
return;
}
} catch (IOException e1) {
LOGGER.info("bad record folder does not exist");
}
- CarbonFile carbonFile = FileFactory.getCarbonFile(badLogStoreLocation, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(badLogStoreLocation);
CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() {
@Override
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
index 2d9ad4a..8ca5ae6 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonLoaderUtil.java
@@ -43,7 +43,6 @@ import org.apache.carbondata.core.datastore.block.TableBlockInfo;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.datastore.filesystem.CarbonFileFilter;
import org.apache.carbondata.core.datastore.impl.FileFactory;
-import org.apache.carbondata.core.datastore.impl.FileFactory.FileType;
import org.apache.carbondata.core.locks.CarbonLockUtil;
import org.apache.carbondata.core.locks.ICarbonLock;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
@@ -114,8 +113,7 @@ public final class CarbonLoaderUtil {
int fileCount = 0;
String segmentPath = CarbonTablePath.getSegmentPath(
loadModel.getTablePath(), currentLoad + "");
- CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath,
- FileFactory.getFileType(segmentPath));
+ CarbonFile carbonFile = FileFactory.getCarbonFile(segmentPath);
CarbonFile[] files = carbonFile.listFiles(new CarbonFileFilter() {
@Override
@@ -139,9 +137,8 @@ public final class CarbonLoaderUtil {
public static void deleteStorePath(String path) {
try {
- FileType fileType = FileFactory.getFileType(path);
- if (FileFactory.isFileExist(path, fileType)) {
- CarbonFile carbonFile = FileFactory.getCarbonFile(path, fileType);
+ if (FileFactory.isFileExist(path)) {
+ CarbonFile carbonFile = FileFactory.getCarbonFile(path);
CarbonUtil.deleteFoldersAndFiles(carbonFile);
}
} catch (IOException | InterruptedException e) {
@@ -229,9 +226,8 @@ public final class CarbonLoaderUtil {
loadModel.getCarbonDataLoadSchema().getCarbonTable().getAbsoluteTableIdentifier();
if (loadModel.isCarbonTransactionalTable()) {
String metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath());
- FileType fileType = FileFactory.getFileType(metadataPath);
- if (!FileFactory.isFileExist(metadataPath, fileType)) {
- FileFactory.mkdirs(metadataPath, fileType);
+ if (!FileFactory.isFileExist(metadataPath)) {
+ FileFactory.mkdirs(metadataPath);
}
}
String tableStatusPath;
@@ -397,7 +393,7 @@ public final class CarbonLoaderUtil {
identifier.getTablePath(), entry.getLoadName());
// add to the deletion list only if file exist else HDFS file system will throw
// exception while deleting the file if file path does not exist
- if (FileFactory.isFileExist(path, FileFactory.getFileType(path))) {
+ if (FileFactory.isFileExist(path)) {
staleFolders.add(FileFactory.getCarbonFile(path));
}
}
@@ -1192,7 +1188,7 @@ public final class CarbonLoaderUtil {
private static void deleteFiles(List<String> filesToBeDeleted) throws IOException {
for (String filePath : filesToBeDeleted) {
- FileFactory.deleteFile(filePath, FileFactory.getFileType(filePath));
+ FileFactory.deleteFile(filePath);
}
}
diff --git a/processing/src/test/java/org/apache/carbondata/lcm/locks/LocalFileLockTest.java b/processing/src/test/java/org/apache/carbondata/lcm/locks/LocalFileLockTest.java
index 957053e..de5982f 100644
--- a/processing/src/test/java/org/apache/carbondata/lcm/locks/LocalFileLockTest.java
+++ b/processing/src/test/java/org/apache/carbondata/lcm/locks/LocalFileLockTest.java
@@ -48,7 +48,7 @@ public class LocalFileLockTest {
@Before public void setUp() throws Exception {
rootPath = new File(this.getClass().getResource("/").getPath()
+ "../../..").getCanonicalPath();
- String storeLocation = rootPath + "/target/store";
+ String storeLocation = rootPath + "/target/store_new/";
CarbonProperties.getInstance()
.addProperty("carbon.storelocation", storeLocation);
}
@@ -60,6 +60,8 @@ public class LocalFileLockTest {
Field f = secretClass.getDeclaredField("lockPath");
f.setAccessible(true);
f.set(secretClass, "");
+ String storeLocation = rootPath + "/target/store_new/";
+ new File(storeLocation).delete();
}
@Test public void testingLocalFileLockingByAcquiring2Locks() {
@@ -88,7 +90,7 @@ public class LocalFileLockTest {
f.set(secretClass, rootPath + "/target/");
AbsoluteTableIdentifier absoluteTableIdentifier = AbsoluteTableIdentifier
.from(CarbonProperties.getInstance().getProperty("carbon.storelocation"), "databaseName",
- "tableName", "1");
+ "tableName1", "1");
ICarbonLock carbonLock = CarbonLockFactory.getCarbonLockObj(absoluteTableIdentifier, LockUsage.TABLE_STATUS_LOCK);
carbonLock.lockWithRetries();
assert (new File(rootPath + "/target/1/LockFiles/tablestatus.lock").exists());
diff --git a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonSchemaReader.java b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonSchemaReader.java
index bc6228a..7929810 100644
--- a/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonSchemaReader.java
+++ b/store/sdk/src/main/java/org/apache/carbondata/sdk/file/CarbonSchemaReader.java
@@ -314,7 +314,7 @@ public class CarbonSchemaReader {
*/
public static String getVersionDetails(String dataFilePath) throws IOException {
long fileSize =
- FileFactory.getCarbonFile(dataFilePath, FileFactory.getFileType(dataFilePath)).getSize();
+ FileFactory.getCarbonFile(dataFilePath).getSize();
FileReader fileReader = FileFactory.getFileHolder(FileFactory.getFileType(dataFilePath));
ByteBuffer buffer =
fileReader.readByteBuffer(FileFactory.getUpdatedFilePath(dataFilePath), fileSize - 8, 8);
diff --git a/store/sdk/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java b/store/sdk/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
index ed2be51..e9dfca6 100644
--- a/store/sdk/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
+++ b/store/sdk/src/main/java/org/apache/carbondata/store/MetaCachedCarbonStore.java
@@ -47,7 +47,7 @@ abstract class MetaCachedCarbonStore implements CarbonStore {
}
String schemaPath = CarbonTablePath.getSchemaFilePath(path);
TableInfo tableInfo;
- if (!FileFactory.isFileExist(schemaPath, FileFactory.getFileType(schemaPath))) {
+ if (!FileFactory.isFileExist(schemaPath)) {
tableInfo = SchemaReader.inferSchema(AbsoluteTableIdentifier.from(path), false);
} else {
org.apache.carbondata.format.TableInfo tableInfoFormat;
diff --git a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CarbonReaderTest.java b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CarbonReaderTest.java
index 0a4aac1..92f819b 100644
--- a/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CarbonReaderTest.java
+++ b/store/sdk/src/test/java/org/apache/carbondata/sdk/file/CarbonReaderTest.java
@@ -218,11 +218,11 @@ public class CarbonReaderTest extends TestCase {
FileUtils.deleteDirectory(new File(path));
- FileFactory.getCarbonFile(path1, FileFactory.getFileType(path1));
- FileFactory.mkdirs(path1, FileFactory.getFileType(path1));
+ FileFactory.getCarbonFile(path1);
+ FileFactory.mkdirs(path1);
- FileFactory.getCarbonFile(path2, FileFactory.getFileType(path2));
- FileFactory.mkdirs(path2, FileFactory.getFileType(path2));
+ FileFactory.getCarbonFile(path2);
+ FileFactory.mkdirs(path2);
Field[] fields = new Field[2];
fields[0] = new Field("name", DataTypes.STRING);
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 1363e13..7ac4a3f 100644
--- a/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
+++ b/streaming/src/main/java/org/apache/carbondata/streaming/CarbonStreamRecordWriter.java
@@ -170,11 +170,10 @@ public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
// initialize data writer and compressor
String filePath = segmentDir + File.separator + fileName;
- FileFactory.FileType fileType = FileFactory.getFileType(filePath);
- CarbonFile carbonFile = FileFactory.getCarbonFile(filePath, fileType);
+ CarbonFile carbonFile = FileFactory.getCarbonFile(filePath);
if (carbonFile.exists()) {
// if the file is existed, use the append api
- outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath, fileType);
+ outputStream = FileFactory.getDataOutputStreamUsingAppend(filePath);
// get the compressor from the fileheader. In legacy store,
// the compressor name is not set and it use snappy compressor
FileHeader header = new CarbonHeaderReader(filePath).readHeader();
@@ -185,7 +184,7 @@ public class CarbonStreamRecordWriter extends RecordWriter<Void, Object> {
}
} else {
// IF the file is not existed, use the create api
- outputStream = FileFactory.getDataOutputStream(filePath, fileType);
+ outputStream = FileFactory.getDataOutputStream(filePath);
compressorName = carbonTable.getTableInfo().getFactTable().getTableProperties().get(
CarbonCommonConstants.COMPRESSOR);
if (null == compressorName) {
diff --git a/streaming/src/main/java/org/apache/carbondata/streaming/segment/StreamSegment.java b/streaming/src/main/java/org/apache/carbondata/streaming/segment/StreamSegment.java
index 7c849c4..c40698a 100644
--- a/streaming/src/main/java/org/apache/carbondata/streaming/segment/StreamSegment.java
+++ b/streaming/src/main/java/org/apache/carbondata/streaming/segment/StreamSegment.java
@@ -317,12 +317,11 @@ public class StreamSegment {
* 2. after job failed (CarbonAppendableStreamSink.writeDataFileJob)
*/
public static void recoverSegmentIfRequired(String segmentDir) throws IOException {
- FileFactory.FileType fileType = FileFactory.getFileType(segmentDir);
- if (FileFactory.isFileExist(segmentDir, fileType)) {
+ if (FileFactory.isFileExist(segmentDir)) {
String indexName = CarbonTablePath.getCarbonStreamIndexFileName();
String indexPath = segmentDir + File.separator + indexName;
- CarbonFile index = FileFactory.getCarbonFile(indexPath, fileType);
- CarbonFile[] files = listDataFiles(segmentDir, fileType);
+ CarbonFile index = FileFactory.getCarbonFile(indexPath);
+ CarbonFile[] files = listDataFiles(segmentDir);
// TODO better to check backup index at first
// index file exists
if (index.exists()) {
@@ -343,7 +342,7 @@ public class StreamSegment {
if (null == size || size == 0) {
file.delete();
} else if (size < file.getSize()) {
- FileFactory.truncateFile(file.getCanonicalPath(), fileType, size);
+ FileFactory.truncateFile(file.getCanonicalPath(), size);
}
}
} finally {
@@ -370,11 +369,10 @@ public class StreamSegment {
String fileName,
String indexName) throws IOException {
- FileFactory.FileType fileType = FileFactory.getFileType(segmentDir);
String filePath = segmentDir + File.separator + fileName;
- CarbonFile file = FileFactory.getCarbonFile(filePath, fileType);
+ CarbonFile file = FileFactory.getCarbonFile(filePath);
String indexPath = segmentDir + File.separator + indexName;
- CarbonFile index = FileFactory.getCarbonFile(indexPath, fileType);
+ CarbonFile index = FileFactory.getCarbonFile(indexPath);
if (file.exists() && index.exists()) {
CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
try {
@@ -385,7 +383,7 @@ public class StreamSegment {
if (blockIndex.getFile_size() == 0) {
file.delete();
} else if (blockIndex.getFile_size() < file.getSize()) {
- FileFactory.truncateFile(filePath, fileType, blockIndex.getFile_size());
+ FileFactory.truncateFile(filePath, blockIndex.getFile_size());
}
break;
}
@@ -399,8 +397,8 @@ public class StreamSegment {
/**
* list all carbondata files of a segment
*/
- public static CarbonFile[] listDataFiles(String segmentDir, FileFactory.FileType fileType) {
- CarbonFile carbonDir = FileFactory.getCarbonFile(segmentDir, fileType);
+ public static CarbonFile[] listDataFiles(String segmentDir) {
+ CarbonFile carbonDir = FileFactory.getCarbonFile(segmentDir);
if (carbonDir.exists()) {
return carbonDir.listFiles(new CarbonFileFilter() {
@Override
@@ -421,10 +419,10 @@ public class StreamSegment {
* @return the list of BlockIndex in the index file
* @throws IOException
*/
- public static List<BlockIndex> readIndexFile(String indexPath, FileFactory.FileType fileType)
+ public static List<BlockIndex> readIndexFile(String indexPath)
throws IOException {
List<BlockIndex> blockIndexList = new ArrayList<>();
- CarbonFile index = FileFactory.getCarbonFile(indexPath, fileType);
+ CarbonFile index = FileFactory.getCarbonFile(indexPath);
if (index.exists()) {
CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
try {
@@ -590,9 +588,9 @@ public class StreamSegment {
* merge new blocklet index and old file index to create new file index
*/
private static void updateStreamFileIndex(Map<String, StreamFileIndex> indexMap,
- String indexPath, FileFactory.FileType fileType, DataType[] msrDataTypes
+ String indexPath, DataType[] msrDataTypes
) throws IOException {
- List<BlockIndex> blockIndexList = readIndexFile(indexPath, fileType);
+ List<BlockIndex> blockIndexList = readIndexFile(indexPath);
for (BlockIndex blockIndex : blockIndexList) {
BlockletMinMaxIndex fileIndex = CarbonMetadataUtil
.convertExternalMinMaxIndex(blockIndex.getBlock_index().getMin_max_index());
@@ -614,20 +612,19 @@ public class StreamSegment {
*/
public static void updateIndexFile(String segmentDir,
StreamFileIndex[] blockIndexes, DataType[] msrDataTypes) throws IOException {
- FileFactory.FileType fileType = FileFactory.getFileType(segmentDir);
String filePath = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);
// update min/max index
Map<String, StreamFileIndex> indexMap = new HashMap<>();
for (StreamFileIndex fileIndex : blockIndexes) {
indexMap.put(fileIndex.getFileName(), fileIndex);
}
- updateStreamFileIndex(indexMap, filePath, fileType, msrDataTypes);
+ updateStreamFileIndex(indexMap, filePath, msrDataTypes);
String tempFilePath = filePath + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
CarbonIndexFileWriter writer = new CarbonIndexFileWriter();
try {
writer.openThriftWriter(tempFilePath);
- CarbonFile[] files = listDataFiles(segmentDir, fileType);
+ CarbonFile[] files = listDataFiles(segmentDir);
BlockIndex blockIndex;
for (CarbonFile file : files) {
blockIndex = new BlockIndex();
@@ -649,7 +646,7 @@ public class StreamSegment {
writer.writeThrift(blockIndex);
}
writer.close();
- CarbonFile tempFile = FileFactory.getCarbonFile(tempFilePath, fileType);
+ CarbonFile tempFile = FileFactory.getCarbonFile(tempFilePath);
if (!tempFile.renameForce(filePath)) {
throw new IOException(
"temporary file renaming failed, src=" + tempFilePath + ", dest=" + filePath);
@@ -669,10 +666,9 @@ public class StreamSegment {
*/
public static long size(String segmentDir) throws IOException {
long size = 0;
- FileFactory.FileType fileType = FileFactory.getFileType(segmentDir);
- if (FileFactory.isFileExist(segmentDir, fileType)) {
+ if (FileFactory.isFileExist(segmentDir)) {
String indexPath = CarbonTablePath.getCarbonStreamIndexFilePath(segmentDir);
- CarbonFile index = FileFactory.getCarbonFile(indexPath, fileType);
+ CarbonFile index = FileFactory.getCarbonFile(indexPath);
if (index.exists()) {
CarbonIndexFileReader indexReader = new CarbonIndexFileReader();
try {