Fixed bugs in PDS catalog caching.

Author: metasim

experimental/src/it/scala/org/locationtech/rasterframes/experimental/datasource/awspds/L8CatalogRelationTest.scala

@@ -32,7 +32,7 @@ import org.locationtech.rasterframes.datasource.raster._
 class L8CatalogRelationTest extends TestEnvironment {
   import spark.implicits._
 
-  val catalog = spark.read.l8Catalog.load()
+  val catalog = spark.read.l8Catalog.load().cache()
 
   val scenes = catalog
     .where($"acquisition_date" === to_timestamp(lit("2017-04-04 15:12:55.394")))
@@ -104,5 +104,28 @@ class L8CatalogRelationTest extends TestEnvironment {
       stats.data_cells should be (512L * 512L)
       stats.mean shouldBe > (10000.0)
     }
+
+    it("should construct an RGB composite") {
+      val aoi = "LINESTRING (31.115 29.963, 31.148 29.99)"
+      val sceneCat = catalog
+        .where(
+          to_date($"acquisition_date") === to_date(lit("2019-07-03")) &&
+            st_intersects(st_geometry($"bounds_wgs84"), st_geomFromWKT(aoi))
+        )
+
+      catalog.orderBy(desc("acquisition_date")).select($"acquisition_date").show(false)
+      catalog.where(to_date($"acquisition_date") === to_date(lit("2019-03-07"))).show(false)
+
+      //sceneCat.show(false)
+
+
+//      val df = spark.read.raster
+//        .fromCatalog(scenes, "B4", "B3", "B2")
+//        .withTileDimensions(128, 128)
+//        .load()
+//        .where
+
+
+    }
   }
 }

experimental/src/main/scala/org/locationtech/rasterframes/experimental/datasource/CachedDatasetRelation.scala

@@ -25,7 +25,7 @@ import com.typesafe.scalalogging.LazyLogging
 import org.apache.hadoop.fs.{FileSystem, Path => HadoopPath}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.{Dataset, Row}
+import org.apache.spark.sql.{Dataset, Row, SaveMode}
 import org.locationtech.rasterframes.util._
 
 /**
@@ -40,12 +40,12 @@ trait CachedDatasetRelation extends ResourceCacheSupport { self: BaseRelation wi
   def buildScan(): RDD[Row] = {
     val conf = sqlContext.sparkContext.hadoopConfiguration
     implicit val fs: FileSystem = FileSystem.get(conf)
-    val catalog = cacheFile.when(fs.exists)
+    val catalog = cacheFile.when(p => fs.exists(p) && !expired(p))
       .map(p ⇒ {logger.debug("Reading " + p); p})
       .map(p ⇒ sqlContext.read.parquet(p.toString))
       .getOrElse {
         val scenes = constructDataset
-        scenes.write.parquet(cacheFile.toString)
+        scenes.write.mode(SaveMode.Overwrite).parquet(cacheFile.toString)
         scenes
       }
 

experimental/src/main/scala/org/locationtech/rasterframes/experimental/datasource/ResourceCacheSupport.scala

@@ -51,7 +51,7 @@ trait ResourceCacheSupport extends DownloadSupport { self: LazyLogging  ⇒
     else {
 
       val time = fs.getFileStatus(p).getModificationTime
-      val exp = Instant.ofEpochMilli(time).isAfter(Instant.now().plus(Duration.ofHours(maxCacheFileAgeHours)))
+      val exp = Instant.ofEpochMilli(time).plus(Duration.ofHours(maxCacheFileAgeHours)).isBefore(Instant.now())
       if(exp) logger.debug(s"'$p' is expired with mod time of '$time'")
       exp
     }

experimental/src/main/scala/org/locationtech/rasterframes/experimental/datasource/awspds/L8CatalogDataSource.scala

@@ -39,7 +39,7 @@ class L8CatalogDataSource extends DataSourceRegister with RelationProvider {
   def shortName = L8CatalogDataSource.SHORT_NAME
 
   def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation = {
-    require(parameters.get("path").isEmpty, "MODISCatalogDataSource doesn't support specifying a path. Please use `load()`.")
+    require(parameters.get("path").isEmpty, "L8CatalogDataSource doesn't support specifying a path. Please use `load()`.")
 
     val conf = sqlContext.sparkContext.hadoopConfiguration
     implicit val fs = FileSystem.get(conf)