더티채킹과 지연쓰기 디버깅

조회와 영속성 컨텍스트에 저장

// SessionImpl
@Override  
public <T> T find(Class<T> entityClass, Object primaryKey, LockModeType lockModeType, Map<String, Object> properties) {  
    // ...
    try {  
        // ...
       return byId( entityClass )  
             .with( determineAppropriateLocalCacheMode( properties ) )  
             .with( lockOptions )  
             .load( primaryKey );  
    }
}

JpaRepository 상속한 Repository 클래스에서 findById를 실행하면 EntityManager의 find를 통해 엔티티를 찾는다. 이때 EntityManager의 구현체인 SessionImpl에서는 byId(entityClass)로 IdentifierLoadAccessImpl가 생성되고 load 메서드를 통해 엔티티 조회를 시작한다.

 

// IdentifierLoadAccessImpl
private Object load(Object id, EventSource eventSource, String entityName, Boolean readOnly) {  
    final LoadEvent event;  
    if ( lockOptions != null ) {  
       event = new LoadEvent( id, entityName, lockOptions, eventSource, readOnly );  
       context.fireLoad( event, LoadEventListener.GET );  
    }  
    else {  
       event = new LoadEvent( id, entityName, false, eventSource, readOnly );  
       boolean success = false;  
       try {  
          context.fireLoad( event, LoadEventListener.GET );  
          success = true;  
       }  
       catch (ObjectNotFoundException e) {  
          // if session cache contains proxy for non-existing object  
       }  
       finally {  
          context.afterOperation( success );  
       }  
    }  
    return event.getResult();  
}

load 메서드에서는 이벤트 클래스로 조회를 위한 정보를 전달하고 context.fireLoad(event, LoadEventListener.GET)를 통해 DefaultLoadEventListener 실행시킨다.

 

// DefaultLoadEventListener
private Object doLoad(LoadEvent event, EntityPersister persister, EntityKey keyToLoad, LoadType options) {  
    final EventSource session = event.getSession();  

    if ( LOG.isTraceEnabled() ) {  
       LOG.tracev(  
             "Attempting to resolve: {0}",  
             infoString( persister, event.getEntityId(), event.getFactory() )  
       );  
    }  

    if ( session.getPersistenceContextInternal().containsDeletedUnloadedEntityKey( keyToLoad ) ) {  
       return null;  
    }  
    else {  
       final PersistenceContextEntry persistenceContextEntry  
             = CacheEntityLoaderHelper.INSTANCE.loadFromSessionCache( event, keyToLoad, options );  
       final Object entity = persistenceContextEntry.getEntity();  
       if ( entity != null ) {  
          if ( persistenceContextEntry.isManaged() ) {  
             initializeIfNecessary( entity );  
             return entity;  
          }  
          else {  
             return null;  
          }  
       }  
       else {  
          return load( event, persister, keyToLoad );  
       }  
    }  
}

DefaultLoadEventListener에서는 우선 CacheEntityLoaderHelper.INSTANCE.loadFromSessionCache(event, keyToLoad, options)를 통해 1차 캐시를 조회하고 persistenceContextEntry.getEntity()를 통해 1차 캐시에 등록된 엔티티 조회를 시도한다.

 

// DefaultLoadEventListener
private Object load(LoadEvent event, EntityPersister persister, EntityKey keyToLoad) {  
    final Object entity = loadFromCacheOrDatasource( event, persister, keyToLoad );  
    if ( entity != null && persister.hasNaturalIdentifier() ) {  
       event.getSession().getPersistenceContextInternal().getNaturalIdResolutions()  
             .cacheResolutionFromLoad(  
                   event.getEntityId(),  
                   persister.getNaturalIdMapping().extractNaturalIdFromEntity( entity ),  
                   persister  
             );  
    }  
    return entity;  
}

private Object loadFromCacheOrDatasource(LoadEvent event, EntityPersister persister, EntityKey keyToLoad) {  
    final Object entity = CacheEntityLoaderHelper.INSTANCE.loadFromSecondLevelCache( event, persister, keyToLoad );  
    if ( entity != null ) {  
       if ( LOG.isTraceEnabled() ) {  
          LOG.tracev(  
                "Resolved object in second-level cache: {0}",  
                infoString( persister, event.getEntityId(), event.getFactory() )  
          );  
       }  
       return entity;  
    }  
    else {  
       if ( LOG.isTraceEnabled() ) {  
          LOG.tracev(  
                "Object not resolved in any cache: {0}",  
                infoString( persister, event.getEntityId(), event.getFactory() )  
          );  
       }  
       return loadFromDatasource( event, persister );  
    }  
}

