diff --git a/CHANGES.txt b/CHANGES.txt index 1d645a060..5df559acd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 0.5.0 ----- + * AbstractSchemaBuilder.build() drops the cdc flag, silently disabling CDC on all tables (CASSANALYTICS-186) * Fix total timeout calculation for getAllNodeSettings (CASSANALYTICS-179) * Support vector data type (CASSANALYTICS-26) * CDC batch-write mixing a CDC-enabled and CDC-disabled table drops the CDC table's mutation (CASSANALYTICS-182) diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java index 2c5a2b251..d73e1976b 100644 --- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java +++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/spark/reader/SchemaBuilderTests.java @@ -19,6 +19,7 @@ package org.apache.cassandra.spark.reader; +import java.util.Collections; import java.util.HashMap; import org.junit.jupiter.api.Test; @@ -32,7 +33,9 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.Types; +import org.apache.cassandra.spark.data.CqlTable; import org.apache.cassandra.spark.data.ReplicationFactor; +import org.apache.cassandra.spark.data.partitioner.Partitioner; import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.spark.reader.SchemaBuilder.rfToMap; @@ -114,4 +117,51 @@ public void testSchemaBuilderWithPartiallyInitializedMetadata() new SchemaBuilder(createTableStatement, keyspaceName, replicationFactor); } + + @Test + public void testBuildPreservesCdcFlag() + { + CassandraBridgeImplementation.setup(); + String keyspaceName = "cdc" + getClass().getSimpleName(); + ReplicationFactor replicationFactor = new ReplicationFactor(ReplicationFactor.ReplicationStrategy.LocalStrategy, new HashMap<>()); + KeyspaceMetadata keyspaceMetadata = KeyspaceMetadata.create(keyspaceName, KeyspaceParams.create(true, rfToMap(replicationFactor))); + Schema.instance.load(keyspaceMetadata); + Keyspace.openWithoutSSTables(keyspaceName); + + String createTableStatement = "CREATE TABLE " + keyspaceName + ".bar (a int PRIMARY KEY) WITH cdc = true"; + TableMetadata tableMetadata = CQLFragmentParser + .parseAny(CqlParser::createTableStatement, createTableStatement, "CREATE TABLE") + .keyspace(keyspaceName) + .prepare(null) + .builder(Types.none()) + .build(); + KeyspaceMetadata keyspace = Schema.instance.getKeyspaceMetadata(keyspaceName); + Schema.instance.load(keyspace.withSwapped(keyspace.tables.with(tableMetadata))); + + SchemaBuilder cdcEnabledBuilder = new SchemaBuilder(createTableStatement, + keyspaceName, + replicationFactor, + Partitioner.Murmur3Partitioner, + bridge -> Collections.emptySet(), + null, + 0, + true); + CqlTable cdcEnabledTable = cdcEnabledBuilder.build(); + assertThat(cdcEnabledTable.cdc()) + .as("build() must preserve enableCdc=true through to the returned CqlTable") + .isTrue(); + + SchemaBuilder cdcDisabledBuilder = new SchemaBuilder(createTableStatement, + keyspaceName, + replicationFactor, + Partitioner.Murmur3Partitioner, + bridge -> Collections.emptySet(), + null, + 0, + false); + CqlTable cdcDisabledTable = cdcDisabledBuilder.build(); + assertThat(cdcDisabledTable.cdc()) + .as("build() must preserve enableCdc=false through to the returned CqlTable") + .isFalse(); + } } 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 e353ceaf4..52ea0fd82 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 @@ -479,7 +479,8 @@ public CqlTable build() replicationFactor, fields, new HashSet<>(udts.values()), - indexCount); + indexCount, + enableCdc); } private Map buildsUdts(KeyspaceMetadata keyspaceMetadata)