This is an automated email from the ASF dual-hosted git repository. ddanielr pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/accumulo.git
commit 1852dbfe9c944bcda6ee84cc3db911be86ade00c Merge: a392a8e650 b874c0ed66 Author: Daniel Roberts ddanielr <[email protected]> AuthorDate: Mon Jun 22 20:00:26 2026 +0000 Merge branch '2.1' .../java/org/apache/accumulo/core/util/Merge.java | 59 +++++------ .../org/apache/accumulo/core/util/MergeTest.java | 24 ++--- .../accumulo/shell/commands/MergeCommand.java | 67 ++++++++---- .../accumulo/shell/commands/MergeCommandTest.java | 113 +++++++++++++++++++++ .../test/functional/MergeTabletsITBase.java | 67 +++++++++++- 5 files changed, 265 insertions(+), 65 deletions(-) diff --cc core/src/main/java/org/apache/accumulo/core/util/Merge.java index df8724ff63,9bd5dcc673..57a0708df6 --- a/core/src/main/java/org/apache/accumulo/core/util/Merge.java +++ b/core/src/main/java/org/apache/accumulo/core/util/Merge.java @@@ -25,10 -25,14 +25,9 @@@ import java.util.ArrayList import java.util.Iterator; import java.util.List; --import org.apache.accumulo.core.cli.ClientOpts; -import org.apache.accumulo.core.client.Accumulo; import org.apache.accumulo.core.client.AccumuloClient; import org.apache.accumulo.core.clientImpl.ClientContext; -import org.apache.accumulo.core.conf.AccumuloConfiguration; -import org.apache.accumulo.core.conf.ConfigurationCopy; import org.apache.accumulo.core.conf.ConfigurationTypeHelper; -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.dataImpl.KeyExtent; @@@ -39,7 -46,10 +38,6 @@@ import org.slf4j.Logger import org.slf4j.LoggerFactory; import com.beust.jcommander.IStringConverter; --import com.beust.jcommander.Parameter; - -import io.opentelemetry.api.trace.Span; -import io.opentelemetry.context.Scope; public class Merge { @@@ -71,23 -81,67 +69,6 @@@ } } -- static class Opts extends ClientOpts { -- @Parameter(names = {"-t", "--table"}, required = true, description = "table to use") -- String tableName; -- @Parameter(names = {"-s", "--size"}, description = "merge goal size", -- converter = MemoryConverter.class) -- Long goalSize = null; -- @Parameter(names = {"-f", "--force"}, -- description = "merge small tablets even if merging them to larger" -- + " tablets might cause a split") -- boolean force = false; -- @Parameter(names = {"-b", "--begin"}, description = "start tablet", -- converter = TextConverter.class) -- Text begin = null; -- @Parameter(names = {"-e", "--end"}, description = "end tablet", converter = TextConverter.class) -- Text end = null; - @Parameter(names = {"--dry-run"}, - description = "Will only list which tablets ranges it plans on merging. No merge will be performed") - boolean dryRun = false; - } - - public void start(String[] args) throws MergeException { - Opts opts = new Opts(); - opts.parseArgs(Merge.class.getName(), args); - Span span = TraceUtil.startSpan(Merge.class, "start"); - try (Scope scope = span.makeCurrent()) { - - try (AccumuloClient client = Accumulo.newClient().from(opts.getClientProps()).build()) { - - if (!client.tableOperations().exists(opts.tableName)) { - System.err.println("table " + opts.tableName + " does not exist"); - return; - } - if (opts.goalSize == null || opts.goalSize < 1) { - AccumuloConfiguration tableConfig = - new ConfigurationCopy(client.tableOperations().getConfiguration(opts.tableName)); - long newGoalSize = tableConfig.getAsBytes(Property.TABLE_SPLIT_THRESHOLD); - message("Invalid goal size: " + opts.goalSize + " Using the " - + Property.TABLE_SPLIT_THRESHOLD.getKey() + " value of : " + newGoalSize); - opts.goalSize = newGoalSize; - } - - message("Merging tablets in table %s to %d bytes", opts.tableName, opts.goalSize); - mergomatic(client, opts.tableName, opts.begin, opts.end, opts.goalSize, opts.force, - opts.dryRun); - } catch (MergeException e) { - TraceUtil.setException(span, e, true); - throw e; - } catch (Exception ex) { - TraceUtil.setException(span, ex, true); - throw new MergeException(ex); - } finally { - span.end(); - } - } - } - - public static void main(String[] args) throws MergeException { - Merge merge = new Merge(); - merge.start(args); -- } -- public static class Size { public Size(KeyExtent extent, long size) { this.extent = extent; @@@ -99,48 -153,26 +80,49 @@@ } public void mergomatic(AccumuloClient client, String table, Text start, Text end, long goalSize, - boolean force) throws MergeException { + boolean force, boolean dryRun) throws MergeException { try { - if (table.equals(SystemTables.METADATA.tableName())) { - throw new IllegalArgumentException("cannot merge tablets on the metadata table"); - if (table.equals(MetadataTable.NAME) || table.equals(RootTable.NAME)) { ++ if (table.equals(SystemTables.METADATA.tableName()) ++ || table.equals(SystemTables.ROOT.tableName())) { + throw new IllegalArgumentException("cannot merge tablets on the " + table + " table"); } List<Size> sizes = new ArrayList<>(); long totalSize = 0; - // Merge any until you get larger than the goal size, and then merge one less tablet - Iterator<Size> sizeIterator = getSizeIterator(client, table, start, end); - while (sizeIterator.hasNext()) { - Size next = sizeIterator.next(); - totalSize += next.size; - sizes.add(next); - if (totalSize > goalSize) { - totalSize = mergeMany(client, table, sizes, goalSize, force, false, dryRun); + + TableId tableId; + ClientContext context = (ClientContext) client; + try { + tableId = context.getTableId(table); + } catch (Exception e) { + throw new MergeException(e); + } + + try (TabletsMetadata tablets = TabletsMetadata.builder(context).scanMetadataTable() + .overRange(new KeyExtent(tableId, end, start).toMetaRange()).fetch(FILES, PREV_ROW) + .build()) { + + Iterator<Size> sizeIterator = + tablets.stream().map(tm -> new Size(tm.getExtent(), tm.getFileSize())).iterator(); + + while (sizeIterator.hasNext()) { + Size next = sizeIterator.next(); + totalSize += next.size; + sizes.add(next); + + if (totalSize > goalSize) { - mergeMany(client, table, sizes, goalSize, force, false); ++ mergeMany(client, table, sizes, goalSize, force, false, dryRun); + sizes.clear(); + sizes.add(next); + totalSize = next.size; + } } } + + // merge one less tablet if (sizes.size() > 1) { - mergeMany(client, table, sizes, goalSize, force, true); + mergeMany(client, table, sizes, goalSize, force, true, dryRun); } + } catch (Exception ex) { throw new MergeException(ex); } @@@ -194,14 -226,16 +176,14 @@@ return result; } - protected void mergeSome(AccumuloClient client, String table, List<Size> sizes, int numToMerge) - throws MergeException { - merge(client, table, sizes, numToMerge); + protected void mergeSome(AccumuloClient client, String table, List<Size> sizes, int numToMerge, + boolean dryRun) throws MergeException { + merge(client, table, sizes, numToMerge, dryRun); - for (int i = 0; i < numToMerge; i++) { - sizes.remove(0); - } + sizes.subList(0, numToMerge).clear(); } - protected void merge(AccumuloClient client, String table, List<Size> sizes, int numToMerge) - throws MergeException { + protected void merge(AccumuloClient client, String table, List<Size> sizes, int numToMerge, + boolean dryRun) throws MergeException { try { Text start = sizes.get(0).extent.prevEndRow(); Text end = sizes.get(numToMerge - 1).extent.endRow(); diff --cc core/src/test/java/org/apache/accumulo/core/util/MergeTest.java index e1a28976a9,ad1a699370..536c09ea14 --- a/core/src/test/java/org/apache/accumulo/core/util/MergeTest.java +++ b/core/src/test/java/org/apache/accumulo/core/util/MergeTest.java @@@ -58,34 -57,8 +58,34 @@@ public class MergeTest protected void message(String format, Object... args) {} @Override - protected Iterator<Size> getSizeIterator(AccumuloClient client, String tableName, - final Text start, final Text end) throws MergeException { + public void mergomatic(AccumuloClient client, String table, Text start, Text end, long goalSize, - boolean force) throws MergeException { ++ boolean force, boolean dryRun) throws MergeException { + if (table.equals(SystemTables.METADATA.tableName())) { + throw new IllegalArgumentException("cannot merge tablets on the metadata table"); + } + + List<Size> sizes = new ArrayList<>(); + long totalSize = 0; + + Iterator<Size> sizeIterator = getSizeIterator(start, end); + + while (sizeIterator.hasNext()) { + Size next = sizeIterator.next(); + totalSize += next.size; + sizes.add(next); + if (totalSize > goalSize) { - mergeMany(client, table, sizes, goalSize, force, false); ++ mergeMany(client, table, sizes, goalSize, force, false, dryRun); + sizes.clear(); + sizes.add(next); + totalSize = next.size; + } + } + if (sizes.size() > 1) { - mergeMany(client, table, sizes, goalSize, force, true); ++ mergeMany(client, table, sizes, goalSize, force, true, dryRun); + } + } + + protected Iterator<Size> getSizeIterator(final Text start, final Text end) { final Iterator<Size> impl = tablets.iterator(); return new Iterator<>() { Size next = skip(); diff --cc shell/src/main/java/org/apache/accumulo/shell/commands/MergeCommand.java index 846b1f73d8,139402c0d0..778fe714c3 --- a/shell/src/main/java/org/apache/accumulo/shell/commands/MergeCommand.java +++ b/shell/src/main/java/org/apache/accumulo/shell/commands/MergeCommand.java @@@ -28,10 -32,7 +32,11 @@@ import org.apache.commons.cli.Options import org.apache.hadoop.io.Text; public class MergeCommand extends Command { - private Option verboseOpt, forceOpt, sizeOpt, allOpt, dryRunOpt; + private Option verboseOpt; + private Option forceOpt; + private Option sizeOpt; + private Option allOpt; ++ private Option dryRunOpt; @Override public int execute(final String fullCommand, final CommandLine cl, final Shell shellState) diff --cc test/src/main/java/org/apache/accumulo/test/functional/MergeTabletsITBase.java index 5a43744bd9,0000000000..9d7d463a57 mode 100644,000000..100644 --- a/test/src/main/java/org/apache/accumulo/test/functional/MergeTabletsITBase.java +++ b/test/src/main/java/org/apache/accumulo/test/functional/MergeTabletsITBase.java @@@ -1,725 -1,0 +1,788 @@@ +/* + * 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.functional; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.apache.accumulo.test.util.FileMetadataUtil.printAndVerifyFileMetadata; +import static org.apache.accumulo.test.util.FileMetadataUtil.verifyMergedMarkerCleared; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; ++import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.UUID; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import org.apache.accumulo.core.client.Accumulo; +import org.apache.accumulo.core.client.AccumuloClient; +import org.apache.accumulo.core.client.AccumuloException; +import org.apache.accumulo.core.client.BatchWriter; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.admin.CompactionConfig; +import org.apache.accumulo.core.client.admin.NewTableConfiguration; +import org.apache.accumulo.core.client.admin.TabletAvailability; +import org.apache.accumulo.core.client.admin.TimeType; +import org.apache.accumulo.core.clientImpl.TabletAvailabilityUtil; +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.Range; +import org.apache.accumulo.core.data.ResourceGroupId; +import org.apache.accumulo.core.data.RowRange; +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.fate.FateId; +import org.apache.accumulo.core.fate.FateInstanceType; +import org.apache.accumulo.core.metadata.ReferencedTabletFile; +import org.apache.accumulo.core.metadata.StoredTabletFile; +import org.apache.accumulo.core.metadata.SystemTables; +import org.apache.accumulo.core.metadata.schema.CompactionMetadata; +import org.apache.accumulo.core.metadata.schema.DataFileValue; +import org.apache.accumulo.core.metadata.schema.ExternalCompactionId; +import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection.TabletColumnFamily; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.accumulo.core.spi.compaction.CompactionKind; +import org.apache.accumulo.core.util.FastFormat; +import org.apache.accumulo.core.util.Merge; +import org.apache.accumulo.harness.SharedMiniClusterBase; +import org.apache.accumulo.test.TestIngest; +import org.apache.accumulo.test.TestIngest.IngestParams; +import org.apache.accumulo.test.VerifyIngest; +import org.apache.accumulo.test.VerifyIngest.VerifyParams; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public abstract class MergeTabletsITBase extends SharedMiniClusterBase { + + private static final Logger log = LoggerFactory.getLogger(MergeTabletsITBase.class); + + SortedSet<Text> splits(String[] points) { + SortedSet<Text> result = new TreeSet<>(); + for (String point : points) { + result.add(new Text(point)); + } + return result; + } + + @Override + protected Duration defaultTimeout() { + return Duration.ofMinutes(8); + } + + @Test + public void tooManyFilesMergeTest() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + c.tableOperations().create(tableName, + new NewTableConfiguration().setProperties(Map.of(Property.TABLE_MAJC_RATIO.getKey(), + "20000", Property.TABLE_MERGE_FILE_MAX.getKey(), "12345"))); + + c.tableOperations().addSplits(tableName, + IntStream.range(1, 10001).mapToObj(i -> String.format("%06d", i)).map(Text::new) + .collect(Collectors.toCollection(TreeSet::new))); + c.tableOperations().addSplits(tableName, + IntStream.range(10001, 20001).mapToObj(i -> String.format("%06d", i)).map(Text::new) + .collect(Collectors.toCollection(TreeSet::new))); + + // add two bogus files to each tablet, creating 40K file entries + c.tableOperations().offline(tableName, true); + try ( + var tablets = getCluster().getServerContext().getAmple().readTablets() + .forTable(getCluster().getServerContext().getTableId(tableName)).build(); + var mutator = getCluster().getServerContext().getAmple().mutateTablets()) { + int fc = 0; + for (var tabletMeta : tablets) { + StoredTabletFile f1 = StoredTabletFile.of(new Path( + "file:///accumulo/tables/1/" + tabletMeta.getDirName() + "/F" + fc++ + ".rf")); + StoredTabletFile f2 = StoredTabletFile.of(new Path( + "file:///accumulo/tables/1/" + tabletMeta.getDirName() + "/F" + fc++ + ".rf")); + DataFileValue dfv1 = new DataFileValue(4200, 42); + DataFileValue dfv2 = new DataFileValue(4200, 42); + mutator.mutateTablet(tabletMeta.getExtent()).putFile(f1, dfv1).putFile(f2, dfv2).mutate(); + } + } + c.tableOperations().online(tableName, true); + + // should fail to merge because there are too many files in the merge range + var exception = assertThrows(AccumuloException.class, + () -> c.tableOperations().merge(tableName, null, null)); + // message should contain the observed number of files + assertTrue(exception.getMessage().contains("40002")); + // message should contain the max files limit it saw + assertTrue(exception.getMessage().contains("12345")); + + assertEquals(20000, c.tableOperations().listSplits(tableName).size()); + + // attempt to merge smaller ranges with less files, should work.. want to make sure the + // aborted merge did not leave the table in a bad state + Text prev = null; + for (int i = 1000; i <= 20000; i += 1000) { + Text end = new Text(String.format("%06d", i)); + c.tableOperations().merge(tableName, prev, end); + prev = end; + } + + assertEquals(20, c.tableOperations().listSplits(tableName).size()); + try (var tablets = getCluster().getServerContext().getAmple().readTablets() + .forTable(getCluster().getServerContext().getTableId(tableName)).build()) { + assertEquals(40002, + tablets.stream().mapToInt(tabletMetadata -> tabletMetadata.getFiles().size()).sum()); + } + } + } + + @Test + public void merge() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + var ntc = new NewTableConfiguration().withSplits(splits("a b c d e f g h i j k".split(" "))); + createTableAndDisableCompactions(c, tableName, ntc); + try (BatchWriter bw = c.createBatchWriter(tableName)) { + for (String row : "a b c d e f g h i j k".split(" ")) { + Mutation m = new Mutation(row); + m.put("cf", "cq", "value"); + bw.addMutation(m); + } + } + c.tableOperations().setTabletAvailability(tableName, RowRange.closed("d", "e"), + TabletAvailability.HOSTED); + c.tableOperations().setTabletAvailability(tableName, RowRange.closed("e", "f"), + TabletAvailability.UNHOSTED); + c.tableOperations().flush(tableName, null, null, true); + c.tableOperations().merge(tableName, new Text("c1"), new Text("f1")); + assertEquals(8, c.tableOperations().listSplits(tableName).size()); + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), + TableId.of(c.tableOperations().tableIdMap().get(tableName))); + try (Scanner s = c.createScanner(SystemTables.METADATA.tableName())) { + String tid = c.tableOperations().tableIdMap().get(tableName); + s.setRange(new Range(tid + ";g")); + TabletColumnFamily.PREV_ROW_COLUMN.fetch(s); + TabletColumnFamily.AVAILABILITY_COLUMN.fetch(s); + assertEquals(2, s.stream().count()); + for (Entry<Key,Value> rows : s) { + if (TabletColumnFamily.PREV_ROW_COLUMN.hasColumns(rows.getKey())) { + assertEquals("c", TabletColumnFamily.decodePrevEndRow(rows.getValue()).toString()); + } else if (TabletColumnFamily.AVAILABILITY_COLUMN.hasColumns(rows.getKey())) { + assertEquals(TabletAvailability.HOSTED, + TabletAvailabilityUtil.fromValue(rows.getValue())); + } else { + fail("Unknown column"); + } + } + } + } + } + + @Test + public void mergeSize() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + NewTableConfiguration ntc = new NewTableConfiguration() + .withSplits(splits("a b c d e f g h i j k l m n o p q r s t u v w x y z".split(" "))); + createTableAndDisableCompactions(c, tableName, ntc); + try (BatchWriter bw = c.createBatchWriter(tableName)) { + for (String row : "c e f y".split(" ")) { + Mutation m = new Mutation(row); + m.put("cf", "cq", "mersydotesanddozeydotesanlittolamsiedives"); + bw.addMutation(m); + } + } + c.tableOperations().flush(tableName, null, null, true); + Merge merge = new Merge(); - merge.mergomatic(c, tableName, null, null, 100, false); ++ merge.mergomatic(c, tableName, null, null, 100, false, false); + assertArrayEquals("b c d e f x y".split(" "), + toStrings(c.tableOperations().listSplits(tableName))); - merge.mergomatic(c, tableName, null, null, 100, true); ++ merge.mergomatic(c, tableName, null, null, 100, true, false); + assertArrayEquals("c e f y".split(" "), toStrings(c.tableOperations().listSplits(tableName))); + } + } + + @Test + public void noChopMergeTest() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + createTableAndDisableCompactions(c, tableName, new NewTableConfiguration()); + final TableId tableId = TableId.of(c.tableOperations().tableIdMap().get(tableName)); + + // First write 1000 rows to a file in the default tablet + ingest(c, 1000, 1, tableName); + c.tableOperations().flush(tableName, null, null, true); + + log.debug("Metadata after Ingest"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 1); + + // Add splits so we end up with 4 tablets + final SortedSet<Text> splits = new TreeSet<>(); + for (int i = 250; i <= 750; i += 250) { + splits.add(new Text("row_" + String.format("%010d", i))); + } + c.tableOperations().addSplits(tableName, splits); + + log.debug("Metadata after Split"); + verify(c, 1000, 1, tableName); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 4); + + // Go through and delete two blocks of rows, 101 - 200 + // and also 301 - 400 so we can test that the data doesn't come + // back on merge + try (BatchWriter bw = c.createBatchWriter(tableName)) { + byte[] COL_PREFIX = "col_".getBytes(UTF_8); + Text colq = new Text(FastFormat.toZeroPaddedString(0, 7, 10, COL_PREFIX)); + + for (int i = 101; i <= 200; i++) { + Mutation m = new Mutation(new Text("row_" + String.format("%010d", i))); + m.putDelete(new Text("colf"), colq); + bw.addMutation(m); + } + for (int i = 301; i <= 400; i++) { + Mutation m = new Mutation(new Text("row_" + String.format("%010d", i))); + m.putDelete(new Text("colf"), colq); + bw.addMutation(m); + } + } + + c.tableOperations().flush(tableName, null, null, true); + + // compact the first 2 tablets so the new files with the deletes are gone + // so we can test that the data does not come back when the 3rd tablet is + // merged back with the other tablets as it still contains the original file + c.tableOperations().compact(tableName, new CompactionConfig().setStartRow(null) + .setEndRow(List.copyOf(splits).get(1)).setWait(true)); + log.debug("Metadata after deleting rows 101 - 200 and 301 - 400"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 4); + + // Merge and print results + c.tableOperations().merge(tableName, null, null); + log.debug("Metadata after Merge"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 4); + + // Verify that the deleted rows can't be read after merge + verify(c, 100, 1, tableName); + verifyNoRows(c, 100, 101, tableName); + verify(c, 100, 201, tableName); + verifyNoRows(c, 100, 301, tableName); + verify(c, 600, 401, tableName); + + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), tableId); + } + } + + @Test + public void noChopMergeDeleteAcrossTablets() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + createTableAndDisableCompactions(c, tableName, new NewTableConfiguration()); + final TableId tableId = TableId.of(c.tableOperations().tableIdMap().get(tableName)); + + // First write 1000 rows to a file in the default tablet + ingest(c, 1000, 1, tableName); + c.tableOperations().flush(tableName, null, null, true); + + log.debug("Metadata after Ingest"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 1); + + // Add splits so we end up with 10 tablets + final SortedSet<Text> splits = new TreeSet<>(); + for (int i = 100; i <= 900; i += 100) { + splits.add(new Text("row_" + String.format("%010d", i))); + } + c.tableOperations().addSplits(tableName, splits); + + log.debug("Metadata after Split"); + verify(c, 1000, 1, tableName); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 10); + + // Go through and delete three blocks of rows + // 151 - 250, 451 - 550, 751 - 850 + try (BatchWriter bw = c.createBatchWriter(tableName)) { + byte[] COL_PREFIX = "col_".getBytes(UTF_8); + Text colq = new Text(FastFormat.toZeroPaddedString(0, 7, 10, COL_PREFIX)); + + for (int j = 0; j <= 2; j++) { + for (int i = 151; i <= 250; i++) { + Mutation m = new Mutation(new Text("row_" + String.format("%010d", i + (j * 300)))); + m.putDelete(new Text("colf"), colq); + bw.addMutation(m); + } + } + } + + c.tableOperations().flush(tableName, null, null, true); + + log.debug("Metadata after deleting rows 151 - 250, 451 - 550, 751 - 850"); + // compact some of the tablets with deletes so we can test that the data does not come back + c.tableOperations().compact(tableName, + new CompactionConfig().setStartRow(new Text("row_" + String.format("%010d", 150))) + .setEndRow(new Text("row_" + String.format("%010d", 250))).setWait(true)); + c.tableOperations().compact(tableName, + new CompactionConfig().setStartRow(new Text("row_" + String.format("%010d", 750))) + .setEndRow(new Text("row_" + String.format("%010d", 850))).setWait(true)); + // Should be 16 files (10 for the original splits plus 2 extra files per deletion range across + // tablets) + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 12); + + c.tableOperations().merge(tableName, null, null); + log.debug("Metadata after Merge"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 12); + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), tableId); + + // Verify that the deleted rows can't be read after merge + verify(c, 150, 1, tableName); + verifyNoRows(c, 100, 151, tableName); + verify(c, 200, 251, tableName); + verifyNoRows(c, 100, 451, tableName); + verify(c, 200, 551, tableName); + verifyNoRows(c, 100, 751, tableName); + verify(c, 150, 851, tableName); + + // Compact and verify we clean up all the files and only 1 left + // Verify only 700 entries + c.tableOperations().compact(tableName, new CompactionConfig().setWait(true)); + log.debug("Metadata after compact"); + // Should just be 1 file with infinite range + Map<StoredTabletFile,DataFileValue> files = + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 1); + assertEquals(new Range(), files.keySet().stream().findFirst().orElseThrow().getRange()); + assertEquals(700, files.values().stream().findFirst().orElseThrow().getNumEntries()); + } + } + + // Multiple splits/deletes/merges to show ranges work and carry forward + // Testing that we can split -> delete, merge, split -> delete, merge + // with deletions across boundaries + @Test + public void noChopMergeDeleteAcrossTabletsMultiple() throws Exception { + // Run initial test to populate table and merge which adds ranges to files + noChopMergeDeleteAcrossTablets(); + + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + final TableId tableId = TableId.of(c.tableOperations().tableIdMap().get(tableName)); + + log.debug("Metadata after initial test run"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, -1); + + // Add splits so we end up with 10 tablets + final SortedSet<Text> splits = new TreeSet<>(); + for (int i = 100; i <= 900; i += 100) { + splits.add(new Text("row_" + String.format("%010d", i))); + } + c.tableOperations().addSplits(tableName, splits); + + log.debug("Metadata after Split for second time"); + // Verify that the deleted rows can't be read after merge + verify(c, 150, 1, tableName); + verifyNoRows(c, 100, 151, tableName); + verify(c, 200, 251, tableName); + verifyNoRows(c, 100, 451, tableName); + verify(c, 200, 551, tableName); + verifyNoRows(c, 100, 751, tableName); + verify(c, 150, 851, tableName); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, -1); + + c.tableOperations().flush(tableName, null, null, true); + + // Go through and also delete 651 - 700 + try (BatchWriter bw = c.createBatchWriter(tableName)) { + byte[] COL_PREFIX = "col_".getBytes(UTF_8); + Text colq = new Text(FastFormat.toZeroPaddedString(0, 7, 10, COL_PREFIX)); + + for (int i = 651; i <= 700; i++) { + Mutation m = new Mutation(new Text("row_" + String.format("%010d", i))); + m.putDelete(new Text("colf"), colq); + bw.addMutation(m); + } + } + + c.tableOperations().flush(tableName, null, null, true); + + log.debug("Metadata after deleting rows 151 - 250, 451 - 550, 651 - 700, 751 - 850"); + // compact some of the tablets with deletes so we can test that the data does not come back + c.tableOperations().compact(tableName, + new CompactionConfig().setStartRow(new Text("row_" + String.format("%010d", 150))) + .setEndRow(new Text("row_" + String.format("%010d", 700))).setWait(true)); + + // Re-merge a second time after deleting more rows + c.tableOperations().merge(tableName, null, null); + log.debug("Metadata after second Merge"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, -1); + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), tableId); + + // Verify that the deleted rows can't be read after merge + verify(c, 150, 1, tableName); + verifyNoRows(c, 100, 151, tableName); + verify(c, 200, 251, tableName); + verifyNoRows(c, 100, 451, tableName); + verify(c, 100, 551, tableName); + verifyNoRows(c, 50, 651, tableName); + verify(c, 50, 701, tableName); + verifyNoRows(c, 100, 751, tableName); + verify(c, 150, 851, tableName); + } + } + + // Tests that after we merge and fence files, we can split and then + // merge a second time the same table which shows splits/merges work + // for files that already have ranges + @Test + public void noChopMergeTestMultipleMerges() throws Exception { + // Do initial merge which will fence off files + noChopMergeTest(); + + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + final TableId tableId = TableId.of(c.tableOperations().tableIdMap().get(tableName)); + + log.debug("Metadata after initial no chop merge test"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, 4); + + // Add splits so we end up with 4 tablets + final SortedSet<Text> splits = new TreeSet<>(); + for (int i = 250; i <= 750; i += 250) { + splits.add(new Text("row_" + String.format("%010d", i))); + } + c.tableOperations().addSplits(tableName, splits); + + log.debug("Metadata after Split"); + // Verify after splitting for the second time + verify(c, 100, 1, tableName); + verifyNoRows(c, 100, 101, tableName); + verify(c, 100, 201, tableName); + verifyNoRows(c, 100, 301, tableName); + verify(c, 600, 401, tableName); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, -1); + + // Re-Merge and print results. This tests merging with files + // that already have a range + c.tableOperations().merge(tableName, null, null); + log.debug("Metadata after Merge"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId, -1); + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), tableId); + + // Verify that the deleted rows can't be read after merge + verify(c, 100, 1, tableName); + verifyNoRows(c, 100, 101, tableName); + verify(c, 100, 201, tableName); + verifyNoRows(c, 100, 301, tableName); + verify(c, 600, 401, tableName); + } + } + + private static void createTableAndDisableCompactions(AccumuloClient c, String tableName, + NewTableConfiguration ntc) throws Exception { + // disable compactions + ntc.setProperties(Map.of(Property.TABLE_MAJC_RATIO.getKey(), "9999")); + c.tableOperations().create(tableName, ntc); + } + + public static void ingest(AccumuloClient accumuloClient, int rows, int offset, String tableName) + throws Exception { + IngestParams params = new IngestParams(accumuloClient.properties(), tableName, rows); + params.cols = 1; + params.dataSize = 10; + params.startRow = offset; + params.columnFamily = "colf"; + params.createTable = true; + TestIngest.ingest(accumuloClient, params); + } + + private static void verify(AccumuloClient accumuloClient, int rows, int offset, String tableName) + throws Exception { + VerifyParams params = new VerifyParams(accumuloClient.properties(), tableName, rows); + params.rows = rows; + params.dataSize = 10; + params.startRow = offset; + params.columnFamily = "colf"; + params.cols = 1; + VerifyIngest.verifyIngest(accumuloClient, params); + } + + private static void verifyNoRows(AccumuloClient accumuloClient, int rows, int offset, + String tableName) throws Exception { + try { + verify(accumuloClient, rows, offset, tableName); + fail("Should have failed"); + } catch (AccumuloException e) { + assertTrue(e.getMessage().contains("Did not read expected number of rows. Saw 0")); + } + } + + private String[] toStrings(Collection<Text> listSplits) { + String[] result = new String[listSplits.size()]; + int i = 0; + for (Text t : listSplits) { + result[i++] = t.toString(); + } + return result; + } + + private String[] ns(String... strings) { + return strings; + } + + @Test + public void mergeTest() throws Exception { + int tc = 0; + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + runMergeTest(c, tableName + tc++, ns(), ns(), ns("l", "m", "n"), ns(null, "l"), + ns(null, "n")); + + runMergeTest(c, tableName + tc++, ns("m"), ns(), ns("l", "m", "n"), ns(null, "l"), + ns(null, "n")); + runMergeTest(c, tableName + tc++, ns("m"), ns("m"), ns("l", "m", "n"), ns("m", "n"), + ns(null, "z")); + runMergeTest(c, tableName + tc++, ns("m"), ns("m"), ns("l", "m", "n"), ns(null, "b"), + ns("l", "m")); + + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns(), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns(null, "a"), ns(null, "s")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("m", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns(null, "a"), ns("c", "m")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns(null, "a"), ns("n", "r")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("b", "c"), ns(null, "s")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "m"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("m", "n"), ns(null, "s")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("b", "c"), ns("q", "r")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "m", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns(null, "a"), ns("aa", "b")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "m", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("r", "s"), ns(null, "z")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "m", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("b", "c"), ns("l", "m")); + runMergeTest(c, tableName + tc++, ns("b", "m", "r"), ns("b", "m", "r"), + ns("a", "b", "c", "l", "m", "n", "q", "r", "s"), ns("m", "n"), ns("q", "r")); + } + } + + private void runMergeTest(AccumuloClient c, String table, String[] splits, + String[] expectedSplits, String[] inserts, String[] start, String[] end) throws Exception { + int count = 0; + + for (String s : start) { + for (String e : end) { + runMergeTest(c, table + "_" + count++, splits, expectedSplits, inserts, s, e); + } + } + } + + private void runMergeTest(AccumuloClient client, String table, String[] splits, + String[] expectedSplits, String[] inserts, String start, String end) throws Exception { + log.debug( + "Running merge test " + table + " " + Arrays.asList(splits) + " " + start + " " + end); + + SortedSet<Text> splitSet = splits(splits); + + NewTableConfiguration ntc = new NewTableConfiguration().setTimeType(TimeType.LOGICAL); + if (!splitSet.isEmpty()) { + ntc = ntc.withSplits(splitSet); + } + createTableAndDisableCompactions(client, table, ntc); + + HashSet<String> expected = new HashSet<>(); + try (BatchWriter bw = client.createBatchWriter(table)) { + for (String row : inserts) { + Mutation m = new Mutation(row); + m.put("cf", "cq", row); + bw.addMutation(m); + expected.add(row); + } + } + + log.debug("Before Merge"); + client.tableOperations().flush(table, null, null, true); + TableId tableId = TableId.of(client.tableOperations().tableIdMap().get(table)); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId); + + client.tableOperations().merge(table, start == null ? null : new Text(start), + end == null ? null : new Text(end)); + + client.tableOperations().flush(table, null, null, true); + log.debug("After Merge"); + printAndVerifyFileMetadata(getCluster().getServerContext(), tableId); + // Verify that the MERGED marker was cleared + verifyMergedMarkerCleared(getCluster().getServerContext(), tableId); + + try (Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) { + + HashSet<String> observed = new HashSet<>(); + for (Entry<Key,Value> entry : scanner) { + String row = entry.getKey().getRowData().toString(); + if (!observed.add(row)) { + throw new Exception("Saw data twice " + table + " " + row); + } + } + + if (!observed.equals(expected)) { + throw new Exception("data inconsistency " + table + " " + observed + " != " + expected); + } + + HashSet<Text> currentSplits = new HashSet<>(client.tableOperations().listSplits(table)); + HashSet<Text> ess = new HashSet<>(); + for (String es : expectedSplits) { + ess.add(new Text(es)); + } + + if (!currentSplits.equals(ess)) { + throw new Exception("split inconsistency " + table + " " + currentSplits + " != " + ess); + } + } + } + + // Test that merge handles metadata from compactions + @Test + public void testCompactionMetadata() throws Exception { + try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { + String tableName = getUniqueNames(1)[0]; + c.tableOperations().create(tableName); + + var split = new Text("m"); + c.tableOperations().addSplits(tableName, new TreeSet<>(List.of(split))); + + TableId tableId = getCluster().getServerContext().getTableId(tableName); + + // add metadata from compactions to tablets prior to merge + try (var tabletsMutator = getCluster().getServerContext().getAmple().mutateTablets()) { + for (var extent : List.of(new KeyExtent(tableId, split, null), + new KeyExtent(tableId, null, split))) { + var tablet = tabletsMutator.mutateTablet(extent); + ExternalCompactionId ecid = ExternalCompactionId.generate(UUID.randomUUID()); + FateInstanceType type = FateInstanceType.fromTableId(tableId); + FateId fateId = FateId.from(type, UUID.randomUUID()); + + ReferencedTabletFile tmpFile = + ReferencedTabletFile.of(new Path("file:///accumulo/tables/t-0/b-0/c1.rf")); + ResourceGroupId ceid = ResourceGroupId.of("G1"); + Set<StoredTabletFile> jobFiles = + Set.of(StoredTabletFile.of(new Path("file:///accumulo/tables/t-0/b-0/b2.rf"))); + CompactionMetadata ecMeta = new CompactionMetadata(jobFiles, tmpFile, "localhost:4444", + CompactionKind.SYSTEM, (short) 2, ceid, false, fateId); + tablet.putExternalCompaction(ecid, ecMeta); + tablet.mutate(); + } + } + + // ensure data is in metadata table as expected + try (var tablets = + getCluster().getServerContext().getAmple().readTablets().forTable(tableId).build()) { + for (var tablet : tablets) { + assertFalse(tablet.getExternalCompactions().isEmpty()); + } + } + + c.tableOperations().merge(tableName, null, null); + + // ensure merge operation remove compaction entries + try (var tablets = + getCluster().getServerContext().getAmple().readTablets().forTable(tableId).build()) { + for (var tablet : tablets) { + assertTrue(tablet.getExternalCompactions().isEmpty()); + } + } + } + } ++ ++ @Test ++ public void testDryRunMerge() throws Exception { ++ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { ++ String tableName = getUniqueNames(1)[0]; ++ c.tableOperations().create(tableName); ++ c.tableOperations().addSplits(tableName, ++ new TreeSet<>(List.of(new Text("row0"), new Text("row1"), new Text("row10")))); ++ try (BatchWriter writer = c.createBatchWriter(tableName)) { ++ for (int i = 0; i < 100; i++) { ++ Mutation m = new Mutation(String.format("row00%d", i)); ++ m.put("cf", "cq", "Test Value"); ++ writer.addMutation(m); ++ } ++ } ++ c.tableOperations().flush(tableName, null, null, true); ++ ++ TableId tableId = getCluster().getServerContext().getTableId(tableName); ++ try (var tablets = ++ getCluster().getServerContext().getAmple().readTablets().forTable(tableId).build()) { ++ assertEquals(4, tablets.stream().count()); ++ } ++ ++ Merge merge = new Merge(); ++ merge.mergomatic(c, tableName, null, null, 1073741824, false, true); ++ try (var tablets = ++ getCluster().getServerContext().getAmple().readTablets().forTable(tableId).build()) { ++ assertEquals(4, tablets.stream().count()); ++ } ++ } ++ } ++ ++ @Test ++ public void testMetadataTableDoesNotMerge() { ++ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { ++ Merge merge = new Merge(); ++ Throwable ex = assertThrows(Merge.MergeException.class, () -> merge.mergomatic(c, ++ SystemTables.METADATA.tableName(), null, new Text("~"), 1, false, false)); ++ assertInstanceOf(IllegalArgumentException.class, ex.getCause()); ++ assertEquals( ++ "java.lang.IllegalArgumentException: cannot merge tablets on the accumulo.metadata table", ++ ex.getMessage()); ++ } ++ } ++ ++ @Test ++ public void testRootTableDoesNotMerge() { ++ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) { ++ Merge merge = new Merge(); ++ Exception e = assertThrows(Merge.MergeException.class, () -> merge.mergomatic(c, ++ SystemTables.ROOT.tableName(), null, new Text("~"), 1, false, false)); ++ assertInstanceOf(IllegalArgumentException.class, e.getCause()); ++ assertEquals( ++ "java.lang.IllegalArgumentException: cannot merge tablets on the accumulo.root table", ++ e.getMessage()); ++ try (var tablets = getCluster().getServerContext().getAmple().readTablets() ++ .forTable(SystemTables.ROOT.tableId()).build()) { ++ assertEquals(1, tablets.stream().count()); ++ } ++ } ++ } ++ +}
