Databend 十一月月报：多模态查询智能

-- 连接到 CityDrive 的 S3 存储桶
CREATE OR REPLACE CONNECTION citydrive_s3
    STORAGE_TYPE='s3'
    ACCESS_KEY_ID='<AWS_ACCESS_KEY_ID>'
    SECRET_ACCESS_KEY='<AWS_SECRET_ACCESS_KEY>';

-- 指向原始多模态导出的共享 Stage
CREATE OR REPLACE STAGE citydrive_stage
    URL='s3://citydrive-lakehouse/raw/'
    CONNECTION=(CONNECTION_NAME='citydrive_s3')
    FILE_FORMAT=(TYPE='PARQUET');

-- 自动化 ETL：每 10 分钟加载一次视频帧元数据
CREATE OR REPLACE TASK task_load_citydrive_videos
    WAREHOUSE='default'
    SCHEDULE=10 MINUTE
AS
COPY INTO citydrive_videos (
    video_id,
    vehicle_id,
    capture_date,
    route_name,
    weather,
    camera_source,
    duration_sec
) FROM (
    SELECT
        video_id::STRING,
        vehicle_id::STRING,
        capture_date::DATE,
        route_name::STRING,
        weather::STRING,
        camera_source::STRING,
        duration_sec::INT
    FROM @citydrive_stage/videos/
) FILE_FORMAT=(TYPE='PARQUET');

-- 自动化 ETL：同步加载帧数据
CREATE OR REPLACE TASK task_load_frame_events
    WAREHOUSE='default'
    SCHEDULE=10 MINUTE
AS
COPY INTO frame_events (
    frame_id,
    video_id,
    frame_index,
    collected_at,
    event_tag,
    risk_score,
    speed_kmh
) FROM (
    SELECT
        frame_id::STRING,
        video_id::STRING,
        frame_index::INT,
        collected_at::TIMESTAMP,
        event_tag::STRING,
        risk_score::DOUBLE,
        speed_kmh::DOUBLE
    FROM @citydrive_stage/frame-events/
) FILE_FORMAT=(TYPE='PARQUET');

-- 激活任务，数据自动导入
ALTER TASK task_load_citydrive_videos RESUME;
ALTER TASK task_load_frame_events RESUME;

CREATE OR REPLACE TABLE citydrive_videos (...);
CREATE OR REPLACE TABLE frame_events (...);

WITH recent_videos AS (
    SELECT * FROM citydrive_videos
    WHERE capture_date >= '2025-01-01' AND capture_date < '2025-01-04'
)
SELECT v.video_id, v.route_name, COUNT(f.frame_id) AS flagged_frames
FROM recent_videos v
LEFT JOIN frame_events f USING(video_id)
GROUP BY v.video_id, v.route_name
ORDER BY flagged_frames DESC;

CREATE OR REPLACE TABLE frame_metadata_catalog (
    doc_id STRING,
    meta_json VARIANT,
    captured_at TIMESTAMP,
    INVERTED INDEX idx_meta_json(meta_json)
);

SELECT doc_id, captured_at
FROM frame_metadata_catalog
WHERE QUERY('meta_json.scene.weather_code:rain AND meta_json.camera.sensor_view:roof')
ORDER BY captured_at;

CREATE OR REPLACE TABLE frame_embeddings (
    frame_id STRING,
    video_id STRING,
    sensor_view STRING,
    embedding VECTOR(512),
    encoder_build STRING,
    created_at TIMESTAMP,
    VECTOR INDEX idx_frame_embeddings(embedding) DISTANCE='cosine'
);

WITH query_embedding AS (
    SELECT embedding FROM frame_embeddings WHERE frame_id='FRAME-0101'
)
SELECT e.frame_id, e.video_id,
       COSINE_DISTANCE(e.embedding, q.embedding) AS distance
FROM frame_embeddings e
CROSS JOIN query_embedding q
ORDER BY distance
LIMIT 3;

CREATE OR REPLACE TABLE frame_geo_points (
    video_id STRING,
    frame_id STRING,
    position_wgs84 GEOMETRY,
    solution_grade INT,
    source_system STRING,
    created_at TIMESTAMP
);

SELECT f.frame_id, ST_DISTANCE(g.position_wgs84, s.signal_position) AS meters_to_signal
FROM frame_geo_points g
JOIN signal_contact_points s USING(frame_id)
JOIN frame_events f USING(frame_id)
WHERE ST_WITHIN(g.position_wgs84,
      TO_GEOMETRY('SRID=4326;POLYGON((114.05 22.54, 114.13 22.54, 114.13 22.57, 114.05 22.57, 114.05 22.54))'));

WITH json_hits AS (
    SELECT doc_id
    FROM frame_metadata_catalog
    WHERE QUERY('meta_json.media_meta.tagging.labels:pedestrian')
),
vector_hits AS (
    SELECT frame_id
    FROM frame_embeddings
    WHERE COSINE_DISTANCE(
          embedding,
          (SELECT embedding FROM frame_embeddings WHERE frame_id='FRAME-0102')
    ) < 0.3
)
SELECT f.frame_id, v.route_name, s.distance_m
FROM frame_events f
JOIN citydrive_videos v USING(video_id)
JOIN json_hits j ON j.doc_id = f.frame_id
JOIN vector_hits vh ON vh.frame_id = f.frame_id
JOIN signal_contact_points s USING(frame_id);