diff --git druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSelectQueryRecordReader.java druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSelectQueryRecordReader.java index 8b40d4d..c30ac56 100644 --- druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSelectQueryRecordReader.java +++ druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidSelectQueryRecordReader.java @@ -76,11 +76,9 @@ public NullWritable getCurrentKey() throws IOException, InterruptedException { public DruidWritable getCurrentValue() throws IOException, InterruptedException { // Create new value DruidWritable value = new DruidWritable(); - value.getValue().put(DruidTable.DEFAULT_TIMESTAMP_COLUMN, current.getTimestamp().getMillis()); - if (values.hasNext()) { - value.getValue().putAll(values.next().getEvent()); - return value; - } + EventHolder e = values.next(); + value.getValue().put(DruidTable.DEFAULT_TIMESTAMP_COLUMN, e.getTimestamp().getMillis()); + value.getValue().putAll(e.getEvent()); return value; } @@ -89,10 +87,9 @@ public boolean next(NullWritable key, DruidWritable value) throws IOException { if (nextKeyValue()) { // Update value value.getValue().clear(); - value.getValue().put(DruidTable.DEFAULT_TIMESTAMP_COLUMN, current.getTimestamp().getMillis()); - if (values.hasNext()) { - value.getValue().putAll(values.next().getEvent()); - } + EventHolder e = values.next(); + value.getValue().put(DruidTable.DEFAULT_TIMESTAMP_COLUMN, e.getTimestamp().getMillis()); + value.getValue().putAll(e.getEvent()); return true; } return false; diff --git druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java index 9c5c65c..1343939 100644 --- druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java +++ druid-handler/src/test/org/apache/hadoop/hive/druid/TestDruidSerDe.java @@ -357,7 +357,7 @@ + " \"segmentId\" : \"wikipedia_2012-12-29T00:00:00.000Z_2013-01-10T08:00:00.000Z_2013-01-10T08:13:47.830Z_v9\", " + " \"offset\" : 2, " + " \"event\" : { " - + " \"timestamp\" : \"2013-01-01T00:00:00.000Z\", " + + " \"timestamp\" : \"2013-01-01T00:00:12.000Z\", " + " \"robot\" : \"0\", " + " \"namespace\" : \"article\", " + " \"anonymous\" : \"0\", " @@ -376,7 +376,7 @@ + " \"segmentId\" : \"wikipedia_2012-12-29T00:00:00.000Z_2013-01-10T08:00:00.000Z_2013-01-10T08:13:47.830Z_v9\", " + " \"offset\" : 3, " + " \"event\" : { " - + " \"timestamp\" : \"2013-01-01T00:00:00.000Z\", " + + " \"timestamp\" : \"2013-01-01T00:00:12.000Z\", " + " \"robot\" : \"0\", " + " \"namespace\" : \"article\", " + " \"anonymous\" : \"0\", " @@ -395,7 +395,7 @@ + " \"segmentId\" : \"wikipedia_2012-12-29T00:00:00.000Z_2013-01-10T08:00:00.000Z_2013-01-10T08:13:47.830Z_v9\", " + " \"offset\" : 4, " + " \"event\" : { " - + " \"timestamp\" : \"2013-01-01T00:00:00.000Z\", " + + " \"timestamp\" : \"2013-01-01T00:00:12.000Z\", " + " \"robot\" : \"0\", " + " \"namespace\" : \"article\", " + " \"anonymous\" : \"0\", " @@ -419,13 +419,13 @@ new Object[] { new TimestampWritable(new Timestamp(1356998400000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), new Text("112_U.S._580"), new Text("en"), new Text("1"), new Text("MZMcBride"), new FloatWritable(1.0F), new FloatWritable(70.0F), new FloatWritable(70.0F), new FloatWritable(70.0F), new FloatWritable(0.0F) } , - new Object[] { new TimestampWritable(new Timestamp(1356998400000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), + new Object[] { new TimestampWritable(new Timestamp(1356998412000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), new Text("113_U.S._243"), new Text("en"), new Text("1"), new Text("MZMcBride"), new FloatWritable(1.0F), new FloatWritable(77.0F), new FloatWritable(77.0F), new FloatWritable(77.0F), new FloatWritable(0.0F) } , - new Object[] { new TimestampWritable(new Timestamp(1356998400000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), + new Object[] { new TimestampWritable(new Timestamp(1356998412000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), new Text("113_U.S._73"), new Text("en"), new Text("1"), new Text("MZMcBride"), new FloatWritable(1.0F), new FloatWritable(70.0F), new FloatWritable(70.0F), new FloatWritable(70.0F), new FloatWritable(0.0F) } , - new Object[] { new TimestampWritable(new Timestamp(1356998400000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), + new Object[] { new TimestampWritable(new Timestamp(1356998412000L)), new Text("0"), new Text("article"), new Text("0"), new Text("0"), new Text("113_U.S._756"), new Text("en"), new Text("1"), new Text("MZMcBride"), new FloatWritable(1.0F), new FloatWritable(68.0F), new FloatWritable(68.0F), new FloatWritable(68.0F), new FloatWritable(0.0F) } };