This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a0c74c1e07030e7bca0c93e24a1e7643937371a7
Author: Timo Walther <twal...@apache.org>
AuthorDate: Mon Dec 6 17:05:36 2021 +0100
[hotfix][table-api-java] Migrate SchemaTranslatorTest to assertj
---
.../flink/table/catalog/SchemaTranslatorTest.java | 372 ++++++++++-----------
1 file changed, 176 insertions(+), 196 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaTranslatorTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaTranslatorTest.java
index c2c2e9f..3bfe128 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaTranslatorTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaTranslatorTest.java
@@ -20,7 +20,6 @@ package org.apache.flink.table.catalog;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.SchemaTranslator.ConsumingResult;
@@ -37,12 +36,18 @@ import java.time.DayOfWeek;
import java.util.Arrays;
import java.util.Optional;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsMessage;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
+import static org.apache.flink.table.api.DataTypes.DECIMAL;
+import static org.apache.flink.table.api.DataTypes.DOUBLE;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.apache.flink.table.api.DataTypes.TIMESTAMP_LTZ;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link SchemaTranslator}. */
public class SchemaTranslatorTest {
@@ -56,63 +61,53 @@ public class SchemaTranslatorTest {
SchemaTranslator.createConsumingResult(
dataTypeFactoryWithRawType(DayOfWeek.class),
inputTypeInfo, null);
- assertEquals(
- DataTypes.ROW(
- DataTypes.FIELD(
- "f0",
- DataTypes.ROW(
- DataTypes.FIELD("f0",
DataTypes.INT()),
- DataTypes.FIELD("f1",
DataTypes.BOOLEAN()))),
- DataTypes.FIELD(
- "f1",
DataTypeFactoryMock.dummyRaw(DayOfWeek.class)))
- .notNull(),
- result.getPhysicalDataType());
-
- assertTrue(result.isTopLevelRecord());
-
- assertEquals(
- Schema.newBuilder()
- .column(
- "f0",
- DataTypes.ROW(
- DataTypes.FIELD("f0", DataTypes.INT()),
- DataTypes.FIELD("f1",
DataTypes.BOOLEAN())))
- .column("f1",
DataTypeFactoryMock.dummyRaw(DayOfWeek.class))
- .build(),
- result.getSchema());
-
- assertNull(result.getProjections());
+ assertThat(result.getPhysicalDataType())
+ .isEqualTo(
+ ROW(
+ FIELD(
+ "f0",
+ ROW(FIELD("f0", INT()),
FIELD("f1", BOOLEAN()))),
+ FIELD("f1",
DataTypeFactoryMock.dummyRaw(DayOfWeek.class)))
+ .notNull());
+
+ assertThat(result.isTopLevelRecord()).isTrue();
+
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("f0", ROW(FIELD("f0", INT()),
FIELD("f1", BOOLEAN())))
+ .column("f1",
DataTypeFactoryMock.dummyRaw(DayOfWeek.class))
+ .build());
+
+ assertThat(result.getProjections()).isNull();
}
@Test
public void testOutputToRowDataType() {
final ResolvedSchema inputSchema =
ResolvedSchema.of(
- Column.physical("c", DataTypes.INT()),
- Column.physical("a", DataTypes.BOOLEAN()),
- Column.physical("B", DataTypes.DOUBLE())); //
case-insensitive mapping
+ Column.physical("c", INT()),
+ Column.physical("a", BOOLEAN()),
+ Column.physical("B", DOUBLE())); // case-insensitive
mapping
final DataType physicalDataType =
- DataTypes.ROW(
- DataTypes.FIELD("a", DataTypes.BOOLEAN()),
- DataTypes.FIELD("b", DataTypes.DOUBLE()),
- DataTypes.FIELD("c", DataTypes.INT()));
+ ROW(FIELD("a", BOOLEAN()), FIELD("b", DOUBLE()), FIELD("c",
INT()));
final ProducingResult result =
SchemaTranslator.createProducingResult(
dataTypeFactory(), inputSchema, physicalDataType);
- assertEquals(Optional.of(Arrays.asList("a", "B", "c")),
result.getProjections());
+ assertThat(result.getProjections()).hasValue(Arrays.asList("a", "B",
"c"));
- assertEquals(
- Schema.newBuilder()
- .column("a", DataTypes.BOOLEAN())
- .column("b", DataTypes.DOUBLE())
- .column("c", DataTypes.INT())
- .build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("a", BOOLEAN())
+ .column("b", DOUBLE())
+ .column("c", INT())
+ .build());
- assertEquals(Optional.of(physicalDataType),
result.getPhysicalDataType());
+ assertThat(result.getPhysicalDataType()).hasValue(physicalDataType);
}
@Test
@@ -123,30 +118,31 @@ public class SchemaTranslatorTest {
SchemaTranslator.createConsumingResult(
dataTypeFactoryWithRawType(Row.class), inputTypeInfo,
null);
- assertEquals(DataTypeFactoryMock.dummyRaw(Row.class),
result.getPhysicalDataType());
+
assertThat(result.getPhysicalDataType()).isEqualTo(DataTypeFactoryMock.dummyRaw(Row.class));
- assertFalse(result.isTopLevelRecord());
+ assertThat(result.isTopLevelRecord()).isFalse();
- assertEquals(
- Schema.newBuilder().column("f0",
DataTypeFactoryMock.dummyRaw(Row.class)).build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("f0",
DataTypeFactoryMock.dummyRaw(Row.class))
+ .build());
- assertNull(result.getProjections());
+ assertThat(result.getProjections()).isNull();
}
@Test
public void testOutputToAtomicDataType() {
- final ResolvedSchema inputSchema =
ResolvedSchema.of(Column.physical("a", DataTypes.INT()));
+ final ResolvedSchema inputSchema =
ResolvedSchema.of(Column.physical("a", INT()));
final ProducingResult result =
- SchemaTranslator.createProducingResult(
- dataTypeFactory(), inputSchema, DataTypes.INT());
+ SchemaTranslator.createProducingResult(dataTypeFactory(),
inputSchema, INT());
- assertEquals(Optional.empty(), result.getProjections());
+ assertThat(result.getProjections()).isEmpty();
- assertEquals(Schema.newBuilder().column("f0",
DataTypes.INT()).build(), result.getSchema());
+
assertThat(result.getSchema()).isEqualTo(Schema.newBuilder().column("f0",
INT()).build());
- assertEquals(Optional.of(DataTypes.INT()),
result.getPhysicalDataType());
+ assertThat(result.getPhysicalDataType()).hasValue(INT());
}
@Test
@@ -163,26 +159,22 @@ public class SchemaTranslatorTest {
.primaryKeyNamed("pk", "f0")
.build());
- assertEquals(
- DataTypes.ROW(
- DataTypes.FIELD("f0", DataTypes.INT()),
- DataTypes.FIELD("f1", DataTypes.BIGINT()))
- .notNull(),
- result.getPhysicalDataType());
-
- assertTrue(result.isTopLevelRecord());
-
- assertEquals(
- Schema.newBuilder()
- .column("f0", DataTypes.INT().notNull()) // not null
due to primary key
- .column("f1", DataTypes.BIGINT())
- .columnByExpression("computed", "f1 + 42")
- .columnByExpression("computed2", "f1 - 1")
- .primaryKeyNamed("pk", "f0")
- .build(),
- result.getSchema());
-
- assertNull(result.getProjections());
+ assertThat(result.getPhysicalDataType())
+ .isEqualTo(ROW(FIELD("f0", INT()), FIELD("f1",
BIGINT())).notNull());
+
+ assertThat(result.isTopLevelRecord()).isTrue();
+
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("f0", INT().notNull()) // not null due
to primary key
+ .column("f1", BIGINT())
+ .columnByExpression("computed", "f1 + 42")
+ .columnByExpression("computed2", "f1 - 1")
+ .primaryKeyNamed("pk", "f0")
+ .build());
+
+ assertThat(result.getProjections()).isNull();
}
@Test
@@ -196,38 +188,38 @@ public class SchemaTranslatorTest {
inputTypeInfo,
Schema.newBuilder()
.primaryKeyNamed("pk", "f0")
- .column("f1", DataTypes.BIGINT()) // reordered
- .column("f0", DataTypes.INT())
+ .column("f1", BIGINT()) // reordered
+ .column("f0", INT())
.columnByExpression("computed", "f1 + 42")
- .column("f2", DataTypes.DECIMAL(10, 2)) //
enriches
+ .column("f2", DECIMAL(10, 2)) // enriches
.columnByExpression("computed2", "f1 - 1")
.build());
- assertEquals(
- DataTypes.ROW(
- DataTypes.FIELD("f0", DataTypes.INT()),
- DataTypes.FIELD("f1", DataTypes.BIGINT()),
- DataTypes.FIELD("f2", DataTypes.DECIMAL(10,
2)),
- DataTypes.FIELD("f3", DataTypes.BOOLEAN()))
- .notNull(),
- result.getPhysicalDataType());
-
- assertTrue(result.isTopLevelRecord());
-
- assertEquals(
- Schema.newBuilder()
- .column("f0", DataTypes.INT())
- .column("f1", DataTypes.BIGINT())
- .column("f2", DataTypes.DECIMAL(10, 2))
- .column("f3", DataTypes.BOOLEAN())
- .columnByExpression("computed", "f1 + 42")
- .columnByExpression("computed2", "f1 - 1")
- .primaryKeyNamed("pk", "f0")
- .build(),
- result.getSchema());
-
- assertEquals(
- Arrays.asList("f1", "f0", "computed", "f2", "computed2"),
result.getProjections());
+ assertThat(result.getPhysicalDataType())
+ .isEqualTo(
+ ROW(
+ FIELD("f0", INT()),
+ FIELD("f1", BIGINT()),
+ FIELD("f2", DECIMAL(10, 2)),
+ FIELD("f3", BOOLEAN()))
+ .notNull());
+
+ assertThat(result.isTopLevelRecord()).isTrue();
+
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("f0", INT())
+ .column("f1", BIGINT())
+ .column("f2", DECIMAL(10, 2))
+ .column("f3", BOOLEAN())
+ .columnByExpression("computed", "f1 + 42")
+ .columnByExpression("computed2", "f1 - 1")
+ .primaryKeyNamed("pk", "f0")
+ .build());
+
+ assertThat(result.getProjections())
+ .isEqualTo(Arrays.asList("f1", "f0", "computed", "f2",
"computed2"));
}
@Test
@@ -240,153 +232,141 @@ public class SchemaTranslatorTest {
inputTypeInfo,
Schema.newBuilder()
.columnByExpression("f0_0", "f0.f0_0")
- .column(
- "f0",
- DataTypes.ROW(
- DataTypes.FIELD("f0_0",
DataTypes.INT()),
- DataTypes.FIELD("f0_1",
DataTypes.BOOLEAN())))
+ .column("f0", ROW(FIELD("f0_0", INT()),
FIELD("f0_1", BOOLEAN())))
+ .columnByExpression("f0_1", "f0.f0_1")
+ .build());
+
+ assertThat(result.getPhysicalDataType())
+ .isEqualTo(ROW(FIELD("f0_0", INT()), FIELD("f0_1",
BOOLEAN())));
+
+ assertThat(result.isTopLevelRecord()).isFalse();
+
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("f0", ROW(FIELD("f0_0", INT()),
FIELD("f0_1", BOOLEAN())))
+ .columnByExpression("f0_0", "f0.f0_0")
.columnByExpression("f0_1", "f0.f0_1")
.build());
- assertEquals(
- DataTypes.ROW(
- DataTypes.FIELD("f0_0", DataTypes.INT()),
- DataTypes.FIELD("f0_1", DataTypes.BOOLEAN())),
- result.getPhysicalDataType());
-
- assertFalse(result.isTopLevelRecord());
-
- assertEquals(
- Schema.newBuilder()
- .column(
- "f0",
- DataTypes.ROW(
- DataTypes.FIELD("f0_0",
DataTypes.INT()),
- DataTypes.FIELD("f0_1",
DataTypes.BOOLEAN())))
- .columnByExpression("f0_0", "f0.f0_0")
- .columnByExpression("f0_1", "f0.f0_1")
- .build(),
- result.getSchema());
-
- assertEquals(Arrays.asList("f0_0", "f0", "f0_1"),
result.getProjections());
+ assertThat(result.getProjections()).isEqualTo(Arrays.asList("f0_0",
"f0", "f0_1"));
}
@Test
public void testInvalidDeclaredSchemaColumn() {
final TypeInformation<?> inputTypeInfo = Types.ROW(Types.INT,
Types.LONG);
- try {
- SchemaTranslator.createConsumingResult(
- dataTypeFactory(),
- inputTypeInfo,
- Schema.newBuilder().column("INVALID",
DataTypes.BIGINT()).build());
- } catch (ValidationException e) {
- assertThat(
- e,
- containsMessage(
- "Unable to find a field named 'INVALID' in the
physical data type"));
- }
+ assertThatThrownBy(
+ () ->
+ SchemaTranslator.createConsumingResult(
+ dataTypeFactory(),
+ inputTypeInfo,
+ Schema.newBuilder().column("INVALID",
BIGINT()).build()))
+ .satisfies(
+ anyCauseMatches(
+ ValidationException.class,
+ "Unable to find a field named 'INVALID' in the
physical data type"));
}
@Test
public void testOutputToNoSchema() {
final ResolvedSchema tableSchema =
ResolvedSchema.of(
- Column.physical("id", DataTypes.BIGINT()),
- Column.metadata("rowtime", DataTypes.TIMESTAMP_LTZ(3),
null, false),
- Column.physical("name", DataTypes.STRING()));
+ Column.physical("id", BIGINT()),
+ Column.metadata("rowtime", TIMESTAMP_LTZ(3), null,
false),
+ Column.physical("name", STRING()));
final ProducingResult result =
SchemaTranslator.createProducingResult(tableSchema, null);
- assertEquals(Optional.empty(), result.getProjections());
+ assertThat(result.getProjections()).isEmpty();
- assertEquals(
- Schema.newBuilder()
- .column("id", DataTypes.BIGINT())
- .column("rowtime", DataTypes.TIMESTAMP_LTZ(3)) //
becomes physical
- .column("name", DataTypes.STRING())
- .build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("id", BIGINT())
+ .column("rowtime", TIMESTAMP_LTZ(3)) //
becomes physical
+ .column("name", STRING())
+ .build());
- assertEquals(Optional.empty(), result.getPhysicalDataType());
+ assertThat(result.getPhysicalDataType()).isEmpty();
}
@Test
public void testOutputToEmptySchema() {
final ResolvedSchema tableSchema =
ResolvedSchema.of(
- Column.physical("id", DataTypes.BIGINT()),
- Column.metadata("rowtime", DataTypes.TIMESTAMP_LTZ(3),
null, false),
- Column.physical("name", DataTypes.STRING()));
+ Column.physical("id", BIGINT()),
+ Column.metadata("rowtime", TIMESTAMP_LTZ(3), null,
false),
+ Column.physical("name", STRING()));
final ProducingResult result =
SchemaTranslator.createProducingResult(tableSchema,
Schema.derived());
- assertEquals(Optional.empty(), result.getProjections());
+ assertThat(result.getProjections()).isEmpty();
- assertEquals(
- Schema.newBuilder()
- .column("id", DataTypes.BIGINT())
- .column("rowtime", DataTypes.TIMESTAMP_LTZ(3)) //
becomes physical
- .column("name", DataTypes.STRING())
- .build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("id", BIGINT())
+ .column("rowtime", TIMESTAMP_LTZ(3)) //
becomes physical
+ .column("name", STRING())
+ .build());
- assertEquals(Optional.empty(), result.getPhysicalDataType());
+ assertThat(result.getPhysicalDataType()).isEmpty();
}
@Test
public void testOutputToPartialSchema() {
final ResolvedSchema tableSchema =
ResolvedSchema.of(
- Column.physical("id", DataTypes.BIGINT().notNull()),
- Column.physical("name", DataTypes.STRING()),
- Column.metadata("rowtime", DataTypes.TIMESTAMP_LTZ(3),
null, false));
+ Column.physical("id", BIGINT().notNull()),
+ Column.physical("name", STRING()),
+ Column.metadata("rowtime", TIMESTAMP_LTZ(3), null,
false));
final ProducingResult result =
SchemaTranslator.createProducingResult(
tableSchema,
Schema.newBuilder()
.columnByExpression("computed", "f1 + 42")
- .columnByMetadata("rowtime",
DataTypes.TIMESTAMP_LTZ(3))
+ .columnByMetadata("rowtime", TIMESTAMP_LTZ(3))
.primaryKey("id")
.build());
- assertEquals(
- Schema.newBuilder()
- .column("id", DataTypes.BIGINT().notNull())
- .column("name", DataTypes.STRING())
- .columnByExpression("computed", "f1 + 42")
- .columnByMetadata("rowtime",
DataTypes.TIMESTAMP_LTZ(3)) // becomes metadata
- .primaryKey("id")
- .build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("id", BIGINT().notNull())
+ .column("name", STRING())
+ .columnByExpression("computed", "f1 + 42")
+ .columnByMetadata("rowtime", TIMESTAMP_LTZ(3))
// becomes metadata
+ .primaryKey("id")
+ .build());
}
@Test
public void testOutputToDeclaredSchema() {
final ResolvedSchema tableSchema =
ResolvedSchema.of(
- Column.physical("id", DataTypes.BIGINT()),
- Column.physical("rowtime", DataTypes.TIMESTAMP_LTZ(3)),
- Column.physical("name", DataTypes.STRING()));
+ Column.physical("id", BIGINT()),
+ Column.physical("rowtime", TIMESTAMP_LTZ(3)),
+ Column.physical("name", STRING()));
final ProducingResult result =
SchemaTranslator.createProducingResult(
tableSchema,
Schema.newBuilder()
- .column("id", DataTypes.BIGINT())
- .columnByMetadata("rowtime",
DataTypes.TIMESTAMP_LTZ(3))
- .column("name",
DataTypes.STRING().bridgedTo(StringData.class))
+ .column("id", BIGINT())
+ .columnByMetadata("rowtime", TIMESTAMP_LTZ(3))
+ .column("name",
STRING().bridgedTo(StringData.class))
.build());
- assertEquals(
- Schema.newBuilder()
- .column("id", DataTypes.BIGINT())
- .columnByMetadata("rowtime",
DataTypes.TIMESTAMP_LTZ(3))
- .column("name",
DataTypes.STRING().bridgedTo(StringData.class))
- .build(),
- result.getSchema());
+ assertThat(result.getSchema())
+ .isEqualTo(
+ Schema.newBuilder()
+ .column("id", BIGINT())
+ .columnByMetadata("rowtime", TIMESTAMP_LTZ(3))
+ .column("name",
STRING().bridgedTo(StringData.class))
+ .build());
}
private static DataTypeFactory dataTypeFactoryWithRawType(Class<?>
rawType) {
