CASSANALYTICS-190 Add overridable extension points for version-varying Cassandra calls

cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java

@@ -279,7 +279,10 @@ protected ClusterBuilderConfiguration testClusterConfiguration()
     {
         ClusterBuilderConfiguration conf = new ClusterBuilderConfiguration();
         // TODO: Shall we read requested compatibility from sidecar (CASSANALYTICS-24)?
-        conf.additionalInstanceConfig(Map.of("storage_compatibility_mode", "NONE"));
+        // Defaults to "NONE"; subclasses or distributions can override the mode as needed.
+        String storageCompatibilityMode =
+            System.getProperty("cassandra.analytics.test.storageCompatibilityMode", "NONE");
+        conf.additionalInstanceConfig(Map.of("storage_compatibility_mode", storageCompatibilityMode));
         return conf;
     }
 

cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/MixedSSTableVersionTest.java

@@ -125,8 +125,8 @@ public void writeAndReadDifferentBigVersions()
 
         Set<String> dataFiles = findSSTableDataFiles(cluster.get(1), table1);
         // check that we produced data files in two different BIG versions
-        assertThat(dataFiles.stream().filter(name -> name.startsWith("nb-"))).isNotEmpty();
-        assertThat(dataFiles.stream().filter(name -> name.startsWith("oa-"))).isNotEmpty();
+        assertThat(dataFiles.stream().filter(name -> hasSSTableVersion(name, "nb"))).isNotEmpty();
+        assertThat(dataFiles.stream().filter(name -> hasSSTableVersion(name, "oa"))).isNotEmpty();
 
         // read the data back through the bulk reader
         // FIVEZERO bridge should be used
@@ -148,7 +148,7 @@ public void writeAndReadDifferentBigVersions()
         .isNotEmpty();
         assertThat(newFiles)
         .as("writer must pick the lowest mutually-compatible version present (4.0 -> big-nb): %s", newFiles)
-        .allMatch(name -> name.startsWith("nb-"));
+        .allMatch(name -> hasSSTableVersion(name, "nb"));
 
         // The bulk reader still returns every row across all three writes.
         Dataset<Row> dfReadAll = bulkReaderDataFrame(table1).load();

cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java

