presto-raptor-legacy/src/main/java/io/prestosql/plugin/raptor/legacy/metadata/ShardCleaner.java/ShardCleaner/cleanLocalShards
class ShardCleaner:    @VisibleForTesting
    synchronized void cleanLocalShards()
    {
        // find all files on the local node
        Set<UUID> local = getLocalShards();

        // get shards assigned to the local node
        Set<UUID> assigned = dao.getNodeShards(currentNode, null).stream()
                .map(ShardMetadata::getShardUuid)
                .collect(toSet());

        // un-mark previously marked files that are now assigned
        for (UUID uuid : assigned) {
            shardsToClean.remove(uuid);
        }

        // mark all files that are not assigned
        for (UUID uuid : local) {
            if (!assigned.contains(uuid)) {
                shardsToClean.putIfAbsent(uuid, ticker.read());
            }
        }

        // delete files marked earlier than the clean interval
        long threshold = ticker.read() - localCleanTime.roundTo(NANOSECONDS);
        Set<UUID> deletions = shardsToClean.entrySet().stream()
                .filter(entry -> entry.getValue() < threshold)
                .map(Map.Entry::getKey)
                .collect(toSet());
        if (deletions.isEmpty()) {
            return;
        }

        for (UUID uuid : deletions) {
            deleteFile(storageService.getStorageFile(uuid));
            shardsToClean.remove(uuid);
        }

        localShardsCleaned.update(deletions.size());
        log.info("Cleaned %s local shards", deletions.size());
    }