GEOMESA-3580 FSDS - Use optimized sorting (#3547)

Author: elahrvivaz

geomesa-fs/geomesa-fs-datastore/src/main/scala/org/locationtech/geomesa/fs/data/FileSystemFeatureStore.scala

@@ -81,6 +81,7 @@ class FileSystemFeatureStore(
         val filter = Option(query.getFilter).filter(_ != Filter.INCLUDE)
         new ManagedScan(new FileSystemScanner(storage, query, readThreads), Timeout(timeout), query.getTypeName, filter)
     }
+
     // get a closeable java iterator that DelegateSimpleFeatureIterator will process correctly
     val iter = new FileSystemFeatureIterator(reader)
 
@@ -91,11 +92,11 @@ class FileSystemFeatureStore(
     new DelegateSimpleFeatureReader(transformSft, new DelegateSimpleFeatureIterator(iter))
   }
 
-  override def canLimit(query: Query): Boolean = false
   override def canTransact: Boolean = false
   override def canEvent: Boolean = false
   override def canReproject: Boolean = false
-  override def canSort(query: Query): Boolean = false
+  override def canLimit(query: Query): Boolean = true
+  override def canSort(query: Query): Boolean = true
   override def canFilter(query: Query): Boolean = true
   override def canRetype(query: Query): Boolean = true
 

geomesa-fs/geomesa-fs-datastore/src/test/scala/org/locationtech/geomesa/fs/data/FileSystemDataStoreTest.scala

@@ -13,9 +13,11 @@ import org.apache.commons.io.FileUtils
 import org.geotools.api.data.{DataStoreFinder, Query, Transaction}
 import org.geotools.api.feature.simple.{SimpleFeature, SimpleFeatureType}
 import org.geotools.api.filter.Filter
+import org.geotools.api.filter.sort.SortOrder
 import org.geotools.filter.text.ecql.ECQL
 import org.geotools.geometry.jts.ReferencedEnvelope
 import org.locationtech.geomesa.features.ScalaSimpleFeature
+import org.locationtech.geomesa.filter.FilterHelper
 import org.locationtech.geomesa.fs.storage.core.StorageKeys
 import org.locationtech.geomesa.utils.collection.CloseableIterator
 import org.locationtech.geomesa.utils.geotools.{CRS_EPSG_4326, FeatureUtils, SimpleFeatureTypes}
@@ -291,6 +293,25 @@ class FileSystemDataStoreTest extends SpecificationWithJUnit with BeforeAfterAll
       }
     }
 
+    "support sorting and limiting" in {
+      foreach(dsParams) { params =>
+        WithClose(DataStoreFinder.getDataStore(params.asJava)) { ds =>
+          foreach(Seq(SortOrder.ASCENDING, SortOrder.DESCENDING)) { sortOrder =>
+            val query = new Query(sft.getTypeName)
+            query.setSortBy(FilterHelper.ff.sort("name", sortOrder))
+            query.setMaxFeatures(1)
+            val results = CloseableIterator(ds.getFeatureReader(query, Transaction.AUTO_COMMIT)).toList
+            results must haveSize(1)
+            if (sortOrder == SortOrder.ASCENDING) {
+              results.head.getID mustEqual "0"
+            } else {
+              results.head.getID mustEqual "9"
+            }
+          }
+        }
+      }
+    }
+
     "support append without fid" in {
       foreach(dsParams) { params =>
         val dir = Files.createTempDirectory("fsds-test-append-without-fid").toFile

geomesa-fs/geomesa-fs-storage/geomesa-fs-storage-core/src/main/scala/org/locationtech/geomesa/fs/storage/core/FileSystemStorage.scala

@@ -26,6 +26,7 @@ import org.locationtech.geomesa.fs.storage.core.utils.FileSize.UpdatingFileSizeE
 import org.locationtech.geomesa.fs.storage.core.utils.{FileSize, FileSystemThreadedReader}
 import org.locationtech.geomesa.index.index.attribute.AttributeIndexKey
 import org.locationtech.geomesa.index.planning.QueryRunner
+import org.locationtech.geomesa.index.utils.SortingSimpleFeatureIterator
 import org.locationtech.geomesa.utils.collection.CloseableIterator
 import org.locationtech.geomesa.utils.io.{CloseQuietly, CloseWithLogging, FlushQuietly, WithClose}
 import org.locationtech.jts.geom.{Envelope, Geometry}
@@ -87,15 +88,17 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
   def getReader(query: Query, threads: Int = 1): CloseableFeatureIterator = {
     import org.locationtech.geomesa.index.conf.QueryHints.RichHints
 
-    val (filter, transform) = {
-      val configured = QueryRunner.configureQuery(metadata.sft, query)
-      val filter = Option(configured.getFilter).getOrElse(Filter.INCLUDE)
-      val transform = configured.getHints.getTransform
-      logger.debug(s"Running query '${query.getTypeName}' ${ECQL.toCQL(filter)}")
-      logger.debug(s"  Original filter: ${ECQL.toCQL(query.getFilter)}")
-      logger.debug(s"  Transforms: " + transform.fold("none") { case (t, _) => if (t.isEmpty) { "empty" } else { t }})
-      (filter, transform)
-    }
+    val configured = QueryRunner.configureQuery(metadata.sft, query)
+    val filter = Option(configured.getFilter).getOrElse(Filter.INCLUDE)
+    val transform = configured.getHints.getTransform
+    val sort = configured.getHints.getSortFields
+    val max = configured.getHints.getMaxFeatures
+
+    logger.debug(s"Running query '${query.getTypeName}' ${ECQL.toCQL(filter)}")
+    logger.debug(s"  Original filter: ${ECQL.toCQL(query.getFilter)}")
+    logger.debug(s"  Transforms: ${transform.fold("none") { case (t, _) => if (t.isEmpty) { "empty" } else { t }}}")
+    logger.debug(s"  Sort: ${sort.fold("none") { fields => fields.map { case (f, rev) => s"$f ${if (rev) "descending" else ""}"}.mkString(", ")}}")
+    logger.debug(s"  Max features: ${max.getOrElse("none")}")
 
     val files = metadata.getFiles(filter)
     logger.debug(s"  Threading the read of ${files.size} files with $threads reader threads")
@@ -105,7 +108,10 @@ abstract class FileSystemStorage(val context: FileSystemContext, val metadata: S
       CloseableIterator.empty
     } else {
       val reader = createReader(Option(filter).filterNot(_ == Filter.INCLUDE), transform)
-      FileSystemThreadedReader(reader, files, threads)
+      val threaded = FileSystemThreadedReader(reader, files, threads)
+      val sorted = sort.fold(threaded)(s => new SortingSimpleFeatureIterator(threaded, s))
+      val limited = max.fold(sorted)(m => sorted.take(m))
+      limited
     }
   }