Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Support pagination in sessions listing API #1173

Closed
wants to merge 27 commits into from
Closed

Support pagination in sessions listing API #1173

Show file tree
Hide file tree
Changes from 21 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
ee4d784
Fix api
tosametal Jul 11, 2019
7b385f4
Add pagination
tosametal Jul 11, 2019
1520d93
Fix test
tosametal Jul 11, 2019
432b8bc
Fix ReferenceError
tosametal Jul 11, 2019
5ce36f4
Revert "Fix ReferenceError"
tosametal Jul 18, 2019
611f54f
Revert "Add pagination"
tosametal Jul 18, 2019
394596a
Restored deleted page size parameters for compatibility
tosametal Jul 18, 2019
b712951
Revert "Fix test"
tosametal Jul 18, 2019
3f4fe5c
Add totalPageCount to RestSessionCollection
tosametal Jul 18, 2019
22431da
Use RestSessionCollection as return class for compatibility
tosametal Jul 18, 2019
f94934a
Fix to get sessions of specified page only
tosametal Jul 23, 2019
f96a154
Fix to get project sessions of specified page only
tosametal Jul 23, 2019
c51f3ba
Fix query and indent
tosametal Jul 23, 2019
3bd36c9
Fix indent
tosametal Jul 23, 2019
16d9f78
Remove offset from getAttempts
tosametal Jul 24, 2019
a4bf756
Refactor duplicated implementation
tosametal Aug 5, 2019
53931ab
Return records number instead of page number
tosametal Aug 5, 2019
6aa8f34
Add test
tosametal Aug 7, 2019
fbf53c0
Add test
tosametal Aug 7, 2019
ff63080
Refactor duplicated implementation
tosametal Aug 14, 2019
2fecfdf
Remove pageSize from RestSessionCollection
tosametal Aug 26, 2019
559b979
Fix variable name
tosametal Sep 3, 2019
8a93b7a
Fix method name
tosametal Sep 3, 2019
d19eff4
Validate page number with QueryParamValidator
tosametal Sep 3, 2019
9d65b0b
Switch the order of pageNumber and lastId
tosametal Oct 4, 2019
09061d8
Add siteId condition to getSessionsCountOfProject query
tosametal Oct 7, 2019
bc0174c
Add test
tosametal Dec 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions digdag-client/src/main/java/io/digdag/client/api/RestSessionCollection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ public interface RestSessionCollection
{
List<RestSession> getSessions();

int getRecordsNumber();

yoyama marked this conversation as resolved.
Show resolved Hide resolved
static ImmutableRestSessionCollection.Builder builder()
{
return ImmutableRestSessionCollection.builder();
Expand Down
99 changes: 88 additions & 11 deletions digdag-core/src/main/java/io/digdag/core/database/DatabaseSessionStoreManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1171,7 +1171,26 @@ public <T> T putAndLockSession(Session session, SessionLockAction<T> func)
@Override
public List<StoredSessionWithLastAttempt> getSessions(int pageSize, Optional<Long> lastId)
{
return autoCommit((handle, dao) -> dao.getSessions(siteId, pageSize, lastId.or(Long.MAX_VALUE)));
return getSessions(pageSize, lastId, 1);
}

@Override
public List<StoredSessionWithLastAttempt> getSessions(int pageSize, Optional<Long> lastId, int pageNumber)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess switch the order of pageNumber and lastId may be understandable.
Other method also same.

{
yoyama marked this conversation as resolved.
Show resolved Hide resolved
int offset = pageSize * (pageNumber - 1);
return autoCommit((handle, dao) -> dao.getSessions(siteId, pageSize, lastId.or(Long.MAX_VALUE), offset));
}

@Override
public Integer getTotalSessionsCount(Optional<Long> lastId)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

getSessionsCount may be better name.

{
return autoCommit((handle, dao) -> dao.getTotalSessionsCount(siteId, lastId.or(Long.MAX_VALUE)));
}

@Override
public Integer getTotalProjectSessionsCount(Optional<Long> lastId, int projectId)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

getSessionsCountOfProject may be better to fit getSessionsOfProject.

{
return autoCommit((handle, dao) -> dao.getTotalProjectSessionsCount(lastId.or(Long.MAX_VALUE), projectId));
}

@Override
Expand All @@ -1186,7 +1205,14 @@ public StoredSessionWithLastAttempt getSessionById(long sessionId)
@Override
public List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(int projectId, String workflowName, int pageSize, Optional<Long> lastId)
{
return autoCommit((handle, dao) -> dao.getSessionsOfWorkflowByName(siteId, projectId, workflowName, pageSize, lastId.or(Long.MAX_VALUE)));
return getSessionsOfWorkflowByName(projectId, workflowName, pageSize, lastId, 1);
}

@Override
public List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(int projectId, String workflowName, int pageSize, Optional<Long> lastId, int pageNumber)
{
int offset = pageSize * (pageNumber - 1);
return autoCommit((handle, dao) -> dao.getSessionsOfWorkflowByName(siteId, projectId, workflowName, pageSize, lastId.or(Long.MAX_VALUE), offset));
}

@Override
Expand Down Expand Up @@ -1214,7 +1240,14 @@ public List<StoredSessionAttemptWithSession> getAttemptsOfProject(boolean withRe
@Override
public List<StoredSessionWithLastAttempt> getSessionsOfProject(int projectId, int pageSize, Optional<Long> lastId)
{
return autoCommit((handle, dao) -> dao.getSessionsOfProject(siteId, projectId, pageSize, lastId.or(Long.MAX_VALUE)));
return getSessionsOfProject(projectId, pageSize, lastId, 1);
}

@Override
public List<StoredSessionWithLastAttempt> getSessionsOfProject(int projectId, int pageSize, Optional<Long> lastId, int pageNumber)
{
int offset = pageSize * (pageNumber - 1);
return autoCommit((handle, dao) -> dao.getSessionsOfProject(siteId, projectId, pageSize, lastId.or(Long.MAX_VALUE), offset));
}

@Override
Expand Down Expand Up @@ -1519,14 +1552,21 @@ public void completeDelayedAttempt(long attemptId)
public interface H2Dao
extends Dao
{
default List<StoredSessionWithLastAttempt> getSessions(int siteId, int limit, long lastId)
{
return getSessions(siteId, limit, lastId, 0);
}

@SqlQuery("select s.*, sa.site_id, sa.attempt_name, sa.workflow_definition_id, sa.state_flags, sa.timezone, sa.params, sa.created_at, sa.finished_at, sa.index" +
" from sessions s" +
" join session_attempts sa on sa.id = s.last_attempt_id" +
" where s.project_id in (select id from projects where site_id = :siteId)" +
" and s.id < :lastId" +
" order by s.id desc" +
" limit :limit")
List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId);
" limit :limit" +
" offset :offset")
List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId, @Bind("offset") int offset);

yoyama marked this conversation as resolved.
Show resolved Hide resolved

// h2's MERGE doesn't reutrn generated id when conflicting row already exists
@SqlUpdate("merge into sessions" +
Expand All @@ -1553,6 +1593,12 @@ void upsertAndLockSession(@Bind("projectId") int projectId,
public interface PgDao
extends Dao
{

default List<StoredSessionWithLastAttempt> getSessions(int siteId, int limit, long lastId)
{
return getSessions(siteId, limit, lastId, 0);
}

// This query uses "sessions.project_id = any(array(select ..))" instead of "in" (semi-join) so that
// PostgreSQL doesn't use a bad query plan which scans almost all records from sessions table.
@SqlQuery("select s.*, sa.site_id, sa.attempt_name, sa.workflow_definition_id, sa.state_flags, sa.timezone, sa.params, sa.created_at, sa.finished_at, sa.index" +
Expand All @@ -1561,8 +1607,10 @@ public interface PgDao
" where s.project_id = any(array(select id from projects where site_id = :siteId))" +
" and s.id < :lastId" +
" order by s.id desc" +
" limit :limit")
List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId);
" limit :limit" +
" offset :offset")
List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId, @Bind("offset") int offset);


@SqlQuery("insert into sessions" +
" (project_id, workflow_name, session_time)" +
Expand Down Expand Up @@ -1595,22 +1643,50 @@ public interface Dao

List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId);

List<StoredSessionWithLastAttempt> getSessions(@Bind("siteId") int siteId, @Bind("limit") int limit, @Bind("lastId") long lastId, @Bind("offset") int offset);

@SqlQuery("select count(1)" +
" from sessions s" +
" join session_attempts sa on sa.id = s.last_attempt_id" +
" where s.project_id in (select id from projects where site_id = :siteId)" +
" and s.id < :lastId")
Integer getTotalSessionsCount(@Bind("siteId") int siteId, @Bind("lastId") long lastId);


@SqlQuery("select count(1)" +
" from sessions s" +
" join session_attempts sa on sa.id = s.last_attempt_id" +
" where s.project_id = :projectId" +
" and s.id < :lastId")
Integer getTotalProjectSessionsCount(@Bind("lastId") long lastId, @Bind("projectId") int projectId);

@SqlQuery("select s.*, sa.site_id, sa.attempt_name, sa.workflow_definition_id, sa.state_flags, sa.timezone, sa.params, sa.created_at, sa.finished_at, sa.index" +
" from sessions s" +
" join session_attempts sa on sa.id = s.last_attempt_id" +
" where s.id = :id" +
" and sa.site_id = :siteId")
StoredSessionWithLastAttempt getSession(@Bind("siteId") int siteId, @Bind("id") long id);

default List<StoredSessionWithLastAttempt> getSessionsOfProject(int siteId, int projId, int limit, long lastId)
{
return getSessionsOfProject(siteId, projId, limit, lastId, 0);
}

@SqlQuery("select s.*, sa.site_id, sa.attempt_name, sa.workflow_definition_id, sa.state_flags, sa.timezone, sa.params, sa.created_at, sa.finished_at, sa.index" +
" from sessions s" +
" join session_attempts sa on sa.id = s.last_attempt_id" +
" where s.project_id = :projId" +
" and sa.site_id = :siteId" +
" and s.id < :lastId" +
" order by s.id desc" +
" limit :limit")
List<StoredSessionWithLastAttempt> getSessionsOfProject(@Bind("siteId") int siteId, @Bind("projId") int projId, @Bind("limit") int limit, @Bind("lastId") long lastId);
" limit :limit" +
" offset :offset")
List<StoredSessionWithLastAttempt> getSessionsOfProject(@Bind("siteId") int siteId, @Bind("projId") int projId, @Bind("limit") int limit, @Bind("lastId") long lastId, @Bind("offset") int offset);

