class StructuredDataSource implements DataSource {

  @VisibleForTesting
  static Optional<String> checkSchemas(
      final LogicalSchema schema,
      final LogicalSchema other
  ) {
    if (!schema.key().equals(other.key())) {
      return Optional.of("Key columns must be identical.");
    }

    final ImmutableSet<Column> colA = ImmutableSet.copyOf(schema.columns());
    final ImmutableSet<Column> colB = ImmutableSet.copyOf(other.columns());

    final SetView<Column> difference = Sets.difference(colA, colB);
    if (!difference.isEmpty()) {
      return Optional.of("The following columns are changed, missing or reordered: " + difference);
    }

    return Optional.empty();
  }

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

}