From e85657d13b7c8432dec37326dbd62397da75a114 Mon Sep 17 00:00:00 2001 From: Volodymyr Rohach Date: Tue, 26 Nov 2024 15:21:11 +0100 Subject: [PATCH] MODSOURCE-824: Endpoint /batch/parsed-records/fetch does not return deleted records (#655) * MODSOURCE-824: Implementation done. * MODSOURCE-824: Tests added+news updated. * MODSOURCE-824: raml-storage updated. * MODSOURCE-824: NEWS updated[2] * MODSOURCE-824: raml-storage updated. * MODSOURCE-824: raml-storage updated[3]. --- NEWS.md | 3 +- .../main/java/org/folio/dao/RecordDao.java | 3 +- .../java/org/folio/dao/RecordDaoImpl.java | 552 +++++++++--------- .../org/folio/services/RecordServiceImpl.java | 3 +- .../rest/impl/SourceStorageBatchApiTest.java | 216 ++++++- .../org/folio/services/RecordServiceTest.java | 137 +++++ ramls/raml-storage | 2 +- 7 files changed, 638 insertions(+), 278 deletions(-) diff --git a/NEWS.md b/NEWS.md index 6ef4de600..98237976d 100644 --- a/NEWS.md +++ b/NEWS.md @@ -1,5 +1,6 @@ -## 2024-xx-xx 5.10.0 +## 2025-xx-xx 5.10.0-SNAPSHOT * [MODSOURCE-816](https://folio-org.atlassian.net/browse/MODSOURCE-816) [RRT] Optimize execution plan for streaming SQL +* [MODSOURCE-824](https://folio-org.atlassian.net/browse/MODSOURCE-824) Endpoint /batch/parsed-records/fetch does not return deleted records ## 2024-10-28 5.9.0 * [MODSOURCE-767](https://folio-org.atlassian.net/browse/MODSOURCE-767) Single record overlay creates duplicate OCLC#/035 diff --git a/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDao.java b/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDao.java index 723f2a406..7ca5ec355 100644 --- a/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDao.java +++ b/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDao.java @@ -70,10 +70,11 @@ public interface RecordDao { * @param externalIds list of ids * @param idType external id type on which source record will be searched * @param recordType record type + * @param includeDeleted searching by deleted records * @param tenantId tenant id * @return {@link Future} of {@link StrippedParsedRecordCollection} */ - Future getStrippedParsedRecords(List externalIds, IdType idType, RecordType recordType, String tenantId); + Future getStrippedParsedRecords(List externalIds, IdType idType, RecordType recordType, Boolean includeDeleted, String tenantId); /** * Searches for {@link Record} by {@link MatchField} with offset and limit diff --git a/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDaoImpl.java b/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDaoImpl.java index e8441c9b1..cfb22242d 100644 --- a/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDaoImpl.java +++ b/mod-source-record-storage-server/src/main/java/org/folio/dao/RecordDaoImpl.java @@ -49,6 +49,7 @@ import io.vertx.reactivex.pgclient.PgPool; import io.vertx.reactivex.sqlclient.SqlConnection; import io.vertx.sqlclient.Row; + import java.sql.Connection; import java.sql.SQLException; import java.time.OffsetDateTime; @@ -67,6 +68,7 @@ import java.util.stream.Collectors; import javax.ws.rs.BadRequestException; import javax.ws.rs.NotFoundException; + import net.sf.jsqlparser.JSQLParserException; import net.sf.jsqlparser.expression.BinaryExpression; import net.sf.jsqlparser.expression.Expression; @@ -210,30 +212,30 @@ public class RecordDaoImpl implements RecordDao { private static final String DELETE_MARC_INDEXERS_TEMP_TABLE = "marc_indexers_deleted_ids"; private static final String DELETE_OLD_MARC_INDEXERS_SQL = "WITH deleted_rows AS ( " + - " delete from marc_indexers mi " + - " where exists( " + - " select 1 " + - " from " + MARC_RECORDS_TRACKING.getName() + " mrt " + - " where mrt.is_dirty = true " + - " and mrt.marc_id = mi.marc_id " + - " and mrt.version > mi.version " + - " ) " + - " returning mi.marc_id), " + - "deleted_rows2 AS ( " + - " delete from marc_indexers mi " + - " where exists( " + - " select 1 " + - " from records_lb " + - " where records_lb.id = mi.marc_id " + - " and records_lb.state = 'OLD' " + - " ) " + - " returning mi.marc_id) " + - "INSERT INTO " + DELETE_MARC_INDEXERS_TEMP_TABLE + " " + - "SELECT DISTINCT marc_id " + - "FROM deleted_rows " + - "UNION " + - "SELECT marc_id " + - "FROM deleted_rows2"; + " delete from marc_indexers mi " + + " where exists( " + + " select 1 " + + " from " + MARC_RECORDS_TRACKING.getName() + " mrt " + + " where mrt.is_dirty = true " + + " and mrt.marc_id = mi.marc_id " + + " and mrt.version > mi.version " + + " ) " + + " returning mi.marc_id), " + + "deleted_rows2 AS ( " + + " delete from marc_indexers mi " + + " where exists( " + + " select 1 " + + " from records_lb " + + " where records_lb.id = mi.marc_id " + + " and records_lb.state = 'OLD' " + + " ) " + + " returning mi.marc_id) " + + "INSERT INTO " + DELETE_MARC_INDEXERS_TEMP_TABLE + " " + + "SELECT DISTINCT marc_id " + + "FROM deleted_rows " + + "UNION " + + "SELECT marc_id " + + "FROM deleted_rows2"; public static final String OR = " or "; public static final String MARC_INDEXERS = "marc_indexers"; public static final Field MARC_INDEXERS_MARC_ID = field(TABLE_FIELD_TEMPLATE, UUID.class, field(MARC_INDEXERS), field(MARC_ID)); @@ -292,11 +294,17 @@ public Future getRecords(Condition condition, RecordType recor } @Override - public Future getStrippedParsedRecords(List externalIds, IdType idType, RecordType recordType, String tenantId) { + public Future getStrippedParsedRecords(List externalIds, IdType idType, RecordType recordType, Boolean includeDeleted, String tenantId) { Name cte = name(CTE); Name prt = name(recordType.getTableName()); - Condition condition = RecordDaoUtil.getExternalIdsCondition(externalIds, idType) - .and(RECORDS_LB.STATE.eq(RecordState.ACTUAL)); + Condition condition; + if (includeDeleted != null && includeDeleted) { + condition = RecordDaoUtil.getExternalIdsCondition(externalIds, idType) + .and(RECORDS_LB.STATE.eq(RecordState.ACTUAL).or(RECORDS_LB.STATE.eq(RecordState.DELETED))); + } else { + condition = RecordDaoUtil.getExternalIdsCondition(externalIds, idType) + .and(RECORDS_LB.STATE.eq(RecordState.ACTUAL)); + } return getQueryExecutor(tenantId).transaction(txQE -> txQE.query(dsl -> dsl .with(cte.as(dsl.selectCount() .from(RECORDS_LB) @@ -622,7 +630,7 @@ private void appendWhere(SelectJoinStep step, ParseLeaderResult parseLeaderResul : DSL.noCondition(); Condition fieldsCondition = parseFieldsResult.isEnabled() ? exists(select(field("*")).from("cte") - .where("records_lb.id = cte.marc_id")) + .where("records_lb.id = cte.marc_id")) : DSL.noCondition(); step.where(leaderCondition) .and(fieldsCondition) @@ -685,9 +693,9 @@ public Future getMatchedRecordsIdentifiers(MatchFi countQuery = select(countDistinct(RECORDS_LB.ID)) .from(RECORDS_LB) .innerJoin(marcIndexersPartitionTable) - .on(RECORDS_LB.ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID)))) + .on(RECORDS_LB.ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID)))) .innerJoin(MARC_RECORDS_TRACKING) - .on(MARC_RECORDS_TRACKING.MARC_ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID))) + .on(MARC_RECORDS_TRACKING.MARC_ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID))) .and(MARC_RECORDS_TRACKING.VERSION.eq(field(TABLE_FIELD_TEMPLATE, Integer.class, marcIndexersPartitionTable, name(VERSION))))) .where(filterRecordByType(typeConnection.getRecordType().value()) .and(filterRecordByState(Record.State.ACTUAL.value())) @@ -702,9 +710,9 @@ public Future getMatchedRecordsIdentifiers(MatchFi .distinctOn(RECORDS_LB.ID) .from(RECORDS_LB) .innerJoin(marcIndexersPartitionTable) - .on(RECORDS_LB.ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID)))) + .on(RECORDS_LB.ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID)))) .innerJoin(MARC_RECORDS_TRACKING) - .on(MARC_RECORDS_TRACKING.MARC_ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID))) + .on(MARC_RECORDS_TRACKING.MARC_ID.eq(field(TABLE_FIELD_TEMPLATE, UUID.class, marcIndexersPartitionTable, name(MARC_ID))) .and(MARC_RECORDS_TRACKING.VERSION.eq(field(TABLE_FIELD_TEMPLATE, Integer.class, marcIndexersPartitionTable, name(VERSION))))) .where(filterRecordByType(typeConnection.getRecordType().value()) .and(filterRecordByState(Record.State.ACTUAL.value())) @@ -794,211 +802,210 @@ public Future saveRecords(RecordCollection recordCollectio logRecordCollection("saveRecords:: Saving", recordCollection, tenantId); Promise finalPromise = Promise.promise(); Context context = Vertx.currentContext(); - if(context == null) return Future.failedFuture("saveRecords must be executed by a Vertx thread"); + if (context == null) return Future.failedFuture("saveRecords must be executed by a Vertx thread"); context.owner().executeBlocking(promise -> { - Set matchedIds = new HashSet<>(); - Set snapshotIds = new HashSet<>(); - Set recordTypes = new HashSet<>(); - - List dbRecords = new ArrayList<>(); - List dbRawRecords = new ArrayList<>(); - List> dbParsedRecords = new ArrayList<>(); - List dbErrorRecords = new ArrayList<>(); - - List errorMessages = new ArrayList<>(); - - recordCollection.getRecords() - .stream() - .map(RecordDaoUtil::ensureRecordHasId) - .map(RecordDaoUtil::ensureRecordHasMatchedId) - .map(RecordDaoUtil::ensureRecordHasSuppressDiscovery) - .map(RecordDaoUtil::ensureRecordForeignKeys) - .forEach(record -> { - // collect unique matched ids to query to determine generation - matchedIds.add(UUID.fromString(record.getMatchedId())); - - // make sure only one snapshot id - snapshotIds.add(record.getSnapshotId()); - if (snapshotIds.size() > 1) { - throw new BadRequestException("Batch record collection only supports single snapshot"); - } + Set matchedIds = new HashSet<>(); + Set snapshotIds = new HashSet<>(); + Set recordTypes = new HashSet<>(); - if(Objects.nonNull(record.getRecordType())) { - recordTypes.add(record.getRecordType().name()); - } else { - throw new BadRequestException(StringUtils.defaultIfEmpty(record.getErrorRecord().getDescription(), String.format("Record with id %s has not record type", record.getId()))); - } + List dbRecords = new ArrayList<>(); + List dbRawRecords = new ArrayList<>(); + List> dbParsedRecords = new ArrayList<>(); + List dbErrorRecords = new ArrayList<>(); - // make sure only one record type - if (recordTypes.size() > 1) { - throw new BadRequestException("Batch record collection only supports single record type"); - } + List errorMessages = new ArrayList<>(); - // if record has parsed record, validate by attempting format - if (Objects.nonNull(record.getParsedRecord())) { - try { - RecordType recordType = toRecordType(record.getRecordType().name()); - recordType.formatRecord(record); - Record2 dbParsedRecord = recordType.toDatabaseRecord2(record.getParsedRecord()); - dbParsedRecords.add(dbParsedRecord); - } catch (Exception e) { - // create error record and remove from record - Object content = Objects.nonNull(record.getParsedRecord()) - ? record.getParsedRecord().getContent() - : null; - ErrorRecord errorRecord = new ErrorRecord() - .withId(record.getId()) - .withDescription(e.getMessage()) - .withContent(content); - errorMessages.add(format(INVALID_PARSED_RECORD_MESSAGE_TEMPLATE, record.getId(), e.getMessage())); - record.withErrorRecord(errorRecord) - .withParsedRecord(null) - .withLeaderRecordStatus(null); + recordCollection.getRecords() + .stream() + .map(RecordDaoUtil::ensureRecordHasId) + .map(RecordDaoUtil::ensureRecordHasMatchedId) + .map(RecordDaoUtil::ensureRecordHasSuppressDiscovery) + .map(RecordDaoUtil::ensureRecordForeignKeys) + .forEach(record -> { + // collect unique matched ids to query to determine generation + matchedIds.add(UUID.fromString(record.getMatchedId())); + + // make sure only one snapshot id + snapshotIds.add(record.getSnapshotId()); + if (snapshotIds.size() > 1) { + throw new BadRequestException("Batch record collection only supports single snapshot"); } - } - if (Objects.nonNull(record.getRawRecord())) { - dbRawRecords.add(RawRecordDaoUtil.toDatabaseRawRecord(record.getRawRecord())); - } - if (Objects.nonNull(record.getErrorRecord())) { - dbErrorRecords.add(ErrorRecordDaoUtil.toDatabaseErrorRecord(record.getErrorRecord())); - } - dbRecords.add(RecordDaoUtil.toDatabaseRecord(record)); - }); - UUID snapshotId = UUID.fromString(snapshotIds.stream().findFirst().orElseThrow()); - - RecordType recordType = toRecordType(recordTypes.stream().findFirst().orElseThrow()); + if (Objects.nonNull(record.getRecordType())) { + recordTypes.add(record.getRecordType().name()); + } else { + throw new BadRequestException(StringUtils.defaultIfEmpty(record.getErrorRecord().getDescription(), String.format("Record with id %s has not record type", record.getId()))); + } - try (Connection connection = getConnection(tenantId)) { - DSL.using(connection).transaction(ctx -> { - DSLContext dsl = DSL.using(ctx); + // make sure only one record type + if (recordTypes.size() > 1) { + throw new BadRequestException("Batch record collection only supports single record type"); + } - // validate snapshot - Optional snapshot = DSL.using(ctx).selectFrom(SNAPSHOTS_LB) - .where(SNAPSHOTS_LB.ID.eq(snapshotId)) - .fetchOptional(); - if (snapshot.isPresent()) { - if (Objects.isNull(snapshot.get().getProcessingStartedDate())) { - throw new BadRequestException(format(SNAPSHOT_NOT_STARTED_MESSAGE_TEMPLATE, snapshot.get().getStatus())); + // if record has parsed record, validate by attempting format + if (Objects.nonNull(record.getParsedRecord())) { + try { + RecordType recordType = toRecordType(record.getRecordType().name()); + recordType.formatRecord(record); + Record2 dbParsedRecord = recordType.toDatabaseRecord2(record.getParsedRecord()); + dbParsedRecords.add(dbParsedRecord); + } catch (Exception e) { + // create error record and remove from record + Object content = Objects.nonNull(record.getParsedRecord()) + ? record.getParsedRecord().getContent() + : null; + ErrorRecord errorRecord = new ErrorRecord() + .withId(record.getId()) + .withDescription(e.getMessage()) + .withContent(content); + errorMessages.add(format(INVALID_PARSED_RECORD_MESSAGE_TEMPLATE, record.getId(), e.getMessage())); + record.withErrorRecord(errorRecord) + .withParsedRecord(null) + .withLeaderRecordStatus(null); + } } - } else { - throw new NotFoundException(format(SNAPSHOT_NOT_FOUND_TEMPLATE, snapshotId)); - } + if (Objects.nonNull(record.getRawRecord())) { + dbRawRecords.add(RawRecordDaoUtil.toDatabaseRawRecord(record.getRawRecord())); + } + if (Objects.nonNull(record.getErrorRecord())) { + dbErrorRecords.add(ErrorRecordDaoUtil.toDatabaseErrorRecord(record.getErrorRecord())); + } + dbRecords.add(RecordDaoUtil.toDatabaseRecord(record)); + }); - List ids = new ArrayList<>(); - Map matchedGenerations = new HashMap<>(); + UUID snapshotId = UUID.fromString(snapshotIds.stream().findFirst().orElseThrow()); - // lookup latest generation by matched id and committed snapshot updated before current snapshot - dsl.select(RECORDS_LB.MATCHED_ID, RECORDS_LB.ID, RECORDS_LB.GENERATION) - .distinctOn(RECORDS_LB.MATCHED_ID) - .from(RECORDS_LB) - .innerJoin(SNAPSHOTS_LB).on(RECORDS_LB.SNAPSHOT_ID.eq(SNAPSHOTS_LB.ID)) - .where(RECORDS_LB.MATCHED_ID.in(matchedIds) - .and(SNAPSHOTS_LB.STATUS.in(JobExecutionStatus.COMMITTED, JobExecutionStatus.ERROR, JobExecutionStatus.CANCELLED)) - .and(SNAPSHOTS_LB.UPDATED_DATE.lessThan(dsl - .select(SNAPSHOTS_LB.PROCESSING_STARTED_DATE) - .from(SNAPSHOTS_LB) - .where(SNAPSHOTS_LB.ID.eq(snapshotId))))) - .orderBy(RECORDS_LB.MATCHED_ID.asc(), RECORDS_LB.GENERATION.desc()) - .fetchStream().forEach(r -> { - UUID id = r.get(RECORDS_LB.ID); - UUID matchedId = r.get(RECORDS_LB.MATCHED_ID); - int generation = r.get(RECORDS_LB.GENERATION); - ids.add(id); - matchedGenerations.put(matchedId, generation); - }); + RecordType recordType = toRecordType(recordTypes.stream().findFirst().orElseThrow()); - // update matching records state - if(!ids.isEmpty()) - { - dsl.update(RECORDS_LB) - .set(RECORDS_LB.STATE, RecordState.OLD) - .where(RECORDS_LB.ID.in(ids)) - .execute(); - } + try (Connection connection = getConnection(tenantId)) { + DSL.using(connection).transaction(ctx -> { + DSLContext dsl = DSL.using(ctx); - // batch insert records updating generation if required - List recordsLoadingErrors = dsl.loadInto(RECORDS_LB) - .batchAfter(1000) - .bulkAfter(500) - .commitAfter(1000) - .onErrorAbort() - .loadRecords(dbRecords.stream().map(record -> { - Integer generation = matchedGenerations.get(record.getMatchedId()); - if (Objects.nonNull(generation)) { - record.setGeneration(generation + 1); - } else if (Objects.isNull(record.getGeneration())) { - record.setGeneration(0); + // validate snapshot + Optional snapshot = DSL.using(ctx).selectFrom(SNAPSHOTS_LB) + .where(SNAPSHOTS_LB.ID.eq(snapshotId)) + .fetchOptional(); + if (snapshot.isPresent()) { + if (Objects.isNull(snapshot.get().getProcessingStartedDate())) { + throw new BadRequestException(format(SNAPSHOT_NOT_STARTED_MESSAGE_TEMPLATE, snapshot.get().getStatus())); } - return record; - }).collect(Collectors.toList())) - .fieldsCorresponding() - .execute() - .errors(); - - recordsLoadingErrors.forEach(error -> { - if (error.exception().sqlState().equals(UNIQUE_VIOLATION_SQL_STATE)) { - throw new DuplicateEventException("SQL Unique constraint violation prevented repeatedly saving the record"); + } else { + throw new NotFoundException(format(SNAPSHOT_NOT_FOUND_TEMPLATE, snapshotId)); } - LOG.warn("saveRecords:: Error occurred on batch execution: {}", error.exception().getCause().getMessage()); - LOG.debug("saveRecords:: Failed to execute statement from batch: {}", error.query()); - }); - // batch insert raw records - dsl.loadInto(RAW_RECORDS_LB) - .batchAfter(250) - .commitAfter(1000) - .onDuplicateKeyUpdate() - .onErrorAbort() - .loadRecords(dbRawRecords) - .fieldsCorresponding() - .execute(); - - // batch insert parsed records - recordType.toLoaderOptionsStep(dsl) - .batchAfter(250) - .commitAfter(1000) - .onDuplicateKeyUpdate() - .onErrorAbort() - .loadRecords(dbParsedRecords) - .fieldsCorresponding() - .execute(); - - if (!dbErrorRecords.isEmpty()) { - // batch insert error records - dsl.loadInto(ERROR_RECORDS_LB) + List ids = new ArrayList<>(); + Map matchedGenerations = new HashMap<>(); + + // lookup latest generation by matched id and committed snapshot updated before current snapshot + dsl.select(RECORDS_LB.MATCHED_ID, RECORDS_LB.ID, RECORDS_LB.GENERATION) + .distinctOn(RECORDS_LB.MATCHED_ID) + .from(RECORDS_LB) + .innerJoin(SNAPSHOTS_LB).on(RECORDS_LB.SNAPSHOT_ID.eq(SNAPSHOTS_LB.ID)) + .where(RECORDS_LB.MATCHED_ID.in(matchedIds) + .and(SNAPSHOTS_LB.STATUS.in(JobExecutionStatus.COMMITTED, JobExecutionStatus.ERROR, JobExecutionStatus.CANCELLED)) + .and(SNAPSHOTS_LB.UPDATED_DATE.lessThan(dsl + .select(SNAPSHOTS_LB.PROCESSING_STARTED_DATE) + .from(SNAPSHOTS_LB) + .where(SNAPSHOTS_LB.ID.eq(snapshotId))))) + .orderBy(RECORDS_LB.MATCHED_ID.asc(), RECORDS_LB.GENERATION.desc()) + .fetchStream().forEach(r -> { + UUID id = r.get(RECORDS_LB.ID); + UUID matchedId = r.get(RECORDS_LB.MATCHED_ID); + int generation = r.get(RECORDS_LB.GENERATION); + ids.add(id); + matchedGenerations.put(matchedId, generation); + }); + + // update matching records state + if (!ids.isEmpty()) { + dsl.update(RECORDS_LB) + .set(RECORDS_LB.STATE, RecordState.OLD) + .where(RECORDS_LB.ID.in(ids)) + .execute(); + } + + // batch insert records updating generation if required + List recordsLoadingErrors = dsl.loadInto(RECORDS_LB) + .batchAfter(1000) + .bulkAfter(500) + .commitAfter(1000) + .onErrorAbort() + .loadRecords(dbRecords.stream().map(record -> { + Integer generation = matchedGenerations.get(record.getMatchedId()); + if (Objects.nonNull(generation)) { + record.setGeneration(generation + 1); + } else if (Objects.isNull(record.getGeneration())) { + record.setGeneration(0); + } + return record; + }).collect(Collectors.toList())) + .fieldsCorresponding() + .execute() + .errors(); + + recordsLoadingErrors.forEach(error -> { + if (error.exception().sqlState().equals(UNIQUE_VIOLATION_SQL_STATE)) { + throw new DuplicateEventException("SQL Unique constraint violation prevented repeatedly saving the record"); + } + LOG.warn("saveRecords:: Error occurred on batch execution: {}", error.exception().getCause().getMessage()); + LOG.debug("saveRecords:: Failed to execute statement from batch: {}", error.query()); + }); + + // batch insert raw records + dsl.loadInto(RAW_RECORDS_LB) .batchAfter(250) .commitAfter(1000) .onDuplicateKeyUpdate() .onErrorAbort() - .loadRecords(dbErrorRecords) + .loadRecords(dbRawRecords) .fieldsCorresponding() .execute(); - } - promise.complete(new RecordsBatchResponse() - .withRecords(recordCollection.getRecords()) - .withTotalRecords(recordCollection.getRecords().size()) - .withErrorMessages(errorMessages)); - }); - } catch (DuplicateEventException e) { - LOG.info("saveRecords:: Skipped saving records due to duplicate event: {}", e.getMessage()); - promise.fail(e); - } catch (SQLException | DataAccessException e) { - LOG.warn("saveRecords:: Failed to save records", e); - promise.fail(e.getCause()); - } - }, - false, - r -> { - if (r.failed()) { - LOG.warn("saveRecords:: Error during batch record save", r.cause()); - finalPromise.fail(r.cause()); - } else { - LOG.debug("saveRecords:: batch record save was successful"); - finalPromise.complete(r.result()); - } - }); + // batch insert parsed records + recordType.toLoaderOptionsStep(dsl) + .batchAfter(250) + .commitAfter(1000) + .onDuplicateKeyUpdate() + .onErrorAbort() + .loadRecords(dbParsedRecords) + .fieldsCorresponding() + .execute(); + + if (!dbErrorRecords.isEmpty()) { + // batch insert error records + dsl.loadInto(ERROR_RECORDS_LB) + .batchAfter(250) + .commitAfter(1000) + .onDuplicateKeyUpdate() + .onErrorAbort() + .loadRecords(dbErrorRecords) + .fieldsCorresponding() + .execute(); + } + + promise.complete(new RecordsBatchResponse() + .withRecords(recordCollection.getRecords()) + .withTotalRecords(recordCollection.getRecords().size()) + .withErrorMessages(errorMessages)); + }); + } catch (DuplicateEventException e) { + LOG.info("saveRecords:: Skipped saving records due to duplicate event: {}", e.getMessage()); + promise.fail(e); + } catch (SQLException | DataAccessException e) { + LOG.warn("saveRecords:: Failed to save records", e); + promise.fail(e.getCause()); + } + }, + false, + r -> { + if (r.failed()) { + LOG.warn("saveRecords:: Error during batch record save", r.cause()); + finalPromise.fail(r.cause()); + } else { + LOG.debug("saveRecords:: batch record save was successful"); + finalPromise.complete(r.result()); + } + }); return finalPromise.future() .onSuccess(response -> response.getRecords() @@ -1011,9 +1018,9 @@ public Future updateRecord(Record record, Map okapiHeade var tenantId = okapiHeaders.get(OKAPI_TENANT_HEADER); LOG.trace("updateRecord:: Updating {} record {} for tenant {}", record.getRecordType(), record.getId(), tenantId); return getQueryExecutor(tenantId).transaction(txQE -> getRecordById(txQE, record.getId()) - .compose(optionalRecord -> optionalRecord - .map(r -> insertOrUpdateRecord(txQE, record)) - .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_TEMPLATE, record.getId())))))) + .compose(optionalRecord -> optionalRecord + .map(r -> insertOrUpdateRecord(txQE, record)) + .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_TEMPLATE, record.getId())))))) .onSuccess(updated -> recordDomainEventPublisher.publishRecordUpdated(updated, okapiHeaders)); } @@ -1092,35 +1099,35 @@ public Future> getSourceRecordByExternalId(String externa public Future> getSourceRecordByCondition(Condition condition, String tenantId) { return getQueryExecutor(tenantId) .transaction(txQE -> txQE.findOneRow(dsl -> dsl.selectFrom(RECORDS_LB) - .where(condition)) - .map(RecordDaoUtil::toOptionalRecord) - .compose(optionalRecord -> { - if (optionalRecord.isPresent()) { - return lookupAssociatedRecords(txQE, optionalRecord.get(), false) - .map(RecordDaoUtil::toSourceRecord) - .map(sourceRecord -> { - if (Objects.nonNull(sourceRecord.getParsedRecord())) { - return Optional.of(sourceRecord); - } - return Optional.empty(); - }); - } - return Future.succeededFuture(Optional.empty()); - })); + .where(condition)) + .map(RecordDaoUtil::toOptionalRecord) + .compose(optionalRecord -> { + if (optionalRecord.isPresent()) { + return lookupAssociatedRecords(txQE, optionalRecord.get(), false) + .map(RecordDaoUtil::toSourceRecord) + .map(sourceRecord -> { + if (Objects.nonNull(sourceRecord.getParsedRecord())) { + return Optional.of(sourceRecord); + } + return Optional.empty(); + }); + } + return Future.succeededFuture(Optional.empty()); + })); } @Override public Future calculateGeneration(ReactiveClassicGenericQueryExecutor txQE, Record record) { return txQE.query(dsl -> dsl.select(max(RECORDS_LB.GENERATION).as(RECORDS_LB.GENERATION)) - .from(RECORDS_LB.innerJoin(SNAPSHOTS_LB).on(RECORDS_LB.SNAPSHOT_ID.eq(SNAPSHOTS_LB.ID))) - .where(RECORDS_LB.MATCHED_ID.eq(UUID.fromString(record.getMatchedId())) - .and(SNAPSHOTS_LB.UPDATED_DATE.lessThan(dsl.select(SNAPSHOTS_LB.PROCESSING_STARTED_DATE) - .from(SNAPSHOTS_LB) - .where(SNAPSHOTS_LB.ID.eq(UUID.fromString(record.getSnapshotId()))))))) - .map(res -> { - Integer generation = res.get(RECORDS_LB.GENERATION); - return Objects.nonNull(generation) ? ++generation : 0; - }); + .from(RECORDS_LB.innerJoin(SNAPSHOTS_LB).on(RECORDS_LB.SNAPSHOT_ID.eq(SNAPSHOTS_LB.ID))) + .where(RECORDS_LB.MATCHED_ID.eq(UUID.fromString(record.getMatchedId())) + .and(SNAPSHOTS_LB.UPDATED_DATE.lessThan(dsl.select(SNAPSHOTS_LB.PROCESSING_STARTED_DATE) + .from(SNAPSHOTS_LB) + .where(SNAPSHOTS_LB.ID.eq(UUID.fromString(record.getSnapshotId()))))))) + .map(res -> { + Integer generation = res.get(RECORDS_LB.GENERATION); + return Objects.nonNull(generation) ? ++generation : 0; + }); } @Override @@ -1129,9 +1136,9 @@ public Future updateParsedRecord(Record record, Map GenericCompositeFuture.all(Lists.newArrayList( - updateExternalIdsForRecord(txQE, record), - ParsedRecordDaoUtil.update(txQE, record.getParsedRecord(), ParsedRecordDaoUtil.toRecordType(record)) - )).onSuccess(updated -> recordDomainEventPublisher.publishRecordUpdated(record, okapiHeaders)) + updateExternalIdsForRecord(txQE, record), + ParsedRecordDaoUtil.update(txQE, record.getParsedRecord(), ParsedRecordDaoUtil.toRecordType(record)) + )).onSuccess(updated -> recordDomainEventPublisher.publishRecordUpdated(record, okapiHeaders)) .map(res -> record.getParsedRecord())); } @@ -1141,7 +1148,7 @@ public Future updateParsedRecords(RecordCollection r logRecordCollection("updateParsedRecords:: Updating", recordCollection, tenantId); Promise promise = Promise.promise(); Context context = Vertx.currentContext(); - if(context == null) return Future.failedFuture("updateParsedRecords must be called by a vertx thread"); + if (context == null) return Future.failedFuture("updateParsedRecords must be called by a vertx thread"); var recordsUpdated = new ArrayList(); context.owner().executeBlocking(blockingPromise -> @@ -1181,8 +1188,8 @@ public Future updateParsedRecords(RecordCollection r String externalId = getExternalId(externalIdsHolder, recordType); String externalHrid = getExternalHrid(externalIdsHolder, recordType); if (StringUtils.isNotEmpty(externalId)) { - updateStep = updateFirstStep - .set(RECORDS_LB.EXTERNAL_ID, UUID.fromString(externalId)); + updateStep = updateFirstStep + .set(RECORDS_LB.EXTERNAL_ID, UUID.fromString(externalId)); } if (StringUtils.isNotEmpty(externalHrid)) { updateStep = (Objects.isNull(updateStep) ? updateFirstStep : updateStep) @@ -1283,16 +1290,16 @@ public Future updateParsedRecords(RecordCollection r blockingPromise.fail(e); } }, - false, - result -> { - if (result.failed()) { - LOG.warn("updateParsedRecords:: Error during update of parsed records", result.cause()); - promise.fail(result.cause()); - } else { - LOG.debug("updateParsedRecords:: Parsed records update was successful"); - promise.complete(result.result()); - } - }); + false, + result -> { + if (result.failed()) { + LOG.warn("updateParsedRecords:: Error during update of parsed records", result.cause()); + promise.fail(result.cause()); + } else { + LOG.debug("updateParsedRecords:: Parsed records update was successful"); + promise.complete(result.result()); + } + }); return promise.future() .onSuccess(response -> @@ -1304,24 +1311,24 @@ public Future updateParsedRecords(RecordCollection r public Future> getRecordByExternalId(String externalId, IdType idType, String tenantId) { return getQueryExecutor(tenantId) - .transaction(txQE -> getRecordByExternalId(txQE, externalId, idType)); + .transaction(txQE -> getRecordByExternalId(txQE, externalId, idType)); } @Override public Future> getRecordByExternalId(ReactiveClassicGenericQueryExecutor txQE, String externalId, IdType idType) { Condition condition = RecordDaoUtil.getExternalIdCondition(externalId, idType) - .and(RECORDS_LB.STATE.eq(RecordState.ACTUAL) - .or(RECORDS_LB.STATE.eq(RecordState.DELETED))); + .and(RECORDS_LB.STATE.eq(RecordState.ACTUAL) + .or(RECORDS_LB.STATE.eq(RecordState.DELETED))); return txQE.findOneRow(dsl -> dsl.selectFrom(RECORDS_LB) - .where(condition) - .orderBy(RECORDS_LB.GENERATION.sort(SortOrder.DESC)) - .limit(1)) - .map(RecordDaoUtil::toOptionalRecord) - .compose(optionalRecord -> optionalRecord - .map(record -> lookupAssociatedRecords(txQE, record, false).map(Optional::of)) - .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_BY_ID_TYPE, idType, externalId))))) - .onFailure(v -> txQE.rollback()); + .where(condition) + .orderBy(RECORDS_LB.GENERATION.sort(SortOrder.DESC)) + .limit(1)) + .map(RecordDaoUtil::toOptionalRecord) + .compose(optionalRecord -> optionalRecord + .map(record -> lookupAssociatedRecords(txQE, record, false).map(Optional::of)) + .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_BY_ID_TYPE, idType, externalId))))) + .onFailure(v -> txQE.rollback()); } @Override @@ -1364,10 +1371,10 @@ public Future saveUpdatedRecord(ReactiveClassicGenericQueryExecutor txQE public Future updateSuppressFromDiscoveryForRecord(String id, IdType idType, Boolean suppress, String tenantId) { LOG.trace("updateSuppressFromDiscoveryForRecord:: Updating suppress from discovery with value {} for record with {} {} for tenant {}", suppress, idType, id, tenantId); return getQueryExecutor(tenantId).transaction(txQE -> getRecordByExternalId(txQE, id, idType) - .compose(optionalRecord -> optionalRecord - .map(record -> RecordDaoUtil.update(txQE, record.withAdditionalInfo(record.getAdditionalInfo().withSuppressDiscovery(suppress)))) - .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_BY_ID_TYPE, idType, id)))))) - .map(u -> true); + .compose(optionalRecord -> optionalRecord + .map(record -> RecordDaoUtil.update(txQE, record.withAdditionalInfo(record.getAdditionalInfo().withSuppressDiscovery(suppress)))) + .orElse(Future.failedFuture(new NotFoundException(format(RECORD_NOT_FOUND_BY_ID_TYPE, idType, id)))))) + .map(u -> true); } @Override @@ -1410,6 +1417,7 @@ public Future deleteRecords(int lastUpdatedDays, int limit, String tenantI /** * Deletes old versions of Marc Indexers based on tenant ID. + * * @param tenantId The ID of the tenant for which the Marc Indexers are being deleted. * @return A Future of Boolean that completes successfully with a value of 'true' if the deletion was successful, * or 'false' if it was not. diff --git a/mod-source-record-storage-server/src/main/java/org/folio/services/RecordServiceImpl.java b/mod-source-record-storage-server/src/main/java/org/folio/services/RecordServiceImpl.java index 5b662d7ee..c39dba0c0 100644 --- a/mod-source-record-storage-server/src/main/java/org/folio/services/RecordServiceImpl.java +++ b/mod-source-record-storage-server/src/main/java/org/folio/services/RecordServiceImpl.java @@ -274,7 +274,8 @@ public Future fetchStrippedParsedRecords(FetchPa } var recordType = toRecordType(fetchRequest.getRecordType().name()); - return recordDao.getStrippedParsedRecords(ids, idType, recordType, tenantId) + var includeDeleted = fetchRequest.getIncludeDeleted(); + return recordDao.getStrippedParsedRecords(ids, idType, recordType, includeDeleted, tenantId) .onComplete(records -> filterFieldsByDataRange(records, fetchRequest)) .onFailure(ex -> { LOG.warn("fetchParsedRecords:: Failed to fetch parsed records. {}", ex.getMessage()); diff --git a/mod-source-record-storage-server/src/test/java/org/folio/rest/impl/SourceStorageBatchApiTest.java b/mod-source-record-storage-server/src/test/java/org/folio/rest/impl/SourceStorageBatchApiTest.java index 9f3c32dee..b2a243294 100644 --- a/mod-source-record-storage-server/src/test/java/org/folio/rest/impl/SourceStorageBatchApiTest.java +++ b/mod-source-record-storage-server/src/test/java/org/folio/rest/impl/SourceStorageBatchApiTest.java @@ -227,8 +227,46 @@ public void shouldPostSourceStorageBatchEdifactRecords(TestContext testContext) } @Test - public void shouldPostFetchParsedRecordsBatch(TestContext testContext) { + public void shouldPostFetchParsedRecordsBatchWithDeletedWhenIncludeDeleteTrue(TestContext testContext) { Async async = testContext.async(); + + Record record_1 = new Record() + .withId(FIRST_UUID) + .withSnapshotId(snapshot_1.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withMatchedId(FIRST_UUID) + .withOrder(0) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + Record record_2 = new Record() + .withId(SECOND_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecord) + .withMatchedId(SECOND_UUID) + .withOrder(11) + .withState(Record.State.DELETED) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(MARC_RECORD_HRID)); + Record record_3 = new Record() + .withId(THIRD_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withErrorRecord(errorRecord) + .withMatchedId(THIRD_UUID) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + var externalIds = List.of( record_1.getExternalIdsHolder().getInstanceId(), record_2.getExternalIdsHolder().getInstanceId(), @@ -243,7 +281,8 @@ public void shouldPostFetchParsedRecordsBatch(TestContext testContext) { FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) .withConditions(conditions) - .withData(emptyList()); + .withData(emptyList()) + .withIncludeDeleted(true); RestAssured.given() .spec(spec) @@ -257,6 +296,147 @@ public void shouldPostFetchParsedRecordsBatch(TestContext testContext) { async.complete(); } + + @Test + public void shouldPostFetchParsedRecordsBatchWithActualWhenIncludeDeleteFalse(TestContext testContext) { + Async async = testContext.async(); + + Record record_1 = new Record() + .withId(FIRST_UUID) + .withSnapshotId(snapshot_1.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withMatchedId(FIRST_UUID) + .withOrder(0) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + Record record_2 = new Record() + .withId(SECOND_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecord) + .withMatchedId(SECOND_UUID) + .withOrder(11) + .withState(Record.State.DELETED) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(MARC_RECORD_HRID)); + Record record_3 = new Record() + .withId(THIRD_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withErrorRecord(errorRecord) + .withMatchedId(THIRD_UUID) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + + var externalIds = List.of( + record_1.getExternalIdsHolder().getInstanceId(), + record_2.getExternalIdsHolder().getInstanceId(), + record_3.getExternalIdsHolder().getInstanceId() + ); + postSnapshots(testContext, snapshot_1, snapshot_2, snapshot_3); + postRecords(testContext, record_1, record_2, record_3); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(externalIds); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(emptyList()) + .withIncludeDeleted(false); + + RestAssured.given() + .spec(spec) + .body(batchRequest) + .when() + .post(SOURCE_STORAGE_BATCH_FETCH_PARSED_RECORDS_PATH) + .then() + .statusCode(HttpStatus.SC_OK) + .body("records.size()", is(externalIds.size()-1)) + .body("totalRecords", is(externalIds.size()-1)); + async.complete(); + } + + @Test + public void shouldPostFetchParsedRecordsBatchWithActualWhenIncludeDeleteNotExists(TestContext testContext) { + Async async = testContext.async(); + + Record record_1 = new Record() + .withId(FIRST_UUID) + .withSnapshotId(snapshot_1.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withMatchedId(FIRST_UUID) + .withOrder(0) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + Record record_2 = new Record() + .withId(SECOND_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecord) + .withMatchedId(SECOND_UUID) + .withOrder(11) + .withState(Record.State.DELETED) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(MARC_RECORD_HRID)); + Record record_3 = new Record() + .withId(THIRD_UUID) + .withSnapshotId(snapshot_2.getJobExecutionId()) + .withRecordType(Record.RecordType.MARC_BIB) + .withRawRecord(rawRecord) + .withParsedRecord(marcRecordWithHrId) + .withErrorRecord(errorRecord) + .withMatchedId(THIRD_UUID) + .withState(Record.State.ACTUAL) + .withExternalIdsHolder(new ExternalIdsHolder() + .withInstanceId(UUID.randomUUID().toString()) + .withInstanceHrid(HRID)); + + var externalIds = List.of( + record_1.getExternalIdsHolder().getInstanceId(), + record_2.getExternalIdsHolder().getInstanceId(), + record_3.getExternalIdsHolder().getInstanceId() + ); + postSnapshots(testContext, snapshot_1, snapshot_2, snapshot_3); + postRecords(testContext, record_1, record_2, record_3); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(externalIds); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(emptyList()); + + RestAssured.given() + .spec(spec) + .body(batchRequest) + .when() + .post(SOURCE_STORAGE_BATCH_FETCH_PARSED_RECORDS_PATH) + .then() + .statusCode(HttpStatus.SC_OK) + .body("records.size()", is(externalIds.size()-1)) + .body("totalRecords", is(externalIds.size()-1)); + async.complete(); + } + + @Test public void shouldPostFetchEmptyParsedRecordsBatch(TestContext testContext) { Async async = testContext.async(); @@ -282,6 +462,38 @@ public void shouldPostFetchEmptyParsedRecordsBatch(TestContext testContext) { async.complete(); } + @Test + public void shouldPostFetchParsedRecordsBatch(TestContext testContext) { + Async async = testContext.async(); + var externalIds = List.of( + record_1.getExternalIdsHolder().getInstanceId(), + record_2.getExternalIdsHolder().getInstanceId(), + record_3.getExternalIdsHolder().getInstanceId() + ); + postSnapshots(testContext, snapshot_1, snapshot_2, snapshot_3); + postRecords(testContext, record_1, record_2, record_3); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(externalIds); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(emptyList()) + .withIncludeDeleted(true); + + RestAssured.given() + .spec(spec) + .body(batchRequest) + .when() + .post(SOURCE_STORAGE_BATCH_FETCH_PARSED_RECORDS_PATH) + .then() + .statusCode(HttpStatus.SC_OK) + .body("records.size()", is(externalIds.size())) + .body("totalRecords", is(externalIds.size())); + async.complete(); + } + @Test public void shouldFailWhenPostSourceStorageBatchRecordsWithMultipleSnapshots(TestContext testContext) { Async async = testContext.async(); diff --git a/mod-source-record-storage-server/src/test/java/org/folio/services/RecordServiceTest.java b/mod-source-record-storage-server/src/test/java/org/folio/services/RecordServiceTest.java index 7745a4de5..e13c4c3eb 100644 --- a/mod-source-record-storage-server/src/test/java/org/folio/services/RecordServiceTest.java +++ b/mod-source-record-storage-server/src/test/java/org/folio/services/RecordServiceTest.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; import javax.ws.rs.BadRequestException; @@ -240,6 +241,142 @@ public void shouldFetchBibRecordsWithOneFieldByExternalId(TestContext context) { }); } + @Test + public void shouldFetchActualAndDeletedBibRecordsWithOneFieldByExternalIdWhenIncludeDeletedExists(TestContext context) { + Async async = context.async(); + List records = TestMocks.getRecords(); + records.get(3).setDeleted(true); + records.get(3).setState(State.DELETED); + records.get(5).setDeleted(true); + records.get(5).setState(State.DELETED); + RecordCollection recordCollection = new RecordCollection() + .withRecords(records) + .withTotalRecords(records.size()); + saveRecords(recordCollection.getRecords()).onComplete(batch -> { + if (batch.failed()) { + context.fail(batch.cause()); + } + + Set externalIds = Set.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc","6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99"); + List data = List.of( + new FieldRange().withFrom("001").withTo("001"), + new FieldRange().withFrom("007").withTo("007") + ); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(List.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc", "6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99")); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(data) + .withIncludeDeleted(true); + + recordService.fetchStrippedParsedRecords(batchRequest, TENANT_ID).onComplete(get -> { + if (get.failed()) { + context.fail(get.cause()); + } + List expected = records.stream() + .filter(r -> r.getRecordType().equals(Record.RecordType.MARC_BIB)) + .filter(r -> externalIds.contains(r.getExternalIdsHolder().getInstanceId())) + .collect(Collectors.toList()); + context.assertEquals(expected.size(), get.result().getTotalRecords()); + async.complete(); + }); + }); + } + + @Test + public void shouldFetchActualBibRecordsWithOneFieldByExternalIdWhenIncludeDeletedNotExists(TestContext context) { + Async async = context.async(); + List records = TestMocks.getRecords(); + records.get(3).setDeleted(true); + records.get(3).setState(State.DELETED); + records.get(5).setDeleted(true); + records.get(5).setState(State.DELETED); + RecordCollection recordCollection = new RecordCollection() + .withRecords(records) + .withTotalRecords(records.size()); + saveRecords(recordCollection.getRecords()).onComplete(batch -> { + if (batch.failed()) { + context.fail(batch.cause()); + } + + Set externalIds = Set.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc","6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99"); + List data = List.of( + new FieldRange().withFrom("001").withTo("001"), + new FieldRange().withFrom("007").withTo("007") + ); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(List.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc", "6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99")); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(data); + + recordService.fetchStrippedParsedRecords(batchRequest, TENANT_ID).onComplete(get -> { + if (get.failed()) { + context.fail(get.cause()); + } + List expected = records.stream() + .filter(r -> r.getRecordType().equals(Record.RecordType.MARC_BIB)) + .filter(r -> externalIds.contains(r.getExternalIdsHolder().getInstanceId())) + .filter(r -> r.getState().equals(State.ACTUAL)) + .collect(Collectors.toList()); + context.assertEquals(expected.size(), get.result().getTotalRecords()); + async.complete(); + }); + }); + } + + @Test + public void shouldFetchActualAndDeletedBibRecordsWithOneFieldByExternalIdWhenIncludeDeletedFalse(TestContext context) { + Async async = context.async(); + List records = TestMocks.getRecords(); + records.get(3).setDeleted(true); + records.get(3).setState(State.DELETED); + records.get(5).setDeleted(true); + records.get(5).setState(State.DELETED); + RecordCollection recordCollection = new RecordCollection() + .withRecords(records) + .withTotalRecords(records.size()); + saveRecords(recordCollection.getRecords()).onComplete(batch -> { + if (batch.failed()) { + context.fail(batch.cause()); + } + + Set externalIds = Set.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc","6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99"); + List data = List.of( + new FieldRange().withFrom("001").withTo("001"), + new FieldRange().withFrom("007").withTo("007") + ); + + Conditions conditions = new Conditions() + .withIdType(IdType.INSTANCE.name()) + .withIds(List.of("3c4ae3f3-b460-4a89-a2f9-78ce3145e4fc", "6b4ae089-e1ee-431f-af83-e1133f8e3da0", "1b74ab75-9f41-4837-8662-a1d99118008d", "c1d3be12-ecec-4fab-9237-baf728575185", "8be05cf5-fb4f-4752-8094-8e179d08fb99")); + FetchParsedRecordsBatchRequest batchRequest = new FetchParsedRecordsBatchRequest() + .withRecordType(FetchParsedRecordsBatchRequest.RecordType.MARC_BIB) + .withConditions(conditions) + .withData(data) + .withIncludeDeleted(false); + + recordService.fetchStrippedParsedRecords(batchRequest, TENANT_ID).onComplete(get -> { + if (get.failed()) { + context.fail(get.cause()); + } + List expected = records.stream() + .filter(r -> r.getRecordType().equals(Record.RecordType.MARC_BIB)) + .filter(r -> externalIds.contains(r.getExternalIdsHolder().getInstanceId())) + .filter(r -> r.getState().equals(State.ACTUAL)) + .collect(Collectors.toList()); + context.assertEquals(expected.size(), get.result().getTotalRecords()); + async.complete(); + }); + }); + } + @Test public void shouldGetMarcAuthorityRecordsBySnapshotId(TestContext context) { getRecordsBySnapshotId(context, "ee561342-3098-47a8-ab6e-0f3eba120b04", RecordType.MARC_AUTHORITY, diff --git a/ramls/raml-storage b/ramls/raml-storage index 1e4076bf5..437a0eba8 160000 --- a/ramls/raml-storage +++ b/ramls/raml-storage @@ -1 +1 @@ -Subproject commit 1e4076bf5e4eae9670cd005adea7a174ccfaf681 +Subproject commit 437a0eba8cae1f8a274d98f43f4d8ccca0a10f4a