class StructuredDataSource implements DataSource {

  @Override
  public Optional<String> canUpgradeTo(final DataSource other) {
    final List<String> issues = PROPERTIES.stream()
        .filter(prop -> !prop.isCompatible(this, other))
        .map(prop -> getCompatMessage(other, prop))
        .collect(Collectors.toList());

    checkSchemas(getSchema(), other.getSchema())
        .map(s -> getCompatMessage(other, SCHEMA_PROP) + ". (" + s + ")")
        .ifPresent(issues::add);

    final String err = String.join("\n\tAND ", issues);
    return err.isEmpty() ? Optional.empty() : Optional.of(err);
  }

    StructuredDataSource(
      final String sqlExpression,
      final SourceName dataSourceName,
      final LogicalSchema schema,
      final Optional<TimestampColumn> tsExtractionPolicy,
      final DataSourceType dataSourceType,
      final boolean casTarget,
      final KsqlTopic ksqlTopic
  );

}