public List<WorkflowInstance> getWorkflowInstancesWithLatestRun(
      String workflowId, long startInstanceId, long endInstanceId, boolean aggregated) {

    List<WorkflowInstance> instances =
        withMetricLogError(
            () ->
                withRetryableQuery(
                    LATEST_RUN_WORKFLOW_INSTANCE_IDS_QUERY,
                    stmt -> {
                      int idx = 0;
                      stmt.setString(++idx, workflowId);
                      stmt.setLong(++idx, endInstanceId);
                      stmt.setLong(++idx, startInstanceId);
                    },
                    result -> {
                      List<WorkflowInstance> rows = new ArrayList<>();
                      while (result.next()) {
                        rows.add(workflowInstanceFromResult(result));
                      }
                      return rows;
                    }),
            "getLatestWorkflowInstanceRuns",
            "Failed to get workflow instances for workflow id: {} between instance id : {} and {}",
            workflowId,
            startInstanceId,
            endInstanceId);
    if (aggregated) {

      instances.forEach(
          instance ->
              instance.setAggregatedInfo(
                  AggregatedViewHelper.computeAggregatedView(instance, true)));
    }
    return instances;
  }
@Test
  public void testGetWorkflowInstanceLatestRuns() throws Exception {
    initializeForGetWorkflowInstancesLatestRun();

    // pagination call to get all the latest runs workflow instance id's in a particular workflow
    // id, null cursor
    List<WorkflowInstance> workflows =
        instanceDao.getWorkflowInstancesWithLatestRun(TEST_WORKFLOW_ID, 1, 100, false);

    workflows.sort(Comparator.comparing(WorkflowInstance::getWorkflowInstanceId).reversed());
    assertNotNull(workflows);
    assertEquals(10, workflows.size());
    for (int i = 0; i < 10; i++) {
      WorkflowInstance instance = workflows.get(i);
      assertNotNull(instance);
      assertEquals(TEST_WORKFLOW_ID, instance.getWorkflowId());
      assertEquals(10 - i, instance.getWorkflowInstanceId());
      assertEquals(109 - i, instance.getWorkflowRunId());
    }

    // pagination NEXT call to get a subset of pages, which fill the entire page request
    workflows = instanceDao.getWorkflowInstancesWithLatestRun(TEST_WORKFLOW_ID, 2, 6, false);
    workflows.sort(Comparator.comparing(WorkflowInstance::getWorkflowInstanceId).reversed());
    assertNotNull(workflows);
    assertEquals(5, workflows.size());
    for (int i = 0; i < workflows.size(); i++) {
      WorkflowInstance instance = workflows.get(i);
      assertNotNull(instance);
      assertEquals(TEST_WORKFLOW_ID, instance.getWorkflowId());
      assertEquals(6 - i, instance.getWorkflowInstanceId());
      assertEquals(105 - i, instance.getWorkflowRunId());
    }

    // pagination NEXT call to get a subset of pages, which fill portion of the page requested
    workflows = instanceDao.getWorkflowInstancesWithLatestRun(TEST_WORKFLOW_ID, 1, 3, false);
    workflows.sort(Comparator.comparing(WorkflowInstance::getWorkflowInstanceId).reversed());
    assertNotNull(workflows);
    assertEquals(3, workflows.size());
    for (int i = 0; i < workflows.size(); i++) {
      WorkflowInstance instance = workflows.get(i);
      assertNotNull(instance);
      assertEquals(TEST_WORKFLOW_ID, instance.getWorkflowId());
      assertEquals(3 - i, instance.getWorkflowInstanceId());
      assertEquals(102 - i, instance.getWorkflowRunId());
    }

    // batch call with nothing
    workflows = instanceDao.getWorkflowInstancesWithLatestRun(TEST_WORKFLOW_ID, 0, 0, false);
    assertNotNull(workflows);
    assertTrue(workflows.isEmpty());

    // non-existing workflow id
    workflows =
        instanceDao.getWorkflowInstancesWithLatestRun("sample-dag-test-random", 1, 1, false);
    assertNotNull(workflows);
    assertTrue(workflows.isEmpty());

    cleanupForGetWorkflowInstancesLatestRun();
  }