public static ProducingResult createProducingResult(
            ResolvedSchema inputSchema, @Nullable Schema declaredSchema) {

        // no schema has been declared by the user,
        // the schema will be entirely derived from the input
        if (declaredSchema == null) {
            // go through data type to erase time attributes
            final DataType physicalDataType = inputSchema.toSourceRowDataType();
            final Schema schema = Schema.newBuilder().fromRowDataType(physicalDataType).build();
            return new ProducingResult(null, schema, null);
        }

        final List<UnresolvedColumn> declaredColumns = declaredSchema.getColumns();

        // the declared schema does not contain physical information,
        // thus, it only replaces physical columns with metadata rowtime or adds a primary key
        if (declaredColumns.stream().noneMatch(SchemaTranslator::isPhysical)) {
            // go through data type to erase time attributes
            final DataType sourceDataType = inputSchema.toSourceRowDataType();
            final DataType physicalDataType =
                    patchDataTypeWithoutMetadataRowtime(sourceDataType, declaredColumns);
            final Schema.Builder builder = Schema.newBuilder();
            builder.fromRowDataType(physicalDataType);
            builder.fromSchema(declaredSchema);
            return new ProducingResult(null, builder.build(), null);
        }

        return new ProducingResult(null, declaredSchema, null);
    }
@Test
    void testOutputToAtomicDataType() {
        final ResolvedSchema inputSchema = ResolvedSchema.of(Column.physical("a", INT()));

        final ProducingResult result =
                SchemaTranslator.createProducingResult(dataTypeFactory(), inputSchema, INT());

        assertThat(result.getProjections()).isEmpty();

        assertThat(result.getSchema()).isEqualTo(Schema.newBuilder().column("f0", INT()).build());

        assertThat(result.getPhysicalDataType()).hasValue(INT());
    }