[FLINK-39084][core] variant accessor methods for handling map or array types in converters

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

@@ -32,8 +32,10 @@
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Base64;
+import java.util.List;
 import java.util.Objects;
 
 import static org.apache.flink.types.variant.BinaryVariantUtil.BINARY_SEARCH_THRESHOLD;
@@ -242,11 +244,32 @@ public Variant getElement(int index) throws VariantTypeException {
         return getElementAtIndex(index);
     }
 
+    @Override
+    public int getArraySize() throws VariantTypeException {
+        return handleArray(value, pos, (size, offsetSize, offsetStart, dataStart) -> size);
+    }
+
     @Override
     public Variant getField(String fieldName) throws VariantTypeException {
         return getFieldByKey(fieldName);
     }
 
+    @Override
+    public List<String> getFieldNames() throws VariantTypeException {
+        return handleObject(
+                value,
+                pos,
+                (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+                    List<String> fieldNames = new ArrayList<>(size);
+                    for (int i = 0; i < size; i++) {
+                        int id = readUnsigned(value, idStart + idSize * i, idSize);
+                        String fieldName = getMetadataKey(metadata, id);
+                        fieldNames.add(fieldName);
+                    }
+                    return fieldNames;
+                });
+    }
+
     @Override
     public String toJson() {
         StringBuilder sb = new StringBuilder();

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

@@ -24,6 +24,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.util.List;
 
 /** Variant represent a semi-structured data. */
 @PublicEvolving
@@ -175,6 +176,14 @@ public interface Variant {
      */
     Variant getElement(int index) throws VariantTypeException;
 
+    /**
+     * Get the size of an array variant.
+     *
+     * @return Number of elements if this variant is an array
+     * @throws VariantTypeException If this variant is not an array.
+     */
+    int getArraySize() throws VariantTypeException;
+
     /**
      * Access value of the specified field of an object variant. If there is no field with the
      * specified name, null is returned.
@@ -186,6 +195,15 @@ public interface Variant {
      */
     Variant getField(String fieldName) throws VariantTypeException;
 
+    /**
+     * Get the field names of an object variant only at top level. Doesn't include the nested
+     * fields.
+     *
+     * @return List of field names if this variant is an object
+     * @throws VariantTypeException If this variant is not an object.
+     */
+    List<String> getFieldNames() throws VariantTypeException;
+
     /** Parses the variant to json. */
     String toJson();
 

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

@@ -113,6 +113,7 @@ void testArrayVariant() {
         assertThat(variant.isPrimitive()).isFalse();
         assertThat(variant.isObject()).isFalse();
         assertThat(variant.getType()).isEqualTo(Variant.Type.ARRAY);
+        assertThat(variant.getArraySize()).isEqualTo(5);
 
         assertThat(variant.getElement(-1)).isNull();
         assertThat(variant.getElement(0).getInt()).isEqualTo(1);
@@ -144,12 +145,16 @@ void testObjectVariant() {
         assertThat(variant.isPrimitive()).isFalse();
         assertThat(variant.isObject()).isTrue();
         assertThat(variant.getType()).isEqualTo(Variant.Type.OBJECT);
+        assertThat(variant.getFieldNames()).containsExactlyInAnyOrder("list", "object", "bb");
 
         assertThat(variant.getField("list").isArray()).isTrue();
+        assertThat(variant.getField("list").getArraySize()).isEqualTo(2);
         assertThat(variant.getField("list").getElement(0).getString()).isEqualTo("hello");
         assertThat(variant.getField("list").getElement(1).getInt()).isEqualTo(1);
 
         assertThat(variant.getField("object").isObject()).isTrue();
+        assertThat(variant.getField("object").getFieldNames())
+                .containsExactlyInAnyOrder("ss", "ff");
         assertThat(variant.getField("object").getField("ss").getShort()).isEqualTo((short) 1);
         assertThat(variant.getField("object").getField("ff").getFloat()).isEqualTo((10.0f));