Merge pull request #765 from sipcapture/fix/ducklake-inlining-and-lp-batching

fix(ducklake): disable data inlining by default, flush it in compaction, batch Line Protocol writes

Author: adubovikov

src/config/config.go

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

src/lineprotoreceiver/ingest.go

@@ -269,34 +269,63 @@ func (i *Ingester) writeRows(ctx context.Context, fqTable string, rows []map[str
 			continue
 		}
 		cols := sortedMapKeys(bucket[0])
-		placeholders := make([]string, len(cols))
-		for j := range placeholders {
-			placeholders[j] = "?"
+		n, err := i.insertBucket(ctx, fqTable, cols, bucket)
+		inserted += n
+		if err != nil {
+			return inserted, err
+		}
+	}
+	return inserted, nil
+}
+
+// lpInsertChunkRows bounds how many rows go into a single multi-row INSERT
+// so a large batch doesn't blow past DuckDB's bound-parameter / statement
+// size limits. Chosen well below any practical limit while still collapsing
+// thousands of per-row commits into a handful of statements.
+const lpInsertChunkRows = 500
+
+// insertBucket writes a homogeneous set of rows (same column subset) using
+// chunked multi-row `INSERT ... VALUES (...),(...),...` statements instead of
+// one Exec per row.
+//
+// Each per-row Exec used to be its own DuckLake transaction — a catalog
+// snapshot plus a tiny Parquet (or inlined) write per row. Under sustained
+// Line Protocol traffic that micro-commit storm bloats the catalog and stalls
+// ingest (the same pattern fixed for OTLP/Python in the sibling ingest
+// service). Bulk inserts cut the transaction/snapshot count by up to
+// lpInsertChunkRows×.
+func (i *Ingester) insertBucket(ctx context.Context, fqTable string, cols []string, bucket []map[string]interface{}) (int, error) {
+	colList := strings.Join(cols, ", ")
+	// "(?, ?, ... ?)" for one row.
+	rowPlaceholder := "(" + strings.TrimSuffix(strings.Repeat("?, ", len(cols)), ", ") + ")"
+
+	inserted := 0
+	for start := 0; start < len(bucket); start += lpInsertChunkRows {
+		end := start + lpInsertChunkRows
+		if end > len(bucket) {
+			end = len(bucket)
+		}
+		chunk := bucket[start:end]
+
+		placeholders := make([]string, len(chunk))
+		args := make([]interface{}, 0, len(chunk)*len(cols))
+		for k, r := range chunk {
+			placeholders[k] = rowPlaceholder
+			for _, c := range cols {
+				args = append(args, r[c])
+			}
 		}
 		stmtSQL := fmt.Sprintf(
-			"INSERT INTO %s (%s) VALUES (%s)",
+			"INSERT INTO %s (%s) VALUES %s",
 			fqTable,
-			strings.Join(cols, ", "),
+			colList,
 			strings.Join(placeholders, ", "),
 		)
-		stmt, err := i.db.PrepareContext(ctx, stmtSQL)
-		if err != nil {
-			metrics.RecordLineProtoWriteError("prepare")
-			return inserted, fmt.Errorf("prepare: %w", err)
-		}
-		for _, r := range bucket {
-			vals := make([]interface{}, len(cols))
-			for j, c := range cols {
-				vals[j] = r[c]
-			}
-			if _, err := stmt.ExecContext(ctx, vals...); err != nil {
-				_ = stmt.Close()
-				metrics.RecordLineProtoWriteError("insert")
-				return inserted, fmt.Errorf("exec: %w", err)
-			}
-			inserted++
+		if _, err := i.db.ExecContext(ctx, stmtSQL, args...); err != nil {
+			metrics.RecordLineProtoWriteError("insert")
+			return inserted, fmt.Errorf("bulk insert: %w", err)
 		}
-		_ = stmt.Close()
+		inserted += len(chunk)
 	}
 	return inserted, nil
 }

src/version.go

@@ -24,7 +24,7 @@ import (
 // Version information for homer-core
 var (
 	// VERSION_APPLICATION is the application version
-	VERSION_APPLICATION = "11.0.231"
+	VERSION_APPLICATION = "11.0.232"
 
 	// BuildDate is the build date
 	BuildDate = ""

src/writer/compaction.go

@@ -424,6 +424,22 @@ func (c *CompactionService) runMerge(tables []string) error {
 		logger.Info("CompactionService: Active snapshots before merge", "count", snapshotCount)
 	}
 
+	// 0. Flush inlined data to Parquet FIRST. DuckLake inlines small writes
+	// (DATA_INLINING_ROW_LIMIT) directly into the catalog DB; with inlining
+	// left enabled and no periodic flush, those rows accumulate inside the
+	// catalog forever — an 800 MB catalog backing only a few dozen Parquet
+	// files is the classic symptom, and DuckLake mirrors the catalog in
+	// memory (multi-GB RSS). Flushing first also lets the subsequent merge /
+	// expire act on freshly written Parquet instead of catalog-resident rows.
+	// Harmless no-op when inlining is disabled (the recommended default).
+	c.withCatalogLock(func() {
+		logger.Info("CompactionService: Flush inlined data", "lake", c.lakeName)
+		flushSQL := fmt.Sprintf("CALL ducklake_flush_inlined_data('%s')", c.lakeName)
+		if _, err := c.execWithRetry(flushSQL); err != nil {
+			logger.Warn("CompactionService: flush_inlined_data failed", "error", err)
+		}
+	})
+
 	// 1. Merge adjacent small files FIRST — lock per table
 	for _, table := range tables {
 		tableName := tableNameFromFQN(table)