SQOOP-1519: Enable HCat/HBase/Accumulo operations with OraOop connection manager

src/java/org/apache/sqoop/manager/ConnManager.java

@@ -794,5 +794,25 @@ public boolean isCharColumn(int columnType) {
   public boolean isDirectModeHCatSupported() {
     return false;
   }
+
+
+  /**
+   * Determine if HBase operations from direct mode of the connector is
+   * allowed.  By default direct mode is not compatible with HBase
+   * @return Whether direct mode is allowed.
+   */
+  public boolean isDirectModeHBaseSupported() {
+    return false;
+  }
+
+  /**
+   * Determine if Accumulo operations from direct mode of the connector is
+   * allowed.  By default direct mode is not compatible with HBase
+   * @return Whether direct mode is allowed.
+   */
+  public boolean isDirectModeAccumuloSupported() {
+    return false;
+  }
+
 }
 

src/java/org/apache/sqoop/manager/oracle/OraOopConnManager.java

@@ -631,5 +631,31 @@ private void explainWhyExportClassCannotBeLoaded(NoClassDefFoundError ex,
             .getJavaClassPath());
     LOG.fatal(msg, ex);
   }
+  /**
+   * Determine if HCat integration from direct mode of the connector is
+   * allowed.  By default direct mode is not compatible with HCat
+   * @return Whether direct mode is allowed.
+   */
+  @Override
+  public boolean isDirectModeHCatSupported() {
+    return true;
+  }
 
+  /**
+   * Determine if HBase operations from direct mode of the connector is
+   * allowed.  By default direct mode is not compatible with HBase
+   * @return Whether direct mode is allowed.
+   */
+  public boolean isDirectModeHBaseSupported() {
+    return true;
+  }
+
+  /**
+   * Determine if Accumulo operations from direct mode of the connector is
+   * allowed.  By default direct mode is not compatible with HBase
+   * @return Whether direct mode is allowed.
+   */
+  public boolean isDirectModeAccumuloSupported() {
+    return true;
+  }
 }

src/java/org/apache/sqoop/mapreduce/ExportJobBase.java

@@ -39,7 +39,9 @@
 import org.apache.sqoop.mapreduce.hcat.SqoopHCatUtilities;
 import org.apache.sqoop.util.LoggingUtils;
 import org.apache.sqoop.util.PerfCounters;
+
 import com.cloudera.sqoop.SqoopOptions;
+import com.cloudera.sqoop.SqoopOptions.InvalidOptionsException;
 import com.cloudera.sqoop.config.ConfigurationHelper;
 import com.cloudera.sqoop.lib.SqoopRecord;
 import com.cloudera.sqoop.manager.ConnManager;
@@ -47,6 +49,7 @@
 import com.cloudera.sqoop.orm.TableClassName;
 import com.cloudera.sqoop.mapreduce.JobBase;
 import com.cloudera.sqoop.util.ExportException;
+
 import org.apache.sqoop.validation.*;
 
 /**
@@ -327,7 +330,16 @@ public void runExport() throws ExportException, IOException {
         + context.getConnManager().getClass().getName()
         + ". Please remove the parameter --direct");
     }
-
+    if (options.getAccumuloTable() != null && options.isDirect()
+        && !cmgr.isDirectModeAccumuloSupported()) {
+      throw new IOException("Direct mode is incompatible with "
+            + "Accumulo. Please remove the parameter --direct");
+    }
+    if (options.getHBaseTable() != null && options.isDirect()
+        && !cmgr.isDirectModeHBaseSupported()) {
+      throw new IOException("Direct mode is incompatible with "
+            + "HBase. Please remove the parameter --direct");
+    }
     if (stagingTableName != null) { // user has specified the staging table
       if (cmgr.supportsStagingForExport()) {
         LOG.info("Data will be staged in the table: " + stagingTableName);

src/java/org/apache/sqoop/mapreduce/ImportJobBase.java

@@ -223,7 +223,16 @@ public void runImport(String tableName, String ormJarFile, String splitByCol,
         + context.getConnManager().getClass().getName()
         + ". Please remove the parameter --direct");
     }
-
+    if (options.getAccumuloTable() != null && options.isDirect()
+        && !getContext().getConnManager().isDirectModeAccumuloSupported()) {
+      throw new IOException("Direct mode is incompatible with "
+            + "Accumulo. Please remove the parameter --direct");
+    }
+    if (options.getHBaseTable() != null && options.isDirect()
+        && !getContext().getConnManager().isDirectModeHBaseSupported()) {
+      throw new IOException("Direct mode is incompatible with "
+            + "HBase. Please remove the parameter --direct");
+    }
     if (null != tableName) {
       LOG.info("Beginning import of " + tableName);
     } else {

src/java/org/apache/sqoop/tool/BaseSqoopTool.java

@@ -1425,10 +1425,7 @@ protected void validateAccumuloOptions(SqoopOptions options)
           "Both --accumulo-table and --accumulo-column-family must be set."
           + HELP_STR);
     }
-    if (options.getAccumuloTable() != null && options.isDirect()) {
+
-      throw new InvalidOptionsException("Direct import is incompatible with "
-            + "Accumulo. Please remove parameter --direct");
-    }
     if (options.getAccumuloTable() != null
         && options.getHBaseTable() != null) {
       throw new InvalidOptionsException("HBase import is incompatible with "
@@ -1576,10 +1573,6 @@ protected void validateHBaseOptions(SqoopOptions options)
           "Both --hbase-table and --column-family must be set together."
           + HELP_STR);
     }
-    if (options.getHBaseTable() != null && options.isDirect()) {
-      throw new InvalidOptionsException("Direct import is incompatible with "
-        + "HBase. Please remove parameter --direct");
-    }
 
     if (options.isBulkLoadEnabled() && options.getHBaseTable() == null) {
       String validationMessage = String.format("Can't run import with %s "