package gc.mda.signal_batch.global.websocket.service;

import gc.mda.signal_batch.global.util.SignalKindCode;
import gc.mda.signal_batch.global.websocket.dto.TrackChunkResponse;
import gc.mda.signal_batch.global.websocket.interceptor.TrackQueryInterceptor;
import gc.mda.signal_batch.domain.vessel.dto.CompactVesselTrack;
import gc.mda.signal_batch.global.websocket.dto.TrackQueryRequest;
import gc.mda.signal_batch.global.websocket.dto.ChunkStats;
import gc.mda.signal_batch.domain.vessel.service.simplification.TrackSimplificationStrategy;
import gc.mda.signal_batch.domain.vessel.service.simplification.TrackSimplificationStrategy.SimplificationLevel;
import lombok.extern.slf4j.Slf4j;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.LineString;
import org.locationtech.jts.io.ParseException;
import org.locationtech.jts.io.WKTReader;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.*;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.function.Consumer;
import gc.mda.signal_batch.global.websocket.dto.QueryStatusUpdate;
import gc.mda.signal_batch.global.websocket.dto.ViewportFilter;
import org.springframework.scheduling.annotation.Async;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 청크 기반 궤적 스트리밍 서비스
 * LineStringM을 압축된 배열 형태로 변환하여 전송
 */
@Slf4j
@Service
public class ChunkedTrackStreamingService {

    private static final org.slf4j.Logger benchmarkLog = org.slf4j.LoggerFactory.getLogger("CACHE_BENCHMARK"); // [BENCHMARK]

    private final JdbcTemplate queryJdbcTemplate;
    private final DataSource queryDataSource;
    @SuppressWarnings("unused")
    private final TrackSimplificationStrategy simplificationStrategy;
    private final ActiveQueryManager activeQueryManager;
    private final TrackQueryInterceptor trackQueryInterceptor;
    private final DailyTrackCacheManager dailyTrackCacheManager;
    private final CacheTrackSimplifier cacheTrackSimplifier;
    private final WKTReader wktReader = new WKTReader();
    @SuppressWarnings("unused")
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    private static final DateTimeFormatter TIMESTAMP_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    private static final int MAX_TRACKS_PER_CHUNK = 500000; // 청크당 최대 트랙 수 (10만 선박 지원)
    private static final int MAX_MESSAGE_SIZE_KB = 1024; // 메시지당 최대 크기 1MB
    private static final int MIN_MESSAGE_SIZE_KB = 256;  // 최소 메시지 크기 256KB
    private static final int DAILY_PAGE_SIZE = 15000; // Daily 테이블 페이지네이션용 (선박 기준)

    // 선박 정보 캐시 (TTL: 1시간)
    private final ConcurrentHashMap<String, VesselInfo> vesselInfoCache = new ConcurrentHashMap<>();
    private static final long VESSEL_CACHE_TTL = 3600_000; // 1시간
    private volatile long lastCacheCleanup = System.currentTimeMillis();

    // 진행률 추적용 변수
    private int estimatedTotalMinutes = 0;
    private long queryStartTime = 0;
    private final Map<String, Integer> processedTimeRanges = new HashMap<>();
    private final AtomicLong pendingBufferSize = new AtomicLong(0);
    private static final long MAX_PENDING_BUFFER = 50 * 1024 * 1024; // 50MB
    @SuppressWarnings("unused")
    private static final long WARNING_BUFFER_THRESHOLD = 40 * 1024 * 1024; // 40MB (80%)

    // 백프레셔 관련 변수
    private final Map<String, BackpressureMetrics> queryMetrics = new ConcurrentHashMap<>();
    @SuppressWarnings("unused")
    private volatile int currentChunkSizeKB = MAX_MESSAGE_SIZE_KB;

    // 쿼리별 취소 플래그 관리
    private final ConcurrentHashMap<String, AtomicBoolean> queryCancelFlags = new ConcurrentHashMap<>();
    
    // track_geom 고정 사용

    public ChunkedTrackStreamingService(
            @Qualifier("queryJdbcTemplate") JdbcTemplate queryJdbcTemplate,
            @Qualifier("queryDataSource") DataSource queryDataSource,
            TrackSimplificationStrategy simplificationStrategy,
            ActiveQueryManager activeQueryManager,
            TrackQueryInterceptor trackQueryInterceptor,
            DailyTrackCacheManager dailyTrackCacheManager,
            CacheTrackSimplifier cacheTrackSimplifier) {
        this.queryJdbcTemplate = queryJdbcTemplate;
        this.queryDataSource = queryDataSource;
        this.simplificationStrategy = simplificationStrategy;
        this.activeQueryManager = activeQueryManager;
        this.trackQueryInterceptor = trackQueryInterceptor;
        this.dailyTrackCacheManager = dailyTrackCacheManager;
        this.cacheTrackSimplifier = cacheTrackSimplifier;
    }

    /**
     * 선박 정보 캐시용 내부 클래스
     */
    private static class VesselInfo {
        String shipName;
        String shipType;
        long cacheTime;

        VesselInfo(String shipName, String shipType) {
            this.shipName = shipName != null ? shipName : "-";
            this.shipType = shipType != null ? shipType : "-";
            this.cacheTime = System.currentTimeMillis();
        }

        boolean isExpired() {
            return System.currentTimeMillis() - cacheTime > VESSEL_CACHE_TTL;
        }
    }

    /**
     * 백프레셔 메트릭스 추적용 내부 클래스
     */
    private static class BackpressureMetrics {
        private final AtomicLong totalBytes = new AtomicLong(0);
        private final AtomicInteger chunkCount = new AtomicInteger(0);
        private final AtomicInteger bufferWarnings = new AtomicInteger(0);
        private final AtomicInteger backpressureEvents = new AtomicInteger(0);
        private volatile long lastWarningTime = 0;
        private volatile int dynamicChunkSizeKB = MAX_MESSAGE_SIZE_KB;
    }

    // [BENCHMARK] 쿼리별 벤치마크 지표 수집용 내부 클래스 - 제거 시 이 클래스 전체 삭제
    private static class QueryBenchmark {
        int cacheHitDays = 0;
        int dbQueryDays = 0;
        int totalTracks = 0;
        int totalPointsBefore = 0;
        int totalPointsAfter = 0;
        int totalBatches = 0;
        int batchesBeforeSimplify = 0;
        long simplifyTimeMs = 0;
        long dbQueryTimeMs = 0;
        // [BENCHMARK] DB 커넥션 세분화 카운터 (기존 단일 dbConnectionCount 대체)
        int connViewportPass1 = 0;   // collectViewportVesselIds DB 쿼리
        int connDailyPages = 0;      // streamDailyTableWithPagination 페이지
        int connVesselInfo = 0;      // preloadVesselInfoWithSessionCache 배치
        int connHourly5min = 0;      // processTableRange (5min/hourly)
        int connTableCheck = 0;      // hasDataInTable 존재 검증
        Integer zoomLevel;

        String determinePath() {
            if (cacheHitDays > 0 && dbQueryDays > 0) return "HYBRID";
            if (cacheHitDays > 0) return "CACHE";
            return "DB";
        }

        int dbConnectionTotal() { // [BENCHMARK]
            return connViewportPass1 + connDailyPages + connVesselInfo + connHourly5min + connTableCheck;
        }

