natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Do not release new SSTables in offline transactions 

patch by Aleksandr Sorokoumov; reviewed by Andrés de la Peña and Branimir Lambov for CASSANDRA-16975
This commit is contained in:
Aleksandr Sorokoumov 2021-09-30 13:01:02 +01:00 committed by Andrés de la Peña
parent 660dace43c
commit 3e6faca572
3 changed files with 100 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0.26:
* Do not release new SSTables in offline transactions (CASSANDRA-16975)
* ArrayIndexOutOfBoundsException in FunctionResource#fromName (CASSANDRA-16977, CASSANDRA-16995)
* CVE-2015-0886 Security vulnerability in jbcrypt is addressed (CASSANDRA-9384)
* Avoid useless SSTable reads during single partition queries (CASSANDRA-16944)

20
src/java/org/apache/cassandra/db/compaction/CompactionTask.java
View File

@ -228,18 +228,14 @@ public class CompactionTask extends AbstractCompactionTask
newSSTableNames.append(reader.descriptor.baseFilename()).append(",");
if (offline)
{
Refs.release(Refs.selfRefs(newSStables));
}
else
{
double mbps = dTime > 0 ? (double) endsize / (1024 * 1024) / ((double) dTime / 1000) : 0;
Summary mergeSummary = updateCompactionHistory(cfs.keyspace.getName(), cfs.getColumnFamilyName(), mergedRowCounts, startsize, endsize);
logger.debug(String.format("Compacted (%s) %d sstables to [%s] to level=%d. %,d bytes to %,d (~%d%% of original) in %,dms = %fMB/s. %,d total partitions merged to %,d. Partition merge counts were {%s}",
taskId, transaction.originals().size(), newSSTableNames.toString(), getLevel(), startsize, endsize, (int) (ratio * 100), dTime, mbps, mergeSummary.totalSourceRows, totalKeysWritten, mergeSummary.partitionMerge));
logger.trace(String.format("CF Total Bytes Compacted: %,d", CompactionTask.addToTotalBytesCompacted(endsize)));
logger.trace("Actual #keys: {}, Estimated #keys:{}, Err%: {}", totalKeysWritten, estimatedKeys, ((double)(totalKeysWritten - estimatedKeys)/totalKeysWritten));
}
return;
double mbps = dTime > 0 ? (double) endsize / (1024 * 1024) / ((double) dTime / 1000) : 0;
Summary mergeSummary = updateCompactionHistory(cfs.keyspace.getName(), cfs.getColumnFamilyName(), mergedRowCounts, startsize, endsize);
logger.debug(String.format("Compacted (%s) %d sstables to [%s] to level=%d. %,d bytes to %,d (~%d%% of original) in %,dms = %fMB/s. %,d total partitions merged to %,d. Partition merge counts were {%s}",
taskId, transaction.originals().size(), newSSTableNames.toString(), getLevel(), startsize, endsize, (int) (ratio * 100), dTime, mbps, mergeSummary.totalSourceRows, totalKeysWritten, mergeSummary.partitionMerge));
logger.trace(String.format("CF Total Bytes Compacted: %,d", CompactionTask.addToTotalBytesCompacted(endsize)));
logger.trace("Actual #keys: {}, Estimated #keys:{}, Err%: {}", totalKeysWritten, estimatedKeys, ((double)(totalKeysWritten - estimatedKeys)/totalKeysWritten));
}
}

91
test/unit/org/apache/cassandra/db/compaction/CompactionTaskTest.java Normal file
View File

@ -0,0 +1,91 @@
/*
* 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.compaction;
import java.util.Set;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.KeyspaceParams;
public class CompactionTaskTest
{
private static ColumnFamilyStore cfs;
@BeforeClass
public static void setUpClass() throws Exception
{
SchemaLoader.prepareServer();
CFMetaData table = CFMetaData.compile("CREATE TABLE tbl (k INT PRIMARY KEY, v INT)", "ks");
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1), table);
cfs = Schema.instance.getColumnFamilyStoreInstance(table.cfId);
}
@Before
public void setUp() throws Exception
{
cfs.getCompactionStrategyManager().enable();
cfs.truncateBlocking();
}
@Test
public void testOfflineCompaction()
{
cfs.getCompactionStrategyManager().disable();
QueryProcessor.executeInternal("INSERT INTO ks.tbl (k, v) VALUES (1, 1);");
cfs.forceBlockingFlush();
QueryProcessor.executeInternal("INSERT INTO ks.tbl (k, v) VALUES (2, 2);");
cfs.forceBlockingFlush();
QueryProcessor.executeInternal("INSERT INTO ks.tbl (k, v) VALUES (3, 3);");
cfs.forceBlockingFlush();
QueryProcessor.executeInternal("INSERT INTO ks.tbl (k, v) VALUES (4, 4);");
cfs.forceBlockingFlush();
Set<SSTableReader> sstables = cfs.getLiveSSTables();
Assert.assertEquals(4, sstables.size());
try (LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.COMPACTION, sstables))
{
Assert.assertEquals(4, txn.tracker.getView().liveSSTables().size());
CompactionTask task = new CompactionTask(cfs, txn, 1000);
task.execute(null);
// Check that new SSTable was not released
Assert.assertEquals(1, txn.tracker.getView().liveSSTables().size());
SSTableReader newSSTable = txn.tracker.getView().liveSSTables().iterator().next();
Assert.assertNotNull(newSSTable.tryRef());
}
finally
{
// SSTables were compacted offline; CFS didn't notice that, so we have to remove them manually
cfs.getTracker().removeUnsafe(sstables);
}
}
}