carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kunalkap...@apache.org
Subject [carbondata] branch master updated: [CARBONDATA-4080] Wrong results for select count on invalid segments
Date Thu, 17 Dec 2020 14:28:03 GMT
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 35c73e6  [CARBONDATA-4080] Wrong results for select count on invalid segments
35c73e6 is described below

commit 35c73e61c55f39f1bf38105e6bf32ff3673396fb
Author: akkio-97 <akshay.nuthala@gmail.com>
AuthorDate: Wed Dec 9 15:15:51 2020 +0530

    [CARBONDATA-4080] Wrong results for select count on invalid segments
    
    Why is this PR needed?
    Wrong results for
    1. select count on marked for delete segment
    2. select count on compacted segment
    Issue comes only when the user explicitly sets deleted/compacted
    segments using the property carbon.input.segments.
    As select * on such segments gives 0 rows as output, in order to
    maintain consistency, select count should also give 0 rows.
    
    What changes were proposed in this PR?
    So this PR focuses on considering only valid segments to be accessed
    during select count query.
    
    This closes #4050
---
 .../hadoop/api/CarbonTableInputFormat.java         |  15 ++-
 .../segmentreading/TestSegmentReading.scala        | 104 +++++++++++++++++++++
 2 files changed, 117 insertions(+), 2 deletions(-)

diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
index a2c162e..f24a2f1 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
@@ -55,6 +55,7 @@ import org.apache.carbondata.core.readcommitter.TableStatusReadCommittedScope;
 import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
 import org.apache.carbondata.core.statusmanager.FileFormat;
 import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
+import org.apache.carbondata.core.statusmanager.SegmentStatus;
 import org.apache.carbondata.core.statusmanager.SegmentStatusManager;
 import org.apache.carbondata.core.statusmanager.SegmentUpdateStatusManager;
 import org.apache.carbondata.core.statusmanager.StageInputCollector;
