new feature: allows user to specify timeZone in TimestampSpec

benchmarks/src/main/java/org/apache/druid/benchmark/FlattenJSONBenchmarkUtil.java

@@ -249,10 +249,23 @@ public String generateNestedEvent() throws Exception
     String[] dimsArray1 = {String.valueOf(rng.nextInt()), String.valueOf(rng.nextInt()), String.valueOf(rng.nextInt())};
     BenchmarkEvent nestedDims2 = new BenchmarkEvent(
         null,
-        null, null, String.valueOf(rng.nextInt()), String.valueOf(rng.nextInt()), String.valueOf(rng.nextInt()), String.valueOf(rng.nextInt()),
-        null, null, null, null,
-        null, null, null, null,
-        dimsArray1, null, null
+        null,
+        null,
+        String.valueOf(rng.nextInt()),
+        String.valueOf(rng.nextInt()),
+        String.valueOf(rng.nextInt()),
+        String.valueOf(rng.nextInt()),
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        null,
+        dimsArray1,
+        null,
+        null
     );
 
     Long[] metricsArray1 = {rng.nextLong(), rng.nextLong(), rng.nextLong(), rng.nextLong()};

benchmarks/src/main/java/org/apache/druid/benchmark/TimeParseBenchmark.java

@@ -53,7 +53,7 @@ public class TimeParseBenchmark
 
   static final String DATA_FORMAT = "MM/dd/yyyy HH:mm:ss Z";
 
-  static Function<String, DateTime> timeFn = TimestampParser.createTimestampParser(DATA_FORMAT);
+  static Function<String, DateTime> timeFn = TimestampParser.createTimestampParser(DATA_FORMAT, null);
 
   private String[] rows;
 

codestyle/joda-time-forbidden-apis.txt

@@ -60,7 +60,7 @@ org.joda.time.MutableInterval#parse(java.lang.String)
 
 org.joda.time.DateTimeZone#forID(java.lang.String) @ Use DateTimes.inferTzFromString() instead
 
-@defaultMessage Uses default time zone, use DateTimes.UtcFormatter to parse.
+@defaultMessage Uses default time zone, use DateTimes.Formatter or DateTimes.UtcFormatter to parse.
 org.joda.time.format.DateTimeFormatter#parseInto(org.joda.time.ReadWritableInstant, java.lang.String, int)
 org.joda.time.format.DateTimeFormatter#parseDateTime(java.lang.String)
 org.joda.time.format.DateTimeFormatter#parseMutableDateTime(java.lang.String)

core/src/main/java/org/apache/druid/data/input/impl/TimeAndDimsParseSpec.java

@@ -37,7 +37,7 @@ public TimeAndDimsParseSpec(
   )
   {
     super(
-        timestampSpec != null ? timestampSpec : new TimestampSpec(null, null, null),
+        timestampSpec != null ? timestampSpec : new TimestampSpec(null, null, null, null),
         dimensionsSpec != null ? dimensionsSpec : new DimensionsSpec(null, null, null)
     );
   }

core/src/main/java/org/apache/druid/data/input/impl/TimestampSpec.java

@@ -21,8 +21,11 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Function;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.druid.guice.annotations.PublicApi;
+import org.apache.druid.java.util.common.DateTimes;
 import org.apache.druid.java.util.common.parsers.TimestampParser;
 import org.joda.time.DateTime;
 
@@ -49,7 +52,10 @@ private static class ParseCtx
   private final String timestampFormat;
   // this value should never be set for production data
   private final DateTime missingValue;
-  /** This field is a derivative of {@link #timestampFormat}; not checked in {@link #equals} and {@link #hashCode} */
+  private final String timeZone;
+  /**
+   * This field is a derivative of {@link #timestampFormat}; not checked in {@link #equals} and {@link #hashCode}
+   */
   private final Function<Object, DateTime> timestampConverter;
 
   // remember last value parsed
@@ -60,17 +66,29 @@ public TimestampSpec(
       @JsonProperty("column") String timestampColumn,
       @JsonProperty("format") String format,
       // this value should never be set for production data
-      @JsonProperty("missingValue") DateTime missingValue
+      @JsonProperty("missingValue") DateTime missingValue,
+      @JsonProperty("timeZone") String timeZone
   )
   {
     this.timestampColumn = (timestampColumn == null) ? DEFAULT_COLUMN : timestampColumn;
     this.timestampFormat = format == null ? DEFAULT_FORMAT : format;
-    this.timestampConverter = TimestampParser.createObjectTimestampParser(timestampFormat);
+    this.timeZone = StringUtils.isBlank(timeZone) ? DateTimes.UTC_TIMEZONE : timeZone;
+    this.timestampConverter = TimestampParser.createObjectTimestampParser(timestampFormat, this.timeZone);
     this.missingValue = missingValue == null
                         ? DEFAULT_MISSING_VALUE
                         : missingValue;
   }
 
