Scenario:

the number of data rows does not change. The data of each column is increasing and changing. At this scenario, it is faster load full data each time using load command. In this case, the query only needs to query the latest segment Need a way to control the table do like this.
apache · Aug 7, 2021 · c7d4143 · c7d4143
1 parent 1e2fc4c
commit c7d4143
Show file tree

Hide file tree

Showing 6 changed files with 498 additions and 14 deletions.
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -484,6 +484,9 @@ private CarbonCommonConstants() {
   // default blocklet size value in MB
   public static final String TABLE_BLOCKLET_SIZE_DEFAULT = "64";
 
+  // does query with latest segment
+  public static final String TABLE_QUERY_LATEST_SEGMENT = "query_latest_segment";
+
   /**
    * set in column level to disable inverted index
    * @Deprecated :This property is deprecated, it is kept just for compatibility

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
@@ -18,13 +18,7 @@
 package org.apache.carbondata.hadoop.api;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
@@ -222,7 +216,9 @@ public List<InputSplit> getSplits(JobContext job) throws IOException {
    */
   private List<Segment> getFilteredSegment(JobContext job, List<Segment> validSegments,
       boolean validationRequired, ReadCommittedScope readCommittedScope) throws IOException {
-    Segment[] segmentsToAccess = getSegmentsToAccess(job, readCommittedScope);
+    // first check for mapreduce.input.carboninputformat.segmentnumbers
+    // second check for table property of latest_segment for query
+    Segment[] segmentsToAccess = getSegmentsToAccess(job, readCommittedScope, validSegments);
     if (segmentsToAccess.length == 0 || segmentsToAccess[0].getSegmentNo().equalsIgnoreCase("*")) {
       return validSegments;
     }
@@ -421,9 +417,11 @@ public void updateLoadMetaDataDetailsToSegments(List<Segment> validSegments,
 
   /**
    * return valid segment to access
+   * check for SET carbon.input.segments.<database_name>.<table_name>
    */
   public Segment[] getSegmentsToAccess(JobContext job, ReadCommittedScope readCommittedScope) {
     String segmentString = job.getConfiguration().get(INPUT_SEGMENT_NUMBERS, "");
+
     if (segmentString.trim().isEmpty()) {
       return new Segment[0];
     }
@@ -601,4 +599,60 @@ public String getSegmentIdFromFilePath(String filePath) {
     }
     return CarbonCommonConstants.INVALID_SEGMENT_ID;
   }
+
+  /**
+   * return valid segment to access
+   * first check for mapreduce.input.carboninputformat.segmentnumbers"
+   * second check for table property of latest_segment for query
+   */
+  public Segment[] getSegmentsToAccess(JobContext job, ReadCommittedScope readCommittedScope,
+                                       List<Segment> validSegments) {
+    String segmentString = job.getConfiguration().get(INPUT_SEGMENT_NUMBERS, "");
+    boolean queryLatestSegment = false;
+    if (null != carbonTable) {
+       queryLatestSegment = Boolean.parseBoolean(carbonTable.getTableInfo()
+              .getFactTable().getTableProperties()
+              .getOrDefault(CarbonCommonConstants.TABLE_QUERY_LATEST_SEGMENT, "false"));
+    }
+    if (segmentString.trim().isEmpty()) {
+      if (!queryLatestSegment) {
+        return new Segment[0];
+      } else {
+        List<Segment> segments = getLatestSegment(validSegments);
+        return  segments.toArray(new Segment[0]);
+      }
+    } else {
+      List<Segment> segments = Segment.toSegmentList(segmentString.split(","), readCommittedScope);
+      if (!queryLatestSegment) {
+        return segments.toArray(new Segment[0]);
+      } else {
+        List<Segment> latestSegment;
+        if (segments.size() > 0 && segments.get(0).getSegmentNo().equalsIgnoreCase("*")) {
+          latestSegment = getLatestSegment(validSegments);
+        } else {
+          latestSegment = getLatestSegment(segments);
+        }
+        return latestSegment.toArray(new Segment[0]);
+      }
+    }
+  }
+
+  /**
+   * get the latest segment
+   * @param validSegments the in put segment for search
+   * @return the latest segment for query
+   */
+  public List<Segment> getLatestSegment(List<Segment> validSegments) {
+    if (validSegments.isEmpty()) {
+      return Collections.emptyList();
+    } else {
+      Segment segment = validSegments.stream().max((a, b) -> {
+        double aNo = Double.parseDouble(a.getSegmentNo());
+        double bNo = Double.parseDouble(b.getSegmentNo());
+        return Double.compare(aNo, bNo);
+      }).get();
+      return Collections.singletonList(segment);
+    }
+  }
+
 }
diff --git a/.../src/main/scala/org/apache/spark/sql/secondaryindex/joins/BroadCastSIFilterPushJoin.scala b/.../src/main/scala/org/apache/spark/sql/secondaryindex/joins/BroadCastSIFilterPushJoin.scala
@@ -273,17 +273,19 @@ object BroadCastSIFilterPushJoin {
     val identifier: AbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
     val readCommittedScope: ReadCommittedScope = carbonTableInputFormat.getReadCommitted(job,
       identifier)
-    val segmentsToAccess: Array[Segment] = carbonTableInputFormat.getSegmentsToAccess(job,
-      readCommittedScope)
-    val segmentsToAccessSet: util.Set[Segment] = new util.HashSet[Segment]
-    for (segId <- segmentsToAccess) {
-      segmentsToAccessSet.add(segId)
-    }
     // get all valid segments and set them into the configuration
     val segmentStatusManager: SegmentStatusManager = new SegmentStatusManager(identifier)
     val segments: SegmentStatusManager.ValidAndInvalidSegmentsInfo = segmentStatusManager
       .getValidAndInvalidSegments(carbonTable.isMV)
     val validSegments: util.List[Segment] = segments.getValidSegments
+
+    val segmentsToAccess: Array[Segment] = carbonTableInputFormat.getSegmentsToAccess (job,
+      readCommittedScope, validSegments)
+    val segmentsToAccessSet: util.Set[Segment] = new util.HashSet[Segment]
+    for (segId <- segmentsToAccess) {
+      segmentsToAccessSet.add(segId)
+    }
+
     // if no segments in table
     val validSegmentsToAccess: util.List[Segment] = new util.ArrayList[Segment]
     if (validSegments.size == 0) {

diff --git a/integration/spark/src/test/resources/latest-table-data.csv b/integration/spark/src/test/resources/latest-table-data.csv
@@ -0,0 +1,11 @@
+shortField,intField,bigintField,doubleField,stringField,timestampField,decimalField,dateField,charField,floatField,complexData
+1,10,1100,48.4,,,1.23,2015/4/23,aaa,2.5,'foo'#'bar'#'world'
+5,17,1140,43.4,,2015/7/27 12:01:02,3.45,2015/7/27,bbb,2.5,'foo'#'bar'#'world'
+1,11,1100,44.4,,2015/5/23 12:01:03,23.23,2015/5/23,ccc,2.5,'foo'#'bar'#'world'
+1,10,1150,43.4,spark,,254.12,2015/7/24,ddd,2.5,'foo'#'bar'#'world'
+1,10,1100,47.4,spark,2015/7/23 12:01:05,876.14,2015/7/23,eeee,3.5,'foo'#'bar'#'world'
+3,14,1160,43.4,hive,2015/7/26 12:01:06,3454.32,2015/7/26,ff,2.5,'foo'#'bar'#'world'
+2,,,43.4,impala,2015/7/23 12:01:07,456.98,2015/7/23,ggg,2.5,'foo'#'bar'#'world'
+,10,1100,43.4,spark,,32.53,2015/5/23,hhh,2.5,'foo'#'bar'#'world'
+4,16,1130,42.4,impala,2015/7/23 12:01:09,67.23,2015/7/23,iii,2.5,
+1,10,1100,43.4,spark,2015/7/23 12:01:10,832.23,2015/7/23,jjj,,'foo'#'bar'#'world'
diff --git a/integration/spark/src/test/resources/latest-table-data1.csv b/integration/spark/src/test/resources/latest-table-data1.csv
@@ -0,0 +1,11 @@
+shortField,intField,bigintField,doubleField,stringField,timestampField,decimalField,dateField,charField,floatField,complexData
+1,10,1100,48.4,spark,2015/4/23 12:01:01,1.23,2015/4/23,aaa,2.5,'foo'#'bar'#'world'
+5,17,1140,43.4,spark,2015/7/27 12:01:02,3.45,2015/7/27,bbb,2.5,'foo'#'bar'#'world'
+1,11,1100,44.4,flink,2015/5/23 12:01:03,23.23,2015/5/23,ccc,2.5,'foo'#'bar'#'world'
+1,10,1150,43.4,spark,2015/7/24 12:01:04,254.12,2015/7/24,ddd,2.5,'foo'#'bar'#'world'
+1,10,1100,47.4,spark,2015/7/23 12:01:05,876.14,2015/7/23,eeee,3.5,'foo'#'bar'#'world'
+3,14,1160,43.4,hive,2015/7/26 12:01:06,3454.32,2015/7/26,ff,2.5,'foo'#'bar'#'world'
+2,10,1100,43.4,impala,2015/7/23 12:01:07,456.98,2015/7/23,ggg,2.5,'foo'#'bar'#'world'
+1,10,1100,43.4,spark,2015/5/23 12:01:08,32.53,2015/5/23,hhh,2.5,'foo'#'bar'#'world'
+4,16,1130,42.4,impala,2015/7/23 12:01:09,67.23,2015/7/23,iii,2.5,'foo'#'bar'#'world'
+1,10,1100,43.4,spark,2015/7/23 12:01:10,832.23,2015/7/23,jjj,2.5,'foo'#'bar'#'world'