Skip to content
Merged
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
35 changes: 32 additions & 3 deletions internal/pruning/partition_pruner.go
Original file line number Diff line number Diff line change
Expand Up @@ -808,13 +808,42 @@ func (p *PartitionPruner) filterExistingRemotePaths(paths []string) []string {
}
}

// Filter paths to only include existing directories
// Filter paths to only include existing directories/files
existingPaths := []string{}
for _, path := range paths {
dir := pathToDir[path]
if existingDirs[dir] {
existingPaths = append(existingPaths, path)
if !existingDirs[dir] {
continue
}

// For day-level paths (5 segments: db/measurement/year/month/day),
// verify .parquet files exist directly at that level (not in subdirs).
// This prevents "No files found" errors when daily compaction hasn't run yet.
prefix := p.extractStoragePrefix(dir + "/")
if parts := strings.Split(strings.Trim(prefix, "/"), "/"); len(parts) == 5 {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
files, err := p.storage.List(ctx, prefix)
cancel()
if err != nil {
p.logger.Debug().Err(err).Str("prefix", prefix).Msg("Failed to list files at day-level path")
continue
}
hasFiles := false
for _, f := range files {
// Check if file is directly in this dir (no / in remaining path after prefix)
remaining := strings.TrimPrefix(f, prefix)
if remaining != "" && !strings.Contains(remaining, "/") && strings.HasSuffix(remaining, ".parquet") {
hasFiles = true
break
}
}
if !hasFiles {
p.logger.Debug().Str("path", path).Msg("Day-level path has no direct parquet files, skipping")
continue
}
}

existingPaths = append(existingPaths, path)
}

p.logger.Info().
Expand Down