diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java index 0577e4290..74a2961f7 100644 --- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/sidecar/testing/SharedClusterIntegrationTestBase.java +++ b/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; } diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/MixedSSTableVersionTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/MixedSSTableVersionTest.java index b1c0758f8..6acdc65af 100644 --- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/MixedSSTableVersionTest.java +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/MixedSSTableVersionTest.java @@ -125,8 +125,8 @@ public void writeAndReadDifferentBigVersions() Set 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 dfReadAll = bulkReaderDataFrame(table1).load(); diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java index c5a0ecb90..6bcec93a7 100644 --- a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java +++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/SharedClusterSparkIntegrationTestBase.java @@ -199,8 +199,7 @@ public void checkSmallDataFrameEquality(Dataset expected, Dataset 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 ---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 ---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}) @@ -208,7 +207,6 @@ public void checkSmallDataFrameEquality(Dataset expected, Dataset actu */ 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,10 +219,10 @@ 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) @@ -232,6 +230,23 @@ protected void assertSSTableFormatOnDisk(QualifiedName table, String format, Str .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 ---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. diff --git a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java b/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java index 5dd648552..b71e12393 100644 --- a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/bridge/CassandraBridgeImplementation.java +++ b/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 partitions = Util.iterToStream(currentScanner); - JsonTransformer.toJson(currentScanner, partitions, false, metadata.get(), output); + SchemaVersionApi.writeSSTableJson(currentScanner, partitions, metadata, output); } catch (IOException exception) { diff --git a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/SSTableTombstoneWriter.java b/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/SSTableTombstoneWriter.java index dfd18bb3e..9258954fd 100644 --- a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/SSTableTombstoneWriter.java +++ b/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; }); diff --git a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/format/bti/BtiReaderUtils.java b/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/format/bti/BtiReaderUtils.java index 9eb88d76b..3e35bb0ac 100644 --- a/cassandra-five-zero-bridge/src/main/java/org/apache/cassandra/io/sstable/format/bti/BtiReaderUtils.java +++ b/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) diff --git a/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/BridgeClientInitializer.java b/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/BridgeClientInitializer.java new file mode 100644 index 000000000..f18b1660a --- /dev/null +++ b/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. + * + *

