skoppu22 commented on code in PR #169: URL: https://github.com/apache/cassandra-analytics/pull/169#discussion_r3539126787
########## cassandra-analytics-core/src/main/java/org/apache/cassandra/bridge/SSTableVersionAnalyzer.java: ########## @@ -0,0 +1,199 @@ +/* + * 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.util.Collection; +import java.util.Comparator; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.apache.cassandra.spark.bulkwriter.BulkSparkConf; + +/** + * Determines which Cassandra bridge to load from the SSTable versions present on a cluster. + */ +public final class SSTableVersionAnalyzer +{ + private SSTableVersionAnalyzer() + { + // Utility class + } + + /** + * Determines the bridge version for bulk write from the SSTable versions present on the cluster. + * Picks the lowest Cassandra version present so the produced SSTables can be imported by every node. + * + * @param sstableVersionsOnCluster set of SSTable versions found on cluster nodes + * @param requestedFormat user's requested format, e.g. "big" or "bti" + * @return the CassandraVersion bridge to load for write + * @throws IllegalStateException if the versions are empty/null or contain an unknown version + * @throws UnsupportedOperationException if the versions are mutually incompatible, or the chosen + * bridge does not support the requested format + */ + public static CassandraVersion determineBridgeVersionForWrite(Set<String> sstableVersionsOnCluster, String requestedFormat) + { + requireSSTableVersionsPresent(sstableVersionsOnCluster); + + // Every observed version must be mutually compatible (readable by the highest version present). + ensureMutuallyCompatibleVersions(sstableVersionsOnCluster, highestCompatibleVersion(sstableVersionsOnCluster)); + + CassandraVersion bridgeVersion = cassandraVersionsFromSSTableVersions(sstableVersionsOnCluster) + .min(Comparator.comparingInt(CassandraVersion::versionNumber)) + .orElseThrow(() -> new IllegalStateException("Unable to determine the lowest SSTable version")); Review Comment: Fixed this ########## cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/data/CassandraDataLayer.java: ########## @@ -844,6 +934,67 @@ private static String readNullable(ObjectInputStream in) throws IOException return null; } + /** + * Validates that all SSTables being read have versions that were observed in gossip info. + * This catches cases where SSTables have unexpected versions that weren't seen during driver initialization. + * This validation runs on executors. + * + * @param sstables list of SSTables to validate + * @throws UnsupportedOperationException if any SSTable has a version not in expected sstable versions + */ + @VisibleForTesting + void validateSStableVersions(List<SSTable> sstables) Review Comment: Done ########## cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java: ########## @@ -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() Review Comment: Done -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
