accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mmil...@apache.org
Subject [accumulo] branch master updated: Make new Bulk Import tableTime take boolean (#880)
Date Wed, 09 Jan 2019 21:31:26 GMT
This is an automated email from the ASF dual-hosted git repository.

mmiller pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/accumulo.git


The following commit(s) were added to refs/heads/master by this push:
     new d379b5f  Make new Bulk Import tableTime take boolean (#880)
d379b5f is described below

commit d379b5f8d3db635046996d4cd39f502ab0dea7d8
Author: Mike Miller <mmiller@apache.org>
AuthorDate: Wed Jan 9 16:31:21 2019 -0500

    Make new Bulk Import tableTime take boolean (#880)
    
    * Also add new test to BulkLoadIT for tableTime
---
 .../core/client/admin/TableOperations.java         |  5 ++-
 .../accumulo/core/clientImpl/BulkImport.java       |  4 +--
 .../org/apache/accumulo/proxy/ProxyServer.java     | 11 ++-----
 .../shell/commands/ImportDirectoryCommand.java     |  9 ++----
 .../accumulo/test/functional/BulkLoadIT.java       | 37 ++++++++++++++++------
 5 files changed, 39 insertions(+), 27 deletions(-)

diff --git a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
index 06a5141..9ffcb1c 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
@@ -567,8 +567,11 @@ public interface TableOperations {
      * timestamp used depends on how the table was created.
      *
      * @see NewTableConfiguration#setTimeType(TimeType)
+     * @param value
+     *          override the time values in the input files, and use the current time for
all
+     *          mutations
      */
-    ImportMappingOptions tableTime();
+    ImportMappingOptions tableTime(boolean value);
 
     /**
      * Loads the files into the table.
diff --git a/core/src/main/java/org/apache/accumulo/core/clientImpl/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/clientImpl/BulkImport.java
index cfa0083..b232d38 100644
--- a/core/src/main/java/org/apache/accumulo/core/clientImpl/BulkImport.java
+++ b/core/src/main/java/org/apache/accumulo/core/clientImpl/BulkImport.java
@@ -106,8 +106,8 @@ public class BulkImport implements ImportDestinationArguments, ImportMappingOpti
   }
 
   @Override
-  public ImportMappingOptions tableTime() {
-    this.setTime = true;
+  public ImportMappingOptions tableTime(boolean value) {
+    this.setTime = value;
     return this;
   }
 
diff --git a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
index 0e147d7..6516623 100644
--- a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
@@ -59,7 +59,6 @@ import org.apache.accumulo.core.client.admin.ActiveCompaction;
 import org.apache.accumulo.core.client.admin.ActiveScan;
 import org.apache.accumulo.core.client.admin.CompactionConfig;
 import org.apache.accumulo.core.client.admin.NewTableConfiguration;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportMappingOptions;
 import org.apache.accumulo.core.client.admin.TimeType;
 import org.apache.accumulo.core.client.security.SecurityErrorCode;
 import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
@@ -1766,13 +1765,9 @@ public class ProxyServer implements AccumuloProxy.Iface {
       org.apache.accumulo.proxy.thrift.AccumuloException,
       org.apache.accumulo.proxy.thrift.AccumuloSecurityException, TException {
     try {
-      ImportMappingOptions loader = getConnector(login).tableOperations().importDirectory(importDir)
-          .to(tableName);
-      if (setTime) {
-        loader.tableTime().load();
-      } else {
-        loader.load();
-      }
+      getConnector(login).tableOperations().importDirectory(importDir).to(tableName)
+          .tableTime(setTime).load();
+
     } catch (Exception e) {
       handleExceptionTNF(e);
     }
diff --git a/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
b/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
index 5bec2bc..cbee631 100644
--- a/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
+++ b/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
@@ -21,7 +21,6 @@ import java.io.IOException;
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.client.admin.TableOperations;
 import org.apache.accumulo.shell.Shell;
 import org.apache.accumulo.shell.Shell.Command;
 import org.apache.commons.cli.CommandLine;
@@ -49,12 +48,8 @@ public class ImportDirectoryCommand extends Command {
     // new bulk import only takes 2 args
     if (args.length == 2) {
       setTime = Boolean.parseBoolean(cl.getArgs()[1]);
-      TableOperations.ImportMappingOptions bulk = shellState.getAccumuloClient().tableOperations()
-          .importDirectory(dir).to(shellState.getTableName());
-      if (setTime)
-        bulk.tableTime().load();
-      else
-        bulk.load();
+      shellState.getAccumuloClient().tableOperations().importDirectory(dir)
+          .to(shellState.getTableName()).tableTime(setTime).load();
     } else if (args.length == 3) {
       // warn using deprecated bulk import
       Shell.log.warn(
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
index 501aec5..c19d62e 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
@@ -42,6 +42,8 @@ import java.util.stream.Collectors;
 import org.apache.accumulo.core.client.AccumuloClient;
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.NewTableConfiguration;
+import org.apache.accumulo.core.client.admin.TimeType;
 import org.apache.accumulo.core.clientImpl.Table;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.crypto.CryptoServiceFactory;
@@ -117,7 +119,8 @@ public class BulkLoadIT extends AccumuloClusterHarness {
     return dir;
   }
 
-  private void testSingleTabletSingleFile(AccumuloClient c, boolean offline) throws Exception
{
+  private void testSingleTabletSingleFile(AccumuloClient c, boolean offline, boolean setTime)
+      throws Exception {
     addSplits(c, tableName, "0333");
 
     if (offline)
@@ -127,12 +130,12 @@ public class BulkLoadIT extends AccumuloClusterHarness {
 
     String h1 = writeData(dir + "/f1.", aconf, 0, 332);
 
-    c.tableOperations().importDirectory(dir).to(tableName).load();
+    c.tableOperations().importDirectory(dir).to(tableName).tableTime(setTime).load();
 
     if (offline)
       c.tableOperations().online(tableName);
 
-    verifyData(c, tableName, 0, 332);
+    verifyData(c, tableName, 0, 332, setTime);
     verifyMetadata(c, tableName,
         ImmutableMap.of("0333", ImmutableSet.of(h1), "null", ImmutableSet.of()));
   }
@@ -140,14 +143,26 @@ public class BulkLoadIT extends AccumuloClusterHarness {
   @Test
   public void testSingleTabletSingleFile() throws Exception {
     try (AccumuloClient client = createAccumuloClient()) {
-      testSingleTabletSingleFile(client, false);
+      testSingleTabletSingleFile(client, false, false);
+    }
+  }
+
+  @Test
+  public void testSetTime() throws Exception {
+    try (AccumuloClient client = createAccumuloClient()) {
+      tableName = "testSetTime_table1";
+      NewTableConfiguration newTableConf = new NewTableConfiguration();
+      // set logical time type so we can set time on bulk import
+      newTableConf.setTimeType(TimeType.LOGICAL);
+      client.tableOperations().create(tableName, newTableConf);
+      testSingleTabletSingleFile(client, false, true);
     }
   }
 
   @Test
   public void testSingleTabletSingleFileOffline() throws Exception {
     try (AccumuloClient client = createAccumuloClient()) {
-      testSingleTabletSingleFile(client, true);
+      testSingleTabletSingleFile(client, true, false);
     }
   }
 
@@ -165,7 +180,7 @@ public class BulkLoadIT extends AccumuloClusterHarness {
     if (offline)
       c.tableOperations().online(tableName);
 
-    verifyData(c, tableName, 0, 333);
+    verifyData(c, tableName, 0, 333, false);
     verifyMetadata(c, tableName, ImmutableMap.of("null", ImmutableSet.of(h1)));
   }
 
@@ -270,7 +285,7 @@ public class BulkLoadIT extends AccumuloClusterHarness {
       if (offline)
         c.tableOperations().online(tableName);
 
-      verifyData(c, tableName, 0, 1999);
+      verifyData(c, tableName, 0, 1999, false);
       verifyMetadata(c, tableName, hashes);
     }
   }
@@ -345,12 +360,13 @@ public class BulkLoadIT extends AccumuloClusterHarness {
     client.tableOperations().addSplits(tableName, splits);
   }
 
-  private void verifyData(AccumuloClient client, String table, int s, int e) throws Exception
{
+  private void verifyData(AccumuloClient client, String table, int start, int end, boolean
setTime)
+      throws Exception {
     try (Scanner scanner = client.createScanner(table, Authorizations.EMPTY)) {
 
       Iterator<Entry<Key,Value>> iter = scanner.iterator();
 
-      for (int i = s; i <= e; i++) {
+      for (int i = start; i <= end; i++) {
         if (!iter.hasNext())
           throw new Exception("row " + i + " not found");
 
@@ -363,6 +379,9 @@ public class BulkLoadIT extends AccumuloClusterHarness {
 
         if (Integer.parseInt(entry.getValue().toString()) != i)
           throw new Exception("unexpected value " + entry + " " + i);
+
+        if (setTime)
+          assertEquals(1L, entry.getKey().getTimestamp());
       }
 
       if (iter.hasNext())


Mime
View raw message