refactor(app): #231 derivation cache 배칭 + 계측 — 가설은 재현 안 됨
#231 은 "캐시가 히트하는데도 우회하고 전량 재임베딩하는 편이 더 빠르다" 고 보고하면서 본문에 "⚠️ 정량 실측 보완 필요 … 수치 미기록" 이라고 표를 비워 뒀다. 그 표를 채우는 것이 이 커밋의 핵심이다. 측정 (나무위키 792문서 / 16,379 chunk, ollama arctic-embed2 1024-dim, --force-reingest 로 전 문서 full re-process): 캐시 히트 경로 139.1초 236 chunk/초 캐시 우회(전량 재임베딩) 1179.6초 28 chunk/초 캐시 히트가 8.5배 빠르다. 가설은 이 환경에서 재현되지 않는다. 새로 넣은 계측으로 139초의 내역을 보면 더 분명하다 (히트 32,758 / 미스 0): Lance upsert + 레코드 구성 74.6초 53% SQLite 문서·청크 기록 56.7초 41% chunk 4.4초 3% 캐시 경로 전체(조회+삽입+touch) 0.6초 0.4% 이슈가 지목한 여섯 원인이 전부 합쳐 run 의 0.4% 다. 다만 "보고가 틀렸다" 로 읽으면 안 된다. 보고 이후 #229 와 #230 이 머지됐고, #231 본문 스스로 #229 를 "같은 Mutex<Connection> 을 공유하므로 상호 증폭" 이라고 적었다. 원 보고 환경에서는 캐시 조회 52,000회가 그 뮤텍스를 잡았다 놓는데 같은 뮤텍스 위에서 chunk 삭제가 FTS5 전체 스캔을 돌리고 있었다. "#229 를 먼저 고치면 체감이 줄어든다" 도 이슈의 예측이다. 이 머신이 61 GB RAM 이라 DB 가 통째로 페이지 캐시에 올라간다는 점도 함께 적어 둔다. 반영한 것: 제안 1·2·4 는 실측과 무관하게 왕복이 줄 뿐 잃는 게 없어 넣었다. 다만 A/B 벽시계는 139.1초 → 139.3초로 측정 오차 안이다. 이 코퍼스에서는 체감이 없다. - derivation_cache_get_many — `WHERE cache_key IN (…)` 배치 조회. 문서 하나가 평균 21 chunk 이라 왕복이 21회에서 1회가 된다. - derivation_cache_put_many — 미스 벡터를 한 트랜잭션에. 기존 단건 put 은 명시 트랜잭션 밖이라 행마다 암묵 커밋이었다. - prepare_cached — get/put/touch 셋 다. query_row 는 호출마다 SQL 을 다시 파싱한다. 제안 5(계측 노출)가 실질 산출물이다. `asset_timings` 에 cache_hit / cache_miss / cache_ms 를 additive 로 실었다. 이전에는 hit/miss 가 tracing::info! 로 stderr 에만 나가 run 이 끝나면 사라졌고, "내 코퍼스에서 캐시가 이득인가" 를 확인할 방법이 없었다. cache_ms 에는 touch 도 포함한다 — 이슈의 가장 날카로운 지적이 "읽기 전용이어야 할 히트 경로가 쓰기를 만든다" 인데, touch 를 빼고 재는 지표로는 그 주장을 검증할 수 없다. 네 out-param 은 CacheStats 구조체로 묶었다. 함께 읽히고 함께 보고되는 값들이고, 셋만 갱신하고 하나를 빠뜨리면 캐시가 공짜인 것처럼 보고된다. 반영하지 않은 것: - 제안 3(touch 를 히트 경로에서 분리). 캐시 경로 전체가 0.6초라 touch 만 떼어낼 이유가 없고, 권한 (c)안은 LRU 를 age 기반 축출로 바꾸는 의미 변경이다. 근거 없이 할 변경이 아니다. - 제안 6(캐시 우회 스위치). 이슈 스스로 "1~4 로 해결되면 불필요 — 플래그부터 만들지 말 것" 이라고 적었다. - 원인 6(4 KB BLOB overflow). page_size 변경은 기존 DB 에서 VACUUM 을 요구하는데 kebab 은 VACUUM 을 실행하지 않는다. 0.4% 에 낼 비용이 아니다. 곁다리로 #228 에서 내가 넣은 flaky test 를 고쳤다. `ingest_log_records_the_deleted_file_sweep` 이 두 run 의 로그 중 뒤엣것을 파일명 정렬로 골랐는데, run id 가 `<초 단위 타임스탬프>-<난수 hex>` 라 같은 초에 끝난 두 run 은 난수 쪽으로 정렬된다. 이번 전체 테스트에서 우연히 터져 잡았다. 첫 run 의 로그 집합을 기록해 두고 차집합으로 고르도록 바꿨다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017c9JwQq8ZkGvYjpKXMiDhF
This commit is contained in:
@@ -23,18 +23,96 @@ impl SqliteStore {
|
||||
/// per batch (cheaper than a write per `get`).
|
||||
pub fn derivation_cache_get(&self, cache_key: &str) -> Result<Option<Vec<u8>>> {
|
||||
let conn = self.lock_conn();
|
||||
// `prepare_cached`, not `query_row`: the latter prepares (and so
|
||||
// re-parses the SQL) on every call. Issue #231.
|
||||
let payload: Option<Vec<u8>> = conn
|
||||
.query_row(
|
||||
"SELECT payload FROM derivation_cache WHERE cache_key = ?",
|
||||
params![cache_key],
|
||||
|row| row.get::<_, Vec<u8>>(0),
|
||||
)
|
||||
.prepare_cached("SELECT payload FROM derivation_cache WHERE cache_key = ?")
|
||||
.map_err(StoreError::from)?
|
||||
.query_row(params![cache_key], |row| row.get::<_, Vec<u8>>(0))
|
||||
.optional()
|
||||
.map_err(StoreError::from)
|
||||
.context("derivation_cache_get")?;
|
||||
Ok(payload)
|
||||
}
|
||||
|
||||
/// Look up many keys at once, returning only the ones present.
|
||||
///
|
||||
/// One statement per `SQLITE_MAX_VARIABLE_NUMBER`-sized batch instead
|
||||
/// of one round trip per key (issue #231). The per-key form made a
|
||||
/// cache *hit* cost as many `lock_conn()` acquisitions as a miss —
|
||||
/// and a miss at least does its expensive half (the embedding) on the
|
||||
/// GPU, batched, outside the lock. Duplicate keys in `keys` are fine;
|
||||
/// the result is keyed by cache_key so they collapse.
|
||||
pub fn derivation_cache_get_many(
|
||||
&self,
|
||||
keys: &[String],
|
||||
) -> Result<std::collections::HashMap<String, Vec<u8>>> {
|
||||
let mut found = std::collections::HashMap::with_capacity(keys.len());
|
||||
if keys.is_empty() {
|
||||
return Ok(found);
|
||||
}
|
||||
let conn = self.lock_conn();
|
||||
// Conservative against the 999-parameter default: the bundled
|
||||
// build allows far more, but the cost of a few extra statements
|
||||
// is nothing next to the per-key round trips this replaces.
|
||||
for batch in keys.chunks(900) {
|
||||
let placeholders = std::iter::repeat_n("?", batch.len())
|
||||
.collect::<Vec<_>>()
|
||||
.join(",");
|
||||
let mut stmt = conn
|
||||
.prepare_cached(&format!(
|
||||
"SELECT cache_key, payload FROM derivation_cache WHERE cache_key IN ({placeholders})"
|
||||
))
|
||||
.map_err(StoreError::from)
|
||||
.context("derivation_cache_get_many: prepare")?;
|
||||
let rows = stmt
|
||||
.query_map(rusqlite::params_from_iter(batch.iter()), |row| {
|
||||
Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
|
||||
})
|
||||
.map_err(StoreError::from)
|
||||
.context("derivation_cache_get_many: query")?;
|
||||
for row in rows {
|
||||
let (k, v) = row.map_err(StoreError::from)?;
|
||||
found.insert(k, v);
|
||||
}
|
||||
}
|
||||
Ok(found)
|
||||
}
|
||||
|
||||
/// Insert many payloads in one transaction.
|
||||
///
|
||||
/// The single-key [`Self::derivation_cache_put`] runs outside any
|
||||
/// explicit transaction, so each call is its own implicit commit — a
|
||||
/// WAL frame write and wal-index update per embedded chunk (issue
|
||||
/// #231). A batch of misses is one logical unit of work and commits
|
||||
/// as one.
|
||||
pub fn derivation_cache_put_many(&self, entries: &[(String, String, Vec<u8>)]) -> Result<()> {
|
||||
if entries.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let now = OffsetDateTime::now_utc()
|
||||
.format(&Rfc3339)
|
||||
.context("format derivation_cache.created_at")?;
|
||||
let mut conn = self.lock_conn();
|
||||
let tx = conn.transaction().map_err(StoreError::from)?;
|
||||
{
|
||||
let mut stmt = tx
|
||||
.prepare_cached(
|
||||
"INSERT OR REPLACE INTO derivation_cache
|
||||
(cache_key, kind, payload, created_at, last_used_at)
|
||||
VALUES (?, ?, ?, ?, ?)",
|
||||
)
|
||||
.map_err(StoreError::from)?;
|
||||
for (key, kind, payload) in entries {
|
||||
stmt.execute(params![key, kind, payload, now, now])
|
||||
.map_err(StoreError::from)
|
||||
.context("derivation_cache_put_many")?;
|
||||
}
|
||||
}
|
||||
tx.commit().map_err(StoreError::from)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Insert (or overwrite) a cached derivation payload.
|
||||
///
|
||||
/// `INSERT OR REPLACE` so a re-computation of the same key (e.g. after a
|
||||
@@ -47,12 +125,13 @@ impl SqliteStore {
|
||||
.format(&Rfc3339)
|
||||
.context("format derivation_cache.created_at")?;
|
||||
let conn = self.lock_conn();
|
||||
conn.execute(
|
||||
conn.prepare_cached(
|
||||
"INSERT OR REPLACE INTO derivation_cache
|
||||
(cache_key, kind, payload, created_at, last_used_at)
|
||||
VALUES (?, ?, ?, ?, ?)",
|
||||
params![cache_key, kind, payload, now, now],
|
||||
)
|
||||
.map_err(StoreError::from)?
|
||||
.execute(params![cache_key, kind, payload, now, now])
|
||||
.map_err(StoreError::from)
|
||||
.context("derivation_cache_put")?;
|
||||
Ok(())
|
||||
@@ -73,7 +152,7 @@ impl SqliteStore {
|
||||
let tx = conn.transaction().map_err(StoreError::from)?;
|
||||
{
|
||||
let mut stmt = tx
|
||||
.prepare("UPDATE derivation_cache SET last_used_at = ? WHERE cache_key = ?")
|
||||
.prepare_cached("UPDATE derivation_cache SET last_used_at = ? WHERE cache_key = ?")
|
||||
.map_err(StoreError::from)?;
|
||||
for key in keys {
|
||||
stmt.execute(params![now, key])
|
||||
@@ -150,6 +229,91 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn get_many_returns_only_present_keys() {
|
||||
let (_d, store) = open_store();
|
||||
store.derivation_cache_put("a", "embedding", b"A").unwrap();
|
||||
store.derivation_cache_put("c", "embedding", b"C").unwrap();
|
||||
|
||||
let found = store
|
||||
.derivation_cache_get_many(&[
|
||||
"a".to_string(),
|
||||
"b".to_string(),
|
||||
"c".to_string(),
|
||||
// A repeat: the embed path derives keys from chunk text,
|
||||
// and a document with two identical chunks produces the
|
||||
// same key twice. The map must collapse it, not trip.
|
||||
"a".to_string(),
|
||||
])
|
||||
.unwrap();
|
||||
assert_eq!(found.len(), 2, "absent keys are simply not in the map");
|
||||
assert_eq!(found.get("a").map(Vec::as_slice), Some(b"A".as_slice()));
|
||||
assert_eq!(found.get("c").map(Vec::as_slice), Some(b"C".as_slice()));
|
||||
assert!(!found.contains_key("b"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn get_many_is_empty_for_no_keys() {
|
||||
let (_d, store) = open_store();
|
||||
assert!(store.derivation_cache_get_many(&[]).unwrap().is_empty());
|
||||
}
|
||||
|
||||
/// The batch splits at 900 parameters. A run that embeds more chunks
|
||||
/// than that in one call must still see every one of them — an
|
||||
/// off-by-one in the chunking would silently turn hits into misses,
|
||||
/// which costs nothing but correctness-invisible re-embedding.
|
||||
#[test]
|
||||
fn get_many_spans_more_keys_than_one_statement_holds() {
|
||||
let (_d, store) = open_store();
|
||||
let keys: Vec<String> = (0..2_000).map(|i| format!("k{i:05}")).collect();
|
||||
for k in &keys {
|
||||
store
|
||||
.derivation_cache_put(k, "embedding", k.as_bytes())
|
||||
.unwrap();
|
||||
}
|
||||
let found = store.derivation_cache_get_many(&keys).unwrap();
|
||||
assert_eq!(found.len(), keys.len());
|
||||
for k in &keys {
|
||||
assert_eq!(
|
||||
found.get(k).map(Vec::as_slice),
|
||||
Some(k.as_bytes()),
|
||||
"key {k} must survive the batch split"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn put_many_writes_all_and_replaces() {
|
||||
let (_d, store) = open_store();
|
||||
store
|
||||
.derivation_cache_put("dup", "embedding", b"old")
|
||||
.unwrap();
|
||||
store
|
||||
.derivation_cache_put_many(&[
|
||||
("x".to_string(), "embedding".to_string(), b"X".to_vec()),
|
||||
("y".to_string(), "embedding".to_string(), b"Y".to_vec()),
|
||||
("dup".to_string(), "embedding".to_string(), b"new".to_vec()),
|
||||
])
|
||||
.unwrap();
|
||||
let found = store
|
||||
.derivation_cache_get_many(&["x".to_string(), "y".to_string(), "dup".to_string()])
|
||||
.unwrap();
|
||||
assert_eq!(found.get("x").map(Vec::as_slice), Some(b"X".as_slice()));
|
||||
assert_eq!(found.get("y").map(Vec::as_slice), Some(b"Y".as_slice()));
|
||||
assert_eq!(
|
||||
found.get("dup").map(Vec::as_slice),
|
||||
Some(b"new".as_slice()),
|
||||
"the batch form keeps INSERT OR REPLACE semantics"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn put_many_with_no_entries_is_noop() {
|
||||
let (_d, store) = open_store();
|
||||
store.derivation_cache_put_many(&[]).unwrap();
|
||||
assert!(store.derivation_cache_get_many(&[]).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn touch_missing_keys_is_noop() {
|
||||
let (_d, store) = open_store();
|
||||
|
||||
Reference in New Issue
Block a user