이때 1차 캐시에 엔티티가 없다면 CacheEntityLoaderHelper.INSTANCE.loadFromSecondLevelCache(event, persister, keyToLoad)를 통해 2차 캐시에서 엔티티 조회를 시도한다. 2차 캐시에도 없다면 loadFromDatasource(event, persister) 를 통해 데이터 소스에서 조회한다.

 

// DefaultLoadEventListener
protected Object loadFromDatasource(final LoadEvent event, final EntityPersister persister) {  
    Object entity = persister.load(  
          event.getEntityId(),  
          event.getInstanceToLoad(),  
          event.getLockOptions(),  
          event.getSession(),  
          event.getReadOnly()  
    );  

    final LazyInitializer lazyInitializer = HibernateProxy.extractLazyInitializer( entity );  
    if ( lazyInitializer != null ) {  
       entity = lazyInitializer.getImplementation();  
    }  

    final StatisticsImplementor statistics = event.getFactory().getStatistics();  
    if ( event.isAssociationFetch() && statistics.isStatisticsEnabled() ) {  
       statistics.fetchEntity( event.getEntityClassName() );  
    }  

    return entity;  
}

DefaultLoadEventListener의 loadFromDatasource에서는 persister.load를 통해 엔티티 조회를 수행한다.

 

// JdbcSelectExecutorStandardImpl#doExecuteQuery
final T result = resultsConsumer.consume(  
       jdbcValues,  
       session,  
       processingOptions,  
       valuesProcessingState,  
       rowProcessingState,  
       rowReader  
);

이때 실제 쿼리는 JdbcSelectExecutorStandardImpl의 doExecuteQuery에서 실행되며 resultsConsumer의 consume 메서드의 반환을 통해 실행 결과를 획득할 수 있다.

 

// ListResultsConsumer
private static <R> int readUniqueAssert(  
       RowProcessingStateStandardImpl rowProcessingState,  
       RowReader<R> rowReader,  
       Results<R> results) {  
    int readRows = 0;  
    while ( rowProcessingState.next() ) {  
       if ( !results.addUnique( rowReader.readRow( rowProcessingState ) ) ) {  
          throw new HibernateException(  
                String.format(  
                      Locale.ROOT,  
                      "Duplicate row was found and `%s` was specified",  
                      UniqueSemantic.ASSERT  
                )  
          );  
       }  
       rowProcessingState.finishRowProcessing( true );  
       readRows++;  
    }  
    return readRows;  
}

// StandardRowReader
@Override  
public T readRow(RowProcessingState rowProcessingState) {  
    coordinateInitializers( rowProcessingState );  

    final T result;  
    if ( componentType != ComponentType.OBJECT ) {  
       result = readPrimitiveRow( rowProcessingState );  
    }  
    else {  
       if ( resultAssemblers.length == 1 && rowTransformer == null ) {  
          //noinspection unchecked  
          result = (T) resultAssemblers[0].assemble( rowProcessingState );  
       }  
       else {  
          final Object[] resultRow = (Object[]) Array.newInstance( resultElementClass, resultAssemblers.length );  
          for ( int i = 0; i < resultAssemblers.length; i++ ) {  
             resultRow[i] = resultAssemblers[i].assemble( rowProcessingState );  
          }  
          //noinspection unchecked  
          result = rowTransformer == null  
                ? (T) resultRow  
                : rowTransformer.transformRow( resultRow );  
       }  
    }  

    finishUpRow();  
    return result;  
}

resultsConsumer의 consume 메서드에서는 StandardRowReader의 rowReader.readRow(rowProcessingState) 를 통해 로우를 읽고 coordinateInitializers(rowProcessingState)에 의해 EntityInitializerImpl 가 초기화한다.

 

// EntityInitializerImpl#initializeEntityInstance
takeSnapshot( data, session, persistenceContext, entityEntry, resolvedEntityState );

