Skip to content

Commit 52e6fac

Browse files
committed
fix(ducklake): disable data inlining by default + flush it in compaction + batch LP writes
Ports the memory/catalog-bloat fixes proven in the sibling ingest service (hepic-lake-ingest) to Homer's DuckLake stack. Same storage class (DuckDB/DuckLake + sqlite catalog + Parquet); the HEP write path here is already well-batched (Appender + double-buffer + bulk flush), but three gaps remained: 1. Data inlining default. DuckLakeConfig.DataInliningRowLimit defaulted to -1 ("leave DuckLake's own default", which inlines ~10-row writes into the catalog DB). Under streaming Line Protocol / OTLP / low- volume HEP subtypes this turns the catalog into the dominant memory + disk consumer (an 800 MB sqlite catalog backing only a few dozen Parquet files, multi-GB RSS when DuckLake mirrors it in memory). Default is now 0 (inlining off, always write Parquet). -1 and >0 are still honoured for operators who want them. 2. No inline flush in maintenance. The CompactionService ran merge / expire / cleanup / delete-orphaned but never ducklake_flush_inlined_data, so anything already inlined (or inlined by an operator who re-enables it) stayed in the catalog forever. Added a flush step at the start of the maintenance cycle (before merge, so merge/expire act on the freshly written Parquet). No-op when inlining is disabled. 3. Line Protocol micro-commits. The generic LP path issued one prepared stmt.ExecContext per row = one DuckLake transaction (snapshot + tiny write) per row. Replaced with chunked multi-row INSERT ... VALUES (500 rows/statement), collapsing the per-row transaction/snapshot churn by up to 500x. hep_proto_* LP and OTLP already batch per request and are unchanged. version.go is intentionally untouched — Homer's version is tag-driven (version-sync.yml updates it from the release tag).
1 parent 1abaef0 commit 52e6fac

3 files changed

Lines changed: 80 additions & 25 deletions

File tree

