QiangCai commented on a change in pull request #4044:
URL: https://github.com/apache/carbondata/pull/4044#discussion_r537289240
##########
File path:
core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
##########
@@ -482,176 +482,6 @@ public boolean accept(CarbonFile file) {
}
- /**
- * Handling of the clean up of old carbondata files, index files , delete
delta,
- * update status files.
- * @param table clean up will be handled on this table.
- * @param forceDelete if true then max query execution timeout will not be
considered.
- */
- public static void cleanUpDeltaFiles(CarbonTable table, boolean forceDelete)
throws IOException {
-
- SegmentStatusManager ssm = new
SegmentStatusManager(table.getAbsoluteTableIdentifier());
-
- LoadMetadataDetails[] details =
- SegmentStatusManager.readLoadMetadata(table.getMetadataPath());
-
- SegmentUpdateStatusManager updateStatusManager = new
SegmentUpdateStatusManager(table);
- SegmentUpdateDetails[] segmentUpdateDetails =
updateStatusManager.getUpdateStatusDetails();
- // hold all the segments updated so that wen can check the delta files in
them, ne need to
- // check the others.
- Set<String> updatedSegments = new HashSet<>();
- for (SegmentUpdateDetails updateDetails : segmentUpdateDetails) {
- updatedSegments.add(updateDetails.getSegmentName());
- }
-
- String validUpdateStatusFile = "";
-
- boolean isAbortedFile = true;
-
- boolean isInvalidFile = false;
-
- // take the update status file name from 0th segment.
- validUpdateStatusFile = ssm.getUpdateStatusFileName(details);
- // scan through each segment.
- for (LoadMetadataDetails segment : details) {
- // if this segment is valid then only we will go for delta file deletion.
- // if the segment is mark for delete or compacted then any way it will
get deleted.
- if (segment.getSegmentStatus() == SegmentStatus.SUCCESS
- || segment.getSegmentStatus() ==
SegmentStatus.LOAD_PARTIAL_SUCCESS) {
- // when there is no update operations done on table, then no need to
go ahead. So
- // just check the update delta start timestamp and proceed if not empty
- if (!segment.getUpdateDeltaStartTimestamp().isEmpty()
- || updatedSegments.contains(segment.getLoadName())) {
- // take the list of files from this segment.
- String segmentPath = CarbonTablePath.getSegmentPath(
- table.getAbsoluteTableIdentifier().getTablePath(),
segment.getLoadName());
- CarbonFile segDir =
- FileFactory.getCarbonFile(segmentPath);
- CarbonFile[] allSegmentFiles = segDir.listFiles();
-
- // now handle all the delete delta files which needs to be deleted.
- // there are 2 cases here .
- // 1. if the block is marked as compacted then the corresponding
delta files
- // can be deleted if query exec timeout is done.
- // 2. if the block is in success state then also there can be delete
- // delta compaction happened and old files can be deleted.
-
- SegmentUpdateDetails[] updateDetails =
updateStatusManager.readLoadMetadata();
- for (SegmentUpdateDetails block : updateDetails) {
- CarbonFile[] completeListOfDeleteDeltaFiles;
- CarbonFile[] invalidDeleteDeltaFiles;
-
- if
(!block.getSegmentName().equalsIgnoreCase(segment.getLoadName())) {
- continue;
- }
-
- // aborted scenario.
- invalidDeleteDeltaFiles = updateStatusManager
- .getDeleteDeltaInvalidFilesList(block, false,
- allSegmentFiles, isAbortedFile);
- for (CarbonFile invalidFile : invalidDeleteDeltaFiles) {
- boolean doForceDelete = true;
- compareTimestampsAndDelete(invalidFile, doForceDelete, false);
- }
-
- // case 1
- if (CarbonUpdateUtil.isBlockInvalid(block.getSegmentStatus())) {
- completeListOfDeleteDeltaFiles = updateStatusManager
- .getDeleteDeltaInvalidFilesList(block, true,
- allSegmentFiles, isInvalidFile);
- for (CarbonFile invalidFile : completeListOfDeleteDeltaFiles) {
- compareTimestampsAndDelete(invalidFile, forceDelete, false);
- }
-
- } else {
- invalidDeleteDeltaFiles = updateStatusManager
- .getDeleteDeltaInvalidFilesList(block, false,
- allSegmentFiles, isInvalidFile);
- for (CarbonFile invalidFile : invalidDeleteDeltaFiles) {
- compareTimestampsAndDelete(invalidFile, forceDelete, false);
- }
- }
- }
- }
- // handle cleanup of merge index files and data files after small
files merge happened for
- // SI table
- cleanUpDataFilesAfterSmallFilesMergeForSI(table, segment);
- }
- }
-
- // delete the update table status files which are old.
- if (null != validUpdateStatusFile && !validUpdateStatusFile.isEmpty()) {
-
- final String updateStatusTimestamp = validUpdateStatusFile
-
.substring(validUpdateStatusFile.lastIndexOf(CarbonCommonConstants.HYPHEN) + 1);
-
- String tablePath = table.getAbsoluteTableIdentifier().getTablePath();
- CarbonFile metaFolder = FileFactory.getCarbonFile(
- CarbonTablePath.getMetadataPath(tablePath));
-
- CarbonFile[] invalidUpdateStatusFiles = metaFolder.listFiles(new
CarbonFileFilter() {
- @Override
- public boolean accept(CarbonFile file) {
- if
(file.getName().startsWith(CarbonCommonConstants.TABLEUPDATESTATUS_FILENAME)) {
- // CHECK if this is valid or not.
- // we only send invalid ones to delete.
- return !file.getName().endsWith(updateStatusTimestamp);
- }
- return false;
- }
- });
-
- for (CarbonFile invalidFile : invalidUpdateStatusFiles) {
- compareTimestampsAndDelete(invalidFile, forceDelete, true);
- }
- }
- }
-
- /**
- * this is the clean up added specifically for SI table, because after we
merge the data files
- * inside the secondary index table, we need to delete the stale carbondata
files.
- * refer org.apache.spark.sql.secondaryindex.rdd.CarbonSIRebuildRDD
- */
- private static void cleanUpDataFilesAfterSmallFilesMergeForSI(CarbonTable
table,
Review comment:
done
[Should clean stale data in success
segments](https://issues.apache.org/jira/browse/CARBONDATA-4074)
cleaning stale data in success segments include the following parts.
1. clean stale delete delta (when force is true)
2. clean stale small files for index table
3. clean stale data files for loading/compaction
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