+  @VisibleForTesting
+  public TimestampSpec(
+      String timestampColumn,
+      String format,
+      DateTime missingValue
+  )
+  {
+    this(timestampColumn, format, missingValue, null);
+  }
+
   @JsonProperty("column")
   public String getTimestampColumn()
   {

core/src/main/java/org/apache/druid/java/util/common/DateTimes.java

@@ -20,10 +20,12 @@
 package org.apache.druid.java.util.common;
 
 import io.netty.util.SuppressForbidden;
+import org.apache.commons.lang3.StringUtils;
 import org.joda.time.Chronology;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.chrono.ISOChronology;
+import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 import org.joda.time.format.ISODateTimeFormat;
 
@@ -35,10 +37,11 @@ public final class DateTimes
   public static final DateTime EPOCH = utc(0);
   public static final DateTime MAX = utc(JodaUtils.MAX_INSTANT);
   public static final DateTime MIN = utc(JodaUtils.MIN_INSTANT);
+  public static final String UTC_TIMEZONE = DateTimeZone.UTC.getID();
 
-  public static final UtcFormatter ISO_DATE_TIME = wrapFormatter(ISODateTimeFormat.dateTime());
-  public static final UtcFormatter ISO_DATE_OPTIONAL_TIME = wrapFormatter(ISODateTimeFormat.dateOptionalTimeParser());
-  public static final UtcFormatter ISO_DATE_OR_TIME_WITH_OFFSET = wrapFormatter(
+  public static final UtcFormatter ISO_DATE_TIME = wrapUtcFormatter(ISODateTimeFormat.dateTime());
+  public static final UtcFormatter ISO_DATE_OPTIONAL_TIME = wrapUtcFormatter(ISODateTimeFormat.dateOptionalTimeParser());
+  public static final UtcFormatter ISO_DATE_OR_TIME_WITH_OFFSET = wrapUtcFormatter(
       ISODateTimeFormat.dateTimeParser().withOffsetParsed()
   );
 
@@ -64,18 +67,19 @@ public static DateTimeZone inferTzFromString(String tzId)
     }
   }
 
-  /**
-   * Simple wrapper class to enforce UTC Chronology in formatter. Specifically, it will use
-   * {@link DateTimeFormatter#withChronology(Chronology)} to set the chronology to
-   * {@link ISOChronology#getInstanceUTC()} on the wrapped {@link DateTimeFormatter}.
-   */
-  public static class UtcFormatter
+  public static class Formatter
   {
-    private final DateTimeFormatter innerFormatter;
+    private DateTimeFormatter innerFormatter;
 
-    private UtcFormatter(final DateTimeFormatter innerFormatter)
+    public Formatter(DateTimeFormatter innerFormatter)
+    {
+      this.innerFormatter = innerFormatter;
+    }
+
+    public Formatter(String format, String timezone)
     {
-      this.innerFormatter = innerFormatter.withChronology(ISOChronology.getInstanceUTC());
+      this.innerFormatter = DateTimeFormat.forPattern(format)
+                                          .withChronology(ISOChronology.getInstance(inferTzFromString(timezone)));
     }
 
     @SuppressForbidden(reason = "DateTimeFormatter#parseDateTime")
@@ -90,12 +94,39 @@ public String print(final DateTime instant)
     }
   }
 
+  /**
+   * Simple wrapper class to enforce UTC Chronology in formatter. Specifically, it will use
+   * {@link DateTimeFormatter#withChronology(Chronology)} to set the chronology to
+   * {@link ISOChronology#getInstanceUTC()} on the wrapped {@link DateTimeFormatter}.
+   */
+  public static class UtcFormatter extends Formatter
+  {
+    private UtcFormatter(final DateTimeFormatter innerFormatter)
+    {
+      super(innerFormatter.withChronology(ISOChronology.getInstanceUTC()));
+    }
+  }
+
+  /**
+   * Create a  {@link Formatter} by args
+   *
+   * @param format
+   * @param timeZone
+   *
+   * @return
+   */
+  public static Formatter wrapFormatter(String format, String timeZone)
+  {
+    timeZone = StringUtils.isNotBlank(timeZone) ? timeZone : DateTimeZone.UTC.getID();
+    return new Formatter(format, timeZone);
+  }
+
   /**
    * Creates a {@link UtcFormatter} that wraps around a {@link DateTimeFormatter}.
    *
    * @param formatter inner {@link DateTimeFormatter} used to parse {@link String}
    */
