CASSANALYTICS-24 and CASSANALYTICS-25 SSTable version based bridge determination

analytics-sidecar-client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java

@@ -244,6 +244,19 @@ public CompletableFuture<GossipInfoResponse> gossipInfo()
         return executor.executeRequestAsync(requestBuilder().gossipInfoRequest().build());
     }
 
+    /**
+     * Executes the gossip info request using the default retry policy and configured selection policy
+     *
+     * @param instance the instance where the request will be executed
+     * @return a completable future of the gossip info
+     */
+    public CompletableFuture<GossipInfoResponse> gossipInfo(SidecarInstance instance)
+    {
+        return executor.executeRequestAsync(requestBuilder().singleInstanceSelectionPolicy(instance)
+                                                            .gossipInfoRequest()
+                                                            .build());
+    }
+
 
     /**
      * Executes the GET gossip health request using the default retry policy and configured selection policy

cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/BridgeInitializationParameters.java

@@ -24,21 +24,20 @@
  */
 public class BridgeInitializationParameters
 {
-    private final String sstableFormat;
+    private final String configuredSSTableFormat;
 
-    public BridgeInitializationParameters(String sstableFormat)
+    public BridgeInitializationParameters(String configuredSSTableFormat)
     {
-        this.sstableFormat = sstableFormat;
+        this.configuredSSTableFormat = configuredSSTableFormat;
     }
 
     public static BridgeInitializationParameters fromEnvironment()
     {
-        String sstableFormat = CassandraVersion.sstableFormat();
-        return new BridgeInitializationParameters(sstableFormat);
+        return new BridgeInitializationParameters(CassandraVersion.configuredSSTableFormat());
     }
 
-    public String getSstableFormat()
+    public String getConfiguredSSTableFormat()
     {
-        return sstableFormat;
+        return configuredSSTableFormat;
     }
 }

cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java

@@ -20,7 +20,9 @@
 package org.apache.cassandra.bridge;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 
@@ -45,22 +47,51 @@
  */
 public enum CassandraVersion
 {
-    THREEZERO(30, "3.0", "three-zero", "big"),
-    FOURZERO(40, "4.0", "four-zero", "big"),
-    FOURONE(41, "4.1", "four-zero", "big"),
-    FIVEZERO(50, "5.0", "five-zero", "big", "bti");
+    THREEZERO(30, "3.0", "three-zero", new String[]{"big"},
+              new String[]{
+                  // Cassandra 3.x native sstable versions
+                  "big-ma",
+                  "big-mb",
+                  "big-mc",
+                  "big-md",
+                  "big-me",
+                  "big-mf"
+              }, 30),
+    FOURZERO(40, "4.0", "four-zero", new String[]{"big"},
+             new String[]{
+                 // Cassandra 4.0 native sstable versions
+                 "big-na",
+                 "big-nb",
+             }, 30),
+    FOURONE(41, "4.1", "four-zero", new String[]{"big"},
+            new String[]{
+                // Cassandra 4.1 did not introduce new native SSTable versions
+            }, 30),
+    FIVEZERO(50, "5.0", "five-zero", new String[]{"big", "bti"},
+             new String[]{
+                 // Cassandra 5.0 native sstable versions
+                 "big-oa",
+                 "bti-da",
+             }, 40);
 
     private final int number;
     private final String name;
     private final String jarBaseName;  // Must match shadowJar.archiveFileName from Gradle configuration (without extension)
     private final Set<String> sstableFormats;
+    private final List<String> nativeSStableVersions;
+    // Lowest Cassandra version number whose SSTables this version can read (inclusive).
+    private final int lowestCompatibleVersionNumber;
 
-    CassandraVersion(int number, String name, String jarBaseName, String... sstableFormats)
+
+    CassandraVersion(int number, String name, String jarBaseName, String[] sstableFormats, String[] nativeSStableVersions,
+                     int lowestCompatibleVersionNumber)
     {
         this.number = number;
         this.name = name;
         this.jarBaseName = jarBaseName;
         this.sstableFormats = new HashSet<>(Arrays.asList(sstableFormats));
+        this.nativeSStableVersions = List.of(nativeSStableVersions);
+        this.lowestCompatibleVersionNumber = lowestCompatibleVersionNumber;
     }
 
     public int versionNumber()
@@ -78,13 +109,69 @@ public String jarBaseName()
         return jarBaseName;
     }
 
