src/java/org/apache/cassandra/service/snapshot/TableSnapshot.java - cassandra - Git at Google

 /*
  * 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.cassandra.service.snapshot;

 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.Path;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Stream;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.lifecycle.SSTableSet;
 import org.apache.cassandra.io.FSReadError;
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.sstable.SSTableId;
 import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.Clock;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.LocalizeString;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.concurrent.Refs;

 import oshi.PlatformEnum;

 public class TableSnapshot
 {
     private static final Logger logger = LoggerFactory.getLogger(TableSnapshot.class);
     private static final PlatformEnum PLATFORM = FBUtilities.getSystemInfo().platform();

     private final String keyspaceName;
     private final String tableName;
     private final String keyspaceTable;
     private final UUID tableId;
     private final String tag;
     private final boolean ephemeral;

     private final Instant createdAt;
     private final Instant expiresAt;

     private final Set<File> snapshotDirs;

     private volatile long sizeOnDisk = 0;

     private volatile long manifestsSize;
     private volatile long schemasSize;

     private volatile boolean inProgress = false;

     public TableSnapshot(String keyspaceName, String tableName, UUID tableId,
                          String tag, Instant createdAt, Instant expiresAt,
                          Set<File> snapshotDirs, boolean ephemeral)
     {
         this.keyspaceName = keyspaceName;
         this.tableName = tableName;
         this.keyspaceTable = keyspaceName + '.' + tableName;
         this.tableId = tableId;
         this.tag = tag;
         this.createdAt = createdAt;
         this.expiresAt = expiresAt;
         this.snapshotDirs = snapshotDirs;
         this.ephemeral = ephemeral;

         manifestsSize = getManifestsSize();
         schemasSize = getSchemasSize();
     }

     public boolean isCompleted()
     {
         return !inProgress;
     }

     public void incomplete()
     {
         inProgress = true;
     }

     public void complete()
     {
         inProgress = false;
     }

     /**
      * Unique identifier of a snapshot. Used
      * only to deduplicate snapshots internally,
      * not exposed externally.
      * <p>
      * Format: "$ks:$table_name:$table_id:$tag"
      */
     public String getId()
     {
         return buildSnapshotId(keyspaceName, tableName, tableId, tag);
     }

     public String getKeyspaceName()
     {
         return keyspaceName;
     }

     public String getTableName()
     {
         return tableName;
     }

     public String getKeyspaceTable()
     {
         return keyspaceTable;
     }

     public String getTag()
     {
         return tag;
     }

     public Instant getCreatedAt()
     {
         if (createdAt == null)
         {
             long minCreation = 0;
             for (File snapshotDir : snapshotDirs)
             {
                 if (!snapshotDir.exists())
                     continue;

                 long lastModified = snapshotDir.lastModified();
                 if (lastModified == 0)
                     continue;

                 if (minCreation == 0 || minCreation > lastModified)
                     minCreation = lastModified;
             }

             if (minCreation != 0)
                 return Instant.ofEpochMilli(minCreation);
         }
         return createdAt;
     }

     public Instant getExpiresAt()
     {
         return expiresAt;
     }

     public boolean isExpired(Instant now)
     {
         if (createdAt == null || expiresAt == null)
         {
             return false;
         }

         return expiresAt.compareTo(now) < 0;
     }

     public boolean exists()
     {
         for (File snapshotDir : snapshotDirs)
             if (snapshotDir.exists())
                 return true;

         return false;
     }

     public boolean isEphemeral()
     {
         return ephemeral;
     }

     public boolean isExpiring()
     {
         return expiresAt != null;
     }

     public long computeSizeOnDiskBytes()
     {
         return computeSizeOnDiskBytes(false);
     }

     /**
      *
      * @param refresh true if a caller wants to recompute otherwise cached size
      * @return on disk bytes
      */
     public long computeSizeOnDiskBytes(boolean refresh)
     {
         long sum = sizeOnDisk;
         if (sum == 0 || refresh)
         {
             sum = 0;
             for (File snapshotDir : snapshotDirs)
                 sum += FileUtils.folderSize(snapshotDir);

             sizeOnDisk = sum;
         }

         return sum;
     }

     public long computeTrueSizeBytes()
     {
         ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(keyspaceName, tableName);
         if (cfs == null)
             return 0;

         return computeTrueSizeBytes(cfs.getFilesOfCfs());
     }

     public long computeTrueSizeBytes(Set<String> files)
     {
         long size = manifestsSize + schemasSize;

         for (File dataPath : getDirectories())
         {
             List<Path> snapshotFiles = listDir(dataPath.toPath());
             for (Path snapshotFile : snapshotFiles)
             {
                 if (!snapshotFile.endsWith("manifest.json") && !snapshotFile.endsWith("schema.cql"))
                 {
                     // files == null means that the underlying table was most probably dropped
                     // so in that case we indeed go to count snapshots file in for true size
                     if (files == null || (!files.contains(getLiveFileFromSnapshotFile(snapshotFile))))
                         size += getFileSize(snapshotFile);
                 }
             }
         }

         return size;
     }

     private List<Path> listDir(Path dir)
     {
         List<Path> paths = new ArrayList<>();
         try (Stream<Path> stream = Files.list(dir))
         {
             stream.forEach(p -> {
                 if (p.getFileName().toString().startsWith("."))
                 {
                     paths.addAll(listDir(p));
                 }
                 else
                 {
                     paths.add(p);
                 }
             });
         }
         catch (Throwable t)
         {
             if (!(t instanceof NoSuchFileException))
                 logger.error("Could not list directory content {}", dir, t);
         }

         return paths;
     }

     private long getFileSize(Path file)
     {
         try
         {
             return Files.size(file);
         }
         catch (Throwable t)
         {
             return 0;
         }
     }

     /**
      * Returns the corresponding live file for a given snapshot file.
      * <p>
      * Example:
      * - Base table:
      * - Snapshot file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/snapshots/1643481737850/me-1-big-Data.db
      * - Live file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/me-1-big-Data.db
      * - Secondary index:
      * - Snapshot file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/snapshots/1643481737850/.tbl_val_idx/me-1-big-Summary.db
      * - Live file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/.tbl_val_idx/me-1-big-Summary.db
      */
     static String getLiveFileFromSnapshotFile(Path snapshotFilePath)
     {
         // Snapshot directory structure format is {data_dir}/snapshots/{snapshot_name}/{snapshot_file}
         Path liveDir = snapshotFilePath.getParent().getParent().getParent();
         if (Directories.isSecondaryIndexFolder(snapshotFilePath.getParent()))
         {
             // Snapshot file structure format is {data_dir}/snapshots/{snapshot_name}/.{index}/{sstable-component}.db
             liveDir = File.getPath(liveDir.getParent().toString(), snapshotFilePath.getParent().getFileName().toString());
         }
         return liveDir.resolve(snapshotFilePath.getFileName().toString()).toAbsolutePath().toString();
     }

     public Collection<File> getDirectories()
     {
         return snapshotDirs;
     }

     /**
      * Returns all manifest files of a snapshot.
      * <p>
      * In practice, there might be multiple manifest files, as many as we have snapshot dirs.
      * Each snapshot dir will hold its view of a snapshot, containing only sstables located in such snapshot dir.
      *
      * @return all manifest files
      */
     public Set<File> getManifestFiles()
     {
         Set<File> manifestFiles = new HashSet<>();
         for (File snapshotDir : snapshotDirs)
         {
             File manifestFile = Directories.getSnapshotManifestFile(snapshotDir);
             if (manifestFile.exists())
                 manifestFiles.add(manifestFile);
         }
         return manifestFiles;
     }

     public boolean hasManifest()
     {
         for (File snapshotDir : snapshotDirs)
         {
             if (Directories.getSnapshotManifestFile(snapshotDir).exists())
                 return true;
         }
         return false;
     }

     /**
      * Returns all schemas files of a snapshot.
      *
      * @return all schema files
      */
     public Set<File> getSchemaFiles()
     {
         Set<File> schemaFiles = new HashSet<>();
         for (File snapshotDir : snapshotDirs)
         {
             File schemaFile = Directories.getSnapshotSchemaFile(snapshotDir);
             if (schemaFile.exists())
                 schemaFiles.add(schemaFile);
         }
         return schemaFiles;
     }

     public long getManifestsSize()
     {
         long size = 0;
         for (File manifestFile : getManifestFiles())
             size += manifestFile.length();

         return size;
     }

     public long getSchemasSize()
     {
         long size = 0;
         for (File schemaFile : getSchemaFiles())
             size += schemaFile.length();

         return size;
     }


     @Override
     public boolean equals(Object o)
     {
         if (this == o) return true;
         if (o == null || getClass() != o.getClass()) return false;
         TableSnapshot snapshot = (TableSnapshot) o;
         return Objects.equals(keyspaceName, snapshot.keyspaceName) &&
                Objects.equals(tableName, snapshot.tableName) &&
                Objects.equals(tableId, snapshot.tableId) &&
                tagsEqual(tag, snapshot.tag);
     }

     private boolean tagsEqual(String tag1, String tag2)
     {
         if (tag1 == null && tag2 == null)
             return true;

         if (tag1 == null || tag2 == null)
             return false;

         // When hardlinks are created for a snapshot with the name "snapshot"
         // and then we take a snapshot with the name "Snapshot", macOS platform thinks
         // that this was already hardlinked because its hardlinking implementation
         // does not seem to be case-sensitive. The fix consists of checking,
         // in a case-insensitive manner, if there is already such snapshot,
         // but only on macOS platform.
         if (PLATFORM == PlatformEnum.MACOS)
             return LocalizeString.toLowerCaseLocalized(tag1).equals(LocalizeString.toLowerCaseLocalized(tag2));

         return Objects.equals(tag1, tag2);
     }

     @Override
     public int hashCode()
     {
         return Objects.hash(keyspaceName, tableName, tableId, tag);
     }

     @Override
     public String toString()
     {
         return "TableSnapshot{" +
                "keyspaceName='" + keyspaceName + '\'' +
                ", tableName='" + tableName + '\'' +
                ", tableId=" + tableId +
                ", tag='" + tag + '\'' +
                ", createdAt=" + createdAt +
                ", expiresAt=" + expiresAt +
                ", snapshotDirs=" + snapshotDirs +
                ", ephemeral=" + ephemeral +
                '}';
     }

     public void updateMetadataSize()
     {
         if (manifestsSize == 0)
             manifestsSize = getManifestsSize();
         if (schemasSize == 0)
             schemasSize = getSchemasSize();
     }

     static class Builder
     {
         private final String keyspaceName;
         private final String tableName;
         private final UUID tableId;
         private final String tag;

         private Instant createdAt = null;
         private Instant expiresAt = null;
         private boolean ephemeral;

         private final Set<File> snapshotDirs = new HashSet<>();

         Builder(String keyspaceName, String tableName, UUID tableId, String tag)
         {
             this.keyspaceName = keyspaceName;
             this.tableName = tableName;
             this.tag = tag;
             this.tableId = tableId;
         }

         void addSnapshotDir(File snapshotDir)
         {
             snapshotDirs.add(snapshotDir);
             File manifestFile = new File(snapshotDir, "manifest.json");
             if (manifestFile.exists() && createdAt == null && expiresAt == null)
                 loadMetadataFromManifest(manifestFile);

             // check if an ephemeral marker file exists only in case it is not already ephemeral
             // by reading it from manifest
             // TODO remove this on Cassandra 4.3 release, see CASSANDRA-16911
             if (!ephemeral && new File(snapshotDir, "ephemeral.snapshot").exists())
                 ephemeral = true;
         }

         private void loadMetadataFromManifest(File manifestFile)
         {
             try
             {
                 logger.trace("Loading snapshot manifest from {}", manifestFile);
                 SnapshotManifest manifest = SnapshotManifest.deserializeFromJsonFile(manifestFile);
                 createdAt = manifest.createdAt;
                 expiresAt = manifest.expiresAt;
                 // a snapshot may be ephemeral when it has a marker file (old way) or flag in manifest (new way)
                 if (!ephemeral)
                     ephemeral = manifest.ephemeral;
             }
             catch (IOException e)
             {
                 logger.warn("Cannot read manifest file {} of snapshot {}.", manifestFile, tag, e);
             }
         }

         TableSnapshot build()
         {
             maybeCreateOrEnrichManifest();
             return new TableSnapshot(keyspaceName, tableName, tableId, tag, createdAt, expiresAt, snapshotDirs, ephemeral);
         }

         private List<File> getSnapshotDirsWithoutManifest()
         {
             List<File> snapshotDirNotHavingManifests = new ArrayList<>();
             for (File snapshotDir : snapshotDirs)
             {
                 if (!new File(snapshotDir.toPath().resolve("manifest.json")).exists())
                     snapshotDirNotHavingManifests.add(snapshotDir);
             }

             return snapshotDirNotHavingManifests;
         }

         private void maybeCreateOrEnrichManifest()
         {
             if (!CassandraRelevantProperties.SNAPSHOT_MANIFEST_ENRICH_OR_CREATE_ENABLED.getBoolean())
                 return;

             // this is caused by not reading any manifest or that snapshot had a basic manifest just with sstables
             // enumerated (pre CASSANDRA-16789), so we just go ahead and enrich it in each snapshot dir

             List<File> snapshotDirsWithoutManifests = getSnapshotDirsWithoutManifest();
             if (createdAt != null && snapshotDirsWithoutManifests.isEmpty())
                 return;

             if (createdAt == null && snapshotDirsWithoutManifests.isEmpty())
                 logger.info("Manifest in the old format for snapshot {} for {}.{} was detected, going to enrich it.",
                             tag, keyspaceName, tableName);

             if (!snapshotDirsWithoutManifests.isEmpty())
                 logger.info("There is no manifest for snapshot {} for {}.{} at paths {}, going to create it.",
                             tag, keyspaceName, tableName, snapshotDirsWithoutManifests);

             long lastModified = createdAt == null ? -1 : createdAt.toEpochMilli();

             if (lastModified == -1)
             {
                 for (File snapshotDir : snapshotDirs)
                 {
                     // we will consider time of the creation the oldest last modified
                     // timestamp on any snapshot directory
                     long currentLastModified = snapshotDir.lastModified();
                     if ((currentLastModified < lastModified || lastModified == -1) && currentLastModified > 0)
                         lastModified = currentLastModified;
                 }
             }

             List<String> allDataFiles = new ArrayList<>();
             for (File snapshotDir : snapshotDirs)
             {
                 List<File> dataFiles = new ArrayList<>();
                 try
                 {
                     List<File> indicesDirs = new ArrayList<>();
                     File[] snapshotFiles = snapshotDir.list(file -> {
                         if (file.isDirectory() && file.name().startsWith("."))
                         {
                             indicesDirs.add(file);
                             return false;
                         }
                         else
                         {
                             return file.name().endsWith('-' + SSTableFormat.Components.DATA.type.repr);
                         }
                     });

                     Collections.addAll(dataFiles, snapshotFiles);

                     for (File indexDir : indicesDirs)
                         dataFiles.addAll(Arrays.asList(indexDir.list(file -> file.name().endsWith('-' + SSTableFormat.Components.DATA.type.repr))));
                 }
                 catch (IOException ex)
                 {
                     logger.error("Unable to list a directory for data components: {}", snapshotDir);
                 }

                 for (File dataFile : dataFiles)
                 {
                     Descriptor descriptor = SSTable.tryDescriptorFromFile(dataFile);
                     if (descriptor != null)
                     {
                         String relativeDataFileName = descriptor.relativeFilenameFor(SSTableFormat.Components.DATA);
                         allDataFiles.add(relativeDataFileName);
                     }
                 }
             }

             // in any case of not being able to resolve it, set it to current time
             if (lastModified < 0)
                 createdAt = Instant.ofEpochMilli(Clock.Global.currentTimeMillis());
             else
                 createdAt = Instant.ofEpochMilli(lastModified);

             for (File snapshotDir : snapshotDirs)
             {
                 SnapshotManifest snapshotManifest = new SnapshotManifest(allDataFiles, null, createdAt, ephemeral);
                 File manifestFile = new File(snapshotDir, "manifest.json");
                 if (snapshotDirsWithoutManifests.contains(snapshotDir))
                 {
                     writeManifest(snapshotManifest, manifestFile);
                 }
                 else
                 {
                     SnapshotManifest existingManifest = readManifest(manifestFile);
                     if (existingManifest != null && existingManifest.createdAt == null)
                         writeManifest(snapshotManifest, manifestFile);
                 }
             }
         }

         private SnapshotManifest readManifest(File manifestFile)
         {
             try
             {
                 return SnapshotManifest.deserializeFromJsonFile(manifestFile);
             }
             catch (IOException ex)
             {
                 logger.error("Unable to read a manifest.json file in {}", manifestFile.absolutePath());
             }

             return null;
         }

         private void writeManifest(SnapshotManifest snapshotManifest, File manifestFile)
         {
             try
             {
                 snapshotManifest.serializeToJsonFile(manifestFile);
             }
             catch (IOException ex)
             {
                 logger.error("Unable to create a manifest.json file in {}", manifestFile.absolutePath());
             }
         }
     }

     protected static String buildSnapshotId(String keyspaceName, String tableName, UUID tableId, String tag)
     {
         return String.format("%s:%s:%s:%s", keyspaceName, tableName, tableId, tag);
     }

     public static Set<Descriptor> getSnapshotDescriptors(String keyspace, String table, String tag)
     {
         try
         {
             Refs<SSTableReader> snapshotSSTableReaders = getSnapshotSSTableReaders(keyspace, table, tag);

             Set<Descriptor> descriptors = new HashSet<>();
             for (SSTableReader ssTableReader : snapshotSSTableReaders)
             {
                 descriptors.add(ssTableReader.descriptor);
             }

             return descriptors;
         }
         catch (IOException e)
         {
             throw Throwables.unchecked(e);
         }
     }

     public static Refs<SSTableReader> getSnapshotSSTableReaders(String keyspace, String table, String tag) throws IOException
     {
         return getSnapshotSSTableReaders(Keyspace.open(keyspace).getColumnFamilyStore(table), tag);
     }

     public static Refs<SSTableReader> getSnapshotSSTableReaders(ColumnFamilyStore cfs, String tag)
     {
         Map<SSTableId, SSTableReader> active = new HashMap<>();
         for (SSTableReader sstable : cfs.getSSTables(SSTableSet.CANONICAL))
             active.put(sstable.descriptor.id, sstable);
         Map<Descriptor, Set<Component>> snapshots = cfs.getDirectories().sstableLister(Directories.OnTxnErr.IGNORE).snapshots(tag).list();
         Refs<SSTableReader> refs = new Refs<>();
         try
         {
             for (Map.Entry<Descriptor, Set<Component>> entries : snapshots.entrySet())
             {
                 // Try to acquire reference to an active sstable instead of snapshot if it exists,
                 // to avoid opening new sstables. If it fails, use the snapshot reference instead.
                 SSTableReader sstable = active.get(entries.getKey().id);
                 if (sstable == null || !refs.tryRef(sstable))
                 {
                     if (logger.isTraceEnabled())
                         logger.trace("using snapshot sstable {}", entries.getKey());
                     // open offline so we don't modify components or track hotness.
                     sstable = SSTableReader.open(cfs, entries.getKey(), entries.getValue(), cfs.metadata, true, true);
                     refs.tryRef(sstable);
                     // release the self ref as we never add the snapshot sstable to DataTracker where it is otherwise released
                     sstable.selfRef().release();
                 }
                 else if (logger.isTraceEnabled())
                 {
                     logger.trace("using active sstable {}", entries.getKey());
                 }
             }
         }
         catch (FSReadError | RuntimeException e)
         {
             // In case one of the snapshot sstables fails to open,
             // we must release the references to the ones we opened so far
             refs.release();
             throw e;
         }
         return refs;
     }

     public static String getTimestampedSnapshotName(String clientSuppliedName, long timestamp)
     {
         String snapshotName = Long.toString(timestamp);
         if (clientSuppliedName != null && !clientSuppliedName.isEmpty())
             snapshotName = snapshotName + '-' + clientSuppliedName;

         return snapshotName;
     }

     public static String getTimestampedSnapshotNameWithPrefix(String clientSuppliedName, long timestamp, String prefix)
     {
         return prefix + '-' + getTimestampedSnapshotName(clientSuppliedName, timestamp);
     }
 }
	/*
	* 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.cassandra.service.snapshot;

	import java.io.IOException;
	import java.nio.file.Files;
	import java.nio.file.NoSuchFileException;
	import java.nio.file.Path;
	import java.time.Instant;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Map;
	import java.util.Objects;
	import java.util.Set;
	import java.util.UUID;
	import java.util.stream.Stream;

	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.cassandra.config.CassandraRelevantProperties;
	import org.apache.cassandra.db.ColumnFamilyStore;
	import org.apache.cassandra.db.Directories;
	import org.apache.cassandra.db.Keyspace;
	import org.apache.cassandra.db.lifecycle.SSTableSet;
	import org.apache.cassandra.io.FSReadError;
	import org.apache.cassandra.io.sstable.Component;
	import org.apache.cassandra.io.sstable.Descriptor;
	import org.apache.cassandra.io.sstable.SSTable;
	import org.apache.cassandra.io.sstable.SSTableId;
	import org.apache.cassandra.io.sstable.format.SSTableFormat;
	import org.apache.cassandra.io.sstable.format.SSTableReader;
	import org.apache.cassandra.io.util.File;
	import org.apache.cassandra.io.util.FileUtils;
	import org.apache.cassandra.utils.Clock;
	import org.apache.cassandra.utils.FBUtilities;
	import org.apache.cassandra.utils.LocalizeString;
	import org.apache.cassandra.utils.Throwables;
	import org.apache.cassandra.utils.concurrent.Refs;

	import oshi.PlatformEnum;

	public class TableSnapshot
	{
	private static final Logger logger = LoggerFactory.getLogger(TableSnapshot.class);
	private static final PlatformEnum PLATFORM = FBUtilities.getSystemInfo().platform();

	private final String keyspaceName;
	private final String tableName;
	private final String keyspaceTable;
	private final UUID tableId;
	private final String tag;
	private final boolean ephemeral;

	private final Instant createdAt;
	private final Instant expiresAt;

	private final Set<File> snapshotDirs;

	private volatile long sizeOnDisk = 0;

	private volatile long manifestsSize;
	private volatile long schemasSize;

	private volatile boolean inProgress = false;

	public TableSnapshot(String keyspaceName, String tableName, UUID tableId,
	String tag, Instant createdAt, Instant expiresAt,
	Set<File> snapshotDirs, boolean ephemeral)
	{
	this.keyspaceName = keyspaceName;
	this.tableName = tableName;
	this.keyspaceTable = keyspaceName + '.' + tableName;
	this.tableId = tableId;
	this.tag = tag;
	this.createdAt = createdAt;
	this.expiresAt = expiresAt;
	this.snapshotDirs = snapshotDirs;
	this.ephemeral = ephemeral;

	manifestsSize = getManifestsSize();
	schemasSize = getSchemasSize();
	}

	public boolean isCompleted()
	{
	return !inProgress;
	}

	public void incomplete()
	{
	inProgress = true;
	}

	public void complete()
	{
	inProgress = false;
	}

	/**
	* Unique identifier of a snapshot. Used
	* only to deduplicate snapshots internally,
	* not exposed externally.
	* <p>
	* Format: "$ks:$table_name:$table_id:$tag"
	*/
	public String getId()
	{
	return buildSnapshotId(keyspaceName, tableName, tableId, tag);
	}

	public String getKeyspaceName()
	{
	return keyspaceName;
	}

	public String getTableName()
	{
	return tableName;
	}

	public String getKeyspaceTable()
	{
	return keyspaceTable;
	}

	public String getTag()
	{
	return tag;
	}

	public Instant getCreatedAt()
	{
	if (createdAt == null)
	{
	long minCreation = 0;
	for (File snapshotDir : snapshotDirs)
	{
	if (!snapshotDir.exists())
	continue;

	long lastModified = snapshotDir.lastModified();
	if (lastModified == 0)
	continue;

	if (minCreation == 0 \|\| minCreation > lastModified)
	minCreation = lastModified;
	}

	if (minCreation != 0)
	return Instant.ofEpochMilli(minCreation);
	}
	return createdAt;
	}

	public Instant getExpiresAt()
	{
	return expiresAt;
	}

	public boolean isExpired(Instant now)
	{
	if (createdAt == null \|\| expiresAt == null)
	{
	return false;
	}

	return expiresAt.compareTo(now) < 0;
	}

	public boolean exists()
	{
	for (File snapshotDir : snapshotDirs)
	if (snapshotDir.exists())
	return true;

	return false;
	}

	public boolean isEphemeral()
	{
	return ephemeral;
	}

	public boolean isExpiring()
	{
	return expiresAt != null;
	}

	public long computeSizeOnDiskBytes()
	{
	return computeSizeOnDiskBytes(false);
	}

	/**
	*
	* @param refresh true if a caller wants to recompute otherwise cached size
	* @return on disk bytes
	*/
	public long computeSizeOnDiskBytes(boolean refresh)
	{
	long sum = sizeOnDisk;
	if (sum == 0 \|\| refresh)
	{
	sum = 0;
	for (File snapshotDir : snapshotDirs)
	sum += FileUtils.folderSize(snapshotDir);

	sizeOnDisk = sum;
	}

	return sum;
	}

	public long computeTrueSizeBytes()
	{
	ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(keyspaceName, tableName);
	if (cfs == null)
	return 0;

	return computeTrueSizeBytes(cfs.getFilesOfCfs());
	}

	public long computeTrueSizeBytes(Set<String> files)
	{
	long size = manifestsSize + schemasSize;

	for (File dataPath : getDirectories())
	{
	List<Path> snapshotFiles = listDir(dataPath.toPath());
	for (Path snapshotFile : snapshotFiles)
	{
	if (!snapshotFile.endsWith("manifest.json") && !snapshotFile.endsWith("schema.cql"))
	{
	// files == null means that the underlying table was most probably dropped
	// so in that case we indeed go to count snapshots file in for true size
	if (files == null \|\| (!files.contains(getLiveFileFromSnapshotFile(snapshotFile))))
	size += getFileSize(snapshotFile);
	}
	}
	}

	return size;
	}

	private List<Path> listDir(Path dir)
	{
	List<Path> paths = new ArrayList<>();
	try (Stream<Path> stream = Files.list(dir))
	{
	stream.forEach(p -> {
	if (p.getFileName().toString().startsWith("."))
	{
	paths.addAll(listDir(p));
	}
	else
	{
	paths.add(p);
	}
	});
	}
	catch (Throwable t)
	{
	if (!(t instanceof NoSuchFileException))
	logger.error("Could not list directory content {}", dir, t);
	}

	return paths;
	}

	private long getFileSize(Path file)
	{
	try
	{
	return Files.size(file);
	}
	catch (Throwable t)
	{
	return 0;
	}
	}

	/**
	* Returns the corresponding live file for a given snapshot file.
	* <p>
	* Example:
	* - Base table:
	* - Snapshot file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/snapshots/1643481737850/me-1-big-Data.db
	* - Live file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/me-1-big-Data.db
	* - Secondary index:
	* - Snapshot file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/snapshots/1643481737850/.tbl_val_idx/me-1-big-Summary.db
	* - Live file: ~/.ccm/test/node1/data0/test_ks/tbl-e03faca0813211eca100c705ea09b5ef/.tbl_val_idx/me-1-big-Summary.db
	*/
	static String getLiveFileFromSnapshotFile(Path snapshotFilePath)
	{
	// Snapshot directory structure format is {data_dir}/snapshots/{snapshot_name}/{snapshot_file}
	Path liveDir = snapshotFilePath.getParent().getParent().getParent();
	if (Directories.isSecondaryIndexFolder(snapshotFilePath.getParent()))
	{
	// Snapshot file structure format is {data_dir}/snapshots/{snapshot_name}/.{index}/{sstable-component}.db
	liveDir = File.getPath(liveDir.getParent().toString(), snapshotFilePath.getParent().getFileName().toString());
	}
	return liveDir.resolve(snapshotFilePath.getFileName().toString()).toAbsolutePath().toString();
	}

	public Collection<File> getDirectories()
	{
	return snapshotDirs;
	}

	/**
	* Returns all manifest files of a snapshot.
	* <p>
	* In practice, there might be multiple manifest files, as many as we have snapshot dirs.
	* Each snapshot dir will hold its view of a snapshot, containing only sstables located in such snapshot dir.
	*
	* @return all manifest files
	*/
	public Set<File> getManifestFiles()
	{
	Set<File> manifestFiles = new HashSet<>();
	for (File snapshotDir : snapshotDirs)
	{
	File manifestFile = Directories.getSnapshotManifestFile(snapshotDir);
	if (manifestFile.exists())
	manifestFiles.add(manifestFile);
	}
	return manifestFiles;
	}

	public boolean hasManifest()
	{
	for (File snapshotDir : snapshotDirs)
	{
	if (Directories.getSnapshotManifestFile(snapshotDir).exists())
	return true;
	}
	return false;
	}

	/**
	* Returns all schemas files of a snapshot.
	*
	* @return all schema files
	*/
	public Set<File> getSchemaFiles()
	{
	Set<File> schemaFiles = new HashSet<>();
	for (File snapshotDir : snapshotDirs)
	{
	File schemaFile = Directories.getSnapshotSchemaFile(snapshotDir);
	if (schemaFile.exists())
	schemaFiles.add(schemaFile);
	}
	return schemaFiles;
	}

	public long getManifestsSize()
	{
	long size = 0;
	for (File manifestFile : getManifestFiles())
	size += manifestFile.length();

	return size;
	}

	public long getSchemasSize()
	{
	long size = 0;
	for (File schemaFile : getSchemaFiles())
	size += schemaFile.length();

	return size;
	}


	@Override
	public boolean equals(Object o)
	{
	if (this == o) return true;
	if (o == null \|\| getClass() != o.getClass()) return false;
	TableSnapshot snapshot = (TableSnapshot) o;
	return Objects.equals(keyspaceName, snapshot.keyspaceName) &&
	Objects.equals(tableName, snapshot.tableName) &&
	Objects.equals(tableId, snapshot.tableId) &&
	tagsEqual(tag, snapshot.tag);
	}

	private boolean tagsEqual(String tag1, String tag2)
	{
	if (tag1 == null && tag2 == null)
	return true;

	if (tag1 == null \|\| tag2 == null)
	return false;

	// When hardlinks are created for a snapshot with the name "snapshot"
	// and then we take a snapshot with the name "Snapshot", macOS platform thinks
	// that this was already hardlinked because its hardlinking implementation
	// does not seem to be case-sensitive. The fix consists of checking,
	// in a case-insensitive manner, if there is already such snapshot,
	// but only on macOS platform.
	if (PLATFORM == PlatformEnum.MACOS)
	return LocalizeString.toLowerCaseLocalized(tag1).equals(LocalizeString.toLowerCaseLocalized(tag2));

	return Objects.equals(tag1, tag2);
	}

	@Override
	public int hashCode()
	{
	return Objects.hash(keyspaceName, tableName, tableId, tag);
	}

	@Override
	public String toString()
	{
	return "TableSnapshot{" +
	"keyspaceName='" + keyspaceName + '\'' +
	", tableName='" + tableName + '\'' +
	", tableId=" + tableId +
	", tag='" + tag + '\'' +
	", createdAt=" + createdAt +
	", expiresAt=" + expiresAt +
	", snapshotDirs=" + snapshotDirs +
	", ephemeral=" + ephemeral +
	'}';
	}

	public void updateMetadataSize()
	{
	if (manifestsSize == 0)
	manifestsSize = getManifestsSize();
	if (schemasSize == 0)
	schemasSize = getSchemasSize();
	}

	static class Builder
	{
	private final String keyspaceName;
	private final String tableName;
	private final UUID tableId;
	private final String tag;

	private Instant createdAt = null;
	private Instant expiresAt = null;
	private boolean ephemeral;

	private final Set<File> snapshotDirs = new HashSet<>();

	Builder(String keyspaceName, String tableName, UUID tableId, String tag)
	{
	this.keyspaceName = keyspaceName;
	this.tableName = tableName;
	this.tag = tag;
	this.tableId = tableId;
	}

	void addSnapshotDir(File snapshotDir)
	{
	snapshotDirs.add(snapshotDir);
	File manifestFile = new File(snapshotDir, "manifest.json");
	if (manifestFile.exists() && createdAt == null && expiresAt == null)
	loadMetadataFromManifest(manifestFile);

	// check if an ephemeral marker file exists only in case it is not already ephemeral
	// by reading it from manifest
	// TODO remove this on Cassandra 4.3 release, see CASSANDRA-16911
	if (!ephemeral && new File(snapshotDir, "ephemeral.snapshot").exists())
	ephemeral = true;
	}

	private void loadMetadataFromManifest(File manifestFile)
	{
	try
	{
	logger.trace("Loading snapshot manifest from {}", manifestFile);
	SnapshotManifest manifest = SnapshotManifest.deserializeFromJsonFile(manifestFile);
	createdAt = manifest.createdAt;
	expiresAt = manifest.expiresAt;
	// a snapshot may be ephemeral when it has a marker file (old way) or flag in manifest (new way)
	if (!ephemeral)
	ephemeral = manifest.ephemeral;
	}
	catch (IOException e)
	{
	logger.warn("Cannot read manifest file {} of snapshot {}.", manifestFile, tag, e);
	}
	}

	TableSnapshot build()
	{
	maybeCreateOrEnrichManifest();
	return new TableSnapshot(keyspaceName, tableName, tableId, tag, createdAt, expiresAt, snapshotDirs, ephemeral);
	}

	private List<File> getSnapshotDirsWithoutManifest()
	{
	List<File> snapshotDirNotHavingManifests = new ArrayList<>();
	for (File snapshotDir : snapshotDirs)
	{
	if (!new File(snapshotDir.toPath().resolve("manifest.json")).exists())
	snapshotDirNotHavingManifests.add(snapshotDir);
	}

	return snapshotDirNotHavingManifests;
	}

	private void maybeCreateOrEnrichManifest()
	{
	if (!CassandraRelevantProperties.SNAPSHOT_MANIFEST_ENRICH_OR_CREATE_ENABLED.getBoolean())
	return;

	// this is caused by not reading any manifest or that snapshot had a basic manifest just with sstables
	// enumerated (pre CASSANDRA-16789), so we just go ahead and enrich it in each snapshot dir

	List<File> snapshotDirsWithoutManifests = getSnapshotDirsWithoutManifest();
	if (createdAt != null && snapshotDirsWithoutManifests.isEmpty())
	return;

	if (createdAt == null && snapshotDirsWithoutManifests.isEmpty())
	logger.info("Manifest in the old format for snapshot {} for {}.{} was detected, going to enrich it.",
	tag, keyspaceName, tableName);

	if (!snapshotDirsWithoutManifests.isEmpty())
	logger.info("There is no manifest for snapshot {} for {}.{} at paths {}, going to create it.",
	tag, keyspaceName, tableName, snapshotDirsWithoutManifests);

	long lastModified = createdAt == null ? -1 : createdAt.toEpochMilli();

	if (lastModified == -1)
	{
	for (File snapshotDir : snapshotDirs)
	{
	// we will consider time of the creation the oldest last modified
	// timestamp on any snapshot directory
	long currentLastModified = snapshotDir.lastModified();
	if ((currentLastModified < lastModified \|\| lastModified == -1) && currentLastModified > 0)
	lastModified = currentLastModified;
	}
	}

	List<String> allDataFiles = new ArrayList<>();
	for (File snapshotDir : snapshotDirs)
	{
	List<File> dataFiles = new ArrayList<>();
	try
	{
	List<File> indicesDirs = new ArrayList<>();
	File[] snapshotFiles = snapshotDir.list(file -> {
	if (file.isDirectory() && file.name().startsWith("."))
	{
	indicesDirs.add(file);
	return false;
	}
	else
	{
	return file.name().endsWith('-' + SSTableFormat.Components.DATA.type.repr);
	}
	});

	Collections.addAll(dataFiles, snapshotFiles);

	for (File indexDir : indicesDirs)
	dataFiles.addAll(Arrays.asList(indexDir.list(file -> file.name().endsWith('-' + SSTableFormat.Components.DATA.type.repr))));
	}
	catch (IOException ex)
	{
	logger.error("Unable to list a directory for data components: {}", snapshotDir);
	}

	for (File dataFile : dataFiles)
	{
	Descriptor descriptor = SSTable.tryDescriptorFromFile(dataFile);
	if (descriptor != null)
	{
	String relativeDataFileName = descriptor.relativeFilenameFor(SSTableFormat.Components.DATA);
	allDataFiles.add(relativeDataFileName);
	}
	}
	}

	// in any case of not being able to resolve it, set it to current time
	if (lastModified < 0)
	createdAt = Instant.ofEpochMilli(Clock.Global.currentTimeMillis());
	else
	createdAt = Instant.ofEpochMilli(lastModified);

	for (File snapshotDir : snapshotDirs)
	{
	SnapshotManifest snapshotManifest = new SnapshotManifest(allDataFiles, null, createdAt, ephemeral);
	File manifestFile = new File(snapshotDir, "manifest.json");
	if (snapshotDirsWithoutManifests.contains(snapshotDir))
	{
	writeManifest(snapshotManifest, manifestFile);
	}
	else
	{
	SnapshotManifest existingManifest = readManifest(manifestFile);
	if (existingManifest != null && existingManifest.createdAt == null)
	writeManifest(snapshotManifest, manifestFile);
	}
	}
	}

	private SnapshotManifest readManifest(File manifestFile)
	{
	try
	{
	return SnapshotManifest.deserializeFromJsonFile(manifestFile);
	}
	catch (IOException ex)
	{
	logger.error("Unable to read a manifest.json file in {}", manifestFile.absolutePath());
	}

	return null;
	}

	private void writeManifest(SnapshotManifest snapshotManifest, File manifestFile)
	{
	try
	{
	snapshotManifest.serializeToJsonFile(manifestFile);
	}
	catch (IOException ex)
	{
	logger.error("Unable to create a manifest.json file in {}", manifestFile.absolutePath());
	}
	}
	}

	protected static String buildSnapshotId(String keyspaceName, String tableName, UUID tableId, String tag)
	{
	return String.format("%s:%s:%s:%s", keyspaceName, tableName, tableId, tag);
	}

	public static Set<Descriptor> getSnapshotDescriptors(String keyspace, String table, String tag)
	{
	try
	{
	Refs<SSTableReader> snapshotSSTableReaders = getSnapshotSSTableReaders(keyspace, table, tag);

	Set<Descriptor> descriptors = new HashSet<>();
	for (SSTableReader ssTableReader : snapshotSSTableReaders)
	{
	descriptors.add(ssTableReader.descriptor);
	}

	return descriptors;
	}
	catch (IOException e)
	{
	throw Throwables.unchecked(e);
	}
	}

	public static Refs<SSTableReader> getSnapshotSSTableReaders(String keyspace, String table, String tag) throws IOException
	{
	return getSnapshotSSTableReaders(Keyspace.open(keyspace).getColumnFamilyStore(table), tag);
	}

	public static Refs<SSTableReader> getSnapshotSSTableReaders(ColumnFamilyStore cfs, String tag)
	{
	Map<SSTableId, SSTableReader> active = new HashMap<>();
	for (SSTableReader sstable : cfs.getSSTables(SSTableSet.CANONICAL))
	active.put(sstable.descriptor.id, sstable);
	Map<Descriptor, Set<Component>> snapshots = cfs.getDirectories().sstableLister(Directories.OnTxnErr.IGNORE).snapshots(tag).list();
	Refs<SSTableReader> refs = new Refs<>();
	try
	{
	for (Map.Entry<Descriptor, Set<Component>> entries : snapshots.entrySet())
	{
	// Try to acquire reference to an active sstable instead of snapshot if it exists,
	// to avoid opening new sstables. If it fails, use the snapshot reference instead.
	SSTableReader sstable = active.get(entries.getKey().id);
	if (sstable == null \|\| !refs.tryRef(sstable))
	{
	if (logger.isTraceEnabled())
	logger.trace("using snapshot sstable {}", entries.getKey());
	// open offline so we don't modify components or track hotness.
	sstable = SSTableReader.open(cfs, entries.getKey(), entries.getValue(), cfs.metadata, true, true);
	refs.tryRef(sstable);
	// release the self ref as we never add the snapshot sstable to DataTracker where it is otherwise released
	sstable.selfRef().release();
	}
	else if (logger.isTraceEnabled())
	{
	logger.trace("using active sstable {}", entries.getKey());
	}
	}
	}
	catch (FSReadError \| RuntimeException e)
	{
	// In case one of the snapshot sstables fails to open,
	// we must release the references to the ones we opened so far
	refs.release();
	throw e;
	}
	return refs;
	}

	public static String getTimestampedSnapshotName(String clientSuppliedName, long timestamp)
	{
	String snapshotName = Long.toString(timestamp);
	if (clientSuppliedName != null && !clientSuppliedName.isEmpty())
	snapshotName = snapshotName + '-' + clientSuppliedName;

	return snapshotName;
	}

	public static String getTimestampedSnapshotNameWithPrefix(String clientSuppliedName, long timestamp, String prefix)
	{
	return prefix + '-' + getTimestampedSnapshotName(clientSuppliedName, timestamp);
	}
	}