From 0e3fb8c747f527a379f825ef6729d9a9a609136a Mon Sep 17 00:00:00 2001 From: Alexandr Dubovikov Date: Mon, 1 Jun 2026 11:07:29 +0200 Subject: [PATCH] fix(ducklake): GC leaked inline tables to stop multi-GB RSS growth DuckLake creates a ducklake_inlined_data__ table per schema version, schema_version bumps on every DDL (incl. our per-table SET PARTITIONED BY / SET SORTED BY), and no maintenance path ever DROPs them (flush only DELETEs rows, expire only drops registry rows, cleanup ignores them). They accumulate forever and the extension rebuilds an in-memory stats map (DuckLakeCatalog::ConstructStatsMap) over all of them on every catalog refresh, growing RSS to multiple GB. Upstream bug duckdb/ducklake#1065. - Startup GC (GCOrphanInlineTables): before ATTACH (exclusive sqlite access, same window as EnableSQLiteWALMode) drop every EMPTY ducklake_inlined_data_* table and its ducklake_inlined_data_tables registry row. Empty => rows already flushed to Parquet, so lossless. Best-effort, non-fatal. - Stop the churn: SET PARTITIONED BY / SET SORTED BY now run only when the HEP/OTLP table is first created (gated on information_schema.tables), not on every startup (each re-issue bumped schema_version and spawned another leaked inline table per restart). Bump 11.0.234 -> 11.0.235. --- src/storage/ducklake/ducklake.go | 99 ++++++++++++++++++++++++++++ src/storage/ducklake/otlp_storage.go | 37 +++++++---- src/storage/ducklake/tables.go | 29 +++++--- src/version.go | 2 +- 4 files changed, 143 insertions(+), 24 deletions(-) diff --git a/src/storage/ducklake/ducklake.go b/src/storage/ducklake/ducklake.go index 4ee8c9da..d83a6dd0 100644 --- a/src/storage/ducklake/ducklake.go +++ b/src/storage/ducklake/ducklake.go @@ -343,6 +343,14 @@ func (mtw *MultiTableWriter) connect() error { if err := EnableSQLiteWALMode(mtw.config.CatalogPath); err != nil { logger.Warn(fmt.Sprintf("Failed to enable WAL mode for SQLite catalog (may cause lock errors): %v", err)) } + // Drain the leaked inline-table backlog before DuckLake opens the catalog + // (upstream duckdb/ducklake#1065). Must run before ATTACH so we have + // exclusive access to the sqlite file. Non-fatal. + if n, err := GCOrphanInlineTables(mtw.config.CatalogPath); err != nil { + logger.Warn("DuckLake inline GC failed (non-fatal)", "err", err) + } else if n > 0 { + logger.Info("DuckLake inline GC: dropped empty ducklake_inlined_data_* tables (upstream #1065)", "dropped", n) + } if err := ApplyDuckDBS3ClientSettings(db, mtw.config.S3Region, @@ -736,3 +744,94 @@ func EnableSQLiteWALMode(catalogPath string) error { return fmt.Errorf("failed to enable WAL mode, got: %s", outputStr) } + +// duckLakeTableExists reports whether a table is already present in the +// attached DuckLake catalog. Used to avoid re-issuing schema-bumping DDL +// (SET PARTITIONED BY / SET SORTED BY) on every startup. Best-effort: any +// query error is treated as "does not exist" so the caller falls back to the +// safe (configure-the-table) path. +func duckLakeTableExists(db *sql.DB, lakeName, tableName string) bool { + row := db.QueryRow( + "SELECT 1 FROM information_schema.tables WHERE table_catalog = ? AND table_name = ? LIMIT 1", + lakeName, tableName) + var x int + return row.Scan(&x) == nil +} + +// runSQLiteCLI feeds a SQL script to the sqlite3 CLI on stdin and returns its +// stdout. Used for catalog maintenance that must run with exclusive access, +// before DuckDB ATTACHes the catalog. +func runSQLiteCLI(catalogPath, sql string) (string, error) { + cmd := exec.Command("sqlite3", "-batch", "-noheader", catalogPath) + cmd.Stdin = strings.NewReader(sql) + out, err := cmd.Output() + return string(out), err +} + +func sqliteLines(catalogPath, sql string) ([]string, error) { + out, err := runSQLiteCLI(catalogPath, sql) + if err != nil { + return nil, err + } + var lines []string + for _, l := range strings.Split(out, "\n") { + l = strings.TrimRight(l, "\r") + if strings.TrimSpace(l) != "" { + lines = append(lines, l) + } + } + return lines, nil +} + +// GCOrphanInlineTables drops empty `ducklake_inlined_data_*` physical tables +// left behind in the catalog. DuckLake never DROPs them: schema_version bumps +// on every DDL (incl. SET PARTITIONED BY / SET SORTED BY), flush only DELETEs +// the rows, and expire/cleanup remove registry rows but not the tables +// (upstream bug duckdb/ducklake#1065). They accumulate per +// (table_id, schema_version) and, although tiny on disk, the DuckLake +// extension rebuilds an in-memory stats map over ALL of them on every catalog +// refresh (DuckLakeCatalog::ConstructStatsMap) — which grows RSS to multiple GB. +// +// Must run BEFORE ATTACH, while the sqlite file is not yet opened by DuckDB, so +// we have exclusive access (same window EnableSQLiteWALMode uses). Only EMPTY +// tables are dropped: their rows were already flushed to Parquet, so this is +// lossless. Best-effort; returns the number of tables dropped. +func GCOrphanInlineTables(catalogPath string) (int, error) { + if _, err := os.Stat(catalogPath); os.IsNotExist(err) { + return 0, nil // fresh catalog, nothing to GC + } + + names, err := sqliteLines(catalogPath, + `SELECT name FROM sqlite_master WHERE type='table' `+ + `AND name LIKE 'ducklake_inlined_data\_%' ESCAPE '\' `+ + `AND name <> 'ducklake_inlined_data_tables';`) + if err != nil || len(names) == 0 { + return 0, err + } + + // Identify the empty ones in a single query (one row per empty table). + var q strings.Builder + for i, n := range names { + if i > 0 { + q.WriteString("\nUNION ALL ") + } + fmt.Fprintf(&q, `SELECT '%s' WHERE (SELECT count(*) FROM "%s")=0`, n, n) + } + empties, err := sqliteLines(catalogPath, q.String()) + if err != nil || len(empties) == 0 { + return 0, err + } + + var script strings.Builder + script.WriteString("BEGIN;\n") + for _, t := range empties { + fmt.Fprintf(&script, "DROP TABLE IF EXISTS \"%s\";\n", t) + fmt.Fprintf(&script, "DELETE FROM ducklake_inlined_data_tables WHERE table_name='%s';\n", t) + } + script.WriteString("COMMIT;\n") + + if _, err := runSQLiteCLI(catalogPath, script.String()); err != nil { + return 0, err + } + return len(empties), nil +} diff --git a/src/storage/ducklake/otlp_storage.go b/src/storage/ducklake/otlp_storage.go index 56e0c32b..afe90c51 100644 --- a/src/storage/ducklake/otlp_storage.go +++ b/src/storage/ducklake/otlp_storage.go @@ -142,24 +142,33 @@ func (s *OTLPStorage) EnsureOTLPSchema(ctx context.Context) error { return fmt.Errorf("otlp storage: nil database handle") } s.once.Do(func() { - stmts := []string{ - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.otlp_traces (%s);", s.lakeName, otlpTracesTableSQL), - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.otlp_metrics (%s);", s.lakeName, otlpMetricsTableSQL), - fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.otlp_logs (%s);", s.lakeName, otlpLogsTableSQL), + tables := []struct { + name string + createSQL string + }{ + {"otlp_traces", otlpTracesTableSQL}, + {"otlp_metrics", otlpMetricsTableSQL}, + {"otlp_logs", otlpLogsTableSQL}, } - for _, q := range stmts { - if _, err := s.db.ExecContext(ctx, q); err != nil { + for _, t := range tables { + fqn := fmt.Sprintf("%s.%s", s.lakeName, t.name) + // Check existence BEFORE create: SET PARTITIONED BY / SET SORTED BY + // bump schema_version on every run, and each bump leaks another + // ducklake_inlined_data_* table (upstream duckdb/ducklake#1065). + // Only configure freshly created tables, not on every restart. + alreadyExisted := duckLakeTableExists(s.db, s.lakeName, t.name) + + if _, err := s.db.ExecContext(ctx, + fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s);", fqn, t.createSQL)); err != nil { s.ddlErr = fmt.Errorf("otlp storage: %w", err) return } - } - // Best-effort: time-partition the OTLP tables the same way HEP - // tables are partitioned so range scans hit a single date - // folder. Failures are logged, not fatal — older DuckLake - // builds without ALTER...PARTITIONED BY simply keep a single - // partition. - for _, t := range []string{"otlp_traces", "otlp_metrics", "otlp_logs"} { - fqn := fmt.Sprintf("%s.%s", s.lakeName, t) + if alreadyExisted { + continue + } + // Best-effort: time-partition + sort the OTLP tables the same way + // HEP tables are. Failures are logged, not fatal — older DuckLake + // builds without ALTER...PARTITIONED BY keep a single partition. if _, err := s.db.ExecContext(ctx, fmt.Sprintf("ALTER TABLE %s SET PARTITIONED BY (date);", fqn)); err != nil { logger.Warn(fmt.Sprintf("otlp storage: failed to set partitioning for %s: %v", fqn, err)) } diff --git a/src/storage/ducklake/tables.go b/src/storage/ducklake/tables.go index 6743363a..14f7d271 100644 --- a/src/storage/ducklake/tables.go +++ b/src/storage/ducklake/tables.go @@ -474,22 +474,33 @@ func NewTableWriter(db *sql.DB, lakeName string, schema *TableSchema, batchSize tw.flushTruncateSQL[i] = "TRUNCATE TABLE " + tw.memTables[i] } + // Did the table already exist before this CREATE? SET PARTITIONED BY / + // SET SORTED BY are DDL that bump the DuckLake schema_version every time + // they run, and each new schema_version spawns another (leaked) + // ducklake_inlined_data_* table (upstream duckdb/ducklake#1065). Re-issuing + // them on every restart is pure churn, so only configure a freshly created + // table. + tableName := fmt.Sprintf("hep_proto_%s", schema.TableSuffix) + alreadyExisted := duckLakeTableExists(db, lakeName, tableName) + // Create DuckLake persistent table createSQL := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s);", tableFQN, schema.CreateSQL) if _, err := db.Exec(createSQL); err != nil { return nil, fmt.Errorf("failed to create table %s: %w", tableFQN, err) } - // Set partitioning by date for efficient time-range queries - partitionSQL := fmt.Sprintf("ALTER TABLE %s SET PARTITIONED BY (date);", tableFQN) - if _, err := db.Exec(partitionSQL); err != nil { - logger.Warn(fmt.Sprintf("Failed to set partitioning for %s: %v", tableFQN, err)) - } + if !alreadyExisted { + // Set partitioning by date for efficient time-range queries + partitionSQL := fmt.Sprintf("ALTER TABLE %s SET PARTITIONED BY (date);", tableFQN) + if _, err := db.Exec(partitionSQL); err != nil { + logger.Warn(fmt.Sprintf("Failed to set partitioning for %s: %v", tableFQN, err)) + } - // Sort rows by timestamp within each file (DuckLake v1.0). - sortSQL := fmt.Sprintf("ALTER TABLE %s SET SORTED BY (timestamp ASC);", tableFQN) - if _, err := db.Exec(sortSQL); err != nil { - logger.Warn(fmt.Sprintf("Failed to set sort order for %s: %v", tableFQN, err)) + // Sort rows by timestamp within each file (DuckLake v1.0). + sortSQL := fmt.Sprintf("ALTER TABLE %s SET SORTED BY (timestamp ASC);", tableFQN) + if _, err := db.Exec(sortSQL); err != nil { + logger.Warn(fmt.Sprintf("Failed to set sort order for %s: %v", tableFQN, err)) + } } // Create both in-memory buffer tables (plain DuckDB, not DuckLake) diff --git a/src/version.go b/src/version.go index 1998bc06..d6156cee 100644 --- a/src/version.go +++ b/src/version.go @@ -24,7 +24,7 @@ import ( // Version information for homer-core var ( // VERSION_APPLICATION is the application version - VERSION_APPLICATION = "11.0.234" + VERSION_APPLICATION = "11.0.235" // BuildDate is the build date BuildDate = ""