This is an automated email from the ASF dual-hosted git repository. kunalkapoor pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push: new f22f623 [HOTFIX] Avoid cleaning segments after reading tablestatus failed f22f623 is described below commit f22f6236ca3329558d419dc4575e94ca16fe6fe3 Author: QiangCai <qiang...@qq.com> AuthorDate: Tue Aug 25 21:11:39 2020 +0800 [HOTFIX] Avoid cleaning segments after reading tablestatus failed Why is this PR needed? 1. After reading tablestatus file failed, method TableProcessingOperations. deletePartialLoadDataIfExist will delete all related segments. 2. If the tablestatus file was removed, the system will delete all original segment before loading. What changes were proposed in this PR? 1. check the result of reading tablestatus file, if it is empty, no need to process. 2. if the table status was removed, the system shouldn't delete any segment. 3. re-factory code to avoid invoking getAbsolutePath method too many times. This closes #3899 --- .../carbondata/core/util/path/CarbonTablePath.java | 2 +- .../testsuite/dataload/TestLoadDataGeneral.scala | 15 ++-- .../loading/TableProcessingOperations.java | 79 +++++++++++++--------- 3 files changed, 57 insertions(+), 39 deletions(-) diff --git a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java index da7403b..7482a1c 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 @@ -41,7 +41,7 @@ public class CarbonTablePath { private static final String DICTIONARY_EXT = ".dict"; public static final String SCHEMA_FILE = "schema"; private static final String FACT_DIR = "Fact"; - private static final String SEGMENT_PREFIX = "Segment_"; + public static final String SEGMENT_PREFIX = "Segment_"; private static final String PARTITION_PREFIX = "Part"; private static final String DATA_PART_PREFIX = "part-"; public static final String BATCH_PREFIX = "_batchno"; diff --git a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala index 6ae9e6b..aaaa88d 100644 --- a/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala +++ b/integration/spark/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataGeneral.scala @@ -19,25 +19,25 @@ package org.apache.carbondata.integration.spark.testsuite.dataload import java.math.BigDecimal +import org.apache.commons.io.FileUtils +import org.apache.commons.lang3.RandomStringUtils import org.apache.spark.sql.Row import org.apache.spark.sql.test.util.QueryTest import org.scalatest.BeforeAndAfterEach -import org.apache.carbondata.core.util.path.CarbonTablePath -import org.apache.carbondata.core.datastore.impl.FileFactory -import org.apache.carbondata.core.metadata.CarbonMetadata import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants} +import org.apache.carbondata.core.datastore.impl.FileFactory import org.apache.carbondata.core.index.Segment +import org.apache.carbondata.core.metadata.CarbonMetadata import org.apache.carbondata.core.util.CarbonProperties +import org.apache.carbondata.core.util.path.CarbonTablePath import org.apache.carbondata.spark.util.BadRecordUtil -import org.apache.commons.io.FileUtils -import org.apache.commons.lang3.RandomStringUtils class TestLoadDataGeneral extends QueryTest with BeforeAndAfterEach { val badRecordAction = CarbonProperties.getInstance() .getProperty(CarbonCommonConstants.CARBON_BAD_RECORDS_ACTION); - val testdata =s"$resourcesPath/MoreThan32KChar.csv" + val testdata = s"$resourcesPath/MoreThan32KChar.csv" val longChar: String = RandomStringUtils.randomAlphabetic(33000) override def beforeEach { @@ -310,7 +310,8 @@ class TestLoadDataGeneral extends QueryTest with BeforeAndAfterEach { val tableStatusFile = CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath) FileFactory.getCarbonFile(tableStatusFile).delete() sql("insert into stale values('k')") - checkAnswer(sql("select * from stale"), Row("k")) + // if table lose tablestatus file, the system should keep all data. + checkAnswer(sql("select * from stale"), Seq(Row("k"), Row("k"))) } test("test data loading with directly writing fact data to hdfs") { 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 f0f14d6..4aba751 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 @@ -19,13 +19,17 @@ package org.apache.carbondata.processing.loading; import java.io.File; import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; 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.filesystem.CarbonFileFilter; import org.apache.carbondata.core.datastore.impl.FileFactory; import org.apache.carbondata.core.metadata.schema.table.CarbonTable; import org.apache.carbondata.core.statusmanager.LoadMetadataDetails; @@ -46,45 +50,58 @@ public class TableProcessingOperations { LogServiceFactory.getLogService(CarbonLoaderUtil.class.getName()); /** - * - * @param carbonTable - * @param isCompactionFlow - * @throws IOException + * delete folder which metadata no exist in tablestatus + * this method don't check tablestatus history. */ public static void deletePartialLoadDataIfExist(CarbonTable carbonTable, final boolean isCompactionFlow) throws IOException { String metaDataLocation = carbonTable.getMetadataPath(); - //delete folder which metadata no exist in tablestatus String partitionPath = CarbonTablePath.getPartitionDir(carbonTable.getTablePath()); if (FileFactory.isFileExist(partitionPath)) { - final LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(metaDataLocation); - CarbonFile carbonFile = FileFactory.getCarbonFile(partitionPath); - CarbonFile[] listFiles = carbonFile.listFiles(new CarbonFileFilter() { - @Override - public boolean accept(CarbonFile path) { - String segmentId = - CarbonTablePath.DataFileUtil.getSegmentIdFromPath(path.getAbsolutePath() + "/dummy"); - boolean found = false; - for (int j = 0; j < details.length; j++) { - if (details[j].getLoadName().equals(segmentId)) { - found = true; - break; + // list all segments before reading tablestatus file. + CarbonFile[] allSegments = FileFactory.getCarbonFile(partitionPath).listFiles(); + // there is no segment + if (allSegments == null || allSegments.length == 0) { + return; + } + LoadMetadataDetails[] details = SegmentStatusManager.readLoadMetadata(metaDataLocation); + // there is no segment or failed to read tablestatus file. + // so it should stop immediately. + if (details == null || details.length == 0) { + return; + } + Set<String> metadataSet = new HashSet<>(details.length); + for (LoadMetadataDetails detail : details) { + metadataSet.add(detail.getLoadName()); + } + List<CarbonFile> staleSegments = new ArrayList<>(allSegments.length); + for (CarbonFile segment : allSegments) { + String segmentName = segment.getName(); + // check segment folder pattern + if (segmentName.startsWith(CarbonTablePath.SEGMENT_PREFIX)) { + String[] parts = segmentName.split(CarbonCommonConstants.UNDERSCORE); + if (parts.length == 2) { + boolean isOriginal = !parts[1].contains("."); + if (isCompactionFlow) { + // in compaction flow, it should be big segment and segment metadata is not exists + if (!isOriginal && !metadataSet.contains(parts[1])) { + staleSegments.add(segment); + } + } else { + // in loading flow, it should be original segment and segment metadata is not exists + if (isOriginal && !metadataSet.contains(parts[1])) { + staleSegments.add(segment); + } } } - return !found; } - }); - for (int k = 0; k < listFiles.length; k++) { - String segmentId = CarbonTablePath.DataFileUtil - .getSegmentIdFromPath(listFiles[k].getAbsolutePath() + "/dummy"); - if (isCompactionFlow) { - if (segmentId.contains(".")) { - CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath()); - } - } else { - if (!segmentId.contains(".")) { - CarbonLoaderUtil.deleteStorePath(listFiles[k].getAbsolutePath()); - } + } + // delete segment one by one + for (CarbonFile staleSegment : staleSegments) { + try { + CarbonUtil.deleteFoldersAndFiles(staleSegment); + } catch (IOException | InterruptedException e) { + LOGGER.error("Unable to delete the given path :: " + e.getMessage(), e); } } }