[Backport][Flink] Fix mapping between Delta's BinaryType and Flink's VarBinaryType (#4554)

#### Which Delta project/connector is this regarding?

- [ ] Spark
- [ ] Standalone
- [X] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

Resolves #3977

This PR fixes a bug where the Delta-Flink connector would incorrectly
map Delta's `BinaryType` (which is variable length) type to Flink's
`BinaryType` (which is fixed length). Instead, this PR fixes it so that
Delta's `BinaryType` is mapped to Flink's `VarBinaryType(MAX_LENGTH)`.
As a comparison, Iceberg does the same

[here](https://github.com/apache/iceberg/blob/8e2ffb35da2d4c5059e96cb78a30fd8c54cfbedf/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java#L125).

This incorrect logic caused the behaviour seen in #3977 because:
- First, we had a source (datagen) table with Flink's BYTES (which is a
synonym for VarBinaryType of MAX_LENGTH)
- Second, we had a target (delta) table. Flink created it with schema
BYTES and the Delta schema had delta type `BinaryType`.
- All good so far. No problems yet.
- When we tried to `INSERT INTO <target> SELECT * FROM <source>`, the
DeltaCatalog would lookup the "flink schema" of the target Delta table.
It would see Delta's BinaryType and map it incorrectly to flink's
BinaryType (not Flink's BYTES or VarBinaryType)
- Hence it would throw the error below

```
org.apache.flink.table.api.ValidationException: Column types of query result and sink for 'print_sink' do not match.
Cause: Incompatible types for sink column 'binary_data' at position 1.
Query schema: [id: BIGINT, binary_data: BYTES]
Sink schema: [id: BIGINT, binary_data: BINARY(1)]
```

## How was this patch tested?

Updated schema conversion tests.

New e2e test which directly tests the problematic scenario brought up in
#3977.

## Does this PR introduce _any_ user-facing changes?

Yes. We fix the mapping of Delta's BinaryType to flinks VarBinaryTYpe

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->

## Does this PR introduce _any_ user-facing changes?

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

Co-authored-by: Scott Sandre <[email protected]>

Author: raveeram-db

connectors/flink/src/main/java/io/delta/flink/source/internal/SchemaConverter.java

@@ -2,7 +2,6 @@
 
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.BigIntType;
-import org.apache.flink.table.types.logical.BinaryType;
 import org.apache.flink.table.types.logical.BooleanType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
@@ -16,6 +15,7 @@
 import org.apache.flink.table.types.logical.SmallIntType;
 import org.apache.flink.table.types.logical.TimestampType;
 import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
 import org.apache.flink.table.types.logical.VarCharType;
 
 import io.delta.standalone.types.DataType;
@@ -64,7 +64,7 @@ public static LogicalType toFlinkDataType(DataType deltaType, boolean nullable)
             case LONG:
                 return new BigIntType(nullable);
             case BINARY:
-                return new BinaryType(nullable, BinaryType.DEFAULT_LENGTH);
+                return new VarBinaryType(nullable, VarBinaryType.MAX_LENGTH);
             case BOOLEAN:
                 return new BooleanType(nullable);
             case BYTE:

connectors/flink/src/test/java/io/delta/flink/source/internal/SchemaConverterTest.java

@@ -5,7 +5,6 @@
 
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.BigIntType;
-import org.apache.flink.table.types.logical.BinaryType;
 import org.apache.flink.table.types.logical.BooleanType;
 import org.apache.flink.table.types.logical.DateType;
 import org.apache.flink.table.types.logical.DecimalType;
@@ -18,6 +17,7 @@
 import org.apache.flink.table.types.logical.SmallIntType;
 import org.apache.flink.table.types.logical.TimestampType;
 import org.apache.flink.table.types.logical.TinyIntType;
+import org.apache.flink.table.types.logical.VarBinaryType;
 import org.apache.flink.table.types.logical.VarCharType;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -71,7 +71,9 @@ private static Stream<Arguments> dataTypes() {
             Arguments.of(new io.delta.standalone.types.ByteType(), new TinyIntType()),
             Arguments.of(new io.delta.standalone.types.ShortType(), new SmallIntType()),
             Arguments.of(new io.delta.standalone.types.LongType(), new BigIntType()),
-            Arguments.of(new io.delta.standalone.types.BinaryType(), new BinaryType()),
+            Arguments.of(
+                new io.delta.standalone.types.BinaryType(),
+                new VarBinaryType(VarBinaryType.MAX_LENGTH)),
             Arguments.of(new io.delta.standalone.types.TimestampType(), new TimestampType()),
             Arguments.of(new io.delta.standalone.types.DateType(), new DateType()),
             Arguments.of(
@@ -163,7 +165,7 @@ private static Stream<Arguments> mapTypes() {
                     new io.delta.standalone.types.ShortType(),
                     true
                 ),
-                new MapType(new BinaryType(), new SmallIntType())),
+                new MapType(new VarBinaryType(VarBinaryType.MAX_LENGTH), new SmallIntType())),
             Arguments.of(
                 new io.delta.standalone.types.MapType(
                     new io.delta.standalone.types.StringType(),

connectors/flink/src/test/java/io/delta/flink/table/it/suite/DeltaEndToEndTableTestSuite.java

@@ -94,6 +94,56 @@ public void setUp() {
             nonPartitionedLargeTablePath);
     }
 
+    /** Tests fix for https://github.com/delta-io/delta/issues/3977 */
+    @Test
+    public void testWriteFromDatagenTableToDeltaTypeWithBytesType() throws Exception {
+        final StreamTableEnvironment tableEnv = setupTableEnvAndDeltaCatalog(false);
+        final String targetTablePath = TEMPORARY_FOLDER.newFolder().getAbsolutePath();
+        final String datagenSourceDDL = ""
+            + "CREATE TABLE source_table ("
+            + " id BIGINT,"
+            + " binary_data BYTES"
+            + ") WITH ("
+            + " 'connector' = 'datagen',"
+            + " 'fields.id.kind' = 'sequence',"
+            + " 'fields.id.start' = '1',"
+            + " 'fields.id.end' = '8',"
+            + " 'number-of-rows' = '8'," // this makes the source BOUNDED
+            + " 'fields.binary_data.kind' = 'random',"
+            + " 'fields.binary_data.length' = '16'"
+            + ")";
+        final String deltaSinkDDL = String.format(""
+                + "CREATE TABLE target_table ("
+                + " id BIGINT,"
+                + " binary_data BYTES"
+                + ") WITH ("
+                + " 'connector' = 'delta',"
+                + " 'table-path' = '%s'"
+                + ")",
+            targetTablePath);
+
+        // Stage 1: Create the source and validate it
+
+        tableEnv.executeSql(datagenSourceDDL).await();
+
+        final List<Row> sourceRows =
+            DeltaTestUtils.readTableResult(tableEnv.executeSql("SELECT * FROM source_table"));
+
+        assertThat(sourceRows).hasSize(8);
+
+        // Stage 2: Create the sink and insert into it and validate it
+
+        tableEnv.executeSql(deltaSinkDDL).await();
+
+        // If our fix for issue #3977 did not work, then this would have thrown an exception.
+        tableEnv.executeSql("INSERT INTO target_table SELECT * FROM source_table").await();
+
+        final List<Row> targetRows =
+            DeltaTestUtils.readTableResult(tableEnv.executeSql("SELECT * FROM target_table"));
+
+        assertThat(targetRows).hasSize(8);
+    }
+
     @Test
     public void shouldReadAndWriteDeltaTable() throws Exception {