Merge branch 'cassandra-4.1' into cassandra-5.0

src/java/org/apache/cassandra/hints/HintsDescriptor.java

@@ -66,6 +66,7 @@ final class HintsDescriptor
 {
     private static final Logger logger = LoggerFactory.getLogger(HintsDescriptor.class);
 
+    static final int VERSION_30 = 1;
     static final int VERSION_40 = 2;
     static final int VERSION_50 = 3;
     static final int CURRENT_VERSION = DatabaseDescriptor.getStorageCompatibilityMode().isBefore(5) ? VERSION_40 : VERSION_50;
@@ -230,6 +231,8 @@ final class HintsDescriptor
     {
         switch (hintsVersion)
         {
+            case VERSION_30:
+                return MessagingService.VERSION_30;
             case VERSION_40:
                 return MessagingService.VERSION_40;
             case VERSION_50:

test/data/legacy-commitlog/3.0.13-encrypted/hash.txt

@@ -0,0 +1,5 @@
+#CommitLog upgrade test, version 3.0.13-SNAPSHOT
+#Mon Nov 06 16:28:38 CET 2023
+cells=1404
+hash=-576744414
+cfid=26be1710-7cb9-11ee-9fce-59fe9713701b

test/data/legacy-commitlog/3.0.29-encrypted/hash.txt

@@ -0,0 +1,5 @@
+#CommitLog upgrade test, version 3.0.30-SNAPSHOT
+#Mon Nov 06 15:09:49 CET 2023
+cells=98
+hash=-325801478
+cfid=241bcee0-7cae-11ee-8a02-094014504250

test/data/legacy-commitlog/4.0.11-encrypted/hash.txt

@@ -0,0 +1,5 @@
+#CommitLog upgrade test, version 4.0.11-SNAPSHOT
+#Mon Nov 06 16:44:40 CET 2023
+cells=355
+hash=-918027651
+cfid=657e5210-7cbb-11ee-9858-c9aa1eb53e69

test/data/legacy-hints/3.0.29/09be0db9-3264-4615-8765-3401a89d050f-1699312935941-1.crc32

@@ -0,0 +1 @@
+57d105c2

test/data/legacy-hints/3.0.29/hash.txt

@@ -0,0 +1,7 @@
+#Hints, version 3.0.30-SNAPSHOT
+#Tue Nov 07 00:22:19 CET 2023
+descriptorTimestamp=1699312935941
+cells=3014
+hash=-2121230191
+cfid=531ba420-7cfb-11ee-824a-01f39c3f61a3
+hostId=09be0db9-3264-4615-8765-3401a89d050f

test/data/legacy-hints/4.1.3/088a61e7-e1b5-46a9-a83d-2ce5a636d531-1699365476115-2.crc32

@@ -0,0 +1 @@
+9068a8dd

test/data/legacy-hints/4.1.3/hash.txt

@@ -0,0 +1,7 @@
+#Hints, version 4.1.4-SNAPSHOT
+#Tue Nov 07 14:57:58 CET 2023
+descriptorTimestamp=1699365476115
+cells=3066
+hash=-1205488531
+cfid=a84e1840-7d75-11ee-98b4-9b56646d896a
+hostId=088a61e7-e1b5-46a9-a83d-2ce5a636d531

test/unit/org/apache/cassandra/db/commitlog/CommitLogUpgradeTest.java

@@ -85,6 +85,7 @@ public class CommitLogUpgradeTest
                      .addPartitionKeyColumn("key", AsciiType.instance)
                      .addClusteringColumn("col", AsciiType.instance)
                      .addRegularColumn("val", BytesType.instance)
+                     .addRegularColumn("val0", BytesType.instance)
                      .compression(SchemaLoader.getCompressionParameters())
                      .build();
 
@@ -102,6 +103,27 @@ public class CommitLogUpgradeTest
         Assert.assertEquals("JVM killed", shouldBeKilled, killerForTests.wasKilled());
     }
 
+    // 30 matches version in MessagingService, 3.0.13 is the latest patch release after 3.0.0 but before 3.0.14
+    @Test
+    public void test30_encrypted() throws Exception
+    {
+        testRestore(DATA_DIR + "3.0.13-encrypted");
+    }
+
+    // 3014 matches version in MessagingService, 3.0.29 is the latest patch release after 3.0.14
+    @Test
+    public void test3014_encrypted() throws Exception
+    {
+        testRestore(DATA_DIR + "3.0.29-encrypted");
+    }
+
+    // 40 matches version in MessagingService, 4.0.11 is the latest patch release on 4.0
+    @Test
+    public void test40_encrypted() throws Exception
+    {
+        testRestore(DATA_DIR + "4.0.11-encrypted");
+    }
+
     @Test
     public void test34_encrypted() throws Exception
     {

test/unit/org/apache/cassandra/hints/HintsMaker.java

@@ -0,0 +1,257 @@
+/*
+ * 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.hints;
+
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Stream;
+
+import io.airlift.airline.Cli;
+import io.airlift.airline.Command;
+import io.airlift.airline.Option;
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.UpdateBuilder;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.utils.FBUtilities;
+
+public class HintsMaker
+{
+    public static void main(String[] args)
+    {
+        System.exit(new HintsMaker().execute(args));
+    }
+
+    public int execute(String[] args)
+    {
+        Cli.CliBuilder<Runnable> builder = Cli.<Runnable>builder("hintsmaker")
+                                              .withCommands(MakeHint.class)
+                                              .withDefaultCommand(MakeHint.class);
+
+        try
+        {
+            builder.build().parse(args).run();
+            return 0;
+        }
+        catch (Throwable t)
+        {
+            t.printStackTrace(System.err);
+        }
+
+        return 1;
+    }
+
+    @Command(name = "make", description = "make file of hints")
+    public static class MakeHint implements Runnable
+    {
+        private static final String KEYSPACE = "Keyspace1";
+        private static final String TABLE = "Standard1";
+        private static final String CELLNAME = "name";
+
+        private static final String DATA_DIR = "test/data/legacy-hints/";
+        private static final String PROPERTIES_FILE = "hash.txt";
+        private static final String HOST_ID_PROPERTY = "hostId";
+        private static final String CFID_PROPERTY = "cfid";
+        private static final String CELLS_PROPERTY = "cells";
+        private static final String DESCRIPTOR_TIMESTAMP_PROPERTY = "descriptorTimestamp";
+        private static final String HASH_PROPERTY = "hash";
+
+        private static ByteBuffer dataSource;
+
+        @Option(name = "dir")
+        private String dir = DATA_DIR;
+
+        @Option(name = HOST_ID_PROPERTY)
+        private UUID hostId = UUID.randomUUID();
+
+        @Option(name = "maxLength") // 1MB by default
+        private long maxLength = 1024 * 1024;
+
+        @Option(name = "randomSize")
+        private boolean randomSize;
+
+        @Option(name = "cellSize")
+        private int cellSize = 256;
+
+        @Option(name = "numCells")
+        private int numCells = 1;
+
+        @Option(name = DESCRIPTOR_TIMESTAMP_PROPERTY)
+        private long descriptorTimestamp = System.currentTimeMillis();
+
+        public void run()
+        {
+            try
+            {
+                initialize();
+
+                maxLength = maxLength == 0 ? DatabaseDescriptor.getMaxHintsFileSize() : maxLength;
+                File dataDir = new File(dir + FBUtilities.getReleaseVersionString());
+
+                Files.createDirectories(dataDir.toPath());
+
+                System.out.printf("Going to generate hints file into directory %s with max length %s, host id %s and " +
+                                  "hints descriptor timestamp %s", dataDir, maxLength, hostId, descriptorTimestamp);
+
+                HintsDescriptor hintsDescriptor = new HintsDescriptor(hostId, descriptorTimestamp);
+
+                final TableMetadata tableMetadata = Schema.instance.getTableMetadata(KEYSPACE, TABLE);
+
+                final AtomicLong counter = new AtomicLong(0);
+                final AtomicInteger hash = new AtomicInteger(0);
+                final AtomicInteger cells = new AtomicInteger(0);
+                final int numCells = this.numCells;
+                final AtomicInteger dataSize = new AtomicInteger(0);
+
+                Iterator<Mutation> mutationIterator = Stream.generate(() -> {
+                    ThreadLocalRandom current = ThreadLocalRandom.current();
+                    ByteBuffer key = randomBytes(16, current);
+                    UpdateBuilder builder = UpdateBuilder.create(tableMetadata, Util.dk(key));
+
+                    for (int i = 0; i < numCells; i++)
+                    {
+                        int sz = randomSize ? current.nextInt(cellSize) : cellSize;
+                        ByteBuffer bytes = randomBytes(sz, current);
+                        builder.newRow(CELLNAME + i).add("val", bytes);
+                        hash.set(hash(hash.get(), bytes));
+                        cells.incrementAndGet();
+                        dataSize.addAndGet(sz);
+                    }
+
+                    counter.incrementAndGet();
+
+                    return (Mutation) builder.makeMutation();
+                }).iterator();
+
+                makeHintFile(dataDir, hintsDescriptor, mutationIterator);
+
+                Properties prop = new Properties();
+                prop.setProperty(HOST_ID_PROPERTY, hostId.toString());
+                prop.setProperty(DESCRIPTOR_TIMESTAMP_PROPERTY, Long.toString(descriptorTimestamp));
+                prop.setProperty(CFID_PROPERTY, Schema.instance.getTableMetadata(KEYSPACE, TABLE).id.toString());
+                prop.setProperty(CELLS_PROPERTY, Integer.toString(cells.get()));
+                prop.setProperty(HASH_PROPERTY, Integer.toString(hash.get()));
+                prop.store(new FileOutputStream(new File(dataDir, PROPERTIES_FILE).toJavaIOFile()),
+                           "Hints, version " + FBUtilities.getReleaseVersionString());
+
+                System.out.println("Done");
+            }
+            catch (Throwable t)
+            {
+                throw new RuntimeException(t);
+            }
+        }
+
+        private void makeHintFile(File dir,
+                                  HintsDescriptor descriptor,
+                                  Iterator<Mutation> mutationIterator)
+        {
+            ByteBuffer buffer = ByteBuffer.allocateDirect(256 * 1024);
+
+            try (HintsWriter writer = HintsWriter.create(dir, descriptor))
+            {
+                try (HintsWriter.Session session = writer.newSession(buffer))
+                {
+                    while (session.position() < maxLength && mutationIterator.hasNext())
+                    {
+                        Hint hint = Hint.create(mutationIterator.next(), System.currentTimeMillis());
+                        session.append(hint);
+                    }
+
+                    System.out.println("Generating finished");
+                }
+            }
+            catch (IOException ex)
+            {
+                throw new FSWriteError(ex, descriptor.fileName());
+            }
+            finally
+            {
+                FileUtils.clean(buffer);
+            }
+        }
+
+        private static int hash(int hash, ByteBuffer bytes)
+        {
+            int shift = 0;
+            for (int i = 0; i < bytes.limit(); i++)
+            {
+                hash += (bytes.get(i) & 0xFF) << shift;
+                shift = (shift + 8) & 0x1F;
+            }
+            return hash;
+        }
+
+        private void initialize() throws Exception
+        {
+            try (FileInputStream fis = new FileInputStream("CHANGES.txt");
+                 FileChannel fileChannel = fis.getChannel())
+            {
+                dataSource = ByteBuffer.allocateDirect((int) fileChannel.size());
+                while (dataSource.hasRemaining())
+                {
+                    fileChannel.read(dataSource);
+                }
+                dataSource.flip();
+            }
+
+            SchemaLoader.loadSchema();
+
+            TableMetadata metadata = TableMetadata.builder(KEYSPACE, TABLE)
+                                                  .addPartitionKeyColumn("key", AsciiType.instance)
+                                                  .addClusteringColumn("col", AsciiType.instance)
+                                                  .addRegularColumn("val", BytesType.instance)
+                                                  .addRegularColumn("val0", BytesType.instance)
+                                                  .compression(SchemaLoader.getCompressionParameters())
+                                                  .build();
+
+            SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(1), metadata);
+        }
+
+        private static ByteBuffer randomBytes(int quantity, ThreadLocalRandom tlr)
+        {
+            ByteBuffer slice = ByteBuffer.allocate(quantity);
+            ByteBuffer source = dataSource.duplicate();
+            source.position(tlr.nextInt(source.capacity() - quantity));
+            source.limit(source.position() + quantity);
+            slice.put(source);
+            slice.flip();
+            return slice;
+        }
+    }
+}

test/unit/org/apache/cassandra/hints/HintsUpgradeTest.java

@@ -0,0 +1,197 @@
+/*
+ * 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.hints;
+
+import java.nio.ByteBuffer;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.function.Consumer;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileInputStreamPlus;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.SchemaTestUtil;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.schema.TableMetadata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class HintsUpgradeTest
+{
+    static
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    private static final String KEYSPACE = "Keyspace1";
+    private static final String TABLE = "Standard1";
+    private static final String CELLNAME = "name";
+
+    private static final String DATA_DIR = "test/data/legacy-hints/";
+    private static final String PROPERTIES_FILE = "hash.txt";
+    private static final String HOST_ID_PROPERTY = "hostId";
+    private static final String CFID_PROPERTY = "cfid";
+    private static final String CELLS_PROPERTY = "cells";
+    private static final String DESCRIPTOR_TIMESTAMP_PROPERTY = "descriptorTimestamp";
+    private static final String HASH_PROPERTY = "hash";
+
+    static TableMetadata.Builder metadataBuilder = TableMetadata.builder(KEYSPACE, TABLE)
+                                                                .addPartitionKeyColumn("key", AsciiType.instance)
+                                                                .addClusteringColumn("col", AsciiType.instance)
+                                                                .addRegularColumn("val", BytesType.instance)
+                                                                .addRegularColumn("val0", BytesType.instance)
+                                                                .compression(SchemaLoader.getCompressionParameters());
+
+    @BeforeClass
+    public static void initialize()
+    {
+        SchemaLoader.loadSchema();
+        SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(1));
+    }
+
+    @After
+    public void teardown()
+    {
+        SchemaTestUtil.announceTableDrop(KEYSPACE, TABLE);
+    }
+
+    private static class HintProperties
+    {
+        private UUID hostId;
+        private String tableId;
+        private int cells;
+        private long descriptorTimestamp;
+        private int hash;
+        private File dir;
+    }
+
+    private HintProperties init(String version) throws Exception
+    {
+        HintProperties properties = loadHintProperties(DATA_DIR + version);
+        SchemaTestUtil.announceNewTable(metadataBuilder.id(TableId.fromString(properties.tableId)).build());
+        return properties;
+    }
+
+    @Test // version 1 of hints
+    public void test30() throws Exception
+    {
+        HintProperties properties = init("3.0.29");
+        readHints(properties);
+    }
+
+    @Test // version 2 of hints
+    public void test41() throws Exception
+    {
+        HintProperties properties = init("4.1.3");
+        readHints(properties);
+    }
+
+    private HintProperties loadHintProperties(String dir) throws Exception
+    {
+        Properties prop = new Properties();
+        prop.load(new FileInputStreamPlus(new File(dir + File.pathSeparator() + PROPERTIES_FILE)));
+
+        HintProperties hintProperties = new HintProperties();
+
+        hintProperties.hostId = UUID.fromString(prop.getProperty(HOST_ID_PROPERTY));
+        hintProperties.tableId = prop.getProperty(CFID_PROPERTY);
+        hintProperties.cells = Integer.parseInt(prop.getProperty(CELLS_PROPERTY));
+        hintProperties.descriptorTimestamp = Long.parseLong(prop.getProperty(DESCRIPTOR_TIMESTAMP_PROPERTY));
+        hintProperties.hash = Integer.parseInt(prop.getProperty(HASH_PROPERTY));
+        hintProperties.dir = new File(dir);
+
+        return hintProperties;
+    }
+
+    private void readHints(HintProperties hintProperties)
+    {
+        HintsCatalog catalog = HintsCatalog.load(hintProperties.dir, ImmutableMap.of());
+        assertTrue(catalog.hasFiles());
+
+        HintsStore store = catalog.getNullable(hintProperties.hostId);
+        assertNotNull(store);
+        assertEquals(hintProperties.hostId, store.hostId);
+
+        HintsDescriptor descriptor = store.poll();
+        assertEquals(hintProperties.descriptorTimestamp, descriptor.timestamp);
+
+        Hasher hasher = new Hasher();
+
+        try (HintsReader hintsReader = HintsReader.open(new File(hintProperties.dir, descriptor.fileName())))
+        {
+            for (HintsReader.Page page : hintsReader)
+                page.hintsIterator().forEachRemaining(hint -> hasher.accept(hint.mutation));
+        }
+
+        assertEquals(hintProperties.hash, hasher.hash);
+        assertEquals(hintProperties.cells, hasher.cells);
+    }
+
+    private static class Hasher implements Consumer<Mutation>
+    {
+        int hash = 0;
+        int cells = 0;
+
+        @Override
+        public void accept(Mutation mutation)
+        {
+            for (PartitionUpdate update : mutation.getPartitionUpdates())
+            {
+                for (Row row : update)
+                {
+                    if (row.clustering().size() > 0 &&
+                        AsciiType.instance.compose(row.clustering().bufferAt(0)).startsWith(CELLNAME))
+                    {
+                        for (Cell<?> cell : row.cells())
+                        {
+                            hash = hash(hash, cell.buffer());
+                            ++cells;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    private static int hash(int hash, ByteBuffer bytes)
+    {
+        int shift = 0;
+        for (int i = 0; i < bytes.limit(); i++)
+        {
+            hash += (bytes.get(i) & 0xFF) << shift;
+            shift = (shift + 8) & 0x1F;
+        }
+        return hash;
+    }
+}