You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by na...@apache.org on 2011/11/04 05:15:37 UTC
svn commit: r1197400 [1/3] - in /hive/trunk: data/files/
ql/src/java/org/apache/hadoop/hive/ql/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/
ql/src/java/org/apache/hadoop/hive/ql/metadata/ ql/src/...
Author: namit
Date: Fri Nov 4 04:15:36 2011
New Revision: 1197400
URL: http://svn.apache.org/viewvc?rev=1197400&view=rev
Log:
HIVE-2278 Support archiving for multiple partitions if the table is partitioned
by multiple columns (Marcin Kurczych and Yongqiang He via namit)
Added:
hive/trunk/data/files/archive_corrupt.rc (with props)
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ArchiveUtils.java
hive/trunk/ql/src/test/queries/clientnegative/archive_insert1.q
hive/trunk/ql/src/test/queries/clientnegative/archive_insert2.q
hive/trunk/ql/src/test/queries/clientnegative/archive_insert3.q
hive/trunk/ql/src/test/queries/clientnegative/archive_insert4.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi1.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi2.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi3.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi4.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi5.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi6.q
hive/trunk/ql/src/test/queries/clientnegative/archive_multi7.q
hive/trunk/ql/src/test/queries/clientnegative/archive_partspec1.q
hive/trunk/ql/src/test/queries/clientnegative/archive_partspec2.q
hive/trunk/ql/src/test/queries/clientnegative/archive_partspec3.q
hive/trunk/ql/src/test/queries/clientnegative/archive_partspec4.q
hive/trunk/ql/src/test/queries/clientnegative/archive_partspec5.q
hive/trunk/ql/src/test/queries/clientpositive/archive_corrupt.q
hive/trunk/ql/src/test/queries/clientpositive/archive_multi.q
hive/trunk/ql/src/test/results/clientnegative/archive_insert1.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_insert2.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_insert3.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_insert4.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi1.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi2.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi3.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi4.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi5.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi6.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_multi7.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_partspec1.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_partspec2.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_partspec3.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_partspec4.q.out
hive/trunk/ql/src/test/results/clientnegative/archive_partspec5.q.out
hive/trunk/ql/src/test/results/clientpositive/archive_corrupt.q.out
hive/trunk/ql/src/test/results/clientpositive/archive_multi.q.out
Modified:
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
hive/trunk/ql/src/test/results/clientnegative/archive1.q.out
hive/trunk/ql/src/test/results/clientnegative/archive2.q.out
Added: hive/trunk/data/files/archive_corrupt.rc
URL: http://svn.apache.org/viewvc/hive/trunk/data/files/archive_corrupt.rc?rev=1197400&view=auto
==============================================================================
Binary file - no diff available.
Propchange: hive/trunk/data/files/archive_corrupt.rc
------------------------------------------------------------------------------
svn:mime-type = application/octet-stream
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/Driver.java Fri Nov 4 04:15:36 2011
@@ -27,6 +27,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -732,18 +733,22 @@ public class Driver implements CommandPr
name = p.getName().split("@")[2];
}
- String partName = name;
String partialName = "";
String[] partns = name.split("/");
int len = p instanceof DummyPartition ? partns.length : partns.length - 1;
+ Map<String, String> partialSpec = new LinkedHashMap<String, String>();
for (int idx = 0; idx < len; idx++) {
String partn = partns[idx];
partialName += partn;
+ String[] nameValue = partn.split("=");
+ assert(nameValue.length == 2);
+ partialSpec.put(nameValue[0], nameValue[1]);
try {
locks.add(new HiveLockObj(
new HiveLockObject(new DummyPartition(p.getTable(), p.getTable().getDbName()
+ "/" + p.getTable().getTableName()
- + "/" + partialName), lockData), mode));
+ + "/" + partialName,
+ partialSpec), lockData), mode));
partialName += "/";
} catch (HiveException e) {
throw new SemanticException(e.getMessage());
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ArchiveUtils.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ArchiveUtils.java?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ArchiveUtils.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ArchiveUtils.java Fri Nov 4 04:15:36 2011
@@ -0,0 +1,363 @@
+/**
+ * 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;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.ql.metadata.Hive;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.metadata.Table;
+
+/**
+ * ArchiveUtils.
+ *
+ */
+@SuppressWarnings("nls")
+public final class ArchiveUtils {
+ private static final Log LOG = LogFactory.getLog(ArchiveUtils.class.getName());
+
+ public static String ARCHIVING_LEVEL = "archiving_level";
+
+ /**
+ * PartSpecInfo keeps fields and values extracted from partial partition info
+ * which is prefix of the full info.
+ */
+ public static class PartSpecInfo {
+ public List<FieldSchema> fields;
+ public List<String> values;
+ private PartSpecInfo(List<FieldSchema> fields, List<String> values) {
+ this.fields = fields;
+ this.values = values;
+ }
+
+ /**
+ * Extract partial prefix specification from table and key-value map
+ *
+ * @param tbl table in which partition is
+ * @param partSpec specification of partition
+ * @return extracted specification
+ */
+ static public PartSpecInfo create(Table tbl, Map<String, String> partSpec)
+ throws HiveException {
+ // we have to check if we receive prefix of partition keys so in table
+ // scheme like table/ds=2011-01-02/hr=13/
+ // ARCHIVE PARTITION (ds='2011-01-02') will work and
+ // ARCHIVE PARTITION(hr='13') won't
+ List<FieldSchema> prefixFields = new ArrayList<FieldSchema>();
+ List<String> prefixValues = new ArrayList<String>();
+ List<FieldSchema> partCols = tbl.getPartCols();
+ Iterator<String> itrPsKeys = partSpec.keySet().iterator();
+ for (FieldSchema fs : partCols) {
+ if (!itrPsKeys.hasNext()) {
+ break;
+ }
+ if (!itrPsKeys.next().toLowerCase().equals(
+ fs.getName().toLowerCase())) {
+ throw new HiveException("Invalid partition specifiation: "
+ + partSpec);
+ }
+ prefixFields.add(fs);
+ prefixValues.add(partSpec.get(fs.getName()));
+ }
+
+ return new PartSpecInfo(prefixFields, prefixValues);
+ }
+
+ /**
+ * Creates path where partitions matching prefix should lie in filesystem
+ * @param tbl table in which partition is
+ * @return expected location of partitions matching prefix in filesystem
+ */
+ public Path createPath(Table tbl) throws HiveException {
+ String prefixSubdir;
+ try {
+ prefixSubdir = Warehouse.makePartName(fields, values);
+ } catch (MetaException e) {
+ throw new HiveException("Unable to get partitions directories prefix", e);
+ }
+ URI tableDir = tbl.getDataLocation();
+ if(tableDir == null) {
+ throw new HiveException("Table has no location set");
+ }
+ return new Path(tableDir.toString(), prefixSubdir);
+ }
+ /**
+ * Generates name for prefix partial partition specification.
+ */
+ public String getName() throws HiveException {
+ try {
+ return Warehouse.makePartName(fields, values);
+ } catch (MetaException e) {
+ throw new HiveException("Unable to create partial name", e);
+ }
+ }
+ }
+
+ /**
+ * HarPathHelper helps to create har:/ URIs for locations inside of archive.
+ */
+ public static class HarPathHelper {
+ boolean parentSettable;
+ private final URI base, originalBase;
+
+ /**
+ * Creates helper for archive.
+ * @param archive absolute location of archive in underlying filesystem
+ * @param originalBase directory for which Hadoop archive was created
+ */
+ public HarPathHelper(HiveConf hconf, URI archive, URI originalBase) throws HiveException {
+ parentSettable = hconf.getBoolVar(HiveConf.ConfVars.HIVEHARPARENTDIRSETTABLE);
+ this.originalBase = addSlash(originalBase);
+ String parentHost = archive.getHost();
+ String harHost = null;
+ if (parentHost == null) {
+ harHost = archive.getScheme();
+ } else {
+ harHost = archive.getScheme() + "-" + parentHost;
+ }
+
+ // have to make sure there's slash after .har, otherwise resolve doesn't work
+ String path = addSlash(archive.getPath());
+ if(!path.endsWith(".har/")) {
+ throw new HiveException("HAR archive path must end with .har");
+ }
+ // harUri is used to access the partition's files, which are in the archive
+ // The format of the RI is something like:
+ // har://underlyingfsscheme-host:port/archivepath
+ try {
+ base = new URI("har", archive.getUserInfo(), harHost, archive.getPort(),
+ path, archive.getQuery(), archive.getFragment());
+ } catch (URISyntaxException e) {
+ throw new HiveException("Couldn't create har URI from archive URI", e);
+ }
+ }
+
+ /**
+ * Creates har URI for file/directory that was put there when creating HAR.
+ *
+ * With older versions of Hadoop, archiving a directory would produce
+ * the same directory structure, reflecting absoulute paths.
+ * If you created myArchive.har of /tmp/myDir the files in /tmp/myDir
+ * will be located under myArchive.har/tmp/myDir/*
+ *
+ * With newer versions, the parent directory can be specified. Assuming
+ * the parent directory was set to /tmp/myDir when creating the archive,
+ * the files can be found under myArchive.har/*
+ *
+ * This is why originalBase is argument - with new versions we can
+ * relativize URI, in older we keep absolute one.
+ *
+ * @param original file/directory path
+ * @return absolute HAR uri
+ */
+ public URI getHarUri(URI original) throws HiveException {
+ URI relative = null;
+ if (!parentSettable) {
+ String dirInArchive = original.getPath();
+ if(dirInArchive.length() > 1 && dirInArchive.charAt(0)=='/') {
+ dirInArchive = dirInArchive.substring(1);
+ }
+ try {
+ relative = new URI(null, null, dirInArchive, null);
+ } catch (URISyntaxException e) {
+ throw new HiveException("Couldn't create har URI for location");
+ } // relative URI with path only
+ }
+ else {
+ relative = originalBase.relativize(original);
+ if(relative.isAbsolute()) {
+ throw new HiveException("Unable to relativize URI");
+ }
+ }
+ return base.resolve(relative);
+ }
+ }
+
+ public static String addSlash(String s) {
+ return s.endsWith("/") ? s : s + "/";
+ }
+
+ /**
+ * Makes sure, that URI points to directory by adding slash to it.
+ * Useful in relativizing URIs.
+ */
+ public static URI addSlash(URI u) throws HiveException {
+ if(u.getPath().endsWith("/")) {
+ return u;
+ } else {
+ try {
+ return new URI(u.getScheme(), u.getAuthority(), u.getPath() + "/", u.getQuery(), u.getFragment());
+ } catch (URISyntaxException e) {
+ throw new HiveException("Couldn't append slash to a URI", e);
+ }
+ }
+ }
+
+ /**
+ * Determines whether a partition has been archived
+ *
+ * @param p
+ * @return
+ */
+ public static boolean isArchived(Partition p) {
+ Map<String, String> params = p.getParameters();
+ if ("true".equalsIgnoreCase(params.get(
+ org.apache.hadoop.hive.metastore.api.Constants.IS_ARCHIVED))) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ /**
+ * Returns archiving level, which is how many fields were set in partial
+ * specification ARCHIVE was run for
+ */
+ public static int getArchivingLevel(Partition p) throws HiveException {
+ if(!isArchived(p)) {
+ throw new HiveException("Getting level of unarchived partition");
+ }
+
+ Map<String, String> params = p.getParameters();
+ String lv = params.get(ArchiveUtils.ARCHIVING_LEVEL);
+ if(lv != null) {
+ return Integer.parseInt(lv);
+ } else { // partitions archived before introducing multiple archiving
+ return p.getValues().size();
+ }
+ }
+
+ /**
+ * Get a prefix of the given parition's string representation. The sencond
+ * argument, level, is used for the prefix length. For example, partition
+ * (ds='2010-01-01', hr='00', min='00'), level 1 will reture 'ds=2010-01-01',
+ * and level 2 will return 'ds=2010-01-01/hr=00'.
+ *
+ * @param p
+ * partition object
+ * @param level
+ * level for prefix depth
+ * @return
+ * @throws HiveException
+ */
+ public static String getPartialName(Partition p, int level) throws HiveException {
+ List<FieldSchema> ffields = p.getTable().getPartCols();
+ List<FieldSchema> fields = new ArrayList<FieldSchema>(level);
+ List<String> fvalues = p.getValues();
+ List<String> values = new ArrayList<String>(level);
+ for(int i =0;i<level;i++) {
+ FieldSchema fs = ffields.get(i);
+ String s = fvalues.get(i);
+ fields.add(fs);
+ values.add(s);
+ }
+ try {
+ return Warehouse.makePartName(fields, values);
+ } catch (MetaException e) {
+ throw new HiveException("Wasn't able to generate name" +
+ " for partial specification");
+ }
+ }
+
+ /**
+ * Determines if one can insert into partition(s), or there's a conflict with
+ * archive. It can be because partition is itself archived or it is to be
+ * created inside existing archive. The second case is when partition doesn't
+ * exist yet, but it would be inside of an archive if it existed. This one is
+ * quite tricky to check, we need to find at least one partition inside of
+ * the parent directory. If it is archived and archiving level tells that
+ * the archival was done of directory partition is in it means we cannot
+ * insert; otherwise we can.
+ * This method works both for full specifications and partial ones - in second
+ * case it checks if any partition that could possibly match such
+ * specification is inside archive.
+ *
+ * @param db - Hive object
+ * @param tbl - table where partition is
+ * @param partSpec - partition specification with possible nulls in case of
+ * dynamic partiton inserts
+ * @return null if partition can be inserted, string with colliding archive
+ * name when it can't
+ * @throws HiveException
+ */
+ public static String conflictingArchiveNameOrNull(Hive db, Table tbl,
+ LinkedHashMap<String, String> partSpec)
+ throws HiveException {
+
+ List<FieldSchema> partKeys = tbl.getPartitionKeys();
+ int partSpecLevel = 0;
+ for (FieldSchema partKey : partKeys) {
+ if (!partSpec.containsKey(partKey.getName())) {
+ break;
+ }
+ partSpecLevel++;
+ }
+
+ if(partSpecLevel != partSpec.size()) {
+ throw new HiveException("partspec " + partSpec
+ + " is wrong for table " + tbl.getTableName());
+ }
+
+ Map<String, String> spec = new HashMap<String, String>(partSpec);
+ List<String> reversedKeys = new LinkedList<String>();
+ for (FieldSchema fs : tbl.getPartCols()) {
+ if (spec.containsKey(fs.getName())) {
+ reversedKeys.add(0, fs.getName());
+ }
+ }
+
+ for (String rk : reversedKeys) {
+ List<Partition> parts = db.getPartitions(tbl, spec, (short) 1);
+ if (parts.size() != 0) {
+ Partition p = parts.get(0);
+ if (!isArchived(p)) {
+ // if archiving was done at this or at upper level, every matched
+ // partition would be archived, so it not being archived means
+ // no archiving was done neither at this nor at upper level
+ return null;
+ } else if (getArchivingLevel(p) > spec.size()) {
+ // if archiving was done at this or at upper level its level
+ // would be lesser or equal to specification size
+ // it is not, which means no archiving at this or upper level
+ return null;
+ } else {
+ return getPartialName(p, getArchivingLevel(p));
+ }
+ }
+ spec.remove(rk);
+ }
+ return null;
+ }
+}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java Fri Nov 4 04:15:36 2011
@@ -74,6 +74,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.QueryPlan;
+import org.apache.hadoop.hive.ql.exec.ArchiveUtils.PartSpecInfo;
import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.io.rcfile.merge.BlockMergeTask;
@@ -1045,22 +1046,37 @@ public class DDLTask extends Task<DDLWor
return 0;
}
- private void setIsArchived(Partition p, boolean state) {
+ /**
+ * Sets archiving flag locally; it has to be pushed into metastore
+ * @param p partition to set flag
+ * @param state desired state of IS_ARCHIVED flag
+ * @param level desired level for state == true, anything for false
+ */
+ private void setIsArchived(Partition p, boolean state, int level) {
Map<String, String> params = p.getParameters();
if (state) {
params.put(org.apache.hadoop.hive.metastore.api.Constants.IS_ARCHIVED,
"true");
+ params.put(ArchiveUtils.ARCHIVING_LEVEL, Integer
+ .toString(level));
} else {
params.remove(org.apache.hadoop.hive.metastore.api.Constants.IS_ARCHIVED);
+ params.remove(ArchiveUtils.ARCHIVING_LEVEL);
}
}
+ /**
+ * Returns original partition of archived partition, null for unarchived one
+ */
private String getOriginalLocation(Partition p) {
Map<String, String> params = p.getParameters();
return params.get(
org.apache.hadoop.hive.metastore.api.Constants.ORIGINAL_LOCATION);
}
+ /**
+ * Sets original location of partition which is to be archived
+ */
private void setOriginalLocation(Partition p, String loc) {
Map<String, String> params = p.getParameters();
if (loc == null) {
@@ -1070,56 +1086,19 @@ public class DDLTask extends Task<DDLWor
}
}
- // Returns only the path component of the URI
- private String getArchiveDirOnly(Path parentDir, String archiveName) {
- URI parentUri = parentDir.toUri();
- Path harDir = new Path(parentUri.getPath(), archiveName);
- return harDir.toString();
- }
-
/**
* Sets the appropriate attributes in the supplied Partition object to mark
* it as archived. Note that the metastore is not touched - a separate
* call to alter_partition is needed.
*
* @param p - the partition object to modify
- * @param parentDir - the parent directory of the archive, which is the
- * original directory that the partition's files resided in
- * @param dirInArchive - the directory within the archive file that contains
- * the partitions files
- * @param archiveName - the name of the archive
- * @throws URISyntaxException
- */
- private void setArchived(Partition p, Path parentDir, String dirInArchive, String archiveName)
- throws URISyntaxException {
- assert(Utilities.isArchived(p) == false);
- Map<String, String> params = p.getParameters();
-
- URI parentUri = parentDir.toUri();
- String parentHost = parentUri.getHost();
- String harHost = null;
- if (parentHost == null) {
- harHost = "";
- } else {
- harHost = parentUri.getScheme() + "-" + parentHost;
- }
-
- // harUri is used to access the partition's files, which are in the archive
- // The format of the RI is something like:
- // har://underlyingfsscheme-host:port/archivepath
- URI harUri = null;
- if (dirInArchive.length() == 0) {
- harUri = new URI("har", parentUri.getUserInfo(), harHost, parentUri.getPort(),
- getArchiveDirOnly(parentDir, archiveName),
- parentUri.getQuery(), parentUri.getFragment());
- } else {
- harUri = new URI("har", parentUri.getUserInfo(), harHost, parentUri.getPort(),
- new Path(getArchiveDirOnly(parentDir, archiveName), dirInArchive).toUri().getPath(),
- parentUri.getQuery(), parentUri.getFragment());
- }
- setIsArchived(p, true);
- setOriginalLocation(p, parentDir.toString());
- p.setLocation(harUri.toString());
+ * @param harPath - new location of partition (har schema URI)
+ */
+ private void setArchived(Partition p, Path harPath, int level) {
+ assert(ArchiveUtils.isArchived(p) == false);
+ setIsArchived(p, true, level);
+ setOriginalLocation(p, p.getLocation());
+ p.setLocation(harPath.toString());
}
/**
@@ -1130,9 +1109,9 @@ public class DDLTask extends Task<DDLWor
* @param p - the partition to modify
*/
private void setUnArchived(Partition p) {
- assert(Utilities.isArchived(p) == true);
+ assert(ArchiveUtils.isArchived(p) == true);
String parentDir = getOriginalLocation(p);
- setIsArchived(p, false);
+ setIsArchived(p, false, 0);
setOriginalLocation(p, null);
assert(parentDir != null);
p.setLocation(parentDir);
@@ -1166,7 +1145,35 @@ public class DDLTask extends Task<DDLWor
}
}
- private int archive(Hive db, AlterTableSimpleDesc simpleDesc, DriverContext driverContext)
+ /**
+ * Checks in partition is in custom (not-standard) location.
+ * @param tbl - table in which partition is
+ * @param p - partition
+ * @return true if partition location is custom, false if it is standard
+ */
+ boolean partitionInCustomLocation(Table tbl, Partition p)
+ throws HiveException {
+ String subdir = null;
+ try {
+ subdir = Warehouse.makePartName(tbl.getPartCols(), p.getValues());
+ } catch (MetaException e) {
+ throw new HiveException("Unable to get partition's directory", e);
+ }
+ URI tableDir = tbl.getDataLocation();
+ if(tableDir == null) {
+ throw new HiveException("Table has no location set");
+ }
+
+ String standardLocation = (new Path(tableDir.toString(), subdir)).toString();
+ if(ArchiveUtils.isArchived(p)) {
+ return !getOriginalLocation(p).equals(standardLocation);
+ } else {
+ return !p.getLocation().equals(standardLocation);
+ }
+ }
+
+ private int archive(Hive db, AlterTableSimpleDesc simpleDesc,
+ DriverContext driverContext)
throws HiveException {
String dbName = simpleDesc.getDbName();
String tblName = simpleDesc.getTableName();
@@ -1174,39 +1181,51 @@ public class DDLTask extends Task<DDLWor
Table tbl = db.getTable(dbName, tblName);
validateAlterTableType(tbl, AlterTableDesc.AlterTableTypes.ARCHIVE);
- Map<String, String> partSpec = simpleDesc.getPartSpec();
- Partition p = db.getPartition(tbl, partSpec, false);
-
if (tbl.getTableType() != TableType.MANAGED_TABLE) {
throw new HiveException("ARCHIVE can only be performed on managed tables");
}
- if (p == null) {
- throw new HiveException("Specified partition does not exist");
- }
+ Map<String, String> partSpec = simpleDesc.getPartSpec();
+ PartSpecInfo partSpecInfo = PartSpecInfo.create(tbl, partSpec);
+ List<Partition> partitions = db.getPartitions(tbl, partSpec);
- if (Utilities.isArchived(p)) {
- // If there were a failure right after the metadata was updated in an
- // archiving operation, it's possible that the original, unarchived files
- // weren't deleted.
- Path originalDir = new Path(getOriginalLocation(p));
- Path leftOverIntermediateOriginal = new Path(originalDir.getParent(),
- originalDir.getName() + INTERMEDIATE_ORIGINAL_DIR_SUFFIX);
+ Path originalDir = null;
- if (pathExists(leftOverIntermediateOriginal)) {
- console.printInfo("Deleting " + leftOverIntermediateOriginal +
- " left over from a previous archiving operation");
- deleteDir(leftOverIntermediateOriginal);
+ // when we have partial partitions specification we must assume partitions
+ // lie in standard place - if they were in custom locations putting
+ // them into one archive would involve mass amount of copying
+ // in full partition specification case we allow custom locations
+ // to keep backward compatibility
+ if (partitions.isEmpty()) {
+ throw new HiveException("No partition matches the specification");
+ } else if(partSpecInfo.values.size() != tbl.getPartCols().size()) {
+ // for partial specifications we need partitions to follow the scheme
+ for(Partition p: partitions){
+ if(partitionInCustomLocation(tbl, p)) {
+ String message = String.format("ARCHIVE cannot run for partition " +
+ "groups with custom locations like %s", p.getLocation());
+ throw new HiveException(message);
+ }
+ }
+ originalDir = partSpecInfo.createPath(tbl);
+ } else {
+ Partition p = partitions.get(0);
+ // partition can be archived if during recovery
+ if(ArchiveUtils.isArchived(p)) {
+ originalDir = new Path(getOriginalLocation(p));
+ } else {
+ originalDir = p.getPartitionPath();
}
-
- throw new HiveException("Specified partition is already archived");
}
- Path originalDir = p.getPartitionPath();
Path intermediateArchivedDir = new Path(originalDir.getParent(),
originalDir.getName() + INTERMEDIATE_ARCHIVED_DIR_SUFFIX);
Path intermediateOriginalDir = new Path(originalDir.getParent(),
originalDir.getName() + INTERMEDIATE_ORIGINAL_DIR_SUFFIX);
+
+ console.printInfo("intermediate.archived is " + intermediateArchivedDir.toString());
+ console.printInfo("intermediate.original is " + intermediateOriginalDir.toString());
+
String archiveName = "data.har";
FileSystem fs = null;
try {
@@ -1215,6 +1234,34 @@ public class DDLTask extends Task<DDLWor
throw new HiveException(e);
}
+ URI archiveUri = (new Path(originalDir, archiveName)).toUri();
+ URI originalUri = ArchiveUtils.addSlash(originalDir.toUri());
+ ArchiveUtils.HarPathHelper harHelper = new ArchiveUtils.HarPathHelper(
+ conf, archiveUri, originalUri);
+
+ // we checked if partitions matching specification are marked as archived
+ // in the metadata; if they are and their levels are the same as we would
+ // set it later it means previous run failed and we have to do the recovery;
+ // if they are different, we throw an error
+ for(Partition p: partitions) {
+ if(ArchiveUtils.isArchived(p)) {
+ if(ArchiveUtils.getArchivingLevel(p) != partSpecInfo.values.size()) {
+ String name = ArchiveUtils.getPartialName(p, ArchiveUtils.getArchivingLevel(p));
+ String m = String.format("Conflict with existing archive %s", name);
+ throw new HiveException(m);
+ } else {
+ throw new HiveException("Partition(s) already archived");
+ }
+ }
+ }
+
+ boolean recovery = false;
+ if (pathExists(intermediateArchivedDir)
+ || pathExists(intermediateOriginalDir)) {
+ recovery = true;
+ console.printInfo("Starting recovery after failed ARCHIVE");
+ }
+
// The following steps seem roundabout, but they are meant to aid in
// recovery if a failure occurs and to keep a consistent state in the FS
@@ -1242,10 +1289,12 @@ public class DDLTask extends Task<DDLWor
// First create the archive in a tmp dir so that if the job fails, the
// bad files don't pollute the filesystem
- Path tmpDir = new Path(driverContext.getCtx().getExternalTmpFileURI(originalDir.toUri()), "partlevel");
+ Path tmpPath = new Path(driverContext.getCtx()
+ .getExternalTmpFileURI(originalDir.toUri()), "partlevel");
- console.printInfo("Creating " + archiveName + " for " + originalDir.toString());
- console.printInfo("in " + tmpDir);
+ console.printInfo("Creating " + archiveName +
+ " for " + originalDir.toString());
+ console.printInfo("in " + tmpPath);
console.printInfo("Please wait... (this may take a while)");
// Create the Hadoop archive
@@ -1254,10 +1303,10 @@ public class DDLTask extends Task<DDLWor
try {
int maxJobNameLen = conf.getIntVar(HiveConf.ConfVars.HIVEJOBNAMELENGTH);
String jobname = String.format("Archiving %s@%s",
- tbl.getTableName(), p.getName());
+ tbl.getTableName(), partSpecInfo.getName());
jobname = Utilities.abbreviate(jobname, maxJobNameLen - 6);
conf.setVar(HiveConf.ConfVars.HADOOPJOBNAME, jobname);
- ret = shim.createHadoopArchive(conf, originalDir, tmpDir, archiveName);
+ ret = shim.createHadoopArchive(conf, originalDir, tmpPath, archiveName);
} catch (Exception e) {
throw new HiveException(e);
}
@@ -1267,11 +1316,11 @@ public class DDLTask extends Task<DDLWor
// Move from the tmp dir to an intermediate directory, in the same level as
// the partition directory. e.g. .../hr=12-intermediate-archived
try {
- console.printInfo("Moving " + tmpDir + " to " + intermediateArchivedDir);
+ console.printInfo("Moving " + tmpPath + " to " + intermediateArchivedDir);
if (pathExists(intermediateArchivedDir)) {
throw new HiveException("The intermediate archive directory already exists.");
}
- fs.rename(tmpDir, intermediateArchivedDir);
+ fs.rename(tmpPath, intermediateArchivedDir);
} catch (IOException e) {
throw new HiveException("Error while moving tmp directory");
}
@@ -1315,38 +1364,28 @@ public class DDLTask extends Task<DDLWor
// Record this change in the metastore
try {
- boolean parentSettable =
- conf.getBoolVar(HiveConf.ConfVars.HIVEHARPARENTDIRSETTABLE);
-
- // dirInArchive is the directory within the archive that has all the files
- // for this partition. With older versions of Hadoop, archiving a
- // a directory would produce the same directory structure
- // in the archive. So if you created myArchive.har of /tmp/myDir, the
- // files in /tmp/myDir would be located under myArchive.har/tmp/myDir/*
- // In this case, dirInArchive should be tmp/myDir
-
- // With newer versions of Hadoop, the parent directory could be specified.
- // Assuming the parent directory was set to /tmp/myDir when creating the
- // archive, the files can be found under myArchive.har/*
- // In this case, dirInArchive should be empty
-
- String dirInArchive = "";
- if (!parentSettable) {
- dirInArchive = originalDir.toUri().getPath();
- if(dirInArchive.length() > 1 && dirInArchive.charAt(0)=='/') {
- dirInArchive = dirInArchive.substring(1);
- }
+ for(Partition p: partitions) {
+ URI originalPartitionUri = ArchiveUtils.addSlash(p.getPartitionPath().toUri());
+ URI harPartitionDir = harHelper.getHarUri(originalPartitionUri);
+ Path harPath = new Path(harPartitionDir.getScheme(),
+ harPartitionDir.getAuthority(),
+ harPartitionDir.getPath()); // make in Path to ensure no slash at the end
+ setArchived(p, harPath, partSpecInfo.values.size());
+ db.alterPartition(tblName, p);
}
- setArchived(p, originalDir, dirInArchive, archiveName);
- db.alterPartition(tblName, p);
} catch (Exception e) {
throw new HiveException("Unable to change the partition info for HAR", e);
}
// If a failure occurs here, the directory containing the original files
// will not be deleted. The user will run ARCHIVE again to clear this up
- deleteDir(intermediateOriginalDir);
+ if(pathExists(intermediateOriginalDir)) {
+ deleteDir(intermediateOriginalDir);
+ }
+ if(recovery) {
+ console.printInfo("Recovery after ARCHIVE succeeded");
+ }
return 0;
}
@@ -1361,118 +1400,152 @@ public class DDLTask extends Task<DDLWor
// Means user specified a table, not a partition
if (simpleDesc.getPartSpec() == null) {
- throw new HiveException("ARCHIVE is for partitions only");
+ throw new HiveException("UNARCHIVE is for partitions only");
}
- Map<String, String> partSpec = simpleDesc.getPartSpec();
- Partition p = db.getPartition(tbl, partSpec, false);
-
if (tbl.getTableType() != TableType.MANAGED_TABLE) {
throw new HiveException("UNARCHIVE can only be performed on managed tables");
}
- if (p == null) {
- throw new HiveException("Specified partition does not exist");
- }
-
- if (!Utilities.isArchived(p)) {
- Path location = new Path(p.getLocation());
- Path leftOverArchiveDir = new Path(location.getParent(),
- location.getName() + INTERMEDIATE_ARCHIVED_DIR_SUFFIX);
+ Map<String, String> partSpec = simpleDesc.getPartSpec();
+ PartSpecInfo partSpecInfo = PartSpecInfo.create(tbl, partSpec);
+ List<Partition> partitions = db.getPartitions(tbl, partSpec);
- if (pathExists(leftOverArchiveDir)) {
- console.printInfo("Deleting " + leftOverArchiveDir + " left over " +
- "from a previous unarchiving operation");
- deleteDir(leftOverArchiveDir);
+ int partSpecLevel = partSpec.size();
+
+ Path originalDir = null;
+
+ // when we have partial partitions specification we must assume partitions
+ // lie in standard place - if they were in custom locations putting
+ // them into one archive would involve mass amount of copying
+ // in full partition specification case we allow custom locations
+ // to keep backward compatibility
+ if (partitions.isEmpty()) {
+ throw new HiveException("No partition matches the specification");
+ } else if(partSpecInfo.values.size() != tbl.getPartCols().size()) {
+ // for partial specifications we need partitions to follow the scheme
+ for(Partition p: partitions){
+ if(partitionInCustomLocation(tbl, p)) {
+ String message = String.format("UNARCHIVE cannot run for partition " +
+ "groups with custom locations like %s", p.getLocation());
+ throw new HiveException(message);
+ }
+ }
+ originalDir = partSpecInfo.createPath(tbl);
+ } else {
+ Partition p = partitions.get(0);
+ if(ArchiveUtils.isArchived(p)) {
+ originalDir = new Path(getOriginalLocation(p));
+ } else {
+ originalDir = new Path(p.getLocation());
}
-
- throw new HiveException("Specified partition is not archived");
+ }
+
+ URI originalUri = ArchiveUtils.addSlash(originalDir.toUri());
+ Path intermediateArchivedDir = new Path(originalDir.getParent(),
+ originalDir.getName() + INTERMEDIATE_ARCHIVED_DIR_SUFFIX);
+ Path intermediateExtractedDir = new Path(originalDir.getParent(),
+ originalDir.getName() + INTERMEDIATE_EXTRACTED_DIR_SUFFIX);
+ boolean recovery = false;
+ if(pathExists(intermediateArchivedDir) || pathExists(intermediateExtractedDir)) {
+ recovery = true;
+ console.printInfo("Starting recovery after failed UNARCHIVE");
+ }
+
+ for(Partition p: partitions) {
+ checkArchiveProperty(partSpecLevel, recovery, p);
+ }
+
+ String archiveName = "data.har";
+ FileSystem fs = null;
+ try {
+ fs = originalDir.getFileSystem(conf);
+ } catch (IOException e) {
+ throw new HiveException(e);
}
- Path originalLocation = new Path(getOriginalLocation(p));
- Path sourceDir = new Path(p.getLocation());
- Path intermediateArchiveDir = new Path(originalLocation.getParent(),
- originalLocation.getName() + INTERMEDIATE_ARCHIVED_DIR_SUFFIX);
- Path intermediateExtractedDir = new Path(originalLocation.getParent(),
- originalLocation.getName() + INTERMEDIATE_EXTRACTED_DIR_SUFFIX);
+ // assume the archive is in the original dir, check if it exists
+ Path archivePath = new Path(originalDir, archiveName);
+ URI archiveUri = archivePath.toUri();
+ ArchiveUtils.HarPathHelper harHelper = new ArchiveUtils.HarPathHelper(conf,
+ archiveUri, originalUri);
+ URI sourceUri = harHelper.getHarUri(originalUri);
+ Path sourceDir = new Path(sourceUri.getScheme(), sourceUri.getAuthority(), sourceUri.getPath());
- Path tmpDir = new Path(driverContext
+ if(!pathExists(intermediateArchivedDir) && !pathExists(archivePath)) {
+ throw new HiveException("Haven't found any archive where it should be");
+ }
+
+ Path tmpPath = new Path(driverContext
.getCtx()
- .getExternalTmpFileURI(originalLocation.toUri()));
+ .getExternalTmpFileURI(originalDir.toUri()));
- FileSystem fs = null;
try {
- fs = tmpDir.getFileSystem(conf);
- // Verify that there are no files in the tmp dir, because if there are, it
- // would be copied to the partition
- FileStatus [] filesInTmpDir = fs.listStatus(tmpDir);
- if (filesInTmpDir != null && filesInTmpDir.length != 0) {
- for (FileStatus file : filesInTmpDir) {
- console.printInfo(file.getPath().toString());
- }
- throw new HiveException("Temporary directory " + tmpDir + " is not empty");
- }
-
+ fs = tmpPath.getFileSystem(conf);
} catch (IOException e) {
throw new HiveException(e);
}
// Some sanity checks
- if (originalLocation == null) {
+ if (originalDir == null) {
throw new HiveException("Missing archive data in the partition");
}
- if (!"har".equals(sourceDir.toUri().getScheme())) {
- throw new HiveException("Location should refer to a HAR");
- }
// Clarification of terms:
- // - The originalLocation directory represents the original directory of the
- // partition's files. They now contain an archived version of those files
+ // - The originalDir directory represents the original directory of the
+ // partitions' files. They now contain an archived version of those files
// eg. hdfs:/warehouse/myTable/ds=1/
// - The source directory is the directory containing all the files that
- // should be in the partition. e.g. har:/warehouse/myTable/ds=1/myTable.har/
+ // should be in the partitions. e.g. har:/warehouse/myTable/ds=1/myTable.har/
// Note the har:/ scheme
// Steps:
// 1. Extract the archive in a temporary folder
// 2. Move the archive dir to an intermediate dir that is in at the same
// dir as originalLocation. Call the new dir intermediate-extracted.
- // 3. Rename the original partition dir to an intermediate dir. Call the
+ // 3. Rename the original partitions dir to an intermediate dir. Call the
// renamed dir intermediate-archive
- // 4. Rename intermediate-extracted to the original partition dir
+ // 4. Rename intermediate-extracted to the original partitions dir
// 5. Change the metadata
- // 6. Delete the archived partition files in intermediate-archive
+ // 6. Delete the archived partitions files in intermediate-archive
if (!pathExists(intermediateExtractedDir) &&
- !pathExists(intermediateArchiveDir)) {
+ !pathExists(intermediateArchivedDir)) {
try {
-
+
// Copy the files out of the archive into the temporary directory
String copySource = sourceDir.toString();
- String copyDest = tmpDir.toString();
+ String copyDest = tmpPath.toString();
List<String> args = new ArrayList<String>();
args.add("-cp");
args.add(copySource);
args.add(copyDest);
console.printInfo("Copying " + copySource + " to " + copyDest);
+ FileSystem srcFs = FileSystem.get(sourceDir.toUri(), conf);
+ srcFs.initialize(sourceDir.toUri(), conf);
+
FsShell fss = new FsShell(conf);
int ret = 0;
try {
ret = ToolRunner.run(fss, args.toArray(new String[0]));
} catch (Exception e) {
+ e.printStackTrace();
throw new HiveException(e);
}
+
if (ret != 0) {
- throw new HiveException("Error while copying files from archive");
+ throw new HiveException("Error while copying files from archive, return code=" + ret);
+ } else {
+ console.printInfo("Succefully Copied " + copySource + " to " + copyDest);
}
- console.printInfo("Moving " + tmpDir + " to " + intermediateExtractedDir);
+ console.printInfo("Moving " + tmpPath + " to " + intermediateExtractedDir);
if (fs.exists(intermediateExtractedDir)) {
throw new HiveException("Invalid state: the intermediate extracted " +
"directory already exists.");
}
- fs.rename(tmpDir, intermediateExtractedDir);
+ fs.rename(tmpPath, intermediateExtractedDir);
} catch (Exception e) {
throw new HiveException(e);
}
@@ -1481,15 +1554,15 @@ public class DDLTask extends Task<DDLWor
// At this point, we know that the extracted files are in the intermediate
// extracted dir, or in the the original directory.
- if (!pathExists(intermediateArchiveDir)) {
+ if (!pathExists(intermediateArchivedDir)) {
try {
- console.printInfo("Moving " + originalLocation + " to " + intermediateArchiveDir);
- fs.rename(originalLocation, intermediateArchiveDir);
+ console.printInfo("Moving " + originalDir + " to " + intermediateArchivedDir);
+ fs.rename(originalDir, intermediateArchivedDir);
} catch (IOException e) {
throw new HiveException(e);
}
} else {
- console.printInfo(intermediateArchiveDir + " already exists. " +
+ console.printInfo(intermediateArchivedDir + " already exists. " +
"Assuming it contains the archived version of the partition");
}
@@ -1499,31 +1572,55 @@ public class DDLTask extends Task<DDLWor
// If the original location exists here, then it must be the extracted files
// because in the previous step, we moved the previous original location
// (containing the archived version of the files) to intermediateArchiveDir
- if (!pathExists(originalLocation)) {
+ if (!pathExists(originalDir)) {
try {
- console.printInfo("Moving " + intermediateExtractedDir + " to " + originalLocation);
- fs.rename(intermediateExtractedDir, originalLocation);
+ console.printInfo("Moving " + intermediateExtractedDir + " to " + originalDir);
+ fs.rename(intermediateExtractedDir, originalDir);
} catch (IOException e) {
throw new HiveException(e);
}
} else {
- console.printInfo(originalLocation + " already exists. " +
+ console.printInfo(originalDir + " already exists. " +
"Assuming it contains the extracted files in the partition");
}
- setUnArchived(p);
- try {
- db.alterPartition(tblName, p);
- } catch (InvalidOperationException e) {
- throw new HiveException(e);
+ for(Partition p: partitions) {
+ setUnArchived(p);
+ try {
+ db.alterPartition(tblName, p);
+ } catch (InvalidOperationException e) {
+ throw new HiveException(e);
+ }
}
+
// If a failure happens here, the intermediate archive files won't be
// deleted. The user will need to call unarchive again to clear those up.
- deleteDir(intermediateArchiveDir);
+ if(pathExists(intermediateArchivedDir)) {
+ deleteDir(intermediateArchivedDir);
+ }
+
+ if(recovery) {
+ console.printInfo("Recovery after UNARCHIVE succeeded");
+ }
return 0;
}
+ private void checkArchiveProperty(int partSpecLevel,
+ boolean recovery, Partition p) throws HiveException {
+ if (!ArchiveUtils.isArchived(p) && !recovery) {
+ throw new HiveException("Partition " + p.getName()
+ + " is not archived.");
+ }
+ int archiveLevel = ArchiveUtils.getArchivingLevel(p);
+ if (partSpecLevel > archiveLevel) {
+ throw new HiveException("Partition " + p.getName()
+ + " is archived at level " + archiveLevel
+ + ", and given partspec only has " + partSpecLevel
+ + " specs.");
+ }
+ }
+
private void validateAlterTableType(
Table tbl, AlterTableDesc.AlterTableTypes alterType) throws HiveException {
@@ -2937,9 +3034,9 @@ public class DDLTask extends Task<DDLWor
} else if (alterTbl.getOp() == AlterTableDesc.AlterTableTypes.ALTERLOCATION) {
String newLocation = alterTbl.getNewLocation();
try {
- URI locURI = new URI(newLocation);
- if (!locURI.isAbsolute() || locURI.getScheme() == null
- || locURI.getScheme().trim().equals("")) {
+ URI locUri = new URI(newLocation);
+ if (!locUri.isAbsolute() || locUri.getScheme() == null
+ || locUri.getScheme().trim().equals("")) {
throw new HiveException(
newLocation
+ " is not absolute or has no scheme information. "
@@ -2948,7 +3045,7 @@ public class DDLTask extends Task<DDLWor
if (part != null) {
part.setLocation(newLocation);
} else {
- tbl.setDataLocation(locURI);
+ tbl.setDataLocation(locUri);
}
} catch (URISyntaxException e) {
throw new HiveException(e);
@@ -3080,14 +3177,32 @@ public class DDLTask extends Task<DDLWor
List<Partition> partsToDelete = new ArrayList<Partition>();
for (Map<String, String> partSpec : dropTbl.getPartSpecs()) {
List<Partition> partitions = db.getPartitions(tbl, partSpec);
+ // this is to prevent dropping archived partition which is archived in a
+ // different level the drop command specified.
+ int partPrefixToDrop = 0;
+ for (FieldSchema fs : tbl.getPartCols()) {
+ if (partSpec.get(fs.getName()) != null) {
+ partPrefixToDrop += 1;
+ } else {
+ break;
+ }
+ }
for (Partition p : partitions) {
if (!p.canDrop()) {
- throw new HiveException("Table " + tbl.getTableName() +
- " Partition " + p.getName() +
- " is protected from being dropped");
+ throw new HiveException("Table " + tbl.getTableName()
+ + " Partition " + p.getName()
+ + " is protected from being dropped");
+ } else if (ArchiveUtils.isArchived(p)) {
+ int partAchiveLevel = ArchiveUtils.getArchivingLevel(p);
+ // trying to drop partitions inside a har, disallow it.
+ if (partAchiveLevel < partPrefixToDrop) {
+ throw new HiveException(
+ "Cannot drop a subset of partitions in an archive, partition "
+ + p.getName());
+ }
}
- partsToDelete.add(p);
}
+ partsToDelete.addAll(partitions);
}
// drop all existing partitions from the list
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Fri Nov 4 04:15:36 2011
@@ -1957,22 +1957,6 @@ public final class Utilities {
}
/**
- * Determines whether a partition has been archived
- *
- * @param p
- * @return
- */
- public static boolean isArchived(Partition p) {
- Map<String, String> params = p.getParameters();
- if ("true".equalsIgnoreCase(params.get(
- org.apache.hadoop.hive.metastore.api.Constants.IS_ARCHIVED))) {
- return true;
- } else {
- return false;
- }
- }
-
- /**
* Check if a function can be pushed down to JDO.
* Now only {=, AND, OR} are supported.
* @param func a generic function.
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lockmgr/zookeeper/ZooKeeperHiveLockManager.java Fri Nov 4 04:15:36 2011
@@ -693,7 +693,7 @@ public class ZooKeeperHiveLockManager im
}
if (partn == null) {
- return new HiveLockObject(new DummyPartition(tab, path), data);
+ return new HiveLockObject(new DummyPartition(tab, path, partSpec), data);
}
return new HiveLockObject(partn, data);
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java Fri Nov 4 04:15:36 2011
@@ -18,8 +18,14 @@
package org.apache.hadoop.hive.ql.metadata;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
/**
* A Hive Table Partition: is a fundamental storage unit within a Table. Currently, Hive does not support
@@ -35,12 +41,20 @@ public class DummyPartition extends Part
.getLog("hive.ql.metadata.DummyPartition");
private String name;
+ private LinkedHashMap<String, String> partSpec;
public DummyPartition() {
}
-
+
public DummyPartition(Table tbl, String name) throws HiveException {
setTable(tbl);
this.name = name;
+ }
+
+ public DummyPartition(Table tbl, String name,
+ Map<String, String> partSpec) throws HiveException {
+ setTable(tbl);
+ this.name = name;
+ this.partSpec = new LinkedHashMap<String, String>(partSpec);
}
public String getName() {
@@ -54,4 +68,19 @@ public class DummyPartition extends Part
public String getCompleteName() {
return getName();
}
+
+ @Override
+ public LinkedHashMap<String, String> getSpec() {
+ return partSpec;
+ }
+
+ @Override
+ public List<String> getValues() {
+ List<String> values = new ArrayList<String>();
+ for (FieldSchema fs : this.getTable().getPartCols()) {
+ values.add(partSpec.get(fs.getName()));
+ }
+ return values;
+ }
+
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java Fri Nov 4 04:15:36 2011
@@ -1574,14 +1574,16 @@ public class Hive {
*
* @param tbl
* object for which partition is needed. Must be partitioned.
+ * @param limit number of partitions to return
* @return list of partition objects
* @throws HiveException
*/
- public List<Partition> getPartitions(Table tbl, Map<String, String> partialPartSpec)
+ public List<Partition> getPartitions(Table tbl, Map<String, String> partialPartSpec,
+ short limit)
throws HiveException {
if (!tbl.isPartitioned()) {
throw new HiveException("Partition spec should only be supplied for a " +
- "partitioned table");
+ "partitioned table");
}
List<String> partialPvals = getPvals(tbl.getPartCols(), partialPartSpec);
@@ -1589,7 +1591,7 @@ public class Hive {
List<org.apache.hadoop.hive.metastore.api.Partition> partitions = null;
try {
partitions = getMSC().listPartitionsWithAuthInfo(tbl.getDbName(), tbl.getTableName(),
- partialPvals, (short) -1, getUserName(), getGroupNames());
+ partialPvals, limit, getUserName(), getGroupNames());
} catch (Exception e) {
throw new HiveException(e);
}
@@ -1609,6 +1611,21 @@ public class Hive {
*
* @param tbl
* object for which partition is needed. Must be partitioned.
+ * @return list of partition objects
+ * @throws HiveException
+ */
+ public List<Partition> getPartitions(Table tbl, Map<String, String> partialPartSpec)
+ throws HiveException {
+ return getPartitions(tbl, partialPartSpec, (short)-1);
+ }
+
+ /**
+ * get all the partitions of the table that matches the given partial
+ * specification. partition columns whose value is can be anything should be
+ * an empty string.
+ *
+ * @param tbl
+ * object for which partition is needed. Must be partitioned.
* @param partialPartSpec
* partial partition specification (some subpartitions can be empty).
* @return list of partition objects
@@ -1884,7 +1901,7 @@ public class Hive {
// Note: there are race conditions here, but I don't believe
// they're worse than what was already present.
int counter = 1;
-
+
// Strip off the file type, if any so we don't make:
// 000000_0.gz -> 000000_0.gz_copy_1
String name = itemStaging.getName();
@@ -1896,10 +1913,10 @@ public class Hive {
} else {
filetype = "";
}
-
+
Path itemDest = new Path(destf, itemStaging.getName());
Path itemStagingBase = new Path(itemStaging.getParent(), name);
-
+
while (fs.exists(itemDest)) {
Path proposedStaging = itemStagingBase
.suffix("_copy_" + counter++).suffix(filetype);
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java Fri Nov 4 04:15:36 2011
@@ -728,26 +728,6 @@ public abstract class BaseSemanticAnalyz
}
}
- private void ErrorPartSpec(Map<String, String> partSpec, List<FieldSchema> parts)
- throws SemanticException {
- StringBuilder sb = new StringBuilder("Partition columns in the table schema are: (");
- for (FieldSchema fs: parts) {
- sb.append(fs.getName()).append(", ");
- }
- sb.setLength(sb.length() - 2); // remove the last ", "
- sb.append("), while the partitions specified in the query are: (");
-
- Iterator<String> itrPsKeys = partSpec.keySet().iterator();
- while (itrPsKeys.hasNext()) {
- sb.append(itrPsKeys.next()).append(", ");
- }
- sb.setLength(sb.length() - 2); // remove the last ", "
- sb.append(").");
-
- throw new SemanticException(
- ErrorMsg.PARTSPEC_DIFFER_FROM_SCHEMA.getMsg(sb.toString()));
- }
-
public Map<String, String> getPartSpec() {
return this.partSpec;
}
@@ -795,6 +775,73 @@ public abstract class BaseSemanticAnalyz
}
return partSpec;
}
+
+ /**
+ * Checks if given specification is proper specification for prefix of
+ * partition cols, for table partitioned by ds, hr, min valid ones are
+ * (ds='2008-04-08'), (ds='2008-04-08', hr='12'), (ds='2008-04-08', hr='12', min='30')
+ * invalid one is for example (ds='2008-04-08', min='30')
+ * @param spec specification key-value map
+ * @return true if the specification is prefix; never returns false, but throws
+ * @throws HiveException
+ */
+ final public boolean isValidPrefixSpec(Table tTable, Map<String, String> spec)
+ throws HiveException {
+
+ // TODO - types need to be checked.
+ List<FieldSchema> partCols = tTable.getPartitionKeys();
+ if (partCols == null || (partCols.size() == 0)) {
+ if (spec != null) {
+ throw new HiveException(
+ "table is not partitioned but partition spec exists: "
+ + spec);
+ } else {
+ return true;
+ }
+ }
+
+ if (spec == null) {
+ throw new HiveException("partition spec is not specified");
+ }
+
+ Iterator<String> itrPsKeys = spec.keySet().iterator();
+ for (FieldSchema fs: partCols) {
+ if(!itrPsKeys.hasNext()) {
+ break;
+ }
+ if (!itrPsKeys.next().toLowerCase().equals(
+ fs.getName().toLowerCase())) {
+ ErrorPartSpec(spec, partCols);
+ }
+ }
+
+ if(itrPsKeys.hasNext()) {
+ ErrorPartSpec(spec, partCols);
+ }
+
+ return true;
+ }
+
+ private static void ErrorPartSpec(Map<String, String> partSpec,
+ List<FieldSchema> parts) throws SemanticException {
+ StringBuilder sb =
+ new StringBuilder(
+ "Partition columns in the table schema are: (");
+ for (FieldSchema fs : parts) {
+ sb.append(fs.getName()).append(", ");
+ }
+ sb.setLength(sb.length() - 2); // remove the last ", "
+ sb.append("), while the partitions specified in the query are: (");
+
+ Iterator<String> itrPsKeys = partSpec.keySet().iterator();
+ while (itrPsKeys.hasNext()) {
+ sb.append(itrPsKeys.next()).append(", ");
+ }
+ sb.setLength(sb.length() - 2); // remove the last ", "
+ sb.append(").");
+ throw new SemanticException(ErrorMsg.PARTSPEC_DIFFER_FROM_SCHEMA
+ .getMsg(sb.toString()));
+ }
public Hive getDb() {
return db;
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java Fri Nov 4 04:15:36 2011
@@ -51,6 +51,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.PrincipalType;
import org.apache.hadoop.hive.ql.Driver;
+import org.apache.hadoop.hive.ql.exec.ArchiveUtils;
import org.apache.hadoop.hive.ql.exec.FetchTask;
import org.apache.hadoop.hive.ql.exec.Task;
import org.apache.hadoop.hive.ql.exec.TaskFactory;
@@ -1158,7 +1159,7 @@ public class DDLSemanticAnalyzer extends
}
bucketCols = part.getBucketCols();
inputFormatClass = part.getInputFormatClass();
- isArchived = Utilities.isArchived(part);
+ isArchived = ArchiveUtils.isArchived(part);
tblPartLoc = part.getDataLocation().toString();
}
} else {
@@ -1759,7 +1760,7 @@ public class DDLSemanticAnalyzer extends
// configured not to fail silently
boolean throwException =
!ifExists && !HiveConf.getBoolVar(conf, ConfVars.DROPIGNORESNONEXISTENT);
- addTablePartsOutputs(tblName, partSpecs, throwException, ast);
+ addTablePartsOutputs(tblName, partSpecs, throwException, false, ast);
}
rootTasks.add(TaskFactory.get(new DDLWork(getInputs(), getOutputs(),
@@ -1954,15 +1955,16 @@ public class DDLSemanticAnalyzer extends
// partition name to value
List<Map<String, String>> partSpecs = getPartitionSpecs(ast);
+ Table tab = null;
try {
- Table tab = db.getTable(db.getCurrentDatabase(), tblName, false);
+ tab = db.getTable(db.getCurrentDatabase(), tblName, false);
if (tab != null) {
inputs.add(new ReadEntity(tab));
}
} catch (HiveException e) {
throw new SemanticException(ErrorMsg.INVALID_TABLE.getMsg(tblName));
}
- addTablePartsOutputs(tblName, partSpecs);
+ addTablePartsOutputs(tblName, partSpecs, true);
if (partSpecs.size() > 1 ) {
throw new SemanticException(isUnArchive ?
@@ -1974,6 +1976,11 @@ public class DDLSemanticAnalyzer extends
}
Map<String,String> partSpec = partSpecs.get(0);
+ try {
+ isValidPrefixSpec(tab, partSpec);
+ } catch(HiveException e) {
+ throw new SemanticException(e.getMessage(), e);
+ }
AlterTableSimpleDesc archiveDesc = new AlterTableSimpleDesc(
db.getCurrentDatabase(), tblName, partSpec,
(isUnArchive ? AlterTableTypes.UNARCHIVE : AlterTableTypes.ARCHIVE));
@@ -2064,7 +2071,16 @@ public class DDLSemanticAnalyzer extends
*/
private void addTablePartsOutputs(String tblName, List<Map<String, String>> partSpecs)
throws SemanticException {
- addTablePartsOutputs(tblName, partSpecs, false, null);
+ addTablePartsOutputs(tblName, partSpecs, false, false, null);
+ }
+
+ /**
+ * Add the table partitions to be modified in the output, so that it is available for the
+ * pre-execution hook. If the partition does not exist, no error is thrown.
+ */
+ private void addTablePartsOutputs(String tblName, List<Map<String, String>> partSpecs, boolean allowMany)
+ throws SemanticException {
+ addTablePartsOutputs(tblName, partSpecs, false, allowMany, null);
}
/**
@@ -2073,7 +2089,7 @@ public class DDLSemanticAnalyzer extends
* throwIfNonExistent is true, otherwise ignore it.
*/
private void addTablePartsOutputs(String tblName, List<Map<String, String>> partSpecs,
- boolean throwIfNonExistent, ASTNode ast)
+ boolean throwIfNonExistent, boolean allowMany, ASTNode ast)
throws SemanticException {
Table tab;
try {
@@ -2085,18 +2101,32 @@ public class DDLSemanticAnalyzer extends
Iterator<Map<String, String>> i;
int index;
for (i = partSpecs.iterator(), index = 1; i.hasNext(); ++index) {
- Map<String, String> partSpec = i.next();
- try {
- Partition part = db.getPartition(tab, partSpec, false);
- if (part == null) {
- continue;
+ Map<String, String> partSpec = i.next();
+ List<Partition> parts = null;
+ if(allowMany) {
+ try {
+ parts = db.getPartitions(tab, partSpec);
+ } catch (HiveException e) {
+ LOG.error("Got HiveException during obtaining list of partitions");
}
- outputs.add(new WriteEntity(part));
- } catch (HiveException e) {
- // Ignore the error if the partition does not exist
- if (throwIfNonExistent) {
- throw new SemanticException(ErrorMsg.INVALID_PARTITION.getMsg(ast.getChild(index)));
- }
+ } else {
+ parts = new ArrayList<Partition>();
+ try {
+ Partition p = db.getPartition(tab, partSpec, false);
+ if(p != null) {
+ parts.add(p);
+ }
+ } catch(HiveException e) {
+ LOG.debug("Wrong specification");
+ }
+ }
+ if (parts.isEmpty()) {
+ if(throwIfNonExistent) {
+ throw new SemanticException(ErrorMsg.INVALID_PARTITION.getMsg(ast.getChild(index)));
+ }
+ }
+ for(Partition p: parts) {
+ outputs.add(new WriteEntity(p));
}
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=1197400&r1=1197399&r2=1197400&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Fri Nov 4 04:15:36 2011
@@ -27,9 +27,9 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
+import java.util.Map.Entry;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
@@ -58,6 +58,7 @@ import org.apache.hadoop.hive.ql.exec.Fu
import org.apache.hadoop.hive.ql.exec.GroupByOperator;
import org.apache.hadoop.hive.ql.exec.JoinOperator;
import org.apache.hadoop.hive.ql.exec.MapRedTask;
+import org.apache.hadoop.hive.ql.exec.ArchiveUtils;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorFactory;
import org.apache.hadoop.hive.ql.exec.RecordReader;
@@ -93,7 +94,6 @@ import org.apache.hadoop.hive.ql.metadat
import org.apache.hadoop.hive.ql.optimizer.GenMRFileSink1;
import org.apache.hadoop.hive.ql.optimizer.GenMROperator;
import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext;
-import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext.GenMapRedCtx;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink1;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink2;
import org.apache.hadoop.hive.ql.optimizer.GenMRRedSink3;
@@ -103,10 +103,14 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils;
import org.apache.hadoop.hive.ql.optimizer.MapJoinFactory;
import org.apache.hadoop.hive.ql.optimizer.Optimizer;
+import org.apache.hadoop.hive.ql.optimizer.GenMRProcContext.GenMapRedCtx;
import org.apache.hadoop.hive.ql.optimizer.physical.PhysicalContext;
import org.apache.hadoop.hive.ql.optimizer.physical.PhysicalOptimizer;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.optimizer.unionproc.UnionProcContext;
+import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.AnalyzeCreateCommonVars;
+import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.RowFormatParams;
+import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.StorageFormat;
import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.tableSpec.SpecType;
import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.plan.CreateTableDesc;
@@ -123,7 +127,6 @@ import org.apache.hadoop.hive.ql.plan.Ex
import org.apache.hadoop.hive.ql.plan.FetchWork;
import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.FilterDesc;
-import org.apache.hadoop.hive.ql.plan.FilterDesc.sampleDesc;
import org.apache.hadoop.hive.ql.plan.ForwardDesc;
import org.apache.hadoop.hive.ql.plan.GroupByDesc;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
@@ -146,12 +149,13 @@ import org.apache.hadoop.hive.ql.plan.Ta
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.ql.plan.UDTFDesc;
import org.apache.hadoop.hive.ql.plan.UnionDesc;
+import org.apache.hadoop.hive.ql.plan.FilterDesc.sampleDesc;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionState.ResourceType;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
-import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFHash;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.Mode;
import org.apache.hadoop.hive.serde.Constants;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.MetadataTypedColumnsetSerDe;
@@ -3835,8 +3839,9 @@ public class SemanticAnalyzer extends Ba
String ppath = dpCtx.getSPPath();
ppath = ppath.substring(0, ppath.length()-1);
DummyPartition p =
- new DummyPartition(dest_tab,
- dest_tab.getDbName() + "@" + dest_tab.getTableName() + "@" + ppath);
+ new DummyPartition(dest_tab, dest_tab.getDbName()
+ + "@" + dest_tab.getTableName() + "@" + ppath,
+ partSpec);
outputs.add(new WriteEntity(p, false));
} catch (HiveException e) {
throw new SemanticException(e.getMessage());
@@ -3858,10 +3863,6 @@ public class SemanticAnalyzer extends Ba
dest_path = new Path(tabPath.toUri().getScheme(), tabPath.toUri()
.getAuthority(), partPath.toUri().getPath());
- if ("har".equalsIgnoreCase(dest_path.toUri().getScheme())) {
- throw new SemanticException(ErrorMsg.OVERWRITE_ARCHIVED_PART
- .getMsg(qb.getParseInfo().getDestForClause(dest)));
- }
queryTmpdir = ctx.getExternalTmpFileURI(dest_path.toUri());
table_desc = Utilities.getTableDesc(dest_tab);
@@ -7517,6 +7518,7 @@ public class SemanticAnalyzer extends Ba
@Override
public void validate() throws SemanticException {
+ LOG.debug("validation start");
// Validate inputs and outputs have right protectmode to execute the query
for (ReadEntity readEntity: getInputs()) {
ReadEntity.Type type = readEntity.getType();
@@ -7549,8 +7551,30 @@ public class SemanticAnalyzer extends Ba
for (WriteEntity writeEntity: getOutputs()) {
WriteEntity.Type type = writeEntity.getType();
+
+ if(type == WriteEntity.Type.PARTITION || type == WriteEntity.Type.DUMMYPARTITION) {
+ String conflictingArchive;
+ try {
+ Partition usedp = writeEntity.getPartition();
+ Table tbl = usedp.getTable();
+
+ LOG.debug("validated " + usedp.getName());
+ LOG.debug(usedp.getTable());
+ conflictingArchive = ArchiveUtils
+ .conflictingArchiveNameOrNull(db, tbl, usedp.getSpec());
+ } catch (HiveException e) {
+ throw new SemanticException(e);
+ }
+ if(conflictingArchive != null) {
+ String message = String.format("Insert conflict with existing archive: %s",
+ conflictingArchive);
+ throw new SemanticException(message);
+ }
+ }
+
if (type != WriteEntity.Type.TABLE &&
type != WriteEntity.Type.PARTITION) {
+ LOG.debug("not validating writeEntity, because entity is neither table nor partition");
continue;
}
@@ -7582,8 +7606,10 @@ public class SemanticAnalyzer extends Ba
" Table " + tbl.getTableName() +
" Partition " + p.getName()));
}
+
}
else {
+ LOG.debug("Not a partition.");
tbl = writeEntity.getTable();
}
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_insert1.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_insert1.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_insert1.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_insert1.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to insert into archived partition.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08', hr='12');
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_insert2.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_insert2.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_insert2.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_insert2.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to insert into archived partition.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_insert3.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_insert3.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_insert3.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_insert3.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to create partition inside of archived directory.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_insert4.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_insert4.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_insert4.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_insert4.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,15 @@
+set hive.archive.enabled = true;
+-- Tests trying to (possible) dynamic insert into archived partition.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08', hr='12');
+
+SET hive.exec.dynamic.partition=true;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr)
+SELECT key, value, hr FROM srcpart WHERE ds='2008-04-08' AND hr='12';
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi1.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi1.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi1.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi1.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive a partition twice.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi2.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi2.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi2.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi2.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,12 @@
+set hive.archive.enabled = true;
+-- Tests trying to unarchive a non-archived partition group
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+drop table tstsrcpart;
+create table tstsrcpart like srcpart;
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+insert overwrite table tstsrcpart partition (ds='2008-04-08', hr='12')
+select key, value from srcpart where ds='2008-04-08' and hr='12';
+
+ALTER TABLE tstsrcpart UNARCHIVE PARTITION (ds='2008-04-08', hr='12');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi3.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi3.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi3.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi3.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive outer partition group containing other partition inside.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08', hr='12');
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi4.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi4.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi4.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi4.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive inner partition contained in archived partition group.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08', hr='12');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi5.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi5.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi5.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi5.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to unarchive outer partition group containing other partition inside.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08', hr='12');
+ALTER TABLE tstsrcpart UNARCHIVE PARTITION (ds='2008-04-08');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi6.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi6.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi6.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi6.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,13 @@
+set hive.archive.enabled = true;
+-- Tests trying to unarchive inner partition contained in archived partition group.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
+ALTER TABLE tstsrcpart UNARCHIVE PARTITION (ds='2008-04-08', hr='12');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_multi7.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_multi7.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_multi7.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_multi7.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,12 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive a partition group with custom locations.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE tstsrcpart LIKE srcpart;
+
+INSERT OVERWRITE TABLE tstsrcpart PARTITION (ds='2008-04-08', hr='11')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='11';
+ALTER TABLE tstsrcpart ADD PARTITION (ds='2008-04-08', hr='12')
+LOCATION "/tmp/tstsrc";
+
+ALTER TABLE tstsrcpart ARCHIVE PARTITION (ds='2008-04-08');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_partspec1.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_partspec1.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_partspec1.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_partspec1.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,10 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive a partition twice.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE srcpart_archived LIKE srcpart;
+
+INSERT OVERWRITE TABLE srcpart_archived PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE srcpart_archived ARCHIVE PARTITION (ds='2008-04-08', nonexistingpart='12');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_partspec2.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_partspec2.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_partspec2.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_partspec2.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,10 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive a partition twice.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE srcpart_archived LIKE srcpart;
+
+INSERT OVERWRITE TABLE srcpart_archived PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE srcpart_archived ARCHIVE PARTITION (hr='12');
Added: hive/trunk/ql/src/test/queries/clientnegative/archive_partspec3.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/archive_partspec3.q?rev=1197400&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/archive_partspec3.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/archive_partspec3.q Fri Nov 4 04:15:36 2011
@@ -0,0 +1,10 @@
+set hive.archive.enabled = true;
+-- Tests trying to archive a partition twice.
+-- EXCLUDE_HADOOP_MAJOR_VERSIONS(0.17, 0.18, 0.19)
+
+CREATE TABLE srcpart_archived LIKE srcpart;
+
+INSERT OVERWRITE TABLE srcpart_archived PARTITION (ds='2008-04-08', hr='12')
+SELECT key, value FROM srcpart WHERE ds='2008-04-08' AND hr='12';
+
+ALTER TABLE srcpart_archived ARCHIVE PARTITION ();