@@ -220,7 +221,7 @@ public class CarbonTableInputFormat<T> extends CarbonInputFormat<T>
{
    * `INPUT_SEGMENT_NUMBERS` in job configuration
    */
   private List<Segment> getFilteredSegment(JobContext job, List<Segment> validSegments,
-      boolean validationRequired, ReadCommittedScope readCommittedScope) {
+      boolean validationRequired, ReadCommittedScope readCommittedScope) throws IOException
{
     Segment[] segmentsToAccess = getSegmentsToAccess(job, readCommittedScope);
     if (segmentsToAccess.length == 0 || segmentsToAccess[0].getSegmentNo().equalsIgnoreCase("*"))
{
       return validSegments;
@@ -241,7 +242,17 @@ public class CarbonTableInputFormat<T> extends CarbonInputFormat<T>
{
     }
     if (!validationRequired && filteredSegmentToAccess.size() != segmentToAccessMap.size())
{
       for (Segment segment : segmentToAccessMap.values()) {
-        if (!filteredSegmentToAccess.containsKey(segment.getSegmentNo())) {
+        boolean isSegmentValid = true;
+        LoadMetadataDetails[] segmentList = readCommittedScope.getSegmentList();
+        for (LoadMetadataDetails validSegment : segmentList) {
+          if (validSegment.getLoadName().equals(segment.getSegmentNo()) && (
+              validSegment.getSegmentStatus().equals(SegmentStatus.MARKED_FOR_DELETE)
+                  || validSegment.getSegmentStatus().equals(SegmentStatus.COMPACTED))) {
+            isSegmentValid = false;
+            break;
+          }
+        }
+        if (isSegmentValid && !filteredSegmentToAccess.containsKey(segment.getSegmentNo()))
{
           filteredSegmentToAccess.put(segment.getSegmentNo(), segment);
         }
       }
diff --git a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/segmentreading/TestSegmentReading.scala
b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/segmentreading/TestSegmentReading.scala
index 8ce8cc5..5c0336c 100644
--- a/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/segmentreading/TestSegmentReading.scala
+++ b/integration/spark/src/test/scala/org/apache/carbondata/spark/testsuite/segmentreading/TestSegmentReading.scala
@@ -22,6 +22,8 @@ import org.apache.spark.sql.test.util.QueryTest
 import org.scalatest.BeforeAndAfterAll
 
 import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.view.rewrite.TestUtil
 
 /**
  * Created by rahul on 19/9/17.
@@ -420,4 +422,106 @@ class TestSegmentReading extends QueryTest with BeforeAndAfterAll {
 
     sql("set spark.sql.adaptive.enabled=false")
   }
+
+  test("Read marked for delete segments") {
+    sql("drop table if exists carbonTable")
+    sql(
+      "create table carbonTable(a string, b int, c string) stored as carbondata ")
+    sql("insert into carbonTable values ('k',1,'k'), ('k',1,'b')")
+    sql("insert into carbonTable values ('a',2,'a')")
+    sql("insert into carbonTable values ('b',2,'b'),('b',2,'b')")
+    sql("insert into carbonTable values ('c',2,'c')")
+
+    sql("delete from table carbonTable where segment.id in (0,3)")
+    sql("set carbon.input.segments.default.carbonTable = 0,2,3")
+
+    checkAnswer(sql("select count(*) from carbonTable"), Seq(Row(2)))
+  }
+
+  test("Read marked for delete segments after SI creation") {
+    sql("drop table if exists carbonTable")
+    sql(
+      "create table carbonTable(a string, b int, c string) stored as carbondata ")
+    sql("insert into carbonTable values ('k',1,'k'), ('k',1,'b')")
+    sql("insert into carbonTable values ('a',2,'a')")
+    sql("insert into carbonTable values ('b',2,'b'),('b',2,'b')")
+    sql("insert into carbonTable values ('c',2,'c')")
+
+    sql("drop index if exists indextable1 on carbonTable")
+    sql("create index indextable1 on table carbonTable (c) AS 'carbondata'")
+
+    sql("delete from table carbonTable where segment.id in (0,3)")
+    sql("set carbon.input.segments.default.carbonTable = 0,2,3")
+
+    checkAnswer(sql("select count(*) from carbonTable"), Seq(Row(2)))
+  }
+
+  test("Read compacted segments") {
+    sql("drop table if exists carbonTable")
+    sql(
+      "create table carbonTable(a string, b int, c string) stored as carbondata")
+    sql("insert into carbonTable values ('k',5,'k'), ('k',5,'b')")
+    sql("insert into carbonTable values ('a',1,'a')")
+    sql("insert into carbonTable values ('b',2,'b'),('b',2,'b')")
+    sql("insert into carbonTable values ('c',3,'c')")
+    sql("alter table carbonTable compact 'major'")
+
+    sql("set carbon.input.segments.default.carbonTable = 0,1,2,3,0.1")
+    checkAnswer(sql("select count(*) from carbonTable"), Seq(Row(6)))
+  }
+
+  test("Read compacted segments after SI creation") {
+    sql("drop table if exists carbonTable")
+    sql(
+      "create table carbonTable(a string, b int, c string) stored as carbondata")
+    sql("insert into carbonTable values ('k',5,'k'), ('k',5,'b')")
+    sql("insert into carbonTable values ('a',1,'a')")
+    sql("insert into carbonTable values ('b',2,'b'),('b',2,'b')")
+    sql("insert into carbonTable values ('c',3,'c')")
+
+    sql("drop index if exists indextable1 on carbonTable")
+    sql("create index indextable1 on table carbonTable (c) AS 'carbondata'")
+
+    sql("alter table carbonTable compact 'major'")
+    sql("set carbon.input.segments.default.carbonTable = 0,1,2,3,0.1")
+    checkAnswer(sql("select count(*) from carbonTable where c = 'b'"), Seq(Row(3)))
+  }
+
+  test("Read marked for delete segments for partition tables") {
+    sql("drop table if exists carbonTable")
+    sql(
+      "create table carbonTable(c1 string, c2 int) PARTITIONED by (c3 string) stored as "
+
+      "carbondata ")
+    sql("insert into carbonTable values ('k',1,'k'), ('k',2,'k')")
+    sql("insert into carbonTable values ('a',2,'a')")
+    sql("insert into carbonTable values ('b',2,'b'),('b',2,'b')")
+    sql("insert into carbonTable values ('c',2,'c')")
+
+    sql("delete from table carbonTable where segment.id in (0,3)")
+    sql("set carbon.input.segments.default.carbonTable = 0,1,2,3")
+
+    checkAnswer(sql("select count(*) from carbonTable where c3 = 'k' or c3 = 'a' or c3 =
'c'"),
+      Seq(Row(1)))
+  }
+
+  test("Read marked for delete segments with MV creation") {
+    sql("drop table if exists carbonTable")
+    sql(" CREATE TABLE carbonTable (id int, name string)  STORED AS carbondata")
+    sql("insert into carbonTable values(1,'a')")
+    sql("insert into carbonTable values(2,'b')")
+    sql("insert into carbonTable values(3,'c')")
+    sql("insert into carbonTable values(4,'d')")
+    sql("delete from table carbonTable where segment.id in (0,3)")
+
+    sql("drop materialized view if exists agg_sale")
+    sql(
+      "CREATE MATERIALIZED VIEW agg_sale AS SELECT id, name, sum(id) FROM carbonTable GROUP
BY " +
+      "id, name")
+
+    sql("set carbon.input.segments.default.carbonTable = 0,1,2,3")
+    checkAnswer(sql("SELECT count(*)  FROM carbonTable where name = 'a' or name = 'd' or
name = " +
+                    "'b'"),
+      Seq(Row(1)))
+  }
+
 }


Mime
View raw message