<commit_before>package org.talend.dataprep.cache;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.commons.io.output.NullOutputStream;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.talend.dataprep.api.preparation.Step;
import org.talend.dataprep.exception.TDPException;
import org.talend.dataprep.exception.error.CommonErrorCodes;

/**
 * HDFS implementation of the content cache.
 *
 * Cache entries path look like this : datasets/{id}/preparations/{id}/steps/{stepid}/{sample} so that cache eviction
 * can happen at different level : dataset, preparation or step (sample is irrelevant because it's tied to the step).
 */
@Component
@EnableScheduling
@ConditionalOnProperty(name = "hdfs.location")
public class HDFSContentCache implements ContentCache {

    /** This class' logger. */
    private static final Logger LOGGER = LoggerFactory.getLogger(HDFSContentCache.class);

    /** Root folder name. */
    private static final String ROOT = "cache";

    /** Hdfs cache location. */
    @Autowired
    private FileSystem fileSystem;

    /**
     * Default empty constructor.
     */
    public HDFSContentCache() {
        LOGGER.info("Using content cache: {}", this.getClass().getName());
    }

    /**
     * Compute the path for the given key.
     *
     * @param key the cache key entry.
     * @return the HDFS path for the entry key.
     */
    private Path computeEntryPath(ContentCacheKey key) {

        String datasetId = key.getDatasetId();
        String preparationId = key.getPreparationId();
        String stepId = key.getStepId();
        String sample = key.getSample();

        if (datasetId == null) {
            throw new IllegalArgumentException("Dataset id cannot be null.");
        }

        if (preparationId == null) {
            throw new IllegalArgumentException("Preparation id cannot be null.");
        }

        if (key.getStepId() == null) {
            throw new IllegalArgumentException("Step id cannot be null.");
        }

        if ("origin".equalsIgnoreCase(stepId)) {
            stepId = Step.ROOT_STEP.id();
        }

        return new Path(ROOT + "/datasets/" + datasetId + "/preparations/" + preparationId + "/steps/" + stepId + '/' + sample);
    }

    /**
     * @see ContentCache#put(ContentCacheKey, TimeToLive)
     */
    @Override
    public OutputStream put(ContentCacheKey key, TimeToLive timeToLive) {
        try {

            // check the step id value
            String stepId = key.getStepId();
            if ("head".equals(stepId) || "origin".equals(stepId)) {
                throw new IllegalArgumentException("Illegal shortcut for preparation step '" + stepId + "'.");
            }

            Path rootPath = computeEntryPath(key);

            // Adds suffix for time to live checks
            final Path path = rootPath.suffix("." + String.valueOf(System.currentTimeMillis() + timeToLive.getTime()));
            final boolean created = fileSystem.createNewFile(path);
            if (!created) {
                LOGGER.error("[{} @{}] Unable to create cache.", key.getPreparationId(), stepId);
                return new NullOutputStream();
            }

            LOGGER.debug("[{}] Cache add.", key);

            return fileSystem.create(path);

        } catch (IOException e) {
            throw new TDPException(CommonErrorCodes.UNEXPECTED_EXCEPTION, e);
        }
    }

    /**
     * @see ContentCache#has(ContentCacheKey)
     */
    @Override
    public boolean has(ContentCacheKey key) {

        Path entry = computeEntryPath(key);

        FileStatus[] entries;
        try {
            entries = fileSystem.listStatus(entry.getParent(), p -> p.getName().startsWith(key.getSample()));
        } catch (IOException e) {
            return false;
        }

        final long now = System.currentTimeMillis();
        for (FileStatus status : entries) {
            final String suffix = StringUtils.substringAfterLast(status.getPath().getName(), ".");
            if (Long.parseLong(suffix) > now) {
                LOGGER.debug("[{}] Cache hit.", key);
                return true;
            }
        }

        LOGGER.debug("[{}] Cache miss.", key);
        return false;
    }

    /**
     * @see ContentCache#get(ContentCacheKey)
     */
    @Override
    public InputStream get(ContentCacheKey key) {

        Path entry = computeEntryPath(key);

        FileStatus[] files;
        try {
            files = fileSystem.listStatus(entry.getParent(), p -> p.getName().startsWith(key.getSample()));
        } catch (IOException e) {
            throw new IllegalArgumentException("No cache for preparation #" + key);
        }

        for (FileStatus file : files) {
            final String suffix = StringUtils.substringAfterLast(file.getPath().getName(), ".");
            if (Long.parseLong(suffix) > System.currentTimeMillis()) {
                try {
                    return fileSystem.open(file.getPath());
                } catch (IOException ioe) {
                    throw new TDPException(CommonErrorCodes.UNEXPECTED_EXCEPTION, ioe);
                }
            }
        }
        throw new IllegalArgumentException("No cache for preparation #" + key);
    }


