Make RaptorPageSink#finish() truly asynchronous

prestodb · Nov 1, 2016 · a9f1b8d · a9f1b8d
1 parent 521a9cd
commit a9f1b8d
Show file tree

Hide file tree

Showing 6 changed files with 36 additions and 31 deletions.
diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorPageSink.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorPageSink.java
@@ -35,7 +35,6 @@
 import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
@@ -47,8 +46,8 @@
 import static com.facebook.presto.spi.type.DateType.DATE;
 import static com.facebook.presto.spi.type.TimestampType.TIMESTAMP;
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.airlift.concurrent.MoreFutures.allAsList;
 import static java.util.Objects.requireNonNull;
-import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.stream.Collectors.toList;
 
@@ -133,18 +132,17 @@ public CompletableFuture<?> appendPage(Page page)
     @Override
     public CompletableFuture<Collection<Slice>> finish()
     {
-        List<ShardInfo> shards = new ArrayList<>();
+        List<CompletableFuture<? extends List<Slice>>> futureSlices = pageWriter.getPageBuffers().stream().map(pageBuffer -> {
-        for (PageBuffer pageBuffer : pageWriter.getPageBuffers()) {
             pageBuffer.flush();
-            shards.addAll(pageBuffer.getStoragePageSink().commit());
+            CompletableFuture<List<ShardInfo>> futureShards = pageBuffer.getStoragePageSink().commit();
-        }
+            return futureShards.thenApply(shards -> shards.stream()
-
+                    .map(shard -> Slices.wrappedBuffer(shardInfoCodec.toJsonBytes(shard)))
-        ImmutableList.Builder<Slice> fragments = ImmutableList.builder();
+                    .collect(toList()));
-        for (ShardInfo shard : shards) {
+        }).collect(toList());
-            fragments.add(Slices.wrappedBuffer(shardInfoCodec.toJsonBytes(shard)));
+
-        }
+        return allAsList(futureSlices).thenApply(lists -> lists.stream()
-        // TODO: process asynchronously
+                .flatMap(Collection::stream)
-        return completedFuture(fragments.build());
+                .collect(toList()));
     }
 
     @Override

diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/OrcStorageManager.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/OrcStorageManager.java
@@ -50,7 +50,6 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import io.airlift.concurrent.MoreFutures;
 import io.airlift.json.JsonCodec;
 import io.airlift.slice.Slice;
 import io.airlift.slice.Slices;
@@ -103,13 +102,15 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Throwables.propagateIfInstanceOf;
+import static io.airlift.concurrent.MoreFutures.allAsList;
 import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
 import static io.airlift.json.JsonCodec.jsonCodec;
 import static java.lang.Math.min;
 import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.CompletableFuture.supplyAsync;
+import static java.util.concurrent.Executors.newCachedThreadPool;
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.stream.Collectors.toList;
 import static org.joda.time.DateTimeZone.UTC;
@@ -134,6 +135,7 @@ public class OrcStorageManager
     private final DataSize minAvailableSpace;
     private final TypeManager typeManager;
     private final ExecutorService deletionExecutor;
+    private final ExecutorService commitExecutor;
 
     @Inject
     public OrcStorageManager(
@@ -200,12 +202,14 @@ public OrcStorageManager(
         this.shardRecorder = requireNonNull(shardRecorder, "shardRecorder is null");
         this.typeManager = requireNonNull(typeManager, "typeManager is null");
         this.deletionExecutor = newFixedThreadPool(deletionThreads, daemonThreadsNamed("raptor-delete-" + connectorId + "-%s"));
+        this.commitExecutor = newCachedThreadPool(daemonThreadsNamed("raptor-commit-" + connectorId + "-%s"));
     }
 
     @PreDestroy
     public void shutdown()
     {
         deletionExecutor.shutdownNow();
+        commitExecutor.shutdown();
     }
 
     @Override
@@ -593,20 +597,19 @@ public void flush()
         }
 
         @Override
-        public List<ShardInfo> commit()
+        public CompletableFuture<List<ShardInfo>> commit()
         {
             checkState(!committed, "already committed");
             committed = true;
 
             flush();
 
-            // backup jobs depend on the staging files, so wait until all backups have finished
+            return allAsList(futures).thenApplyAsync(ignored -> {
-            futures.forEach(MoreFutures::getFutureValue);
+                for (ShardInfo shard : shards) {
-
+                    writeShard(shard.getShardUuid());
-            for (ShardInfo shard : shards) {
+                }
-                writeShard(shard.getShardUuid());
+                return ImmutableList.copyOf(shards);
-            }
+            }, commitExecutor);
-            return ImmutableList.copyOf(shards);
         }
 
         @SuppressWarnings("ResultOfMethodCallIgnored")

diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/StoragePageSink.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/StoragePageSink.java
@@ -17,6 +17,7 @@
 import com.facebook.presto.spi.Page;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 public interface StoragePageSink
 {
@@ -30,7 +31,7 @@ public interface StoragePageSink
 
     void flush();
 
-    List<ShardInfo> commit();
+    CompletableFuture<List<ShardInfo>> commit();
 
     void rollback();
 }
