public static Pair<List<RowData>, String[]> transposeColumnStatsIndex(List<RowData> colStats, String[] queryColumns, RowType tableSchema) {

    Map<String, LogicalType> tableFieldTypeMap = tableSchema.getFields().stream()
        .collect(Collectors.toMap(RowType.RowField::getName, RowType.RowField::getType));

    // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
    //       w/in the transposed dataset: since some files might not have all the columns indexed
    //       either due to the Column Stats Index config changes, schema evolution, etc. we have
    //       to make sure that all the rows w/in transposed data-frame are properly padded (with null
    //       values) for such file-column combinations
    Set<String> indexedColumns = colStats.stream().map(row -> row.getString(ORD_COL_NAME)
        .toString()).collect(Collectors.toSet());

    // NOTE: We're sorting the columns to make sure final index schema matches layout
    //       of the transposed table
    TreeSet<String> sortedTargetColumns = Arrays.stream(queryColumns).sorted()
        .filter(indexedColumns::contains)
        .collect(Collectors.toCollection(TreeSet::new));

    final Map<LogicalType, AvroToRowDataConverters.AvroToRowDataConverter> converters = new ConcurrentHashMap<>();
    Map<StringData, List<RowData>> fileNameToRows = colStats.stream().parallel()
        .filter(row -> sortedTargetColumns.contains(row.getString(ORD_COL_NAME).toString()))
        .map(row -> {
          if (row.isNullAt(ORD_MIN_VAL) && row.isNullAt(ORD_MAX_VAL)) {
            // Corresponding row could be null in either of the 2 cases
            //    - Column contains only null values (in that case both min/max have to be nulls)
            //    - This is a stubbed Column Stats record (used as a tombstone)
            return row;
          } else {
            String colName = row.getString(ORD_COL_NAME).toString();
            LogicalType colType = tableFieldTypeMap.get(colName);
            return unpackMinMaxVal(row, colType, converters);
          }
        }).collect(Collectors.groupingBy(rowData -> rowData.getString(ORD_FILE_NAME)));

    return Pair.of(foldRowsByFiles(sortedTargetColumns, fileNameToRows), sortedTargetColumns.toArray(new String[0]));
  }
@Test
  void testTransposeColumnStatsIndex() throws Exception {
    final String path = tempFile.getAbsolutePath();
    Configuration conf = TestConfigurations.getDefaultConf(path);
    conf.setBoolean(FlinkOptions.METADATA_ENABLED, true);
    conf.setBoolean(FlinkOptions.READ_DATA_SKIPPING_ENABLED, true);
    conf.setString("hoodie.metadata.index.column.stats.enable", "true");

    HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder()
        .enable(true)
        .withMetadataIndexColumnStats(true)
        .build();
    TestData.writeData(TestData.DATA_SET_INSERT, conf);

    // explicit query columns
    String[] queryColumns1 = {"uuid", "age"};
    List<RowData> indexRows1 = ColumnStatsIndices.readColumnStatsIndex(path, metadataConfig, queryColumns1);
    Pair<List<RowData>, String[]> transposedIndexTable1 = ColumnStatsIndices
        .transposeColumnStatsIndex(indexRows1, queryColumns1, TestConfigurations.ROW_TYPE);
    assertThat("The schema columns should sort by natural order",
        Arrays.toString(transposedIndexTable1.getRight()), is("[age, uuid]"));
    List<RowData> transposed1 = filterOutFileNames(transposedIndexTable1.getLeft());
    assertThat(transposed1.size(), is(4));
    final String expected = "["
        + "+I(2,18,20,0,id5,id6,0), "
        + "+I(2,23,33,0,id1,id2,0), "
        + "+I(2,31,53,0,id3,id4,0), "
        + "+I(2,44,56,0,id7,id8,0)]";
    assertThat(transposed1.toString(), is(expected));

    // no query columns, only for tests
    assertThrows(IllegalArgumentException.class,
        () -> ColumnStatsIndices.readColumnStatsIndex(path, metadataConfig, new String[0]));
  }