This is an automated email from the ASF dual-hosted git repository. DomGarguilo pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit a273e93d5e4c15e7549dfee65019a9e17c100778 Merge: 94940b81a0 ed7e1fd7d3 Author: Dom G <[email protected]> AuthorDate: Fri Aug 28 14:40:25 2026 -0400 Merge branch '2.1' into main core/pom.xml | 5 ++++ .../accumulo/core/client/rfile/RFileScanner.java | 3 +++ .../accumulo/core/clientImpl/ScannerOptions.java | 3 +++ .../TabletServerBatchReaderIterator.java | 3 +++ .../core/clientImpl/ThriftTransportPool.java | 4 ++++ .../core/crypto/streams/BlockedInputStream.java | 3 +++ .../fate/zookeeper/DistributedReadWriteLock.java | 5 ++++ .../accumulo/core/file/BloomFilterLayer.java | 3 +++ .../impl/SeekableByteArrayInputStream.java | 3 +++ .../org/apache/accumulo/core/file/rfile/RFile.java | 4 ++++ .../rfile/bcfile/SimpleBufferedOutputStream.java | 3 +++ .../file/streams/BoundedRangeFileInputStream.java | 3 +++ .../system/ColumnFamilySkippingIterator.java | 3 +++ .../core/iteratorsImpl/system/StatsIterator.java | 3 +++ .../core/metadata/schema/TabletsMetadata.java | 3 +++ .../compaction/RatioBasedCompactionPlanner.java | 6 ++--- .../accumulo/core/spi/crypto/AESCryptoService.java | 2 ++ .../accumulo/core/util/CountingInputStream.java | 3 +++ server/base/pom.xml | 5 ++++ .../server/compaction/CountingIterator.java | 3 +++ .../org/apache/accumulo/server/fs/FileManager.java | 3 +++ .../server/problems/ProblemReportingIterator.java | 3 +++ .../delegation/AuthenticationTokenKeyManager.java | 8 +++++++ .../apache/accumulo/server/tablets/TabletTime.java | 14 ++++------- .../main/java/org/apache/accumulo/gc/GCRun.java | 27 +++++++++++----------- server/tserver/pom.xml | 5 ++++ .../org/apache/accumulo/tserver/InMemoryMap.java | 3 +++ .../org/apache/accumulo/tserver/NativeMap.java | 3 +++ .../tserver/TabletServerResourceManager.java | 2 +- .../accumulo/tserver/tablet/ScanDataSource.java | 3 +++ .../apache/accumulo/tserver/tablet/Scanner.java | 26 ++++++++++++++++----- .../org/apache/accumulo/tserver/tablet/Tablet.java | 2 +- test/pom.xml | 5 ++++ .../test/functional/ErrorThrowingIterator.java | 3 +++ 34 files changed, 144 insertions(+), 33 deletions(-) diff --cc core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java index 92b75ae57e,f926a4fffc..a7ed159d1e --- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java +++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java @@@ -78,7 -79,8 +80,8 @@@ import org.apache.hadoop.io.Text import com.google.common.base.Preconditions; + @NotThreadSafe -class RFileScanner extends ScannerOptions implements Scanner { +final class RFileScanner extends ScannerOptions implements Scanner { private static class RFileScannerEnvironmentImpl extends ClientServiceEnvironmentImpl { diff --cc core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java index 2643546372,51e777a2e2..0918e71a7d --- a/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java +++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/TabletServerBatchReaderIterator.java @@@ -45,9 -46,10 +45,11 @@@ import java.util.concurrent.atomic.Atom import java.util.function.Supplier; import java.util.stream.Collectors; + import javax.annotation.concurrent.NotThreadSafe; + import org.apache.accumulo.core.client.AccumuloException; import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.InvalidTabletHostingRequestException; import org.apache.accumulo.core.client.SampleNotPresentException; import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel; import org.apache.accumulo.core.client.TableDeletedException; diff --cc core/src/main/java/org/apache/accumulo/core/crypto/streams/BlockedInputStream.java index b5206e9f91,4777b34153..7f41560b52 --- a/core/src/main/java/org/apache/accumulo/core/crypto/streams/BlockedInputStream.java +++ b/core/src/main/java/org/apache/accumulo/core/crypto/streams/BlockedInputStream.java @@@ -23,14 -23,13 +23,17 @@@ import java.io.EOFException import java.io.IOException; import java.io.InputStream; + import javax.annotation.concurrent.NotThreadSafe; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; + /** * Reader corresponding to BlockedOutputStream. Expects all data to be in the form of size (int) * data (size bytes) junk (however many bytes it takes to complete a block) */ +@SuppressFBWarnings(value = "CT_CONSTRUCTOR_THROW", + justification = "Constructor validation is required for proper initialization") + @NotThreadSafe public class BlockedInputStream extends InputStream { byte[] array; // ReadPos is where to start reading diff --cc core/src/main/java/org/apache/accumulo/core/fate/zookeeper/DistributedReadWriteLock.java index afa55e73b7,50ec14df8f..c1301cf610 --- a/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/DistributedReadWriteLock.java +++ b/core/src/main/java/org/apache/accumulo/core/fate/zookeeper/DistributedReadWriteLock.java @@@ -27,12 -29,9 +27,14 @@@ import java.util.SortedMap import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; +import java.util.function.BiPredicate; +import java.util.function.Supplier; + import javax.annotation.concurrent.NotThreadSafe; + +import org.apache.accumulo.core.fate.FateId; +import org.apache.accumulo.core.fate.zookeeper.FateLock.FateLockEntry; +import org.apache.accumulo.core.util.Timer; import org.apache.accumulo.core.util.UtilWaitThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@@ -62,23 -109,18 +64,26 @@@ public class DistributedReadWriteLock i private static final Logger log = LoggerFactory.getLogger(DistributedReadWriteLock.class); + public interface DistributedLock extends Lock { + LockType getType(); + + LockRange getRange(); + } + + // Intended usage of instances of this class is that it's only used by a single thread and + // supports locking across threads/processes + @NotThreadSafe - static class ReadLock implements Lock { + static class ReadLock implements DistributedLock { - QueueLock qlock; - byte[] userData; + final QueueLock qlock; + final FateId fateId; long entry = -1; + final LockRange range; - ReadLock(QueueLock qlock, byte[] userData) { + ReadLock(QueueLock qlock, FateId fateId, LockRange range) { this.qlock = qlock; - this.userData = userData; + this.fateId = fateId; + this.range = range; } // for recovery diff --cc core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/SeekableByteArrayInputStream.java index 9aaa67b3c9,732123d199..f2e8e7437a --- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/SeekableByteArrayInputStream.java +++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/SeekableByteArrayInputStream.java @@@ -22,19 -22,31 +22,22 @@@ import static java.util.Objects.require import java.io.IOException; import java.io.InputStream; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiFunction; +import java.util.function.IntBinaryOperator; + import javax.annotation.concurrent.NotThreadSafe; + -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; - /** * This class is like byte array input stream with two differences. It supports seeking and avoids * synchronization. */ + @NotThreadSafe public class SeekableByteArrayInputStream extends InputStream { - // making this volatile for the following case - // * thread 1 creates and initializes byte array - // * thread 2 reads from bye array - // spotbugs complains about this because thread2 may not see any changes to the byte array after - // thread 1 set the volatile, - // however the expectation is that the byte array is static. In the case of it being static, - // volatile ensures that - // thread 2 sees all of thread 1 changes before setting the volatile. - @SuppressFBWarnings(value = "VO_VOLATILE_REFERENCE_TO_ARRAY", - justification = "see explanation above") - private volatile byte[] buffer; - private int cur; - private int max; + private final byte[] buffer; + private final AtomicInteger cur = new AtomicInteger(0); + private final int max; @Override public int read() { diff --cc core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/SimpleBufferedOutputStream.java index 4696fb7897,e1bd64ed78..052829f1d6 --- a/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/SimpleBufferedOutputStream.java +++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/SimpleBufferedOutputStream.java @@@ -26,8 -28,9 +28,9 @@@ import javax.annotation.concurrent.NotT * A simplified BufferedOutputStream with borrowed buffer, and allow users to see how much data have * been buffered. */ + @NotThreadSafe class SimpleBufferedOutputStream extends FilterOutputStream { - protected byte[] buf; // the borrowed buffer + protected final byte[] buf; // the borrowed buffer protected int count = 0; // bytes used in buffer. // Constructor diff --cc core/src/main/java/org/apache/accumulo/core/file/streams/BoundedRangeFileInputStream.java index 0602d70972,ff6a7be71b..3198e6cf70 --- a/core/src/main/java/org/apache/accumulo/core/file/streams/BoundedRangeFileInputStream.java +++ b/core/src/main/java/org/apache/accumulo/core/file/streams/BoundedRangeFileInputStream.java @@@ -28,7 -30,8 +30,8 @@@ import org.apache.hadoop.fs.Seekable * regular input stream. One can create multiple BoundedRangeFileInputStream on top of the same * FSDataInputStream and they would not interfere with each other. */ + @NotThreadSafe -public class BoundedRangeFileInputStream extends InputStream { +public final class BoundedRangeFileInputStream extends InputStream { private volatile boolean closed = false; private final InputStream in; diff --cc core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java index b9dc881228,90095cc5bb..897af940b3 --- a/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/schema/TabletsMetadata.java @@@ -80,9 -91,7 +82,10 @@@ import com.google.common.collect.Iterat */ public class TabletsMetadata implements Iterable<TabletMetadata>, AutoCloseable { + private static final Logger LOG = LoggerFactory.getLogger(TabletsMetadata.class); + private static final Collection<ByteSequence> EMPTY_COL_FAMS = new ArrayList<>(); + + @NotThreadSafe public static class Builder implements TableRangeOptions, TableOptions, RangeOptions, Options { private final List<Text> families = new ArrayList<>(); diff --cc core/src/main/java/org/apache/accumulo/core/spi/compaction/RatioBasedCompactionPlanner.java index 630f3a5bb7,0000000000..e64b9b783e mode 100644,000000..100644 --- a/core/src/main/java/org/apache/accumulo/core/spi/compaction/RatioBasedCompactionPlanner.java +++ b/core/src/main/java/org/apache/accumulo/core/spi/compaction/RatioBasedCompactionPlanner.java @@@ -1,671 -1,0 +1,671 @@@ +/* + * 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 + * + * https://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.accumulo.core.spi.compaction; + +import static org.apache.accumulo.core.util.LazySingletons.GSON; + +import java.lang.reflect.Field; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +import org.apache.accumulo.core.client.TableNotFoundException; +import org.apache.accumulo.core.client.admin.compaction.CompactableFile; +import org.apache.accumulo.core.conf.ConfigurationTypeHelper; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.metadata.CompactableFileImpl; +import org.apache.accumulo.core.spi.common.ServiceEnvironment; +import org.apache.accumulo.core.util.NumUtil; +import org.apache.accumulo.core.util.compaction.CompactionJobPrioritizer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.common.base.Preconditions; +import com.google.gson.JsonArray; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import com.google.gson.JsonParseException; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; + +/** + * Finds the largest continuous set of small files that meet the compaction ratio and do not prevent + * future compactions. + * + * <p> + * The following configuration options are supported. Replace {@code <service>} with the name of the + * compaction service you are configuring. + * + * <ul> + * <li>Note that the CompactionCoordinator and at least one running Compactor must be assigned to + * the "large" compactor group. + * <li>{@code compaction.service.<service>.planner.opts.maxOpen} This determines the maximum number + * of files that will be included in a single compaction. + * <li>{@code compaction.service.<service>.planner.opts.groups} This is a json array of compactor + * group objects which have the following fields: + * <table> + * <caption>Default Compaction Planner Group options</caption> + * <tr> + * <th>Field Name</th> + * <th>Description</th> + * </tr> + * <tr> + * <td>group</td> + * <td>name of the compactor resource group (required)</td> + * </tr> + * <tr> + * <td>maxSize</td> + * <td>threshold sum of the input files (required for all but one of the configs)</td> + * </tr> + * </table> + * <br> + * This 'groups' object provides information that is used for mapping a compaction job to a + * compactor group. The maxSize field determines the maximum size of compaction that will run in a + * group. The maxSize field can have a suffix of K,M,G for kilobytes, megabytes, or gigabytes and + * represents the sum of the input files for a given compaction. One group can have no max size and + * it will run everything that is too large for the other groups. If all groups have a max size, + * then system compactions will only run for compactions smaller than the largest max size. User and + * selector compactions will always run, even if there is no group for their size. These compactions + * will run on the group with the largest max size. The following example value for this property + * will create three separate compactor groups. "small" will run compactions of files whose file + * size sum is less than 100M, "medium" will run compactions of files whose file size sum is less + * than 500M, and "large" will run all other compactions on Compactors configured to pull jobs from + * the large group. + * + * <pre> + * {@code + * [ + * {"group":"small", "maxSize":"100M"}, + * {"group":"medium", "maxSize":"500M"}, + * {"group": "large"} + * ]} + * </pre> + * </ul> + * + * <p> + * Starting with Accumulo 2.1.3, this plugin will use the table config option + * {@code "table.file.max"}. When the following four conditions are met, then this plugin will try + * to find a lower compaction ratio that will result in a compaction: + * <ol> + * <li>When a tablet has no compactions running</li> + * <li>Its number of files exceeds table.file.max</li> + * <li>System compactions are not finding anything to compact</li> + * <li>No files are selected for user compaction</li> + * </ol> + * For example, given a tablet with 20 files, and table.file.max is 15 and no compactions are + * planned. If the compaction ratio is set to 3, then this plugin will find the largest compaction + * ratio less than 3 that results in a compaction. The lowest compaction ratio that will be + * considered in this search defaults to 1.1. Starting in 2.1.4, the lower bound for the search can + * be set using {@code tserver.compaction.major.service.<service>.opts.lowestRatio} + * + * @since 4.0.0 + * @see org.apache.accumulo.core.spi.compaction + */ +public class RatioBasedCompactionPlanner implements CompactionPlanner { + + private final static Logger log = LoggerFactory.getLogger(RatioBasedCompactionPlanner.class); + + private static class GroupConfig { + String group; + String maxSize; + } + + private static class CompactionGroup { + final ResourceGroupId cgid; + final Long maxSize; + + public CompactionGroup(ResourceGroupId cgid, Long maxSize) { + Preconditions.checkArgument(maxSize == null || maxSize > 0, "Invalid value for maxSize"); + this.cgid = Objects.requireNonNull(cgid, "Compaction ID is null"); + this.maxSize = maxSize; + } + + Long getMaxSize() { + return maxSize; + } + + @Override + public String toString() { + return "[cgid=" + cgid + ", maxSize=" + maxSize + "]"; + } + } + + private static class FakeFileGenerator { + + private int count = 0; + + public CompactableFile create(long size) { + try { + count++; + return new CompactableFileImpl( + new URI("hdfs://fake/accumulo/tables/adef/t-zzFAKEzz/FAKE-0000" + count + ".rf"), size, + 0); + } catch (URISyntaxException e) { + throw new IllegalStateException(e); + } + } + } + - private List<CompactionGroup> groups; - private int maxFilesToCompact; - private double lowestRatio; ++ private volatile List<CompactionGroup> groups; ++ private volatile int maxFilesToCompact; ++ private volatile double lowestRatio; + + @SuppressFBWarnings(value = {"UWF_UNWRITTEN_FIELD", "NP_UNWRITTEN_FIELD"}, + justification = "Field is written by Gson") + @Override + public void init(InitParameters params) { + List<CompactionGroup> tmpGroups = new ArrayList<>(); + String values; + + if (params.getOptions().containsKey("groups") && !params.getOptions().get("groups").isBlank()) { + values = params.getOptions().get("groups"); + + // Generate a list of fields from the desired object. + final List<String> groupFields = Arrays.stream(GroupConfig.class.getDeclaredFields()) + .map(Field::getName).collect(Collectors.toList()); + + for (JsonElement element : GSON.get().fromJson(values, JsonArray.class)) { + validateConfig(element, groupFields, GroupConfig.class.getName()); + GroupConfig groupConfig = GSON.get().fromJson(element, GroupConfig.class); + + Long maxSize = groupConfig.maxSize == null ? null + : ConfigurationTypeHelper.getFixedMemoryAsBytes(groupConfig.maxSize); + + ResourceGroupId cgid; + String group = Objects.requireNonNull(groupConfig.group, "'group' must be specified"); + cgid = params.getGroupManager().getGroup(group); + tmpGroups.add(new CompactionGroup(cgid, maxSize)); + } + } + + if (tmpGroups.size() < 1) { + throw new IllegalStateException("No defined compactor groups for this planner"); + } + + tmpGroups.sort(Comparator.comparing(CompactionGroup::getMaxSize, + Comparator.nullsLast(Comparator.naturalOrder()))); + + groups = List.copyOf(tmpGroups); + + if (groups.stream().filter(g -> g.getMaxSize() == null).count() > 1) { + throw new IllegalArgumentException( + "Can only have one group w/o a maxSize. " + params.getOptions().get("groups")); + } + + // use the add method on the Set interface to check for duplicate maxSizes + Set<Long> maxSizes = new HashSet<>(); + groups.forEach(g -> { + if (!maxSizes.add(g.getMaxSize())) { + throw new IllegalArgumentException( + "Duplicate maxSize set in groups. " + params.getOptions().get("groups")); + } + }); + + lowestRatio = Double.parseDouble(params.getOptions().getOrDefault("lowestRatio", "1.1")); + Preconditions.checkArgument(lowestRatio >= 1.0, "lowestRatio must be >= 1.0 not %s", + lowestRatio); + + determineMaxFilesToCompact(params); + } + + private void determineMaxFilesToCompact(InitParameters params) { + + String maxOpen = params.getOptions().get("maxOpen"); + if (maxOpen == null) { + maxOpen = Property.COMPACTION_SERVICE_DEFAULT_MAX_OPEN.getDefaultValue(); + log.trace("default maxOpen not set, defaulting to {}", maxOpen); + } + this.maxFilesToCompact = Integer.parseInt(maxOpen); + } + + private void validateConfig(JsonElement json, List<String> fields, String className) { + + JsonObject jsonObject = GSON.get().fromJson(json, JsonObject.class); + + List<String> objectProperties = new ArrayList<>(jsonObject.keySet()); + HashSet<String> classFieldNames = new HashSet<>(fields); + + if (!classFieldNames.containsAll(objectProperties)) { + objectProperties.removeAll(classFieldNames); + throw new JsonParseException( + "Invalid fields: " + objectProperties + " provided for class: " + className); + } + } + + @Override + public CompactionPlan makePlan(PlanningParameters params) { + if (params.getCandidates().isEmpty()) { + return params.createPlanBuilder().build(); + } + + Set<CompactableFile> filesCopy = new HashSet<>(params.getCandidates()); + + FakeFileGenerator fakeFileGenerator = new FakeFileGenerator(); + + long maxSizeToCompact = getMaxSizeToCompact(params.getKind()); + + // This set represents future files that will be produced by running compactions. If the optimal + // set of files to compact is computed and contains one of these files, then it's optimal to + // wait for this compaction to finish. + Set<CompactableFile> expectedFiles = new HashSet<>(); + params.getRunningCompactions().stream().filter(job -> job.getKind() == params.getKind()) + .map(job -> getExpected(job.getFiles(), fakeFileGenerator)) + .forEach(compactableFile -> Preconditions.checkState(expectedFiles.add(compactableFile))); + Preconditions.checkState(Collections.disjoint(expectedFiles, filesCopy)); + filesCopy.addAll(expectedFiles); + + List<Collection<CompactableFile>> compactionJobs = new ArrayList<>(); + + while (true) { + var filesToCompact = + findDataFilesToCompact(filesCopy, params.getRatio(), maxFilesToCompact, maxSizeToCompact); + if (filesToCompact.isEmpty()) { + break; + } + + if (Collections.disjoint(filesToCompact, expectedFiles)) { + compactionJobs.add(filesToCompact); + } + + filesCopy.removeAll(filesToCompact); + + // A compaction job will be created for these files, so lets add an expected file for that + // planned compaction job. Then if future iterations of this loop will include that file then + // they will not compact. + var expectedFile = getExpected(filesToCompact, fakeFileGenerator); + Preconditions.checkState(expectedFiles.add(expectedFile)); + Preconditions.checkState(filesCopy.add(expectedFile)); + + if (filesToCompact.size() < maxFilesToCompact && !compactionJobs.isEmpty()) { + // Only continue looking for more compaction jobs when a set of files is found equals + // maxFilesToCompact in size. When the files found is less than the max size its an + // indication that the compaction ratio was no longer met and therefore it would be + // suboptimal to look for more jobs because the smallest optimal set was found. + break; + } + } + + if (compactionJobs.size() == 1 && params.getKind() == CompactionKind.USER + && compactionJobs.get(0).size() < params.getCandidates().size() + && compactionJobs.get(0).size() <= maxFilesToCompact) { + // USER compactions must eventually compact all files. When a subset of files + // that meets the compaction ratio is selected, look ahead and see if the next compaction + // would also meet the compaction ratio. If not then compact everything to avoid doing + // more than logarithmic work across multiple comapctions. + + var group = compactionJobs.get(0); + var candidatesCopy = new HashSet<>(params.getCandidates()); + + candidatesCopy.removeAll(group); + Preconditions.checkState(candidatesCopy.add(getExpected(group, fakeFileGenerator))); + + if (findDataFilesToCompact(candidatesCopy, params.getRatio(), maxFilesToCompact, + maxSizeToCompact).isEmpty()) { + // The next possible compaction does not meet the compaction ratio, so compact + // everything. + compactionJobs.set(0, Set.copyOf(params.getCandidates())); + } + } + + int maxTabletFiles = 0; + if (compactionJobs.isEmpty()) { + if (params.getKind() == CompactionKind.USER && params.getRunningCompactions().stream() + .noneMatch(job -> job.getKind() == params.getKind())) { + // These kinds of compaction require files to compact even if none of the files meet the + // compaction ratio. No files were found using the compaction ratio and no compactions are + // running, so force a compaction. + compactionJobs = findMaximalRequiredSetToCompact(params.getCandidates(), maxFilesToCompact); + } else if (params.getKind() == CompactionKind.SYSTEM + && params.getRunningCompactions().isEmpty() + && params.getAll().size() == params.getCandidates().size()) { + maxTabletFiles = + getMaxTabletFiles(params.getServiceEnvironment().getConfiguration(params.getTableId())); + if (params.getAll().size() > maxTabletFiles) { + // The tablet is above its max files, there are no compactions running, all files are + // candidates for a system compaction, and no files were found to compact. Attempt to + // find a set of files to compact by lowering the compaction ratio. + compactionJobs = + findFilesToCompactWithLowerRatio(params, maxSizeToCompact, maxTabletFiles); + } + } + } + + var builder = params.createPlanBuilder(); + for (Collection<CompactableFile> job : compactionJobs) { + try { + builder.addJob(createPriority(params, job, maxTabletFiles), getGroup(job), job); + } catch (TableNotFoundException e) { + throw new RuntimeException("Error getting namespace for table: " + params.getTableId(), e); + } + } + return builder.build(); + } + + static int getMaxTabletFiles(ServiceEnvironment.Configuration configuration) { + int maxTabletFiles = Integer.parseInt(configuration.get(Property.TABLE_FILE_MAX.getKey())); + if (maxTabletFiles <= 0) { + maxTabletFiles = + Integer.parseInt(configuration.get(Property.TSERV_SCAN_MAX_OPENFILES.getKey())) - 1; + } + return maxTabletFiles; + } + + /** + * Searches for the highest compaction ratio that is less than the configured ratio that will + * lower the number of files. + */ + private List<Collection<CompactableFile>> findFilesToCompactWithLowerRatio( + PlanningParameters params, long maxSizeToCompact, int maxTabletFiles) { + + var candidates = Set.copyOf(params.getCandidates()); + List<CompactableFile> sortedFiles = sortAndLimitByMaxSize(candidates, maxSizeToCompact); + + List<CompactableFile> found = List.of(); + double largestRatioSeen = Double.MIN_VALUE; + + if (sortedFiles.size() > 1) { + int windowStart = 0; + int windowEnd = Math.min(sortedFiles.size(), maxFilesToCompact); + + while (windowEnd <= sortedFiles.size()) { + var filesInWindow = sortedFiles.subList(windowStart, windowEnd); + + long sum = filesInWindow.get(0).getEstimatedSize(); + for (int i = 1; i < filesInWindow.size(); i++) { + long size = filesInWindow.get(i).getEstimatedSize(); + sum += size; + if (size > 0) { + // This is the compaction ratio needed to compact these files + double neededCompactionRatio = sum / (double) size; + log.trace("neededCompactionRatio:{} files:{}", neededCompactionRatio, + filesInWindow.subList(0, i + 1)); + if (neededCompactionRatio >= largestRatioSeen) { + largestRatioSeen = neededCompactionRatio; + found = filesInWindow.subList(0, i + 1); + } + } else { + log.warn("Unexpected size seen for file {} {} {}", params.getTabletId(), + filesInWindow.get(i).getFileName(), size); + } + } + + windowStart++; + windowEnd++; + } + } // else all of the files are too large + + if (found.isEmpty() || largestRatioSeen <= lowestRatio) { + var examinedFiles = sortAndLimitByMaxSize(candidates, maxSizeToCompact); + var excludedBecauseMaxSize = candidates.size() - examinedFiles.size(); + var tabletId = params.getTabletId(); + + log.warn("Unable to plan compaction for {} that has too many files. {}:{} num_files:{} " + + "excluded_large_files:{} max_compaction_size:{} ratio:{} largestRatioSeen:{} lowestRatio:{}", + tabletId, Property.TABLE_FILE_MAX.getKey(), maxTabletFiles, candidates.size(), + excludedBecauseMaxSize, NumUtil.bigNumberForSize(maxSizeToCompact), params.getRatio(), + largestRatioSeen, lowestRatio); + if (log.isDebugEnabled()) { + var sizesOfExamined = examinedFiles.stream() + .map(compactableFile -> NumUtil.bigNumberForSize(compactableFile.getEstimatedSize())) + .collect(Collectors.toList()); + HashSet<CompactableFile> excludedFiles = new HashSet<>(candidates); + examinedFiles.forEach(excludedFiles::remove); + var sizesOfExcluded = excludedFiles.stream() + .map(compactableFile -> NumUtil.bigNumberForSize(compactableFile.getEstimatedSize())) + .collect(Collectors.toList()); + log.debug("Failed planning details for {} examined_file_sizes:{} excluded_file_sizes:{}", + tabletId, sizesOfExamined, sizesOfExcluded); + } + found = List.of(); + } else { + log.info( + "For {} found {} files to compact lowering compaction ratio from {} to {} because the tablet " + + "exceeded {} files, it had {}", + params.getTabletId(), found.size(), params.getRatio(), largestRatioSeen, maxTabletFiles, + params.getCandidates().size()); + } + if (found.isEmpty()) { + return List.of(); + } else { + return List.of(found); + } + } + + private static short createPriority(PlanningParameters params, Collection<CompactableFile> group, + int maxTabletFiles) throws TableNotFoundException { + return CompactionJobPrioritizer.createPriority(params.getNamespaceId(), params.getTableId(), + params.getKind(), params.getAll().size(), group.size(), maxTabletFiles); + } + + private long getMaxSizeToCompact(CompactionKind kind) { + if (kind == CompactionKind.SYSTEM) { + Long max = groups.get(groups.size() - 1).maxSize; + if (max != null) { + return max; + } + } + return Long.MAX_VALUE; + } + + private CompactableFile getExpected(Collection<CompactableFile> files, + FakeFileGenerator fakeFileGenerator) { + long size = files.stream().mapToLong(CompactableFile::getEstimatedSize).sum(); + return fakeFileGenerator.create(size); + } + + private static List<Collection<CompactableFile>> + findMaximalRequiredSetToCompact(Collection<CompactableFile> files, int maxFilesToCompact) { + + if (files.size() <= maxFilesToCompact) { + return List.of(files); + } + + List<CompactableFile> sortedFiles = sortByFileSize(files); + + // compute the number of full compaction jobs with full files that could run and then subtract + // 1. The 1 is subtracted because the last job is a special case. + int batches = sortedFiles.size() / maxFilesToCompact - 1; + + if (batches > 0) { + ArrayList<Collection<CompactableFile>> jobs = new ArrayList<>(batches); + for (int i = 0; i < batches; i++) { + jobs.add(sortedFiles.subList(i * maxFilesToCompact, (i + 1) * maxFilesToCompact)); + } + return jobs; + } else { + int numToCompact = maxFilesToCompact; + + if (sortedFiles.size() > maxFilesToCompact && sortedFiles.size() < 2 * maxFilesToCompact) { + // On the second to last compaction pass, compact the minimum amount of files possible. This + // is done to avoid unnecessarily compacting the largest files more than once. + numToCompact = sortedFiles.size() - maxFilesToCompact + 1; + } + + return List.of(sortedFiles.subList(0, numToCompact)); + } + } + + /** + * @return a list of the smallest files where the sum of the sizes is less than maxSizeToCompact + */ + static List<CompactableFile> sortAndLimitByMaxSize(Set<CompactableFile> files, + long maxSizeToCompact) { + + // sort files from smallest to largest. So position 0 has the smallest file. + List<CompactableFile> sortedFiles = sortByFileSize(files); + + if (maxSizeToCompact == Long.MAX_VALUE) { + return sortedFiles; + } + + int maxSizeIndex = sortedFiles.size(); + long sum = 0; + for (int i = 0; i < sortedFiles.size(); i++) { + sum += sortedFiles.get(i).getEstimatedSize(); + if (sum > maxSizeToCompact) { + maxSizeIndex = i; + break; + } + } + + if (maxSizeIndex < sortedFiles.size()) { + return sortedFiles.subList(0, maxSizeIndex); + } else { + return sortedFiles; + } + } + + static Collection<CompactableFile> findDataFilesToCompact(Set<CompactableFile> files, + double ratio, int maxFilesToCompact, long maxSizeToCompact) { + + if (files.size() <= 1) { + return Collections.emptySet(); + } + + List<CompactableFile> sortedFiles = sortAndLimitByMaxSize(files, maxSizeToCompact); + if (sortedFiles.size() <= 1) { + return Collections.emptySet(); + } + + int windowStart = 0; + int windowEnd = Math.min(sortedFiles.size(), maxFilesToCompact); + + while (windowEnd <= sortedFiles.size()) { + var filesToCompact = + findDataFilesToCompact(sortedFiles.subList(windowStart, windowEnd), ratio); + if (!filesToCompact.isEmpty()) { + return filesToCompact; + } + + windowStart++; + windowEnd++; + } + + return Collections.emptySet(); + + } + + /** + * Find the largest set of contiguous small files that meet the compaction ratio. For a set of + * file size like [101M,102M,103M,104M,4M,3M,3M,3M,3M], it would be nice compact the smaller files + * [4M,3M,3M,3M,3M] followed by the larger ones. The reason to do the smaller ones first is to + * more quickly reduce the number of files. However, all compactions should still follow the + * compaction ratio in order to ensure the amount of data rewriting is logarithmic. + * + * <p> + * A set of files meets the compaction ratio when the largestFileinSet * compactionRatio < + * sumOfFileSizesInSet. This algorithm grows the set of small files until it meets the compaction + * ratio, then keeps growing it while it continues to meet the ratio. Once a set does not meet the + * compaction ratio, the last set that did is returned. Growing the set of small files means + * adding the smallest file not in the set. + * + * <p> + * There is one caveat to the algorithm mentioned above, if a smaller set of files would prevent a + * future compaction then do not select it. This code in this function performs a look ahead to + * see if a candidate set will prevent future compactions. + * + * <p> + * As an example of a small set of files that could prevent a future compaction, consider the + * files sizes [100M,99M,33M,33M,33M,33M]. For a compaction ratio of 3, the set + * [100M,99M,33M,33M,33M,33M] and [33M,33M,33M,33M] both meet the compaction ratio. If the set + * [33M,33M,33M,33M] is compacted, then it will result in a tablet having [132M, 100M, 99M] which + * does not meet the compaction ration. So in this case, choosing the set [33M,33M,33M,33M] + * prevents a future compaction that could have occurred. This function will not choose the + * smaller set because of it would prevent the future compaction. + */ + private static Collection<CompactableFile> + findDataFilesToCompact(List<CompactableFile> sortedFiles, double ratio) { + + int larsmaIndex = -1; + long larsmaSum = Long.MIN_VALUE; + + // index into sortedFiles, everything at and below this index meets the compaction ratio + int goodIndex = -1; + + long sum = sortedFiles.get(0).getEstimatedSize(); + + for (int c = 1; c < sortedFiles.size(); c++) { + long currSize = sortedFiles.get(c).getEstimatedSize(); + + // ensure data is sorted + Preconditions.checkArgument(currSize >= sortedFiles.get(c - 1).getEstimatedSize()); + + sum += currSize; + + if (currSize * ratio < sum) { + goodIndex = c; + } else if (c - 1 == goodIndex) { + // The previous file met the compaction ratio, but the current file does not. So all of the + // previous files are candidates. However we must ensure that any candidate set produces a + // file smaller than the next largest file in the next candidate set to ensure future + // compactions are not prevented. + if (larsmaIndex == -1 || larsmaSum > sortedFiles.get(goodIndex).getEstimatedSize()) { + larsmaIndex = goodIndex; + larsmaSum = sum - currSize; + } else { + break; + } + } + } + + if (sortedFiles.size() - 1 == goodIndex + && (larsmaIndex == -1 || larsmaSum > sortedFiles.get(goodIndex).getEstimatedSize())) { + larsmaIndex = goodIndex; + } + + if (larsmaIndex == -1) { + return Collections.emptySet(); + } + + return sortedFiles.subList(0, larsmaIndex + 1); + } + + ResourceGroupId getGroup(Collection<CompactableFile> files) { + + long size = files.stream().mapToLong(CompactableFile::getEstimatedSize).sum(); + + for (CompactionGroup group : groups) { + if (group.maxSize == null || size < group.maxSize) { + return group.cgid; + } + } + + return groups.get(groups.size() - 1).cgid; + } + + private static List<CompactableFile> sortByFileSize(Collection<CompactableFile> files) { + ArrayList<CompactableFile> sortedFiles = new ArrayList<>(files); + + // sort from smallest file to largest + Collections.sort(sortedFiles, Comparator.comparingLong(CompactableFile::getEstimatedSize) + .thenComparing(CompactableFile::getUri)); + + return sortedFiles; + } +} diff --cc server/base/src/main/java/org/apache/accumulo/server/problems/ProblemReportingIterator.java index 5cfd27063b,e72cd6dab4..ea77bb6f60 --- a/server/base/src/main/java/org/apache/accumulo/server/problems/ProblemReportingIterator.java +++ b/server/base/src/main/java/org/apache/accumulo/server/problems/ProblemReportingIterator.java @@@ -31,8 -33,9 +33,9 @@@ import org.apache.accumulo.core.data.Va import org.apache.accumulo.core.iterators.IteratorEnvironment; import org.apache.accumulo.core.iterators.SortedKeyValueIterator; import org.apache.accumulo.core.iteratorsImpl.system.InterruptibleIterator; -import org.apache.accumulo.server.ServerContext; +import org.apache.accumulo.core.logging.TabletLogger; + @NotThreadSafe public class ProblemReportingIterator implements InterruptibleIterator { private final SortedKeyValueIterator<Key,Value> source; private boolean sawError = false; diff --cc server/base/src/main/java/org/apache/accumulo/server/tablets/TabletTime.java index ae6c249ada,ead338961d..bc99948334 --- a/server/base/src/main/java/org/apache/accumulo/server/tablets/TabletTime.java +++ b/server/base/src/main/java/org/apache/accumulo/server/tablets/TabletTime.java @@@ -86,7 -87,7 +86,7 @@@ public abstract class TabletTime } @Override - public void updateTimeIfGreater(long time) { - public synchronized void useMaxTimeFromWALog(long time) { ++ public synchronized void updateTimeIfGreater(long time) { if (time > lastTime) { lastTime = time; } @@@ -155,12 -156,8 +155,8 @@@ } @Override - public void useMaxTimeFromWALog(long time) { + public void updateTimeIfGreater(long time) { - time++; - - if (this.nextTime.get() < time) { - this.nextTime.set(time); - } + this.nextTime.getAndUpdate(val -> Math.max(val, time + 1)); } @Override diff --cc server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java index 0fe140ce33,567682c720..f13cff3bd1 --- a/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java +++ b/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java @@@ -337,8 -349,8 +338,8 @@@ public class GCRun implements GarbageCo } else { // this failure, we still want to remove the metadata entry removeFlag = true; - errors++; + errors.increment(); - String[] parts = pathToDel.toString().split(Constants.ZTABLES)[1].split("/"); + String[] parts = pathToDel.toString().split(Constants.HDFS_TABLES_DIR)[1].split("/"); if (parts.length > 2) { TableId tableId = TableId.of(parts[1]); String tabletDir = parts[2]; @@@ -418,9 -431,34 +419,9 @@@ @Override public void incrementInUseStat(long i) { - inUse += i; + inUse.add(i); } - @Override - @Deprecated - public Iterator<Map.Entry<String,Replication.Status>> getReplicationNeededIterator() { - AccumuloClient client = context; - try { - Scanner s = org.apache.accumulo.core.replication.ReplicationTable.getScanner(client); - org.apache.accumulo.core.replication.ReplicationSchema.StatusSection.limit(s); - return Iterators.transform(s.iterator(), input -> { - String file = input.getKey().getRow().toString(); - Replication.Status stat; - try { - stat = Replication.Status.parseFrom(input.getValue().get()); - } catch (InvalidProtocolBufferException e) { - log.warn("Could not deserialize protobuf for: {}", input.getKey()); - stat = null; - } - return Maps.immutableEntry(file, stat); - }); - } catch (org.apache.accumulo.core.replication.ReplicationTableOfflineException - | TableOfflineException e) { - // No elements that we need to preclude - return Collections.emptyIterator(); - } - } - @VisibleForTesting static void minimizeDeletes(SortedMap<String,GcCandidate> confirmedDeletes, List<GcCandidate> processedDeletes, VolumeManager fs, Logger logger, diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java index e6779ada30,be327e0f46..34e1a680e3 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java @@@ -433,9 -412,10 +433,9 @@@ public class TabletServerResourceManage public static class AssignmentWatcher implements Runnable { private static final Logger log = LoggerFactory.getLogger(AssignmentWatcher.class); - private static long longAssignments = 0; + private static volatile long longAssignments = 0; private final Map<KeyExtent,RunnableStartedAt> activeAssignments; - private final AccumuloConfiguration conf; private final ServerContext context; public static long getLongAssignments() { diff --cc server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java index c1a4b5675a,9fef32b6fd..0e0f8d1d08 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/tablet/Tablet.java @@@ -203,7 -226,8 +203,7 @@@ public class Tablet extends TabletBase private final Rate ingestByteRate = new Rate(0.95); private final Rate scannedRate = new Rate(0.95); - private long lastMinorCompactionFinishTime = 0; + private volatile long lastMinorCompactionFinishTime = 0; - private volatile long lastMapFileImportTime = 0; private volatile long numEntries = 0; private volatile long numEntriesInMemory = 0;
