public static TimestampExtractionPolicy create(
      final KsqlConfig ksqlConfig,
      final LogicalSchema schema,
      final Optional<TimestampColumn> timestampColumn
  ) {
    if (!timestampColumn.isPresent()) {
      return new MetadataTimestampExtractionPolicy(getDefaultTimestampExtractor(ksqlConfig));
    }

    final ColumnName col = timestampColumn.get().getColumn();
    final Optional<String> timestampFormat = timestampColumn.get().getFormat();

    final Column column = schema.findColumn(col)
        .orElseThrow(() -> new KsqlException(
            "The TIMESTAMP column set in the WITH clause does not exist in the schema: '"
                + col.toString(FormatOptions.noEscape()) + "'"));

    final SqlBaseType tsColumnType = column.type().baseType();
    if (tsColumnType == SqlBaseType.STRING) {
      final String format = timestampFormat.orElseThrow(() -> new KsqlException(
          "A String timestamp field has been specified without"
              + " also specifying the "
              + CommonCreateConfigs.TIMESTAMP_FORMAT_PROPERTY.toLowerCase()));

      return new StringTimestampExtractionPolicy(col, format);
    }

    if (timestampFormat.isPresent()) {
      throw new KsqlException("'" + CommonCreateConfigs.TIMESTAMP_FORMAT_PROPERTY
          + "' set in the WITH clause can only be used "
          + "when the timestamp column is of type STRING.");
    }

    if (tsColumnType == SqlBaseType.BIGINT) {
      return new LongColumnTimestampExtractionPolicy(col);
    }

    if (tsColumnType == SqlBaseType.TIMESTAMP) {
      return new TimestampColumnTimestampExtractionPolicy(col);
    }

    throw new KsqlException(
        "Timestamp column, " + col + ", should be LONG(INT64), TIMESTAMP,"
            + " or a String with a "
            + CommonCreateConfigs.TIMESTAMP_FORMAT_PROPERTY.toLowerCase()
            + " specified.");
  }
@Test
  public void shouldThrowIfLongTimestampTypeAndFormatIsSupplied() {
    // Given:
    final String timestamp = "timestamp";
    final LogicalSchema schema = schemaBuilder2
        .valueColumn(ColumnName.of(timestamp.toUpperCase()), SqlTypes.BIGINT)
        .build();

    // When:
    assertThrows(
        KsqlException.class,
        () -> TimestampExtractionPolicyFactory
            .create(ksqlConfig,
                schema,
                Optional.of(
                    new TimestampColumn(
                        ColumnName.of(timestamp.toUpperCase()),
                        Optional.of("b")
                    )
                )
            )
    );
  }