>From Ritik Raj <[email protected]>: Ritik Raj has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21625?usp=email )
Change subject: [NO ISSUE][MTD] Spike: compress secondary btree indexes ...................................................................... [NO ISSUE][MTD] Spike: compress secondary btree indexes - user model changes: no - storage format changes: no (per-index, opt-in via existing DDL) - interface changes: no SPIKE, not a proposal: the code is marked as such and the scope question below is deliberately left open. Committed so the measurements have something to sit against. Details: - Compression was restricted to the primary index by the original ASTERIXDB-2422, whose message scoped it as "pages of the primary index can be compressed". No blocker was recorded; it was initial scope. - Both secondary btree providers now take the dataset's own scheme instead of hardcoding NoOpCompressorDecompressorFactory. The scheme is already a dataset property, so the existing storage-block-compression DDL option is the toggle and no new knob is added. - Upgrade-safe without migration: compressorDecompressorFactory is persisted per index in the local resource, not resolved from the dataset at open time, so indexes built before this keep NoOp and stay readable. Had it been resolved at open time, every pre-existing secondary index of a compressed dataset would find no look-aside file and land in CompressedFileManager.State.INVALID. - RTree and inverted-index providers are untouched; they never pass a compression factory at all. Measured, space (storage_size, includes the .dic look-aside file): - array index, 1M entries: 33,711,388 -> 14,826,800 (56.0%) - secondary on a string field: 6,998,948 -> 3,369,972 (51.9%) - secondary on a bigint field: 4,999,672 -> 3,194,071 (36.1%) - With one array index present the secondaries save nearly twice what the primary does, and the array index is larger than the primary. Measured, cost (SecondaryIndexCompressionBenchmark, added here): - The whole read-path cost is decompression CPU. Decompression measured in isolation accounts for the full per-page delta; the extra look-aside pin and the staging copy contribute nothing measurable. - Per page LOAD, not per access: +3 us at 6% saved rising to +18 us at 39% saved. Cost rises with the ratio achieved, because a well compressed page is back-reference heavy while a barely compressed one decodes at memcpy speed. - Zero for a cached working set: the query harness measured a 100% buffer cache hit ratio and an identical page read count either way. Compression shrinks the on-disk form, never the logical page count, so there is no cache-occupancy benefit to offset the CPU. Open, and why this is not yet a proposal: - Real buffer cache miss rates for secondary indexes under production workloads are unknown, and they are what converts us-per-page-load into query impact. The two bounds measured here are 0% and 97%. - Decompression cost above 39% saved is unmeasured, so the array index case is bounded only from below. - cloud_storage/dic-merge-misalign covers the look-aside write path for a column primary index only. This change puts every secondary index on that path, which has already produced one cloud misalignment bug (ASTERIXDB-3788), so it wants a secondary-index variant. Tests: - SqlppExecutionTest 3174 run, 1 pre-existing failure (cbo-join/ch2, ASX3077 missing CH2 data). - CloudStorageTest 3036 run, CloudStorageMergeTest 1/1, MetricsExecutionTest 5/5. - The one spike-caused failure, ddl/analyze-dataset-with-indexes, was a latent page-count bug fixed separately in the preceding commit. - SecondaryIndexCompressionBenchmark 4/4. It is named ...Benchmark so surefire's *Test.java include pattern never runs it in CI. Co-Authored-By: Claude Opus 5 <[email protected]> Change-Id: I894eed2fbe262c5d53a077740cf804bda1b30230 --- M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/ArrayBTreeResourceFactoryProvider.java M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/BTreeResourceFactoryProvider.java A hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/SecondaryIndexCompressionBenchmark.java 3 files changed, 488 insertions(+), 12 deletions(-) git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/25/21625/1 diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/ArrayBTreeResourceFactoryProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/ArrayBTreeResourceFactoryProvider.java index a61db6f..519acee 100644 --- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/ArrayBTreeResourceFactoryProvider.java +++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/ArrayBTreeResourceFactoryProvider.java @@ -50,7 +50,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMPageWriteCallbackFactory; import org.apache.hyracks.storage.common.IResourceFactory; import org.apache.hyracks.storage.common.IStorageManager; -import org.apache.hyracks.storage.common.compression.NoOpCompressorDecompressorFactory; public class ArrayBTreeResourceFactoryProvider implements IResourceFactoryProvider { @@ -87,13 +86,13 @@ AsterixVirtualBufferCacheProvider vbcProvider = new AsterixVirtualBufferCacheProvider(dataset.getDatasetId()); - final ICompressorDecompressorFactory compDecompFactory; if (index.isPrimaryIndex()) { throw new CompilationException(ErrorCode.COMPILATION_ILLEGAL_STATE, "Array indexes cannot be " + "primary indexes."); - } else { - compDecompFactory = NoOpCompressorDecompressorFactory.INSTANCE; } + // SPIKE: array indexes are secondary btrees, so they follow the dataset's scheme too. + final ICompressorDecompressorFactory compDecompFactory = + mdProvider.getCompressionManager().getFactory(dataset.getCompressionScheme()); return new LSMBTreeLocalResourceFactory(storageManager, typeTraits, cmpFactories, filterTypeTraits, filterCmpFactories, filterFields, opTrackerFactory, ioOpCallbackFactory, diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/BTreeResourceFactoryProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/BTreeResourceFactoryProvider.java index c27e707..5c358a8 100644 --- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/BTreeResourceFactoryProvider.java +++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/BTreeResourceFactoryProvider.java @@ -57,7 +57,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMPageWriteCallbackFactory; import org.apache.hyracks.storage.common.IResourceFactory; import org.apache.hyracks.storage.common.IStorageManager; -import org.apache.hyracks.storage.common.compression.NoOpCompressorDecompressorFactory; public class BTreeResourceFactoryProvider implements IResourceFactoryProvider { @@ -95,13 +94,12 @@ AsterixVirtualBufferCacheProvider vbcProvider = new AsterixVirtualBufferCacheProvider(dataset.getDatasetId()); - final ICompressorDecompressorFactory compDecompFactory; - if (index.isPrimaryIndex()) { - //Compress only primary index - compDecompFactory = mdProvider.getCompressionManager().getFactory(dataset.getCompressionScheme()); - } else { - compDecompFactory = NoOpCompressorDecompressorFactory.INSTANCE; - } + // SPIKE: compress every btree index of the dataset, not only the primary. The scheme is a + // dataset property, so the existing "storage-block-compression" DDL option is the toggle; + // no new knob. Persisted per index in the local resource, so indexes created before this + // keep NoOp and stay readable. + final ICompressorDecompressorFactory compDecompFactory = + mdProvider.getCompressionManager().getFactory(dataset.getCompressionScheme()); boolean isSecondaryNoIncrementalMaintenance = index.getIndexType() == DatasetConfig.IndexType.SAMPLE; diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/SecondaryIndexCompressionBenchmark.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/SecondaryIndexCompressionBenchmark.java new file mode 100644 index 0000000..b5223b7 --- /dev/null +++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test/src/test/java/org/apache/hyracks/storage/common/SecondaryIndexCompressionBenchmark.java @@ -0,0 +1,479 @@ +/* + * 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.hyracks.storage.common; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +import org.apache.hyracks.api.compression.ICompressorDecompressor; +import org.apache.hyracks.api.context.IHyracksTaskContext; +import org.apache.hyracks.api.exceptions.HyracksDataException; +import org.apache.hyracks.api.io.FileReference; +import org.apache.hyracks.api.io.IIOManager; +import org.apache.hyracks.storage.common.buffercache.HaltOnFailureCallback; +import org.apache.hyracks.storage.common.buffercache.IBufferCache; +import org.apache.hyracks.storage.common.buffercache.ICachedPage; +import org.apache.hyracks.storage.common.buffercache.IFIFOPageWriter; +import org.apache.hyracks.storage.common.buffercache.NoOpPageWriteCallback; +import org.apache.hyracks.storage.common.buffercache.context.write.DefaultBufferCacheWriteContext; +import org.apache.hyracks.storage.common.compression.SnappyCompressorDecompressorFactory; +import org.apache.hyracks.storage.common.compression.file.CompressedFileReference; +import org.apache.hyracks.storage.common.compression.file.ICompressedPageWriter; +import org.apache.hyracks.storage.common.file.BufferedFileHandle; +import org.apache.hyracks.test.support.TestStorageManagerComponentHolder; +import org.apache.hyracks.test.support.TestUtils; +import org.apache.hyracks.util.annotations.AiProvenance; +import org.junit.AfterClass; +import org.junit.Test; + +/** + * Measures what block compression costs on the <em>read</em> path of an index, to decide whether + * secondary indexes should be compressed like the primary already is. + * <p> + * The class name deliberately ends in {@code Benchmark}, not {@code Test}, so surefire's + * {@code **}{@code /*Test.java} include pattern never picks it up in CI. Run it explicitly: + * + * <pre> + * mvn -o -pl hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-common-test test \ + * -Dtest=SecondaryIndexCompressionBenchmark + * </pre> + * + * <h2>Two regimes, because they answer opposite halves of the question</h2> + * In both, the buffer cache is sized to a small fraction of the file, so nearly every pin misses and + * has to reach the file. What differs is where the bytes come from: + * <ol> + * <li><b>Warm OS page cache</b> ({@link #reportRandomPinCost}, {@link #reportSequentialScanCost}) — + * the file was just written, so no physical device read happens on either side. This isolates the + * decompression CPU and grants compression none of its benefit, making it the <b>worst case for + * compression</b>: an overhead that is acceptable here is acceptable anywhere.</li> + * <li><b>Cold device</b> ({@link #reportColdDeviceCost}) — the page cache is evicted first, so a pin + * is a real device read. This is the regime that matters for an index larger than memory, and the + * one where reading fewer bytes can pay for the decompression. Without it the benchmark would only + * ever report costs and never the corresponding benefit.</li> + * </ol> + * Both report <b>microseconds per page access</b>, not just totals, since per-access latency is what + * a secondary-index probe actually pays. + * <p> + * Page content is synthesised to span a range of compressibility rather than to a single guess, + * because the achieved ratio drives both the space win and the decompression cost. The measured + * ratios on real data — 36% for a bigint secondary index, 52% for a string one — fall inside the + * swept range, so the relevant row can be read off directly. + */ +@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_CLI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Spike benchmark for compressing secondary indexes; read-path cost of snappy per page pin") +public class SecondaryIndexCompressionBenchmark { + + private static final int PAGE_SIZE = 32 * 1024; + /** 2048 pages = 64 MiB of logical pages, well beyond the cache below. */ + private static final int FILE_PAGES = 2048; + /** 64 pages = 2 MiB, so a random pin misses ~97% of the time. */ + private static final int CACHE_PAGES = 64; + private static final int MAX_OPEN_FILES = 8; + private static final int RANDOM_PROBES = 20_000; + private static final int REPEATS = 7; + private static final int WARMUP_REPEATS = 2; + private static final int DECOMPRESS_ITERATIONS = 2_000; + + /** + * Fraction of each page filled with incompressible bytes. Sweeping this sweeps the achieved + * compression ratio, which is the axis that actually matters. + */ + private static final double[] INCOMPRESSIBLE_FRACTIONS = { 0.0, 0.10, 0.20, 0.30, 0.50, 0.70, 0.85 }; + + /** 32768 pages = 1 GiB logical: big enough that a cold pin is a real device read. */ + private static final int COLD_FILE_PAGES = 32 * 1024; + private static final int COLD_PROBES = 4_000; + private static final int COLD_PROBE_SEED = 99; + /** + * Calibrated by running the warm sweep and reading the achieved ratio off it: 0.20 lands near the + * 36% saving measured on a real bigint secondary index. An earlier guess of 0.70 achieved only + * 12%, which understated both the cost and the benefit. + */ + private static final double COLD_INCOMPRESSIBLE_FRACTION = 0.20; + /** GiB of memory pressure applied to evict the page cache; the machine has 32 GiB. */ + private static final int BALLAST_GIB = 22; + private static final long BALLAST_TIMEOUT_SECONDS = 300; + + private static final List<String> createdFiles = new ArrayList<>(); + + private final IHyracksTaskContext ctx = TestUtils.create(PAGE_SIZE); + + private static final class Measurement { + private final long dataBytes; + private final long lafBytes; + private final double medianMillis; + + private Measurement(long dataBytes, long lafBytes, double medianMillis) { + this.dataBytes = dataBytes; + this.lafBytes = lafBytes; + this.medianMillis = medianMillis; + } + + private long totalBytes() { + return dataBytes + lafBytes; + } + + private double microsPerAccess(int accesses) { + return medianMillis * 1000.0 / accesses; + } + } + + /** + * Attributes the read-path gap. The warm regimes show snappy costing +2 to +22 us per page access, + * and there are three candidate sources: decompression CPU, the extra buffer-cache pin of the + * look-aside file that {@code CompressedBufferedFileHandle} needs on every read to locate a page, + * and the staging copy through the compressed buffer. + * <p> + * This measures decompression alone -- same page content, same codec, no buffer cache, no file, no + * LAF. Whatever the warm delta shows beyond this number is not CPU. + */ + @Test + public void reportDecompressionCpuOnly() throws Exception { + ICompressorDecompressor compDecomp = new SnappyCompressorDecompressorFactory().createInstance(); + StringBuilder out = new StringBuilder(); + out.append("\n=== decompression CPU in isolation (no buffer cache, no file, no LAF) ===\n"); + out.append(String.format("%-14s %13s %8s %14s%n", "incompressible", "compressed", "saved", "decompress us")); + + ByteBuffer uBuffer = ByteBuffer.allocate(PAGE_SIZE); + ByteBuffer cBuffer = ByteBuffer.allocate(compDecomp.computeCompressedBufferSize(PAGE_SIZE)); + ByteBuffer outBuffer = ByteBuffer.allocate(PAGE_SIZE); + for (double incompressible : INCOMPRESSIBLE_FRACTIONS) { + fillLikeIndexLeaf(uBuffer, 1, incompressible); + uBuffer.position(0).limit(PAGE_SIZE); + cBuffer.clear(); + compDecomp.compress(uBuffer, cBuffer); + int compressedSize = cBuffer.limit(); + + double[] millis = new double[REPEATS]; + for (int repeat = 0; repeat < REPEATS; repeat++) { + long start = System.nanoTime(); + for (int i = 0; i < DECOMPRESS_ITERATIONS; i++) { + cBuffer.position(0).limit(compressedSize); + outBuffer.clear(); + compDecomp.uncompress(cBuffer, outBuffer); + } + millis[repeat] = (System.nanoTime() - start) / 1_000_000.0; + } + double us = medianAfterWarmup(millis) * 1000.0 / DECOMPRESS_ITERATIONS; + out.append(String.format("%13.0f%% %13d %7.1f%% %14.2f%n", incompressible * 100, compressedSize, + (1.0 - (double) compressedSize / PAGE_SIZE) * 100, us)); + } + System.out.println(out); // NOSONAR + } + + @Test + public void reportRandomPinCost() throws Exception { + report("random pin (" + RANDOM_PROBES + " scattered pins)", false); + } + + @Test + public void reportSequentialScanCost() throws Exception { + report("sequential scan (" + FILE_PAGES + " pages in order)", true); + } + + private void report(String label, boolean sequential) throws Exception { + int accesses = sequential ? FILE_PAGES : RANDOM_PROBES; + StringBuilder out = new StringBuilder(); + out.append("\n=== secondary index compression, ").append(label).append(" (OS page cache WARM) ===\n"); + out.append(String.format("%-14s %13s %13s %7s %10s %10s %10s %10s%n", "incompressible", "none bytes", + "snappy bytes", "saved", "none us", "snappy us", "delta us", "overhead")); + + for (double incompressible : INCOMPRESSIBLE_FRACTIONS) { + Measurement plain = measure(false, incompressible, sequential); + Measurement snappy = measure(true, incompressible, sequential); + + double saved = 1.0 - (double) snappy.totalBytes() / plain.totalBytes(); + double plainUs = plain.microsPerAccess(accesses); + double snappyUs = snappy.microsPerAccess(accesses); + out.append(String.format("%13.0f%% %13d %13d %6.1f%% %10.2f %10.2f %10.2f %9.1f%%%n", incompressible * 100, + plain.totalBytes(), snappy.totalBytes(), saved * 100, plainUs, snappyUs, snappyUs - plainUs, + (snappyUs / plainUs - 1.0) * 100)); + } + // A benchmark's whole output is its result, so print it rather than log it. + System.out.println(out); // NOSONAR + } + + /** + * The case the warm sweep cannot reach: the index does not fit in memory, so a pin costs a real + * device read. Here compression trades decompression CPU for fewer bytes off the device, which is + * the trade that decides whether secondary indexes should be compressed. + * <p> + * A file larger than RAM is not reachable on this class of machine (32 GiB RAM against 33 GiB of + * free disk), and {@code purge} needs root. So instead the file stays moderate and the OS page + * cache is evicted by putting the machine under memory pressure — see + * {@link #evictOsPageCache()}. Each side is then read once cold and once warm; the cold/warm gap + * is the evidence that eviction actually happened, and is printed so a run where it did not is + * self-evident rather than silently reported as a cold number. + */ + @Test + public void reportColdDeviceCost() throws Exception { + if (!canEvictOsPageCache()) { + System.out.println("\n=== cold-device: SKIPPED, no python3 to apply memory pressure ===\n"); // NOSONAR + return; + } + StringBuilder out = new StringBuilder(); + out.append("\n=== secondary index compression, cold device (OS page cache EVICTED) ===\n"); + out.append(String.format("file %d MiB logical, buffer cache %d MiB%n", + (long) COLD_FILE_PAGES * PAGE_SIZE / (1024 * 1024), (long) CACHE_PAGES * PAGE_SIZE / (1024 * 1024))); + out.append(String.format("%-12s %-8s %13s %7s %10s %10s %10s %12s%n", "pattern", "scheme", "bytes", "saved", + "cold us", "warm us", "cold/warm", "cold MB/s")); + + // Random probes are latency-bound and sequential scans are bandwidth-bound, and compression + // only helps the second. Reporting one without the other would answer half the question: an + // ordinary secondary index is probed, but the sample index is fully scanned by the CBO. + for (boolean sequential : new boolean[] { false, true }) { + int accesses = sequential ? COLD_FILE_PAGES : COLD_PROBES; + long plainBytes = -1; + for (boolean compressed : new boolean[] { false, true }) { + ColdMeasurement m = measureCold(compressed, sequential); + if (!compressed) { + plainBytes = m.totalBytes; + } + double coldUs = m.coldMillis * 1000.0 / accesses; + double warmUs = m.warmMillis * 1000.0 / accesses; + double coldMbPerSec = (double) accesses * PAGE_SIZE / (1024 * 1024) / (m.coldMillis / 1000.0); + double saved = 1.0 - (double) m.totalBytes / plainBytes; + out.append(String.format("%-12s %-8s %13d %6.1f%% %10.2f %10.2f %10.1fx %12.1f%n", + sequential ? "sequential" : "random", compressed ? "snappy" : "none", m.totalBytes, saved * 100, + coldUs, warmUs, coldUs / warmUs, coldMbPerSec)); + } + } + System.out.println(out); // NOSONAR + } + + private static final class ColdMeasurement { + private final long totalBytes; + private final double coldMillis; + private final double warmMillis; + + private ColdMeasurement(long totalBytes, double coldMillis, double warmMillis) { + this.totalBytes = totalBytes; + this.coldMillis = coldMillis; + this.warmMillis = warmMillis; + } + } + + private ColdMeasurement measureCold(boolean compressed, boolean sequential) throws Exception { + TestStorageManagerComponentHolder.init(PAGE_SIZE, CACHE_PAGES, MAX_OPEN_FILES); + IIOManager ioManager = TestStorageManagerComponentHolder.getIOManager(); + IBufferCache bufferCache = + TestStorageManagerComponentHolder.getBufferCache(ctx.getJobletContext().getServiceContext()); + FileReference fileRef = newFileReference(ioManager, compressed, COLD_INCOMPRESSIBLE_FRACTION, sequential); + try { + int fileId = bufferCache.createFile(fileRef); + writePages(bufferCache, fileId, COLD_INCOMPRESSIBLE_FRACTION, COLD_FILE_PAGES); + long totalBytes = ioManager.getSize(fileRef) + + (compressed ? ioManager.getSize(((CompressedFileReference) fileRef).getLAFFileReference()) : 0L); + + evictOsPageCache(); + + bufferCache.openFile(fileId); + // Identical access sequence in both passes, so cold and warm differ only in where the + // bytes came from. + long start = System.nanoTime(); + coldAccess(bufferCache, fileId, sequential); + double coldMillis = (System.nanoTime() - start) / 1_000_000.0; + + start = System.nanoTime(); + coldAccess(bufferCache, fileId, sequential); + double warmMillis = (System.nanoTime() - start) / 1_000_000.0; + + bufferCache.closeFile(fileId); + bufferCache.deleteFile(fileId); + return new ColdMeasurement(totalBytes, coldMillis, warmMillis); + } finally { + bufferCache.close(); + } + } + + private static void coldAccess(IBufferCache bufferCache, int fileId, boolean sequential) + throws HyracksDataException { + if (sequential) { + scanPages(bufferCache, fileId, COLD_FILE_PAGES); + } else { + pinRandomPages(bufferCache, fileId, COLD_PROBE_SEED, COLD_PROBES, COLD_FILE_PAGES); + } + } + + private static boolean canEvictOsPageCache() { + try { + return new ProcessBuilder("python3", "-c", "pass").start().waitFor() == 0; + } catch (Exception e) { + return false; + } + } + + /** + * Evicts file-backed pages by allocating and touching most of RAM in a short-lived subprocess. + * Clean file pages are the first thing the OS gives up under this pressure, so the benchmark file + * leaves the page cache. A subprocess is used because the surefire {@code argLine} in the parent + * pom pins this JVM to {@code -Xmx2048m}, and it cannot be overridden from the command line. + */ + private static void evictOsPageCache() throws Exception { + // The ballast must be INCOMPRESSIBLE. macOS compresses anonymous memory before it gives up + // file-backed pages, and zero-filled ballast compresses to nearly nothing -- the page cache + // would survive and every "cold" number would silently be a warm one. So every page is filled + // from a random block: the compressor works per page, so identical random pages are still + // individually incompressible, which lets one 16 MiB urandom block seed all of it cheaply. + String script = "import os\n" + "block = os.urandom(1 << 24)\n" + "held = []\n" + "for _ in range(" + + BALLAST_GIB + "):\n" + " held.append(bytearray(block * 64))\n" + "del held\n"; + Process p = new ProcessBuilder("python3", "-c", script).redirectErrorStream(true).start(); + if (!p.waitFor(BALLAST_TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + p.destroyForcibly(); + } + } + + private Measurement measure(boolean compressed, double incompressibleFraction, boolean sequential) + throws Exception { + TestStorageManagerComponentHolder.init(PAGE_SIZE, CACHE_PAGES, MAX_OPEN_FILES); + IIOManager ioManager = TestStorageManagerComponentHolder.getIOManager(); + IBufferCache bufferCache = + TestStorageManagerComponentHolder.getBufferCache(ctx.getJobletContext().getServiceContext()); + FileReference fileRef = newFileReference(ioManager, compressed, incompressibleFraction, sequential); + try { + int fileId = bufferCache.createFile(fileRef); + writePages(bufferCache, fileId, incompressibleFraction, FILE_PAGES); + + long dataBytes = ioManager.getSize(fileRef); + long lafBytes = + compressed ? ioManager.getSize(((CompressedFileReference) fileRef).getLAFFileReference()) : 0L; + + bufferCache.openFile(fileId); + double[] millis = new double[REPEATS]; + for (int repeat = 0; repeat < REPEATS; repeat++) { + long start = System.nanoTime(); + if (sequential) { + scanPages(bufferCache, fileId, FILE_PAGES); + } else { + pinRandomPages(bufferCache, fileId, repeat, RANDOM_PROBES, FILE_PAGES); + } + millis[repeat] = (System.nanoTime() - start) / 1_000_000.0; + } + bufferCache.closeFile(fileId); + bufferCache.deleteFile(fileId); + return new Measurement(dataBytes, lafBytes, medianAfterWarmup(millis)); + } finally { + bufferCache.close(); + } + } + + private void writePages(IBufferCache bufferCache, int fileId, double incompressibleFraction, int filePages) + throws HyracksDataException { + bufferCache.openFile(fileId); + ICompressedPageWriter compressedPageWriter = bufferCache.getCompressedPageWriter(fileId); + IFIFOPageWriter pageWriter = bufferCache.createFIFOWriter(NoOpPageWriteCallback.INSTANCE, + HaltOnFailureCallback.INSTANCE, DefaultBufferCacheWriteContext.INSTANCE); + for (int pageId = 0; pageId < filePages; pageId++) { + long dpid = BufferedFileHandle.getDiskPageId(fileId, pageId); + ICachedPage page = bufferCache.confiscatePage(dpid); + compressedPageWriter.prepareWrite(page); + fillLikeIndexLeaf(page.getBuffer(), pageId, incompressibleFraction); + pageWriter.write(page); + } + compressedPageWriter.endWriting(); + bufferCache.closeFile(fileId); + } + + /** + * Writes something shaped like a btree leaf of a secondary index: a run of ascending keys, + * which compress well because their high-order bytes repeat, followed by a block of + * incompressible bytes standing in for high-entropy primary keys. The split between the two is + * what {@code incompressibleFraction} controls. + * <p> + * Content is seeded from the page id alone, so the compressed and uncompressed runs of a given + * configuration see byte-identical input. + */ + private static void fillLikeIndexLeaf(ByteBuffer buf, int pageId, double incompressibleFraction) { + Random rnd = new Random(pageId); + int incompressibleBytes = (int) (PAGE_SIZE * incompressibleFraction); + int structuredBytes = PAGE_SIZE - incompressibleBytes; + + buf.position(0); + long key = (long) pageId * 100_000L; + int written = 0; + while (written + Long.BYTES <= structuredBytes) { + buf.putLong(key); + key += 3; + written += Long.BYTES; + } + byte[] noise = new byte[PAGE_SIZE - written]; + rnd.nextBytes(noise); + buf.put(noise); + buf.position(0); + } + + private static void pinRandomPages(IBufferCache bufferCache, int fileId, int seed, int probes, int filePages) + throws HyracksDataException { + Random rnd = new Random(seed); + for (int probe = 0; probe < probes; probe++) { + long dpid = BufferedFileHandle.getDiskPageId(fileId, rnd.nextInt(filePages)); + ICachedPage page = bufferCache.pin(dpid); + bufferCache.unpin(page); + } + } + + private static void scanPages(IBufferCache bufferCache, int fileId, int filePages) throws HyracksDataException { + for (int pageId = 0; pageId < filePages; pageId++) { + ICachedPage page = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, pageId)); + bufferCache.unpin(page); + } + } + + /** Drops the first repeats as JIT warm-up, then takes the median so a single stall cannot skew it. */ + private static double medianAfterWarmup(double[] millis) { + double[] measured = Arrays.copyOfRange(millis, WARMUP_REPEATS, millis.length); + Arrays.sort(measured); + int mid = measured.length / 2; + return measured.length % 2 == 1 ? measured[mid] : (measured[mid - 1] + measured[mid]) / 2; + } + + private static FileReference newFileReference(IIOManager ioManager, boolean compressed, + double incompressibleFraction, boolean sequential) throws HyracksDataException { + String name = String.format("sicb-%s-%s-%02d", sequential ? "scan" : "rand", compressed ? "snappy" : "none", + (int) (incompressibleFraction * 100)); + FileReference fileRef = ioManager.resolve(name); + // A run that died before its cleanup would otherwise leave a file that fails createFile with + // HYR0082 forever, so start from a clean slate rather than depending on the previous run. + deleteIfExists(ioManager, fileRef); + createdFiles.add(name); + if (!compressed) { + return fileRef; + } + ICompressorDecompressor compDecomp = new SnappyCompressorDecompressorFactory().createInstance(); + createdFiles.add(name + ".dic"); + CompressedFileReference cFileRef = new CompressedFileReference(fileRef.getDeviceHandle(), compDecomp, + fileRef.getRelativePath(), fileRef.getRelativePath() + ".dic"); + deleteIfExists(ioManager, cFileRef.getLAFFileReference()); + return cFileRef; + } + + private static void deleteIfExists(IIOManager ioManager, FileReference fileRef) throws HyracksDataException { + if (ioManager.exists(fileRef)) { + ioManager.delete(fileRef); + } + } + + @AfterClass + public static void cleanup() { + createdFiles.clear(); + } +} -- To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21625?usp=email To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email Gerrit-MessageType: newchange Gerrit-Project: asterixdb Gerrit-Branch: master Gerrit-Change-Id: I894eed2fbe262c5d53a077740cf804bda1b30230 Gerrit-Change-Number: 21625 Gerrit-PatchSet: 1 Gerrit-Owner: Ritik Raj <[email protected]>