@@ -199,16 +199,14 @@ public void checkSmallDataFrameEquality(Dataset<Row> expected, Dataset<Row> actu
     /**
      * Asserts that every on-disk SSTable data file for the given table matches the expected SSTable format and
      * version across all running nodes. Data file names follow the pattern
-     * {@code <version>-<generation>-<format>-Data.db} (e.g. {@code oa-1-big-Data.db}). The generation component is
-     * matched loosely since it may be sequence- or UUID-based depending on cluster configuration.
+     * {@code <version>-<generation>-<format>-Data.db} (e.g. {@code oa-1-big-Data.db}).
      *
      * @param table           the table whose on-disk SSTables are inspected
      * @param format          the expected SSTable format (e.g. {@code big})
      * @param expectedVersion the expected SSTable version (e.g. {@code oa} or {@code nb})
      */
     protected void assertSSTableFormatOnDisk(QualifiedName table, String format, String expectedVersion)
     {
-        String dataFileRegex = expectedVersion + "-[^-]+-" + format + "-Data\\.db";
         boolean foundDataFiles = false;
         for (int i = 1; i <= cluster.size(); i++)
         {
@@ -221,17 +219,34 @@ protected void assertSSTableFormatOnDisk(QualifiedName table, String format, Str
             for (String fileName : findSSTableDataFiles(instance, table))
             {
                 foundDataFiles = true;
-                assertThat(fileName)
+                assertThat(hasSSTableVersionAndFormat(fileName, expectedVersion, format))
                 .as("SSTable data file for %s on node %d should be in %s format with version %s: %s",
                     table, i, format, expectedVersion, fileName)
-                .matches(dataFileRegex);
+                .isTrue();
             }
         }
         assertThat(foundDataFiles)
         .as("Expected to find at least one SSTable data file for %s on a running node", table)
         .isTrue();
     }
 
+    /**
+     * Whether the given SSTable data-file name carries the given version token (e.g. {@code nb}, {@code oa}).
+     */
+    protected static boolean hasSSTableVersion(String dataFileName, String version)
+    {
+        return dataFileName.startsWith(version + "-") || dataFileName.contains("-" + version + "-");
+    }
+
+    /**
+     * Whether the given SSTable data-file name carries the expected {@code <version>-<generation>-<format>-Data.db}
+     * tail. The generation is matched loosely (it may be sequence- or UUID-based).
+     */
+    protected static boolean hasSSTableVersionAndFormat(String dataFileName, String version, String format)
+    {
+        return dataFileName.matches("(?:.*-)?" + version + "-[^-]+-" + format + "-Data\\.db");
+    }
+
     /**
      * Finds the names of all SSTable {@code *-Data.db} files belonging to the given table on a single node,
      * scanning every configured data directory and scoping to the table's own data subdirectory.

cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java

@@ -79,7 +79,6 @@
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileOutputStreamPlus;
-import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.TableMetadataRef;
 import org.apache.cassandra.spark.data.CassandraTypes;
@@ -111,7 +110,6 @@
 import org.apache.cassandra.spark.utils.Pair;
 import org.apache.cassandra.spark.utils.SparkClassLoaderOverride;
 import org.apache.cassandra.spark.utils.TimeProvider;
-import org.apache.cassandra.tools.JsonTransformer;
 import org.apache.cassandra.tools.Util;
 import org.apache.cassandra.util.CompressionUtil;
 import org.apache.cassandra.util.IntWrapper;
@@ -631,7 +629,7 @@ public SSTableSummary getSSTableSummary(@NotNull String keyspace,
                                             @NotNull String table,
                                             @NotNull SSTable ssTable)
     {
-        TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
+        TableMetadata metadata = SchemaVersionApi.schemaInstance().getTableMetadata(keyspace, table);
         if (metadata == null)
         {
             throw new RuntimeException("Could not create table metadata needed for reading SSTable summaries for keyspace: " + keyspace);
@@ -739,7 +737,7 @@ public void sstableToJson(Path dataDbFile, OutputStream output) throws FileNotFo
             SSTableReader ssTable = SSTableReader.openNoValidation(null, desc, metadata);
             ISSTableScanner currentScanner = ssTable.getScanner();
             Stream<UnfilteredRowIterator> partitions = Util.iterToStream(currentScanner);
-            JsonTransformer.toJson(currentScanner, partitions, false, metadata.get(), output);
+            SchemaVersionApi.writeSSTableJson(currentScanner, partitions, metadata, output);
         }
         catch (IOException exception)
         {

cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/SSTableTombstoneWriter.java

@@ -34,6 +34,7 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.cassandra.bridge.CassandraSchema;
+import org.apache.cassandra.bridge.SchemaUpdater;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.ColumnSpecification;
 import org.apache.cassandra.cql3.QueryOptions;
@@ -416,7 +417,7 @@ public SSTableTombstoneWriter build()
             TableMetadata tableMetadata = CassandraSchema.apply(schema -> {
                 if (schema.getKeyspaceMetadata(SchemaConstants.SYSTEM_KEYSPACE_NAME) == null)
                 {
-                    schema.transform(SchemaTransformations.addKeyspace(SystemKeyspace.metadata(), false));
+                    SchemaUpdater.apply(schema, SchemaTransformations.addKeyspace(SystemKeyspace.metadata(), false));
                 }
 
                 String keyspaceName = schemaStatement.keyspace();
@@ -429,7 +430,7 @@ public SSTableTombstoneWriter build()
                                                                    Views.none(),
                                                                    Types.none(),
                                                                    UserFunctions.none());
-                    schema.transform(SchemaTransformations.addKeyspace(ksm, false));
+                    SchemaUpdater.apply(schema, SchemaTransformations.addKeyspace(ksm, false));
                 }
 
                 KeyspaceMetadata ksm = schema.getKeyspaceMetadata(keyspaceName);
@@ -440,7 +441,7 @@ public SSTableTombstoneWriter build()
                     Types types = createTypes(keyspaceName);
                     table = createTable(types);
                     TableMetadata finalTable = table;
-                    schema.transform(st -> st.withAddedOrUpdated(ksm.withSwapped(ksm.tables.with(finalTable)).withSwapped(types)));
+                    SchemaUpdater.applyKeyspacesOp(schema, st -> st.withAddedOrUpdated(ksm.withSwapped(ksm.tables.with(finalTable)).withSwapped(types)));
                 }
                 return table;
             });

cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/format/bti/BtiReaderUtils.java

@@ -140,6 +140,13 @@ public static Long startOffsetInDataFile(@NotNull SSTable ssTable,
                 BtiTableReader btiTableReader = new BtiTableReader.Builder(descriptor)
                                                 .setDataFile(dataFileHandle)
                                                 .setPartitionIndex(partitionIndex)
+                                                // Populate first/last from the partition index. This offline
+                                                // Builder.build() bypasses the online BtiTableReaderLoadingBuilder
+                                                // that normally sets them; leaving them null makes
+                                                // getPositionsForRanges dereference null on engines that read
+                                                // first/last directly (e.g. pre-CASSANDRA-20092). No-op on stock 5.x.
+                                                .setFirst(partitionIndex.firstKey())
+                                                .setLast(partitionIndex.lastKey())
                                                 .setRowIndexFile(rowFileHandle)
                                                 .setComponents(indexComponents)
                                                 .setTableMetadataRef(metadataRef)

cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/BridgeClientInitializer.java

@@ -0,0 +1,79 @@
+/*
+ * 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.bridge;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.UUID;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DataStorageSpec;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.locator.SimpleSnitch;
+
+/**
+ * Performs the one-time client-mode initialization of the embedded Cassandra engine used by the bridge.
+ *
+ * <p>{@link #initialize(BridgeInitializationParameters)} here is the stock Apache Cassandra 5.0 sequence. The
+ * client-init sequence is genuinely version-entangled — the snitch mechanism, whether the partitioner is set
+ * before or after {@code clientInitialization}, and any cluster-metadata bootstrap — so a Cassandra
+ * distribution whose client-init API differs simply subclasses this and overrides {@code initialize}, then
+ * registers the subclass via {@link java.util.ServiceLoader} (a
+ * {@code META-INF/services/org.apache.cassandra.bridge.BridgeClientInitializer} entry).
+ * {@code CassandraTypesImplementation} loads the registered subclass, falling back to this default when none
+ * is registered.</p>
+ */
+public class BridgeClientInitializer
+{
+    public void initialize(BridgeInitializationParameters params)
+    {
+        // We never want to enable mbean registration in the Cassandra code we use so disable it here
+        System.setProperty("org.apache.cassandra.disable_mbean_registration", "true");
+        System.setProperty("cassandra.schema.force_load_local_keyspaces", "true");
+        Config.setClientMode(true);
+        // When we create a TableStreamScanner, we will set the partitioner directly on the table metadata
+        // using the supplied IIndexStreamScanner.Partitioner. CFMetaData::compile requires a partitioner to
+        // be set in DatabaseDescriptor before we can do that though, so we set one here in preparation.
+        DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
+        Config config = new Config();
+        config.memtable_flush_writers = 8;
+        config.diagnostic_events_enabled = false;
+        config.max_mutation_size = new DataStorageSpec.IntKibibytesBound(config.commitlog_segment_size.toKibibytes() / 2);
+        config.concurrent_compactors = 4;
+        config.sstable.selected_format = params.getConfiguredSSTableFormat();
+        Path tempDirectory;
+        try
+        {
+            tempDirectory = Files.createTempDirectory(UUID.randomUUID().toString());
+        }
+        catch (IOException exception)
+        {
+            throw new RuntimeException(exception);
+        }
+        config.data_file_directories = new String[]{tempDirectory.toString()};
+        DatabaseDescriptor.clientInitialization(true, () -> config);
+        CassandraTypesImplementation.setupCommitLogConfigs(tempDirectory);
+        DatabaseDescriptor.setEndpointSnitch(new SimpleSnitch());
+        Keyspace.setInitialized();
+    }
+}

cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraTypesImplementation.java

@@ -19,19 +19,14 @@
 
 package org.apache.cassandra.bridge;
 
-import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.UUID;
+import java.util.ServiceLoader;
 
 import com.esotericsoftware.kryo.io.Input;
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DataStorageSpec;
 import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.commitlog.CommitLogSegmentManagerStandard;
-import org.apache.cassandra.dht.Murmur3Partitioner;
-import org.apache.cassandra.locator.SimpleSnitch;
 import org.apache.cassandra.security.EncryptionContext;
 import org.apache.cassandra.spark.data.CqlField;
 import org.apache.cassandra.spark.data.complex.CqlVector;
@@ -44,38 +39,22 @@ public static synchronized void setup(BridgeInitializationParameters params)
     {
         if (!CassandraTypesImplementation.setup)
         {
-            // We never want to enable mbean registration in the Cassandra code we use so disable it here
-            System.setProperty("org.apache.cassandra.disable_mbean_registration", "true");
-            System.setProperty("cassandra.schema.force_load_local_keyspaces", "true");
-            Config.setClientMode(true);
-            // When we create a TableStreamScanner, we will set the partitioner directly on the table metadata
-            // using the supplied IIndexStreamScanner.Partitioner. CFMetaData::compile requires a partitioner to
-            // be set in DatabaseDescriptor before we can do that though, so we set one here in preparation.
-            DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
-            Config config = new Config();
-            config.memtable_flush_writers = 8;
-            config.diagnostic_events_enabled = false;
-            config.max_mutation_size = new DataStorageSpec.IntKibibytesBound(config.commitlog_segment_size.toKibibytes() / 2);
-            config.concurrent_compactors = 4;
-            config.sstable.selected_format = params.getConfiguredSSTableFormat();
-            Path tempDirectory;
-            try
-            {
-                tempDirectory = Files.createTempDirectory(UUID.randomUUID().toString());
-            }
-            catch (IOException exception)
-            {
-                throw new RuntimeException(exception);
-            }
-            config.data_file_directories = new String[]{tempDirectory.toString()};
-            DatabaseDescriptor.clientInitialization(true, () -> config);
-            setupCommitLogConfigs(tempDirectory);
-            DatabaseDescriptor.setEndpointSnitch(new SimpleSnitch());
-            Keyspace.setInitialized();
+            // Client-mode engine initialization is version-specific (a distribution may change the snitch
+            // mechanism, the partitioner ordering, or add cluster-metadata bootstrap), so it is delegated to a
+            // BridgeClientInitializer discovered via ServiceLoader; absent a registered one, the stock Apache
+            // default is used. The synchronized one-time guard stays here.
+            resolveClientInitializer().initialize(params);
             setup = true;
         }
     }
 
+    private static BridgeClientInitializer resolveClientInitializer()
+    {
+        return ServiceLoader.load(BridgeClientInitializer.class, BridgeClientInitializer.class.getClassLoader())
+                            .findFirst()
+                            .orElseGet(BridgeClientInitializer::new);
+    }
+
     protected static void setupCommitLogConfigs(Path path)
     {
         Path commitLogPath = path.resolve("commitlog");