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

Merge branch 'cassandra-4.1' into cassandra-5.0

This commit is contained in:
Stefan Miklosovic 2024-05-02 10:42:36 +02:00
parent 1b1f88e38a 85039aa8c5
commit d1101eeccc
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
4 changed files with 111 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -41,6 +41,7 @@ Merged from 4.1:
* Memoize Cassandra verion and add a backoff interval for failed schema pulls (CASSANDRA-18902)
* Fix StackOverflowError on ALTER after many previous schema changes (CASSANDRA-19166)
Merged from 4.0:
* Make nodetool import congruent with the documentation by not relying on the folder structure of the imported SSTable files (CASSANDRA-19401)
* IR may leak SSTables with pending repair when coming from streaming (CASSANDRA-19182)
* Streaming exception race creates corrupt transaction log files that prevent restart (CASSANDRA-18736)
* Fix CQL tojson timestamp output on negative timestamp values before Gregorian calendar reform in 1582 (CASSANDRA-19566)

55
src/java/org/apache/cassandra/db/Directories.java
View File

@ -1034,7 +1034,21 @@ public class Directories
public Map<Descriptor, Set<Component>> list()
{
filter();
return list(false);
}
/**
* This method is used upon SSTable importing (nodetool import) as there is no strict requirement to
* place SSTables in a directory structure where name of a directory equals to table name
* and parent of such directory equals to keyspace name. For such cases, we want to include such SSTables too,
* rendering the parameter to be set to true.
*
* @param includeForeignTables whether descriptors not matching metadata of this lister should be included
* @return found descriptors and related set of components
*/
public Map<Descriptor, Set<Component>> list(boolean includeForeignTables)
{
filter(includeForeignTables);
return ImmutableMap.copyOf(components);
}
@ -1052,7 +1066,12 @@ public class Directories
public List<File> listFiles()
{
filter();
return listFiles(false);
}
public List<File> listFiles(boolean includeForeignTables)
{
filter(includeForeignTables);
List<File> l = new ArrayList<>(nbFiles);
for (Map.Entry<Descriptor, Set<Component>> entry : components.entrySet())
{
@ -1064,7 +1083,7 @@ public class Directories
return l;
}
private void filter()
private void filter(boolean includeForeignTables)
{
if (filtered)
return;
@ -1077,24 +1096,24 @@ public class Directories
if (snapshotName != null)
{
Optional<File> maybeSnapshotDir = getSnapshotDirectoryIfExists(location, snapshotName);
maybeSnapshotDir.ifPresent(dir -> LifecycleTransaction.getFiles(dir.toPath(), getFilter(), onTxnErr));
maybeSnapshotDir.ifPresent(dir -> LifecycleTransaction.getFiles(dir.toPath(), getFilter(includeForeignTables), onTxnErr));
continue;
}
if (!onlyBackups)
LifecycleTransaction.getFiles(location.toPath(), getFilter(), onTxnErr);
LifecycleTransaction.getFiles(location.toPath(), getFilter(includeForeignTables), onTxnErr);
if (includeBackups)
{
Optional<File> maybeBackupsDir = getBackupsDirectoryIfExists(location);
maybeBackupsDir.ifPresent(dir -> LifecycleTransaction.getFiles(dir.toPath(), getFilter(), onTxnErr));
maybeBackupsDir.ifPresent(dir -> LifecycleTransaction.getFiles(dir.toPath(), getFilter(includeForeignTables), onTxnErr));
}
}
filtered = true;
}
private BiPredicate<File, FileType> getFilter()
private BiPredicate<File, FileType> getFilter(boolean includeForeignTables)
{
// This function always return false since it adds to the components map
return (file, type) ->
@ -1112,15 +1131,31 @@ public class Directories
if (pair == null)
return false;
Descriptor descriptor = null;
// we are only interested in the SSTable files that belong to the specific ColumnFamily
if (!pair.left.ksname.equals(metadata.keyspace) || !pair.left.cfname.equals(metadata.name))
return false;
{
if (!includeForeignTables)
return false;
Set<Component> previous = components.get(pair.left);
descriptor = new Descriptor(pair.left.version.toString(),
pair.left.directory,
metadata.keyspace,
metadata.name,
pair.left.id,
pair.left.getFormat());
}
else
{
descriptor = pair.left;
}
Set<Component> previous = components.get(descriptor);
if (previous == null)
{
previous = new HashSet<>();
components.put(pair.left, previous);
components.put(descriptor, previous);
}
previous.add(pair.right);
nbFiles++;

4
src/java/org/apache/cassandra/db/SSTableImporter.java
View File

@ -93,7 +93,7 @@ public class SSTableImporter
{
Directories.SSTableLister lister = listerPair.left;
String dir = listerPair.right;
for (Map.Entry<Descriptor, Set<Component>> entry : lister.list().entrySet())
for (Map.Entry<Descriptor, Set<Component>> entry : lister.list(true).entrySet())
{
Descriptor descriptor = entry.getKey();
if (!currentDescriptors.contains(entry.getKey()))
@ -164,7 +164,7 @@ public class SSTableImporter
Set<MovedSSTable> movedSSTables = new HashSet<>();
Set<SSTableReader> newSSTablesPerDirectory = new HashSet<>();
for (Map.Entry<Descriptor, Set<Component>> entry : lister.list().entrySet())
for (Map.Entry<Descriptor, Set<Component>> entry : lister.list(true).entrySet())
{
try
{

71
test/unit/org/apache/cassandra/db/ImportTest.java
View File

@ -31,6 +31,7 @@ import java.util.Set;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
@ -69,6 +70,11 @@ import static org.junit.Assert.assertTrue;
public class ImportTest extends CQLTester
{
@After
public void afterTest()
{
SSTableReader.resetTidying();
}
@Test
public void basicImportByMovingTest() throws Throwable
@ -237,13 +243,10 @@ public class ImportTest extends CQLTester
assertFalse(sstable.isRepaired());
}
private File moveToBackupDir(Set<SSTableReader> sstables) throws IOException
private File moveToBackupDir(Set<SSTableReader> sstables, String keyspace, String table) throws IOException
{
Path temp = Files.createTempDirectory("importtest");
SSTableReader sst = sstables.iterator().next();
String tabledir = sst.descriptor.directory.name();
String ksdir = sst.descriptor.directory.parent().name();
Path backupdir = createDirectories(temp.toString(), ksdir, tabledir);
Path backupdir = createDirectories(temp.toString(), keyspace, table);
for (SSTableReader sstable : sstables)
{
sstable.selfRef().release();
@ -262,6 +265,14 @@ public class ImportTest extends CQLTester
return new File(backupdir);
}
private File moveToBackupDir(Set<SSTableReader> sstables) throws IOException
{
SSTableReader sst = sstables.iterator().next();
String tabledir = sst.descriptor.directory.name();
String ksdir = sst.descriptor.directory.parent().name();
return moveToBackupDir(sstables, ksdir, tabledir);
}
private Path createDirectories(String base, String ... subdirs)
{
File b = new File(base);
@ -425,9 +436,9 @@ public class ImportTest extends CQLTester
TokenMetadata tmd = StorageService.instance.getTokenMetadata();
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 5), InetAddressAndPort.getByName("127.0.0.1"));
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 5), InetAddressAndPort.getByName("127.0.0.2"));
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 5), InetAddressAndPort.getByName("127.0.0.3"));
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 15), InetAddressAndPort.getByName("127.0.0.1"));
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 15), InetAddressAndPort.getByName("127.0.0.2"));
tmd.updateNormalTokens(BootStrapper.getRandomTokens(tmd, 15), InetAddressAndPort.getByName("127.0.0.3"));
File backupdir = moveToBackupDir(sstables);
@ -550,6 +561,7 @@ public class ImportTest extends CQLTester
beforeFirstImport.forEach(s -> s.selfRef().release());
options = SSTableImporter.Options.options(backupdir.toString()).verifySSTables(true).verifyTokens(true).invalidateCaches(true).build();
importer.importNewSSTables(options);
Thread.sleep(2000);
assertEquals(10, CacheService.instance.rowCache.size());
it = CacheService.instance.rowCache.keyIterator();
while (it.hasNext())
@ -721,6 +733,49 @@ public class ImportTest extends CQLTester
}
}
/**
* This test verifies that we successfully import SSTables which are in a directory structure
* where table name and keyspace name (current dir name and parent dir name) do not match the keyspace and
* table arguments on the command line.
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-19401">CASSANDRA-19401</a>
*/
@Test
public void importFromNonMatchingKeyspaceTableDir() throws Throwable
{
String table = "nonmatchingtable";
try
{
schemaChange(String.format("CREATE TABLE %s.%s (id int primary key, d int)", KEYSPACE, table));
for (int i = 0; i < 10; i++)
execute(String.format("INSERT INTO %s.%s (id, d) values (?, ?)", KEYSPACE, table), i, i);
ColumnFamilyStore cfs = getColumnFamilyStore(KEYSPACE, table);
Util.flush(cfs);
Set<SSTableReader> sstables = cfs.getLiveSSTables();
cfs.clearUnsafe();
File backupDir = moveToBackupDir(sstables, "randomdir1", "randomdir2");
assertEquals(0, execute(String.format("SELECT * FROM %s.%s", KEYSPACE, table)).size());
SSTableImporter importer = new SSTableImporter(cfs);
SSTableImporter.Options options = SSTableImporter.Options.options(backupDir.toString()).copyData(true).build();
List<String> failedDirectories = importer.importNewSSTables(options);
assertTrue(failedDirectories.isEmpty());
assertEquals(10, execute(String.format("select * from %s.%s", KEYSPACE, table)).size());
// files are left there as they were just copied
Assert.assertNotEquals(0, countFiles(backupDir));
}
finally
{
execute(String.format("DROP TABLE IF EXISTS %s.%s", KEYSPACE, table));
}
}
@Test
public void mustNotFailOnBuiltSAIIndexesWhenRequiredTest() throws Throwable
{