You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by dl...@apache.org on 2026/07/20 18:20:20 UTC
(accumulo) branch main updated (6152686cfe -> af04221510)
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
from 6152686cfe Use new LeaseRecoverable interface added in HADOOP-18671 (#5241)
add 94c35c7a5d Use FileSystem.openFile with FileStatus to reduce NameNode RPCs (#6460)
new af04221510 Merge branch '2.1'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../core/client/rfile/RFileScannerBuilder.java | 29 +++++++++-
.../accumulo/core/clientImpl/bulk/BulkImport.java | 28 +++++-----
.../apache/accumulo/core/file/FileOperations.java | 39 ++++++++++++--
.../file/blockfile/impl/CachableBlockFile.java | 63 ++++++++++++++++------
.../accumulo/core/file/rfile/RFileOperations.java | 9 +++-
5 files changed, 132 insertions(+), 36 deletions(-)
(accumulo) 01/01: Merge branch '2.1'
Posted by dl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit af04221510bc8f629dacfe1b4e441b775f833ab6
Merge: 6152686cfe 94c35c7a5d
Author: Dave Marion <dl...@apache.org>
AuthorDate: Mon Jul 20 18:20:07 2026 +0000
Merge branch '2.1'
.../core/client/rfile/RFileScannerBuilder.java | 29 +++++++++-
.../accumulo/core/clientImpl/bulk/BulkImport.java | 28 +++++-----
.../apache/accumulo/core/file/FileOperations.java | 39 ++++++++++++--
.../file/blockfile/impl/CachableBlockFile.java | 63 ++++++++++++++++------
.../accumulo/core/file/rfile/RFileOperations.java | 9 +++-
5 files changed, 132 insertions(+), 36 deletions(-)
diff --cc core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
index c8e41f87fe,64d77d2c52..4e4966c811
--- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScannerBuilder.java
@@@ -58,11 -58,31 +62,32 @@@ class RFileScannerBuilder implements RF
RFileSource[] getSources() throws IOException {
if (sources == null) {
- sources = new RFileSource[paths.length];
- for (int i = 0; i < paths.length; i++) {
- FileSystem fs = getFileSystem(paths[i]);
- FileStatus status = fs.getFileStatus(paths[i]);
+ sources = new RFileSource[rFiles.length];
+ for (int i = 0; i < rFiles.length; i++) {
+ final Path path = rFiles[i].getPath();
- sources[i] = new RFileSource(getFileSystem(path).open(path),
- getFileSystem(path).getFileStatus(path).getLen(), rFiles[i].getFence());
++ FileSystem fs = getFileSystem(path);
++ FileStatus status = fs.getFileStatus(path);
+ CompletableFuture<FSDataInputStream> future =
- fs.openFile(paths[i]).withFileStatus(status).build();
++ fs.openFile(path).withFileStatus(status).build();
+ while (!future.isDone()) {
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
- throw new IOException("Interrupted while opening file: " + paths[i], e);
++ throw new IOException("Interrupted while opening file: " + path, e);
+ }
+ }
+ try {
+ FSDataInputStream is = future.get();
- sources[i] = new RFileSource(is, status.getLen());
++ sources[i] = new RFileSource(is, status.getLen(), rFiles[i].getFence());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
- throw new IOException("Interrupted while opening file: " + paths[i], e);
++ throw new IOException("Interrupted while opening file: " + path, e);
+ } catch (CancellationException e) {
- throw new IOException("Cancelled while opening file: " + paths[i], e);
++ throw new IOException("Cancelled while opening file: " + path, e);
+ } catch (ExecutionException e) {
- throw new IOException("Error trying to open file: " + paths[i], e);
++ throw new IOException("Error trying to open file: " + path, e);
+ }
}
} else {
for (int i = 0; i < sources.length; i++) {
diff --cc core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java
index 9a925de345,575a221b8b..44717cf741
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/bulk/BulkImport.java
@@@ -265,9 -265,11 +265,11 @@@ public class BulkImport implements Impo
long l;
}
- public static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration acuConf, Path mapFile,
- FileStatus status, Collection<KeyExtent> extents, FileSystem ns,
- Cache<String,Long> fileLenCache, CryptoService cs) throws IOException {
+ public static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration acuConf,
- UnreferencedTabletFile dataFile, long fileSize, Collection<KeyExtent> extents, FileSystem ns,
- Cache<String,Long> fileLenCache, CryptoService cs) throws IOException {
++ UnreferencedTabletFile dataFile, FileStatus status, Collection<KeyExtent> extents,
++ FileSystem ns, Cache<String,Long> fileLenCache, CryptoService cs) throws IOException {
+
+ final long fileSize = status.getLen();
if (extents.size() == 1) {
return Collections.singletonMap(extents.iterator().next(), fileSize);
@@@ -281,9 -283,11 +283,10 @@@
Text row = new Text();
- try (FileSKVIterator index =
- FileOperations.getInstance().newIndexReaderBuilder().forFile(dataFile, ns, ns.getConf(), cs)
- .withTableConfiguration(acuConf).withFileLenCache(fileLenCache).build()) {
- FileSKVIterator index = FileOperations.getInstance().newIndexReaderBuilder()
- .forFile(mapFile.toString(), ns, ns.getConf(), cs, status).withTableConfiguration(acuConf)
- .withFileLenCache(fileLenCache).build();
++ try (FileSKVIterator index = FileOperations.getInstance().newIndexReaderBuilder()
++ .forFile(dataFile, ns, ns.getConf(), cs, status).withTableConfiguration(acuConf)
++ .withFileLenCache(fileLenCache).build()) {
+
- try {
while (index.hasTop()) {
Key key = index.getTopKey();
totalIndexEntries++;
@@@ -353,11 -366,12 +356,12 @@@
}
public static List<KeyExtent> findOverlappingTablets(ClientContext context,
- KeyExtentCache keyExtentCache, Path file, FileSystem fs, Cache<String,Long> fileLenCache,
- CryptoService cs, FileStatus status) throws IOException {
- try (FileSKVIterator reader = FileOperations.getInstance().newReaderBuilder()
- .forFile(file.toString(), fs, fs.getConf(), cs, status)
- .withTableConfiguration(context.getConfiguration()).withFileLenCache(fileLenCache)
- .seekToBeginning().build()) {
+ KeyExtentCache keyExtentCache, UnreferencedTabletFile file, FileSystem fs,
- Cache<String,Long> fileLenCache, CryptoService cs) throws IOException {
- try (FileSKVIterator reader = FileOperations.getInstance().newReaderBuilder()
- .forFile(file, fs, fs.getConf(), cs).withTableConfiguration(context.getConfiguration())
- .withFileLenCache(fileLenCache).seekToBeginning().build()) {
++ Cache<String,Long> fileLenCache, CryptoService cs, FileStatus status) throws IOException {
++ try (FileSKVIterator reader =
++ FileOperations.getInstance().newReaderBuilder().forFile(file, fs, fs.getConf(), cs, status)
++ .withTableConfiguration(context.getConfiguration()).withFileLenCache(fileLenCache)
++ .seekToBeginning().build()) {
Collection<ByteSequence> columnFamilies = Collections.emptyList();
NextRowFunction nextRowFunction = row -> {
@@@ -558,18 -576,18 +562,18 @@@
CompletableFuture<Map<KeyExtent,Bulk.FileInfo>> future = CompletableFuture.supplyAsync(() -> {
try {
long t1 = System.currentTimeMillis();
- List<KeyExtent> extents = findOverlappingTablets(context, extentCache, filePath, fs,
- fileLensCache, cs, fileStatus);
+ List<KeyExtent> extents =
- findOverlappingTablets(context, extentCache, file, fs, fileLensCache, cs);
++ findOverlappingTablets(context, extentCache, file, fs, fileLensCache, cs, fileStatus);
// make sure file isn't going to too many tablets
- checkTabletCount(maxTablets, extents.size(), filePath.toString());
- Map<KeyExtent,Long> estSizes = estimateSizes(context.getConfiguration(), filePath,
- fileStatus, extents, fs, fileLensCache, cs);
- Map<KeyExtent,Bulk.FileInfo> pathLocations = new HashMap<>();
+ checkTabletCount(maxTablets, extents.size(), file.toString());
- Map<KeyExtent,Long> estSizes = estimateSizes(context.getConfiguration(), file,
- fileStatus.getLen(), extents, fs, fileLensCache, cs);
++ Map<KeyExtent,Long> estSizes = estimateSizes(context.getConfiguration(), file, fileStatus,
++ extents, fs, fileLensCache, cs);
+ Map<KeyExtent,Bulk.FileInfo> pathLocations = new HashMap<>(extents.size(), 1.0f);
for (KeyExtent ke : extents) {
- pathLocations.put(ke, new Bulk.FileInfo(filePath, estSizes.getOrDefault(ke, 0L)));
+ pathLocations.put(ke, new Bulk.FileInfo(file.getPath(), estSizes.getOrDefault(ke, 0L)));
}
long t2 = System.currentTimeMillis();
- log.debug("Mapped {} to {} tablets in {}ms", filePath, pathLocations.size(), t2 - t1);
+ log.debug("Mapped {} to {} tablets in {}ms", file, pathLocations.size(), t2 - t1);
return pathLocations;
} catch (Exception e) {
throw new CompletionException(e);
diff --cc core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
index ecbc595664,dd5b518d01..1ea4fb1384
--- a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
@@@ -32,16 -32,15 +32,17 @@@ import org.apache.accumulo.core.data.Ra
import org.apache.accumulo.core.data.TableId;
import org.apache.accumulo.core.file.blockfile.impl.CacheProvider;
import org.apache.accumulo.core.file.rfile.RFile;
+import org.apache.accumulo.core.metadata.TabletFile;
+import org.apache.accumulo.core.metadata.UnreferencedTabletFile;
import org.apache.accumulo.core.spi.crypto.CryptoService;
-import org.apache.accumulo.core.util.ratelimit.RateLimiter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
+ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileOutputCommitter;
-import com.google.common.cache.Cache;
+import com.github.benmanes.caffeine.cache.Cache;
public abstract class FileOperations {
@@@ -191,17 -192,20 +192,19 @@@
public final Set<ByteSequence> columnFamilies;
public final boolean inclusive;
public final boolean dropCacheBehind;
+ public final FileStatus status;
- public FileOptions(TableId tableId, AccumuloConfiguration tableConfiguration, String filename,
- FileSystem fs, Configuration fsConf, RateLimiter rateLimiter, String compression,
+ protected FileOptions(TableId tableId, AccumuloConfiguration tableConfiguration,
+ TabletFile file, FileSystem fs, Configuration fsConf, String compression,
FSDataOutputStream outputStream, boolean enableAccumuloStart, CacheProvider cacheProvider,
Cache<String,Long> fileLenCache, boolean seekToBeginning, CryptoService cryptoService,
- Range range, Set<ByteSequence> columnFamilies, boolean inclusive, boolean dropCacheBehind) {
+ Range range, Set<ByteSequence> columnFamilies, boolean inclusive, boolean dropCacheBehind,
+ FileStatus status) {
this.tableId = tableId;
this.tableConfiguration = tableConfiguration;
- this.filename = filename;
+ this.file = Objects.requireNonNull(file);
this.fs = fs;
this.fsConf = fsConf;
- this.rateLimiter = rateLimiter;
this.compression = compression;
this.outputStream = outputStream;
this.enableAccumuloStart = enableAccumuloStart;
@@@ -282,11 -291,13 +286,12 @@@
public static class FileHelper {
private TableId tableId;
private AccumuloConfiguration tableConfiguration;
- private String filename;
+ private TabletFile file;
private FileSystem fs;
private Configuration fsConf;
- private RateLimiter rateLimiter;
private CryptoService cryptoService;
private boolean dropCacheBehind = false;
+ private FileStatus status;
protected FileHelper table(TableId tid) {
this.tableId = tid;
@@@ -323,30 -339,36 +328,36 @@@
return this;
}
+ protected FileHelper fileStatus(FileStatus status) {
+ this.status = status;
+ return this;
+ }
+
protected FileOptions toWriterBuilderOptions(String compression,
FSDataOutputStream outputStream, boolean startEnabled) {
- return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter,
- compression, outputStream, startEnabled, NULL_PROVIDER, null, false, cryptoService, null,
- null, true, dropCacheBehind, status);
+ return new FileOptions(tableId, tableConfiguration, file, fs, fsConf, compression,
+ outputStream, startEnabled, NULL_PROVIDER, null, false, cryptoService, null, null, true,
- dropCacheBehind);
++ dropCacheBehind, status);
}
protected FileOptions toReaderBuilderOptions(CacheProvider cacheProvider,
Cache<String,Long> fileLenCache, boolean seekToBeginning) {
- return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null,
- null, false, cacheProvider == null ? NULL_PROVIDER : cacheProvider, fileLenCache,
- seekToBeginning, cryptoService, null, null, true, dropCacheBehind, status);
+ return new FileOptions(tableId, tableConfiguration, file, fs, fsConf, null, null, false,
+ cacheProvider == null ? NULL_PROVIDER : cacheProvider, fileLenCache, seekToBeginning,
- cryptoService, null, null, true, dropCacheBehind);
++ cryptoService, null, null, true, dropCacheBehind, status);
}
protected FileOptions toIndexReaderBuilderOptions(Cache<String,Long> fileLenCache) {
- return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null,
- null, false, NULL_PROVIDER, fileLenCache, false, cryptoService, null, null, true,
- dropCacheBehind, status);
+ return new FileOptions(tableId, tableConfiguration, file, fs, fsConf, null, null, false,
- NULL_PROVIDER, fileLenCache, false, cryptoService, null, null, true, dropCacheBehind);
++ NULL_PROVIDER, fileLenCache, false, cryptoService, null, null, true, dropCacheBehind,
++ status);
}
protected FileOptions toScanReaderBuilderOptions(Range range, Set<ByteSequence> columnFamilies,
boolean inclusive) {
- return new FileOptions(tableId, tableConfiguration, filename, fs, fsConf, rateLimiter, null,
- null, false, NULL_PROVIDER, null, false, cryptoService, range, columnFamilies, inclusive,
+ return new FileOptions(tableId, tableConfiguration, file, fs, fsConf, null, null, false,
+ NULL_PROVIDER, null, false, cryptoService, range, columnFamilies, inclusive,
- dropCacheBehind);
+ dropCacheBehind, status);
}
protected AccumuloConfiguration getTableConfiguration() {
@@@ -426,6 -451,12 +437,12 @@@
return this;
}
- public ReaderTableConfiguration forFile(String filename, FileSystem fs, Configuration fsConf,
++ public ReaderTableConfiguration forFile(TabletFile file, FileSystem fs, Configuration fsConf,
+ CryptoService cs, FileStatus status) {
- filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
++ file(file).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public ReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) {
tableConfiguration(tableConfiguration);
@@@ -489,6 -525,12 +506,12 @@@
return this;
}
- public IndexReaderTableConfiguration forFile(String filename, FileSystem fs,
++ public IndexReaderTableConfiguration forFile(TabletFile file, FileSystem fs,
+ Configuration fsConf, CryptoService cs, FileStatus status) {
- filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
++ file(file).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public IndexReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) {
tableConfiguration(tableConfiguration);
@@@ -521,6 -563,12 +544,12 @@@
return this;
}
- public ScanReaderTableConfiguration forFile(String filename, FileSystem fs,
++ public ScanReaderTableConfiguration forFile(TabletFile file, FileSystem fs,
+ Configuration fsConf, CryptoService cs, FileStatus status) {
- filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
++ file(file).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public ScanReaderBuilder withTableConfiguration(AccumuloConfiguration tableConfiguration) {
tableConfiguration(tableConfiguration);
diff --cc core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java
index 0d57ffd66e,2a29023581..6a6412e303
--- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/impl/CachableBlockFile.java
@@@ -40,10 -44,14 +43,12 @@@ import org.apache.accumulo.core.spi.cac
import org.apache.accumulo.core.spi.crypto.CryptoService;
import org.apache.accumulo.core.trace.ScanInstrumentation;
import org.apache.accumulo.core.util.CountingInputStream;
-import org.apache.accumulo.core.util.ratelimit.RateLimiter;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
+ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+ import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.Seekable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --cc core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java
index 0927b6b374,db930730d2..7cc22b5197
--- a/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/rfile/RFileOperations.java
@@@ -60,17 -60,23 +60,22 @@@ public class RFileOperations extends Fi
private static final Collection<ByteSequence> EMPTY_CF_SET = Collections.emptySet();
private static final AtomicBoolean SYNC_CAPABILITY_LOGGED = new AtomicBoolean(false);
- private static RFile.Reader getReader(FileOptions options) throws IOException {
+ private static RFileSKVIterator getReader(FileOptions options) throws IOException {
CachableBuilder cb = new CachableBuilder()
- .fsPath(options.getFileSystem(), options.getFile().getPath(), options.dropCacheBehind)
- .fsPath(options.getFileSystem(), new Path(options.getFilename()), options.dropCacheBehind,
++ .fsPath(options.getFileSystem(), options.getFile().getPath(), options.dropCacheBehind,
+ options.status)
.conf(options.getConfiguration()).fileLen(options.getFileLenCache())
- .cacheProvider(options.cacheProvider).readLimiter(options.getRateLimiter())
- .cryptoService(options.getCryptoService());
- return new RFile.Reader(cb);
+ .cacheProvider(options.cacheProvider).cryptoService(options.getCryptoService());
+ return RFile.getReader(cb, options.getFile());
}
@Override
protected long getFileSize(FileOptions options) throws IOException {
- return options.getFileSystem().getFileStatus(options.getFile().getPath()).getLen();
+ if (options.status == null) {
- return options.getFileSystem().getFileStatus(new Path(options.getFilename())).getLen();
++ return options.getFileSystem().getFileStatus(options.getFile().getPath()).getLen();
+ } else {
+ return options.status.getLen();
+ }
}
@Override