You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2024/09/25 02:55:04 UTC
(flink) 03/03: [FLINK-36311][format/csv] Remove deprecated APIs in Flink Csv format
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 483a94a15aed6661a333255139280affb935d52c
Author: Dian Fu <di...@apache.org>
AuthorDate: Wed Sep 18 16:12:05 2024 +0800
[FLINK-36311][format/csv] Remove deprecated APIs in Flink Csv format
---
.../flink/formats/csv/RowCsvInputFormat.java | 188 +++++++++-
.../formats/csv/CsvRowDataSerDeSchemaTest.java | 4 +-
.../csv/CsvRowDeSerializationSchemaTest.java | 403 ---------------------
.../formats/avro/AvroRowDeserializationSchema.java | 1 -
.../formats/csv/CsvRowDeserializationSchema.java | 10 +-
.../formats/csv/CsvRowSerializationSchema.java | 8 +-
6 files changed, 191 insertions(+), 423 deletions(-)
diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowCsvInputFormat.java b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowCsvInputFormat.java
index 53634c5cc83..51ff3a44b08 100644
--- a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowCsvInputFormat.java
+++ b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowCsvInputFormat.java
@@ -18,11 +18,14 @@
package org.apache.flink.formats.csv;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.core.fs.FileInputSplit;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.formats.csv.CsvRowDeserializationSchema.RuntimeConverter;
import org.apache.flink.types.Row;
import org.apache.flink.util.jackson.JacksonMapperFactory;
@@ -32,11 +35,19 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.Csv
import java.io.IOException;
import java.io.Serializable;
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.NoSuchElementException;
-import static org.apache.flink.formats.csv.CsvRowDeserializationSchema.createFieldRuntimeConverters;
-import static org.apache.flink.formats.csv.CsvRowDeserializationSchema.validateArity;
+import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_FORMAT;
+import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -140,6 +151,177 @@ public class RowCsvInputFormat extends AbstractCsvInputFormat<Row> {
return returnRecord;
}
+ // --------------------------------------------------------------------------------------------
+
+ interface RuntimeConverter extends Serializable {
+ Object convert(JsonNode node);
+ }
+
+ private static RuntimeConverter createRowRuntimeConverter(
+ RowTypeInfo rowTypeInfo, boolean ignoreParseErrors, boolean isTopLevel) {
+ final TypeInformation<?>[] fieldTypes = rowTypeInfo.getFieldTypes();
+ final String[] fieldNames = rowTypeInfo.getFieldNames();
+
+ final RuntimeConverter[] fieldConverters =
+ createFieldRuntimeConverters(ignoreParseErrors, fieldTypes);
+
+ return assembleRowRuntimeConverter(
+ ignoreParseErrors, isTopLevel, fieldNames, fieldConverters);
+ }
+
+ static RuntimeConverter[] createFieldRuntimeConverters(
+ boolean ignoreParseErrors, TypeInformation<?>[] fieldTypes) {
+ final RuntimeConverter[] fieldConverters = new RuntimeConverter[fieldTypes.length];
+ for (int i = 0; i < fieldTypes.length; i++) {
+ fieldConverters[i] = createNullableRuntimeConverter(fieldTypes[i], ignoreParseErrors);
+ }
+ return fieldConverters;
+ }
+
+ private static RuntimeConverter assembleRowRuntimeConverter(
+ boolean ignoreParseErrors,
+ boolean isTopLevel,
+ String[] fieldNames,
+ RuntimeConverter[] fieldConverters) {
+ final int rowArity = fieldNames.length;
+
+ return (node) -> {
+ final int nodeSize = node.size();
+
+ if (nodeSize != 0) {
+ validateArity(rowArity, nodeSize, ignoreParseErrors);
+ } else {
+ return null;
+ }
+
+ final Row row = new Row(rowArity);
+ for (int i = 0; i < Math.min(rowArity, nodeSize); i++) {
+ // Jackson only supports mapping by name in the first level
+ if (isTopLevel) {
+ row.setField(i, fieldConverters[i].convert(node.get(fieldNames[i])));
+ } else {
+ row.setField(i, fieldConverters[i].convert(node.get(i)));
+ }
+ }
+ return row;
+ };
+ }
+
+ private static RuntimeConverter createNullableRuntimeConverter(
+ TypeInformation<?> info, boolean ignoreParseErrors) {
+ final RuntimeConverter valueConverter = createRuntimeConverter(info, ignoreParseErrors);
+ return (node) -> {
+ if (node.isNull()) {
+ return null;
+ }
+ try {
+ return valueConverter.convert(node);
+ } catch (Throwable t) {
+ if (!ignoreParseErrors) {
+ throw t;
+ }
+ return null;
+ }
+ };
+ }
+
+ private static RuntimeConverter createRuntimeConverter(
+ TypeInformation<?> info, boolean ignoreParseErrors) {
+ if (info.equals(Types.VOID)) {
+ return (node) -> null;
+ } else if (info.equals(Types.STRING)) {
+ return JsonNode::asText;
+ } else if (info.equals(Types.BOOLEAN)) {
+ return (node) -> Boolean.valueOf(node.asText().trim());
+ } else if (info.equals(Types.BYTE)) {
+ return (node) -> Byte.valueOf(node.asText().trim());
+ } else if (info.equals(Types.SHORT)) {
+ return (node) -> Short.valueOf(node.asText().trim());
+ } else if (info.equals(Types.INT)) {
+ return (node) -> Integer.valueOf(node.asText().trim());
+ } else if (info.equals(Types.LONG)) {
+ return (node) -> Long.valueOf(node.asText().trim());
+ } else if (info.equals(Types.FLOAT)) {
+ return (node) -> Float.valueOf(node.asText().trim());
+ } else if (info.equals(Types.DOUBLE)) {
+ return (node) -> Double.valueOf(node.asText().trim());
+ } else if (info.equals(Types.BIG_DEC)) {
+ return (node) -> new BigDecimal(node.asText().trim());
+ } else if (info.equals(Types.BIG_INT)) {
+ return (node) -> new BigInteger(node.asText().trim());
+ } else if (info.equals(Types.SQL_DATE)) {
+ return (node) -> Date.valueOf(node.asText());
+ } else if (info.equals(Types.SQL_TIME)) {
+ return (node) -> Time.valueOf(node.asText());
+ } else if (info.equals(Types.SQL_TIMESTAMP)) {
+ return (node) -> Timestamp.valueOf(node.asText());
+ } else if (info.equals(Types.LOCAL_DATE)) {
+ return (node) -> Date.valueOf(node.asText()).toLocalDate();
+ } else if (info.equals(Types.LOCAL_TIME)) {
+ return (node) -> Time.valueOf(node.asText()).toLocalTime();
+ } else if (info.equals(Types.LOCAL_DATE_TIME)) {
+ return (node) -> LocalDateTime.parse(node.asText().trim(), SQL_TIMESTAMP_FORMAT);
+ } else if (info.equals(Types.INSTANT)) {
+ return (node) ->
+ LocalDateTime.parse(node.asText(), SQL_TIMESTAMP_WITH_LOCAL_TIMEZONE_FORMAT)
+ .toInstant(ZoneOffset.UTC);
+ } else if (info instanceof RowTypeInfo) {
+ final RowTypeInfo rowTypeInfo = (RowTypeInfo) info;
+ return createRowRuntimeConverter(rowTypeInfo, ignoreParseErrors, false);
+ } else if (info instanceof BasicArrayTypeInfo) {
+ return createObjectArrayRuntimeConverter(
+ ((BasicArrayTypeInfo<?, ?>) info).getComponentInfo(), ignoreParseErrors);
+ } else if (info instanceof ObjectArrayTypeInfo) {
+ return createObjectArrayRuntimeConverter(
+ ((ObjectArrayTypeInfo<?, ?>) info).getComponentInfo(), ignoreParseErrors);
+ } else if (info instanceof PrimitiveArrayTypeInfo
+ && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
+ return createByteArrayRuntimeConverter(ignoreParseErrors);
+ } else {
+ throw new RuntimeException("Unsupported type information '" + info + "'.");
+ }
+ }
+
+ private static RuntimeConverter createObjectArrayRuntimeConverter(
+ TypeInformation<?> elementType, boolean ignoreParseErrors) {
+ final Class<?> elementClass = elementType.getTypeClass();
+ final RuntimeConverter elementConverter =
+ createNullableRuntimeConverter(elementType, ignoreParseErrors);
+
+ return (node) -> {
+ final int nodeSize = node.size();
+ final Object[] array = (Object[]) Array.newInstance(elementClass, nodeSize);
+ for (int i = 0; i < nodeSize; i++) {
+ array[i] = elementConverter.convert(node.get(i));
+ }
+ return array;
+ };
+ }
+
+ private static RuntimeConverter createByteArrayRuntimeConverter(boolean ignoreParseErrors) {
+ return (node) -> {
+ try {
+ return node.binaryValue();
+ } catch (IOException e) {
+ if (!ignoreParseErrors) {
+ throw new RuntimeException("Unable to deserialize byte array.", e);
+ }
+ return null;
+ }
+ };
+ }
+
+ static void validateArity(int expected, int actual, boolean ignoreParseErrors) {
+ if (expected != actual && !ignoreParseErrors) {
+ throw new RuntimeException(
+ "Row length mismatch. "
+ + expected
+ + " fields expected but was "
+ + actual
+ + ".");
+ }
+ }
+
/** Create a builder. */
public static Builder builder(TypeInformation<Row> typeInfo, Path... filePaths) {
return new Builder(typeInfo, filePaths);
diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java
index 402168a3d8d..87f109721e4 100644
--- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java
+++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java
@@ -474,7 +474,7 @@ class CsvRowDataSerDeSchemaTest {
CsvRowDataSerializationSchema schema =
InstantiationUtil.deserializeObject(
InstantiationUtil.serializeObject(serSchemaBuilder.build()),
- CsvRowDeSerializationSchemaTest.class.getClassLoader());
+ CsvRowDataSerDeSchemaTest.class.getClassLoader());
open(schema);
return schema.serialize(row);
}
@@ -487,7 +487,7 @@ class CsvRowDataSerDeSchemaTest {
CsvRowDataDeserializationSchema schema =
InstantiationUtil.deserializeObject(
InstantiationUtil.serializeObject(deserSchemaBuilder.build()),
- CsvRowDeSerializationSchemaTest.class.getClassLoader());
+ CsvRowDataSerDeSchemaTest.class.getClassLoader());
open(schema);
return schema.deserialize(csv != null ? csv.getBytes() : null);
}
diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java
deleted file mode 100644
index b26732b371e..00000000000
--- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDeSerializationSchemaTest.java
+++ /dev/null
@@ -1,403 +0,0 @@
-/*
- * 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.formats.csv;
-
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.common.typeinfo.Types;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.InstantiationUtil;
-
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.sql.Date;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.util.function.Consumer;
-
-import static org.apache.flink.connector.testutils.formats.SchemaTestUtils.open;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** Tests for {@link CsvRowSerializationSchema} and {@link CsvRowDeserializationSchema}. */
-class CsvRowDeSerializationSchemaTest {
-
- @Test
- @SuppressWarnings("unchecked")
- public void testSerializeDeserialize() throws Exception {
-
- testNullableField(Types.LONG, "null", null);
- testNullableField(Types.STRING, "null", null);
- testNullableField(Types.VOID, "null", null);
- testNullableField(Types.STRING, "\"This is a test.\"", "This is a test.");
- testNullableField(Types.STRING, "\"This is a test\n\r.\"", "This is a test\n\r.");
- testNullableField(Types.BOOLEAN, "true", true);
- testNullableField(Types.BOOLEAN, "null", null);
- testNullableField(Types.BYTE, "124", (byte) 124);
- testNullableField(Types.SHORT, "10000", (short) 10000);
- testNullableField(Types.INT, "1234567", 1234567);
- testNullableField(Types.LONG, "12345678910", 12345678910L);
- testNullableField(Types.FLOAT, "0.33333334", 0.33333334f);
- testNullableField(Types.DOUBLE, "0.33333333332", 0.33333333332d);
- testNullableField(
- Types.BIG_DEC,
- "\"1234.0000000000000000000000001\"",
- new BigDecimal("1234.0000000000000000000000001"));
- testNullableField(
- Types.BIG_INT,
- "\"123400000000000000000000000000\"",
- new BigInteger("123400000000000000000000000000"));
- testNullableField(Types.SQL_DATE, "2018-10-12", Date.valueOf("2018-10-12"));
- testNullableField(Types.SQL_TIME, "12:12:12", Time.valueOf("12:12:12"));
- testNullableField(
- Types.SQL_TIMESTAMP,
- "\"2018-10-12 12:12:12.0\"",
- Timestamp.valueOf("2018-10-12 12:12:12"));
- testNullableField(Types.LOCAL_DATE, "2018-10-12", Date.valueOf("2018-10-12").toLocalDate());
- testNullableField(Types.LOCAL_TIME, "12:12:12", Time.valueOf("12:12:12").toLocalTime());
- testNullableField(
- Types.LOCAL_DATE_TIME,
- "\"2018-10-12 12:12:12\"",
- LocalDateTime.parse("2018-10-12T12:12:12"));
- testNullableField(
- Types.INSTANT,
- "\"1970-01-01 00:00:01.123456789Z\"",
- Instant.ofEpochMilli(1123).plusNanos(456789));
- testNullableField(Types.INSTANT, "\"1970-01-01 00:00:12Z\"", Instant.ofEpochSecond(12));
- testNullableField(
- Types.ROW(Types.STRING, Types.INT, Types.BOOLEAN),
- "Hello;42;false",
- Row.of("Hello", 42, false));
- testNullableField(Types.OBJECT_ARRAY(Types.STRING), "a;b;c", new String[] {"a", "b", "c"});
- testNullableField(Types.OBJECT_ARRAY(Types.BYTE), "12;4;null", new Byte[] {12, 4, null});
- testNullableField(
- (TypeInformation<byte[]>) Types.PRIMITIVE_ARRAY(Types.BYTE),
- "awML",
- new byte[] {107, 3, 11});
- }
-
- @Test
- void testSerializeDeserializeCustomizedProperties() throws Exception {
-
- final Consumer<CsvRowSerializationSchema.Builder> serConfig =
- (serSchemaBuilder) ->
- serSchemaBuilder
- .setEscapeCharacter('*')
- .setQuoteCharacter('\'')
- .setArrayElementDelimiter(":")
- .setFieldDelimiter(';');
-
- final Consumer<CsvRowDeserializationSchema.Builder> deserConfig =
- (deserSchemaBuilder) ->
- deserSchemaBuilder
- .setEscapeCharacter('*')
- .setQuoteCharacter('\'')
- .setArrayElementDelimiter(":")
- .setFieldDelimiter(';');
-
- testField(Types.STRING, "123*'4**", "123'4*", deserConfig, ";");
- testField(Types.STRING, "'123''4**'", "123'4*", serConfig, deserConfig, ";");
- testField(Types.STRING, "'a;b*'c'", "a;b'c", deserConfig, ";");
- testField(Types.STRING, "'a;b''c'", "a;b'c", serConfig, deserConfig, ";");
- testField(Types.INT, " 12 ", 12, deserConfig, ";");
- testField(Types.INT, "12", 12, serConfig, deserConfig, ";");
- testField(
- Types.LOCAL_DATE_TIME,
- " 2018-10-12 12:12:12 ",
- LocalDateTime.parse("2018-10-12T12:12:12"),
- deserConfig,
- ";");
- testField(
- Types.ROW(Types.STRING, Types.STRING),
- "1:hello",
- Row.of("1", "hello"),
- deserConfig,
- ";");
- testField(
- Types.ROW(Types.STRING, Types.STRING),
- "'1:hello'",
- Row.of("1", "hello"),
- serConfig,
- deserConfig,
- ";");
- testField(
- Types.ROW(Types.STRING, Types.STRING),
- "'1:hello world'",
- Row.of("1", "hello world"),
- serConfig,
- deserConfig,
- ";");
- testField(
- Types.STRING,
- "null",
- "null",
- serConfig,
- deserConfig,
- ";"); // string because null literal has not been set
- }
-
- @Test
- void testDeserializeParseError() {
- assertThatThrownBy(() -> testDeserialization(false, false, "Test,null,Test"))
- .isInstanceOf(IOException.class);
- }
-
- @Test
- void testDeserializeUnsupportedNull() throws Exception {
- // unsupported null for integer
- assertThat(testDeserialization(true, false, "Test,null,Test"))
- .isEqualTo(Row.of("Test", null, "Test"));
- }
-
- @Test
- void testDeserializeIncompleteRow() throws Exception {
- // last two columns are missing
- assertThat(testDeserialization(true, false, "Test")).isEqualTo(Row.of("Test", null, null));
- }
-
- @Test
- void testDeserializeMoreColumnsThanExpected() throws Exception {
- // one additional string column
- assertThat(testDeserialization(true, false, "Test,12,Test,Test")).isNull();
- }
-
- @Test
- void testDeserializeIgnoreComment() throws Exception {
- // # is part of the string
- assertThat(testDeserialization(false, false, "#Test,12,Test"))
- .isEqualTo(Row.of("#Test", 12, "Test"));
- }
-
- @Test
- void testDeserializeAllowComment() throws Exception {
- // entire row is ignored
- assertThat(testDeserialization(true, true, "#Test,12,Test")).isNull();
- }
-
- @Test
- void testSerializationProperties() throws Exception {
- final TypeInformation<Row> rowInfo = Types.ROW(Types.STRING, Types.INT, Types.STRING);
- final CsvRowSerializationSchema.Builder serSchemaBuilder =
- new CsvRowSerializationSchema.Builder(rowInfo).setLineDelimiter("\r");
-
- assertThat(serialize(serSchemaBuilder, Row.of("Test", 12, "Hello")))
- .isEqualTo("Test,12,Hello\r".getBytes());
-
- serSchemaBuilder.setQuoteCharacter('#');
-
- assertThat(serialize(serSchemaBuilder, Row.of("Test", 12, "2019-12-26 12:12:12")))
- .isEqualTo("Test,12,#2019-12-26 12:12:12#\r".getBytes());
-
- serSchemaBuilder.disableQuoteCharacter();
-
- assertThat(serialize(serSchemaBuilder, Row.of("Test", 12, "2019-12-26 12:12:12")))
- .isEqualTo("Test,12,2019-12-26 12:12:12\r".getBytes());
- }
-
- @Test
- void testEmptyLineDelimiter() throws Exception {
- final TypeInformation<Row> rowInfo = Types.ROW(Types.STRING, Types.INT, Types.STRING);
- final CsvRowSerializationSchema.Builder serSchemaBuilder =
- new CsvRowSerializationSchema.Builder(rowInfo).setLineDelimiter("");
-
- assertThat(serialize(serSchemaBuilder, Row.of("Test", 12, "Hello")))
- .isEqualTo("Test,12,Hello".getBytes());
- }
-
- @Test
- void testInvalidNesting() {
- assertThatThrownBy(
- () ->
- testNullableField(
- Types.ROW(Types.ROW(Types.STRING)),
- "FAIL",
- Row.of(Row.of("FAIL"))))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- void testInvalidType() {
- assertThatThrownBy(
- () ->
- testNullableField(
- Types.GENERIC(java.util.Date.class),
- "FAIL",
- new java.util.Date()))
- .isInstanceOf(IllegalArgumentException.class);
- }
-
- @Test
- void testSerializeDeserializeNestedTypes() throws Exception {
- final TypeInformation<Row> subDataType0 =
- Types.ROW(
- Types.STRING,
- Types.INT,
- Types.STRING,
- Types.LOCAL_DATE_TIME,
- Types.INSTANT);
- final TypeInformation<Row> subDataType1 =
- Types.ROW(
- Types.STRING,
- Types.INT,
- Types.STRING,
- Types.LOCAL_DATE_TIME,
- Types.INSTANT);
- final TypeInformation<Row> rowInfo = Types.ROW(subDataType0, subDataType1);
-
- // serialization
- CsvRowSerializationSchema.Builder serSchemaBuilder =
- new CsvRowSerializationSchema.Builder(rowInfo);
- // deserialization
- CsvRowDeserializationSchema.Builder deserSchemaBuilder =
- new CsvRowDeserializationSchema.Builder(rowInfo);
-
- Row normalRow =
- Row.of(
- Row.of(
- "hello",
- 1,
- "This is 1st top column",
- LocalDateTime.parse("1970-01-01T01:02:03"),
- Instant.ofEpochMilli(1000)),
- Row.of(
- "world",
- 2,
- "This is 2nd top column",
- LocalDateTime.parse("1970-01-01T01:02:04"),
- Instant.ofEpochMilli(2000)));
- testSerDeConsistency(normalRow, serSchemaBuilder, deserSchemaBuilder);
-
- Row nullRow =
- Row.of(
- null,
- Row.of(
- "world",
- 2,
- "This is 2nd top column after null",
- LocalDateTime.parse("1970-01-01T01:02:05"),
- Instant.ofEpochMilli(3000)));
- testSerDeConsistency(nullRow, serSchemaBuilder, deserSchemaBuilder);
- }
-
- private <T> void testNullableField(TypeInformation<T> fieldInfo, String string, T value)
- throws Exception {
- testField(
- fieldInfo,
- string,
- value,
- (deserSchema) -> deserSchema.setNullLiteral("null"),
- (serSchema) -> serSchema.setNullLiteral("null"),
- ",");
- }
-
- private <T> void testField(
- TypeInformation<T> fieldInfo,
- String csvValue,
- T value,
- Consumer<CsvRowSerializationSchema.Builder> serializationConfig,
- Consumer<CsvRowDeserializationSchema.Builder> deserializationConfig,
- String fieldDelimiter)
- throws Exception {
- final TypeInformation<Row> rowInfo = Types.ROW(Types.STRING, fieldInfo, Types.STRING);
- final String expectedCsv = "BEGIN" + fieldDelimiter + csvValue + fieldDelimiter + "END\n";
- final Row expectedRow = Row.of("BEGIN", value, "END");
-
- // serialization
- final CsvRowSerializationSchema.Builder serSchemaBuilder =
- new CsvRowSerializationSchema.Builder(rowInfo);
- serializationConfig.accept(serSchemaBuilder);
- final byte[] serializedRow = serialize(serSchemaBuilder, expectedRow);
- assertThat(new String(serializedRow)).isEqualTo(expectedCsv);
-
- // deserialization
- final CsvRowDeserializationSchema.Builder deserSchemaBuilder =
- new CsvRowDeserializationSchema.Builder(rowInfo);
- deserializationConfig.accept(deserSchemaBuilder);
- final Row deserializedRow = deserialize(deserSchemaBuilder, expectedCsv);
- assertThat(deserializedRow).isEqualTo(expectedRow);
- }
-
- private <T> void testField(
- TypeInformation<T> fieldInfo,
- String csvValue,
- T value,
- Consumer<CsvRowDeserializationSchema.Builder> deserializationConfig,
- String fieldDelimiter)
- throws Exception {
- final TypeInformation<Row> rowInfo = Types.ROW(Types.STRING, fieldInfo, Types.STRING);
- final String csv = "BEGIN" + fieldDelimiter + csvValue + fieldDelimiter + "END\n";
- final Row expectedRow = Row.of("BEGIN", value, "END");
-
- // deserialization
- final CsvRowDeserializationSchema.Builder deserSchemaBuilder =
- new CsvRowDeserializationSchema.Builder(rowInfo);
- deserializationConfig.accept(deserSchemaBuilder);
- final Row deserializedRow = deserialize(deserSchemaBuilder, csv);
- assertThat(deserializedRow).isEqualTo(expectedRow);
- }
-
- private Row testDeserialization(
- boolean allowParsingErrors, boolean allowComments, String string) throws Exception {
- final TypeInformation<Row> rowInfo = Types.ROW(Types.STRING, Types.INT, Types.STRING);
- final CsvRowDeserializationSchema.Builder deserSchemaBuilder =
- new CsvRowDeserializationSchema.Builder(rowInfo)
- .setIgnoreParseErrors(allowParsingErrors)
- .setAllowComments(allowComments);
- return deserialize(deserSchemaBuilder, string);
- }
-
- private void testSerDeConsistency(
- Row originalRow,
- CsvRowSerializationSchema.Builder serSchemaBuilder,
- CsvRowDeserializationSchema.Builder deserSchemaBuilder)
- throws Exception {
- Row deserializedRow =
- deserialize(
- deserSchemaBuilder, new String(serialize(serSchemaBuilder, originalRow)));
- assertThat(originalRow).isEqualTo(deserializedRow);
- }
-
- private static byte[] serialize(CsvRowSerializationSchema.Builder serSchemaBuilder, Row row)
- throws Exception {
- // we serialize and deserialize the schema to test runtime behavior
- // when the schema is shipped to the cluster
- final CsvRowSerializationSchema schema =
- InstantiationUtil.deserializeObject(
- InstantiationUtil.serializeObject(serSchemaBuilder.build()),
- CsvRowDeSerializationSchemaTest.class.getClassLoader());
- open(schema);
- return schema.serialize(row);
- }
-
- private static Row deserialize(
- CsvRowDeserializationSchema.Builder deserSchemaBuilder, String csv) throws Exception {
- // we serialize and deserialize the schema to test runtime behavior
- // when the schema is shipped to the cluster
- final CsvRowDeserializationSchema schema =
- InstantiationUtil.deserializeObject(
- InstantiationUtil.serializeObject(deserSchemaBuilder.build()),
- CsvRowDeSerializationSchemaTest.class.getClassLoader());
- open(schema);
- return schema.deserialize(csv.getBytes());
- }
-}
diff --git a/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
index 852ee54f850..923783f6cef 100644
--- a/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
+++ b/flink-python/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
@@ -20,7 +20,6 @@ package org.apache.flink.formats.avro;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
-import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java
similarity index 97%
rename from flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java
rename to flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java
index b2a91a9fb34..1f1831303f4 100644
--- a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java
+++ b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowDeserializationSchema.java
@@ -18,7 +18,7 @@
package org.apache.flink.formats.csv;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
@@ -57,13 +57,8 @@ import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOC
* Row}.
*
* <p>Failure during deserialization are forwarded as wrapped {@link IOException}s.
- *
- * @deprecated The format was developed for the Table API users and will not be maintained for
- * DataStream API users anymore. Either use Table API or switch to Data Stream, defining your
- * own {@link DeserializationSchema}.
*/
-@PublicEvolving
-@Deprecated
+@Internal
public final class CsvRowDeserializationSchema implements DeserializationSchema<Row> {
private static final long serialVersionUID = 2135553495874539201L;
@@ -98,7 +93,6 @@ public final class CsvRowDeserializationSchema implements DeserializationSchema<
}
/** A builder for creating a {@link CsvRowDeserializationSchema}. */
- @PublicEvolving
public static class Builder {
private final RowTypeInfo typeInfo;
diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java
similarity index 98%
rename from flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java
rename to flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java
index 53f4edf16e3..358d4ec56fa 100644
--- a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java
+++ b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvRowSerializationSchema.java
@@ -18,6 +18,7 @@
package org.apache.flink.formats.csv;
+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.common.typeinfo.BasicArrayTypeInfo;
@@ -57,13 +58,8 @@ import static org.apache.flink.formats.common.TimeFormats.SQL_TIMESTAMP_WITH_LOC
*
* <p>Result <code>byte[]</code> messages can be deserialized using {@link
* CsvRowDeserializationSchema}.
- *
- * @deprecated The format was developed for the Table API users and will not be maintained for
- * DataStream API users anymore. Either use Table API or switch to Data Stream, defining your
- * own {@link SerializationSchema}.
*/
-@PublicEvolving
-@Deprecated
+@Internal
public final class CsvRowSerializationSchema implements SerializationSchema<Row> {
private static final long serialVersionUID = 2098447220136965L;