[HUDI-7930] Flink Support for Array of Row and Map of Row value

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java

@@ -18,15 +18,19 @@
 
 package org.apache.hudi.table;
 
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.source.ExpressionPredicates;
 import org.apache.hudi.source.prune.DataPruner;
 import org.apache.hudi.source.prune.PrimaryKeyPruners;
 import org.apache.hudi.storage.StoragePath;
 import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.table.format.cow.CopyOnWriteInputFormat;
 import org.apache.hudi.table.format.mor.MergeOnReadInputFormat;
 import org.apache.hudi.util.SerializableSchema;
+import org.apache.hudi.utils.SchemaBuilder;
 import org.apache.hudi.utils.TestConfigurations;
 import org.apache.hudi.utils.TestData;
 
@@ -62,6 +66,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.stream.Collectors;
 
 import static org.apache.hudi.keygen.constant.KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED;
@@ -70,8 +75,11 @@
 import static org.hamcrest.core.Is.is;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
 
 /**
  * Test cases for HoodieTableSource.
@@ -151,6 +159,91 @@ void testGetTableAvroSchema() {
     assertThat(schemaFields, is(expected));
   }
 
+  @Test
+  void readArrayOfRows() throws Exception {
+    final String path = tempFile.getAbsolutePath();
+    conf = TestConfigurations.getDefaultConf(path);
+    conf.setString(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_COPY_ON_WRITE);
+    conf.setString(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH,
+            Objects.requireNonNull(Thread.currentThread()
+                    .getContextClassLoader().getResource("test_read_array_of_rows_schema.avsc")).toString());
+    TestData.writeDataAsBatch(TestData.DATA_SET_ARRAY_OF_ROWS, conf);
+    HoodieTableSource tableSource = new HoodieTableSource(
+            SerializableSchema.create( SchemaBuilder.instance()
+            .fields(TestData.ARRAY_OF_ROWS_TYPE.getFieldNames(), TestData.ARRAY_OF_ROWS_DATA_TYPE.getChildren())
+            .build()),
+            new StoragePath(conf.getString(FlinkOptions.PATH)),
+            Arrays.asList(conf.getString(FlinkOptions.PARTITION_PATH_FIELD).split(",")),
+            "default-par",
+            conf);
+    List<StoragePathInfo> fileList = tableSource.getReadFiles();
+    assertNotNull(fileList);
+    assertThat(fileList.size(), is(4));
+    CopyOnWriteInputFormat inputFormat = (CopyOnWriteInputFormat) tableSource.getInputFormat();
+
+    inputFormat.open(inputFormat.createInputSplits(1)[0]);
+    while (!inputFormat.reachedEnd()) {
+      RowData row = inputFormat.nextRecord(null);
+      assertNotNull(row);
+      ArrayData rowArray = row.getArray(3);
+      assertNotNull(rowArray);
+      for(int i = 0; i < rowArray.size(); ++i) {
+        RowData subRow = rowArray.getRow(i, 2);
+        assertNotNull(subRow);
+        assertNotNull(subRow.getString(0));
+        subRow.getInt(1);
+      }
+
+    }
+  }
+
+  @Test
+  void readMapOfRows() throws Exception {
+    final String path = tempFile.getAbsolutePath();
+    conf = TestConfigurations.getDefaultConf(path);
+    conf.setString(FlinkOptions.TABLE_TYPE, FlinkOptions.TABLE_TYPE_COPY_ON_WRITE);
+    conf.setString(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH,
+            Objects.requireNonNull(Thread.currentThread()
+                    .getContextClassLoader().getResource("test_read_map_of_rows_schema.avsc")).toString());
+    TestData.writeDataAsBatch(TestData.DATA_SET_MAP_OF_ROWS, conf);
+    HoodieTableSource tableSource = new HoodieTableSource(
+            SerializableSchema.create( SchemaBuilder.instance()
+                    .fields(TestData.MAP_OF_ROWS_TYPE.getFieldNames(), TestData.MAP_OF_ROWS_DATA_TYPE.getChildren())
+                    .build()),
+            new StoragePath(conf.getString(FlinkOptions.PATH)),
+            Arrays.asList(conf.getString(FlinkOptions.PARTITION_PATH_FIELD).split(",")),
+            "default-par",
+            conf);
+    List<StoragePathInfo> fileList = tableSource.getReadFiles();
+    assertNotNull(fileList);
+    assertThat(fileList.size(), is(4));
+    CopyOnWriteInputFormat inputFormat = (CopyOnWriteInputFormat) tableSource.getInputFormat();
+
+    inputFormat.open(inputFormat.createInputSplits(1)[0]);
+    while (!inputFormat.reachedEnd()) {
+      RowData row = inputFormat.nextRecord(null);
+      assertNotNull(row);
+       MapData rowMap = row.getMap(3);
+      assertNotNull(rowMap);
+      for(int i = 0; i < rowMap.size(); ++i) {
+        String key = rowMap.keyArray().getString(i).toString();
+        assertNotNull(key);
+        RowData subRow = rowMap.valueArray().getRow(i, 2);
+        assertNotNull(subRow);
+        assertNotNull(subRow.getString(0));
+        if(key.endsWith("1")) {
+          assertFalse(subRow.getString(0).toString().endsWith("2"));
+          assertTrue(subRow.getInt(1) < 200);
+        } else {
+          assertTrue(subRow.getString(0).toString().endsWith("2"));
+          assertTrue(subRow.getInt(1) > 200);
+        }
+      }
+
+    }
+  }
+
+
   @Test
   void testDataSkippingFilterShouldBeNotNullWhenTableSourceIsCopied() {
     HoodieTableSource tableSource = getEmptyStreamingSource();

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java

@@ -18,6 +18,10 @@
 
 package org.apache.hudi.utils;
 
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.types.DataType;
 import org.apache.hudi.client.common.HoodieFlinkEngineContext;
 import org.apache.hudi.common.config.HoodieCommonConfig;
 import org.apache.hudi.common.fs.FSUtils;
@@ -29,6 +33,7 @@
 import org.apache.hudi.common.table.TableSchemaResolver;
 import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.MapUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.configuration.FlinkOptions;
@@ -48,6 +53,7 @@
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.table.data.GenericArrayData;
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.StringData;
@@ -413,6 +419,139 @@ public class TestData {
             TimestampData.fromEpochMillis(i), i, DecimalData.fromBigDecimal(new BigDecimal(String.format("%d.%d%d", i, i, i)), 3, 2))));
   }
 
+  public static final DataType ARRAY_OF_ROWS_DATA_TYPE = DataTypes.ROW(
+                  DataTypes.FIELD("uuid", DataTypes.VARCHAR(20)),// record key
+                  DataTypes.FIELD("partition", DataTypes.VARCHAR(10)),
+                  DataTypes.FIELD("ts", DataTypes.TIMESTAMP(3)), // precombine field
+                  DataTypes.FIELD("data", DataTypes.ARRAY( DataTypes.ROW(
+                    DataTypes.FIELD("name", DataTypes.VARCHAR(10)),
+                    DataTypes.FIELD("age", DataTypes.INT())
+                  ))))
+                 .notNull();
+
+  public static final RowType ARRAY_OF_ROWS_TYPE = (RowType) ARRAY_OF_ROWS_DATA_TYPE.getLogicalType();
+
+  public static List<RowData> DATA_SET_ARRAY_OF_ROWS = Arrays.asList(
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id1"), StringData.fromString("par1"),
+                  TimestampData.fromEpochMillis(3),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Danny"), 23),
+                          GenericRowData.of(StringData.fromString("Danny2"), 223)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id2"), StringData.fromString("par1"),
+                  TimestampData.fromEpochMillis(2),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Stephen"), 33),
+                          GenericRowData.of(StringData.fromString("Stephen2"), 233)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id3"), StringData.fromString("par2"),
+                  TimestampData.fromEpochMillis(3),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Julian"), 53),
+                          GenericRowData.of(StringData.fromString("Julian2"), 253)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id4"), StringData.fromString("par2"),
+                  TimestampData.fromEpochMillis(4),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Fabian"), 31),
+                          GenericRowData.of(StringData.fromString("Fabian2"), 231)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id5"), StringData.fromString("par3"),
+                  TimestampData.fromEpochMillis(5),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Sophia"), 18),
+                          GenericRowData.of(StringData.fromString("Sophia2"), 218)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id6"), StringData.fromString("par3"),
+                  TimestampData.fromEpochMillis(6),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Emma"), 20),
+                          GenericRowData.of(StringData.fromString("Emma2"), 220)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id7"), StringData.fromString("par4"),
+                  TimestampData.fromEpochMillis(7),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Bob"), 44),
+                          GenericRowData.of(StringData.fromString("Bob2"), 244)
+                  })),
+          insertRow(ARRAY_OF_ROWS_TYPE, StringData.fromString("id8"), StringData.fromString("par4"),
+                  TimestampData.fromEpochMillis(8),
+                  new GenericArrayData(new Object[]{
+                          GenericRowData.of(StringData.fromString("Han"), 56),
+                          GenericRowData.of(StringData.fromString("Han2"), 256)
+                  }))
+  );
+
+  public static final DataType MAP_OF_ROWS_DATA_TYPE = DataTypes.ROW(
+                  DataTypes.FIELD("uuid", DataTypes.VARCHAR(20)),// record key
+                  DataTypes.FIELD("partition", DataTypes.VARCHAR(10)),
+                  DataTypes.FIELD("ts", DataTypes.TIMESTAMP(3)), // precombine field
+                  DataTypes.FIELD("data", DataTypes.MAP( DataTypes.VARCHAR(10), DataTypes.ROW(
+                          DataTypes.FIELD("name", DataTypes.VARCHAR(10)),
+                          DataTypes.FIELD("age", DataTypes.INT())
+                  ))))
+          .notNull();
+
+  public static final RowType MAP_OF_ROWS_TYPE = (RowType) MAP_OF_ROWS_DATA_TYPE.getLogicalType();
+
+  public static List<RowData> DATA_SET_MAP_OF_ROWS = Arrays.asList(
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id1"), StringData.fromString("par1"),
+                  TimestampData.fromEpochMillis(3),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Danny"), 23),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Danny2"), 223)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id2"), StringData.fromString("par1"),
+                  TimestampData.fromEpochMillis(2),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Stephen"), 33),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Stephen2"), 233)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id3"), StringData.fromString("par2"),
+                  TimestampData.fromEpochMillis(3),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Julian"), 53),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Julian2"), 253)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id4"), StringData.fromString("par2"),
+                  TimestampData.fromEpochMillis(4),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Fabian"), 31),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Fabian2"), 231)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id5"), StringData.fromString("par3"),
+                  TimestampData.fromEpochMillis(5),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Sophia"), 18),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Sophia2"), 218)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id6"), StringData.fromString("par3"),
+                  TimestampData.fromEpochMillis(6),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Emma"), 20),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Emma2"), 220)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id7"), StringData.fromString("par4"),
+                  TimestampData.fromEpochMillis(7),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Bob"), 44),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Bob2"), 244)
+                  ))),
+          insertRow(MAP_OF_ROWS_TYPE, StringData.fromString("id8"), StringData.fromString("par4"),
+                  TimestampData.fromEpochMillis(8),
+                  new GenericMapData( MapOf(
+                          StringData.fromString("key1"), GenericRowData.of(StringData.fromString("Han"), 56),
+                          StringData.fromString("key2"), GenericRowData.of(StringData.fromString("Han2"), 256)
+                  )))
+  );
+
+  public static <K,V> Map<K,V> MapOf(K key1, V value1, K key2, V value2){
+    Map<K,V> map = new HashMap<>();
+    map.put(key1, value1);
+    map.put(key2, value2);
+    return map;
+  }
+
   public static List<RowData> dataSetInsert(int... ids) {
     List<RowData> inserts = new ArrayList<>();
     Arrays.stream(ids).forEach(i -> inserts.add(

hudi-flink-datasource/hudi-flink/src/test/resources/test_read_array_of_rows_schema.avsc

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+{
+  "type" : "record",
+  "name" : "record",
+  "fields" : [ {
+    "name" : "uuid",
+    "type" : [ "null", "string" ],
+    "default" : null
+  }, {
+    "name" : "partition",
+    "type" : [ "null", "string" ],
+    "default" : null
+  }, {
+    "name" : "ts",
+    "type" : [ "null", {
+      "type" : "long",
+      "logicalType" : "timestamp-millis"
+    } ],
+    "default" : null
+  }, {
+    "name" : "data",
+    "type" : {
+      "type":"array",
+      "items" : {
+        "name":"person",
+        "type":"record",
+        "fields": [
+          {
+            "name" : "name",
+            "type" : [ "null", "string" ],
+            "default" : null
+          }, {
+            "name" : "age",
+            "type" : [ "null", "int" ],
+            "default" : null
+          }
+        ]
+      }
+    }
+  } ]
+}