        String toJson(String queryId, int uniqueVessels, long totalElapsedMs) {
            return String.format(
                "{\"queryId\":\"%s\",\"timestamp\":\"%s\",\"path\":\"%s\"," +
                "\"zoomLevel\":%s,\"dateRanges\":%d,\"cacheHitDays\":%d,\"dbQueryDays\":%d," +
                "\"totalTracks\":%d,\"totalPointsBefore\":%d,\"totalPointsAfter\":%d," +
                "\"pointReductionPct\":%d,\"totalBatches\":%d,\"batchesBeforeSimplify\":%d," +
                "\"simplifyTimeMs\":%d,\"dbQueryTimeMs\":%d,\"totalElapsedMs\":%d," +
                "\"dbConnectionTotal\":%d,\"dbConnViewportPass1\":%d,\"dbConnDailyPages\":%d," +
                "\"dbConnVesselInfo\":%d,\"dbConnHourly5min\":%d,\"dbConnTableCheck\":%d," +
                "\"uniqueVessels\":%d}",
                queryId,
                LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),
                determinePath(),
                zoomLevel != null ? zoomLevel.toString() : "null",
                cacheHitDays + dbQueryDays,
                cacheHitDays, dbQueryDays,
                totalTracks, totalPointsBefore, totalPointsAfter,
                totalPointsBefore > 0 ? Math.round((1 - (double) totalPointsAfter / totalPointsBefore) * 100) : 0,
                totalBatches, batchesBeforeSimplify,
                simplifyTimeMs, dbQueryTimeMs, totalElapsedMs,
                dbConnectionTotal(), connViewportPass1, connDailyPages,
                connVesselInfo, connHourly5min, connTableCheck,
                uniqueVessels);
        }
    }

    /**
     * 테이블 범위 처리 - LineStringM을 압축된 배열로 변환
     */
    // 선박 데이터 누적용 내부 클래스
    private static class VesselAccumulator {
        String mmsi;
        String shipName;
        String shipType;
        String shipKindCode;
        List<double[]> geometry = new ArrayList<>(500);
        List<String> timestamps = new ArrayList<>(500);
        List<Double> speeds = new ArrayList<>(500);
        double totalDistance = 0;
        double maxSpeed = 0;
        int pointCount = 0;
    }

    /**
     * 선박 정보 조회 (캐시 우선)
     */
    private VesselInfo getVesselInfo(String mmsi) {
        // 캐시 청소 (10분마다)
        if (System.currentTimeMillis() - lastCacheCleanup > 600_000) {
            cleanupVesselCache();
        }

        // 캐시에서 조회
        VesselInfo cached = vesselInfoCache.get(mmsi);
        if (cached != null && !cached.isExpired()) {
            return cached;
        }

        // DB에서 조회
        try {
            String sql = "SELECT ship_nm, vessel_type FROM signal.t_ais_position " +
                    "WHERE mmsi = ? LIMIT 1";

            VesselInfo info = queryJdbcTemplate.queryForObject(sql,
                    (rs, rowNum) -> new VesselInfo(
                            rs.getString("ship_nm"),
                            rs.getString("vessel_type")
                    ),
                    mmsi
            );

            vesselInfoCache.put(mmsi, info);
            log.debug("Vessel info loaded from DB and cached: {} - {} ({})",
                    mmsi, info.shipName, info.shipType);
            return info;

        } catch (Exception e) {
            log.debug("No vessel info found for {}, using defaults", mmsi);
            VesselInfo defaultInfo = new VesselInfo(null, null);
            vesselInfoCache.put(mmsi, defaultInfo);
            return defaultInfo;
        }
    }

    /**
     * 선박 캐시 정리
     */
    private void cleanupVesselCache() {
        int before = vesselInfoCache.size();
        vesselInfoCache.entrySet().removeIf(entry -> entry.getValue().isExpired());
        int after = vesselInfoCache.size();
        lastCacheCleanup = System.currentTimeMillis();
        log.info("Vessel cache cleanup: {} -> {} entries", before, after);
    }

    /**
     * 선박 정보 배치 조회
     */
    @SuppressWarnings("unused")
    private Map<String, VesselInfo> batchGetVesselInfo(Set<String> vesselIds) {
        Map<String, VesselInfo> result = new HashMap<>();
        List<String> uncachedIds = new ArrayList<>();

        // 캐시 확인
        for (String vesselId : vesselIds) {
            VesselInfo cached = vesselInfoCache.get(vesselId);
            if (cached != null && !cached.isExpired()) {
                result.put(vesselId, cached);
            } else {
                uncachedIds.add(vesselId);
            }
        }

        // 캐시에 없는 것들은 DB에서 배치 조회
        if (!uncachedIds.isEmpty()) {
            try {
                String sql = "SELECT mmsi, ship_nm, vessel_type " +
                        "FROM signal.t_ais_position " +
                        "WHERE mmsi IN (" +
                        String.join(",", Collections.nCopies(uncachedIds.size(), "?")) + ")";

                queryJdbcTemplate.query(sql, rs -> {
                    String vesselId = rs.getString("mmsi");
                    VesselInfo info = new VesselInfo(
                            rs.getString("ship_nm"),
                            rs.getString("vessel_type")
                    );
                    result.put(vesselId, info);
                    vesselInfoCache.put(vesselId, info);
                }, uncachedIds.toArray());

                log.info("Batch loaded {} vessel infos from DB", result.size() - (vesselIds.size() - uncachedIds.size()));
            } catch (Exception e) {
                log.warn("Failed to batch load vessel info: {}", e.getMessage());
                // 기본값 설정
                for (String vesselId : uncachedIds) {
                    if (!result.containsKey(vesselId)) {
                        VesselInfo defaultInfo = new VesselInfo(null, null);
                        result.put(vesselId, defaultInfo);
                        vesselInfoCache.put(vesselId, defaultInfo);
                    }
                }
            }
        }

        return result;
    }

    /**
     * 선박 정보 배치 프리로드 - 세션 캐시 사용 버전
     * 순서: 세션 캐시 → 전역 캐시 → DB 조회
     */
    private void preloadVesselInfoWithSessionCache(Set<String> vesselIds, Map<String, VesselInfo> sessionCache) { // [BENCHMARK] benchmark 파라미터 오버로드 아래 참조
        preloadVesselInfoWithSessionCache(vesselIds, sessionCache, null);
    }

    private void preloadVesselInfoWithSessionCache(Set<String> vesselIds, Map<String, VesselInfo> sessionCache, QueryBenchmark benchmark) { // [BENCHMARK]
        // 세션 캐시와 전역 캐시에 없는 선박만 필터링
        List<String> uncachedIds = vesselIds.stream()
                .filter(id -> {
                    // 세션 캐시에 있으면 스킵
                    if (sessionCache.containsKey(id)) {
                        return false;
                    }
                    // 전역 캐시에 있으면 세션 캐시로 복사하고 스킵
                    VesselInfo globalCached = vesselInfoCache.get(id);
                    if (globalCached != null && !globalCached.isExpired()) {
                        sessionCache.put(id, globalCached);
                        return false;
                    }
                    return true;
                })
                .collect(Collectors.toList());

        if (uncachedIds.isEmpty()) {
            log.debug("All {} vessel infos found in cache (session: {}, global: {})",
                    vesselIds.size(), sessionCache.size(), vesselInfoCache.size());
            return;
        }

        int sessionHits = vesselIds.size() - uncachedIds.size();
        log.info("Batch loading {} vessel infos from DB (cache hits: {})",
                uncachedIds.size(), sessionHits);

        // 배치 크기 제한 (1000개씩)
        int batchSize = 1000;
        for (int i = 0; i < uncachedIds.size(); i += batchSize) {
            List<String> batch = uncachedIds.subList(i, Math.min(i + batchSize, uncachedIds.size()));

            try {
                String placeholders = batch.stream()
                        .map(id -> "?")
                        .collect(Collectors.joining(","));

                String sql = "SELECT mmsi, ship_nm, vessel_type " +
                        "FROM signal.t_ais_position " +
                        "WHERE mmsi IN (" + placeholders + ")";

                Set<String> foundIds = new HashSet<>();

                queryJdbcTemplate.query(sql, rs -> {
                    String visselId = rs.getString("mmsi");
                    VesselInfo info = new VesselInfo(
                            rs.getString("ship_nm"),
                            rs.getString("vessel_type")
                    );
                    // 세션 캐시와 전역 캐시 모두에 저장
                    sessionCache.put(visselId, info);
                    vesselInfoCache.put(visselId, info);
                    foundIds.add(visselId);
                }, batch.toArray());
                if (benchmark != null) benchmark.connVesselInfo++; // [BENCHMARK]

                // DB에 없는 선박은 기본값으로 세션 캐시에 저장 (전역 캐시에는 저장 안함)
                for (String visselId : batch) {
                    if (!foundIds.contains(visselId)) {
                        sessionCache.put(visselId, new VesselInfo(null, null));
                    }
                }

            } catch (Exception e) {
                log.warn("Failed to batch load vessel info: {}", e.getMessage());
                for (String visselId : batch) {
                    if (!sessionCache.containsKey(visselId)) {
                        sessionCache.put(visselId, new VesselInfo(null, null));
                    }
                }
            }
        }
    }

    /**
     * 선박 정보 배치 프리로드 (캐시에 없는 것만 DB에서 조회)
     */
    private void preloadVesselInfoBatch(Set<String> vesselIds) {
        // 캐시에 없는 선박만 필터링
        List<String> uncachedIds = vesselIds.stream()
                .filter(id -> {
                    VesselInfo cached = vesselInfoCache.get(id);
                    return cached == null || cached.isExpired();
                })
                .collect(Collectors.toList());

        if (uncachedIds.isEmpty()) {
            log.debug("All {} vessel infos found in cache", vesselIds.size());
            return;
        }

        log.info("Batch loading {} vessel infos from DB (cached: {})",
                uncachedIds.size(), vesselIds.size() - uncachedIds.size());

        // 배치 크기 제한 (1000개씩)
        int batchSize = 1000;
        for (int i = 0; i < uncachedIds.size(); i += batchSize) {
            List<String> batch = uncachedIds.subList(i, Math.min(i + batchSize, uncachedIds.size()));

            try {
                // IN 절용 파라미터 생성
                String placeholders = batch.stream()
                        .map(id -> "?")
                        .collect(Collectors.joining(","));

                String sql = "SELECT mmsi, ship_nm, vessel_type " +
                        "FROM signal.t_ais_position " +
                        "WHERE mmsi IN (" + placeholders + ")";

                Set<String> foundIds = new HashSet<>();

                queryJdbcTemplate.query(sql, rs -> {
                    String vesselId = rs.getString("mmsi");
                    VesselInfo info = new VesselInfo(
                            rs.getString("ship_nm"),
                            rs.getString("vessel_type")
                    );
                    vesselInfoCache.put(vesselId, info);
                    foundIds.add(vesselId);
                }, batch.toArray());

                // DB에 없는 선박은 기본값으로 캐시
                for (String vesselId : batch) {
                    if (!foundIds.contains(vesselId)) {
                        vesselInfoCache.put(vesselId, new VesselInfo(null, null));
                    }
                }

            } catch (Exception e) {
                log.warn("Failed to batch load vessel info: {}", e.getMessage());
                // 실패 시 기본값으로 캐시
                for (String vesselId : batch) {
                    if (!vesselInfoCache.containsKey(vesselId)) {
                        vesselInfoCache.put(vesselId, new VesselInfo(null, null));
                    }
                }
            }
        }
    }

    /**
     * 2-pass 뷰포트 필터링 - Pass 1: 뷰포트에 교차하는 선박 ID 수집
     * 전체 쿼리 시간 범위에서 뷰포트 영역을 지나는 모든 선박을 식별하여,
     * Pass 2에서 해당 선박의 전체 항적(뷰포트 밖 포함)을 조회할 수 있도록 함
     */
    private Set<String> collectViewportVesselIds(Map<TableStrategy, List<TimeRange>> strategyMap, TrackQueryRequest request) { // [BENCHMARK] benchmark 파라미터 오버로드 아래 참조
        return collectViewportVesselIds(strategyMap, request, null);
    }

    private Set<String> collectViewportVesselIds(Map<TableStrategy, List<TimeRange>> strategyMap, TrackQueryRequest request, QueryBenchmark benchmark) { // [BENCHMARK]
        ViewportFilter viewport = request.getViewport();
        if (viewport == null || !viewport.isValid()) {
            return null;
        }

        long startMs = System.currentTimeMillis();
        Set<String> vesselIds = new HashSet<>();
        int cacheHitDays = 0;
        int dbQueryDays = 0;

        for (Map.Entry<TableStrategy, List<TimeRange>> entry : strategyMap.entrySet()) {
            TableStrategy strategy = entry.getKey();
            String tableName = strategy.getTableName();

            for (TimeRange range : entry.getValue()) {
                // Daily 전략이고 캐시에 있으면 → 메모리에서 뷰포트 체크 (DB 커넥션 절약)
                if (strategy == TableStrategy.DAILY && dailyTrackCacheManager.isEnabled()) {
                    LocalDate rangeDate = range.getStart().toLocalDate();
                    if (dailyTrackCacheManager.isCached(rangeDate)) {
                        List<CompactVesselTrack> cachedTracks = dailyTrackCacheManager.getCachedTracks(
                                rangeDate, viewport.getMinLon(), viewport.getMinLat(),
                                viewport.getMaxLon(), viewport.getMaxLat());
                        for (CompactVesselTrack track : cachedTracks) {
                            vesselIds.add(track.getVesselId());
                        }
                        cacheHitDays++;
                        continue;
                    }
                }

                // DB 쿼리로 뷰포트 교차 선박 수집
                dbQueryDays++;
                StringBuilder sql = new StringBuilder();
                sql.append("SELECT DISTINCT mmsi FROM ").append(tableName);
                sql.append(" WHERE time_bucket >= ? AND time_bucket < ?");
                sql.append(" AND public.ST_Intersects(track_geom, public.ST_MakeEnvelope(?, ?, ?, ?, 4326))");

                if (request.getMinAvgSpeed() != null) {
                    sql.append(" AND avg_speed >= ").append(request.getMinAvgSpeed());
                }
                if (request.getMaxAvgSpeed() != null) {
                    sql.append(" AND avg_speed <= ").append(request.getMaxAvgSpeed());
                }

                try (Connection conn = queryDataSource.getConnection();
                     PreparedStatement ps = conn.prepareStatement(sql.toString())) {
                    ps.setTimestamp(1, Timestamp.valueOf(range.getStart()));
                    ps.setTimestamp(2, Timestamp.valueOf(range.getEnd()));
                    ps.setDouble(3, viewport.getMinLon());
                    ps.setDouble(4, viewport.getMinLat());
                    ps.setDouble(5, viewport.getMaxLon());
                    ps.setDouble(6, viewport.getMaxLat());

                    try (ResultSet rs = ps.executeQuery()) {
                        while (rs.next()) {
                            vesselIds.add(rs.getString("mmsi"));
                        }
                    }
                    if (benchmark != null) benchmark.connViewportPass1++; // [BENCHMARK]
                } catch (SQLException e) {
                    log.error("Error collecting viewport vessel IDs from {}: {}", tableName, e.getMessage());
                }
            }
        }

        long elapsed = System.currentTimeMillis() - startMs;
        log.info("2-pass viewport filter Pass 1: {} vessels, cacheHit={} days, dbQuery={} days, {}ms",
                vesselIds.size(), cacheHitDays, dbQueryDays, elapsed);
        return vesselIds;
    }

    private List<CompactVesselTrack> processTableRange(TrackQueryRequest request, TableStrategy strategy, TimeRange range) {
        return processTableRange(request, strategy, range, null, null);
    }

    private List<CompactVesselTrack> processTableRange(TrackQueryRequest request, TableStrategy strategy, TimeRange range, String sessionId) {
        return processTableRange(request, strategy, range, sessionId, null);
    }

    private List<CompactVesselTrack> processTableRange(TrackQueryRequest request, TableStrategy strategy, TimeRange range, String sessionId, Set<String> viewportVesselIds) {
        Map<String, VesselAccumulator> vesselMap = new HashMap<>(20000); // 예상 선박 수

        String tableName = strategy.getTableName();

        // 간소화 레벨 결정
        SimplificationLevel simplificationLevel = determineSimplificationLevel(request,
                new TimeChunk(range.getStart(), range.getEnd()));
        log.info("Using {} table with simplification level {} for range [{} - {}]",
                strategy, simplificationLevel, range.getStart(), range.getEnd());

        String sql = buildRangeQuery(tableName, request, range, simplificationLevel, viewportVesselIds);

        // Connection을 직접 사용하여 스트리밍 처리
        try (Connection conn = queryDataSource.getConnection();
             PreparedStatement ps = conn.prepareStatement(sql)) {

            // 파라미터 바인딩
            int paramIndex = 1;
            ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getStart()));
            ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getEnd()));

            // 2-pass 뷰포트 필터: vessel ID 배열 또는 기존 viewport 좌표 바인딩
            if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
                Array vesselArray = conn.createArrayOf("text", viewportVesselIds.toArray(new String[0]));
                ps.setArray(paramIndex++, vesselArray);
            } else if (request.getViewport() != null) {
                ViewportFilter vp = request.getViewport();
                ps.setDouble(paramIndex++, vp.getMinLon());
                ps.setDouble(paramIndex++, vp.getMinLat());
                ps.setDouble(paramIndex++, vp.getMaxLon());
                ps.setDouble(paramIndex++, vp.getMaxLat());
            }

            // PostgreSQL 스트리밍 설정
            // FetchSize 최적화 - 메모리가 충분하므로 기존 유지
            ps.setFetchSize(5000);

            try (ResultSet rs = ps.executeQuery()) {
                int trackCount = 0;
                int vesselCount = 0;

                while (rs.next() && trackCount < MAX_TRACKS_PER_CHUNK) {
                    // 세션 연결 끊김 체크 (1000개마다)
                    if (sessionId != null && trackCount % 1000 == 0) {
                        if (isQueryCancelled(sessionId)) {
                            log.warn("Query cancelled by disconnect: sessionId={}, table={}, trackCount={}",
                                    sessionId, tableName, trackCount);
                            return Collections.emptyList();
                        }
                    }
                    String mmsi = rs.getString("mmsi");
                    String vesselId = mmsi;

                    // LineStringM 파싱
                    String trackGeomWkt = rs.getString("track_geom");
                    Timestamp timeBucket = rs.getTimestamp("time_bucket");

                    // start_position, end_position에서 시간 정보 추출
                    String startTimeStr = null;
                    String endTimeStr = null;
                    try {
                        startTimeStr = rs.getString("start_time");
                        endTimeStr = rs.getString("end_time");
                    } catch (SQLException ignored) {
                        // 5min 테이블은 이 필드가 없을 수 있음
                    }

                    // 선박 객체는 geometry가 비어있어도 생성 (선박 누락 방지)
                    VesselAccumulator accumulator = vesselMap.get(vesselId);
                    if (accumulator == null) {
                        vesselCount++;
                        accumulator = new VesselAccumulator();
                        accumulator.mmsi = mmsi;

                        VesselInfo vesselInfo = getVesselInfo(mmsi);
                        accumulator.shipName = vesselInfo.shipName;
                        accumulator.shipType = vesselInfo.shipType;
                        accumulator.shipKindCode = SignalKindCode.resolve(vesselInfo.shipType, null).getCode();

                        vesselMap.put(vesselId, accumulator);
                    }

                    if (trackGeomWkt != null && !trackGeomWkt.isEmpty() && !"LINESTRING EMPTY".equals(trackGeomWkt)) {
                        try {
                            LineString lineString = (LineString) wktReader.read(trackGeomWkt);

                            // 빈 LineString은 건너뛰되, 선박 객체는 이미 생성됨
                            if (lineString.getNumPoints() == 0) {
                                trackCount++;
                                continue;
                            }

                            // 좌표 배열로 변환
                            List<double[]> geometry = new ArrayList<>();
                            List<String> timestamps = new ArrayList<>();
                            List<Double> speeds = new ArrayList<>();

                            // 기준 시간 결정 (start_time이 있으면 사용, 없으면 timeBucket 사용)
                            LocalDateTime baseTime = timeBucket.toLocalDateTime();
                            if (startTimeStr != null && !startTimeStr.isEmpty()) {
                                try {
                                    baseTime = LocalDateTime.parse(startTimeStr, TIMESTAMP_FORMATTER);
                                } catch (Exception e) {
                                    log.debug("Failed to parse start_time: {}", startTimeStr);
                                }
                            }

                            Coordinate prevCoord = null;
                            long prevTimeMillis = 0;

                            for (int i = 0; i < lineString.getNumPoints(); i++) {
                                Coordinate coord = lineString.getCoordinateN(i);
                                geometry.add(new double[]{coord.x, coord.y});

                                // Unix timestamp (초 단위) - String으로 저장
                                long unixTimestamp = (long)coord.getM();
                                timestamps.add(String.valueOf(unixTimestamp));

                                // 속도 계산 (두 점 사이의 거리와 시간차를 이용)
                                double speed = 0.0;
                                long currentTimeMillis;

                                // Unix timestamp (초 단위를 밀리초로 변환)
                                currentTimeMillis = (long)coord.getM() * 1000;

                                if (prevCoord != null && i > 0) {
                                    double distance = calculateDistance(prevCoord, coord); // 해리(nm)
                                    double timeDiff = (currentTimeMillis - prevTimeMillis) / 3600000.0; // 시간(hour)
                                    if (timeDiff > 0) {
                                        speed = distance / timeDiff; // knots
                                    }
                                }
                                speeds.add(speed);

                                prevCoord = coord;
                                prevTimeMillis = currentTimeMillis;
                            }

                            // 데이터 직접 추가 (성능 개선)
                            accumulator.geometry.addAll(geometry);
                            accumulator.timestamps.addAll(timestamps);
                            accumulator.speeds.addAll(speeds);
                            accumulator.totalDistance += rs.getDouble("distance_nm");
                            accumulator.maxSpeed = Math.max(accumulator.maxSpeed, rs.getDouble("max_speed"));
                            accumulator.pointCount += geometry.size();

                        } catch (ParseException e) {
                            log.error("Error parsing geometry: {}", e.getMessage());
                        }
                    }
                    trackCount++;
                }

                // 시간 범위 계산
//                long rangeMinutes = java.time.Duration.between(range.getStart(), range.getEnd()).toMinutes();
//                int expectedBuckets = (int) Math.ceil(rangeMinutes / 5.0);

//                log.info("[{}] Range [{} - {}] = {} minutes, {} buckets expected",
//                    tableName, range.getStart(), range.getEnd(), rangeMinutes, expectedBuckets);
//                log.info("[{}] Processed {} tracks for {} unique vessels (avg {} tracks/vessel)",
//                    tableName, trackCount, vesselCount,
//                    vesselCount > 0 ? String.format("%.1f", (double)trackCount/vesselCount) : "0.0");
//                log.debug("[{}] Vessel count in this chunk: {} (cumulative in vesselMap: {})",
//                    tableName, vesselCount, vesselMap.size());
            }
        } catch (SQLException e) {
            log.error("Error processing table range: {}", e.getMessage());
        }

        // VesselAccumulator를 CompactVesselTrack로 변환
        return vesselMap.entrySet().stream()
                .map(entry -> {
                    String vesselId = entry.getKey();
                    VesselAccumulator acc = entry.getValue();

                    // 평균속도 계산 (전체 궤적 기반)
                    double avgSpeed = 0.0;
                    if (acc.totalDistance > 0 && acc.timestamps.size() > 1) {
                        String firstTs = acc.timestamps.get(0);
                        String lastTs = acc.timestamps.get(acc.timestamps.size() - 1);
                        
                        LocalDateTime startTime;
                        LocalDateTime endTime;
                        
                        // Unix timestamp 감지 (10자리 이상 숫자)
                        if (firstTs.matches("\\d{10,}")) {
                            startTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(firstTs)), 
                                java.time.ZoneId.systemDefault());
                            endTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(lastTs)), 
                                java.time.ZoneId.systemDefault());
                        } else {
                            startTime = LocalDateTime.parse(firstTs, TIMESTAMP_FORMATTER);
                            endTime = LocalDateTime.parse(lastTs, TIMESTAMP_FORMATTER);
                        }
                        
                        double hours = Duration.between(startTime, endTime).toMinutes() / 60.0;
                        if (hours > 0) {
                            avgSpeed = acc.totalDistance / hours;
                        }
                    }

                    return CompactVesselTrack.builder()
                            .vesselId(vesselId)
                            .nationalCode(acc.mmsi != null && acc.mmsi.length() >= 3 ? acc.mmsi.substring(0, 3) : null)
                            .shipName(acc.shipName)
                            .shipType(acc.shipType)
                            .shipKindCode(acc.shipKindCode)
                            .geometry(acc.geometry)
                            .timestamps(acc.timestamps)
                            .speeds(acc.speeds)
                            .totalDistance(acc.totalDistance)
                            .avgSpeed(avgSpeed)
                            .maxSpeed(acc.maxSpeed)
                            .pointCount(acc.pointCount)
                            .build();
                })
                .collect(Collectors.toList());
    }

    /**
     * 쿼리를 테이블 전략별로 처리
     */
    public List<TrackChunkResponse> processQueryInChunks(TrackQueryRequest request, String queryId) {
        log.info("Processing chunked query: {}", queryId);
        queryStartTime = System.currentTimeMillis();
        processedTimeRanges.clear();

        // 시간 범위별 테이블 전략 분할
        Map<TableStrategy, List<TimeRange>> strategyMap = splitTimeRangeByStrategy(
                request.getStartTime(), request.getEndTime()
        );

        log.info("Query {} using strategies: {}", queryId, strategyMap.keySet());

        // 2-pass 뷰포트 필터링: 전체 시간 범위에서 뷰포트 교차 선박 수집
        Set<String> viewportVesselIds = collectViewportVesselIds(strategyMap, request);

        List<TrackChunkResponse> responses = new ArrayList<>();
        int globalChunkIndex = 0;
        Set<String> uniqueVesselIds = new HashSet<>();

        // 테이블 전략별 처리 (daily → hourly → 5min 순서)
        for (TableStrategy strategy : new TableStrategy[]{TableStrategy.DAILY, TableStrategy.HOURLY, TableStrategy.FIVE_MINUTE}) {
            if (!strategyMap.containsKey(strategy)) continue;

            List<TimeRange> ranges = strategyMap.get(strategy);
            log.info("Processing {} strategy with {} ranges", strategy, ranges.size());

            // Daily 테이블의 경우 캐시 우선 → 페이지네이션 폴백
            if (strategy == TableStrategy.DAILY) {
                for (TimeRange range : ranges) {
                    try {
                        List<CompactVesselTrack> compactTracks;
                        LocalDate rangeDate = range.getStart().toLocalDate();

                        // 캐시 히트 시 메모리에서 가져옴
                        if (dailyTrackCacheManager.isEnabled() && dailyTrackCacheManager.isCached(rangeDate)) {
                            // viewportVesselIds가 있으면 vessel ID 필터만 (공간 필터 재적용 금지)
                            if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
                                compactTracks = dailyTrackCacheManager.getCachedTracks(rangeDate).stream()
                                        .filter(t -> viewportVesselIds.contains(t.getVesselId()))
                                        .collect(Collectors.toList());
                            } else {
                                compactTracks = dailyTrackCacheManager.getCachedTracks(rangeDate);
                            }
                            log.info("Daily cache HIT for {} in processQueryInChunks: {} tracks", rangeDate, compactTracks.size());
                        } else {
                            compactTracks = processDailyTableWithPagination(request, range, null, viewportVesselIds);
                        }

                        if (!compactTracks.isEmpty()) {
                            // 메시지 크기로 분할
                            List<List<CompactVesselTrack>> batches = splitByMessageSize(compactTracks);
                            for (List<CompactVesselTrack> batch : batches) {
                                TrackChunkResponse response = new TrackChunkResponse();
                                response.setQueryId(queryId);
                                response.setChunkIndex(globalChunkIndex++);
                                response.setIsLastChunk(false); // 나중에 설정
                                response.setTotalChunks(-1); // 마지막 청크에서 설정
                                response.setCompactTracks(batch);
                                // 처리된 시간 계산
                                String rangeKey = "daily_" + range.getStart();
                                processedTimeRanges.put(rangeKey, (int)Duration.between(range.getStart(), range.getEnd()).toMinutes());
                                int processedMinutes = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
                                response.setStats(createChunkStats(batch, uniqueVesselIds, processedMinutes));
                                responses.add(response);

                                // 유니크 선박 추가
                                batch.forEach(track -> uniqueVesselIds.add(track.getVesselId()));

                                // 배치 크기 로그 (Daily에도 추가)
                                int batchSize = batch.stream().mapToInt(t -> estimateTrackSize(t)).sum() / 1024;
                                log.debug("[{}] Batch {} size: {}KB, tracks: {}",
                                        strategy, response.getChunkIndex(), batchSize, batch.size());
                            }
                        }
                    } catch (Exception e) {
                        log.error("Error processing daily range {}: {}", range, e.getMessage());
                    }
                }
            } else {
                // Hourly/5min 테이블은 청크로 분할
                List<TimeChunk> chunks = divideRangesIntoChunks(ranges, strategy);

                for (TimeChunk chunk : chunks) {
                    try {
                        List<CompactVesselTrack> compactTracks = processTableRange(request, strategy,
                                new TimeRange(chunk.start, chunk.end), null, viewportVesselIds);

                        if (!compactTracks.isEmpty()) {
                            // 메시지 크기로 분할 (5min/hourly에도 적용)
                            List<List<CompactVesselTrack>> batches = splitByMessageSize(compactTracks);
                            for (List<CompactVesselTrack> batch : batches) {
                                TrackChunkResponse response = new TrackChunkResponse();
                                response.setQueryId(queryId);
                                response.setChunkIndex(globalChunkIndex++);
                                response.setIsLastChunk(false);
                                response.setTotalChunks(-1); // 마짉 청크에서 설정
                                response.setCompactTracks(batch);
                                // 처리된 시간 업데이트
                                String chunkKey = strategy + "_" + chunk.start;
                                processedTimeRanges.put(chunkKey, (int)Duration.between(chunk.start, chunk.end).toMinutes());
                                int processedMin = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
                                response.setStats(createChunkStats(batch, uniqueVesselIds, processedMin));
                                responses.add(response);

                                // 유니크 선박 추가
                                batch.forEach(track -> uniqueVesselIds.add(track.getVesselId()));

                                // 배치 크기 로그
                                int batchSize = batch.stream().mapToInt(t -> estimateTrackSize(t)).sum() / 1024;
                                log.debug("[{}] Batch {} size: {}KB, tracks: {}",
                                        strategy, response.getChunkIndex(), batchSize, batch.size());
                            }
                        }
                    } catch (Exception e) {
                        log.error("Error processing chunk {}: {}", chunk, e.getMessage());
                    }
                }
            }
        }

        // 마지막 청크 표시
        if (!responses.isEmpty()) {
            responses.get(responses.size() - 1).setIsLastChunk(true);

            // 총 선박 수 계산
            responses.forEach(response -> {
                if (response.getCompactTracks() != null) {
                    response.getCompactTracks().forEach(track -> uniqueVesselIds.add(track.getVesselId()));
                }
            });
            log.info("Query {} completed: Total {} chunks, {} unique vessels processed",
                    queryId, responses.size(), uniqueVesselIds.size());
        }

        // 전체 응답 크기 체크
        long totalResponseSize = responses.stream()
                .mapToLong(r -> r.getCompactTracks().stream()
                        .mapToInt(t -> estimateTrackSize(t))
                        .sum())
                .sum();

        log.info("Query {} total response size: {} MB across {} chunks",
                queryId, totalResponseSize / (1024 * 1024), responses.size());

        return responses;
    }

    /**
     * 비동기 스트리밍 메서드 (컨트롤러에서 호출)
     */
    @Async("trackStreamingExecutor")
    public void streamChunkedTracks(TrackQueryRequest request,
                                    String queryId,
                                    String sessionId,
                                    Consumer<TrackChunkResponse> chunkConsumer,
                                    Consumer<QueryStatusUpdate> statusConsumer) {
        boolean slotAcquired = false;
        try {
            // 글로벌 동시 쿼리 슬롯 즉시 획득 시도
            slotAcquired = activeQueryManager.tryAcquireQuerySlotImmediate(queryId);
            if (!slotAcquired) {
                // 대기열 진입 → 2초 간격으로 QUEUED 상태 전송하며 슬롯 대기
                log.info("Query {} entering wait queue: active={}/{}", queryId,
                        activeQueryManager.getGlobalActiveQueryCount(), activeQueryManager.getMaxConcurrentGlobal());

                int maxWaitSeconds = 120; // 최대 2분 대기
                long deadline = System.currentTimeMillis() + (maxWaitSeconds * 1000L);
                activeQueryManager.getWaitingQueue().offer(queryId);
                try {
                    int heapRetryCount = 0;
                    while (System.currentTimeMillis() < deadline) {
                        slotAcquired = activeQueryManager.tryAcquireQuerySlot(queryId);
                        if (slotAcquired) {
                            if (activeQueryManager.isHeapPressureHigh()) {
                                activeQueryManager.releaseQuerySlot(queryId);
                                slotAcquired = false;
                                heapRetryCount++;
                                if (heapRetryCount <= 3) {
                                    log.warn("Query {} slot acquired but heap pressure high, waiting 3s before retry (attempt {})",
                                            queryId, heapRetryCount);
                                }
                                Thread.sleep(3000);  // GC 회복 대기 (busy-spin 방지)
                                continue;
                            } else {
                                break;
                            }
                        }
                        // QUEUED 상태 전송
                        int position = activeQueryManager.getQueuePosition(queryId);
                        int totalInQueue = activeQueryManager.getQueueSize();
                        QueryStatusUpdate queueStatus = new QueryStatusUpdate(
                                queryId, "QUEUED",
                                String.format("Queue position: %d/%d. Waiting for available slot...", position, totalInQueue),
                                0.0
                        );
                        queueStatus.setQueuePosition(position);
                        queueStatus.setTotalInQueue(totalInQueue);
                        statusConsumer.accept(queueStatus);
                    }
                } finally {
                    activeQueryManager.getWaitingQueue().remove(queryId);
                }

                if (!slotAcquired) {
                    log.warn("Query {} timed out in queue after {}s", queryId, maxWaitSeconds);
                    statusConsumer.accept(new QueryStatusUpdate(
                            queryId, "ERROR",
                            "Server is busy. Queue timeout exceeded. Please retry later.",
                            0.0
                    ));
                    return;
                }
            }

            // 취소 플래그 등록
            AtomicBoolean cancelFlag = new AtomicBoolean(false);
            queryCancelFlags.put(queryId, cancelFlag);

            log.info("Starting chunked streaming for query: {}", queryId);
            queryStartTime = System.currentTimeMillis();
            processedTimeRanges.clear();

            // [BENCHMARK] 벤치마크 지표 초기화
            QueryBenchmark benchmark = new QueryBenchmark();
            benchmark.zoomLevel = request.getZoomLevel();

            // 백프레셔 메트릭스 초기화
            BackpressureMetrics metrics = new BackpressureMetrics();
            queryMetrics.put(queryId, metrics);
            pendingBufferSize.set(0);

            log.info("[BACKPRESSURE] Query {} initialized - Buffer size: 0, Max buffer: {}MB",
                    queryId, MAX_PENDING_BUFFER / (1024 * 1024));

            // 시간 범위별 테이블 전략 분할
            Map<TableStrategy, List<TimeRange>> strategyMap = splitTimeRangeByStrategy(
                    request.getStartTime(), request.getEndTime(), benchmark // [BENCHMARK]
            );

            // 전체 시간 계산
            estimatedTotalMinutes = (int)Duration.between(request.getStartTime(), request.getEndTime()).toMinutes();
            log.info("Total time range: {} minutes", estimatedTotalMinutes);

            // 2-pass 뷰포트 필터링: 전체 시간 범위에서 뷰포트 교차 선박 수집
            Set<String> viewportVesselIds = collectViewportVesselIds(strategyMap, request, benchmark); // [BENCHMARK]

            int globalChunkIndex = 0;
            int totalVessels = 0;
            Set<String> uniqueVesselIds = new HashSet<>();

            // 테이블 전략별 처리 (daily → hourly → 5min 순서)
            for (TableStrategy strategy : new TableStrategy[]{TableStrategy.DAILY, TableStrategy.HOURLY, TableStrategy.FIVE_MINUTE}) {
                if (!strategyMap.containsKey(strategy)) continue;

                // 취소 확인
                if (cancelFlag.get()) {
                    log.info("Query {} cancelled before processing strategy {}", queryId, strategy);
                    statusConsumer.accept(new QueryStatusUpdate(queryId, "CANCELLED", "Query cancelled by user", 0.0));
                    return;
                }

                List<TimeRange> ranges = strategyMap.get(strategy);
                log.info("Processing {} strategy with {} ranges", strategy, ranges.size());

                if (strategy == TableStrategy.DAILY) {
                    // Daily는 기존 방식 유지 (이미 일 단위)
                    processDailyStrategy(ranges, request, queryId, chunkConsumer, statusConsumer,
                            globalChunkIndex, uniqueVesselIds, viewportVesselIds, benchmark);
                    globalChunkIndex = getCurrentChunkIndex();
                } else {
                    // Hourly/5min은 6시간 단위로 그룹화하여 처리
                    Map<String, List<TimeRange>> timeGroups = groupRangesByTimeWindow(ranges, 6);

                    for (Map.Entry<String, List<TimeRange>> groupEntry : timeGroups.entrySet()) {
                        String groupKey = groupEntry.getKey();
                        List<TimeRange> groupRanges = groupEntry.getValue();
                        log.info("[{}] Processing time window {} with {} ranges", strategy, groupKey, groupRanges.size());

                        // 시간 그룹 데이터를 병합
                        Map<String, VesselAccumulator> mergedMap = new HashMap<>(20000);
                        LocalDateTime baseTime = null;

                        for (TimeRange range : groupRanges) {
                            try {
                                // 첫 범위의 시작 시간을 기준으로 설정
                                if (baseTime == null) {
                                    baseTime = range.getStart();
                                }

                                List<CompactVesselTrack> compactTracks = processTableRangeWithBaseTime(
                                        request, strategy, range, baseTime, viewportVesselIds);
                                if (benchmark != null) benchmark.connHourly5min++; // [BENCHMARK]

                                // 선박별로 볕합
                                for (CompactVesselTrack track : compactTracks) {
                                    String vesselId = track.getVesselId();
                                    VesselAccumulator accumulator = mergedMap.get(vesselId);

                                    if (accumulator == null) {
                                        accumulator = new VesselAccumulator();
                                        accumulator.mmsi = track.getVesselId();

                                        VesselInfo vesselInfo = getVesselInfo(track.getVesselId());
                                        accumulator.shipName = vesselInfo.shipName;
                                        accumulator.shipType = vesselInfo.shipType;
                                        accumulator.shipKindCode = SignalKindCode.resolve(vesselInfo.shipType, null).getCode();

                                        mergedMap.put(vesselId, accumulator);
                                    }

                                    // 데이터 병합
                                    accumulator.geometry.addAll(track.getGeometry());
                                    accumulator.timestamps.addAll(track.getTimestamps());
                                    accumulator.speeds.addAll(track.getSpeeds());
                                    accumulator.totalDistance += track.getTotalDistance();
                                    accumulator.maxSpeed = Math.max(accumulator.maxSpeed, track.getMaxSpeed());
                                    accumulator.pointCount += track.getPointCount();
                                }
                            } catch (Exception e) {
                                log.error("Error processing {} range {}: {}", strategy, range, e.getMessage());
                            }
                        }

                        // 병합된 데이터를 청크로 분할하여 전송
                        if (!mergedMap.isEmpty()) {
                            List<CompactVesselTrack> mergedTracks = mergedMap.entrySet().stream()
                                    .map(entry -> {
                                        String vesselId = entry.getKey();
                                        VesselAccumulator acc = entry.getValue();

                                        double avgSpeed = 0.0;
                                        if (acc.totalDistance > 0 && acc.timestamps.size() > 1) {
                                            String firstTs = acc.timestamps.get(0);
                                            String lastTs = acc.timestamps.get(acc.timestamps.size() - 1);
                                            
                                            LocalDateTime start;
                                            LocalDateTime end;
                                            
                                            // Unix timestamp 감지 (10자리 이상 숫자)
                                            if (firstTs.matches("\\d{10,}")) {
                                                start = LocalDateTime.ofInstant(
                                                    java.time.Instant.ofEpochSecond(Long.parseLong(firstTs)), 
                                                    java.time.ZoneId.systemDefault());
                                                end = LocalDateTime.ofInstant(
                                                    java.time.Instant.ofEpochSecond(Long.parseLong(lastTs)), 
                                                    java.time.ZoneId.systemDefault());
                                            } else {
                                                start = LocalDateTime.parse(firstTs, TIMESTAMP_FORMATTER);
                                                end = LocalDateTime.parse(lastTs, TIMESTAMP_FORMATTER);
                                            }
                                            
                                            double hours = Duration.between(start, end).toMinutes() / 60.0;
                                            if (hours > 0) {
                                                avgSpeed = acc.totalDistance / hours;
                                            }
                                        }

                                        return CompactVesselTrack.builder()
                                                .vesselId(vesselId)
                                                .nationalCode(acc.mmsi != null && acc.mmsi.length() >= 3 ? acc.mmsi.substring(0, 3) : null)
                                                .shipName(acc.shipName)
                                                .shipType(acc.shipType)
                                                .shipKindCode(acc.shipKindCode)
                                                .geometry(acc.geometry)
                                                .timestamps(acc.timestamps)
                                                .speeds(acc.speeds)
                                                .totalDistance(acc.totalDistance)
                                                .avgSpeed(avgSpeed)
                                                .maxSpeed(acc.maxSpeed)
                                                .pointCount(acc.pointCount)
                                                .build();
                                    })
                                    .collect(Collectors.toList());

                            // 전체 포인트 통계 계산
                            int totalOriginalPoints = mergedTracks.stream()
                                    .mapToInt(t -> t.getPointCount())
                                    .sum();

                            log.info("[{}] Time window {} - Merged {} vessels, Total {} points",
                                    strategy, groupKey, mergedTracks.size(), totalOriginalPoints);

                            List<List<CompactVesselTrack>> batches = splitByMessageSize(mergedTracks, queryId);

                            for (List<CompactVesselTrack> batch : batches) {
                                TrackChunkResponse response = new TrackChunkResponse();
                                response.setQueryId(queryId);
                                response.setChunkIndex(globalChunkIndex++);
                                response.setIsLastChunk(false);
                                response.setTotalChunks(-1); // 마짉 청크에서 설정
                                response.setCompactTracks(batch);
                                // 처리된 시간 추가
                                String timeKey = groupKey + "_" + strategy;
                                processedTimeRanges.put(timeKey, groupRanges.stream()
                                        .mapToInt(r -> (int)Duration.between(r.getStart(), r.getEnd()).toMinutes())
                                        .sum());
                                int currentProcessedMin = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
                                response.setStats(createChunkStats(batch, uniqueVesselIds, currentProcessedMin));

                                // 버퍼 크기 계산 및 추가
                                int chunkSize = batch.stream().mapToInt(t -> estimateTrackSize(t)).sum();
                                long currentBufferSize = pendingBufferSize.addAndGet(chunkSize);
                                metrics.totalBytes.addAndGet(chunkSize);
                                metrics.chunkCount.incrementAndGet();

                                // 버퍼 사용률 로그
                                double bufferUsage = (double) currentBufferSize / MAX_PENDING_BUFFER * 100;
                                if (bufferUsage > 80 && System.currentTimeMillis() - metrics.lastWarningTime > 5000) {
                                    metrics.bufferWarnings.incrementAndGet();
                                    metrics.lastWarningTime = System.currentTimeMillis();
                                    log.warn("[BACKPRESSURE] Query {} - Buffer usage high: {}% ({} MB / {} MB)",
                                            queryId, String.format("%.1f", bufferUsage), currentBufferSize / (1024 * 1024),
                                            MAX_PENDING_BUFFER / (1024 * 1024));
                                }

                                // 버퍼가 가듍 찬 경우 대기 및 동적 청크 크기 조절
                                int backpressureWaitCount = 0;
                                while (pendingBufferSize.get() > MAX_PENDING_BUFFER) {
                                    if (backpressureWaitCount == 0) {
                                        metrics.backpressureEvents.incrementAndGet();
                                        log.warn("[BACKPRESSURE] Query {} - Buffer full! Waiting for buffer to drain. Current: {} MB",
                                                queryId, pendingBufferSize.get() / (1024 * 1024));
                                    }
                                    Thread.sleep(50);
                                    backpressureWaitCount++;

                                    // 500ms 이상 대기 시 청크 크기 감소
                                    if (backpressureWaitCount > 10 && metrics.dynamicChunkSizeKB > MIN_MESSAGE_SIZE_KB) {
                                        int oldSize = metrics.dynamicChunkSizeKB;
                                        metrics.dynamicChunkSizeKB = Math.max(MIN_MESSAGE_SIZE_KB,
                                                metrics.dynamicChunkSizeKB - 512);
                                        log.info("[BACKPRESSURE] Query {} - Reducing chunk size: {} KB -> {} KB",
                                                queryId, oldSize, metrics.dynamicChunkSizeKB);
                                    }
                                }

                                if (backpressureWaitCount > 0) {
                                    log.info("[BACKPRESSURE] Query {} - Buffer drained after {} ms wait",
                                            queryId, backpressureWaitCount * 50);
                                }

                                // 전송 및 버퍼 추적 (전송 완료 후 즉시 감소)
                                try {
                                    chunkConsumer.accept(response);
                                } finally {
                                    pendingBufferSize.addAndGet(-chunkSize);
                                }

                                // 유니크 선박 카운트
                                batch.forEach(track -> uniqueVesselIds.add(track.getVesselId()));

                                // 진행률 업데이트 (시간 기반만 사용)
                                double timeProgress = (double) currentProcessedMin / estimatedTotalMinutes * 100;

                                statusConsumer.accept(new QueryStatusUpdate(
                                        queryId,
                                        "PROCESSING",
                                        "Processing chunk " + globalChunkIndex,
                                        Math.min(99.0, timeProgress)
                                ));

                                // 버퍼 사용률에 따른 적응형 대기
                                double currentBufferUsage = (double) pendingBufferSize.get() / MAX_PENDING_BUFFER;
                                int waitTime;
                                if (currentBufferUsage > 0.8) waitTime = 200;       // 80%↑: 강한 억제
                                else if (currentBufferUsage > 0.5) waitTime = 100;  // 50%↑: 중간 억제
                                else if (currentBufferUsage > 0.3) waitTime = 50;   // 30%↑: 약한 억제
                                else waitTime = 10;                                  // 정상: 최소 지연
                                Thread.sleep(waitTime);

                                // 진행 상황 로그 (매 10번째 청크마다)
                                if (globalChunkIndex % 10 == 0) {
                                    log.info("Progress: chunk {}, vessels: {}, time progress: {}%",
                                            globalChunkIndex, uniqueVesselIds.size(),
                                            Math.round(timeProgress));
                                }
                            }
                            batches.clear();        // 메모리 즉시 해제: 배치 분할 리스트
                            mergedTracks.clear();   // 메모리 즉시 해제: 병합 항적 리스트
                        }
                        mergedMap.clear();          // 메모리 즉시 해제: 선박 누적 맵
                    }
                }
            }

            // 마지막 청크 표시
            if (globalChunkIndex > 0) {
                // 마지막 청크 재전송 (마지막 플래그 설정)
                TrackChunkResponse lastChunkMarker = new TrackChunkResponse();
                lastChunkMarker.setQueryId(queryId);
                lastChunkMarker.setChunkIndex(globalChunkIndex - 1);
                lastChunkMarker.setTotalChunks(globalChunkIndex);
                lastChunkMarker.setIsLastChunk(true);
                lastChunkMarker.setCompactTracks(new ArrayList<>());
                int finalProcessedMin = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
                lastChunkMarker.setStats(createChunkStats(new ArrayList<>(), uniqueVesselIds, finalProcessedMin));
                chunkConsumer.accept(lastChunkMarker);
            }

            // [BENCHMARK] 벤치마크 JSON 기록
            long totalElapsedMs = System.currentTimeMillis() - queryStartTime;
            benchmark.totalBatches = globalChunkIndex;
            try {
                benchmarkLog.info(benchmark.toJson(queryId, uniqueVesselIds.size(), totalElapsedMs));
            } catch (Exception e) {
                log.debug("Failed to write benchmark log: {}", e.getMessage());
            }

            log.info("Query {} completed: {} chunks, {} unique vessels",
                    queryId, globalChunkIndex, uniqueVesselIds.size());

            // 백프레셔 통계 출력
            if (metrics != null) {
                log.info("[BACKPRESSURE] Query {} statistics:", queryId);
                log.info("[BACKPRESSURE] - Total bytes sent: {} MB", metrics.totalBytes.get() / (1024 * 1024));
                log.info("[BACKPRESSURE] - Total chunks: {}", metrics.chunkCount.get());
                log.info("[BACKPRESSURE] - Buffer warnings: {}", metrics.bufferWarnings.get());
                log.info("[BACKPRESSURE] - Backpressure events: {}", metrics.backpressureEvents.get());
                log.info("[BACKPRESSURE] - Final chunk size: {} KB (original: {} KB)",
                        metrics.dynamicChunkSizeKB, MAX_MESSAGE_SIZE_KB);

                if (metrics.chunkCount.get() > 0) {
                    double avgChunkSize = (double) metrics.totalBytes.get() / metrics.chunkCount.get() / 1024;
                    log.info("[BACKPRESSURE] - Average chunk size: {} KB", String.format("%.1f", avgChunkSize));
                }
            }

            // 완료 상태
            statusConsumer.accept(new QueryStatusUpdate(
                    queryId,
                    "COMPLETED",
                    "Query completed successfully",
                    100.0
            ));

        } catch (Exception e) {
            log.error("Error in chunked streaming: {}", e.getMessage(), e);

            // 에러 시에도 백프레셔 통계 출력
            BackpressureMetrics errorMetrics = queryMetrics.get(queryId);
            if (errorMetrics != null) {
                log.error("[BACKPRESSURE] Query {} failed with statistics:", queryId);
                log.error("[BACKPRESSURE] - Chunks sent before error: {}", errorMetrics.chunkCount.get());
                log.error("[BACKPRESSURE] - Backpressure events: {}", errorMetrics.backpressureEvents.get());
            }

            statusConsumer.accept(new QueryStatusUpdate(
                    queryId,
                    "ERROR",
                    "Error: " + e.getMessage(),
                    0.0
            ));
        } finally {
            // 리소스 반환 (순서: 데이터 정리 → 취소 플래그 → 글로벌 슬롯 → 세션 카운트 → 메트릭스)
            processedTimeRanges.clear();  // 메모리 즉시 해제: 처리 시간 범위 맵
            queryCancelFlags.remove(queryId);
            if (slotAcquired) {
                activeQueryManager.releaseQuerySlot(queryId);
            }
            trackQueryInterceptor.releaseQuery(sessionId);
            activeQueryManager.completeQuery(sessionId);
            cleanupQueryMetrics(queryId);

            // Humongous 영역 조기 회수: G1GC에서 8MB+ 객체는 Young GC로 회수 불가
            // Concurrent Mark → Mixed GC 사이클이 필요하므로, 대규모 해제 후 GC 힌트 제공
            // -XX:+ExplicitGCInvokesConcurrent 플래그와 함께 사용 시 STW 없이 concurrent GC 트리거
            if (activeQueryManager.isHeapPressureHigh()) {
                log.info("Query {} triggering concurrent GC for Humongous reclaim (session={})", queryId, sessionId);
                System.gc();
            }

            log.info("Query {} resources fully released (session={})", queryId, sessionId);
        }
    }

    /**
     * 쿼리 취소
     */
    public void cancelQuery(String queryId) {
        log.info("Cancelling chunked query: {}", queryId);
        AtomicBoolean flag = queryCancelFlags.get(queryId);
        if (flag != null) {
            flag.set(true);
            log.info("Cancel flag set for chunked query: {}", queryId);
        }
        cleanupQueryMetrics(queryId);
    }

    /**
     * 쿼리 메트릭스 정리
     */
    private void cleanupQueryMetrics(String queryId) {
        BackpressureMetrics removed = queryMetrics.remove(queryId);
        if (removed != null) {
            log.debug("[BACKPRESSURE] Cleaned up metrics for query {}", queryId);
        }
    }

    /**
     * 메시지 크기 기반으로 트랙 분할 (동적 크기 적용)
     */
    private List<List<CompactVesselTrack>> splitByMessageSize(List<CompactVesselTrack> tracks) {
        return splitByMessageSize(tracks, null);
    }

    private List<List<CompactVesselTrack>> splitByMessageSize(List<CompactVesselTrack> tracks, String queryId) {
        // 쿼리에 대한 동적 청크 크기 가져오기
        int maxChunkSizeKB = MAX_MESSAGE_SIZE_KB;
        if (queryId != null && queryMetrics.containsKey(queryId)) {
            maxChunkSizeKB = queryMetrics.get(queryId).dynamicChunkSizeKB;
        }

        List<List<CompactVesselTrack>> batches = new ArrayList<>();
        List<CompactVesselTrack> currentBatch = new ArrayList<>();
        int currentSize = 0;

        for (CompactVesselTrack track : tracks) {
            int trackSize = estimateTrackSize(track);

            // 현재 배치가 크기 제한을 초과하고 비어있지 않으면 새 배치 시작
            if (currentSize + trackSize > maxChunkSizeKB * 1024 && !currentBatch.isEmpty()) {
                batches.add(new ArrayList<>(currentBatch));
                currentBatch.clear();
                currentSize = 0;
            }

            currentBatch.add(track);
            currentSize += trackSize;
        }

        // 마지막 배치 추가
        if (!currentBatch.isEmpty()) {
            batches.add(currentBatch);
        }

        log.info("[splitByMessageSize] {} tracks split into {} batches (max size: {}KB)",
                tracks.size(), batches.size(), maxChunkSizeKB);

        return batches;
    }

    /**
     * 트랙 크기 추정 (바이트)
     */
    private int estimateTrackSize(CompactVesselTrack track) {
        // 각 좌표: 2 * 8 bytes (double), 타임스탬프: ~20 chars, 속도: 8 bytes
        int pointSize = 16 + 20 + 8; // ~44 bytes per point
        int metadataSize = 100; // vessel ID, 기타 필드
        return track.getPointCount() * pointSize + metadataSize;
    }

    /**
     * 두 좌표 사이의 거리 계산 (Haversine 공식, 해리)
     */
    private double calculateDistance(Coordinate from, Coordinate to) {
        double lat1 = Math.toRadians(from.y);
        double lat2 = Math.toRadians(to.y);
        double deltaLat = lat2 - lat1;
        double deltaLon = Math.toRadians(to.x - from.x);

        double a = Math.sin(deltaLat / 2) * Math.sin(deltaLat / 2) +
                Math.cos(lat1) * Math.cos(lat2) *
                        Math.sin(deltaLon / 2) * Math.sin(deltaLon / 2);
        double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));

        // 지구 반지름 (nm)
        double R = 3440.065; // nautical miles
        return R * c;
    }

    /**
     * 두 좌표 사이의 거리 계산 (double 배열 버전)
     */
    private double calculateDistance(double lat1, double lon1, double lat2, double lon2) {
        double lat1Rad = Math.toRadians(lat1);
        double lat2Rad = Math.toRadians(lat2);
        double deltaLat = lat2Rad - lat1Rad;
        double deltaLon = Math.toRadians(lon2 - lon1);

        double a = Math.sin(deltaLat / 2) * Math.sin(deltaLat / 2) +
                Math.cos(lat1Rad) * Math.cos(lat2Rad) *
                        Math.sin(deltaLon / 2) * Math.sin(deltaLon / 2);
        double c = 2 * Math.atan2(Math.sqrt(a), Math.sqrt(1 - a));

        double R = 3440.065; // nautical miles
        return R * c;
    }

    /**
     * 시간 범위별 테이블 선택 로직 (StompTrackStreamingService 참고)
     */
    private Map<TableStrategy, List<TimeRange>> splitTimeRangeByStrategy(LocalDateTime start, LocalDateTime end) { // [BENCHMARK] benchmark 파라미터 오버로드 아래 참조
        return splitTimeRangeByStrategy(start, end, null);
    }

    private Map<TableStrategy, List<TimeRange>> splitTimeRangeByStrategy(LocalDateTime start, LocalDateTime end, QueryBenchmark benchmark) { // [BENCHMARK]
        Map<TableStrategy, List<TimeRange>> strategyMap = new LinkedHashMap<>();
        LocalDateTime now = LocalDateTime.now();

        // 5분 데이터: 최근 1시간 이내
        LocalDateTime hourAgo = now.minusHours(1);

        // 1시간 데이터: 1시간 전 ~ 24시간 전
        LocalDateTime dayAgo = now.minusDays(1);

        // 집계 수준별 시간 범위 분할
        List<TimeRange> fiveMinRanges = new ArrayList<>();
        List<TimeRange> hourlyRanges = new ArrayList<>();
        List<TimeRange> dailyRanges = new ArrayList<>();

        LocalDateTime current = start;

        while (current.isBefore(end)) {
            LocalDateTime rangeEnd;

            // 현재 시간에서 얼마나 오래된 데이터인지에 따라 처리
            if (current.isAfter(hourAgo)) {
                // 최근 1시간 이내 - 5분 단위로 처리
                rangeEnd = current.plusMinutes(5);
                if (rangeEnd.isAfter(end)) rangeEnd = end;
                if (rangeEnd.isAfter(now)) rangeEnd = now;

                fiveMinRanges.add(new TimeRange(current, rangeEnd));
            } else if (current.isAfter(dayAgo)) {
                // 1시간 ~ 24시간 - 시간 단위로 처리
                rangeEnd = current.plusHours(1).withMinute(0).withSecond(0).withNano(0);
                if (rangeEnd.isAfter(end)) rangeEnd = end;

                hourlyRanges.add(new TimeRange(current, rangeEnd));
            } else {
                // 24시간 이전 - 일 단위로 처리
                // 일별 테이블은 전체 날짜를 포함해야 함
                LocalDateTime dayStart = current.toLocalDate().atStartOfDay();
                LocalDateTime dayEnd = dayStart.plusDays(1);

                // 요청된 시간 범위를 초과하지 않도록 조정
                if (dayEnd.isAfter(end)) {
                    dayEnd = end.toLocalDate().plusDays(1).atStartOfDay();
                }

                boolean alreadyAdded = dailyRanges.stream()
                        .anyMatch(r -> r.getStart().toLocalDate().equals(dayStart.toLocalDate()));

                if (!alreadyAdded && dayStart.isBefore(end)) {
                    dailyRanges.add(new TimeRange(dayStart, dayEnd));
                }

                rangeEnd = dayEnd;
            }

            current = rangeEnd;
        }

        // 결과를 strategyMap에 추가 (데이터 존재 여부 확인 후)
        if (!dailyRanges.isEmpty()) {
            List<TimeRange> mergedDailyRanges = mergeDailyRanges(dailyRanges);
            List<TimeRange> validDailyRanges = new ArrayList<>();
            List<TimeRange> fallbackToHourly = new ArrayList<>();

            // 각 일별 범위에 대해 데이터 존재 여부 확인
            for (TimeRange dailyRange : mergedDailyRanges) {
                if (benchmark != null) benchmark.connTableCheck++; // [BENCHMARK]
                if (hasDataInTable(TableStrategy.DAILY.getTableName(), dailyRange.getStart(), dailyRange.getEnd())) {
                    validDailyRanges.add(dailyRange);
                    log.debug("Daily data found for range: {}", dailyRange);
                } else {
                    log.info("No daily data for range {}, falling back to hourly", dailyRange);
                    LocalDateTime hourlyStart = dailyRange.getStart();
                    while (hourlyStart.isBefore(dailyRange.getEnd())) {
                        LocalDateTime rangeEnd = hourlyStart.plusHours(1).withMinute(0).withSecond(0).withNano(0);
                        if (rangeEnd.isAfter(dailyRange.getEnd())) rangeEnd = dailyRange.getEnd();
                        fallbackToHourly.add(new TimeRange(hourlyStart, rangeEnd));
                        hourlyStart = rangeEnd;
                    }
                }
            }

            if (!validDailyRanges.isEmpty()) {
                strategyMap.put(TableStrategy.DAILY, validDailyRanges);
            }
            hourlyRanges.addAll(fallbackToHourly);
        }

        if (!hourlyRanges.isEmpty()) {
            List<TimeRange> validHourlyRanges = new ArrayList<>();
            List<TimeRange> fallbackTo5Min = new ArrayList<>();

            for (TimeRange range : hourlyRanges) {
                if (benchmark != null) benchmark.connTableCheck++; // [BENCHMARK]
                if (hasDataInTable(TableStrategy.HOURLY.getTableName(), range.getStart(), range.getEnd())) {
                    validHourlyRanges.add(range);
                } else {
                    log.info("No data found in hourly table for range {}, falling back to 5min", range);
                    LocalDateTime fiveMinStart = range.getStart();
                    while (fiveMinStart.isBefore(range.getEnd())) {
                        LocalDateTime rangeEnd = fiveMinStart.plusMinutes(5);
                        if (rangeEnd.isAfter(range.getEnd())) rangeEnd = range.getEnd();
                        fallbackTo5Min.add(new TimeRange(fiveMinStart, rangeEnd));
                        fiveMinStart = rangeEnd;
                    }
                }
            }

            if (!validHourlyRanges.isEmpty()) {
                strategyMap.put(TableStrategy.HOURLY, validHourlyRanges);
            }
            fiveMinRanges.addAll(fallbackTo5Min);
        }

        if (!fiveMinRanges.isEmpty()) {
            strategyMap.put(TableStrategy.FIVE_MINUTE, fiveMinRanges);
        }

        log.info("Time range split - Daily: {} ranges, Hourly: {} ranges, 5min: {} ranges",
                dailyRanges.size(), hourlyRanges.size(), fiveMinRanges.size());

        return strategyMap;
    }

    /**
     * 테이블에 데이터가 존재하는지 확인
     */
    private boolean hasDataInTable(String tableName, LocalDateTime start, LocalDateTime end) {
        String sql = "SELECT EXISTS(SELECT 1 FROM " + tableName +
                " WHERE time_bucket >= ? AND time_bucket < ? LIMIT 1)";
        try {
            return queryJdbcTemplate.queryForObject(sql, Boolean.class, start, end);
        } catch (Exception e) {
            log.warn("Failed to check data existence in {}: {}", tableName, e.getMessage());
            return false;
        }
    }

    /**
     * 일별 범위를 날짜별로 병합
     */
    private List<TimeRange> mergeDailyRanges(List<TimeRange> ranges) {
        Map<LocalDateTime, LocalDateTime> dayMap = new TreeMap<>();

        for (TimeRange range : ranges) {
            LocalDateTime dayStart = range.getStart().toLocalDate().atStartOfDay();
            LocalDateTime currentEnd = dayMap.get(dayStart);

            if (currentEnd == null || range.getEnd().isAfter(currentEnd)) {
                dayMap.put(dayStart, range.getEnd());
            }
        }

        return dayMap.entrySet().stream()
                .map(e -> new TimeRange(e.getKey(), e.getValue()))
                .collect(Collectors.toList());
    }

    /**
     * 범위를 청크로 분할
     */
    private List<TimeChunk> divideRangesIntoChunks(List<TimeRange> ranges, TableStrategy strategy) {
        List<TimeChunk> chunks = new ArrayList<>();

        for (TimeRange range : ranges) {
            if (strategy == TableStrategy.HOURLY) {
                // Hourly는 3시간 단위로 분할
                LocalDateTime chunkStart = range.getStart();
                while (chunkStart.isBefore(range.getEnd())) {
                    LocalDateTime chunkEnd = chunkStart.plusHours(3);
                    if (chunkEnd.isAfter(range.getEnd())) {
                        chunkEnd = range.getEnd();
                    }
                    chunks.add(new TimeChunk(chunkStart, chunkEnd));
                    chunkStart = chunkEnd;
                }
            } else {
                // 5min은 30분 단위로 분할
                LocalDateTime chunkStart = range.getStart();
                while (chunkStart.isBefore(range.getEnd())) {
                    LocalDateTime chunkEnd = chunkStart.plusMinutes(30);
                    if (chunkEnd.isAfter(range.getEnd())) {
                        chunkEnd = range.getEnd();
                    }
                    chunks.add(new TimeChunk(chunkStart, chunkEnd));
                    chunkStart = chunkEnd;
                }
            }
        }

        return chunks;
    }

    /**
     * 청크 처리 - 시간 범위별 테이블 전략 사용
     */
    @SuppressWarnings("unused")
    private List<CompactVesselTrack> processChunk(TrackQueryRequest request, TimeChunk chunk) {
        Map<String, CompactVesselTrack.CompactVesselTrackBuilder> vesselMap = new HashMap<>();

        // 시간 범위별 테이블 전략 분할
        Map<TableStrategy, List<TimeRange>> strategyMap = splitTimeRangeByStrategy(
                chunk.start, chunk.end
        );

        // 각 테이블 전략별로 처리
        for (Map.Entry<TableStrategy, List<TimeRange>> entry : strategyMap.entrySet()) {
            TableStrategy strategy = entry.getKey();
            List<TimeRange> ranges = entry.getValue();

            for (TimeRange range : ranges) {
                List<CompactVesselTrack> tracks = processTableRange(request, strategy, range);
                // 선박별로 병합
                for (CompactVesselTrack track : tracks) {
                    CompactVesselTrack.CompactVesselTrackBuilder builder = vesselMap.get(track.getVesselId());
                    if (builder == null) {
                        builder = CompactVesselTrack.builder()
                                .vesselId(track.getVesselId())
                                .nationalCode(track.getNationalCode())
                                .geometry(new ArrayList<>())
                                .timestamps(new ArrayList<>())
                                .speeds(new ArrayList<>())
                                .totalDistance(0.0)
                                .avgSpeed(0.0)
                                .maxSpeed(0.0)
                                .pointCount(0);
                        vesselMap.put(track.getVesselId(), builder);
                    }

                    // 데이터 병합
                    builder.geometry(mergeGeometry(builder.build().getGeometry(), track.getGeometry()));
                    builder.timestamps(mergeTimestamps(builder.build().getTimestamps(), track.getTimestamps()));
                    builder.speeds(mergeSpeeds(builder.build().getSpeeds(), track.getSpeeds()));
                    builder.totalDistance(builder.build().getTotalDistance() + track.getTotalDistance());
                    builder.maxSpeed(Math.max(builder.build().getMaxSpeed(), track.getMaxSpeed()));
                    builder.pointCount(builder.build().getPointCount() + track.getPointCount());
                }
            }
        }

        // 평균 속도 계산 및 최종 변환
        return vesselMap.values().stream()
                .map(builder -> {
                    CompactVesselTrack track = builder.build();
                    if (track.getTotalDistance() > 0 && track.getTimestamps().size() > 1) {
                        // Unix timestamp 지원
                        String firstTs = track.getTimestamps().get(0);
                        String lastTs = track.getTimestamps().get(track.getTimestamps().size() - 1);
                        
                        LocalDateTime startTime;
                        LocalDateTime endTime;
                        
                        // Unix timestamp 감지 (10자리 이상 숫자)
                        if (firstTs.matches("\\d{10,}")) {
                            startTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(firstTs)), 
                                java.time.ZoneId.systemDefault());
                            endTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(lastTs)), 
                                java.time.ZoneId.systemDefault());
                        } else {
                            startTime = LocalDateTime.parse(firstTs, TIMESTAMP_FORMATTER);
                            endTime = LocalDateTime.parse(lastTs, TIMESTAMP_FORMATTER);
                        }
                        
                        double hours = java.time.Duration.between(startTime, endTime).toMinutes() / 60.0;
                        if (hours > 0) {
                            track.setAvgSpeed(track.getTotalDistance() / hours);
                        }
                    }
                    return track;
                })
                .collect(Collectors.toList());
    }


    /**
     * 테이블 선택 (집계 지연 고려)
     */
    @SuppressWarnings("unused")
    private String selectTableByTimeRange(LocalDateTime start, LocalDateTime end) {
        LocalDateTime now = LocalDateTime.now();

        // Daily 데이터는 매일 01:00에 집계 (전일 데이터)
        LocalDateTime lastDailyAggregation = now.toLocalDate().atTime(1, 0);
        if (now.isBefore(lastDailyAggregation)) {
            lastDailyAggregation = lastDailyAggregation.minusDays(1);
        }

        // Hourly 데이터는 매시 10분에 집계 (이전 시간 데이터)
        LocalDateTime lastHourlyAggregation = now.truncatedTo(java.time.temporal.ChronoUnit.HOURS).plusMinutes(10);
        if (now.isBefore(lastHourlyAggregation)) {
            lastHourlyAggregation = lastHourlyAggregation.minusHours(1);
        }

        // 조회 시간이 완전히 daily 집계 완료 범위에 있으면
        if (end.isBefore(lastDailyAggregation.minusDays(1))) {
            return "signal.t_vessel_tracks_daily";
        }

        // 조회 시간이 완전히 hourly 집계 완료 범위에 있고 1일 이상이면
        if (end.isBefore(lastHourlyAggregation.minusHours(1)) &&
                java.time.Duration.between(start, end).toHours() >= 24) {
            return "signal.t_vessel_tracks_daily";
        }

        // 조회 시간이 완전히 hourly 집계 완료 범위에 있으면
        if (end.isBefore(lastHourlyAggregation.minusHours(1))) {
            return "signal.t_vessel_tracks_hourly";
        }

        // 그 외의 경우 5분 데이터 사용
        return "signal.t_vessel_tracks_5min";
    }

    /**
     * 쿼리 생성 (간소화 적용, 2-pass 뷰포트 필터링 지원)
     */
    private String buildRangeQuery(String tableName, TrackQueryRequest request, TimeRange range,
            SimplificationLevel simplificationLevel, Set<String> viewportVesselIds) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT mmsi, time_bucket, ");

        // 간소화 적용
        if (simplificationLevel != SimplificationLevel.NONE && simplificationLevel.getTolerance() > 0) {
            sql.append("public.ST_AsText(public.ST_Simplify(track_geom, ").append(simplificationLevel.getTolerance())
                    .append(")) as track_geom, ");
        } else {
            sql.append("public.ST_AsText(track_geom) as track_geom, ");
        }

        sql.append("distance_nm, avg_speed, max_speed, point_count");

        if (!tableName.contains("5min")) {
            sql.append(", start_position->>'time' as start_time");
            sql.append(", end_position->>'time' as end_time");
        }

        sql.append(" FROM ").append(tableName).append(" ");
        sql.append("WHERE time_bucket >= ? ");
        sql.append("AND time_bucket < ? ");

        if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
            sql.append("AND mmsi = ANY(?) ");
        } else if (request.getViewport() != null) {
            sql.append("AND public.ST_Intersects(track_geom, public.ST_MakeEnvelope(?, ?, ?, ?, 4326)) ");
        }

        if (request.getMinAvgSpeed() != null) {
            sql.append("AND avg_speed >= ").append(request.getMinAvgSpeed()).append(" ");
        }
        if (request.getMaxAvgSpeed() != null) {
            sql.append("AND avg_speed <= ").append(request.getMaxAvgSpeed()).append(" ");
        }

        sql.append("ORDER BY mmsi, time_bucket ");
        sql.append("LIMIT ").append(MAX_TRACKS_PER_CHUNK);

        return sql.toString();
    }

    /**
     * Daily 테이블용 페이지네이션 쿼리 생성
     */
    private String buildDailyPaginationQuery(String tableName, TrackQueryRequest request, TimeRange range,
            double tolerance, String lastMmsi, String lastTargetId, Set<String> viewportVesselIds) {
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT mmsi, time_bucket, ");

        if (tolerance > 0) {
            sql.append("public.ST_AsText(public.ST_Simplify(track_geom, ").append(tolerance)
                    .append(")) as track_geom, ");
        } else {
            sql.append("public.ST_AsText(track_geom) as track_geom, ");
        }

        sql.append("distance_nm, avg_speed, max_speed, point_count");
        sql.append(", start_position->>'time' as start_time");
        sql.append(", end_position->>'time' as end_time");

        sql.append(" FROM ").append(tableName).append(" ");
        sql.append("WHERE time_bucket >= ? ");
        sql.append("AND time_bucket < ? ");

        if (lastMmsi != null) {
            sql.append("AND mmsi > ? ");
        }

        if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
            sql.append("AND mmsi = ANY(?) ");
        } else if (request.getViewport() != null) {
            sql.append("AND public.ST_Intersects(track_geom, public.ST_MakeEnvelope(?, ?, ?, ?, 4326)) ");
        }

        if (request.getMinAvgSpeed() != null) {
            sql.append("AND avg_speed >= ").append(request.getMinAvgSpeed()).append(" ");
        }
        if (request.getMaxAvgSpeed() != null) {
            sql.append("AND avg_speed <= ").append(request.getMaxAvgSpeed()).append(" ");
        }

        sql.append("ORDER BY mmsi, time_bucket ");
        sql.append("LIMIT ").append(DAILY_PAGE_SIZE);

        return sql.toString();
    }

    /**
     * Daily 테이블 처리 - 선박 기준 페이지네이션으로 모든 데이터 수집 후 병합
     */
    private List<CompactVesselTrack> processDailyTableWithPagination(TrackQueryRequest request, TimeRange range, String sessionId) {
        return processDailyTableWithPagination(request, range, sessionId, null);
    }

    private List<CompactVesselTrack> processDailyTableWithPagination(TrackQueryRequest request, TimeRange range, String sessionId, Set<String> viewportVesselIds) {
        Map<String, VesselAccumulator> vesselMap = new HashMap<>(50000); // 예상 선박 수
        String tableName = TableStrategy.DAILY.getTableName();

        // 줌 레벨에 따른 강화된 간소화 tolerance
        double tolerance = getEnhancedDailyTolerance(request.getZoomLevel());
        log.info("Daily pagination started for range [{} - {}] with tolerance {} (zoom: {})",
                range.getStart(), range.getEnd(), tolerance, request.getZoomLevel());

        String lastMmsi = null;
        int pageNum = 0;
        int totalTrackCount = 0;

        // 페이지네이션 루프
        while (true) {
            String sql = buildDailyPaginationQuery(tableName, request, range, tolerance, lastMmsi, null, viewportVesselIds);

            try (Connection conn = queryDataSource.getConnection();
                 PreparedStatement ps = conn.prepareStatement(sql)) {

                int paramIndex = 1;
                ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getStart()));
                ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getEnd()));

                // 페이지네이션 파라미터
                if (lastMmsi != null) {
                    ps.setString(paramIndex++, lastMmsi);
                }

                // 2-pass 뷰포트 필터: vessel ID 배열 또는 기존 viewport 좌표 바인딩
                if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
                    Array vesselArray = conn.createArrayOf("text", viewportVesselIds.toArray(new String[0]));
                    ps.setArray(paramIndex++, vesselArray);
                } else if (request.getViewport() != null) {
                    ViewportFilter vp = request.getViewport();
                    ps.setDouble(paramIndex++, vp.getMinLon());
                    ps.setDouble(paramIndex++, vp.getMinLat());
                    ps.setDouble(paramIndex++, vp.getMaxLon());
                    ps.setDouble(paramIndex++, vp.getMaxLat());
                }

                ps.setFetchSize(5000);

                try (ResultSet rs = ps.executeQuery()) {
                    int pageTrackCount = 0;
                    String currentMmsi = null;

                    while (rs.next()) {
                        // 세션 연결 끊김 체크 (1000개마다)
                        if (sessionId != null && pageTrackCount % 1000 == 0) {
                            if (isQueryCancelled(sessionId)) {
                                log.warn("Query cancelled during daily pagination: sessionId={}, page={}, trackCount={}",
                                        sessionId, pageNum, totalTrackCount);
                                return Collections.emptyList();
                            }
                        }

                        currentMmsi = rs.getString("mmsi");
                        String vesselId = currentMmsi;

                        // LineStringM 파싱
                        String trackGeomWkt = rs.getString("track_geom");
                        Timestamp timeBucket = rs.getTimestamp("time_bucket");

                        String startTimeStr = null;
                        String endTimeStr = null;
                        try {
                            startTimeStr = rs.getString("start_time");
                            endTimeStr = rs.getString("end_time");
                        } catch (SQLException ignored) {}

                        // 선박 객체는 geometry가 비어있어도 생성 (선박 누락 방지)
                        final String finalMmsi = currentMmsi;
                        VesselAccumulator accumulator = vesselMap.computeIfAbsent(vesselId, k -> {
                            VesselInfo info = getVesselInfo(finalMmsi);
                            VesselAccumulator acc = new VesselAccumulator();
                            acc.mmsi = finalMmsi;
                            acc.shipName = info.shipName;
                            acc.shipType = info.shipType;
                            acc.shipKindCode = SignalKindCode.resolve(info.shipType, null).getCode();
                            return acc;
                        });

                        if (trackGeomWkt != null && !trackGeomWkt.isEmpty() && !"LINESTRING EMPTY".equals(trackGeomWkt)) {
                            try {
                                LineString lineString = (LineString) wktReader.read(trackGeomWkt);

                                // 빈 LineString은 건너뛰되, 선박 객체는 이미 생성됨
                                if (lineString.getNumPoints() == 0) {
                                    pageTrackCount++;
                                    continue;
                                }

                                // 좌표 배열로 변환
                                List<double[]> geometry = new ArrayList<>();
                                List<String> timestamps = new ArrayList<>();
                                List<Double> speeds = new ArrayList<>();

                                LocalDateTime baseTime = timeBucket.toLocalDateTime();
                                if (startTimeStr != null && !startTimeStr.isEmpty()) {
                                    try {
                                        baseTime = LocalDateTime.parse(startTimeStr, TIMESTAMP_FORMATTER);
                                    } catch (Exception e) {
                                        log.debug("Failed to parse start_time: {}", startTimeStr);
                                    }
                                }

                                Coordinate prevCoord = null;
                                long prevTimeMillis = 0;

                                for (int i = 0; i < lineString.getNumPoints(); i++) {
                                    Coordinate coord = lineString.getCoordinateN(i);
                                    geometry.add(new double[]{coord.x, coord.y});

                                    long unixTimestamp = (long)coord.getM();
                                    timestamps.add(String.valueOf(unixTimestamp));

                                    double speed = 0.0;
                                    long currentTimeMillis = (long)coord.getM() * 1000;

                                    if (prevCoord != null && prevTimeMillis > 0) {
                                        double distance = calculateDistance(prevCoord.y, prevCoord.x, coord.y, coord.x);
                                        long timeDiff = currentTimeMillis - prevTimeMillis;
                                        if (timeDiff > 0) {
                                            speed = (distance / (timeDiff / 3600000.0));
                                        }
                                    }
                                    speeds.add(speed);

                                    prevCoord = coord;
                                    prevTimeMillis = currentTimeMillis;
                                }

                                accumulator.geometry.addAll(geometry);
                                accumulator.timestamps.addAll(timestamps);
                                accumulator.speeds.addAll(speeds);
                                accumulator.totalDistance += rs.getDouble("distance_nm");
                                accumulator.maxSpeed = Math.max(accumulator.maxSpeed, rs.getDouble("max_speed"));
                                accumulator.pointCount += geometry.size();

                            } catch (ParseException e) {
                                log.error("Error parsing geometry: {}", e.getMessage());
                            }
                        }
                        pageTrackCount++;
                    }

                    totalTrackCount += pageTrackCount;

                    // 페이지가 완전히 채워지지 않았으면 마지막 페이지
                    if (pageTrackCount < DAILY_PAGE_SIZE) {
                        log.info("Daily pagination completed: {} pages, {} total tracks, {} unique vessels",
                                pageNum + 1, totalTrackCount, vesselMap.size());
                        break;
                    }

                    // 다음 페이지를 위한 마지막 MMSI 저장
                    lastMmsi = currentMmsi;
                    pageNum++;

                    log.debug("Daily pagination page {} completed: {} tracks (total: {}, vessels: {})",
                            pageNum, pageTrackCount, totalTrackCount, vesselMap.size());
                }
            } catch (SQLException e) {
                log.error("Error in daily pagination page {}: {}", pageNum, e.getMessage());
                break;
            }
        }

        // VesselAccumulator를 CompactVesselTrack로 변환
        return vesselMap.entrySet().stream()
                .map(entry -> {
                    String vesselId = entry.getKey();
                    VesselAccumulator acc = entry.getValue();

                    double avgSpeed = 0.0;
                    if (acc.totalDistance > 0 && acc.timestamps.size() > 1) {
                        String firstTs = acc.timestamps.get(0);
                        String lastTs = acc.timestamps.get(acc.timestamps.size() - 1);

                        LocalDateTime startTime;
                        LocalDateTime endTime;

                        if (firstTs.matches("\\d{10,}")) {
                            startTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(firstTs)),
                                java.time.ZoneId.systemDefault());
                            endTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(lastTs)),
                                java.time.ZoneId.systemDefault());
                        } else {
                            startTime = LocalDateTime.parse(firstTs, TIMESTAMP_FORMATTER);
                            endTime = LocalDateTime.parse(lastTs, TIMESTAMP_FORMATTER);
                        }

                        double hours = Duration.between(startTime, endTime).toMinutes() / 60.0;
                        if (hours > 0) {
                            avgSpeed = acc.totalDistance / hours;
                        }
                    }

                    return CompactVesselTrack.builder()
                            .vesselId(vesselId)
                            .nationalCode(acc.mmsi != null && acc.mmsi.length() >= 3 ? acc.mmsi.substring(0, 3) : null)
                            .shipName(acc.shipName)
                            .shipType(acc.shipType)
                            .shipKindCode(acc.shipKindCode)
                            .geometry(acc.geometry)
                            .timestamps(acc.timestamps)
                            .speeds(acc.speeds)
                            .totalDistance(acc.totalDistance)
                            .avgSpeed(avgSpeed)
                            .maxSpeed(acc.maxSpeed)
                            .pointCount(acc.pointCount)
                            .build();
                })
                .collect(Collectors.toList());
    }

    /**
     * Daily 테이블용 강화된 간소화 tolerance 결정
     */
    private double getEnhancedDailyTolerance(Integer zoomLevel) {
        if (zoomLevel == null) zoomLevel = 10;

        return switch (zoomLevel) {
            case 0, 1, 2, 3, 4, 5 -> 0.05;   // 약 5km - 극단적 간소화
            case 6, 7 -> 0.02;               // 약 2km
            case 8, 9 -> 0.01;               // 약 1km
            case 10, 11 -> 0.005;            // 약 500m
            default -> 0.002;                // 약 200m (zoom 12+)
        };
    }

    /**
     * 간소화 레벨 결정
     */
    private SimplificationLevel determineSimplificationLevel(TrackQueryRequest request, TimeChunk chunk) {
        // 줌 레벨 우선 적용
        Integer zoom = request.getZoomLevel();
        if (zoom == null) zoom = 10; // 기본값

        // 줌 레벨에 따른 간소화 레벨 결정 (가장 우선순위가 높음)
        if (zoom < 6) {
            return SimplificationLevel.EXTREME;      // 원거리: 90% 제거
        } else if (zoom < 8) {
            return SimplificationLevel.VERY_HEAVY;   // 80% 제거
        } else if (zoom < 10) {
            return SimplificationLevel.HEAVY;        // 70% 제거
        } else if (zoom < 12) {
            return SimplificationLevel.MODERATE;     // 50% 제거
        }

        // 시간 범위 계산 (보조 기준)
        Duration duration = Duration.between(chunk.start, chunk.end);
        long hours = duration.toHours();

        // 뷰포트 크기 계산
        double viewportArea = 180 * 90; // 기본값 (전체)
        if (request.getViewport() != null) {
            ViewportFilter vp = request.getViewport();
            double width = vp.getMaxLon() - vp.getMinLon();
            double height = vp.getMaxLat() - vp.getMinLat();
            viewportArea = width * height;
        }

        // 간소화 모드 확인
        if (request.getSimplificationMode() != null) {
            // if ("NONE".equals(request.getSimplificationMode())) {
            if ("NONE".equals(request.getSimplificationMode().toString())) {
                return SimplificationLevel.NONE;
            } else if ("AGGRESSIVE".equals(request.getSimplificationMode().toString())) {
                return SimplificationLevel.EXTREME;
            }
        }

        // 줌인 상태에서도 긴 시간 범위는 추가 간소화
        if (hours > 24 && zoom >= 12) {
            return SimplificationLevel.MODERATE;
        }

        // 근거리 뷰 (줌 12+): 경량 간소화
        return SimplificationLevel.LIGHT;
    }

    // 도우미 메서드들
    private List<double[]> mergeGeometry(List<double[]> existing, List<double[]> newData) {
        if (existing == null) existing = new ArrayList<>();
        existing.addAll(newData);
        return existing;
    }

    private List<String> mergeTimestamps(List<String> existing, List<String> newData) {
        if (existing == null) existing = new ArrayList<>();
        existing.addAll(newData);
        return existing;
    }

    private List<Double> mergeSpeeds(List<Double> existing, List<Double> newData) {
        if (existing == null) existing = new ArrayList<>();
        existing.addAll(newData);
        return existing;
    }

    // 내부 클래스들
    private static class TimeChunk {
        LocalDateTime start;
        LocalDateTime end;

        TimeChunk(LocalDateTime start, LocalDateTime end) {
            this.start = start;
            this.end = end;
        }
    }

    // 테이블 전략
    private enum TableStrategy {
        FIVE_MINUTE("signal.t_vessel_tracks_5min"),
        HOURLY("signal.t_vessel_tracks_hourly"),
        DAILY("signal.t_vessel_tracks_daily");

        private final String tableName;

        TableStrategy(String tableName) {
            this.tableName = tableName;
        }

        public String getTableName() {
            return tableName;
        }
    }

    // 시간 범위
    private static class TimeRange {
        private final LocalDateTime start;
        private final LocalDateTime end;

        TimeRange(LocalDateTime start, LocalDateTime end) {
            this.start = start;
            this.end = end;
        }

        public LocalDateTime getStart() {
            return start;
        }

        public LocalDateTime getEnd() {
            return end;
        }

        @Override
        public String toString() {
            return "[" + start + " ~ " + end + "]";
        }
    }

    // ChunkStats 생성 헬퍼 메서드
    private ChunkStats createChunkStats(List<CompactVesselTrack> compactTracks,
                                        Set<String> uniqueVesselIds, int processedMinutes) {
        ChunkStats stats = new ChunkStats();
        stats.setTrackCount(compactTracks.size());
        stats.setTotalTracks(compactTracks.stream()
                .mapToInt(CompactVesselTrack::getPointCount)
                .sum());

        // 시간 기반 진행률만 사용
        double timeProgress = estimatedTotalMinutes > 0 ?
                (double) processedMinutes / estimatedTotalMinutes * 100 : 0;
        stats.setProgressPercentage(Math.min(100.0, timeProgress));
        stats.setTimeProgress(timeProgress);

        // 기본 정보
        stats.setProcessedVessels(uniqueVesselIds.size());
        stats.setProcessedMinutes(processedMinutes);
        stats.setTotalMinutes(estimatedTotalMinutes);

        // 경과 시간
        stats.setElapsedMillis(System.currentTimeMillis() - queryStartTime);

        return stats;
    }

    // 현재 청크 인덱스 추적용
    private int currentGlobalChunkIndex = 0;
    private int vesselLogCount = 0; // 로그 카운트 추가

    private int getCurrentChunkIndex() {
        return currentGlobalChunkIndex;
    }

    /**
     * 범위를 날짜별로 그룹화
     */
    @SuppressWarnings("unused")
    private Map<LocalDate, List<TimeRange>> groupRangesByDate(List<TimeRange> ranges) {
        Map<LocalDate, List<TimeRange>> dateGroups = new TreeMap<>();

        for (TimeRange range : ranges) {
            LocalDate date = range.getStart().toLocalDate();
            dateGroups.computeIfAbsent(date, k -> new ArrayList<>()).add(range);
        }

        return dateGroups;
    }

    /**
     * 범위를 시간 윈도우로 그룹화 (예: 6시간 단위)
     */
    private Map<String, List<TimeRange>> groupRangesByTimeWindow(List<TimeRange> ranges, int windowHours) {
        Map<String, List<TimeRange>> timeGroups = new LinkedHashMap<>();

        for (TimeRange range : ranges) {
            LocalDateTime windowStart = range.getStart()
                    .withHour((range.getStart().getHour() / windowHours) * windowHours)
                    .withMinute(0).withSecond(0).withNano(0);

            String key = windowStart.format(DateTimeFormatter.ofPattern("yyyy-MM-dd_HH"));
            timeGroups.computeIfAbsent(key, k -> new ArrayList<>()).add(range);
        }

        return timeGroups;
    }

    /**
     * 기준 시간을 사용하여 테이블 범위 처리 (M값 재계산)
     */
    private List<CompactVesselTrack> processTableRangeWithBaseTime(
            TrackQueryRequest request, TableStrategy strategy, TimeRange range, LocalDateTime dayBaseTime,
            Set<String> viewportVesselIds) {

        // 기본 처리 후 M값 보정
        List<CompactVesselTrack> tracks = processTableRange(request, strategy, range, null, viewportVesselIds);

        // 각 트랙의 timestamp를 dayBaseTime 기준으로 재계산 및 간소화
        for (CompactVesselTrack track : tracks) {
            List<String> adjustedTimestamps = new ArrayList<>();
            List<double[]> simplifiedGeometry = new ArrayList<>();
            List<Double> simplifiedSpeeds = new ArrayList<>();
            List<String> simplifiedTimestamps = new ArrayList<>();

            // 시간 조정 (Unix timestamp 지원)
            for (String timestamp : track.getTimestamps()) {
                LocalDateTime time;
                
                // Unix timestamp 감지 (10자리 이상 숫자)
                if (timestamp.matches("\\d{10,}")) {
                    // Unix timestamp는 이미 UTC이므로 바로 변환
                    time = LocalDateTime.ofInstant(
                        java.time.Instant.ofEpochSecond(Long.parseLong(timestamp)), 
                        java.time.ZoneId.systemDefault());
                    // Unix timestamp는 조정 없이 그대로 사용
                    adjustedTimestamps.add(timestamp);
                } else {
                    time = LocalDateTime.parse(timestamp, TIMESTAMP_FORMATTER);
                    long secondsFromBase = ChronoUnit.SECONDS.between(dayBaseTime, time);
                    LocalDateTime adjustedTime = dayBaseTime.plusSeconds(secondsFromBase);
                    adjustedTimestamps.add(TIMESTAMP_FORMATTER.format(adjustedTime));
                }
            }

            // 간소화 적용 (시간 기반 샘플링 포함)
            if (track.getGeometry().size() > 1) {
                int originalSize = track.getGeometry().size();
                double[] prevPoint = null;
                LocalDateTime prevTime = null;

                for (int i = 0; i < track.getGeometry().size(); i++) {
                    double[] point = track.getGeometry().get(i);
                    LocalDateTime currentTime;
                    String tsStr = adjustedTimestamps.get(i);
                    
                    // Unix timestamp 감지
                    if (tsStr.matches("\\d{10,}")) {
                        currentTime = LocalDateTime.ofInstant(
                            java.time.Instant.ofEpochSecond(Long.parseLong(tsStr)), 
                            java.time.ZoneId.systemDefault());
                    } else {
                        currentTime = LocalDateTime.parse(tsStr, TIMESTAMP_FORMATTER);
                    }
                    boolean include = false;

                    // 첫 포인트나 마지막 포인트는 항상 포함
                    if (i == 0 || i == track.getGeometry().size() - 1) {
                        include = true;
                    } else if (prevPoint != null && prevTime != null) {
                        double distance = calculateDistance(prevPoint[1], prevPoint[0], point[1], point[0]);
                        long minutesSincePrev = ChronoUnit.MINUTES.between(prevTime, currentTime);

                        // 거리 또는 시간 기반 포함 여부 결정 (균형잡힌 간소화)
                        if (strategy == TableStrategy.DAILY) {
                            // Daily: 이미 집계된 데이터 - 추가 간소화 최소화
                            include = distance > 0.05 || minutesSincePrev >= 60;
                        } else if (strategy == TableStrategy.HOURLY) {
                            // Hourly: 2km(1.08nm) 이상 이동 또는 60분 이상 경과
                            include = distance > 1.08 || minutesSincePrev >= 60;
                        } else {
                            // 5min: 1km(0.54nm) 이상 이동 또는 30분 이상 경과
                            include = distance > 0.54 || minutesSincePrev >= 30;
                        }

                        // 저속 선박은 더 강하게 간소화 (속도 정보가 있을 때만)
                        if (track.getAvgSpeed() > 0 && track.getAvgSpeed() < 5.0) {
                            // 5 knots 미만 저속 선박: 2.8km(1.5nm) 이상 이동 또는 45분 이상 경과
                            include = distance > 1.5 || minutesSincePrev >= 45;
                        }
                    }

                    if (include) {
                        simplifiedGeometry.add(point);
                        simplifiedTimestamps.add(adjustedTimestamps.get(i));
                        simplifiedSpeeds.add(track.getSpeeds().get(i));
                        prevPoint = point;
                        prevTime = currentTime;
                    }
                }

                // 줌 레벨에 따른 추가 샘플링
                if (request.getZoomLevel() != null && request.getZoomLevel() < 10) {
                    int sampleRate = request.getZoomLevel() < 6 ? 10 :
                            request.getZoomLevel() < 8 ? 5 : 2;

                    // 매 N번째 포인트만 유지 (첨 포인트와 마지막 포인트는 항상 포함)
                    List<double[]> sampledGeometry = new ArrayList<>();
                    List<String> sampledTimestamps = new ArrayList<>();
                    List<Double> sampledSpeeds = new ArrayList<>();

                    for (int i = 0; i < simplifiedGeometry.size(); i++) {
                        if (i % sampleRate == 0 || i == simplifiedGeometry.size() - 1) {
                            sampledGeometry.add(simplifiedGeometry.get(i));
                            sampledTimestamps.add(simplifiedTimestamps.get(i));
                            sampledSpeeds.add(simplifiedSpeeds.get(i));
                        }
                    }

                    track.setGeometry(sampledGeometry);
                    track.setTimestamps(sampledTimestamps);
                    track.setSpeeds(sampledSpeeds);
                    track.setPointCount(sampledGeometry.size());

                    // 간소화 결과 로그 (처음 10개 선박만)
                    if (vesselLogCount++ < 10) {
                        double reductionRate = (1 - (double)sampledGeometry.size() / originalSize) * 100;
                        log.info("[{}] Vessel {} simplified: {} -> {} points ({}% reduced, zoom: {}, speed: {} knots)",
                                strategy, track.getVesselId(), originalSize, sampledGeometry.size(),
                                Math.round(reductionRate), request.getZoomLevel(), track.getAvgSpeed());
                    }
                } else {
                    track.setGeometry(simplifiedGeometry);
                    track.setTimestamps(simplifiedTimestamps);
                    track.setSpeeds(simplifiedSpeeds);
                    track.setPointCount(simplifiedGeometry.size());

                    // 간소화 결과 로그 (처음 10개 선박만)
                    if (vesselLogCount++ < 10) {
                        double reductionRate = (1 - (double)simplifiedGeometry.size() / originalSize) * 100;
                        log.info("[{}] Vessel {} simplified: {} -> {} points ({}% reduced, speed: {} knots)",
                                strategy, track.getVesselId(), originalSize, simplifiedGeometry.size(),
                                Math.round(reductionRate), track.getAvgSpeed());
                    }
                }
            } else {
                track.setTimestamps(adjustedTimestamps);
            }
        }

        return tracks;
    }

    /**
     * Daily 전략 처리 - 캐시 우선 조회 + 페이지별 즉시 스트리밍 (타임아웃 방지)
     */
    private void processDailyStrategy(List<TimeRange> ranges, TrackQueryRequest request, String queryId,
                                      Consumer<TrackChunkResponse> chunkConsumer,
                                      Consumer<QueryStatusUpdate> statusConsumer,
                                      int startChunkIndex,
                                      Set<String> uniqueVesselIds,
                                      Set<String> viewportVesselIds,
                                      QueryBenchmark benchmark) throws Exception {

        currentGlobalChunkIndex = startChunkIndex;

        // 전체 쿼리 동안 유지되는 세션 캐시 (모든 날짜에 걸쳐 공유)
        Map<String, VesselInfo> sessionVesselCache = new HashMap<>(20000);
        log.info("Session vessel cache initialized for {} date ranges", ranges.size());

        for (TimeRange range : ranges) {
            LocalDate rangeDate = range.getStart().toLocalDate();

            // 캐시 히트 체크: 해당 날짜가 인메모리 캐시에 있으면 DB 조회 생략
            if (dailyTrackCacheManager.isEnabled() && dailyTrackCacheManager.isCached(rangeDate)) {
                log.info("Daily cache HIT for {}: serving from memory", rangeDate);
                List<CompactVesselTrack> cachedTracks;

                // viewportVesselIds가 있으면 (2-pass 완료) → vessel ID로만 필터링
                // 공간 필터 재적용 금지: 다른 날짜에서 뷰포트 통과한 선박의 이 날짜 항적도 포함해야 함
                if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
                    cachedTracks = dailyTrackCacheManager.getCachedTracks(rangeDate);
                    cachedTracks = cachedTracks.stream()
                            .filter(t -> viewportVesselIds.contains(t.getVesselId()))
                            .collect(Collectors.toList());
                } else if (request.getViewport() != null) {
                    // 2-pass 없이 직접 뷰포트 필터 (processQueryInChunks 등)
                    ViewportFilter vp = request.getViewport();
                    cachedTracks = dailyTrackCacheManager.getCachedTracks(
                            rangeDate, vp.getMinLon(), vp.getMinLat(), vp.getMaxLon(), vp.getMaxLat());
                } else {
                    cachedTracks = dailyTrackCacheManager.getCachedTracks(rangeDate);
                }

                if (!cachedTracks.isEmpty()) {
                    // 간소화 전 지표 측정
                    int preTrackCount = cachedTracks.size();
                    int prePointCount = cachedTracks.stream()
                            .mapToInt(t -> t.getGeometry() != null ? t.getGeometry().size() : 0).sum();
                    List<List<CompactVesselTrack>> preBatches = splitByMessageSize(cachedTracks);
                    int preBatchCount = preBatches.size();
                    preBatches.clear();
                    preBatches = null;

                    // 방어적 복사 후 간소화 (캐시 원본 보호)
                    long simplifyStart = System.currentTimeMillis();
                    cachedTracks = cachedTracks.stream()
                            .map(t -> t.toBuilder().build())
                            .collect(Collectors.toList());
                    cachedTracks = cacheTrackSimplifier.simplify(cachedTracks, request.getZoomLevel());
                    long simplifyElapsed = System.currentTimeMillis() - simplifyStart;

                    // 간소화 후 지표 측정
                    int postPointCount = cachedTracks.stream()
                            .mapToInt(t -> t.getGeometry() != null ? t.getGeometry().size() : 0).sum();

                    // 메시지 크기로 분할하여 전송
                    List<List<CompactVesselTrack>> batches = splitByMessageSize(cachedTracks);
                    int postBatchCount = batches.size();
                    for (List<CompactVesselTrack> batch : batches) {
                        sendChunkResponse(batch, queryId, chunkConsumer, uniqueVesselIds);
                    }
                    batches.clear();
                    batches = null;
                    cachedTracks.clear();
                    cachedTracks = null;

                    log.info("[CacheHIT] date={}, zoom={}, tracks={}, points: {} -> {} ({}% 감소), batches: {} -> {} ({}% 감소), simplify: {}ms",
                            rangeDate, request.getZoomLevel(), preTrackCount,
                            prePointCount, postPointCount,
                            prePointCount > 0 ? Math.round((1 - (double) postPointCount / prePointCount) * 100) : 0,
                            preBatchCount, postBatchCount,
                            preBatchCount > 0 ? Math.round((1 - (double) postBatchCount / preBatchCount) * 100) : 0,
                            simplifyElapsed);

                    // [BENCHMARK] 벤치마크 누적 (캐시 경로)
                    if (benchmark != null) {
                        benchmark.cacheHitDays++;
                        benchmark.totalTracks += preTrackCount;
                        benchmark.totalPointsBefore += prePointCount;
                        benchmark.totalPointsAfter += postPointCount;
                        benchmark.batchesBeforeSimplify += preBatchCount;
                        benchmark.simplifyTimeMs += simplifyElapsed;
                    }
                }
            } else {
                // 캐시 미스: 기존 DB 페이지네이션으로 처리
                if (dailyTrackCacheManager.isEnabled()) {
                    log.info("Daily cache MISS for {}: falling back to DB", rangeDate);
                }
                if (benchmark != null) {
                    benchmark.dbQueryDays++; // [BENCHMARK]
                }
                streamDailyTableWithPagination(request, range, queryId, chunkConsumer, statusConsumer, uniqueVesselIds, sessionVesselCache, viewportVesselIds, benchmark);
            }
        }

        log.info("Session vessel cache final size: {} vessels cached", sessionVesselCache.size());
        sessionVesselCache.clear();
    }

    /**
     * 캐시 데이터를 청크 응답으로 전송하는 헬퍼 메서드
     */
    private void sendChunkResponse(List<CompactVesselTrack> batch, String queryId,
                                   Consumer<TrackChunkResponse> chunkConsumer,
                                   Set<String> uniqueVesselIds) {
        TrackChunkResponse response = new TrackChunkResponse();
        response.setQueryId(queryId);
        response.setChunkIndex(currentGlobalChunkIndex++);
        response.setIsLastChunk(false);
        response.setTotalChunks(-1);
        response.setCompactTracks(batch);

        int processedMin = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
        response.setStats(createChunkStats(batch, uniqueVesselIds, processedMin));

        chunkConsumer.accept(response);
        batch.forEach(track -> uniqueVesselIds.add(track.getVesselId()));

        // 백프레셔 적용
        int batchSize = batch.stream().mapToInt(this::estimateTrackSize).sum();
        pendingBufferSize.addAndGet(batchSize);
    }

    /**
     * Daily 테이블 페이지별 즉시 스트리밍 - 각 페이지 완료 즉시 청크 전송
     */
    private void streamDailyTableWithPagination(TrackQueryRequest request, TimeRange range, String queryId,
                                                 Consumer<TrackChunkResponse> chunkConsumer,
                                                 Consumer<QueryStatusUpdate> statusConsumer,
                                                 Set<String> uniqueVesselIds,
                                                 Map<String, VesselInfo> sessionVesselCache,
                                                 Set<String> viewportVesselIds,
                                                 QueryBenchmark benchmark) {
        String tableName = TableStrategy.DAILY.getTableName();

        // 줌 레벨에 따른 강화된 간소화 tolerance
        double tolerance = getEnhancedDailyTolerance(request.getZoomLevel());
        log.info("Daily streaming pagination started for range [{} - {}] with tolerance {} (zoom: {}, cacheSize: {})",
                range.getStart(), range.getEnd(), tolerance, request.getZoomLevel(), sessionVesselCache.size());

        // 세션 ID 추출 (쿼리 취소 확인용)
        String sessionId = queryId != null ? queryId.split("_")[0] : null;

        String lastMmsi = null;
        int pageNum = 0;
        int totalTrackCount = 0;
        int totalVesselsSent = 0;

        // 페이지네이션 루프 - 각 페이지 완료 즉시 전송
        while (true) {
            // 세션 연결 끊김 체크 (각 페이지 시작 전)
            if (sessionId != null && isQueryCancelled(sessionId)) {
                log.warn("Query cancelled before page {}: sessionId={}, totalTracks={}",
                        pageNum + 1, sessionId, totalTrackCount);
                return;
            }

            Map<String, VesselAccumulator> pageVesselMap = new HashMap<>(20000);
            String sql = buildDailyPaginationQuery(tableName, request, range, tolerance, lastMmsi, null, viewportVesselIds);

            try (Connection conn = queryDataSource.getConnection();
                 PreparedStatement ps = conn.prepareStatement(sql)) {

                int paramIndex = 1;
                ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getStart()));
                ps.setTimestamp(paramIndex++, Timestamp.valueOf(range.getEnd()));

                // 페이지네이션 파라미터
                if (lastMmsi != null) {
                    ps.setString(paramIndex++, lastMmsi);
                }

                // 2-pass 뷰포트 필터: vessel ID 배열 또는 기존 viewport 좌표 바인딩
                if (viewportVesselIds != null && !viewportVesselIds.isEmpty()) {
                    Array vesselArray = conn.createArrayOf("text", viewportVesselIds.toArray(new String[0]));
                    ps.setArray(paramIndex++, vesselArray);
                } else if (request.getViewport() != null) {
                    ViewportFilter vp = request.getViewport();
                    ps.setDouble(paramIndex++, vp.getMinLon());
                    ps.setDouble(paramIndex++, vp.getMinLat());
                    ps.setDouble(paramIndex++, vp.getMaxLon());
                    ps.setDouble(paramIndex++, vp.getMaxLat());
                }

                ps.setFetchSize(5000);

                long queryStartTime = System.currentTimeMillis();
                try (ResultSet rs = ps.executeQuery()) {
                    long queryEndTime = System.currentTimeMillis();
                    long pageQueryTime = queryEndTime - queryStartTime;
                    log.info("Daily page {} query executed in {}ms", pageNum + 1, pageQueryTime);

                    // [BENCHMARK] DB 쿼리 시간 누적
                    if (benchmark != null) {
                        benchmark.dbQueryTimeMs += pageQueryTime;
                        benchmark.connDailyPages++; // [BENCHMARK]
                    }

                    int pageTrackCount = 0;
                    String currentMmsi3 = null;

                    // 1단계: 먼저 모든 선박 ID를 수집
                    List<String[]> trackDataList = new ArrayList<>(DAILY_PAGE_SIZE);
                    Set<String> vesselIdsInPage = new HashSet<>();

                    long collectStartTime = System.currentTimeMillis();
                    while (rs.next()) {
                        String mmsi = rs.getString("mmsi");
                        vesselIdsInPage.add(mmsi);

                        // ResultSet 데이터를 임시 저장
                        trackDataList.add(new String[]{
                            mmsi,
                            rs.getString("track_geom"),
                            rs.getString("time_bucket"),
                            rs.getString("start_time"),
                            rs.getString("end_time"),
                            String.valueOf(rs.getDouble("distance_nm")),
                            String.valueOf(rs.getDouble("max_speed"))
                        });
                    }
                    long collectEndTime = System.currentTimeMillis();
                    log.info("Daily page {} data collected: {} tracks, {} vessels in {}ms",
                            pageNum + 1, trackDataList.size(), vesselIdsInPage.size(), collectEndTime - collectStartTime);

                    // 2단계: 선박 정보 배치 조회 (세션 캐시 → 전역 캐시 → DB 순서)
                    long batchStartTime = System.currentTimeMillis();
                    preloadVesselInfoWithSessionCache(vesselIdsInPage, sessionVesselCache, benchmark); // [BENCHMARK]
                    long batchEndTime = System.currentTimeMillis();
                    log.info("Daily page {} vessel info batch loaded in {}ms (session cache size: {})",
                            pageNum + 1, batchEndTime - batchStartTime, sessionVesselCache.size());

                    // 3단계: 수집된 데이터 처리
                    long processStartTime = System.currentTimeMillis();
                    for (String[] trackData : trackDataList) {
                        // 세션 연결 끊김 체크 (1000개마다)
                        if (sessionId != null && pageTrackCount % 1000 == 0 && pageTrackCount > 0) {
                            if (isQueryCancelled(sessionId)) {
                                log.warn("Query cancelled during page {} processing: sessionId={}, pageTrackCount={}",
                                        pageNum + 1, sessionId, pageTrackCount);
                                return;
                            }
                        }

                        currentMmsi3 = trackData[0];
                        String vesselId = currentMmsi3;
                        String trackGeomWkt = trackData[1];
                        String timeBucketStr = trackData[2];
                        String startTimeStr = trackData[3];
                        String endTimeStr = trackData[4];
                        double distanceNm = Double.parseDouble(trackData[5]);
                        double maxSpeed = Double.parseDouble(trackData[6]);

                        // 선박 객체는 geometry가 비어있어도 생성 (선박 누락 방지)
                        final String finalMmsi3 = currentMmsi3;
                        VesselAccumulator accumulator = pageVesselMap.computeIfAbsent(vesselId, k -> {
                            // 세션 캐시에서 조회 (이미 preload됨)
                            VesselInfo info = sessionVesselCache.get(finalMmsi3);
                            if (info == null) {
                                info = new VesselInfo(null, null);
                            }
                            VesselAccumulator acc = new VesselAccumulator();
                            acc.mmsi = finalMmsi3;
                            acc.shipName = info.shipName;
                            acc.shipType = info.shipType;
                            acc.shipKindCode = SignalKindCode.resolve(info.shipType, null).getCode();
                            return acc;
                        });

                        if (trackGeomWkt != null && !trackGeomWkt.isEmpty() && !"LINESTRING EMPTY".equals(trackGeomWkt)) {
                            try {
                                LineString lineString = (LineString) wktReader.read(trackGeomWkt);

                                // 빈 LineString은 건너뛰되, 선박 객체는 이미 생성됨
                                if (lineString.getNumPoints() == 0) {
                                    pageTrackCount++;
                                    continue;
                                }

                                // 좌표 배열로 변환
                                List<double[]> geometry = new ArrayList<>();
                                List<String> timestamps = new ArrayList<>();
                                List<Double> speeds = new ArrayList<>();

                                Coordinate prevCoord = null;
                                long prevTimeMillis = 0;

                                for (int i = 0; i < lineString.getNumPoints(); i++) {
                                    Coordinate coord = lineString.getCoordinateN(i);
                                    geometry.add(new double[]{coord.x, coord.y});

                                    long unixTimestamp = (long)coord.getM();
                                    timestamps.add(String.valueOf(unixTimestamp));

                                    double speed = 0.0;
                                    long currentTimeMillis = (long)coord.getM() * 1000;

                                    if (prevCoord != null && prevTimeMillis > 0) {
                                        double distance = calculateDistance(prevCoord.y, prevCoord.x, coord.y, coord.x);
                                        long timeDiff = currentTimeMillis - prevTimeMillis;
                                        if (timeDiff > 0) {
                                            speed = (distance / (timeDiff / 3600000.0));
                                        }
                                    }
                                    speeds.add(speed);

                                    prevCoord = coord;
                                    prevTimeMillis = currentTimeMillis;
                                }

                                accumulator.geometry.addAll(geometry);
                                accumulator.timestamps.addAll(timestamps);
                                accumulator.speeds.addAll(speeds);
                                accumulator.totalDistance += distanceNm;
                                accumulator.maxSpeed = Math.max(accumulator.maxSpeed, maxSpeed);
                                accumulator.pointCount += geometry.size();

                            } catch (ParseException e) {
                                log.error("Error parsing geometry: {}", e.getMessage());
                            }
                        }
                        pageTrackCount++;
                    }

                    long processEndTime = System.currentTimeMillis();
                    log.info("Daily page {} data processed in {}ms", pageNum + 1, processEndTime - processStartTime);

                    totalTrackCount += trackDataList.size();
                    trackDataList.clear();      // 메모리 즉시 해제: ResultSet 임시 데이터
                    vesselIdsInPage.clear();    // 메모리 즉시 해제: 페이지 선박 ID 집합

                    // 페이지 데이터를 즉시 청크로 전송
                    if (!pageVesselMap.isEmpty()) {
                        List<CompactVesselTrack> pageTracks = convertAccumulatorsToTracks(pageVesselMap);

                        // 통합선박 필터링 제거됨 (SNP API 전환: MMSI 단일 식별자)

                        totalVesselsSent += pageTracks.size();

                        // [BENCHMARK] DB 경로 포인트 수 누적
                        if (benchmark != null) {
                            int pagePoints = pageTracks.stream()
                                    .mapToInt(t -> t.getPointCount() != null ? t.getPointCount() : 0).sum();
                            benchmark.totalTracks += pageTracks.size();
                            benchmark.totalPointsBefore += pagePoints;
                            benchmark.totalPointsAfter += pagePoints; // DB 경로는 ST_Simplify 적용 후이므로 동일
                        }

                        // 메시지 크기로 분할하여 전송
                        List<List<CompactVesselTrack>> batches = splitByMessageSize(pageTracks);
                        for (List<CompactVesselTrack> batch : batches) {
                            TrackChunkResponse response = new TrackChunkResponse();
                            response.setQueryId(queryId);
                            response.setChunkIndex(currentGlobalChunkIndex++);
                            response.setIsLastChunk(false);
                            response.setTotalChunks(-1);
                            response.setCompactTracks(batch);

                            // 처리된 시간 계산
                            String dailyKey = "daily_" + range.getStart() + "_page" + pageNum;
                            processedTimeRanges.put(dailyKey, (int)Duration.between(range.getStart(), range.getEnd()).toMinutes() / Math.max(1, pageNum + 1));
                            int processedMin = processedTimeRanges.values().stream().mapToInt(Integer::intValue).sum();
                            response.setStats(createChunkStats(batch, uniqueVesselIds, processedMin));

                            // metrics 업데이트 (BACKPRESSURE 통계용)
                            BackpressureMetrics metrics = queryMetrics.get(queryId);
                            if (metrics != null) {
                                int chunkSize = batch.stream().mapToInt(t -> estimateTrackSize(t)).sum();
                                metrics.totalBytes.addAndGet(chunkSize);
                                metrics.chunkCount.incrementAndGet();
                            }

                            chunkConsumer.accept(response);

                            batch.forEach(track -> uniqueVesselIds.add(track.getVesselId()));

                            double timeProgress = (double) processedMin / estimatedTotalMinutes * 100;
                            statusConsumer.accept(new QueryStatusUpdate(
                                    queryId,
                                    "PROCESSING",
                                    String.format("Daily page %d sent (%d vessels)", pageNum + 1, batch.size()),
                                    Math.min(95.0, timeProgress)
                            ));

                            try {
                                Thread.sleep(10);
                            } catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                                log.warn("Thread interrupted during daily streaming");
                                return;
                            }
                        }

                        int pageVesselCount = pageTracks.size();
                        batches.clear();        // 메모리 즉시 해제: 배치 분할 리스트
                        pageTracks.clear();     // 메모리 즉시 해제: 페이지 항적 리스트

                        log.info("Daily streaming page {} sent: {} tracks, {} vessels (total: {} tracks, {} vessels)",
                                pageNum + 1, pageTrackCount, pageVesselCount, totalTrackCount, totalVesselsSent);
                    }

                    // 페이지가 완전히 채워지지 않았으면 마지막 페이지
                    if (pageTrackCount < DAILY_PAGE_SIZE) {
                        pageVesselMap.clear();  // 메모리 즉시 해제: 마지막 페이지 선박 맵
                        log.info("Daily streaming pagination completed: {} pages, {} total tracks, {} total vessels sent",
                                pageNum + 1, totalTrackCount, totalVesselsSent);
                        break;
                    }

                    // 다음 페이지를 위한 마지막 MMSI 저장
                    lastMmsi = currentMmsi3;
                    pageNum++;
                    pageVesselMap.clear();  // 메모리 즉시 해제: 페이지 선박 누적 맵
                }
            } catch (SQLException e) {
                log.error("Error in daily streaming pagination page {}: {}", pageNum, e.getMessage());
                break;
            }
        }
    }

    /**
     * VesselAccumulator Map을 CompactVesselTrack 리스트로 변환
     */
    private List<CompactVesselTrack> convertAccumulatorsToTracks(Map<String, VesselAccumulator> vesselMap) {
        return vesselMap.entrySet().stream()
                .map(entry -> {
                    String visselId = entry.getKey();
                    VesselAccumulator acc = entry.getValue();

                    double avgSpeed = 0.0;
                    if (acc.totalDistance > 0 && acc.timestamps.size() > 1) {
                        String firstTs = acc.timestamps.get(0);
                        String lastTs = acc.timestamps.get(acc.timestamps.size() - 1);

                        LocalDateTime startTime;
                        LocalDateTime endTime;

                        if (firstTs.matches("\\d{10,}")) {
                            startTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(firstTs)),
                                java.time.ZoneId.systemDefault());
                            endTime = LocalDateTime.ofInstant(
                                java.time.Instant.ofEpochSecond(Long.parseLong(lastTs)),
                                java.time.ZoneId.systemDefault());
                        } else {
                            startTime = LocalDateTime.parse(firstTs, TIMESTAMP_FORMATTER);
                            endTime = LocalDateTime.parse(lastTs, TIMESTAMP_FORMATTER);
                        }

                        double hours = Duration.between(startTime, endTime).toMinutes() / 60.0;
                        if (hours > 0) {
                            avgSpeed = acc.totalDistance / hours;
                        }
                    }

                    return CompactVesselTrack.builder()
                            .vesselId(visselId)
                            .nationalCode(acc.mmsi != null && acc.mmsi.length() >= 3 ? acc.mmsi.substring(0, 3) : null)
                            .shipName(acc.shipName)
                            .shipType(acc.shipType)
                            .shipKindCode(acc.shipKindCode)
                            .geometry(acc.geometry)
                            .timestamps(acc.timestamps)
                            .speeds(acc.speeds)
                            .totalDistance(acc.totalDistance)
                            .avgSpeed(avgSpeed)
                            .maxSpeed(acc.maxSpeed)
                            .pointCount(acc.pointCount)
                            .build();
                })
                .collect(Collectors.toList());
    }

    /**
     * 쿼리 중단 여부 확인
     */
    private boolean isQueryCancelled(String sessionId) {
        if (sessionId == null) {
            return false;
        }

        // 1. queryCancelFlags 확인 (cancelQuery() 호출 시 설정됨)
        for (Map.Entry<String, AtomicBoolean> entry : queryCancelFlags.entrySet()) {
            if (entry.getValue().get()) {
                return true;
            }
        }

        // 2. ActiveQueryManager 확인 (세션 종료 시 설정됨)
        ActiveQueryManager.ActiveQuery query = activeQueryManager.getQuery(sessionId);
        return query != null && query.isCancelled();
    }

}