Repository: apache/flink
File: flink-table/flink-table-common/src/test/java/org/apache/flink/table/api/SchemaTest.java (Java)

Code before:
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.api;

import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TestSchemaResolver;

import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

class SchemaTest {
    @Nested
    class Builder {
        @Test
        void testFromResolvedSchema() {
            ResolvedSchema originalSchema =
                    ResolvedSchema.of(
                            Column.physical("timestamp", DataTypes.TIMESTAMP_LTZ(3).notNull())
                                    .withComment("The timestamp when the event occurred."));
            Schema newSchema = Schema.newBuilder().fromResolvedSchema(originalSchema).build();

            assertThat(newSchema.resolve(new TestSchemaResolver())).isEqualTo(originalSchema);
        }
    }
}


Review comment (suggestion):
I suggest adding a test for the other 2 column types we have affected with the fix

Code after:
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.api;

import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TestSchemaResolver;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.expressions.utils.ResolvedExpressionMock;

import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;

import static org.assertj.core.api.Assertions.assertThat;

class SchemaTest {
    private static final String COMPUTED_SQL = "orig_ts - INTERVAL '60' MINUTE";

    private static final ResolvedExpression COMPUTED_COLUMN_RESOLVED =
            new ResolvedExpressionMock(DataTypes.TIMESTAMP(3), () -> COMPUTED_SQL);

    @Nested
    class Builder {
        @Test
        void testFromResolvedSchema() {
            ResolvedSchema originalSchema =
                    ResolvedSchema.of(
                            Column.physical("event_timestamp", DataTypes.TIMESTAMP_LTZ(3).notNull())
                                    .withComment("The timestamp when the event occurred."),
                            Column.computed("rowtime", COMPUTED_COLUMN_RESOLVED)
                                    .withComment("The rowtime derived from the event timestamp."),
                            Column.metadata("topic", DataTypes.STRING(), null, true)
                                    .withComment("kafka topic"));
            Schema newSchema = Schema.newBuilder().fromResolvedSchema(originalSchema).build();

            assertThat(newSchema.resolve(new TestSchemaResolver())).isEqualTo(originalSchema);
        }
    }
}