gc/signal-batch
7
0
포크 0
코드 이슈 풀 리퀘스트 Actions 위키 활동
8f784de358
signal-batch/scripts/test-hourly-aggregation-fixed.sql
htlee 2e9361ee58 refactor: SNP API 전환 및 레거시 코드 전면 정리 
- CollectDB 다중 신호 수집 → S&P Global AIS API 단일 수집으로 전환
- sig_src_cd + target_id 이중 식별자 → mmsi(VARCHAR) 단일 식별자
- t_vessel_latest_position → t_ais_position 테이블 전환
- 레거시 배치/유틸 ~30개 클래스 삭제 (VesselAggregationJobConfig, ShipKindCodeConverter 등)
- AisTargetCacheManager 기반 캐시 이중 구조 (최신위치 + 트랙 버퍼)
- CacheBasedVesselTrackDataReader + CacheBasedTrackJobListener 신규 추가
- VesselStaticStepConfig: 정적정보 CDC 변경 검출 + hourly job 편승
- SignalKindCode enum: vesselType/extraInfo 기반 선종 자동 분류
- WebSocket/STOMP 전체 mmsi 전환 (StompTrackStreamingService ~40곳)
- 모니터링/성능 최적화 코드 mmsi 기반 전환
- DataSource 설정 통합 (snpdb 단일 DB)
- AreaBoundaryCache Polygon→Geometry 캐스트 수정 (MULTIPOLYGON 지원)
- ConcurrentHashMap 적용 (VesselTrackStepConfig 동시성 버그 수정)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-19 09:59:49 +09:00

485 lines
18 KiB
SQL
Raw Blame 히스토리

