You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by GitBox <gi...@apache.org> on 2020/10/06 17:23:51 UTC
[GitHub] [accumulo] ctubbsii commented on a change in pull request #1723: WIP Accumulo t#1628
ctubbsii commented on a change in pull request #1723:
URL: https://github.com/apache/accumulo/pull/1723#discussion_r500458655
##########
File path: core/src/main/java/org/apache/accumulo/core/client/admin/compaction/CompactionSelector.java
##########
@@ -78,5 +78,5 @@ public Selection(Collection<CompactableFile> filesToCompact) {
}
}
- Selection select(SelectionParameters sparams);
Review comment:
We probably don't want to throw checked exceptions here. The assumption should be that this method doesn't throw any normal exceptions, and that the selector is responsible for handling everything. That way, the only things that can fall through here are RuntimeExceptions.
##########
File path: test/src/main/java/org/apache/accumulo/test/CompactionIT.java
##########
@@ -74,595 +70,729 @@
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
public class CompactionIT extends SharedMiniClusterBase {
- @SuppressFBWarnings(value = "PREDICTABLE_RANDOM",
- justification = "predictable random is okay for testing")
- public static class TestPlanner implements CompactionPlanner {
-
- private int filesPerCompaction;
- private List<CompactionExecutorId> executorIds;
- private EnumSet<CompactionKind> kindsToProcess = EnumSet.noneOf(CompactionKind.class);
- private Random rand = new Random();
-
- @Override
- public void init(InitParameters params) {
- var executors = Integer.parseInt(params.getOptions().get("executors"));
- this.filesPerCompaction = Integer.parseInt(params.getOptions().get("filesPerCompaction"));
- this.executorIds = new ArrayList<>();
- for (String kind : params.getOptions().get("process").split(",")) {
- kindsToProcess.add(CompactionKind.valueOf(kind.toUpperCase()));
- }
-
- for (int i = 0; i < executors; i++) {
- var ceid = params.getExecutorManager().createExecutor("e" + i, 2);
- executorIds.add(ceid);
- }
+ public static final Logger log = LoggerFactory.getLogger(CompactionIT.class);
- }
+ @SuppressFBWarnings(value = "PREDICTABLE_RANDOM",
+ justification = "predictable random is okay for testing")
+ public static class TestPlanner implements CompactionPlanner {
+
+ private int filesPerCompaction;
+ private List<CompactionExecutorId> executorIds;
+ private EnumSet<CompactionKind> kindsToProcess = EnumSet.noneOf(CompactionKind.class);
+ private Random rand = new Random();
+
+ @Override
+ public void init(InitParameters params) {
+ var executors = Integer.parseInt(params.getOptions().get("executors"));
+ this.filesPerCompaction = Integer.parseInt(params.getOptions().get("filesPerCompaction"));
+ this.executorIds = new ArrayList<>();
+ for (String kind : params.getOptions().get("process").split(",")) {
+ kindsToProcess.add(CompactionKind.valueOf(kind.toUpperCase()));
+ }
+
+ for (int i = 0; i < executors; i++) {
+ var ceid = params.getExecutorManager().createExecutor("e" + i, 2);
+ executorIds.add(ceid);
+ }
+
+ }
+
+ static String getFirstChar(CompactableFile cf) {
+ return cf.getFileName().substring(0, 1);
+ }
+
+ @Override
+ public CompactionPlan makePlan(PlanningParameters params) {
- static String getFirstChar(CompactableFile cf) {
- return cf.getFileName().substring(0, 1);
+ if (Boolean.parseBoolean(params.getExecutionHints().getOrDefault("compact_all", "false"))) {
+ return params.createPlanBuilder()
+ .addJob(1, executorIds.get(rand.nextInt(executorIds.size())), params.getCandidates())
+ .build();
+ }
+
+ if (kindsToProcess.contains(params.getKind())) {
+ var planBuilder = params.createPlanBuilder();
+
+ // Group files by first char, like F for flush files or C for compaction produced files.
+ // This prevents F and C files from compacting together, which makes it easy to reason about
+ // the number of expected files produced by compactions from known number of F files.
+ params.getCandidates().stream().collect(Collectors.groupingBy(TestPlanner::getFirstChar))
+ .values().forEach(files -> {
+ for (int i = filesPerCompaction; i <= files.size(); i += filesPerCompaction) {
+ planBuilder.addJob(1, executorIds.get(rand.nextInt(executorIds.size())),
+ files.subList(i - filesPerCompaction, i));
+ }
+ });
+
+ return planBuilder.build();
+ } else {
+ return params.createPlanBuilder().build();
+ }
+ }
}
- @Override
- public CompactionPlan makePlan(PlanningParameters params) {
-
- if (Boolean.parseBoolean(params.getExecutionHints().getOrDefault("compact_all", "false"))) {
- return params.createPlanBuilder()
- .addJob(1, executorIds.get(rand.nextInt(executorIds.size())), params.getCandidates())
- .build();
- }
-
- if (kindsToProcess.contains(params.getKind())) {
- var planBuilder = params.createPlanBuilder();
-
- // Group files by first char, like F for flush files or C for compaction produced files.
- // This prevents F and C files from compacting together, which makes it easy to reason about
- // the number of expected files produced by compactions from known number of F files.
- params.getCandidates().stream().collect(Collectors.groupingBy(TestPlanner::getFirstChar))
- .values().forEach(files -> {
- for (int i = filesPerCompaction; i <= files.size(); i += filesPerCompaction) {
- planBuilder.addJob(1, executorIds.get(rand.nextInt(executorIds.size())),
- files.subList(i - filesPerCompaction, i));
- }
- });
+ @BeforeClass
+ public static void setup() throws Exception {
+ SharedMiniClusterBase.startMiniClusterWithConfig((miniCfg, coreSite) -> {
+ Map<String, String> siteCfg = new HashMap<>();
+
+ var csp = Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey();
+ siteCfg.put(csp + "cs1.planner", TestPlanner.class.getName());
+ siteCfg.put(csp + "cs1.planner.opts.executors", "3");
+ siteCfg.put(csp + "cs1.planner.opts.filesPerCompaction", "5");
+ siteCfg.put(csp + "cs1.planner.opts.process", "SYSTEM");
+
+ siteCfg.put(csp + "cs2.planner", TestPlanner.class.getName());
+ siteCfg.put(csp + "cs2.planner.opts.executors", "2");
+ siteCfg.put(csp + "cs2.planner.opts.filesPerCompaction", "7");
+ siteCfg.put(csp + "cs2.planner.opts.process", "SYSTEM");
+
+ siteCfg.put(csp + "cs3.planner", TestPlanner.class.getName());
+ siteCfg.put(csp + "cs3.planner.opts.executors", "1");
+ siteCfg.put(csp + "cs3.planner.opts.filesPerCompaction", "3");
+ siteCfg.put(csp + "cs3.planner.opts.process", "USER");
+
+ siteCfg.put(csp + "cs4.planner", TestPlanner.class.getName());
+ siteCfg.put(csp + "cs4.planner.opts.executors", "2");
+ siteCfg.put(csp + "cs4.planner.opts.filesPerCompaction", "11");
+ siteCfg.put(csp + "cs4.planner.opts.process", "USER");
+
+ // this is meant to be dynamically reconfigured
+ siteCfg.put(csp + "recfg.planner", TestPlanner.class.getName());
+ siteCfg.put(csp + "recfg.planner.opts.executors", "2");
+ siteCfg.put(csp + "recfg.planner.opts.filesPerCompaction", "11");
+ siteCfg.put(csp + "recfg.planner.opts.process", "SYSTEM");
+
+ miniCfg.setSiteConfig(siteCfg);
+ });
+ }
- return planBuilder.build();
- } else {
- return params.createPlanBuilder().build();
- }
+ @AfterClass
+ public static void teardown() {
+ SharedMiniClusterBase.stopMiniCluster();
}
- }
-
- @BeforeClass
- public static void setup() throws Exception {
- SharedMiniClusterBase.startMiniClusterWithConfig((miniCfg, coreSite) -> {
- Map<String,String> siteCfg = new HashMap<>();
-
- var csp = Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey();
- siteCfg.put(csp + "cs1.planner", TestPlanner.class.getName());
- siteCfg.put(csp + "cs1.planner.opts.executors", "3");
- siteCfg.put(csp + "cs1.planner.opts.filesPerCompaction", "5");
- siteCfg.put(csp + "cs1.planner.opts.process", "SYSTEM");
-
- siteCfg.put(csp + "cs2.planner", TestPlanner.class.getName());
- siteCfg.put(csp + "cs2.planner.opts.executors", "2");
- siteCfg.put(csp + "cs2.planner.opts.filesPerCompaction", "7");
- siteCfg.put(csp + "cs2.planner.opts.process", "SYSTEM");
-
- siteCfg.put(csp + "cs3.planner", TestPlanner.class.getName());
- siteCfg.put(csp + "cs3.planner.opts.executors", "1");
- siteCfg.put(csp + "cs3.planner.opts.filesPerCompaction", "3");
- siteCfg.put(csp + "cs3.planner.opts.process", "USER");
-
- siteCfg.put(csp + "cs4.planner", TestPlanner.class.getName());
- siteCfg.put(csp + "cs4.planner.opts.executors", "2");
- siteCfg.put(csp + "cs4.planner.opts.filesPerCompaction", "11");
- siteCfg.put(csp + "cs4.planner.opts.process", "USER");
-
- // this is meant to be dynamically reconfigured
- siteCfg.put(csp + "recfg.planner", TestPlanner.class.getName());
- siteCfg.put(csp + "recfg.planner.opts.executors", "2");
- siteCfg.put(csp + "recfg.planner.opts.filesPerCompaction", "11");
- siteCfg.put(csp + "recfg.planner.opts.process", "SYSTEM");
-
- miniCfg.setSiteConfig(siteCfg);
- });
- }
-
- @AfterClass
- public static void teardown() {
- SharedMiniClusterBase.stopMiniCluster();
- }
-
- @After
- public void cleanup() {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- client.tableOperations().list().stream()
- .filter(tableName -> !tableName.startsWith("accumulo.")).forEach(tableName -> {
- try {
- client.tableOperations().delete(tableName);
- } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
- throw new RuntimeException(e);
- }
- });
+
+ @After
+ public void cleanup() {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ client.tableOperations().list().stream()
+ .filter(tableName -> !tableName.startsWith("accumulo.")).forEach(tableName -> {
+ try {
+ client.tableOperations().delete(tableName);
+ } catch (AccumuloException | AccumuloSecurityException | TableNotFoundException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
}
- }
- @Test
- public void testReconfigureCompactionService() throws Exception {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- createTable(client, "rctt", "recfg");
+ @Test
+ public void testReconfigureCompactionService() throws Exception {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ createTable(client, "rctt", "recfg");
- addFiles(client, "rctt", 22);
+ addFiles(client, "rctt", 22);
- while (getFiles(client, "rctt").size() > 2) {
- Thread.sleep(100);
- }
+ while (getFiles(client, "rctt").size() > 2) {
+ Thread.sleep(100);
+ }
- assertEquals(2, getFiles(client, "rctt").size());
+ assertEquals(2, getFiles(client, "rctt").size());
- client.instanceOperations().setProperty(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey()
- + "recfg.planner.opts.filesPerCompaction", "5");
- client.instanceOperations().setProperty(
- Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "recfg.planner.opts.executors", "1");
+ client.instanceOperations().setProperty(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey()
+ + "recfg.planner.opts.filesPerCompaction", "5");
+ client.instanceOperations().setProperty(
+ Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "recfg.planner.opts.executors", "1");
- addFiles(client, "rctt", 10);
+ addFiles(client, "rctt", 10);
- while (getFiles(client, "rctt").size() > 4) {
- Thread.sleep(100);
- }
+ while (getFiles(client, "rctt").size() > 4) {
+ Thread.sleep(100);
+ }
- assertEquals(4, getFiles(client, "rctt").size());
+ assertEquals(4, getFiles(client, "rctt").size());
+ }
}
- }
-
- @Test
- public void testAddCompactionService() throws Exception {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- client.instanceOperations().setProperty(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey()
- + "newcs.planner.opts.filesPerCompaction", "7");
- client.instanceOperations().setProperty(
- Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.process",
- "SYSTEM");
- client.instanceOperations().setProperty(
- Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.executors", "3");
- client.instanceOperations().setProperty(
- Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner",
- TestPlanner.class.getName());
-
- createTable(client, "acst", "newcs");
-
- addFiles(client, "acst", 42);
-
- while (getFiles(client, "acst").size() > 6) {
- Thread.sleep(100);
- }
-
- assertEquals(6, getFiles(client, "acst").size());
+
+ @Test
+ public void testAddCompactionService() throws Exception {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ client.instanceOperations().setProperty(Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey()
+ + "newcs.planner.opts.filesPerCompaction", "7");
+ client.instanceOperations().setProperty(
+ Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.process",
+ "SYSTEM");
+ client.instanceOperations().setProperty(
+ Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner.opts.executors", "3");
+ client.instanceOperations().setProperty(
+ Property.TSERV_COMPACTION_SERVICE_PREFIX.getKey() + "newcs.planner",
+ TestPlanner.class.getName());
+
+ createTable(client, "acst", "newcs");
+
+ addFiles(client, "acst", 42);
+
+ while (getFiles(client, "acst").size() > 6) {
+ Thread.sleep(100);
+ }
+
+ assertEquals(6, getFiles(client, "acst").size());
+ }
}
- }
-
- /**
- * Test ensures that system compactions are dispatched to a configured compaction service. The
- * compaction services produce a very specific number of files, so the test indirectly checks
- * dispatching worked by observing how many files a tablet ends up with.
- */
- @Test
- public void testDispatchSystem() throws Exception {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- createTable(client, "dst1", "cs1");
- createTable(client, "dst2", "cs2");
-
- addFiles(client, "dst1", 14);
- addFiles(client, "dst2", 13);
-
- assertTrue(getFiles(client, "dst1").size() >= 6);
- assertTrue(getFiles(client, "dst2").size() >= 7);
-
- addFiles(client, "dst1", 1);
- addFiles(client, "dst2", 1);
-
- while (getFiles(client, "dst1").size() > 3 || getFiles(client, "dst2").size() > 2) {
- Thread.sleep(100);
- }
-
- assertEquals(3, getFiles(client, "dst1").size());
- assertEquals(2, getFiles(client, "dst2").size());
+
+ /**
+ * Test ensures that system compactions are dispatched to a configured compaction service. The
+ * compaction services produce a very specific number of files, so the test indirectly checks
+ * dispatching worked by observing how many files a tablet ends up with.
+ */
+ @Test
+ public void testDispatchSystem() throws Exception {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ createTable(client, "dst1", "cs1");
+ createTable(client, "dst2", "cs2");
+
+ addFiles(client, "dst1", 14);
+ addFiles(client, "dst2", 13);
+
+ assertTrue(getFiles(client, "dst1").size() >= 6);
+ assertTrue(getFiles(client, "dst2").size() >= 7);
+
+ addFiles(client, "dst1", 1);
+ addFiles(client, "dst2", 1);
+
+ while (getFiles(client, "dst1").size() > 3 || getFiles(client, "dst2").size() > 2) {
+ Thread.sleep(100);
+ }
+
+ assertEquals(3, getFiles(client, "dst1").size());
+ assertEquals(2, getFiles(client, "dst2").size());
+ }
}
- }
- @Test
- public void testDispatchUser() throws Exception {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- createTable(client, "dut1", "cs3");
- createTable(client, "dut2", "cs3", "special", "cs4");
+ @Test
+ public void testDispatchUser() throws Exception {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ createTable(client, "dut1", "cs3");
+ createTable(client, "dut2", "cs3", "special", "cs4");
- addFiles(client, "dut1", 6);
- addFiles(client, "dut2", 33);
+ addFiles(client, "dut1", 6);
+ addFiles(client, "dut2", 33);
- assertEquals(6, getFiles(client, "dut1").size());
- assertEquals(33, getFiles(client, "dut2").size());
+ assertEquals(6, getFiles(client, "dut1").size());
+ assertEquals(33, getFiles(client, "dut2").size());
- client.tableOperations().compact("dut1", new CompactionConfig().setWait(false));
+ client.tableOperations().compact("dut1", new CompactionConfig().setWait(false));
- // The hint should cause the compaction to dispatch to service cs4 which will produce a
- // different number of files.
- client.tableOperations().compact("dut2", new CompactionConfig().setWait(false)
- .setExecutionHints(Map.of("compaction_type", "special")));
+ // The hint should cause the compaction to dispatch to service cs4 which will produce a
+ // different number of files.
+ client.tableOperations().compact("dut2", new CompactionConfig().setWait(false)
+ .setExecutionHints(Map.of("compaction_type", "special")));
- while (getFiles(client, "dut1").size() > 2 || getFiles(client, "dut2").size() > 3) {
- Thread.sleep(100);
- }
+ while (getFiles(client, "dut1").size() > 2 || getFiles(client, "dut2").size() > 3) {
+ Thread.sleep(100);
+ }
- assertEquals(2, getFiles(client, "dut1").size());
- assertEquals(3, getFiles(client, "dut2").size());
+ assertEquals(2, getFiles(client, "dut1").size());
+ assertEquals(3, getFiles(client, "dut2").size());
- // The way the compaction services were configured, they would never converge to one file for
- // the user compactions. However Accumulo will keep asking the planner for a plan until a user
- // compaction converges to one file. So cancel the compactions.
- client.tableOperations().cancelCompaction("dut1");
- client.tableOperations().cancelCompaction("dut2");
+ // The way the compaction services were configured, they would never converge to one file for
+ // the user compactions. However Accumulo will keep asking the planner for a plan until a user
+ // compaction converges to one file. So cancel the compactions.
+ client.tableOperations().cancelCompaction("dut1");
+ client.tableOperations().cancelCompaction("dut2");
- assertEquals(2, getFiles(client, "dut1").size());
- assertEquals(3, getFiles(client, "dut2").size());
+ assertEquals(2, getFiles(client, "dut1").size());
+ assertEquals(3, getFiles(client, "dut2").size());
- client.tableOperations().compact("dut1",
- new CompactionConfig().setWait(true).setExecutionHints(Map.of("compact_all", "true")));
- client.tableOperations().compact("dut2",
- new CompactionConfig().setWait(true).setExecutionHints(Map.of("compact_all", "true")));
+ client.tableOperations().compact("dut1",
+ new CompactionConfig().setWait(true).setExecutionHints(Map.of("compact_all", "true")));
+ client.tableOperations().compact("dut2",
+ new CompactionConfig().setWait(true).setExecutionHints(Map.of("compact_all", "true")));
+
+ assertEquals(1, getFiles(client, "dut1").size());
+ assertEquals(1, getFiles(client, "dut2").size());
+ }
- assertEquals(1, getFiles(client, "dut1").size());
- assertEquals(1, getFiles(client, "dut2").size());
}
- }
+ @Test
+ public void testTooManyDeletes() throws Exception {
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ Map<String,
+ String> props = Map.of(Property.TABLE_COMPACTION_SELECTOR.getKey(),
+ TooManyDeletesSelector.class.getName(),
+ Property.TABLE_COMPACTION_SELECTOR_OPTS.getKey() + "threshold", ".4");
+ var deleteSummarizerCfg =
+ SummarizerConfiguration.builder(DeletesSummarizer.class.getName()).build();
+ client.tableOperations().create("tmd_selector", new NewTableConfiguration()
+ .setProperties(props).enableSummarization(deleteSummarizerCfg));
+ client.tableOperations().create("tmd_control1",
+ new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
+ client.tableOperations().create("tmd_control2",
+ new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
+ client.tableOperations().create("tmd_control3",
+ new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
+
+ addFile(client, "tmd_selector", 1, 1000, false);
+ addFile(client, "tmd_selector", 1, 1000, true);
+
+ addFile(client, "tmd_control1", 1, 1000, false);
+ addFile(client, "tmd_control1", 1, 1000, true);
+
+ addFile(client, "tmd_control2", 1, 1000, false);
+ addFile(client, "tmd_control2", 1000, 2000, false);
+
+ addFile(client, "tmd_control3", 1, 2000, false);
+ addFile(client, "tmd_control3", 1, 1000, true);
+
+ assertEquals(2, getFiles(client, "tmd_control1").size());
+ assertEquals(2, getFiles(client, "tmd_control2").size());
+ assertEquals(2, getFiles(client, "tmd_control3").size());
+
+ while (getFiles(client, "tmd_selector").size() != 0) {
+ Thread.sleep(100);
+ }
+
+ assertEquals(2, getFiles(client, "tmd_control1").size());
+ assertEquals(2, getFiles(client, "tmd_control2").size());
+ assertEquals(2, getFiles(client, "tmd_control3").size());
+
+ var cc1 = new CompactionConfig()
+ .setSelector(
+ new PluginConfig(TooManyDeletesSelector.class.getName(), Map.of("threshold", ".99")))
+ .setWait(true);
+
+ client.tableOperations().compact("tmd_control1", cc1);
+ client.tableOperations().compact("tmd_control2", cc1);
+ client.tableOperations().compact("tmd_control3", cc1);
+
+ assertEquals(0, getFiles(client, "tmd_control1").size());
+ assertEquals(2, getFiles(client, "tmd_control2").size());
+ assertEquals(2, getFiles(client, "tmd_control3").size());
+
+ var cc2 = new CompactionConfig()
+ .setSelector(
+ new PluginConfig(TooManyDeletesSelector.class.getName(), Map.of("threshold", ".40")))
+ .setWait(true);
- @Test
- public void testTooManyDeletes() throws Exception {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- Map<String,
- String> props = Map.of(Property.TABLE_COMPACTION_SELECTOR.getKey(),
- TooManyDeletesSelector.class.getName(),
- Property.TABLE_COMPACTION_SELECTOR_OPTS.getKey() + "threshold", ".4");
- var deleteSummarizerCfg =
- SummarizerConfiguration.builder(DeletesSummarizer.class.getName()).build();
- client.tableOperations().create("tmd_selector", new NewTableConfiguration()
- .setProperties(props).enableSummarization(deleteSummarizerCfg));
- client.tableOperations().create("tmd_control1",
- new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
- client.tableOperations().create("tmd_control2",
- new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
- client.tableOperations().create("tmd_control3",
- new NewTableConfiguration().enableSummarization(deleteSummarizerCfg));
+ client.tableOperations().compact("tmd_control1", cc2);
+ client.tableOperations().compact("tmd_control2", cc2);
+ client.tableOperations().compact("tmd_control3", cc2);
- addFile(client, "tmd_selector", 1, 1000, false);
- addFile(client, "tmd_selector", 1, 1000, true);
+ assertEquals(0, getFiles(client, "tmd_control1").size());
+ assertEquals(2, getFiles(client, "tmd_control2").size());
+ assertEquals(1, getFiles(client, "tmd_control3").size());
- addFile(client, "tmd_control1", 1, 1000, false);
- addFile(client, "tmd_control1", 1, 1000, true);
+ client.tableOperations().compact("tmd_control2", new CompactionConfig().setWait(true));
- addFile(client, "tmd_control2", 1, 1000, false);
- addFile(client, "tmd_control2", 1000, 2000, false);
+ assertEquals(1, getFiles(client, "tmd_control2").size());
- addFile(client, "tmd_control3", 1, 2000, false);
- addFile(client, "tmd_control3", 1, 1000, true);
+ }
+ }
+
+ @Test
+ public void testIteratorsWithRange() throws Exception {
+
+ String tableName = "tiwr";
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ client.tableOperations().create(tableName);
+ SortedSet<Text> splits = new TreeSet<>();
+ splits.add(new Text("f"));
+ splits.add(new Text("m"));
+ splits.add(new Text("r"));
+ splits.add(new Text("t"));
+ client.tableOperations().addSplits(tableName, splits);
+
+ Map<String, String> expected = new TreeMap<>();
+
+ try (var writer = client.createBatchWriter(tableName)) {
+ int v = 0;
+ for (String row : List.of("a", "h", "o", "s", "x")) {
+ Mutation m = new Mutation(row);
+ for (int q = 0; q < 10; q++) {
+ String qual = String.format("%03d", q);
+ String val = "v" + v++;
+ m.at().family("f").qualifier(qual).put(val);
+ expected.put(row + ":f:" + qual, val);
+ }
+ writer.addMutation(m);
+ }
+ }
- assertEquals(2, getFiles(client, "tmd_control1").size());
- assertEquals(2, getFiles(client, "tmd_control2").size());
- assertEquals(2, getFiles(client, "tmd_control3").size());
+ IteratorSetting iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
+ RegExFilter.setRegexs(iterSetting, null, null, "004|007", null, false);
+ RegExFilter.setNegate(iterSetting, true);
+ client.tableOperations().compact(tableName,
+ new CompactionConfig().setStartRow(new Text("b")).setEndRow(new Text("m"))
+ .setIterators(List.of(iterSetting)).setWait(true).setFlush(true));
- while (getFiles(client, "tmd_selector").size() != 0) {
- Thread.sleep(100);
- }
+ for (String row : List.of("a", "h")) {
+ assertTrue(expected.remove(row + ":f:004") != null);
+ assertTrue(expected.remove(row + ":f:007") != null);
+ }
- assertEquals(2, getFiles(client, "tmd_control1").size());
- assertEquals(2, getFiles(client, "tmd_control2").size());
- assertEquals(2, getFiles(client, "tmd_control3").size());
+ Map<String, String> actual = scanTable(client, tableName);
+ assertEquals(expected, actual);
- var cc1 = new CompactionConfig()
- .setSelector(
- new PluginConfig(TooManyDeletesSelector.class.getName(), Map.of("threshold", ".99")))
- .setWait(true);
+ iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
+ RegExFilter.setRegexs(iterSetting, null, null, "002|005|009", null, false);
+ RegExFilter.setNegate(iterSetting, true);
+ client.tableOperations().compact(tableName, new CompactionConfig().setStartRow(new Text("m"))
+ .setEndRow(new Text("u")).setIterators(List.of(iterSetting)).setWait(true));
- client.tableOperations().compact("tmd_control1", cc1);
- client.tableOperations().compact("tmd_control2", cc1);
- client.tableOperations().compact("tmd_control3", cc1);
+ for (String row : List.of("o", "s", "x")) {
+ assertTrue(expected.remove(row + ":f:002") != null);
+ assertTrue(expected.remove(row + ":f:005") != null);
+ assertTrue(expected.remove(row + ":f:009") != null);
+ }
- assertEquals(0, getFiles(client, "tmd_control1").size());
- assertEquals(2, getFiles(client, "tmd_control2").size());
- assertEquals(2, getFiles(client, "tmd_control3").size());
+ actual = scanTable(client, tableName);
+ assertEquals(expected, actual);
- var cc2 = new CompactionConfig()
- .setSelector(
- new PluginConfig(TooManyDeletesSelector.class.getName(), Map.of("threshold", ".40")))
- .setWait(true);
+ iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
+ RegExFilter.setRegexs(iterSetting, null, null, "00[18]", null, false);
+ RegExFilter.setNegate(iterSetting, true);
+ client.tableOperations().compact(tableName,
+ new CompactionConfig().setIterators(List.of(iterSetting)).setWait(true));
- client.tableOperations().compact("tmd_control1", cc2);
- client.tableOperations().compact("tmd_control2", cc2);
- client.tableOperations().compact("tmd_control3", cc2);
+ for (String row : List.of("a", "h", "o", "s", "x")) {
+ assertTrue(expected.remove(row + ":f:001") != null);
+ assertTrue(expected.remove(row + ":f:008") != null);
+ }
- assertEquals(0, getFiles(client, "tmd_control1").size());
- assertEquals(2, getFiles(client, "tmd_control2").size());
- assertEquals(1, getFiles(client, "tmd_control3").size());
+ actual = scanTable(client, tableName);
+ assertEquals(expected, actual);
+
+ // add all data back and force a compaction to ensure iters do not run again
+ try (var writer = client.createBatchWriter(tableName)) {
+ int v = 1000;
+ for (String row : List.of("a", "h", "o", "s", "x")) {
+ Mutation m = new Mutation(row);
+ for (int q = 0; q < 10; q++) {
+ String qual = String.format("%03d", q);
+ String val = "v" + v++;
+ m.at().family("f").qualifier(qual).put(val);
+ expected.put(row + ":f:" + qual, val);
+ }
+ writer.addMutation(m);
+ }
+ }
- client.tableOperations().compact("tmd_control2", new CompactionConfig().setWait(true));
+ client.tableOperations().compact(tableName,
+ new CompactionConfig().setWait(true).setFlush(true));
- assertEquals(1, getFiles(client, "tmd_control2").size());
+ actual = scanTable(client, tableName);
+ assertEquals(expected, actual);
+ }
}
- }
-
- @Test
- public void testIteratorsWithRange() throws Exception {
-
- String tableName = "tiwr";
-
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- client.tableOperations().create(tableName);
- SortedSet<Text> splits = new TreeSet<>();
- splits.add(new Text("f"));
- splits.add(new Text("m"));
- splits.add(new Text("r"));
- splits.add(new Text("t"));
- client.tableOperations().addSplits(tableName, splits);
-
- Map<String,String> expected = new TreeMap<>();
-
- try (var writer = client.createBatchWriter(tableName)) {
- int v = 0;
- for (String row : List.of("a", "h", "o", "s", "x")) {
- Mutation m = new Mutation(row);
- for (int q = 0; q < 10; q++) {
- String qual = String.format("%03d", q);
- String val = "v" + v++;
- m.at().family("f").qualifier(qual).put(val);
- expected.put(row + ":f:" + qual, val);
- }
- writer.addMutation(m);
+
+ @Test
+ public void testConfigurer() throws Exception {
+ String tableName = "tcc";
+
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ var ntc = new NewTableConfiguration()
+ .setProperties(Map.of(Property.TABLE_FILE_COMPRESSION_TYPE.getKey(), "none"));
+ client.tableOperations().create(tableName, ntc);
+
+ byte[] data = new byte[100000];
+ Arrays.fill(data, (byte) 65);
+
+ try (var writer = client.createBatchWriter(tableName)) {
+ for (int row = 0; row < 10; row++) {
+ Mutation m = new Mutation(row + "");
+ m.at().family("big").qualifier("stuff").put(data);
+ writer.addMutation(m);
+ }
+ }
+
+ client.tableOperations().flush(tableName, null, null, true);
+
+ // without compression, expect file to be large
+ long sizes = getFileSizes(client, tableName);
+ assertTrue("Unexpected files sizes : " + sizes,
+ sizes > data.length * 10 && sizes < data.length * 11);
+
+ client.tableOperations().compact(tableName,
+ new CompactionConfig().setWait(true)
+ .setConfigurer(new PluginConfig(CompressionConfigurer.class.getName(),
+ Map.of(CompressionConfigurer.LARGE_FILE_COMPRESSION_TYPE, "gz",
+ CompressionConfigurer.LARGE_FILE_COMPRESSION_THRESHOLD, data.length + ""))));
+
+ // after compacting with compression, expect small file
+ sizes = getFileSizes(client, tableName);
+ assertTrue("Unexpected files sizes : " + sizes, sizes < data.length);
+
+ client.tableOperations().compact(tableName, new CompactionConfig().setWait(true));
+
+ // after compacting without compression, expect big files again
+ sizes = getFileSizes(client, tableName);
+ assertTrue("Unexpected files sizes : " + sizes,
+ sizes > data.length * 10 && sizes < data.length * 11);
+
}
- }
-
- IteratorSetting iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
- RegExFilter.setRegexs(iterSetting, null, null, "004|007", null, false);
- RegExFilter.setNegate(iterSetting, true);
- client.tableOperations().compact(tableName,
- new CompactionConfig().setStartRow(new Text("b")).setEndRow(new Text("m"))
- .setIterators(List.of(iterSetting)).setWait(true).setFlush(true));
-
- for (String row : List.of("a", "h")) {
- assertTrue(expected.remove(row + ":f:004") != null);
- assertTrue(expected.remove(row + ":f:007") != null);
- }
-
- Map<String,String> actual = scanTable(client, tableName);
- assertEquals(expected, actual);
-
- iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
- RegExFilter.setRegexs(iterSetting, null, null, "002|005|009", null, false);
- RegExFilter.setNegate(iterSetting, true);
- client.tableOperations().compact(tableName, new CompactionConfig().setStartRow(new Text("m"))
- .setEndRow(new Text("u")).setIterators(List.of(iterSetting)).setWait(true));
-
- for (String row : List.of("o", "s", "x")) {
- assertTrue(expected.remove(row + ":f:002") != null);
- assertTrue(expected.remove(row + ":f:005") != null);
- assertTrue(expected.remove(row + ":f:009") != null);
- }
-
- actual = scanTable(client, tableName);
- assertEquals(expected, actual);
-
- iterSetting = new IteratorSetting(20, "rf", RegExFilter.class.getName());
- RegExFilter.setRegexs(iterSetting, null, null, "00[18]", null, false);
- RegExFilter.setNegate(iterSetting, true);
- client.tableOperations().compact(tableName,
- new CompactionConfig().setIterators(List.of(iterSetting)).setWait(true));
-
- for (String row : List.of("a", "h", "o", "s", "x")) {
- assertTrue(expected.remove(row + ":f:001") != null);
- assertTrue(expected.remove(row + ":f:008") != null);
- }
-
- actual = scanTable(client, tableName);
- assertEquals(expected, actual);
-
- // add all data back and force a compaction to ensure iters do not run again
- try (var writer = client.createBatchWriter(tableName)) {
- int v = 1000;
- for (String row : List.of("a", "h", "o", "s", "x")) {
- Mutation m = new Mutation(row);
- for (int q = 0; q < 10; q++) {
- String qual = String.format("%03d", q);
- String val = "v" + v++;
- m.at().family("f").qualifier(qual).put(val);
- expected.put(row + ":f:" + qual, val);
- }
- writer.addMutation(m);
+ }
+
+ private long getFileSizes(AccumuloClient client, String tableName) {
+ var tableId = TableId.of(client.tableOperations().tableIdMap().get(tableName));
+
+ try (var tabletsMeta =
+ TabletsMetadata.builder().forTable(tableId).fetch(ColumnType.FILES).build(client)) {
+ return tabletsMeta.stream().flatMap(tm -> tm.getFiles().stream()).mapToLong(stf -> {
+ try {
+ return FileSystem.getLocal(new Configuration()).getFileStatus(stf.getPath()).getLen();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ ).sum();
}
- }
+ }
+
+ @Test
+ public void testIncorrectSelectorType() throws Exception {
+ String tableName = "tist";
- client.tableOperations().compact(tableName,
- new CompactionConfig().setWait(true).setFlush(true));
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ client.tableOperations().create(tableName);
+ addFiles(client, tableName, 5);
- actual = scanTable(client, tableName);
- assertEquals(expected, actual);
+ var msg = assertThrows(AccumuloException.class, () -> {
+ client.tableOperations().compact(tableName, new CompactionConfig()
+ .setSelector(new PluginConfig(CompressionConfigurer.class.getName())).setWait(true));
+ }).getMessage();
+ assertTrue("Unexcpected message : " + msg,
+ msg.contains("TabletServer could not load CompactionSelector"));
+
+ }
}
- }
- @Test
- public void testConfigurer() throws Exception {
- String tableName = "tcc";
+ @Test
+ public void testIncorrectConfigurerType() throws Exception {
+ String tableName = "tict";
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- var ntc = new NewTableConfiguration()
- .setProperties(Map.of(Property.TABLE_FILE_COMPRESSION_TYPE.getKey(), "none"));
- client.tableOperations().create(tableName, ntc);
+ try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
+ client.tableOperations().create(tableName);
+ addFiles(client, tableName, 5);
- byte[] data = new byte[100000];
- Arrays.fill(data, (byte) 65);
+ var msg = assertThrows(AccumuloException.class, () -> {
+ client.tableOperations().compact(tableName, new CompactionConfig()
+ .setConfigurer(new PluginConfig(TooManyDeletesSelector.class.getName())).setWait(true));
+ }).getMessage();
- try (var writer = client.createBatchWriter(tableName)) {
- for (int row = 0; row < 10; row++) {
- Mutation m = new Mutation(row + "");
- m.at().family("big").qualifier("stuff").put(data);
- writer.addMutation(m);
+ assertTrue("Unexcpected message : " + msg,
+ msg.contains("TabletServer could not load CompactionConfigurer"));
}
- }
+ }
- client.tableOperations().flush(tableName, null, null, true);
+ @Test
+ public void testCompactionSelector() throws Exception {
+ PluginConfig csc = new PluginConfig(CompactionIT.FooSelector.class.getName());
+ CompactionConfig compactConfig = new CompactionConfig().setSelector(csc);
+ compactionTest(compactConfig);
+ }
+
+ private void write(BatchWriter bw, String row, String family, String qualifier, String value)
+ throws MutationsRejectedException {
+ Mutation m1 = new Mutation(row);
+ m1.put(family, qualifier, value);
+ bw.addMutation(m1);
+ }
+
+
+ public void compactionTest(CompactionConfig compactConfig) throws Exception {
+ final Random rand = new Random();
+ final String table = getUniqueNames(1)[0];
+ //long fooCount = 1 + (long) rand.nextInt(5), barCount = 1 + (long) rand.nextInt(5);
+ int i, k, count = 2;
+ long fooCount = 3, barCount = 4;
+
+ try (AccumuloClient c = Accumulo.newClient().from(getClientProps()).build()) {
+ NewTableConfiguration ntc = new NewTableConfiguration();
+ SummarizerConfiguration sc1 =
+ SummarizerConfiguration.builder(SummaryIT.FooCounter.class.getName()).build();
+ ntc.enableSummarization(sc1);
+ c.tableOperations().create(table, ntc);
+
+
+ for (i = 0; i < barCount; i++) {
+ try (BatchWriter bw = c.createBatchWriter(table)) {
+ write(bw, "bar" + i, "f" + i, "q" + i, "v" + i);
+ }
+ }
+ for (k = 0; k < fooCount; k++) {
+ try (BatchWriter bw = c.createBatchWriter(table)) {
+ write(bw, "foo" + k, "f" + k, "q" + k, "v" + k);
+ }
+ }
- // without compression, expect file to be large
- long sizes = getFileSizes(client, tableName);
- assertTrue("Unexpected files sizes : " + sizes,
- sizes > data.length * 10 && sizes < data.length * 11);
+ if (fooCount > barCount) {
+ count = 1;
+ fooCount = 0;
+ }
+
+ List<IteratorSetting> iterators =
+ Collections.singletonList(new IteratorSetting(100, SummaryIT.FooFilter.class));
+ compactConfig = compactConfig.setFlush(true).setIterators(iterators).setWait(true);
- client.tableOperations().compact(tableName,
- new CompactionConfig().setWait(true)
- .setConfigurer(new PluginConfig(CompressionConfigurer.class.getName(),
- Map.of(CompressionConfigurer.LARGE_FILE_COMPRESSION_TYPE, "gz",
- CompressionConfigurer.LARGE_FILE_COMPRESSION_THRESHOLD, data.length + ""))));
- // after compacting with compression, expect small file
- sizes = getFileSizes(client, tableName);
- assertTrue("Unexpected files sizes : " + sizes, sizes < data.length);
+ c.tableOperations().compact(table, compactConfig);
- client.tableOperations().compact(tableName, new CompactionConfig().setWait(true));
+ try (Scanner scanner = c.createScanner(table, Authorizations.EMPTY)) {
+ Stream<Entry<Key, Value>> stream = StreamSupport.stream(scanner.spliterator(), false);
+ Map<String, Long> counts = stream.map(e -> e.getKey().getRowData().toString()) // convert to
+ // row
+ .map(r -> r.replaceAll("[0-9]+", "")) // strip numbers off row
+ .collect(groupingBy(identity(), counting())); // count different row types
- // after compacting without compression, expect big files again
- sizes = getFileSizes(client, tableName);
- assertTrue("Unexpected files sizes : " + sizes,
- sizes > data.length * 10 && sizes < data.length * 11);
+
+ assertEquals(fooCount, (long) counts.getOrDefault("foo", 0L));
+ assertEquals(barCount, (long) counts.getOrDefault("bar", 0L));
+ assertEquals(count, counts.size());
+ }
+
+
+ }
}
- }
-
- private long getFileSizes(AccumuloClient client, String tableName) {
- var tableId = TableId.of(client.tableOperations().tableIdMap().get(tableName));
-
- try (var tabletsMeta =
- TabletsMetadata.builder().forTable(tableId).fetch(ColumnType.FILES).build(client)) {
- return tabletsMeta.stream().flatMap(tm -> tm.getFiles().stream()).mapToLong(stf -> {
- try {
- return FileSystem.getLocal(new Configuration()).getFileStatus(stf.getPath()).getLen();
- } catch (IOException e) {
- throw new UncheckedIOException(e);
- }
- }
- ).sum();
+ static class fooSelectorException extends RuntimeException {
+ public fooSelectorException(String message) {
+ super(message);
+ }
}
- }
- @Test
- public void testIncorrectSelectorType() throws Exception {
- String tableName = "tist";
+ public static class FooSelector implements CompactionSelector {
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- client.tableOperations().create(tableName);
- addFiles(client, tableName, 5);
- var msg = assertThrows(AccumuloException.class, () -> {
- client.tableOperations().compact(tableName, new CompactionConfig()
- .setSelector(new PluginConfig(CompressionConfigurer.class.getName())).setWait(true));
- }).getMessage();
+ boolean odd = false;
- assertTrue("Unexcpected message : " + msg,
- msg.contains("TabletServer could not load CompactionSelector"));
- }
- }
+ @Override
+ public void init(InitParamaters iparams) {
+ }
- @Test
- public void testIncorrectConfigurerType() throws Exception {
- String tableName = "tict";
+ @Override
+ public Selection select(SelectionParameters sparams){
+ Collection<Summary> summaries = sparams.getSummaries(sparams.getAvailableFiles(),
+ conf -> conf.getClassName().contains("FooCounter"));
- try (AccumuloClient client = Accumulo.newClient().from(getClientProps()).build()) {
- client.tableOperations().create(tableName);
- addFiles(client, tableName, 5);
+ if (summaries.size() == 1) {
+ Summary summary = summaries.iterator().next();
+ Long foos = summary.getStatistics().getOrDefault("foos", 0L);
+ Long bars = summary.getStatistics().getOrDefault("bars", 0L);
+ odd = isOdd(foos);
- var msg = assertThrows(AccumuloException.class, () -> {
- client.tableOperations().compact(tableName, new CompactionConfig()
- .setConfigurer(new PluginConfig(TooManyDeletesSelector.class.getName())).setWait(true));
- }).getMessage();
- assertTrue("Unexcpected message : " + msg,
- msg.contains("TabletServer could not load CompactionConfigurer"));
- }
- }
-
- private Map<String,String> scanTable(AccumuloClient client, String tableName)
- throws TableNotFoundException, AccumuloSecurityException, AccumuloException {
- Map<String,String> actual = new TreeMap<>();
- try (var scanner = client.createScanner(tableName)) {
- for (Entry<Key,Value> e : scanner) {
- var k = e.getKey();
- actual.put(
- k.getRowData() + ":" + k.getColumnFamilyData() + ":" + k.getColumnQualifierData(),
- e.getValue().toString());
- }
+
+ try {
+ if (odd)
+ throw new fooSelectorException("Exception Thrown");
+
+
+ } catch ( fooSelectorException e){
+ log.debug(e.getMessage());
+ }
+
+ if (foos > bars)
+ return new Selection(sparams.getAvailableFiles());
+
+
+ }
+ return new Selection(Set.of());
+
+ }
+
+ boolean isOdd(long foos) {
+
+ if ((foos & 0x1) == 1) {
+ return true;
+ }
+ return false;
Review comment:
```suggestion
return (foos & 0x1) == 1;
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org