This is an automated email from the ASF dual-hosted git repository. mapohl pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit d0b7a03432dfb6f9318fc716c15d395cc469e26c Author: Alexander Preuß <[email protected]> AuthorDate: Tue Apr 12 15:09:52 2022 +0200 [FLINK-27527][test-utils] Create a file-based upsert sink connector for testing internal components --- flink-test-utils-parent/flink-test-utils/pom.xml | 7 + .../upserttest/sink/ImmutableByteArrayWrapper.java | 67 ++++++++ .../upserttest/sink/UpsertTestFileUtil.java | 174 +++++++++++++++++++++ .../connector/upserttest/sink/UpsertTestSink.java | 73 +++++++++ .../upserttest/sink/UpsertTestSinkBuilder.java | 98 ++++++++++++ .../upserttest/sink/UpsertTestSinkWriter.java | 82 ++++++++++ .../table/UpsertTestConnectorOptions.java | 53 +++++++ .../table/UpsertTestDynamicTableSink.java | 119 ++++++++++++++ .../table/UpsertTestDynamicTableSinkFactory.java | 85 ++++++++++ .../org.apache.flink.table.factories.Factory | 19 +++ flink-tests/pom.xml | 15 ++ .../sink/ImmutableByteArrayWrapperTest.java | 54 +++++++ .../upserttest/sink/UpsertTestSinkBuilderTest.java | 83 ++++++++++ .../sink/UpsertTestSinkWriterITCase.java | 119 ++++++++++++++ .../table/UpsertTestDynamicTableSinkITCase.java | 96 ++++++++++++ .../completeness/TypeInfoTestCoverageTest.java | 37 ++++- .../TypeSerializerTestCoverageTest.java | 91 ++++++++++- 17 files changed, 1268 insertions(+), 4 deletions(-) diff --git a/flink-test-utils-parent/flink-test-utils/pom.xml b/flink-test-utils-parent/flink-test-utils/pom.xml index b16df155738..70d01c408d1 100644 --- a/flink-test-utils-parent/flink-test-utils/pom.xml +++ b/flink-test-utils-parent/flink-test-utils/pom.xml @@ -35,6 +35,13 @@ under the License. <dependencies> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-common</artifactId> + <version>${project.version}</version> + <scope>compile</scope> + </dependency> + <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-test-utils-junit</artifactId> diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapper.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapper.java new file mode 100644 index 00000000000..d772af15ace --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapper.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.annotation.VisibleForTesting; + +import java.util.Arrays; +import java.util.Map; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * This class is a slim wrapper around {@code byte[]} that defensively copies the array during + * construction and retrieval to prevent its contents from changing. It also implements a + * content-dependent {@link #hashCode} method to allow usage as keys in a {@link Map}. + */ +class ImmutableByteArrayWrapper { + + @VisibleForTesting final byte[] bytes; + + ImmutableByteArrayWrapper(byte[] bytes) { + checkNotNull(bytes); + this.bytes = bytes.clone(); + } + + /** + * Returns a reference-free copy of the underlying byte[]. + * + * @return the copied byte[] + */ + byte[] array() { + return bytes.clone(); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ImmutableByteArrayWrapper that = (ImmutableByteArrayWrapper) o; + return Arrays.equals(bytes, that.bytes); + } + + @Override + public int hashCode() { + return Arrays.hashCode(bytes); + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestFileUtil.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestFileUtil.java new file mode 100644 index 00000000000..fca20a43fe5 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestFileUtil.java @@ -0,0 +1,174 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.DeserializationSchema; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.IOException; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** Collection uf utility methods for reading and writing files from the {@link UpsertTestSink}. */ +@Internal +public class UpsertTestFileUtil { + static final byte MAGIC_BYTE = 13; + + private UpsertTestFileUtil() {} + + /** + * Writes a Map of records serialized by the {@link UpsertTestSinkWriter} to the given + * BufferedOutputStream. + * + * @param bos the BufferedOutputStream to write to + * @param records the Map of records created by the UpsertTestSinkWriter + * @throws IOException + */ + public static void writeRecords( + BufferedOutputStream bos, + Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> records) + throws IOException { + checkNotNull(bos); + for (Map.Entry<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> record : + records.entrySet()) { + byte[] key = record.getKey().array(); + byte[] value = record.getValue().array(); + bos.write(MAGIC_BYTE); + bos.write(key.length); + bos.write(key); + bos.write(value.length); + bos.write(value); + } + bos.flush(); + } + + /** + * Returns the total number of records written using the {@link UpsertTestSinkWriter} to the + * given File. + * + * @param bis The BufferedInputStream to read from + * @return the number of records + * @throws IOException + */ + public static int getNumberOfRecords(BufferedInputStream bis) throws IOException { + Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> records = readRecords(bis); + return records.size(); + } + + /** + * Returns the total number of records written using the {@link UpsertTestSinkWriter} to the + * given File. + * + * @param file The File to read from + * @return the number of records + * @throws IOException + */ + public static int getNumberOfRecords(File file) throws IOException { + checkNotNull(file); + FileInputStream fs = new FileInputStream(file); + BufferedInputStream bis = new BufferedInputStream(fs); + return getNumberOfRecords(bis); + } + + /** + * Reads records that were written using the {@link UpsertTestSinkWriter} from the given + * InputStream and converts them using the provided {@link DeserializationSchema}s. + * + * @param bis The BufferedInputStream to read from + * @param keyDeserializationSchema The key's DeserializationSchema + * @param valueDeserializationSchema The value's DeserializationSchema + * @return Map containing the deserialized key-value pairs + * @throws IOException + */ + public static <K, V> Map<K, V> readRecords( + BufferedInputStream bis, + DeserializationSchema<K> keyDeserializationSchema, + DeserializationSchema<V> valueDeserializationSchema) + throws IOException { + checkNotNull(bis); + Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> bytesMap = readRecords(bis); + Map<K, V> typedMap = new HashMap<>(bytesMap.size()); + + Iterator<Map.Entry<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper>> it = + bytesMap.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> entry = it.next(); + K key = keyDeserializationSchema.deserialize(entry.getKey().array()); + V value = valueDeserializationSchema.deserialize(entry.getValue().array()); + typedMap.put(key, value); + } + return typedMap; + } + + /** + * Reads records that were written using the {@link UpsertTestSinkWriter} from the given File + * and converts them using the provided {@link DeserializationSchema}s. + * + * @param file The File to read from + * @param keyDeserializationSchema The key's DeserializationSchema + * @param valueDeserializationSchema The value's DeserializationSchema + * @return Map containing the deserialized key-value pairs + * @throws IOException + */ + public static <K, V> Map<K, V> readRecords( + File file, + DeserializationSchema<K> keyDeserializationSchema, + DeserializationSchema<V> valueDeserializationSchema) + throws IOException { + checkNotNull(file); + FileInputStream fs = new FileInputStream(file); + BufferedInputStream bis = new BufferedInputStream(fs); + return readRecords(bis, keyDeserializationSchema, valueDeserializationSchema); + } + + /** + * Reads records that were written using the {@link UpsertTestSinkWriter} from the given + * InputStream. + * + * @param bis The BufferedInputStream to read from + * @return Map containing the read ImmutableByteArrayWrapper key-value pairs + * @throws IOException + */ + private static Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> readRecords( + BufferedInputStream bis) throws IOException { + checkNotNull(bis); + Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> records = new HashMap<>(); + int magicByte; + while ((magicByte = bis.read()) != -1) { + if (magicByte != MAGIC_BYTE) { + throw new IOException("Data was serialized incorrectly or is corrupted."); + } + int keyLength = bis.read(); + byte[] key = new byte[keyLength]; + bis.read(key); + int valueLength = bis.read(); + byte[] value = new byte[valueLength]; + bis.read(value); + records.put(new ImmutableByteArrayWrapper(key), new ImmutableByteArrayWrapper(value)); + } + return records; + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java new file mode 100644 index 00000000000..27b57187a7a --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.api.connector.sink2.Sink; +import org.apache.flink.api.connector.sink2.SinkWriter; + +import java.io.File; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Flink Sink to upsert test data into a file. This Sink is intended for testing internal + * functionality and is **not** production-ready. + * + * <p>Please note that the UpsertTestSink needs to run with a parallelism of 1 to function + * correctly. There is currently no support for using multiple writers at once. + * + * @param <IN> type of records written to the file + * @see UpsertTestSinkBuilder on how to construct an UpsertTestSink + */ +@PublicEvolving +public class UpsertTestSink<IN> implements Sink<IN> { + + private final File outputFile; + private final SerializationSchema<IN> keySerializationSchema; + private final SerializationSchema<IN> valueSerializationSchema; + + UpsertTestSink( + File outputFile, + SerializationSchema<IN> keySerializationSchema, + SerializationSchema<IN> valueSerializationSchema) { + this.outputFile = checkNotNull(outputFile); + this.keySerializationSchema = checkNotNull(keySerializationSchema); + this.valueSerializationSchema = checkNotNull(valueSerializationSchema); + } + + /** + * Create a {@link UpsertTestSinkBuilder} to construct a new {@link UpsertTestSink}. + * + * @param <IN> type of incoming records + * @return {@link UpsertTestSinkBuilder} + */ + public static <IN> UpsertTestSinkBuilder<IN> builder() { + return new UpsertTestSinkBuilder<>(); + } + + @Internal + @Override + public SinkWriter<IN> createWriter(InitContext context) { + return new UpsertTestSinkWriter<>( + outputFile, keySerializationSchema, valueSerializationSchema); + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilder.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilder.java new file mode 100644 index 00000000000..daf8b93d50b --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilder.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.api.common.serialization.SerializationSchema; + +import java.io.File; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * Builder to construct {@link UpsertTestSink}. + * + * <p>The following example shows the minimum setup to create a UpsertTestSink that writes {@code + * Tuple2<String, String>} values to a file. + * + * <pre>{@code + * UpsertTestSink<Tuple2<String, String>> sink = UpsertTestSink + * .<Tuple2<String, String>>builder + * .setOutputFile(MY_OUTPUT_FILE) + * .setKeySerializationSchema(MY_KEY_SERIALIZER) + * .setValueSerializationSchema(MY_VALUE_SERIALIZER) + * .build(); + * }</pre> + * + * @param <IN> type of the records written to the file + */ +@PublicEvolving +public class UpsertTestSinkBuilder<IN> { + + private File outputFile; + private SerializationSchema<IN> keySerializationSchema; + private SerializationSchema<IN> valueSerializationSchema; + + /** + * Sets the output {@link File} to write to. + * + * @param outputFile + * @return {@link UpsertTestSinkBuilder} + */ + public UpsertTestSinkBuilder<IN> setOutputFile(File outputFile) { + this.outputFile = checkNotNull(outputFile); + return this; + } + + /** + * Sets the key {@link SerializationSchema} that transforms incoming records to byte[]. + * + * @param keySerializationSchema + * @return {@link UpsertTestSinkBuilder} + */ + public UpsertTestSinkBuilder<IN> setKeySerializationSchema( + SerializationSchema<IN> keySerializationSchema) { + this.keySerializationSchema = checkNotNull(keySerializationSchema); + return this; + } + + /** + * Sets the value {@link SerializationSchema} that transforms incoming records to byte[]. + * + * @param valueSerializationSchema + * @return {@link UpsertTestSinkBuilder} + */ + public UpsertTestSinkBuilder<IN> setValueSerializationSchema( + SerializationSchema<IN> valueSerializationSchema) { + this.valueSerializationSchema = checkNotNull(valueSerializationSchema); + return this; + } + + /** + * Constructs the {@link UpsertTestSink} with the configured properties. + * + * @return {@link UpsertTestSink} + */ + public UpsertTestSink<IN> build() { + checkNotNull(outputFile); + checkNotNull(keySerializationSchema); + checkNotNull(valueSerializationSchema); + return new UpsertTestSink<>(outputFile, keySerializationSchema, valueSerializationSchema); + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriter.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriter.java new file mode 100644 index 00000000000..c6c8889e951 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriter.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.api.connector.sink2.SinkWriter; +import org.apache.flink.util.FlinkRuntimeException; + +import java.io.BufferedOutputStream; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * This class is responsible for writing records into a file in an upsert-fashion. On every + * checkpoint each key-value pair currently in the map is written to the file. + * + * @param <IN> The type of the input elements. + */ +class UpsertTestSinkWriter<IN> implements SinkWriter<IN> { + + private final SerializationSchema<IN> keySerializationSchema; + private final SerializationSchema<IN> valueSerializationSchema; + private final Map<ImmutableByteArrayWrapper, ImmutableByteArrayWrapper> records = + new HashMap<>(); + private final BufferedOutputStream bufferedOutputStream; + + UpsertTestSinkWriter( + File outputFile, + SerializationSchema<IN> keySerializationSchema, + SerializationSchema<IN> valueSerializationSchema) { + this.keySerializationSchema = checkNotNull(keySerializationSchema); + this.valueSerializationSchema = checkNotNull(valueSerializationSchema); + checkNotNull(outputFile); + try { + this.bufferedOutputStream = + new BufferedOutputStream(new FileOutputStream(outputFile, true)); + } catch (FileNotFoundException e) { + throw new FlinkRuntimeException("Could not find file", e); + } + } + + @Override + public void write(IN element, Context context) { + byte[] key = keySerializationSchema.serialize(element); + byte[] value = valueSerializationSchema.serialize(element); + records.put(new ImmutableByteArrayWrapper(key), new ImmutableByteArrayWrapper(value)); + } + + @Override + public void flush(boolean endOfInput) throws IOException { + UpsertTestFileUtil.writeRecords(bufferedOutputStream, records); + records.clear(); + } + + @Override + public void close() throws Exception { + flush(true); + bufferedOutputStream.close(); + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestConnectorOptions.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestConnectorOptions.java new file mode 100644 index 00000000000..1587d382a65 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestConnectorOptions.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.table; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.table.factories.FactoryUtil; + +import static org.apache.flink.configuration.ConfigOptions.key; + +/** Options for the UpsertTestConnector. */ +@PublicEvolving +public class UpsertTestConnectorOptions { + + public static final ConfigOption<String> OUTPUT_FILEPATH_OPTION = + key("output-filepath") + .stringType() + .noDefaultValue() + .withDescription("The filepath for the output file."); + + public static final ConfigOption<String> KEY_FORMAT_OPTION = + ConfigOptions.key("key" + FactoryUtil.FORMAT_SUFFIX) + .stringType() + .noDefaultValue() + .withDescription( + "Defines the format identifier for encoding key data. " + + "The identifier is used to discover a suitable format factory."); + + public static final ConfigOption<String> VALUE_FORMAT_OPTION = + ConfigOptions.key("value" + FactoryUtil.FORMAT_SUFFIX) + .stringType() + .noDefaultValue() + .withDescription( + "Defines the format identifier for encoding value data. " + + "The identifier is used to discover a suitable format factory."); +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSink.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSink.java new file mode 100644 index 00000000000..bc1e41b0806 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSink.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.connector.upserttest.sink.UpsertTestSink; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.connector.format.EncodingFormat; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.connector.sink.SinkV2Provider; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.types.DataType; +import org.apache.flink.types.RowKind; + +import java.io.File; +import java.util.Objects; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A {@link DynamicTableSink} that describes how to create a {@link UpsertTestSink} from a logical + * description. + */ +@Internal +class UpsertTestDynamicTableSink implements DynamicTableSink { + + private final DataType physicalRowDataType; + private final EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat; + private final EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat; + private final String outputFilePath; + + UpsertTestDynamicTableSink( + DataType physicalRowDataType, + EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat, + EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat, + String outputFilePath) { + this.physicalRowDataType = checkNotNull(physicalRowDataType); + this.keyEncodingFormat = checkNotNull(keyEncodingFormat); + this.valueEncodingFormat = checkNotNull(valueEncodingFormat); + this.outputFilePath = checkNotNull(outputFilePath); + } + + @Override + public ChangelogMode getChangelogMode(ChangelogMode requestedMode) { + ChangelogMode.Builder builder = ChangelogMode.newBuilder(); + for (RowKind kind : requestedMode.getContainedKinds()) { + if (kind != RowKind.UPDATE_BEFORE) { + builder.addContainedKind(kind); + } + } + return builder.build(); + } + + @Override + public SinkRuntimeProvider getSinkRuntimeProvider(Context context) { + final File outputFile = new File(outputFilePath); + final SerializationSchema<RowData> keySerialization = + keyEncodingFormat.createRuntimeEncoder(context, physicalRowDataType); + final SerializationSchema<RowData> valueSerialization = + valueEncodingFormat.createRuntimeEncoder(context, physicalRowDataType); + + final UpsertTestSink<RowData> sink = + UpsertTestSink.<RowData>builder() + .setOutputFile(outputFile) + .setKeySerializationSchema(keySerialization) + .setValueSerializationSchema(valueSerialization) + .build(); + return SinkV2Provider.of(sink, 1); + } + + @Override + public DynamicTableSink copy() { + return new UpsertTestDynamicTableSink( + physicalRowDataType, keyEncodingFormat, valueEncodingFormat, outputFilePath); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + UpsertTestDynamicTableSink that = (UpsertTestDynamicTableSink) o; + return Objects.equals(physicalRowDataType, that.physicalRowDataType) + && Objects.equals(keyEncodingFormat, that.keyEncodingFormat) + && Objects.equals(valueEncodingFormat, that.valueEncodingFormat) + && Objects.equals(outputFilePath, that.outputFilePath); + } + + @Override + public int hashCode() { + return Objects.hash( + physicalRowDataType, keyEncodingFormat, valueEncodingFormat, outputFilePath); + } + + @Override + public String asSummaryString() { + return "UpsertTestSink"; + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkFactory.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkFactory.java new file mode 100644 index 00000000000..bd524c99340 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkFactory.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.table; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.connector.format.EncodingFormat; +import org.apache.flink.table.connector.sink.DynamicTableSink; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.factories.DynamicTableSinkFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.factories.SerializationFormatFactory; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.apache.flink.connector.upserttest.table.UpsertTestConnectorOptions.KEY_FORMAT_OPTION; +import static org.apache.flink.connector.upserttest.table.UpsertTestConnectorOptions.OUTPUT_FILEPATH_OPTION; +import static org.apache.flink.connector.upserttest.table.UpsertTestConnectorOptions.VALUE_FORMAT_OPTION; + +/** A {@link DynamicTableSinkFactory} for discovering {@link UpsertTestDynamicTableSink}. */ +@Internal +public class UpsertTestDynamicTableSinkFactory implements DynamicTableSinkFactory { + public static final String IDENTIFIER = "upsert-files"; + + @Override + public DynamicTableSink createDynamicTableSink(Context context) { + final FactoryUtil.TableFactoryHelper helper = + FactoryUtil.createTableFactoryHelper(this, context); + helper.validate(); + + EncodingFormat<SerializationSchema<RowData>> keyEncodingFormat = + helper.discoverEncodingFormat(SerializationFormatFactory.class, KEY_FORMAT_OPTION); + EncodingFormat<SerializationSchema<RowData>> valueEncodingFormat = + helper.discoverEncodingFormat( + SerializationFormatFactory.class, VALUE_FORMAT_OPTION); + + final ReadableConfig tableOptions = helper.getOptions(); + final String outputFilePath = tableOptions.get(OUTPUT_FILEPATH_OPTION); + + return new UpsertTestDynamicTableSink( + context.getPhysicalRowDataType(), + keyEncodingFormat, + valueEncodingFormat, + outputFilePath); + } + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + final Set<ConfigOption<?>> options = new HashSet<>(); + options.add(OUTPUT_FILEPATH_OPTION); + options.add(KEY_FORMAT_OPTION); + options.add(VALUE_FORMAT_OPTION); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } +} diff --git a/flink-test-utils-parent/flink-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-test-utils-parent/flink-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory new file mode 100644 index 00000000000..0d38495e6c7 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory @@ -0,0 +1,19 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# 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. +# + +org.apache.flink.connector.upserttest.table.UpsertTestDynamicTableSinkFactory diff --git a/flink-tests/pom.xml b/flink-tests/pom.xml index dab6ed70b00..592d0a8cd90 100644 --- a/flink-tests/pom.xml +++ b/flink-tests/pom.xml @@ -43,6 +43,13 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-test-utils</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-files</artifactId> @@ -252,6 +259,14 @@ under the License. <scope>test</scope> </dependency> + <!-- For UpsertTestDynamicTableSinkITCase --> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-json</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <!-- utility to scan classpaths --> <dependency> <groupId>org.reflections</groupId> diff --git a/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapperTest.java b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapperTest.java new file mode 100644 index 00000000000..8b201ece9e8 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/ImmutableByteArrayWrapperTest.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.util.TestLoggerExtension; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit test for {@link ImmutableByteArrayWrapper}. */ +@ExtendWith(TestLoggerExtension.class) +class ImmutableByteArrayWrapperTest { + + @Test + void testConstructorCopy() { + byte[] array = "immutability of constructor".getBytes(); + byte[] clonedArray = new ImmutableByteArrayWrapper(array).bytes; + assertCopyIsReferenceFree(array, clonedArray); + } + + @Test + void testGetterCopy() { + byte[] array = "immutability of getter".getBytes(); + byte[] clonedArray = new ImmutableByteArrayWrapper(array).array(); + assertCopyIsReferenceFree(array, clonedArray); + } + + private static void assertCopyIsReferenceFree(byte[] original, byte[] clone) { + assertThat(clone).isNotSameAs(original); + assertThat(clone).isEqualTo(original); + Arrays.fill(original, (byte) 0); + assertThat(clone).isNotEqualTo(original); + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilderTest.java b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilderTest.java new file mode 100644 index 00000000000..4a3d82e1409 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkBuilderTest.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.util.TestLoggerExtension; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; + +import static org.assertj.core.api.Assertions.assertThatNoException; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link UpsertTestSinkBuilder}. */ +@ExtendWith(TestLoggerExtension.class) +class UpsertTestSinkBuilderTest { + + @TempDir static File tempFile; + + @Test + void testValidBuilder() { + SerializationSchema<String> dummySerializationSchema = element -> null; + UpsertTestSinkBuilder<String> builder = + new UpsertTestSinkBuilder<String>() + .setOutputFile(tempFile) + .setKeySerializationSchema(dummySerializationSchema) + .setValueSerializationSchema(dummySerializationSchema); + + assertThatNoException().isThrownBy(builder::build); + } + + @Test + void testThrowIfFileNotSet() { + SerializationSchema<String> dummySerializationSchema = element -> null; + UpsertTestSinkBuilder<String> builder = + new UpsertTestSinkBuilder<String>() + .setKeySerializationSchema(dummySerializationSchema) + .setValueSerializationSchema(dummySerializationSchema); + + assertThatThrownBy(builder::build).isInstanceOf(NullPointerException.class); + } + + @Test + void testThrowIfKeySerializerNotSet() { + SerializationSchema<String> dummySerializationSchema = element -> null; + UpsertTestSinkBuilder<String> builder = + new UpsertTestSinkBuilder<String>() + .setOutputFile(tempFile) + .setValueSerializationSchema(dummySerializationSchema); + + assertThatThrownBy(builder::build).isInstanceOf(NullPointerException.class); + } + + @Test + void testThrowIfValueSerializerNotSet() { + SerializationSchema<String> dummySerializationSchema = element -> null; + UpsertTestSinkBuilder<String> builder = + new UpsertTestSinkBuilder<String>() + .setOutputFile(tempFile) + .setValueSerializationSchema(dummySerializationSchema); + + assertThatThrownBy(builder::build).isInstanceOf(NullPointerException.class); + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriterITCase.java b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriterITCase.java new file mode 100644 index 00000000000..3493a4dec53 --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/sink/UpsertTestSinkWriterITCase.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.sink; + +import org.apache.flink.api.common.serialization.DeserializationSchema; +import org.apache.flink.api.common.serialization.SerializationSchema; +import org.apache.flink.api.common.serialization.SimpleStringSchema; +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.util.TestLoggerExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link UpsertTestSinkWriter}. */ +@ExtendWith(TestLoggerExtension.class) +class UpsertTestSinkWriterITCase { + + @RegisterExtension + public static final MiniClusterExtension MINI_CLUSTER_RESOURCE = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(1) + .build()); + + @TempDir private File tempDir; + private File outputFile; + private UpsertTestSinkWriter<Tuple2<String, String>> writer; + private List<Tuple2<String, String>> expectedRecords; + + @BeforeEach + void setup() { + outputFile = new File(tempDir, "records.out"); + writer = createSinkWriter(outputFile); + expectedRecords = writeTestData(writer); + } + + @AfterEach + void tearDown() throws Exception { + writer.close(); + } + + @Test + void testWrite() throws Exception { + writer.close(); + testRecordPresence(outputFile, expectedRecords); + } + + @Test + void testWriteOnCheckpoint() throws Exception { + writer.flush(false); + testRecordPresence(outputFile, expectedRecords); + } + + private UpsertTestSinkWriter<Tuple2<String, String>> createSinkWriter(File outputFile) { + SerializationSchema<Tuple2<String, String>> keySerializationSchema = + element -> element.f0.getBytes(); + SerializationSchema<Tuple2<String, String>> valueSerializationSchema = + element -> element.f1.getBytes(); + + return new UpsertTestSinkWriter<>( + outputFile, keySerializationSchema, valueSerializationSchema); + } + + private List<Tuple2<String, String>> writeTestData( + UpsertTestSinkWriter<Tuple2<String, String>> writer) { + final List<Tuple2<String, String>> expectedRecords = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + Tuple2<String, String> record = Tuple2.of("Key #" + i, "Value #" + i); + expectedRecords.add(record); + writer.write(record, null); + } + return expectedRecords; + } + + private void testRecordPresence(File outputFile, List<Tuple2<String, String>> expectedRecords) + throws IOException { + DeserializationSchema<String> keyDeserializationSchema = new SimpleStringSchema(); + DeserializationSchema<String> valueDeserializationSchema = new SimpleStringSchema(); + + Map<String, String> resultMap = + UpsertTestFileUtil.readRecords( + outputFile, keyDeserializationSchema, valueDeserializationSchema); + + for (Tuple2<String, String> record : expectedRecords) { + assertThat(resultMap).containsEntry(record.f0, record.f1); + } + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkITCase.java b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkITCase.java new file mode 100644 index 00000000000..647a3beae1a --- /dev/null +++ b/flink-tests/src/test/java/org/apache/flink/connector/upserttest/table/UpsertTestDynamicTableSinkITCase.java @@ -0,0 +1,96 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.connector.upserttest.table; + +import org.apache.flink.connector.upserttest.sink.UpsertTestFileUtil; +import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.test.junit5.MiniClusterExtension; +import org.apache.flink.util.TestLoggerExtension; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; + +import static org.assertj.core.api.Assertions.assertThat; + +/** IT Tests for {@link UpsertTestDynamicTableSink}. */ +@ExtendWith(TestLoggerExtension.class) +class UpsertTestDynamicTableSinkITCase { + + @RegisterExtension + public static final MiniClusterExtension MINI_CLUSTER_RESOURCE = + new MiniClusterExtension( + new MiniClusterResourceConfiguration.Builder() + .setNumberTaskManagers(1) + .setNumberSlotsPerTaskManager(1) + .build()); + + @Test + void testWritingDocumentsInBatchMode(@TempDir File tempDir) throws Exception { + TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inBatchMode()); + + String format = "json"; + File outputFile = new File(tempDir, "records.out"); + String outputFilepath = outputFile.toString(); + + final String createTable = + String.format( + "CREATE TABLE UpsertFileSinkTable (\n" + + " user_id INT,\n" + + " user_name STRING,\n" + + " user_count BIGINT,\n" + + " PRIMARY KEY (user_id) NOT ENFORCED\n" + + " ) WITH (\n" + + " 'connector' = '%s',\n" + + " 'key.format' = '%s',\n" + + " 'value.format' = '%s',\n" + + " 'output-filepath' = '%s'\n" + + " );", + UpsertTestDynamicTableSinkFactory.IDENTIFIER, + format, + format, + outputFilepath); + + tEnv.executeSql(createTable); + + String insertSql = + "INSERT INTO UpsertFileSinkTable\n" + + "SELECT user_id, user_name, COUNT(*) AS user_count\n" + + "FROM (VALUES (1, 'Bob'), (22, 'Tom'), (42, 'Kim'), (42, 'Kim'), (42, 'Kim'), (1, 'Bob'))\n" + + " AS UserCountTable(user_id, user_name)\n" + + "GROUP BY user_id, user_name"; + tEnv.executeSql(insertSql).await(); + + /* + | user_id | user_name | user_count | + | ------- | --------- | ---------- | + | 1 | Bob | 2 | + | 22 | Tom | 1 | + | 42 | Kim | 3 | + */ + + int numberOfResultRecords = UpsertTestFileUtil.getNumberOfRecords(outputFile); + assertThat(numberOfResultRecords).isEqualTo(3); + } +} diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java index 32c9548a914..4d9c3dabfd3 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java +++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeInfoTestCoverageTest.java @@ -20,12 +20,28 @@ package org.apache.flink.test.completeness; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.api.common.typeutils.TypeInformationTestBase; +import org.apache.flink.table.dataview.ListViewTypeInfo; +import org.apache.flink.table.dataview.MapViewTypeInfo; +import org.apache.flink.table.runtime.typeutils.BigDecimalTypeInfo; +import org.apache.flink.table.runtime.typeutils.DecimalDataTypeInfo; +import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.runtime.typeutils.LegacyInstantTypeInfo; +import org.apache.flink.table.runtime.typeutils.LegacyLocalDateTimeTypeInfo; +import org.apache.flink.table.runtime.typeutils.LegacyTimestampTypeInfo; +import org.apache.flink.table.runtime.typeutils.SortedMapTypeInfo; +import org.apache.flink.table.runtime.typeutils.StringDataTypeInfo; +import org.apache.flink.table.runtime.typeutils.TimestampDataTypeInfo; +import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo; +import org.apache.flink.table.typeutils.TimeIntervalTypeInfo; import org.apache.flink.util.TestLogger; import org.junit.Test; import org.reflections.Reflections; import java.lang.reflect.Modifier; +import java.util.Arrays; +import java.util.List; import java.util.Set; import java.util.stream.Collectors; @@ -46,6 +62,24 @@ public class TypeInfoTestCoverageTest extends TestLogger { .map(Class::getName) .collect(Collectors.toSet()); + // type info whitelist for TypeInformationTestBase test coverage (see FLINK-27725) + final List<String> typeInfoTestBaseWhitelist = + Arrays.asList( + LegacyTimestampTypeInfo.class.getName(), + InternalTypeInfo.class.getName(), + LegacyLocalDateTimeTypeInfo.class.getName(), + TimeIntervalTypeInfo.class.getName(), + TimeIndicatorTypeInfo.class.getName(), + TimestampDataTypeInfo.class.getName(), + MapViewTypeInfo.class.getName(), + LegacyInstantTypeInfo.class.getName(), + ListViewTypeInfo.class.getName(), + StringDataTypeInfo.class.getName(), + SortedMapTypeInfo.class.getName(), + ExternalTypeInfo.class.getName(), + BigDecimalTypeInfo.class.getName(), + DecimalDataTypeInfo.class.getName()); + // check if a test exists for each type information for (Class<? extends TypeInformation> typeInfo : typeInfos) { // we skip abstract classes and inner classes to skip type information defined in test @@ -61,7 +95,8 @@ public class TypeInfoTestCoverageTest extends TestLogger { } final String testToFind = typeInfo.getName() + "Test"; - if (!typeInfoTestNames.contains(testToFind)) { + if (!typeInfoTestNames.contains(testToFind) + && !typeInfoTestBaseWhitelist.contains(typeInfo.getName())) { fail( "Could not find test '" + testToFind diff --git a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java index 8c2aa180cbd..20b01b4df57 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java +++ b/flink-tests/src/test/java/org/apache/flink/test/completeness/TypeSerializerTestCoverageTest.java @@ -54,6 +54,16 @@ import org.apache.flink.api.scala.typeutils.OptionSerializer; import org.apache.flink.api.scala.typeutils.TraversableSerializer; import org.apache.flink.api.scala.typeutils.TrySerializer; import org.apache.flink.api.scala.typeutils.UnitSerializer; +import org.apache.flink.cep.nfa.DeweyNumber; +import org.apache.flink.cep.nfa.NFA; +import org.apache.flink.cep.nfa.NFAStateSerializer; +import org.apache.flink.cep.nfa.SharedBuffer; +import org.apache.flink.cep.nfa.sharedbuffer.EventId; +import org.apache.flink.cep.nfa.sharedbuffer.Lockable; +import org.apache.flink.cep.nfa.sharedbuffer.NodeId; +import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferEdge; +import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferNode; +import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferNodeSerializer; import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy; import org.apache.flink.runtime.state.ArrayListSerializer; import org.apache.flink.runtime.state.heap.TestDuplicateSerializer; @@ -65,6 +75,23 @@ import org.apache.flink.streaming.api.operators.co.IntervalJoinOperator; import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer; +import org.apache.flink.table.dataview.ListViewSerializer; +import org.apache.flink.table.dataview.MapViewSerializer; +import org.apache.flink.table.dataview.NullAwareMapSerializer; +import org.apache.flink.table.dataview.NullSerializer; +import org.apache.flink.table.runtime.operators.window.CountWindow; +import org.apache.flink.table.runtime.typeutils.ArrayDataSerializer; +import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer; +import org.apache.flink.table.runtime.typeutils.DecimalDataSerializer; +import org.apache.flink.table.runtime.typeutils.ExternalSerializer; +import org.apache.flink.table.runtime.typeutils.LinkedListSerializer; +import org.apache.flink.table.runtime.typeutils.MapDataSerializer; +import org.apache.flink.table.runtime.typeutils.RawValueDataSerializer; +import org.apache.flink.table.runtime.typeutils.RowDataSerializer; +import org.apache.flink.table.runtime.typeutils.SortedMapSerializer; +import org.apache.flink.table.runtime.typeutils.StringDataSerializer; +import org.apache.flink.table.runtime.typeutils.TimestampDataSerializer; +import org.apache.flink.table.runtime.typeutils.WindowKeySerializer; import org.apache.flink.util.TestLogger; import org.junit.Test; @@ -102,7 +129,7 @@ public class TypeSerializerTestCoverageTest extends TestLogger { .map(Class::getSimpleName) .collect(Collectors.toSet()); - // type serializer whitelist for SerializerTestBase test coverage + // type serializer whitelist for SerializerTestBase test coverage (see FLINK-27725) final List<String> serializerTestBaseWhitelist = Arrays.asList( "org.apache.flink.streaming.api.operators.sort.KeyAndValueSerializer", @@ -139,7 +166,36 @@ public class TypeSerializerTestCoverageTest extends TestLogger { NothingSerializer.class.getName(), TrySerializer.class.getName(), "org.apache.flink.api.scala.package$Tuple2CaseClassSerializer", - TestDuplicateSerializer.class.getName()); + TestDuplicateSerializer.class.getName(), + LinkedListSerializer.class.getName(), + SharedBuffer.SharedBufferSerializer.class.getName(), + WindowKeySerializer.class.getName(), + DeweyNumber.DeweyNumberSerializer.class.getName(), + SharedBufferNodeSerializer.class.getName(), + SortedMapSerializer.class.getName(), + BinaryRowDataSerializer.class.getName(), + NFAStateSerializer.class.getName(), + StringDataSerializer.class.getName(), + EventId.EventIdSerializer.class.getName(), + ExternalSerializer.class.getName(), + RawValueDataSerializer.class.getName(), + TimestampDataSerializer.class.getName(), + org.apache.flink.table.runtime.operators.window.TimeWindow.Serializer.class + .getName(), + ArrayDataSerializer.class.getName(), + NullSerializer.class.getName(), + NodeId.NodeIdSerializer.class.getName(), + MapDataSerializer.class.getName(), + NullAwareMapSerializer.class.getName(), + MapViewSerializer.class.getName(), + CountWindow.Serializer.class.getName(), + Lockable.LockableTypeSerializer.class.getName(), + ListViewSerializer.class.getName(), + SharedBufferEdge.SharedBufferEdgeSerializer.class.getName(), + RowDataSerializer.class.getName(), + DecimalDataSerializer.class.getName(), + SharedBufferNode.SharedBufferNodeSerializer.class.getName(), + NFA.NFASerializer.class.getName()); // type serializer whitelist for TypeSerializerUpgradeTestBase test coverage final List<String> typeSerializerUpgradeTestBaseWhitelist = @@ -173,7 +229,36 @@ public class TypeSerializerTestCoverageTest extends TestLogger { NothingSerializer.class.getName(), TrySerializer.class.getName(), "org.apache.flink.api.scala.package$Tuple2CaseClassSerializer", - TestDuplicateSerializer.class.getName()); + TestDuplicateSerializer.class.getName(), + LinkedListSerializer.class.getName(), + SharedBuffer.SharedBufferSerializer.class.getName(), + WindowKeySerializer.class.getName(), + DeweyNumber.DeweyNumberSerializer.class.getName(), + SharedBufferNodeSerializer.class.getName(), + SortedMapSerializer.class.getName(), + BinaryRowDataSerializer.class.getName(), + NFAStateSerializer.class.getName(), + StringDataSerializer.class.getName(), + EventId.EventIdSerializer.class.getName(), + ExternalSerializer.class.getName(), + RawValueDataSerializer.class.getName(), + TimestampDataSerializer.class.getName(), + org.apache.flink.table.runtime.operators.window.TimeWindow.Serializer.class + .getName(), + ArrayDataSerializer.class.getName(), + NullSerializer.class.getName(), + NodeId.NodeIdSerializer.class.getName(), + MapDataSerializer.class.getName(), + NullAwareMapSerializer.class.getName(), + MapViewSerializer.class.getName(), + CountWindow.Serializer.class.getName(), + Lockable.LockableTypeSerializer.class.getName(), + ListViewSerializer.class.getName(), + SharedBufferEdge.SharedBufferEdgeSerializer.class.getName(), + RowDataSerializer.class.getName(), + DecimalDataSerializer.class.getName(), + SharedBufferNode.SharedBufferNodeSerializer.class.getName(), + NFA.NFASerializer.class.getName()); // check if a test exists for each type serializer for (Class<? extends TypeSerializer> typeSerializer : typeSerializers) {
