Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 99 additions & 0 deletions src/storage/ducklake/ducklake.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
}
37 changes: 23 additions & 14 deletions src/storage/ducklake/otlp_storage.go
Original file line number Diff line number Diff line change
Expand Up @@ -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))
}
Expand Down
29 changes: 20 additions & 9 deletions src/storage/ducklake/tables.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion src/version.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 = ""
Expand Down
Loading