This is an automated email from the ASF dual-hosted git repository.
dlmarion pushed a commit to branch 2.1
in repository https://gitbox.apache.org/repos/asf/accumulo.git
The following commit(s) were added to refs/heads/2.1 by this push:
new 94c35c7a5d Use FileSystem.openFile with FileStatus to reduce NameNode
RPCs (#6460)
94c35c7a5d is described below
commit 94c35c7a5d1b2fb632b34484e8627cb464143d96
Author: Dave Marion <[email protected]>
AuthorDate: Mon Jul 20 13:10:28 2026 -0400
Use FileSystem.openFile with FileStatus to reduce NameNode RPCs (#6460)
This commit changes how we open Hadoop files for reading. Instead
of calling Filesystem.open this changes the code to use
FileSystem.openFile. The openFile method returns a Builder object
that has a setter method for a FileStatus object. HDFS-17593 adds
logic to the DFSClient to use the located blocks in the FileStatus
to reduce NameNode RPCs to get the block locations. This is useful
in code where we happen to already have the FileStatus object for
the associated file that we want to open.
---
.../core/client/rfile/RFileScannerBuilder.java | 29 +++++++++-
.../accumulo/core/clientImpl/bulk/BulkImport.java | 18 ++++---
.../apache/accumulo/core/file/FileOperations.java | 38 +++++++++++--
.../file/blockfile/impl/CachableBlockFile.java | 62 ++++++++++++++++------
.../accumulo/core/file/rfile/RFileOperations.java | 9 +++-
.../accumulo/server/client/BulkImporter.java | 59 ++++++++++++--------
.../org/apache/accumulo/server/util/FileUtil.java | 10 ++--
.../accumulo/server/client/BulkImporterTest.java | 6 +--
.../accumulo/tserver/log/RecoveryLogsIterator.java | 26 ++++-----
9 files changed, 183 insertions(+), 74 deletions(-)
diff --git
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
index d9ad691cac..64d77d2c52 100644
---
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
@@ -23,6 +23,9 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.rfile.RFile.ScannerFSOptions;
@@ -30,6 +33,7 @@ import
org.apache.accumulo.core.client.rfile.RFile.ScannerOptions;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -56,8 +60,29 @@ class RFileScannerBuilder implements RFile.InputArguments,
RFile.ScannerFSOption
if (sources == null) {
sources = new RFileSource[paths.length];
for (int i = 0; i < paths.length; i++) {
- sources[i] = new RFileSource(getFileSystem(paths[i]).open(paths[i]),
- getFileSystem(paths[i]).getFileStatus(paths[i]).getLen());
+ FileSystem fs = getFileSystem(paths[i]);
+ FileStatus status = fs.getFileStatus(paths[i]);
+ CompletableFuture<FSDataInputStream> future =
+ fs.openFile(paths[i]).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);
+ }
+ }
+ try {
+ FSDataInputStream is = future.get();
+ sources[i] = new RFileSource(is, status.getLen());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while opening file: " +
paths[i], e);
+ } catch (CancellationException e) {
+ throw new IOException("Cancelled while opening file: " + paths[i],
e);
+ } catch (ExecutionException e) {
+ throw new IOException("Error trying to open file: " + paths[i], e);
+ }
}
} else {
for (int i = 0; i < sources.length; i++) {
diff --git
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
index 51a2302652..575a221b8b 100644
---
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
@@ -266,8 +266,10 @@ public class BulkImport implements
ImportDestinationArguments, ImportMappingOpti
}
public static Map<KeyExtent,Long> estimateSizes(AccumuloConfiguration
acuConf, Path mapFile,
- long fileSize, Collection<KeyExtent> extents, FileSystem ns,
Cache<String,Long> fileLenCache,
- CryptoService cs) throws IOException {
+ 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);
@@ -282,7 +284,7 @@ public class BulkImport implements
ImportDestinationArguments, ImportMappingOpti
Text row = new Text();
FileSKVIterator index =
FileOperations.getInstance().newIndexReaderBuilder()
- .forFile(mapFile.toString(), ns, ns.getConf(),
cs).withTableConfiguration(acuConf)
+ .forFile(mapFile.toString(), ns, ns.getConf(), cs,
status).withTableConfiguration(acuConf)
.withFileLenCache(fileLenCache).build();
try {
@@ -365,9 +367,9 @@ public class BulkImport implements
ImportDestinationArguments, ImportMappingOpti
public static List<KeyExtent> findOverlappingTablets(ClientContext context,
KeyExtentCache keyExtentCache, Path file, FileSystem fs,
Cache<String,Long> fileLenCache,
- CryptoService cs) throws IOException {
+ CryptoService cs, FileStatus status) throws IOException {
try (FileSKVIterator reader =
FileOperations.getInstance().newReaderBuilder()
- .forFile(file.toString(), fs, fs.getConf(), cs)
+ .forFile(file.toString(), fs, fs.getConf(), cs, status)
.withTableConfiguration(context.getConfiguration()).withFileLenCache(fileLenCache)
.seekToBeginning().build()) {
@@ -574,12 +576,12 @@ public class BulkImport implements
ImportDestinationArguments, ImportMappingOpti
CompletableFuture<Map<KeyExtent,Bulk.FileInfo>> future =
CompletableFuture.supplyAsync(() -> {
try {
long t1 = System.currentTimeMillis();
- List<KeyExtent> extents =
- findOverlappingTablets(context, extentCache, filePath, fs,
fileLensCache, cs);
+ List<KeyExtent> extents = findOverlappingTablets(context,
extentCache, filePath, 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.getLen(), extents, fs, fileLensCache, cs);
+ fileStatus, extents, fs, fileLensCache, cs);
Map<KeyExtent,Bulk.FileInfo> pathLocations = new HashMap<>();
for (KeyExtent ke : extents) {
pathLocations.put(ke, new Bulk.FileInfo(filePath,
estSizes.getOrDefault(ke, 0L)));
diff --git
a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
index ce5c392fdc..dd5b518d01 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/FileOperations.java
@@ -36,6 +36,7 @@ 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.mapred.FileOutputCommitter;
@@ -191,12 +192,14 @@ public abstract class FileOperations {
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,
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;
@@ -214,6 +217,7 @@ public abstract class FileOperations {
this.columnFamilies = columnFamilies;
this.inclusive = inclusive;
this.dropCacheBehind = dropCacheBehind;
+ this.status = status;
}
public TableId getTableId() {
@@ -293,6 +297,7 @@ public abstract class FileOperations {
private RateLimiter rateLimiter;
private CryptoService cryptoService;
private boolean dropCacheBehind = false;
+ private FileStatus status;
protected FileHelper table(TableId tid) {
this.tableId = tid;
@@ -334,31 +339,36 @@ public abstract class FileOperations {
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);
+ null, true, 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);
+ seekToBeginning, 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);
+ 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,
- dropCacheBehind);
+ dropCacheBehind, status);
}
protected AccumuloConfiguration getTableConfiguration() {
@@ -441,6 +451,12 @@ public abstract class FileOperations {
return this;
}
+ public ReaderTableConfiguration forFile(String filename, FileSystem fs,
Configuration fsConf,
+ CryptoService cs, FileStatus status) {
+
filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public ReaderBuilder withTableConfiguration(AccumuloConfiguration
tableConfiguration) {
tableConfiguration(tableConfiguration);
@@ -509,6 +525,12 @@ public abstract class FileOperations {
return this;
}
+ public IndexReaderTableConfiguration forFile(String filename, FileSystem
fs,
+ Configuration fsConf, CryptoService cs, FileStatus status) {
+
filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public IndexReaderBuilder withTableConfiguration(AccumuloConfiguration
tableConfiguration) {
tableConfiguration(tableConfiguration);
@@ -541,6 +563,12 @@ public abstract class FileOperations {
return this;
}
+ public ScanReaderTableConfiguration forFile(String filename, FileSystem fs,
+ Configuration fsConf, CryptoService cs, FileStatus status) {
+
filename(filename).fs(fs).fsConf(fsConf).cryptoService(cs).fileStatus(status);
+ return this;
+ }
+
@Override
public ScanReaderBuilder withTableConfiguration(AccumuloConfiguration
tableConfiguration) {
tableConfiguration(tableConfiguration);
diff --git
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
index 19033e4394..2a29023581 100644
---
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
@@ -26,6 +26,8 @@ import java.io.UncheckedIOException;
import java.util.Collections;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
@@ -45,7 +47,9 @@ 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;
@@ -87,29 +91,57 @@ public class CachableBlockFile {
}
public CachableBuilder fsPath(FileSystem fs, Path dataFile) {
- return fsPath(fs, dataFile, false);
+ return fsPath(fs, dataFile, false, null);
}
- public CachableBuilder fsPath(FileSystem fs, Path dataFile, boolean
dropCacheBehind) {
+ public CachableBuilder fsPath(FileSystem fs, Path dataFile, FileStatus
status) {
+ return fsPath(fs, dataFile, false, status);
+ }
+
+ public CachableBuilder fsPath(FileSystem fs, Path dataFile, boolean
dropCacheBehind,
+ FileStatus status) {
this.cacheId = pathToCacheId(dataFile);
this.inputSupplier = () -> {
- FSDataInputStream is = fs.open(dataFile);
- if (dropCacheBehind) {
- // Tell the DataNode that the write ahead log does not need to be
cached in the OS page
- // cache
+ FutureDataInputStreamBuilder builder = fs.openFile(dataFile);
+ if (status != null) {
+ builder.withFileStatus(status);
+ }
+ CompletableFuture<FSDataInputStream> future = builder.build();
+ while (!future.isDone()) {
try {
- is.setDropBehind(Boolean.TRUE);
- log.trace("Called setDropBehind(TRUE) for stream reading file {}",
dataFile);
- } catch (UnsupportedOperationException e) {
- log.debug("setDropBehind not enabled for wal file: {}", dataFile);
- } catch (IOException e) {
- log.debug("IOException setting drop behind for file: {}, msg: {}",
dataFile,
- e.getMessage());
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while opening file: " +
dataFile, e);
+ }
+ }
+ try {
+ FSDataInputStream is = future.get();
+ if (dropCacheBehind) {
+ // Tell the DataNode that the write ahead log does not need to be
cached in the OS page
+ // cache
+ try {
+ is.setDropBehind(Boolean.TRUE);
+ log.trace("Called setDropBehind(TRUE) for stream reading file
{}", dataFile);
+ } catch (UnsupportedOperationException e) {
+ log.debug("setDropBehind not enabled for wal file: {}",
dataFile);
+ } catch (IOException e) {
+ log.debug("IOException setting drop behind for file: {}, msg:
{}", dataFile,
+ e.getMessage());
+ }
}
+ return is;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Interrupted while opening file: " + dataFile,
e);
+ } catch (CancellationException e) {
+ throw new IOException("Cancelled while opening file: " + dataFile,
e);
+ } catch (ExecutionException e) {
+ throw new IOException("Error trying to open file: " + dataFile, e);
}
- return is;
};
- this.lengthSupplier = () -> fs.getFileStatus(dataFile).getLen();
+ this.lengthSupplier =
+ () -> status == null ? fs.getFileStatus(dataFile).getLen() :
status.getLen();
return this;
}
diff --git
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
index c87b071642..db930730d2 100644
---
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
@@ -62,7 +62,8 @@ public class RFileOperations extends FileOperations {
private static RFile.Reader getReader(FileOptions options) throws
IOException {
CachableBuilder cb = new CachableBuilder()
- .fsPath(options.getFileSystem(), new Path(options.getFilename()),
options.dropCacheBehind)
+ .fsPath(options.getFileSystem(), new Path(options.getFilename()),
options.dropCacheBehind,
+ options.status)
.conf(options.getConfiguration()).fileLen(options.getFileLenCache())
.cacheProvider(options.cacheProvider).readLimiter(options.getRateLimiter())
.cryptoService(options.getCryptoService());
@@ -71,7 +72,11 @@ public class RFileOperations extends FileOperations {
@Override
protected long getFileSize(FileOptions options) throws IOException {
- return options.getFileSystem().getFileStatus(new
Path(options.getFilename())).getLen();
+ if (options.status == null) {
+ return options.getFileSystem().getFileStatus(new
Path(options.getFilename())).getLen();
+ } else {
+ return options.status.getLen();
+ }
}
@Override
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
index 8d224b6b45..61457d907e 100644
---
a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
+++
b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java
@@ -65,6 +65,7 @@ import org.apache.accumulo.core.util.threads.ThreadPools;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
@@ -126,6 +127,8 @@ public class BulkImporter {
final Map<Path,List<KeyExtent>> completeFailures =
Collections.synchronizedSortedMap(new TreeMap<>());
+ final Map<Path,FileStatus> fileStatuses = new HashMap<>();
+
ClientService.Client client = null;
final TabletLocator locator = TabletLocator.getLocator(context, tableId);
@@ -143,8 +146,10 @@ public class BulkImporter {
Runnable getAssignments = () -> {
List<TabletLocation> tabletsToAssignMapFileTo =
Collections.emptyList();
try {
- tabletsToAssignMapFileTo =
- findOverlappingTablets(context, fs, locator, mapFile,
tableConf.getCryptoService());
+ FileStatus status = fs.getFileStatus(mapFile);
+ fileStatuses.put(mapFile, status);
+ tabletsToAssignMapFileTo = findOverlappingTablets(context, fs,
locator, mapFile,
+ tableConf.getCryptoService(), status);
} catch (Exception ex) {
log.warn("Unable to find tablets that overlap file " + mapFile,
ex);
}
@@ -173,7 +178,7 @@ public class BulkImporter {
assignmentStats.attemptingAssignments(assignments);
Map<Path,List<KeyExtent>> assignmentFailures =
- assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads);
+ assignMapFiles(fs, assignments, paths, numAssignThreads, numThreads,
fileStatuses);
assignmentStats.assignmentsFailed(assignmentFailures);
Map<Path,Integer> failureCount = new TreeMap<>();
@@ -212,8 +217,9 @@ public class BulkImporter {
timer.start(Timers.QUERY_METADATA);
try {
- tabletsToAssignMapFileTo.addAll(findOverlappingTablets(context,
fs, locator,
- entry.getKey(), ke, tableConf.getCryptoService()));
+ Path p = entry.getKey();
+ tabletsToAssignMapFileTo.addAll(findOverlappingTablets(context,
fs, locator, p, ke,
+ tableConf.getCryptoService(), fileStatuses.get(p)));
keListIter.remove();
} catch (Exception ex) {
log.warn("Exception finding overlapping tablets, will retry
tablet " + ke, ex);
@@ -228,7 +234,7 @@ public class BulkImporter {
assignmentStats.attemptingAssignments(assignments);
Map<Path,List<KeyExtent>> assignmentFailures2 =
- assignMapFiles(fs, assignments, paths, numAssignThreads,
numThreads);
+ assignMapFiles(fs, assignments, paths, numAssignThreads,
numThreads, fileStatuses);
assignmentStats.assignmentsFailed(assignmentFailures2);
// merge assignmentFailures2 into assignmentFailures
@@ -347,15 +353,21 @@ public class BulkImporter {
}
private Map<Path,List<AssignmentInfo>> estimateSizes(final VolumeManager vm,
- Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int
numThreads) {
+ Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int
numThreads,
+ Map<Path,FileStatus> statuses) {
long t1 = System.currentTimeMillis();
final Map<Path,Long> mapFileSizes = new TreeMap<>();
try {
for (Path path : paths) {
- FileSystem fs = vm.getFileSystemByPath(path);
- mapFileSizes.put(path, fs.getContentSummary(path).getLength());
+ FileStatus status = statuses.get(path);
+ if (status == null) {
+ FileSystem fs = vm.getFileSystemByPath(path);
+ mapFileSizes.put(path, fs.getContentSummary(path).getLength());
+ } else {
+ mapFileSizes.put(path, status.getLen());
+ }
}
} catch (IOException e) {
log.error("Failed to get map files in for {}: {}", paths,
e.getMessage(), e);
@@ -386,9 +398,9 @@ public class BulkImporter {
Path mapFile = entry.getKey();
FileSystem ns =
context.getVolumeManager().getFileSystemByPath(mapFile);
- estimatedSizes =
BulkImport.estimateSizes(context.getConfiguration(), mapFile,
- mapFileSizes.get(entry.getKey()), extentsOf(entry.getValue()),
ns, null,
- tableConf.getCryptoService());
+ estimatedSizes =
+ BulkImport.estimateSizes(context.getConfiguration(), mapFile,
statuses.get(mapFile),
+ extentsOf(entry.getValue()), ns, null,
tableConf.getCryptoService());
} catch (IOException e) {
log.warn("Failed to estimate map file sizes {}", e.getMessage());
}
@@ -447,10 +459,10 @@ public class BulkImporter {
private Map<Path,List<KeyExtent>> assignMapFiles(VolumeManager fs,
Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int
numThreads,
- int numMapThreads) {
+ int numMapThreads, Map<Path,FileStatus> statuses) {
timer.start(Timers.EXAMINE_MAP_FILES);
Map<Path,List<AssignmentInfo>> assignInfo =
- estimateSizes(fs, assignments, paths, numMapThreads);
+ estimateSizes(fs, assignments, paths, numMapThreads, statuses);
timer.stop(Timers.EXAMINE_MAP_FILES);
Map<Path,List<KeyExtent>> ret;
@@ -623,15 +635,16 @@ public class BulkImporter {
}
public static List<TabletLocation> findOverlappingTablets(ServerContext
context, VolumeManager fs,
- TabletLocator locator, Path file, CryptoService cs) throws Exception {
- return findOverlappingTablets(context, fs, locator, file, null, null, cs);
+ TabletLocator locator, Path file, CryptoService cs, FileStatus status)
throws Exception {
+ return findOverlappingTablets(context, fs, locator, file, null, null, cs,
status);
}
public static List<TabletLocation> findOverlappingTablets(ServerContext
context, VolumeManager fs,
- TabletLocator locator, Path file, KeyExtent failed, CryptoService cs)
throws Exception {
+ TabletLocator locator, Path file, KeyExtent failed, CryptoService cs,
FileStatus status)
+ throws Exception {
locator.invalidateCache(failed);
Text start = getStartRowForExtent(failed);
- return findOverlappingTablets(context, fs, locator, file, start,
failed.endRow(), cs);
+ return findOverlappingTablets(context, fs, locator, file, start,
failed.endRow(), cs, status);
}
protected static Text getStartRowForExtent(KeyExtent extent) {
@@ -648,16 +661,16 @@ public class BulkImporter {
static final byte[] byte0 = {0};
public static List<TabletLocation> findOverlappingTablets(ServerContext
context, VolumeManager vm,
- TabletLocator locator, Path file, Text startRow, Text endRow,
CryptoService cs)
- throws Exception {
+ TabletLocator locator, Path file, Text startRow, Text endRow,
CryptoService cs,
+ FileStatus status) throws Exception {
List<TabletLocation> result = new ArrayList<>();
Collection<ByteSequence> columnFamilies = Collections.emptyList();
String filename = file.toString();
// log.debug(filename + " finding overlapping tablets " + startRow + " ->
" + endRow);
FileSystem fs = vm.getFileSystemByPath(file);
- try (FileSKVIterator reader =
- FileOperations.getInstance().newReaderBuilder().forFile(filename, fs,
fs.getConf(), cs)
-
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) {
+ try (FileSKVIterator reader =
FileOperations.getInstance().newReaderBuilder()
+ .forFile(filename, fs, fs.getConf(), cs, status)
+
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) {
Text row = startRow;
if (row == null) {
row = new Text();
diff --git
a/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java
b/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java
index 3c189644df..3e909b330f 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/util/FileUtil.java
@@ -48,6 +48,7 @@ import org.apache.accumulo.core.metadata.TabletFile;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.conf.TableConfiguration;
import org.apache.accumulo.server.fs.VolumeManager;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
@@ -434,14 +435,15 @@ public class FileUtil {
FileSKVIterator reader = null;
Path path = new Path(file);
FileSystem ns = context.getVolumeManager().getFileSystemByPath(path);
+ FileStatus status = ns.getFileStatus(path);
try {
if (useIndex) {
reader = FileOperations.getInstance().newIndexReaderBuilder()
- .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService())
+ .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService(), status)
.withTableConfiguration(tableConf).build();
} else {
reader = FileOperations.getInstance().newScanReaderBuilder()
- .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService())
+ .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService(), status)
.withTableConfiguration(tableConf)
.overRange(new Range(prevEndRow, false, null, true), Set.of(),
false).build();
}
@@ -468,11 +470,11 @@ public class FileUtil {
if (useIndex) {
readers.add(FileOperations.getInstance().newIndexReaderBuilder()
- .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService())
+ .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService(), status)
.withTableConfiguration(tableConf).build());
} else {
readers.add(FileOperations.getInstance().newScanReaderBuilder()
- .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService())
+ .forFile(path.toString(), ns, ns.getConf(),
tableConf.getCryptoService(), status)
.withTableConfiguration(tableConf)
.overRange(new Range(prevEndRow, false, null, true), Set.of(),
false).build());
}
diff --git
a/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java
b/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java
index 9b69efa196..5e0647d697 100644
---
a/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java
+++
b/server/base/src/test/java/org/apache/accumulo/server/client/BulkImporterTest.java
@@ -144,8 +144,8 @@ public class BulkImporterTest {
writer.append(new Key("xyzzy", "cf", "cq"), empty);
writer.close();
try (var vm = VolumeManagerImpl.getLocalForTesting("file:///")) {
- List<TabletLocation> overlaps =
- BulkImporter.findOverlappingTablets(context, vm, locator, new
Path(file), null, null, cs);
+ List<TabletLocation> overlaps =
BulkImporter.findOverlappingTablets(context, vm, locator,
+ new Path(file), null, null, cs, null);
assertEquals(5, overlaps.size());
Collections.sort(overlaps);
assertEquals(new KeyExtent(tableId, new Text("a"), null),
overlaps.get(0).tablet_extent);
@@ -158,7 +158,7 @@ public class BulkImporterTest {
assertEquals(new KeyExtent(tableId, null, new Text("l")),
overlaps.get(4).tablet_extent);
List<TabletLocation> overlaps2 =
BulkImporter.findOverlappingTablets(context, vm, locator,
- new Path(file), new KeyExtent(tableId, new Text("h"), new
Text("b")), cs);
+ new Path(file), new KeyExtent(tableId, new Text("h"), new
Text("b")), cs, null);
assertEquals(3, overlaps2.size());
assertEquals(new KeyExtent(tableId, new Text("d"), new Text("cm")),
overlaps2.get(0).tablet_extent);
diff --git
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
index f2a1a5c990..b56612314a 100644
---
a/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
+++
b/server/tserver/src/main/java/org/apache/accumulo/tserver/log/RecoveryLogsIterator.java
@@ -26,8 +26,7 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
-import java.util.SortedSet;
-import java.util.TreeSet;
+import java.util.TreeMap;
import org.apache.accumulo.core.crypto.CryptoEnvironmentImpl;
import org.apache.accumulo.core.data.Key;
@@ -80,7 +79,7 @@ public class RecoveryLogsIterator
for (Path logDir : recoveryLogDirs) {
LOG.debug("Opening recovery log dir {}", logDir.getName());
- SortedSet<Path> logFiles = getFiles(vm, logDir);
+ TreeMap<Path,FileStatus> logFiles = getFiles(vm, logDir);
var fs = vm.getFileSystemByPath(logDir);
// only check the first key once to prevent extra iterator creation and
seeking
@@ -88,9 +87,9 @@ public class RecoveryLogsIterator
validateFirstKey(context, cryptoService, fs, logFiles, logDir);
}
- for (Path log : logFiles) {
+ for (Entry<Path,FileStatus> entry : logFiles.entrySet()) {
FileSKVIterator fileIter =
FileOperations.getInstance().newReaderBuilder()
- .forFile(log.toString(), fs, fs.getConf(), cryptoService)
+ .forFile(entry.getKey().toString(), fs, fs.getConf(),
cryptoService, entry.getValue())
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build();
if (range != null) {
fileIter.seek(range, Collections.emptySet(), false);
@@ -98,11 +97,13 @@ public class RecoveryLogsIterator
Iterator<Entry<Key,Value>> scanIter = new IteratorAdapter(fileIter);
if (scanIter.hasNext()) {
- LOG.debug("Write ahead log {} has data in range {} {}",
log.getName(), start, end);
+ LOG.debug("Write ahead log {} has data in range {} {}",
entry.getKey().getName(), start,
+ end);
iterators.add(scanIter);
fileIters.add(fileIter);
} else {
- LOG.debug("Write ahead log {} has no data in range {} {}",
log.getName(), start, end);
+ LOG.debug("Write ahead log {} has no data in range {} {}",
entry.getKey().getName(),
+ start, end);
fileIter.close();
}
}
@@ -137,12 +138,12 @@ public class RecoveryLogsIterator
/**
* Check for sorting signal files (finished/failed) and get the logs in the
provided directory.
*/
- private SortedSet<Path> getFiles(VolumeManager fs, Path directory) throws
IOException {
+ private TreeMap<Path,FileStatus> getFiles(VolumeManager fs, Path directory)
throws IOException {
boolean foundFinish = false;
// Path::getName compares the last component of each Path value. In this
case, the last
// component should
// always have the format 'part-r-XXXXX.rf', where XXXXX are one-up values.
- SortedSet<Path> logFiles = new
TreeSet<>(Comparator.comparing(Path::getName));
+ TreeMap<Path,FileStatus> logFiles = new
TreeMap<>(Comparator.comparing(Path::getName));
for (FileStatus child : fs.listStatus(directory)) {
if (child.getPath().getName().startsWith("_")) {
continue;
@@ -156,7 +157,7 @@ public class RecoveryLogsIterator
}
FileSystem ns = fs.getFileSystemByPath(child.getPath());
Path fullLogPath = ns.makeQualified(child.getPath());
- logFiles.add(fullLogPath);
+ logFiles.put(fullLogPath, child);
}
if (!foundFinish) {
throw new IOException(
@@ -169,9 +170,10 @@ public class RecoveryLogsIterator
* Check that the first entry in the WAL is OPEN. Only need to do this once.
*/
private void validateFirstKey(ServerContext context, CryptoService cs,
FileSystem fs,
- SortedSet<Path> logFiles, Path fullLogPath) throws IOException {
+ TreeMap<Path,FileStatus> logFiles, Path fullLogPath) throws IOException {
+ Entry<Path,FileStatus> first = logFiles.firstEntry();
try (FileSKVIterator fileIter =
FileOperations.getInstance().newReaderBuilder()
- .forFile(logFiles.first().toString(), fs, fs.getConf(), cs)
+ .forFile(first.getKey().toString(), fs, fs.getConf(), cs,
first.getValue())
.withTableConfiguration(context.getConfiguration()).seekToBeginning().build()) {
Iterator<Entry<Key,Value>> iterator = new IteratorAdapter(fileIter);
