cerc-io/watcher-ts
16
1
Fork 0
mirror of https://github.com/cerc-io/watcher-ts synced 2025-02-08 11:02:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support nested filters for plural queries on subgraph entities (#446)

Browse Source 
* Handle nested filters while building the database query

* Build nested filters for relational fields

* Avoid queries to get canonical block hashes for loading relations

* Handle nested filters in all query types

* Give precedence to block hash over number in time travel queries
This commit is contained in:
prathamesh0 2023-11-02 10:37:33 +05:30 committed by GitHub
parent 9fb51e89f6
commit 92f3fb8252
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 170 additions and 134 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
packages/graph-node/src/watcher.ts
View File

@ -28,7 +28,10 @@ import {
Transaction,
EthClient,
DEFAULT_LIMIT,
FILTER_CHANGE_BLOCK
FILTER_CHANGE_BLOCK,
Where,
Filter,
OPERATOR_MAP
} from '@cerc-io/util';
import { Context, GraphData, instantiate } from './loader';
@ -322,50 +325,7 @@ export class GraphWatcher {
const dbTx = await this._database.createTransactionRunner();
try {
where = Object.entries(where).reduce((acc: { [key: string]: any }, [fieldWithSuffix, value]) => {
if (fieldWithSuffix === FILTER_CHANGE_BLOCK) {
assert(value.number_gte && typeof value.number_gte === 'number');
// Maintain util.Where type
acc[FILTER_CHANGE_BLOCK] = [{
value: value.number_gte
}];
return acc;
}
const [field, ...suffix] = fieldWithSuffix.split('_');
if (!acc[field]) {
acc[field] = [];
}
const filter = {
value,
not: false,
operator: 'equals'
};
let operator = suffix.shift();
if (operator === 'not') {
filter.not = true;
operator = suffix.shift();
}
if (operator) {
filter.operator = operator;
}
// If filter field ends with "nocase", use case insensitive version of the operator
if (suffix[suffix.length - 1] === 'nocase') {
filter.operator = `${operator}_nocase`;
}
acc[field].push(filter);
return acc;
}, {});
where = this._buildFilter(where);
if (!queryOptions.limit) {
queryOptions.limit = DEFAULT_LIMIT;
@ -498,6 +458,65 @@ export class GraphWatcher {
return transaction;
}
_buildFilter (where: { [key: string]: any } = {}): Where {
return Object.entries(where).reduce((acc: Where, [fieldWithSuffix, value]) => {
if (fieldWithSuffix === FILTER_CHANGE_BLOCK) {
assert(value.number_gte && typeof value.number_gte === 'number');
acc[FILTER_CHANGE_BLOCK] = [{
value: value.number_gte,
not: false
}];
return acc;
}
const [field, ...suffix] = fieldWithSuffix.split('_');
if (!acc[field]) {
acc[field] = [];
}
const filter: Filter = {
value,
not: false,
operator: 'equals'
};
let operator = suffix.shift();
// If the operator is "" (different from undefined), it means it's a nested filter on a relation field
if (operator === '') {
acc[field].push({
// Parse nested filter value
value: this._buildFilter(value),
not: false,
operator: 'nested'
});
return acc;
}
if (operator === 'not') {
filter.not = true;
operator = suffix.shift();
}
if (operator) {
filter.operator = operator as keyof typeof OPERATOR_MAP;
}
// If filter field ends with "nocase", use case insensitive version of the operator
if (suffix[suffix.length - 1] === 'nocase') {
filter.operator = `${operator}_nocase` as keyof typeof OPERATOR_MAP;
}
acc[field].push(filter);
return acc;
}, {});
}
}
export const getGraphDbAndWatcher = async (

63
packages/util/src/database.ts
View File

@ -5,6 +5,7 @@
import assert from 'assert';
import {
Between,
Brackets,
Connection,
ConnectionOptions,
createConnection,
@ -38,7 +39,8 @@ export const OPERATOR_MAP = {
ends: 'LIKE',
contains_nocase: 'ILIKE',
starts_nocase: 'ILIKE',
ends_nocase: 'ILIKE'
ends_nocase: 'ILIKE',
nested: ''
};
const INSERT_EVENTS_BATCH = 100;
@ -48,6 +50,10 @@ export interface BlockHeight {
hash?: string;
}
export interface CanonicalBlockHeight extends BlockHeight {
canonicalBlockHashes?: string[];
}
export enum OrderDirection {
asc = 'asc',
desc = 'desc'
@ -60,12 +66,15 @@ export interface QueryOptions {
orderDirection?: OrderDirection;
}
export interface Filter {
// eslint-disable-next-line no-use-before-define
value: any | Where;
not: boolean;
operator?: keyof typeof OPERATOR_MAP;
}
export interface Where {
[key: string]: [{
value: any;
not: boolean;
operator: keyof typeof OPERATOR_MAP;
}]
[key: string]: Filter[];
}
export type Relation = string | { property: string, alias: string }
@ -820,7 +829,9 @@ export class Database {
buildQuery<Entity extends ObjectLiteral> (
repo: Repository<Entity>,
selectQueryBuilder: SelectQueryBuilder<Entity>,
where: Where = {},
where: Readonly<Where> = {},
relations: Readonly<{ [key: string]: any }> = {},
block: Readonly<CanonicalBlockHeight> = {},
alias?: string
): SelectQueryBuilder<Entity> {
if (!alias) {
@ -828,11 +839,42 @@ export class Database {
}
Object.entries(where).forEach(([field, filters]) => {
// TODO: Handle nested filters on derived and array fields
const columnMetadata = repo.metadata.findColumnWithPropertyName(field);
assert(columnMetadata);
filters.forEach((filter, index) => {
// Form the where clause.
let { not, operator, value } = filter;
const columnMetadata = repo.metadata.findColumnWithPropertyName(field);
assert(columnMetadata);
// Handle nested relation filter
const relation = relations[field];
if (operator === 'nested' && relation) {
const relationRepo = this.conn.getRepository<any>(relation.entity);
const relationTableName = relationRepo.metadata.tableName;
let relationSubQuery: SelectQueryBuilder<any> = relationRepo.createQueryBuilder(relationTableName, repo.queryRunner)
.select('1')
.where(`${relationTableName}.id = "${alias}"."${columnMetadata.databaseName}"`);
// canonicalBlockHashes take precedence over block number if provided
if (block.canonicalBlockHashes) {
relationSubQuery = relationSubQuery
.andWhere(new Brackets(qb => {
qb.where(`${relationTableName}.block_hash IN (:...relationBlockHashes)`, { relationBlockHashes: block.canonicalBlockHashes })
.orWhere(`${relationTableName}.block_number <= :relationCanonicalBlockNumber`, { relationCanonicalBlockNumber: block.number });
}));
} else if (block.number) {
relationSubQuery = relationSubQuery.andWhere(`${relationTableName}.block_number <= :blockNumber`, { blockNumber: block.number });
}
relationSubQuery = this.buildQuery(relationRepo, relationSubQuery, value);
selectQueryBuilder = selectQueryBuilder
.andWhere(`EXISTS (${relationSubQuery.getQuery()})`)
.setParameters(relationSubQuery.getParameters());
return;
}
// Form the where clause.
let whereClause = `"${alias}"."${columnMetadata.databaseName}" `;
if (columnMetadata.relationMetadata) {
@ -850,6 +892,7 @@ export class Database {
}
}
assert(operator);
whereClause += `${OPERATOR_MAP[operator]} `;
value = this._transformBigIntValues(value);

132
packages/util/src/graph/database.ts
View File

@ -22,7 +22,7 @@ import { SelectionNode } from 'graphql';
import _ from 'lodash';
import debug from 'debug';
import { BlockHeight, Database as BaseDatabase, QueryOptions, Where } from '../database';
import { BlockHeight, Database as BaseDatabase, QueryOptions, Where, CanonicalBlockHeight } from '../database';
import { BlockProgressInterface } from '../types';
import { cachePrunedEntitiesCount, eventProcessingLoadEntityCacheHitCount, eventProcessingLoadEntityCount, eventProcessingLoadEntityDBQueryDuration } from '../metrics';
import { ServerConfig } from '../config';
@ -350,7 +350,7 @@ export class GraphDatabase {
queryRunner: QueryRunner,
entityType: new () => Entity,
relationsMap: Map<any, { [key: string]: any }>,
block: BlockHeight = {},
block: CanonicalBlockHeight = {},
where: Where = {},
queryOptions: QueryOptions = {},
selections: ReadonlyArray<SelectionNode> = []
@ -375,6 +375,7 @@ export class GraphDatabase {
queryRunner,
entityType,
latestEntityType,
relationsMap,
where,
queryOptions,
selections
@ -384,21 +385,21 @@ export class GraphDatabase {
// Use different suitable query patterns based on entities.
switch (this._entityQueryTypeMap.get(entityType)) {
case ENTITY_QUERY_TYPE.SINGULAR:
entities = await this.getEntitiesSingular(queryRunner, entityType, block, where);
entities = await this.getEntitiesSingular(queryRunner, entityType, relationsMap, block, where);
break;
case ENTITY_QUERY_TYPE.UNIQUE:
entities = await this.getEntitiesUnique(queryRunner, entityType, block, where, queryOptions);
entities = await this.getEntitiesUnique(queryRunner, entityType, relationsMap, block, where, queryOptions);
break;
case ENTITY_QUERY_TYPE.DISTINCT_ON:
entities = await this.getEntitiesDistinctOn(queryRunner, entityType, block, where, queryOptions);
entities = await this.getEntitiesDistinctOn(queryRunner, entityType, relationsMap, block, where, queryOptions);
break;
case ENTITY_QUERY_TYPE.GROUP_BY:
default:
// Use group by query if entity query type is not specified in map.
entities = await this.getEntitiesGroupBy(queryRunner, entityType, block, where, queryOptions);
entities = await this.getEntitiesGroupBy(queryRunner, entityType, relationsMap, block, where, queryOptions);
break;
}
}
@ -417,7 +418,8 @@ export class GraphDatabase {
async getEntitiesGroupBy<Entity extends ObjectLiteral> (
queryRunner: QueryRunner,
entityType: new () => Entity,
block: BlockHeight,
relationsMap: Map<any, { [key: string]: any }>,
block: CanonicalBlockHeight,
where: Where = {},
queryOptions: QueryOptions = {}
): Promise<Entity[]> {
@ -440,19 +442,7 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
if (block.hash) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
subQuery = subQuery
.andWhere(new Brackets(qb => {
qb.where('subTable.block_hash IN (:...blockHashes)', { blockHashes })
.orWhere('subTable.block_number <= :canonicalBlockNumber', { canonicalBlockNumber });
}));
}
if (block.number) {
subQuery = subQuery.andWhere('subTable.block_number <= :blockNumber', { blockNumber: block.number });
}
subQuery = await this._applyBlockHeightFilter(queryRunner, subQuery, block, 'subTable');
let selectQueryBuilder = repo.createQueryBuilder(tableName)
.innerJoin(
@ -462,7 +452,7 @@ export class GraphDatabase {
)
.setParameters(subQuery.getParameters());
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where);
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, relationsMap.get(entityType), block);
if (queryOptions.orderBy) {
selectQueryBuilder = this._baseDatabase.orderQuery(repo, selectQueryBuilder, queryOptions);
@ -486,7 +476,8 @@ export class GraphDatabase {
async getEntitiesDistinctOn<Entity extends ObjectLiteral> (
queryRunner: QueryRunner,
entityType: new () => Entity,
block: BlockHeight,
relationsMap: Map<any, { [key: string]: any }>,
block: CanonicalBlockHeight,
where: Where = {},
queryOptions: QueryOptions = {}
): Promise<Entity[]> {
@ -508,21 +499,9 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
if (block.hash) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
subQuery = await this._applyBlockHeightFilter(queryRunner, subQuery, block, 'subTable');
subQuery = subQuery
.andWhere(new Brackets(qb => {
qb.where('subTable.block_hash IN (:...blockHashes)', { blockHashes })
.orWhere('subTable.block_number <= :canonicalBlockNumber', { canonicalBlockNumber });
}));
}
if (block.number) {
subQuery = subQuery.andWhere('subTable.block_number <= :blockNumber', { blockNumber: block.number });
}
subQuery = this._baseDatabase.buildQuery(repo, subQuery, where);
subQuery = this._baseDatabase.buildQuery(repo, subQuery, where, relationsMap.get(entityType), block);
let selectQueryBuilder = queryRunner.manager.createQueryBuilder()
.from(
@ -555,6 +534,7 @@ export class GraphDatabase {
async getEntitiesSingular<Entity extends ObjectLiteral> (
queryRunner: QueryRunner,
entityType: new () => Entity,
relationsMap: Map<any, { [key: string]: any }>,
block: BlockHeight,
where: Where = {}
): Promise<Entity[]> {
@ -571,21 +551,9 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
if (block.hash) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
selectQueryBuilder = await this._applyBlockHeightFilter(queryRunner, selectQueryBuilder, block, tableName);
selectQueryBuilder = selectQueryBuilder
.andWhere(new Brackets(qb => {
qb.where(`${tableName}.block_hash IN (:...blockHashes)`, { blockHashes })
.orWhere(`${tableName}.block_number <= :canonicalBlockNumber`, { canonicalBlockNumber });
}));
}
if (block.number) {
selectQueryBuilder = selectQueryBuilder.andWhere(`${tableName}.block_number <= :blockNumber`, { blockNumber: block.number });
}
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where);
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, relationsMap.get(entityType), block);
const entities = await selectQueryBuilder.getMany();
@ -595,6 +563,7 @@ export class GraphDatabase {
async getEntitiesUnique<Entity extends ObjectLiteral> (
queryRunner: QueryRunner,
entityType: new () => Entity,
relationsMap: Map<any, { [key: string]: any }>,
block: BlockHeight,
where: Where = {},
queryOptions: QueryOptions = {}
@ -610,21 +579,9 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
if (block.hash) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
selectQueryBuilder = await this._applyBlockHeightFilter(queryRunner, selectQueryBuilder, block, tableName);
selectQueryBuilder = selectQueryBuilder
.andWhere(new Brackets(qb => {
qb.where(`${tableName}.block_hash IN (:...blockHashes)`, { blockHashes })
.orWhere(`${tableName}.block_number <= :canonicalBlockNumber`, { canonicalBlockNumber });
}));
}
if (block.number) {
selectQueryBuilder = selectQueryBuilder.andWhere(`${tableName}.block_number <= :blockNumber`, { blockNumber: block.number });
}
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where);
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, relationsMap.get(entityType), block);
if (queryOptions.orderBy) {
selectQueryBuilder = this._baseDatabase.orderQuery(repo, selectQueryBuilder, queryOptions);
@ -649,6 +606,7 @@ export class GraphDatabase {
queryRunner: QueryRunner,
entityType: new () => Entity,
latestEntity: new () => any,
relationsMap: Map<any, { [key: string]: any }>,
where: Where = {},
queryOptions: QueryOptions = {},
selections: ReadonlyArray<SelectionNode> = []
@ -685,7 +643,7 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, 'latest');
selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, relationsMap.get(entityType), {}, 'latest');
if (queryOptions.orderBy) {
selectQueryBuilder = this._baseDatabase.orderQuery(repo, selectQueryBuilder, queryOptions, '', 'latest');
@ -726,19 +684,7 @@ export class GraphDatabase {
delete where[FILTER_CHANGE_BLOCK];
}
if (block.hash) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
subQuery = subQuery
.andWhere(new Brackets(qb => {
qb.where('subTable.block_hash IN (:...blockHashes)', { blockHashes })
.orWhere('subTable.block_number <= :canonicalBlockNumber', { canonicalBlockNumber });
}));
}
if (block.number) {
subQuery = subQuery.andWhere('subTable.block_number <= :blockNumber', { blockNumber: block.number });
}
subQuery = await this._applyBlockHeightFilter(queryRunner, subQuery, block, 'subTable');
let selectQueryBuilder = latestEntityRepo.createQueryBuilder('latest')
.select('*')
@ -752,7 +698,7 @@ export class GraphDatabase {
'result'
) as SelectQueryBuilder<Entity>;
selectQueryBuilder = this._baseDatabase.buildQuery(latestEntityRepo, selectQueryBuilder, where, 'latest');
selectQueryBuilder = this._baseDatabase.buildQuery(latestEntityRepo, selectQueryBuilder, where, {}, {}, 'latest');
if (queryOptions.orderBy) {
selectQueryBuilder = this._baseDatabase.orderQuery(latestEntityRepo, selectQueryBuilder, queryOptions, '', 'latest');
@ -1283,6 +1229,34 @@ export class GraphDatabase {
);
}
async _applyBlockHeightFilter<Entity> (
queryRunner: QueryRunner,
queryBuilder: SelectQueryBuilder<Entity>,
block: CanonicalBlockHeight,
alias: string
): Promise<SelectQueryBuilder<Entity>> {
// Block hash takes precedence over number if provided
if (block.hash) {
if (!block.canonicalBlockHashes) {
const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
// Update the block field to avoid firing the same query further
block.number = canonicalBlockNumber;
block.canonicalBlockHashes = blockHashes;
}
queryBuilder = queryBuilder
.andWhere(new Brackets(qb => {
qb.where(`${alias}.block_hash IN (:...blockHashes)`, { blockHashes: block.canonicalBlockHashes })
.orWhere(`${alias}.block_number <= :canonicalBlockNumber`, { canonicalBlockNumber: block.number });
}));
} else if (block.number) {
queryBuilder = queryBuilder.andWhere(`${alias}.block_number <= :blockNumber`, { blockNumber: block.number });
}
return queryBuilder;
}
_measureCachedPrunedEntities (): void {
const totalEntities = Array.from(this.cachedEntities.latestPrunedEntities.values())
.reduce((acc, idEntitiesMap) => acc + idEntitiesMap.size, 0);