default List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(int siteId, int projId, String workflowName, int limit, long lastId)
{
return getSessionsOfWorkflowByName(siteId, projId, workflowName, limit, lastId, 0);
}

@SqlQuery("select s.*, sa.site_id, sa.attempt_name, sa.workflow_definition_id, sa.state_flags, sa.timezone, sa.params, sa.created_at, sa.finished_at, sa.index" +
" from sessions s" +
Expand All @@ -1620,8 +1696,9 @@ public interface Dao
" and sa.site_id = :siteId" +
" and s.id < :lastId" +
" order by s.id desc" +
" limit :limit")
List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(@Bind("siteId") int siteId, @Bind("projId") int projId, @Bind("workflowName") String workflowName, @Bind("limit") int limit, @Bind("lastId") long lastId);
" limit :limit" +
" offset :offset")
List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(@Bind("siteId") int siteId, @Bind("projId") int projId, @Bind("workflowName") String workflowName, @Bind("limit") int limit, @Bind("lastId") long lastId, @Bind("offset") int offset);

@SqlQuery("select sa.*, s.session_uuid, s.workflow_name, s.session_time" +
" from session_attempts sa" +
Expand Down
10 changes: 10 additions & 0 deletions digdag-core/src/main/java/io/digdag/core/session/SessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,23 @@ public interface SessionStore
{
List<StoredSessionWithLastAttempt> getSessions(int pageSize, Optional<Long> lastId);

List<StoredSessionWithLastAttempt> getSessions(int pageSize, Optional<Long> lastId, int pageNumber);

Integer getTotalSessionsCount(Optional<Long> lastId);

Integer getTotalProjectSessionsCount(Optional<Long> lastId, int projectId);

StoredSessionWithLastAttempt getSessionById(long sessionId)
throws ResourceNotFoundException;

List<StoredSessionWithLastAttempt> getSessionsOfProject(int projectId, int pageSize, Optional<Long> lastId);

List<StoredSessionWithLastAttempt> getSessionsOfProject(int projectId, int pageSize, Optional<Long> lastId, int pageNumber);

List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(int projectId, String workflowName, int pageSize, Optional<Long> lastId);

List<StoredSessionWithLastAttempt> getSessionsOfWorkflowByName(int projectId, String workflowName, int pageSize, Optional<Long> lastId, int pageNumber);

List<StoredSessionAttemptWithSession> getAttempts(boolean withRetriedAttempts, int pageSize, Optional<Long> lastId);

List<StoredSessionAttemptWithSession> getAttemptsOfProject(boolean withRetriedAttempts, int projectId, int pageSize, Optional<Long> lastId);
Expand Down
28 changes: 28 additions & 0 deletions digdag-core/src/test/java/io/digdag/core/database/DatabaseSessionStoreManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,9 +260,15 @@ public void testGetAndNotFounds()

assertSessionAndLastAttemptEquals(session1, attempt1);
assertThat(ImmutableList.of(session1, otherProjSession1), is(store.getSessions(100, Optional.absent())));
assertThat(ImmutableList.of(otherProjSession1), is(store.getSessions(1, Optional.absent(), 2)));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfProject(proj.getId(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 1)));
assertEmpty(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 2));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 1)));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 2));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf2.getName(), 100, Optional.absent()));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf2.getName(), 1, Optional.absent(), 2));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf1.getName(), 100, Optional.absent()), containsInAnyOrder(attempt1));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf2.getName(), 100, Optional.absent()), is(Matchers.empty()));

