Skip to content

Fix cursor pagination with optional tie-breaker #2080

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

Merged
ravikiranvm merged 10 commits into from
Mar 17, 2026
Merged

Fix cursor pagination with optional tie-breaker #2080

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1635750
Fix cursor pagination with optional tie-breaker
ravikiranvm Mar 6, 2026
0e1098e
Remove some unnecessary changes
ravikiranvm Mar 9, 2026
76010d0
Use better names
ravikiranvm Mar 9, 2026
758ff11
Add integration test cases
ravikiranvm Mar 9, 2026
7bd00f4
Guard secondary column to be used only with primary column
ravikiranvm Mar 9, 2026
4e1d6f3
Fix sonar comments
ravikiranvm Mar 9, 2026
8fc7a8c
Fix sonar
ravikiranvm Mar 9, 2026
38489e8
Extract a func to be used in multiple test cases
ravikiranvm Mar 9, 2026
7e8d234
Merge branch 'main' into ops-3810
ravikiranvm Mar 16, 2026
612ddac
Remove unnecessary func
ravikiranvm Mar 16, 2026
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
5 changes: 5 additions & 0 deletions packages/server/api/src/app/flows/flow/flow.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,11 @@ export const flowService = {
columnName: 'fv.updated',
columnType: 'timestamp with time zone',
},
customPaginationSecondaryColumn: {
columnPath: 'id',
columnName: 'flow.id',
columnType: 'string',
},
});

const queryWhere: Record<string, unknown> = {
Expand Down
41 changes: 35 additions & 6 deletions packages/server/api/src/app/helper/pagination/build-paginator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,16 +8,28 @@ export type PagingQuery = {
order?: Order | 'ASC' | 'DESC';
};

type CustomPaginationColumnOptions = {
columnPath: string;
columnName: string;
columnType?: string;
};

// Secondary custom pagination is only valid when primary custom pagination is configured.
type CustomPaginationColumns =
| {
customPaginationColumn?: never;
customPaginationSecondaryColumn?: never;
}
| {
customPaginationColumn: CustomPaginationColumnOptions;
customPaginationSecondaryColumn?: CustomPaginationColumnOptions;
};

export type PaginationOptions<Entity> = {
entity: EntitySchema<Entity>;
alias?: string;
query?: PagingQuery;
customPaginationColumn?: {
columnPath: string;
columnName: string;
columnType?: string;
};
};
} & CustomPaginationColumns;