protected void takeSnapshot(  
       EntityInitializerData data,  
       SharedSessionContractImplementor session,  
       PersistenceContext persistenceContext,  
       EntityEntry entityEntry,  
       Object[] resolvedEntityState) {  
    if ( isReallyReadOnly( data, session ) ) {  
       //no need to take a snapshot - this is a  
       //performance optimization, but not really       //important, except for entities with huge       //mutable property values       persistenceContext.setEntryStatus( entityEntry, Status.READ_ONLY );  
    }  
    else {  
       //take a snapshot  
       deepCopy( data.concreteDescriptor, resolvedEntityState, resolvedEntityState );  
       persistenceContext.setEntryStatus( entityEntry, Status.MANAGED );  
    }  
}

EntityInitializerImpl가 초기화되는 과정에서 takeSnapshot를 통해 해당 작업이 리드온리가 아니라면 persistenceContext에 엔티티를 저장한다.

 

이러한 과정을 거쳐 조회를 요청한 엔티티를 영속성 컨텍스트인 1차 캐시에 저장할 수 있다.

 

더티채킹

비즈니스 로직을 작성하다 보면 영속성 컨텍스트에 의해 관리되는 엔티티들이 수정하게 된다. 이때 더티채킹이라는 과정을 통해 별도의 Repository 클래스를 사용하지 않고 수정된 값을 저장할 수 있다.

 

// DefaultFlushEntityEventListener
@Override  
public int[] findDirty(Object[] currentState, Object[] previousState, Object entity, SharedSessionContractImplementor session)  
       throws HibernateException {  
    int[] props = DirtyHelper.findDirty(  
          entityMetamodel.getDirtyCheckablePropertyTypes(),  
          currentState,  
          previousState,  
          propertyColumnUpdateable,  
          session  
    );  
    if ( props == null ) {  
       return null;  
    }  
    else {  
       logDirtyProperties( props );  
       return props;  
    }  
}

조금 더 구체적으로 플러시 과정에서 DefaultFlushEntityEventListener가 findDirty를 통해 더티채킹이 필요한 프로퍼티를 찾아낸다.

 

// DefaultFlushEntityEventListener#performDirtyCheck
event.setDirtyProperties( dirtyProperties );
event.setDirtyCheckPossible( dirtyCheckPossible );

그리고 DefaultFlushEntityEventListener의 performDirtyCheck 메서드에서 해당 프로퍼티를 DirtyProperties로 지정하고 더티채킹이 필요하다는 사실도 저장한다.

 

// DefaultFlushEntityEventListener#onFlushEntity
if ( isUpdateNecessary( event, mightBeDirty ) ) {  
    substitute = scheduleUpdate( event ) || substitute;  
}

위의 과정은 DefaultFlushEntityEventListener의 onFlushEntity 메서드 내부 isUpdateNecessary에서 수행된다.

 

// DefaultFlushEntityEventListener
private boolean isUpdateNecessary(final FlushEntityEvent event, final boolean mightBeDirty) {  
    final EntityEntry entry = event.getEntityEntry();  
    if ( mightBeDirty || entry.getStatus() == Status.DELETED ) {  
       // compare to cached state (ignoring collections unless versioned) 
       dirtyCheck( event );  
       if ( isUpdateNecessary( event ) ) {  
          return true;  
       }  
       else {  
          final Object entity = event.getEntity();  
          processIfSelfDirtinessTracker( entity, SelfDirtinessTracker::$$_hibernate_clearDirtyAttributes );  
          processIfManagedEntity( entity, DefaultFlushEntityEventListener::useTracker );  
          event.getFactory()  
                .getCustomEntityDirtinessStrategy()  
                .resetDirty( entity, entry.getPersister(), event.getSession() );  
          return false;  
       }  
    }  
    else {  
       return hasDirtyCollections( event );  
    }  
}

isUpdateNecessary에서는 해당 세션에서 dirtyCheck의 결과를 바탕으로 업데이트가 필요한 작업이 존재하는지 판단한다.

 

