public DateTimeFormatSpec(String format) {
    Preconditions.checkArgument(StringUtils.isNotEmpty(format), "Must provide format");

    if (Character.isDigit(format.charAt(0))) {
      // Colon format

      String[] tokens = StringUtil.split(format, COLON_SEPARATOR, COLON_FORMAT_MAX_TOKENS);
      Preconditions.checkArgument(tokens.length >= COLON_FORMAT_MIN_TOKENS && tokens.length <= COLON_FORMAT_MAX_TOKENS,
          "Invalid format: %s, must be of format 'size:timeUnit:timeFormat(:patternWithTz)'", format);

      TimeFormat timeFormat;
      try {
        timeFormat = TimeFormat.valueOf(tokens[COLON_FORMAT_TIME_FORMAT_POSITION]);
      } catch (Exception e) {
        throw new IllegalArgumentException(
            String.format("Invalid time format: %s in format: %s", tokens[COLON_FORMAT_TIME_FORMAT_POSITION], format));
      }

      switch (timeFormat) {
        case EPOCH:
          String sizeStr = tokens[COLON_FORMAT_SIZE_POSITION];
          try {
            _size = Integer.parseInt(sizeStr);
          } catch (Exception e) {
            throw new IllegalArgumentException(String.format("Invalid size: %s in format: %s", sizeStr, format));
          }
          Preconditions.checkArgument(_size > 0, "Invalid size: %s in format: %s, must be positive", _size, format);
          String timeUnitStr = tokens[COLON_FORMAT_TIME_UNIT_POSITION];
          try {
            _unitSpec = new DateTimeFormatUnitSpec(timeUnitStr);
          } catch (Exception e) {
            throw new IllegalArgumentException(
                String.format("Invalid time unit: %s in format: %s", timeUnitStr, format));
          }
          _patternSpec = DateTimeFormatPatternSpec.EPOCH;
          break;
        case TIMESTAMP:
          _size = 1;
          _unitSpec = DateTimeFormatUnitSpec.MILLISECONDS;
          _patternSpec = DateTimeFormatPatternSpec.TIMESTAMP;
          break;
        case SIMPLE_DATE_FORMAT:
          _size = 1;
          _unitSpec = DateTimeFormatUnitSpec.MILLISECONDS;
          String patternStr =
              tokens.length > COLON_FORMAT_PATTERN_POSITION ? tokens[COLON_FORMAT_PATTERN_POSITION] : null;
          try {
            _patternSpec = new DateTimeFormatPatternSpec(TimeFormat.SIMPLE_DATE_FORMAT, patternStr);
          } catch (Exception e) {
            throw new IllegalArgumentException(
                String.format("Invalid SIMPLE_DATE_FORMAT pattern: %s in format: %s", patternStr, format));
          }
          break;
        default:
          throw new IllegalStateException("Unsupported time format: " + timeFormat);
      }
    } else {
      // Pipe format

      String[] tokens = StringUtil.split(format, PIPE_SEPARATOR, PIPE_FORMAT_MAX_TOKENS);
      Preconditions.checkArgument(tokens.length >= PIPE_FORMAT_MIN_TOKENS && tokens.length <= PIPE_FORMAT_MAX_TOKENS,
          "Invalid format: %s, must be of format 'EPOCH|<timeUnit>(|<size>)' or "
              + "'SIMPLE_DATE_FORMAT|<pattern>(|<timeZone>)' or 'TIMESTAMP'", format);

      TimeFormat timeFormat;
      try {
        timeFormat = TimeFormat.valueOf(tokens[PIPE_FORMAT_TIME_FORMAT_POSITION]);
      } catch (Exception e) {
        throw new IllegalArgumentException(
            String.format("Invalid time format: %s in format: %s", tokens[PIPE_FORMAT_TIME_FORMAT_POSITION], format));
      }

      switch (timeFormat) {
        case EPOCH:
          if (tokens.length > PIPE_FORMAT_SIZE_POSITION) {
            try {
              _size = Integer.parseInt(tokens[PIPE_FORMAT_SIZE_POSITION]);
            } catch (Exception e) {
              throw new IllegalArgumentException(
                  String.format("Invalid size: %s in format: %s", tokens[COLON_FORMAT_SIZE_POSITION], format));
            }
            Preconditions.checkArgument(_size > 0, "Invalid size: %s in format: %s, must be positive", _size, format);
          } else {
            _size = 1;
          }
          try {
            _unitSpec = tokens.length > PIPE_FORMAT_TIME_UNIT_POSITION ? new DateTimeFormatUnitSpec(
                tokens[PIPE_FORMAT_TIME_UNIT_POSITION]) : DateTimeFormatUnitSpec.MILLISECONDS;
          } catch (Exception e) {
            throw new IllegalArgumentException(
                String.format("Invalid time unit: %s in format: %s", tokens[PIPE_FORMAT_TIME_UNIT_POSITION], format));
          }
          _patternSpec = DateTimeFormatPatternSpec.EPOCH;
          break;
        case TIMESTAMP:
          _size = 1;
          _unitSpec = DateTimeFormatUnitSpec.MILLISECONDS;
          _patternSpec = DateTimeFormatPatternSpec.TIMESTAMP;
          break;
        case SIMPLE_DATE_FORMAT:
          _size = 1;
          _unitSpec = DateTimeFormatUnitSpec.MILLISECONDS;
          if (tokens.length > PIPE_FORMAT_TIME_ZONE_POSITION) {
            try {
              _patternSpec =
                  new DateTimeFormatPatternSpec(TimeFormat.SIMPLE_DATE_FORMAT, tokens[PIPE_FORMAT_PATTERN_POSITION],
                      tokens[PIPE_FORMAT_TIME_ZONE_POSITION]);
            } catch (Exception e) {
              throw new IllegalArgumentException(
                  String.format("Invalid SIMPLE_DATE_FORMAT pattern: %s, time zone: %s in format: %s",
                      tokens[PIPE_FORMAT_PATTERN_POSITION], tokens[PIPE_FORMAT_TIME_ZONE_POSITION], format));
            }
          } else {
            try {
              String pattern =
                  tokens.length > PIPE_FORMAT_PATTERN_POSITION ? tokens[PIPE_FORMAT_PATTERN_POSITION] : null;
              _patternSpec = new DateTimeFormatPatternSpec(TimeFormat.SIMPLE_DATE_FORMAT, pattern);
            } catch (Exception e) {
              throw new IllegalArgumentException(String.format("Invalid SIMPLE_DATE_FORMAT pattern: %s in format: %s",
                  tokens[PIPE_FORMAT_PATTERN_POSITION], format));
            }
          }
          break;
        default:
          throw new IllegalStateException("Unsupported time format: " + timeFormat);
      }
    }
  }
