[CARBONDATA-4263]support query with latestSegment

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

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);
+    }
+  }
+
 }

integration/spark/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) {

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'

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'