{@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.

+ */ +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(); + } +} diff --git a/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraTypesImplementation.java b/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraTypesImplementation.java index 085afed19..8c00994cc 100644 --- a/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraTypesImplementation.java +++ b/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"); diff --git a/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaUpdater.java b/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaUpdater.java index 399eabe9f..e6e937852 100644 --- a/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaUpdater.java +++ b/cassandra-five-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaUpdater.java @@ -19,35 +19,94 @@ package org.apache.cassandra.bridge; +import java.util.function.UnaryOperator; + import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.SchemaTransformation; import org.apache.cassandra.schema.SchemaTransformations; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.Types; +/** + * Applies schema mutations for the bridge. The {@code static} entry points ({@code load}/{@code updateTable}) + * are used by the shared {@code cassandra-four-zero-types} schema classes and by the bridge's + * {@code SSTableTombstoneWriter}; they express every mutation in terms of two primitives that delegate to a + * registered {@link #instance}: + *
    + *
  • {@link #apply(Schema, SchemaTransformation)} — apply a prebuilt {@link SchemaTransformation};
    • + *
  • {@link #applyKeyspacesOp(Schema, UnaryOperator)} — apply an in-place edit of the keyspace graph + * (its lambda can't be written in shared source because the {@link SchemaTransformation} SAM differs + * across distributions).
    • + *
+ * + *

This class is the default (Apache C* 5.0) implementation, mutating through {@code Schema.transform(...)}. + * A distribution whose mutation path differs registers a subclass via {@link #setInstance(SchemaUpdater)} that + * overrides the two {@code do*} primitives. + */ public class SchemaUpdater { - private SchemaUpdater() + private static volatile SchemaUpdater instance = new SchemaUpdater(); + + protected SchemaUpdater() { } + /** Registers the distribution-specific implementation. */ + public static void setInstance(SchemaUpdater impl) + { + instance = impl; + } + + // ------------------------------------------------------------------------------------------------------ + // Static entry points (call sites use these). + // ------------------------------------------------------------------------------------------------------ + public static void load(Schema schema, KeyspaceMetadata keyspaceMetadata) { - schema.transform(SchemaTransformations.addKeyspace(keyspaceMetadata, false)); + apply(schema, SchemaTransformations.addKeyspace(keyspaceMetadata, false)); } public static void load(Schema schema, KeyspaceMetadata keyspaceMetadata, TableMetadata tableMetadata) { - schema.transform(SchemaTransformations.addTable(tableMetadata, false)); + apply(schema, SchemaTransformations.addTable(tableMetadata, false)); } public static void load(Schema schema, KeyspaceMetadata keyspaceMetadata, Types userTypes) { - schema.transform(SchemaTransformations.addTypes(userTypes, true)); + apply(schema, SchemaTransformations.addTypes(userTypes, true)); } public static void updateTable(Schema schema, KeyspaceMetadata keyspaceMetadata, TableMetadata tableMetadata) { - schema.transform(st -> st.withAddedOrUpdated(keyspaceMetadata.withSwapped(keyspaceMetadata.tables.withSwapped(tableMetadata)))); + applyKeyspacesOp(schema, keyspaces -> + keyspaces.withAddedOrUpdated(keyspaceMetadata.withSwapped(keyspaceMetadata.tables.withSwapped(tableMetadata)))); + } + + /** Applies a prebuilt {@link SchemaTransformation}. */ + public static void apply(Schema schema, SchemaTransformation transformation) + { + instance.doApply(schema, transformation); + } + + /** Applies an in-place edit of the keyspace graph. */ + public static void applyKeyspacesOp(Schema schema, UnaryOperator keyspacesOp) + { + instance.doApplyKeyspacesOp(schema, keyspacesOp); + } + + // ------------------------------------------------------------------------------------------------------ + // Overridable defaults (Apache C* 5.0 behavior). + // ------------------------------------------------------------------------------------------------------ + + protected void doApply(Schema schema, SchemaTransformation transformation) + { + schema.transform(transformation); + } + + protected void doApplyKeyspacesOp(Schema schema, UnaryOperator keyspacesOp) + { + schema.transform(keyspaces -> keyspacesOp.apply(keyspaces)); } } diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java index bf90bbf02..7a8bc154b 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/CassandraSchema.java @@ -39,6 +39,7 @@ import org.apache.cassandra.cdc.api.TableIdLookup; import org.apache.cassandra.cql3.CQLFragmentParser; import org.apache.cassandra.cql3.CqlParser; +import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; import org.apache.cassandra.cql3.statements.schema.CreateTypeStatement; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.schema.KeyspaceMetadata; @@ -70,9 +71,10 @@ private CassandraSchema() */ public static void update(Consumer updater) { - synchronized (Schema.instance) + Schema schema = SchemaVersionApi.schemaInstance(); + synchronized (schema) { - updater.accept(Schema.instance); + updater.accept(schema); } } @@ -85,9 +87,10 @@ public static void update(Consumer updater) */ public static T apply(Function updater) { - synchronized (Schema.instance) + Schema schema = SchemaVersionApi.schemaInstance(); + synchronized (schema) { - return updater.apply(Schema.instance); + return updater.apply(schema); } } @@ -122,11 +125,9 @@ public static TableMetadata buildTableMetadata(String keyspace, @Nullable UUID tableId, boolean enableCdc) { - TableMetadata.Builder builder = CQLFragmentParser.parseAny(CqlParser::createTableStatement, createStmt, "CREATE TABLE") - .keyspace(keyspace) - .prepare(null) - .builder(types) - .partitioner(CassandraTypesImplementation.getPartitioner(partitioner)); + CreateTableStatement.Raw createTable = CQLFragmentParser.parseAny(CqlParser::createTableStatement, createStmt, "CREATE TABLE"); + TableMetadata.Builder builder = SchemaVersionApi.tableMetadataBuilder(createTable, keyspace, types) + .partitioner(CassandraTypesImplementation.getPartitioner(partitioner)); if (tableId != null) { @@ -170,7 +171,7 @@ public static Optional getKeyspaceMetadata(Schema schema, Stri public static Optional getTable(String keyspace, String table) { - return getTable(Schema.instance, keyspace, table); + return getTable(SchemaVersionApi.schemaInstance(), keyspace, table); } public static Optional getTable(Schema schema, String keyspace, String table) @@ -197,7 +198,7 @@ public static boolean isCdcEnabled(Schema schema, CqlTable cqlTable) public static boolean isCdcEnabled(String keyspace, String table) { - return isCdcEnabled(Schema.instance, keyspace, table); + return isCdcEnabled(SchemaVersionApi.schemaInstance(), keyspace, table); } public static boolean isCdcEnabled(Schema schema, String keyspace, String table) @@ -234,7 +235,7 @@ public static void updateCdcSchema(@NotNull Set cdcTables, @NotNull Partitioner partitioner, @NotNull TableIdLookup tableIdLookup) { - updateCdcSchema(Schema.instance, cdcTables, partitioner, tableIdLookup); + updateCdcSchema(SchemaVersionApi.schemaInstance(), cdcTables, partitioner, tableIdLookup); } public static void maybeUpdateSchema(Schema schema, @@ -380,7 +381,7 @@ private static void disableCdcOnStaleTables(Schema schema, Set */ public static void unregisterNonCdcTables(@NotNull Set tables) { - unregisterNonCdcTables(Schema.instance, tables); + unregisterNonCdcTables(SchemaVersionApi.schemaInstance(), tables); } public static void unregisterNonCdcTables(@NotNull Schema schema, @NotNull Set tables) diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaVersionApi.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaVersionApi.java new file mode 100644 index 000000000..fdebef7d3 --- /dev/null +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/bridge/SchemaVersionApi.java @@ -0,0 +1,155 @@ +/* + * 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.io.OutputStream; +import java.util.stream.Stream; + +import org.apache.cassandra.cql3.statements.schema.CreateTableStatement; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.io.sstable.ISSTableScanner; +import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.schema.TableMetadataRef; +import org.apache.cassandra.schema.Types; +import org.apache.cassandra.tools.JsonTransformer; + +/** + * Facade for the handful of {@code cassandra-all} calls whose shape or behavior differs across Cassandra + * distributions. The call sites live in the shared {@code cassandra-four-zero-types} classes + * ({@code CassandraSchema}, {@code AbstractSchemaBuilder}, {@code CqlUdt}) and the bridge + * ({@code CassandraBridgeImplementation}); they invoke the {@code static} entry points below, which delegate + * to a registered {@link #instance}. + * + *

This class is the default (Apache C* 4.0/5.0) implementation — its bodies inline the expressions the call + * sites previously used inline. A distribution whose {@code cassandra-all} API differs registers a subclass + * via {@link #setInstance(SchemaVersionApi)} that overrides only the divergent {@code do*} methods, so the + * large shared classes are consumed unchanged (no same-FQN overlay, no drift gate). + */ +public class SchemaVersionApi +{ + private static volatile SchemaVersionApi instance = new SchemaVersionApi(); + + protected SchemaVersionApi() + { + } + + /** + * Registers the distribution-specific implementation. Called once, before any schema access, from the + * bridge's client initialization. + */ + public static void setInstance(SchemaVersionApi impl) + { + instance = impl; + } + + // ------------------------------------------------------------------------------------------------------ + // Static entry points (call sites use these) -> delegate to the registered instance. + // ------------------------------------------------------------------------------------------------------ + + public static Schema schemaInstance() + { + return instance.doSchemaInstance(); + } + + public static void openKeyspaceInstance(String keyspaceName) + { + instance.doOpenKeyspaceInstance(keyspaceName); + } + + public static void reopenKeyspaceInstance(String keyspaceName) + { + instance.doReopenKeyspaceInstance(keyspaceName); + } + + public static void initColumnFamily(Schema schema, String keyspaceName, TableMetadata table) + { + instance.doInitColumnFamily(schema, keyspaceName, table); + } + + public static TableMetadata.Builder tableMetadataBuilder(CreateTableStatement.Raw createTable, + String keyspace, + Types types) + { + return instance.doTableMetadataBuilder(createTable, keyspace, types); + } + + public static void writeSSTableJson(ISSTableScanner scanner, + Stream partitions, + TableMetadataRef metadata, + OutputStream output) throws IOException + { + instance.doWriteSSTableJson(scanner, partitions, metadata, output); + } + + // ------------------------------------------------------------------------------------------------------ + // Overridable defaults (Apache C* behavior). + // ------------------------------------------------------------------------------------------------------ + + /** + * Returns the active schema. Always return the same instance on every call: callers lock on this object, + * so handing back a new one each time would break that locking. + */ + protected Schema doSchemaInstance() + { + return Schema.instance; + } + + /** Ensures the keyspace's runtime instance exists in the schema. */ + protected void doOpenKeyspaceInstance(String keyspaceName) + { + Keyspace.openWithoutSSTables(keyspaceName); + } + + /** + * Re-asserts the keyspace's runtime instance immediately before post-build validation. A no-op by default; + * a distribution whose schema mutations can transiently clear keyspace instances overrides this. + */ + protected void doReopenKeyspaceInstance(String keyspaceName) + { + // no-op by default + } + + /** Initializes the column-family store for a table whose keyspace instance is already open. */ + protected void doInitColumnFamily(Schema schema, String keyspaceName, TableMetadata table) + { + schema.getKeyspaceInstance(keyspaceName) + .initCf(TableMetadataRef.forOfflineTools(table), false); + } + + /** Prepares a CREATE TABLE statement and returns its {@link TableMetadata.Builder}. */ + protected TableMetadata.Builder doTableMetadataBuilder(CreateTableStatement.Raw createTable, + String keyspace, + Types types) + { + return createTable.keyspace(keyspace).prepare(null).builder(types); + } + + /** Serializes the given SSTable {@code partitions} to JSON on {@code output}. */ + protected void doWriteSSTableJson(ISSTableScanner scanner, + Stream partitions, + TableMetadataRef metadata, + OutputStream output) throws IOException + { + JsonTransformer.toJson(scanner, partitions, false, metadata.get(), output); + } +} diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlUdt.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlUdt.java index 31a8187d1..3deee748a 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlUdt.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/data/complex/CqlUdt.java @@ -38,12 +38,12 @@ import com.esotericsoftware.kryo.io.Input; import com.esotericsoftware.kryo.io.Output; import org.apache.cassandra.bridge.CassandraVersion; +import org.apache.cassandra.bridge.SchemaVersionApi; import org.apache.cassandra.cql3.functions.types.SettableByIndexData; import org.apache.cassandra.cql3.functions.types.UDTValue; import org.apache.cassandra.cql3.functions.types.UserType; import org.apache.cassandra.cql3.functions.types.UserTypeHelper; import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.cassandra.schema.Schema; import org.apache.cassandra.serializers.TypeSerializer; import org.apache.cassandra.serializers.UTF8Serializer; import org.apache.cassandra.spark.data.CassandraTypes; @@ -180,7 +180,7 @@ public AbstractType dataType() public AbstractType dataType(boolean isMultiCell) { // Get UserTypeSerializer from Schema instance to ensure fields are deserialized in correct order - return Schema.instance.getKeyspaceMetadata(keyspace()).types + return SchemaVersionApi.schemaInstance().getKeyspaceMetadata(keyspace()).types .get(UTF8Serializer.instance.serialize(name())) .orElseThrow(() -> new RuntimeException(String.format("UDT '%s' not initialized", name()))); } @@ -190,7 +190,7 @@ public AbstractType dataType(boolean isMultiCell) public TypeSerializer serializer() { // Get UserTypeSerializer from Schema instance to ensure fields are deserialized in correct order - return (TypeSerializer) Schema.instance.getKeyspaceMetadata(keyspace()).types + return (TypeSerializer) SchemaVersionApi.schemaInstance().getKeyspaceMetadata(keyspace()).types .get(UTF8Serializer.instance.serialize(name())) .orElseThrow(() -> new RuntimeException(String.format("UDT '%s' not initialized", name()))) .getSerializer(); diff --git a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java index 52ea0fd82..7503b46f9 100644 --- a/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java +++ b/cassandra-four-zero-types/src/main/java/org/apache/cassandra/spark/reader/AbstractSchemaBuilder.java @@ -41,6 +41,7 @@ import org.apache.cassandra.bridge.CassandraSchema; import org.apache.cassandra.bridge.CassandraTypesImplementation; import org.apache.cassandra.bridge.SchemaUpdater; +import org.apache.cassandra.bridge.SchemaVersionApi; import org.apache.cassandra.cql3.CQL3Type; import org.apache.cassandra.cql3.CQLFragmentParser; import org.apache.cassandra.cql3.CqlParser; @@ -61,7 +62,6 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.schema.TableMetadataRef; import org.apache.cassandra.schema.Types; import org.apache.cassandra.spark.data.CassandraTypes; import org.apache.cassandra.spark.data.CqlField; @@ -201,10 +201,7 @@ private static Pair updateSchema(Schema schema, tableId = maybeExistingTableMetadata.id.asUUID(); } - TableMetadata.Builder builder = createTable - .keyspace(keyspace) - .prepare(null) - .builder(types) + TableMetadata.Builder builder = SchemaVersionApi.tableMetadataBuilder(createTable, keyspace, types) .partitioner(cassPartitioner); if (tableId != null) @@ -226,6 +223,8 @@ private static Pair updateSchema(Schema schema, tableMetadata.columns().forEach(columnValidator); setupTableAndUdt(schema, keyspace, tableMetadata, types); + // Re-assert the keyspace instance before post-build validation (no-op by default). + SchemaVersionApi.reopenKeyspaceInstance(keyspace); return validateKeyspaceTable(schema, keyspace, tableMetadata.name); } @@ -363,7 +362,7 @@ private static void setupKeyspace(Schema schema, LOGGER.info("Setting up keyspace instance in schema keyspace={} rfStrategy={} partitioner={}", keyspaceName, replicationFactor.getReplicationStrategy().name(), partitioner); // Create keyspace instance and also initCf (cfs) for the table - Keyspace.openWithoutSSTables(keyspaceName); + SchemaVersionApi.openKeyspaceInstance(keyspaceName); } } @@ -409,13 +408,12 @@ private static void setupTableAndUdt(Schema schema, if (keyspaceInstanceExists(schema, keyspaceName)) { // initCf (cfs) in the opened keyspace - schema.getKeyspaceInstance(keyspaceName) - .initCf(TableMetadataRef.forOfflineTools(currentTable), false); + SchemaVersionApi.initColumnFamily(schema, keyspaceName, currentTable); } else { // The keyspace has not yet opened, create/open keyspace instance and also initCf (cfs) for the table - Keyspace.openWithoutSSTables(keyspaceName); + SchemaVersionApi.openKeyspaceInstance(keyspaceName); } }