diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/ReferencedTabletFile.java b/core/src/main/java/org/apache/accumulo/core/metadata/ReferencedTabletFile.java index 2bcae4be9d5..66f2adce2a2 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/ReferencedTabletFile.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/ReferencedTabletFile.java @@ -18,15 +18,10 @@ */ package org.apache.accumulo.core.metadata; -import static org.apache.accumulo.core.Constants.HDFS_TABLES_DIR; - -import java.net.URI; import java.util.Comparator; import java.util.Objects; import org.apache.accumulo.core.data.Range; -import org.apache.accumulo.core.data.TableId; -import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.fs.Path; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,114 +44,7 @@ justification = "Constructor validation is required for proper initialization") public class ReferencedTabletFile extends AbstractTabletFile { - public static class FileParts { - - // parts of an absolute URI, like "hdfs://1.2.3.4/accumulo/tables/2a/t-0003/C0004.rf" - // volume: hdfs://1.2.3.4/accumulo - // tableId: 2a - // tabletDir: t-0003 - // fileName: C0004.rf - // normalizedPath: hdfs://1.2.3.4/accumulo/tables/2a/t-0003/C0004.rf - private final String volume; - private final TableId tableId; - private final String tabletDir; - private final String fileName; - private final String normalizedPath; - - public FileParts(String volume, TableId tableId, String tabletDir, String fileName, - String normalizedPath) { - this.volume = volume; - this.tableId = tableId; - this.tabletDir = tabletDir; - this.fileName = fileName; - this.normalizedPath = normalizedPath; - } - - public String getVolume() { - return volume; - } - - public TableId getTableId() { - return tableId; - } - - public String getTabletDir() { - return tabletDir; - } - - public String getFileName() { - return fileName; - } - - public String getNormalizedPath() { - return normalizedPath; - } - - } - - private static String constructErrorMsg(Path filePath) { - return "Missing or invalid part of tablet file metadata entry: " + filePath; - } - - public static FileParts parsePath(Path filePath) { - // File name construct: //// - // Example: hdfs://namenode:9020/accumulo/tables/1/default_tablet/F00001.rf - // Example compaction tmp file: - // hdfs://namenode:9020/accumulo/tables/1/default_tablet/F00001.rf_tmp_ECID- - final URI uri = filePath.toUri(); - - // validate that this is a fully qualified uri - Preconditions.checkArgument(uri.getScheme() != null, constructErrorMsg(filePath)); - - final String path = uri.getPath(); // ex: /accumulo/tables/1/default_tablet/F00001.rf - final String[] parts = path.split("/"); - final int numParts = parts.length; // should contain tables, 1, default_tablet, F00001.rf - - if (numParts < 4) { - throw new IllegalArgumentException(constructErrorMsg(filePath)); - } - - final String fileName = parts[numParts - 1]; - final String tabletDirectory = parts[numParts - 2]; - final TableId tableId = TableId.of(parts[numParts - 3]); - final String tablesPath = parts[numParts - 4]; - - // determine where file path starts, the rest is the volume - final String computedFilePath = - HDFS_TABLES_DIR + "/" + tableId.canonical() + "/" + tabletDirectory + "/" + fileName; - final String uriString = uri.toString(); - int idx = uriString.lastIndexOf(computedFilePath); - - if (idx == -1) { - throw new IllegalArgumentException(constructErrorMsg(filePath)); - } - - // The volume is the beginning portion of the uri up to the start - // of the file path. - final String volume = uriString.substring(0, idx); - - if (StringUtils.isBlank(fileName) || StringUtils.isBlank(tabletDirectory) - || StringUtils.isBlank(tablesPath) || StringUtils.isBlank(volume)) { - throw new IllegalArgumentException(constructErrorMsg(filePath)); - } - ValidationUtil.validateFileName(fileName); - Preconditions.checkArgument(tablesPath.equals(HDFS_TABLES_DIR_NAME), - "tables directory name is not " + HDFS_TABLES_DIR_NAME + ", is " + tablesPath); - - final String normalizedPath = volume + computedFilePath; - - if (!normalizedPath.equals(uriString)) { - throw new RuntimeException("Error parsing file path, " + normalizedPath + " != " + uriString); - } - - return new FileParts(volume, tableId, tabletDirectory, fileName, normalizedPath); - - } - - private final FileParts parts; - private static final Logger log = LoggerFactory.getLogger(ReferencedTabletFile.class); - private static final String HDFS_TABLES_DIR_NAME = HDFS_TABLES_DIR.substring(1); private static final Comparator comparator = Comparator.comparing(ReferencedTabletFile::getNormalizedPathStr) @@ -171,26 +59,19 @@ public ReferencedTabletFile(Path metaPath) { * qualify an absolute path or create a new file. */ public ReferencedTabletFile(Path metaPath, Range range) { - super(Objects.requireNonNull(metaPath), range); - log.trace("Parsing TabletFile from {}", metaPath); - parts = parsePath(metaPath); - } - - public String getVolume() { - return parts.getVolume(); - } - - public TableId getTableId() { - return parts.getTableId(); + super(validatePath(metaPath), range); + log.trace("Creating tablet file reference from {}", metaPath); } - public String getTabletDir() { - return parts.getTabletDir(); + private static Path validatePath(Path metaPath) { + Path path = ValidationUtil.validate(Objects.requireNonNull(metaPath)); + Preconditions.checkArgument(!path.getName().isEmpty(), "Path must identify a file: %s", path); + return path; } @Override public String getFileName() { - return parts.getFileName(); + return path.getName(); } /** @@ -198,7 +79,7 @@ public String getFileName() { * metadata. */ public String getNormalizedPathStr() { - return parts.getNormalizedPath(); + return path.toString(); } /** @@ -216,20 +97,19 @@ public int compareTo(ReferencedTabletFile o) { @Override public boolean equals(Object obj) { if (obj instanceof ReferencedTabletFile that) { - return parts.getNormalizedPath().equals(that.parts.getNormalizedPath()) - && range.equals(that.range); + return path.equals(that.path) && range.equals(that.range); } return false; } @Override public int hashCode() { - return Objects.hash(parts.getNormalizedPath(), range); + return Objects.hash(path, range); } @Override public String toString() { - return parts.getNormalizedPath(); + return path.toString(); } public static ReferencedTabletFile of(final Path path) { diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFile.java b/core/src/main/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFile.java index c289c38a47e..5da7394b1a1 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFile.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFile.java @@ -19,37 +19,39 @@ package org.apache.accumulo.core.metadata; import java.net.URI; +import java.util.Collection; +import java.util.HashSet; import java.util.Objects; +import java.util.Set; +import java.util.TreeSet; import java.util.UUID; import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.data.Value; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; public class ScanServerRefTabletFile extends ReferencedTabletFile { - private final Value NULL_VALUE = new Value(new byte[0]); private final Text serverAddress; private final Text uuid; + private final Set tableIds; - public ScanServerRefTabletFile(UUID serverLockUUID, String serverAddress, String file) { + public ScanServerRefTabletFile(UUID serverLockUUID, String serverAddress, String file, + Collection tableIds) { super(new Path(URI.create(file))); this.serverAddress = new Text(serverAddress); uuid = new Text(serverLockUUID.toString()); + this.tableIds = copyAndValidateTableIds(tableIds); } - public ScanServerRefTabletFile(String file, String serverAddress, UUID serverLockUUID) { - super(new Path(URI.create(file))); - this.serverAddress = new Text(serverAddress); - this.uuid = new Text(serverLockUUID.toString()); - } - - public ScanServerRefTabletFile(Key k) { + public ScanServerRefTabletFile(Key k, Value value) { super(new Path(URI.create(k.getColumnQualifier().toString()))); serverAddress = k.getColumnFamily(); uuid = k.getRow(); + tableIds = decodeTableIds(Objects.requireNonNull(value).toString()); } public Mutation putMutation() { @@ -73,7 +75,42 @@ public UUID getServerLockUUID() { } public Value getValue() { - return NULL_VALUE; + TreeSet sortedIds = new TreeSet<>(); + for (TableId tableId : tableIds) { + sortedIds.add(tableId.canonical()); + } + return new Value(String.join(",", sortedIds)); + } + + public Set getTableIds() { + return Set.copyOf(tableIds); + } + + private static Set decodeTableIds(String serialized) { + if (serialized.isEmpty()) { + throw new IllegalArgumentException("Serialized table IDs must not be empty"); + } + + Set tableIds = new HashSet<>(); + for (String tableId : serialized.split(",")) { + if (tableId.isEmpty()) { + throw new IllegalArgumentException("Serialized table IDs must not contain empty IDs"); + } + tableIds.add(TableId.of(tableId)); + } + return copyAndValidateTableIds(tableIds); + } + + private static Set copyAndValidateTableIds(Collection tableIds) { + Objects.requireNonNull(tableIds, "Table IDs must be supplied"); + Set copy = new HashSet<>(tableIds); + if (copy.isEmpty()) { + throw new IllegalArgumentException("Table IDs must not be empty"); + } + for (TableId tableId : copy) { + Objects.requireNonNull(tableId, "Table IDs must not contain null"); + } + return copy; } public Text getServerAddress() { diff --git a/core/src/main/java/org/apache/accumulo/core/metadata/StoredTabletFile.java b/core/src/main/java/org/apache/accumulo/core/metadata/StoredTabletFile.java index b1f80ccd75a..08ab90a7b8c 100644 --- a/core/src/main/java/org/apache/accumulo/core/metadata/StoredTabletFile.java +++ b/core/src/main/java/org/apache/accumulo/core/metadata/StoredTabletFile.java @@ -30,7 +30,6 @@ import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.Range; -import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.util.json.ByteArrayToBase64TypeAdapter; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.DataInputBuffer; @@ -109,10 +108,6 @@ public ReferencedTabletFile getTabletFile() { return referencedTabletFile; } - public TableId getTableId() { - return referencedTabletFile.getTableId(); - } - @Override public String getFileName() { return referencedTabletFile.getFileName(); diff --git a/core/src/test/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFileTest.java b/core/src/test/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFileTest.java new file mode 100644 index 00000000000..c225b4a7422 --- /dev/null +++ b/core/src/test/java/org/apache/accumulo/core/metadata/ScanServerRefTabletFileTest.java @@ -0,0 +1,66 @@ +/* + * 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.metadata; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Set; +import java.util.UUID; + +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.data.Value; +import org.apache.hadoop.io.Text; +import org.junit.jupiter.api.Test; + +public class ScanServerRefTabletFileTest { + + @Test + public void tableIdsRoundTripForExternalPaths() { + UUID serverLockUuid = UUID.randomUUID(); + String serverAddress = "localhost:1234"; + String file = "hdfs://nn/external/rfiles/source.rf"; + Set tableIds = Set.of(TableId.of("2a"), TableId.of("3b")); + var reference = new ScanServerRefTabletFile(serverLockUuid, serverAddress, file, tableIds); + var key = new Key(new Text(serverLockUuid.toString()), new Text(serverAddress), + reference.getFilePath(), new Text(), 1); + + var decoded = new ScanServerRefTabletFile(key, reference.getValue()); + + assertEquals(file, decoded.getNormalizedPathStr()); + assertEquals(tableIds, decoded.getTableIds()); + } + + @Test + public void tableIdsAreRequired() { + String serverAddress = "localhost:1234"; + String file = "hdfs://nn/external/rfiles/source.rf"; + + assertThrows(NullPointerException.class, + () -> new ScanServerRefTabletFile(UUID.randomUUID(), serverAddress, file, null)); + assertThrows(IllegalArgumentException.class, + () -> new ScanServerRefTabletFile(UUID.randomUUID(), serverAddress, file, Set.of())); + + var key = new Key(new Text(UUID.randomUUID().toString()), new Text(serverAddress), + new Text(file), new Text(), 1); + assertThrows(IllegalArgumentException.class, + () -> new ScanServerRefTabletFile(key, new Value())); + } +} diff --git a/core/src/test/java/org/apache/accumulo/core/metadata/schema/ReferencedTabletFileTest.java b/core/src/test/java/org/apache/accumulo/core/metadata/schema/ReferencedTabletFileTest.java index 04c38f9f9b8..78ebe76db88 100644 --- a/core/src/test/java/org/apache/accumulo/core/metadata/schema/ReferencedTabletFileTest.java +++ b/core/src/test/java/org/apache/accumulo/core/metadata/schema/ReferencedTabletFileTest.java @@ -25,7 +25,6 @@ import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.PartialKey; import org.apache.accumulo.core.data.Range; -import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.metadata.ReferencedTabletFile; import org.apache.accumulo.core.metadata.StoredTabletFile; import org.apache.hadoop.fs.Path; @@ -34,94 +33,46 @@ public class ReferencedTabletFileTest { - private ReferencedTabletFile test(String metadataFile, String volume, String tableId, - String tabletDir, String fileName) { + private ReferencedTabletFile test(String metadataFile, String fileName) { String metadataPath = StoredTabletFile.serialize(metadataFile); StoredTabletFile storedTabletFile = new StoredTabletFile(metadataPath); ReferencedTabletFile tabletFile = storedTabletFile.getTabletFile(); // Make sure original file name wasn't changed when serialized assertTrue(metadataPath.contains(metadataFile)); - assertEquals(volume, tabletFile.getVolume()); assertEquals(metadataPath, storedTabletFile.getMetadata()); - assertEquals(TableId.of(tableId), tabletFile.getTableId()); - assertEquals(tabletDir, tabletFile.getTabletDir()); assertEquals(fileName, tabletFile.getFileName()); return tabletFile; } @Test public void testValidPaths() { - test("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf", - "hdfs://localhost:8020/accumulo", "2a", "default_tablet", "F0000070.rf"); - test("hdfs://nn1:9000/accumulo/tables/5a/t-0005/C0009.rf", "hdfs://nn1:9000/accumulo", "5a", - "t-0005", "C0009.rf"); - test( - "file:/home/dude/workspace/accumulo/test/target/mini-tests/org.apache.accumulo.test.VolumeIT_test/volumes/v1/tables/1/t-0000003/F0000006.rf", - "file:/home/dude/workspace/accumulo/test/target/mini-tests/org.apache.accumulo.test.VolumeIT_test/volumes/v1", - "1", "t-0000003", "F0000006.rf"); + test("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf", "F0000070.rf"); + test("hdfs://nn1:9000/unrelated/external/location/C0009.rf", "C0009.rf"); + test("file:/home/dude/external-data/F0000006.rf", "F0000006.rf"); } @Test public void testBadPaths() { - // 2a< srv:dir - final String message = "Failed to throw error on bad path"; - - assertThrows(IllegalArgumentException.class, - () -> test("C0004.rf", "", "2a", "t-0003", "C0004.rf"), message); - assertThrows(IllegalArgumentException.class, () -> test("dir", "", "2a", "", ""), message); - - assertThrows(IllegalArgumentException.class, - () -> test("hdfs://localhost:8020/accumulo/tablets/2a/default_tablet/F0000070.rf", - "hdfs://localhost:8020/accumulo", "2a", "default_tablet", "F0000070.rf"), - message); - assertThrows(IllegalArgumentException.class, - () -> test("hdfs://localhost:8020/accumulo/2a/default_tablet/F0000070.rf", - " hdfs://localhost:8020/accumulo", "2a", "default_tablet", " F0000070.rf"), - message); - assertThrows(IllegalArgumentException.class, - () -> test("/accumulo/tables/2a/default_tablet/F0000070.rf", "", "2a", "default_tablet", - "F0000070.rf"), - message); assertThrows(IllegalArgumentException.class, - () -> test("hdfs://localhost:8020/accumulo/tables/2a/F0000070.rf", - "hdfs://localhost:8020/accumulo", "2a", "", "F0000070.rf"), - message); - assertThrows(IllegalArgumentException.class, - () -> test("hdfs://localhost:8020/accumulo/tables/F0000070.rf", - "hdfs://localhost:8020/accumulo", null, "", "F0000070.rf"), - message); - - } - - private final String id = "2a"; - private final String dir = "t-0003"; - private final String filename = "C0004.rf"; - - @Test - public void testFullPathWithVolume() { - String volume = "hdfs://1.2.3.4/accumulo"; - String metadataEntry = volume + "/tables/" + id + "/" + dir + "/" + filename; - test(metadataEntry, volume, id, dir, filename); + () -> new ReferencedTabletFile(new Path("C0004.rf"))); + assertThrows(IllegalArgumentException.class, () -> new ReferencedTabletFile(new Path("dir"))); } @Test public void testNormalizePath() { - String uglyVolume = "hdfs://nn.somewhere.com:86753/accumulo/blah/.././/bad/bad2/../.././/////"; - String metadataEntry = uglyVolume + "/tables/" + id + "/" + dir + "/" + filename; - ReferencedTabletFile uglyFile = - test(metadataEntry, "hdfs://nn.somewhere.com:86753/accumulo", id, dir, filename); + String uglyPath = + "hdfs://nn.somewhere.com:86753/external/blah/.././/bad/bad2/../.././/////file.rf"; + ReferencedTabletFile uglyFile = test(uglyPath, "file.rf"); ReferencedTabletFile niceFile = StoredTabletFile - .of(new Path( - "hdfs://nn.somewhere.com:86753/accumulo/tables/" + id + "/" + dir + "/" + filename)) - .getTabletFile(); + .of(new Path("hdfs://nn.somewhere.com:86753/external/file.rf")).getTabletFile(); assertEquals(niceFile, uglyFile); assertEquals(niceFile.hashCode(), uglyFile.hashCode()); } @Test public void testNonRowRange() { - Path testPath = new Path("hdfs://localhost:8020/accumulo/tables/2a/default_tablet/F0000070.rf"); + Path testPath = new Path("hdfs://localhost:8020/outside/F0000070.rf"); // range where start key is not a row key Range r1 = new Range(new Key("r1", "f1"), true, null, false); diff --git a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java index e41237b087b..9932a26e2fb 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/fs/VolumeUtil.java @@ -62,23 +62,39 @@ public static Path removeTrailingSlash(Path path) { } public static Path switchVolume(Path path, FileType ft, Map replacements) { - Path replacement = null; - if (!replacements.isEmpty()) { - // removing trailing slash for exact match comparison on the volume itself - Path volume = removeTrailingSlash(ft.getVolume(requireNonNull(path))); - replacement = replacements.entrySet().stream() - .filter(entry -> removeTrailingSlash(entry.getKey()).equals(volume)) - .map(entry -> new Path(entry.getValue(), requireNonNull(ft.removeVolume(path)))) - .findFirst().orElse(null); - } - if (replacement != null) { - log.trace("Replacing {} with {} for {}", path, replacement, ft); - return replacement; + requireNonNull(path); + for (Entry entry : replacements.entrySet()) { + Path source = removeTrailingSlash(entry.getKey()); + Path relative = getRelativePath(source, path); + if (relative != null) { + Path replacement = relative.toString().equals(".") ? entry.getValue() + : new Path(entry.getValue(), relative); + log.trace("Replacing {} with {} for {}", path, replacement, ft); + return replacement; + } } log.trace("No replacement available for {} at {}", ft, path); return null; } + private static Path getRelativePath(Path parent, Path path) { + ArrayList names = new ArrayList<>(); + Path current = path; + while (current != null && !current.equals(parent)) { + names.add(current.getName()); + current = current.getParent(); + } + if (current == null) { + return null; + } + + Path relative = null; + for (int i = names.size() - 1; i >= 0; i--) { + relative = relative == null ? new Path(names.get(i)) : new Path(relative, names.get(i)); + } + return relative == null ? new Path(".") : relative; + } + public static LogEntry switchVolume(LogEntry le, Map replacements) { Path switchedPath = switchVolume(new Path(le.getPath()), FileType.WAL, replacements); return switchedPath == null ? null : LogEntry.fromPath(switchedPath.toString()); diff --git a/server/base/src/main/java/org/apache/accumulo/server/metadata/ScanServerRefStoreImpl.java b/server/base/src/main/java/org/apache/accumulo/server/metadata/ScanServerRefStoreImpl.java index 7b7dd77aabd..148352ccf0f 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/metadata/ScanServerRefStoreImpl.java +++ b/server/base/src/main/java/org/apache/accumulo/server/metadata/ScanServerRefStoreImpl.java @@ -67,7 +67,7 @@ public Stream list() { try { Scanner scanner = context.createScanner(tableName, Authorizations.EMPTY); return scanner.stream().onClose(scanner::close) - .map(e -> new ScanServerRefTabletFile(e.getKey())); + .map(e -> new ScanServerRefTabletFile(e.getKey(), e.getValue())); } catch (TableNotFoundException e) { throw new IllegalStateException(tableName + " not found!", e); } @@ -81,7 +81,8 @@ public void delete(String serverAddress, UUID scanServerLockUUID) { scanner.setRange(new Range(scanServerLockUUID.toString())); Set refsToDelete = StreamSupport.stream(scanner.spliterator(), false) - .map(e -> new ScanServerRefTabletFile(e.getKey())).collect(Collectors.toSet()); + .map(e -> new ScanServerRefTabletFile(e.getKey(), e.getValue())) + .collect(Collectors.toSet()); if (!refsToDelete.isEmpty()) { this.delete(refsToDelete); diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java index b73eac09149..208243af6c0 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/MetadataTableUtil.java @@ -76,6 +76,7 @@ import org.apache.accumulo.core.util.Pair; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.gc.AllVolumesDirectory; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -110,30 +111,33 @@ public static void deleteTable(TableId tableId, boolean insertDeletes, ServerCon Ample ample = context.getAmple(); ms.setRange(new KeyExtent(tableId, null, null).toMetaRange()); - // insert deletes before deleting data from metadata... this makes the code fault-tolerant + // Insert delete markers before deleting data from metadata. When table deletion is + // optimized, files outside the table directories in configured volumes still need explicit + // candidates because deleting those directories cannot remove them. + ms.fetchColumnFamily(DataFileColumnFamily.NAME); if (insertDeletes) { - - ms.fetchColumnFamily(DataFileColumnFamily.NAME); ServerColumnFamily.DIRECTORY_COLUMN.fetch(ms); + } - for (Entry cell : ms) { - Key key = cell.getKey(); + for (Entry cell : ms) { + Key key = cell.getKey(); - if (key.getColumnFamily().equals(DataFileColumnFamily.NAME)) { - StoredTabletFile stf = new StoredTabletFile(key.getColumnQualifierData().toString()); + if (key.getColumnFamily().equals(DataFileColumnFamily.NAME)) { + StoredTabletFile stf = new StoredTabletFile(key.getColumnQualifierData().toString()); + if (insertDeletes || !isInTableDirectories(context, tableId, stf.getPath())) { bw.addMutation(ample.createDeleteMutation(ReferenceFile.forFile(tableId, stf))); } + } - if (ServerColumnFamily.DIRECTORY_COLUMN.hasColumns(key)) { - var uri = new AllVolumesDirectory(tableId, cell.getValue().toString()); - bw.addMutation(ample.createDeleteMutation(uri)); - } + if (insertDeletes && ServerColumnFamily.DIRECTORY_COLUMN.hasColumns(key)) { + var uri = new AllVolumesDirectory(tableId, cell.getValue().toString()); + bw.addMutation(ample.createDeleteMutation(uri)); } + } - bw.flush(); + bw.flush(); - ms.clearColumns(); - } + ms.clearColumns(); for (Entry cell : ms) { Key key = cell.getKey(); @@ -161,6 +165,20 @@ public static void deleteTable(TableId tableId, boolean insertDeletes, ServerCon } } + private static boolean isInTableDirectories(ServerContext context, TableId tableId, Path file) { + for (String tablesDir : context.getTablesDirs()) { + Path tableDir = new Path(tablesDir, tableId.canonical()); + Path parent = file; + while (parent != null) { + if (parent.equals(tableDir)) { + return true; + } + parent = parent.getParent(); + } + } + return false; + } + public static Pair,SortedMap> getFileAndLogEntries(ServerContext context, KeyExtent extent) throws IOException { TreeMap sizes = new TreeMap<>(); diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java b/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java index b0aebbb2288..93136c11d29 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/TableDiskUsage.java @@ -194,7 +194,7 @@ public static Map,Long> getDiskUsage(Set tableIds, // read the metadata try (var tabletsMetadata = ((ClientContext) client).getAmple().readTablets().forTable(tableId) .fetch(FILES).build()) { - final Set allFiles = new HashSet<>(); + final Set allFiles = new HashSet<>(); for (var tm : tabletsMetadata) { final Map tmFiles = tm.getFilesMap(); @@ -202,27 +202,17 @@ public static Map,Long> getDiskUsage(Set tableIds, for (var file : tmFiles.entrySet()) { final var stf = file.getKey(); final var dataFileValue = file.getValue(); - - // get the table referenced by the file which may not be the same as the current - // table we are scanning if the file is shared between multiple tables - final TableId fileTableRef = stf.getTableId(); - - // if this is a ref to a different table than the one we are scanning then we need - // to make sure the table is also linked for this shared file if the table is - // part of the set of tables we are running du on so we can track shared usages - if (!fileTableRef.equals(tableId) && tableIds.contains(fileTableRef)) { - // link the table and the shared file for computing shared sizes - tdu.linkFileAndTable(fileTableRef, stf.getFileName()); - } - - // link the file to the table we are scanning for - tdu.linkFileAndTable(tableId, stf.getFileName()); + // The metadata row identifies the table that references this file. Use the full path + // to identify shared files because their URI does not encode a table ID and basenames + // are not necessarily unique. + final String filePath = stf.getNormalizedPathStr(); + tdu.linkFileAndTable(tableId, filePath); // add the file size for the table if not already seen for this scan - if (allFiles.add(stf)) { + if (allFiles.add(filePath)) { // This tracks the file size for individual files for computing shared file statistics // later - tdu.addFileSize(stf.getFileName(), dataFileValue.getSize()); + tdu.addFileSize(filePath, dataFileValue.getSize()); } } } diff --git a/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java b/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java index 90d585b6748..62b16d4669b 100644 --- a/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java +++ b/server/base/src/main/java/org/apache/accumulo/server/util/adminCommand/ListVolumesUsed.java @@ -72,11 +72,18 @@ public void execute(JCommander cl, ServerOpts options) throws Exception { } private String getTableURI(String rootTabletDir) { - Path ret = FileType.TABLE.getVolume(new Path(rootTabletDir)); - if (ret == null) { + Path path = new Path(rootTabletDir); + try { + Path ret = FileType.TABLE.getVolume(path); + if (ret != null) { + return ret.toString(); + } return "RELATIVE"; + } catch (IllegalArgumentException e) { + // External file references need not contain the Accumulo tables directory. Include their + // full URI so this command remains useful without inventing a volume root. + return "EXTERNAL: " + path; } - return ret.toString(); } private String getLogURI(String logEntry) { diff --git a/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java b/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java index c9beb51c06a..ca46797556f 100644 --- a/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java +++ b/server/base/src/test/java/org/apache/accumulo/server/constraints/MetadataConstraintsTest.java @@ -315,14 +315,15 @@ public void testBulkFileCheck() { new Value(fateId1.canonical())); assertViolation(mc, m, (short) 3100); - // Missing tables directory in path + // External files do not need to be in the tables directory m = new Mutation(new Text("0;foo")); - m.put(BulkFileColumnFamily.NAME, + Text externalFileMetadata = new Text(StoredTabletFile.of(new Path("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile")) .getMetadata().replace("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile", - "hdfs://1.2.3.4/accumulo/2a/t-0003/someFile")), - new Value(fateId1.canonical())); - assertViolation(mc, m, (short) 3100); + "hdfs://1.2.3.4/accumulo/2a/t-0003/someFile")); + m.put(BulkFileColumnFamily.NAME, externalFileMetadata, new Value(fateId1.canonical())); + m.put(DataFileColumnFamily.NAME, externalFileMetadata, new DataFileValue(1, 1).encodeAsValue()); + assertTrue(mc.check(createEnv(), m).isEmpty()); m = new Mutation(new Text("0;foo")); m.put( @@ -477,14 +478,14 @@ columnFamily, new Text(StoredTabletFile value); assertViolation(mc, m, (short) 3100); - // Missing tables directory in path + // External files do not need to be in the tables directory m = new Mutation(new Text("0;foo")); m.put(columnFamily, new Text(StoredTabletFile.of(new Path("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile")) .getMetadata().replace("hdfs://1.2.3.4/accumulo/tables/2a/t-0003/someFile", "hdfs://1.2.3.4/accumulo/2a/t-0003/someFile")), new DataFileValue(1, 1).encodeAsValue()); - assertViolation(mc, m, (short) 3100); + assertTrue(mc.check(createEnv(), m).isEmpty()); // Should pass validation (inf range) m = new Mutation(new Text("0;foo")); diff --git a/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java b/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java index 033cf172125..c144ea4bb8b 100644 --- a/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java +++ b/server/base/src/test/java/org/apache/accumulo/server/fs/VolumeUtilTest.java @@ -52,6 +52,10 @@ public void testSwitchVolume() { FileType.TABLE, replacements)); assertNull(VolumeUtil.switchVolume(new Path("file:/nn1/a/accumulo/tables/t-00000/C000.rf"), FileType.TABLE, replacements)); + assertEquals(new Path("viewfs:/a/accumulo/external/C000.rf"), VolumeUtil.switchVolume( + new Path("hdfs://nn1/accumulo/external/C000.rf"), FileType.TABLE, replacements)); + assertNull(VolumeUtil.switchVolume(new Path("hdfs://nn1/accumulo-sibling/external/C000.rf"), + FileType.TABLE, replacements)); replacements.clear(); replacements.put(new Path("hdfs://nn1/d1/accumulo"), new Path("viewfs:/a/accumulo")); diff --git a/server/base/src/test/java/org/apache/accumulo/server/util/TableDiskUsageTest.java b/server/base/src/test/java/org/apache/accumulo/server/util/TableDiskUsageTest.java index cab48d722fe..822e31a5d0b 100644 --- a/server/base/src/test/java/org/apache/accumulo/server/util/TableDiskUsageTest.java +++ b/server/base/src/test/java/org/apache/accumulo/server/util/TableDiskUsageTest.java @@ -78,11 +78,11 @@ public void testSingleTableMultipleTablets() throws Exception { final TabletsMetadata mockTabletsMetadata = mockTabletsMetadata(client, tableId1); List realTabletsMetadata = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0001.rf"), 1024); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0002.rf"), 1024); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName2, "C0003.rf"), 2048); mockTabletsMetadataIter(mockTabletsMetadata, realTabletsMetadata.iterator()); @@ -108,13 +108,13 @@ public void testMultipleVolumes() throws Exception { final TabletsMetadata mockTabletsMetadata = mockTabletsMetadata(client, tableId1); List realTabletsMetadata = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0001.rf"), 1024); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0002.rf"), 1024); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume2, tableId1, tabletName2, "C0003.rf"), 2048); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume2, tableId1, tabletName2, "C0004.rf"), 10000); mockTabletsMetadataIter(mockTabletsMetadata, realTabletsMetadata.iterator()); @@ -140,8 +140,8 @@ public void testMetadataTable() throws Exception { mockTabletsMetadata(client, SystemTables.METADATA.tableId()); List realTabletsMetadata = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata, getTabletFile(volume1, SystemTables.METADATA.tableId(), - SystemTables.METADATA.tableName(), "C0001.rf"), 1024); + appendFileMetadata(realTabletsMetadata, SystemTables.METADATA.tableId(), getTabletFile(volume1, + SystemTables.METADATA.tableId(), SystemTables.METADATA.tableName(), "C0001.rf"), 1024); mockTabletsMetadataIter(mockTabletsMetadata, realTabletsMetadata.iterator()); EasyMock.replay(client, mockTabletsMetadata); @@ -165,9 +165,9 @@ public void testDuplicateFile() throws Exception { final TabletsMetadata mockTabletsMetadata = mockTabletsMetadata(client, tableId1); List realTabletsMetadata = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0001.rf"), 1024); - appendFileMetadata(realTabletsMetadata, + appendFileMetadata(realTabletsMetadata, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0001.rf"), 1024); mockTabletsMetadataIter(mockTabletsMetadata, realTabletsMetadata.iterator()); @@ -216,30 +216,30 @@ public void testMultipleTables() throws Exception { final TabletsMetadata mockTabletsMetadata1 = mockTabletsMetadata(client, tableId1); List realTabletsMetadata1 = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata1, + appendFileMetadata(realTabletsMetadata1, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0001.rf"), 1024); - appendFileMetadata(realTabletsMetadata1, + appendFileMetadata(realTabletsMetadata1, tableId1, getTabletFile(volume1, tableId1, tabletName1, "C0002.rf"), 4096); mockTabletsMetadataIter(mockTabletsMetadata1, realTabletsMetadata1.iterator()); final TabletsMetadata mockTabletsMetadata2 = mockTabletsMetadata(client, tableId2); List realTabletsMetadata2 = new ArrayList<>(); - appendFileMetadata(realTabletsMetadata2, + appendFileMetadata(realTabletsMetadata2, tableId2, getTabletFile(volume1, tableId2, tabletName2, "C0003.rf"), 2048); - appendFileMetadata(realTabletsMetadata2, + appendFileMetadata(realTabletsMetadata2, tableId2, getTabletFile(volume1, tableId2, tabletName2, "C0004.rf"), 3000); mockTabletsMetadataIter(mockTabletsMetadata2, realTabletsMetadata2.iterator()); final TabletsMetadata mockTabletsMetadata3 = mockTabletsMetadata(client, tableId3); List realTabletsMetadata3 = new ArrayList<>(); // shared file - appendFileMetadata(realTabletsMetadata3, + appendFileMetadata(realTabletsMetadata3, tableId3, getTabletFile(volume1, tableId2, tabletName2, "C0003.rf"), 2048); - appendFileMetadata(realTabletsMetadata3, + appendFileMetadata(realTabletsMetadata3, tableId3, getTabletFile(volume1, tableId3, tabletName3, "C0005.rf"), 84520); - appendFileMetadata(realTabletsMetadata3, + appendFileMetadata(realTabletsMetadata3, tableId3, getTabletFile(volume1, tableId3, tabletName3, "C0006.rf"), 3000); - appendFileMetadata(realTabletsMetadata3, + appendFileMetadata(realTabletsMetadata3, tableId3, getTabletFile(volume1, tableId3, tabletName4, "C0007.rf"), 98456); mockTabletsMetadataIter(mockTabletsMetadata3, realTabletsMetadata3.iterator()); @@ -268,6 +268,44 @@ public void testMultipleTables() throws Exception { EasyMock.verify(client, mockTabletsMetadata1, mockTabletsMetadata2, mockTabletsMetadata3); } + @Test + public void testExternalFilesAreSharedByFullPath() throws Exception { + final ClientContext client = EasyMock.createMock(ClientContext.class); + EasyMock.expect(client.createTableIdToQualifiedNameMap()).andReturn(tableIdToNameMap); + + final TabletsMetadata mockTabletsMetadata1 = mockTabletsMetadata(client, tableId1); + final List table1Metadata = new ArrayList<>(); + appendFileMetadata(table1Metadata, tableId1, + new StoredTabletFile(StoredTabletFile.serialize("hdfs://nn/external/files/shared.rf")), + 100); + mockTabletsMetadataIter(mockTabletsMetadata1, table1Metadata.iterator()); + + final TabletsMetadata mockTabletsMetadata2 = mockTabletsMetadata(client, tableId2); + final List table2Metadata = new ArrayList<>(); + appendFileMetadata(table2Metadata, tableId2, + new StoredTabletFile(StoredTabletFile.serialize("hdfs://nn/external/files/shared.rf")), + 100); + mockTabletsMetadataIter(mockTabletsMetadata2, table2Metadata.iterator()); + + final TabletsMetadata mockTabletsMetadata3 = mockTabletsMetadata(client, tableId3); + final List table3Metadata = new ArrayList<>(); + appendFileMetadata(table3Metadata, tableId3, + new StoredTabletFile(StoredTabletFile.serialize("hdfs://other/external/files/shared.rf")), + 200); + mockTabletsMetadataIter(mockTabletsMetadata3, table3Metadata.iterator()); + + EasyMock.replay(client, mockTabletsMetadata1, mockTabletsMetadata2, mockTabletsMetadata3); + + Map,Long> result = + TableDiskUsage.getDiskUsage(tableSet(tableId1, tableId2, tableId3), client); + + assertEquals(100, result.get(tableNameSet(tableId1, tableId2))); + assertEquals(200, result.get(tableNameSet(tableId3))); + assertEquals(2, result.size()); + + EasyMock.verify(client, mockTabletsMetadata1, mockTabletsMetadata2, mockTabletsMetadata3); + } + private static TreeSet tableNameSet(TableId... tableIds) { return Set.of(tableIds).stream().map(TableDiskUsageTest::getTableName) .collect(Collectors.toCollection(TreeSet::new)); @@ -289,9 +327,9 @@ private static String getTableName(TableId tableId) { return tableIdToNameMap.get(tableId); } - private static void appendFileMetadata(List realTabletsMetadata, + private static void appendFileMetadata(List realTabletsMetadata, TableId tableId, StoredTabletFile file, long size) { - TabletMetadata tm = TabletMetadata.builder(new KeyExtent(file.getTableId(), null, null)) + TabletMetadata tm = TabletMetadata.builder(new KeyExtent(tableId, null, null)) .putFile(file, new DataFileValue(size, 1)).build(); realTabletsMetadata.add(tm); } diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java b/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java index f13cff3bd1e..48e4575288d 100644 --- a/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java +++ b/server/gc/src/main/java/org/apache/accumulo/gc/GCRun.java @@ -35,7 +35,6 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Set; import java.util.SortedMap; import java.util.concurrent.ThreadPoolExecutor; @@ -69,7 +68,6 @@ import org.apache.accumulo.core.security.Authorizations; import org.apache.accumulo.core.util.Timer; import org.apache.accumulo.core.util.threads.ThreadPools; -import org.apache.accumulo.core.volume.Volume; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.fs.VolumeManager; import org.apache.accumulo.server.fs.VolumeUtil; @@ -221,8 +219,8 @@ public Stream getReferences() { return fileStream; }); - var scanServerRefs = context.getAmple().scanServerRefs().list() - .map(sfr -> ReferenceFile.forScan(sfr.getTableId(), sfr)); + var scanServerRefs = context.getAmple().scanServerRefs().list().flatMap( + sfr -> sfr.getTableIds().stream().map(tableId -> ReferenceFile.forScan(tableId, sfr))); return Stream.concat(tabletReferences, scanServerRefs); } @@ -339,21 +337,7 @@ public void deleteConfirmedCandidates(SortedMap confirmedDel // this failure, we still want to remove the metadata entry removeFlag = true; errors.increment(); - 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]; - TableState tableState = context.getTableManager().getTableState(tableId); - if (tableState != TableState.UNKNOWN && tableState != TableState.DELETING) { - // clone directories don't always exist - if (!tabletDir.startsWith(Constants.CLONE_PREFIX)) { - log.debug("{} File doesn't exist: {}", fileActionPrefix, pathToDel); - } - } - } else { - log.warn("{} Delete failed due to invalid file path format: {}", fileActionPrefix, - delete.getPath()); - } + log.debug("{} File doesn't exist: {}", fileActionPrefix, pathToDel); } } @@ -426,57 +410,34 @@ public void incrementInUseStat(long i) { static void minimizeDeletes(SortedMap confirmedDeletes, List processedDeletes, VolumeManager fs, Logger logger, Duration loggingInterval) { - Set seenVolumes = new HashSet<>(); + Set deleteDirs = new HashSet<>(); + for (Map.Entry entry : confirmedDeletes.entrySet()) { + if (entry.getKey().startsWith(GarbageCollectionAlgorithm.DIRECTORY_CANDIDATE_KEY_PREFIX)) { + Path dirPath = new Path(entry.getValue().getPath()); + deleteDirs.addAll(GcVolumeUtil.expandAllVolumesUri(fs, dirPath)); + } + } - // when deleting a dir and all files in that dir, only need to delete the dir. - // The dir will sort right before the files... so remove the files in this case - // to minimize namenode ops + // When a directory and its files are all being deleted, deleting the directory is sufficient. Iterator> cdIter = confirmedDeletes.entrySet().iterator(); - - String lastDirRel = null; - Path lastDirAbs = null; Timer progressTimer = Timer.startNew(); int progressCount = 0; int totalDeletes = confirmedDeletes.size(); while (cdIter.hasNext()) { progressCount++; Map.Entry entry = cdIter.next(); - String relPath = entry.getKey(); - Path absPath = new Path(entry.getValue().getPath()); - - if (SimpleGarbageCollector.isDir(relPath)) { - lastDirRel = relPath; - lastDirAbs = absPath; - } else if (lastDirRel != null) { - if (relPath.startsWith(lastDirRel)) { - Path vol = VolumeManager.FileType.TABLE.getVolume(absPath); - - boolean sameVol = false; - - if (GcVolumeUtil.isAllVolumesUri(lastDirAbs)) { - if (seenVolumes.contains(vol)) { - sameVol = true; - } else { - for (Volume cvol : fs.getVolumes()) { - if (cvol.containsPath(vol)) { - seenVolumes.add(vol); - sameVol = true; - } - } - } - } else { - sameVol = Objects.equals(VolumeManager.FileType.TABLE.getVolume(lastDirAbs), vol); - } - - if (sameVol) { + if (!entry.getKey().startsWith(GarbageCollectionAlgorithm.DIRECTORY_CANDIDATE_KEY_PREFIX)) { + Path filePath = new Path(entry.getValue().getPath()); + Path parent = filePath.getParent(); + while (parent != null) { + if (deleteDirs.contains(parent)) { logger.info("{} Ignoring {} because {} exist", fileActionPrefix, - entry.getValue().getPath(), lastDirAbs); + entry.getValue().getPath(), parent); processedDeletes.add(entry.getValue()); cdIter.remove(); + break; } - } else { - lastDirRel = null; - lastDirAbs = null; + parent = parent.getParent(); } } if (progressTimer.hasElapsed(loggingInterval)) { diff --git a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java index 462d593cf4e..6b9a4ed23cc 100644 --- a/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java +++ b/server/gc/src/main/java/org/apache/accumulo/gc/GarbageCollectionAlgorithm.java @@ -29,6 +29,7 @@ import java.util.Iterator; import java.util.List; import java.util.Locale; +import java.util.Map.Entry; import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; @@ -42,6 +43,7 @@ import org.apache.accumulo.core.metadata.schema.Ample.GcCandidateType; import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.ServerColumnFamily; import org.apache.accumulo.core.trace.TraceUtil; +import org.apache.hadoop.fs.Path; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,6 +55,8 @@ public class GarbageCollectionAlgorithm { private static final Logger log = LoggerFactory.getLogger(GarbageCollectionAlgorithm.class); + static final String DIRECTORY_CANDIDATE_KEY_PREFIX = "DIR|"; + private static final String FILE_CANDIDATE_KEY_PREFIX = "FILE|"; /** * This method takes a file or directory path and returns a relative path in 1 of 2 forms: @@ -119,18 +123,72 @@ private String makeRelative(String path, int expectedLen) { return relPath; } - private SortedMap makeRelative(Collection candidates) { + private String makeCandidateKey(String path) { + try { + String relative = makeRelative(path, 0); + if (SimpleGarbageCollector.isDir(relative)) { + return DIRECTORY_CANDIDATE_KEY_PREFIX + relative; + } + } catch (IllegalArgumentException e) { + // Paths for external files do not have the Accumulo tables// structure. + // Their full URI is their identity, so don't discard them or compare by path suffix. + } + + return fileCandidateKey(path); + } + + private String fileCandidateKey(String path) { + if (path.endsWith("/")) { + throw new IllegalArgumentException("GC file candidate is a directory: " + path); + } + Path candidatePath = new Path(path); + if (candidatePath.toUri().getScheme() != null) { + if (candidatePath.getName().isEmpty()) { + throw new IllegalArgumentException("GC file candidate does not identify a file: " + path); + } + return FILE_CANDIDATE_KEY_PREFIX + candidatePath; + } + + // Relative paths are retained for compatibility with older metadata entries. Once paths are + // absolute, full URI identity is required so files at distinct locations cannot alias. + String relative = tryMakeRelative(path, 3); + if (relative == null) { + throw new IllegalArgumentException("Invalid relative GC file candidate: " + path); + } + return FILE_CANDIDATE_KEY_PREFIX + "RELATIVE|" + relative; + } + + private String directoryCandidateKey(String relativePath) { + return DIRECTORY_CANDIDATE_KEY_PREFIX + relativePath; + } + + private String tabletDirectoryKey(String filePath) { + String relativeFile = tryMakeRelative(filePath, 3); + if (relativeFile == null) { + // External files are not in an Accumulo tablet directory and have no directory candidate to + // suppress. + return null; + } + return directoryCandidateKey(relativeFile.substring(0, relativeFile.lastIndexOf('/'))); + } + + private String tryMakeRelative(String path, int expectedLen) { + try { + return makeRelative(path, expectedLen); + } catch (IllegalArgumentException e) { + return null; + } + } + + private SortedMap makeCandidateMap(Collection candidates) { SortedMap ret = new TreeMap<>(); for (GcCandidate candidate : candidates) { - String relPath; try { - relPath = makeRelative(candidate.getPath(), 0); - } catch (IllegalArgumentException iae) { + ret.put(makeCandidateKey(candidate.getPath()), candidate); + } catch (IllegalArgumentException e) { log.warn("Ignoring invalid deletion candidate {}", candidate); - continue; } - ret.put(relPath, candidate); } return ret; @@ -151,7 +209,7 @@ private void removeCandidatesInUse(GarbageCollectionEnvironment gce, String dir = "/" + ref.getTableId() + "/" + ref.getMetadataPath(); - dir = makeRelative(dir, 2); + dir = directoryCandidateKey(makeRelative(dir, 2)); GcCandidate gcTemp = candidateMap.remove(dir); if (gcTemp != null) { @@ -169,24 +227,49 @@ private void removeCandidatesInUse(GarbageCollectionEnvironment gce, throw new RuntimeException("Bad file reference " + reference); } - String relativePath = makeRelative(reference, 3); + if (new Path(reference).toUri().getScheme() == null + && tryMakeRelative(reference, 3) == null) { + throw new IllegalArgumentException("Invalid relative file reference " + reference); + } - // WARNING: This line is EXTREMELY IMPORTANT. - // You MUST REMOVE candidates that are still in use - GcCandidate gcTemp = candidateMap.remove(relativePath); + List inUseCandidates = new ArrayList<>(); + GcCandidate gcTemp = candidateMap.remove(fileCandidateKey(reference)); if (gcTemp != null) { - log.debug("File Candidate was still in use: {}", relativePath); + inUseCandidates.add(gcTemp); + } + + // Old relative references do not identify a filesystem location. Conservatively protect + // any managed-volume candidate with the corresponding legacy suffix. Absolute references + // always use exact URI identity and cannot alias an external file by suffix. + if (new Path(reference).toUri().getScheme() == null) { + String relativePath = tryMakeRelative(reference, 3); + if (relativePath != null) { + Iterator> candidateIter = + candidateMap.entrySet().iterator(); + while (candidateIter.hasNext()) { + Entry candidateEntry = candidateIter.next(); + if (candidateEntry.getKey().startsWith(FILE_CANDIDATE_KEY_PREFIX) && relativePath + .equals(tryMakeRelative(candidateEntry.getValue().getPath(), 0))) { + inUseCandidates.add(candidateEntry.getValue()); + candidateIter.remove(); + } + } + } + } + + for (GcCandidate inUseCandidate : inUseCandidates) { + log.debug("File Candidate was still in use: {}", reference); // Prevent deletion of candidates that are still in use by scans, because they won't be // recreated once the scan is finished. if (!ref.isScan()) { - candidateEntriesToBeDeleted.add(gcTemp); + candidateEntriesToBeDeleted.add(inUseCandidate); } } - String dir = relativePath.substring(0, relativePath.lastIndexOf('/')); - GcCandidate gcT = candidateMap.remove(dir); + String dirKey = tabletDirectoryKey(reference); + GcCandidate gcT = dirKey == null ? null : candidateMap.remove(dirKey); if (gcT != null) { - log.debug("Directory Candidate was still in use by file ref: {}", relativePath); + log.debug("Directory Candidate was still in use by file ref: {}", reference); // Do not add dir candidates to candidateEntriesToBeDeleted as they are only created // once. } @@ -202,8 +285,9 @@ private void removeCandidatesInUse(GarbageCollectionEnvironment gce, private long removeBlipCandidates(GarbageCollectionEnvironment gce, SortedMap candidateMap) throws TableNotFoundException { long blipCount = 0; - boolean checkForBulkProcessingFiles = candidateMap.keySet().stream().anyMatch( - relativePath -> relativePath.toLowerCase(Locale.ENGLISH).contains(Constants.BULK_PREFIX)); + boolean checkForBulkProcessingFiles = + candidateMap.values().stream().anyMatch(candidate -> candidate.getPath() + .toLowerCase(Locale.ENGLISH).contains(Constants.BULK_PREFIX)); if (checkForBulkProcessingFiles) { try (Stream blipStream = gce.getBlipPaths()) { @@ -216,18 +300,31 @@ private long removeBlipCandidates(GarbageCollectionEnvironment gce, while (blipiter.hasNext()) { blipCount++; String blipPath = blipiter.next(); - blipPath = makeRelative(blipPath, 2); - - Iterator tailIter = candidateMap.tailMap(blipPath).keySet().iterator(); + Path blip = new Path(blipPath); + boolean absoluteBlip = blip.toUri().getScheme() != null; + String relativeBlip = null; + try { + relativeBlip = makeRelative(blipPath, 2); + } catch (IllegalArgumentException e) { + // External bulk-load paths are matched using their full URI below. + } + Iterator> candidateIter = candidateMap.entrySet().iterator(); int count = 0; - - while (tailIter.hasNext()) { - if (tailIter.next().startsWith(blipPath)) { + while (candidateIter.hasNext()) { + Entry candidateEntry = candidateIter.next(); + Path candidatePath = new Path(candidateEntry.getValue().getPath()); + if (isDescendantOrSame(blip, candidatePath)) { + count++; + candidateIter.remove(); + } else if (absoluteBlip && candidatePath.toUri().getScheme() == null + && hasRelativeBulkPath(relativeBlip, candidateEntry.getValue().getPath())) { + count++; + candidateIter.remove(); + } else if (!absoluteBlip + && hasRelativeBulkPath(relativeBlip, candidateEntry.getValue().getPath())) { count++; - tailIter.remove(); - } else { - break; + candidateIter.remove(); } } @@ -241,6 +338,30 @@ private long removeBlipCandidates(GarbageCollectionEnvironment gce, return blipCount; } + private boolean hasRelativeBulkPath(String relativeDir, String candidatePath) { + if (relativeDir == null) { + return false; + } + try { + String relativeCandidate = makeRelative(candidatePath, 0); + return relativeCandidate.equals(relativeDir) + || relativeCandidate.startsWith(relativeDir + "/"); + } catch (IllegalArgumentException e) { + return false; + } + } + + private static boolean isDescendantOrSame(Path ancestor, Path candidate) { + Path current = candidate; + while (current != null) { + if (current.equals(ancestor)) { + return true; + } + current = current.getParent(); + } + return false; + } + /** * Double check no tables were missed during GC */ @@ -287,10 +408,14 @@ private void cleanUpDeletedTableDirs(GarbageCollectionEnvironment gce, // find the table ids that had dirs deleted for (String delete : candidateMap.keySet()) { - String[] tokens = delete.split("/"); + if (!delete.startsWith(DIRECTORY_CANDIDATE_KEY_PREFIX)) { + continue; + } + String relativeDir = delete.substring(DIRECTORY_CANDIDATE_KEY_PREFIX.length()); + String[] tokens = relativeDir.split("/"); if (tokens.length == 2) { // its a directory - TableId tableId = TableId.of(delete.split("/")[0]); + TableId tableId = TableId.of(tokens[0]); tableIdsWithDeletes.add(tableId); } } @@ -373,7 +498,7 @@ private long deleteBatch(GarbageCollectionEnvironment gce, List cur gce.incrementCandidatesStat(origSize); log.info("Batch {} total deletion candidates: {}", batchCount, origSize); - SortedMap candidateMap = makeRelative(currentBatch); + SortedMap candidateMap = makeCandidateMap(currentBatch); long blips = confirmDeletesTrace(gce, candidateMap); gce.incrementInUseStat(origSize - candidateMap.size()); diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java index 72cc61e474b..5d64ee81d04 100644 --- a/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java +++ b/server/gc/src/test/java/org/apache/accumulo/gc/GarbageCollectionTest.java @@ -18,6 +18,7 @@ */ package org.apache.accumulo.gc; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -261,6 +262,32 @@ public void minimalDelete() throws Exception { assertCandidateRemoved(gce, GcCandidateType.INUSE, candOne, candTwo, candFour); } + @Test + public void externalFilesAreMatchedByFullUri() throws Exception { + TestGCE gce = new TestGCE(); + + var sameNameDifferentLocation = gce.addCandidate("hdfs://nn1/external/a/shared.rf"); + var referencedCandidate = gce.addCandidate("hdfs://nn2/external/b/shared.rf"); + gce.addFileReference("4", null, referencedCandidate.getPath()); + + new GarbageCollectionAlgorithm().collect(gce); + + assertFileDeleted(gce, sameNameDifferentLocation); + assertCandidateRemoved(gce, GcCandidateType.INUSE, referencedCandidate); + } + + @Test + public void externalScanReferenceProtectsCandidate() throws Exception { + TestGCE gce = new TestGCE(); + var candidate = gce.addCandidate("hdfs://nn/external/rfiles/source.rf"); + gce.addScanReference("4", null, candidate.getPath()); + + new GarbageCollectionAlgorithm().collect(gce); + + assertFileDeleted(gce); + assertNoCandidatesRemoved(gce); + } + @Test public void testBasic() throws Exception { TestGCE gce = new TestGCE(); @@ -671,15 +698,13 @@ public void testBadFileRef2() { } @Test - public void testBadFileRef3() { - assertThrows(IllegalArgumentException.class, - () -> badRefTest("hdfs://foo.com:6000/accumulo/F00.rf")); + public void testExternalFileReferenceDoesNotNeedTabletPathLayout() { + assertDoesNotThrow(() -> badRefTest("hdfs://foo.com:6000/accumulo/F00.rf")); } @Test - public void testBadFileRef4() { - assertThrows(IllegalArgumentException.class, - () -> badRefTest("hdfs://foo.com:6000/accumulo/tbls/5/F00.rf")); + public void testExternalFileReferenceCanHaveArbitraryPath() { + assertDoesNotThrow(() -> badRefTest("hdfs://foo.com:6000/accumulo/tbls/5/F00.rf")); } @Test @@ -712,7 +737,8 @@ public void testBadDeletes() throws Exception { gca.collect(gce); System.out.println(gce.fileDeletions); - assertFileDeleted(gce); + assertFileDeleted(gce, new GcCandidate("hdfs://foo.com:6000/accumulo/tbls/5/F00.rf", 8L), + new GcCandidate("hdfs://foo.com:6000/user/foo/tables/a/t-0/t-1/F00.rf", 11L)); } @Test diff --git a/server/gc/src/test/java/org/apache/accumulo/gc/SimpleGarbageCollectorTest.java b/server/gc/src/test/java/org/apache/accumulo/gc/SimpleGarbageCollectorTest.java index 9cc2e7f8b22..b5bd4a52a1e 100644 --- a/server/gc/src/test/java/org/apache/accumulo/gc/SimpleGarbageCollectorTest.java +++ b/server/gc/src/test/java/org/apache/accumulo/gc/SimpleGarbageCollectorTest.java @@ -19,6 +19,7 @@ package org.apache.accumulo.gc; import static org.easymock.EasyMock.anyObject; +import static org.easymock.EasyMock.anyString; import static org.easymock.EasyMock.createMock; import static org.easymock.EasyMock.expect; import static org.easymock.EasyMock.getCurrentArguments; @@ -146,11 +147,15 @@ public void testMinimizeDeletes() { expect(vol1.containsPath(anyObject())) .andAnswer(() -> getCurrentArguments()[0].toString().startsWith("hdfs://nn1/accumulo")) .anyTimes(); + expect(vol1.prefixChild(anyString())) + .andAnswer(() -> new Path("hdfs://nn1/accumulo/" + getCurrentArguments()[0])).anyTimes(); Volume vol2 = createMock(Volume.class); expect(vol2.containsPath(anyObject())) .andAnswer(() -> getCurrentArguments()[0].toString().startsWith("hdfs://nn2/accumulo")) .anyTimes(); + expect(vol2.prefixChild(anyString())) + .andAnswer(() -> new Path("hdfs://nn2/accumulo/" + getCurrentArguments()[0])).anyTimes(); Collection vols = Arrays.asList(vol1, vol2); @@ -160,24 +165,24 @@ public void testMinimizeDeletes() { replay(vol1, vol2, volMgr2); TreeMap confirmed = new TreeMap<>(); - confirmed.put("5a/t-0001", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001", 0L)); - confirmed.put("5a/t-0001/F0001.rf", + confirmed.put("DIR|5a/t-0001", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001", 0L)); + confirmed.put("FILE|hdfs://nn1/accumulo/tables/5a/t-0001/F0001.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001/F0001.rf", 1L)); - confirmed.put("5a/t-0001/F0002.rf", + confirmed.put("FILE|hdfs://nn1/accumulo/tables/5a/t-0001/F0002.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001/F0002.rf", 2L)); - confirmed.put("5a/t-0002/F0001.rf", + confirmed.put("FILE|hdfs://nn1/accumulo/tables/5a/t-0002/F0001.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0002/F0001.rf", 3L)); var allVolumesDirectory = new AllVolumesDirectory(TableId.of("5b"), "t-0003"); - confirmed.put("5b/t-0003", new GcCandidate(allVolumesDirectory.getMetadataPath(), 4L)); - confirmed.put("5b/t-0003/F0001.rf", + confirmed.put("DIR|5b/t-0003", new GcCandidate(allVolumesDirectory.getMetadataPath(), 4L)); + confirmed.put("FILE|hdfs://nn1/accumulo/tables/5b/t-0003/F0001.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5b/t-0003/F0001.rf", 5L)); - confirmed.put("5b/t-0003/F0002.rf", + confirmed.put("FILE|hdfs://nn2/accumulo/tables/5b/t-0003/F0002.rf", new GcCandidate("hdfs://nn2/accumulo/tables/5b/t-0003/F0002.rf", 6L)); - confirmed.put("5b/t-0003/F0003.rf", + confirmed.put("FILE|hdfs://nn3/accumulo/tables/5b/t-0003/F0003.rf", new GcCandidate("hdfs://nn3/accumulo/tables/5b/t-0003/F0003.rf", 7L)); allVolumesDirectory = new AllVolumesDirectory(TableId.of("5b"), "t-0004"); - confirmed.put("5b/t-0004", new GcCandidate(allVolumesDirectory.getMetadataPath(), 8L)); - confirmed.put("5b/t-0004/F0001.rf", + confirmed.put("DIR|5b/t-0004", new GcCandidate(allVolumesDirectory.getMetadataPath(), 8L)); + confirmed.put("FILE|hdfs://nn1/accumulo/tables/5b/t-0004/F0001.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5b/t-0004/F0001.rf", 9L)); List processedDeletes = new ArrayList<>(); @@ -185,23 +190,23 @@ public void testMinimizeDeletes() { GCRun.minimizeDeletes(confirmed, processedDeletes, volMgr2, log, Duration.ofMinutes(1)); TreeMap expected = new TreeMap<>(); - expected.put("5a/t-0001", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001", 0L)); - expected.put("5a/t-0002/F0001.rf", + expected.put("DIR|5a/t-0001", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001", 0L)); + expected.put("FILE|hdfs://nn1/accumulo/tables/5a/t-0002/F0001.rf", new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0002/F0001.rf", 3L)); allVolumesDirectory = new AllVolumesDirectory(TableId.of("5b"), "t-0003"); - expected.put("5b/t-0003", new GcCandidate(allVolumesDirectory.getMetadataPath(), 4L)); - expected.put("5b/t-0003/F0003.rf", + expected.put("DIR|5b/t-0003", new GcCandidate(allVolumesDirectory.getMetadataPath(), 4L)); + expected.put("FILE|hdfs://nn3/accumulo/tables/5b/t-0003/F0003.rf", new GcCandidate("hdfs://nn3/accumulo/tables/5b/t-0003/F0003.rf", 7L)); allVolumesDirectory = new AllVolumesDirectory(TableId.of("5b"), "t-0004"); - expected.put("5b/t-0004", new GcCandidate(allVolumesDirectory.getMetadataPath(), 8L)); + expected.put("DIR|5b/t-0004", new GcCandidate(allVolumesDirectory.getMetadataPath(), 8L)); assertEquals(expected, confirmed); assertEquals( Arrays.asList(new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001/F0001.rf", 1L), new GcCandidate("hdfs://nn1/accumulo/tables/5a/t-0001/F0002.rf", 2L), new GcCandidate("hdfs://nn1/accumulo/tables/5b/t-0003/F0001.rf", 5L), - new GcCandidate("hdfs://nn2/accumulo/tables/5b/t-0003/F0002.rf", 6L), - new GcCandidate("hdfs://nn1/accumulo/tables/5b/t-0004/F0001.rf", 9L)), + new GcCandidate("hdfs://nn1/accumulo/tables/5b/t-0004/F0001.rf", 9L), + new GcCandidate("hdfs://nn2/accumulo/tables/5b/t-0003/F0002.rf", 6L)), processedDeletes); } } diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java index 0d3363c302c..08cabf75125 100644 --- a/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java +++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/ScanServer.java @@ -34,7 +34,6 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.UUID; @@ -593,6 +592,7 @@ private Map getTabletMetadata(Collection ex static class ReservedFile { final Set activeReservations = new ConcurrentSkipListSet<>(); + final Set tableIds = ConcurrentHashMap.newKeySet(); final AtomicLong lastUseTime = new AtomicLong(0); boolean shouldDelete(long expireTimeMs) { @@ -695,10 +695,12 @@ private Map reserveFilesInner(Collection ex } } - Map allFiles = new HashMap<>(extents.size()); + Map> allFiles = new HashMap<>(extents.size()); tabletsMetadata.forEach((extent, tm) -> { - tm.getFiles().forEach(file -> allFiles.put(file, extent)); + for (StoredTabletFile file : tm.getFiles()) { + allFiles.computeIfAbsent(file, ignored -> new HashSet<>()).add(extent); + } }); // The read lock prevents anything from being removed from reservedFiles while adding @@ -706,8 +708,25 @@ private Map reserveFilesInner(Collection ex // wait on each other their files are already reserved. reservationsReadLock.lock(); try { - if (reservedFiles.keySet().containsAll(allFiles.keySet())) { - // all files already have reservations in the metadata table, so we can add ourself + boolean allFilesAndTableIdsReserved = reservedFiles.keySet().containsAll(allFiles.keySet()); + if (allFilesAndTableIdsReserved) { + for (Map.Entry> entry : allFiles.entrySet()) { + ReservedFile reservedFile = reservedFiles.get(entry.getKey()); + for (KeyExtent extent : entry.getValue()) { + if (!reservedFile.tableIds.contains(extent.tableId())) { + allFilesAndTableIdsReserved = false; + break; + } + } + if (!allFilesAndTableIdsReserved) { + break; + } + } + } + + if (allFilesAndTableIdsReserved) { + // all files and table references already have reservations in metadata, so we can add + // ourself without updating the stored reference values. for (StoredTabletFile file : allFiles.keySet()) { if (!reservedFiles.get(file).activeReservations.add(myReservationId)) { throw new IllegalStateException("reservation id unexpectedly already in set"); @@ -743,25 +762,56 @@ private Map reserveFilesInner(Collection ex try { Set filesToReserve = new HashSet<>(); + Map> refTableIds = new HashMap<>(); + Set pathsWithReservedRefs = new HashSet<>(); + for (StoredTabletFile reservedFile : reservedFiles.keySet()) { + pathsWithReservedRefs.add(reservedFile.getNormalizedPathStr()); + } + Set newRefPaths = new HashSet<>(); List refs = new ArrayList<>(); Set tabletsToCheck = new HashSet<>(); String serverAddress = getAdvertiseAddress().toString(); - for (StoredTabletFile file : allFiles.keySet()) { - if (!reservedFiles.containsKey(file)) { - refs.add(new ScanServerRefTabletFile(serverLockUUID, serverAddress, - file.getNormalizedPathStr())); + for (Map.Entry> entry : allFiles.entrySet()) { + StoredTabletFile file = entry.getKey(); + Set tableIds = new HashSet<>(); + for (KeyExtent extent : entry.getValue()) { + tableIds.add(extent.tableId()); + } + + ReservedFile reservedFile = reservedFiles.get(file); + if (reservedFile == null) { filesToReserve.add(file); - tabletsToCheck.add(Objects.requireNonNull(allFiles.get(file))); - LOG.trace("RFFS {} need to add scan ref for file {}", myReservationId, file); + tabletsToCheck.addAll(entry.getValue()); + } + + if (reservedFile == null || !reservedFile.tableIds.containsAll(tableIds)) { + Set idsForRef = new HashSet<>(tableIds); + if (reservedFile != null) { + idsForRef.addAll(reservedFile.tableIds); + } + String filePath = file.getNormalizedPathStr(); + if (!pathsWithReservedRefs.contains(filePath)) { + newRefPaths.add(filePath); + pathsWithReservedRefs.add(filePath); + } + refTableIds.computeIfAbsent(filePath, ignored -> new HashSet<>()).addAll(idsForRef); + LOG.trace("RFFS {} need to add scan ref for file {} and tables {}", myReservationId, file, + idsForRef); } } - if (!filesToReserve.isEmpty()) { + if (!refTableIds.isEmpty()) { + for (Map.Entry> entry : refTableIds.entrySet()) { + refs.add(new ScanServerRefTabletFile(serverLockUUID, serverAddress, entry.getKey(), + entry.getValue())); + } scanServerMetrics.recordWriteOutReservationTime( () -> getContext().getAmple().scanServerRefs().put(refs)); + } + if (!filesToReserve.isEmpty()) { // After we insert the scan server refs we need to check and see if the tablet is still // using the file. As long as the tablet is still using the files then the Accumulo GC // should not have deleted the files. This assumes the Accumulo GC reads scan server refs @@ -793,7 +843,15 @@ private Map reserveFilesInner(Collection ex if (!filesToReserve.isEmpty()) { LOG.info("RFFS {} tablet files changed while attempting to reference files {}", myReservationId, filesToReserve); - getContext().getAmple().scanServerRefs().delete(refs); + List refsToDelete = new ArrayList<>(newRefPaths.size()); + for (ScanServerRefTabletFile ref : refs) { + if (newRefPaths.contains(ref.getNormalizedPathStr())) { + refsToDelete.add(ref); + } + } + if (!refsToDelete.isEmpty()) { + getContext().getAmple().scanServerRefs().delete(refsToDelete); + } scanServerMetrics.incrementReservationConflictCount(); return null; } @@ -801,13 +859,17 @@ private Map reserveFilesInner(Collection ex // we do not hold a lock but the files we are adding are in influxFiles so its ok to add to // reservedFiles - for (StoredTabletFile file : allFiles.keySet()) { - if (!reservedFiles.computeIfAbsent(file, k -> new ReservedFile()).activeReservations - .add(myReservationId)) { + for (Map.Entry> entry : allFiles.entrySet()) { + ReservedFile reservedFile = + reservedFiles.computeIfAbsent(entry.getKey(), k -> new ReservedFile()); + for (KeyExtent extent : entry.getValue()) { + reservedFile.tableIds.add(extent.tableId()); + } + if (!reservedFile.activeReservations.add(myReservationId)) { throw new IllegalStateException("reservation id unexpectedly already in set"); } - LOG.trace("RFFS {} reserved reference for startScan {}", myReservationId, file); + LOG.trace("RFFS {} reserved reference for startScan {}", myReservationId, entry.getKey()); } return tabletsMetadata; @@ -952,8 +1014,8 @@ private void cleanUpReservedFiles(long expireTimeMs) { // then adding the file to influxFiles will make it wait until we finish influxFiles.add(file); confirmed.add(file); - refsToDelete.add(new ScanServerRefTabletFile(file.getNormalizedPathStr(), serverAddress, - serverLockUUID)); + refsToDelete.add(new ScanServerRefTabletFile(serverLockUUID, serverAddress, + file.getNormalizedPathStr(), entry.getValue().tableIds)); // remove the entry from the map while holding the write lock ensuring no new // reservations are added to the map values while the metadata operation to delete is diff --git a/test/src/main/java/org/apache/accumulo/test/ComprehensiveTableOperationsIT.java b/test/src/main/java/org/apache/accumulo/test/ComprehensiveTableOperationsIT.java index 46e97cac895..f35f494bbf4 100644 --- a/test/src/main/java/org/apache/accumulo/test/ComprehensiveTableOperationsIT.java +++ b/test/src/main/java/org/apache/accumulo/test/ComprehensiveTableOperationsIT.java @@ -885,7 +885,8 @@ private Text createScanRefTableRow() { UUID serverLockUUID = UUID.randomUUID(); Set scanRefs = Stream.of("F0000070.rf", "F0000071.rf") .map(f -> "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/" + f) - .map(f -> new ScanServerRefTabletFile(f, server.toString(), serverLockUUID)) + .map(f -> new ScanServerRefTabletFile(serverLockUUID, server.toString(), f, + Set.of(TableId.of("2a")))) .collect(Collectors.toSet()); getCluster().getServerContext().getAmple().scanServerRefs().put(scanRefs); return new Text(getCluster().getServerContext().getAmple().scanServerRefs().list() diff --git a/test/src/main/java/org/apache/accumulo/test/ExternalTabletFileIT.java b/test/src/main/java/org/apache/accumulo/test/ExternalTabletFileIT.java new file mode 100644 index 00000000000..63894da8bc9 --- /dev/null +++ b/test/src/main/java/org/apache/accumulo/test/ExternalTabletFileIT.java @@ -0,0 +1,192 @@ +/* + * 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.test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.time.Duration; + +import org.apache.accumulo.core.client.Accumulo; +import org.apache.accumulo.core.client.AccumuloClient; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.admin.CompactionConfig; +import org.apache.accumulo.core.client.rfile.RFile; +import org.apache.accumulo.core.client.rfile.RFileWriter; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.TableId; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.dataImpl.KeyExtent; +import org.apache.accumulo.core.gc.GcCandidate; +import org.apache.accumulo.core.metadata.StoredTabletFile; +import org.apache.accumulo.core.metadata.schema.Ample; +import org.apache.accumulo.core.metadata.schema.DataFileValue; +import org.apache.accumulo.core.metadata.schema.TabletMetadata; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.accumulo.minicluster.ServerType; +import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl; +import org.apache.accumulo.test.functional.ConfigurableMacBase; +import org.apache.accumulo.test.util.Wait; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.junit.jupiter.api.Test; + +/** Exercises tablet files stored outside Accumulo's configured table directories. */ +public class ExternalTabletFileIT extends ConfigurableMacBase { + + private static final Duration GC_WAIT = Duration.ofSeconds(45); + private static final long GC_POLL_MS = 100; + + @Override + public void configure(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) { + cfg.useMiniDFS(true); + cfg.setProperty(Property.GC_CYCLE_START, "1s"); + cfg.setProperty(Property.GC_CYCLE_DELAY, "1s"); + cfg.setProperty(Property.MANAGER_TABLE_DELETE_OPTIMIZATION, "true"); + super.configure(cfg, hadoopCoreSite); + } + + @Override + protected Duration defaultTimeout() { + return Duration.ofMinutes(3); + } + + @Test + public void scanAndCompactExternalFile() throws Exception { + try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) { + FileSystem fs = getCluster().getMiniDfs().getFileSystem(); + String table = getUniqueNames(1)[0]; + client.tableOperations().create(table); + + Path externalFile = writeRFile(fs, "scan-compact"); + addFileReference(client, table, externalFile, fs.getFileStatus(externalFile).getLen(), 2); + + assertRows(client, table); + + client.tableOperations().compact(table, new CompactionConfig().setWait(true)); + + assertRows(client, table); + assertTrue(hasConfiguredVolumeFile(client, table)); + Wait.waitFor(() -> !fs.exists(externalFile), GC_WAIT.toMillis(), GC_POLL_MS, + "GC did not remove compacted external RFile " + externalFile); + } + } + + @Test + public void optimizedTableDeleteGcExternalFileAfterLastReference() throws Exception { + try (AccumuloClient client = Accumulo.newClient().from(getClientProperties()).build()) { + FileSystem fs = getCluster().getMiniDfs().getFileSystem(); + String[] tables = getUniqueNames(2); + client.tableOperations().create(tables[0]); + client.tableOperations().create(tables[1]); + + Path externalFile = writeRFile(fs, "delete-shared"); + long fileSize = fs.getFileStatus(externalFile).getLen(); + addFileReference(client, tables[0], externalFile, fileSize, 2); + addFileReference(client, tables[1], externalFile, fileSize, 2); + assertRows(client, tables[0]); + assertRows(client, tables[1]); + + getCluster().getClusterControl().stop(ServerType.GARBAGE_COLLECTOR); + client.tableOperations().delete(tables[0]); + + assertTrue(fs.exists(externalFile), "Deleting one table must preserve the shared file"); + assertTrue(hasGcCandidate(externalFile), + "Optimized deletion must enqueue an external-file GC candidate"); + + getCluster().getClusterControl().start(ServerType.GARBAGE_COLLECTOR); + Wait.waitFor(() -> !hasGcCandidate(externalFile), GC_WAIT.toMillis(), GC_POLL_MS, + "GC did not clear the in-use external-file candidate"); + assertTrue(fs.exists(externalFile), "GC deleted a file still referenced by another table"); + assertRows(client, tables[1]); + + getCluster().getClusterControl().stop(ServerType.GARBAGE_COLLECTOR); + client.tableOperations().delete(tables[1]); + assertTrue(hasGcCandidate(externalFile), + "Deleting the last referencing table must enqueue a GC candidate"); + + getCluster().getClusterControl().start(ServerType.GARBAGE_COLLECTOR); + Wait.waitFor(() -> !fs.exists(externalFile), GC_WAIT.toMillis(), GC_POLL_MS, + "GC did not delete the external RFile after its last reference was removed"); + } + } + + private Path writeRFile(FileSystem fs, String name) throws Exception { + Path externalFile = fs.makeQualified(new Path("/external-rfiles/" + name + ".rf")); + fs.mkdirs(externalFile.getParent()); + try (RFileWriter writer = + RFile.newWriter().to(externalFile.toString()).withFileSystem(fs).build()) { + writer.startDefaultLocalityGroup(); + writer.append(new Key("r1", "cf", "cq"), new Value("v1")); + writer.append(new Key("r2", "cf", "cq"), new Value("v2")); + } + return externalFile; + } + + private void addFileReference(AccumuloClient client, String table, Path file, long size, + long entries) throws Exception { + client.tableOperations().offline(table, true); + TableId tableId = TableId.of(client.tableOperations().tableIdMap().get(table)); + getCluster().getServerContext().getAmple().mutateTablet(new KeyExtent(tableId, null, null)) + .putFile(StoredTabletFile.of(file), new DataFileValue(size, entries)).mutate(); + client.tableOperations().online(table, true); + } + + private void assertRows(AccumuloClient client, String table) throws Exception { + try (Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) { + int count = 0; + for (var entry : scanner) { + String row = entry.getKey().getRow().toString(); + assertEquals("r" + (count + 1), row); + assertEquals("v" + (count + 1), entry.getValue().toString()); + count++; + } + assertEquals(2, count); + } + } + + private boolean hasConfiguredVolumeFile(AccumuloClient client, String table) throws Exception { + TableId tableId = TableId.of(client.tableOperations().tableIdMap().get(table)); + try (var tablets = getCluster().getServerContext().getAmple().readTablets().forTable(tableId) + .fetch(TabletMetadata.ColumnType.FILES).build()) { + for (var tablet : tablets) { + for (StoredTabletFile file : tablet.getFilesMap().keySet()) { + if (!file.getPath().toString().contains("/external-rfiles/")) { + return true; + } + } + } + } + return false; + } + + private boolean hasGcCandidate(Path path) { + var candidates = + getCluster().getServerContext().getAmple().getGcCandidates(Ample.DataLevel.USER); + while (candidates.hasNext()) { + GcCandidate candidate = candidates.next(); + if (new Path(candidate.getPath()).equals(path)) { + return true; + } + } + return false; + } +} diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesCleanIT_SimpleSuite.java b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesCleanIT_SimpleSuite.java index abe78e99c2b..4198fddd9a7 100644 --- a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesCleanIT_SimpleSuite.java +++ b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesCleanIT_SimpleSuite.java @@ -27,6 +27,7 @@ import java.util.stream.Collectors; import java.util.stream.Stream; +import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.metadata.ScanServerRefTabletFile; import org.apache.accumulo.server.ServerContext; import org.apache.accumulo.server.util.ScanServerMetadataEntries; @@ -58,7 +59,8 @@ public void testServerContextMethods() { Set scanRefs = Stream.of("F0000070.rf", "F0000071.rf") .map(f -> "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/" + f) - .map(f -> new ScanServerRefTabletFile(f, server.toString(), serverLockUUID)) + .map(f -> new ScanServerRefTabletFile(serverLockUUID, server.toString(), f, + Set.of(TableId.of("2a")))) .collect(Collectors.toSet()); ServerContext ctx = getCluster().getServerContext(); diff --git a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java index 4bf1aa54ab1..e33cefe0ccd 100644 --- a/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java +++ b/test/src/main/java/org/apache/accumulo/test/ScanServerMetadataEntriesIT.java @@ -109,7 +109,8 @@ public void testServerContextMethods() { Set scanRefs = Stream.of("F0000070.rf", "F0000071.rf") .map(f -> "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/" + f) - .map(f -> new ScanServerRefTabletFile(f, server.toString(), serverLockUUID)) + .map(f -> new ScanServerRefTabletFile(serverLockUUID, server.toString(), f, + Set.of(TableId.of("2a")))) .collect(Collectors.toSet()); ServerContext ctx = getCluster().getServerContext(); @@ -240,7 +241,8 @@ public void testGcRunScanServerReferences() throws Exception { Set metadataScanFileRefs = new HashSet<>(); metadataEntries.forEach(m -> { - metadataScanFileRefs.add(new ScanServerRefTabletFile(m.getKey()).getFilePath().toString()); + metadataScanFileRefs + .add(new ScanServerRefTabletFile(m.getKey(), m.getValue()).getFilePath().toString()); }); assertEquals(fileCount, metadataScanFileRefs.size()); diff --git a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java index 5bec14ad872..b984a0222b7 100644 --- a/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java +++ b/test/src/main/java/org/apache/accumulo/test/compaction/ExternalCompaction_1_IT.java @@ -627,15 +627,14 @@ public void init(InitParameters iparams) {} @Override public Overrides override(InputParameters params) { - // This will validate the paths looks like a tablet file path and throw an exception if it - // does not + // Validate the output file's URI and the generated RFile name. var parts = ReferencedTabletFile.of(new Path(params.getOutputFile())); // For this test should be producing A files Preconditions.checkArgument( parts.getFileName().startsWith(FilePrefix.FULL_COMPACTION.getPrefix() + "")); Preconditions.checkArgument(parts.getFileName().endsWith(RFile.EXTENSION)); - if (parts.getTabletDir() + if (parts.getPath().getParent().getName() .equals(MetadataSchema.TabletsSection.ServerColumnFamily.DEFAULT_TABLET_DIR_NAME)) { return new Overrides(Map.of(Property.TABLE_FILE_COMPRESSION_TYPE.getKey(), "gz")); } else { diff --git a/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorTrashBase.java b/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorTrashBase.java index 8904dadd676..0cef282d056 100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorTrashBase.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/GarbageCollectorTrashBase.java @@ -28,9 +28,9 @@ import java.util.Iterator; import java.util.stream.Collectors; +import org.apache.accumulo.core.Constants; import org.apache.accumulo.core.client.AccumuloClient; import org.apache.accumulo.core.data.TableId; -import org.apache.accumulo.core.metadata.ReferencedTabletFile; import org.apache.accumulo.core.metadata.StoredTabletFile; import org.apache.accumulo.core.metadata.schema.TabletMetadata.ColumnType; import org.apache.accumulo.core.metadata.schema.TabletsMetadata; @@ -113,9 +113,9 @@ protected long countFilesInTrash(FileSystem fs, TableId tid) if (lfs.isDirectory()) { continue; } - ReferencedTabletFile tf = new ReferencedTabletFile(lfs.getPath()); - LOG.debug("File in trash: {}, tableId: {}", lfs.getPath(), tf.getTableId()); - if (tid.equals(tf.getTableId())) { + String tablePath = Constants.HDFS_TABLES_DIR + "/" + tid + "/"; + LOG.debug("File in trash: {}, looking for table path {}", lfs.getPath(), tablePath); + if (lfs.getPath().toString().contains(tablePath)) { count++; } } diff --git a/test/src/main/java/org/apache/accumulo/test/upgrade/ScanServerUpgrade11to12TestIT.java b/test/src/main/java/org/apache/accumulo/test/upgrade/ScanServerUpgrade11to12TestIT.java index 345b90da818..677f392ee2d 100644 --- a/test/src/main/java/org/apache/accumulo/test/upgrade/ScanServerUpgrade11to12TestIT.java +++ b/test/src/main/java/org/apache/accumulo/test/upgrade/ScanServerUpgrade11to12TestIT.java @@ -38,6 +38,7 @@ import org.apache.accumulo.core.conf.Property; import org.apache.accumulo.core.data.Key; import org.apache.accumulo.core.data.Mutation; +import org.apache.accumulo.core.data.TableId; import org.apache.accumulo.core.data.Value; import org.apache.accumulo.core.metadata.ScanServerRefTabletFile; import org.apache.accumulo.core.metadata.schema.Ample; @@ -101,7 +102,8 @@ private void testMetadataScanServerRefRemoval() { Set scanRefs = Stream.of("F0000070.rf", "F0000071.rf", "F0000072.rf") .map(f -> "hdfs://localhost:8020/accumulo/tables/2a/default_tablet/" + f) - .map(f -> new ScanServerRefTabletFile(serverLockUUID, server.toString(), f)) + .map(f -> new ScanServerRefTabletFile(serverLockUUID, server.toString(), f, + Set.of(TableId.of("2a")))) .collect(Collectors.toSet()); ServerContext ctx = getCluster().getServerContext();