src/config/config.go

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -598,9 +598,19 @@ type DuckLakeConfig struct {
598598
FlushQueue *bool `json:"flush_queue" mapstructure:"flush_queue"` // nil = auto (true for SQLite), explicit true/false overrides
599599
// DataInliningRowLimit controls the DuckLake data inlining threshold (DuckLake v1.0).
600600
// Writes of ≤N rows are stored directly in the catalog database instead of creating
601-
// small Parquet files. Default -1 means use DuckLake's built-in default (10 rows).
602-
// Set to 0 to disable inlining entirely (always write Parquet).
603-
DataInliningRowLimit int `json:"data_inlining_row_limit" mapstructure:"data_inlining_row_limit" default:"-1"`
601+
// small Parquet files.
602+
//
603+
// Default 0 = inlining DISABLED (every write goes to a Parquet file). This is
604+
// deliberate: DuckLake's own default inlines small writes into the catalog DB, and
605+
// under streaming ingest with many small writes (Line Protocol, OTLP, low-volume
606+
// HEP subtypes) that turns the catalog (sqlite) into the dominant memory + disk
607+
// consumer — an 800 MB catalog backing only a few dozen Parquet files, and multi-GB
608+
// RSS when DuckLake mirrors the catalog in memory. The CompactionService now also
609+
// flushes inlined data on each maintenance cycle as a safety net.
610+
// * 0 -> disable inlining (recommended; always write Parquet)
611+
// * >0 -> inline writes smaller than N rows (only for low write cardinality)
612+
// * -1 -> leave DuckLake's own default (inlines ~10 rows) — avoid for streaming
613+
DataInliningRowLimit int `json:"data_inlining_row_limit" mapstructure:"data_inlining_row_limit" default:"0"`
604614
Tuning DuckDBTuning `json:"tuning" mapstructure:"tuning"`
605615
S3 S3Config `json:"s3" mapstructure:"s3"`
606616
Compaction CompactionConfig `json:"compaction" mapstructure:"compaction"`

src/lineprotoreceiver/ingest.go

Lines changed: 51 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -269,34 +269,63 @@ func (i *Ingester) writeRows(ctx context.Context, fqTable string, rows []map[str
269269
continue
270270
}
271271
cols := sortedMapKeys(bucket[0])
272-
placeholders := make([]string, len(cols))
273-
for j := range placeholders {
274-
placeholders[j] = "?"
272+
n, err := i.insertBucket(ctx, fqTable, cols, bucket)
273+
inserted += n
274+
if err != nil {
275+
return inserted, err
276+
}
277+
}
278+
return inserted, nil
279+
}
280+
281+
// lpInsertChunkRows bounds how many rows go into a single multi-row INSERT
282+
// so a large batch doesn't blow past DuckDB's bound-parameter / statement
283+
// size limits. Chosen well below any practical limit while still collapsing
284+
// thousands of per-row commits into a handful of statements.
285+
const lpInsertChunkRows = 500
286+
287+
// insertBucket writes a homogeneous set of rows (same column subset) using
288+
// chunked multi-row `INSERT ... VALUES (...),(...),...` statements instead of
289+
// one Exec per row.
290+
//
291+
// Each per-row Exec used to be its own DuckLake transaction — a catalog
292+
// snapshot plus a tiny Parquet (or inlined) write per row. Under sustained
293+
// Line Protocol traffic that micro-commit storm bloats the catalog and stalls
294+
// ingest (the same pattern fixed for OTLP/Python in the sibling ingest
295+
// service). Bulk inserts cut the transaction/snapshot count by up to
296+
// lpInsertChunkRows×.
297+
func (i *Ingester) insertBucket(ctx context.Context, fqTable string, cols []string, bucket []map[string]interface{}) (int, error) {
298+
colList := strings.Join(cols, ", ")
299+
// "(?, ?, ... ?)" for one row.
300+
rowPlaceholder := "(" + strings.TrimSuffix(strings.Repeat("?, ", len(cols)), ", ") + ")"
301+
302+
inserted := 0
303+
for start := 0; start < len(bucket); start += lpInsertChunkRows {
304+
end := start + lpInsertChunkRows
305+
if end > len(bucket) {
306+
end = len(bucket)
307+
}
308+
chunk := bucket[start:end]
309+
310+
placeholders := make([]string, len(chunk))
311+
args := make([]interface{}, 0, len(chunk)*len(cols))
312+
for k, r := range chunk {
313+
placeholders[k] = rowPlaceholder
314+
for _, c := range cols {
315+
args = append(args, r[c])
316+
}
275317
}
276318
stmtSQL := fmt.Sprintf(
277-
"INSERT INTO %s (%s) VALUES (%s)",
319+
"INSERT INTO %s (%s) VALUES %s",
278320
fqTable,
279-
strings.Join(cols, ", "),
321+
colList,
280322
strings.Join(placeholders, ", "),
281323
)
282-
stmt, err := i.db.PrepareContext(ctx, stmtSQL)
283-
if err != nil {
284-
metrics.RecordLineProtoWriteError("prepare")
285-
return inserted, fmt.Errorf("prepare: %w", err)
286-
}
287-
for _, r := range bucket {
288-
vals := make([]interface{}, len(cols))
289-
for j, c := range cols {
290-
vals[j] = r[c]
291-
}
292-
if _, err := stmt.ExecContext(ctx, vals...); err != nil {
293-
_ = stmt.Close()
294-
metrics.RecordLineProtoWriteError("insert")
295-
return inserted, fmt.Errorf("exec: %w", err)
296-
}
297-
inserted++
324+
if _, err := i.db.ExecContext(ctx, stmtSQL, args...); err != nil {
325+
metrics.RecordLineProtoWriteError("insert")
326+
return inserted, fmt.Errorf("bulk insert: %w", err)
298327
}
299-
_ = stmt.Close()
328+
inserted += len(chunk)
300329
}
301330
return inserted, nil
302331
}

src/writer/compaction.go

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -424,6 +424,22 @@ func (c *CompactionService) runMerge(tables []string) error {
424424
logger.Info("CompactionService: Active snapshots before merge", "count", snapshotCount)
425425
}
426426

427+
// 0. Flush inlined data to Parquet FIRST. DuckLake inlines small writes
428+
// (DATA_INLINING_ROW_LIMIT) directly into the catalog DB; with inlining
429+
// left enabled and no periodic flush, those rows accumulate inside the
430+
// catalog forever — an 800 MB catalog backing only a few dozen Parquet
431+
// files is the classic symptom, and DuckLake mirrors the catalog in
432+
// memory (multi-GB RSS). Flushing first also lets the subsequent merge /
433+
// expire act on freshly written Parquet instead of catalog-resident rows.
434+
// Harmless no-op when inlining is disabled (the recommended default).
435+
c.withCatalogLock(func() {
436+
logger.Info("CompactionService: Flush inlined data", "lake", c.lakeName)
437+
flushSQL := fmt.Sprintf("CALL ducklake_flush_inlined_data('%s')", c.lakeName)
438+
if _, err := c.execWithRetry(flushSQL); err != nil {
439+
logger.Warn("CompactionService: flush_inlined_data failed", "error", err)
440+
}
441+
})
442+
427443
// 1. Merge adjacent small files FIRST — lock per table
428444
for _, table := range tables {
429445
tableName := tableNameFromFQN(table)

0 commit comments

Comments
 (0)