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 <[email protected]> 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
