#46 improved join and ordering tests

Author: adersberger

chronix-solr-cloud-storage/src/main/java/de/qaware/chronix/storage/solr/timeseries/metric/MetricTimeSeriesOrdering.java

@@ -17,8 +17,11 @@
 
 import com.google.common.collect.Ordering;
 import de.qaware.chronix.timeseries.MetricTimeSeries;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.util.Date;
 
 /**
  * Orders MetricTimeSeries by their start timestamp
@@ -31,6 +34,9 @@ public class MetricTimeSeriesOrdering extends Ordering<MetricTimeSeries> impleme
 
     private static final long serialVersionUID = 42L;
 
+    private final Logger LOGGER = LoggerFactory.getLogger(MetricTimeSeriesOrdering.class);
+
+
     @Override
     public int compare(MetricTimeSeries left, MetricTimeSeries right) {
         if (left == null && right == null) {
@@ -40,11 +46,30 @@ public int compare(MetricTimeSeries left, MetricTimeSeries right) {
         } else if (right == null) {
             return 1;
         } else if (left.getStart() < right.getStart()) {
+            if (left.getEnd() >= right.getStart()) printTimeseriesDescription(left, right);
             return -1;
         } else if (left.getStart() == right.getStart()) {
             return 0;
         } else {
+            if (left.getEnd() <= right.getStart()) printTimeseriesDescription(left, right);
             return 1;
         }
     }
+
+    private void printTimeseriesDescription(MetricTimeSeries left, MetricTimeSeries right) {
+        LOGGER.error("Detected intersecting time series chunks:");
+        LOGGER.error(
+                "LEFT: {}", getTimeseriesIntervalDescription(left)
+        );
+        LOGGER.error(
+                "RIGHT: {}", getTimeseriesIntervalDescription(right)
+        );
+    }
+
+    private String getTimeseriesIntervalDescription(MetricTimeSeries mts) {
+        String start = new Date(mts.getStart()).toString();
+        String end = new Date(mts.getEnd()).toString();
+        StringBuffer sb = new StringBuffer();
+        return sb.append(start).append(", ").append(end).toString();
+    }
 }

chronix-spark-ts-rdd/src/test/groovy/de/qaware/chronix/spark/api/java/TestChronixSparkContext.groovy

@@ -15,6 +15,7 @@
  */
 package de.qaware.chronix.spark.api.java
 
+import de.qaware.chronix.storage.solr.timeseries.metric.MetricTimeSeriesKey
 import de.qaware.chronix.timeseries.MetricTimeSeries
 import org.apache.solr.client.solrj.SolrQuery
 import org.apache.spark.SparkConf
@@ -49,7 +50,7 @@ class TestChronixSparkContext extends Specification {
         then:
         Assert.assertTrue(timeSeries.size() == 5)
         for (MetricTimeSeries ts : timeSeries) {
-            System.out.println(ts.toString());
+            System.out.println(ts.toString())
         }
     }
 
@@ -62,9 +63,45 @@ class TestChronixSparkContext extends Specification {
         long joined = result.count()
         println "Chunked: " + chunked
         println "Joined: " + joined
+        Assert.assertTrue(resultChunked.count() >= result.count())
+
+        Set<MetricTimeSeriesKey> keys = new HashSet<>();
+        int incorrectCnt = 0;
+        int correctCnt = 0;
+
+        println "CHUNKED *****************"
+        for (MetricTimeSeries mts : resultChunked.collect()) {
+            //check ordering
+            long prevTimeStamp = 0.0
+            for (Long timeStamp : mts.getTimestampsAsArray()) {
+                Assert.assertTrue(mts.getStart() <= timeStamp &&
+                        timeStamp <= mts.getEnd())
+                Assert.assertTrue(timeStamp > prevTimeStamp)
+                prevTimeStamp = timeStamp
+            }
+        }
+
+        println "JOINED ******************"
         for (MetricTimeSeries mts : result.collect()) {
-            println mts
+            //check identity
+            MetricTimeSeriesKey mtsKey = new MetricTimeSeriesKey(mts)
+            Assert.assertTrue(!keys.contains(mtsKey))
+            keys.add(mtsKey)
+            //check ordering
+            long prevTimeStamp = 0.0
+            for (Long timeStamp : mts.getTimestampsAsArray()) {
+                Assert.assertTrue(mts.getStart() <= timeStamp &&
+                        timeStamp <= mts.getEnd())
+                if (timeStamp > prevTimeStamp) correctCnt++
+                else {
+                    incorrectCnt++
+                    println mts.metric
+                    println "Incorrect timestamps: " + timeStamp + " < " + prevTimeStamp
+                }
+                prevTimeStamp = timeStamp
+            }
         }
-        Assert.assertTrue(resultChunked.count() >= result.count())
+        println "Incorrect ordering: " + incorrectCnt
+        println "Correct ordering: " + correctCnt
     }
 }