-- ========================================
-- 시간별 집계 쿼리 검증 스크립트
-- CAST 및 타입 호환성 테스트
-- ========================================
-- 1. 임시 테스트 테이블 생성
DROP TABLE IF EXISTS test_vessel_tracks_5min CASCADE;
DROP TABLE IF EXISTS test_vessel_tracks_hourly CASCADE;
CREATE TABLE test_vessel_tracks_5min (
sig_src_cd VARCHAR(10),
target_id VARCHAR(20),
time_bucket TIMESTAMP,
track_geom geometry(LineStringM, 4326),
distance_nm NUMERIC(10,2),
avg_speed NUMERIC(6,2),
max_speed NUMERIC(6,2),
point_count INTEGER,
start_position JSONB,
end_position JSONB,
PRIMARY KEY (sig_src_cd, target_id, time_bucket)
);
CREATE TABLE test_vessel_tracks_hourly (
sig_src_cd VARCHAR(10),
target_id VARCHAR(20),
time_bucket TIMESTAMP,
track_geom geometry(LineStringM, 4326),
distance_nm NUMERIC(10,2),
avg_speed NUMERIC(6,2),
max_speed NUMERIC(6,2),
point_count INTEGER,
start_position JSONB,
end_position JSONB,
PRIMARY KEY (sig_src_cd, target_id, time_bucket)
);
-- 2. 샘플 데이터 삽입 (1시간치 5분 간격 데이터)
-- 시나리오 1: 정상 이동 선박
INSERT INTO test_vessel_tracks_5min VALUES
(
'000001',
'TEST001',
'2025-01-07 10:00:00',
public.ST_GeomFromText('LINESTRING M(126.5 37.5 1736215200, 126.51 37.51 1736215260, 126.52 37.52 1736215320)', 4326),
0.5,
10.5,
12.0,
3,
'{"lat": 37.5, "lon": 126.5, "time": "2025-01-07 10:00:00", "sog": 10.5}'::jsonb,
'{"lat": 37.52, "lon": 126.52, "time": "2025-01-07 10:02:00", "sog": 11.0}'::jsonb
),
(
'000001',
'TEST001',
'2025-01-07 10:05:00',
public.ST_GeomFromText('LINESTRING M(126.52 37.52 1736215500, 126.53 37.53 1736215560, 126.54 37.54 1736215620)', 4326),
0.6,
11.0,
13.0,
3,
'{"lat": 37.52, "lon": 126.52, "time": "2025-01-07 10:05:00", "sog": 11.0}'::jsonb,
'{"lat": 37.54, "lon": 126.54, "time": "2025-01-07 10:07:00", "sog": 12.0}'::jsonb
),
(
'000001',
'TEST001',
'2025-01-07 10:10:00',
public.ST_GeomFromText('LINESTRING M(126.54 37.54 1736215800, 126.55 37.55 1736215860)', 4326),
0.4,
9.5,
11.0,
2,
'{"lat": 37.54, "lon": 126.54, "time": "2025-01-07 10:10:00", "sog": 9.5}'::jsonb,
'{"lat": 37.55, "lon": 126.55, "time": "2025-01-07 10:11:00", "sog": 10.0}'::jsonb
);
-- 시나리오 2: 정박 선박 (같은 좌표 반복)
INSERT INTO test_vessel_tracks_5min VALUES
(
'000002',
'TEST002',
'2025-01-07 10:00:00',
public.ST_GeomFromText('LINESTRING M(129.0 35.0 1736215200, 129.0 35.0 1736215260)', 4326),
0.0,
0.0,
0.5,
2,
'{"lat": 35.0, "lon": 129.0, "time": "2025-01-07 10:00:00", "sog": 0.0}'::jsonb,
'{"lat": 35.0, "lon": 129.0, "time": "2025-01-07 10:01:00", "sog": 0.0}'::jsonb
),
(
'000002',
'TEST002',
'2025-01-07 10:05:00',
public.ST_GeomFromText('LINESTRING M(129.0 35.0 1736215500, 129.0 35.0 1736215560)', 4326),
0.0,
0.0,
0.3,
2,
'{"lat": 35.0, "lon": 129.0, "time": "2025-01-07 10:05:00", "sog": 0.0}'::jsonb,
'{"lat": 35.0, "lon": 129.0, "time": "2025-01-07 10:06:00", "sog": 0.0}'::jsonb
);
-- 시나리오 3: 단일 포인트 (중복 포인트로 유효한 LineString)
INSERT INTO test_vessel_tracks_5min VALUES
(
'000003',
'TEST003',
'2025-01-07 10:00:00',
public.ST_GeomFromText('LINESTRING M(130.0 36.0 1736215200, 130.0 36.0 1736215200)', 4326),
0.0,
0.0,
0.0,
1,
'{"lat": 36.0, "lon": 130.0, "time": "2025-01-07 10:00:00", "sog": 0.0}'::jsonb,
'{"lat": 36.0, "lon": 130.0, "time": "2025-01-07 10:00:00", "sog": 0.0}'::jsonb
);
-- 3. 입력 데이터 검증
SELECT
'=== INPUT DATA VALIDATION ===' as section,
sig_src_cd,
target_id,
time_bucket,
public.ST_NPoints(track_geom) as points,
public.ST_IsValid(track_geom) as is_valid,
public.ST_AsText(track_geom) as wkt
FROM test_vessel_tracks_5min
ORDER BY sig_src_cd, target_id, time_bucket;
-- 4. 실제 HourlyTrackProcessor SQL 실행 (CAST 사용)
-- Vessel: 000001_TEST001, Hour: 2025-01-07 10:00:00
WITH ordered_tracks AS (
SELECT *
FROM test_vessel_tracks_5min
WHERE sig_src_cd = '000001'
AND target_id = 'TEST001'
AND time_bucket >= CAST('2025-01-07 10:00:00' AS timestamp)
AND time_bucket < CAST('2025-01-07 11:00:00' AS timestamp)
AND track_geom IS NOT NULL
AND public.ST_NPoints(track_geom) > 0
ORDER BY time_bucket
),
merged_coords AS (
SELECT
sig_src_cd,
target_id,
string_agg(
COALESCE(
substring(public.ST_AsText(track_geom) from 'LINESTRING\\s*M\\s*\\((.+)\\)'),
substring(public.ST_AsText(track_geom) from '\\((.+)\\)')
),
','
ORDER BY time_bucket
) FILTER (WHERE track_geom IS NOT NULL) as all_coords
FROM ordered_tracks
GROUP BY sig_src_cd, target_id
),
merged_tracks AS (
SELECT
mc.sig_src_cd,
mc.target_id,
CAST('2025-01-07 10:00:00' AS timestamp) as time_bucket,
public.ST_GeomFromText('LINESTRING M(' || mc.all_coords || ')', 4326) as merged_geom,
(SELECT MAX(max_speed) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as max_speed,
(SELECT SUM(point_count) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as total_points,
(SELECT MIN(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as start_time,
(SELECT MAX(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as end_time,
(SELECT start_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket LIMIT 1) as start_pos,
(SELECT end_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket DESC LIMIT 1) as end_pos
FROM merged_coords mc
),
calculated_tracks AS (
SELECT
*,
public.ST_Length(merged_geom::geography) / 1852.0 as total_distance,
CASE
WHEN public.ST_NPoints(merged_geom) > 0 THEN
public.ST_M(public.ST_PointN(merged_geom, public.ST_NPoints(merged_geom))) -
public.ST_M(public.ST_PointN(merged_geom, 1))
ELSE
EXTRACT(EPOCH FROM
CAST(end_pos->>'time' AS timestamp) - CAST(start_pos->>'time' AS timestamp)
)
END as time_diff_seconds
FROM merged_tracks
)
SELECT
'=== HOURLY AGGREGATION RESULT (VESSEL 000001_TEST001) ===' as section,
sig_src_cd,
target_id,
time_bucket,
public.ST_NPoints(merged_geom) as merged_points,
public.ST_IsValid(merged_geom) as is_valid,
total_distance,
CASE
WHEN time_diff_seconds > 0 THEN
CAST(LEAST((total_distance / (time_diff_seconds / 3600.0)), 9999.99) AS numeric(6,2))
ELSE 0
END as avg_speed,
max_speed,
total_points,
start_time,
end_time,
start_pos,
end_pos,
public.ST_AsText(merged_geom) as geom_text
FROM calculated_tracks;
-- 5. INSERT 테스트 (CAST 호환성 검증)
INSERT INTO test_vessel_tracks_hourly
WITH ordered_tracks AS (
SELECT *
FROM test_vessel_tracks_5min
WHERE sig_src_cd = '000001'
AND target_id = 'TEST001'
AND time_bucket >= CAST('2025-01-07 10:00:00' AS timestamp)
AND time_bucket < CAST('2025-01-07 11:00:00' AS timestamp)
AND track_geom IS NOT NULL
AND public.ST_NPoints(track_geom) > 0
ORDER BY time_bucket
),
merged_coords AS (
SELECT
sig_src_cd,
target_id,
string_agg(
COALESCE(
substring(public.ST_AsText(track_geom) from 'LINESTRING\\s*M\\s*\\((.+)\\)'),
substring(public.ST_AsText(track_geom) from '\\((.+)\\)')
),
','
ORDER BY time_bucket
) FILTER (WHERE track_geom IS NOT NULL) as all_coords
FROM ordered_tracks
GROUP BY sig_src_cd, target_id
),
merged_tracks AS (
SELECT
mc.sig_src_cd,
mc.target_id,
CAST('2025-01-07 10:00:00' AS timestamp) as time_bucket,
public.ST_GeomFromText('LINESTRING M(' || mc.all_coords || ')', 4326) as merged_geom,
(SELECT MAX(max_speed) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as max_speed,
(SELECT SUM(point_count) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as total_points,
(SELECT MIN(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as start_time,
(SELECT MAX(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as end_time,
(SELECT start_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket LIMIT 1) as start_pos,
(SELECT end_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket DESC LIMIT 1) as end_pos
FROM merged_coords mc
),
calculated_tracks AS (
SELECT
*,
public.ST_Length(merged_geom::geography) / 1852.0 as total_distance,
CASE
WHEN public.ST_NPoints(merged_geom) > 0 THEN
public.ST_M(public.ST_PointN(merged_geom, public.ST_NPoints(merged_geom))) -
public.ST_M(public.ST_PointN(merged_geom, 1))
ELSE
EXTRACT(EPOCH FROM
CAST(end_pos->>'time' AS timestamp) - CAST(start_pos->>'time' AS timestamp)
)
END as time_diff_seconds
FROM merged_tracks
)
SELECT
sig_src_cd,
target_id,
time_bucket,
merged_geom as track_geom,
total_distance as distance_nm,
CASE
WHEN time_diff_seconds > 0 THEN
CAST(LEAST((total_distance / (time_diff_seconds / 3600.0)), 9999.99) AS numeric(6,2))
ELSE 0
END as avg_speed,
max_speed,
total_points as point_count,
start_pos as start_position,
end_pos as end_position
FROM calculated_tracks;
-- 6. 정박 선박 INSERT 테스트
INSERT INTO test_vessel_tracks_hourly
WITH ordered_tracks AS (
SELECT *
FROM test_vessel_tracks_5min
WHERE sig_src_cd = '000002'
AND target_id = 'TEST002'
AND time_bucket >= CAST('2025-01-07 10:00:00' AS timestamp)
AND time_bucket < CAST('2025-01-07 11:00:00' AS timestamp)
AND track_geom IS NOT NULL
AND public.ST_NPoints(track_geom) > 0
ORDER BY time_bucket
),
merged_coords AS (
SELECT
sig_src_cd,
target_id,
string_agg(
COALESCE(
substring(public.ST_AsText(track_geom) from 'LINESTRING\\s*M\\s*\\((.+)\\)'),
substring(public.ST_AsText(track_geom) from '\\((.+)\\)')
),
','
ORDER BY time_bucket
) FILTER (WHERE track_geom IS NOT NULL) as all_coords
FROM ordered_tracks
GROUP BY sig_src_cd, target_id
),
merged_tracks AS (
SELECT
mc.sig_src_cd,
mc.target_id,
CAST('2025-01-07 10:00:00' AS timestamp) as time_bucket,
public.ST_GeomFromText('LINESTRING M(' || mc.all_coords || ')', 4326) as merged_geom,
(SELECT MAX(max_speed) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as max_speed,
(SELECT SUM(point_count) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as total_points,
(SELECT MIN(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as start_time,
(SELECT MAX(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as end_time,
(SELECT start_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket LIMIT 1) as start_pos,
(SELECT end_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket DESC LIMIT 1) as end_pos
FROM merged_coords mc
),
calculated_tracks AS (
SELECT
*,
public.ST_Length(merged_geom::geography) / 1852.0 as total_distance,
CASE
WHEN public.ST_NPoints(merged_geom) > 0 THEN
public.ST_M(public.ST_PointN(merged_geom, public.ST_NPoints(merged_geom))) -
public.ST_M(public.ST_PointN(merged_geom, 1))
ELSE
EXTRACT(EPOCH FROM
CAST(end_pos->>'time' AS timestamp) - CAST(start_pos->>'time' AS timestamp)
)
END as time_diff_seconds
FROM merged_tracks
)
SELECT
sig_src_cd,
target_id,
time_bucket,
merged_geom as track_geom,
total_distance as distance_nm,
CASE
WHEN time_diff_seconds > 0 THEN
CAST(LEAST((total_distance / (time_diff_seconds / 3600.0)), 9999.99) AS numeric(6,2))
ELSE 0
END as avg_speed,
max_speed,
total_points as point_count,
start_pos as start_position,
end_pos as end_position
FROM calculated_tracks;
-- 7. 단일 포인트 선박 INSERT 테스트
INSERT INTO test_vessel_tracks_hourly
WITH ordered_tracks AS (
SELECT *
FROM test_vessel_tracks_5min
WHERE sig_src_cd = '000003'
AND target_id = 'TEST003'
AND time_bucket >= CAST('2025-01-07 10:00:00' AS timestamp)
AND time_bucket < CAST('2025-01-07 11:00:00' AS timestamp)
AND track_geom IS NOT NULL
AND public.ST_NPoints(track_geom) > 0
ORDER BY time_bucket
),
merged_coords AS (
SELECT
sig_src_cd,
target_id,
string_agg(
COALESCE(
substring(public.ST_AsText(track_geom) from 'LINESTRING\\s*M\\s*\\((.+)\\)'),
substring(public.ST_AsText(track_geom) from '\\((.+)\\)')
),
','
ORDER BY time_bucket
) FILTER (WHERE track_geom IS NOT NULL) as all_coords
FROM ordered_tracks
GROUP BY sig_src_cd, target_id
),
merged_tracks AS (
SELECT
mc.sig_src_cd,
mc.target_id,
CAST('2025-01-07 10:00:00' AS timestamp) as time_bucket,
public.ST_GeomFromText('LINESTRING M(' || mc.all_coords || ')', 4326) as merged_geom,
(SELECT MAX(max_speed) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as max_speed,
(SELECT SUM(point_count) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as total_points,
(SELECT MIN(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as start_time,
(SELECT MAX(time_bucket) FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id) as end_time,
(SELECT start_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket LIMIT 1) as start_pos,
(SELECT end_position FROM ordered_tracks WHERE sig_src_cd = mc.sig_src_cd AND target_id = mc.target_id ORDER BY time_bucket DESC LIMIT 1) as end_pos
FROM merged_coords mc
),
calculated_tracks AS (
SELECT
*,
public.ST_Length(merged_geom::geography) / 1852.0 as total_distance,
CASE
WHEN public.ST_NPoints(merged_geom) > 0 THEN
public.ST_M(public.ST_PointN(merged_geom, public.ST_NPoints(merged_geom))) -
public.ST_M(public.ST_PointN(merged_geom, 1))
ELSE
EXTRACT(EPOCH FROM
CAST(end_pos->>'time' AS timestamp) - CAST(start_pos->>'time' AS timestamp)
)
END as time_diff_seconds
FROM merged_tracks
)
SELECT
sig_src_cd,
target_id,
time_bucket,
merged_geom as track_geom,
total_distance as distance_nm,
CASE
WHEN time_diff_seconds > 0 THEN
CAST(LEAST((total_distance / (time_diff_seconds / 3600.0)), 9999.99) AS numeric(6,2))
ELSE 0
END as avg_speed,
max_speed,
total_points as point_count,
start_pos as start_position,
end_pos as end_position
FROM calculated_tracks;
-- 8. 최종 결과 검증
SELECT
'=== FINAL HOURLY AGGREGATION RESULTS ===' as section,
sig_src_cd,
target_id,
time_bucket,
public.ST_NPoints(track_geom) as points,
public.ST_IsValid(track_geom) as is_valid,
distance_nm,
avg_speed,
max_speed,
point_count,
public.ST_AsText(track_geom) as wkt
FROM test_vessel_tracks_hourly
ORDER BY sig_src_cd, target_id;
-- 9. 타입 검증
SELECT
'=== DATA TYPE VALIDATION ===' as section,
pg_typeof(time_bucket) as time_bucket_type,
pg_typeof(track_geom) as track_geom_type,
pg_typeof(distance_nm) as distance_type,
pg_typeof(avg_speed) as avg_speed_type,
pg_typeof(max_speed) as max_speed_type,
pg_typeof(point_count) as point_count_type,
pg_typeof(start_position) as start_position_type
FROM test_vessel_tracks_hourly
LIMIT 1;
-- 10. 시간 순서 검증 (M값이 증가하는지 확인)
SELECT
'=== TIME ORDERING VALIDATION ===' as section,
sig_src_cd,
target_id,
public.ST_M(public.ST_PointN(track_geom, 1)) as first_m_value,
public.ST_M(public.ST_PointN(track_geom, public.ST_NPoints(track_geom))) as last_m_value,
CASE
WHEN public.ST_M(public.ST_PointN(track_geom, public.ST_NPoints(track_geom))) >=
public.ST_M(public.ST_PointN(track_geom, 1))
THEN 'PASS'
ELSE 'FAIL'
END as time_order_check
FROM test_vessel_tracks_hourly;
-- 11. 정리
DROP TABLE IF EXISTS test_vessel_tracks_5min CASCADE;
DROP TABLE IF EXISTS test_vessel_tracks_hourly CASCADE;
-- ========================================
-- 테스트 완료
-- 모든 INSERT가 성공하고 타입 에러가 없으면 CAST 사용이 정상
-- ========================================
Reference in New Issue View Git Blame Copy Permalink