Skip to content

Commit 00c692f

Browse files
adutraadutra
authored
Merge pull request apache#745 from datastax/java1287
JAVA-1287: Add CDC to TableOptionsMetadata and Schema Builder
2 parents 7980ab9 + 5b15f69 commit 00c692f

8 files changed

Lines changed: 199 additions & 12 deletions

File tree

‎changelog/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
- [bug] JAVA-1397: Handle duration as native datatype in protocol v5+.
1010
- [improvement] JAVA-1308: CodecRegistry performance improvements.
1111
- [improvement] JAVA-1241: Upgrade Netty to 4.1.x.
12+
- [improvement] JAVA-1287: Add CDC to TableOptionsMetadata and Schema Builder.
1213

1314
Merged from 3.1.x branch:
1415

‎driver-core/src/main/java/com/datastax/driver/core/AbstractTableMetadata.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -269,6 +269,9 @@ protected StringBuilder appendOptions(StringBuilder sb, boolean formatted) {
269269
if (cassandraVersion.getMajor() > 2) {
270270
and(sb, formatted).append("crc_check_chance = ").append(options.getCrcCheckChance());
271271
}
272+
if (cassandraVersion.getMajor() > 3 || (cassandraVersion.getMajor() == 3 && cassandraVersion.getMinor() >= 8)) {
273+
and(sb, formatted).append("cdc = ").append(options.isCDC());
274+
}
272275
sb.append(';');
273276
return sb;
274277
}

