/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.connector.file.src;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.connector.file.src.util.CheckpointedPosition;
import org.apache.flink.core.fs.Path;
import org.apache.flink.util.StringUtils;

import javax.annotation.Nullable;

import java.io.Serializable;
import java.util.Arrays;
import java.util.Optional;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * A {@link SourceSplit} that represents a file, or a region of a file.
 *
 * <p>The split has an offset and an end, which defines the region of the file represented by
 * the split. For splits representing the while file, the offset is zero and the length is the
 * file size.
 *
 * <p>The split may furthermore have a "reader position", which is the checkpointed position from
 * a reader previously reading this split. This position is typically null when the split is assigned
 * from the enumerator to the readers, and is non-null when the readers checkpoint their state
 * in a file source split.
 *
 * <p>This class is {@link Serializable} for convenience. For Flink's internal serialization (both for
 * RPC and for checkpoints), the {@link FileSourceSplitSerializer} is used.
 */
@PublicEvolving
public class FileSourceSplit implements SourceSplit, Serializable {

	private static final long serialVersionUID = 1L;

	private static final String[] NO_HOSTS = StringUtils.EMPTY_STRING_ARRAY;

	/** The unique ID of the split. Unique within the scope of this source. */
	private final String id;

	/** The path of the file referenced by this split. */
	private final Path filePath;

	/** The position of the first byte in the file to process. */
	private final long offset;

	/** The number of bytes in the file to process. */
	private final long length;

	/** The names of the hosts storing this range of the file. Empty, if no host information is available. */
	private final String[] hostnames;

	/** The precise reader position in the split, to resume from. */
	@Nullable
	private final CheckpointedPosition readerPosition;

	/** The splits are frequently serialized into checkpoints.
	 * Caching the byte representation makes repeated serialization cheap.
	 * This field is used by {@link FileSourceSplitSerializer}. */
	@Nullable
	transient byte[] serializedFormCache;

	// --------------------------------------------------------------------------------------------

	/**
	 * Constructs a split with host information.
	 *
	 * @param id The unique ID of this source split.
	 * @param filePath The path to the file.
	 * @param offset The start (inclusive) of the split's rage in the file.
	 * @param length The number of bytes in the split (starting from the offset)
	 */
	public FileSourceSplit(String id, Path filePath, long offset, long length) {
		this(id, filePath, offset, length, NO_HOSTS);
	}

	/**
	 * Constructs a split with host information.
	 *
	 * @param filePath The path to the file.
	 * @param offset The start (inclusive) of the split's rage in the file.
	 * @param length The number of bytes in the split (starting from the offset)
	 * @param hostnames The hostnames of the nodes storing the split's file range.
	 */
	public FileSourceSplit(String id, Path filePath, long offset, long length, String... hostnames) {
		this(id, filePath, offset, length, hostnames, null, null);
	}

	/**
	 * Constructs a split with host information.
	 *
	 * @param filePath The path to the file.
	 * @param offset The start (inclusive) of the split's rage in the file.
	 * @param length The number of bytes in the split (starting from the offset)
	 * @param hostnames The hostnames of the nodes storing the split's file range.
	 */
	public FileSourceSplit(
			String id,
			Path filePath,
			long offset,
			long length,
			String[] hostnames,
			@Nullable CheckpointedPosition readerPosition) {
		this(id, filePath, offset, length, hostnames, readerPosition, null);
	}

	/**
	 * Package private constructor, used by the serializers to directly cache the serialized form.
	 */
	FileSourceSplit(
			String id,
			Path filePath,
			long offset,
			long length,
			String[] hostnames,
			@Nullable CheckpointedPosition readerPosition,
			@Nullable byte[] serializedForm) {

		checkArgument(offset >= 0, "offset must be >= 0");
		checkArgument(length >= 0, "length must be >= 0");
		checkNoNullHosts(hostnames);

		this.id = checkNotNull(id);
		this.filePath = checkNotNull(filePath);
		this.offset = offset;
		this.length = length;
		this.hostnames = hostnames;
		this.readerPosition = readerPosition;
		this.serializedFormCache = serializedForm;
	}

	// ------------------------------------------------------------------------
	//  split properties
	// ------------------------------------------------------------------------

	@Override
	public String splitId() {
		return id;
	}

	/**
	 * Gets the file's path.
	 */
	public Path path() {
		return filePath;
	}

	/**
	 * Returns the start of the file region referenced by this source split.
	 * The position is inclusive, the value indicates the first byte that is part of the split.
	 */
	public long offset() {
		return offset;
	}

	/**
	 * Returns the number of bytes in the file region described by this source split.
	 */
	public long length() {
		return length;
	}

	/**
	 * Gets the hostnames of the nodes storing the file range described by this split.
	 * The returned array is empty, if no host information is available.
	 *
	 * <p>Host information is typically only available on specific file systems, like HDFS.
	 */
	public String[] hostnames() {
		return hostnames;
	}

	/**
	 * Gets the (checkpointed) position of the reader, if set.
	 * This value is typically absent for splits when assigned from the enumerator to the readers,
	 * and present when the splits are recovered from a checkpoint.
	 */
	public Optional<CheckpointedPosition> getReaderPosition() {
		return Optional.ofNullable(readerPosition);
	}

	// ------------------------------------------------------------------------
	//  utils
	// ------------------------------------------------------------------------

	@Override
	public String toString() {
		final String hosts = hostnames.length == 0 ? "(no host info)" : " hosts=" + Arrays.toString(hostnames);
		return String.format("FileSourceSplit: %s [%d, %d) %s ID=%s position=%s",
				filePath, offset, offset + length, hosts, id, readerPosition);
	}

	private static void checkNoNullHosts(String[] hosts) {
		checkNotNull(hosts, "hostnames array must not be null");
		for (String host : hosts) {
			checkArgument(host != null, "the hostnames must not contain null entries");
		}
	}
}
