Merge pull request #767 from sipcapture/fix/ducklake-inline-table-leak

fix(ducklake): GC leaked inline tables to stop multi-GB RSS growth

Author: adubovikov

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
+}

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))
 			}

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)

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 = ""