-  public static UtcFormatter wrapFormatter(final DateTimeFormatter formatter)
+  public static UtcFormatter wrapUtcFormatter(final DateTimeFormatter formatter)
   {
     return new UtcFormatter(formatter);
   }

core/src/main/java/org/apache/druid/java/util/common/parsers/TimestampParser.java

@@ -26,7 +26,6 @@
 import org.apache.druid.java.util.common.IAE;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
-import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 import org.joda.time.format.DateTimeFormatterBuilder;
 import org.joda.time.format.DateTimeParser;
@@ -37,12 +36,13 @@
 public class TimestampParser
 {
   public static Function<String, DateTime> createTimestampParser(
-      final String format
+      final String format,
+      final String userTimeZone
   )
   {
     if ("auto".equalsIgnoreCase(format)) {
       // Could be iso or millis
-      final DateTimes.UtcFormatter parser = DateTimes.wrapFormatter(createAutoParser());
+      final DateTimes.UtcFormatter parser = DateTimes.wrapUtcFormatter(createAutoParser());
       return (String input) -> {
         Preconditions.checkArgument(!Strings.isNullOrEmpty(input), "null timestamp");
 
@@ -87,7 +87,7 @@ public static Function<String, DateTime> createTimestampParser(
       };
     } else {
       try {
-        final DateTimes.UtcFormatter formatter = DateTimes.wrapFormatter(DateTimeFormat.forPattern(format));
+        final DateTimes.Formatter formatter = DateTimes.wrapFormatter(format, userTimeZone);
         return input -> {
           Preconditions.checkArgument(!Strings.isNullOrEmpty(input), "null timestamp");
           return formatter.parse(ParserUtils.stripQuotes(input));
@@ -117,10 +117,11 @@ public static Function<Number, DateTime> createNumericTimestampParser(
   }
 
   public static Function<Object, DateTime> createObjectTimestampParser(
-      final String format
+      final String format,
+      final String timeZone
   )
   {
-    final Function<String, DateTime> stringFun = createTimestampParser(format);
+    final Function<String, DateTime> stringFun = createTimestampParser(format, timeZone);
     final Function<Number, DateTime> numericFun = createNumericTimestampParser(format);
 
     return o -> {

core/src/main/java/org/apache/druid/math/expr/Function.java

@@ -835,7 +835,7 @@ public ExprEval apply(List<Expr> args, Expr.ObjectBinding bindings)
         if (format.type() != ExprType.STRING) {
           throw new IAE("second argument should be string type but got %s type", format.type());
         }
-        formatter = DateTimes.wrapFormatter(DateTimeFormat.forPattern(format.asString()));
+        formatter = DateTimes.wrapUtcFormatter(DateTimeFormat.forPattern(format.asString()));
       }
       DateTime date;
       try {

core/src/test/java/org/apache/druid/data/input/impl/TimestampSpecTest.java

@@ -61,12 +61,38 @@ public void testContextualTimestampList()
         };
     TimestampSpec spec = new TimestampSpec("TIMEstamp", DATE_FORMAT, null);
 
-    DateTimes.UtcFormatter formatter = DateTimes.wrapFormatter(ISODateTimeFormat.dateHourMinuteSecond());
+    DateTimes.UtcFormatter formatter = DateTimes.wrapUtcFormatter(ISODateTimeFormat.dateHourMinuteSecond());
 
     for (String date : dates) {
       DateTime dateTime = spec.extractTimestamp(ImmutableMap.of("TIMEstamp", date));
       DateTime expectedDateTime = formatter.parse(date);
       Assert.assertEquals(expectedDateTime, dateTime);
     }
   }
+
+  @Test
+  public void testExtractTimestampWithTimezone()
+  {
+    String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
+    String timeZone = "Asia/Shanghai";
+    String[] dates = new String[]{
+        "2000-01-01 00:00:00",
+        "2000-01-01 08:00:00",
+        "2000-01-01 12:00:01",
+        "2000-01-01 23:59:59",
+        };
+    TimestampSpec spec = new TimestampSpec("log_time", DATE_FORMAT, null, timeZone);
+
+    DateTimes.Formatter expectedFormatter = DateTimes.wrapFormatter(DATE_FORMAT, timeZone);
+
+    for (String date : dates) {
+      DateTime actualDateTime = spec.extractTimestamp(ImmutableMap.<String, Object>of(
+          "log_time", date,
+          "dim1", "value1"
+      ));
+      long expectedTs = expectedFormatter.parse(date).getMillis();
+      Assert.assertEquals(expectedTs, actualDateTime.getMillis());
+    }
+  }
+
 }