public static Iterator<TransferableBlock> splitBlock(TransferableBlock block, DataBlock.Type type, int maxBlockSize) {
    List<TransferableBlock> blockChunks = new ArrayList<>();
    if (type == DataBlock.Type.ROW) {
      // Use estimated row size, this estimate is not accurate and is used to estimate numRowsPerChunk only.
      int estimatedRowSizeInBytes = block.getDataSchema().getColumnNames().length * MEDIAN_COLUMN_SIZE_BYTES;
      int numRowsPerChunk = maxBlockSize / estimatedRowSizeInBytes;
      Preconditions.checkState(numRowsPerChunk > 0, "row size too large for query engine to handle, abort!");

      int totalNumRows = block.getNumRows();
      List<Object[]> allRows = block.getContainer();
      int currentRow = 0;
      while (currentRow < totalNumRows) {
        List<Object[]> chunk = allRows.subList(currentRow, Math.min(currentRow + numRowsPerChunk, allRows.size()));
        currentRow += numRowsPerChunk;
        blockChunks.add(new TransferableBlock(chunk, block.getDataSchema(), block.getType()));
      }
      return blockChunks.iterator();
    } else if (type == DataBlock.Type.METADATA) {
      return Iterators.singletonIterator(block);
    } else {
      throw new IllegalArgumentException("Unsupported data block type: " + type);
    }
  }
@Test(dataProvider = "splitRowCountProvider")
  public void testSplitBlockUtils(int splitRowCount)
      throws Exception {
    DataSchema dataSchema = getDataSchema();
    // compare serialized split
    int estRowSizeInBytes = dataSchema.size() * TEST_EST_BYTES_PER_COLUMN;
    List<Object[]> rows = DataBlockTestUtils.getRandomRows(dataSchema, TOTAL_ROW_COUNT, 1);
    RowDataBlock rowBlock = DataBlockBuilder.buildFromRows(rows, dataSchema);
    validateBlocks(TransferableBlockUtils.splitBlock(new TransferableBlock(rowBlock), DataBlock.Type.ROW,
        estRowSizeInBytes * splitRowCount + 1), rows, dataSchema);
    // compare non-serialized split
    validateBlocks(TransferableBlockUtils.splitBlock(new TransferableBlock(rows, dataSchema, DataBlock.Type.ROW),
        DataBlock.Type.ROW, estRowSizeInBytes * splitRowCount + 1), rows, dataSchema);
  }