Merge pull request #520 from bonitoo-io/issue_517

fix issue #517 : missing millis and nanos in MsgPack
influxdata · Sep 11, 2018 · ad97d00 · ad97d00
2 parents 215fe46 + f111d84
commit ad97d00
Show file tree

Hide file tree

Showing 3 changed files with 38 additions and 13 deletions.
diff --git a/src/main/java/org/influxdb/msgpack/MessagePackTraverser.java b/src/main/java/org/influxdb/msgpack/MessagePackTraverser.java
@@ -8,6 +8,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.influxdb.InfluxDBException;
 import org.influxdb.dto.QueryResult;
@@ -27,6 +28,7 @@
  */
 public class MessagePackTraverser {
 
+ private static final byte MSG_PACK_TIME_EXT_TYPE = 5;
  private String lastStringNode;
 
  /**
@@ -229,14 +231,17 @@ void traverse(final MessageUnpacker unpacker, final QueryResultModelPath queryRe
  }
  break;
  case EXTENSION:
- final byte msgPackTimeExtType = (byte) 5;
- final int timeOffset = 0;
- final int timeByteArrayLength = 8;
+ final int nanosStartIndex = 8;
  extension = unpacker.unpackExtensionTypeHeader();
- if (extension.getType() == msgPackTimeExtType) {
+ if (extension.getType() == MSG_PACK_TIME_EXT_TYPE) {
+ //decode epoch nanos in accordance with https://github.com/tinylib/msgp/blob/master/msgp/write.go#L594
+
  dst = new byte[extension.getLength()];
  unpacker.readPayload(dst);
- o = ByteBuffer.wrap(dst, timeOffset, timeByteArrayLength).getLong();
+ ByteBuffer bf = ByteBuffer.wrap(dst, 0, extension.getLength());
+ long epochSeconds = bf.getLong();
+ int nanosOffset = bf.getInt(nanosStartIndex);
+ o = TimeUnit.SECONDS.toNanos(epochSeconds) + nanosOffset;
  }
  break;
 

diff --git a/src/test/java/org/influxdb/MessagePackInfluxDBTest.java b/src/test/java/org/influxdb/MessagePackInfluxDBTest.java
@@ -5,6 +5,7 @@
 import static org.mockito.Mockito.spy;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -91,9 +92,17 @@ public void testWriteBatchWithPrecision() throws Exception {
 
  // THEN the measure points have a timestamp with second precision
  QueryResult queryResult = this.influxDB.query(new Query("SELECT * FROM " + measurement, dbName));
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0), t1);
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0), t2);
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0), t3);
+ long bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0));
+ Assertions.assertEquals(bySecond, t1);
+
+ bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0));
+ Assertions.assertEquals(bySecond, t2);
+
+ bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0));
+ Assertions.assertEquals(bySecond, t3);
 
  this.influxDB.deleteDatabase(dbName);
  }
@@ -182,9 +191,19 @@ public void testWriteRecordsWithPrecision() throws Exception {
  // THEN the measure points have a timestamp with second precision
  QueryResult queryResult = this.influxDB.query(new Query("SELECT * FROM " + measurement, dbName));
  Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().size(), 3);
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0), timeP1);
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0), timeP2);
- Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0), timeP3);
+
+ long bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0));
+ Assertions.assertEquals(bySecond, timeP1);
+
+ bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0));
+ Assertions.assertEquals(bySecond, timeP2);
+
+ bySecond = TimeUnit.NANOSECONDS.toSeconds(
+ (Long) queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0));
+ Assertions.assertEquals(bySecond, timeP3);
+
  this.influxDB.deleteDatabase(dbName);
  }
 

diff --git a/src/test/java/org/influxdb/msgpack/MessagePackTraverserTest.java b/src/test/java/org/influxdb/msgpack/MessagePackTraverserTest.java
@@ -33,7 +33,8 @@ public void testTraverseMethod() {
  QueryResult result = iter.next();
  List<List<Object>> values = result.getResults().get(0).getSeries().get(0).getValues();
  Assertions.assertEquals(2, values.size());
- assertEquals(1532325083L, values.get(0).get(0));
+
+ assertEquals(1532325083803052600L, values.get(0).get(0));
  assertEquals("b", values.get(1).get(1));
 
  assertTrue(iter.hasNext());
@@ -56,7 +57,7 @@ public void testParseMethodOnNonEmptyResult() {
  QueryResult queryResult = traverser.parse(MessagePackTraverserTest.class.getResourceAsStream("msgpack_2.bin"));
  List<List<Object>> values = queryResult.getResults().get(0).getSeries().get(0).getValues();
  Assertions.assertEquals(3, values.size());
- assertEquals(1485273600L, values.get(0).get(0));
+ assertEquals(1485273600000000000L, values.get(0).get(0));
  assertEquals("two", values.get(1).get(1));
  assertEquals(3.0, values.get(2).get(2));
  }