Expand All @@ -277,8 +283,13 @@ public void testGetAndNotFounds()

assertSessionAndLastAttemptEquals(session2, attempt2);
assertThat(ImmutableList.of(session2, session1, otherProjSession1), is(store.getSessions(100, Optional.absent())));
assertThat(ImmutableList.of(otherProjSession1), is(store.getSessions(1, Optional.absent(), 3)));
assertThat(ImmutableList.of(session2, session1), is(store.getSessionsOfProject(proj.getId(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 2)));
assertEmpty(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 3));
assertThat(ImmutableList.of(session2, session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 2)));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 3));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf2.getName(), 100, Optional.absent()));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf1.getName(), 100, Optional.absent()), containsInAnyOrder(attempt1, attempt2));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf2.getName(), 100, Optional.absent()), is(Matchers.empty()));
Expand All @@ -299,8 +310,13 @@ public void testGetAndNotFounds()
assertSessionAndLastAttemptEquals(session2AfterRetry, attempt3);
assertThat(session2AfterRetry.getLastAttempt().getRetryAttemptName(), is(Optional.of(retryAttemptName)));
assertThat(ImmutableList.of(session2AfterRetry, session1, otherProjSession1), is(store.getSessions(100, Optional.absent())));
assertThat(ImmutableList.of(session2AfterRetry, session1), is(store.getSessions(2, Optional.absent(), 1)));
assertThat(ImmutableList.of(session2AfterRetry, session1), is(store.getSessionsOfProject(proj.getId(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 2)));
assertEmpty(store.getSessionsOfProject(proj.getId(), 1, Optional.absent(), 3));
assertThat(ImmutableList.of(session2AfterRetry, session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 100, Optional.absent())));
assertThat(ImmutableList.of(session1), is(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 2)));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf1.getName(), 1, Optional.absent(), 3));
assertEmpty(store.getSessionsOfWorkflowByName(proj.getId(), wf2.getName(), 100, Optional.absent()));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf1.getName(), 100, Optional.absent()), containsInAnyOrder(attempt1, attempt3));
assertThat(store.getActiveAttemptsOfWorkflow(proj.getId(), wf2.getName(), 100, Optional.absent()), is(Matchers.empty()));
Expand All @@ -320,16 +336,28 @@ public void testGetAndNotFounds()
//
assertThat(ImmutableList.of(session2AfterRetry, session1, otherProjSession1),
is(store.getSessions(100, Optional.absent())));
assertThat(ImmutableList.of(session1),
is(store.getSessions(1, Optional.absent(), 2)));
assertThat(ImmutableList.of(session2AfterRetry, session1),
is(store.getSessions(2, Optional.absent())));
assertThat(ImmutableList.of(otherProjSession1),
is(store.getSessions(2, Optional.absent(), 2)));
assertThat(ImmutableList.of(session2AfterRetry),
is(store.getSessions(1, Optional.absent())));
assertThat(ImmutableList.of(session1),
is(store.getSessions(1, Optional.absent(), 2)));
assertThat(ImmutableList.of(session1, otherProjSession1),
is(store.getSessions(100, Optional.of(session2AfterRetry.getId()))));
assertThat(ImmutableList.of(otherProjSession1),
is(store.getSessions(1, Optional.of(session2AfterRetry.getId()), 2)));
assertThat(ImmutableList.of(otherProjSession1),
is(store.getSessions(100, Optional.of(session1.getId()))));
assertThat(ImmutableList.of(otherProjSession1),
is(store.getSessions(100, Optional.of(session1.getId()), 1)));
assertEmpty(store.getSessions(100, Optional.of(otherProjSession1.getId())));
assertEmpty(store.getSessions(100, Optional.of(otherProjSession1.getId()), 2));
assertEmpty(anotherSite.getSessions(100, Optional.absent()));
assertEmpty(anotherSite.getSessions(100, Optional.absent(), 2));

