[FLINK-37154] Introduce ByteArraySchema (#26080)

Author: christeoh

Diff for: flink-core/src/main/java/org/apache/flink/api/common/serialization/ByteArraySchema.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.api.common.serialization;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+
+import java.io.IOException;
+
+/** Simple serialization schema for bytes. */
+@PublicEvolving
+public class ByteArraySchema implements DeserializationSchema<byte[]>, SerializationSchema<byte[]> {
+
+    private static final long serialVersionUID = 1L;
+
+    // ------------------------------------------------------------------------
+    //  Kafka Serialization
+    // ------------------------------------------------------------------------
+
+    @Override
+    public byte[] deserialize(byte[] message) throws IOException {
+        return message;
+    }
+
+    @Override
+    public boolean isEndOfStream(byte[] nextElement) {
+        return false;
+    }
+
+    @Override
+    public byte[] serialize(byte[] element) {
+        return element;
+    }
+
+    @Override
+    public TypeInformation<byte[]> getProducedType() {
+        return TypeInformation.of(byte[].class);
+    }
+}

Diff for: flink-core/src/test/java/org/apache/flink/api/common/serialization/ByteArraySchemaTest.java

@@ -0,0 +1,33 @@
+/*
+ * 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.api.common.serialization;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ByteArraySchemaTest {
+    @Test
+    void testSimpleSerialisation() throws IOException {
+        final byte[] testBytes = "hello world".getBytes();
+        assertThat(new ByteArraySchema().serialize(testBytes)).isEqualTo(testBytes);
+        assertThat(new ByteArraySchema().deserialize(testBytes)).isEqualTo(testBytes);
+    }
+}

Diff for: flink-python/pyflink/common/__init__.py

@@ -45,8 +45,9 @@
       :class:`~pyflink.datastream.formats.csv.CsvRowSerializationSchema`,
       :class:`~pyflink.datastream.formats.csv.CsvRowDeserializationSchema`,
       :class:`~pyflink.datastream.formats.avro.AvroRowSerializationSchema`,
-      :class:`~pyflink.datastream.formats.avro.AvroRowDeserializationSchema` and
-      :class:`~SimpleStringSchema` for more details.
+      :class:`~pyflink.datastream.formats.avro.AvroRowDeserializationSchema`,
+      :class:`~SimpleStringSchema` and
+      :class:`~SimpleByteSchema` for more details.
 """
 from pyflink.common.completable_future import CompletableFuture
 from pyflink.common.config_options import ConfigOption, ConfigOptions
@@ -59,11 +60,11 @@
 from pyflink.common.job_status import JobStatus
 from pyflink.common.restart_strategy import RestartStrategies, RestartStrategyConfiguration
 from pyflink.common.serialization import SerializationSchema, DeserializationSchema, \
-    SimpleStringSchema, Encoder
+    SimpleStringSchema, ByteArraySchema, Encoder
 from pyflink.common.serializer import TypeSerializer
+from pyflink.common.time import Duration, Instant, Time
 from pyflink.common.typeinfo import Types, TypeInformation
 from pyflink.common.types import Row, RowKind
-from pyflink.common.time import Duration, Instant, Time
 from pyflink.common.watermark_strategy import WatermarkStrategy, \
     AssignerWithPeriodicWatermarksWrapper
 
@@ -78,6 +79,7 @@
     'SerializationSchema',
     'DeserializationSchema',
     'SimpleStringSchema',
+    'ByteArraySchema',
     'Encoder',
     'CompletableFuture',
     'InputDependencyConstraint',

Diff for: flink-python/pyflink/common/serialization.py

@@ -23,6 +23,7 @@
     'SerializationSchema',
     'DeserializationSchema',
     'SimpleStringSchema',
+    'ByteArraySchema',
     'Encoder',
     'BulkWriterFactory'
 ]
@@ -68,6 +69,21 @@ def __init__(self, charset: str = 'UTF-8'):
             self, j_deserialization_schema=j_simple_string_serialization_schema)
 
 
+class ByteArraySchema(SerializationSchema, DeserializationSchema):
+    """
+    Simple serialization/deserialization schema for bytes.
+    """
+
+    def __init__(self):
+        gate_way = get_gateway()
+        j_simple_byte_serialization_schema = gate_way \
+            .jvm.org.apache.flink.api.common.serialization.ByteArraySchema()
+        SerializationSchema.__init__(self,
+                                     j_serialization_schema=j_simple_byte_serialization_schema)
+        DeserializationSchema.__init__(
+            self, j_deserialization_schema=j_simple_byte_serialization_schema)
+
+
 class Encoder(object):
     """
     Encoder is used by the file sink to perform the actual writing of the

Diff for: flink-python/pyflink/common/tests/test_serialization_schemas.py

@@ -15,7 +15,7 @@
 #  See the License for the specific language governing permissions and
 # limitations under the License.
 ################################################################################
-from pyflink.common.serialization import SimpleStringSchema
+from pyflink.common.serialization import SimpleStringSchema, ByteArraySchema
 from pyflink.testing.test_case_utils import PyFlinkTestCase
 
 
@@ -29,3 +29,13 @@ def test_simple_string_schema(self):
 
         self.assertEqual(expected_string, simple_string_schema._j_deserialization_schema
                          .deserialize(expected_string.encode(encoding='utf-8')))
+
+
+class SimpleByteSchemaTests(PyFlinkTestCase):
+    def test_simple_byte_schema(self):
+        expected_bytes = "test bytes".encode(encoding='utf-8')
+        simple_byte_schema = ByteArraySchema()
+        self.assertEqual(expected_bytes,
+                         simple_byte_schema._j_serialization_schema.serialize(expected_bytes))
+        self.assertEqual(expected_bytes, simple_byte_schema._j_deserialization_schema
+                         .deserialize(expected_bytes))

Diff for: flink-python/pyflink/datastream/connectors/tests/test_kafka.py

@@ -16,14 +16,13 @@
 # limitations under the License.
 ################################################################################
 import json
-from typing import Dict
 import unittest
+from typing import Dict
 
 import pyflink.datastream.data_stream as data_stream
 from pyflink.common import typeinfo
-
 from pyflink.common.configuration import Configuration
-from pyflink.common.serialization import SimpleStringSchema, DeserializationSchema
+from pyflink.common.serialization import SimpleStringSchema, ByteArraySchema, DeserializationSchema
 from pyflink.common.typeinfo import Types
 from pyflink.common.types import Row
 from pyflink.common.watermark_strategy import WatermarkStrategy
@@ -318,6 +317,7 @@ def _check(schema: DeserializationSchema, class_name: str):
                              class_name)
 
         _check(SimpleStringSchema(), 'org.apache.flink.api.common.serialization.SimpleStringSchema')
+        _check(ByteArraySchema(), 'org.apache.flink.api.common.serialization.ByteArraySchema')
         _check(
             JsonRowDeserializationSchema.builder().type_info(Types.ROW([Types.STRING()])).build(),
             'org.apache.flink.formats.json.JsonRowDeserializationSchema'
@@ -635,6 +635,10 @@ def _check_serialization_schema_implementations(check_function):
             SimpleStringSchema(),
             'org.apache.flink.api.common.serialization.SimpleStringSchema'
         )
+        check_function(
+            ByteArraySchema(),
+            'org.apache.flink.api.common.serialization.ByteArraySchema'
+        )
 
 
 class MockDataStream(data_stream.DataStream):