HIVE-23324: Parallelise compaction directory cleaning process #1275

adesh-rao · 2020-07-17T17:09:09Z

NOTICE

Please create an issue in ASF JIRA before opening a pull request,
and you need to set the title of the pull request which starts with
the corresponding JIRA issue number. (e.g. HIVE-XXXXX: Fix a typo in YYY)
For more details, please see https://cwiki.apache.org/confluence/display/Hive/HowToContribute

adesh-rao · 2020-07-21T09:21:24Z

@pvary Can you please take a look at this PR?

pvary · 2020-07-21T09:42:47Z

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java


  private ReplChangeManager replChangeManager;

  @Override
  public void init(AtomicBoolean stop) throws Exception {
    super.init(stop);
    replChangeManager = ReplChangeManager.getInstance(conf);
+    cleanerExecutor = Executors.newFixedThreadPool(conf.getIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_REQUEST_QUEUE));


I think it would be a good idea to shut down the executor when we finished the run loop.
What do you think?

Yeah, makes sense. Added it.

I prefer the way how it was done in Worker.java:

Named threads

Priority down

Set daemon on/off

Stopped in finally

Shall we do it here, or should we create a follow-up jira for creating and cleaning up the executor threads for Cleaner and Initiator as well?

Thanks,
Peter

Just saw the approach in Worker.java, it looks much better/cleaner. I will make the changes here only ()for both cleaner/intiator) and update once it is ready.

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java

belugabehr · 2020-07-21T14:43:40Z

Let us not re-invent the wheel here:

  public static void main(String[] args) throws IOException
  {

    ExecutorService service = Executors.newFixedThreadPool(2);
    CompletionService<String> completionService = new ExecutorCompletionService<>(service);

    for (int i = 0; i < 10; i++)
    {
      final int t = i;
      completionService.submit(() -> {
        return clean("File" + t);
      });
    }

    for (int i = 0; i < 10; ++i) {
      try {
          String path = completionService.take().get();
          System.out.println(path);
      } catch (Exception ignore) {}
  }
  }

  public static String clean(String path)
  {
    return "Cleaned: " + path;
  }

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorCompletionService.html

adesh-rao · 2020-07-21T17:36:58Z

@belugabehr

clean method does not return anything (void). I couldn't find a way to create a CompletionService on void return type. This means that to use CompletionService, clean method's definition need to be changed.
we don't care about which executor threads are getting completed when, and also, all the related logging about directory/files deletion are being done in clean method only.

Therefore,
`
for (int i = 0; i < 10; ++i) {

  try {

      String path = completionService.take().get();

      System.out.println(path);

  } catch (Exception ignore) {}

}`

the above code will be extra lines of code when we can just use join() instead (I think that might have been the initial intention too).

So, the suggestion that you have provided, if necessary, we can take it up in a separate jira (This will require similar changes in Initiator thread too).

belugabehr · 2020-07-21T20:14:14Z

Void type:

  public static void main(String[] args) throws IOException
  {

    ExecutorService service = Executors.newFixedThreadPool(2);
    CompletionService<Void> completionService = new ExecutorCompletionService<>(service);

    for (int i = 0; i < 10; i++)
    {
      final int t = i;
      completionService.submit(() -> {
        clean("File" + t);
        return null;
      });
    }

    for (int i = 0; i < 10; ++i) {
      try {
          completionService.take().get();
      } catch (Exception ignore) {}
  }
  }

  public static String clean(String path)
  {
    return "Cleaned: " + path;
  }

adesh-rao · 2020-07-22T06:15:08Z

@belugabehr thanks for suggestion. Implemented this.

pvary · 2020-07-22T08:03:21Z

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java

