Skip to content

Commit

Permalink
MODFQMMGR-225: Enable cross-tenant querying in ECS environments
Browse files Browse the repository at this point in the history
  • Loading branch information
@bvsharp
bvsharp committed Jul 31, 2024
1 parent 10654ef commit fb58ed3
Show file tree
Hide file tree
Showing 31 changed files with 563 additions and 242 deletions.
2 changes: 1 addition & 1 deletion src/main/java/org/folio/fqm/aspect/EntityTypePermissionsAspect.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ Object validatePermissionsWithContentsRequest(ProceedingJoinPoint joinPoint) thr
}

private EntityType getEntityTypeFromId(UUID entityTypeId) {
return entityTypeRepository.getEntityTypeDefinition(entityTypeId)
return entityTypeRepository.getEntityTypeDefinition(entityTypeId, null)
.orElseThrow(() -> new EntityTypeNotFoundException(entityTypeId));
}

Expand Down
14 changes: 9 additions & 5 deletions src/main/java/org/folio/fqm/repository/EntityTypeRepository.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,18 +48,22 @@ public class EntityTypeRepository {
private final DSLContext jooqContext;
private final ObjectMapper objectMapper;

public Optional<EntityType> getEntityTypeDefinition(UUID entityTypeId) {
return getEntityTypeDefinitions(Collections.singleton(entityTypeId)).findFirst();
public Optional<EntityType> getEntityTypeDefinition(UUID entityTypeId, String tenantId) {
var returnValue = getEntityTypeDefinitions(Collections.singleton(entityTypeId), tenantId).findFirst();
return returnValue;
}

public Stream<EntityType> getEntityTypeDefinitions(Collection<UUID> entityTypeIds) {
public Stream<EntityType> getEntityTypeDefinitions(Collection<UUID> entityTypeIds, String tenantId) {
String tableName = tenantId == null ? TABLE_NAME : tenantId + "_mod_fqm_manager." + TABLE_NAME;
log.info("Getting definitions name for entity type ID: {}", entityTypeIds);

Field<String> definitionField = field(DEFINITION_FIELD_NAME, String.class);
Condition entityTypeIdCondition = isEmpty(entityTypeIds) ? trueCondition() : field("id").in(entityTypeIds);
return readerJooqContext
.select(definitionField)
.from(table(TABLE_NAME))
// .from(table(TABLE_NAME))
// .from(table("college_mod_fqm_manager." + TABLE_NAME))
.from(table(tableName))
.where(entityTypeIdCondition)
.fetch(definitionField)
.stream()
Expand Down Expand Up @@ -99,7 +103,7 @@ public void replaceEntityTypeDefinitions(List<EntityType> entityTypes) {

private List<EntityTypeColumn> fetchColumnNamesForCustomFields(UUID entityTypeId) {
log.info("Getting columns for entity type ID: {}", entityTypeId);
EntityType entityTypeDefinition = getEntityTypeDefinition(entityTypeId)
EntityType entityTypeDefinition = getEntityTypeDefinition(entityTypeId, null)
.orElseThrow(() -> new EntityTypeNotFoundException(entityTypeId));
String sourceViewName = entityTypeDefinition.getSourceView();
String sourceViewExtractor = entityTypeDefinition.getSourceViewExtractor();
Expand Down
131 changes: 84 additions & 47 deletions src/main/java/org/folio/fqm/repository/IdStreamer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,20 @@
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.folio.fqm.model.IdsWithCancelCallback;
import org.folio.fqm.service.CrossTenantQueryService;
import org.folio.fqm.service.EntityTypeFlatteningService;
import org.folio.fqm.service.FqlToSqlConverterService;
import org.folio.fqm.utils.IdColumnUtils;
import org.folio.fqm.utils.EntityTypeUtils;
import org.folio.fqm.utils.StreamHelper;
import org.folio.fql.model.Fql;
import org.folio.querytool.domain.dto.EntityType;
import org.folio.querytool.domain.dto.EntityTypeDefaultSort;
import org.folio.spring.FolioExecutionContext;
import org.jooq.Condition;
import org.jooq.Cursor;
import org.jooq.DSLContext;
import org.jooq.ResultQuery;
import org.jooq.Select;
import org.jooq.SortField;
import org.jooq.impl.DSL;
import org.springframework.beans.factory.annotation.Autowired;
Expand All @@ -22,6 +25,7 @@
import org.jooq.Record1;
import org.jooq.Field;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;
Expand All @@ -30,8 +34,9 @@
import java.util.stream.Stream;

import static org.apache.commons.lang3.ObjectUtils.isEmpty;
import static org.folio.fqm.utils.IdColumnUtils.RESULT_ID_FIELD;
import static org.folio.fqm.utils.EntityTypeUtils.RESULT_ID_FIELD;
import static org.jooq.impl.DSL.field;
import static org.jooq.impl.DSL.select;
import static org.jooq.impl.DSL.table;

@Repository
Expand All @@ -42,6 +47,8 @@ public class IdStreamer {
@Qualifier("readerJooqContext")
private final DSLContext jooqContext;
private final EntityTypeFlatteningService entityTypeFlatteningService;
private final FolioExecutionContext executionContext;
private final CrossTenantQueryService crossTenantQueryService;

/**
* Executes the given Fql Query and stream the result Ids back.
Expand All @@ -52,7 +59,11 @@ public int streamIdsInBatch(EntityType entityType,
int batchSize,
Consumer<IdsWithCancelCallback> idsConsumer) {
Condition sqlWhereClause = FqlToSqlConverterService.getSqlCondition(fql.fqlCondition(), entityType);
return this.streamIdsInBatch(entityType, sortResults, sqlWhereClause, batchSize, idsConsumer);
List<String> tenantsToQuery = crossTenantQueryService.getTenantsToQuery(UUID.fromString(entityType.getId()));
// return this.streamIdsInBatch(entityType, sortResults, sqlWhereClause, batchSize, idsConsumer);
// TODO: temporarily using hard-coded sortResults=true, change back when done testing
return this.streamIdsInBatchCrossTenant(entityType, sortResults, sqlWhereClause, batchSize, idsConsumer, tenantsToQuery);
// return this.streamIdsInBatchCrossTenant(entityType, true, sqlWhereClause, batchSize, idsConsumer, tenantsToQuery);
}

public List<List<String>> getSortedIds(String derivedTableName,
Expand All @@ -74,68 +85,94 @@ public List<List<String>> getSortedIds(String derivedTableName,
.toList();
}

private int streamIdsInBatch(EntityType entityType,
boolean sortResults,
Condition sqlWhereClause,
int batchSize,
Consumer<IdsWithCancelCallback> idsConsumer) {
String finalJoinClause = entityTypeFlatteningService.getJoinClause(entityType);
Field<String[]> idValueGetter = IdColumnUtils.getResultIdValueGetter(entityType);
ResultQuery<Record1<String[]>> query = null;
if (!isEmpty(entityType.getGroupByFields())) {
Field<?>[] groupByFields = entityType
.getColumns()
.stream()
.filter(col -> entityType.getGroupByFields().contains(col.getName()))
.map(col -> col.getFilterValueGetter() == null ? col.getValueGetter() : col.getFilterValueGetter())
.map(DSL::field)
.toArray(Field[]::new);
query = jooqContext.dsl()
.select(field(idValueGetter))
.from(finalJoinClause)
.where(sqlWhereClause)
.groupBy(groupByFields)
.orderBy(getSortFields(entityType, sortResults))
.fetchSize(batchSize);
} else {
query = jooqContext.dsl()
.select(field(idValueGetter))
.from(finalJoinClause)
.where(sqlWhereClause)
.orderBy(getSortFields(entityType, sortResults))
.fetchSize(batchSize);
private int streamIdsInBatchCrossTenant(EntityType entityType,
boolean sortResults,
Condition sqlWhereClause,
int batchSize,
Consumer<IdsWithCancelCallback> idsConsumer, List<String> tenantsToQuery) {
UUID entityTypeId = UUID.fromString(entityType.getId());
log.info("Streaming ids. Cross tenant: {}", tenantsToQuery.size() > 1);


log.info("List of tenants to query: ");
tenantsToQuery.forEach(System.out::println);
List<ResultQuery<Record1<String[]>>> innerQueries = new ArrayList<>();

// TODO: CAN'T USE THE SAME IDVALUEGETTER, HAVE TO GET INDIVIDUALLY
Field<String[]> idValueGetter = EntityTypeUtils.getResultIdValueGetter(entityType);
// TODO: think about handling batchSize and sortResults

Select<Record1<String[]>> fullQuery = null;

for (String tenantId : tenantsToQuery) {
log.info("Creating query for tenant {}", tenantId);
log.info("Getting entity type definition for tenant {}", tenantId);
EntityType entityTypeDefinition = entityTypeFlatteningService.getFlattenedEntityType(entityTypeId, true, tenantId);
log.info("Got entity type: ", entityTypeDefinition);

var currentIdValueGetter = EntityTypeUtils.getResultIdValueGetter(entityTypeDefinition);
String innerJoinClause = entityTypeFlatteningService.getJoinClause(entityTypeDefinition, tenantId);
ResultQuery<Record1<String[]>> innerQuery = buildQuery(entityTypeDefinition, currentIdValueGetter, innerJoinClause, sqlWhereClause, sortResults, batchSize);
log.info("Inner query: {}", innerQuery);
innerQueries.add(innerQuery);
}

log.info("Built all inner queries");

for (var innerQuery : innerQueries) {
if (fullQuery == null) {
fullQuery = (Select<Record1<String[]>>) innerQuery;
} else {
fullQuery = fullQuery.unionAll((Select<Record1<String[]>>) innerQuery);
}
}

log.info("Full query: {}", fullQuery);

try (
Cursor<Record1<String[]>> idsCursor = query.fetchLazy();
Cursor<Record1<String[]>> idsCursor = fullQuery.fetchLazy();
Stream<String[]> idStream = idsCursor
.stream()
.map(row -> row.getValue(idValueGetter));
.map(row -> {
log.info("row: {}", row); // TODO: remove logging
return row.getValue(idValueGetter);
});
Stream<List<String[]>> idsStream = StreamHelper.chunk(idStream, batchSize)
) {
var total = new AtomicInteger();
idsStream.map(ids -> new IdsWithCancelCallback(ids, idsStream::close))
.forEach(idsWithCancelCallback -> {
// problem is duplicate (id, source_tenant_id) tuple for shared instances with shadow records in data tenant
idsConsumer.accept(idsWithCancelCallback);
total.addAndGet(idsWithCancelCallback.ids().size());
});
return total.get();
}
}

private List<SortField<Object>> getSortFields(EntityType entityType, boolean sortResults) {
if (sortResults && !isEmpty(entityType.getDefaultSort())) {
return entityType
.getDefaultSort()
private ResultQuery<Record1<String[]>> buildQuery(EntityType entityType, Field<String[]> idValueGetter, String finalJoinClause, Condition sqlWhereClause, boolean sortResults, int batchSize) {
if (!isEmpty(entityType.getGroupByFields())) {
Field<?>[] groupByFields = entityType
.getColumns()
.stream()
.map(IdStreamer::toSortField)
.toList();
.filter(col -> entityType.getGroupByFields().contains(col.getName()))
.map(col -> col.getFilterValueGetter() == null ? col.getValueGetter() : col.getFilterValueGetter())
.map(DSL::field)
.toArray(Field[]::new);
return jooqContext.dsl()
.select(field(idValueGetter))
.from(finalJoinClause)
.where(sqlWhereClause)
.groupBy(groupByFields)
.orderBy(EntityTypeUtils.getSortFields(entityType, sortResults))
.fetchSize(batchSize);
} else {
return jooqContext.dsl()
.select(field(idValueGetter))
.from(finalJoinClause)
.where(sqlWhereClause)
.orderBy(EntityTypeUtils.getSortFields(entityType, sortResults))
.fetchSize(batchSize);
}
return List.of();
}

private static SortField<Object> toSortField(EntityTypeDefaultSort entityTypeDefaultSort) {
Field<Object> field = field(entityTypeDefaultSort.getColumnName());
return entityTypeDefaultSort.getDirection() == EntityTypeDefaultSort.DirectionEnum.DESC ? field.desc() : field.asc();
}
}
2 changes: 1 addition & 1 deletion src/main/java/org/folio/fqm/repository/QueryResultsRepository.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
import java.util.List;
import java.util.UUID;

import static org.folio.fqm.utils.IdColumnUtils.RESULT_ID_FIELD;
import static org.folio.fqm.utils.EntityTypeUtils.RESULT_ID_FIELD;
import static org.jooq.impl.DSL.table;
import static org.jooq.impl.DSL.field;

Expand Down
Loading

0 comments on commit fb58ed3

Please sign in to comment.