‎driver-core/src/main/java/com/datastax/driver/core/TableOptionsMetadata.java‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@ public class TableOptionsMetadata {
4545
private static final String MAX_INDEX_INTERVAL = "max_index_interval";
4646
private static final String CRC_CHECK_CHANCE = "crc_check_chance";
4747
private static final String EXTENSIONS = "extensions";
48+
private static final String CDC = "cdc";
4849

4950
private static final boolean DEFAULT_REPLICATE_ON_WRITE = true;
5051
private static final double DEFAULT_BF_FP_CHANCE = 0.01;
@@ -56,6 +57,7 @@ public class TableOptionsMetadata {
5657
private static final int DEFAULT_MIN_INDEX_INTERVAL = 128;
5758
private static final int DEFAULT_MAX_INDEX_INTERVAL = 2048;
5859
private static final double DEFAULT_CRC_CHECK_CHANCE = 1.0;
60+
private static final boolean DEFAULT_CDC = false;
5961

6062
private final boolean isCompactStorage;
6163

@@ -77,12 +79,15 @@ public class TableOptionsMetadata {
7779
private final Map<String, String> compression;
7880
private final Double crcCheckChance;
7981
private final Map<String, ByteBuffer> extensions;
82+
private final boolean cdc;
8083

8184
TableOptionsMetadata(Row row, boolean isCompactStorage, VersionNumber version) {
8285

8386
boolean is120 = version.getMajor() < 2;
8487
boolean is200 = version.getMajor() == 2 && version.getMinor() == 0;
8588
boolean is210 = version.getMajor() == 2 && version.getMinor() >= 1;
89+
boolean is400OrHigher = version.getMajor() > 3;
90+
boolean is380OrHigher = is400OrHigher || version.getMajor() == 3 && version.getMinor() >= 8;
8691
boolean is300OrHigher = version.getMajor() > 2;
8792
boolean is210OrHigher = is210 || is300OrHigher;
8893

@@ -154,6 +159,13 @@ public class TableOptionsMetadata {
154159
this.extensions = ImmutableMap.copyOf(row.getMap(EXTENSIONS, String.class, ByteBuffer.class));
155160
else
156161
this.extensions = ImmutableMap.of();
162+
163+
if (is380OrHigher)
164+
this.cdc = isNullOrAbsent(row, CDC)
165+
? DEFAULT_CDC
166+
: row.getBool(CDC);
167+
else
168+
this.cdc = DEFAULT_CDC;
157169
}
158170

159171
private static boolean isNullOrAbsent(Row row, String name) {
@@ -366,6 +378,17 @@ public Map<String, ByteBuffer> getExtensions() {
366378
return extensions;
367379
}
368380

381+
/**
382+
* Returns whether or not change data capture is enabled for this table.
383+
* <p/>
384+
* For Cassandra versions prior to 3.8.0, this method always returns false.
385+
*
386+
* @return whether or not change data capture is enabled for this table.
387+
*/
388+
public boolean isCDC() {
389+
return cdc;
390+
}
391+
369392
@Override
370393
public boolean equals(Object other) {
371394
if (other == this)
@@ -385,6 +408,7 @@ public boolean equals(Object other) {
385408
this.populateCacheOnFlush == that.populateCacheOnFlush &&
386409
this.memtableFlushPeriodMs == that.memtableFlushPeriodMs &&
387410
this.defaultTTL == that.defaultTTL &&
411+
this.cdc == that.cdc &&
388412
Objects.equal(this.speculativeRetry, that.speculativeRetry) &&
389413
Objects.equal(this.indexInterval, that.indexInterval) &&
390414
Objects.equal(this.minIndexInterval, that.minIndexInterval) &&
@@ -399,6 +423,7 @@ public boolean equals(Object other) {
399423
public int hashCode() {
400424
return Objects.hashCode(isCompactStorage, comment, readRepair, localReadRepair, replicateOnWrite, gcGrace,
401425
bfFpChance, caching, populateCacheOnFlush, memtableFlushPeriodMs, defaultTTL, speculativeRetry,
402-
indexInterval, minIndexInterval, maxIndexInterval, compaction, compression, crcCheckChance, extensions);
426+
indexInterval, minIndexInterval, maxIndexInterval, compaction, compression, crcCheckChance, extensions,
427+
cdc);
403428
}
404429
}

‎driver-core/src/main/java/com/datastax/driver/core/schemabuilder/TableOptions.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,8 @@ public abstract class TableOptions<T extends TableOptions> extends SchemaStateme
6767

6868
private Optional<SpeculativeRetryValue> speculativeRetry = Optional.absent();
6969

70+
private Optional<Boolean> cdc = Optional.absent();
71+
7072
private List<String> customOptions = new ArrayList<String>();
7173

7274
@SuppressWarnings("unchecked")
@@ -357,6 +359,21 @@ public T speculativeRetry(SpeculativeRetryValue speculativeRetry) {
357359
return self;
358360
}
359361

362+
/**
363+
* Define whether or not change data capture is enabled on this table.
364+
* <p/>
365+
* Note that using this option with a version of Apache Cassandra less than 3.8 will raise a syntax error.
366+
* <p/>
367+
* If no call is made to this method, the default value set by Cassandra is {@code false}.
368+
*
369+
* @param cdc Whether or not change data capture should be enabled for this table.
370+
* @return this {@code TableOptions} object.
371+
*/
372+
public T cdc(Boolean cdc) {
373+
this.cdc = Optional.fromNullable(cdc);
374+
return self;
375+
}
376+
360377
/**
361378
* Define a free-form option as a key/value pair.
362379
* <p/>
@@ -448,6 +465,10 @@ private List<String> buildCommonOptions() {
448465
options.add("speculative_retry = " + speculativeRetry.get().value());
449466
}
450467

468+
if (cdc.isPresent()) {
469+
options.add("cdc = " + cdc.get());
470+
}
471+
451472
options.addAll(customOptions);
452473

453474
return options;

‎driver-core/src/test/java/com/datastax/driver/core/TableMetadataCDCTest.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
* Copyright (C) 2012-2015 DataStax Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.datastax.driver.core;
17+
18+
import com.datastax.driver.core.utils.CassandraVersion;
19+
import org.testng.annotations.Test;
20+
21+
import static org.assertj.core.api.Assertions.assertThat;
22+
23+
@CCMConfig(config = {"cdc_enabled:true"})
24+
@CassandraVersion(major = 3.8, description = "Requires CASSANDRA-12041 added in 3.8")
25+
public class TableMetadataCDCTest extends CCMTestsSupport {
26+
27+
/**
28+
* Ensures that if a table is configured with change data capture enabled that
29+
* {@link TableOptionsMetadata#isCDC()} returns true for that table.
30+
*
31+
* @test_category metadata
32+
* @jira_ticket JAVA-1287
33+
* @jira_ticket CASSANDRA-12041
34+
*/
35+
@Test(groups = "short")
36+
public void should_parse_cdc_from_table_options() {
37+
// given
38+
// create a simple table with cdc as true.
39+
String cql = String.format("CREATE TABLE %s.cdc_table (\n"
40+
+ " k text,\n"
41+
+ " c int,\n"
42+
+ " v timeuuid,\n"
43+
+ " PRIMARY KEY (k, c)\n"
44+
+ ") WITH cdc=true;", keyspace);
45+
session().execute(cql);
46+
47+
// when retrieving the table's metadata.
48+
TableMetadata table = cluster().getMetadata().getKeyspace(keyspace).getTable("cdc_table");
49+
// then the table's options should have cdc as true.
50+
assertThat(table.getOptions().isCDC()).isEqualTo(true);
51+
assertThat(table.asCQLQuery(true)).contains("cdc = true");
52+
}
53+
}

‎driver-core/src/test/java/com/datastax/driver/core/TableMetadataTest.java‎

Lines changed: 51 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -234,8 +234,8 @@ public void should_parse_table_options() {
234234
assertThat(table.getColumns().get(3)).isNotNull().hasName("i").isRegularColumn().hasType(cint());
235235
assertThat(table);
236236

237-
// Cassandra 3.0 +
238-
if (version.getMajor() > 2) {
237+
// Cassandra 3.8 +
238+
if (version.getMajor() > 3 || (version.getMajor() == 3 && version.getMinor() >= 8)) {
239239

240240
assertThat(table.getOptions().getReadRepairChance()).isEqualTo(0.5);
241241
assertThat(table.getOptions().getLocalReadRepairChance()).isEqualTo(0.6);
@@ -273,8 +273,51 @@ public void should_parse_table_options() {
273273
.contains("min_index_interval = 128")
274274
.contains("max_index_interval = 2048")
275275
.contains("crc_check_chance = 0.5")
276+
.contains("cdc = false")
276277
.doesNotContain(" index_interval")
277278
.doesNotContain("replicate_on_write");
279+
// Cassandra 3.0 +
280+
} else if (version.getMajor() > 2) {
281+
282+
assertThat(table.getOptions().getReadRepairChance()).isEqualTo(0.5);
283+
assertThat(table.getOptions().getLocalReadRepairChance()).isEqualTo(0.6);
284+
assertThat(table.getOptions().getGcGraceInSeconds()).isEqualTo(42);
285+
assertThat(table.getOptions().getBloomFilterFalsePositiveChance()).isEqualTo(0.01);
286+
assertThat(table.getOptions().getComment()).isEqualTo("My awesome table");
287+
assertThat(table.getOptions().getCaching()).contains(entry("keys", "ALL"));
288+
assertThat(table.getOptions().getCaching()).contains(entry("rows_per_partition", "10"));
289+
assertThat(table.getOptions().getCompaction()).contains(entry("class", "org.apache.cassandra.db.compaction.LeveledCompactionStrategy"));
290+
assertThat(table.getOptions().getCompaction()).contains(entry("sstable_size_in_mb", "15"));
291+
assertThat(table.getOptions().getCompression()).contains(entry("class", "org.apache.cassandra.io.compress.SnappyCompressor")); // sstable_compression becomes class
292+
assertThat(table.getOptions().getCompression()).contains(entry("chunk_length_in_kb", "128")); // note the "in" prefix
293+
assertThat(table.getOptions().getDefaultTimeToLive()).isEqualTo(0);
294+
assertThat(table.getOptions().getSpeculativeRetry()).isEqualTo("99.9PERCENTILE");
295+
assertThat(table.getOptions().getIndexInterval()).isNull();
296+
assertThat(table.getOptions().getMinIndexInterval()).isEqualTo(128);
297+
assertThat(table.getOptions().getMaxIndexInterval()).isEqualTo(2048);
298+
assertThat(table.getOptions().getReplicateOnWrite()).isTrue(); // default
299+
assertThat(table.getOptions().getCrcCheckChance()).isEqualTo(0.5);
300+
assertThat(table.getOptions().getExtensions()).isEmpty(); // default
301+
assertThat(table.asCQLQuery())
302+
.contains("read_repair_chance = 0.5")
303+
.contains("dclocal_read_repair_chance = 0.6")
304+
.contains("gc_grace_seconds = 42")
305+
.contains("bloom_filter_fp_chance = 0.01")
306+
.contains("comment = 'My awesome table'")
307+
.contains("'keys' : 'ALL'")
308+
.contains("'rows_per_partition' : 10")
309+
.contains("'class' : 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'")
310+
.contains("'sstable_size_in_mb' : 15")
311+
.contains("'class' : 'org.apache.cassandra.io.compress.SnappyCompressor'") // sstable_compression becomes class
312+
.contains("'chunk_length_in_kb' : 128") // note the "in" prefix
313+
.contains("default_time_to_live = 0")
314+
.contains("speculative_retry = '99.9PERCENTILE'")
315+
.contains("min_index_interval = 128")
316+
.contains("max_index_interval = 2048")
317+
.contains("crc_check_chance = 0.5")
318+
.doesNotContain(" index_interval")
319+
.doesNotContain("replicate_on_write")
320+
.doesNotContain("cdc"); // 3.8+
278321

279322
// Cassandra 2.1 and 2.2
280323
} else if (version.getMajor() == 2 && version.getMinor() > 0) {
@@ -315,7 +358,8 @@ public void should_parse_table_options() {
315358
.contains("min_index_interval = 128")
316359
.contains("max_index_interval = 2048")
317360
.doesNotContain(" index_interval")
318-
.doesNotContain("replicate_on_write");
361+
.doesNotContain("replicate_on_write")
362+
.doesNotContain("cdc");
319363

320364
// Cassandra 2.0
321365
} else if (version.getMajor() == 2 && version.getMinor() == 0) {
@@ -354,7 +398,8 @@ public void should_parse_table_options() {
354398
.contains("speculative_retry = '99.0PERCENTILE'")
355399
.contains("default_time_to_live = 0")
356400
.doesNotContain("min_index_interval") // 2.1 +
357-
.doesNotContain("max_index_interval"); // 2.1 +
401+
.doesNotContain("max_index_interval") // 2.1 +
402+
.doesNotContain("cdc");
358403

359404
// Cassandra 1.2
360405
} else {
@@ -393,7 +438,8 @@ public void should_parse_table_options() {
393438
.doesNotContain("min_index_interval") // 2.1 +
394439
.doesNotContain("max_index_interval") // 2.1 +
395440
.doesNotContain("speculative_retry") // 2.0 +
396-
.doesNotContain("default_time_to_live"); // 2.0 +
441+
.doesNotContain("default_time_to_live") // 2.0 +
442+
.doesNotContain("cdc");
397443

398444
}
399445

‎driver-core/src/test/java/com/datastax/driver/core/schemabuilder/AlterTest.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,8 @@ public void should_alter_table_options() throws Exception {
107107
.populateIOCacheOnFlush(true)
108108
.replicateOnWrite(true)
109109
.readRepairChance(0.42)
110-
.speculativeRetry(always());
110+
.speculativeRetry(always())
111+
.cdc(true);
111112

112113
SchemaStatement statementWith21Caching = alterTable("test").withOptions()
113114
.caching(KeyCaching.NONE, rows(100));
@@ -129,7 +130,8 @@ public void should_alter_table_options() throws Exception {
129130
"AND populate_io_cache_on_flush = true " +
130131
"AND read_repair_chance = 0.42 " +
131132
"AND replicate_on_write = true " +
132-
"AND speculative_retry = 'ALWAYS'");
133+
"AND speculative_retry = 'ALWAYS' " +
134+
"AND cdc = true");
133135

134136
assertThat(statementWith21Caching.getQueryString()).isEqualTo("\n\tALTER TABLE test\n\t" +
135137
"WITH caching = {'keys' : 'none', 'rows_per_partition' : 100}");

‎driver-core/src/test/java/com/datastax/driver/core/schemabuilder/CreateTest.java‎

Lines changed: 40 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,8 @@ public void should_create_table_with_udt_partition_key() throws Exception {
4848

4949
//Then
5050
assertThat(statement.getQueryString()).isEqualTo("\n\tCREATE TABLE test(\n\t\t" +
51-
"u frozen<user>,\n\t\t" +
52-
"PRIMARY KEY(u))"
51+
"u frozen<user>,\n\t\t" +
52+
"PRIMARY KEY(u))"
5353
);
5454
}
5555

@@ -324,7 +324,8 @@ public void should_create_table_with_all_options() throws Exception {
324324
.populateIOCacheOnFlush(true)
325325
.readRepairChance(0.05)
326326
.replicateOnWrite(true)
327-
.speculativeRetry(always());
327+
.speculativeRetry(always())
328+
.cdc(true);
328329

329330
//Then
330331
assertThat(statement.getQueryString()).isEqualTo("\n\tCREATE TABLE test(\n\t\t" +
@@ -346,7 +347,8 @@ public void should_create_table_with_all_options() throws Exception {
346347
"AND populate_io_cache_on_flush = true " +
347348
"AND read_repair_chance = 0.05 " +
348349
"AND replicate_on_write = true " +
349-
"AND speculative_retry = 'ALWAYS' AND CLUSTERING ORDER BY(col1 ASC, col2 DESC) AND COMPACT STORAGE");
350+
"AND speculative_retry = 'ALWAYS' " +
351+
"AND cdc = true AND CLUSTERING ORDER BY(col1 ASC, col2 DESC) AND COMPACT STORAGE");
350352
}
351353

352354
@Test(groups = "unit")
@@ -484,6 +486,40 @@ public void should_create_table_with_speculative_retry_in_milli_secs() throws Ex
484486
"WITH speculative_retry = '12ms'");
485487
}
486488

489+
@Test(groups = "unit")
490+
public void should_create_table_with_cdc_true() throws Exception {
491+
//When
492+
SchemaStatement statement = createTable("test")
493+
.addPartitionKey("id", DataType.bigint())
494+
.addColumn("name", DataType.text())
495+
.withOptions()
496+
.cdc(true);
497+
498+
//Then
499+
assertThat(statement.getQueryString()).isEqualTo("\n\tCREATE TABLE test(\n\t\t" +
500+
"id bigint,\n\t\t" +
501+
"name text,\n\t\t" +
502+
"PRIMARY KEY(id))\n\t" +
503+
"WITH cdc = true");
504+
}
505+
506+
@Test(groups = "unit")
507+
public void should_create_table_with_cdc_false() throws Exception {
508+
//When
509+
SchemaStatement statement = createTable("test")
510+
.addPartitionKey("id", DataType.bigint())
511+
.addColumn("name", DataType.text())
512+
.withOptions()
513+
.cdc(false);
514+
515+
//Then
516+
assertThat(statement.getQueryString()).isEqualTo("\n\tCREATE TABLE test(\n\t\t" +
517+
"id bigint,\n\t\t" +
518+
"name text,\n\t\t" +
519+
"PRIMARY KEY(id))\n\t" +
520+
"WITH cdc = false");
521+
}
522+
487523
@Test(groups = "unit", expectedExceptions = IllegalStateException.class,
488524
expectedExceptionsMessageRegExp = "The '\\[pk\\]' columns can not be declared as partition keys and clustering keys at the same time")
489525
public void should_fail_if_same_partition_and_clustering_column() throws Exception {

0 commit comments

Comments
 (0)