커서를 사용한 쿼리

커서를 사용한 쿼리 (Query Using Cursors)

커서를 사용하면 클라이언트가 데이터베이스 커서처럼 Pinot에서 큰 결과 집합을 청크 단위로 가져올 수 있어요. 이는 클라이언트와 서버 양쪽의 메모리 소비를 줄여줘요.

출처: 문서

본문

아키텍처 개요 (Architecture Overview)

커서 응답은 이제 브로커 수준에서 관리되며, 각 브로커가 자체 로컬 정리 스케줄 실행기를 실행해요. controller는 더 이상 커서 응답 정리를 관리하지 않아요.

구성 (Configuration)

브로커 커서 정리 구성

브로커 구성에 다음 속성을 추가하세요.

pinot.broker.response_store.cleanup_interval_ms: 브로커가 만료된 커서 응답을 정리하는 간격(밀리초)(기본값: 3600000, 즉 1시간)
pinot.broker.response_store.max_age_ms: 커서 응답이 만료로 간주되기 전의 최대 수명(밀리초)

브로커 구성 예시

pinot.broker.response_store.cleanup_interval_ms=3600000
pinot.broker.response_store.max_age_ms=604800000

REST API

커서 응답 가져오기

커서 ID를 사용해 다음 결과 청크를 가져와요.

엔드포인트: GET /responseStore/{requestId}

쿼리 파라미터:

requestId (필수): 커서/응답 ID

예시:

curl -X GET "http://localhost:8099/responseStore/my-cursor-id"

특정 커서 응답 삭제

ID로 특정 커서 응답을 삭제해요.

엔드포인트: DELETE /responseStore/{requestId}

예시:

curl -X DELETE "http://localhost:8099/responseStore/my-cursor-id"

만료된 커서 응답 일괄 삭제 (NEW)

컷오프 시간 기준으로 만료된 모든 커서 응답을 삭제해요. 예약된 정리를 기다리는 대신 수동으로 정리를 트리거하려는 운영자에게 유용해요.

엔드포인트: DELETE /responseStore/

쿼리 파라미터:

expiredBefore (선택): 에포크 밀리초 컷오프 시간. expirationTimeMs가 이 값 이하인 응답이 삭제돼요. 생략하면 현재 시간이 기본값이에요.

예시:

# 현재 시간 이전에 만료된 모든 응답 삭제
curl -X DELETE "http://localhost:8099/responseStore/"

# 특정 타임스탬프(예: 한 시간 전) 이전에 만료된 모든 응답 삭제
curl -X DELETE "http://localhost:8099/responseStore/?expiredBefore=1682083200000"

응답:

{
  "message": "Deleted 42 expired response(s)."
}

하위 호환되지 않는 변경 내용 (Backward-Incompatible Change)

경고: PR #18203이 포함된 Apache Pinot 버전부터 커서 응답 정리가 controller에서 각 브로커로 이동했어요. controller는 더 이상 ResponseStoreCleaner 주기 작업을 실행하지 않아요.

롤링 업그레이드 중 운영 영향:

먼저 업그레이드된 controller는 더 이상 커서 응답을 정리하지 않음

브로커가 자체 정리를 활성화하려면 새 버전으로 업그레이드해야 함

브로커가 즉시 업그레이드되지 않으면 만료된 커서 응답이 일시적으로 축적될 수 있음

권장: controller 업그레이드 직후 모든 브로커를 새 버전으로 업그레이드하세요.

전환 중 수동 정리가 필요하면 새 일괄 삭제 API인 DELETE /responseStore/를 사용하세요.

커서로 큰 결과 집합 가져오기: 사용 예시

import requests
import json

BASE_URL = "http://localhost:8099"

# Execute query and get initial results
response = requests.get(f"{BASE_URL}/query", params={
    "sql": "SELECT * FROM my_table LIMIT 1000000"
})

data = response.json()
cursor_id = data.get("responseId")
results = data.get("resultTable", {}).get("rows", [])

print(f"Cursor ID: {cursor_id}")
print(f"Initial rows: {len(results)}")

# Fetch next chunks using cursor
while cursor_id:
    response = requests.get(f"{BASE_URL}/responseStore/{cursor_id}")
    if response.status_code != 200:
        break
    
    data = response.json()
    results = data.get("rows", [])
    cursor_id = data.get("nextCursorId")
    
    print(f"Fetched {len(results)} more rows")
    # Process results...

# Clean up: delete cursor when done (optional, cleaned up automatically)
# requests.delete(f"{BASE_URL}/responseStore/{cursor_id}")

커서를 사용한 쿼리