// DefaultFlushEntityEventListener
private boolean scheduleUpdate(final FlushEntityEvent event) {  
    final EntityEntry entry = event.getEntityEntry();  
    final EventSource session = event.getSession();  
    final Object entity = event.getEntity();  
    final Status status = entry.getStatus();  
    final EntityPersister persister = entry.getPersister();  
    final Object[] values = event.getPropertyValues();  

    logScheduleUpdate( entry, event.getFactory(), status, persister );  

    final boolean intercepted = !entry.isBeingReplicated() && handleInterception( event );  

    // increment the version number (if necessary)  
    final Object nextVersion = getNextVersion( event );  

    int[] dirtyProperties = getDirtyProperties( event, intercepted );  

    // check nullability but do not doAfterTransactionCompletion command execute  
    // we'll use scheduled updates for that.    
    new Nullability( session ).checkNullability( values, persister, true );  

    // schedule the update  
    // note that we intentionally do _not_ pass in currentPersistentState!    
    session.getActionQueue().addAction(  
          new EntityUpdateAction(  
                entry.getId(),  
                values,  
                dirtyProperties,  
                event.hasDirtyCollection(),  
                status == Status.DELETED && !entry.isModifiableEntity()  
                      ? persister.getValues( entity )  
                      : entry.getLoadedState(),  
                entry.getVersion(),  
                nextVersion,  
                entity,  
                entry.getRowId(),  
                persister,  
                session  
          )  
    );  

    return intercepted;  
}

업데이트가 필요한 작업이 있다면 getDirtyProperties를 통해 업데이트가 필요한 DirtyProperties를 조회하고 EntityUpdateAction를 생성하여 세션의 ActionQueue에 추가한다.

 

이러한 과정을 통해 우리는 더티채킹이 필요한 엔티티를 판단하고 필요한 작업을 정의해 ActionQueue에 추가한다. 이를 통해 알 수 있는 것은 더티채킹 역시 해당 작업이 필요하다 판단 이후 바로 실행되는 것이 아닌 ActionQueue에 추가되며 실행이 지연된다는 것이다.

 

지연 쓰기

// DefaultFlushEventListener
public void onFlush(FlushEvent event) throws HibernateException {  
    final EventSource source = event.getSession();  
    final PersistenceContext persistenceContext = source.getPersistenceContextInternal();  
    final EventManager eventManager = source.getEventManager();  
    if ( persistenceContext.getNumberOfManagedEntities() > 0  
          || persistenceContext.getCollectionEntriesSize() > 0 ) {  
       final HibernateMonitoringEvent flushEvent = eventManager.beginFlushEvent();  
       try {  
          source.getEventListenerManager().flushStart();  

          flushEverythingToExecutions( event );  
          performExecutions( source );  
          postFlush( source );  
       }  
       finally {  
          eventManager.completeFlushEvent( flushEvent, event );  
          source.getEventListenerManager().flushEnd(  
                event.getNumberOfEntitiesProcessed(),  
                event.getNumberOfCollectionsProcessed()  
          );  
       }  

       postPostFlush( source );  

       final StatisticsImplementor statistics = source.getFactory().getStatistics();  
       if ( statistics.isStatisticsEnabled() ) {  
          statistics.flush();  
       }  
    }  
    else if ( source.getActionQueue().hasAnyQueuedActions() ) {  
       // execute any queued unloaded-entity deletions  
       performExecutions( source );  
    }  
}

Repository 클래스를 사용하며 그리고 flushEverythingToExecutions를 통해 쓰기 작업들은 ActionQueue에 추가된다.

 

// AbstractFlushingEventListener
 protected void performExecutions(EventSource session) {  
    LOG.trace( "Executing flush" );  

    final PersistenceContext persistenceContext = session.getPersistenceContextInternal();  
    final JdbcCoordinator jdbcCoordinator = session.getJdbcCoordinator();  
    try {  
       jdbcCoordinator.flushBeginning();  
       persistenceContext.setFlushing( true );  

       final ActionQueue actionQueue = session.getActionQueue();  
       actionQueue.prepareActions();  
       actionQueue.executeActions();  
    }  
    finally {  
       persistenceContext.setFlushing( false );  
       jdbcCoordinator.flushEnding();  
    }  
}

해당 세션에서 ActionQue에 추가된 쓰기 작업들은 performExecutions 메서드 내에서 조회하고 수행된다.

 

이렇게 DefaultFlushEventListener에서 ActionQue를 사용하여 지연한 더티채킹을 포함한 지연된 쓰기 작업을 완료한다. 쓰기 작업을 트랜잭션의 비즈니스 로직이 실행되고 마지막 과정인 커밋 과정 속 플러시에서 처리하며 지연 쓰기는 데이터베이스 로우에 락이 걸리는 시간을 최소화할 수 있다는 장점을 지니게 된다.