This is an automated email from the ASF dual-hosted git repository.
lpinter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 5a21e94 HIVE-22862: Remove unnecessary calls to isEnoughToCompact
(Karen Coppage via Laszlo Pinter)
5a21e94 is described below
commit 5a21e946b1adbca25859e3c57516ba691162ae4c
Author: Karen Coppage <karen.copp...@cloudera.com>
AuthorDate: Mon Feb 24 11:16:13 2020 +0100
HIVE-22862: Remove unnecessary calls to isEnoughToCompact (Karen Coppage
via Laszlo Pinter)
---
.../hadoop/hive/ql/txn/compactor/CompactorMR.java | 3 --
.../hive/ql/txn/compactor/MinorQueryCompactor.java | 3 --
.../ql/txn/compactor/MmMajorQueryCompactor.java | 4 --
.../hive/ql/txn/compactor/QueryCompactor.java | 57 --------------------
.../hadoop/hive/ql/txn/compactor/Worker.java | 63 +++++++++++++++++++++-
5 files changed, 61 insertions(+), 69 deletions(-)
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
index b4bc3c7..25c14e0 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
@@ -231,9 +231,6 @@ public class CompactorMR {
}
JobConf job = createBaseJobConf(conf, jobName, t, sd, writeIds, ci);
- if (!QueryCompactor.Util.isEnoughToCompact(ci.isMajorCompaction(), dir,
sd)) {
- return;
- }
List<AcidUtils.ParsedDelta> parsedDeltas = dir.getCurrentDirectories();
int maxDeltasToHandle =
conf.getIntVar(HiveConf.ConfVars.COMPACTOR_MAX_NUM_DELTA);
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MinorQueryCompactor.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MinorQueryCompactor.java
index f96a048..59dcf2c 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MinorQueryCompactor.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MinorQueryCompactor.java
@@ -56,9 +56,6 @@ final class MinorQueryCompactor extends QueryCompactor {
AcidUtils.Directory dir = AcidUtils
.getAcidState(null, new Path(storageDescriptor.getLocation()),
hiveConf, writeIds, Ref.from(false), false,
table.getParameters(), false);
- if (!Util.isEnoughToCompact(compactionInfo.isMajorCompaction(), dir,
storageDescriptor)) {
- return;
- }
// Set up the session for driver.
HiveConf conf = new HiveConf(hiveConf);
conf.set(HiveConf.ConfVars.HIVE_QUOTEDID_SUPPORT.varname, "column");
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MmMajorQueryCompactor.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MmMajorQueryCompactor.java
index 48387c9..41fdd7e 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MmMajorQueryCompactor.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/MmMajorQueryCompactor.java
@@ -54,10 +54,6 @@ final class MmMajorQueryCompactor extends QueryCompactor {
table.getParameters(), false);
MmQueryCompactorUtils.removeFilesForMmTable(hiveConf, dir);
- if (!Util.isEnoughToCompact(compactionInfo.isMajorCompaction(), dir,
storageDescriptor)) {
- return;
- }
-
String tmpLocation = Util.generateTmpPath(storageDescriptor);
Path baseLocation = new Path(tmpLocation, "_base");
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
index d234910..3ce4dde 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/QueryCompactor.java
@@ -40,10 +40,8 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.UUID;
-import java.util.stream.Collectors;
/**
* Common interface for query based compactions.
@@ -144,61 +142,6 @@ abstract class QueryCompactor {
*/
static class Util {
/**
- * Determine if compaction can run in a specified directory.
- * @param isMajorCompaction type of compaction.
- * @param dir the delta directory
- * @param sd resolved storage descriptor
- * @return true, if compaction can run.
- */
- static boolean isEnoughToCompact(boolean isMajorCompaction,
AcidUtils.Directory dir, StorageDescriptor sd) {
- int deltaCount = dir.getCurrentDirectories().size();
- int origCount = dir.getOriginalFiles().size();
-
- StringBuilder deltaInfo = new StringBuilder().append(deltaCount);
- boolean isEnoughToCompact;
-
- if (isMajorCompaction) {
- isEnoughToCompact = (origCount > 0 || deltaCount +
(dir.getBaseDirectory() == null ? 0 : 1) > 1);
-
- } else {
- isEnoughToCompact = (deltaCount > 1);
-
- if (deltaCount == 2) {
- Map<String, Long> deltaByType =
dir.getCurrentDirectories().stream().collect(Collectors
- .groupingBy(delta -> (delta.isDeleteDelta() ?
AcidUtils.DELETE_DELTA_PREFIX : AcidUtils.DELTA_PREFIX),
- Collectors.counting()));
-
- isEnoughToCompact = (deltaByType.size() != deltaCount);
- deltaInfo.append(" ").append(deltaByType);
- }
- }
-
- if (!isEnoughToCompact) {
- LOG.debug("Not compacting {}; current base: {}, delta files: {},
originals: {}", sd.getLocation(),
- dir.getBaseDirectory(), deltaInfo, origCount);
- }
- return isEnoughToCompact;
- }
-
- /**
- * Check for obsolete directories, and return true if any exist and
Cleaner should be
- * run. For example if we insert overwrite into a table with only deltas,
a new base file with
- * the highest writeId is created so there will be no live delta
directories, only obsolete
- * ones. Compaction is not needed, but the cleaner should still be run.
- *
- * @return true if cleaning is needed
- */
- public static boolean needsCleaning(AcidUtils.Directory dir,
StorageDescriptor sd) {
- int numObsoleteDirs = dir.getObsolete().size();
- boolean needsJustCleaning = numObsoleteDirs > 0;
- if (needsJustCleaning) {
- LOG.debug("{} obsolete directories in {} found; marked for cleaning.",
- numObsoleteDirs, sd.getLocation());
- }
- return needsJustCleaning;
- }
-
- /**
* Generate a random tmp path, under the provided storage.
* @param sd storage descriptor, must be not null.
* @return path, always not null
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
index 383969a..90c699a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
@@ -58,6 +58,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
/**
* A class to do compactions. This will run in a separate thread. It will
spin on the
@@ -188,8 +189,8 @@ public class Worker extends RemoteCompactorThread
implements MetaStoreThread {
// Don't start compaction or cleaning if not necessary
AcidUtils.Directory dir = AcidUtils.getAcidState(null, new
Path(sd.getLocation()), conf,
tblValidWriteIds, Ref.from(false), true, null, false);
- if (!QueryCompactor.Util.isEnoughToCompact(ci.isMajorCompaction(),
dir, sd)) {
- if (QueryCompactor.Util.needsCleaning(dir, sd)) {
+ if (!isEnoughToCompact(ci.isMajorCompaction(), dir, sd)) {
+ if (needsCleaning(dir, sd)) {
msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci));
} else {
// do nothing
@@ -436,4 +437,62 @@ public class Worker extends RemoteCompactorThread
implements MetaStoreThread {
}
}
}
+
+ /**
+ * Determine if compaction can run in a specified directory.
+ * @param isMajorCompaction type of compaction.
+ * @param dir the delta directory
+ * @param sd resolved storage descriptor
+ * @return true, if compaction can run.
+ */
+ static boolean isEnoughToCompact(boolean isMajorCompaction,
AcidUtils.Directory dir,
+ StorageDescriptor sd) {
+ int deltaCount = dir.getCurrentDirectories().size();
+ int origCount = dir.getOriginalFiles().size();
+
+ StringBuilder deltaInfo = new StringBuilder().append(deltaCount);
+ boolean isEnoughToCompact;
+
+ if (isMajorCompaction) {
+ isEnoughToCompact =
+ (origCount > 0 || deltaCount + (dir.getBaseDirectory() == null ? 0 :
1) > 1);
+
+ } else {
+ isEnoughToCompact = (deltaCount > 1);
+
+ if (deltaCount == 2) {
+ Map<String, Long> deltaByType =
dir.getCurrentDirectories().stream().collect(Collectors
+ .groupingBy(delta -> (delta
+ .isDeleteDelta() ? AcidUtils.DELETE_DELTA_PREFIX :
AcidUtils.DELTA_PREFIX),
+ Collectors.counting()));
+
+ isEnoughToCompact = (deltaByType.size() != deltaCount);
+ deltaInfo.append(" ").append(deltaByType);
+ }
+ }
+
+ if (!isEnoughToCompact) {
+ LOG.debug("Not compacting {}; current base: {}, delta files: {},
originals: {}",
+ sd.getLocation(), dir.getBaseDirectory(), deltaInfo, origCount);
+ }
+ return isEnoughToCompact;
+ }
+
+ /**
+ * Check for obsolete directories, and return true if any exist and Cleaner
should be
+ * run. For example if we insert overwrite into a table with only deltas, a
new base file with
+ * the highest writeId is created so there will be no live delta
directories, only obsolete
+ * ones. Compaction is not needed, but the cleaner should still be run.
+ *
+ * @return true if cleaning is needed
+ */
+ public static boolean needsCleaning(AcidUtils.Directory dir,
StorageDescriptor sd) {
+ int numObsoleteDirs = dir.getObsolete().size();
+ boolean needsJustCleaning = numObsoleteDirs > 0;
+ if (needsJustCleaning) {
+ LOG.debug("{} obsolete directories in {} found; marked for cleaning.",
numObsoleteDirs,
+ sd.getLocation());
+ }
+ return needsJustCleaning;
+ }
}
