DRILL-5978: Updating of Apache and MapR Hive libraries to 2.3.2 and 2.1.2-mapr-1710 versions respectively

contrib/storage-hive/core/pom.xml

@@ -63,12 +63,6 @@
     <dependency>
       <groupId>org.apache.hive</groupId>
       <artifactId>hive-hbase-handler</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>org.apache.hive</groupId>
-          <artifactId>hive-exec</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.hbase</groupId>

contrib/storage-hive/core/src/main/codegen/templates/HiveRecordReaders.java

@@ -130,7 +130,7 @@ public int next() {
       while (!recordsInspector.isBatchFull() && hasNextValue(recordsInspector.getValueHolder())) {
         Object value = recordsInspector.getNextValue();
         if (value != null) {
-          Object deSerializedValue = partitionSerDe.deserialize((Writable) value);
+          Object deSerializedValue = partitionDeserializer.deserialize((Writable) value);
           if (partTblObjectInspectorConverter != null) {
             deSerializedValue = partTblObjectInspectorConverter.convert(deSerializedValue);
           }
@@ -159,7 +159,7 @@ public int next() {
     try {
       int recordCount = 0;
       while (recordCount < TARGET_RECORD_COUNT && hasNextValue(value)) {
-        Object deSerializedValue = partitionSerDe.deserialize((Writable) value);
+        Object deSerializedValue = partitionDeserializer.deserialize((Writable) value);
         if (partTblObjectInspectorConverter != null) {
           deSerializedValue = partTblObjectInspectorConverter.convert(deSerializedValue);
         }

contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveMetadataProvider.java

@@ -264,6 +264,7 @@ public List<LogicalInputSplit> run() throws Exception {
           final List<LogicalInputSplit> splits = Lists.newArrayList();
           final JobConf job = new JobConf(hiveConf);
           HiveUtilities.addConfToJob(job, properties);
+          HiveUtilities.verifyAndAddTransactionalProperties(job, sd);
           job.setInputFormat(HiveUtilities.getInputFormatClass(job, sd, hiveReadEntry.getTable()));
           final Path path = new Path(sd.getLocation());
           final FileSystem fs = path.getFileSystem(job);

contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveUtilities.java

@@ -17,8 +17,11 @@
  */
 package org.apache.drill.exec.store.hive;
 
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
 import io.netty.buffer.DrillBuf;
 import org.apache.drill.common.exceptions.DrillRuntimeException;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
@@ -51,10 +54,14 @@
 import org.apache.drill.exec.work.ExecErrorConstants;
 
 import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.MetaStoreUtils;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
+import org.apache.hadoop.hive.ql.io.IOConstants;
 import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
 import org.apache.hadoop.hive.ql.metadata.HiveUtils;
 import org.apache.hadoop.hive.serde.serdeConstants;
@@ -70,6 +77,7 @@
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 
+import javax.annotation.Nullable;
 import java.math.BigDecimal;
 import java.sql.Date;
 import java.sql.Timestamp;
@@ -104,8 +112,7 @@ public static Object convertPartitionType(TypeInfo typeInfo, String value, final
           return Boolean.parseBoolean(value);
         case DECIMAL: {
           DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) typeInfo;
-          return HiveDecimalUtils.enforcePrecisionScale(HiveDecimal.create(value),
-              decimalTypeInfo.precision(), decimalTypeInfo.scale());
+          return HiveDecimalUtils.enforcePrecisionScale(HiveDecimal.create(value), decimalTypeInfo);
         }
         case DOUBLE:
           return Double.parseDouble(value);
@@ -507,5 +514,59 @@ public static boolean hasHeaderOrFooter(HiveTableWithColumnCache table) {
     int skipFooter = retrieveIntProperty(tableProperties, serdeConstants.FOOTER_COUNT, -1);
     return skipHeader > 0 || skipFooter > 0;
   }
+
+  /**
+   * This method checks whether the table is transactional and set necessary properties in {@link JobConf}.
+   * If schema evolution properties aren't set in job conf for the input format, method sets the column names
+   * and types from table/partition properties or storage descriptor.
+   *
+   * @param job the job to update
+   * @param sd storage descriptor
+   */
+  public static void verifyAndAddTransactionalProperties(JobConf job, StorageDescriptor sd) {
+
+    if (AcidUtils.isTablePropertyTransactional(job)) {
+      AcidUtils.setTransactionalTableScan(job, true);
+
+      // No work is needed, if schema evolution is used
+      if (Utilities.isSchemaEvolutionEnabled(job, true) && job.get(IOConstants.SCHEMA_EVOLUTION_COLUMNS) != null &&
+          job.get(IOConstants.SCHEMA_EVOLUTION_COLUMNS_TYPES) != null) {
+        return;
+      }
+
+      String colNames;
+      String colTypes;
+
+      // Try to get get column names and types from table or partition properties. If they are absent there, get columns
+      // data from storage descriptor of the table
+      colNames = job.get(serdeConstants.LIST_COLUMNS);
+      colTypes = job.get(serdeConstants.LIST_COLUMN_TYPES);
+
+      if (colNames == null || colTypes == null) {
+        colNames = Joiner.on(",").join(Lists.transform(sd.getCols(), new Function<FieldSchema, String>()
+        {
+          @Nullable
+          @Override
+          public String apply(@Nullable FieldSchema input)
+          {
+            return input.getName();
+          }
+        }));
+
+        colTypes = Joiner.on(",").join(Lists.transform(sd.getCols(), new Function<FieldSchema, String>()
+        {
+          @Nullable
+          @Override
+          public String apply(@Nullable FieldSchema input)
+          {
+            return input.getType();
+          }
+        }));
+      }
+
+      job.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS, colNames);
+      job.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS_TYPES, colTypes);
+    }
+  }
 }
 

contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/readers/HiveAbstractReader.java

@@ -50,7 +50,7 @@
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
-import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.hive.serde2.Deserializer;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorConverters.Converter;
@@ -85,10 +85,10 @@ public abstract class HiveAbstractReader extends AbstractRecordReader {
   protected List<TypeInfo> selectedPartitionTypes = Lists.newArrayList();
   protected List<Object> selectedPartitionValues = Lists.newArrayList();
 
-  // SerDe of the reading partition (or table if the table is non-partitioned)
-  protected SerDe partitionSerDe;
+  // Deserializer of the reading partition (or table if the table is non-partitioned)
+  protected Deserializer partitionDeserializer;
 
-  // ObjectInspector to read data from partitionSerDe (for a non-partitioned table this is same as the table
+  // ObjectInspector to read data from partitionDeserializer (for a non-partitioned table this is same as the table
   // ObjectInspector).
   protected StructObjectInspector partitionOI;
 
@@ -143,19 +143,20 @@ private void init() throws ExecutionSetupException {
               HiveUtilities.getPartitionMetadata(partition, table);
       HiveUtilities.addConfToJob(job, partitionProperties);
 
-      final SerDe tableSerDe = createSerDe(job, table.getSd().getSerdeInfo().getSerializationLib(), tableProperties);
-      final StructObjectInspector tableOI = getStructOI(tableSerDe);
+      final Deserializer tableDeserializer = createDeserializer(job, table.getSd().getSerdeInfo().getSerializationLib(), tableProperties);
+      final StructObjectInspector tableOI = getStructOI(tableDeserializer);
 
       if (partition != null) {
-        partitionSerDe = createSerDe(job, partition.getSd().getSerdeInfo().getSerializationLib(), partitionProperties);
-        partitionOI = getStructOI(partitionSerDe);
+        partitionDeserializer = createDeserializer(job, partition.getSd().getSerdeInfo().getSerializationLib(), partitionProperties);
+        partitionOI = getStructOI(partitionDeserializer);
 
         finalOI = (StructObjectInspector)ObjectInspectorConverters.getConvertedOI(partitionOI, tableOI);
         partTblObjectInspectorConverter = ObjectInspectorConverters.getConverter(partitionOI, finalOI);
         job.setInputFormat(HiveUtilities.getInputFormatClass(job, partition.getSd(), table));
+        HiveUtilities.verifyAndAddTransactionalProperties(job, table.getSd());
       } else {
         // For non-partitioned tables, there is no need to create converter as there are no schema changes expected.
-        partitionSerDe = tableSerDe;
+        partitionDeserializer = tableDeserializer;
         partitionOI = tableOI;
         partTblObjectInspectorConverter = null;
         finalOI = tableOI;
@@ -166,7 +167,7 @@ private void init() throws ExecutionSetupException {
         for (StructField field: finalOI.getAllStructFieldRefs()) {
           logger.trace("field in finalOI: {}", field.getClass().getName());
         }
-        logger.trace("partitionSerDe class is {} {}", partitionSerDe.getClass().getName());
+        logger.trace("partitionDeserializer class is {} {}", partitionDeserializer.getClass().getName());
       }
       // Get list of partition column names
       final List<String> partitionNames = Lists.newArrayList();
@@ -176,8 +177,8 @@ private void init() throws ExecutionSetupException {
 
       // We should always get the columns names from ObjectInspector. For some of the tables (ex. avro) metastore
       // may not contain the schema, instead it is derived from other sources such as table properties or external file.
-      // SerDe object knows how to get the schema with all the config and table properties passed in initialization.
-      // ObjectInspector created from the SerDe object has the schema.
+      // Deserializer object knows how to get the schema with all the config and table properties passed in initialization.
+      // ObjectInspector created from the Deserializer object has the schema.
       final StructTypeInfo sTypeInfo = (StructTypeInfo) TypeInfoUtils.getTypeInfoFromObjectInspector(finalOI);
       final List<String> tableColumnNames = sTypeInfo.getAllStructFieldNames();
 
@@ -201,7 +202,20 @@ private void init() throws ExecutionSetupException {
           }
         }
       }
-      ColumnProjectionUtils.appendReadColumns(job, columnIds, selectedColumnNames);
+      ColumnProjectionUtils.appendReadColumns(job, columnIds);
+
+      // TODO: Use below overloaded method instead of above simpler version of it, once Hive client dependencies
+      // (from all profiles) will be updated to 2.3 version or above
+//      ColumnProjectionUtils.appendReadColumns(job, columnIds, selectedColumnNames,
+//          Lists.newArrayList(Iterables.transform(getColumns(), new Function<SchemaPath, String>()
+//      {
+//        @Nullable
+//        @Override
+//        public String apply(@Nullable SchemaPath path)
+//        {
+//          return path.getRootSegmentPath();
+//        }
+//      })));
 
       for (String columnName : selectedColumnNames) {
         StructField fieldRef = finalOI.getStructFieldRef(columnName);
@@ -269,18 +283,19 @@ protected boolean initNextReader(JobConf job) throws ExecutionSetupException {
   }
 
   /**
-   * Utility method which creates a SerDe object for given SerDe class name and properties.
+   * Utility method which creates a Deserializer object for given Deserializer class name and properties.
+   * TODO: Replace Deserializer interface with AbstractSerDe, once all Hive clients is upgraded to 2.3 version
    */
-  private static SerDe createSerDe(final JobConf job, final String sLib, final Properties properties) throws Exception {
-    final Class<? extends SerDe> c = Class.forName(sLib).asSubclass(SerDe.class);
-    final SerDe serde = c.getConstructor().newInstance();
-    serde.initialize(job, properties);
+  private static Deserializer createDeserializer(final JobConf job, final String sLib, final Properties properties) throws Exception {
+    final Class<? extends Deserializer> c = Class.forName(sLib).asSubclass(Deserializer.class);
+    final Deserializer deserializer = c.getConstructor().newInstance();
+    deserializer.initialize(job, properties);
 
-    return serde;
+    return deserializer;
   }
 
-  private static StructObjectInspector getStructOI(final SerDe serDe) throws Exception {
-    ObjectInspector oi = serDe.getObjectInspector();
+  private static StructObjectInspector getStructOI(final Deserializer deserializer) throws Exception {
+    ObjectInspector oi = deserializer.getObjectInspector();
     if (oi.getCategory() != ObjectInspector.Category.STRUCT) {
       throw new UnsupportedOperationException(String.format("%s category not supported", oi.getCategory()));
     }

contrib/storage-hive/core/src/test/java/org/apache/drill/exec/hive/TestHiveStorage.java

@@ -392,6 +392,16 @@ public void readFromAlteredPartitionedTable() throws Exception {
         .go();
   }
 
+  @Test // DRILL-3938
+  public void readFromAlteredPartitionedTableWithEmptyGroupType() throws Exception {
+    testBuilder()
+        .sqlQuery("SELECT newcol FROM hive.kv_parquet LIMIT 1")
+        .unOrdered()
+        .baselineColumns("newcol")
+        .baselineValues(new Object[]{null})
+        .go();
+  }
+
   @Test // DRILL-3938
   public void nativeReaderIsDisabledForAlteredPartitionedTable() throws Exception {
     try {

contrib/storage-hive/core/src/test/java/org/apache/drill/exec/impersonation/hive/BaseTestHiveImpersonation.java

@@ -1,4 +1,4 @@
-/**
+/*
  * 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
@@ -72,6 +72,9 @@ protected static void prepHiveConfAndData() throws Exception {
 
     hiveConf.set(ConfVars.SCRATCHDIR.varname, "file://" + scratchDir.getAbsolutePath());
     hiveConf.set(ConfVars.LOCALSCRATCHDIR.varname, localScratchDir.getAbsolutePath());
+    hiveConf.set(ConfVars.METASTORE_SCHEMA_VERIFICATION.varname, "false");
+    hiveConf.set(ConfVars.METASTORE_AUTO_CREATE_ALL.varname, "true");
+    hiveConf.set(ConfVars.HIVE_CBO_ENABLED.varname, "false");
 
     // Set MiniDFS conf in HiveConf
     hiveConf.set(FS_DEFAULT_NAME_KEY, dfsConf.get(FS_DEFAULT_NAME_KEY));

contrib/storage-hive/core/src/test/java/org/apache/drill/exec/impersonation/hive/TestSqlStdBasedAuthorization.java

@@ -1,4 +1,4 @@
-/**
+/*
  * 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
@@ -41,9 +41,12 @@
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_AUTHENTICATOR_MANAGER;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_AUTHORIZATION_MANAGER;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_CBO_ENABLED;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_AUTO_CREATE_ALL;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_EXECUTE_SET_UGI;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_SCHEMA_VERIFICATION;
 
 @Category({SlowTest.class, HiveStorageTest.class})
 public class TestSqlStdBasedAuthorization extends BaseTestHiveImpersonation {
@@ -101,6 +104,9 @@ private static Map<String, String> getHivePluginConfig() {
     hiveConfig.put(HIVE_AUTHORIZATION_ENABLED.varname, hiveConf.get(HIVE_AUTHORIZATION_ENABLED.varname));
     hiveConfig.put(HIVE_AUTHENTICATOR_MANAGER.varname, SessionStateUserAuthenticator.class.getName());
     hiveConfig.put(HIVE_AUTHORIZATION_MANAGER.varname, SQLStdHiveAuthorizerFactory.class.getName());
+    hiveConfig.put(METASTORE_SCHEMA_VERIFICATION.varname, hiveConf.get(METASTORE_SCHEMA_VERIFICATION.varname));
+    hiveConfig.put(METASTORE_AUTO_CREATE_ALL.varname, hiveConf.get(METASTORE_AUTO_CREATE_ALL.varname));
+    hiveConfig.put(HIVE_CBO_ENABLED.varname, hiveConf.get(HIVE_CBO_ENABLED.varname));
     return hiveConfig;
   }
 

contrib/storage-hive/core/src/test/java/org/apache/drill/exec/impersonation/hive/TestStorageBasedHiveAuthorization.java

@@ -41,13 +41,16 @@
 
 import static org.apache.drill.exec.hive.HiveTestUtilities.executeQuery;
 import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_CBO_ENABLED;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_METASTORE_AUTHENTICATOR_MANAGER;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_METASTORE_AUTHORIZATION_AUTH_READS;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_METASTORE_AUTHORIZATION_MANAGER;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTOREURIS;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_AUTO_CREATE_ALL;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_EXECUTE_SET_UGI;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_PRE_EVENT_LISTENERS;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.METASTORE_SCHEMA_VERIFICATION;
 import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.DYNAMICPARTITIONINGMODE;
 
 @Category({SlowTest.class, HiveStorageTest.class})
@@ -136,6 +139,9 @@ private static Map<String, String> getHivePluginConfig() {
     hiveConfig.put(FS_DEFAULT_NAME_KEY, dfsConf.get(FS_DEFAULT_NAME_KEY));
     hiveConfig.put(HIVE_SERVER2_ENABLE_DOAS.varname, hiveConf.get(HIVE_SERVER2_ENABLE_DOAS.varname));
     hiveConfig.put(METASTORE_EXECUTE_SET_UGI.varname, hiveConf.get(METASTORE_EXECUTE_SET_UGI.varname));
+    hiveConfig.put(METASTORE_SCHEMA_VERIFICATION.varname, hiveConf.get(METASTORE_SCHEMA_VERIFICATION.varname));
+    hiveConfig.put(METASTORE_AUTO_CREATE_ALL.varname, hiveConf.get(METASTORE_AUTO_CREATE_ALL.varname));
+    hiveConfig.put(HIVE_CBO_ENABLED.varname, hiveConf.get(HIVE_CBO_ENABLED.varname));
     return hiveConfig;
   }
 

contrib/storage-hive/core/src/test/java/org/apache/drill/exec/store/hive/HiveTestDataGenerator.java

@@ -149,6 +149,9 @@ private void generateTestData() throws Exception {
     conf.set(ConfVars.SCRATCHDIR.varname,  scratchDir.getAbsolutePath());
     conf.set(ConfVars.LOCALSCRATCHDIR.varname, localScratchDir.getAbsolutePath());
     conf.set(ConfVars.DYNAMICPARTITIONINGMODE.varname, "nonstrict");
+    conf.set(ConfVars.METASTORE_AUTO_CREATE_ALL.varname, "true");
+    conf.set(ConfVars.METASTORE_SCHEMA_VERIFICATION.varname, "false");
+    conf.set(ConfVars.HIVE_CBO_ENABLED.varname, "false");
 
     SessionState ss = new SessionState(conf);
     SessionState.start(ss);