http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java index 0000000,a1187c4..6354f77 mode 000000,100644..100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/repl/bootstrap/load/table/LoadTable.java @@@ -1,0 -1,243 +1,245 @@@ + /* + 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 + + http://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.hadoop.hive.ql.exec.repl.bootstrap.load.table; + + import org.apache.hadoop.fs.Path; + import org.apache.hadoop.hive.metastore.MetaStoreUtils; + import org.apache.hadoop.hive.metastore.TableType; + import org.apache.hadoop.hive.metastore.api.Database; + import org.apache.hadoop.hive.metastore.api.MetaException; + import org.apache.hadoop.hive.ql.ErrorMsg; + import org.apache.hadoop.hive.ql.exec.ReplCopyTask; + import org.apache.hadoop.hive.ql.exec.Task; + import org.apache.hadoop.hive.ql.exec.TaskFactory; + import org.apache.hadoop.hive.ql.exec.Utilities; + import org.apache.hadoop.hive.ql.exec.repl.ReplStateLogWork; + import org.apache.hadoop.hive.ql.exec.repl.bootstrap.ReplLoadTask; + import org.apache.hadoop.hive.ql.exec.repl.bootstrap.events.TableEvent; + import org.apache.hadoop.hive.ql.exec.repl.bootstrap.load.TaskTracker; + import org.apache.hadoop.hive.ql.exec.repl.bootstrap.load.util.Context; + import org.apache.hadoop.hive.ql.metadata.Table; + import org.apache.hadoop.hive.ql.parse.EximUtil; + import org.apache.hadoop.hive.ql.parse.ImportSemanticAnalyzer; + import org.apache.hadoop.hive.ql.parse.ReplicationSpec; + import org.apache.hadoop.hive.ql.parse.SemanticException; + import org.apache.hadoop.hive.ql.parse.repl.ReplLogger; + import org.apache.hadoop.hive.ql.plan.ImportTableDesc; + import org.apache.hadoop.hive.ql.plan.LoadTableDesc; + import org.apache.hadoop.hive.ql.plan.MoveWork; ++import org.apache.hadoop.hive.ql.session.SessionState; + import org.slf4j.Logger; + import org.slf4j.LoggerFactory; + + import java.io.IOException; + import java.io.Serializable; + import java.util.ArrayList; + import java.util.HashSet; + import java.util.List; + import java.util.TreeMap; + + import static org.apache.hadoop.hive.ql.parse.ImportSemanticAnalyzer.isPartitioned; + + public class LoadTable { + private final static Logger LOG = LoggerFactory.getLogger(LoadTable.class); + // private final Helper helper; + private final Context context; + private final ReplLogger replLogger; + private final TableContext tableContext; + private final TaskTracker tracker; + private final TableEvent event; + + public LoadTable(TableEvent event, Context context, ReplLogger replLogger, + TableContext tableContext, TaskTracker limiter) + throws SemanticException, IOException { + this.event = event; + this.context = context; + this.replLogger = replLogger; + this.tableContext = tableContext; + this.tracker = new TaskTracker(limiter); + } + + private void createTableReplLogTask(String tableName, TableType tableType) throws SemanticException { + ReplStateLogWork replLogWork = new ReplStateLogWork(replLogger,tableName, tableType); + Task<ReplStateLogWork> replLogTask = TaskFactory.get(replLogWork, context.hiveConf); + ReplLoadTask.dependency(tracker.tasks(), replLogTask); + + if (tracker.tasks().isEmpty()) { + tracker.addTask(replLogTask); + } else { + ReplLoadTask.dependency(tracker.tasks(), replLogTask); + + List<Task<? extends Serializable>> visited = new ArrayList<>(); + tracker.updateTaskCount(replLogTask, visited); + } + } + + public TaskTracker tasks() throws SemanticException { + // Path being passed to us is a table dump location. We go ahead and load it in as needed. + // If tblName is null, then we default to the table name specified in _metadata, which is good. + // or are both specified, in which case, that's what we are intended to create the new table as. + try { + if (event.shouldNotReplicate()) { + return tracker; + } + String dbName = tableContext.dbNameToLoadIn; //this can never be null or empty; + // Create table associated with the import + // Executed if relevant, and used to contain all the other details about the table if not. + ImportTableDesc tableDesc = tableContext.overrideProperties(event.tableDesc(dbName)); + Table table = ImportSemanticAnalyzer.tableIfExists(tableDesc, context.hiveDb); + ReplicationSpec replicationSpec = event.replicationSpec(); + + // Normally, on import, trying to create a table or a partition in a db that does not yet exist + // is a error condition. However, in the case of a REPL LOAD, it is possible that we are trying + // to create tasks to create a table inside a db that as-of-now does not exist, but there is + // a precursor Task waiting that will create it before this is encountered. Thus, we instantiate + // defaults and do not error out in that case. + // the above will change now since we are going to split replication load in multiple execution + // tasks and hence we could have created the database earlier in which case the waitOnPrecursor will + // be false and hence if db Not found we should error out. + Database parentDb = context.hiveDb.getDatabase(tableDesc.getDatabaseName()); + if (parentDb == null) { + if (!tableContext.waitOnPrecursor()) { + throw new SemanticException( + ErrorMsg.DATABASE_NOT_EXISTS.getMsg(tableDesc.getDatabaseName())); + } + } + + if (table == null) { + // If table doesn't exist, allow creating a new one only if the database state is older than the update. + if ((parentDb != null) && (!replicationSpec + .allowReplacementInto(parentDb.getParameters()))) { + // If the target table exists and is newer or same as current update based on repl.last.id, then just noop it. + return tracker; + } + } else { + if (!replicationSpec.allowReplacementInto(table.getParameters())) { + // If the target table exists and is newer or same as current update based on repl.last.id, then just noop it. + return tracker; + } + } + + if (tableDesc.getLocation() == null) { + tableDesc.setLocation(location(tableDesc, parentDb)); + } + + + /* Note: In the following section, Metadata-only import handling logic is + interleaved with regular repl-import logic. The rule of thumb being + followed here is that MD-only imports are essentially ALTERs. They do + not load data, and should not be "creating" any metadata - they should + be replacing instead. The only place it makes sense for a MD-only import + to create is in the case of a table that's been dropped and recreated, + or in the case of an unpartitioned table. In all other cases, it should + behave like a noop or a pure MD alter. + */ + if (table == null) { + newTableTasks(tableDesc); + } else { + existingTableTasks(tableDesc, table, replicationSpec); + } + + if (!isPartitioned(tableDesc)) { + createTableReplLogTask(tableDesc.getTableName(), tableDesc.tableType()); + } + return tracker; + } catch (Exception e) { + throw new SemanticException(e); + } + } + + private void existingTableTasks(ImportTableDesc tblDesc, Table table, + ReplicationSpec replicationSpec) { + if (!table.isPartitioned()) { + + LOG.debug("table non-partitioned"); + if (!replicationSpec.allowReplacementInto(table.getParameters())) { + return; // silently return, table is newer than our replacement. + } + + Task<? extends Serializable> alterTableTask = alterTableTask(tblDesc, replicationSpec); + if (replicationSpec.isMetadataOnly()) { + tracker.addTask(alterTableTask); + } else { + Task<?> loadTableTask = + loadTableTask(table, replicationSpec, event.metadataPath(), event.metadataPath()); + alterTableTask.addDependentTask(loadTableTask); + tracker.addTask(alterTableTask); + } + } + } + + private void newTableTasks(ImportTableDesc tblDesc) throws SemanticException { + Table table; + table = new Table(tblDesc.getDatabaseName(), tblDesc.getTableName()); + // Either we're dropping and re-creating, or the table didn't exist, and we're creating. + Task<?> createTableTask = + tblDesc.getCreateTableTask(new HashSet<>(), new HashSet<>(), context.hiveConf); + if (event.replicationSpec().isMetadataOnly()) { + tracker.addTask(createTableTask); + return; + } + if (!isPartitioned(tblDesc)) { + LOG.debug("adding dependent CopyWork/MoveWork for table"); + Task<?> loadTableTask = + loadTableTask(table, event.replicationSpec(), new Path(tblDesc.getLocation()), + event.metadataPath()); + createTableTask.addDependentTask(loadTableTask); + } + tracker.addTask(createTableTask); + } + + private String location(ImportTableDesc tblDesc, Database parentDb) + throws MetaException, SemanticException { + if (!tableContext.waitOnPrecursor()) { + return context.warehouse.getDefaultTablePath(parentDb, tblDesc.getTableName()).toString(); + } else { + Path tablePath = new Path( + context.warehouse.getDefaultDatabasePath(tblDesc.getDatabaseName()), + MetaStoreUtils.encodeTableName(tblDesc.getTableName().toLowerCase()) + ); + return context.warehouse.getDnsPath(tablePath).toString(); + } + } + + private Task<?> loadTableTask(Table table, ReplicationSpec replicationSpec, Path tgtPath, + Path fromURI) { + Path dataPath = new Path(fromURI, EximUtil.DATA_PATH_NAME); + Path tmpPath = context.utils.getExternalTmpPath(tgtPath); + Task<?> copyTask = + ReplCopyTask.getLoadCopyTask(replicationSpec, dataPath, tmpPath, context.hiveConf); + + LoadTableDesc loadTableWork = new LoadTableDesc( - tmpPath, Utilities.getTableDesc(table), new TreeMap<>(), replicationSpec.isReplace() ++ tmpPath, Utilities.getTableDesc(table), new TreeMap<>(), replicationSpec.isReplace(), ++ SessionState.get().getTxnMgr().getCurrentTxnId() + ); + MoveWork moveWork = + new MoveWork(new HashSet<>(), new HashSet<>(), loadTableWork, null, false); + Task<?> loadTableTask = TaskFactory.get(moveWork, context.hiveConf); + copyTask.addDependentTask(loadTableTask); + return copyTask; + } + + private Task<? extends Serializable> alterTableTask(ImportTableDesc tableDesc, + ReplicationSpec replicationSpec) { + tableDesc.setReplaceMode(true); + if ((replicationSpec != null) && (replicationSpec.isInReplicationScope())) { + tableDesc.setReplicationSpec(replicationSpec); + } + return tableDesc.getCreateTableTask(new HashSet<>(), new HashSet<>(), context.hiveConf); + } + }
http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java index cfa4e2c,90d1372..b5733ec --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorizedRowBatchCtx.java @@@ -145,12 -162,18 +162,18 @@@ public class VectorizedRowBatchCtx public static void getPartitionValues(VectorizedRowBatchCtx vrbCtx, Configuration hiveConf, FileSplit split, Object[] partitionValues) throws IOException { + // TODO: this is invalid for SMB. Keep this for now for legacy reasons. See the other overload. + MapWork mapWork = Utilities.getMapWork(hiveConf); + getPartitionValues(vrbCtx, mapWork, split, partitionValues); + } - Map<Path, PartitionDesc> pathToPartitionInfo = Utilities - .getMapWork(hiveConf).getPathToPartitionInfo(); + public static void getPartitionValues(VectorizedRowBatchCtx vrbCtx, + MapWork mapWork, FileSplit split, Object[] partitionValues) + throws IOException { + Map<Path, PartitionDesc> pathToPartitionInfo = mapWork.getPathToPartitionInfo(); PartitionDesc partDesc = HiveFileFormatUtils - .getPartitionDescFromPathRecursively(pathToPartitionInfo, + .getFromPathRecursively(pathToPartitionInfo, split.getPath(), IOPrepareCache.get().getPartitionDescMap()); getPartitionValues(vrbCtx, partDesc, partitionValues); http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java index f325c0e,feacdd8..9c9a2d3 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java @@@ -36,9 -37,9 +37,10 @@@ import org.apache.hadoop.hive.conf.Hive import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.metastore.api.DataOperationType; import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants; +import org.apache.hadoop.hive.metastore.MetaStoreUtils; import org.apache.hadoop.hive.metastore.TransactionalValidationListener; import org.apache.hadoop.hive.ql.ErrorMsg; + import org.apache.hadoop.hive.ql.io.orc.OrcRecordUpdater; import org.apache.hadoop.hive.ql.metadata.Table; import org.apache.hadoop.hive.shims.HadoopShims; import org.apache.hadoop.hive.shims.HadoopShims.HdfsFileStatusWithId; @@@ -345,11 -384,7 +385,10 @@@ public class AcidUtils public static final String SPLIT_UPDATE_STRING = "split_update"; public static final int HASH_BASED_MERGE_BIT = 0x02; public static final String HASH_BASED_MERGE_STRING = "hash_merge"; + public static final int INSERT_ONLY_BIT = 0x04; + public static final String INSERT_ONLY_STRING = "insert_only"; public static final String DEFAULT_VALUE_STRING = TransactionalValidationListener.DEFAULT_TRANSACTIONAL_PROPERTY; - public static final String LEGACY_VALUE_STRING = TransactionalValidationListener.LEGACY_TRANSACTIONAL_PROPERTY; + public static final String INSERTONLY_VALUE_STRING = TransactionalValidationListener.INSERTONLY_TRANSACTIONAL_PROPERTY; private AcidOperationalProperties() { } @@@ -401,12 -414,6 +430,9 @@@ if (propertiesStr.equalsIgnoreCase(DEFAULT_VALUE_STRING)) { return AcidOperationalProperties.getDefault(); } - if (propertiesStr.equalsIgnoreCase(LEGACY_VALUE_STRING)) { - return AcidOperationalProperties.getLegacy(); - } + if (propertiesStr.equalsIgnoreCase(INSERTONLY_VALUE_STRING)) { + return AcidOperationalProperties.getInsertOnly(); + } AcidOperationalProperties obj = new AcidOperationalProperties(); String[] options = propertiesStr.split("\\|"); for (String option : options) { http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/HiveFileFormatUtils.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/MapredParquetInputFormat.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/io/parquet/MapredParquetInputFormat.java index a70fde6,38aaeed..6178de2 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/MapredParquetInputFormat.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/MapredParquetInputFormat.java @@@ -14,16 -14,12 +14,20 @@@ package org.apache.hadoop.hive.ql.io.parquet; import java.io.IOException; +import java.util.Map; + import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; + import org.apache.hadoop.hive.common.io.DataCache; + import org.apache.hadoop.hive.common.io.FileMetadataCache; import org.apache.hadoop.hive.ql.exec.vector.VectorizedInputFormatInterface; +import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils; + import org.apache.hadoop.hive.ql.io.LlapCacheOnlyInputFormatInterface; +import org.apache.hadoop.hive.ql.io.parquet.serde.ParquetTableUtils; +import org.apache.hadoop.hive.ql.plan.MapWork; +import org.apache.hadoop.hive.ql.plan.PartitionDesc; +import org.apache.hadoop.mapred.FileSplit; +import org.apache.hadoop.mapred.JobConf; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.hadoop.hive.ql.exec.Utilities; http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/io/parquet/ProjectionPusher.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java index 01e8a48,5dec791..82804b2 --- a/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/DbTxnManager.java @@@ -423,11 -423,19 +423,19 @@@ public final class DbTxnManager extend compBuilder.setExclusive(); compBuilder.setOperationType(DataOperationType.NO_TXN); break; - + case INSERT_OVERWRITE: + t = getTable(output); + if (AcidUtils.isAcidTable(t)) { + compBuilder.setSemiShared(); + compBuilder.setOperationType(DataOperationType.UPDATE); + } else { + compBuilder.setExclusive(); + compBuilder.setOperationType(DataOperationType.NO_TXN); + } + break; case INSERT: assert t != null; - if(AcidUtils.isAcidTable(t)) { + if(AcidUtils.isFullAcidTable(t)) { compBuilder.setShared(); } else { http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java index ae451e2,d661f10..a0b735c --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java @@@ -65,9 -62,9 +65,10 @@@ import org.apache.hadoop.fs.FileStatus import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.PathFilter; + import org.apache.hadoop.hive.common.BlobStorageUtils; import org.apache.hadoop.hive.common.FileUtils; import org.apache.hadoop.hive.common.HiveStatsUtils; +import org.apache.hadoop.hive.common.JavaUtils; import org.apache.hadoop.hive.common.ObjectPair; import org.apache.hadoop.hive.common.StatsSetupConst; import org.apache.hadoop.hive.common.classification.InterfaceAudience.LimitedPrivate; @@@ -3338,8 -3250,9 +3389,9 @@@ private void constructOneLBLocationMap( // (1) Do not delete the dest dir before doing the move operation. // (2) It is assumed that subdir and dir are in same encryption zone. // (3) Move individual files from scr dir to dest dir. - boolean destIsSubDir = isSubDir(srcf, destf, srcFs, destFs, isSrcLocal); + boolean srcIsSubDirOfDest = isSubDir(srcf, destf, srcFs, destFs, isSrcLocal), + destIsSubDirOfSrc = isSubDir(destf, srcf, destFs, srcFs, false); + final String msg = "Unable to move source " + srcf + " to destination " + destf; - try { if (replace) { try{ http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/optimizer/BucketingSortingReduceSinkOptimizer.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java index 7fd8f04,a054abb..021507f --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java @@@ -1507,9 -1517,14 +1530,15 @@@ public class DDLSemanticAnalyzer extend } Table tab = getTable(tableName, true); + // cascade only occurs with partitioned table + if (isCascade && !tab.isPartitioned()) { + throw new SemanticException( + ErrorMsg.ALTER_TABLE_NON_PARTITIONED_TABLE_CASCADE_NOT_SUPPORTED); + } + // Determine the lock type to acquire - WriteEntity.WriteType writeType = WriteEntity.determineAlterTableWriteType(op); + WriteEntity.WriteType writeType = doForceExclusive + ? WriteType.DDL_EXCLUSIVE : WriteEntity.determineAlterTableWriteType(op); if (!alterPartitions) { inputs.add(new ReadEntity(tab)); http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java index 2401036,40c34bf..4109de4 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/EximUtil.java @@@ -140,15 -141,15 +141,16 @@@ public class EximUtil * Initialize the URI where the exported data collection is * to created for export, or is present for import */ - static URI getValidatedURI(HiveConf conf, String dcPath) throws SemanticException { + public static URI getValidatedURI(HiveConf conf, String dcPath) throws SemanticException { try { - boolean testMode = conf.getBoolVar(HiveConf.ConfVars.HIVETESTMODE); + boolean testMode = conf.getBoolVar(HiveConf.ConfVars.HIVETESTMODE) + || conf.getBoolVar(HiveConf.ConfVars.HIVEEXIMTESTMODE); URI uri = new Path(dcPath).toUri(); - String scheme = uri.getScheme(); + FileSystem fs = FileSystem.get(uri, conf); + // Get scheme from FileSystem + String scheme = fs.getScheme(); String authority = uri.getAuthority(); String path = uri.getPath(); - FileSystem fs = FileSystem.get(uri, conf); LOG.info("Path before norm :" + path); // generate absolute path relative to home directory http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/ExportSemanticAnalyzer.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/ExportSemanticAnalyzer.java index e534272,b8c6ea9..54ee7ae --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/ExportSemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/ExportSemanticAnalyzer.java @@@ -18,42 -18,12 +18,46 @@@ package org.apache.hadoop.hive.ql.parse; + +import org.apache.hadoop.hive.ql.metadata.HiveException; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.Serializable; +import java.net.URI; +import java.util.HashSet; +import java.util.List; + +import org.antlr.runtime.tree.Tree; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.common.FileUtils; +import org.apache.hadoop.hive.common.ValidReadTxnList; +import org.apache.hadoop.hive.common.ValidTxnList; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.MetaStoreUtils; +import org.apache.hadoop.hive.ql.Context; +import org.apache.hadoop.hive.ql.ErrorMsg; +import org.apache.hadoop.hive.ql.QueryState; +import org.apache.hadoop.hive.ql.exec.ReplCopyTask; +import org.apache.hadoop.hive.ql.exec.Task; +import org.apache.hadoop.hive.ql.exec.TaskFactory; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.hooks.ReadEntity; +import org.apache.hadoop.hive.ql.hooks.WriteEntity; +import org.apache.hadoop.hive.ql.metadata.Hive; + import org.antlr.runtime.tree.Tree; + import org.apache.hadoop.hive.ql.ErrorMsg; + import org.apache.hadoop.hive.ql.QueryState; import org.apache.hadoop.hive.ql.metadata.InvalidTableException; +import org.apache.hadoop.hive.ql.metadata.Partition; +import org.apache.hadoop.hive.ql.metadata.PartitionIterable; import org.apache.hadoop.hive.ql.metadata.Table; +import org.apache.hadoop.hive.ql.plan.CopyWork; +import org.slf4j.Logger; + import org.apache.hadoop.hive.ql.parse.repl.dump.TableExport; /** * ExportSemanticAnalyzer. http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java index 5ac15af,7f3460f..387582b --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java @@@ -62,12 -59,10 +62,12 @@@ import org.apache.hadoop.hive.ql.metada import org.apache.hadoop.hive.ql.metadata.InvalidTableException; import org.apache.hadoop.hive.ql.metadata.Table; import org.apache.hadoop.hive.ql.parse.repl.load.MetaData; + import org.apache.hadoop.hive.ql.parse.repl.load.UpdatedMetaDataTracker; import org.apache.hadoop.hive.ql.plan.AddPartitionDesc; +import org.apache.hadoop.hive.ql.plan.CopyWork; +import org.apache.hadoop.hive.ql.plan.CreateTableDesc; import org.apache.hadoop.hive.ql.plan.ImportTableDesc; import org.apache.hadoop.hive.ql.plan.DDLWork; - import org.apache.hadoop.hive.ql.plan.DropTableDesc; import org.apache.hadoop.hive.ql.plan.LoadTableDesc; import org.apache.hadoop.hive.ql.plan.MoveWork; import org.apache.hadoop.hive.ql.session.SessionState; @@@ -223,9 -228,8 +234,9 @@@ public class ImportSemanticAnalyzer ext } catch (Exception e) { throw new HiveException(e); } + boolean isSourceMm = MetaStoreUtils.isInsertOnlyTable(tblDesc.getTblProps()); - if ((replicationSpec!= null) && replicationSpec.isInReplicationScope()){ + if ((replicationSpec != null) && replicationSpec.isInReplicationScope()){ tblDesc.setReplicationSpec(replicationSpec); } @@@ -318,8 -306,8 +324,8 @@@ } else { createReplImportTasks( tblDesc, partitionDescs, - isPartSpecSet, replicationSpec, waitOnPrecursor, table, - fromURI, fs, wh, x, txnId, stmtId, isSourceMm); + replicationSpec, waitOnPrecursor, table, - fromURI, fs, wh, x, updatedMetadata); ++ fromURI, fs, wh, x, txnId, stmtId, isSourceMm, updatedMetadata); } return tableExists; } @@@ -800,8 -734,8 +799,8 @@@ // ensure if destination is not empty only for regular import Path tgtPath = new Path(table.getDataLocation().toString()); FileSystem tgtFs = FileSystem.get(tgtPath.toUri(), x.getConf()); - checkTargetLocationEmpty(tgtFs, tgtPath, replicationSpec, x); + checkTargetLocationEmpty(tgtFs, tgtPath, replicationSpec, x.getLOG()); - loadTable(fromURI, table, false, tgtPath, replicationSpec,x); + loadTable(fromURI, table, false, tgtPath, replicationSpec, x, txnId, stmtId, isSourceMm); } // Set this to read because we can't overwrite any existing partitions x.getOutputs().add(new WriteEntity(table, WriteEntity.WriteType.DDL_NO_LOCK)); @@@ -838,14 -769,8 +837,14 @@@ tablePath = wh.getDefaultTablePath(parentDb, tblDesc.getTableName()); } FileSystem tgtFs = FileSystem.get(tablePath.toUri(), x.getConf()); - checkTargetLocationEmpty(tgtFs, tablePath, replicationSpec,x); + checkTargetLocationEmpty(tgtFs, tablePath, replicationSpec,x.getLOG()); - t.addDependentTask(loadTable(fromURI, table, false, tablePath, replicationSpec, x)); + if (isSourceMm) { // since target table doesn't exist, it should inherit soruce table's properties + Map<String, String> tblproperties = table.getParameters(); + tblproperties.put("transactional", "true"); + tblproperties.put("transactional_properties", "insert_only"); + table.setParameters(tblproperties); + } + t.addDependentTask(loadTable(fromURI, table, false, tablePath, replicationSpec, x, txnId, stmtId, isSourceMm)); } } x.getTasks().add(t); @@@ -866,29 -783,14 +865,15 @@@ private static void createReplImportTasks( ImportTableDesc tblDesc, List<AddPartitionDesc> partitionDescs, - boolean isPartSpecSet, ReplicationSpec replicationSpec, boolean waitOnPrecursor, + ReplicationSpec replicationSpec, boolean waitOnPrecursor, Table table, URI fromURI, FileSystem fs, Warehouse wh, - EximUtil.SemanticAnalyzerWrapperContext x, Long txnId, int stmtId, boolean isSourceMm) - EximUtil.SemanticAnalyzerWrapperContext x, ++ EximUtil.SemanticAnalyzerWrapperContext x, Long txnId, int stmtId, boolean isSourceMm, + UpdatedMetaDataTracker updatedMetadata) throws HiveException, URISyntaxException, IOException, MetaException { + Task<?> dr = null; WriteEntity.WriteType lockType = WriteEntity.WriteType.DDL_NO_LOCK; - if ((table != null) && (isPartitioned(tblDesc) != table.isPartitioned())){ - // If destination table exists, but is partitioned, and we think we're writing to an unpartitioned - // or if destination table exists, but is unpartitioned and we think we're writing to a partitioned - // table, then this can only happen because there are drops in the queue that are yet to be processed. - // So, we check the repl.last.id of the destination, and if it's newer, we no-op. If it's older, we - // drop and re-create. - if (replicationSpec.allowReplacementInto(table)){ - dr = dropTableTask(table, x); - lockType = WriteEntity.WriteType.DDL_EXCLUSIVE; - table = null; // null it out so we go into the table re-create flow. - } else { - return; // noop out of here. - } - } - // Normally, on import, trying to create a table or a partition in a db that does not yet exist // is a error condition. However, in the case of a REPL LOAD, it is possible that we are trying // to create tasks to create a table inside a db that as-of-now does not exist, but there is @@@ -942,21 -863,18 +949,18 @@@ for (AddPartitionDesc addPartitionDesc : partitionDescs) { addPartitionDesc.setReplicationSpec(replicationSpec); t.addDependentTask( - addSinglePartition(fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x)); + addSinglePartition(fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x, txnId, stmtId, isSourceMm, ict)); + if (updatedMetadata != null) { + updatedMetadata.addPartition(addPartitionDesc.getPartition(0).getPartSpec()); + } } } else { x.getLOG().debug("adding dependent CopyWork/MoveWork for table"); - t.addDependentTask(loadTable(fromURI, table, true, new Path(tblDesc.getLocation()),replicationSpec, x)); + t.addDependentTask(loadTable(fromURI, table, true, new Path(tblDesc.getLocation()), replicationSpec, x, txnId, stmtId, isSourceMm)); } } - if (dr == null){ - // Simply create - x.getTasks().add(t); - } else { - // Drop and recreate - dr.addDependentTask(t); - x.getTasks().add(dr); - } + // Simply create + x.getTasks().add(t); } else { // Table existed, and is okay to replicate into, not dropping and re-creating. if (table.isPartitioned()) { @@@ -971,15 -887,18 +975,18 @@@ if ((ptn = x.getHive().getPartition(table, partSpec, false)) == null) { if (!replicationSpec.isMetadataOnly()){ x.getTasks().add(addSinglePartition( - fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x)); + fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x, txnId, stmtId, isSourceMm, ict)); + if (updatedMetadata != null) { + updatedMetadata.addPartition(addPartitionDesc.getPartition(0).getPartSpec()); + } } } else { // If replicating, then the partition already existing means we need to replace, maybe, if // the destination ptn's repl.last.id is older than the replacement's. - if (replicationSpec.allowReplacementInto(ptn)){ + if (replicationSpec.allowReplacementInto(ptn.getParameters())){ if (!replicationSpec.isMetadataOnly()){ x.getTasks().add(addSinglePartition( - fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x)); + fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, x, txnId, stmtId, isSourceMm, ict)); } else { x.getTasks().add(alterSinglePartition( fromURI, fs, tblDesc, table, wh, addPartitionDesc, replicationSpec, ptn, x)); @@@ -1002,13 -921,9 +1009,10 @@@ } } else { x.getLOG().debug("table non-partitioned"); - if (!replicationSpec.allowReplacementInto(table)){ - return; // silently return, table is newer than our replacement. - } if (!replicationSpec.isMetadataOnly()) { // repl-imports are replace-into unless the event is insert-into - loadTable(fromURI, table, replicationSpec.isReplace(), new Path(fromURI), replicationSpec, x); + loadTable(fromURI, table, replicationSpec.isReplace(), new Path(fromURI), + replicationSpec, x, txnId, stmtId, isSourceMm); } else { x.getTasks().add(alterTableTask(tblDesc, x, replicationSpec)); } http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java index 9b39cd0,fa79700..5e70863 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java @@@ -227,6 -228,9 +229,9 @@@ public class LoadSemanticAnalyzer exten + " and use 'insert... select' to allow Hive to enforce bucketing. " + error); } - if(AcidUtils.isAcidTable(ts.tableHandle)) { ++ if(AcidUtils.isAcidTable(ts.tableHandle) && !MetaStoreUtils.isInsertOnlyTable(ts.tableHandle.getParameters())) { + throw new SemanticException(ErrorMsg.LOAD_DATA_ON_ACID_TABLE, ts.tableHandle.getCompleteName()); + } // make sure the arguments make sense List<FileStatus> files = applyConstraintsAndGetFiles(fromURI, fromTree, isLocal); http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java index 19882eb,6f379da..ab3cbe1 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java @@@ -6851,17 -6896,13 +6855,21 @@@ public class SemanticAnalyzer extends B AcidUtils.Operation acidOp = AcidUtils.Operation.NOT_ACID; if (destTableIsAcid) { acidOp = getAcidType(table_desc.getOutputFileFormatClass(), dest); - checkAcidConstraints(qb, table_desc, dest_tab, acidOp); + checkAcidConstraints(qb, table_desc, dest_tab); } - ltd = new LoadTableDesc(queryTmpdir, table_desc, dpCtx, acidOp); + if (MetaStoreUtils.isInsertOnlyTable(table_desc.getProperties())) { + acidOp = getAcidType(table_desc.getOutputFileFormatClass(), dest); + } + if (isMmTable) { + txnId = SessionState.get().getTxnMgr().getCurrentTxnId(); + } + boolean isReplace = !qb.getParseInfo().isInsertIntoTable( + dest_tab.getDbName(), dest_tab.getTableName()); + ltd = new LoadTableDesc(queryTmpdir, table_desc, dpCtx, acidOp, isReplace, txnId); + // For Acid table, Insert Overwrite shouldn't replace the table content. We keep the old + // deltas and base and leave them up to the cleaner to clean up + ltd.setReplace(!qb.getParseInfo().isInsertIntoTable(dest_tab.getDbName(), + dest_tab.getTableName()) && !destTableIsAcid); ltd.setLbCtx(lbCtx); loadTableWork.add(ltd); } else { @@@ -6915,17 -7011,13 +6923,19 @@@ AcidUtils.Operation acidOp = AcidUtils.Operation.NOT_ACID; if (destTableIsAcid) { acidOp = getAcidType(table_desc.getOutputFileFormatClass(), dest); - checkAcidConstraints(qb, table_desc, dest_tab, acidOp); + checkAcidConstraints(qb, table_desc, dest_tab); } - ltd = new LoadTableDesc(queryTmpdir, table_desc, dest_part.getSpec(), acidOp); + if (MetaStoreUtils.isInsertOnlyTable(dest_part.getTable().getParameters())) { + acidOp = getAcidType(table_desc.getOutputFileFormatClass(), dest); + } + if (isMmTable) { + txnId = SessionState.get().getTxnMgr().getCurrentTxnId(); + } + ltd = new LoadTableDesc(queryTmpdir, table_desc, dest_part.getSpec(), acidOp, txnId); + // For Acid table, Insert Overwrite shouldn't replace the table content. We keep the old + // deltas and base and leave them up to the cleaner to clean up ltd.setReplace(!qb.getParseInfo().isInsertIntoTable(dest_tab.getDbName(), - dest_tab.getTableName())); + dest_tab.getTableName()) && !destTableIsAcid); ltd.setLbCtx(lbCtx); loadTableWork.add(ltd); @@@ -7091,122 -7226,18 +7101,122 @@@ genPartnCols(dest, input, qb, table_desc, dest_tab, rsCtx); } - FileSinkDesc fileSinkDesc = new FileSinkDesc( - queryTmpdir, - table_desc, - conf.getBoolVar(HiveConf.ConfVars.COMPRESSRESULT), - currentTableId, - rsCtx.isMultiFileSpray(), - canBeMerged, - rsCtx.getNumFiles(), - rsCtx.getTotalFiles(), - rsCtx.getPartnCols(), - dpCtx, - dest_path); + FileSinkDesc fileSinkDesc = createFileSinkDesc(dest, table_desc, dest_part, + dest_path, currentTableId, destTableIsAcid, destTableIsTemporary, + destTableIsMaterialization, queryTmpdir, rsCtx, dpCtx, lbCtx, fsRS, - canBeMerged, txnId, isMmCtas); ++ canBeMerged, dest_tab, txnId, isMmCtas); + if (isMmCtas) { + // Add FSD so that the LoadTask compilation could fix up its path to avoid the move. + tableDesc.setWriter(fileSinkDesc); + } + + if (SessionState.get().isHiveServerQuery() && + null != table_desc && + table_desc.getSerdeClassName().equalsIgnoreCase(ThriftJDBCBinarySerDe.class.getName()) && + HiveConf.getBoolVar(conf,HiveConf.ConfVars.HIVE_SERVER2_THRIFT_RESULTSET_SERIALIZE_IN_TASKS)) { + fileSinkDesc.setIsUsingThriftJDBCBinarySerDe(true); + } else { + fileSinkDesc.setIsUsingThriftJDBCBinarySerDe(false); + } + + Operator output = putOpInsertMap(OperatorFactory.getAndMakeChild( + fileSinkDesc, fsRS, input), inputRR); + + handleLineage(ltd, output); + + if (LOG.isDebugEnabled()) { + LOG.debug("Created FileSink Plan for clause: " + dest + "dest_path: " + + dest_path + " row schema: " + inputRR.toString()); + } + + FileSinkOperator fso = (FileSinkOperator) output; + fso.getConf().setTable(dest_tab); + // the following code is used to collect column stats when + // hive.stats.autogather=true + // and it is an insert overwrite or insert into table + if (dest_tab != null && conf.getBoolVar(ConfVars.HIVESTATSAUTOGATHER) + && conf.getBoolVar(ConfVars.HIVESTATSCOLAUTOGATHER) + && ColumnStatsAutoGatherContext.canRunAutogatherStats(fso)) { + if (dest_type.intValue() == QBMetaData.DEST_TABLE) { + genAutoColumnStatsGatheringPipeline(qb, table_desc, partSpec, input, qb.getParseInfo() + .isInsertIntoTable(dest_tab.getDbName(), dest_tab.getTableName())); + } else if (dest_type.intValue() == QBMetaData.DEST_PARTITION) { + genAutoColumnStatsGatheringPipeline(qb, table_desc, dest_part.getSpec(), input, qb + .getParseInfo().isInsertIntoTable(dest_tab.getDbName(), dest_tab.getTableName())); + + } + } + return output; + } + + private ColsAndTypes deriveFileSinkColTypes( + RowResolver inputRR, List<FieldSchema> field_schemas) throws SemanticException { + ColsAndTypes result = new ColsAndTypes("", ""); + ArrayList<ColumnInfo> colInfos = inputRR.getColumnInfos(); + boolean first = true; + for (ColumnInfo colInfo : colInfos) { + String[] nm = inputRR.reverseLookup(colInfo.getInternalName()); + + if (nm[1] != null) { // non-null column alias + colInfo.setAlias(nm[1]); + } + + String colName = colInfo.getInternalName(); //default column name + if (field_schemas != null) { + FieldSchema col = new FieldSchema(); + if (!("".equals(nm[0])) && nm[1] != null) { + colName = unescapeIdentifier(colInfo.getAlias()).toLowerCase(); // remove `` + } + colName = fixCtasColumnName(colName); + col.setName(colName); + String typeName = colInfo.getType().getTypeName(); + // CTAS should NOT create a VOID type + if (typeName.equals(serdeConstants.VOID_TYPE_NAME)) { + throw new SemanticException(ErrorMsg.CTAS_CREATES_VOID_TYPE.getMsg(colName)); + } + col.setType(typeName); + field_schemas.add(col); + } + + if (!first) { + result.cols = result.cols.concat(","); + result.colTypes = result.colTypes.concat(":"); + } + + first = false; + result.cols = result.cols.concat(colName); + + // Replace VOID type with string when the output is a temp table or + // local files. + // A VOID type can be generated under the query: + // + // select NULL from tt; + // or + // insert overwrite local directory "abc" select NULL from tt; + // + // where there is no column type to which the NULL value should be + // converted. + // + String tName = colInfo.getType().getTypeName(); + if (tName.equals(serdeConstants.VOID_TYPE_NAME)) { + result.colTypes = result.colTypes.concat(serdeConstants.STRING_TYPE_NAME); + } else { + result.colTypes = result.colTypes.concat(tName); + } + } + return result; + } + + private FileSinkDesc createFileSinkDesc(String dest, TableDesc table_desc, + Partition dest_part, Path dest_path, int currentTableId, + boolean destTableIsAcid, boolean destTableIsTemporary, + boolean destTableIsMaterialization, Path queryTmpdir, + SortBucketRSCtx rsCtx, DynamicPartitionCtx dpCtx, ListBucketingCtx lbCtx, - RowSchema fsRS, boolean canBeMerged, Long mmWriteId, boolean isMmCtas) throws SemanticException { ++ RowSchema fsRS, boolean canBeMerged, Table dest_tab, Long mmWriteId, boolean isMmCtas) throws SemanticException { + FileSinkDesc fileSinkDesc = new FileSinkDesc(queryTmpdir, table_desc, + conf.getBoolVar(HiveConf.ConfVars.COMPRESSRESULT), currentTableId, rsCtx.isMultiFileSpray(), + canBeMerged, rsCtx.getNumFiles(), rsCtx.getTotalFiles(), rsCtx.getPartnCols(), dpCtx, + dest_path, mmWriteId, isMmCtas); boolean isHiveServerQuery = SessionState.get().isHiveServerQuery(); fileSinkDesc.setHiveServerQuery(isHiveServerQuery); @@@ -7448,14 -7395,12 +7456,9 @@@ */ conf.set(AcidUtils.CONF_ACID_KEY, "true"); - if (!Operation.NOT_ACID.equals(acidOp)) { - if (table.getNumBuckets() < 1) { - throw new SemanticException(ErrorMsg.ACID_OP_ON_NONACID_TABLE, table.getTableName()); - } - if (table.getSortCols() != null && table.getSortCols().size() > 0) { - throw new SemanticException(ErrorMsg.ACID_NO_SORTED_BUCKETS, table.getTableName()); - } + if (table.getSortCols() != null && table.getSortCols().size() > 0) { + throw new SemanticException(ErrorMsg.ACID_NO_SORTED_BUCKETS, table.getTableName()); } - - - } /** http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/spark/GenSparkUtils.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/parse/spark/GenSparkUtils.java index d0a82af,4f8ba6f..ffc2daf --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/spark/GenSparkUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/spark/GenSparkUtils.java @@@ -355,6 -396,52 +396,52 @@@ public class GenSparkUtils } /** + * Create and add any dependent move tasks. + * + * This is forked from {@link GenMapRedUtils}. The difference is that it doesn't check + * 'isLinkedFileSink' and does not set parent dir for the linked file sinks. + */ + public static Path createMoveTask(Task<? extends Serializable> currTask, boolean chDir, + FileSinkOperator fsOp, ParseContext parseCtx, List<Task<MoveWork>> mvTasks, + HiveConf hconf, DependencyCollectionTask dependencyTask) { + + Path dest = null; ++ FileSinkDesc fileSinkDesc = fsOp.getConf(); + + if (chDir) { + dest = fsOp.getConf().getFinalDirName(); + + // generate the temporary file + // it must be on the same file system as the current destination + Context baseCtx = parseCtx.getContext(); + + Path tmpDir = baseCtx.getExternalTmpPath(dest); + - FileSinkDesc fileSinkDesc = fsOp.getConf(); + // Change all the linked file sink descriptors + if (fileSinkDesc.getLinkedFileSinkDesc() != null) { + for (FileSinkDesc fsConf : fileSinkDesc.getLinkedFileSinkDesc()) { + fsConf.setDirName(tmpDir); + } + } else { + fileSinkDesc.setDirName(tmpDir); + } + } + + Task<MoveWork> mvTask = null; + + if (!chDir) { - mvTask = GenMapRedUtils.findMoveTask(mvTasks, fsOp); ++ mvTask = GenMapRedUtils.findMoveTaskForFsopOutput(mvTasks, fileSinkDesc.getFinalDirName(), false); + } + + // Set the move task to be dependent on the current task + if (mvTask != null) { + GenMapRedUtils.addDependentMoveTasks(mvTask, hconf, currTask, dependencyTask); + } + + return dest; + } + + /** * Populate partition pruning information from the pruning sink operator to the * target MapWork (the MapWork for the big table side). The information include the source table * name, column name, and partition key expression. It also set up the temporary path used to http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/parse/spark/SparkProcessAnalyzeTable.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java ---------------------------------------------------------------------- diff --cc ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java index 955e0e5,a3df166..4732f0a --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java @@@ -20,9 -20,9 +20,10 @@@ package org.apache.hadoop.hive.ql.plan import java.util.ArrayList; import java.util.List; + import java.util.Objects; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.metastore.MetaStoreUtils; import org.apache.hadoop.hive.ql.io.AcidUtils; import org.apache.hadoop.hive.ql.metadata.Table; import org.apache.hadoop.hive.ql.plan.Explain.Level; http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/plan/TableScanDesc.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java ---------------------------------------------------------------------- diff --cc ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java index 3600a6b,0f129fc..5a8ac31 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java @@@ -487,79 -439,7 +439,6 @@@ public class TestTxnCommands extends Te } } - /** - * takes raw data and turns it into a string as if from Driver.getResults() - * sorts rows in dictionary order - */ - private List<String> stringifyValues(int[][] rowsIn) { - assert rowsIn.length > 0; - int[][] rows = rowsIn.clone(); - Arrays.sort(rows, new RowComp()); - List<String> rs = new ArrayList<String>(); - for(int[] row : rows) { - assert row.length > 0; - StringBuilder sb = new StringBuilder(); - for(int value : row) { - sb.append(value).append("\t"); - } - sb.setLength(sb.length() - 1); - rs.add(sb.toString()); - } - return rs; - } - private static final class RowComp implements Comparator<int[]> { - @Override - public int compare(int[] row1, int[] row2) { - assert row1 != null && row2 != null && row1.length == row2.length; - for(int i = 0; i < row1.length; i++) { - int comp = Integer.compare(row1[i], row2[i]); - if(comp != 0) { - return comp; - } - } - return 0; - } - } - private String makeValuesClause(int[][] rows) { - assert rows.length > 0; - StringBuilder sb = new StringBuilder("values"); - for(int[] row : rows) { - assert row.length > 0; - if(row.length > 1) { - sb.append("("); - } - for(int value : row) { - sb.append(value).append(","); - } - sb.setLength(sb.length() - 1);//remove trailing comma - if(row.length > 1) { - sb.append(")"); - } - sb.append(","); - } - sb.setLength(sb.length() - 1);//remove trailing comma - return sb.toString(); - } - - private List<String> runStatementOnDriver(String stmt) throws Exception { - LOG.info("Running " + stmt); - CommandProcessorResponse cpr = d.run(stmt); - if(cpr.getResponseCode() != 0) { - throw new RuntimeException(stmt + " failed: " + cpr); - } - List<String> rs = new ArrayList<String>(); - d.getResults(rs); - return rs; - } - private CommandProcessorResponse runStatementOnDriverNegative(String stmt) throws Exception { - CommandProcessorResponse cpr = d.run(stmt); - if(cpr.getResponseCode() != 0) { - return cpr; - } - throw new RuntimeException("Didn't get expected failure!"); - } -- - // @Ignore @Test public void exchangePartition() throws Exception { runStatementOnDriver("create database ex1"); http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java ---------------------------------------------------------------------- diff --cc ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java index 7a73a17,21b4a2c..bd6e6a0 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java @@@ -85,16 -84,16 +84,17 @@@ public class TestTxnCommands2 protected HiveConf hiveConf; protected Driver d; - protected static enum Table { + protected enum Table { ACIDTBL("acidTbl"), - ACIDTBLPART("acidTblPart"), + ACIDTBLPART("acidTblPart", "p"), NONACIDORCTBL("nonAcidOrcTbl"), - NONACIDPART("nonAcidPart"), - NONACIDPART2("nonAcidPart2"), - ACIDNESTEDPART("acidNestedPart"), + NONACIDPART("nonAcidPart", "p"), + NONACIDPART2("nonAcidPart2", "p2"), - ACIDNESTEDPART("acidNestedPart", "p,q"); ++ ACIDNESTEDPART("acidNestedPart", "p,q"), + MMTBL("mmTbl"); private final String name; + private final String partitionColumns; @Override public String toString() { return name; @@@ -738,14 -841,13 +843,13 @@@ // There should be only 1 directory left: base_0000001. // Original bucket files, delta directories and previous base directory should have been cleaned up. status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + - (Table.NONACIDORCTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + (Table.NONACIDORCTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); Assert.assertEquals(1, status.length); - Assert.assertEquals("base_0000023", status[0].getPath().getName()); + Assert.assertEquals("base_0000025", status[0].getPath().getName()); FileStatus[] buckets = fs.listStatus(status[0].getPath(), FileUtils.STAGING_DIR_PATH_FILTER); Arrays.sort(buckets); - Assert.assertEquals(BUCKET_COUNT, buckets.length); - Assert.assertEquals("bucket_00000", buckets[0].getPath().getName()); - Assert.assertEquals("bucket_00001", buckets[1].getPath().getName()); + Assert.assertEquals(1, buckets.length); + Assert.assertEquals("bucket_00001", buckets[0].getPath().getName()); rs = runStatementOnDriver("select a,b from " + Table.NONACIDORCTBL); resultData = new int[][] {{1, 3}, {3, 4}}; Assert.assertEquals(stringifyValues(resultData), rs); @@@ -1643,65 -1755,241 +1757,299 @@@ Assert.assertEquals(stringifyValues(rExpected), r); } + @Test + public void testBucketCodec() throws Exception { + d.destroy(); + //insert data in "legacy" format + hiveConf.setIntVar(HiveConf.ConfVars.TESTMODE_BUCKET_CODEC_VERSION, 0); + d = new Driver(hiveConf); + + int[][] targetVals = {{2,1},{4,3},{5,6},{7,8}}; + runStatementOnDriver("insert into " + Table.ACIDTBL + " " + makeValuesClause(targetVals)); + + d.destroy(); + hiveConf.setIntVar(HiveConf.ConfVars.TESTMODE_BUCKET_CODEC_VERSION, 1); + d = new Driver(hiveConf); + //do some operations with new format + runStatementOnDriver("update " + Table.ACIDTBL + " set b=11 where a in (5,7)"); + runStatementOnDriver("insert into " + Table.ACIDTBL + " values(11,11)"); + runStatementOnDriver("delete from " + Table.ACIDTBL + " where a = 7"); + + //make sure we get the right data back before/after compactions + List<String> r = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a,b"); + int[][] rExpected = {{2,1},{4,3},{5,11},{11,11}}; + Assert.assertEquals(stringifyValues(rExpected), r); + + runStatementOnDriver("ALTER TABLE " + Table.ACIDTBL + " COMPACT 'MINOR'"); + runWorker(hiveConf); + + r = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a,b"); + Assert.assertEquals(stringifyValues(rExpected), r); + + runStatementOnDriver("ALTER TABLE " + Table.ACIDTBL + " COMPACT 'MAJOR'"); + runWorker(hiveConf); + + r = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a,b"); + Assert.assertEquals(stringifyValues(rExpected), r); + } + /** + * Test the scenario when IOW comes in before a MAJOR compaction happens + * @throws Exception + */ + @Test + public void testInsertOverwrite1() throws Exception { + FileSystem fs = FileSystem.get(hiveConf); + FileStatus[] status; + + // 1. Insert two rows to an ACID table + runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(1,2)"); + runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(3,4)"); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs in the location + Assert.assertEquals(2, status.length); + for (int i = 0; i < status.length; i++) { + Assert.assertTrue(status[i].getPath().getName().matches("delta_.*")); + } + + // 2. INSERT OVERWRITE + // Prepare data for the source table + runStatementOnDriver("insert into " + Table.NONACIDORCTBL + "(a,b) values(5,6),(7,8)"); + // Insert overwrite ACID table from source table + runStatementOnDriver("insert overwrite table " + Table.ACIDTBL + " select a,b from " + Table.NONACIDORCTBL); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs, plus a base dir in the location + Assert.assertEquals(3, status.length); + boolean sawBase = false; + String baseDir = ""; + int deltaCount = 0; + for (int i = 0; i < status.length; i++) { + String dirName = status[i].getPath().getName(); + if (dirName.matches("delta_.*")) { + deltaCount++; + } else { + sawBase = true; + baseDir = dirName; + Assert.assertTrue(baseDir.matches("base_.*")); + } + } + Assert.assertEquals(2, deltaCount); + Assert.assertTrue(sawBase); + // Verify query result + List<String> rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + int [][] resultData = new int[][] {{5,6},{7,8}}; + Assert.assertEquals(stringifyValues(resultData), rs); + + // 3. Perform a major compaction. Nothing should change. Both deltas and base dirs should have the same name. + // Re-verify directory layout and query result by using the same logic as above + runStatementOnDriver("alter table "+ Table.ACIDTBL + " compact 'MAJOR'"); + runWorker(hiveConf); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs, plus a base dir in the location + Assert.assertEquals(3, status.length); + sawBase = false; + deltaCount = 0; + for (int i = 0; i < status.length; i++) { + String dirName = status[i].getPath().getName(); + if (dirName.matches("delta_.*")) { + deltaCount++; + } else { + sawBase = true; + Assert.assertTrue(dirName.matches("base_.*")); + Assert.assertEquals(baseDir, dirName); + } + } + Assert.assertEquals(2, deltaCount); + Assert.assertTrue(sawBase); + // Verify query result + rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + + // 4. Run Cleaner. It should remove the 2 delta dirs. + runCleaner(hiveConf); + // There should be only 1 directory left: base_xxxxxxx. + // The delta dirs should have been cleaned up. + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + Assert.assertEquals(1, status.length); + Assert.assertTrue(status[0].getPath().getName().matches("base_.*")); + Assert.assertEquals(baseDir, status[0].getPath().getName()); + // Verify query result + rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + } + + /** + * Test the scenario when IOW comes in after a MAJOR compaction happens + * @throws Exception + */ + @Test + public void testInsertOverwrite2() throws Exception { + FileSystem fs = FileSystem.get(hiveConf); + FileStatus[] status; + + // 1. Insert two rows to an ACID table + runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(1,2)"); + runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(3,4)"); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs in the location + Assert.assertEquals(2, status.length); + for (int i = 0; i < status.length; i++) { + Assert.assertTrue(status[i].getPath().getName().matches("delta_.*")); + } + + // 2. Perform a major compaction. There should be an extra base dir now. + runStatementOnDriver("alter table "+ Table.ACIDTBL + " compact 'MAJOR'"); + runWorker(hiveConf); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs, plus a base dir in the location + Assert.assertEquals(3, status.length); + boolean sawBase = false; + int deltaCount = 0; + for (int i = 0; i < status.length; i++) { + String dirName = status[i].getPath().getName(); + if (dirName.matches("delta_.*")) { + deltaCount++; + } else { + sawBase = true; + Assert.assertTrue(dirName.matches("base_.*")); + } + } + Assert.assertEquals(2, deltaCount); + Assert.assertTrue(sawBase); + // Verify query result + int [][] resultData = new int[][] {{1,2},{3,4}}; + List<String> rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + + // 3. INSERT OVERWRITE + // Prepare data for the source table + runStatementOnDriver("insert into " + Table.NONACIDORCTBL + "(a,b) values(5,6),(7,8)"); + // Insert overwrite ACID table from source table + runStatementOnDriver("insert overwrite table " + Table.ACIDTBL + " select a,b from " + Table.NONACIDORCTBL); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs, plus 2 base dirs in the location + Assert.assertEquals(4, status.length); + int baseCount = 0; + deltaCount = 0; + for (int i = 0; i < status.length; i++) { + String dirName = status[i].getPath().getName(); + if (dirName.matches("delta_.*")) { + deltaCount++; + } else { + baseCount++; + } + } + Assert.assertEquals(2, deltaCount); + Assert.assertEquals(2, baseCount); + // Verify query result + resultData = new int[][] {{5,6},{7,8}}; + rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + + // 4. Perform another major compaction. Nothing should change. Both deltas and both base dirs + // should have the same name. + // Re-verify directory layout and query result by using the same logic as above + runStatementOnDriver("alter table "+ Table.ACIDTBL + " compact 'MAJOR'"); + runWorker(hiveConf); + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + // There should be 2 delta dirs, plus 2 base dirs in the location + Assert.assertEquals(4, status.length); + baseCount = 0; + deltaCount = 0; + for (int i = 0; i < status.length; i++) { + String dirName = status[i].getPath().getName(); + if (dirName.matches("delta_.*")) { + deltaCount++; + } else { + Assert.assertTrue(dirName.matches("base_.*")); + baseCount++; + } + } + Assert.assertEquals(2, deltaCount); + Assert.assertEquals(2, baseCount); + // Verify query result + rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + + // 5. Run Cleaner. It should remove the 2 delta dirs and 1 old base dir. + runCleaner(hiveConf); + // There should be only 1 directory left: base_xxxxxxx. + // The delta dirs should have been cleaned up. + status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.ACIDTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + Assert.assertEquals(1, status.length); + Assert.assertTrue(status[0].getPath().getName().matches("base_.*")); + // Verify query result + rs = runStatementOnDriver("select a,b from " + Table.ACIDTBL + " order by a"); + Assert.assertEquals(stringifyValues(resultData), rs); + } + /** + * Test compaction for Micro-managed table + * 1. Regular compaction shouldn't impact any valid subdirectories of MM tables + * 2. Compactions will only remove subdirectories for aborted transactions of MM tables, if any + * @throws Exception + */ + @Test + public void testMmTableCompaction() throws Exception { + // 1. Insert some rows into MM table + runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)"); + runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)"); + // There should be 2 delta directories + verifyDirAndResult(2); + + // 2. Perform a MINOR compaction. Since nothing was aborted, subdirs should stay. + runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'"); + runWorker(hiveConf); + verifyDirAndResult(2); + + // 3. Let a transaction be aborted + hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true); + runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)"); + hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false); + // There should be 3 delta directories. The new one is the aborted one. + verifyDirAndResult(3); + + // 4. Perform a MINOR compaction again. This time it will remove the subdir for aborted transaction. + runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'"); + runWorker(hiveConf); + // The worker should remove the subdir for aborted transaction + verifyDirAndResult(2); + + // 5. Run Cleaner. Shouldn't impact anything. + runCleaner(hiveConf); + verifyDirAndResult(2); + } + + private void verifyDirAndResult(int expectedDeltas) throws Exception { + FileSystem fs = FileSystem.get(hiveConf); + // Verify the content of subdirs + FileStatus[] status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + + (Table.MMTBL).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); + int sawDeltaTimes = 0; + for (int i = 0; i < status.length; i++) { + Assert.assertTrue(status[i].getPath().getName().matches("delta_.*")); + sawDeltaTimes++; + FileStatus[] files = fs.listStatus(status[i].getPath(), FileUtils.STAGING_DIR_PATH_FILTER); + Assert.assertEquals(1, files.length); + Assert.assertTrue(files[0].getPath().getName().equals("000000_0")); + } + Assert.assertEquals(expectedDeltas, sawDeltaTimes); + + // Verify query result + int [][] resultData = new int[][] {{1,2}, {3,4}}; + List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL); + Assert.assertEquals(stringifyValues(resultData), rs); + } + + /** * takes raw data and turns it into a string as if from Driver.getResults() * sorts rows in dictionary order */ http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java ---------------------------------------------------------------------- diff --cc ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java index 6b19eb1,4c30732..ccd7d8e --- a/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java @@@ -40,10 -46,15 +46,7 @@@ import java.util.TreeSet import org.apache.commons.codec.binary.Base64; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.BlockLocation; -import org.apache.hadoop.fs.FSDataInputStream; --import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.FSInputStream; --import org.apache.hadoop.fs.FileStatus; --import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.LocatedFileStatus; --import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.RemoteIterator; ++import org.apache.hadoop.fs.*; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hive.common.ValidTxnList; import org.apache.hadoop.hive.common.type.HiveDecimal; http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/org/apache/hadoop/hive/ql/optimizer/TestGenMapRedUtilsCreateConditionalTask.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/42a38577/ql/src/test/queries/clientpositive/mm_all.q ---------------------------------------------------------------------- diff --cc ql/src/test/queries/clientpositive/mm_all.q index 6001e9f,0000000..e23260f mode 100644,000000..100644 --- a/ql/src/test/queries/clientpositive/mm_all.q +++ b/ql/src/test/queries/clientpositive/mm_all.q @@@ -1,370 -1,0 +1,325 @@@ +set hive.mapred.mode=nonstrict; +set hive.explain.user=false; +set hive.fetch.task.conversion=none; +set tez.grouping.min-size=1; +set tez.grouping.max-size=2; +set hive.exec.dynamic.partition.mode=nonstrict; +set hive.support.concurrency=true; +set hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager; + + +-- Force multiple writers when reading +drop table intermediate; +create table intermediate(key int) partitioned by (p int) stored as orc; +insert into table intermediate partition(p='455') select distinct key from src where key >= 0 order by key desc limit 2; +insert into table intermediate partition(p='456') select distinct key from src where key is not null order by key asc limit 2; +insert into table intermediate partition(p='457') select distinct key from src where key >= 100 order by key asc limit 2; + + +drop table part_mm; +create table part_mm(key int) partitioned by (key_mm int) stored as orc tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +explain insert into table part_mm partition(key_mm=455) select key from intermediate; +insert into table part_mm partition(key_mm=455) select key from intermediate; +insert into table part_mm partition(key_mm=456) select key from intermediate; +insert into table part_mm partition(key_mm=455) select key from intermediate; +select * from part_mm order by key, key_mm; + +-- TODO: doesn't work truncate table part_mm partition(key_mm=455); +select * from part_mm order by key, key_mm; +truncate table part_mm; +select * from part_mm order by key, key_mm; +drop table part_mm; + +drop table simple_mm; +create table simple_mm(key int) stored as orc tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +insert into table simple_mm select key from intermediate; +select * from simple_mm order by key; +insert into table simple_mm select key from intermediate; +select * from simple_mm order by key; +truncate table simple_mm; +select * from simple_mm; +drop table simple_mm; + + +-- simple DP (no bucketing) +drop table dp_mm; + +set hive.exec.dynamic.partition.mode=nonstrict; + +set hive.merge.mapredfiles=false; +set hive.merge.sparkfiles=false; +set hive.merge.tezfiles=false; + +create table dp_mm (key int) partitioned by (key1 string, key2 int) stored as orc + tblproperties ("transactional"="true", "transactional_properties"="insert_only"); + +insert into table dp_mm partition (key1='123', key2) select key, key from intermediate; + +select * from dp_mm order by key; + +drop table dp_mm; + + +-- union + +create table union_mm(id int) tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +insert into table union_mm +select temps.p from ( +select key as p from intermediate +union all +select key + 1 as p from intermediate ) temps; + +select * from union_mm order by id; + +insert into table union_mm +select p from +( +select key + 1 as p from intermediate +union all +select key from intermediate +) tab group by p +union all +select key + 2 as p from intermediate; + +select * from union_mm order by id; + +insert into table union_mm +SELECT p FROM +( + SELECT key + 1 as p FROM intermediate + UNION ALL + SELECT key as p FROM ( + SELECT distinct key FROM ( + SELECT key FROM ( + SELECT key + 2 as key FROM intermediate + UNION ALL + SELECT key FROM intermediate + )t1 + group by key)t2 + )t3 +)t4 +group by p; + + +select * from union_mm order by id; +drop table union_mm; + + +create table partunion_mm(id int) partitioned by (key int) tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +insert into table partunion_mm partition(key) +select temps.* from ( +select key as p, key from intermediate +union all +select key + 1 as p, key + 1 from intermediate ) temps; + +select * from partunion_mm order by id; +drop table partunion_mm; + + + +create table skew_mm(k1 int, k2 int, k4 int) skewed by (k1, k4) on ((0,0),(1,1),(2,2),(3,3)) + stored as directories tblproperties ("transactional"="true", "transactional_properties"="insert_only"); + +insert into table skew_mm +select key, key, key from intermediate; + +select * from skew_mm order by k2, k1, k4; +drop table skew_mm; + + +create table skew_dp_union_mm(k1 int, k2 int, k4 int) partitioned by (k3 int) +skewed by (k1, k4) on ((0,0),(1,1),(2,2),(3,3)) stored as directories tblproperties ("transactional"="true", "transactional_properties"="insert_only"); + +insert into table skew_dp_union_mm partition (k3) +select key as i, key as j, key as k, key as l from intermediate +union all +select key +1 as i, key +2 as j, key +3 as k, key +4 as l from intermediate; + + +select * from skew_dp_union_mm order by k2, k1, k4; +drop table skew_dp_union_mm; + + + +set hive.merge.orcfile.stripe.level=true; +set hive.merge.tezfiles=true; +set hive.merge.mapfiles=true; +set hive.merge.mapredfiles=true; + + +create table merge0_mm (id int) stored as orc tblproperties("transactional"="true", "transactional_properties"="insert_only"); + +insert into table merge0_mm select key from intermediate; +select * from merge0_mm; + +set tez.grouping.split-count=1; +insert into table merge0_mm select key from intermediate; +set tez.grouping.split-count=0; +select * from merge0_mm; + +drop table merge0_mm; + + +create table merge2_mm (id int) tblproperties("transactional"="true", "transactional_properties"="insert_only"); + +insert into table merge2_mm select key from intermediate; +select * from merge2_mm; + +set tez.grouping.split-count=1; +insert into table merge2_mm select key from intermediate; +set tez.grouping.split-count=0; +select * from merge2_mm; + +drop table merge2_mm; + + +create table merge1_mm (id int) partitioned by (key int) stored as orc tblproperties("transactional"="true", "transactional_properties"="insert_only"); + +insert into table merge1_mm partition (key) select key, key from intermediate; +select * from merge1_mm order by id, key; + +set tez.grouping.split-count=1; +insert into table merge1_mm partition (key) select key, key from intermediate; +set tez.grouping.split-count=0; +select * from merge1_mm order by id, key; + +drop table merge1_mm; + +set hive.merge.tezfiles=false; +set hive.merge.mapfiles=false; +set hive.merge.mapredfiles=false; + +-- TODO: need to include merge+union+DP, but it's broken for now + + +drop table ctas0_mm; +create table ctas0_mm tblproperties ("transactional"="true", "transactional_properties"="insert_only") as select * from intermediate; +select * from ctas0_mm; +drop table ctas0_mm; + +drop table ctas1_mm; +create table ctas1_mm tblproperties ("transactional"="true", "transactional_properties"="insert_only") as + select * from intermediate union all select * from intermediate; +select * from ctas1_mm; +drop table ctas1_mm; + + - drop table load0_mm; - create table load0_mm (key string, value string) stored as textfile tblproperties("transactional"="true", "transactional_properties"="insert_only"); - load data local inpath '../../data/files/kv1.txt' into table load0_mm; - select count(1) from load0_mm; - load data local inpath '../../data/files/kv2.txt' into table load0_mm; - select count(1) from load0_mm; - load data local inpath '../../data/files/kv2.txt' overwrite into table load0_mm; - select count(1) from load0_mm; - drop table load0_mm; - - - drop table intermediate2; - create table intermediate2 (key string, value string) stored as textfile - location 'file:${system:test.tmp.dir}/intermediate2'; - load data local inpath '../../data/files/kv1.txt' into table intermediate2; - load data local inpath '../../data/files/kv2.txt' into table intermediate2; - load data local inpath '../../data/files/kv3.txt' into table intermediate2; - - drop table load1_mm; - create table load1_mm (key string, value string) stored as textfile tblproperties("transactional"="true", "transactional_properties"="insert_only"); - load data inpath 'file:${system:test.tmp.dir}/intermediate2/kv2.txt' into table load1_mm; - load data inpath 'file:${system:test.tmp.dir}/intermediate2/kv1.txt' into table load1_mm; - select count(1) from load1_mm; - load data local inpath '../../data/files/kv1.txt' into table intermediate2; - load data local inpath '../../data/files/kv2.txt' into table intermediate2; - load data local inpath '../../data/files/kv3.txt' into table intermediate2; - load data inpath 'file:${system:test.tmp.dir}/intermediate2/kv*.txt' overwrite into table load1_mm; - select count(1) from load1_mm; - load data local inpath '../../data/files/kv2.txt' into table intermediate2; - load data inpath 'file:${system:test.tmp.dir}/intermediate2/kv2.txt' overwrite into table load1_mm; - select count(1) from load1_mm; - drop table load1_mm; - - drop table load2_mm; - create table load2_mm (key string, value string) - partitioned by (k int, l int) stored as textfile tblproperties("transactional"="true", "transactional_properties"="insert_only"); - load data local inpath '../../data/files/kv1.txt' into table intermediate2; - load data local inpath '../../data/files/kv2.txt' into table intermediate2; - load data local inpath '../../data/files/kv3.txt' into table intermediate2; - load data inpath 'file:${system:test.tmp.dir}/intermediate2/kv*.txt' into table load2_mm partition(k=5, l=5); - select count(1) from load2_mm; - drop table load2_mm; - drop table intermediate2; - - +drop table multi0_1_mm; +drop table multi0_2_mm; +create table multi0_1_mm (key int, key2 int) tblproperties("transactional"="true", "transactional_properties"="insert_only"); +create table multi0_2_mm (key int, key2 int) tblproperties("transactional"="true", "transactional_properties"="insert_only"); + +--from intermediate +--insert overwrite table multi0_1_mm select key, p +--insert overwrite table multi0_2_mm select p, key; +insert into table multi0_1_mm select key, p from intermediate; +insert into table multi0_2_mm select p, key from intermediate; + +select * from multi0_1_mm order by key, key2; +select * from multi0_2_mm order by key, key2; + +set hive.merge.mapredfiles=true; +set hive.merge.sparkfiles=true; +set hive.merge.tezfiles=true; + +--from intermediate +--insert into table multi0_1_mm select p, key +--insert overwrite table multi0_2_mm select key, p; +insert into table multi0_1_mm select p, key from intermediate; +insert into table multi0_2_mm select key, p from intermediate; +select * from multi0_1_mm order by key, key2; +select * from multi0_2_mm order by key, key2; + +set hive.merge.mapredfiles=false; +set hive.merge.sparkfiles=false; +set hive.merge.tezfiles=false; + +drop table multi0_1_mm; +drop table multi0_2_mm; + + +drop table multi1_mm; +create table multi1_mm (key int, key2 int) partitioned by (p int) tblproperties("transactional"="true", "transactional_properties"="insert_only"); +from intermediate +insert into table multi1_mm partition(p=1) select p, key +insert into table multi1_mm partition(p=2) select key, p; +select * from multi1_mm order by key, key2, p; +--from intermediate +--insert into table multi1_mm partition(p=2) select p, key +--insert overwrite table multi1_mm partition(p=1) select key, p; +insert into table multi1_mm partition(p=2) select p, key from intermediate; +insert into table multi1_mm partition(p=1) select key, p from intermediate; +select * from multi1_mm order by key, key2, p; + +from intermediate +insert into table multi1_mm partition(p) select p, key, p +insert into table multi1_mm partition(p=1) select key, p; +select key, key2, p from multi1_mm order by key, key2, p; + +from intermediate +insert into table multi1_mm partition(p) select p, key, 1 +insert into table multi1_mm partition(p=1) select key, p; +select key, key2, p from multi1_mm order by key, key2, p; +drop table multi1_mm; + + + + +set datanucleus.cache.collections=false; +set hive.stats.autogather=true; + +drop table stats_mm; +create table stats_mm(key int) tblproperties("transactional"="true", "transactional_properties"="insert_only"); +--insert overwrite table stats_mm select key from intermediate; +insert into table stats_mm select key from intermediate; +desc formatted stats_mm; + +insert into table stats_mm select key from intermediate; +desc formatted stats_mm; +drop table stats_mm; + +drop table stats2_mm; +create table stats2_mm tblproperties("transactional"="true", "transactional_properties"="insert_only") as select array(key, value) from src; +desc formatted stats2_mm; +drop table stats2_mm; + + +set hive.optimize.skewjoin=true; +set hive.skewjoin.key=2; +set hive.optimize.metadataonly=false; + +CREATE TABLE skewjoin_mm(key INT, value STRING) STORED AS TEXTFILE tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +FROM src src1 JOIN src src2 ON (src1.key = src2.key) INSERT into TABLE skewjoin_mm SELECT src1.key, src2.value; +select count(distinct key) from skewjoin_mm; +drop table skewjoin_mm; + +set hive.optimize.skewjoin=false; + +set hive.optimize.index.filter=true; +set hive.auto.convert.join=false; +CREATE TABLE parquet1_mm(id INT) STORED AS PARQUET tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +INSERT INTO parquet1_mm VALUES(1), (2); +CREATE TABLE parquet2_mm(id INT, value STRING) STORED AS PARQUET tblproperties ("transactional"="true", "transactional_properties"="insert_only"); +INSERT INTO parquet2_mm VALUES(1, 'value1'); +INSERT INTO parquet2_mm VALUES(1, 'value2'); +select parquet1_mm.id, t1.value, t2.value FROM parquet1_mm + JOIN parquet2_mm t1 ON parquet1_mm.id=t1.id + JOIN parquet2_mm t2 ON parquet1_mm.id=t2.id +where t1.value = 'value1' and t2.value = 'value2'; +drop table parquet1_mm; +drop table parquet2_mm; + +set hive.auto.convert.join=true; + + +DROP TABLE IF EXISTS temp1; +CREATE TEMPORARY TABLE temp1 (a int) TBLPROPERTIES ("transactional"="true", "transactional_properties"="insert_only"); +INSERT INTO temp1 SELECT key FROM intermediate; +DESC EXTENDED temp1; +SELECT * FROM temp1; + + +drop table intermediate; + + +