diff --git a/...-raptor/src/main/java/com/facebook/presto/raptor/storage/organization/ShardCompactor.java b/...-raptor/src/main/java/com/facebook/presto/raptor/storage/organization/ShardCompactor.java
@@ -45,6 +45,7 @@
 
 import static com.facebook.presto.raptor.storage.Row.extractRow;
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.units.Duration.nanosSince;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;
@@ -107,7 +108,7 @@ private List<ShardInfo> compact(StoragePageSink storagePageSink, OptionalInt buc
                 }
             }
         }
-        return storagePageSink.commit();
+        return getFutureValue(storagePageSink.commit());
     }
 
     public List<ShardInfo> compactSorted(long transactionId, OptionalInt bucketNumber, Set<UUID> uuids, List<ColumnInfo> columns, List<Long> sortColumnIds, List<SortOrder> sortOrders)
@@ -151,7 +152,7 @@ public List<ShardInfo> compactSorted(long transactionId, OptionalInt bucketNumbe
                 rowSources.add(rowSource);
             }
             outputPageSink.flush();
-            List<ShardInfo> shardInfos = outputPageSink.commit();
+            List<ShardInfo> shardInfos = getFutureValue(outputPageSink.commit());
 
             updateStats(uuids.size(), shardInfos.size(), nanosSince(start).toMillis());
 

diff --git a/presto-raptor/src/test/java/com/facebook/presto/raptor/storage/TestOrcStorageManager.java b/presto-raptor/src/test/java/com/facebook/presto/raptor/storage/TestOrcStorageManager.java
@@ -84,6 +84,7 @@
 import static com.google.common.hash.Hashing.md5;
 import static com.google.common.io.Files.createTempDir;
 import static com.google.common.io.Files.hash;
+import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.json.JsonCodec.jsonCodec;
 import static io.airlift.slice.Slices.utf8Slice;
 import static io.airlift.slice.Slices.wrappedBuffer;
@@ -183,7 +184,7 @@ public void testWriter()
         List<RecordedShard> recordedShards = shardRecorder.getShards();
         assertEquals(recordedShards.size(), 1);
 
-        List<ShardInfo> shards = sink.commit();
+        List<ShardInfo> shards = getFutureValue(sink.commit());
 
         assertEquals(shards.size(), 1);
         ShardInfo shardInfo = Iterables.getOnlyElement(shards);
@@ -262,7 +263,7 @@ public void testReader()
                 .build();
 
         sink.appendPages(pages);
-        List<ShardInfo> shards = sink.commit();
+        List<ShardInfo> shards = getFutureValue(sink.commit());
 
         assertEquals(shards.size(), 1);
         UUID uuid = Iterables.getOnlyElement(shards).getShardUuid();
@@ -321,7 +322,7 @@ public void testRewriter()
                 .row(456L, "bye")
                 .build();
         sink.appendPages(pages);
-        List<ShardInfo> shards = sink.commit();
+        List<ShardInfo> shards = getFutureValue(sink.commit());
 
         assertEquals(shardRecorder.getShards().size(), 1);
 
@@ -658,7 +659,7 @@ private List<ColumnStats> columnStats(List<Type> columnTypes, Object[]... rows)
         OrcStorageManager manager = createOrcStorageManager();
         StoragePageSink sink = createStoragePageSink(manager, columnIds, columnTypes);
         sink.appendPages(rowPagesBuilder(columnTypes).rows(rows).build());
-        List<ShardInfo> shards = sink.commit();
+        List<ShardInfo> shards = getFutureValue(sink.commit());
 
         assertEquals(shards.size(), 1);
         return Iterables.getOnlyElement(shards).getColumnStats();

diff --git a/...tor/src/test/java/com/facebook/presto/raptor/storage/organization/TestShardCompactor.java b/...tor/src/test/java/com/facebook/presto/raptor/storage/organization/TestShardCompactor.java
@@ -57,6 +57,7 @@
 import static com.facebook.presto.testing.TestingConnectorSession.SESSION;
 import static com.facebook.presto.tests.QueryAssertions.assertEqualsIgnoreOrder;
 import static com.google.common.io.Files.createTempDir;
+import static io.airlift.concurrent.MoreFutures.getFutureValue;
 import static io.airlift.testing.FileUtils.deleteRecursively;
 import static io.airlift.units.DataSize.Unit.MEGABYTE;
 import static java.util.Collections.nCopies;
@@ -264,7 +265,7 @@ private static List<ShardInfo> createSortedShards(StorageManager storageManager,
         for (int shardNum = 0; shardNum < shardCount; shardNum++) {
             createSortedShard(columnTypes, sortChannels, sortOrders, sink);
         }
-        return sink.commit();
+        return getFutureValue(sink.commit());
     }
 
     private static void createSortedShard(List<Type> columnTypes, List<Integer> sortChannels, List<SortOrder> sortOrders, StoragePageSink sink)
@@ -293,7 +294,7 @@ private static List<ShardInfo> createShards(StorageManager storageManager, List<
             sink.appendPages(createPages(columnTypes));
             sink.flush();
         }
-        return sink.commit();
+        return getFutureValue(sink.commit());
     }
 
     private static StoragePageSink createStoragePageSink(StorageManager manager, List<Long> columnIds, List<Type> columnTypes)