    /**<commit_msg>(code coverage & sonar) Improves exception handling in HDFS content cache.<commit_after>package org.talend.dataprep.cache;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.commons.io.output.NullOutputStream;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.talend.dataprep.api.preparation.Step;
import org.talend.dataprep.exception.TDPException;
import org.talend.dataprep.exception.error.CommonErrorCodes;

/**
 * HDFS implementation of the content cache.
 *
 * Cache entries path look like this : datasets/{id}/preparations/{id}/steps/{stepid}/{sample} so that cache eviction
 * can happen at different level : dataset, preparation or step (sample is irrelevant because it's tied to the step).
 */
@Component
@EnableScheduling
@ConditionalOnProperty(name = "hdfs.location")
public class HDFSContentCache implements ContentCache {

    /** This class' logger. */
    private static final Logger LOGGER = LoggerFactory.getLogger(HDFSContentCache.class);

    /** Root folder name. */
    private static final String ROOT = "cache";

    /** Hdfs cache location. */
    @Autowired
    private FileSystem fileSystem;

    /**
     * Default empty constructor.
     */
    public HDFSContentCache() {
        LOGGER.info("Using content cache: {}", this.getClass().getName());
    }

    /**
     * Compute the path for the given key.
     *
     * @param key the cache key entry.
     * @return the HDFS path for the entry key.
     */
    private Path computeEntryPath(ContentCacheKey key) {

        String datasetId = key.getDatasetId();
        String preparationId = key.getPreparationId();
        String stepId = key.getStepId();
        String sample = key.getSample();

        if (datasetId == null) {
            throw new IllegalArgumentException("Dataset id cannot be null.");
        }

        if (preparationId == null) {
            throw new IllegalArgumentException("Preparation id cannot be null.");
        }

        if (key.getStepId() == null) {
            throw new IllegalArgumentException("Step id cannot be null.");
        }

        if ("origin".equalsIgnoreCase(stepId)) {
            stepId = Step.ROOT_STEP.id();
        }

        return new Path(ROOT + "/datasets/" + datasetId + "/preparations/" + preparationId + "/steps/" + stepId + '/' + sample);
    }

    /**
     * @see ContentCache#put(ContentCacheKey, TimeToLive)
     */
    @Override
    public OutputStream put(ContentCacheKey key, TimeToLive timeToLive) {
        try {

            // check the step id value
            String stepId = key.getStepId();
            if ("head".equals(stepId) || "origin".equals(stepId)) {
                throw new IllegalArgumentException("Illegal shortcut for preparation step '" + stepId + "'.");
            }

            Path rootPath = computeEntryPath(key);

            // Adds suffix for time to live checks
            final Path path = rootPath.suffix("." + String.valueOf(System.currentTimeMillis() + timeToLive.getTime()));
            final boolean created = fileSystem.createNewFile(path);
            if (!created) {
                LOGGER.error("[{} @{}] Unable to create cache.", key.getPreparationId(), stepId);
                return new NullOutputStream();
            }

            LOGGER.debug("[{}] Cache add.", key);

            return fileSystem.create(path);

        } catch (IOException e) {
            throw new TDPException(CommonErrorCodes.UNEXPECTED_EXCEPTION, e);
        }
    }

    /**
     * @see ContentCache#has(ContentCacheKey)
     */
    @Override
    public boolean has(ContentCacheKey key) {

        Path entry = computeEntryPath(key);

        FileStatus[] entries;
        try {
            entries = fileSystem.listStatus(entry.getParent(), p -> p.getName().startsWith(key.getSample()));
        } catch (IOException e) {
            LOGGER.debug("Unable to list status of for cache entry {}.", key, e);
            return false;
        }

        final long now = System.currentTimeMillis();
        for (FileStatus status : entries) {
            final String suffix = StringUtils.substringAfterLast(status.getPath().getName(), ".");
            if (Long.parseLong(suffix) > now) {
                LOGGER.debug("[{}] Cache hit.", key);
                return true;
            }
        }

        LOGGER.debug("[{}] Cache miss.", key);
        return false;
    }

    /**
     * @see ContentCache#get(ContentCacheKey)
     */
    @Override
    public InputStream get(ContentCacheKey key) {

        Path entry = computeEntryPath(key);

        FileStatus[] files;
        try {
            files = fileSystem.listStatus(entry.getParent(), p -> p.getName().startsWith(key.getSample()));
        } catch (IOException e) {
            throw new IllegalArgumentException("No cache for preparation #" + key, e);
        }

        for (FileStatus file : files) {
            final String suffix = StringUtils.substringAfterLast(file.getPath().getName(), ".");
            if (Long.parseLong(suffix) > System.currentTimeMillis()) {
                try {
                    return fileSystem.open(file.getPath());
                } catch (IOException ioe) {
                    throw new TDPException(CommonErrorCodes.UNEXPECTED_EXCEPTION, ioe);
                }
            }
        }
        throw new IllegalArgumentException("No cache for preparation #" + key);
    }


    /**
{'fraction_non_alphanumeric': 0.11543134872417983, 'fraction_numerical': 0.0003471619510501649, 'mean_word_length': 2.4438135086670654, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1567510', 'n_tokens_mistral': 3098, 'n_tokens_neox': 3019, 'n_words': 770}