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 shouldCreateMetadataPolicyWhenTimestampFieldNotProvided() {
    // When:
    final TimestampExtractionPolicy result = TimestampExtractionPolicyFactory
        .create(
            ksqlConfig,
            schemaBuilder2.build(),
            Optional.empty()
        );

    // Then:
    assertThat(result, instanceOf(MetadataTimestampExtractionPolicy.class));
    final TimestampExtractor timestampExtractor = result.create(Optional.empty(), true, logger);
    assertThat(timestampExtractor, instanceOf(LoggingTimestampExtractor.class));
    assertThat(
        ((LoggingTimestampExtractor) timestampExtractor).getDelegate(),
        instanceOf(MetadataTimestampExtractor.class));
  }