Skip to content

fix(compaction): batch filename collision causes up to 84% data loss … #359

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xe-nvdk wants to merge 1 commit into from
Closed

fix(compaction): batch filename collision causes up to 84% data loss … #359

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
26 changes: 17 additions & 9 deletions internal/compaction/daily.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,28 +277,36 @@ func extractNewestFileTime(files []string) time.Time {
// Remove .parquet extension
filename = strings.TrimSuffix(filename, ".parquet")

// Check if it's a daily compacted file: measurement_YYYYMMDD_HHMMSS_daily
// Check if it's a tier-compacted file: measurement_YYYYMMDD_HHMMSS_{nanos}_{daily|compacted}

@gemini-code-assist gemini-code-assist Bot Mar 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The comment on this line describes the tier-compacted file format as measurement_YYYYMMDD_HHMMSS_{nanos}_{daily|compacted}. However, based on the new filename generation logic in job.go, the _nanos part is inserted before the _daily or _compacted suffix. A more accurate description of the filename structure (before stripping the .parquet extension) would be measurement_YYYYMMDD_HHMMSS_nanos_{daily|compacted}. Updating this comment would improve clarity and consistency with the actual file naming convention.

Suggested change
// Check if it's a tier-compacted file: measurement_YYYYMMDD_HHMMSS_{nanos}_{daily|compacted}
// Check if it's a tier-compacted file: measurement_YYYYMMDD_HHMMSS_nanos_{daily|compacted}

// Strip the tier suffix, then handle like a raw file
tierSuffix := ""
if strings.HasSuffix(filename, "_daily") {
// Remove _daily suffix
filename = strings.TrimSuffix(filename, "_daily")
tierSuffix = "_daily"
} else if strings.HasSuffix(filename, "_compacted") {
tierSuffix = "_compacted"
}

if tierSuffix != "" {
filename = strings.TrimSuffix(filename, tierSuffix)
fileParts := strings.Split(filename, "_")
if len(fileParts) < 3 {
continue
}
// Last two parts are date and time: YYYYMMDD_HHMMSS
// Try parsing last two parts as YYYYMMDD_HHMMSS (old format without nanos)
dateTimePart := fileParts[len(fileParts)-2] + "_" + fileParts[len(fileParts)-1]
// Parse timestamp: YYYYMMDD_HHMMSS
fileTime, err := time.Parse("20060102_150405", dateTimePart)
if err != nil {
continue
if err != nil && len(fileParts) >= 4 {
// New format with nanos: ..._YYYYMMDD_HHMMSS_nanos — skip nanos, take 3rd and 2nd from end
dateTimePart = fileParts[len(fileParts)-3] + "_" + fileParts[len(fileParts)-2]
fileTime, err = time.Parse("20060102_150405", dateTimePart)
}
if fileTime.After(newest) {
if err == nil && fileTime.After(newest) {
newest = fileTime
}
continue
}

// Handle hourly file: measurement_YYYYMMDD_HHMMSS_nanos
// Handle raw hourly file: measurement_YYYYMMDD_HHMMSS_nanos
fileParts := strings.Split(filename, "_")
if len(fileParts) < 3 {
continue
Expand Down
11 changes: 8 additions & 3 deletions internal/compaction/job.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -472,13 +472,18 @@ func (j *Job) downloadSingleFile(ctx context.Context, tempDir string, index int,
// It validates each file and only compacts valid ones, storing the list of
// successfully compacted files' storage keys in j.compactedFiles.
func (j *Job) compactFiles(ctx context.Context, files []downloadedFile, tempDir string) (string, error) {
// Generate output filename with tier-specific suffix (use UTC for consistency)
timestamp := time.Now().UTC().Format("20060102_150405")
// Generate output filename with tier-specific suffix.
// Include UnixNano to guarantee uniqueness when multiple batches run
// sequentially for the same partition (SplitCandidateIntoBatches).
// Second-precision timestamps caused batch N to overwrite batch N-1's output,
// destroying up to 84% of data.
now := time.Now().UTC()
timestamp := now.Format("20060102_150405")
suffix := "compacted"
if j.Tier != "hourly" {
suffix = j.Tier
}
outputFile := filepath.Join(tempDir, fmt.Sprintf("%s_%s_%s.parquet", j.Measurement, timestamp, suffix))
outputFile := filepath.Join(tempDir, fmt.Sprintf("%s_%s_%d_%s.parquet", j.Measurement, timestamp, now.UnixNano(), suffix))

// Use the shared DuckDB connection instead of creating a new one
// This prevents memory retention from DuckDB's jemalloc not releasing memory on Close()
Expand Down
Loading