[CARBONDATA-3911]Fix NullPointerException in case of multiple updates and clean files
Why is this PR needed?
when multiple load and multiple updates are performed, the segment files will be created
for each update, when clean files performed, if two segments are there, first segment will
clean other segments's segment file also which leads to nullpointer.
What changes were proposed in this PR?
Once all the segments stale segment files are collected, then delete the stale files at once.
This closes #3843
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 791c422..4610377 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
@@ -610,6 +610,7 @@
List<Segment> segmentFilesToBeUpdatedLatest = new ArrayList<>();
CarbonFile segmentFilesLocation =
FileFactory.getCarbonFile(CarbonTablePath.getSegmentFilesLocation(table.getTablePath()));
+ Set<String> segmentFilesNotToDelete = new HashSet<>();
for (Segment segment : segmentFilesToBeUpdated) {
SegmentFileStore fileStore =
new SegmentFileStore(table.getTablePath(), segment.getSegmentFileName());
@@ -618,13 +619,15 @@
.writeSegmentFile(table, segment.getSegmentNo(), UUID,
CarbonTablePath.getSegmentPath(table.getTablePath(), segment.getSegmentNo()),
segment.getSegmentMetaDataInfo());
+ segmentFilesNotToDelete.add(updatedSegmentFile);
segmentFilesToBeUpdatedLatest.add(new Segment(segment.getSegmentNo(), updatedSegmentFile));
-
+ }
+ if (segmentFilesNotToDelete.size() > 0) {
// delete the old segment files
CarbonFile[] invalidSegmentFiles = segmentFilesLocation.listFiles(new CarbonFileFilter() {
@Override
public boolean accept(CarbonFile file) {
- return !file.getName().equalsIgnoreCase(updatedSegmentFile);
+ return !segmentFilesNotToDelete.contains(file.getName());
}
});
for (CarbonFile invalidSegmentFile : invalidSegmentFiles) {
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
index 2f455af..ec89570 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
@@ -71,6 +71,21 @@
sql("""drop table iud.zerorows""")
}
+ test("test update operation with multiple loads and clean files operation") {
+ sql("""drop table if exists iud.zerorows""").show
+ sql("""create table iud.zerorows (c1 string,c2 int,c3 string,c5 string) STORED AS carbondata""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.zerorows""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/dest.csv' INTO table iud.zerorows""")
+ sql("""update zerorows d set (d.c2) = (d.c2 + 1) where d.c1 = 'a'""").show()
+ sql("""update zerorows d set (d.c2) = (d.c2 + 1) where d.c1 = 'b'""").show()
+ sql("clean files for table iud.zerorows")
+ checkAnswer(
+ sql("""select c1,c2,c3,c5 from iud.zerorows"""),
+ Seq(Row("a",2,"aa","aaa"),Row("b",3,"bb","bbb"),Row("c",3,"cc","ccc"),Row("d",4,"dd","ddd"),Row("e",5,"ee","eee"),Row("a",2,"aa","aaa"),Row("b",3,"bb","bbb"),Row("c",3,"cc","ccc"),Row("d",4,"dd","ddd"),Row("e",5,"ee","eee"))
+ )
+ sql("""drop table iud.zerorows""")
+ }
+
test("update carbon table[select from source table with where and exist]") {
sql("""drop table if exists iud.dest11""").show
