[Ray] ExecutionCache 도입으로 데이터셋 캐싱 로직 통합 및 간소화

PR 링크: ray-project/ray#60996 상태: Merged | 변경: +236 / -150

들어가며

Ray Data의 ExecutionPlan에는 실행 결과를 캐싱하는 여러 변수가 산재되어 있었습니다: _snapshot_bundle, _snapshot_operator, _snapshot_stats, _snapshot_metadata_schema, _schema. 이들은 execute()와 execute_to_iterator()에서 각각 다른 방식으로 설정되어, 캐시 무효화(stale check) 로직이 불일치했습니다. 이 PR은 ExecutionCache 클래스로 이들을 통합합니다.

핵심 코드 분석

Before: 산재된 스냅샷 변수

class ExecutionPlan:
    def __init__(self, ...):
        self._snapshot_operator = None
        self._snapshot_stats = None
        self._snapshot_bundle = None
        self._snapshot_metadata_schema = None
        self._schema = None

    def schema(self, fetch_if_missing=True):
        if self._schema is not None:
            return self._schema
        if self.has_computed_output():
            schema = self._snapshot_bundle.schema
        # ...

    def copy(self):
        plan_copy._snapshot_bundle = self._snapshot_bundle
        plan_copy._snapshot_operator = self._snapshot_operator
        plan_copy._snapshot_stats = self._snapshot_stats

After: ExecutionCache로 통합

class ExecutionPlan:
    def __init__(self, ...):
        self._cache = _ExecutionCache()

    def schema(self, fetch_if_missing=True):
        schema = self._cache.get_schema(self._logical_plan.dag)
        if schema is None:
            schema = self._logical_plan.dag.infer_schema()
        # ...

    def copy(self):
        plan_copy._cache = self._cache.copy()

Iterator 완료 시 캐시 업데이트도 통합됩니다:

# Before: 별도의 메타데이터 스냅샷
plan._snapshot_metadata_schema = meta_with_schema

# After: 개별 필드 설정
dag = plan._logical_plan.dag
plan._cache.set_num_rows(dag, self._num_rows)
plan._cache.set_size_bytes(dag, self._size_bytes)
plan._cache.set_schema(dag, schema)

왜 이게 좋은가

• 일관된 캐시 무효화: execute()와 execute_to_iterator() 모두 현재 논리 플랜 DAG와 캐시된 DAG를 비교하여 stale 데이터를 감지합니다.
• 코드 응집도: 5개의 산재된 변수가 하나의 클래스로 통합되어, 캐시 관련 로직이 한 곳에서 관리됩니다.
• 안전한 복사: copy()와 deep_copy()가 ExecutionCache의 메서드를 호출하여 복사 누락이 불가능합니다.
• 성능 재사용: meta_count와 schema가 반복 실행 후 캐시된 데이터를 활용하여 불필요한 재계산을 방지합니다.

참고 자료

• Ray Data ExecutionPlan 아키텍처
• 캐시 무효화 패턴