작업 요건은 단순했습니다. 시스템 전체에 걸쳐 TEN_ID 컬럼의 값을 일괄 변경해야 하는 상황이었습니다. TEN_ID 는 테넌트를 식별하는 공통 키 컬럼으로, 수십 개의 테이블에 걸쳐 존재하는 컬럼이었습니다.

처음에는 단순하게 접근했습니다. USER_TAB_COLUMNS에서 TEN_ID 컬럼을 가진 테이블 목록을 뽑아 루프를 돌면서 테이블마다 UPDATE 를 치는 방식이었습니다.

그런데 곧 문제가 생겼습니다. FK 제약조건 때문이었습니다.

TEN_ID를 참조하는 자식 테이블이 존재하는 상황에서 부모 테이블을 먼저 UPDATE하면 Oracle은 즉시 오류를 발생시킵니다. 그렇다고 자식 테이블을 먼저 UPDATE하자니, 이번엔 또 그 자식 테이블을 참조하는 다른 테이블이 있습니다. 320개 테이블의 FK 참조 관계를 모두 파악해서 UPDATE 순서를 수작업으로 정하는 것은 사실상 불가능한 일이었습니다.

결국 방향을 바꿨습니다. UPDATE 전에 제약조건과 인덱스를 비활성화하고, 병렬 처리를 적용한 뒤, 작업이 끝난 후 다시 복원하는 방식입니다. 결과적으로 320개 테이블, 2,800만 건의 데이터를 15분 안에 처리할 수 있었습니다.

2. 단순 UPDATE가 실패하는 이유

2-1. FK 제약조건의 벽

관계형 데이터베이스에서 외래 키(Foreign Key, FK)는 두 테이블 사이의 참조 무결성을 보장하는 제약조건입니다. 예를 들어 ORDERS 테이블의 TEN_ID 컬럼이 TEN 테이블의 TEN_ID를 참조하고 있다면, TEN 테이블의 TEN_ID 값을 변경하는 순간 ORDERS 테이블의 참조가 깨지기 때문에 Oracle은 즉시 오류를 발생시킵니다.

ORA-02292: integrity constraint violated - child record found

이 문제를 단순하게 해결하려면 참조하는 자식 테이블을 먼저 UPDATE하고, 그다음에 부모 테이블을 UPDATE해야 합니다. 그런데 이 의존 관계가 수십 개의 테이블에 걸쳐 있고, 테이블 간 참조 관계가 복잡하게 얽혀 있다면 순서를 파악하는 것 자체가 쉽지 않습니다. 테이블이 수백 개라면 사실상 수작업으로는 불가능합니다.

가장 근본적인 해결책은 UPDATE를 수행하는 동안 FK 제약조건 자체를 잠시 비활성화하는 것입니다.

2-2. 인덱스가 성능을 잡아먹는 구조

인덱스는 SELECT 성능을 높여주지만, DML(INSERT/UPDATE/DELETE) 성능에는 반대로 작용합니다. 특히 UPDATE의 경우, Oracle은 값이 변경될 때마다 해당 컬럼에 걸린 인덱스 엔트리를 삭제하고 새로운 엔트리를 삽입하는 작업을 내부적으로 수행합니다.

대상 테이블이 수십 개이고, 각 테이블마다 수십만 건의 레코드가 있으며, 해당 컬럼에 인덱스가 걸려 있다면 UPDATE 한 건당 인덱스 유지 비용이 실제 데이터 변경 비용보다 훨씬 커지는 상황이 발생합니다. 전체 작업 시간의 대부분이 인덱스 유지에 소비되는 것입니다.

이 문제를 해결하는 전략은 UPDATE 전에 인덱스를 UNUSABLE 상태로 만들고, UPDATE가 완료된 후에 인덱스를 일괄 재생성하는 것입니다. 인덱스를 처음부터 다시 쌓는 비용이 건건이 유지하는 비용보다 훨씬 저렴하기 때문입니다.

3. 설계 전략 개요

위의 문제들을 해결하기 위한 프로시저의 전체 흐름은 다음과 같습니다.

STEP 1. 제약조건 비활성화 (FK → UNIQUE → PK 순서)
STEP 2. 비고유 인덱스 UNUSABLE 처리
STEP 3. USER_TAB_COLUMNS 기반 동적 병렬 UPDATE
STEP 4. UNUSABLE 인덱스 PARALLEL REBUILD
STEP 5. PK 제약조건 재활성화
STEP 6. UNIQUE 제약조건 재활성화
STEP 7. FK 제약조건 재활성화

각 단계는 독립적으로 보이지만, 순서가 바뀌면 오류가 발생하거나 성능이 크게 저하됩니다. 이 흐름 자체가 이 작업의 핵심 설계입니다.

4. STEP 1 — 제약조건 비활성화

4-1. 제약조건 종류별 처리 순서가 중요한 이유

Oracle의 제약조건 중 DML 성능에 영향을 미치는 제약 조건은 세 가지가 있습니다.

• PK (Primary Key): 테이블의 기본 키입니다. 다른 테이블의 FK가 참조하는 대상이 됩니다.
• UNIQUE: 특정 컬럼의 유일성을 보장합니다. FK 참조 대상이 될 수도 있습니다.
• FK (Foreign Key): 다른 테이블의 PK 또는 UNIQUE를 참조하는 제약조건입니다.

비활성화할 때는 FK를 먼저 끄고, 그다음에 PK와 UNIQUE를 꺼야 합니다. PK를 먼저 비활성화하려고 하면, 해당 PK를 참조하는 FK가 아직 살아있기 때문에 Oracle이 오류를 발생시킵니다.

-- 비활성화 순서: FK → UNIQUE → PK
-- 활성화 순서는 반대:  PK → UNIQUE → FK

코드 상에서는 CONSTRAINT_TYPE DESC 정렬을 활용합니다. Oracle이 제약조건 타입에 사용하는 코드값은 P(Primary Key), R(Referential, 즉 FK), U(Unique)입니다. 이를 알파벳 역순으로 정렬하면 U → R → P 순서가 되며, 덕분에 FK(R)가 PK(P)보다 먼저 비활성화됩니다.

FOR C IN (
    SELECT CONSTRAINT_NAME, TABLE_NAME, CONSTRAINT_TYPE
    FROM USER_CONSTRAINTS
    WHERE CONSTRAINT_TYPE IN ('R', 'P', 'U')
    ORDER BY CONSTRAINT_TYPE DESC  -- U → R → P 순서
) LOOP
    ...
END LOOP;

4-2. CASCADE 옵션

PK를 비활성화할 때는 CASCADE 옵션이 필요할 수 있습니다. CASCADE는 해당 PK를 참조하는 FK들도 함께 비활성화하라는 의미입니다. FK를 먼저 모두 끈 후에 PK를 끄는 구조라면 이미 FK가 비활성화되어 있어 CASCADE가 실질적으로 동작하지 않을 수 있지만, 예외 상황에 대비해 PK에는 CASCADE를 붙여두는 것이 안전합니다.

IF C.CONSTRAINT_TYPE = 'P' THEN
    EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                    ' DISABLE CONSTRAINT ' || C.CONSTRAINT_NAME || ' CASCADE';
ELSE
    EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                    ' DISABLE CONSTRAINT ' || C.CONSTRAINT_NAME;
END IF;

모든 작업은 EXCEPTION WHEN OTHERS THEN NULL로 감싸서, 특정 테이블의 제약조건 비활성화가 실패하더라도 전체 루프가 중단되지 않도록 합니다. 이미 비활성화된 제약조건이나 권한 문제 등으로 일부 실패가 발생할 수 있기 때문입니다.

5. STEP 2 — 인덱스 UNUSABLE 처리

5-1. NONUNIQUE 인덱스만 대상으로 하는 이유

인덱스를 비활성화할 때 주의할 점이 있습니다. UNIQUE 인덱스는 건드리지 않습니다.

UNIQUE 인덱스는 UNIQUE 제약조건 또는 PK 제약조건과 연결되어 있습니다. 이 인덱스를 UNUSABLE로 만들면 제약조건 자체가 무력화되거나, 이후 활성화 과정에서 예기치 않은 오류가 발생할 수 있습니다. 또한 PK 인덱스가 UNUSABLE 상태에서는 해당 테이블에 대한 DML 자체가 거부되는 경우도 있습니다.

따라서 성능 최적화를 위해 비활성화하는 대상은 NONUNIQUE 인덱스로 한정합니다.

FOR IDX IN (
    SELECT INDEX_NAME
    FROM USER_INDEXES
    WHERE INDEX_TYPE NOT IN ('LOB')
      AND UNIQUENESS = 'NONUNIQUE'
) LOOP
    EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME || ' UNUSABLE';
END LOOP;

INDEX_TYPE NOT IN ('LOB')을 추가하는 이유는 LOB 컬럼에 연결된 인덱스는 일반적인 방법으로 UNUSABLE 처리가 되지 않기 때문입니다. 이 조건 없이 LOB 인덱스에 UNUSABLE을 시도하면 오류가 발생합니다.

인덱스가 UNUSABLE 상태가 되면 Oracle은 해당 인덱스를 DML 과정에서 완전히 무시합니다. UPDATE 시 인덱스 유지 비용이 사라지므로 대용량 변경 작업에서 극적인 성능 향상을 기대할 수 있습니다.

6. STEP 3 — 병렬 UPDATE 실행

6-1. USER_TAB_COLUMNS로 대상 테이블 동적 수집

이 작업의 핵심은 변경 대상 컬럼이 존재하는 테이블을 자동으로 찾아내는 것입니다. 하드코딩으로 테이블 목록을 관리하면, 테이블이 추가되거나 삭제될 때마다 프로시저를 수정해야 합니다. 대신 Oracle의 데이터 딕셔너리 뷰인 USER_TAB_COLUMNS를 활용하면 현재 스키마에서 해당 컬럼을 가진 모든 테이블을 실행 시점에 동적으로 수집할 수 있습니다.

FOR COL IN (
    SELECT TABLE_NAME
    FROM USER_TAB_COLUMNS
    WHERE COLUMN_NAME = 'TEN_ID'  -- 변경 대상 컬럼명
    ORDER BY TABLE_NAME
) LOOP
    ...
END LOOP;

USER_TAB_COLUMNS는 현재 사용자 소유의 테이블 컬럼 정보를 모두 담고 있습니다. ALL_TAB_COLUMNS나 DBA_TAB_COLUMNS를 사용하면 다른 스키마의 테이블까지 포함할 수 있습니다.

6-2. PARALLEL 힌트 동작 원리

각 테이블에 대한 UPDATE는 PARALLEL 힌트를 사용해 병렬로 실행합니다.

EXECUTE IMMEDIATE
    'UPDATE /*+ PARALLEL(t, 4) */ ' || COL.TABLE_NAME || ' t ' ||
    'SET TEN_ID = :1 ' ||
    'WHERE TEN_ID = :2'
    USING AFTR_VALUE, PREV_VALUE;

PARALLEL(t, 4)는 t라는 별칭을 가진 테이블에 대해 4개의 병렬 프로세스를 사용하라는 지시입니다. Oracle은 이 힌트를 받으면 테이블의 데이터를 4개의 구간으로 나누어 각 병렬 프로세스가 동시에 UPDATE를 수행하도록 실행 계획을 수립합니다.

병렬 처리가 유효하게 동작하려면 몇 가지 전제 조건이 필요합니다.

• 데이터베이스 레벨에서 병렬 처리가 활성화되어 있어야 합니다(PARALLEL_MAX_SERVERS 파라미터).
• 해당 세션에 병렬 DML이 허용되어 있어야 합니다. 필요하다면 ALTER SESSION ENABLE PARALLEL DML을 사전에 실행해야 합니다.
• 너무 많은 병렬 프로세스는 오히려 리소스 경합을 유발할 수 있으므로, CPU 코어 수와 서버 부하를 고려해 병렬도를 결정해야 합니다.

7. STEP 4 — 인덱스 REBUILD (PARALLEL + NOLOGGING)

UPDATE가 완료된 후에는 UNUSABLE 상태의 인덱스를 재생성해야 합니다. STEP 2에서 UNUSABLE로 만들었기 때문에, STATUS = 'UNUSABLE' 조건으로 대상을 찾으면 됩니다.

FOR IDX IN (
    SELECT INDEX_NAME
    FROM USER_INDEXES
    WHERE STATUS = 'UNUSABLE'
) LOOP
    EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME ||
                    ' REBUILD PARALLEL 4 NOLOGGING';
    EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME || ' NOPARALLEL';
END LOOP;

PARALLEL 4는 인덱스 재생성 자체도 4개의 병렬 프로세스로 수행하라는 의미입니다. 대용량 테이블의 인덱스를 순차적으로 재생성하면 시간이 매우 오래 걸리기 때문에, 병렬로 처리해 재생성 시간을 크게 단축합니다.

NOLOGGING은 인덱스 재생성 과정에서 Redo 로그 생성을 최소화하는 옵션입니다. 일반적인 DML은 장애 발생 시 복구를 위해 모든 변경 내역을 Redo 로그에 기록합니다. 그런데 인덱스 REBUILD는 데이터 자체가 아닌 파생 구조물을 만드는 작업이므로, 혹시 장애가 발생하더라도 인덱스를 다시 재생성하면 그만입니다. 따라서 Redo 로그 기록을 생략해 I/O 부하를 줄이는 NOLOGGING 옵션이 매우 효과적입니다.

REBUILD가 끝난 후 NOPARALLEL을 명시적으로 설정하는 이유는, REBUILD 시 사용한 병렬 설정이 인덱스 속성으로 남아 이후 일반 DML에도 불필요하게 병렬 처리를 시도하는 부작용을 방지하기 위함입니다.

8. STEP 5~7 — 제약조건 ENABLE NOVALIDATE 전략

8-1. VALIDATE vs NOVALIDATE 차이

제약조건을 다시 활성화할 때 두 가지 옵션 중 하나를 선택할 수 있습니다.

• ENABLE VALIDATE (기본값): 제약조건을 활성화하면서 기존 데이터 전체를 검사합니다. 모든 데이터가 제약조건을 만족하는지 확인하므로 데이터가 많을수록 시간이 오래 걸립니다.
• ENABLE NOVALIDATE: 제약조건을 활성화하되, 기존 데이터는 검사하지 않습니다. 이후 신규로 입력되는 데이터에 대해서만 제약조건을 적용합니다.

이번 작업에서는 ENABLE NOVALIDATE를 사용합니다. 이유는 명확합니다. UPDATE 작업을 통해 모든 데이터를 일관성 있게 변경했기 때문에 기존 데이터를 다시 검증할 필요가 없습니다. 불필요한 전체 데이터 스캔을 건너뜀으로써 제약조건 활성화 시간을 크게 단축할 수 있습니다.

EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                ' ENABLE NOVALIDATE CONSTRAINT ' || C.CONSTRAINT_NAME;

8-2. PK → UNIQUE → FK 순서의 이유

제약조건 활성화 순서는 비활성화 순서의 정반대입니다. FK는 PK나 UNIQUE를 참조하기 때문에, 참조 대상인 PK와 UNIQUE가 먼저 활성화되어 있어야 FK를 활성화할 수 있습니다. 만약 FK를 먼저 활성화하려고 하면 참조 대상이 아직 비활성화 상태이므로 오류가 발생합니다.

STEP 5. PK     활성화 — 참조 대상 먼저 준비
STEP 6. UNIQUE 활성화 — 참조 대상 준비 완료
STEP 7. FK     활성화 — 참조 대상이 모두 준비된 후 마지막으로 활성화

9. 트랜잭션 설계 — 중간 COMMIT 전략

대용량 일괄 UPDATE에서 트랜잭션 관리는 매우 중요합니다. 모든 테이블의 UPDATE를 하나의 트랜잭션으로 묶으면 두 가지 문제가 발생합니다.

첫째, Undo 세그먼트 부족 문제입니다. Oracle은 트랜잭션 롤백을 위해 변경 전 데이터를 Undo 세그먼트에 저장합니다. 수백만 건의 UPDATE를 하나의 트랜잭션으로 처리하면 Undo 세그먼트가 고갈되어 ORA-01555: snapshot too old 오류가 발생할 수 있습니다.

둘째, 락(Lock) 경합 문제입니다. UPDATE된 레코드는 COMMIT이 이루어지기 전까지 행 수준 잠금이 유지됩니다. 트랜잭션이 길어질수록 잠금 보유 시간이 늘어나고, 다른 세션과의 경합 가능성이 높아집니다.

이를 해결하기 위해 각 테이블 단위로 COMMIT을 수행합니다.

EXECUTE IMMEDIATE 'UPDATE ... SET COL = :1 WHERE COL = :2' ...;
V_COUNT := SQL%ROWCOUNT;
COMMIT;  -- 테이블 단위 중간 COMMIT

테이블 단위로 COMMIT을 하면 각 테이블의 UPDATE가 독립적인 트랜잭션으로 처리됩니다. 중간에 특정 테이블에서 오류가 발생하더라도 이미 COMMIT된 테이블은 롤백되지 않으므로, 오류 발생 테이블만 재처리할 수 있습니다.

단, 이 전략은 중간 실패 시 일부 테이블만 변경된 불일치 상태가 남을 수 있다는 점을 인지해야 합니다. 이 프로시저는 운영 중단 상태에서 수행하는 일회성 작업을 전제로 설계된 것이므로, 서비스 중 실행보다는 점검 시간에 수행하는 것이 안전합니다.

10. 전체 프로시저 코드

지금까지 설명한 모든 내용을 담은 전체 프로시저 코드입니다.

CREATE OR REPLACE PROCEDURE UPDATE_COLUMN_VALUE_FAST (
    P_COLUMN_NAME IN VARCHAR2,  -- 변경 대상 컬럼명
    PREV_VALUE    IN VARCHAR2,  -- 변경 전 값
    AFTR_VALUE    IN VARCHAR2   -- 변경 후 값
) IS
    V_TOTAL_UPDATED NUMBER := 0;
    V_TABLE_COUNT   NUMBER := 0;
    V_COUNT         NUMBER;