-    private static final String sstableFormat;
+    /**
+     * Get the set of SSTable formats supported by this Cassandra version.
+     *
+     * @return Set of supported SSTable format strings
+     */
+    public Set<String> sstableFormats()
+    {
+        return sstableFormats;
+    }
+
+    /**
+     * Get the list of native SSTable version strings for this Cassandra version.
+     *
+     * @return List of native SSTable version strings
+     */
+    public List<String> getNativeSStableVersions()
+    {
+        return nativeSStableVersions;
+    }
+
+    /**
+     * Get the set of SSTable version strings that this Cassandra version can read.
+     * This includes the native versions of every Cassandra version from {@link #lowestCompatibleVersionNumber}
+     * (inclusive) up to this version (inclusive). For example, Cassandra 5.0 can read 4.0, 4.1 and 5.0
+     * SSTables (lowestCompatible=40) but NOT 3.x.
+     *
+     * @return Set of full SSTable version strings that can be read
+     */
+    public Set<String> getSupportedSStableVersionsForRead()
+    {
+        Set<String> readableVersions = new HashSet<>();
+        for (CassandraVersion version : CassandraVersion.values())
+        {
+            if (version.number >= lowestCompatibleVersionNumber && version.number <= this.number)
+            {
+                readableVersions.addAll(version.nativeSStableVersions);
+            }
+        }
+
+        return Collections.unmodifiableSet(readableVersions);
+    }
+
+    /**
+     * Whether this Cassandra version can read (and therefore import) SSTables natively written by {@code other}.
+     * A version can read SSTables from every version in the inclusive range
+     * [{@link #lowestCompatibleVersionNumber}, this version]. For example, Cassandra 5.0 can read 4.0/4.1/5.0
+     * SSTables but not 3.x.
+     *
+     * @param other the Cassandra version whose SSTables would be read
+     * @return {@code true} if this version can read {@code other}'s SSTables
+     */
+    public boolean canRead(CassandraVersion other)
+    {
+        return other.number >= this.lowestCompatibleVersionNumber && other.number <= this.number;
+    }
+
+    private static final String configuredSSTableFormat;
     private static final CassandraVersion[] implementedVersions;
     private static final String[] supportedVersions;
 
     static
     {
-        sstableFormat = System.getProperty("cassandra.analytics.bridges.sstable_format", "big");
+        configuredSSTableFormat = System.getProperty("cassandra.analytics.bridges.sstable_format", "big");
 
         // NOTE: These default enum names must stay in sync with cassandraVersionEnumMap in build.gradle.
         // FOURONE is intentionally excluded from local-dev defaults to keep iteration fast;
@@ -93,25 +180,25 @@ public String jarBaseName()
                                                               String.join(",", FOURZERO.name(), FIVEZERO.name()));
         implementedVersions = Arrays.stream(providedVersionsOrDefault.split(","))
                                     .map(CassandraVersion::valueOf)
-                                    .filter(v -> v.sstableFormats.contains(sstableFormat))
+                                    .filter(v -> v.sstableFormats().contains(configuredSSTableFormat))
                                     .toArray(CassandraVersion[]::new);
 
         // NOTE: These default versions must stay in sync with cassandraFullVersionMap in build.gradle.
         String providedSupportedVersionsOrDefault = System.getProperty("cassandra.analytics.bridges.supported_versions",
                                                                        "cassandra-4.0.17,cassandra-5.0.5");
         supportedVersions = Arrays.stream(providedSupportedVersionsOrDefault.split(","))
                                   .filter(version -> CassandraVersion.fromVersion(version)
-                                                                     .filter(v -> v.sstableFormats.contains(sstableFormat))
+                                                                     .filter(v -> v.sstableFormats().contains(configuredSSTableFormat))
                                                                      .isPresent())
                                   .toArray(String[]::new);
 
         Preconditions.checkArgument(implementedVersions.length > 0 && supportedVersions.length > 0,
                                     "No versions available");
     }
 
-    public static String sstableFormat()
+    public static String configuredSSTableFormat()
     {
-        return sstableFormat;
+        return configuredSSTableFormat;
     }
 
     public static Optional<CassandraVersion> fromVersion(String cassandraVersion)
@@ -122,6 +209,37 @@ public static Optional<CassandraVersion> fromVersion(String cassandraVersion)
                      .findAny();
     }
 
+    /**
+     * Find the Cassandra version that originally writes SSTables with this version string.
+     * Returns the native Cassandra version that introduced this SSTable version.
+     *
+     * @param sstableVersion full version string including format (e.g., "big-na", "bti-da")
+     * @return Optional containing the CassandraVersion that natively writes this format,
+     *         or Optional.empty() if:
+     *         <ul>
+     *           <li>The version string is null</li>
+     *           <li>The version string is unrecognized (not in any enum's nativeSStableVersions)</li>
+     *           <li>The version format is invalid or doesn't match expected pattern</li>
+     *         </ul>
+     */
+    public static Optional<CassandraVersion> fromSSTableVersion(String sstableVersion)
+    {
+        if (sstableVersion == null)
+        {
+            return Optional.empty();
+        }
+
+        for (CassandraVersion version : CassandraVersion.values())
+        {
+            if (version.nativeSStableVersions.contains(sstableVersion))
+            {
+                return Optional.of(version);
+            }
+        }
+
+        return Optional.empty();
+    }
+
     public static CassandraVersion[] implementedVersions()
     {
         return implementedVersions;