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