export function buildPaginator<Entity extends ObjectLiteral>(
options: PaginationOptions<Entity>,
Expand All @@ -32,6 +44,15 @@ export function buildPaginator<Entity extends ObjectLiteral>(

paginator.setAlias(alias);

if (
options.customPaginationSecondaryColumn &&
!options.customPaginationColumn
) {
throw new Error(
'customPaginationSecondaryColumn requires customPaginationColumn',
);
}

if (options.customPaginationColumn) {
paginator.setPaginationColumn(
options.customPaginationColumn.columnPath,
Expand All @@ -40,6 +61,14 @@ export function buildPaginator<Entity extends ObjectLiteral>(
);
}

if (options.customPaginationSecondaryColumn) {
paginator.setPaginationSecondaryColumn(
options.customPaginationSecondaryColumn.columnPath,
options.customPaginationSecondaryColumn.columnName,
options.customPaginationSecondaryColumn.columnType,
);
}

if (query.afterCursor) {
paginator.setAfterCursor(query.afterCursor);
}
Expand Down
243 changes: 221 additions & 22 deletions packages/server/api/src/app/helper/pagination/paginator.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,16 @@ export type PagingResult<Entity> = {
cursor: CursorResult;
};

type CursorContext = {
primaryColumnName: string;
primaryParamName: string;
secondaryColumnName: string | null;
secondaryParamName: string | null;
};

const PAGINATION_KEY = 'created';
const CUSTOM_PAGINATION_KEY = 'custom_pagination';
const CUSTOM_PAGINATION_SECONDARY_KEY = 'custom_pagination_tie_breaker';
const DEFAULT_TIMESTAMP_TYPE = 'timestamp with time zone';

export default class Paginator<Entity extends ObjectLiteral> {
Expand All @@ -56,6 +64,12 @@ export default class Paginator<Entity extends ObjectLiteral> {

private paginationColumnType: string | null = null;

private paginationSecondaryColumnPath: string | null = null;

private paginationSecondaryColumnName: string | null = null;

private paginationSecondaryColumnType: string | null = null;

public constructor(private readonly entity: EntitySchema) {}

public setPaginationColumn(
Expand All @@ -72,6 +86,16 @@ export default class Paginator<Entity extends ObjectLiteral> {
this.alias = alias;
}

public setPaginationSecondaryColumn(
columnPath: string,
columnName: string,
columnType = 'string',
): void {
this.paginationSecondaryColumnPath = columnPath;
this.paginationSecondaryColumnName = columnName;
this.paginationSecondaryColumnType = columnType;
}

public setAfterCursor(cursor: string): void {
this.afterCursor = cursor;
}
Expand Down Expand Up @@ -171,28 +195,26 @@ export default class Paginator<Entity extends ObjectLiteral> {
): void {
const dbType = system.get(AppSystemProp.DB_TYPE);
const operator = this.getOperator();
let queryString: string;

const isCustomColumn =
this.paginationColumnName && cursors[CUSTOM_PAGINATION_KEY];
const columnName = isCustomColumn
? this.paginationColumnName
: `${this.alias}.${PAGINATION_KEY}`;
const paramName = isCustomColumn ? CUSTOM_PAGINATION_KEY : PAGINATION_KEY;
Comment on lines -176 to -181
Copy link
Copy Markdown
Contributor Author

@ravikiranvm ravikiranvm Mar 16, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

We compute whether the custom pagination key is being used and derive columnName and paramName from that. I moved that logic into resolveCursorContext, and it now also supports resolving the secondary column and parameter name when composite pagination is active.

openops/packages/server/api/src/app/helper/pagination/paginator.ts

Line 396 in 7e8d234

private resolveCursorContext(cursors: CursorParam): CursorContext {


if (dbType === DatabaseType.SQLITE3) {
queryString = `${columnName} ${operator} :${paramName}`;
} else if (dbType === DatabaseType.POSTGRES) {
if (this.hasBeforeCursor() && !this.hasAfterCursor()) {
queryString = `${columnName} ${operator} (:${paramName}::timestamp + INTERVAL '1 millisecond')`;
} else {
queryString = `${columnName} ${operator} :${paramName}::timestamp`;
Comment on lines -186 to -189
Copy link
Copy Markdown
Contributor Author

@ravikiranvm ravikiranvm Mar 16, 2026

Choose a reason for hiding this comment

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

The query-string construction logic is now split across three helpers:

  • applySingleColumnCursorFilter
  • applyCompositeCursorFilter
  • buildComparisonClause

where the first two decide the shape of the filter, and buildComparisonClause builds the DB comparison fragments they use.

}
} else {
throw new Error('Unsupported database type');
const context = this.resolveCursorContext(cursors);

if (context.secondaryColumnName && context.secondaryParamName) {
this.applyCompositeCursorFilter(
where,
cursors,
dbType,
operator,
context,
);
return;
}

where.orWhere(queryString, cursors);
this.applySingleColumnCursorFilter(
where,
cursors,
dbType,
operator,
context,
);
}

private getOperator(): string {
Expand Down Expand Up @@ -222,6 +244,10 @@ export default class Paginator<Entity extends ObjectLiteral> {
orderByCondition[`${this.alias}.${PAGINATION_KEY}`] = order;
}

if (this.paginationColumnName && this.paginationSecondaryColumnName) {
orderByCondition[this.paginationSecondaryColumnName] = order;
}

return orderByCondition;
}

Expand Down Expand Up @@ -257,9 +283,31 @@ export default class Paginator<Entity extends ObjectLiteral> {
this.paginationColumnType || DEFAULT_TIMESTAMP_TYPE,
value,
);
const payload = `${CUSTOM_PAGINATION_KEY}:${encodedValue}`;
const payload = [`${CUSTOM_PAGINATION_KEY}:${encodedValue}`];

return btoa(payload);
if (
this.paginationSecondaryColumnPath &&
this.paginationSecondaryColumnName
) {
const secondaryValue = getValueByPath(
entity,
this.paginationSecondaryColumnPath,
);
if (secondaryValue === null || secondaryValue === undefined) {
throw new Error(
`Pagination secondary column not found at path: ${this.paginationSecondaryColumnPath}`,
);
}
const encodedSecondaryValue = encodeByType(
this.paginationSecondaryColumnType || 'string',
secondaryValue,
);
payload.push(
`${CUSTOM_PAGINATION_SECONDARY_KEY}:${encodedSecondaryValue}`,
);
}

return btoa(payload.join(','));
}

private decode(cursor: string): CursorParam {
Expand All @@ -279,6 +327,9 @@ export default class Paginator<Entity extends ObjectLiteral> {
if (key === CUSTOM_PAGINATION_KEY) {
return this.paginationColumnType || DEFAULT_TIMESTAMP_TYPE;
}
if (key === CUSTOM_PAGINATION_SECONDARY_KEY) {
return this.paginationSecondaryColumnType || 'string';
}

const col = this.entity.options.columns[key];
if (col === undefined) {
Expand All @@ -291,6 +342,154 @@ export default class Paginator<Entity extends ObjectLiteral> {
return order === Order.ASC ? Order.DESC : Order.ASC;
}

private buildComparisonClause({
Copy link
Copy Markdown
Contributor Author

@ravikiranvm ravikiranvm Mar 16, 2026 •
edited
Loading

Choose a reason for hiding this comment

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

We used to build the DB specific comparison inline for a single pagination key. This logic is now extracted and expanded to support both timestamp and non timestamp fields, because pagination now needs to compare both the primary key and the secondary key.

The main functional addition here is the timestamp = handling with a 1ms window, which is used to treat rows in the same timestamp bucket as equal before applying the secondary key.

dbType,
columnName,
paramName,
operator,
}: {
dbType: string | undefined;
columnName: string;
paramName: string;
operator: string;
}): string {
if (dbType === DatabaseType.SQLITE3) {
return `${columnName} ${operator} :${paramName}`;
}

if (dbType === DatabaseType.POSTGRES) {
const type = this.getEntityPropertyType(paramName);
if (this.isTimestampType(type)) {
if (operator === '<') {
return `${columnName} < :${paramName}::timestamptz`;
}
if (operator === '>') {
return `${columnName} >= (:${paramName}::timestamptz + INTERVAL '1 millisecond')`;
}
if (operator === '=') {
return `(${columnName} >= :${paramName}::timestamptz AND ${columnName} < (:${paramName}::timestamptz + INTERVAL '1 millisecond'))`;
}
return `${columnName} ${operator} :${paramName}::timestamptz`;
}
return `${columnName} ${operator} :${paramName}`;
}

throw new Error('Unsupported database type');
}

private isTimestampType(type: string): boolean {
return (
type === 'timestamp with time zone' ||
type === 'datetime' ||
type === 'date'
);
}

private resolveCursorContext(cursors: CursorParam): CursorContext {
const customPaginationColumnName = this.paginationColumnName;
const hasCustomPaginationCursor =
customPaginationColumnName !== null &&
cursors[CUSTOM_PAGINATION_KEY] !== undefined;

const primaryColumnName =
hasCustomPaginationCursor && customPaginationColumnName
? customPaginationColumnName
: `${this.alias}.${PAGINATION_KEY}`;
const primaryParamName = hasCustomPaginationCursor
? CUSTOM_PAGINATION_KEY
: PAGINATION_KEY;

const hasCustomSecondaryCursor =
this.paginationSecondaryColumnName !== null &&
cursors[CUSTOM_PAGINATION_SECONDARY_KEY] !== undefined;

if (hasCustomPaginationCursor && hasCustomSecondaryCursor) {
return {
primaryColumnName,
primaryParamName,
secondaryColumnName: this.paginationSecondaryColumnName,
secondaryParamName: CUSTOM_PAGINATION_SECONDARY_KEY,
};
}

return {
primaryColumnName,
primaryParamName,
secondaryColumnName: null,
secondaryParamName: null,
};
}

private applySingleColumnCursorFilter(
where: WhereExpressionBuilder,
cursors: CursorParam,
dbType: string | undefined,
operator: string,
context: CursorContext,
): void {
where.orWhere(
this.buildComparisonClause({
dbType,
columnName: context.primaryColumnName,
paramName: context.primaryParamName,
operator,
}),
cursors,
);
}

private applyCompositeCursorFilter(
Copy link
Copy Markdown
Contributor Author

@ravikiranvm ravikiranvm Mar 16, 2026

Choose a reason for hiding this comment

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

This is the main functional change for the fix. It builds two branches in the WHERE condition: the first branch handles rows that are already beyond the cursor based on the primary key, and the second branch handles rows that have the same primary value as the cursor row and therefore need the secondary key to decide paging order correctly. Earlier only the first branch existed, which is why rows with the same primary value could be skipped.

where: WhereExpressionBuilder,
cursors: CursorParam,
dbType: string | undefined,
operator: string,
context: CursorContext,
): void {
const {
primaryColumnName,
primaryParamName,
secondaryColumnName,
secondaryParamName,
} = context;
if (!secondaryColumnName || !secondaryParamName) {
throw new Error('Pagination secondary context is not configured');
}

where.orWhere(
this.buildComparisonClause({
dbType,
columnName: primaryColumnName,
paramName: primaryParamName,
operator,
}),
cursors,
);

// Lexicographic cursor compare: primary equals, then compare secondary key.
where.orWhere(
new Brackets((nestedWhere) => {
nestedWhere.where(
this.buildComparisonClause({
dbType,
columnName: primaryColumnName,
paramName: primaryParamName,
operator: '=',
}),
cursors,
);
nestedWhere.andWhere(
this.buildComparisonClause({
dbType,
columnName: secondaryColumnName,
paramName: secondaryParamName,
operator,
}),
cursors,
);
}),
);
}

private toPagingResult<Entity>(entities: Entity[]): PagingResult<Entity> {
return {
data: entities,
Expand Down
Loading
Loading