diff --git a/CHANGES.txt b/CHANGES.txt index 04e62e29b0..deb3623c50 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 7.0 + * Rebuild the Bloom filter for CQLSSTableWriter-produced SSTables so it is sized for the real partition count (CASSANDRA-21423) * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975) * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394) * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java index 2487e66fd2..97b1974c06 100644 --- a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java @@ -25,11 +25,14 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.Objects; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.stream.Stream; +import com.google.common.base.Preconditions; + import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.RegularAndStaticColumns; @@ -40,9 +43,13 @@ import org.apache.cassandra.db.rows.EncodingStats; import org.apache.cassandra.index.Index; import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.sstable.format.SSTableReaderLoadingBuilder; +import org.apache.cassandra.io.sstable.format.SSTableReaderWithFilter; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.schema.TableMetadataRef; import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.utils.IFilter; /** * Base class for the sstable writers used by CQLSSTableWriter. @@ -115,12 +122,70 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable sstableProducedListener.accept(sstables); } - protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException + /** + * Rebuilds the Bloom filter of the sstable just produced by {@code txnWriter} and overwrites its {@code Filter.db}. + *

+ * The underlying SSTable writer sizes the filter from a hardcoded partition count of {@code 0} (this writer + * cannot know the count up front), which yields a negligibly small, saturated filter (see CASSANDRA-21423). Once the + * sstable is fully written its true partition count is known, so we rebuild the filter from the on-disk index here. + *

+ * If the sstables were opened (see {@link #shouldOpenSSTables()}), the in-memory readers still hold the original + * (broken) filter, so each is replaced with a clone carrying the rebuilt filter and the original is released. + * + * @param txnWriter the writer that produced the sstable + * @param produced the readers returned by {@code finish} (may contain {@code null} entries when sstables were not opened) + * @return the readers to hand to the produced-sstable listener, with any opened readers carrying the rebuilt filter + */ + protected Collection rebuildBloomFilter(SSTableTxnWriter txnWriter, Collection produced) throws IOException + { + Descriptor descriptor = SSTable.tryDescriptorFromFile(new File(txnWriter.getFilename())); + if (descriptor == null) + return produced; + + TableMetrics metrics = owner != null ? owner.getMetrics() : null; + SSTableReaderLoadingBuilder loadingBuilder = descriptor.getFormat() + .getReaderFactory() + .loadingBuilder(descriptor, metadata, null); + IFilter filter = loadingBuilder.rebuildFilter(metrics); + if (filter == null || produced == null) + return produced; + + try + { + List result = new ArrayList<>(produced.size()); + for (SSTableReader reader : produced) + { + if (reader instanceof SSTableReaderWithFilter) + { + // We rebuilt the filter for exactly one descriptor (derived from txnWriter.getFilename()); these + // writers produce a single sstable per finish(). Guard against ever applying it to a reader for a + // different sstable, which would install the wrong filter. + Preconditions.checkState(descriptor.equals(reader.descriptor), + "Rebuilt bloom filter for %s but produced reader is for %s", + descriptor, reader.descriptor); + result.add(((SSTableReaderWithFilter) reader).cloneAndReplace(filter.sharedCopy())); + reader.selfRef().release(); + } + else + { + // sstable was not opened (null) - the on-disk filter is already corrected, nothing to swap + result.add(reader); + } + } + return result; + } + finally + { + filter.close(); + } + } + + protected SSTableTxnWriter createWriter(SSTable.Owner owner, long keyCount) throws IOException { SerializationHeader header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS); if (makeRangeAware) - return SSTableTxnWriter.createRangeAware(metadata, 0, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, format, header); + return SSTableTxnWriter.createRangeAware(metadata, keyCount, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, format, header); SSTable.Owner effectiveOwner; @@ -138,7 +203,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable return SSTableTxnWriter.create(metadata, createDescriptor(directory, metadata.keyspace, metadata.name, format), - 0, + keyCount, ActiveRepairService.UNREPAIRED_SSTABLE, ActiveRepairService.NO_PENDING_REPAIR, false, @@ -147,6 +212,11 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable effectiveOwner); } + protected SSTableTxnWriter createWriter(SSTable.Owner owner) throws IOException + { + return createWriter(owner, 0); + } + private static Descriptor createDescriptor(File directory, final String keyspace, final String columnFamily, final SSTableFormat fmt) throws IOException { SSTableId nextGen = getNextId(directory, columnFamily); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java index d1c6a5fbc0..b3fd640499 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java @@ -225,7 +225,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter if (b == SENTINEL) return; - try (SSTableTxnWriter writer = createWriter(null)) + try (SSTableTxnWriter writer = createWriter(null, b.size())) { for (Map.Entry entry : b.entrySet()) writer.append(entry.getValue().build().unfilteredIterator()); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java index 7b5fb8a89e..3aa44511a7 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableSimpleWriter.java @@ -161,7 +161,7 @@ class SSTableSimpleWriter extends AbstractSSTableSimpleWriter return; Collection finished = writer.finish(shouldOpenSSTables()); - notifySSTableProduced(finished); + notifySSTableProduced(rebuildBloomFilter(writer, finished)); } catch (Throwable t) { diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReaderLoadingBuilder.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReaderLoadingBuilder.java index 056b4de12a..44a1056e0f 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReaderLoadingBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReaderLoadingBuilder.java @@ -35,12 +35,16 @@ import org.apache.cassandra.io.sstable.IOOptions; import org.apache.cassandra.io.sstable.KeyReader; import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; +import org.apache.cassandra.io.sstable.metadata.MetadataType; import org.apache.cassandra.io.sstable.metadata.ValidationMetadata; import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.TableMetadataRef; import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.FilterFactory; +import org.apache.cassandra.utils.FilterKeys; +import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.JVMStabilityInspector; import static com.google.common.base.Preconditions.checkArgument; @@ -115,6 +119,55 @@ public abstract class SSTableReaderLoadingBuilder + * This is used to correct filters emitted by writers that cannot estimate the partition count up front - notably + * {@link org.apache.cassandra.io.sstable.CQLSSTableWriter}, which sizes its filter from a hardcoded count of {@code 0} + * and therefore produces a negligibly small, saturated filter (see CASSANDRA-21423). + * + * @param tableMetrics metrics to associate with the transient key reader, may be {@code null} + * @return the rebuilt filter, of which the caller takes ownership and must close, or {@code null} if the table has + * Bloom filtering disabled ({@code bloom_filter_fp_chance == 1.0}) and therefore has no filter component + * @throws IOException if the index cannot be read or the filter cannot be saved + */ + public IFilter rebuildFilter(TableMetrics tableMetrics) throws IOException + { + TableMetadata metadata = tableMetadataRef.getLocal(); + double fpChance = metadata.params.bloomFilterFpChance; + if (!FilterComponent.shouldUseBloomFilter(fpChance)) + return null; + + StatsComponent statsComponent = StatsComponent.load(descriptor, MetadataType.STATS); + long numKeys = statsComponent.statsMetadata().totalRows; + + IFilter bf = null; + try (KeyReader keyReader = buildKeyReader(tableMetrics)) + { + bf = FilterFactory.getFilter(numKeys, fpChance); + FilterKeys.Reusable filterKey = FilterKeys.reusable(); + while (!keyReader.isExhausted()) + { + bf.add(filterKey.reset(keyReader.key())); + keyReader.advance(); + } + FilterComponent.save(bf, descriptor, true); + return bf; + } + catch (IOException | RuntimeException | Error ex) + { + if (bf != null) + bf.close(); + // A failure during the index walk (before save) would otherwise leave the original broken Filter.db in + // place. Remove it so the sstable has no filter rather than an invalid one - matching the deleteOnFailure + // path in FilterComponent.save (uses the same fileFor(Components.FILTER)). + descriptor.fileFor(Components.FILTER).deleteIfExists(); + throw ex; + } + } + protected abstract void openComponents(B builder, SSTable.Owner owner, boolean validate, boolean online) throws IOException; /** diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigSSTableReaderLoadingBuilder.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigSSTableReaderLoadingBuilder.java index 24b53ae94e..97b53263f4 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigSSTableReaderLoadingBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigSSTableReaderLoadingBuilder.java @@ -50,6 +50,7 @@ import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.CacheService; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FilterFactory; +import org.apache.cassandra.utils.FilterKeys; import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.Throwables; @@ -210,22 +211,23 @@ public class BigSSTableReaderLoadingBuilder extends SortedTableReaderLoadingBuil if (rebuildFilter) bf = FilterFactory.getFilter(estimatedRowsNumber, tableMetadata.params.bloomFilterFpChance); + FilterKeys.Reusable filterKey = FilterKeys.reusable(); try (IndexSummaryBuilder summaryBuilder = !rebuildSummary ? null : new IndexSummaryBuilder(estimatedRowsNumber, tableMetadata.params.minIndexInterval, Downsampling.BASE_SAMPLING_LEVEL)) { while (!keyReader.isExhausted()) { - key = tableMetadata.partitioner.decorateKey(keyReader.key()); if (rebuildSummary) { + key = tableMetadata.partitioner.decorateKey(keyReader.key()); if (first == null) first = key; summaryBuilder.maybeAddEntry(key, keyReader.keyPositionForSecondaryIndex()); } if (rebuildFilter) - bf.add(key); + bf.add(filterKey.reset(keyReader.key())); keyReader.advance(); } diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java index 6a2c488289..3c9985ee7f 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReaderLoadingBuilder.java @@ -24,7 +24,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.db.ColumnFamilyStore; -import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.compression.CompressionDictionaryManager; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.io.compress.CompressionMetadata; @@ -43,6 +42,7 @@ import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.FilterFactory; +import org.apache.cassandra.utils.FilterKeys; import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.Throwables; @@ -167,11 +167,10 @@ public class BtiTableReaderLoadingBuilder extends SortedTableReaderLoadingBuilde TableMetadata tableMetadata = tableMetadataRef.getLocal(); bf = FilterFactory.getFilter(statsMetadata.totalRows, tableMetadata.params.bloomFilterFpChance); + FilterKeys.Reusable filterKey = FilterKeys.reusable(); while (!keyReader.isExhausted()) { - DecoratedKey key = tableMetadata.partitioner.decorateKey(keyReader.key()); - bf.add(key); - + bf.add(filterKey.reset(keyReader.key())); keyReader.advance(); } } diff --git a/src/java/org/apache/cassandra/utils/FilterKeys.java b/src/java/org/apache/cassandra/utils/FilterKeys.java new file mode 100644 index 0000000000..54f2b8e030 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/FilterKeys.java @@ -0,0 +1,58 @@ +/* + * 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.utils; + +import java.nio.ByteBuffer; + +/** + * Helpers for {@link IFilter.FilterKey} construction without allocating a {@link org.apache.cassandra.db.DecoratedKey} + * per key. Bloom filters hash the raw partition key bytes only; decorating keys to obtain a token is unnecessary when + * bulk-populating a filter from on-disk index keys. + */ +public final class FilterKeys +{ + private FilterKeys() + { + } + + public static Reusable reusable() + { + return new Reusable(); + } + + /** + * A zero-allocation {@link IFilter.FilterKey} that can be repointed at successive {@link ByteBuffer} keys. + * Not thread-safe. + */ + public static final class Reusable implements IFilter.FilterKey + { + private ByteBuffer key; + + public Reusable reset(ByteBuffer key) + { + this.key = key; + return this; + } + + @Override + public void filterHash(long[] dest) + { + MurmurHash.hash3_x64_128(key, key.position(), key.remaining(), 0, dest); + } + } +} diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java index a6b4cc1b87..8a36484cf7 100644 --- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java @@ -203,6 +203,109 @@ public abstract class CQLSSTableWriterTest } } + @Test + public void testBloomFilterRebuiltBig() throws Exception + { + assertBloomFilterRebuilt(BigFormat.getInstance()); + } + + @Test + public void testBloomFilterRebuiltBti() throws Exception + { + assertBloomFilterRebuilt(new BtiFormat.BtiFormatFactory().getInstance(Collections.emptyMap())); + } + + /** + * Before CASSANDRA-21423 the writer sized the Bloom filter from a hardcoded partition count of {@code 0}, producing + * a negligibly small (~16 byte) saturated filter regardless of the number of keys. The filter must now be rebuilt + * from the on-disk index and sized for the real key count. + */ + private void assertBloomFilterRebuilt(SSTableFormat format) throws Exception + { + String schema = "CREATE TABLE " + qualifiedTable + " (k int PRIMARY KEY, v int)"; + String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)"; + try (CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .withFormat(format) + .using(insert) + .build()) + { + for (int i = 0; i < 10_000; i++) + writer.addRow(i, i); + } + + List filterFiles = filterComponentFiles(); + assertFalse("no Filter.db was produced", filterFiles.isEmpty()); + for (Path filterFile : filterFiles) + assertThat(Files.size(filterFile)).isGreaterThan(1024L); + } + + @Test + public void testBloomFilterRebuildSkippedWhenDisabled() throws Exception + { + String schema = "CREATE TABLE " + qualifiedTable + " (k int PRIMARY KEY, v int) " + + "WITH bloom_filter_fp_chance = 1.0"; + String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)"; + try (CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .using(insert) + .build()) + { + for (int i = 0; i < 1_000; i++) + writer.addRow(i, i); + } + + // With Bloom filtering disabled the writer emits no Filter.db and the rebuild must be a no-op. + assertThat(filterComponentFiles()).isEmpty(); + } + + /** + * A single writer session can roll over into several sstables when {@code maxSSTableSizeInMiB} is exceeded. Each + * produced sstable is finished (and its filter rebuilt) independently, so verify that every emitted + * Filter.db is correctly sized for its own partition count - not just the first one. + */ + @Test + public void testBloomFilterRebuiltForEachRotatedSSTable() throws Exception + { + String schema = "CREATE TABLE " + qualifiedTable + " (k int PRIMARY KEY, v text)"; + String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)"; + // ~512 bytes of payload per row so that a 1 MiB size cap is crossed many times, forcing several rotations. + StringBuilder payload = new StringBuilder(); + for (int i = 0; i < 512; i++) + payload.append('x'); + String value = payload.toString(); + + try (CQLSSTableWriter writer = CQLSSTableWriter.builder() + .inDirectory(dataDir) + .forTable(schema) + .using(insert) + .withMaxSSTableSizeInMiB(1) + .build()) + { + for (int i = 0; i < 40_000; i++) + writer.addRow(i, value); + } + + List filterFiles = filterComponentFiles(); + assertThat(filterFiles.size()) + .as("expected the writer to roll over into multiple sstables") + .isGreaterThan(1); + for (Path filterFile : filterFiles) + assertThat(Files.size(filterFile)) + .as("Filter.db %s should be sized for its partition count, not the broken ~16 byte filter", filterFile) + .isGreaterThan(1024L); + } + + private List filterComponentFiles() throws IOException + { + try (Stream paths = Files.list(dataDir.toPath())) + { + return paths.filter(p -> p.toString().endsWith("Filter.db")).collect(Collectors.toList()); + } + } + @Test public void testCompressedWriteAndReadBack() throws Exception { diff --git a/test/unit/org/apache/cassandra/utils/BloomFilterTest.java b/test/unit/org/apache/cassandra/utils/BloomFilterTest.java index 51e6c9fb4c..5067ee04c9 100644 --- a/test/unit/org/apache/cassandra/utils/BloomFilterTest.java +++ b/test/unit/org/apache/cassandra/utils/BloomFilterTest.java @@ -230,6 +230,23 @@ public class BloomFilterTest filter2.close(); } + @Test + public void testFilterKeysMatchesDecoratedKey() + { + IPartitioner partitioner = Murmur3Partitioner.instance; + Iterator gen = new KeyGenerator.RandomStringGenerator(new Random().nextInt(), FilterTestHelper.ELEMENTS); + FilterKeys.Reusable filterKey = FilterKeys.reusable(); + long[] expected = new long[2]; + long[] actual = new long[2]; + while (gen.hasNext()) + { + ByteBuffer key = gen.next(); + partitioner.decorateKey(key).filterHash(expected); + filterKey.reset(key).filterHash(actual); + Assert.assertArrayEquals(expected, actual); + } + } + @Test public void testMurmur3FilterHash() {