lukasz-antoniak commented on code in PR #169: URL: https://github.com/apache/cassandra-analytics/pull/169#discussion_r3448012339
########## cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/BulkRoundtripSSTableVersionBridgeTestBase.java: ########## @@ -0,0 +1,268 @@ +/* + * 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.analytics; + +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IInstance; +import org.apache.cassandra.sidecar.testing.QualifiedName; +import org.apache.cassandra.testing.ClusterBuilderConfiguration; +import org.apache.spark.SparkConf; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructField; +import org.apache.spark.sql.types.StructType; + +import static org.apache.cassandra.testing.TestUtils.DC1_RF3; +import static org.apache.cassandra.testing.TestUtils.TEST_KEYSPACE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * Shared roundtrip integration tests for SSTable version-based bridge selection. The same test logic runs + * for both the feature-enabled and feature-disabled cases. + * + * <p>Each test exercises a full write/read cycle and asserts the on-disk SSTable files use the format and + * version expected for the Cassandra version under test.</p> + * <ul> + * <li>{@link #testBulkWriteReadRoundtrip()} writes via the bulk writer, then reads back via the bulk reader.</li> + * <li>{@link #testBulkReadAcrossMultipleSSTables()} writes via CQL across several flushed SSTables, then bulk reads.</li> + * <li>{@link #testBulkReadFromSnapshot()} writes via CQL, snapshots, then bulk reads from the snapshot.</li> + * </ul> + */ +abstract class BulkRoundtripSSTableVersionBridgeTestBase extends SharedClusterSparkIntegrationTestBase +{ + static final List<String> DATASET = Arrays.asList("alpha", "beta", "gamma", "delta", "epsilon"); + static final QualifiedName TABLE_ROUNDTRIP = new QualifiedName(TEST_KEYSPACE, "test_roundtrip"); + static final QualifiedName TABLE_MULTIPLE = new QualifiedName(TEST_KEYSPACE, "test_multiple"); + static final QualifiedName TABLE_SNAPSHOT = new QualifiedName(TEST_KEYSPACE, "test_snapshot"); + + static final StructType SCHEMA = DataTypes.createStructType(new StructField[]{ + DataTypes.createStructField("id", DataTypes.IntegerType, false), + DataTypes.createStructField("value", DataTypes.StringType, false) + }); + + /** + * @return whether SSTable version-based bridge selection should be enabled for this test class + */ + protected abstract boolean sstableVersionBridgeEnabled(); + + /** + * @return the SSTable format the analytics bulk writer should produce ({@code big} by default; + * overridden to {@code bti} by the BTI variant). + */ + protected String sstableFormat() + { + return "big"; + } + + @Override + protected ClusterBuilderConfiguration testClusterConfiguration() + { + ClusterBuilderConfiguration conf = super.testClusterConfiguration() + .nodesPerDc(3); + // Preserve the base instance config (e.g. storage_compatibility_mode) and pin the node SSTable format. + Map<String, Object> instanceConfig = new HashMap<>(); + if (conf.additionalInstanceConfig != null) + { + instanceConfig.putAll(conf.additionalInstanceConfig); + } + instanceConfig.put("sstable.selected_format", sstableFormat()); + return conf.additionalInstanceConfig(instanceConfig); + } + + @Override + protected void beforeClusterProvisioning() + { + // Set before CassandraVersion class initialization (which reads this property once, statically) so the bulk + // writer generates SSTables in the configured format. This runs at the start of cluster provisioning, before + // any analytics bridge usage, and relies on forkEvery = 1 (the integration-test default) for a fresh JVM per + // test class. + System.setProperty("cassandra.analytics.bridges.sstable_format", sstableFormat()); + if ("bti".equals(sstableFormat())) + { + // BTI (bti-da) is a Cassandra 5.0+ format; skip on older versions. + assumeTrue(testVersion.version().startsWith("5."), Review Comment: NIT: Just to support Cassandra 6, maybe we can do: ``` Semver version = new Semver(testVersion.version(), Semver.SemverType.LOOSE); assumeTrue(version.isGreaterThanOrEqualTo(new Semver("5.0", Semver.SemverType.LOOSE)), "BTI format (bti-da) requires Cassandra 5.0+, but test version is " + testVersion.version()); ``` -- 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]
