[FLINK-38268] Fix incorrect serialization of VARIANT type FLOAT

Author: JinkunLiu

flink-core/src/main/java/org/apache/flink/types/variant/BinaryVariantInternalBuilder.java

@@ -290,7 +290,7 @@ public void appendTimestamp(long microsSinceEpoch) {
     public void appendFloat(float f) {
         checkCapacity(1 + 4);
         writeBuffer[writePos++] = primitiveHeader(FLOAT);
-        writeLong(writeBuffer, writePos, Float.floatToIntBits(f), 8);
+        writeLong(writeBuffer, writePos, Float.floatToIntBits(f), 4);
         writePos += 4;
     }
 

flink-core/src/test/java/org/apache/flink/types/variant/BinaryVariantInternalBuilderTest.java

@@ -24,6 +24,7 @@
 import java.math.BigDecimal;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class BinaryVariantInternalBuilderTest {
@@ -116,4 +117,17 @@ void testParseJsonObject() throws IOException {
         assertThat(variant.getField("k1").getByte()).isEqualTo((byte) 2);
         assertThat(variant.getField("k2").getDecimal()).isEqualTo(BigDecimal.valueOf(1.5));
     }
+
+    @Test
+    void testAppendFloat(){
+        BinaryVariantInternalBuilder builder = new BinaryVariantInternalBuilder(false);
+        float[] floats = new float[25];
+        java.util.Arrays.fill(floats, 4.2f);
+
+        assertThatCode(()->{
+            for (float f : floats) {
+                builder.appendFloat(f);
+            }
+        }).doesNotThrowAnyException();
+    }
 }