////
// public attempt listings
Expand Down
13 changes: 9 additions & 4 deletions digdag-server/src/main/java/io/digdag/server/rs/ProjectResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package io.digdag.server.rs;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
import java.net.URI;
Expand Down Expand Up @@ -415,7 +416,8 @@ public RestSessionCollection getSessions(
@PathParam("id") int projectId,
@QueryParam("workflow") String workflowName,
@QueryParam("last_id") Long lastId,
@QueryParam("page_size") Integer pageSize)
@QueryParam("page_size") Integer pageSize,
@QueryParam("page_number") Integer pageNumber)
throws ResourceNotFoundException
{
int validPageSize = QueryParamValidator.validatePageSize(Optional.fromNullable(pageSize), MAX_SESSIONS_PAGE_SIZE, DEFAULT_SESSIONS_PAGE_SIZE);
Expand All @@ -425,15 +427,18 @@ public RestSessionCollection getSessions(
SessionStore ss = ssm.getSessionStore(getSiteId());

StoredProject proj = ensureNotDeletedProject(ps.getProjectById(projectId));
Integer page = Optional.fromNullable(pageNumber).or(1);
Copy link
Contributor

@yoyama yoyama Sep 3, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as SessionResource. Please change var name.


Integer sessionRecordsNumber = ss.getTotalProjectSessionsCount(Optional.fromNullable(lastId), projectId);

List<StoredSessionWithLastAttempt> sessions;
if (workflowName != null) {
sessions = ss.getSessionsOfWorkflowByName(proj.getId(), workflowName, validPageSize, Optional.fromNullable(lastId));
sessions = ss.getSessionsOfWorkflowByName(proj.getId(), workflowName, validPageSize, Optional.fromNullable(lastId), page);
} else {
sessions = ss.getSessionsOfProject(proj.getId(), validPageSize, Optional.fromNullable(lastId));
sessions = ss.getSessionsOfProject(proj.getId(), validPageSize, Optional.fromNullable(lastId), page);
}

return RestModels.sessionCollection(ps, sessions);
return RestModels.sessionCollection(ps, sessions, sessionRecordsNumber);
}, ResourceNotFoundException.class);
}

Expand Down
Loading