+            try {
+              completionService.take().get();
+            } catch (InterruptedException| ExecutionException ignore) {
+              // What should we do here?


Minimally log the error on info level?

pvary · 2020-07-22T08:03:52Z

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java

+            count++;
+          }
+
+          for(int i=0; i<count; i++) {


nit of the nit: formatting:
for (int i = 0; i < count; i++) {

pvary · 2020-07-22T08:05:23Z

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java


  private ReplChangeManager replChangeManager;

  @Override
  public void init(AtomicBoolean stop) throws Exception {
    super.init(stop);
    replChangeManager = ReplChangeManager.getInstance(conf);
+    ThreadFactory threadFactory = new ThreadFactoryBuilder()


Shall we move this to the run method too? It would make it easier to understand the code IMHO

Or maybe use this as an utility method to create the executor like and use this in the run method to create the executor:
WhateverUtil.createExecutor(String name, int size) {}

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java

adesh-rao · 2020-07-29T06:25:17Z

@pvary Can you please take one more look at this? There were few modifications after your last comment.

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java

sankarh · 2020-07-28T08:34:40Z

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorUtil.java

+import java.util.concurrent.ThreadFactory;
+
+public class CompactorUtil {
+    public interface ThrowingRunnable<E extends Exception> {


nit: Make it 2 space tabs.

sankarh · 2020-07-28T08:49:06Z

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -3028,6 +3028,9 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal

    HIVE_COMPACTOR_CLEANER_RUN_INTERVAL("hive.compactor.cleaner.run.interval", "5000ms",
        new TimeValidator(TimeUnit.MILLISECONDS), "Time between runs of the cleaner thread"),
+    HIVE_COMPACTOR_CLEANER_REQUEST_QUEUE("hive.compactor.cleaner.request.queue", 1,


The config name can be relevant. It actually represent how many threads that we use for parallelly run the cleaner. But, the name sounds like Queue name. Can we change it to "HIVE_COMPACTOR_CLEANER_THREADS_NUM"?

sankarh · 2020-07-29T08:14:37Z

ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java

+    Table t = newTable("default", "camipc", true);
+    List<Partition> partitions = new ArrayList<>();
+    Partition p = null;
+    for(int i=0; i<10; i++) {


nit: Make it "for (int i = 0; i < 10; i++)". Check other places in this patch.

sankarh · 2020-07-29T08:16:25Z

ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java

+      addDeltaFile(t, p, 21L, 24L, 4);
+      partitions.add(p);
+    }
+    burnThroughTransactions("default", "camipc", 25);


nit: Need a blank line after closing braces,

sankarh · 2020-07-29T08:17:36Z

ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java

+      Assert.assertEquals(2, paths.size());
+      boolean sawBase = false, sawDelta = false;
+      for (Path path : paths) {
+        if (path.getName().equals("base_20")) sawBase = true;


nit: Keep the code block of "if", "else if" and "else" in new line with a tab space.

…cess (Adesh Kumar Rao, reviewed by Peter Vary, Beluga Behr, Sankar Hariappan, Denys Kuzmenko) Closes (apache#1275) (cherry picked from commit a206833) Change-Id: Iffb7d6bcfda0bfa64c7d907d0a5d01cb7b1f9c8e

kgyrtkirk added tests pending tests passed and removed tests pending labels Jul 17, 2020

pvary reviewed Jul 21, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java Outdated Show resolved Hide resolved

kgyrtkirk added tests pending tests passed and removed tests passed tests pending labels Jul 21, 2020

belugabehr reviewed Jul 21, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java Outdated Show resolved Hide resolved

adesh-rao force-pushed the HIVE-23324 branch from ce1780f to 8227680 Compare July 21, 2020 17:04

kgyrtkirk added tests pending and removed tests passed labels Jul 21, 2020

kgyrtkirk added tests passed tests pending and removed tests pending tests passed labels Jul 21, 2020

kgyrtkirk added tests unstable and removed tests pending labels Jul 22, 2020

pvary reviewed Jul 22, 2020

View reviewed changes

deniskuzZ reviewed Jul 22, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java Outdated Show resolved Hide resolved

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java Outdated Show resolved Hide resolved

deniskuzZ reviewed Jul 22, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java Outdated Show resolved Hide resolved

deniskuzZ reviewed Jul 22, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java Outdated Show resolved Hide resolved

adesh-rao added 2 commits July 28, 2020 18:02

Add executor shutdown

6039bbf

Remove redundant check

5f16fd2

kgyrtkirk added tests unstable and removed tests pending labels Jul 28, 2020

adesh-rao force-pushed the HIVE-23324 branch from c6ef5dd to 5f16fd2 Compare July 28, 2020 16:00

kgyrtkirk added tests pending tests passed and removed tests unstable tests pending labels Jul 28, 2020

deniskuzZ requested changes Jul 29, 2020

View reviewed changes

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java Outdated Show resolved Hide resolved

sankarh reviewed Jul 29, 2020

View reviewed changes

Log exception

f3d1dc3

kgyrtkirk added tests pending and removed tests passed labels Jul 29, 2020

fix review comments

423fa2a

kgyrtkirk added tests failed tests pending tests unstable and removed tests pending tests failed labels Jul 29, 2020

Exit if cleaner thread got interrupted

0b61fa6

kgyrtkirk added tests pending and removed tests unstable labels Jul 30, 2020

deniskuzZ approved these changes Jul 30, 2020

View reviewed changes

kgyrtkirk added tests passed and removed tests pending labels Jul 30, 2020

deniskuzZ merged commit a206833 into apache:master Jul 31, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

HIVE-23324: Parallelise compaction directory cleaning process #1275

HIVE-23324: Parallelise compaction directory cleaning process #1275

adesh-rao commented Jul 17, 2020

adesh-rao commented Jul 21, 2020

pvary Jul 21, 2020

adesh-rao Jul 21, 2020

pvary Jul 21, 2020 •

edited

adesh-rao Jul 21, 2020

adesh-rao Jul 21, 2020

belugabehr commented Jul 21, 2020

adesh-rao commented Jul 21, 2020

belugabehr commented Jul 21, 2020 •

edited

adesh-rao commented Jul 22, 2020

pvary Jul 22, 2020

pvary Jul 22, 2020

pvary Jul 22, 2020

pvary Jul 22, 2020 •

edited

adesh-rao Jul 22, 2020

adesh-rao commented Jul 29, 2020

sankarh Jul 28, 2020

sankarh Jul 28, 2020

sankarh Jul 29, 2020

sankarh Jul 29, 2020

adesh-rao Jul 29, 2020

sankarh Jul 29, 2020

HIVE-23324: Parallelise compaction directory cleaning process #1275

HIVE-23324: Parallelise compaction directory cleaning process #1275

Conversation

adesh-rao commented Jul 17, 2020

NOTICE

adesh-rao commented Jul 21, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvary Jul 21, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

belugabehr commented Jul 21, 2020

adesh-rao commented Jul 21, 2020

belugabehr commented Jul 21, 2020 • edited

adesh-rao commented Jul 22, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvary Jul 22, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

adesh-rao commented Jul 29, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvary Jul 21, 2020 •

edited

belugabehr commented Jul 21, 2020 •

edited

pvary Jul 22, 2020 •

edited