BEGIN
    DBMS_OUTPUT.ENABLE(1000000);

    -- STEP 1: 제약조건 비활성화 (U → R → P 순서)
    FOR C IN (
        SELECT CONSTRAINT_NAME, TABLE_NAME, CONSTRAINT_TYPE
        FROM USER_CONSTRAINTS
        WHERE CONSTRAINT_TYPE IN ('R', 'P', 'U')
        ORDER BY CONSTRAINT_TYPE DESC
    ) LOOP
        BEGIN
            IF C.CONSTRAINT_TYPE = 'P' THEN
                EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                    ' DISABLE CONSTRAINT ' || C.CONSTRAINT_NAME || ' CASCADE';
            ELSE
                EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                    ' DISABLE CONSTRAINT ' || C.CONSTRAINT_NAME;
            END IF;
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    -- STEP 2: NONUNIQUE 인덱스 UNUSABLE 처리
    FOR IDX IN (
        SELECT INDEX_NAME
        FROM USER_INDEXES
        WHERE INDEX_TYPE NOT IN ('LOB')
          AND UNIQUENESS = 'NONUNIQUE'
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME || ' UNUSABLE';
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    -- STEP 3: 동적 병렬 UPDATE
    FOR COL IN (
        SELECT TABLE_NAME
        FROM USER_TAB_COLUMNS
        WHERE COLUMN_NAME = P_COLUMN_NAME
        ORDER BY TABLE_NAME
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE
                'UPDATE /*+ PARALLEL(t, 4) */ ' || COL.TABLE_NAME || ' t ' ||
                'SET ' || P_COLUMN_NAME || ' = :1 ' ||
                'WHERE ' || P_COLUMN_NAME || ' = :2'
                USING AFTR_VALUE, PREV_VALUE;

            V_COUNT := SQL%ROWCOUNT;

            IF V_COUNT > 0 THEN
                V_TABLE_COUNT   := V_TABLE_COUNT + 1;
                V_TOTAL_UPDATED := V_TOTAL_UPDATED + V_COUNT;
            END IF;

            COMMIT;
        EXCEPTION
            WHEN OTHERS THEN
                ROLLBACK;
        END;
    END LOOP;

    -- STEP 4: 인덱스 REBUILD (PARALLEL + NOLOGGING)
    FOR IDX IN (
        SELECT INDEX_NAME
        FROM USER_INDEXES
        WHERE STATUS = 'UNUSABLE'
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME ||
                ' REBUILD PARALLEL 4 NOLOGGING';
            EXECUTE IMMEDIATE 'ALTER INDEX ' || IDX.INDEX_NAME || ' NOPARALLEL';
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    -- STEP 5: PK 활성화
    FOR C IN (
        SELECT CONSTRAINT_NAME, TABLE_NAME
        FROM USER_CONSTRAINTS
        WHERE CONSTRAINT_TYPE = 'P' AND STATUS = 'DISABLED'
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                ' ENABLE NOVALIDATE CONSTRAINT ' || C.CONSTRAINT_NAME;
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    -- STEP 6: UNIQUE 활성화
    FOR C IN (
        SELECT CONSTRAINT_NAME, TABLE_NAME
        FROM USER_CONSTRAINTS
        WHERE CONSTRAINT_TYPE = 'U' AND STATUS = 'DISABLED'
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                ' ENABLE NOVALIDATE CONSTRAINT ' || C.CONSTRAINT_NAME;
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    -- STEP 7: FK 활성화
    FOR C IN (
        SELECT CONSTRAINT_NAME, TABLE_NAME
        FROM USER_CONSTRAINTS
        WHERE CONSTRAINT_TYPE = 'R' AND STATUS = 'DISABLED'
    ) LOOP
        BEGIN
            EXECUTE IMMEDIATE 'ALTER TABLE ' || C.TABLE_NAME ||
                ' ENABLE NOVALIDATE CONSTRAINT ' || C.CONSTRAINT_NAME;
        EXCEPTION
            WHEN OTHERS THEN NULL;
        END;
    END LOOP;

    COMMIT;

    DBMS_OUTPUT.PUT_LINE('완료: ' || V_TABLE_COUNT || '개 테이블, ' ||
        TO_CHAR(V_TOTAL_UPDATED, '999,999,999') || '건 업데이트');

EXCEPTION
    WHEN OTHERS THEN
        ROLLBACK;
        DBMS_OUTPUT.PUT_LINE('오류 발생: ' || SQLERRM);
        RAISE;
END;

AUTOTRACE는 쿼리 실행 결과뿐 아니라 실행 계획(Execution Plan)과 통계(Statistics) 정보를 함께 보여주는 기능입니다. 그러나 통계 정보를 출력하려면 내부적으로 V$SESSTAT, V$STATNAME, V$MYSTAT, V$SESSION 등의 성능 뷰를 참조해야 하며, 일반 사용자에게는 접근 권한이 없습니다. Oracle에서는 이를 위해 PLUSTRACE 롤을 사용하도록 설계되어 있습니다.

문제 상황

PLUSTRACE 롤이 없어 발생한 오류

SP2-0618: 세션 식별자를 찾을 수 없습니다. PLUSTRACE 롤이 사용으로 설정되었는지 점검하십시오
SP2-0611: STATISTICS 레포트를 사용 가능시 오류가 생겼습니다

Oracle 12c 이상에서 CDB 루트에 접속해 발생한 오류

ORA-65096: 공통 사용자 또는 롤 이름은 C## 접두어로 시작해야 합니다

해결 방법

1. PDB로 전환 후 롤 생성

우선 SYSDBA 로 접근 후 현재 컨테이너 확인합니다.

SHOW CON_NAME;

사용 가능한 PDB 확인합니다.

SHOW PDBS;

PDB로 전환합니다.

ALTER SESSION SET CONTAINER = ORCLPDB1;

롤 생성 및 권한을 부여합니다.

CREATE ROLE PLUSTRACE;

GRANT SELECT ON V_$SESSTAT  TO PLUSTRACE;
GRANT SELECT ON V_$STATNAME TO PLUSTRACE;
GRANT SELECT ON V_$MYSTAT   TO PLUSTRACE;
GRANT SELECT ON V_$SESSION  TO PLUSTRACE;

GRANT PLUSTRACE TO SCOTT;

또는 해당 롤은 plustrce.sql이라는 파일로 존재하기에, 위 대신 다음과 같이 간단하게 수행도 가능합니다.

@?/sqlplus/admin/plustrce.sql

2. CDB 루트에서 공통 롤로 생성

이 방식은 CDB 에서 직접 권한을 주는 방법으로 비추천하는 방법입니다.

CREATE ROLE C##PLUSTRACE CONTAINER=ALL;

GRANT SELECT ON V_$SESSTAT  TO C##PLUSTRACE CONTAINER=ALL;
GRANT SELECT ON V_$STATNAME TO C##PLUSTRACE CONTAINER=ALL;
GRANT SELECT ON V_$MYSTAT   TO C##PLUSTRACE CONTAINER=ALL;
GRANT SELECT ON V_$SESSION  TO C##PLUSTRACE CONTAINER=ALL;

GRANT C##PLUSTRACE TO SCOTT CONTAINER=ALL;

AUTOTRACE를 이쁘게 보는 방법

1. SQL*Plus 화면 포맷 조정

SET AUTOTRACE ON
SET LINESIZE 200
SET PAGESIZE 1000
SET TRIMOUT ON
SET TRIMSPOOL ON
SET TAB OFF
COLUMN "Operation" FORMAT A30
COLUMN "Name" FORMAT A15
COLUMN "Cost (%CPU)" FORMAT A15

• SET AUTOTRACE ON : 쿼리를 실행할 때 자동으로 실행 계획과 통계를 함께 출력합니다. 실행 결과만 보는 것이 아니라 성능 분석까지 바로 확인할 수 있는 핵심 옵션입니다.

  • 결과만 보고 싶으면: SET AUTOTRACE OFF

  • 실행 계획만 보고 싶으면: SET AUTOTRACE TRACEONLY EXPLAIN

  • 통계만 보고 싶으면: SET AUTOTRACE TRACEONLY STATISTICS

• SET LINESIZE 200 : 한 줄에 출력할 최대 문자 수를 지정합니다. 기본값이 작으면 실행 계획 표가 줄 바꿈되어 읽기 불편하므로, 200 정도로 설정하면 대부분의 계획이 한 줄에 깔끔하게 표시됩니다.

• SET PAGESIZE 1000 : 한 페이지에 표시할 최대 줄 수를 지정합니다. 작게 설정하면 실행 계획과 통계가 페이지 단위로 나뉘어 보기가 불편합니다. 큰 값으로 설정하면 전체 계획이 한 번에 표시됩니다.

• SET TRIMOUT ON / SET TRIMSPOOL ON : 출력과 SPOOL 파일에서 각 줄 끝의 공백을 제거합니다. 정렬이 깨지지 않고 깔끔하게 표시됩니다.

• SET TAB OFF : 컬럼 구분 시 탭 문자가 아닌 공백을 사용합니다. 탭 때문에 표가 들쭉날쭉해지는 문제를 방지합니다.

• COLUMN "Operation" FORMAT A30, "Name" FORMAT A15, "Cost (%CPU)" FORMAT A15 : 실행 계획의 주요 컬럼 폭을 지정합니다. 기본 폭보다 좁으면 글자가 잘리거나 정렬이 깨지기 때문에, 폭을 지정해 가독성을 높입니다.

2. DBMS_XPLAN 사용

EXPLAIN PLAN FOR
SELECT * FROM EMP WHERE DEPTNO = 10;

SELECT * FROM TABLE(DBMS_XPLAN.DISPLAY);

실제 실행 통계까지 확인:

SELECT * FROM TABLE(DBMS_XPLAN.DISPLAY_CURSOR(NULL, NULL, 'ALLSTATS LAST'));

실제 실행하기

오라클 성능 고도화 원리와 해법2 에 나오는 스크립트를 실행해봅니다.

CREATE TABLE T_EMP
AS
SELECT *
FROM  EMP, (SELECT ROWNUM NO FROM DUAL CONNECT BY LEVEL <= 100000);

ALTER SESSION SET WORKAREA_SIZE_POLICY = MANUAL;

ALTER SESSION SET SORT_AREA_SIZE = 1048576;

우선, 필요한 테이블을 만들고, 디스크 소트가 유발되는지 확인하기 위해 WORKAREA_SIZE_POLICY를 manual로, SORT_AREA_SIZE 를 1 MB 로 낮게 설정하였습니다.

SET AUTOTRACE ON
SET LINESIZE 200
SET PAGESIZE 1000
SET TRIMOUT ON
SET TRIMSPOOL ON
SET TAB OFF
COLUMN "Operation" FORMAT A30
COLUMN "Name" FORMAT A15
COLUMN "Cost (%CPU)" FORMAT A15

SELECT *
FROM (
         SELECT NO, EMPNO, ENAME, JOB, MGR, SAL
              , AVG(SAL) OVER (PARTITION BY TO_CHAR(NO), DEPTNO) AVG_SAL
         FROM   T_EMP
     )
WHERE  NO = 1
ORDER BY SAL DESC;

실행 결과는 아래와 같습니다.

        NO      EMPNO ENAME                JOB                         MGR        SAL    AVG_SAL
---------- ---------- -------------------- -------------------- ---------- ---------- ----------
         1       1009 KING                 PRESIDENT                             5000       2250
         1       1019 KIM                  HR MANAGER                 1009       3500 2816.66667
...

50 행이 선택되었습니다.


Execution Plan
----------------------------------------------------------
Plan hash value: 4263631893

--------------------------------------------------------------------------------------
| Id  | Operation            | Name  | Rows  | Bytes |TempSpc| Cost (%CPU)| Time     |
--------------------------------------------------------------------------------------
|   0 | SELECT STATEMENT     |       |  5000K|   424M|       |   156K  (1)| 00:00:07 |
|   1 |  SORT ORDER BY       |       |  5000K|   424M|   520M|   156K  (1)| 00:00:07 |
|*  2 |   VIEW               |       |  5000K|   424M|       | 54296   (1)| 00:00:03 |
|   3 |    WINDOW SORT       |       |  5000K|   162M|   248M| 54296   (1)| 00:00:03 |
|   4 |     TABLE ACCESS FULL| T_EMP |  5000K|   162M|       |  8875   (1)| 00:00:01 |
--------------------------------------------------------------------------------------

Predicate Information (identified by operation id):
---------------------------------------------------

   2 - filter("NO"=1)


Statistics
----------------------------------------------------------
        253  recursive calls
         13  db block gets
      32393  consistent gets
      96097  physical reads
        668  redo size
       3524  bytes sent via SQL*Net to client
        183  bytes received via SQL*Net from client
          5  SQL*Net roundtrips to/from client
          1  sorts (memory)
          1  sorts (disk)
         50  rows processed

• recursive calls : Oracle 내부에서 발생한 재귀 SQL문 실행 횟수입니다.

• db block gets : 쓰기 가능한 데이터 블록을 읽은 논리적 읽기 횟수로, 인덱스 검색 시 주로 증가합니다.

• consistent gets : 일관성 있는 데이터 블록을 읽은 횟수로, 대부분 메모리에서 처리됩니다.

• physical reads : 디스크에서 직접 읽은 블록 수입니다. 값이 높으면 I/O 부하가 크다는 신호입니다.

• redo size : 쿼리 수행 중 생성된 redo 로그 크기(bytes)입니다.

• bytes sent via SQL*Net to client : 서버에서 클라이언트로 전송된 데이터 크기입니다. 결과가 많으면 값이 커집니다.

• bytes received via SQL*Net from client : 클라이언트에서 서버로 전송된 SQL문이나 파라미터 크기입니다.

• SQL*Net roundtrips to/from client : 서버와 클라이언트 간의 왕복 통신 횟수입니다. 네트워크 비용을 확인할 때 참고합니다.

• sorts (memory) : 메모리에서 수행된 정렬 연산 횟수입니다.

• sorts (disk) : 메모리 부족 시 디스크에 임시 파일을 만들어 수행한 정렬 횟수입니다.

• rows processed : 쿼리에서 실제로 처리된 행(row) 수입니다. SELECT는 반환 행, DML은 영향을 받은 행 수를 나타냅니다.

필요성

SSAFY 다니던 시절 저는 인프라 구성에 대해서는 아예 모르고 있기 때문에 다른 세계의 이야기라고 생각하고 있었습니다.

하지만 프로젝트를 혼자서 해보고 싶다는 생각을 한 저는 인프라를 모르고서는 프로젝트가 의미가 있는지 생각해보게 되었습니다.

마침 주변에 인프라 하시는 분이 있어 CCP & SAA 를 추천을 받아 공부하게 되었습니다.

어떤 시험을 봐야할까?

아래 사진은 현재 기준 모든 AWS 자격증입니다. (SysOps의 경우 CloudOps로 변경되었습니다)

여러 자격증들이 있는데, 도메인 특정되지 않고 입문으로 가장 많이 치르는 가격증이 이제 AWS CCP 또는 AWS SAA 가 있습니다.

우선 AWS CCP(Certified Cloud Prationer, 시험 코드 CLF-C02)의 경우 Foundational 레벨로 AWS의 전반적인 서비스를 물어보는 시험입니다. AWS 서비스 어떤게 있는지 알고 있니?를 묻는 시험입니다.

AWS SAA(Certified Solutions Architect Associate, 시험 코드 SAA-C03)는 Associate 레벨로 CCP 단계에서 더 나아가 실무적인 관점에서의 서비스가 결합되는 것들을 물어봅니다. 문제도 실무에서 있을 법한 상황이 주어지고, 이를 어떻게 AWS 서비스를 사용해서 개선하는지를 묻는 시험입니다.

AWS SAA 부터 시도해도 된다고 인터넷 상에서 글이 많이 보이는데, AWS에 대한 기초 지식이 없으시면 AWS CCP부터 시작하는 것이 좋습니다. 일부 겹치는 내용이 있고, CCP에서 공부한 내용들이나 용어들이 SAA를 도전하는게 아무래도 조금 더 탄탄하기 때문입니다.

만일 본인이 AWS 서비스를 어느정도 사용해보셨다(실무나 아니면 실제 프로젝트 운영 경험)면 SAA부터 시작하셔도 무방합니다.

공통 팁

응시 언어

시험 접수시 응시 언어를 선택할 수 있는데 반드시 한국어로 하시기 바랍니다. 번역이 이상할 수 있지만, 시험 중에 영어 원문도 볼 수 있기 때문에 한국어로 하지 않을 이유가 없습니다.

또한 비영어권 사용자들을 위해 30분이 추가 제공되는데, https://cp.certmetrics.com/amazon/en/schedule/accommodations 에서 가능합니다.

이렇게 30분 추가하면 추후 볼 시험에도 무제한 적용되니 꼭 적용하시기 바랍니다.

시험 선택

시험장을 선택하는 방법과 원격으로 하는 방법이 있는데, CCP의 경우 원격으로, SAA의 경우에는 시험장 방문을 선택했습니다. 장단점이 있는데, 개인적으로는 시험장이 근처에 있다면 시험장 방문을 추천드립니다.

물론 사정상 원격밖에 안되면 원격으로 보셔도 무방합니다만, 개인적으로는 불편한 점들이 있었습니다.

• 모니터가 있으면 절대 안됩니다. 스터디룸 빌려서 시험 봤었는데, 거기 모니터가 있었어서 큰 모니터 어떻게든 치우고 했던 기억이 납니다.
• 감독관 언어로 한국어는 없으니 영어를 선택하실껀데, 높은 확률로 인도식 영어 억양을 듣게 되실껍니다. 익숙하시면 괜찮지만, 전 전혀 익숙하지 않아서 그냥 채팅으로 지시사항 알려달라고 했습니다.
• 주변에서 좀 제한이 빡빡해서 시험 응시 못한 분도 있다고 듣긴 했습니다. 이건 그냥 들은거긴 해서...

마음 편하게 시험장 가는게 낫긴 합니다. 단, 시험장 가실 때는 본인 명의 카드 하나 있어야 된다는 것만 주의하면 됩니다.

AWS CCP

공부 방법

수습 기간 동안 개인 공부가 가능해서 그 기간(대략 1달) 정도만에 준비해서 취득한 시험입니다. 나름 시간이 있어서 압박감은 없이 치를 수 있었습니다.

강의와 유데미 덤프로 학습했고, 유데미 덤프는 2회독 했습니다.

강의

https://www.udemy.com/course/best-aws-certified-cloud/?couponCode=KEEPLEARNING

CCP를 도전하시는 분들은 AWS 서비스가 처음인 경우가 많아서 강의로 어느정도 감을 잡아가면서 공부하시는 것 추천드립니다. CCP의 경우 해당 강의를 완강했고, 강의에 있는 모의고사 하나만 남겨두고 시험이 얼마 남지 않은 시점에 풀었습니다.

또 유데미가 할인을 종종하는데, 회사에서 지원하는거 아니면 할인기간 중에 사는 것 추천드립니다.

덤프

보통 Examtopics 덤프를 많이 보고, 추가적으로 유데미 덤프를 많이 봅니다.

저의 경우에는 유데미만 2회독하고 응시하였습니다. examtopics 덤프는 기출 문제고, 유데미 덤프도 일부 겹치는데 전반적인 난이도는 examtopics < udemy 라는 평이 많습니다.

기타

강의와 덤프를 노션에 정리하면서 푸는데, 잘 이해가 안되는 부분이 있을수 있습니다. 그 경우에는 LLM 을 적극 활용하시는 것을 매우 추천드립니다. 공부 시간을 많이 줄일 수 있습니다. 검증이 필요하다면 역시 공식 문서를 틈틈히 읽어두는 것도 좋은 방법입니다.

시험 후기

CCP는 AWS에 어떤 서비스가 있는지 아는지를 묻는 시험입니다. 그렇다보니 넓고 얇게 알아야 됩니다. 그렇다 보니 기간은 짧았지만, 공부하면서 이걸 다 외워야되나...? 라는 생각이 들었습니다(당시 SAA를 맛보지 않은 시점에서는...ㅎㅎ).

CCP만으로는 스펙이 되는 것도 아니고, AWS 서비스를 잘 만진다는 소리를 듣는 것도 절대 아닙니다. 다만, 가장 많이 응시하는 SAA를 대비하는데 있어 도움이 많이 됩니다. CCP에서 익혔던 개념들은 빠르게 스킵하고 넘어갈 수 있습니다.

AWS SAA

공부 방법

회사에서 업무를 하고 퇴근 후 공부를 하였는데, 퇴근 후 공부가 정말 쉽지 않았음을 느꼈습니다. 그래서 한번에 못땄다면 많이 힘들었을 것 같은데... 다행히 한번에 땄습니다.

CCP를 취득하고, SAA 따야되나 고민하고 있을 때, 주변에 취득하신 분이 SAA 덤프 문제를 보여준 적이 있었습니다. 처음 딱 보고 느낀 점은 쉽지 않겠다고 생각했습니다. 지문 길이나 선지 길이부터 CCP 보다 1.5 ~ 2배는 긴 것 같았습니다.

무엇보다 선지 중에 답 자체는 맞지만, 다소 효율적이지 않아서 오답인 선지들이 많아서 찍는 상황도 존재했습니다.

강의

https://www.udemy.com/course/best-aws-certified-solutions-architect-associate/

회사 다니면서 완강하기가 너무 벅차긴 했습니다. 70퍼 정도까지 듣고 덤프 풀이에 들어갔습니다. 다만 강의 중에 솔루션 아키텍쳐 토론 강의가 있는데, AWS 아키텍쳐의 기본 구조를 잡는데 도움이 되기 때문에 강의가 있으시면 이 부분은 듣는 것을 추천드립니다.

덤프

examtopics 와 유데미 덤프를 모두 풀었습니다. 유데미를 먼저 풀었고, examtopics 는 200 문제정도 푼 것 같습니다.

공부시간

강의는 지하철 출퇴근 시간을 사용해서 들어도 봤는데, 집중이 잘 안되긴 합니다. 따로 시간을 잡아서 듣는 편이 좋습니다.

문제 풀이의 경우 유데미는 모바일로 출퇴근 시간에 풀고, 집에서는 필요한 부분 정리하는 방식으로 진행했습니다.

시험 후기

다행히 한번에 합격했습니다. 직장 다니면서 공부하는건 정말 쉬운 일이 아닙니다. 게다가 현재 SQLP 와 병행하고 있어 온전히 퇴근 후 AWS에 쏟기는 힘든 상황이라 며칠은 못할 때도 있었습니다.

시험 치는 주에는 계속 틀렸던거 틀리고 해서 좀 불안불안 했습니다. 그리고 시험 직전에 마지막 모의고사를 풀었는데 딱 커트라인 72%가 나와서 더욱 그랬습니다.

시험 때는 둘 중 하나 중에 찍는 상황이 많았고, 평소에 잘 봤던 S3는 쉽게 출제 되고, 잘 보지도 않던 EKS가 꽤 나왔습니다(제 생각에는 채점되지 않는 문제들에 속할지도 모르겠습니다.).

다행히 저녁에 뱃지가 와서 안도의 한숨을 내쉬었던 기억이 납니다.

마무리

SAA까지 땄는데 그러면 인프라 구성 잘하겠네? 하면 절대 아닙니다. 다만 SAA를 취득함으로써 이제야 인프라 아키텍쳐가 좀 보이고, 시작이구나를 느낍니다. 가야할 길은 멀지만, SAA가 이 길의 초석이 될 것 같습니다.

그리고 퇴근 후 공부하는 모든 직장인 분들, 화이팅입니다.

Docker 환경 구성

도커 이미지는 여러가지가 있지만 오라클 공식 사이트에서 제공하는 것을 기준으로 작성하였습니다.

도커 이미지만 다운로드

docker pull container-registry.oracle.com/database/free:latest

도커로 이미지 다운받아 실행

docker run -d -p 1521:1521 --name oracle -e ORACLE_PASSWORD=oracle container-registry.oracle.com/database/free:latest

• -d : detach, 컨테이너가 백그라운드에서 실행될 수 있게 합니다(데몬 모드).
• -p : port, 각 포트를 설정합니다(오라클의 기본포트는 1521).
• -e : 컨테이너 내에서 사용될 환경변수를 설정합니다.

도커 프로세스 눈으로 확인

docker ps -a

도커 프로세스 실행상태 로그 출력

docker logs -f oracle

아래 메세지가 나올 때 까지 기다립니다.

#########################
# DATABASE IS READY TO USE!
#########################

만약 sys나 system의 암호을 잊었다면

docker exec oracle resetPassword 원하는암호

중단

docker stop oracle

시작

docker start oracle

해당 컨테이너의 system 계정으로 접속

docker exec -it oracle /bin/bash

sqlplus / as sysdba

• -i, --interactive=false : 표준 입력(stdin)을 활성화하며 컨테이너와 연결(attach)되어 있지 않더라도 표준 입력을 유지한다.
• -t, --tty=false : TTY 모드(pseudo-TTY)를 사용한다. Bash를 사용하려면 이 옵션을 설정해야 한다. 이 옵션을 설정하지 않으면 명령을 입력할 수는 있지만 셸이 표시되지 않는다.
• sqlplus / as sysdba : 오라클 데이터베이스로 system 계정을 가지고 접속

접속 후 User 만들기

PDB에 접속

유저를 생성할 PDB에 접속합니다.

SQL> show pdbs

    CON_ID CON_NAME                       OPEN MODE  RESTRICTED
---------- ------------------------------ ---------- ----------
         2 PDB$SEED                       READ ONLY  NO
         3 FREEPDB1                       READ WRITE NO

## PDB에 접속
SQL> alter session set container=FREEPDB1; 

Session altered.

User 생성

다음과 같은 예시 유저를 만들고 실습을 진행합니다.

• 유저명 : docker
• PW : docker123

## user 생성
SQL> create user docker identified by "docker123";

User created.

## 접속, resource 권한 부여
SQL> grant connect, resource to docker;

Grant succeeded.

## users 테이블스페이스 사용 권한 설정
SQL> alter user docker quota unlimited on users;

User altered.

Client Tool (DBeaver, Datagrip, …) 접속

• driver : thin
• SID 대신 서비스명 으로 접속할 것
• 서비스명 : 처음 session 설정한 PDB(FREEPDB1)
• jdbc:oracle:thin:@localhost:1521/FREEPDB1

SQL*Plus 접근 방법

• 실습 하다보면 SQL*Plus로 터미널로 직접 접근 할 일이 생김
• 방법 3가지 중 선택

## 1. Docker Oracle 컨테이너에서의 일반적인 접속
docker exec -it oracle sqlplus docker/docker123@FREEPDB1

## 2. exec 환경에서 PDB 지정하여 접속
## bash 는 /bin/bash 과 동일
docker exec -it oracle bash
sqlplus docker/docker123@FREEPDB1

## 3. exec 환경에서 관리자 접근 후 세션에서 PDB 변경 후 접속
docker exec -it oracle bash
sqlplus / as sysdba
ALTER SESSION SET CONTAINER = FREEPDB1;
CONNECT docker/docker123;

PDB?

중간에 PDB가 나와서 뭐지...? 생각하셨을 수도 있어서 첨부합니다.

아래는 오라클 컨테이너 데이터베이스의 기본 구조입니다. CDB(Container Database) 가 중앙으로 관리하고, PDB(Pluggable Database)를 여러 개 붙이는 형태입니다. 이를 오라클 멀티테넌트(Multitenant)라 합니다.

CDB (Container Database) - 컨테이너 데이터베이스
├── CDB$ROOT - 루트 컨테이너
├── PDB$SEED - 시드 PDB (템플릿)
├── PDB1 - 첫 번째 Pluggable Database
├── PDB2 - 두 번째 Pluggable Database
└── PDB3 - 세 번째 Pluggable Database

특징

• 각 PDB는 논리적으로 완전히 분리
• PDB를 다른 CDB로 쉽게 이동 가능, 마치 USB 처럼
• 리소스 효율성 - 메모리, CPU 등 시스템 리소스 공유

주요 명령어

• 현재 환경 확인 ```sql
• • 현재 컨테이너 확인 SHOW CON_NAME;

-- 모든 PDB 목록 조회 SELECT name, open_mode FROM v$pdbs;

-- CDB인지 확인 SELECT cdb FROM v$database;

- PDB 간 이동
```sql
-- 특정 PDB로 세션 변경
ALTER SESSION SET CONTAINER = XEPDB1;

-- 루트 컨테이너로 이동
ALTER SESSION SET CONTAINER = CDB$ROOT;

• PDB 관리 ```sql
• • PDB 열기 ALTER PLUGGABLE DATABASE pdb1 OPEN;

-- PDB 닫기 ALTER PLUGGABLE DATABASE pdb1 CLOSE;

-- PDB 생성 CREATE PLUGGABLE DATABASE pdb_new FROM pdb$seed;

-- PDB 삭제 DROP PLUGGABLE DATABASE pdb_old INCLUDING DATAFILES;

### 접속 방법

```bash
## 1. Docker Oracle 컨테이너에서의 일반적인 접속
docker exec -it oracle sqlplus docker/docker123@FREEPDB1

## 2. DataGrip 연결
jdbc:oracle:thin:@localhost:1521/FREEPDB1

## 3. exec 환경에서 PDB 지정하여 접속
docker exec -it oracle bash
sqlplus docker/docker123@FREEPDB1

## 4. exec 환경에서 관리자 접근 후 세션에서 PDB 변경 후 접속
docker exec -it oracle bash
sqlplus / as sysdba
ALTER SESSION SET CONTAINER = FREEPDB1;
CONNECT docker/docker123;

PDB 환경 확인

-- 1. CDB 환경인지 확인
SELECT cdb FROM v$database;

-- 2. 현재 위치 확인
SHOW CON_NAME;

-- 3. 모든 PDB 조회
SELECT name, open_mode, restricted FROM v$pdbs;

-- 4. 서비스 이름 확인
SELECT name FROM v$services;

SQL> show pdbs

    CON_ID CON_NAME                       OPEN MODE  RESTRICTED
---------- ------------------------------ ---------- ----------
         2 PDB$SEED                       READ ONLY  NO
         3 FREEPDB1                       READ WRITE NO

• PDB에 접속

  SQL> alter session set container=FREEPDB1; 
  

Session altered. ```

SQL 최적화

• SQL 파싱 과정
  • 파싱 트리 생성 : SQL문 이루는 개별 구성요소를 분석, 파싱 트리 생성
  • Syntax 체크 : 문법적 오류 확인
  • Semantic 체크 : 의미상 오류 확인
• SQL 최적화 - Optimizer
• 로우 소스 생성

SQL 옵티마이저

• 옵티마이저의 최적화 단계
  • 사용자로부터 전달받은 쿼리를 수행하는데 후보군이 될만한 실행계획들을 찾아냄
  • 데이터 딕셔너리에 미리 수집해 둔 오브젝트 통계 및 시스템 통계정보를 이용해 각 실행계획의 예상비용을 산정
  • 최저 비용을 나타내는 실행계획을 세움

실행계획과 비용

• 옵티마이저는 비용에 따라 인덱스를 선택 및 실행 계획 수립

    -- 테스트용 테이블 생성
    CREATE TABLE T AS
    SELECT D.NO, E.*
    FROM (
         SELECT * FROM EMPLOYEES
             JOIN DEPT_EMP USING (EMP_NO)
             JOIN DEPARTMENTS USING (DEPT_NO)
    ) E, (SELECT ROWNUM NO FROM DUAL CONNECT BY LEVEL <= 100) D;
  
    -- 3316만 건
    SELECT COUNT(*) FROM T;
  
    -- 인덱스 생성
    CREATE INDEX T_X01 ON T(DEPT_NO, NO);
    CREATE INDEX T_X02 ON T(DEPT_NO, FIRST_NAME, LAST_NAME, NO);

  별다른 힌트 없이 쿼리 실행 : 옵티마이저의 판단에 따라 T_X01 선택

    SELECT * FROM T
    WHERE DEPT_NO = 'd005'
    AND NO = 1;

  T_X02 로 옵티마이저 힌트를 줄 때 : cost 증가

    ---------------------------------------------------------------------------------------------
    | Id  | Operation                           | Name  | Rows  | Bytes | Cost (%CPU)| Time     |
    ---------------------------------------------------------------------------------------------
    |   0 | SELECT STATEMENT                    |       | 36845 |  2266K|  3260   (1)| 00:00:01 |
    |   1 |  TABLE ACCESS BY INDEX ROWID BATCHED| T     | 36845 |  2266K|  3260   (1)| 00:00:01 |
    |*  2 |   INDEX RANGE SCAN                  | T_X01 | 36845 |       |   100   (0)| 00:00:01 |
    ---------------------------------------------------------------------------------------------

  Full scan : cost 증가

    SELECT /*+ INDEX(T T_X02) */ * FROM T
    WHERE DEPT_NO = 'd005'
    AND NO = 1;

    ---------------------------------------------------------------------------------------------
    | Id  | Operation                           | Name  | Rows  | Bytes | Cost (%CPU)| Time     |
    ---------------------------------------------------------------------------------------------
    |   0 | SELECT STATEMENT                    |       | 36845 |  2266K| 54548   (1)| 00:00:03 |
    |   1 |  TABLE ACCESS BY INDEX ROWID BATCHED| T     | 36845 |  2266K| 54548   (1)| 00:00:03 |
    |*  2 |   INDEX RANGE SCAN                  | T_X02 | 36845 |       | 17699   (1)| 00:00:01 |
    ---------------------------------------------------------------------------------------------

    SELECT /*+ FULL(T) */ * FROM T
    WHERE DEPT_NO = 'd005'
    AND NO = 1;

    --------------------------------------------------------------------------
    | Id  | Operation         | Name | Rows  | Bytes | Cost (%CPU)| Time     |
    --------------------------------------------------------------------------
    |   0 | SELECT STATEMENT  |      | 36845 |  2266K| 86009   (1)| 00:00:04 |
    |*  1 |  TABLE ACCESS FULL| T    | 36845 |  2266K| 86009   (1)| 00:00:04 |
    --------------------------------------------------------------------------

• 개발자가 직접 데이터 엑세스 경로를 변경 가능 → 옵티마이저 힌트

• cost는 예상치이기에 실제 시간과는 차이 존재

옵티마이저 힌트

• 사용법

    SELECT /*+ INDEX(A 고객_PK) */
                고객명, 연락처, 주소, 가입일시
    FROM 고객 A
    WHERE 고객ID = '1111'

• 주의 사항

    -- 콤마 사용은 인자에서만 가능
    /*+ INDEX(A A_X01 INDEX(B,B_X03) */ -- 모두 유효
    /*+ INDEX(C), FULL(D) */ -- 첫 번째 힌트만 유효
  
    -- 테이블을 지정시 스키마 명 명시 금지
    SELECT /*+ FULL(PUBLIC.EMP) */ -- 무효
    FROM EMP
  
    -- FROM 에서 ALIAS 지정시 힌트에도 ALIAS 사용
    SELECT /*+ FULL(EMP) */ -- 무효
    FROM EMP E

자주 사용하는 힌트 목록

SQL 공유 및 재사용

소프트 파싱 vs 하드 파싱

• 라이브러리 캐시 : SQL 파싱, 최적화, 로우 소스 생성 과정을 거쳐 생성한 내부 프로시저를 반복 재사용할 수 있도록 캐싱해두는 메모리 공간

• SGA(System Global Area) : 서버 프로세스와 백그라운드 프로세스가 공통으로 액세스하는 데이터와 제어 구조를 캐싱하는 메모리 공간

• 소프트 파싱 : SQL을 캐시에서 찾아 곧바로 실행 단계에서 넘어가는 것

• 하드 파싱 : SQL을 캐시에서 찾는데 실패해 최적화 및 로우 소스 생성까지 거치는 것

• 하나의 쿼리를 수행하는데 있어 후보군이 될만한 무수히 많은 실행경로를 도출하고, 짧은 순간에 딕셔너리와 통계정보를 읽어 각각에 대한 효율성을 판단하는 과정은 무거움

  • DB의 대부분의 작업은 I/O, but 하드 파싱은 CPU를 소비
  • 옵티마이저가 사용하는 정보들
    • 테이블, 컬럼, 인덱스 구조에 관한 기본 정보
    • 오브젝트 통계 : 테이블 통계, 인덱스 통계, 컬럼 통계
    • 시스템 통계 : CPU 속도, Single Block I/O 속도, Multiblock I/O 속도 등
    • 옵티마이저 관련 파라미터
  • 캐시로 저장해서 하드 파싱 횟수를 줄이기

바인드 변수의 중요성

• 라이브러리 캐시에서 SQL을 찾는 법 - SQL 문 그 자체

  • SQL 텍스트가 변하면 SQL ID도 변함
  • 개발 과정 중에는 수시로 SQL이 바뀜 - 계속 적재되면 캐시에서 찾는게 느려짐

• 로그인 ID를 받는 로그인을 생각

  • 아래와 같이 로그인ID를 파라미터로 받는 프로시저 하나 공유 → 바인드 변수

      CREATE PROCEDURE LOGIN (login_id in varchar2) { ... }
    
      -- 라이브러리 캐시 : 하드 파싱 최초 1회만 발생
      SELECT * FROM CUSTOMER WHERE LOGIN_ID = :1

데이터 저장 구조 및 I/O 메커니즘

SQL이 느린 이유

• 프로세스는 디스크 I/O를 요청할 시에 CPU를 OS에 반환하고 잠시 waiting 상태에서 I/O가 완료될 때까지 대기 진행 → 대기 큐에서 머물러 있음 → 작동해야 될 프로세스가 가만히 대기 → 성능 저하

데이터베이스 저장 구조

테이블스페이스 → 세그먼트 → 익스텐트 → 블록 → 로우

• 테이블스페이스 : 세그먼트를 담는 컨테이너
• 세그먼트 : 테이블, 인덱스, 파티션, LOB(Large Object) 같이 저장 공간이 필요한 영역
  • 세그먼트의 공간이 부족해지면 익스텐트를 추가로 할당 받음
  • 하지만 할당된 모든 익스텐트가 같은 데이터 파일에 위치하지 않을 수도 있음 - 분산 저장
  • DBA_EXTENTS 뷰에서 세그먼트에 할당된 익스텐트 목록 조회 가능
• 익스텐트 : 공간을 확장하는 단위, 여러 블록들로 구성
• 블록 : 레코드를 실제로 저장하는 공간, 페이지라고도 불림
  • 한 블록은 하나의 테이블이 독점 → 한 블록에 저장된 레코드는 모두 같은 테이블 레코드
  • DBA(Data Block Address) : 디스크 상에서 몇 번째 블록인지 나타냄
  • 테이블 스캔 시 : 익스텐트 맵 활용 → 첫 번째 블록 DBA → 블록은 연속이므로 Scan 가능
• 데이터파일 : 디스크 상의 물리적인 OS 파일

익스텐트 내 블록은 서로 인접, 익스텐트 끼리는 불연속!

• DBA(Data Block Address) : 블록의 고유 주소값
  • 인덱스를 이용해 레코드를 읽을 때는 ROWID (DBA + 로우 번호)로 조회
  • 테이블 스캔 시에는 세그먼트 헤더에 저장된 익스텐트 맵 이용 → 첫번째 DBA 파악

블록 단위 I/O

DBMS에서 데이터를 읽고 쓰는 단위

V$PARAMETER 조회

SQL> show parameter block_size

NAME                                 TYPE        VALUE
------------------------------------ ----------- ------------------------------
db_block_size                        integer     8192

SQL> select value from v$parameter where name='db_block_size';

VALUE
--------------------------------------------------------------------------------
8192

시퀀셜 액세스 vs 랜덤 액세스

• 시퀀셜 : 연결된 순서에 따라서
  • 하지만 테이블 블록간 논리적인 연결고리는 없음
    • 익스텐트 목록을 세그먼트 헤더에 map으로 관리
    • 익스텐트 맵 활용 → 첫 번째 블록 DBA → 블록은 연속이므로 Scan 가능
• 랜덤 : 순서X, 한 블록씩 접근

논리적 I/O vs 물리적 I/O

• 논리적 : SQL을 처리하는 과정에 발생한 총 블록 I/O (캐시를 모두 거침 - 사실상 메모리 I/O)
• 물리적 : 디스크에서 발생항 총 블록 I/O → 시간 오래 걸림

버퍼캐시 히트율(Buffer Cache Hit Ratio, BCHR)

BCHR = (캐시에서 곧바로 찾은 블록 수 / 촌 읽은 블록 수) * 100
    = ((논리적 I/O - 물리적 I/O) / (논리적 I/O)) * 100
    = (1 - (물리적 I/O) / (논리적 I/O)) * 100

물리적 I/O = 논리적 I/O * (100% - BCHR)

• 논리적 I/O 줄이기 - SQL을 튜닝해서 읽는 총 블록 개수 줄이기

Q. BCHR는 SQL 성능을 좌우 하지만, BCHR이 높다고 효율적인 SQL을 의미하진 않는다??

• 같은 블록을 비효율적으로 반복해서 읽으면 BCHR이 상승

  SELECT e.empno, e.ename, d.dname
  FROM emp e
  JOIN dept d ON e.deptno = d.deptno;

• emp가 100만건, dept가 10만건 이라 가정
• 옵티마이저가 dept 인덱스를 매번 스캔하면, dept 블록은 한번 읽어서 캐시에 있지만, 100만 번의 논리적 I/O가 발생 → BCHR는 높지만, CPU는 혹사

Single Block I/O vs Multiblock I/O

• 한번에 한 블록씩 요청해서 메모리 적재 → Single Block I/O
  • 인덱스를 이용할 시 인덱스와 테이블 블록 모두 Single Block I/O
    • 인덱스 루트 블록을 읽을 때
    • 인덱스 루트 블록에서 얻은 주소 정보로 브랜치 블록을 읽을 때
    • 인덱스 브랜치 블록에서 얻은 주소 정보로 리프 블록을 읽을 때
    • 인덱스 리프 블록에서 얻은 주소 정보로 테이블 블록을 읽을 때
• 한번에 여러 블록씩 요청해서 메모리 적재 → Multiblock I/O
  • 많은 데이터 블록을 읽을 때 유리 → 테이블 풀 스캔
  • OS 레벨 I/O 단위 : 1 MB, 오라클 레벨 I/O 단위 : 8 KB
  • 확인 - db_file_multiblock_read_count 파라미터
• Multiblock I/O 중간에 Single Block I/O 가 나타나는 이유
  • 익스텐트 맵은 테이블 블록에 대한 인덱스
  • Multiblock I/O는 Batch I/O
  • Full Scan 중 Chain 이 발생할 로루를 읽을 때

Table Full Scan vs Index Range Scan

• 테이블 전체 스캔 vs 인덱스로 일정량만 스캔
• Table Full Scan
  • 시퀀셜 액세스 & Multi Block I/O 방식으로 디스크 블록 읽음
  • 소량의 데이터만 가져올 때는 불리, 가져오는 데이터의 사이즈가 클 경우 유리
• Index Range Scan
  • 랜덤 액세스 & Single Block I/O
  • 소량의 데이터만 가져올 때는 유리, 가져오는 데이터의 사이즈가 클 경우 불리

캐시 탐색 메커니즘

• Direct Path I/O를 제외한 모든 블록 I/O는 메모리 버퍼 캐시를 경유
  • 인덱스 루트 블록을 읽을 때
  • 인덱스 루트 블록에서 얻은 주소 정보로 브랜치 블록을 읽을 때
  • 인덱스 브랜치 블록에서 얻은 주소 정보로 리프 블록을 읽을 때
  • 인덱스 리프 블록에서 얻은 주소 정보로 테이블 블록을 읽을 때
  • 테이블 블록을 Full Scan 할 때
• 해시 체이닝을 통한 해시 구조 관리 : Hash Table을 생각
• SGA 내부에 있는 버퍼 캐시는 공유자원이기 때문에 직렬화가 필수
  • 직렬화 : 줄세우기
    • 여러 프로세스가 동시에 접근할 수 없도록 순서대로 해시 체인을 탐색
    • Latch를 통해 접근한 프로세스만 체인에 관여할 수 있도록 하여, 탐색 과정 중 변경을 방지 ⇒ LRU 채인 래치
  • 버퍼 Lock을 통해, 데이터 정합성을 방지
    • 선행 프로세스가 작업을 완료하고 Latch를 풀기 전, 버퍼 캐시에 I/O 동안 접근하지 못하도록 Lock을 수행
• 가급적 SQL 튜닝을 통해 쿼리 일량(논리적 I/O) 자체를 줄여야 함

[PRG] 입양 시각 구하기

접근 방법

문제를 쪼개어서 접근하면 다음과 같습니다.

• DATETIME 에서 시간 추출
• 시간별로 동물 세기

시간 추출

시간을 추출하는 방법은 여러 방법이 있지만, 가장 쉬운 방법은 TO_CHAR(DATETIME, 'HH24') 를 사용하는 것 입니다. 주의할 것은 시간을 추출할 때 'HH' 가 아닌 'HH24' 를 사용하여 24시간을 표현해야 하는 것입니다. 이렇게 해야 [0, 24) 의 값이 나오게 됩니다.

마침 과거에 간단한 문법을 제가 정리한 내용이 있어 이를 첨부합니다.

시간별로 동물 세기

이는 기본적인 GROUP BY 를 시간에 대해서 진행하면 됩니다.

현재까지 정리한 내용을 SQL로 나타내면 다음과 같습니다.

  SELECT TO_CHAR(DATETIME, 'HH24') AS HOUR,
      COUNT(ANIMAL_ID) AS COUNT
  FROM ANIMAL_OUTS
  GROUP BY TO_CHAR(DATETIME, 'HH24')

문제점

하지만 문제가 있습니다. 바로, [0, 24) 까지의 모든 시간이 반영되어 있지 않다는 점입니다.

만약 0시 기록이 없을 경우 COUNT 를 0으로 처리해두어야 하는데, 애초에 시간 데이터 자체가 없어서 표시 자체를 해둘 수가 없습니다.

뭔가 저 데이터와 상관없이 [0, 24) 를 모두 가지는 테이블 하나가 있으면 좋겠다는 생각이 듭니다.

만드는 방법에는 직접 작성하는 방법도 있긴 하겠지만 너무 귀찮습니다. 그렇다고 PL/SQL 까지 가는 것도 무겁습니다. 바로 이 때 재귀 쿼리를 생성하면 됩니다.

재귀 쿼리 역시 정리한 것이 있어 이 포스팅을 참고하면 됩니다. 다만 해당 포스팅은 MySQL 기반이라 Oracle 기반은 여기를 참고하면 됩니다.

Oracle 계층형 쿼리

MySQL은 CTE + RECURSIVE 를 사용하는데, Oracle은 CONNECT BY 를 사용합니다. 사용 예시를 [0, 24) 를 모두 가지는 테이블인 TIME_TABLE 을 만들어서 해결해보겠습니다.

WITH TIME_TABLE AS (
    SELECT LEVEL - 1 AS HOUR
    FROM DUAL
    CONNECT BY LEVEL <= 24
)

사실 이 CONNECT BY 는 무려 CTE 없이도 사용 가능합니다!

SELECT column1, column2, ...
FROM table_name
START WITH condition
CONNECT BY [PRIOR] condition

기본적인 문법은 다음과 같이 사용되고, 해당 쿼리를 CTE 말고 단일 쿼리든, 서브 쿼리든 어디든 사용이 가능합니다. Oracle이 CTE 제공 이전부터 계층형 쿼리가 필요할 때 제공하던 강력한 문법이고, 다른 RDBMS에서는 적용이 되지 않으므로 주의하기 바랍니다.

START WITH

START WITH MGR_ID IS NULL --MGR_ID 가 NULL인 ROW를 최상위로 시작

트리에서 루트 노드 설정하는 것처럼 계층 구조에서 최상위 ROW를 선정하는 조건입니다.

CONNECT BY

CONNECT BY PRIOR 부모노드 = 자식노드  
CONNECT BY 자식노드 = PRIOR 부모노드

어떻게 연결되어 있는지를 명시해주는 부분입니다. 기본적으로 자식은 PRIOR 부모라 생각하시면 편합니다.

유용한 함수들

이런 CONNECT BY 와 활용하기좋은 함수들이 몇가지 있는데, 저의 경우에는 LEVEL을 사용해서 계층의 깊이를 표시하였습니다. 이 문제에서는 필수긴 하지만, 알고리즘 풀 때도 깊이 정보 정도는 마킹하는 만큼 습관이 되면 좋습니다.

LEVEL

현재 계층의 깊이를 나타냅니다.

SELECT LEVEL, 'Level ' || LEVEL AS DESCRIPTION
FROM DUAL
CONNECT BY LEVEL <= 5;

SYS_CONNECT_BY_PATH

루트부터 현재 노드까지의 경로를 보여줍니다.

SELECT EMP_NAME,
       SYS_CONNECT_BY_PATH(EMP_NAME, ' -> ') AS PATH
FROM EMPLOYEES
START WITH MGR_ID IS NULL
CONNECT BY PRIOR EMP_ID = MGR_ID;

CONNECT_BY_ROOT

루트 노드의 값을 가져옵니다.

SELECT EMP_NAME,
       CONNECT_BY_ROOT EMP_NAME AS TOP_MANAGER
FROM EMPLOYEES
START WITH MGR_ID IS NULL
CONNECT BY PRIOR EMP_ID = MGR_ID;

예시 답안

WITH TIME_TABLE AS (
    SELECT LEVEL - 1 AS HOUR
    FROM DUAL
    CONNECT BY LEVEL <= 24
), ANIMAL_LIST AS (
    SELECT TO_CHAR(DATETIME, 'HH24') AS HOUR,
        COUNT(ANIMAL_ID) AS COUNT
    FROM ANIMAL_OUTS
    GROUP BY TO_CHAR(DATETIME, 'HH24')
)

SELECT T.HOUR,
    NVL(L.COUNT, 0) AS COUNT
FROM TIME_TABLE T
LEFT JOIN ANIMAL_LIST L ON (T.HOUR = L.HOUR)
ORDER BY HOUR

NULL 처리는 Oracle 제공인 NVL 을 사용하였지만, 다음과 같은 COALESCE 사용을 추천하고 있습니다.

-- COALESCE(expr1, expr2, expr3, ...)
-- 첫 번째 NULL이 아닌 값을 반환
-- 여러 개의 표현식 중에서 선택 가능
WITH TIME_TABLE AS (
    SELECT LEVEL - 1 AS HOUR
    FROM DUAL
    CONNECT BY LEVEL <= 24
), ANIMAL_LIST AS (
    SELECT TO_CHAR(DATETIME, 'HH24') AS HOUR,
        COUNT(ANIMAL_ID) AS COUNT
    FROM ANIMAL_OUTS
    GROUP BY TO_CHAR(DATETIME, 'HH24')
)

SELECT T.HOUR,
    COALESCE(L.COUNT, 0) AS COUNT
FROM TIME_TABLE T
LEFT JOIN ANIMAL_LIST L ON (T.HOUR = L.HOUR)
ORDER BY HOUR

CTE

CTE(Common Table Expression)은 단일 쿼리 내부에서 임시로 쿼리 결과를 저장하고, 해당 쿼리 내에서 반복적으로 사용 가능한 개념입니다.

이걸 들었을 때 가장 먼저 생각나는건 아무래도 가상 테이블 VIEW 인데, 차이점이 있습니다.

• VIEW 는 만들기 위해서 특별한 권한이 필요하고, 쿼리 실행 전 사전에 정의 되어 있어야 하며, 쿼리문이 끝나도 사용 가능합니다.
• 반면 CTE는 특별한 권한은 필요 없고, 쿼리가 끝나면 사라지는 임시 테이블입니다.

이를 쓰는 이유는 서브 쿼리와 달리 동일 쿼리 내에서 재사용이 가능하고, 또 자체 참조로 재귀 쿼리를 작성할 수 있음에 있습니다.

활용

기본 활용

기본적인 문법은 아래와 같습니다. WITH 절을 사용하여 정의합니다.

WITH cte_name AS (
    SELECT column1, column2, ...
    FROM table_name
    WHERE condition
)
SELECT *
FROM cte_name
WHERE additional_condition;

테이블명 옆에 컬럼명을 새로 지정하여 사용할 수도 있습니다.

WITH cte_name (column1, column2, ...) AS (
    SELECT column1, column2, ...
    FROM table_name
    WHERE condition
)
SELECT *
FROM cte_name
WHERE additional_condition;

여러 개 만들기

여러 개를 만드는 것은 단순히 , 로 이어주면 됩니다.

WITH 
    cte1 AS (
        SELECT column1, column2
        FROM table1
        WHERE condition1
    ),
    cte2 AS (
        SELECT column3, column4
        FROM table2
        WHERE condition2
    )
SELECT c1.column1, c2.column3
FROM cte1 c1
JOIN cte2 c2 ON c1.column2 = c2.column4;

재귀 쿼리

재귀 쿼리는 재귀 CTE를 사용하여 구현합니다. 재귀 CTE는 자기 자신을 참조하여 반복적으로 실행되는 임시 결과 집합입니다. 쿼리로 알고리즘 문제 풀 듯 DFS가 가능합니다.

가장 대표적으로 쓰이는 사례는 트리 형태로 되어있는 자료 검색입니다.

보통 부서도를 볼 때 다음과 같은 형태로 되어 있는 걸 볼 수 있습니다. 보통 이런 스키마의 경우 ID 와 PARENT_ID 를 따로 두는 편입니다.

CEO 김대표
├── 부장 이부장
│   └── 과장 최과장
│       └── 사원 정사원
└── 부장 박부장

여기서 최과장을 검색한다고 생각해보면, 아래와 같이 검색 결과가 나오는 것이 자연스럽습니다. 즉 계층 관계를 모두 표현해야 됩니다. 이럴 때 사용하는 것이 바로 재귀 쿼리입니다.

CEO 김대표
└── 부장 이부장
    └── 과장 최과장
        └── 사원 정사원

문법은 WITH RECURSIVE 를 사용하여서 표현합니다.

WITH RECURSIVE recur_cte AS (
    -- 1. Anchor Member (기준점)
    SELECT 초기값들...,
        1 AS DEPTH -- 시작 깊이는 1
    FROM 테이블
    WHERE 시작조건

    UNION ALL

    -- 2. Recursive Member (재귀 부분)
    SELECT 계산값들...,
        R.DEPTH + 1 -- 기존 깊이에 1 추가
    FROM 테이블 T
    JOIN recur_cte R ON 조인조건
    WHERE 종료조건
)
SELECT * FROM recur_cte;

문법을 보시면 아시겠지만, 호출을 위해 recur_cte 내부 본인을 재귀적으로 호출하고 있습니다.

기준점과 재귀 파트가 있는데, 이 둘을 UNION ALL 연산을 통해 모두 합치는 과정을 거칩니다. 합치는 것이 종료되는 시점은 바로 재귀 파트에서 더이상 새로운 row가 없을 때 종료됩니다.

또한 재귀 깊이는 SQL이 아닌 알고리즘 문제에서도 dfs할 때 depth 형태로 많이 쓰이는 만큼 쿼리를 짤 때도 해당 칼럼은 추가해주는 편이 좋습니다.

멸종위기의 대장균 찾기

[PRG] 멸종위기의 대장균 찾기

문제에서 요구하는 것은 다음과 같습니다.

• 각 대장균들의 세대를 파악해야 됩니다.
• 각 대장균들 중 자식이 없는 대장균을 파악해야 됩니다.
• 각 세대별 자식이 없는 개체의 수(COUNT)와 세대(GENERATION)를 출력합니다.

CTE 생성

각 세대를 마킹하기 위해서는 알고리즘 문제 풀 듯 dfs 진행하면 됩니다. 즉 재귀 쿼리가 적절한 선택입니다.

WITH RECURSIVE CTE AS (
    SELECT ID,
        PARENT_ID,
        1 AS GENERATION
    FROM ECOLI_DATA
    WHERE PARENT_ID IS NULL
    UNION ALL
    SELECT E.ID,
        E.PARENT_ID,
        C.GENERATION + 1
    FROM ECOLI_DATA E
    JOIN CTE C ON (E.PARENT_ID = C.ID)
)

여기서 depth 역할을 하는 것이 바로 GENERATION 칼럼입니다.

쿼리

자식이 없다라는 말은 해당 대장균과 연결된 PARENT_ID 가 없다는 말과 동일합니다.

이를 구현하는 방법은 크게 두 가지가 있습니다.

• CTE 에 원래 테이블(ECOLI_DATA) 을 LEFT JOIN 을 걸어서 ECOLI_DATA 의 ID가 NULL 인 것이 바로 자식이 없는 대장균입니다.

    SELECT 
        COUNT(C.GENERATION) AS COUNT,
        C.GENERATION
    FROM CTE C
    LEFT JOIN ECOLI_DATA E ON (E.PARENT_ID = C.ID)
    WHERE E.ID IS NULL
    GROUP BY GENERATION

• NOT EXISTS 로 ECOLI_DATA 의 PARENT_ID 에 매핑되는 CTE의 ID 를 제외해줍니다.

    SELECT 
        COUNT(GENERATION) AS COUNT,
        GENERATION
    FROM CTE
    WHERE NOT EXISTS (
        SELECT 1 FROM ECOLI_DATA E WHERE E.PARENT_ID = CTE.ID
    )
    GROUP BY GENERATION

해당 상황은 NOT EXISTS 가 더 적합합니다. 해당 주제로 추후 포스팅 예정이지만, NOT EXISTS 의 동작은 중간에 EXISTS 조건을 만족하는 행을 찾으면 즉시 중단하기에 수행 횟수 측면에서 이득이 있습니다.

전체 예시 답안

WITH RECURSIVE CTE AS (
    SELECT ID,
        PARENT_ID,
        1 AS GENERATION
    FROM ECOLI_DATA
    WHERE PARENT_ID IS NULL
    UNION ALL
    SELECT E.ID,
        E.PARENT_ID,
        C.GENERATION + 1
    FROM ECOLI_DATA E
    JOIN CTE C ON (E.PARENT_ID = C.ID)
) 

-- NOT EXIST 활용
SELECT 
    COUNT(GENERATION) AS COUNT,
    GENERATION
FROM CTE
WHERE NOT EXISTS (
    SELECT 1 FROM ECOLI_DATA E WHERE E.PARENT_ID = CTE.ID
)
GROUP BY GENERATION

https://www.acmicpc.net/problem/1626 문제는 아주 간단합니다. MST 가 아닌 SMST(Second Minimum Spanning Tree)를 구하면 됩니다.

개요

우선 희소 배열에 대한 사전 지식이 있어야 하고, #15481 그래프와 MST를 미리 푸는 것이 좋습니다.

여기서는 희소 배열에 대한 것은 안다고 생각하고, 15481번 과 1626번 순으로 풀이하겠습니다.

15481 - 그래프와 MST(P1)

각각의 간선에 대해서 각 간선을 포함하는 MST를 구할 때 그 가중치 합들을 각각 구하면 되는 문제입니다. 정점의 수는 N, 간선의 수는 M이라 가정합니다.

우선 시간 복잡도를 고려하지 않은 풀이를 생각하면 다음과 같습니다.

• 간선 정보를 모두 저장한다.
• 각 간선에 대해서 해당하는 간선이 연결되어 있다고 생각하고 MST를 구한다
• 가중치 합을 구한다.

MST를 크루스칼 알고리즘으로 구한다고 하면 기본적으로 O(MlogM)이 나오고, 각 간선에 대해서 모두 따지면 O(M * MlogM) 이라는 TLE 받기 좋은 시간 복잡도가 나오게 됩니다. 그래서 조금은 다른 방법을 생각해봅니다.

여기서 크루스칼 알고리즘을 M번 돌리는게 문제라는 걸 알 수 있습니다. 그래서 이를 최소한으로, 가능하면 딱 한번만 돌려서 판단해보면 좋을 것 같습니다. 여기서 파생해서 이러한 아이디어를 생각할 수 있습니다.

• 일단 MST를 구한다.
• 각 간선에 대해서
  • 해당 간선이 MST에 포함되면 추가 연산이 불필요하다.
  • 하지만, MST에 포함되지 않으면 추가 연산이 필요하다.

추가 연산이 MST 다시 구하는거면 너무 무겁습니다. 하지만 이렇게 생각해볼 수 있습니다. 결국 해당 간선이 MST에 해당하지 않으면 해당하는 정점이 해당 간선으로는 연결이 되지 않았다는 것입니다. 다른 경로가 MST에 있다는 것입니다. 각 정점을 u, v라 하겠습니다.

여기서 MST는 기본적으로 트리이기 때문에 u, v 사이 경로가 유일함이 보장됩니다. 그렇기에 MST의 u, v 사이 경로 중 최대가 되는 값을 우리가 원하는 간선으로 대치 해주면 원하는 간선을 포함한 MST가 되는 것입니다. 유일한 경로을 끊어주고 새로운 유일한 경로를 만들어주는 느낌입니다.

즉, MST를 그리고 각 정점 사이 간선의 최대값 을 저장해주면 됩니다. 그리고 이를 저장할 때 유용한 자료구조가 바로 희소 배열이고, 최소 공통 조상(LCA) 알고리즘을 통해서 저장, 조회를 진행하면 됩니다.

코드에서는 특별히 해당 간선이 MST인지 아닌지를 별도로 구분하진 않았습니다. 어차피 MST에 포함된다면 본인을 끊었다 다시 본인으로 연결하는 꼴이기 때문입니다.

#include <iostream>
#include <vector>
#include <algorithm>

using namespace std;

typedef long long ll;
typedef pair<int, ll> pil;

struct Edge {
    int u, v; ll w;
    Edge(int _u, int _v, ll _w) : u(_u), v(_v), w(_w) {}
};

const int MAX = 200'001;
int n, m, p[MAX], depth[MAX], pp[20][MAX];
ll wmax[20][MAX];
vector<pil> g[MAX], mst[MAX];
vector<Edge> edges, sorted_edges;

int find_p(int x) { return p[x] == x ? x : p[x] = find_p(p[x]); }
void union_p(int x, int y) { (x > y ? p[x] : p[y]) = (x > y ? y : x); }
bool comp(Edge e1, Edge e2) { return e1.w < e2.w; }

void dfs(int cur, int d) {
    depth[cur] = d;
    for(auto nxt : mst[cur]) {
        if(depth[nxt.first] == 0) {
            pp[0][nxt.first] = cur;
            wmax[0][nxt.first] = nxt.second;
            dfs(nxt.first, d + 1);
        }
    }
}

ll lca(int u, int v) {
    ll res = 0;
    if(depth[u] < depth[v]) swap(u, v);
    for(int i = 19; i >= 0; --i) {
        if(depth[pp[i][u]] >= depth[v]) {
            res = max(res, wmax[i][u]);
            u = pp[i][u];
        }
    }
    if(u == v) return res;
    for(int i = 19; i >= 0; --i) {
        if(pp[i][u] != pp[i][v]) {
            res = max(res, max(wmax[i][u], wmax[i][v]));
            u = pp[i][u];
            v = pp[i][v];
        }
    }
    res = max(res, max(wmax[0][u], wmax[0][v]));
    return res;
}

int main() {
    ios::sync_with_stdio(0); cin.tie(0); cout.tie(0);

    cin >> n >> m;
    int u, v; ll w;
    for(int i = 0; i < m; ++i) {
        cin >> u >> v >> w;
        g[u].emplace_back(v, w);
        g[v].emplace_back(u, w);
        Edge e(u, v, w);
        edges.emplace_back(e);
        sorted_edges.emplace_back(e);
    }

    ll mst_val = 0;
    int cnt = 0;
    sort(sorted_edges.begin(), sorted_edges.end(), comp);
    for(int i = 1; i <= n; ++i) p[i] = i;
    for(auto& e : sorted_edges) {
        int u = find_p(e.u), v = find_p(e.v);
        if (u != v) {
            union_p(u, v);
            mst_val += e.w;
            mst[e.u].emplace_back(e.v, e.w);
            mst[e.v].emplace_back(e.u, e.w);
            if (++cnt == n - 1) break;
        }
    }

    dfs(1, 1);
    for(int i = 1; i < 20; ++i) {
        for(int j = 1; j <= n; ++j) {
            pp[i][j] = pp[i - 1][pp[i - 1][j]];
            wmax[i][j] = max(wmax[i - 1][j], wmax[i - 1][pp[i - 1][j]]);
        }
    }

    for (auto& e : edges) cout << mst_val - lca(e.u, e.v) + e.w << '\n';

    return 0;
}

1626 - 두 번째로 작은 스패닝 트리(D4)

위 문제를 풀었다면, 이 문제 아이디어도 어렵지 않게 생각할 수 있습니다. 다만 몇가지 실수할 여지가 있습니다.

이 문제는 SMST(Second Minimum Spanning Tree)를 구하면 되는 것이고 핵심은 MST보다 크다 에 있습니다. 그렇기에 MST를 만들 수 없거나 아니면 MST 밖에 못만드는 상황에서는 SMST를 만들 수 없습니다.

처음 생각(오답)

위 그래프와 MST 문제처럼 각 간선을 택했을 때 최댓값을 변경하는 방식으로 진행했습니다. 그래서 나온 코드는 아래와 같습니다.

#include <iostream>
#include <vector>
#include <algorithm>

using namespace std;
typedef pair<int, int> pii;
typedef long long ll;

struct Edge {
    int u, v, w;
    Edge(int _u, int _v, int _w) : u(_u), v(_v), w(_w) {}
};

const int MAX = 50'001;

vector<pii> g[MAX], mst[MAX];
vector<Edge> edges;

int n, m;
int par[MAX], dep[MAX], pp[17][MAX], wmax[17][MAX];

int find_p(int x) { return (par[x] == x ? x : par[x] = find_p(par[x])); }
void union_p(int x, int y) { (x > y ? par[x] : par[y]) = (x > y ? y : x); }
bool comp(Edge e1, Edge e2) { return e1.w < e2.w; }

int kruskal() {
    int res = 0, cnt = 0;
    sort(edges.begin(), edges.end(), comp);
    for(int i = 1; i <= n; ++i) par[i] = i;
    for(auto e : edges) {
        int pu = find_p(e.u);
        int pv = find_p(e.v);
        if(pu != pv) {
            union_p(pu, pv);
            res += e.w;
            mst[e.u].emplace_back(e.v, e.w);
            mst[e.v].emplace_back(e.u, e.w);
            if(++cnt == n - 1) return res;
        }
    }
    return -1;
}

void dfs(int cur, int d) {
    dep[cur] = d;
    for(auto nxt : mst[cur]) {
        if(dep[nxt.first] == 0) {
            pp[0][nxt.first] = cur;
            wmax[0][nxt.first] = nxt.second;
            dfs(nxt.first, d + 1);
        }
    }
}

int lca(int u, int v) {
    int res = 0;
    if(dep[u] < dep[v]) swap(u, v);
    for(int i = 16; i >= 0; --i) {
        if(dep[pp[i][u]] >= dep[v]) {
            res = max(res, wmax[i][u]);
            u = pp[i][u];
        }
    }
    if(u == v) return res;
    for(int i = 16; i >= 0; --i) {
        if(pp[i][u] != pp[i][v]) {
            res = max(res, max(wmax[i][u], wmax[i][v]));
            u = pp[i][u];
            v = pp[i][v];
        }
    }
    res = max(res, max(wmax[0][u], wmax[0][v]));
    return res;
}

int main() {
    ios::sync_with_stdio(0); cin.tie(0); cout.tie(0);
    cin >> n >> m;
    int u, v, w;
    for(int i = 0; i < m; ++i) {
        cin >> u >> v >> w;
        g[u].emplace_back(v, w);
        g[v].emplace_back(u, w);
        Edge e(u, v, w);
        edges.emplace_back(e);
    }

    int mst_val = kruskal();
    if(mst_val == -1) {
        cout << mst_val << '\n';
        return 0;
    }

    dfs(1, 1);
    for(int i = 1; i < 17; ++i) {
        for(int j = 1; j <= n; ++j) {
            pp[i][j] = pp[i - 1][pp[i - 1][j]];
            wmax[i][j] = max(wmax[i - 1][j], wmax[i - 1][pp[i - 1][j]]);
        }
    }

    ll smst = 1e18;
    for(auto e : edges) {
        int l = lca(e.u, e.v);
        if(l < e.w) smst = min(smst, ll(mst_val - l + e.w));
    }

    cout << (smst == 1e18 ? -1 : smst) << '\n';
    return 0;
}

최종(정답)

하지만, 위 코드로 제출할 경우 반례가 있습니다. 바로 최댓값에 해당하는 간선의 가중치와 포함하려는 간선의 가중치가 동일한 경우 입니다. 이런 경우는 SMST가 아닌 그냥 새로운 MST를 구하는 꼴입니다.

이를 해결하기 위해서는 바로 최댓값 뿐만 아니라 두번째로 큰 간선도 같이 저장해줍니다. 각각 wmax와 swmax로 저장을 해줍니다.

두번째로 큰 값 구하는게 조금 귀찮은데, 여러 방법이 있지만 다음과 같이 일일이 순회하면서 구현했습니다.

// {최대, 두번째 최대} 구하기
// vector<int> nxt - 기존 최대, 기존 두번째 최대, 새로운 값들 등을 넣는 인자
pii calc(vector<int> nxt) {
    int mm = -1, sm = -1;
    for(int x : nxt) {
        if(x == -1) continue;
        if(mm < x) {
            sm = mm;
            mm = x;
        } else if(sm < x && x < mm) {
            sm = x;
        }
    }
    return { mm, sm };
}

그래서 다음과 같이 희소 배열 초기화할 때나

    dfs(1, 1);
    for(int i = 1; i < 17; ++i) {
        for(int j = 1; j <= n; ++j) {
            pp[i][j] = pp[i - 1][pp[i - 1][j]];
            pii res = calc({ wmax[i - 1][j], swmax[i - 1][j], wmax[i - 1][pp[i - 1][j]], swmax[i - 1][pp[i - 1][j]] });
            wmax[i][j] = res.first;
            swmax[i][j] = res.second;
        }
    }

아니면 lca 구할 때

pii lca(int u, int v) {
    int mm = -1, sm = -1;
    pii res;
    if(dep[u] < dep[v]) swap(u, v);
    for(int i = 16; i >= 0; --i) {
        if(dep[pp[i][u]] >= dep[v]) {
            res = calc({ mm, sm, wmax[i][u], swmax[i][u] });
            mm = res.first;
            sm = res.second;
            u = pp[i][u];
        }
    }
    if(u == v) return { mm, sm };
    for(int i = 16; i >= 0; --i) {
        if(pp[i][u] != pp[i][v]) {
            res = calc({ mm, sm, wmax[i][u], swmax[i][u], wmax[i][v], swmax[i][v] });
            mm = res.first;
            sm = res.second;
            u = pp[i][u];
            v = pp[i][v];
        }
    }
    res = calc({ mm, sm, wmax[0][u], swmax[0][u], wmax[0][v], swmax[0][v] });
    mm = res.first;
    sm = res.second;
    return { mm, sm };
}

이 때 중간중간 넣어주면서 구현해주면 됩니다.

최종 코드

#include <iostream>
#include <vector>
#include <algorithm>

using namespace std;
typedef pair<int, int> pii;
typedef long long ll;

struct Edge {
    int u, v, w;
    Edge(int _u, int _v, int _w) : u(_u), v(_v), w(_w) {}
};

const int MAX = 50001;

bool check[200000];
vector<pii> g[MAX], mst[MAX];
vector<Edge> edges;

int n, m;
int par[MAX], dep[MAX], pp[17][MAX], wmax[17][MAX], swmax[17][MAX];

int find_p(int x) { return (par[x] == x ? x : par[x] = find_p(par[x])); }
void union_p(int x, int y) { (x > y ? par[x] : par[y]) = (x > y ? y : x); }
bool comp(Edge e1, Edge e2) { return e1.w < e2.w; }

int kruskal() {
    int res = 0, cnt = 0;
    sort(edges.begin(), edges.end(), comp);
    for(int i = 1; i <= n; ++i) par[i] = i;
    for(int i = 0; i < m; ++i) {
        Edge e = edges[i];
        int pu = find_p(e.u);
        int pv = find_p(e.v);
        if(pu != pv) {
            union_p(pu, pv);
            res += e.w;
            mst[e.u].emplace_back(e.v, e.w);
            mst[e.v].emplace_back(e.u, e.w);
            check[i] = true;
            if(++cnt == n - 1) return res;
        }
    }
    return -1;
}

void dfs(int cur, int d) {
    dep[cur] = d;
    for(auto nxt : mst[cur]) {
        if(dep[nxt.first] == 0) {
            pp[0][nxt.first] = cur;
            wmax[0][nxt.first] = nxt.second;
            swmax[0][nxt.first] = -1;
            dfs(nxt.first, d + 1);
        }
    }
}

pii calc(vector<int> nxt) {
    int mm = -1, sm = -1;
    for(int x : nxt) {
        if(x == -1) continue;
        if(mm < x) {
            sm = mm;
            mm = x;
        } else if(sm < x && x < mm) {
            sm = x;
        }
    }
    return { mm, sm };
}

pii lca(int u, int v) {
    int mm = -1, sm = -1;
    pii res;
    if(dep[u] < dep[v]) swap(u, v);
    for(int i = 16; i >= 0; --i) {
        if(dep[pp[i][u]] >= dep[v]) {
            res = calc({ mm, sm, wmax[i][u], swmax[i][u] });
            mm = res.first;
            sm = res.second;
            u = pp[i][u];
        }
    }
    if(u == v) return { mm, sm };
    for(int i = 16; i >= 0; --i) {
        if(pp[i][u] != pp[i][v]) {
            res = calc({ mm, sm, wmax[i][u], swmax[i][u], wmax[i][v], swmax[i][v] });
            mm = res.first;
            sm = res.second;
            u = pp[i][u];
            v = pp[i][v];
        }
    }
    res = calc({ mm, sm, wmax[0][u], swmax[0][u], wmax[0][v], swmax[0][v] });
    mm = res.first;
    sm = res.second;
    return { mm, sm };
}

int main() {
    ios::sync_with_stdio(0); cin.tie(0); cout.tie(0);
    cin >> n >> m;
    int u, v, w;
    for(int i = 0; i < m; ++i) {
        cin >> u >> v >> w;
        g[u].emplace_back(v, w);
        g[v].emplace_back(u, w);
        Edge e(u, v, w);
        edges.emplace_back(e);
    }

    int mst_val = kruskal();
    if(mst_val == -1) {
        cout << mst_val << '\n';
        return 0;
    }

    dfs(1, 1);
    for(int i = 1; i < 17; ++i) {
        for(int j = 1; j <= n; ++j) {
            pp[i][j] = pp[i - 1][pp[i - 1][j]];
            pii res = calc({ wmax[i - 1][j], swmax[i - 1][j], wmax[i - 1][pp[i - 1][j]], swmax[i - 1][pp[i - 1][j]] });
            wmax[i][j] = res.first;
            swmax[i][j] = res.second;
        }
    }

    ll smst = 1e18;
    int mm, sm;
    for(int i = 0; i < m; ++i) {
        if(check[i]) continue;
        Edge e = edges[i];
        pii res = lca(e.u, e.v);
        mm = res.first, sm = res.second;
        if(e.w > mm) smst = min(smst, ll(mst_val - mm + e.w));
        else if(e.w > sm && sm != -1) smst = min(smst, ll(mst_val - sm + e.w));
    }

    cout << (smst == 1e18 ? -1 : smst) << '\n';
    return 0;
}

조금씩 풀다보니 다이아5가 되었습니다!

풀다보니 골드가 없어져 있어서 플레티넘 4이상이 아니라면 제자리 걸음...

나름 1일 1알고를 하려 노력을 했습니다. 8월 중에는 SSAFY에서 SWEA에 집중한다고 백준을 잠깐 손놓긴 했었지만, 프로젝트 기간에 밤을 새더라도 문제는 풀자는 주의로 웬만하면 문제를 풀었습니다.

아무래도 집중해서 문제만 풀 수는 없는 입장인지라, 플레티넘에서 다이아까지는 굉장히 오랜시간이 걸렸던 것 같습니다.

흔히 PS계에서 티어 올리기 쉬운 세그먼트 트리 등 구간 쿼리 문제들도 많이 풀긴 했는데, 확실히 세그먼트 트리가 알면 티어에 비해서는 쉬운편이 맞는 것 같습니다. 그래도 금광은 어려웠...

개인적으로 재밌게 푼 알고리즘 분류들은

• Dynamic Programing
• Sparse Table
• Network Flow & MCMF

이 분류들이 재밌었습니다.

수학, 기하학을 풀면 사실 좀 더 빨리 찍었을 것 같기도 한데, 사실 기업 코딩 테스트에는 거의 안나오는 유형이기도 해서(라고 하기엔 코딩 테스트에 나오는 알고리즘은 넘어서 푼 것 같긴 하지만) 배제하고 있었습니다. 하지만 이제 DP도 CHT(Convex Hull Trick) 등 최적화를 요하는 문제가 슬슬 나와서 관련 문제도 슬슬 풀어야겠다고 생각은 듭니다.

무엇보다 다각형이 이쁘게 나오고 싶긴 합니다ㅎㅎ

그리고 문자열을 Trie, Rabin-Karp 몇 문제만 풀어봐서 많이 못풀었는데 문자열도 좀 풀어야 될 것 같네요

지금까지 기억나는 문제들

승급을 시켜준 마지막 문제 : 1396 - 크루스칼의 공(P1)

티어가 가장 높은 문제 : 1626 - 두 번째로 작은 스패닝 트리(D4) / 풀이

가장 많이 틀린 문제 : 10217 - KCM Travel(P4)

얼마나 틀렸냐고요? 사진 한장으로 대체합니다...ㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋㅋ

가장 오래 풀었던 문제 : 5372 - 큐빙(P5)

후기

다이아는 늦어도 올 중순까지는 찍고 싶긴 했는데 찍게 되어서 우선 기쁩니다.

사실 티어를 올리는 건 이제 크게 의미는 없긴 합니다. 대학생도 아니고, 소소한 일반인 대상 대회 정도는 생각 있지만 그런거 외에는 의미가 없기도 합니다.

개발을 하면서 느끼는 건, 알고리즘 역량이 정말 개발에 도움이 되냐 했을 때 느끼는 건 아직까진 반반인 것 같습니다.

물론 아직 실무에서 핵심 비즈니스 로직을 설계해본 경험은 없어서 이 부분은 모르겠지만, 이렇게 문제 풀면서 구현 능력이나 문제 상황에서 바로 코드부터 작성하는 대신 어떻게 접근을 해야되는지 설계하는 능력은 확실히 많이 늘어난 것 같습니다.

이제 그만 풀 것인가? 라고 하기에는 알고리즘 공부 자체가 많이 재밌어졌습니다. 그냥 게임 대신 한번 스트레스 풀겸 종종 풀 것 같습니다. 대신 백준은 조금 쉬엄쉬엄하고 코드포스도 좀 풀어야겠습니다.

들어오기 전

개발의 시작

대학원 자퇴생입니다. 대학원을 나온 이유는 명확합니다. 뭔가 해보고 싶은 업무의 방향성이 달랐기 때문입니다.

기계공학을 전공하면서 수학과 물리학은 좋아했지만 정작 기계 그 자체에는 별 흥미가 없던 것이 큽니다. 대신 산업군의 시스템 관리 및 최적화와 관련된 업무를 하고 싶다는 생각을 많이 했습니다.

이를 구현하기 위해서 가장 필요한 것이 백엔드 개발이라 생각하였고, 결국은 SSAFY에 지원하게 되었습니다.

한번의 고배

사실 전 SSAFY 11기에 지원했었고, 지원이야기가 나올 시점에 Hello World 한번 쳐봤습니다. 컴퓨터공학과는 사실 큰 접점이 없었습니다. 그리고 시험도 잘 못쳐서 서류 탈락이라는 고배를 마시게 되었습니다.

그래서 SSAFY 탈락 후 뭐라도 해야겠다는 심정으로 국비학원을 다녔고, 학원을 다니면서도 SSAFY를 가고 싶다 생각하였습니다.

그 결과 SSAFY 12기 비전공자로 합격하였습니다!

더불어 입과시험으로 전공 Java 반에 입과하였고, 본격적으로 SSAFY 생활이 시작되었습니다.

1학기

전공자들과의 만남

국비학원 다니면서 완전 기초부터 시작해 알고리즘을 계속 풀어와서 백준 티어가 플레티넘 4인 상태에서 들어왔습니다. 전공자들은 다이아, 루비들이 많겠지라는 막연한 기대감이 있던 것 같습니다.

알고리즘은 의외로 제가 가장 잘했습니다. 다만 문제가 된 부분은 바로 그냥 일상적인 대화였습니다.

전공자들이 개발과 관련된 용어들을 그냥 자연스럽게 이야기하는데 전 이해한 척만 하고 뭔소리지...?를 맘속으로 많이 생각했던 것 같습니다.

그래서 스터디를 좀 진행했었습니다. 싸피 내 스터디를 3개 다니면서 부족한 부분을 채우려고 노력 많이했던 것 같습니다.

그리고 전공자 반이 좋았던게, 인사이트를 얻을 수 있는 사람들이 많았습니다. 사람이 많다는게 싸피 최고의 장점이라 생각하는데, 개발 인맥도 쌓고 취업 정보 및 개발 관련 지식을 쌓는데 좋은 환경이였다 생각합니다.

이렇게 해서 1학기를 성적 우수(2등) 으로 마치게 됩니다.

B형

싸피의 또다른 혜택으로는 삼성 SW 역량테스트 B형 응시 기회가 주어진다는 것입니다. 알고리즘에 어느정도 자신 있는 저로써 B형도 쉽게 취득하리라 생각했던 것 같습니다. 하지만 그렇지 않았습니다.

기본적으로 구현력이 발목을 잡았습니다. 삼성에서 내는 구현 스타일의 문제가 있는데, 이게 익숙하지 않았던 탓이 큽니다. 또한 너무 수준을 얕보고 이정도 시간이면 되겠지... 하고 넘겼던 것도 큽니다.

사실 1학기의 목표는 취업 아니면 B형이였는데, 둘 다 이루지 못하고 2학기에 진학하게 되었습니다.

2학기

공통

임베디드를 맡게 되었고, Java는 MQTT 설정하는 부분을 제외하고는 거의 C, python, shell script만 주구장창 만지게 되었습니다.

공통을 생각하면 아쉬운 부분이 많습니다. 기획을 좀 더 빨리하면 어땠을까, 처음부터 임베 전공자가 있었으면 어땠을까, 시연 실수가 없었으면 어땠을까 등 다양한 아쉬운 부분들이 있습니다.

다만 얻어간 부분이 몇가지 있습니다.

• 스마트팩토리에서 필요한 IoT와 MQTT 에 대해서는 어느 정도 설명을 할 수 있다.
• 다양한 개발 경험으로 자소서에 어필할 만한 것이 생긴다.

그리고 임베디드 개발도 재미있기도 했습니다. 수상은 못했고 아쉬운 부분도 많지만, 분명히 얻어갈 점도 있긴 했습니다.

특화

특화 기간은 팀장으로 활동했습니다. 그리고 이 기간을 저는 이렇게 표현하고 싶습니다.

SSAFY 에서의 가장 좋았던 순간, 그리고 나가고 싶었던 순간이 공존하는 상태

정말 가장 기뻤던 일과 가장 힘들었던 순간이 동시에 왔습니다.

1학기 저는 B형을 따지 못했다는 아쉬움에 바로 뜻이 맞는 사람들과 B형 스터디를 결성하였습니다.

그리고 공통에서 특화가 넘어가는 순간 2025년 1회차 B형을 응시하였고, 운이 좋게도 1시간만에 답을 내고 B형을 합격하였습니다.

하지만 동시에 이런 생각이 들었습니다.

B형도 땄네...? 난 B형만 보고 왔는데...? 이제 뭐하지...?

그리고 한동안 Java를 안보고 있었어서 백엔드에 대한 감이 떨어진 상태였고, Spring 거의 처음 본 시점 같은 기분을 느꼈습니다.

계속 앞만 보고 달려와서 번아웃이 와버렸습니다. 물론 팀장이니 팀장으로써 할 일들에 대해서는 해왔지만, 정작 개발에 대한 방향성을 상실했습니다.

하지만 팀원 운이 너무 좋았습니다. 팀원들이 긍정에너지가 가득해서 많이 힘든 시기였는데도 도움을 많이 주었습니다. 그리고 나온 결과도 좋아서 입상도 하게 되었습니다! 그리고 자율 때는 특화 때 개발적으로 아쉬웠던 부분들 보완하고자 다짐하였습니다.

자율, 그리고 싸탈

자율 팀원들도 좋은 팀원들과 함께 하게 되었는데, 별로 개발을 진행하지 않은 시점에서 지원했던 기업에 최종합격 소식을 받게 되었습니다.

2025년 상반기를 생각해보면 정말 채용 시장이 얼어있었습니다. 2024년 하반기 때도 역대급 한파다 이런 이야기가 많았는데, 이번에는 더더욱 그랬습니다.

작년 서류합격해서 면접까지 간 기업들이 있는데, 이번에는 서류에서 바로 탈락하였습니다. 당연히 서류는 합격할 줄 알았는데, 바로 떨어지는걸 보고 정말 이번 채용시장 많이 힘들구나 생각했습니다.

그런데, 코딩 테스트야 평소에 자신 있으니 논외로 하고, 다른 사람들이 다 떨어졌다는데 제가 딱 하나 붙고, 1차 면접도 긴장 안했고, 2차 면접은 아쉬웠는데 운이 좋게 최종 합격까지 가버렸습니다...!!

관심있던 기업에 합격을 해서 싸탈을 하게 되었습니다.

성과

• 1학기 성적우수(2위)
• 배틀싸피 전국대항전 준우승
• 2학기 Best Member 2회(공통, 특화)
• 특화 프로젝트 입상(3위)
• 삼성 SW 역량테스트 B형(우수 풀이 선정)
• 우수 스터디 선정 다수
• 취업 싸탈
• 작성 시점 BOJ Platinum 1

개인적으로 싸피 수료전 다이아는 달고 싶었는데 이건 아쉽긴 합니다. - 결국 달았네요 그래도 취업이 더 좋으니까..ㅎ

결론

싸피를 다니면서 크게 느낀 장점은 다음과 같습니다.

• 100만원을 포함한 지원
• 삼성 SW 역량테스트 B형
• 사람이 많고, 그 중 좋은 사람들이 많아 인맥 쌓기 좋습니다

다니면서 다른 것도 다른건데, 정말 사람이 중요하다 생각합니다. 싸피에서 좋은 사람들을 많이 만나 좋은 성과를 많이 낸 것 같습니다.

크롤링을 로컬 PC(Macbook)으로 돌려서 데이터 수집을 진행했었습니다. 다만 저장한 데이터가 로컬 docker로 띄운 MongoDB 안에 있었는데, 이를 EC2 내부 docker로 띄운 개발용 MongoDB와 배포용 MongoDB 모두에 반영해야 되었습니다. Dump 파일을 옮기는 Docker 명령어 및 MongoDB 관련 사용법을 메모하고자 작성하게 되었습니다.

1. 로컬 Docker 컨테이너에서 MongoDB 덤프 생성

로컬에서 실행 중인 Docker MongoDB 컨테이너에서 데이터를 덤프합니다. 예를 들어 컨테이너 이름이 <컨테이너_이름>, 데이터베이스 이름이 <DB_이름>, 계정 정보가 다음과 같다고 가정합니다.

• 사용자명: <USERNAME>
• 비밀번호: <PASSWORD>
• 인증 DB: admin

docker exec <컨테이너_이름> mongodump \
  -u <USERNAME> -p <PASSWORD> \
  --authenticationDatabase admin \
  --db <DB_이름> \
  --out /dump

이 명령은 컨테이너 내부의 /dump 디렉토리에 <DB_이름> DB를 .bson 파일로 덤프합니다.

2. 컨테이너 내부 덤프 파일을 로컬로 복사

컨테이너에서 로컬 PC로 덤프 파일을 복사합니다.

docker cp <컨테이너_이름>:/dump ./<로컬_저장_폴더>

복사 후 ./<로컬_저장_폴더>/<DB_이름> 경로에 .bson 파일이 생성되어 있어야 합니다.

3. EC2로 덤프 파일 전송

.pem 키 파일의 권한이 너무 널널하면 SSH 클라이언트에서 접속을 차단하므로 권한을 제한합니다.

chmod 400 <키_파일_경로>.pem

권한 설정 후, scp 명령어를 사용해 덤프 폴더를 EC2로 전송합니다.

scp -i <키_파일_경로>.pem -r ./<로컬_저장_폴더> ubuntu@<EC2_IP>:/home/ubuntu/

4. EC2에서 MongoDB 컨테이너로 덤프 복사

EC2에서 SSH 접속 후, 덤프 폴더를 MongoDB 컨테이너로 복사합니다.

docker cp ~/ <로컬_저장_폴더> <컨테이너_이름>:/<컨테이너_경로>

예시:

docker cp ~/dump mongo-dev:/dump
docker cp ~/dump mongo-deploy:/dump

5. MongoDB 덤프 복원 실행

개발용 컨테이너 복원

docker exec -it mongo-dev mongorestore \
  -u <USERNAME> -p <PASSWORD> \
  --authenticationDatabase admin \
  --db <DB_이름> /dump/<DB_이름>

배포용 컨테이너 복원

docker exec -it mongo-deploy mongorestore \
  -u <USERNAME> -p <PASSWORD> \
  --authenticationDatabase admin \
  --db <DB_이름> /dump/<DB_이름>

6. 복원 확인

복원이 완료되었는지 확인하기 위해 MongoDB에 접속합니다.

docker exec -it mongo-dev mongosh -u <USERNAME> -p <PASSWORD> --authenticationDatabase admin

MongoDB 셸 내에서 다음 명령을 실행합니다.

use <DB_이름>
show collections

update 할 때 Dirty Checking 방식을 사용하는 것이 유지보수에 좋아요!

이를 정리하기 위해 블로그 포스팅을 남기게 되었습니다.

1. 기존의 코드

초기에 JPA의 @Modifying과 @Query를 사용하여 Bulk Update 방식으로 데이터를 업데이트하는 코드를 작성했습니다.

기존 코드

서비스 코드

@Service
@RequiredArgsConstructor
@Transactional
public class TownService {

    private final TownRepository townRepository;

    public void updateTownName(User user, TownNameUpdateDto townNameUpdateDto) {
        int updatedCount = townRepository.updateTownName(user.getId(), townNameUpdateDto.townName());
        if(updatedCount == 0) {
            throw new InvalidArgumentException("해당하는 마을을 찾을 수 없습니다.");
        }
    }
}

레포지토리 코드

public interface TownRepository extends JpaRepository<Town, Long> {
    @Modifying(clearAutomatically = true)
    @Query("UPDATE Town t SET t.name = :townName WHERE t.id = :townId")
    int updateTownName(Long townId, String townName);
}

기존 코드의 문제점

@Modifying을 사용하여 직접 SQL UPDATE 쿼리를 실행하기 때문에 영속성 컨텍스트와 무관하게 DB가 변경됩니다. 따라서 1차 캐시가 갱신되지 않아서, 이후 동일한 엔티티를 조회할 경우 데이터 불일치 문제가 발생할 수 있습니다.

또한, 코드 가독성이 떨어지고 유지보수성이 낮습니다. 그리고 트랜잭션 롤백이 발생해도 이미 실행된 쿼리는 취소되지 않습니다.

2. Dirty Checking

위 문제를 해결하기 위해 Dirty Checking 방식으로 변경했습니다. JPA의 영속성 컨텍스트를 활용하여 엔티티의 필드 값이 변경되면 트랜잭션이 종료될 때 자동으로 업데이트 쿼리가 실행됩니다.

변경된 코드

서비스 코드

@Service
@RequiredArgsConstructor
@Transactional
public class TownService {

    private final TownRepository townRepository;

    public void updateTownName(User user, TownNameUpdateRequest townNameUpdateRequest) {
        user.getTown().updateTownName(townNameUpdateRequest.townName());
    }
}

• townRepository.updateTownName()을 호출하여 직접 쿼리를 실행했던 방식에서 user.getTown().updateTownName()으로 변경되었습니다.
• 기존에는 @Modifying을 사용하여 즉시 쿼리를 실행했지만, Dirty Checking 방식에서는 엔티티의 상태를 변경하는 것만으로도 변경 사항이 자동 반영됩니다.

레포지토리 코드

public interface TownRepository extends JpaRepository<Town, Long> {
}

• updateTownName()과 같은 명시적인 @Query 메서드가 필요하지 않습니다.
• JPA는 영속성 컨텍스트 내에서 엔티티가 변경되었음을 감지하고 자동으로 UPDATE 쿼리를 실행하기 때문에 직접 명시하지 않아도 됩니다.

엔티티 코드

@Entity
public class Town {

    public void updateTownName(String townName) {
        this.name = townName;
    }
}

• townName 필드의 값을 변경하면 JPA는 이를 감지하고, 트랜잭션이 종료될 때 해당 엔티티의 변경 사항을 반영하는 UPDATE 쿼리를 실행합니다.

변경된 코드의 특징

엔티티를 조회한 후 필드 값을 변경하면, 트랜잭션 종료 시점에 자동으로 UPDATE 쿼리가 실행됩니다. 여기서 영속성 컨텍스트가 유지되므로 데이터 불일치 문제가 발생하지 않습니다.

그리고 트랜잭션이 롤백될 경우, 변경 사항도 함께 롤백됩니다. 마지막으로 코드가 간결해지고 유지보수가 쉬워집니다.

JPA의 Dirty Checking 동작 원리

JPA는 영속성 컨텍스트에서 엔티티를 관리하며, 엔티티의 스냅샷(초기 상태)을 저장합니다.

이후 트랜잭션이 종료될 때 스냅샷과 현재 엔티티의 상태를 비교하여 변경된 경우 UPDATE 쿼리를 자동 실행합니다. 따라서 별도의 save() 호출 없이도 변경 사항이 자동 반영됩니다.

3. 각각의 장단점

지금까지 보았을 때 Dirty Checking 방식이 무조건 좋아보이지만, 각각의 장단점이 존재합니다.

Bulk Update 방식

특징

• @Modifying을 활용하여 직접 SQL 실행
• 즉시 반영되지만 영속성 컨텍스트와는 무관

장점

• 대량 데이터 업데이트 시 성능이 뛰어남
• 한 번의 SQL 실행으로 여러 행을 업데이트 가능

단점

• 영속성 컨텍스트 미반영 문제
• 트랜잭션 롤백이 되지 않음
• 코드가 복잡해지고 유지보수가 어려워짐

Dirty Checking 방식

특징

• 엔티티를 조회한 후 필드 값을 변경하면 트랜잭션 종료 시점에 자동 반영
• 영속성 컨텍스트를 활용하여 변경 사항을 감지

장점

• 데이터 일관성이 보장(영속성 컨텍스트 활용)
• 트랜잭션 롤백이 쉬움
• 코드 유지보수성이 높아짐

단점

• 대량 업데이트 시 개별 쿼리가 발생하여 성능이 저하될 수 있음
• 엔티티 조회 후 변경하는 과정이 필요

4. 결론

Bulk Update 방식은 대량 데이터를 빠르게 업데이트할 때 유리하지만, 영속성 컨텍스트를 무시하기 때문에 데이터 일관성이 깨질 위험이 있습니다. 반면 Dirty Checking 방식은 JPA의 변경 감지를 활용하여 코드 유지보수성이 높고, 트랜잭션을 활용할 수 있어 더 안정적입니다.

어떤거 선택해야 되나요?

• 대량 업데이트가 필요한 경우: Bulk Update 방식이 유리
• 트랜잭션 내에서 엔티티를 변경하는 경우: Dirty Checking 방식이 적합

색깔 트리라는 문제를 풀어보며 다음과 같은 의문이 들었습니다.

만약 트리의 최대 깊이가 100이라서 단순 dfs로도 구현이 되는데, 만일 LCA 때처럼 10만이란 단위가 나온다면...?

여러가지 조건을 여기에 적용 가능하지만, 편의를 위해 다음과 같은 조건을 추가해보겠습니다.

대신 트리의 구조 자체는 변하지 않고 부모 노트 색 칠할 때 자식 노드도 모두 칠해지는 조건만 가져와서 적용해보자

Segment Tree with Lazy Propagation 을 적용하기에 딱 좋아보입니다. 그리고 마침 백준에 이에 해당하는 문제가 있습니다. 구조는 안바뀌는데, 뭔가 누적되는 것은 똑같아 보입니다. 하지만, O(depth) 로 처리 시에는 분명히 문제가 됩니다. 뭔가 O(1) 아니면 O(log N) 으로 접근하면 됩니다.

바로 세그먼트 트리로 관리 해주되, 해당되는 트리의 구간을 Lazy Propagation 을 활용해서 O(log N) 으로 진행하면 됩니다. 그런데 문제가 있습니다.

우리가 세그먼트 트리 만들 때는 분명히 완전 이진 트리여서 1차원 배열로 작성이 가능했습니다. 근데 저건 그런게 아니라 완전히 일반화된 트리입니다. 일반화된 트리를 저렇게 1차원 배열 형식으로 가능…은 할까요…?

네 가능합니다. 대신 특별한 기술이 들어갑니다. 바로 Euler Tour Technique(ETT) 입니다.

Euler Tour Technique(ETT)

ETT는 트리를 1차원 배열로 변환하는 기법으로, 트리에서 특정 노드의 서브트리를 연속된 구간으로 변환하여 세그먼트 트리와 같은 자료구조를 활용할 수 있도록 합니다.

트리는 일반적으로 계층적 구조를 가지고 있어서 1차원 배열을 기반으로 동작하는 세그먼트 트리나 펜윅 트리를 직접 적용하기 어렵습니다.

하지만 dfs를 수행하며 트리를 펼쳐서 1차원 배열로 표현하면 O(log N)의 시간 복잡도로 서브트리 연산이 가능해집니다.

ETT 과정

ETT의 과정은 다음과 같습니다.

1. dfs를 수행하며 정점을 방문하는 순서를 기록합니다.
2. 각 노드의 "들어온 시간 (In-Time)"과 "나가는 시간 (Out-Time)"을 저장합니다.
3. In-Time과 Out-Time을 이용해 서브트리를 하나의 연속된 구간으로 표현합니다.

뭔가 복잡해 보이지만, 사실은 단순한 개념입니다. 트리를 dfs로 순회하며 각 노드가 언제 탐색을 시작하고 끝나는지만 기록하면 됩니다.

이 방식은 dfs numbering 이라고도 부릅니다. 코드로도 아주 간단합니다.

int cnt = 0;
void dfs(int node) {
    in[node] = ++cnt; // 시작 번호 마킹
    // 탐색 부분
    out[node] = cnt;
}

아래는 트리에 대해서 dfs numbering을 진행한 결과입니다.

표로 나타내면 다음과 같습니다.

이렇게 만들면 트리를 1차원 배열로 만들 준비가 끝났습니다. 적용은 다음과 같이 하면 됩니다.

이렇게 구간으로 관리할 수 있습니다. 이제는 Lazy Propagation 활용해서 구간 관리하면 됩니다.

[BOJ] 16404 / 주식회사 승범이네

위 과정으로 얻은 서브트리 구간을 Segment tree with Lazy Propagation 을 통해 구간 갱신, 현재 노드 값은 시작 값을 가져와서 그대로 Segment tree에서 가져오면 됩니다.

#include <iostream>
#include <vector>

using namespace std;
typedef long long ll;

const int MAX = 100'001;

int n;
int num = 0;
int depth[MAX], s[MAX], e[MAX];
ll lazy[4 * MAX], tree[4 * MAX];
vector<int> g[MAX];

void dfs(int cur, int d) {
    depth[cur] = d;
    s[cur] = ++num;
    for (int nxt : g[cur]) {
        if (depth[nxt] == 0) {
            dfs(nxt, d + 1);
        }
    }
    e[cur] = num;
}

void updateLazy(int node, int s, int e) {
    if (lazy[node] == 0) return;
    tree[node] += (e - s + 1) * lazy[node];
    if (s != e) {
        lazy[node << 1] += lazy[node];
        lazy[node << 1 | 1] += lazy[node];
    }
    lazy[node] = 0;
}

void updateTree(int node, int s, int e, int ts, int te, int val) {
    updateLazy(node, s, e);
    if (te < s || e < ts) return;
    if (ts <= s && e <= te) {
        tree[node] += val;
        if (s != e) {
            lazy[node << 1] += val;
            lazy[node << 1 | 1] += val;
        }
        return;
    }

    int m = (s + e) >> 1;
    updateTree(node << 1, s, m, ts, te, val);
    updateTree(node << 1 | 1, m + 1, e, ts, te, val);
    tree[node] = tree[node << 1] + tree[node << 1 | 1];
}

ll get(int node, int s, int e, int ts, int te) {
    updateLazy(node, s, e);
    if (te < s || e < ts) return 0;
    if (ts <= s && e <= te) return tree[node];

    int m = (s + e) >> 1;
    return get(node << 1, s, m, ts, te) + get(node << 1 | 1, m + 1, e, ts, te);
}

void update(int i, int w) {
    updateTree(1, 1, n, s[i], e[i], w);
}

ll balance(int i) {
    return get(1, 1, n, s[i], s[i]);
}

int main() {
    ios::sync_with_stdio(0); cin.tie(0); cout.tie(0);
    int m, p;
    cin >> n >> m;
    for (int i = 1; i <= n; ++i) {
        cin >> p;
        if (p == -1) continue;
        g[p].emplace_back(i);
    }

    dfs(1, 1);
    fill(lazy, lazy + (4 * n + 1), 0);
    fill(tree, tree + (4 * n + 1), 0);

    int q, i, w;
    while (m--) {
        cin >> q;
        if (q == 1) {
            cin >> i >> w;
            update(i, w);
        }
        else {
            cin >> i;
            cout << balance(i) << '\n';
        }
    }

    return 0;
}

백엔드로 센서 데이터를 보내서 처리를 할 필요가 있어 Spring Boot로 보내는 작업에 대해 작성합니다.

의존성

plugins {
    java
    id("org.springframework.boot") version "3.4.2"
    id("io.spring.dependency-management") version "1.1.7"
}

group = "com.example"
version = "0.0.1-SNAPSHOT"

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(17)
    }
}

repositories {
    mavenCentral()
}

dependencies {
    implementation("org.springframework.boot:spring-boot-starter-web")
    testImplementation("org.springframework.boot:spring-boot-starter-test")
    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
    implementation("org.springframework.boot:spring-boot-starter-integration")
    implementation("org.springframework.integration:spring-integration-mqtt")
}

tasks.withType<Test> {
    useJUnitPlatform()
}

build.gradle 입니다. 모바일 환경을 고려하고 있어 우선 Kotlin 형식으로 받기는 했지만, 다른 방식도 크게 다르지는 않습니다. 다만, 기존 Spring Web 이외에 다른 의존성이 몇가지 필요합니다.

• spring-boot-starter-integration : HTTP, TCP 등과 같은 전송 로직을 추상화하는 역할을 합니다.
• spring-integration-mqtt : 이름 그대로 MQTT를 사용할 수 있습니다.

패키지 구조

실제 프로젝트를 진행하면 이것보다 훨씬 복잡한 구조를 가지겠지만, 현재는 메세지 테스트만을 진행하므로 간단하게 구성하였습니다.

src
 ├── main
 │   ├── java/com/example/demo
 │   │   ├── config
 │   │   │   ├── MqttConfig.java  <-- MQTT 설정 파일
 │   │   ├── handler
 │   │   │   ├── MqttMessageHandler.java  <-- MQTT 메시지 처리 핸들러
 │   │   ├── DemoApplication.java  <-- Spring Boot 메인 애플리케이션
 │   ├── resources
 │   │   ├── application.properties  <-- MQTT 브로커 설정

MqttConfig

• 현재 서버로 데이터가 들어오는 것을 확인하는 것이기에 inboundChannel로 구성
• 공식 문서 등에 여러 구현체들이 나와 있어 해당 문서를 참고하면서 구현하는 것이 좋습니다.

package com.example.demo.config;

import com.example.demo.handler.MqttMessageSubscriber;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Configuration
public class MqttConfig {

    @Value("${mqtt.broker.url}")
    private String brokerUrl;

    @Value("${mqtt.client.id}")
    private String clientId;

    @Value("${mqtt.topic}")
    private String topic;

    @Value("${mqtt.username}")
    private String username;

    @Value("${mqtt.password}")
    private String password;

    @Bean
    public MqttPahoClientFactory mqttClientFactory() { // MQTT 클라이언트 관련 설정
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setAutomaticReconnect(true);
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public MessageProducer inboundChannel(
            MqttPahoClientFactory mqttClientFactory,
            MessageChannel mqttInputChannel
    ) { // inboundChannel 어댑터
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                brokerUrl,
                clientId,
                mqttClientFactory,
                topic
        );
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel);
        return adapter;
    }

    @Bean
    public MessageChannel mqttInputChannel() { // MQTT 구독 채널 생성
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel") // MQTT 구독 핸들러
    public MessageHandler inboundMessageHandler() {
        return new MqttMessageSubscriber();
    }
}

MqttMessageSubscriber

• 메세지 핸들러를 받아 단순히 받은 메세지를 출력

package com.example.demo.handler;

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Component;

@Component
public class MqttMessageSubscriber implements MessageHandler {

    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        System.out.println("Received MQTT message: " + message.getPayload());
    }
}

application.properties

mqtt.broker.url = tcp://[브로커 IP 주소]:1883
mqtt.client.id = test-client
mqtt.topic = floor/room/temp
mqtt.username = [유저 이름]
mqtt.password = [비밀 번호]

Test

우선 Raspberry PI에 mosquitto 설정을 마쳤으면 다음과 같이 실행해줍니다. Raspberry PI가 브로커의 역할을 합니다.

Raspberry PI

브로커 설정

cd /etc/mosquitto # 경로 변경
sudo /etc/init.d/mosquitto stop # 실행 중인 브로커 중지
sudo mosquitto -c mosquitto.conf -v # 설정 파일에 맞게 브로커 실행

Publisher

mosquitto_pub -h [브로커의 IP 주소] -t floor/room/temp -m "Hello Spring Boot!"

Spring Boot

console

Received MQTT message: Hello Spring Boot!

참고

• https://godekdls.github.io/Spring%20Boot/spring-integration/
• https://docs.spring.io/spring-integration/reference/mqtt.html
• https://junhyunny.github.io/spring-boot/integration/mqtt-subscriber-spring-boot/

MQTT

• MQTT(Message Queuing Telemetry Transport)는 발행-구독(Publish-Subscribe) 기반의 메시지 송수신 프로토콜
• IoT와 같이 제한된 또는 대규모 트래픽 전송을 위해 만들어진 프로토콜

구성요소

• 발행자(Publisher) - 메시지를 발행(메시지를 브로커로 전송)하는 역할을 담당하는 구성요소
• 구독자(Subscriber) - 메시지를 구독(브로커로부터 메시지를 수신)하는 역할을 담당하는 구성요소
• 브로커 - 발행자가 발행한 메시지를 수신하여 구독자에게 전달하는 중계장 역할을 담당하는 구성요소

특징

1. 연결 지향적 (Connection Oriented)

   • TCP/IP 소켓 연결 이휴 연결이 끊어질 때까지 상태 유지
   • Live라는 하트비트와 Topic에 발행되는 메시지를 통해 연결을 유지하고 메시지 송수신을 하게 됨
   • 연결이 끊어지면 재접속 가능

2. 브로커를 통한 통신

   • 오로지 브로커를 통해서만 통신이 가능
   • 개설된 Topic에 메시지를 발행 시 해당 Topic를 구독하는 클라이언트들에게 메시지를 발행
   • 일대일, 일대다 통신 모두 가능

3. QoS(Quality of Service)

   • 0 : 보낸 다음 잊어버림

     • 최대 1회 전송.
     • Topic을 통해 메시지를 전송할 뿐 보장은 하지 않음.

   • 1 : 확인 응답을 거치는 전달

     • 최소 1회 전송.
     • 구독하는 클라이언트가 메시지를 받았는지 불확실하면 정해진 횟수만큼 재전송.
     • 메시지의 핸드셰이킹 과정을 엄밀하게 추적하지는 않으므로 중복의 위험성 존재.

   • 2 : 보장된 전달

     • 구독하는 클라이언트가 요구된 메시지를 정확히 한 번 수신할 수 있도록 보장
     • 메시지의 핸드셰이킹 과정을 추적
     • 높은 품질을 보장, but 성능 요구

   • TCP/IP 데이터 전송의 처리에 영향을 주지 않으며, MQTT 송신자와 수신자 간에만 사용

   • 글자 수 제한이 없으므로, 긴 메시지나 JSON 포맷 또는 파일 전송 가능

   • 0에 가까울수록 메시지 처리에 대한 부하가 줄어들고, 메시지 손실의 위험이 높음.

4. 메시지 유형

   • 연결하기 - 서버와의 연결 수립을 기다린 다음 노드 간 링크
   • 연결 끊기 - MQTT 클라이언트가 해야 할 일을 기다리고 인터넷 프로토콜 스위트 세션의 연결이 끊어지기를 기다림
   • 발행하기 - MQTT 클라이언트에 요청이 전달된 직후 어플리케이션 스레드에 즉시 반환
   • 각각의 메시지의 event에 따라 MQTT 브로커가 알림을 주어 대응

5. 다양한 개발언어의 다양한 클라이언트 지원

mosquitto

mosquitto

• 대표적인 오픈소스 MQTT 브로커 중 하나

• Raspberry Pi에 mosquitto 설치

    sudo apt install mosquitto mosquitto-clients

• 브로커 동작 확인

    sudo systemctl status mosquitto

  결과 예시 :

    ● mosquitto.service - Mosquitto MQTT Broker
         Loaded: loaded (/lib/systemd/system/mosquitto.service; enabled; preset: enabled)
         Active: active (running) since Thu 2025-01-23 19:41:01 GMT; 12s ago
           Docs: man:mosquitto.conf(5)
                 man:mosquitto(8)
        Process: 724 ExecStartPre=/bin/mkdir -m 740 -p /var/log/mosquitto (code=exited, status=0/SUCCESS)
        Process: 731 ExecStartPre=/bin/chown mosquitto /var/log/mosquitto (code=exited, status=0/SUCCESS)
        Process: 733 ExecStartPre=/bin/mkdir -m 740 -p /run/mosquitto (code=exited, status=0/SUCCESS)
        Process: 735 ExecStartPre=/bin/chown mosquitto /run/mosquitto (code=exited, status=0/SUCCESS)
       Main PID: 736 (mosquitto)
          Tasks: 1 (limit: 4443)
            CPU: 17ms
         CGroup: /system.slice/mosquitto.service
                 └─736 /usr/sbin/mosquitto -c /etc/mosquitto/mosquitto.conf
  
    Jan 23 19:41:01 raspberrypi systemd[1]: Starting mosquitto.service - Mosquitto MQTT Broker...
    Jan 23 19:41:01 raspberrypi systemd[1]: Started mosquitto.service - Mosquitto MQTT Broker.

브로커

/etc/mosquitto 경로에 mosquitto 관련 설정 파일이 있습니다. 브로커를 실행하기 위해서는 해당 경로로 들어가서 해당 설정 파일에 있는 내용에 맞게 실행해야 되므로 경로를 바꾸고 브로커를 실행합니다.

트러블 슈팅

단 그냥 접속 시, Error: Connection refused 를 볼 수있습니다. 따라서 다음과 같이 수정을 해줍니다.

1. sudo vi /etc/mosquitto/mosquitto.conf 를 통해 설정 파일을 vi editor로 열어줍니다.

2. 다음 내용을 해당 파일 제일 마지막에 추가해줍니다.

    port 1883
    allow_anonymous true

브로커 실행

cd /etc/mosquitto # 경로 변경
sudo /etc/init.d/mosquitto stop # 실행 중인 브로커 중지
sudo mosquitto -c mosquitto.conf -v # 설정 파일에 맞게 브로커 실행

이렇게 실행하면 The 'port' option is now deprecated and will be removed in a future 라고 해서 port 관련 설정을 준 것이 deprecated 되었다고 경고가 뜨는데, 지금은 크게 신경쓸 필요는 없습니다.

mosquitto_sub

mosquitto_sub -h [MQTT 브로커 접속주소(IP)] -p [통신포트 : 기본값 1883] -t [구독할 토픽 : "#"는 모든 토픽을 구독]

참고로 [MQTT 브로커 접속주소(IP)] 는 라즈베리파이에서 ifconfig 로 확인할 수 있는 그 IP로 접속하면 됩니다.

mosquitto_pub

mosquitto_pub -h [MQTT 브로커 접속주소(IP)] -p [통신포트 : 기본값 1883] -t [발행할 메시지의 토픽] -m [발행할 메시지]

Test

• Raspberry Pi 터미널을 열어서 실행

mosquitto_sub -h [MQTT 브로커 접속주소(IP)] -p 1883 -t "#" 
# #은 topic에 해당하는 하위 항목 모두 다 포함
# mosquitto_pub 입력마다 Hello 출력

mosquitto_pub -h [MQTT 브로커 접속주소(IP)] -p 1883 -t "test" -m "Hello"

참고

https://underflow101.tistory.com/22?category=826163

1학기를 마치며, 느낀 사실이 있습니다.

개발자라는 직업이라는 것을 생각하기 이전에는 취업이라는 선택지가 저에게 있어서는 전혀 존재하지 않았습니다. 학계에 남고 싶어 대학원 준비를 하였으니 말입니다. 학점은 나름 높았지만, 취업 준비와는 한 활동들이 거리가 멀었습니다.

하지만, 뜻하지 않게 대학원 학위를 얻지 못하였고, 여러가지 분야들을 공부하다 개발 분야에 뜻하지 않게 개발자 진로를 희망하게 되었습니다.

어쩌다보니 본의아니게 취업과 관련된 첫 면접이 SSAFY가 되었고, 운좋게 합격하여 비전공자인데 분반 테스트 결과도 잘 나오게 되어 전공반에 가게되었습니다.

또 거기서도 운좋게 성적우수(2등)을 하게 되었고, 배틀싸피라는 새로 생긴 SSAFY 내 게이미피케이션 프로그램에서도 전국대항전 준우승이란 성과를 달성하게 되었습니다.

이런 것을 보면 새옹지마라는 표현이 딱 떠오릅니다.

그 결과...

이런 상품도 얻었습니다...ㅎㅎ

About 상품

상품에 대해 저건 뭐야? 하실까봐 간단하게 언급하고 지나가겠습니다.

평소 클래식 음악을 즐겨듣는터라 비싼 스피커를 받게 되어서 기쁜 마음이 컸습니다. 다만 받은 순간에는 저게 꽤 무거워서 (체감상 5kg) 저거 들고 집에 어떻게 가지... 란 생각을 했습니다.

블루투스 스피커인데, 기본적으로 모듈 자체가 커서 성능 특히 저음부를 잘 살릴 수 있지 않을까 생각을 했습니다. 그리고 곡은 3가지 정도를 먼저 들어봤습니다.

• F.Liszt, Réminiscences des Norma - 피아노 솔로
• M.Ravel, La valse - 피아노 솔로 & 오케스트라
• A.Bruckner, 교향곡 9번 1악장 - 오케스트라

특히, 두번째 La valse의 경우에는 컨트라바순으로 시작하는 초저음부터 확인 가능하고, 모리스 라벨의 오케스트레이션으로 화려하고 귀가 즐거운 음악이기에 제가 항상 음질 테스트시 사용하던 작품입니다.

기본적으로 손잡이가 달려있어 들고 다니기는 무게에 비해서 편하긴 하였고, 기본적으로 유선 포트가 따로 있는 것으로 확인되어서 일반적인 블루투스 스피커에 비해서 유선 스피커에 가깝게 음질이 좋았습니다.

음 해상도가 명확한 편이고 저음부 역시 빵빵한 편이여서 여건만 괜찮으면 크게 듣고 싶은데... 환경이 고시원이라 제한적이긴 한게 아쉽긴 합니다. Mahler나 Shostakovich 작품들 들으면 기깔날 것 같다는 생각이 드네요.

내장 마이크가 있어서 주변 환경에 적응을 하는 측면도 있고, 기기도 하나만 페어링 되는 것이 아닌 두개가 가능해서 기술적으로도 볼만한 사항이 있던 것 같습니다. 안그래도 이번 프로젝트를 IoT쪽으로 구상하고 있어 사용해보면서 어느정도 참고할 만한 측면이 있는 것 같습니다.

SSAFY의 장점?

이제는 프로젝트 주간입니다. 프로젝트 주간에서 팀과 논의를 해보면서 아직 새롭게 배울 것이 많아감을 요즘 느끼고 있습니다.

SSAFY에서 가장 크게 얻어가고 있는 것(또는 얻어갈 것)이 무엇인가 라고 한다면 다음과 같습니다.

진로

한번도 취업준비에 대해 생각해본적 없는 제 입장에서 가장 막막한 것은 능력도, 비전공도 아닌 뭘 하고 싶은지에 대한 고민이였습니다. SSAFY 에서 사람들을 많이 많나고 많은 도메인을 가진 사람들과 교류하다보니 지금은 어느정도 안정화되는 것 같습니다. 또한 취업지원프로그램 역시도 많이 있다보니 취업준비로 무엇을 해야되는지 정도 감이 잡혀가는 것 같습니다.

겸손

이에 따라 많은 능력자들을 보게 되었습니다. 그런 능력자들을 볼 때마다 항상 작아지기도 하지만, 가끔 자만할 때 계영배의 역할을 해 주는 것 같습니다.

(아직은 못딴...) B형

결국 원하는 직무가 삼성전자에 있어 B형이 거의 저에게 있어서는 무엇보다도 필수적입니다. B형 스터디를 운영하고 있는만큼, 또 지금까지 B형 탈락한 만큼 이왕 딸꺼 최상위권으로 통과하고 싶습니다.

프로젝트 경험

기본적으로 자소서에 쓸 프로젝트 3개(관통 프로젝트까지 하면 4개)가 생긴다는 점이 경쟁력이 확실히 있는 것 같습니다. 전문적인 코칭을 현재 받고 있는데, 열심히해서 좋은 성과 있으면 좋겠습니다.

다이아는 정말... 힘드네요

https://www.acmicpc.net/problem/10167 점과 가중치가 있을 때 x, y축과 평행한 직사각형으로 감싼 최대 2차원 연속합을 구하는 문제입니다.

풀이

연습 문제

연습문제로 [BOJ] 16992 / 연속합과 쿼리 문제를 먼저 풀이하는 것을 추천 드립니다.

연습 문제 풀이

연속합은 떨어지는 구간이 발생해서는 안됩니다. 따라서 기본적으로 연속합에 해당하는 구간이 왼쪽 끝, 오른쪽 끝에 닿는지 여부를 판단하는 것이 중요합니다. 이 여부에 따라 두 구간이 붙을 수 있는지 없는지에 대한 판단 기준이 됩니다.

결국 관리해야 되는 것은 4가지 입니다.

왼쪽 끝이 닿는 구간합(left)

오른쪽 끝이 닿는 구간합(right)

중간 부분에 걸친 구간합(mid)

전체 합(all)

여기서 구간 내 최댓값은 사실상 모든 기준을 포함하는 mid 가 됩니다. 따라서 마지막에 get 할 때 mid 값을 가져오면 됩니다.

그리고 4개의 값을 가져와야 되므로 트리를 4개 만들어야 되는가 생각할 수 있는데, 4개나 따로 구현하기란 너무 귀찮은 일입니다. 이 4개를 하나로 통합하는 Node 를 만들어 관리하면 됩니다.

이제 남은 것은 두 구간합을 합치는 연산입니다. n1 노드와 n2 노드를 합치는 연산은 다음과 같습니다.

• left
  • 왼쪽 노드만 생각할 것인가 → n1.left
  • 오른쪽 노드도 고려할 것인가 → 왼쪽 전체 선택하고 오른쪽 노드의 left → n1.all + n2.left

    max(n1.left, n1.all + n2.left);

• right
  • 오른쪽 노드만 생각할 것인가 → n1.right
  • 왼쪽 노드도 고려할 것인가 → 오른쪽 전체 선택하고 왼쪽 노드의 right → n1.right + n2.all

    max(n2.right, n1.right + n2.all);

• mid
  • 왼쪽 노드만 생각할 것인가 → n1.mid
  • 오른쪽 노드만 생각할 것인가 → n2.mid
  • 둘을 이어 붙일 것인가 → 왼쪽 노드의 right + 오른쪽 노드의 left → n1.right + n2.left

    max(n1.mid, n2.mid, n1.right + n2.left);

• all
  • 둘 다 전체 선택한 거 더하면 됩니다.

    n1.all + n2.all

이 연산에 대한 항등원(세그먼트 트리 범위 밖일 경우 반환)도 생각해보면 다음과 같습니다.

• left, right, mid : max 연산의 항등원 → -INF
• all : + 연산의 항등원 → 0

참고로 정식 명칭은 MSP(Maximum Subarray Problem)라 하는데, 국내에서는 아래에 풀이할 금광 덕분에 금광세그라는 이름으로 더 많이 알려져 있습니다.

연습 문제 코드

import java.io.*;
import java.util.*;

public class Main {
    static class Node {
        int left, right, mid, all;

        Node() {
            this(-1_000_000_000);
            all = 0;
        }

        Node(int val) {
            left = right = mid = all = val;
        }
    }

    static int n, size;
    static int[] arr;
    static Node[] tree;

    static Node merge(Node n1, Node n2) {
        Node node = new Node();
        node.left = Math.max(n1.left, n1.all + n2.left);
        node.right = Math.max(n2.right, n1.right + n2.all);
        node.mid = Math.max(Math.max(n1.mid, n2.mid), n1.right + n2.left);
        node.all = n1.all + n2.all;
        return node;
    }

    static void init(int node, int s, int e) {
        if (s == e) {
            tree[node] = new Node(arr[s]);
            return;
        }
        int mid = (s + e) >> 1;
        init(node << 1, s, mid);
        init(node << 1 | 1, mid + 1, e);
        tree[node] = merge(tree[node << 1], tree[node << 1 | 1]);
    }

    static int query(int s, int e) {
        return get(1, 1, n, s, e).mid;
    }

    static Node get(int node, int s, int e, int ts, int te) {
        if (e < ts || te < s) return new Node();
        if (ts <= s && e <= te) return tree[node];
        int mid = (s + e) >> 1;
        Node l = get(node << 1, s, mid, ts, te);
        Node r = get(node << 1 | 1, mid + 1, e, ts, te);
        return merge(l, r);
    }

    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringBuilder sb = new StringBuilder();
        StringTokenizer st;

        n = Integer.parseInt(br.readLine());
        size = 1 << ((int) Math.ceil(Math.log(n) / Math.log(2)) + 1);
        arr = new int[n + 1];
        tree = new Node[size];

        st = new StringTokenizer(br.readLine(), " ");
        for (int i = 1; i <= n; i++) {
            arr[i] = Integer.parseInt(st.nextToken());
        }

        init(1, 1, n);
        int m = Integer.parseInt(br.readLine());
        while (m-- > 0) {
            st = new StringTokenizer(br.readLine(), " ");
            int i = Integer.parseInt(st.nextToken());
            int j = Integer.parseInt(st.nextToken());
            sb.append(query(i, j)).append('\n');
        }

        System.out.print(sb);
        br.close();
    }
}

금광 풀이

지금까지 금광세그라는 것을 알아보았습니다. 하지만 이 세그를 알아도 풀이가 쉽지는 않습니다. 세그를 일단 생각하지 않고 접근 해보겠습니다.

처음 든 생각

[x1, x2] , [y1, y2] 범위의 점들을 선택해서 이 범위에 해당하는 점들의 합을 구합니다.

• [x1, x2] : O(N^2)
• [y1, y2] : O(N^2)
• 구간 내 점들의 합 : O(N^2)
• 전체 : O(N^6)

당연히 이렇게 풀면 TLE입니다. 여기서 힌트가 되는 것이 바로 서브테스크가 됩니다.

1. N ≤ 100 : O(N^6) 가능
2. N ≤ 500 : 조금 더 줄여야 됨
3. wi < 0 (1 개) : 연속합 중간에 끊기는 부분이 딱 하나 존재
4. y = 0 : 일직선 상에 존재하는 연속합의 최대, 위에서 설명한 금광세그 사용
5. 제약조건 없음 : 종합

점들 선택하는 과정

문제가 되는 부분은 바로 [x1, x2] , [y1, y2] 범위를 모두 본다는 것에 있습니다. 이중에서 하나만 보고 나머지는 뭔가 최적화하는 방법이 있을 것인지 생각해보는 것이 어떨까요?

마침 서브테스크 중에 y값이 같을 경우 어떻게 해결할 것인지 보는 것이 있습니다. [y1, y2] 에 대한 선택만 하도록 합니다 → O(N^2)

그러면 점들에 대한 순서 역시 필요하게 되고, 그 과정은 다음과 같습니다.

• y 기준으로 정렬
• y1을 고정하여 세그먼트 트리를 초기화
• 정점들을 하나씩 넣어서 세그먼트 트리를 갱신

반례 잡기

이렇게 풀이하면 되는데, 반례 하나가 있습니다.

4
2 2 4
2 1 6
1 2 7
1 1 -1000

최대가 나올 수 있는 부분은 (1, 2) 와 (2, 2) 을 선택하면 됩니다. (답 : 11)

하지만 현재 풀이대로만 진행하면 17이 나옵니다. 문제의 원인은 다음과 같습니다.

• 정렬 순서 (1, 1) → (1, 2) → (2, 1) → (2, 2)
• (1, 1) 에서 4개의 점 삽입 시에는 문제 없음
• (1, 2) 에서 같은 높이에 있는 (1, 1)이 누락 → (1, 2), (2, 1), (2, 2) 만 포함된 범위가 계산됨 → error

즉, 같은 높이일 때는 해당하는 같은 높이의 모든 점을 다 삽입해주어야 됩니다.

다만, 조금 관점을 다르게 생각해서 x 좌표가 제일 작은 시점에서만 모두 넣어주고, 나머지는 모두 무시해보는 방법도 생각해볼 수 있습니다. 즉 중복은 무시해주면 됩니다.

for (int i = 0; i < n; i++) {
    // 같은 높이는 중복 처리 X
    if (i > 0 && point[i - 1].y == point[i].y) continue;
    init();
    for (int j = i; j < n; j++) {
        update((int) point[j].x, point[j].v);
        // 같은 높이는 모두 삽입된 후 처리
        if (j == n - 1 || point[j].y != point[j + 1].y) {
            res = Math.max(res, tree[1].mid);
        }
    }
}

코드

import java.io.*;
import java.util.*;

public class Main {
    static class Point implements Comparable<Point> {
        long x, y, v;
        int i;

        public Point(long x, long y, long v, int i) {
            this.x = x;
            this.y = y;
            this.v = v;
            this.i = i;
        }

        @Override
        public String toString() {
            return "[" + x + ", " + y + ", " + v + ", " + i + "]";
        }

        @Override
        public int compareTo(Point p) {
            return this.y == p.y ? Long.compare(this.x, p.x) : Long.compare(this.y, p.y);
        }
    }

    static class Pair {
        int idx;
        long val;

        public Pair(int idx, long val) {
            this.idx = idx;
            this.val = val;
        }

        @Override
        public String toString() {
            return "[" + idx + ", " + val + "]";
        }
    }

    static class Node {
        long left, right, mid, all;
    }

    static int n, m, size;
    static Node[] tree;
    static Point[] point;
    static Pair[] tmpX, tmpY;

    static void compression() {
        tmpX = new Pair[n];
        tmpY = new Pair[n];
        for (Point p : point) {
            tmpX[p.i] = new Pair(p.i, p.x);
            tmpY[p.i] = new Pair(p.i, p.y);
        }

        Arrays.sort(tmpX, (o1, o2) -> Long.compare(o1.val, o2.val));
        Arrays.sort(tmpY, (o1, o2) -> Long.compare(o1.val, o2.val));

        int ix = 0, iy = 0;
        for (int i = 0; i < n - 1; i++) {
            point[tmpX[i].idx].x = ix;
            if (tmpX[i].val != tmpX[i + 1].val) ix++;
        }
        point[tmpX[n - 1].idx].x = ix;

        for (int i = 0; i < n - 1; i++) {
            point[tmpY[i].idx].y = iy;
            if (tmpY[i].val != tmpY[i + 1].val) iy++;
        }
        point[tmpY[n - 1].idx].y = iy;

        Arrays.sort(point);
    }

    static void init() {
        for (int i = 0; i < size; i++) {
            tree[i].left = tree[i].right = tree[i].mid = (long) (-1e14);
            tree[i].all = 0;
        }
    }

    static Node merge(Node n1, Node n2) {
        Node node = new Node();
        node.left = Math.max(n1.left, n1.all + n2.left);
        node.right = Math.max(n2.right, n1.right + n2.all);
        node.mid = Math.max(Math.max(n1.mid, n2.mid), n1.right + n2.left);
        node.all = n1.all + n2.all;
        return node;
    }

    static void update(int idx, long val) {
        idx += m;
        tree[idx].all += val;
        tree[idx].left = tree[idx].right = tree[idx].mid = tree[idx].all;
        while ((idx >>= 1) > 0) {
            tree[idx] = merge(tree[idx << 1], tree[idx << 1 | 1]);
        }
    }

    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st;

        n = Integer.parseInt(br.readLine());
        m = 1 << (int) Math.ceil(Math.log(n) / Math.log(2));
        size = m << 1;
        tree = new Node[size];
        for (int i = 0; i < size; i++) tree[i] = new Node();
        point = new Point[n];
        for (int i = 0; i < n; i++) {
            st = new StringTokenizer(br.readLine(), " ");
            long x = Long.parseLong(st.nextToken());
            long y = Long.parseLong(st.nextToken());
            long v = Long.parseLong(st.nextToken());
            point[i] = new Point(x, y, v, i);
        }
        compression();

        long res = 0;
        for (int i = 0; i < n; i++) {
            if (i > 0 && point[i - 1].y == point[i].y) continue;
            init();
            for (int j = i; j < n; j++) {
                update((int) point[j].x, point[j].v);
                if (j == n - 1 || point[j].y != point[j + 1].y) {
                    res = Math.max(res, tree[1].mid);
                }
            }
        }

        System.out.println(res);
    }
}

• update 과정은 가급적 O(depth)로 처리!
• 런타임 에러 조심할 것!

구현할 API 설명

void init()

• 최초 1회 실행
• 0번 root 노드는 주어지지 않음
• 알람은 부모로 전파 됨

Q : [1, 100000] // 쿼리 수
N : [1, 100000] // 0번 제외 노드 수
D : [1, 20] // 이진 트리 최대 깊이, depth
pi : [0, N] // 부모 노드 번호
ai : [1, N] // 권한 세기

void updateAlarm(int c)

• c번 노드 알람 전파 ON/OFF toggle

c : [1, N]

void updateAuth(int c, int power)

• c번 노드 권한 세기 변경

c : [1, N]
power : [1, N]

void updateParents(int c1, int c2)

• c1번 노드, c2번 노드 부모 변경
• c1번 노드와 c2번 노드는 동일한 depth

c1, c2 : [1, N]

int get(int c)

• c번 노드에 알림을 보낼 수 있는 하위 채팅방(노드) 개수(본인은 제외)

c : [1, N]

풀이

권한 설정

전파 거리 관리

어떻게 전파 거리를 고려하여 값을 가지고 있는가에 대한 고민을 하면 됩니다. 사실 가장 기초적인 방법으로는 Node 구현 시 List<Node> 형태로 해당 노드로 전파 가능한 하위 노드들을 들고 있으면 되긴 합니다.

하지만 삽입 삭제 과정이 빈번하게 일어나는 상황에서 이는 바람직하지 않습니다. 어떤 방법을 쓰면 될까요?

이 때 우리는 조건 D <= 20 이 수상합니다. 이걸 활용해서 가능합니다.

어차피 우리에게 필요한 건 하위 노드의 종류가 아닌 전파 거리만이 필요합니다. 그래서 이를 관리해주는 카운팅 배열 cnt[D] 를 만들어 줍니다.

cnt[i] : 전파 거리가 i 만큼 남은 하위 노드의 개수
ex) 0 - 1(auth = 1) - 2(auth = 1) - 3(auth = 2) 와 같은 트리에서
3 - cnt : [0, 0, 0, 0, ...]
2 - cnt : [0, 1, 0, 0, ...] // 노드 3의 전파 거리 2 - 1
1 - cnt : [2, 0, 0, 0, ...] // 노드 2의 전파 거리 1 - 1, 노드 3의 전파 거리 2 - 1 - 1
0 - cnt : [1, 0, 0, 0, ...] // 음수가 되면 사라짐, 노드 1의 전파 거리 1 - 1

그리고 update나 get을 할 때 이 배열 전체를 순회하면 됩니다. 배열의 사이즈는 D 로 잡으면 충분합니다. 어차피 제일 깊은 곳에서 root 까지 올라와도 겨우 20회 확인하기 때문입니다.

setValue & setSub

각각의 역할은 다음과 같습니다.

• setValue : 전파 거리 딱 하나 바뀌었을 때 update 수행
• setSub : 서브 트리 자체가 바뀌었을 때 update 수행

setValue(int id, int val, int power)

• id번 노드의 auth 값이 power 로 변경이 될 때 val 만큼 cnt를 변경해주는 update
• 단일 값이 수정 될 경우 사용하고, root거나 전파 거리가 부족하거나 알람 OFF라면 break
• 자식에서 부모로 이동하는 것이 핵심 포인트!
• O(D)

    static void setValue(int id, int val, int power) {
        while(true) {
            if(parents[id] == -1 || power-- == 0 || alarm[id] == -1) break;
            node[parents[id]].cnt[power] += val;
            id = parents[id];
        }
    }

setSub(int id, int val)

• id번 노드를 root로 하는 서브트리를 그 부모 노드에 할당 또는 제거하는 연산
• 알람 ON/OFF, parents 바꾸기 등에 사용
• setValue 를 root(id번 노드) 의 auth를 한번 적용한 것과 cnt의 크기(D) 만큼 수행
• O(D^2)

    static void setSub(int id, int val) {
        setValue(id, val, node[id].auth);
        for(int power = 1; power < 20; power++) {
            setValue(id, val * node[id].cnt[power], power);
        }
    }

추가 메소드의 사용처

• setValue : updateAuth - O(D)
• setSub : updateAlarm , updateParents - O(D^2)

주의 사항

권한 세기

• 권한 세기를 정할 때 p 가 D 보다도 크게 나오는 경우가 존재 - 런타임 에러
• setter 를 통해서 값을 변경하도록 함 - 캡슐화 필요

    static class Node {
        int id, auth;
        int[] cnt;

        public Node(int id, int auth) {
            this.id = id;
            setAuth(auth);
            cnt = new int[20];
        }

        public void setAuth(int auth) { // setter를 만들어 두기
            this.auth = auth > 20 ? 20 : auth;
        }

        @Override
        public String toString() {
            return "Node [id=" + id + ", auth=" + auth + ", cnt=" + Arrays.toString(cnt) + "]";
        }
    }

• 값을 바꿀 때는 setter를 거칩니다.

    static void updateAuth(int c, int power) {
        setValue(c, -1, node[c].auth);
        node[c].setAuth(power); // setter를 통한 값 변경
        setValue(c, 1, node[c].auth);
    }

알람 ON/OFF

• setSub 도 사용하는 setValue 를 보면 알겠지만 알람 OFF의 경우에는 update X
  • 알람 ON : alarm[i] == 1 , 알람 OFF : alarm[i] == -1
• 알람의 초기 상태에 따라 동작 과정이 다름
  • 알람 ON → OFF
    • setSub(id, -1) : 상위 노드 cnt 하나씩 빼주기
    • alarm[id] = -1 : 알람 OFF
  • 알람 OFF → ON
    • alarm[id] = 1 : 알람 ON
    • setSub(id, 1) : 상위 노드 cnt 하나씩 더해주기

시간 복잡도

API 별

• init : O(D * N)
• updateAlarm : O(D^2)
• updateAuth : O(D)
• updateParents : O(D^2)
• get : O(D)

전체

O(D^2 * Q + D * N)

코드

import java.io.*;
import java.util.*;

public class Main {
    static class Node {
        int id, auth;
        int[] cnt;

        public Node(int id, int auth) {
            this.id = id;
            setAuth(auth);
            cnt = new int[20];
        }

        public void setAuth(int auth) {
            this.auth = auth > 20 ? 20 : auth;
        }

        @Override
        public String toString() {
            return "Node [id=" + id + ", auth=" + auth + ", cnt=" + Arrays.toString(cnt) + "]";
        }
    }

    static int N, Q;
    static int[] parents, auth, alarm;
    static Node[] node;

    static void init() {
        parents[0] = -1;
        node = new Node[N + 1];
        alarm = new int[N + 1];
        Arrays.fill(alarm, 1);
        for(int i = 0; i <= N; i++) {
            node[i] = new Node(i, auth[i]);
        }
        for(int i = 0; i <= N; i++) {
            setValue(i, 1, node[i].auth);
        }
    }

    static void setValue(int id, int val, int power) {
        while(true) {
            if(parents[id] == -1 || power-- == 0 || alarm[id] == -1) break;
            node[parents[id]].cnt[power] += val;
            id = parents[id];
        }
    }

    static void setSub(int id, int val) {
        setValue(id, val, node[id].auth);
        for(int power = 1; power < 20; power++) {
            setValue(id, val * node[id].cnt[power], power);
        }
    }

    static void updateAlarm(int c) {
        if(alarm[c] == -1) {
            alarm[c] = 1; setSub(c, 1);
        } else {
            setSub(c, -1); alarm[c] = -1;
        }
    }

    static void updateAuth(int c, int power) {
        setValue(c, -1, node[c].auth);
        node[c].setAuth(power);
        setValue(c, 1, node[c].auth);
    }

    static void updateParents(int c1, int c2) {
        setSub(c1, -1); setSub(c2, -1);
        int p1 = parents[c1], p2 = parents[c2];
        parents[c1] = p2; parents[c2] = p1;
        setSub(c1, 1); setSub(c2, 1);
    }

    static int get(int c) {
        int ans = 0;
        int[] cnt = node[c].cnt;
        for(int i : cnt) {
            ans += i;
        }
        return ans;
    }

    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringTokenizer st = null;
        st = new StringTokenizer(br.readLine(), " ");
        N = Integer.parseInt(st.nextToken());
        Q = Integer.parseInt(st.nextToken());
        parents = new int[N + 1];
        auth = new int[N + 1];
        StringBuilder sb = new StringBuilder();
        for(int i = 0; i < Q; i++) {
            st = new StringTokenizer(br.readLine());
            int query = Integer.parseInt(st.nextToken());
            int c, c1, c2, power;
            switch(query) {
            case 100:
                for(int j = 1; j <= N; j++) parents[j] = Integer.parseInt(st.nextToken());
                for(int j = 1; j <= N; j++) auth[j] = Integer.parseInt(st.nextToken());
                init();
                break;
            case 200:
                c = Integer.parseInt(st.nextToken());
                updateAlarm(c);
                break;
            case 300:
                c = Integer.parseInt(st.nextToken());
                power = Integer.parseInt(st.nextToken());
                updateAuth(c, power);
                break;
            case 400:
                c1 = Integer.parseInt(st.nextToken());
                c2 = Integer.parseInt(st.nextToken());
                updateParents(c1, c2);
                break;
            case 500:
                c = Integer.parseInt(st.nextToken());
                sb.append(get(c)).append('\n');
                break;
            }
        }
        System.out.print(sb);
        br.close();
    }
}

다음 문제를 바로 살펴봅시다. 문제 링크

문제에서 처리하는 쿼리들을 정리하면 다음과 같습니다.

• 1 i x : i번 지역에 x 만큼 추가
• 2 i y : i번 지역에 y 만큼 추가
• 3 L R : [L, R] 범위의 지역 수
• 4 T : 내림차순으로 T번째 높이

이거만 보면 일반적인 세그먼트 트리(이하 세그) 문제라 생각할 수 있습니다. k번째 찾는 쿼리라 k번째 찾는 방법을 쓰면 풀릴 것 같습니다. (단 내림차순이라 N - T + 1 번째를 찾는 것임에 유의)

그런데... 높이 범위가 10의 18승으로... 일반적인 방법으로 저장은 절대 불가능하다는 것을 알 수 있습니다.

방법으로는 오프라인 쿼리를 활용하여 좌표 압축을 진행하는 방법도 있습니다. 쿼리에 나오는 모든 높이를 받아서 이에 해당하는 배열만 만드는 방법입니다.

하지만, 온라인으로 문제를 풀어야 되는 상황이라면...? 트리의 노드를 동적으로 저장해야됩니다. 이를 다이나믹 세그먼트 트리라 합니다.

다이나믹 세그먼트 트리

동작원리는 일반 세그먼트 트리를 잘 안다면 쉽게 이해할 수 있습니다. 딱 두가지만 지키면 됩니다.

• update 시 필요한 곳의 노드만 만들어 주자
• 탐색 했는데 하위 노드 없으면 그냥 return 해주자

일반 세그먼트 트리는 이미 트리를 완전 이진 트리 형태로 다 만들어 놓고 거기에서 값만 변경하는 방식입니다.

하지만 다이나믹 세그먼트 트리는 update 쿼리 수행시 필요한 곳에서만 노드를 만들어 주고 있기에 이진 트리는 이진 트리인데 완전 이진 트리는 아니고 많이 비어 있는 모습을 띄게 됩니다.

이렇게 필요한 부분만 만들기에 메모리를 많이 절약할 수 있게 됩니다.

그리고 눈치가 빠른 분들은 기본 세그의 노드 번호는 고정이 되어있지만, 다이나믹 세그 노드 번호는 이와 다르다는게 보일겁니다(사진상의 노드 번호가 보면 서로 다릅니다.)

이는 노드가 새롭게 부여될 때마다 노드 번호가 증가하는 방식입니다. 그래서 노드 번호는 할당된 순으로 부여됩니다.

따라서 노드 선언할 때 값 뿐만 아니라 왼쪽과 오른쪽 노드 정보도 같이 들고 있어야 한다는 것이 핵심입니다.

구현

이 구현은 JusticeHui님의 구현 방식을 많이 참고하였습니다. 자바로 구현한 포스팅은 못보았기에 구현해보게 되었습니다.

Node

우선 Node를 정의합니다.

class Node {
    long val = 0;
    int l = -1, r = -1;
}

그리고 Node를 관리할 리스트를 만들어 둡니다.

int size = 1; // 초기 트리는 루트 노트 딱 하나만 존재
List<Node> tree;

// 초기화 시
tree = new ArrayList<>();
tree.add(new Node()); // 루트 노드 저장

즉 필요한 노드만 관리하고, 노드 내에서 저장을 해당하는 노드의 인덱스를 저장하는 방식을 택합니다. 여기서는 root는 0번 노드로 설정했으며, -1은 노드가 할당되지 않음을 의미합니다.

update

우선 기존 세그입니다.

void update(int node, int s, int e, int idx, long val) {
    if(e < idx || idx < s) return;
    if(s == e) {
        tree[node] += val;
        return;
    }
    int mid = (s + e) / 2;
    update(2 * node, s, mid, idx, val);
    update(2 * node + 1, mid + 1, e, idx, val);
    tree[node] = tree[2 * node] + tree[2 * node + 1];
}

위 코드는 update 하기 위해서 양쪽 모두를 탐색하고 있습니다. 하지만 다이나믹 세그의 경우는 필요한 곳만 할당하기에 탐색 과정에서 조건문을 하나 추가로 적어줍니다.

void update(int node, long s, long e, long idx, long val) {
    if(s == e) {
        tree.get(node).val += val;
        return;
    }
    long mid = (s + e) / 2;
    if(idx <= mid) {
        if(tree.get(node).l < 0) {
            tree.add(new Node());
            tree.get(node).l = size++;
        }
        update(tree.get(node).l, s, mid, idx, val);
    } else {
        if(tree.get(node).r < 0) {
            tree.add(new Node());
            tree.get(node).r = size++;
        }
        update(tree.get(node).r, mid + 1, e, idx, val);
    }
    int left = tree.get(node).l;
    int right = tree.get(node).r;
    long lval = left < 0 ? 0 : tree.get(left).val;
    long rval = right < 0 ? 0 : tree.get(right).val;
    tree.get(node).val = lval + rval;
}

get

get은 비슷하지만 없는 노드가 나오면 탐색을 중단한다는 점만 조심하면 됩니다.

long getTotal(int node, long s, long e, long ts, long te) {
    if(node < 0) return 0;
    if(e < ts || te < s) return 0;
    if(ts <= s && e <= te) return tree.get(node).val;
    long mid = (s + e) / 2;
    long left = getTotal(tree.get(node).l, s, mid, ts, te);
    long right = getTotal(tree.get(node).r, mid + 1, e, ts, te);
    return left + right;
}

T번째 항목 얻기

이게 처음 생각하면 잘 생각이 안날껀데 의외로 기존 세그와 비슷합니다. 다만, 없는 노드는 탐색하지 않는다는 점만 주의하면 됩니다.

long getTth(int node, long s, long e, long t) {
    if (s == e) return s;
    long mid = (s + e) / 2;
    long lval = (tree.get(node).l < 0) ? 0 : tree.get(tree.get(node).l).val;
    if (lval >= t) return getTth(tree.get(node).l, s, mid, t);
    return getTth(tree.get(node).r, mid + 1, e, t - lval);
}

문제 풀이

14577 일기 예보 - P3

import java.io.*;
import java.util.*;

public class Main {
    static class Node {
        long val = 0;
        int l = -1, r = -1;
    }

    static int N, M, size;
    static final long MAX_VALUE = (long) 1e18;
    static long[] arr;
    static List<Node> tree;

    static void update(long idx, long val) {
        update(0, 0, MAX_VALUE, idx, val);
    }

    static long getTotal(long l, long r) {
        return getTotal(0, 0, MAX_VALUE, l, r);
    }

    static long getTth(long t) {
        return getTth(0, 0, MAX_VALUE, t);
    }

    static void update(int node, long s, long e, long idx, long val) {
        if(s == e) {
            tree.get(node).val += val;
            return;
        }
        long mid = (s + e) / 2;
        if(idx <= mid) {
            if(tree.get(node).l < 0) {
                tree.add(new Node());
                tree.get(node).l = size++;
            }
            update(tree.get(node).l, s, mid, idx, val);
        } else {
            if(tree.get(node).r < 0) {
                tree.add(new Node());
                tree.get(node).r = size++;
            }
            update(tree.get(node).r, mid + 1, e, idx, val);
        }
        int left = tree.get(node).l;
        int right = tree.get(node).r;
        long lval = left < 0 ? 0 : tree.get(left).val;
        long rval = right < 0 ? 0 : tree.get(right).val;
        tree.get(node).val = lval + rval;
    }

    static long getTotal(int node, long s, long e, long ts, long te) {
        if(node < 0) return 0;
        if(e < ts || te < s) return 0;
        if(ts <= s && e <= te) return tree.get(node).val;
        long mid = (s + e) / 2;
        long left = getTotal(tree.get(node).l, s, mid, ts, te);
        long right = getTotal(tree.get(node).r, mid + 1, e, ts, te);
        return left + right;
    }

    static long getTth(int node, long s, long e, long t) {
        if (s == e) return s;
        long mid = (s + e) / 2;
        long lval = (tree.get(node).l < 0) ? 0 : tree.get(tree.get(node).l).val;
        if (lval >= t) return getTth(tree.get(node).l, s, mid, t);
        return getTth(tree.get(node).r, mid + 1, e, t - lval);
    }

    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        StringBuilder sb = new StringBuilder();
        StringTokenizer st = null;
        st = new StringTokenizer(br.readLine(), " ");
        N = Integer.parseInt(st.nextToken());
        M = Integer.parseInt(st.nextToken());
        size = 1;
        arr = new long[N + 1];
        tree = new ArrayList<>();
        tree.add(new Node());

        st = new StringTokenizer(br.readLine(), " ");
        for(int i = 1; i <= N; i++) {
            arr[i] = Integer.parseInt(st.nextToken());
            update(arr[i], 1);
        }

        int q, idx;
        long val, l, r, t;
        while(M-- > 0) {
            st = new StringTokenizer(br.readLine(), " ");
            q = Integer.parseInt(st.nextToken());
            switch(q) {
            case 1:
                idx = Integer.parseInt(st.nextToken());
                val = Long.parseLong(st.nextToken());
                update(arr[idx], -1);
                arr[idx] += val;
                update(arr[idx], 1);
                break;
            case 2:
                idx = Integer.parseInt(st.nextToken());
                val = Long.parseLong(st.nextToken());
                update(arr[idx], -1);
                arr[idx] -= val;
                update(arr[idx], 1);
                break;
            case 3:
                l = Long.parseLong(st.nextToken());
                r = Long.parseLong(st.nextToken());
                sb.append(getTotal(l, r)).append('\n');
                break;
            case 4:
                t = Long.parseLong(st.nextToken());
                sb.append(getTth(N - t + 1)).append('\n');
                break;
            }
        }

        System.out.print(sb);
        br.close();
    }
}

참고

• https://justicehui.github.io/medium-algorithm/2020/02/28/DynamicSeg/
• https://nicotina04.tistory.com/265
• https://www.acmicpc.net/blog/view/86

먼저 읽어보기 - Sparse Table

Sparse Table 포스팅을 작성하고 관련 피드백을 하나 들었습니다.

관련해서 이 포스팅 한번 참고하시면 좋을 것 같아요!

바로 CPU Cache 때문에 배열의 선언 방법에 따라 TLE가 날 수도 있다고 합니다. 관련 이슈 및 개념을 정리하면서 이를 확인해보기 위해 최소 공통 조상에서 어떻게 성능 차이가 나는지 알아보겠습니다.

CPU 캐시(Cache)란?

캐시에 대해서 간단하게 먼저 짚고 넘어가겠습니다.

캐시 메모리(Cache Memory)는 데이터를 저장하는 임시 저장소로, CPU가 데이터를 더 빠르고 효율적으로 액세스할 수 있도록 돕습니다. 이는 CPU와 메모리 사이에 위치하며, 속도가 빠른 SRAM 기반으로 설계된 저장 장치입니다. 캐시는 특히 동일한 데이터에 반복적으로 접근하거나 많은 연산이 필요한 경우 컴퓨터 성능을 크게 향상시킵니다.

캐시 메모리의 동작 원리

1. 시스템이 데이터 요청 시, 먼저 캐시에서 해당 데이터를 찾습니다.
2. 캐시에 데이터가 있으면(Cache Hit), 해당 데이터를 바로 제공합니다.
3. 데이터가 없거나 오래되어 최신성을 잃은 경우(Cache Miss), 원본 데이터 소스에 접근합니다.
4. 캐시 공간이 부족하면, 잘 사용되지 않는 데이터를 삭제하여 공간을 확보합니다.

캐시의 계층 구조

캐시는 보통 L1, L2, L3 캐시로 나뉘며, 각 계층은 속도와 용량이 다릅니다.

• L1 캐시: 코어별로 존재하며, 가장 작고 빠름
  • 명령어 저장(L1I)과 데이터 저장(L1D)으로 나뉘기도 함 - 분리형 캐시
• L2 캐시: L1보다 크고 느리고, 코어별로 존재.
• L3 캐시: 여러 코어가 공유, 가장 크고 느림.

참조 지역성(Locality of Reference)

시간 지역성(Time Locality) : 최근에 접근한 데이터에 다시 접근하는 경향.

   for (int i = 0; i < 10; i++) {
       arr[i] = i; // 변수 i를 여러 번 참조.
   }

공간 지역성(Spatial Locality) : 최근 접근한 데이터 주변 공간에 다시 접근하는 경향.

   int[] arr = new int[10];
   for (int i = 0; i < 10; i++) {
       arr[i] = i; // 배열 요소에 연속적으로 접근.
   }

캐시의 필요성

1. 병목 현상 완화: CPU와 메인 메모리(RAM) 사이의 병목 현상을 줄이기 위해, 크기는 작지만 속도가 빠른 캐시를 사용합니다. CPU가 자주 사용하는 데이터를 캐시에 저장하여, 데이터를 더 빠르게 가져옵니다.

   • 병목 현상: 시스템 성능이 특정 구성 요소의 제한으로 인해 저하되는 현상.

2. 데이터 접근 시간 감소: 캐싱은 데이터를 빠르게 액세스할 수 있는 위치에 저장함으로써, 디스크나 데이터베이스와 같은 느린 저장 장치에 대한 접근을 줄입니다.

3. 서버 부하 분산: 캐시에서 데이터를 제공함으로써 데이터베이스 서버나 파일 시스템에 가해지는 부하를 줄이고, 전체 시스템 성능을 향상시킵니다.

캐시 관련 용어

• 캐시 히트(Cache Hit): CPU가 요청한 데이터가 캐시에 있는 경우.
• 캐시 미스(Cache Miss): CPU가 요청한 데이터가 캐시에 없는 경우.
• 캐시 적중률(Cache Hit Ratio): 캐시 히트 / (캐시 히트 + 캐시 미스).
• 캐시 일관성(Cache Coherence): 멀티 코어 환경에서 각 코어의 캐시에 저장된 데이터가 동일하게 유지되는 것을 보장.

캐시와 이차원 배열 성능

다음과 같이 간단하게 c++ 로 배열 원소의 주소를 출력하면 다음과 같습니다.

#include <iostream>

using namespace std;

int arr[20][101010];

int main() {
    for(int i = 0; i < 3; i++) {
        for(int j = 0; j < 3; j++) {
            cout << "arr[" << i << "]" << "[" << j << "]" << " : " << &arr[i][j] << '\n';
        }
    }
    return 0;
}

arr[0][0] : 0x7ff6a3957040
arr[0][1] : 0x7ff6a3957044
arr[0][2] : 0x7ff6a3957048
arr[1][0] : 0x7ff6a39b9a88
arr[1][1] : 0x7ff6a39b9a8c
arr[1][2] : 0x7ff6a39b9a90
arr[2][0] : 0x7ff6a3a1c4d0
arr[2][1] : 0x7ff6a3a1c4d4
arr[2][2] : 0x7ff6a3a1c4d8

arr[0][0], arr[0][1], arr[0][2] 와 같은 경우에는 int 가 4 byte 이므로 연속적입니다. 배열에 원소가 연속적으로 있으면 앞서 언급한 참조 지역성으로 캐시는 자주 사용되거나 인접한 데이터를 미리 가져와서 처리 속도를 높이고, 캐시 히트율이 높습니다. 즉 효율적입니다.

단, 2차원 배열에서 두번째 항목의 크기가 101010으로 크기에, arr[0] 에서 arr[1] 로 넘어가면 기존 캐시에 저장되어 있는 데이터가 존재하지 않아 캐시 미스가 일어나게 됩니다. 이렇게 불연속적이면 캐시 미스가 일어날 가능성이 높습니다.

행-열 순서

for (int i = 0; i < 1000; i++) {
    for (int j = 0; j < 1000; j++) {
        a[i][j] = 0; // 연속된 메모리 공간 접근.
    }
}

배열의 행은 메모리에 연속적으로 배치되기 때문에, CPU는 한 번 데이터를 가져올 때 여러 인접 요소를 캐시에 저장합니다. 결과적으로 캐시 히트율이 높아지고 성능이 향상됩니다.

열-행 순서

for (int i = 0; i < 1000; i++) {
    for (int j = 0; j < 1000; j++) {
        a[j][i] = 0; // 비연속적인 메모리 공간 접근.
    }
}

반대로, 열 단위로 접근할 경우 메모리의 비연속적으로 띄엄띄엄 참조하게 됩니다. 이는 캐시에 저장된 데이터 블록을 효과적으로 활용하지 못하므로, 캐시 미스가 발생할 확률이 높아집니다.

Sparse Table을 다시 보자!

아래 두 코드는 모두 제가 제출한 최소 공통 조상 문제의 해설코드입니다.

1. http://boj.kr/5974934546014b85adc0b7faaa2ac017
2. http://boj.kr/9cef16a5874343e58676c88709ba6a61

두 코드 변수명, 공백까지 다 같은데 딱 다른 점이 하나 있습니다.

• table을 [size + 1][N + 1] 로 생성한 것이 1번,
• table을 [N + 1][size + 1] 로 생성한 것이 2번입니다.

Sparse Table 에서는 1번 방식으로 표기하였고, 실제 시간도 1번(680 ms)가 2번(724 ms)보다 더 빠른 것을 확인할 수 있었습니다. 단순한 우연일지 확인해보도록 해보겠습니다.

table 초기화

    // 1번 코드
    dfs(1, 1);
    for(int i = 1; i <= size; i++) {
        for(int j = 1; j <= N; j++) {
            dp[i][j] = dp[i - 1][dp[i - 1][j]];
        }
    }

    // 2번 코드
    dfs(1, 1);
    for (int i = 1; i <= size; i++) {
        for (int j = 1; j <= N; j++) {
            dp[j][i] = dp[dp[j][i - 1]][i - 1];
          }
    }

벌써 눈치채신 분들도 있겠지만, 바로 i, j가 1, 2번 코드가 서로 바뀌어 있습니다. sparse table의 초기화 시에는 이동 횟수별로 각 노드가 이동했을 때 도착 노드가 초기화 된 뒤 다음 이동 횟수로 넘어가야 초기화가 되기에, for-loop 자체는 table 생성 순서와 관계 없이 동일해야 됩니다. 그러면 행과 열을 반대로 작성해야 되는데, 그러면 이차원 배열 예시와 동일한 이유로 1번 코드가 2번보다 빠르게 됩니다.

참고

• https://namnamseo.tistory.com/entry/Sparse-Table
• 혼자 공부하는 컴퓨터 구조+운영체제
• https://frogred8.github.io/docs/014_cache_line/