Handle Invalid timestamps

pinot-core/src/test/java/org/apache/pinot/core/util/SchemaUtilsTest.java

@@ -334,6 +334,33 @@ public void testDateTimeFieldSpec()
             + "\"format\":\"1:DAYS:SIMPLE_DATE_FORMAT\",\"granularity\":\"1:DAYS\"}]}");
     checkValidationFails(schema);
 
+    schema = Schema.fromString(
+        "{\"schemaName\":\"testSchema\"," + "\"dimensionFieldSpecs\":[ {\"name\":\"dim1\",\"dataType\":\"STRING\"}],"
+            + "\"dateTimeFieldSpecs\":[{\"name\":\"dt1\",\"dataType\":\"INT\","
+            + "\"format\":\"1:DAYS:SIMPLE_DATE_FORMAT:yyyyMMdd\",\"granularity\":\"1:DAYS\","
+            + " \"sampleValue\" : \"19700101\"}]}");
+    checkValidationFails(schema);
+
+    schema = Schema.fromString(
+        "{\"schemaName\":\"testSchema\"," + "\"dimensionFieldSpecs\":[ {\"name\":\"dim1\",\"dataType\":\"STRING\"}],"
+            + "\"dateTimeFieldSpecs\":[{\"name\":\"dt1\",\"dataType\":\"INT\","
+            + "\"format\":\"1:DAYS:SIMPLE_DATE_FORMAT:yyyyMMdd\",\"granularity\":\"1:DAYS\","
+            + " \"sampleValue\" : \"19720101\"}]}");
+    SchemaUtils.validate(schema);
+
+    schema = Schema.fromString(
+        "{\"schemaName\":\"testSchema\"," + "\"dimensionFieldSpecs\":[ {\"name\":\"dim1\",\"dataType\":\"STRING\"}],"
+            + "\"dateTimeFieldSpecs\":[{\"name\":\"dt1\",\"dataType\":\"LONG\","
+            + "\"format\":\"1:SECONDS:EPOCH\",\"granularity\":\"1:SECONDS\", \"sampleValue\" : 1663170923}]}");
+    SchemaUtils.validate(schema);
+
+    schema = Schema.fromString(
+        "{\"schemaName\":\"testSchema\"," + "\"dimensionFieldSpecs\":[ {\"name\":\"dim1\",\"dataType\":\"STRING\"}],"
+            + "\"dateTimeFieldSpecs\":[{\"name\":\"dt1\",\"dataType\":\"LONG\","
+            + "\"format\":\"1:MILLISECONDS:EPOCH\",\"granularity\":\"1:MILLISECONDS\","
+            + " \"sampleValue\" : 1663170923}]}");
+    checkValidationFails(schema);
+
     schema = Schema.fromString(
         "{\"schemaName\":\"testSchema\"," + "\"dimensionFieldSpecs\":[ {\"name\":\"dim1\",\"dataType\":\"STRING\"}],"
             + "\"dateTimeFieldSpecs\":[{\"name\":\"dt1\",\"dataType\":\"INT\",\"format\":\"1:HOURS:EPOCH\","

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SchemaUtils.java

@@ -36,6 +36,7 @@
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.data.TimeFieldSpec;
 import org.apache.pinot.spi.data.TimeGranularitySpec;
+import org.apache.pinot.spi.utils.TimeUtils;
 
 
 /**
@@ -198,6 +199,23 @@ private static void validateDateTimeFieldSpec(DateTimeFieldSpec dateTimeFieldSpe
       }
     }
 
+    Object sampleValue = dateTimeFieldSpec.getSampleValue();
+    if (sampleValue != null) {
+      long sampleTimestampValue;
+      try {
+        sampleTimestampValue = formatSpec.fromFormatToMillis(sampleValue.toString());
+      } catch (Exception e) {
+        throw new IllegalArgumentException(
+            String.format("Cannot format provided sample value: %s with provided date time spec: %s", sampleValue,
+                formatSpec));
+      }
+      boolean isValidTimestamp = TimeUtils.timeValueInValidRange(sampleTimestampValue);
+      Preconditions.checkArgument(isValidTimestamp,
+          "Incorrect date time format. "
+              + "Converted sample value %s for date-time field spec is not in valid time-range: %s and %s",
+          sampleTimestampValue, TimeUtils.VALID_MIN_TIME_MILLIS, TimeUtils.VALID_MAX_TIME_MILLIS);
+    }
+
     DateTimeGranularitySpec granularitySpec;
     try {
       granularitySpec = dateTimeFieldSpec.getGranularitySpec();

pinot-spi/src/main/java/org/apache/pinot/spi/data/DateTimeFieldSpec.java

@@ -25,12 +25,12 @@
 import javax.annotation.Nullable;
 import org.apache.pinot.spi.utils.EqualityUtils;
 
-
 @SuppressWarnings("unused")
 @JsonIgnoreProperties(ignoreUnknown = true)
 public final class DateTimeFieldSpec extends FieldSpec {
   private String _format;
   private String _granularity;
+  private Object _sampleValue;
   private transient DateTimeFormatSpec _formatSpec;
   private transient DateTimeGranularitySpec _granularitySpec;
 
@@ -70,13 +70,19 @@ public DateTimeFieldSpec() {
    *       2) if a time column is defined in hoursSinceEpoch (format=1:HOURS:EPOCH), and the data buckets are 1 hours,
    *          the granularity will be 1:HOURS
    */
-  public DateTimeFieldSpec(String name, DataType dataType, String format, String granularity) {
+  public DateTimeFieldSpec(String name, DataType dataType, String format, String granularity,
+      @Nullable Object sampleValue) {
     super(name, dataType, true);
 
     _format = format;
     _granularity = granularity;
     _formatSpec = new DateTimeFormatSpec(format);
     _granularitySpec = new DateTimeGranularitySpec(granularity);
+    _sampleValue = sampleValue;
+  }
+
+  public DateTimeFieldSpec(String name, DataType dataType, String format, String granularity) {
+    this(name, dataType, format, granularity, null);
   }
 
   /**
@@ -86,7 +92,17 @@ public DateTimeFieldSpec(String name, DataType dataType, String format, String g
    */
   public DateTimeFieldSpec(String name, DataType dataType, String format, String granularity,
       @Nullable Object defaultNullValue, @Nullable String transformFunction) {
-    this(name, dataType, format, granularity);
+    this(name, dataType, format, granularity, defaultNullValue, transformFunction, null);
+  }
+
+  /**
+   * Constructs a DateTimeFieldSpec with basic fields - name, dataType, format, granularity - and also with
+   * defaultNullValue and
+   * transformFunction
+   */
+  public DateTimeFieldSpec(String name, DataType dataType, String format, String granularity,
+      @Nullable Object defaultNullValue, @Nullable String transformFunction, @Nullable String sampleValue) {
+    this(name, dataType, format, granularity, sampleValue);
     setDefaultNullValue(defaultNullValue);
     setTransformFunction(transformFunction);
   }
@@ -119,6 +135,7 @@ public DateTimeFormatSpec getFormatSpec() {
       formatSpec = new DateTimeFormatSpec(_format);
       _formatSpec = formatSpec;
     }
+
     return formatSpec;
   }
 
@@ -131,6 +148,14 @@ public void setGranularity(String granularity) {
     _granularity = granularity;
   }
 
+  public Object getSampleValue() {
+    return _sampleValue;
+  }
+
+  public void setSampleValue(String sampleValue) {
+    _sampleValue = sampleValue;
+  }
+
   @JsonIgnore
   public DateTimeGranularitySpec getGranularitySpec() {
     DateTimeGranularitySpec granularitySpec = _granularitySpec;
@@ -162,13 +187,15 @@ public boolean equals(Object o) {
     }
 
     DateTimeFieldSpec that = (DateTimeFieldSpec) o;
-    return EqualityUtils.isEqual(_format, that._format) && EqualityUtils.isEqual(_granularity, that._granularity);
+    return EqualityUtils.isEqual(_format, that._format) && EqualityUtils.isEqual(_granularity, that._granularity)
+        && EqualityUtils.isEqual(_sampleValue, that._sampleValue);
   }
 
   @Override
   public int hashCode() {
     int result = EqualityUtils.hashCodeOf(super.hashCode(), _format);
     result = EqualityUtils.hashCodeOf(result, _granularity);
+    result = EqualityUtils.hashCodeOf(result, _sampleValue);
     return result;
   }
 }