This is an automated email from the ASF dual-hosted git repository.
marcuse pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new aaa4a60 Extract an AbstractCompactionController to allow for custom
implementations
aaa4a60 is described below
commit aaa4a602ef1f658bd3898ceeedc3dc1f0fdc4fcd
Author: James Berragan <[email protected]>
AuthorDate: Tue Aug 20 14:47:26 2019 -0700
Extract an AbstractCompactionController to allow for custom implementations
Patch by James Berragan; reviewed by marcuse for CASSANDRA-15286
---
CHANGES.txt | 1 +
.../cassandra/db/AbstractCompactionController.java | 61 ++++++++++++++++++++++
.../db/compaction/CompactionController.java | 37 +++----------
.../db/compaction/CompactionIterator.java | 14 ++---
4 files changed, 77 insertions(+), 36 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 96be055..a901441 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0-alpha2
+ * Extract an AbstractCompactionController to allow for custom implementations
(CASSANDRA-15286)
* Move chronicle-core version from snapshot to stable, and include
carrotsearch in generated pom.xml (CASSANDRA-15321)
* Untangle RepairMessage sub-hierarchy of messages, use new messaging (more)
correctly (CASSANDRA-15163)
* Add `allocate_tokens_for_local_replication_factor` option for token
allocation (CASSANDRA-15260)
diff --git a/src/java/org/apache/cassandra/db/AbstractCompactionController.java
b/src/java/org/apache/cassandra/db/AbstractCompactionController.java
new file mode 100644
index 0000000..99193f8
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/AbstractCompactionController.java
@@ -0,0 +1,61 @@
+/*
+ * 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.db;
+
+import java.util.function.LongPredicate;
+
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.schema.CompactionParams;
+
+/**
+ * AbstractCompactionController allows custom implementations of the
CompactionController for use in tooling, without being tied to the
SSTableReader and local filesystem
+ */
+public abstract class AbstractCompactionController implements AutoCloseable
+{
+ public final ColumnFamilyStore cfs;
+ public final int gcBefore;
+ public final CompactionParams.TombstoneOption tombstoneOption;
+
+ public AbstractCompactionController(final ColumnFamilyStore cfs, final int
gcBefore, CompactionParams.TombstoneOption tombstoneOption)
+ {
+ assert cfs != null;
+ this.cfs = cfs;
+ this.gcBefore = gcBefore;
+ this.tombstoneOption = tombstoneOption;
+ }
+
+ public abstract boolean compactingRepaired();
+
+ public String getKeyspace()
+ {
+ return cfs.keyspace.getName();
+ }
+
+ public String getColumnFamily()
+ {
+ return cfs.name;
+ }
+
+ public Iterable<UnfilteredRowIterator> shadowSources(DecoratedKey key,
boolean tombstoneOnly)
+ {
+ return null;
+ }
+
+ public abstract LongPredicate getPurgeEvaluator(DecoratedKey key);
+}
diff --git
a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
index 59bba0a..e1b0f32 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionController.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionController.java
@@ -19,26 +19,21 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import java.util.function.LongPredicate;
-import java.util.function.Predicate;
-
-import org.apache.cassandra.config.Config;
-import org.apache.cassandra.db.Memtable;
-import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterables;
import com.google.common.util.concurrent.RateLimiter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.Partition;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.db.*;
import org.apache.cassandra.utils.AlwaysPresentFilter;
import org.apache.cassandra.utils.OverlapIterator;
import org.apache.cassandra.utils.concurrent.Refs;
@@ -48,13 +43,12 @@ import static
org.apache.cassandra.db.lifecycle.SSTableIntervalTree.buildInterva
/**
* Manage compaction options.
*/
-public class CompactionController implements AutoCloseable
+public class CompactionController extends AbstractCompactionController
{
private static final Logger logger =
LoggerFactory.getLogger(CompactionController.class);
private static final String NEVER_PURGE_TOMBSTONES_PROPERTY =
Config.PROPERTY_PREFIX + "never_purge_tombstones";
static final boolean NEVER_PURGE_TOMBSTONES =
Boolean.getBoolean(NEVER_PURGE_TOMBSTONES_PROPERTY);
- public final ColumnFamilyStore cfs;
private final boolean compactingRepaired;
// note that overlapIterator and overlappingSSTables will be null if
NEVER_PURGE_TOMBSTONES is set - this is a
// good thing so that noone starts using them and thinks that if
overlappingSSTables is empty, there
@@ -64,11 +58,8 @@ public class CompactionController implements AutoCloseable
private final Iterable<SSTableReader> compacting;
private final RateLimiter limiter;
private final long minTimestamp;
- final TombstoneOption tombstoneOption;
final Map<SSTableReader, FileDataInput> openDataFiles = new HashMap<>();
- public final int gcBefore;
-
protected CompactionController(ColumnFamilyStore cfs, int maxValue)
{
this(cfs, null, maxValue);
@@ -82,13 +73,10 @@ public class CompactionController implements AutoCloseable
public CompactionController(ColumnFamilyStore cfs, Set<SSTableReader>
compacting, int gcBefore, RateLimiter limiter, TombstoneOption tombstoneOption)
{
- assert cfs != null;
- this.cfs = cfs;
- this.gcBefore = gcBefore;
+ super(cfs, gcBefore, tombstoneOption);
this.compacting = compacting;
this.limiter = limiter;
compactingRepaired = compacting != null &&
compacting.stream().allMatch(SSTableReader::isRepaired);
- this.tombstoneOption = tombstoneOption;
this.minTimestamp = compacting != null && !compacting.isEmpty()
// check needed for test
?
compacting.stream().mapToLong(SSTableReader::getMinTimestamp).min().getAsLong()
: 0;
@@ -246,16 +234,6 @@ public class CompactionController implements AutoCloseable
return getFullyExpiredSSTables(cfStore, compacting, overlapping,
gcBefore, false);
}
- public String getKeyspace()
- {
- return cfs.keyspace.getName();
- }
-
- public String getColumnFamily()
- {
- return cfs.name;
- }
-
/**
* @param key
* @return a predicate for whether tombstones marked for deletion at the
given time for the given partition are
@@ -263,6 +241,7 @@ public class CompactionController implements AutoCloseable
* containing his partition and not participating in the compaction. This
means there isn't any data in those
* sstables that might still need to be suppressed by a tombstone at this
timestamp.
*/
+ @Override
public LongPredicate getPurgeEvaluator(DecoratedKey key)
{
if (NEVER_PURGE_TOMBSTONES || !compactingRepaired() ||
cfs.getNeverPurgeTombstones())
diff --git
a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index 1c56a87..789d1ee 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -57,7 +57,7 @@ public class CompactionIterator extends CompactionInfo.Holder
implements Unfilte
private static final long UNFILTERED_TO_UPDATE_PROGRESS = 100;
private final OperationType type;
- private final CompactionController controller;
+ private final AbstractCompactionController controller;
private final List<ISSTableScanner> scanners;
private final ImmutableSet<SSTableReader> sstables;
private final int nowInSec;
@@ -77,13 +77,13 @@ public class CompactionIterator extends
CompactionInfo.Holder implements Unfilte
private final UnfilteredPartitionIterator compacted;
private final ActiveCompactionsTracker activeCompactions;
- public CompactionIterator(OperationType type, List<ISSTableScanner>
scanners, CompactionController controller, int nowInSec, UUID compactionId)
+ public CompactionIterator(OperationType type, List<ISSTableScanner>
scanners, AbstractCompactionController controller, int nowInSec, UUID
compactionId)
{
this(type, scanners, controller, nowInSec, compactionId,
ActiveCompactionsTracker.NOOP);
}
@SuppressWarnings("resource") // We make sure to close mergedIterator in
close() and CompactionIterator is itself an AutoCloseable
- public CompactionIterator(OperationType type, List<ISSTableScanner>
scanners, CompactionController controller, int nowInSec, UUID compactionId,
ActiveCompactionsTracker activeCompactions)
+ public CompactionIterator(OperationType type, List<ISSTableScanner>
scanners, AbstractCompactionController controller, int nowInSec, UUID
compactionId, ActiveCompactionsTracker activeCompactions)
{
this.controller = controller;
this.type = type;
@@ -259,14 +259,14 @@ public class CompactionIterator extends
CompactionInfo.Holder implements Unfilte
private class Purger extends PurgeFunction
{
- private final CompactionController controller;
+ private final AbstractCompactionController controller;
private DecoratedKey currentKey;
private LongPredicate purgeEvaluator;
private long compactedUnfiltered;
- private Purger(CompactionController controller, int nowInSec)
+ private Purger(AbstractCompactionController controller, int nowInSec)
{
super(nowInSec, controller.gcBefore,
controller.compactingRepaired() ? Integer.MAX_VALUE : Integer.MIN_VALUE,
controller.cfs.getCompactionStrategyManager().onlyPurgeRepairedTombstones(),
@@ -510,10 +510,10 @@ public class CompactionIterator extends
CompactionInfo.Holder implements Unfilte
*/
private static class GarbageSkipper extends
Transformation<UnfilteredRowIterator>
{
- final CompactionController controller;
+ final AbstractCompactionController controller;
final boolean cellLevelGC;
- private GarbageSkipper(CompactionController controller)
+ private GarbageSkipper(AbstractCompactionController controller)
{
this.controller = controller;
cellLevelGC = controller.tombstoneOption == TombstoneOption.CELL;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
