001/*- 002 * #%L 003 * HAPI FHIR JPA Server 004 * %% 005 * Copyright (C) 2014 - 2024 Smile CDR, Inc. 006 * %% 007 * Licensed under the Apache License, Version 2.0 (the "License"); 008 * you may not use this file except in compliance with the License. 009 * You may obtain a copy of the License at 010 * 011 * http://www.apache.org/licenses/LICENSE-2.0 012 * 013 * Unless required by applicable law or agreed to in writing, software 014 * distributed under the License is distributed on an "AS IS" BASIS, 015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 016 * See the License for the specific language governing permissions and 017 * limitations under the License. 018 * #L% 019 */ 020package ca.uhn.fhir.jpa.dao.data; 021 022import ca.uhn.fhir.batch2.model.WorkChunkStatusEnum; 023import ca.uhn.fhir.jpa.entity.Batch2WorkChunkEntity; 024import org.springframework.data.domain.Pageable; 025import org.springframework.data.jpa.repository.JpaRepository; 026import org.springframework.data.jpa.repository.Modifying; 027import org.springframework.data.jpa.repository.Query; 028import org.springframework.data.repository.query.Param; 029 030import java.util.Collection; 031import java.util.Date; 032import java.util.List; 033import java.util.Set; 034import java.util.stream.Stream; 035 036public interface IBatch2WorkChunkRepository 037 extends JpaRepository<Batch2WorkChunkEntity, String>, IHapiFhirJpaRepository { 038 039 // NOTE we need a stable sort so paging is reliable. 040 // Warning: mySequence is not unique - it is reset for every chunk. So we also sort by myId. 041 @Query( 042 "SELECT e FROM Batch2WorkChunkEntity e WHERE e.myInstanceId = :instanceId ORDER BY e.mySequence ASC, e.myId ASC") 043 List<Batch2WorkChunkEntity> fetchChunks(Pageable thePageRequest, @Param("instanceId") String theInstanceId); 044 045 /** 046 * A projection query to avoid fetching the CLOB over the wire. 047 * Otherwise, the same as fetchChunks. 048 */ 049 @Query("SELECT new Batch2WorkChunkEntity(" 050 + "e.myId, e.mySequence, e.myJobDefinitionId, e.myJobDefinitionVersion, e.myInstanceId, e.myTargetStepId, e.myStatus," 051 + "e.myCreateTime, e.myStartTime, e.myUpdateTime, e.myEndTime," 052 + "e.myErrorMessage, e.myErrorCount, e.myRecordsProcessed, e.myWarningMessage" 053 + ") FROM Batch2WorkChunkEntity e WHERE e.myInstanceId = :instanceId ORDER BY e.mySequence ASC, e.myId ASC") 054 List<Batch2WorkChunkEntity> fetchChunksNoData(Pageable thePageRequest, @Param("instanceId") String theInstanceId); 055 056 @Query( 057 "SELECT DISTINCT e.myStatus from Batch2WorkChunkEntity e where e.myInstanceId = :instanceId AND e.myTargetStepId = :stepId") 058 Set<WorkChunkStatusEnum> getDistinctStatusesForStep( 059 @Param("instanceId") String theInstanceId, @Param("stepId") String theStepId); 060 061 @Query( 062 "SELECT e FROM Batch2WorkChunkEntity e WHERE e.myInstanceId = :instanceId AND e.myTargetStepId = :targetStepId ORDER BY e.mySequence ASC") 063 Stream<Batch2WorkChunkEntity> fetchChunksForStep( 064 @Param("instanceId") String theInstanceId, @Param("targetStepId") String theTargetStepId); 065 066 @Modifying 067 @Query("UPDATE Batch2WorkChunkEntity e SET e.myStatus = :status, e.myEndTime = :et, " 068 + "e.myRecordsProcessed = :rp, e.myErrorCount = e.myErrorCount + :errorRetries, e.mySerializedData = null, e.mySerializedDataVc = null, " 069 + "e.myWarningMessage = :warningMessage WHERE e.myId = :id") 070 void updateChunkStatusAndClearDataForEndSuccess( 071 @Param("id") String theChunkId, 072 @Param("et") Date theEndTime, 073 @Param("rp") int theRecordsProcessed, 074 @Param("errorRetries") int theErrorRetries, 075 @Param("status") WorkChunkStatusEnum theInProgress, 076 @Param("warningMessage") String theWarningMessage); 077 078 @Modifying 079 @Query( 080 "UPDATE Batch2WorkChunkEntity e SET e.myStatus = :status, e.myEndTime = :et, e.mySerializedData = null, e.mySerializedDataVc = null, e.myErrorMessage = :em WHERE e.myId IN(:ids)") 081 void updateAllChunksForInstanceStatusClearDataAndSetError( 082 @Param("ids") List<String> theChunkIds, 083 @Param("et") Date theEndTime, 084 @Param("status") WorkChunkStatusEnum theInProgress, 085 @Param("em") String theError); 086 087 @Modifying 088 @Query( 089 "UPDATE Batch2WorkChunkEntity e SET e.myStatus = :status, e.myEndTime = :et, e.myErrorMessage = :em, e.myErrorCount = e.myErrorCount + 1 WHERE e.myId = :id") 090 int updateChunkStatusAndIncrementErrorCountForEndError( 091 @Param("id") String theChunkId, 092 @Param("et") Date theEndTime, 093 @Param("em") String theErrorMessage, 094 @Param("status") WorkChunkStatusEnum theInProgress); 095 096 @Modifying 097 @Query( 098 "UPDATE Batch2WorkChunkEntity e SET e.myStatus = :status, e.myStartTime = :st WHERE e.myId = :id AND e.myStatus IN :startStatuses") 099 int updateChunkStatusForStart( 100 @Param("id") String theChunkId, 101 @Param("st") Date theStartedTime, 102 @Param("status") WorkChunkStatusEnum theInProgress, 103 @Param("startStatuses") Collection<WorkChunkStatusEnum> theStartStatuses); 104 105 @Modifying 106 @Query("DELETE FROM Batch2WorkChunkEntity e WHERE e.myInstanceId = :instanceId") 107 int deleteAllForInstance(@Param("instanceId") String theInstanceId); 108 109 @Query( 110 "SELECT e.myId from Batch2WorkChunkEntity e where e.myInstanceId = :instanceId AND e.myTargetStepId = :stepId AND e.myStatus = :status") 111 List<String> fetchAllChunkIdsForStepWithStatus( 112 @Param("instanceId") String theInstanceId, 113 @Param("stepId") String theStepId, 114 @Param("status") WorkChunkStatusEnum theStatus); 115}