@Test
  public void testDateTimeFormatSpec() {
    DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec("5:DAYS:EPOCH");
    assertEquals(dateTimeFormatSpec.getTimeFormat(), DateTimeFieldSpec.TimeFormat.EPOCH);
    assertEquals(dateTimeFormatSpec.getColumnSize(), 5);
    assertEquals(dateTimeFormatSpec.getColumnUnit(), TimeUnit.DAYS);
    assertEquals(dateTimeFormatSpec.getColumnDateTimeTransformUnit(),
        DateTimeFormatUnitSpec.DateTimeTransformUnit.DAYS);
    assertNull(dateTimeFormatSpec.getSDFPattern());

    assertEquals(new DateTimeFormatSpec("EPOCH|DAYS|5"), dateTimeFormatSpec);

    dateTimeFormatSpec = new DateTimeFormatSpec("1:DAYS:TIMESTAMP");
    assertEquals(dateTimeFormatSpec.getTimeFormat(), DateTimeFieldSpec.TimeFormat.TIMESTAMP);
    assertEquals(dateTimeFormatSpec.getColumnSize(), 1);
    assertEquals(dateTimeFormatSpec.getColumnUnit(), TimeUnit.MILLISECONDS);
    assertEquals(dateTimeFormatSpec.getColumnDateTimeTransformUnit(),
        DateTimeFormatUnitSpec.DateTimeTransformUnit.MILLISECONDS);
    assertNull(dateTimeFormatSpec.getSDFPattern());

    assertEquals(new DateTimeFormatSpec("TIMESTAMP"), dateTimeFormatSpec);

    dateTimeFormatSpec = new DateTimeFormatSpec("1:DAYS:SIMPLE_DATE_FORMAT:yyyyMMdd");
    assertEquals(dateTimeFormatSpec.getTimeFormat(), DateTimeFieldSpec.TimeFormat.SIMPLE_DATE_FORMAT);
    assertEquals(dateTimeFormatSpec.getColumnSize(), 1);
    assertEquals(dateTimeFormatSpec.getColumnUnit(), TimeUnit.MILLISECONDS);
    assertEquals(dateTimeFormatSpec.getColumnDateTimeTransformUnit(),
        DateTimeFormatUnitSpec.DateTimeTransformUnit.MILLISECONDS);
    assertEquals(dateTimeFormatSpec.getSDFPattern(), "yyyyMMdd");
    assertEquals(dateTimeFormatSpec.getDateTimezone(), DateTimeZone.UTC);

    assertEquals(new DateTimeFormatSpec("SIMPLE_DATE_FORMAT|yyyyMMdd"), dateTimeFormatSpec);

    dateTimeFormatSpec = new DateTimeFormatSpec("1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd tz(CST)");
    assertEquals(dateTimeFormatSpec.getTimeFormat(), DateTimeFieldSpec.TimeFormat.SIMPLE_DATE_FORMAT);
    assertEquals(dateTimeFormatSpec.getColumnSize(), 1);
    assertEquals(dateTimeFormatSpec.getColumnUnit(), TimeUnit.MILLISECONDS);
    assertEquals(dateTimeFormatSpec.getColumnDateTimeTransformUnit(),
        DateTimeFormatUnitSpec.DateTimeTransformUnit.MILLISECONDS);
    assertEquals(dateTimeFormatSpec.getSDFPattern(), "yyyy-MM-dd");
    assertEquals(dateTimeFormatSpec.getDateTimezone(), DateTimeZone.forTimeZone(TimeZone.getTimeZone("CST")));

    assertEquals(new DateTimeFormatSpec("SIMPLE_DATE_FORMAT|yyyy-MM-dd|CST"), dateTimeFormatSpec);

    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("1:DAY"));

    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("one:DAYS:EPOCH"));
    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("EPOCH|DAYS|one"));

    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("1:DAY:EPOCH"));
    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("EPOCH|DAY"));

    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("1:DAY:EPOCH:yyyyMMdd"));
    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("EPOCH|yyyyMMdd"));

    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("1:DAY:SIMPLE_DATE_FORMAT:yyycMMdd"));
    assertThrows(IllegalArgumentException.class, () -> new DateTimeFormatSpec("SIMPLE_DATE_FORMAT|yyycMMdd"));
  }