class TypeMappingUtils {

	public static int[] computePhysicalIndicesOrTimeAttributeMarkers(
			TableSource<?> tableSource,
			List<TableColumn> logicalColumns,
			boolean streamMarkers,
			Function<String, String> nameRemapping) {
		Optional<String> proctimeAttribute = getProctimeAttribute(tableSource);
		List<String> rowtimeAttributes = getRowtimeAttributes(tableSource);

		List<TableColumn> columnsWithoutTimeAttributes = logicalColumns.stream().filter(col ->
			!rowtimeAttributes.contains(col.getName())
				&& proctimeAttribute.map(attr -> !attr.equals(col.getName())).orElse(true))
			.collect(Collectors.toList());

		Map<TableColumn, Integer> columnsToPhysicalIndices = TypeMappingUtils.computePhysicalIndices(
			columnsWithoutTimeAttributes.stream(),
			tableSource.getProducedDataType(),
			nameRemapping
		);

		return logicalColumns.stream().mapToInt(logicalColumn -> {
			if (proctimeAttribute.map(attr -> attr.equals(logicalColumn.getName())).orElse(false)) {
				verifyTimeAttributeType(logicalColumn, "Proctime");

				if (streamMarkers) {
					return TimeIndicatorTypeInfo.PROCTIME_STREAM_MARKER;
				} else {
					return TimeIndicatorTypeInfo.PROCTIME_BATCH_MARKER;
				}
			} else if (rowtimeAttributes.contains(logicalColumn.getName())) {
				verifyTimeAttributeType(logicalColumn, "Rowtime");

				if (streamMarkers) {
					return TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER;
				} else {
					return TimeIndicatorTypeInfo.ROWTIME_BATCH_MARKER;
				}
			} else {
				return columnsToPhysicalIndices.get(logicalColumn);
			}
		}).toArray();
	}

	private  TypeMappingUtils();

}