Cursor 와 Transaction 의 관계 (feat. fetch size)

DB 에 저정되어 있는 대량 데이터를 모두 SELECT 후 어떤 처리를 해야 하는 경우 

Mybatis result handler 를 이용해서 처리하는 경우도 있는데 

가장 쉽게 대응할 수 있는 방법중 하나는 Cursor 를 이용하는 것이다.

 

사용법은 간단한다. 

return Type  을 List<T> 에서 Cursor<T> 형태로 변경하면 된다.

 

주의해야 할 것은 하나의 Transaction 안에서 동작해야 한다는 것이다.

생각해보면 당연하다. 

Cursor 를 return 하고 Cursor<T> 를 loop 돌면 next, next ... 할 것이기 때문에 Transaction 이 끝나야 Cursor 가 close 되지 않을까?

 

일단위 로그 테이블의 데이터를 가져와 어떤 처리를 한다고 가정했을때

아래와 같이 코드를 작성할 수 있다. 

@Mapper
@Repository
public interface TestMapper {
  Cursor<LogData> getLogs(Param param) throws Exception;
}

@Component 
public LogProcessor {
  
  @Transactional
  public void analyzeLogs(Param param) {
    // 일단위 로그 테이블에 대해서 loop 돌면서 가져와서 처리 
    while (oldLogTime.isBefore(now)) {
      try (Cursor<LogData> logList = testMapper.getLogs(param)) {
        for (LogData log : logList) {
          transfer.parseLog(log);
        }
      } catch (Exception e) {
        // ....
      }
    }
  }
}

@Component
public class LogTrasfer {
  public void parseLog(LogData log) { 
    // parsing and save to file
  }
}

잘못된 것이 있을까?

사실 위 코드는 정상적으로 동작 한다. 

다만, 개선해야 할 부분이 있다.

while (oldLogTime.isBefore(now)) {
  try (Cursor<LogData> logList = testMapper.getLogs(param)) {
    // ...
  } catch (Exception e) {
    // ....
  }
}

이렇게 일별 테이블에 대한 LOOP 로직이 하나의 Transactional 로 묶이게 되면 

cursor 는 transaction 이 끝날때 까지 close 되지 않게 된다. 

때문에 가져온 logData 에 대한 후처리를 진행하는 LogTrasnfer 에서 parsing & save to file 처리 할 때 

logData 에 big data 가 있다면 (ex. TEXT, MEDIUMTEXT 등) 해당 정보를 담고 있는 String 변수에 따른

Memory 사용량이 높아지게 된다. 

 

아래와 같이 transaction 으로 묶여 있는 analyzeLogs method 를 가볍게(?) 하고 외부에서 business logic (일별 loop 및 parameter 설정 등) 을 처리하도록 분리하면 문제는 가볍게 해결된다. 

..
while (oldLogTime.isBefore(now)) {
  logProcessor.analyzeLogs(param);
}
..

@Component 
public LogProcessor {
  
  @Transactional
  public void analyzeLogs(Param param) {
    try (Cursor<LogData> logList = testMapper.getLogs(param)) {
      for (LogData log : logList) {
        transfer.parseLog(log);
      }
    } catch (Exception e) {
      // ....
    } 
  }
}

 

P.S

 - cursor 를 이용할 경우 fetchSize 또한 적당한 (select 결과 사이즈에 따라, target db network 상황 등에 따라) 사이즈로 설 정 하는 것이 좋다. 

 - mysql, mariadb 의 경우 fetchSize 적용이 잘 안되는 경우가 있는데 jdbcUrl 옵션에 useCursorFetch=true&fetchSize=1000 와 같이 추가하거나 Mybatis 의 Interceptor plugin  기능을 이용해서 설정하면 된다. 

@Intercepts(@Signature(
        type = StatementHandler.class,
        method = "queryCursor", // cursor 사용이 아닌 일반 query 일 경우 method = "query" 이용
        args = {
                Statement.class}))
public class FetchSizePlugin implements Interceptor {
    private static final ThreadLocal<Integer> FETCH_SIZE = ThreadLocal.withInitial(() -> 1000);

    public FetchSizePlugin(int fetchSize) {
        FETCH_SIZE.set(fetchSize);
    }

    @Override
    public Object intercept(Invocation invocation) throws Throwable {
        Statement statement = (Statement) invocation.getArgs()[0];
        statement.setFetchSize(FETCH_SIZE.get());
        return invocation.proceed();
    }

    @Override
    public Object plugin(Object target) {
        return Plugin.wrap(target, this);
    }
}


@Configuration
@MapperScan(basePackages = "com.yhkim.test.repo", sqlSessionFactoryRef = "yhkimSqlSessionFactory")
public class TestDbConfig {
  ....
  @Bean(name = "testSqlSessionFactory")
  @Primary
  public SqlSessionFactory sqlSessionFactory(@Qualifier("testDs") DataSource dataSource) throws Exception {
	SqlSessionFactoryBean sqlSessionFactory = new SqlSessionFactoryBean();
	sqlSessionFactory.setDataSource(dataSource);
	sqlSessionFactory.setTypeAliasesPackage("com.yhkim.test.domain");
	sqlSessionFactory.setMapperLocations(new PathMatchingResourcePatternResolver().getResources("classpath:/mapper/test.xml"));
	sqlSessionFactory.setTypeHandlers(new BooleanTypeHandler());
	sqlSessionFactory.setPlugins(new FetchSizePlugin(1000)); // fetch size plugin 적용 
	return sqlSessionFactory.getObject();
  }
}