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();
    }
@Test
    void testMappingWithStreamTimeAttributes() {
        TestTableSource tableSource =
                new TestTableSource(
                        DataTypes.BIGINT(), Collections.singletonList("rowtime"), "proctime");
        int[] indices =
                TypeMappingUtils.computePhysicalIndicesOrTimeAttributeMarkers(
                        tableSource,
                        TableSchema.builder()
                                .field("a", Types.LONG)
                                .field("rowtime", Types.SQL_TIMESTAMP)
                                .field("proctime", Types.SQL_TIMESTAMP)
                                .build()
                                .getTableColumns(),
                        true,
                        Function.identity());

        assertThat(indices).isEqualTo(new int[] {0, -1, -2});
    }