You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by am...@apache.org on 2017/02/13 03:50:40 UTC
[1/3] drill git commit: DRILL-5080: Memory-managed version of
external sort
Repository: drill
Updated Branches:
refs/heads/master c9a6ac4fc -> 300e9349a
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierTemplate.java
new file mode 100644
index 0000000..81856fa
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopierTemplate.java
@@ -0,0 +1,170 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import io.netty.buffer.DrillBuf;
+
+import java.io.IOException;
+import java.util.List;
+
+import javax.inject.Named;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.vector.AllocationHelper;
+
+public abstract class PriorityQueueCopierTemplate implements PriorityQueueCopier {
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PriorityQueueCopierTemplate.class);
+
+ private SelectionVector4 vector4;
+ private List<BatchGroup> batchGroups;
+ private VectorAccessible hyperBatch;
+ private VectorAccessible outgoing;
+ private int size;
+ private int queueSize = 0;
+
+ @Override
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch, List<BatchGroup> batchGroups,
+ VectorAccessible outgoing) throws SchemaChangeException {
+ this.hyperBatch = hyperBatch;
+ this.batchGroups = batchGroups;
+ this.outgoing = outgoing;
+ this.size = batchGroups.size();
+
+ @SuppressWarnings("resource")
+ final DrillBuf drillBuf = allocator.buffer(4 * size);
+ vector4 = new SelectionVector4(drillBuf, size, Character.MAX_VALUE);
+ doSetup(context, hyperBatch, outgoing);
+
+ queueSize = 0;
+ for (int i = 0; i < size; i++) {
+ vector4.set(i, i, batchGroups.get(i).getNextIndex());
+ siftUp();
+ queueSize++;
+ }
+ }
+
+ @Override
+ public int next(int targetRecordCount) {
+ allocateVectors(targetRecordCount);
+ for (int outgoingIndex = 0; outgoingIndex < targetRecordCount; outgoingIndex++) {
+ if (queueSize == 0) {
+ return 0;
+ }
+ int compoundIndex = vector4.get(0);
+ int batch = compoundIndex >>> 16;
+ assert batch < batchGroups.size() : String.format("batch: %d batchGroups: %d", batch, batchGroups.size());
+ doCopy(compoundIndex, outgoingIndex);
+ int nextIndex = batchGroups.get(batch).getNextIndex();
+ if (nextIndex < 0) {
+ vector4.set(0, vector4.get(--queueSize));
+ } else {
+ vector4.set(0, batch, nextIndex);
+ }
+ if (queueSize == 0) {
+ setValueCount(++outgoingIndex);
+ return outgoingIndex;
+ }
+ siftDown();
+ }
+ setValueCount(targetRecordCount);
+ return targetRecordCount;
+ }
+
+ private void setValueCount(int count) {
+ for (VectorWrapper<?> w: outgoing) {
+ w.getValueVector().getMutator().setValueCount(count);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ vector4.clear();
+ for (final VectorWrapper<?> w: outgoing) {
+ w.getValueVector().clear();
+ }
+ for (final VectorWrapper<?> w : hyperBatch) {
+ w.clear();
+ }
+
+ for (BatchGroup batchGroup : batchGroups) {
+ batchGroup.close();
+ }
+ }
+
+ private void siftUp() {
+ int p = queueSize;
+ while (p > 0) {
+ if (compare(p, (p - 1) / 2) < 0) {
+ swap(p, (p - 1) / 2);
+ p = (p - 1) / 2;
+ } else {
+ break;
+ }
+ }
+ }
+
+ private void allocateVectors(int targetRecordCount) {
+ for (VectorWrapper<?> w: outgoing) {
+ AllocationHelper.allocateNew(w.getValueVector(), targetRecordCount);
+ }
+ }
+
+ private void siftDown() {
+ int p = 0;
+ int next;
+ while (p * 2 + 1 < queueSize) { // While the current node has at least one child
+ if (p * 2 + 2 >= queueSize) { // if current node has only one child, then we only look at it
+ next = p * 2 + 1;
+ } else {
+ if (compare(p * 2 + 1, p * 2 + 2) <= 0) {//if current node has two children, we must first determine which one has higher priority
+ next = p * 2 + 1;
+ } else {
+ next = p * 2 + 2;
+ }
+ }
+ if (compare(p, next) > 0) { // compare current node to highest priority child and swap if necessary
+ swap(p, next);
+ p = next;
+ } else {
+ break;
+ }
+ }
+ }
+
+ public void swap(int sv0, int sv1) {
+ int tmp = vector4.get(sv0);
+ vector4.set(sv0, vector4.get(sv1));
+ vector4.set(sv1, tmp);
+ }
+
+ public int compare(int leftIndex, int rightIndex) {
+ int sv1 = vector4.get(leftIndex);
+ int sv2 = vector4.get(rightIndex);
+ return doEval(sv1, sv2);
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorAccessible incoming, @Named("outgoing") VectorAccessible outgoing);
+ public abstract int doEval(@Named("leftIndex") int leftIndex, @Named("rightIndex") int rightIndex);
+ public abstract void doCopy(@Named("inIndex") int inIndex, @Named("outIndex") int outIndex);
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
index f8b4334..26a77ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -17,6 +17,8 @@
*/
package org.apache.drill.exec.server.options;
+import static com.google.common.base.Preconditions.checkArgument;
+
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
@@ -24,25 +26,22 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
-
import org.apache.commons.collections.IteratorUtils;
import org.apache.drill.common.config.LogicalPlanPersistence;
-import org.apache.drill.common.map.CaseInsensitiveMap;
import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.map.CaseInsensitiveMap;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.compile.ClassCompilerSelector;
import org.apache.drill.exec.compile.ClassTransformer;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.server.options.OptionValue.OptionType;
-import org.apache.drill.exec.server.options.TypeValidators.BooleanValidator;
import org.apache.drill.exec.store.sys.PersistentStore;
import org.apache.drill.exec.store.sys.PersistentStoreConfig;
import org.apache.drill.exec.store.sys.PersistentStoreProvider;
import org.apache.drill.exec.util.AssertionUtil;
-import static com.google.common.base.Preconditions.checkArgument;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
/**
* {@link OptionManager} that holds options within {@link org.apache.drill.exec.server.DrillbitContext}.
@@ -163,7 +162,8 @@ public class SystemOptionManager extends BaseOptionManager implements AutoClosea
ExecConstants.IMPLICIT_FILEPATH_COLUMN_LABEL_VALIDATOR,
ExecConstants.CODE_GEN_EXP_IN_METHOD_SIZE_VALIDATOR,
ExecConstants.CREATE_PREPARE_STATEMENT_TIMEOUT_MILLIS_VALIDATOR,
- ExecConstants.DYNAMIC_UDF_SUPPORT_ENABLED_VALIDATOR
+ ExecConstants.DYNAMIC_UDF_SUPPORT_ENABLED_VALIDATOR,
+ ExecConstants.EXTERNAL_SORT_DISABLE_MANAGED_OPTION
};
final Map<String, OptionValidator> tmp = new HashMap<>();
for (final OptionValidator validator : validators) {
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
index 678167f..d06424e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/MemoryAllocationUtilities.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -68,7 +68,11 @@ public class MemoryAllocationUtilities {
logger.debug("Max sort alloc: {}", maxSortAlloc);
for(final ExternalSort externalSort : sortList) {
- externalSort.setMaxAllocation(maxSortAlloc);
+ // Ensure that the sort receives the minimum memory needed to make progress.
+ // Without this, the math might work out to allocate too little memory.
+
+ long alloc = Math.max(maxSortAlloc, externalSort.getInitialAllocation());
+ externalSort.setMaxAllocation(alloc);
}
}
plan.getProperties().hasResourcePlan = true;
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/resources/drill-module.conf
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 735ba2f..1c702d7 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -103,10 +103,10 @@ drill.exec: {
}
},
zk: {
- connect: "localhost:2181",
- root: "drill",
- refresh: 500,
- timeout: 5000,
+ connect: "localhost:2181",
+ root: "drill",
+ refresh: 500,
+ timeout: 5000,
retry: {
count: 7200,
delay: 500
@@ -177,13 +177,47 @@ drill.exec: {
sort: {
purge.threshold : 1000,
external: {
- batch.size : 4000,
+ // Drill uses the managed External Sort Batch by default.
+ // Set this to true to use the legacy, unmanaged version.
+ // Disabled in the intial commit, to be enabled after
+ // tests are committed.
+ disable_managed: true
+ // Limit on the number of batches buffered in memory.
+ // Primarily for testing.
+ // 0 = unlimited
+ batch_limit: 0
+ // Limit on the amount of memory used for xsort. Overrides the
+ // value provided by Foreman. Primarily for testing.
+ // 0 = unlimited, Supports HOCON memory suffixes.
+ mem_limit: 0
+ // Limit on the number of spilled batches that can be merged in
+ // a single pass. Limits the number of open file handles.
+ // 0 = unlimited
+ merge_limit: 0
spill: {
- batch.size : 4000,
- group.size : 40000,
- threshold : 40000,
- directories : [ "/tmp/drill/spill" ],
- fs : "file:///"
+ // Deprecated for managed xsort; used only by legacy xsort
+ group.size: 40000,
+ // Deprecated for managed xsort; used only by legacy xsort
+ threshold: 40000,
+ // File system to use. Local file system by default.
+ fs: "file:///"
+ // List of directories to use. Directories are created
+ // if they do not exist.
+ directories: [ "/tmp/drill/spill" ],
+ // Size of the batches written to, and read from, the spill files.
+ // Determines the ratio of memory to input data size for a single-
+ // generation sort. Smaller values give larger ratios, but at a
+ // (high) cost of much greater disk seek times.
+ spill_batch_size = 8M,
+ // Preferred file size for "first-generation" spill files.
+ // Set large enough to get long, continuous writes, but not so
+ // large as to overwhelm a temp directory.
+ // Supports HOCON memory suffixes.
+ file_size: 256M,
+ // Size of the batch sent downstream from the sort operator during
+ // the merge phase. Don't change this unless you know what you are doing,
+ // larger sizes can result in memory fragmentation.
+ merge_batch_size = 16M
}
}
},
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/TestExternalSort.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/TestExternalSort.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/TestExternalSort.java
index 3e55d9d..52ebd57 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/TestExternalSort.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/xsort/TestExternalSort.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -19,6 +19,8 @@ package org.apache.drill.exec.physical.impl.xsort;
import org.apache.drill.BaseTestQuery;
import org.apache.drill.TestBuilder;
+import org.apache.drill.exec.ExecConstants;
+import org.junit.Ignore;
import org.junit.Test;
import java.io.BufferedOutputStream;
@@ -28,28 +30,37 @@ import java.io.FileOutputStream;
public class TestExternalSort extends BaseTestQuery {
@Test
- public void testNumericTypes() throws Exception {
+ public void testNumericTypesManaged() throws Exception {
+ testNumericTypes( false );
+ }
+
+ @Test
+ public void testNumericTypesLegacy() throws Exception {
+ testNumericTypes( true );
+ }
+
+ private void testNumericTypes(boolean testLegacy) throws Exception {
final int record_count = 10000;
String dfs_temp = getDfsTestTmpSchemaLocation();
System.out.println(dfs_temp);
File table_dir = new File(dfs_temp, "numericTypes");
table_dir.mkdir();
- BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")));
- String format = "{ a : %d }%n";
- for (int i = 0; i <= record_count; i += 2) {
- os.write(String.format(format, i).getBytes());
+ try(BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")))) {
+ String format = "{ a : %d }%n";
+ for (int i = 0; i <= record_count; i += 2) {
+ os.write(String.format(format, i).getBytes());
+ }
}
- os.close();
- os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")));
- format = "{ a : %.2f }%n";
- for (int i = 1; i <= record_count; i+=2) {
- os.write(String.format(format, (float) i).getBytes());
+ try(BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")))) {
+ String format = "{ a : %.2f }%n";
+ for (int i = 1; i <= record_count; i+=2) {
+ os.write(String.format(format, (float) i).getBytes());
+ }
}
- os.close();
String query = "select * from dfs_test.tmp.numericTypes order by a desc";
TestBuilder builder = testBuilder()
.sqlQuery(query)
- .optionSettingQueriesForTestQuery("alter session set `exec.enable_union_type` = true")
+ .optionSettingQueriesForTestQuery(getOptions(testLegacy))
.ordered()
.baselineColumns("a");
for (int i = record_count; i >= 0;) {
@@ -61,30 +72,48 @@ public class TestExternalSort extends BaseTestQuery {
builder.go();
}
+ private String getOptions(boolean testLegacy) {
+ String options = "alter session set `exec.enable_union_type` = true";
+ options += ";alter session set `" + ExecConstants.EXTERNAL_SORT_DISABLE_MANAGED_OPTION.getOptionName() + "` = " +
+ Boolean.toString(testLegacy);
+ return options;
+ }
+
+ @Test
+ @Ignore("Schema changes are disabled in external sort")
+ public void testNumericAndStringTypesManaged() throws Exception {
+ testNumericAndStringTypes(false);
+ }
+
@Test
- public void testNumericAndStringTypes() throws Exception {
+ @Ignore("Schema changes are disabled in external sort")
+ public void testNumericAndStringTypesLegacy() throws Exception {
+ testNumericAndStringTypes(true);
+ }
+
+ private void testNumericAndStringTypes(boolean testLegacy) throws Exception {
final int record_count = 10000;
String dfs_temp = getDfsTestTmpSchemaLocation();
System.out.println(dfs_temp);
File table_dir = new File(dfs_temp, "numericAndStringTypes");
table_dir.mkdir();
- BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")));
- String format = "{ a : %d }%n";
- for (int i = 0; i <= record_count; i += 2) {
- os.write(String.format(format, i).getBytes());
+ try (BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")))) {
+ String format = "{ a : %d }%n";
+ for (int i = 0; i <= record_count; i += 2) {
+ os.write(String.format(format, i).getBytes());
+ }
}
- os.close();
- os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")));
- format = "{ a : \"%05d\" }%n";
- for (int i = 1; i <= record_count; i+=2) {
- os.write(String.format(format, i).getBytes());
+ try (BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")))) {
+ String format = "{ a : \"%05d\" }%n";
+ for (int i = 1; i <= record_count; i+=2) {
+ os.write(String.format(format, i).getBytes());
+ }
}
- os.close();
String query = "select * from dfs_test.tmp.numericAndStringTypes order by a desc";
TestBuilder builder = testBuilder()
.sqlQuery(query)
.ordered()
- .optionSettingQueriesForTestQuery("alter session set `exec.enable_union_type` = true")
+ .optionSettingQueriesForTestQuery(getOptions(testLegacy))
.baselineColumns("a");
// Strings come first because order by is desc
for (int i = record_count; i >= 0;) {
@@ -101,30 +130,40 @@ public class TestExternalSort extends BaseTestQuery {
}
@Test
- public void testNewColumns() throws Exception {
+ public void testNewColumnsManaged() throws Exception {
+ testNewColumns(false);
+ }
+
+
+ @Test
+ public void testNewColumnsLegacy() throws Exception {
+ testNewColumns(true);
+ }
+
+ private void testNewColumns(boolean testLegacy) throws Exception {
final int record_count = 10000;
String dfs_temp = getDfsTestTmpSchemaLocation();
System.out.println(dfs_temp);
File table_dir = new File(dfs_temp, "newColumns");
table_dir.mkdir();
- BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")));
- String format = "{ a : %d, b : %d }%n";
- for (int i = 0; i <= record_count; i += 2) {
- os.write(String.format(format, i, i).getBytes());
+ try (BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "a.json")))) {
+ String format = "{ a : %d, b : %d }%n";
+ for (int i = 0; i <= record_count; i += 2) {
+ os.write(String.format(format, i, i).getBytes());
+ }
}
- os.close();
- os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")));
- format = "{ a : %d, c : %d }%n";
- for (int i = 1; i <= record_count; i+=2) {
- os.write(String.format(format, i, i).getBytes());
+ try (BufferedOutputStream os = new BufferedOutputStream(new FileOutputStream(new File(table_dir, "b.json")))) {
+ String format = "{ a : %d, c : %d }%n";
+ for (int i = 1; i <= record_count; i+=2) {
+ os.write(String.format(format, i, i).getBytes());
+ }
}
- os.close();
String query = "select a, b, c from dfs_test.tmp.newColumns order by a desc";
// Test framework currently doesn't handle changing schema (i.e. new columns) on the client side
TestBuilder builder = testBuilder()
.sqlQuery(query)
.ordered()
- .optionSettingQueriesForTestQuery("alter session set `exec.enable_union_type` = true")
+ .optionSettingQueriesForTestQuery(getOptions(testLegacy))
.baselineColumns("a", "b", "c");
for (int i = record_count; i >= 0;) {
builder.baselineValues((long) i, (long) i--, null);
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/memory/base/src/main/java/org/apache/drill/exec/memory/BaseAllocator.java
----------------------------------------------------------------------
diff --git a/exec/memory/base/src/main/java/org/apache/drill/exec/memory/BaseAllocator.java b/exec/memory/base/src/main/java/org/apache/drill/exec/memory/BaseAllocator.java
index 8c7e7ca..1245e86 100644
--- a/exec/memory/base/src/main/java/org/apache/drill/exec/memory/BaseAllocator.java
+++ b/exec/memory/base/src/main/java/org/apache/drill/exec/memory/BaseAllocator.java
@@ -40,6 +40,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
public static final String DEBUG_ALLOCATOR = "drill.memory.debug.allocator";
+ @SuppressWarnings("unused")
private static final AtomicLong ID_GENERATOR = new AtomicLong(0);
private static final int CHUNK_SIZE = AllocationManager.INNER_ALLOCATOR.getChunkSize();
@@ -98,9 +99,9 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
historicalLog = null;
childLedgers = null;
}
-
}
+ @Override
public void assertOpen() {
if (AssertionUtil.ASSERT_ENABLED) {
if (isClosed) {
@@ -289,6 +290,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
}
}
+ @Override
public boolean add(final int nBytes) {
assertOpen();
@@ -310,6 +312,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
return true;
}
+ @Override
public DrillBuf allocateBuffer() {
assertOpen();
@@ -321,14 +324,17 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
return drillBuf;
}
+ @Override
public int getSize() {
return nBytes;
}
+ @Override
public boolean isUsed() {
return used;
}
+ @Override
public boolean isClosed() {
return closed;
}
@@ -366,6 +372,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
closed = true;
}
+ @Override
public boolean reserve(int nBytes) {
assertOpen();
@@ -511,6 +518,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
}
+ @Override
public String toString() {
final Verbosity verbosity = logger.isTraceEnabled() ? Verbosity.LOG_WITH_STACKTRACE
: Verbosity.BASIC;
@@ -525,6 +533,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
*
* @return A Verbose string of current allocator state.
*/
+ @Override
public String toVerboseString() {
final StringBuilder sb = new StringBuilder();
print(sb, 0, Verbosity.LOG_WITH_STACKTRACE);
@@ -542,7 +551,7 @@ public abstract class BaseAllocator extends Accountant implements BufferAllocato
* An integer value.
* @return The closest power of two of that value.
*/
- static int nextPowerOfTwo(int val) {
+ public static int nextPowerOfTwo(int val) {
int highestBit = Integer.highestOneBit(val);
if (highestBit == val) {
return val;
[3/3] drill git commit: DRILL-5080: Memory-managed version of
external sort
Posted by am...@apache.org.
DRILL-5080: Memory-managed version of external sort
Please see JIRA entry for reasons for revision, design spec and list of
changes.
This PR covers the changes to the external sort itself. Tests for this
operator require the test framework in DRILL-5126 and the mock data
source in DRILL-5152. Tests for this operator will be issued as a
separate PR once those two dependencies are committed.
Until then, the new operator is disabled by default. It can be enabled
using drill.sort.external.disable_managed: false.
The operator now spills before receiving a new batch. Revised memory calcs and
merge calcs to make them a bit clearer and provide more margin of error
for the power-of-two allocations used when allocating vectors.
We have two external sort implementations, but only one operator code
for both. They can use only one Metrics enum between them. When adding
new metrics to the new version, didn\u2019t add matching metrics to the old
one. This fixes that issue. (The issue will go away once the old one is
retired.)
Revised memory calculations to reflect limit of 16 MB per vector.
Current revision limits to 16 MB per output batch to be safe. Next
revision will enforce per-vector limits to allow the overall batch to
be larger when possible.
Also simplified the merge-time calculations.
Original code provided only crude methods to learn the size of a record
batch. Adds a "RecordBatchSizer" to provide detailed analysis so the
sort can know the amount of memory used to buffer a batch, the number
of rows, and the expected row width once the rows are copied to a
spill file or the output.
Moved generic spill classes to a separate package.
Created parameters for spill batch size and merge batch size. Separated
these values in code. Deprecated the min, max spill parameters as they
no longer add much value. Minor code rearranging.
Bug fix
Fixes a corner case of merging spilled files in a low-memory condition.
Fixes from code review
close apache/drill#717
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/300e9349
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/300e9349
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/300e9349
Branch: refs/heads/master
Commit: 300e9349a52df7f6ad866a78868602005eadb392
Parents: c9a6ac4
Author: Paul Rogers <pr...@maprtech.com>
Authored: Thu Dec 15 19:54:05 2016 -0800
Committer: Aman Sinha <as...@maprtech.com>
Committed: Sun Feb 12 19:41:34 2017 -0800
----------------------------------------------------------------------
.../org/apache/drill/exec/ExecConstants.java | 16 +
.../drill/exec/physical/base/AbstractBase.java | 20 +-
.../exec/physical/config/ExternalSort.java | 34 +-
.../physical/impl/spill/RecordBatchSizer.java | 293 ++++
.../exec/physical/impl/spill/SpillSet.java | 431 ++++++
.../exec/physical/impl/spill/package-info.java | 22 +
.../physical/impl/xsort/ExternalSortBatch.java | 16 +-
.../impl/xsort/ExternalSortBatchCreator.java | 26 +-
.../physical/impl/xsort/managed/BatchGroup.java | 367 +++++
.../impl/xsort/managed/CopierHolder.java | 307 ++++
.../impl/xsort/managed/ExternalSortBatch.java | 1456 ++++++++++++++++++
.../impl/xsort/managed/MSortTemplate.java | 237 +++
.../physical/impl/xsort/managed/MSorter.java | 40 +
.../physical/impl/xsort/managed/MergeSort.java | 167 ++
.../xsort/managed/OperatorCodeGenerator.java | 259 ++++
.../impl/xsort/managed/PriorityQueueCopier.java | 40 +
.../managed/PriorityQueueCopierTemplate.java | 170 ++
.../server/options/SystemOptionManager.java | 16 +-
.../exec/util/MemoryAllocationUtilities.java | 8 +-
.../src/main/resources/drill-module.conf | 54 +-
.../physical/impl/xsort/TestExternalSort.java | 113 +-
.../apache/drill/exec/memory/BaseAllocator.java | 13 +-
22 files changed, 4011 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 07cc3a7..d739f88 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -63,13 +63,29 @@ public interface ExecConstants {
String SPOOLING_BUFFER_DELETE = "drill.exec.buffer.spooling.delete";
String SPOOLING_BUFFER_MEMORY = "drill.exec.buffer.spooling.size";
String BATCH_PURGE_THRESHOLD = "drill.exec.sort.purge.threshold";
+
+ // External Sort Boot configuration
+
String EXTERNAL_SORT_TARGET_BATCH_SIZE = "drill.exec.sort.external.batch.size";
String EXTERNAL_SORT_TARGET_SPILL_BATCH_SIZE = "drill.exec.sort.external.spill.batch.size";
String EXTERNAL_SORT_SPILL_GROUP_SIZE = "drill.exec.sort.external.spill.group.size";
String EXTERNAL_SORT_SPILL_THRESHOLD = "drill.exec.sort.external.spill.threshold";
String EXTERNAL_SORT_SPILL_DIRS = "drill.exec.sort.external.spill.directories";
String EXTERNAL_SORT_SPILL_FILESYSTEM = "drill.exec.sort.external.spill.fs";
+ String EXTERNAL_SORT_SPILL_FILE_SIZE = "drill.exec.sort.external.spill.file_size";
String EXTERNAL_SORT_MSORT_MAX_BATCHSIZE = "drill.exec.sort.external.msort.batch.maxsize";
+ String EXTERNAL_SORT_DISABLE_MANAGED = "drill.exec.sort.external.disable_managed";
+ String EXTERNAL_SORT_MERGE_LIMIT = "drill.exec.sort.external.merge_limit";
+ String EXTERNAL_SORT_SPILL_BATCH_SIZE = "drill.exec.sort.external.spill.spill_batch_size";
+ String EXTERNAL_SORT_MERGE_BATCH_SIZE = "drill.exec.sort.external.spill.merge_batch_size";
+ String EXTERNAL_SORT_MAX_MEMORY = "drill.exec.sort.external.mem_limit";
+ String EXTERNAL_SORT_BATCH_LIMIT = "drill.exec.sort.external.batch_limit";
+
+ // External Sort Runtime options
+
+ BooleanValidator EXTERNAL_SORT_DISABLE_MANAGED_OPTION = new BooleanValidator("exec.sort.disable_managed", false);
+
+
String TEXT_LINE_READER_BATCH_SIZE = "drill.exec.storage.file.text.batch.size";
String TEXT_LINE_READER_BUFFER_SIZE = "drill.exec.storage.file.text.buffer.size";
String HAZELCAST_SUBNETS = "drill.exec.cache.hazel.subnets";
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractBase.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractBase.java
index c7b0e7e..526d728 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractBase.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractBase.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -27,8 +27,8 @@ public abstract class AbstractBase implements PhysicalOperator{
private final String userName;
- protected long initialAllocation = 1000000L;
- protected long maxAllocation = 10000000000L;
+ protected long initialAllocation = 1_000_000L;
+ protected long maxAllocation = 10_000_000_000L;
private int id;
private double cost;
@@ -78,19 +78,33 @@ public abstract class AbstractBase implements PhysicalOperator{
return SelectionVectorMode.NONE;
}
+ // Not available. Presumably because Drill does not currently use
+ // this value, though it does appear in some test physical plans.
+// public void setInitialAllocation(long alloc) {
+// initialAllocation = alloc;
+// }
+
@Override
public long getInitialAllocation() {
return initialAllocation;
}
+ @Override
public double getCost() {
return cost;
}
+ @Override
public void setCost(double cost) {
this.cost = cost;
}
+ // Not available. Presumably because Drill does not currently use
+ // this value, though it does appear in some test physical plans.
+// public void setMaxAllocation(long alloc) {
+// maxAllocation = alloc;
+// }
+
@Override
public long getMaxAllocation() {
return maxAllocation;
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/ExternalSort.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/ExternalSort.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/ExternalSort.java
index 456b9c0..17848d0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/ExternalSort.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/ExternalSort.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -21,7 +21,6 @@ import java.util.List;
import org.apache.drill.common.logical.data.Order.Ordering;
import org.apache.drill.exec.physical.base.PhysicalOperator;
-import org.apache.drill.exec.physical.base.PhysicalVisitor;
import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
import com.fasterxml.jackson.annotation.JsonCreator;
@@ -32,26 +31,10 @@ import com.fasterxml.jackson.annotation.JsonTypeName;
public class ExternalSort extends Sort {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExternalSort.class);
- private long initialAllocation = 20000000;
-
@JsonCreator
public ExternalSort(@JsonProperty("child") PhysicalOperator child, @JsonProperty("orderings") List<Ordering> orderings, @JsonProperty("reverse") boolean reverse) {
super(child, orderings, reverse);
- }
-
- @Override
- public List<Ordering> getOrderings() {
- return orderings;
- }
-
- @Override
- public boolean getReverse() {
- return reverse;
- }
-
- @Override
- public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E{
- return physicalVisitor.visitSort(this, value);
+ initialAllocation = 20_000_000;
}
@Override
@@ -66,13 +49,12 @@ public class ExternalSort extends Sort {
return CoreOperatorType.EXTERNAL_SORT_VALUE;
}
- public void setMaxAllocation(long maxAllocation) {
- this.maxAllocation = Math.max(initialAllocation, maxAllocation);
- }
+ // Set here, rather than the base class, because this is the only
+ // operator, at present, that makes use of the maximum allocation.
+ // Remove this, in favor of the base class version, when Drill
+ // sets the memory allocation for all operators.
- @Override
- public long getInitialAllocation() {
- return initialAllocation;
+ public void setMaxAllocation(long maxAllocation) {
+ this.maxAllocation = maxAllocation;
}
-
}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
new file mode 100644
index 0000000..05354e5
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/RecordBatchSizer.java
@@ -0,0 +1,293 @@
+/*
+ * 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.drill.exec.physical.impl.spill;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.memory.BaseAllocator;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.vector.BaseDataValueVector;
+import org.apache.drill.exec.vector.FixedWidthVector;
+import org.apache.drill.exec.vector.NullableVarCharVector;
+import org.apache.drill.exec.vector.NullableVector;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarCharVector;
+
+import io.netty.buffer.DrillBuf;
+
+/**
+ * Given a record batch or vector container, determines the actual memory
+ * consumed by each column, the average row, and the entire record batch.
+ */
+
+public class RecordBatchSizer {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(RecordBatchSizer.class);
+
+ /**
+ * Column size information.
+ */
+ public static class ColumnSize {
+ public final MaterializedField metadata;
+
+ /**
+ * Assumed size from Drill metadata.
+ */
+ public int stdSize;
+ /**
+ * Actual memory consumed by all the vectors associated with this column.
+ */
+ public int totalSize;
+ /**
+ * Actual average column width as determined from actual memory use. This
+ * size is larger than the actual data size since this size includes per-
+ * column overhead such as any unused vector space, etc.
+ */
+ public int estSize;
+
+ /**
+ * The size of the data vector backing the column. Useful for detecting
+ * cases of possible direct memory fragmentation.
+ */
+ public int dataVectorSize;
+ public int capacity;
+ public int density;
+ public int dataSize;
+
+ @SuppressWarnings("resource")
+ public ColumnSize(VectorWrapper<?> vw) {
+ metadata = vw.getField();
+ stdSize = TypeHelper.getSize(metadata.getType());
+
+ // Can't get size estimates if this is an empty batch.
+
+ ValueVector v = vw.getValueVector();
+ int rowCount = v.getAccessor().getValueCount();
+ if (rowCount == 0) {
+ return;
+ }
+ DrillBuf[] bufs = v.getBuffers(false);
+ for (DrillBuf buf : bufs) {
+ totalSize += buf.capacity();
+ }
+
+ // Capacity is the number of values that the vector could
+ // contain. This is useful only for fixed-length vectors.
+
+ capacity = v.getValueCapacity();
+
+ // Crude way to get the size of the buffer underlying simple (scalar) values.
+ // Ignores maps, lists and other esoterica. Uses a crude way to subtract out
+ // the null "bit" (really byte) buffer size for nullable vectors.
+
+ if (v instanceof BaseDataValueVector) {
+ dataVectorSize = totalSize;
+ if (v instanceof NullableVector) {
+ dataVectorSize -= bufs[0].getActualMemoryConsumed();
+ }
+ }
+
+ // Determine "density" the number of rows compared to potential
+ // capacity. Low-density batches occur at block boundaries, ends
+ // of files and so on. Low-density batches throw off our estimates
+ // for Varchar columns because we don't know the actual number of
+ // bytes consumed (that information is hidden behind the Varchar
+ // implementation where we can't get at it.)
+ //
+ // A better solution is to have each vector do this calc rather
+ // than trying to do it generically, but that increases the code
+ // change footprint and slows the commit process.
+
+ if (v instanceof FixedWidthVector) {
+ dataSize = stdSize * rowCount;
+ } else if ( v instanceof VarCharVector ) {
+ VarCharVector vv = (VarCharVector) v;
+ dataSize = vv.getOffsetVector().getAccessor().get(rowCount);
+ } else if ( v instanceof NullableVarCharVector ) {
+ NullableVarCharVector vv = (NullableVarCharVector) v;
+ dataSize = vv.getValuesVector().getOffsetVector().getAccessor().get(rowCount);
+ } else {
+ dataSize = 0;
+ }
+ if (dataSize > 0) {
+ density = roundUp(dataSize * 100, dataVectorSize);
+ estSize = roundUp(dataSize, rowCount);
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder buf = new StringBuilder();
+ buf.append(metadata.getName());
+ buf.append("(std col. size: ");
+ buf.append(stdSize);
+ buf.append(", actual col. size: ");
+ buf.append(estSize);
+ buf.append(", total size: ");
+ buf.append(totalSize);
+ buf.append(", vector size: ");
+ buf.append(dataVectorSize);
+ buf.append(", data size: ");
+ buf.append(dataSize);
+ buf.append(", row capacity: ");
+ buf.append(capacity);
+ buf.append(", density: ");
+ buf.append(density);
+ buf.append(")");
+ return buf.toString();
+ }
+ }
+
+ List<ColumnSize> columnSizes = new ArrayList<>();
+
+ /**
+ * Number of records (rows) in the batch.
+ */
+ private int rowCount;
+ /**
+ * Standard row width using Drill meta-data.
+ */
+ private int stdRowWidth;
+ /**
+ * Actual batch size summing all buffers used to store data
+ * for the batch.
+ */
+ private int totalBatchSize;
+ /**
+ * Actual row width computed by dividing total batch memory by the
+ * record count.
+ */
+ private int grossRowWidth;
+ /**
+ * Actual row width computed by summing columns. Use this if the
+ * vectors are partially full; prevents overestimating row width.
+ */
+ private int netRowWidth;
+ private boolean hasSv2;
+ private int sv2Size;
+ private int avgDensity;
+
+ public RecordBatchSizer(VectorAccessible va) {
+ rowCount = va.getRecordCount();
+ for (VectorWrapper<?> vw : va) {
+ measureField(vw);
+ }
+
+ if (rowCount > 0) {
+ grossRowWidth = roundUp(totalBatchSize, rowCount);
+ }
+
+ hasSv2 = va.getSchema().getSelectionVectorMode() == BatchSchema.SelectionVectorMode.TWO_BYTE;
+ if (hasSv2) {
+ @SuppressWarnings("resource")
+ SelectionVector2 sv2 = va.getSelectionVector2();
+ sv2Size = sv2.getBuffer().capacity();
+ grossRowWidth += sv2Size;
+ netRowWidth += 2;
+ }
+
+ int totalDensity = 0;
+ int usableCount = 0;
+ for (ColumnSize colSize : columnSizes) {
+ if ( colSize.density > 0 ) {
+ usableCount++;
+ }
+ totalDensity += colSize.density;
+ }
+ avgDensity = roundUp(totalDensity, usableCount);
+ }
+
+ public void applySv2() {
+ if (hasSv2) {
+ return;
+ }
+
+ sv2Size = BaseAllocator.nextPowerOfTwo(2 * rowCount);
+ grossRowWidth += roundUp(sv2Size, rowCount);
+ totalBatchSize += sv2Size;
+ }
+
+ private void measureField(VectorWrapper<?> vw) {
+ ColumnSize colSize = new ColumnSize(vw);
+ columnSizes.add(colSize);
+
+ stdRowWidth += colSize.stdSize;
+ totalBatchSize += colSize.totalSize;
+ netRowWidth += colSize.estSize;
+ }
+
+ public static int roundUp(int num, int denom) {
+ if(denom == 0) {
+ return 0;
+ }
+ return (int) Math.ceil((double) num / denom);
+ }
+
+ public int rowCount() { return rowCount; }
+ public int stdRowWidth() { return stdRowWidth; }
+ public int grossRowWidth() { return grossRowWidth; }
+ public int netRowWidth() { return netRowWidth; }
+ public int actualSize() { return totalBatchSize; }
+ public boolean hasSv2() { return hasSv2; }
+ public int getAvgDensity() { return avgDensity; }
+
+ public static final int MAX_VECTOR_SIZE = 16 * 1024 * 1024; // 16 MiB
+
+ /**
+ * Look for columns backed by vectors larger than the 16 MiB size
+ * employed by the Netty allocator. Such large blocks can lead to
+ * memory fragmentation and unexpected OOM errors.
+ * @return true if any column is oversized
+ */
+ public boolean checkOversizeCols() {
+ boolean hasOversize = false;
+ for (ColumnSize colSize : columnSizes) {
+ if ( colSize.dataVectorSize > MAX_VECTOR_SIZE) {
+ logger.warn( "Column is wider than 256 characters: OOM due to memory fragmentation is possible - " + colSize.metadata.getPath() );
+ hasOversize = true;
+ }
+ }
+ return hasOversize;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder buf = new StringBuilder();
+ buf.append("Actual batch schema & sizes {\n");
+ for (ColumnSize colSize : columnSizes) {
+ buf.append(" ");
+ buf.append(colSize.toString());
+ buf.append("\n");
+ }
+ buf.append( " Records: " );
+ buf.append(rowCount);
+ buf.append(", Total size: ");
+ buf.append(totalBatchSize);
+ buf.append(", Row width:");
+ buf.append(grossRowWidth);
+ buf.append(", Density:");
+ buf.append(avgDensity);
+ buf.append("}");
+ return buf.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/SpillSet.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/SpillSet.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/SpillSet.java
new file mode 100644
index 0000000..4615500
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/SpillSet.java
@@ -0,0 +1,431 @@
+/*
+ * 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.drill.exec.physical.impl.spill;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Iterator;
+import java.util.Set;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Sets;
+
+/**
+ * Generates the set of spill files for this sort session.
+ */
+
+public class SpillSet {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SpillSet.class);
+
+ /**
+ * Spilling on the Mac using the HDFS file system is very inefficient,
+ * affects performance numbers. This interface allows using HDFS in
+ * production, but to bypass the HDFS file system when needed.
+ */
+
+ private interface FileManager {
+
+ void deleteOnExit(String fragmentSpillDir) throws IOException;
+
+ OutputStream createForWrite(String fileName) throws IOException;
+
+ InputStream openForInput(String fileName) throws IOException;
+
+ void deleteFile(String fileName) throws IOException;
+
+ void deleteDir(String fragmentSpillDir) throws IOException;
+
+ /**
+ * Given a manager-specific output stream, return the current write position.
+ * Used to report total write bytes.
+ *
+ * @param outputStream output stream created by the file manager
+ * @return
+ */
+ long getWriteBytes(OutputStream outputStream);
+
+ /**
+ * Given a manager-specific input stream, return the current read position.
+ * Used to report total read bytes.
+ *
+ * @param outputStream input stream created by the file manager
+ * @return
+ */
+ long getReadBytes(InputStream inputStream);
+ }
+
+ /**
+ * Normal implementation of spill files using the HDFS file system.
+ */
+
+ private static class HadoopFileManager implements FileManager{
+ /**
+ * The HDFS file system (for local directories, HDFS storage, etc.) used to
+ * create the temporary spill files. Allows spill files to be either on local
+ * disk, or in a DFS. (The admin can choose to put spill files in DFS when
+ * nodes provide insufficient local disk space)
+ */
+
+ private FileSystem fs;
+
+ protected HadoopFileManager(String fsName) {
+ Configuration conf = new Configuration();
+ conf.set("fs.default.name", fsName);
+ try {
+ fs = FileSystem.get(conf);
+ } catch (IOException e) {
+ throw UserException.resourceError(e)
+ .message("Failed to get the File System for external sort")
+ .build(logger);
+ }
+ }
+
+ @Override
+ public void deleteOnExit(String fragmentSpillDir) throws IOException {
+ fs.deleteOnExit(new Path(fragmentSpillDir));
+ }
+
+ @Override
+ public OutputStream createForWrite(String fileName) throws IOException {
+ return fs.create(new Path(fileName));
+ }
+
+ @Override
+ public InputStream openForInput(String fileName) throws IOException {
+ return fs.open(new Path(fileName));
+ }
+
+ @Override
+ public void deleteFile(String fileName) throws IOException {
+ Path path = new Path(fileName);
+ if (fs.exists(path)) {
+ fs.delete(path, false);
+ }
+ }
+
+ @Override
+ public void deleteDir(String fragmentSpillDir) throws IOException {
+ Path path = new Path(fragmentSpillDir);
+ if (path != null && fs.exists(path)) {
+ if (fs.delete(path, true)) {
+ fs.cancelDeleteOnExit(path);
+ }
+ }
+ }
+
+ @Override
+ public long getWriteBytes(OutputStream outputStream) {
+ try {
+ return ((FSDataOutputStream) outputStream).getPos();
+ } catch (IOException e) {
+ // Just used for logging, not worth dealing with the exception.
+ return 0;
+ }
+ }
+
+ @Override
+ public long getReadBytes(InputStream inputStream) {
+ try {
+ return ((FSDataInputStream) inputStream).getPos();
+ } catch (IOException e) {
+ // Just used for logging, not worth dealing with the exception.
+ return 0;
+ }
+ }
+ }
+
+ public static class CountingInputStream extends InputStream
+ {
+ private InputStream in;
+ private long count;
+
+ public CountingInputStream(InputStream in) {
+ this.in = in;
+ }
+
+ @Override
+ public int read() throws IOException {
+ int b = in.read();
+ if (b != -1) {
+ count++;
+ }
+ return b;
+ }
+
+ @Override
+ public int read(byte b[]) throws IOException {
+ int n = in.read(b);
+ if (n != -1) {
+ count += n;
+ }
+ return n;
+ }
+
+ @Override
+ public int read(byte b[], int off, int len) throws IOException {
+ int n = in.read(b, off, len);
+ if (n != -1) {
+ count += n;
+ }
+ return n;
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ return in.skip(n);
+ }
+
+ @Override
+ public void close() throws IOException {
+ in.close();
+ }
+
+ public long getCount() { return count; }
+ }
+
+ public static class CountingOutputStream extends OutputStream {
+
+ private OutputStream out;
+ private long count;
+
+ public CountingOutputStream(OutputStream out) {
+ this.out = out;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ count++;
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ count += b.length;
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ count += len;
+ out.write(b, off, len);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ out.close();
+ }
+
+ public long getCount() { return count; }
+ }
+
+ /**
+ * Performance-oriented direct access to the local file system which
+ * bypasses HDFS.
+ */
+
+ private static class LocalFileManager implements FileManager {
+
+ private File baseDir;
+
+ public LocalFileManager(String fsName) {
+ baseDir = new File(fsName.replace("file://", ""));
+ }
+
+ @Override
+ public void deleteOnExit(String fragmentSpillDir) throws IOException {
+ File dir = new File(baseDir, fragmentSpillDir);
+ dir.mkdirs();
+ dir.deleteOnExit();
+ }
+
+ @SuppressWarnings("resource")
+ @Override
+ public OutputStream createForWrite(String fileName) throws IOException {
+ return new CountingOutputStream(
+ new BufferedOutputStream(
+ new FileOutputStream(new File(baseDir, fileName))));
+ }
+
+ @SuppressWarnings("resource")
+ @Override
+ public InputStream openForInput(String fileName) throws IOException {
+ return new CountingInputStream(
+ new BufferedInputStream(
+ new FileInputStream(new File(baseDir, fileName))));
+ }
+
+ @Override
+ public void deleteFile(String fileName) throws IOException {
+ new File(baseDir, fileName).delete();
+ }
+
+ @Override
+ public void deleteDir(String fragmentSpillDir) throws IOException {
+ new File(baseDir, fragmentSpillDir).delete();
+ }
+
+ @Override
+ public long getWriteBytes(OutputStream outputStream) {
+ return ((CountingOutputStream) outputStream).getCount();
+ }
+
+ @Override
+ public long getReadBytes(InputStream inputStream) {
+ return ((CountingInputStream) inputStream).getCount();
+ }
+ }
+
+ private final Iterator<String> dirs;
+
+ /**
+ * Set of directories to which this operator should write spill files in a round-robin
+ * fashion. The operator requires at least one spill directory, but can
+ * support any number. The admin must ensure that sufficient space exists
+ * on all directories as this operator does not check space availability
+ * before writing to the directories.
+ */
+
+ private Set<String> currSpillDirs = Sets.newTreeSet();
+
+ /**
+ * The base part of the file name for spill files. Each file has this
+ * name plus an appended spill serial number.
+ */
+
+ private final String spillDirName;
+
+ private int fileCount = 0;
+
+ private FileManager fileManager;
+
+ private long readBytes;
+
+ private long writeBytes;
+
+ public SpillSet(FragmentContext context, PhysicalOperator popConfig) {
+ DrillConfig config = context.getConfig();
+ dirs = Iterators.cycle(config.getStringList(ExecConstants.EXTERNAL_SORT_SPILL_DIRS));
+
+ // Use the high-performance local file system if the local file
+ // system is selected and impersonation is off. (We use that
+ // as a proxy for a non-production Drill setup.)
+
+ String spillFs = config.getString(ExecConstants.EXTERNAL_SORT_SPILL_FILESYSTEM);
+ boolean impersonationEnabled = config.getBoolean(ExecConstants.IMPERSONATION_ENABLED);
+ if (spillFs.startsWith("file:///") && ! impersonationEnabled) {
+ fileManager = new LocalFileManager(spillFs);
+ } else {
+ fileManager = new HadoopFileManager(spillFs);
+ }
+ FragmentHandle handle = context.getHandle();
+ spillDirName = String.format("%s_major%s_minor%s_op%s", QueryIdHelper.getQueryId(handle.getQueryId()),
+ handle.getMajorFragmentId(), handle.getMinorFragmentId(), popConfig.getOperatorId());
+ }
+
+ public String getNextSpillFile() {
+
+ // Identify the next directory from the round-robin list to
+ // the file created from this round of spilling. The directory must
+ // must have sufficient space for the output file.
+
+ String spillDir = dirs.next();
+ String currSpillPath = Joiner.on("/").join(spillDir, spillDirName);
+ currSpillDirs.add(currSpillPath);
+ String outputFile = Joiner.on("/").join(currSpillPath, "spill" + ++fileCount);
+ try {
+ fileManager.deleteOnExit(currSpillPath);
+ } catch (IOException e) {
+ // since this is meant to be used in a batches's spilling, we don't propagate the exception
+ logger.warn("Unable to mark spill directory " + currSpillPath + " for deleting on exit", e);
+ }
+ return outputFile;
+ }
+
+ public boolean hasSpilled() {
+ return fileCount > 0;
+ }
+
+ public int getFileCount() { return fileCount; }
+
+ public InputStream openForInput(String fileName) throws IOException {
+ return fileManager.openForInput(fileName);
+ }
+
+ public OutputStream openForOutput(String fileName) throws IOException {
+ return fileManager.createForWrite(fileName);
+ }
+
+ public void delete(String fileName) throws IOException {
+ fileManager.deleteFile(fileName);
+ }
+
+ public long getWriteBytes() { return writeBytes; }
+ public long getReadBytes() { return readBytes; }
+
+ public void close() {
+ for (String path : currSpillDirs) {
+ try {
+ fileManager.deleteDir(path);
+ } catch (IOException e) {
+ // since this is meant to be used in a batches's cleanup, we don't propagate the exception
+ logger.warn("Unable to delete spill directory " + path, e);
+ }
+ }
+ }
+
+ public long getPosition(InputStream inputStream) {
+ return fileManager.getReadBytes(inputStream);
+ }
+
+ public long getPosition(OutputStream outputStream) {
+ return fileManager.getWriteBytes(outputStream);
+ }
+
+ public void tallyReadBytes(long readLength) {
+ readBytes += readLength;
+ }
+
+ public void tallyWriteBytes(long writeLength) {
+ writeBytes += writeLength;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/package-info.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/package-info.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/package-info.java
new file mode 100644
index 0000000..6cc74f5
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/spill/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Collection of classes shared by operators that implement spill-to-disk.
+ */
+
+package org.apache.drill.exec.physical.impl.spill;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
index 8fe05f0..c009cc0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatch.java
@@ -133,10 +133,22 @@ public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
public static final String INTERRUPTION_AFTER_SETUP = "after-setup";
public static final String INTERRUPTION_WHILE_SPILLING = "spilling";
+ // Be careful here! This enum is used in TWO places! First, it is used
+ // in this code to build up metrics. Easy enough. But, it is also used
+ // in OperatorMetricRegistry to define the metrics for the
+ // operator ID defined in CoreOperatorType. As a result, the values
+ // defined here are shared between this legacy version AND the new
+ // managed version. (Though the new, managed version has its own
+ // copy of this enum.) The two enums MUST be identical.
+
public enum Metric implements MetricDef {
SPILL_COUNT, // number of times operator spilled to disk
- PEAK_SIZE_IN_MEMORY, // peak value for totalSizeInMemory
- PEAK_BATCHES_IN_MEMORY; // maximum number of batches kept in memory
+ RETIRED1, // Was: peak value for totalSizeInMemory
+ // But operator already provides this value
+ PEAK_BATCHES_IN_MEMORY, // maximum number of batches kept in memory
+ MERGE_COUNT, // Used only by the managed version.
+ MIN_BUFFER, // Used only by the managed version.
+ INPUT_BATCHES; // Used only by the managed version.
@Override
public int metricId() {
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatchCreator.java
index b9f6396..e579fc2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/ExternalSortBatchCreator.java
@@ -19,23 +19,41 @@ package org.apache.drill.exec.physical.impl.xsort;
import java.util.List;
+import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.config.ExternalSort;
import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.record.AbstractRecordBatch;
import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.server.options.OptionManager;
import com.google.common.base.Preconditions;
public class ExternalSortBatchCreator implements BatchCreator<ExternalSort>{
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExternalSortBatchCreator.class);
@Override
- public ExternalSortBatch getBatch(FragmentContext context, ExternalSort config, List<RecordBatch> children)
+ public AbstractRecordBatch<ExternalSort> getBatch(FragmentContext context, ExternalSort config, List<RecordBatch> children)
throws ExecutionSetupException {
Preconditions.checkArgument(children.size() == 1);
- return new ExternalSortBatch(config, context, children.iterator().next());
- }
+ // Prefer the managed version, but provide runtime and boot-time options
+ // to disable it and revert to the "legacy" version. The legacy version
+ // is retained primarily to allow cross-check testing against the managed
+ // version, and as a fall back in the first release of the managed version.
+ OptionManager optionManager = context.getOptions();
+ boolean disableManaged = optionManager.getOption(ExecConstants.EXTERNAL_SORT_DISABLE_MANAGED_OPTION);
+ if ( ! disableManaged ) {
+ DrillConfig drillConfig = context.getConfig();
+ disableManaged = drillConfig.hasPath(ExecConstants.EXTERNAL_SORT_DISABLE_MANAGED) &&
+ drillConfig.getBoolean(ExecConstants.EXTERNAL_SORT_DISABLE_MANAGED);
+ }
+ if (disableManaged) {
+ return new ExternalSortBatch(config, context, children.iterator().next());
+ } else {
+ return new org.apache.drill.exec.physical.impl.xsort.managed.ExternalSortBatch(config, context, children.iterator().next());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
new file mode 100644
index 0000000..cd5cd1f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/BatchGroup.java
@@ -0,0 +1,367 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Iterator;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.cache.VectorAccessibleSerializable;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.spill.SpillSet;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.SchemaUtil;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.WritableBatch;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+
+import com.google.common.base.Stopwatch;
+
+/**
+ * Represents a group of batches spilled to disk.
+ * <p>
+ * The batches are defined by a schema which can change over time. When the schema changes,
+ * all existing and new batches are coerced into the new schema. Provides a
+ * uniform way to iterate over records for one or more batches whether
+ * the batches are in memory or on disk.
+ * <p>
+ * The <code>BatchGroup</code> operates in two modes as given by the two
+ * subclasses:
+ * <ul>
+ * <li>Input mode (@link InputBatchGroup): Used to buffer in-memory batches
+ * prior to spilling.</li>
+ * <li>Spill mode (@link SpilledBatchGroup): Holds a "memento" to a set
+ * of batches written to disk. Acts as both a reader and writer for
+ * those batches.</li>
+ */
+
+public abstract class BatchGroup implements VectorAccessible, AutoCloseable {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(BatchGroup.class);
+
+ /**
+ * The input batch group gathers batches buffered in memory before
+ * spilling. The structure of the data is:
+ * <ul>
+ * <li>Contains a single batch received from the upstream (input)
+ * operator.</li>
+ * <li>Associated selection vector that provides a sorted
+ * indirection to the values in the batch.</li>
+ * </ul>
+ */
+
+ public static class InputBatch extends BatchGroup {
+ private SelectionVector2 sv2;
+
+ public InputBatch(VectorContainer container, SelectionVector2 sv2, OperatorContext context, long batchSize) {
+ super(container, context, batchSize);
+ this.sv2 = sv2;
+ }
+
+ public SelectionVector2 getSv2() {
+ return sv2;
+ }
+
+ @Override
+ public int getRecordCount() {
+ if (sv2 != null) {
+ return sv2.getCount();
+ } else {
+ return super.getRecordCount();
+ }
+ }
+
+ @Override
+ public int getNextIndex() {
+ int val = super.getNextIndex();
+ if (val == -1) {
+ return val;
+ }
+ return sv2.getIndex(val);
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ super.close();
+ }
+ finally {
+ if (sv2 != null) {
+ sv2.clear();
+ }
+ }
+ }
+ }
+
+ /**
+ * Holds a set of spilled batches, represented by a file on disk.
+ * Handles reads from, and writes to the spill file. The data structure
+ * is:
+ * <ul>
+ * <li>A pointer to a file that contains serialized batches.</li>
+ * <li>When writing, each batch is appended to the output file.</li>
+ * <li>When reading, iterates over each spilled batch, and for each
+ * of those, each spilled record.</li>
+ * </ul>
+ * <p>
+ * Starts out with no current batch. Defines the current batch to be the
+ * (shell: schema without data) of the last batch spilled to disk.
+ * <p>
+ * When reading, has destructive read-once behavior: closing the
+ * batch (after reading) deletes the underlying spill file.
+ * <p>
+ * This single class does three tasks: load data, hold data and
+ * read data. This should be split into three separate classes. But,
+ * the original (combined) structure is retained for expedience at
+ * present.
+ */
+
+ public static class SpilledRun extends BatchGroup {
+ private InputStream inputStream;
+ private OutputStream outputStream;
+ private String path;
+ private SpillSet spillSet;
+ private BufferAllocator allocator;
+ private int spilledBatches = 0;
+
+ public SpilledRun(SpillSet spillSet, String path, OperatorContext context, long batchSize) throws IOException {
+ super(null, context, batchSize);
+ this.spillSet = spillSet;
+ this.path = path;
+ this.allocator = context.getAllocator();
+ outputStream = spillSet.openForOutput(path);
+ }
+
+ public void addBatch(VectorContainer newContainer) throws IOException {
+ int recordCount = newContainer.getRecordCount();
+ @SuppressWarnings("resource")
+ WritableBatch batch = WritableBatch.getBatchNoHVWrap(recordCount, newContainer, false);
+ VectorAccessibleSerializable outputBatch = new VectorAccessibleSerializable(batch, allocator);
+ Stopwatch watch = Stopwatch.createStarted();
+ outputBatch.writeToStream(outputStream);
+ newContainer.zeroVectors();
+ logger.trace("Wrote {} records in {} us", recordCount, watch.elapsed(TimeUnit.MICROSECONDS));
+ spilledBatches++;
+
+ // Hold onto the husk of the last added container so that we have a
+ // current container when starting to read rows back later.
+
+ currentContainer = newContainer;
+ currentContainer.setRecordCount(0);
+ }
+
+ @Override
+ public int getNextIndex() {
+ if (pointer == getRecordCount()) {
+ if (spilledBatches == 0) {
+ return -1;
+ }
+ try {
+ currentContainer.zeroVectors();
+ getBatch();
+ } catch (IOException e) {
+ // Release any partially-loaded data.
+ currentContainer.clear();
+ throw UserException.dataReadError(e)
+ .message("Failure while reading spilled data")
+ .build(logger);
+ }
+
+ // The pointer indicates the NEXT index, not the one we
+ // return here. At this point, we just started reading a
+ // new batch and have returned index 0. So, the next index
+ // is 1.
+
+ pointer = 1;
+ return 0;
+ }
+ return super.getNextIndex();
+ }
+
+ private VectorContainer getBatch() throws IOException {
+ if (inputStream == null) {
+ inputStream = spillSet.openForInput(path);
+ }
+ VectorAccessibleSerializable vas = new VectorAccessibleSerializable(allocator);
+ Stopwatch watch = Stopwatch.createStarted();
+ vas.readFromStream(inputStream);
+ VectorContainer c = vas.get();
+ if (schema != null) {
+ c = SchemaUtil.coerceContainer(c, schema, context);
+ }
+ logger.trace("Read {} records in {} us", c.getRecordCount(), watch.elapsed(TimeUnit.MICROSECONDS));
+ spilledBatches--;
+ currentContainer.zeroVectors();
+ Iterator<VectorWrapper<?>> wrapperIterator = c.iterator();
+ for (@SuppressWarnings("rawtypes") VectorWrapper w : currentContainer) {
+ TransferPair pair = wrapperIterator.next().getValueVector().makeTransferPair(w.getValueVector());
+ pair.transfer();
+ }
+ currentContainer.setRecordCount(c.getRecordCount());
+ c.zeroVectors();
+ return c;
+ }
+
+ /**
+ * Close resources owned by this batch group. Each can fail; report
+ * only the first error. This is cluttered because this class tries
+ * to do multiple tasks. TODO: Split into multiple classes.
+ */
+
+ @Override
+ public void close() throws IOException {
+ IOException ex = null;
+ try {
+ super.close();
+ } catch (IOException e) {
+ ex = e;
+ }
+ try {
+ closeOutputStream();
+ } catch (IOException e) {
+ ex = ex == null ? e : ex;
+ }
+ try {
+ closeInputStream();
+ } catch (IOException e) {
+ ex = ex == null ? e : ex;
+ }
+ try {
+ spillSet.delete(path);
+ } catch (IOException e) {
+ ex = ex == null ? e : ex;
+ }
+ if (ex != null) {
+ throw ex;
+ }
+ }
+
+ private void closeInputStream() throws IOException {
+ if (inputStream == null) {
+ return;
+ }
+ long readLength = spillSet.getPosition(inputStream);
+ spillSet.tallyReadBytes(readLength);
+ inputStream.close();
+ inputStream = null;
+ logger.trace("Summary: Read {} bytes from {}", readLength, path);
+ }
+
+ public long closeOutputStream() throws IOException {
+ if (outputStream == null) {
+ return 0;
+ }
+ long posn = spillSet.getPosition(outputStream);
+ spillSet.tallyWriteBytes(posn);
+ outputStream.close();
+ outputStream = null;
+ logger.trace("Summary: Wrote {} bytes to {}", posn, path);
+ return posn;
+ }
+ }
+
+ protected VectorContainer currentContainer;
+ protected int pointer = 0;
+ protected OperatorContext context;
+ protected BatchSchema schema;
+ protected long dataSize;
+
+ public BatchGroup(VectorContainer container, OperatorContext context, long dataSize) {
+ this.currentContainer = container;
+ this.context = context;
+ this.dataSize = dataSize;
+ }
+
+ /**
+ * Updates the schema for this batch group. The current as well as any
+ * deserialized batches will be coerced to this schema.
+ * @param schema
+ */
+ public void setSchema(BatchSchema schema) {
+ currentContainer = SchemaUtil.coerceContainer(currentContainer, schema, context);
+ this.schema = schema;
+ }
+
+ public int getNextIndex() {
+ if (pointer == getRecordCount()) {
+ return -1;
+ }
+ int val = pointer++;
+ assert val < currentContainer.getRecordCount();
+ return val;
+ }
+
+ public VectorContainer getContainer() {
+ return currentContainer;
+ }
+
+ @Override
+ public void close() throws IOException {
+ currentContainer.zeroVectors();
+ }
+
+ @Override
+ public VectorWrapper<?> getValueAccessorById(Class<?> clazz, int... ids) {
+ return currentContainer.getValueAccessorById(clazz, ids);
+ }
+
+ @Override
+ public TypedFieldId getValueVectorId(SchemaPath path) {
+ return currentContainer.getValueVectorId(path);
+ }
+
+ @Override
+ public BatchSchema getSchema() {
+ return currentContainer.getSchema();
+ }
+
+ @Override
+ public int getRecordCount() {
+ return currentContainer.getRecordCount();
+ }
+
+ public int getUnfilteredRecordCount() {
+ return currentContainer.getRecordCount();
+ }
+
+ public long getDataSize() { return dataSize; }
+
+ @Override
+ public Iterator<VectorWrapper<?>> iterator() {
+ return currentContainer.iterator();
+ }
+
+ @Override
+ public SelectionVector2 getSelectionVector2() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public SelectionVector4 getSelectionVector4() {
+ throw new UnsupportedOperationException();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/CopierHolder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/CopierHolder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/CopierHolder.java
new file mode 100644
index 0000000..4fa520d
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/CopierHolder.java
@@ -0,0 +1,307 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.TypeHelper;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.xsort.managed.ExternalSortBatch.SortResults;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.vector.ValueVector;
+
+import com.google.common.base.Stopwatch;
+
+/**
+ * Manages a {@link PriorityQueueCopier} instance produced from code generation.
+ * Provides a wrapper around a copier "session" to simplify reading batches
+ * from the copier.
+ */
+
+public class CopierHolder {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CopierHolder.class);
+
+ private PriorityQueueCopier copier;
+
+ private final FragmentContext context;
+ private final BufferAllocator allocator;
+ private OperatorCodeGenerator opCodeGen;
+
+ public CopierHolder(FragmentContext context, BufferAllocator allocator, OperatorCodeGenerator opCodeGen) {
+ this.context = context;
+ this.allocator = allocator;
+ this.opCodeGen = opCodeGen;
+ }
+
+ /**
+ * Start a merge operation using a temporary vector container. Used for
+ * intermediate merges.
+ *
+ * @param schema
+ * @param batchGroupList
+ * @param targetRecordCount
+ * @return
+ */
+
+ public CopierHolder.BatchMerger startMerge(BatchSchema schema, List<? extends BatchGroup> batchGroupList, int targetRecordCount) {
+ return new BatchMerger(this, schema, batchGroupList, targetRecordCount);
+ }
+
+ /**
+ * Start a merge operation using the specified vector container. Used for
+ * the final merge operation.
+ *
+ * @param schema
+ * @param batchGroupList
+ * @param outputContainer
+ * @param targetRecordCount
+ * @return
+ */
+ public CopierHolder.BatchMerger startFinalMerge(BatchSchema schema, List<? extends BatchGroup> batchGroupList, VectorContainer outputContainer, int targetRecordCount) {
+ return new BatchMerger(this, schema, batchGroupList, outputContainer, targetRecordCount);
+ }
+
+ /**
+ * Prepare a copier which will write a collection of vectors to disk. The copier
+ * uses generated code to do the actual writes. If the copier has not yet been
+ * created, generate code and create it. If it has been created, close it and
+ * prepare it for a new collection of batches.
+ *
+ * @param batch the (hyper) batch of vectors to be copied
+ * @param batchGroupList same batches as above, but represented as a list
+ * of individual batches
+ * @param outputContainer the container into which to copy the batches
+ */
+
+ @SuppressWarnings("unchecked")
+ private void createCopier(VectorAccessible batch, List<? extends BatchGroup> batchGroupList, VectorContainer outputContainer) {
+ if (copier != null) {
+ opCodeGen.closeCopier();
+ } else {
+ copier = opCodeGen.getCopier(batch);
+ }
+
+ // Initialize the value vectors for the output container
+
+ for (VectorWrapper<?> i : batch) {
+ @SuppressWarnings("resource")
+ ValueVector v = TypeHelper.getNewVector(i.getField(), allocator);
+ outputContainer.add(v);
+ }
+ try {
+ copier.setup(context, allocator, batch, (List<BatchGroup>) batchGroupList, outputContainer);
+ } catch (SchemaChangeException e) {
+ throw UserException.unsupportedError(e)
+ .message("Unexpected schema change - likely code error.")
+ .build(logger);
+ }
+ }
+
+ public BufferAllocator getAllocator() { return allocator; }
+
+ public void close() {
+ opCodeGen.closeCopier();
+ copier = null;
+ }
+
+ /**
+ * We've gathered a set of batches, each of which has been sorted. The batches
+ * may have passed through a filter and thus may have "holes" where rows have
+ * been filtered out. We will spill records in blocks of targetRecordCount.
+ * To prepare, copy that many records into an outputContainer as a set of
+ * contiguous values in new vectors. The result is a single batch with
+ * vectors that combine a collection of input batches up to the
+ * given threshold.
+ * <p>
+ * Input. Here the top line is a selection vector of indexes.
+ * The second line is a set of batch groups (separated by underscores)
+ * with letters indicating individual records:<pre>
+ * [3 7 4 8 0 6 1] [5 3 6 8 2 0]
+ * [eh_ad_ibf] [r_qm_kn_p]</pre>
+ * <p>
+ * Output, assuming blocks of 5 records. The brackets represent
+ * batches, the line represents the set of batches copied to the
+ * spill file.<pre>
+ * [abcde] [fhikm] [npqr]</pre>
+ * <p>
+ * The copying operation does a merge as well: copying
+ * values from the sources in ordered fashion. Consider a different example,
+ * we want to merge two input batches to produce a single output batch:
+ * <pre>
+ * Input: [aceg] [bdfh]
+ * Output: [abcdefgh]</pre>
+ * <p>
+ * In the above, the input consists of two sorted batches. (In reality,
+ * the input batches have an associated selection vector, but that is omitted
+ * here and just the sorted values shown.) The output is a single batch
+ * with the merged records (indicated by letters) from the two input batches.
+ * <p>
+ * Here we bind the copier to the batchGroupList of sorted, buffered batches
+ * to be merged. We bind the copier output to outputContainer: the copier will write its
+ * merged "batches" of records to that container.
+ * <p>
+ * Calls to the {@link #next()} method sequentially return merged batches
+ * of the desired row count.
+ */
+
+ public static class BatchMerger implements SortResults, AutoCloseable {
+
+ private CopierHolder holder;
+ private VectorContainer hyperBatch;
+ private VectorContainer outputContainer;
+ private int targetRecordCount;
+ private int copyCount;
+ private int batchCount;
+
+ /**
+ * Creates a merger with an temporary output container.
+ *
+ * @param holder the copier that does the work
+ * @param schema schema for the input and output batches
+ * @param batchGroupList the input batches
+ * @param targetRecordCount number of records for each output batch
+ */
+ private BatchMerger(CopierHolder holder, BatchSchema schema, List<? extends BatchGroup> batchGroupList,
+ int targetRecordCount) {
+ this(holder, schema, batchGroupList, new VectorContainer(), targetRecordCount);
+ }
+
+ /**
+ * Creates a merger with the specified output container
+ *
+ * @param holder the copier that does the work
+ * @param schema schema for the input and output batches
+ * @param batchGroupList the input batches
+ * @param outputContainer merges output batch into the given output container
+ * @param targetRecordCount number of records for each output batch
+ */
+ private BatchMerger(CopierHolder holder, BatchSchema schema, List<? extends BatchGroup> batchGroupList,
+ VectorContainer outputContainer, int targetRecordCount) {
+ this.holder = holder;
+ hyperBatch = constructHyperBatch(schema, batchGroupList);
+ copyCount = 0;
+ this.targetRecordCount = targetRecordCount;
+ this.outputContainer = outputContainer;
+ holder.createCopier(hyperBatch, batchGroupList, outputContainer);
+ }
+
+ /**
+ * Return the output container.
+ *
+ * @return the output container
+ */
+ public VectorContainer getOutput() {
+ return outputContainer;
+ }
+
+ /**
+ * Read the next merged batch. The batch holds the specified row count, but
+ * may be less if this is the last batch.
+ *
+ * @return the number of rows in the batch, or 0 if no more batches
+ * are available
+ */
+
+ @Override
+ public boolean next() {
+ Stopwatch w = Stopwatch.createStarted();
+ int count = holder.copier.next(targetRecordCount);
+ copyCount += count;
+ if (count > 0) {
+ long t = w.elapsed(TimeUnit.MICROSECONDS);
+ logger.trace("Took {} us to merge {} records", t, count);
+ } else {
+ logger.trace("copier returned 0 records");
+ }
+ batchCount++;
+
+ // Identify the schema to be used in the output container. (Since
+ // all merged batches have the same schema, the schema we identify
+ // here should be the same as that which we already had.
+
+ outputContainer.buildSchema(BatchSchema.SelectionVectorMode.NONE);
+
+ // The copier does not set the record count in the output
+ // container, so do that here.
+
+ outputContainer.setRecordCount(count);
+
+ return count > 0;
+ }
+
+ /**
+ * Construct a vector container that holds a list of batches, each represented as an
+ * array of vectors. The entire collection of vectors has a common schema.
+ * <p>
+ * To build the collection, we go through the current schema (which has been
+ * devised to be common for all batches.) For each field in the schema, we create
+ * an array of vectors. To create the elements, we iterate over all the incoming
+ * batches and search for the vector that matches the current column.
+ * <p>
+ * Finally, we build a new schema for the combined container. That new schema must,
+ * because of the way the container was created, match the current schema.
+ *
+ * @param schema schema for the hyper batch
+ * @param batchGroupList list of batches to combine
+ * @return a container where each column is represented as an array of vectors
+ * (hence the "hyper" in the method name)
+ */
+
+ private VectorContainer constructHyperBatch(BatchSchema schema, List<? extends BatchGroup> batchGroupList) {
+ VectorContainer cont = new VectorContainer();
+ for (MaterializedField field : schema) {
+ ValueVector[] vectors = new ValueVector[batchGroupList.size()];
+ int i = 0;
+ for (BatchGroup group : batchGroupList) {
+ vectors[i++] = group.getValueAccessorById(
+ field.getValueClass(),
+ group.getValueVectorId(SchemaPath.getSimplePath(field.getPath())).getFieldIds())
+ .getValueVector();
+ }
+ cont.add(vectors);
+ }
+ cont.buildSchema(BatchSchema.SelectionVectorMode.FOUR_BYTE);
+ return cont;
+ }
+
+ @Override
+ public void close() {
+ hyperBatch.clear();
+ holder.close();
+ }
+
+ @Override
+ public int getRecordCount() {
+ return copyCount;
+ }
+
+ @Override
+ public int getBatchCount() {
+ return batchCount;
+ }
+ }
+}
[2/3] drill git commit: DRILL-5080: Memory-managed version of
external sort
Posted by am...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/ExternalSortBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/ExternalSortBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/ExternalSortBatch.java
new file mode 100644
index 0000000..783865c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/ExternalSortBatch.java
@@ -0,0 +1,1456 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.OutOfMemoryException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.MetricDef;
+import org.apache.drill.exec.physical.config.ExternalSort;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.physical.impl.spill.RecordBatchSizer;
+import org.apache.drill.exec.physical.impl.spill.SpillSet;
+import org.apache.drill.exec.physical.impl.xsort.MSortTemplate;
+import org.apache.drill.exec.physical.impl.xsort.SingleBatchSorter;
+import org.apache.drill.exec.physical.impl.xsort.managed.BatchGroup.InputBatch;
+import org.apache.drill.exec.record.AbstractRecordBatch;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.SchemaUtil;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.record.WritableBatch;
+import org.apache.drill.exec.record.selection.SelectionVector2;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.drill.exec.testing.ControlsInjector;
+import org.apache.drill.exec.testing.ControlsInjectorFactory;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.complex.AbstractContainerVector;
+
+import com.google.common.collect.Lists;
+
+/**
+ * External sort batch: a sort batch which can spill to disk in
+ * order to operate within a defined memory footprint.
+ * <p>
+ * <h4>Basic Operation</h4>
+ * The operator has three key phases:
+ * <p>
+ * <ul>
+ * <li>The load phase in which batches are read from upstream.</li>
+ * <li>The merge phase in which spilled batches are combined to
+ * reduce the number of files below the configured limit. (Best
+ * practice is to configure the system to avoid this phase.)
+ * <li>The delivery phase in which batches are combined to produce
+ * the final output.</li>
+ * </ul>
+ * During the load phase:
+ * <p>
+ * <ul>
+ * <li>The incoming (upstream) operator provides a series of batches.</li>
+ * <li>This operator sorts each batch, and accumulates them in an in-memory
+ * buffer.</li>
+ * <li>If the in-memory buffer becomes too large, this operator selects
+ * a subset of the buffered batches to spill.</li>
+ * <li>Each spill set is merged to create a new, sorted collection of
+ * batches, and each is spilled to disk.</li>
+ * <li>To allow the use of multiple disk storage, each spill group is written
+ * round-robin to a set of spill directories.</li>
+ * </ul>
+ * <p>
+ * Data is spilled to disk as a "run". A run consists of one or more (typically
+ * many) batches, each of which is itself a sorted run of records.
+ * <p>
+ * During the sort/merge phase:
+ * <p>
+ * <ul>
+ * <li>When the input operator is complete, this operator merges the accumulated
+ * batches (which may be all in memory or partially on disk), and returns
+ * them to the output (downstream) operator in chunks of no more than
+ * 64K records.</li>
+ * <li>The final merge must combine a collection of in-memory and spilled
+ * batches. Several limits apply to the maximum "width" of this merge. For
+ * example, each open spill run consumes a file handle, and we may wish
+ * to limit the number of file handles. Further, memory must hold one batch
+ * from each run, so we may need to reduce the number of runs so that the
+ * remaining runs can fit into memory. A consolidation phase combines
+ * in-memory and spilled batches prior to the final merge to control final
+ * merge width.</li>
+ * <li>A special case occurs if no batches were spilled. In this case, the input
+ * batches are sorted in memory without merging.</li>
+ * </ul>
+ * <p>
+ * Many complex details are involved in doing the above; the details are explained
+ * in the methods of this class.
+ * <p>
+ * <h4>Configuration Options</h4>
+ * <dl>
+ * <dt>drill.exec.sort.external.spill.fs</dt>
+ * <dd>The file system (file://, hdfs://, etc.) of the spill directory.</dd>
+ * <dt>drill.exec.sort.external.spill.directories</dt>
+ * <dd>The comma delimited list of directories, on the above file
+ * system, to which to spill files in round-robin fashion. The query will
+ * fail if any one of the directories becomes full.</dt>
+ * <dt>drill.exec.sort.external.spill.file_size</dt>
+ * <dd>Target size for first-generation spill files Set this to large
+ * enough to get nice long writes, but not so large that spill directories
+ * are overwhelmed.</dd>
+ * <dt>drill.exec.sort.external.mem_limit</dt>
+ * <dd>Maximum memory to use for the in-memory buffer. (Primarily for testing.)</dd>
+ * <dt>drill.exec.sort.external.batch_limit</dt>
+ * <dd>Maximum number of batches to hold in memory. (Primarily for testing.)</dd>
+ * <dt>drill.exec.sort.external.spill.max_count</dt>
+ * <dd>Maximum number of batches to add to \u201cfirst generation\u201d files.
+ * Defaults to 0 (no limit). (Primarily for testing.)</dd>
+ * <dt>drill.exec.sort.external.spill.min_count</dt>
+ * <dd>Minimum number of batches to add to \u201cfirst generation\u201d files.
+ * Defaults to 0 (no limit). (Primarily for testing.)</dd>
+ * <dt>drill.exec.sort.external.merge_limit</dt>
+ * <dd>Sets the maximum number of runs to be merged in a single pass (limits
+ * the number of open files.)</dd>
+ * </dl>
+ * <p>
+ * The memory limit observed by this operator is the lesser of:
+ * <ul>
+ * <li>The maximum allocation allowed the allocator assigned to this batch
+ * as set by the Foreman, or</li>
+ * <li>The maximum limit configured in the mem_limit parameter above. (Primarily for
+ * testing.</li>
+ * </ul>
+ * <h4>Output</h4>
+ * It is helpful to note that the sort operator will produce one of two kinds of
+ * output batches.
+ * <ul>
+ * <li>A large output with sv4 if data is sorted in memory. The sv4 addresses
+ * the entire in-memory sort set. A selection vector remover will copy results
+ * into new batches of a size determined by that operator.</li>
+ * <li>A series of batches, without a selection vector, if the sort spills to
+ * disk. In this case, the downstream operator will still be a selection vector
+ * remover, but there is nothing for that operator to remove. Each batch is
+ * of the size set by {@link #MAX_MERGED_BATCH_SIZE}.</li>
+ * </ul>
+ * Note that, even in the in-memory sort case, this operator could do the copying
+ * to eliminate the extra selection vector remover. That is left as an exercise
+ * for another time.
+ * <h4>Logging</h4>
+ * Logging in this operator serves two purposes:
+ * <li>
+ * <ul>
+ * <li>Normal diagnostic information.</li>
+ * <li>Capturing the essence of the operator functionality for analysis in unit
+ * tests.</li>
+ * </ul>
+ * Test logging is designed to capture key events and timings. Take care
+ * when changing or removing log messages as you may need to adjust unit tests
+ * accordingly.
+ */
+
+public class ExternalSortBatch extends AbstractRecordBatch<ExternalSort> {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExternalSortBatch.class);
+ protected static final ControlsInjector injector = ControlsInjectorFactory.getInjector(ExternalSortBatch.class);
+
+ /**
+ * Smallest allowed output batch size. The smallest output batch
+ * created even under constrained memory conditions.
+ */
+ private static final int MIN_MERGED_BATCH_SIZE = 256 * 1024;
+
+ /**
+ * The preferred amount of memory to set aside to output batches
+ * expressed as a ratio of available memory.
+ */
+
+ private static final float MERGE_BATCH_ALLOWANCE = 0.10F;
+
+ /**
+ * In the bizarre case where the user gave us an unrealistically low
+ * spill file size, set a floor at some bare minimum size. (Note that,
+ * at this size, big queries will create a huge number of files, which
+ * is why the configuration default is one the order of hundreds of MB.)
+ */
+
+ private static final long MIN_SPILL_FILE_SIZE = 1 * 1024 * 1024;
+
+ public static final String INTERRUPTION_AFTER_SORT = "after-sort";
+ public static final String INTERRUPTION_AFTER_SETUP = "after-setup";
+ public static final String INTERRUPTION_WHILE_SPILLING = "spilling";
+ public static final long DEFAULT_SPILL_BATCH_SIZE = 8L * 1024 * 1024;
+ public static final long MIN_SPILL_BATCH_SIZE = 256 * 1024;
+
+ private final RecordBatch incoming;
+
+ /**
+ * Memory allocator for this operator itself. Incoming batches are
+ * transferred into this allocator. Intermediate batches used during
+ * merge also reside here.
+ */
+
+ private final BufferAllocator allocator;
+
+ /**
+ * Schema of batches that this operator produces.
+ */
+
+ private BatchSchema schema;
+
+ private LinkedList<BatchGroup.InputBatch> bufferedBatches = Lists.newLinkedList();
+ private LinkedList<BatchGroup.SpilledRun> spilledRuns = Lists.newLinkedList();
+ private SelectionVector4 sv4;
+
+ /**
+ * The number of records to add to each output batch sent to the
+ * downstream operator or spilled to disk.
+ */
+
+ private int mergeBatchRowCount;
+ private int peakNumBatches = -1;
+
+ private long memoryLimit;
+
+ /**
+ * Iterates over the final, sorted results.
+ */
+
+ private SortResults resultsIterator;
+
+ /**
+ * Manages the set of spill directories and files.
+ */
+
+ private final SpillSet spillSet;
+
+ /**
+ * Manages the copier used to merge a collection of batches into
+ * a new set of batches.
+ */
+
+ private final CopierHolder copierHolder;
+
+ private enum SortState { START, LOAD, DELIVER, DONE }
+ private SortState sortState = SortState.START;
+ private int inputRecordCount = 0;
+ private int inputBatchCount = 0; // total number of batches received so far
+ private final OperatorCodeGenerator opCodeGen;
+
+ /**
+ * Estimated size of the records for this query, updated on each
+ * new batch received from upstream.
+ */
+
+ private int estimatedRowWidth;
+
+ /**
+ * Size of the merge batches that this operator produces. Generally
+ * the same as the merge batch size, unless low memory forces a smaller
+ * value.
+ */
+
+ private long targetMergeBatchSize;
+
+ /**
+ * Estimate of the input batch size based on the largest batch seen
+ * thus far.
+ */
+ private long estimatedInputBatchSize;
+
+ /**
+ * Maximum number of batches to hold in memory.
+ * (Primarily for testing.)
+ */
+
+ private int bufferedBatchLimit;
+ private int mergeLimit;
+ private long spillFileSize;
+ private long minimumBufferSpace;
+
+ /**
+ * Minimum memory level before spilling occurs. That is, we can buffer input
+ * batches in memory until we are down to the level given by the spill point.
+ */
+
+ private long spillPoint;
+ private long mergeMemoryPool;
+ private long preferredMergeBatchSize;
+ private long bufferMemoryPool;
+ private boolean hasOversizeCols;
+ private long totalInputBytes;
+ private Long spillBatchSize;
+ private int maxDensity;
+
+ /**
+ * Estimated number of rows that fit into a single spill batch.
+ */
+
+ private int spillBatchRowCount;
+
+ // WARNING: The enum here is used within this class. But, the members of
+ // this enum MUST match those in the (unmanaged) ExternalSortBatch since
+ // that is the enum used in the UI to display metrics for the query profile.
+
+ public enum Metric implements MetricDef {
+ SPILL_COUNT, // number of times operator spilled to disk
+ RETIRED1, // Was: peak value for totalSizeInMemory
+ // But operator already provides this value
+ PEAK_BATCHES_IN_MEMORY, // maximum number of batches kept in memory
+ MERGE_COUNT, // Number of second+ generation merges
+ MIN_BUFFER, // Minimum memory level observed in operation.
+ SPILL_MB; // Number of MB of data spilled to disk. This
+ // amount is first written, then later re-read.
+ // So, disk I/O is twice this amount.
+
+ @Override
+ public int metricId() {
+ return ordinal();
+ }
+ }
+
+ /**
+ * Iterates over the final sorted results. Implemented differently
+ * depending on whether the results are in-memory or spilled to
+ * disk.
+ */
+
+ public interface SortResults {
+ boolean next();
+ void close();
+ int getBatchCount();
+ int getRecordCount();
+ }
+
+ public ExternalSortBatch(ExternalSort popConfig, FragmentContext context, RecordBatch incoming) {
+ super(popConfig, context, true);
+ this.incoming = incoming;
+ allocator = oContext.getAllocator();
+ opCodeGen = new OperatorCodeGenerator(context, popConfig);
+
+ spillSet = new SpillSet(context, popConfig);
+ copierHolder = new CopierHolder(context, allocator, opCodeGen);
+ configure(context.getConfig());
+ }
+
+ private void configure(DrillConfig config) {
+
+ // The maximum memory this operator can use as set by the
+ // operator definition (propagated to the allocator.)
+
+ memoryLimit = allocator.getLimit();
+
+ // Optional configured memory limit, typically used only for testing.
+
+ long configLimit = config.getBytes(ExecConstants.EXTERNAL_SORT_MAX_MEMORY);
+ if (configLimit > 0) {
+ memoryLimit = Math.min(memoryLimit, configLimit);
+ }
+
+ // Optional limit on the number of buffered in-memory batches.
+ // 0 means no limit. Used primarily for testing. Must allow at least two
+ // batches or no merging can occur.
+
+ bufferedBatchLimit = getConfigLimit(config, ExecConstants.EXTERNAL_SORT_BATCH_LIMIT, Integer.MAX_VALUE, 2);
+
+ // Optional limit on the number of spilled runs to merge in a single
+ // pass. Limits the number of open file handles. Must allow at least
+ // two batches to merge to make progress.
+
+ mergeLimit = getConfigLimit(config, ExecConstants.EXTERNAL_SORT_MERGE_LIMIT, Integer.MAX_VALUE, 2);
+
+ // Limits the size of first-generation spill files.
+
+ spillFileSize = config.getBytes(ExecConstants.EXTERNAL_SORT_SPILL_FILE_SIZE);
+
+ // Ensure the size is reasonable.
+
+ spillFileSize = Math.max(spillFileSize, MIN_SPILL_FILE_SIZE);
+
+ // The spill batch size. This is a critical setting for performance.
+ // Set too large and the ratio between memory and input data sizes becomes
+ // small. Set too small and disk seek times dominate performance.
+
+ spillBatchSize = config.getBytes(ExecConstants.EXTERNAL_SORT_SPILL_BATCH_SIZE);
+ spillBatchSize = Math.max(spillBatchSize, MIN_SPILL_BATCH_SIZE);
+
+ // Set the target output batch size. Use the maximum size, but only if
+ // this represents less than 10% of available memory. Otherwise, use 10%
+ // of memory, but no smaller than the minimum size. In any event, an
+ // output batch can contain no fewer than a single record.
+
+ preferredMergeBatchSize = config.getBytes(ExecConstants.EXTERNAL_SORT_MERGE_BATCH_SIZE);
+ long maxAllowance = (long) (memoryLimit * MERGE_BATCH_ALLOWANCE);
+ preferredMergeBatchSize = Math.min(maxAllowance, preferredMergeBatchSize);
+ preferredMergeBatchSize = Math.max(preferredMergeBatchSize, MIN_MERGED_BATCH_SIZE);
+
+ logger.debug("Config: memory limit = {}, batch limit = {}, " +
+ "spill file size = {}, batch size = {}, merge limit = {}, merge batch size = {}",
+ memoryLimit, bufferedBatchLimit, spillFileSize, spillBatchSize, mergeLimit,
+ preferredMergeBatchSize);
+ }
+
+ private int getConfigLimit(DrillConfig config, String paramName, int valueIfZero, int minValue) {
+ int limit = config.getInt(paramName);
+ if (limit > 0) {
+ limit = Math.max(limit, minValue);
+ } else {
+ limit = valueIfZero;
+ }
+ return limit;
+ }
+
+ @Override
+ public int getRecordCount() {
+ if (sv4 != null) {
+ return sv4.getCount();
+ }
+ return container.getRecordCount();
+ }
+
+ @Override
+ public SelectionVector4 getSelectionVector4() {
+ return sv4;
+ }
+
+ private void closeBatchGroups(Collection<? extends BatchGroup> groups) {
+ for (BatchGroup group: groups) {
+ try {
+ group.close();
+ } catch (Exception e) {
+ // collect all failure and make sure to cleanup all remaining batches
+ // Originally we would have thrown a RuntimeException that would propagate to FragmentExecutor.closeOutResources()
+ // where it would have been passed to context.fail()
+ // passing the exception directly to context.fail(e) will let the cleanup process continue instead of stopping
+ // right away, this will also make sure we collect any additional exception we may get while cleaning up
+ context.fail(e);
+ }
+ }
+ }
+
+ /**
+ * Called by {@link AbstractRecordBatch} as a fast-path to obtain
+ * the first record batch and setup the schema of this batch in order
+ * to quickly return the schema to the client. Note that this method
+ * fetches the first batch from upstream which will be waiting for
+ * us the first time that {@link #innerNext()} is called.
+ */
+
+ @Override
+ public void buildSchema() {
+ IterOutcome outcome = next(incoming);
+ switch (outcome) {
+ case OK:
+ case OK_NEW_SCHEMA:
+ for (VectorWrapper<?> w : incoming) {
+ @SuppressWarnings("resource")
+ ValueVector v = container.addOrGet(w.getField());
+ if (v instanceof AbstractContainerVector) {
+ w.getValueVector().makeTransferPair(v); // Can we remove this hack?
+ v.clear();
+ }
+ v.allocateNew(); // Can we remove this? - SVR fails with NPE (TODO)
+ }
+ container.buildSchema(SelectionVectorMode.NONE);
+ container.setRecordCount(0);
+ break;
+ case STOP:
+ state = BatchState.STOP;
+ break;
+ case OUT_OF_MEMORY:
+ state = BatchState.OUT_OF_MEMORY;
+ break;
+ case NONE:
+ state = BatchState.DONE;
+ break;
+ default:
+ throw new IllegalStateException("Unexpected iter outcome: " + outcome);
+ }
+ }
+
+ /**
+ * Process each request for a batch. The first request retrieves
+ * all the incoming batches and sorts them, optionally spilling to
+ * disk as needed. Subsequent calls retrieve the sorted results in
+ * fixed-size batches.
+ */
+
+ @Override
+ public IterOutcome innerNext() {
+ switch (sortState) {
+ case DONE:
+ return IterOutcome.NONE;
+ case START:
+ case LOAD:
+ return load();
+ case DELIVER:
+ return nextOutputBatch();
+ default:
+ throw new IllegalStateException("Unexpected sort state: " + sortState);
+ }
+ }
+
+ private IterOutcome nextOutputBatch() {
+ if (resultsIterator.next()) {
+ return IterOutcome.OK;
+ } else {
+ logger.trace("Deliver phase complete: Returned {} batches, {} records",
+ resultsIterator.getBatchCount(), resultsIterator.getRecordCount());
+ sortState = SortState.DONE;
+ return IterOutcome.NONE;
+ }
+ }
+
+ /**
+ * Load and process a single batch, handling schema changes. In general, the
+ * external sort accepts only one schema.
+ *
+ * @return return code depending on the amount of data read from upstream
+ */
+
+ private IterOutcome loadBatch() {
+
+ // If this is the very first batch, then AbstractRecordBatch
+ // already loaded it for us in buildSchema().
+
+ IterOutcome upstream;
+ if (sortState == SortState.START) {
+ sortState = SortState.LOAD;
+ upstream = IterOutcome.OK_NEW_SCHEMA;
+ } else {
+ upstream = next(incoming);
+ }
+ switch (upstream) {
+ case NONE:
+ case STOP:
+ return upstream;
+ case OK_NEW_SCHEMA:
+ case OK:
+ setupSchema(upstream);
+
+ // Add the batch to the in-memory generation, spilling if
+ // needed.
+
+ processBatch();
+ break;
+ case OUT_OF_MEMORY:
+
+ // Note: it is highly doubtful that this code actually works. It
+ // requires that the upstream batches got to a safe place to run
+ // out of memory and that no work as in-flight and thus abandoned.
+ // Consider removing this case once resource management is in place.
+
+ logger.debug("received OUT_OF_MEMORY, trying to spill");
+ if (bufferedBatches.size() > 2) {
+ spillFromMemory();
+ } else {
+ logger.debug("not enough batches to spill, sending OUT_OF_MEMORY downstream");
+ return IterOutcome.OUT_OF_MEMORY;
+ }
+ break;
+ default:
+ throw new IllegalStateException("Unexpected iter outcome: " + upstream);
+ }
+ return IterOutcome.OK;
+ }
+
+ /**
+ * Load the results and sort them. May bail out early if an exceptional
+ * condition is passed up from the input batch.
+ *
+ * @return return code: OK_NEW_SCHEMA if rows were sorted,
+ * NONE if no rows
+ */
+
+ private IterOutcome load() {
+ logger.trace("Start of load phase");
+
+ // Clear the temporary container created by
+ // buildSchema().
+
+ container.clear();
+
+ // Loop over all input batches
+
+ for (;;) {
+ IterOutcome result = loadBatch();
+
+ // None means all batches have been read.
+
+ if (result == IterOutcome.NONE) {
+ break; }
+
+ // Any outcome other than OK means something went wrong.
+
+ if (result != IterOutcome.OK) {
+ return result; }
+ }
+
+ // Anything to actually sort?
+
+ if (inputRecordCount == 0) {
+ sortState = SortState.DONE;
+ return IterOutcome.NONE;
+ }
+ logger.debug("Completed load phase: read {} batches, spilled {} times, total input bytes: {}",
+ inputBatchCount, spilledRuns.size(), totalInputBytes);
+
+ // Do the merge of the loaded batches. The merge can be done entirely in memory if
+ // the results fit; else we have to do a disk-based merge of
+ // pre-sorted spilled batches.
+
+ if (canUseMemoryMerge()) {
+ return sortInMemory();
+ } else {
+ return mergeSpilledRuns();
+ }
+ }
+
+ /**
+ * All data has been read from the upstream batch. Determine if we
+ * can use a fast in-memory sort, or must use a merge (which typically,
+ * but not always, involves spilled batches.)
+ *
+ * @return whether sufficient resources exist to do an in-memory sort
+ * if all batches are still in memory
+ */
+
+ private boolean canUseMemoryMerge() {
+ if (spillSet.hasSpilled()) { return false; }
+
+ // Do we have enough memory for MSorter (the in-memory sorter)?
+
+ long allocMem = allocator.getAllocatedMemory();
+ long availableMem = memoryLimit - allocMem;
+ long neededForInMemorySort = MSortTemplate.memoryNeeded(inputRecordCount);
+ if (availableMem < neededForInMemorySort) { return false; }
+
+ // Make sure we don't exceed the maximum number of batches SV4 can address.
+
+ if (bufferedBatches.size() > Character.MAX_VALUE) { return false; }
+
+ // We can do an in-memory merge.
+
+ return true;
+ }
+
+ /**
+ * Handle a new schema from upstream. The ESB is quite limited in its ability
+ * to handle schema changes.
+ *
+ * @param upstream the status code from upstream: either OK or OK_NEW_SCHEMA
+ */
+
+ private void setupSchema(IterOutcome upstream) {
+
+ // First batch: we won't have a schema.
+
+ if (schema == null) {
+ schema = incoming.getSchema();
+
+ // Subsequent batches, nothing to do if same schema.
+
+ } else if (upstream == IterOutcome.OK) {
+ return;
+
+ // Only change in the case that the schema truly changes. Artificial schema changes are ignored.
+
+ } else if (incoming.getSchema().equals(schema)) {
+ return;
+ } else if (unionTypeEnabled) {
+ schema = SchemaUtil.mergeSchemas(schema, incoming.getSchema());
+
+ // New schema: must generate a new sorter and copier.
+
+ opCodeGen.setSchema(schema);
+ } else {
+ throw UserException.unsupportedError()
+ .message("Schema changes not supported in External Sort. Please enable Union type.")
+ .build(logger);
+ }
+
+ // Coerce all existing batches to the new schema.
+
+ for (BatchGroup b : bufferedBatches) {
+// System.out.println("Before: " + allocator.getAllocatedMemory()); // Debug only
+ b.setSchema(schema);
+// System.out.println("After: " + allocator.getAllocatedMemory()); // Debug only
+ }
+ for (BatchGroup b : spilledRuns) {
+ b.setSchema(schema);
+ }
+ }
+
+ /**
+ * Convert an incoming batch into the agree-upon format. (Also seems to
+ * make a persistent shallow copy of the batch saved until we are ready
+ * to sort or spill.)
+ *
+ * @return the converted batch, or null if the incoming batch is empty
+ */
+
+ private VectorContainer convertBatch() {
+
+ // Must accept the batch even if no records. Then clear
+ // the vectors to release memory since we won't do any
+ // further processing with the empty batch.
+
+ VectorContainer convertedBatch = SchemaUtil.coerceContainer(incoming, schema, oContext);
+ if (incoming.getRecordCount() == 0) {
+ for (VectorWrapper<?> w : convertedBatch) {
+ w.clear();
+ }
+ return null;
+ }
+ return convertedBatch;
+ }
+
+ private SelectionVector2 makeSelectionVector() {
+ if (incoming.getSchema().getSelectionVectorMode() == BatchSchema.SelectionVectorMode.TWO_BYTE) {
+ return incoming.getSelectionVector2().clone();
+ } else {
+ return newSV2();
+ }
+ }
+
+ /**
+ * Process the converted incoming batch by adding it to the in-memory store
+ * of data, or spilling data to disk when necessary.
+ */
+
+ @SuppressWarnings("resource")
+ private void processBatch() {
+
+ // Skip empty batches (such as the first one.)
+
+ if (incoming.getRecordCount() == 0) {
+ return;
+ }
+
+ // Determine actual sizes of the incoming batch before taking
+ // ownership. Allows us to figure out if we need to spill first,
+ // to avoid overflowing memory simply due to ownership transfer.
+
+ RecordBatchSizer sizer = analyzeIncomingBatch();
+
+ // The heart of the external sort operator: spill to disk when
+ // the in-memory generation exceeds the allowed memory limit.
+ // Preemptively spill BEFORE accepting the new batch into our memory
+ // pool. The allocator will throw an OOM exception if we accept the
+ // batch when we are near the limit - despite the fact that the batch
+ // is already in memory and no new memory is allocated during the transfer.
+
+ if ( isSpillNeeded(sizer.actualSize())) {
+ spillFromMemory();
+ }
+
+ // Sanity check. We should now be above the spill point.
+
+ long startMem = allocator.getAllocatedMemory();
+ if (memoryLimit - startMem < spillPoint) {
+ logger.error( "ERROR: Failed to spill below the spill point. Spill point = {}, free memory = {}",
+ spillPoint, memoryLimit - startMem);
+ }
+
+ // Convert the incoming batch to the agreed-upon schema.
+ // No converted batch means we got an empty input batch.
+ // Converting the batch transfers memory ownership to our
+ // allocator. This gives a round-about way to learn the batch
+ // size: check the before and after memory levels, then use
+ // the difference as the batch size, in bytes.
+
+ VectorContainer convertedBatch = convertBatch();
+ if (convertedBatch == null) {
+ return;
+ }
+
+ SelectionVector2 sv2 = makeSelectionVector();
+
+ // Compute batch size, including allocation of an sv2.
+
+ long endMem = allocator.getAllocatedMemory();
+ long batchSize = endMem - startMem;
+ int count = sv2.getCount();
+ inputRecordCount += count;
+ inputBatchCount++;
+ totalInputBytes += sizer.actualSize();
+
+ // Update the minimum buffer space metric.
+
+ if (minimumBufferSpace == 0) {
+ minimumBufferSpace = endMem;
+ } else {
+ minimumBufferSpace = Math.min(minimumBufferSpace, endMem);
+ }
+ stats.setLongStat(Metric.MIN_BUFFER, minimumBufferSpace);
+
+ // Update the size based on the actual record count, not
+ // the effective count as given by the selection vector
+ // (which may exclude some records due to filtering.)
+
+ updateMemoryEstimates(batchSize, sizer);
+
+ // Sort the incoming batch using either the original selection vector,
+ // or a new one created here.
+
+ SingleBatchSorter sorter;
+ sorter = opCodeGen.getSorter(convertedBatch);
+ try {
+ sorter.setup(context, sv2, convertedBatch);
+ } catch (SchemaChangeException e) {
+ convertedBatch.clear();
+ throw UserException.unsupportedError(e)
+ .message("Unexpected schema change.")
+ .build(logger);
+ }
+ try {
+ sorter.sort(sv2);
+ } catch (SchemaChangeException e) {
+ convertedBatch.clear();
+ throw UserException.unsupportedError(e)
+ .message("Unexpected schema change.")
+ .build(logger);
+ }
+ RecordBatchData rbd = new RecordBatchData(convertedBatch, allocator);
+ try {
+ rbd.setSv2(sv2);
+ bufferedBatches.add(new BatchGroup.InputBatch(rbd.getContainer(), rbd.getSv2(), oContext, batchSize));
+ if (peakNumBatches < bufferedBatches.size()) {
+ peakNumBatches = bufferedBatches.size();
+ stats.setLongStat(Metric.PEAK_BATCHES_IN_MEMORY, peakNumBatches);
+ }
+
+ } catch (Throwable t) {
+ rbd.clear();
+ throw t;
+ }
+ }
+
+ /**
+ * Scan the vectors in the incoming batch to determine batch size and if
+ * any oversize columns exist. (Oversize columns cause memory fragmentation.)
+ *
+ * @return an analysis of the incoming batch
+ */
+
+ private RecordBatchSizer analyzeIncomingBatch() {
+ RecordBatchSizer sizer = new RecordBatchSizer(incoming);
+ sizer.applySv2();
+ if (! hasOversizeCols) {
+ hasOversizeCols = sizer.checkOversizeCols();
+ }
+ if (inputBatchCount == 0) {
+ logger.debug("{}", sizer.toString());
+ }
+ return sizer;
+ }
+
+ /**
+ * Update the data-driven memory use numbers including:
+ * <ul>
+ * <li>The average size of incoming records.</li>
+ * <li>The estimated spill and output batch size.</li>
+ * <li>The estimated number of average-size records per
+ * spill and output batch.</li>
+ * <li>The amount of memory set aside to hold the incoming
+ * batches before spilling starts.</li>
+ * </ul>
+ *
+ * @param actualBatchSize the overall size of the current batch received from
+ * upstream
+ * @param actualRecordCount the number of actual (not filtered) records in
+ * that upstream batch
+ */
+
+ private void updateMemoryEstimates(long memoryDelta, RecordBatchSizer sizer) {
+ long actualBatchSize = sizer.actualSize();
+ int actualRecordCount = sizer.rowCount();
+
+ if (actualBatchSize < memoryDelta) {
+ logger.debug("Memory delta: {}, actual batch size: {}, Diff: {}",
+ memoryDelta, actualBatchSize, memoryDelta - actualBatchSize);
+ }
+
+ // The record count should never be zero, but better safe than sorry...
+
+ if (actualRecordCount == 0) {
+ return; }
+
+ // If the vector is less than 75% full, just ignore it, except in the
+ // unfortunate case where it is the first batch. Low-density batches generally
+ // occur only at the end of a file or at the end of a DFS block. In such a
+ // case, we will continue to rely on estimates created on previous, high-
+ // density batches.
+ // We actually track the max density seen, and compare to 75% of that since
+ // Parquet produces very low density record batches.
+
+ if (sizer.getAvgDensity() < maxDensity * 0.75) {
+ logger.debug("Saw low density batch. Density: {}", sizer.getAvgDensity());
+ return;
+ }
+ maxDensity = Math.max(maxDensity, sizer.getAvgDensity());
+
+ // We know the batch size and number of records. Use that to estimate
+ // the average record size. Since a typical batch has many records,
+ // the average size is a fairly good estimator. Note that the batch
+ // size includes not just the actual vector data, but any unused space
+ // resulting from power-of-two allocation. This means that we don't
+ // have to do size adjustments for input batches as we will do below
+ // when estimating the size of other objects.
+
+ int batchRowWidth = sizer.netRowWidth();
+
+ // Record sizes may vary across batches. To be conservative, use
+ // the largest size observed from incoming batches.
+
+ int origRowEstimate = estimatedRowWidth;
+ estimatedRowWidth = Math.max(estimatedRowWidth, batchRowWidth);
+
+ // Maintain an estimate of the incoming batch size: the largest
+ // batch yet seen. Used to reserve memory for the next incoming
+ // batch.
+
+ long origInputBatchSize = estimatedInputBatchSize;
+ estimatedInputBatchSize = Math.max(estimatedInputBatchSize, actualBatchSize);
+
+ // Go no further if nothing changed.
+
+ if (estimatedRowWidth == origRowEstimate && estimatedInputBatchSize == origInputBatchSize) {
+ return; }
+
+ // Estimate the total size of each incoming batch plus sv2. Note that, due
+ // to power-of-two rounding, the allocated size might be twice the data size.
+
+ long estimatedInputSize = estimatedInputBatchSize + 4 * actualRecordCount;
+
+ // Determine the number of records to spill per spill batch. The goal is to
+ // spill batches of either 64K records, or as many records as fit into the
+ // amount of memory dedicated to each spill batch, whichever is less.
+
+ spillBatchRowCount = (int) Math.max(1, spillBatchSize / estimatedRowWidth);
+ spillBatchRowCount = Math.min(spillBatchRowCount, Character.MAX_VALUE);
+
+ // Determine the number of records per batch per merge step. The goal is to
+ // merge batches of either 64K records, or as many records as fit into the
+ // amount of memory dedicated to each merge batch, whichever is less.
+
+ targetMergeBatchSize = preferredMergeBatchSize;
+ mergeBatchRowCount = (int) Math.max(1, targetMergeBatchSize / estimatedRowWidth);
+ mergeBatchRowCount = Math.min(mergeBatchRowCount, Character.MAX_VALUE);
+
+ // Determine the minimum memory needed for spilling. Spilling is done just
+ // before accepting a batch, so we must spill if we don't have room for a
+ // (worst case) input batch. To spill, we need room for the output batch created
+ // by merging the batches already in memory. Double this to allow for power-of-two
+ // memory allocations.
+
+ spillPoint = estimatedInputBatchSize + 2 * spillBatchSize;
+
+ // The merge memory pool assumes we can spill all input batches. To make
+ // progress, we must have at least two merge batches (same size as an output
+ // batch) and one output batch. Again, double to allow for power-of-two
+ // allocation and add one for a margin of error.
+
+ int minMergeBatches = 2 * 3 + 1;
+ long minMergeMemory = minMergeBatches * targetMergeBatchSize;
+
+ // If we are in a low-memory condition, then we might not have room for the
+ // default output batch size. In that case, pick a smaller size.
+
+ long minMemory = Math.max(spillPoint, minMergeMemory);
+ if (minMemory > memoryLimit) {
+
+ // Figure out the minimum output batch size based on memory, but can't be
+ // any smaller than the defined minimum.
+
+ targetMergeBatchSize = Math.max(MIN_MERGED_BATCH_SIZE, memoryLimit / minMergeBatches);
+
+ // Regardless of anything else, the batch must hold at least one
+ // complete row.
+
+ targetMergeBatchSize = Math.max(estimatedRowWidth, targetMergeBatchSize);
+ spillPoint = estimatedInputBatchSize + 2 * spillBatchSize;
+ minMergeMemory = minMergeBatches * targetMergeBatchSize;
+ }
+
+ // Determine the minimum total memory we would need to receive two input
+ // batches (the minimum needed to make progress) and the allowance for the
+ // output batch.
+
+ long minLoadMemory = spillPoint + estimatedInputSize;
+
+ // Determine how much memory can be used to hold in-memory batches of spilled
+ // runs when reading from disk.
+
+ bufferMemoryPool = memoryLimit - spillPoint;
+ mergeMemoryPool = Math.max(minMergeMemory,
+ (long) ((memoryLimit - 3 * targetMergeBatchSize) * 0.95));
+
+ // Sanity check: if we've been given too little memory to make progress,
+ // issue a warning but proceed anyway. Should only occur if something is
+ // configured terribly wrong.
+
+ long minMemoryNeeds = Math.max(minLoadMemory, minMergeMemory);
+ if (minMemoryNeeds > memoryLimit) {
+ logger.warn("Potential memory overflow! " +
+ "Minumum needed = {} bytes, actual available = {} bytes",
+ minMemoryNeeds, memoryLimit);
+ }
+
+ // Log the calculated values. Turn this on if things seem amiss.
+ // Message will appear only when the values change.
+
+ logger.debug("Memory Estimates: record size = {} bytes; input batch = {} bytes, {} records; " +
+ "merge batch size = {} bytes, {} records; " +
+ "output batch size = {} bytes, {} records; " +
+ "Available memory: {}, spill point = {}, min. merge memory = {}",
+ estimatedRowWidth, estimatedInputBatchSize, actualRecordCount,
+ spillBatchSize, spillBatchRowCount,
+ targetMergeBatchSize, mergeBatchRowCount,
+ memoryLimit, spillPoint, minMergeMemory);
+ }
+
+ /**
+ * Determine if spill is needed before receiving the new record batch.
+ * Spilling is driven purely by memory availability (and an optional
+ * batch limit for testing.)
+ *
+ * @return true if spilling is needed, false otherwise
+ */
+
+ private boolean isSpillNeeded(int incomingSize) {
+
+ // Can't spill if less than two batches else the merge
+ // can't make progress.
+
+ if (bufferedBatches.size() < 2) {
+ return false; }
+
+ // Must spill if we are below the spill point (the amount of memory
+ // needed to do the minimal spill.)
+
+ if (allocator.getAllocatedMemory() + incomingSize >= bufferMemoryPool) {
+ return true; }
+
+ // For test purposes, configuration may have set a limit on the number of
+ // batches in memory. Spill if we exceed this limit. (By default the number
+ // of in-memory batches is unlimited.)
+
+ return bufferedBatches.size() > bufferedBatchLimit;
+ }
+
+ /**
+ * Perform an in-memory sort of the buffered batches. Obviously can
+ * be used only for the non-spilling case.
+ *
+ * @return DONE if no rows, OK_NEW_SCHEMA if at least one row
+ */
+
+ private IterOutcome sortInMemory() {
+ logger.info("Starting in-memory sort. Batches = {}, Records = {}, Memory = {}",
+ bufferedBatches.size(), inputRecordCount, allocator.getAllocatedMemory());
+
+ // Note the difference between how we handle batches here and in the spill/merge
+ // case. In the spill/merge case, this class decides on the batch size to send
+ // downstream. However, in the in-memory case, we must pass along all batches
+ // in a single SV4. Attempts to do paging will result in errors. In the memory
+ // merge case, the downstream Selection Vector Remover will split the one
+ // big SV4 into multiple smaller batches to send further downstream.
+
+ // If the sort fails or is empty, clean up here. Otherwise, cleanup is done
+ // by closing the resultsIterator after all results are returned downstream.
+
+ MergeSort memoryMerge = new MergeSort(context, allocator, opCodeGen);
+ try {
+ sv4 = memoryMerge.merge(bufferedBatches, this, container);
+ if (sv4 == null) {
+ sortState = SortState.DONE;
+ return IterOutcome.STOP;
+ } else {
+ logger.info("Completed in-memory sort. Memory = {}",
+ allocator.getAllocatedMemory());
+ resultsIterator = memoryMerge;
+ memoryMerge = null;
+ sortState = SortState.DELIVER;
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+ } finally {
+ if (memoryMerge != null) {
+ memoryMerge.close();
+ }
+ }
+ }
+
+ /**
+ * Perform merging of (typically spilled) batches. First consolidates batches
+ * as needed, then performs a final merge that is read one batch at a time
+ * to deliver batches to the downstream operator.
+ *
+ * @return always returns OK_NEW_SCHEMA
+ */
+
+ private IterOutcome mergeSpilledRuns() {
+ logger.info("Starting consolidate phase. Batches = {}, Records = {}, Memory = {}, In-memory batches {}, spilled runs {}",
+ inputBatchCount, inputRecordCount, allocator.getAllocatedMemory(),
+ bufferedBatches.size(), spilledRuns.size());
+
+ // Consolidate batches to a number that can be merged in
+ // a single last pass.
+
+ int mergeCount = 0;
+ while (consolidateBatches()) {
+ mergeCount++;
+ }
+ stats.addLongStat(Metric.MERGE_COUNT, mergeCount);
+
+ // Merge in-memory batches and spilled runs for the final merge.
+
+ List<BatchGroup> allBatches = new LinkedList<>();
+ allBatches.addAll(bufferedBatches);
+ bufferedBatches.clear();
+ allBatches.addAll(spilledRuns);
+ spilledRuns.clear();
+
+ logger.info("Starting merge phase. Runs = {}, Alloc. memory = {}", allBatches.size(), allocator.getAllocatedMemory());
+
+ // Do the final merge as a results iterator.
+
+ CopierHolder.BatchMerger merger = copierHolder.startFinalMerge(schema, allBatches, container, mergeBatchRowCount);
+ merger.next();
+ resultsIterator = merger;
+ sortState = SortState.DELIVER;
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+
+ private boolean consolidateBatches() {
+
+ // Determine additional memory needed to hold one batch from each
+ // spilled run.
+
+ int inMemCount = bufferedBatches.size();
+ int spilledRunsCount = spilledRuns.size();
+
+ // Can't merge more than will fit into memory at one time.
+
+ int maxMergeWidth = (int) (mergeMemoryPool / targetMergeBatchSize);
+ maxMergeWidth = Math.min(mergeLimit, maxMergeWidth);
+
+ // If we can't fit all batches in memory, must spill any in-memory
+ // batches to make room for multiple spill-merge-spill cycles.
+
+ if (inMemCount > 0) {
+ if (spilledRunsCount > maxMergeWidth) {
+ spillFromMemory();
+ return true;
+ }
+
+ // If we just plain have too many batches to merge, spill some
+ // in-memory batches to reduce the burden.
+
+ if (inMemCount + spilledRunsCount > mergeLimit) {
+ spillFromMemory();
+ return true;
+ }
+
+ // If the on-disk batches and in-memory batches need more memory than
+ // is available, spill some in-memory batches.
+
+ long allocated = allocator.getAllocatedMemory();
+ long totalNeeds = spilledRunsCount * targetMergeBatchSize + allocated;
+ if (totalNeeds > mergeMemoryPool) {
+ spillFromMemory();
+ return true;
+ }
+ }
+
+ // Merge on-disk batches if we have too many.
+
+ int mergeCount = spilledRunsCount - maxMergeWidth;
+ if (mergeCount <= 0) {
+ return false;
+ }
+
+ // Must merge at least 2 batches to make progress.
+
+ mergeCount = Math.max(2, mergeCount);
+
+ // We will merge. This will create yet another spilled
+ // run. Account for that.
+
+ mergeCount += 1;
+
+ mergeCount = Math.min(mergeCount, maxMergeWidth);
+
+ // If we are going to merge, and we have batches in memory,
+ // spill them and try again. We need to do this to ensure we
+ // have adequate memory to hold the merge batches. We are into
+ // a second-generation sort/merge so there is no point in holding
+ // onto batches in memory.
+
+ if (inMemCount > 0) {
+ spillFromMemory();
+ return true;
+ }
+
+ // Do the merge, then loop to try again in case not
+ // all the target batches spilled in one go.
+
+ logger.trace("Merging {} on-disk runs, Alloc. memory = {}",
+ mergeCount, allocator.getAllocatedMemory());
+ mergeAndSpill(spilledRuns, mergeCount);
+ return true;
+ }
+
+ /**
+ * This operator has accumulated a set of sorted incoming record batches.
+ * We wish to spill some of them to disk. To do this, a "copier"
+ * merges the target batches to produce a stream of new (merged) batches
+ * which are then written to disk.
+ * <p>
+ * This method spills only half the accumulated batches
+ * minimizing unnecessary disk writes. The exact count must lie between
+ * the minimum and maximum spill counts.
+ */
+
+ private void spillFromMemory() {
+
+ // Determine the number of batches to spill to create a spill file
+ // of the desired size. The actual file size might be a bit larger
+ // or smaller than the target, which is expected.
+
+ long estSize = 0;
+ int spillCount = 0;
+ for (InputBatch batch : bufferedBatches) {
+ estSize += batch.getDataSize();
+ if (estSize > spillFileSize) {
+ break; }
+ spillCount++;
+ }
+
+ // Should not happen, but just to be sure...
+
+ if (spillCount == 0) {
+ return; }
+
+ // Do the actual spill.
+
+ logger.trace("Starting spill from memory. Memory = {}, Buffered batch count = {}, Spill batch count = {}",
+ allocator.getAllocatedMemory(), bufferedBatches.size(), spillCount);
+ mergeAndSpill(bufferedBatches, spillCount);
+ }
+
+ private void mergeAndSpill(LinkedList<? extends BatchGroup> source, int count) {
+ if (count == 0) {
+ return; }
+ spilledRuns.add(doMergeAndSpill(source, count));
+ }
+
+ private BatchGroup.SpilledRun doMergeAndSpill(LinkedList<? extends BatchGroup> batchGroups, int spillCount) {
+ List<BatchGroup> batchesToSpill = Lists.newArrayList();
+ spillCount = Math.min(batchGroups.size(), spillCount);
+ assert spillCount > 0 : "Spill count to mergeAndSpill must not be zero";
+ long spillSize = 0;
+ for (int i = 0; i < spillCount; i++) {
+ @SuppressWarnings("resource")
+ BatchGroup batch = batchGroups.pollFirst();
+ assert batch != null : "Encountered a null batch during merge and spill operation";
+ batchesToSpill.add(batch);
+ spillSize += batch.getDataSize();
+ }
+
+ // Merge the selected set of matches and write them to the
+ // spill file. After each write, we release the memory associated
+ // with the just-written batch.
+
+ String outputFile = spillSet.getNextSpillFile();
+ stats.setLongStat(Metric.SPILL_COUNT, spillSet.getFileCount());
+ BatchGroup.SpilledRun newGroup = null;
+ try (AutoCloseable ignored = AutoCloseables.all(batchesToSpill);
+ CopierHolder.BatchMerger merger = copierHolder.startMerge(schema, batchesToSpill, spillBatchRowCount)) {
+ logger.trace("Merging and spilling to {}", outputFile);
+ newGroup = new BatchGroup.SpilledRun(spillSet, outputFile, oContext, spillSize);
+
+ // The copier will merge records from the buffered batches into
+ // the outputContainer up to targetRecordCount number of rows.
+ // The actual count may be less if fewer records are available.
+
+ while (merger.next()) {
+
+ // Add a new batch of records (given by merger.getOutput()) to the spill
+ // file, opening the file if not yet open, and creating the target
+ // directory if it does not yet exist.
+ //
+ // note that addBatch also clears the merger's output container
+
+ newGroup.addBatch(merger.getOutput());
+ }
+ injector.injectChecked(context.getExecutionControls(), INTERRUPTION_WHILE_SPILLING, IOException.class);
+ newGroup.closeOutputStream();
+ logger.trace("mergeAndSpill: completed, memory = {}, spilled {} records to {}",
+ allocator.getAllocatedMemory(), merger.getRecordCount(), outputFile);
+ return newGroup;
+ } catch (Throwable e) {
+ // we only need to clean up newGroup if spill failed
+ try {
+ if (newGroup != null) {
+ AutoCloseables.close(e, newGroup);
+ }
+ } catch (Throwable t) { /* close() may hit the same IO issue; just ignore */ }
+
+ // Here the merger is holding onto a partially-completed batch.
+ // It will release the memory in the close() call.
+
+ try {
+ // Rethrow so we can organize how to handle the error.
+
+ throw e;
+ }
+
+ // If error is a User Exception, just use as is.
+
+ catch (UserException ue) { throw ue; }
+ catch (Throwable ex) {
+ throw UserException.resourceError(ex)
+ .message("External Sort encountered an error while spilling to disk")
+ .build(logger);
+ }
+ }
+ }
+
+ /**
+ * Allocate and initialize the selection vector used as the sort index.
+ * Assumes that memory is available for the vector since memory management
+ * ensured space is available.
+ *
+ * @return a new, populated selection vector 2
+ */
+
+ private SelectionVector2 newSV2() {
+ SelectionVector2 sv2 = new SelectionVector2(allocator);
+ if (!sv2.allocateNewSafe(incoming.getRecordCount())) {
+ throw UserException.resourceError(new OutOfMemoryException("Unable to allocate sv2 buffer"))
+ .build(logger);
+ }
+ for (int i = 0; i < incoming.getRecordCount(); i++) {
+ sv2.setIndex(i, (char) i);
+ }
+ sv2.setRecordCount(incoming.getRecordCount());
+ return sv2;
+ }
+
+ @Override
+ public WritableBatch getWritableBatch() {
+ throw new UnsupportedOperationException("A sort batch is not writable.");
+ }
+
+ @Override
+ protected void killIncoming(boolean sendUpstream) {
+ incoming.kill(sendUpstream);
+ }
+
+ /**
+ * Extreme paranoia to avoid leaving resources unclosed in the case
+ * of an error. Since generally only the first error is of interest,
+ * we track only the first exception, not potential cascading downstream
+ * exceptions.
+ * <p>
+ * Some Drill code ends up calling close() two or more times. The code
+ * here protects itself from these undesirable semantics.
+ */
+
+ @Override
+ public void close() {
+ if (spillSet.getWriteBytes() > 0) {
+ logger.debug("End of sort. Total write bytes: {}, Total read bytes: {}",
+ spillSet.getWriteBytes(), spillSet.getWriteBytes());
+ }
+ stats.setLongStat(Metric.SPILL_MB,
+ (int) Math.round( spillSet.getWriteBytes() / 1024.0D / 1024.0 ) );
+ RuntimeException ex = null;
+ try {
+ if (bufferedBatches != null) {
+ closeBatchGroups(bufferedBatches);
+ bufferedBatches = null;
+ }
+ } catch (RuntimeException e) {
+ ex = e;
+ }
+ try {
+ if (spilledRuns != null) {
+ closeBatchGroups(spilledRuns);
+ spilledRuns = null;
+ }
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ if (sv4 != null) {
+ sv4.clear();
+ }
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ if (resultsIterator != null) {
+ resultsIterator.close();
+ resultsIterator = null;
+ }
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ copierHolder.close();
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ spillSet.close();
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ opCodeGen.close();
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+
+ // The call to super.close() clears out the output container.
+ // Doing so requires the allocator here, so it must be closed
+ // after the super call.
+
+ try {
+ super.close();
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ try {
+ allocator.close();
+ } catch (RuntimeException e) {
+ ex = (ex == null) ? e : ex;
+ }
+ if (ex != null) {
+ throw ex;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSortTemplate.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSortTemplate.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSortTemplate.java
new file mode 100644
index 0000000..31475d2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSortTemplate.java
@@ -0,0 +1,237 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.util.Queue;
+
+import javax.inject.Named;
+
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BaseAllocator;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+import org.apache.hadoop.util.IndexedSortable;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Queues;
+
+import io.netty.buffer.DrillBuf;
+
+public abstract class MSortTemplate implements MSorter, IndexedSortable {
+// private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MSortTemplate.class);
+
+ private SelectionVector4 vector4;
+ private SelectionVector4 aux;
+ @SuppressWarnings("unused")
+ private long compares;
+
+ /**
+ * Holds offsets into the SV4 of the start of each batch
+ * (sorted run.)
+ */
+
+ private Queue<Integer> runStarts = Queues.newLinkedBlockingQueue();
+ private FragmentContext context;
+
+ /**
+ * Controls the maximum size of batches exposed to downstream
+ */
+ private int desiredRecordBatchCount;
+
+ @Override
+ public void setup(final FragmentContext context, final BufferAllocator allocator, final SelectionVector4 vector4,
+ final VectorContainer hyperBatch, int outputBatchSize) throws SchemaChangeException{
+ // we pass in the local hyperBatch since that is where we'll be reading data.
+ Preconditions.checkNotNull(vector4);
+ this.vector4 = vector4.createNewWrapperCurrent();
+ this.context = context;
+ vector4.clear();
+ doSetup(context, hyperBatch, null);
+
+ // Populate the queue with the offset in the SV4 of each
+ // batch. Note that this is expensive as it requires a scan
+ // of all items to be sorted: potentially millions.
+
+ runStarts.add(0);
+ int batch = 0;
+ final int totalCount = this.vector4.getTotalCount();
+ for (int i = 0; i < totalCount; i++) {
+ final int newBatch = this.vector4.get(i) >>> 16;
+ if (newBatch == batch) {
+ continue;
+ } else if (newBatch == batch + 1) {
+ runStarts.add(i);
+ batch = newBatch;
+ } else {
+ throw new UnsupportedOperationException(String.format("Missing batch. batch: %d newBatch: %d", batch, newBatch));
+ }
+ }
+
+ // Create a temporary SV4 to hold the merged results.
+
+ @SuppressWarnings("resource")
+ final DrillBuf drillBuf = allocator.buffer(4 * totalCount);
+ desiredRecordBatchCount = Math.min(outputBatchSize, Character.MAX_VALUE);
+ desiredRecordBatchCount = Math.min(desiredRecordBatchCount, totalCount);
+ aux = new SelectionVector4(drillBuf, totalCount, desiredRecordBatchCount);
+ }
+
+ /**
+ * For given recordCount how much memory does MSorter needs for its own purpose. This is used in
+ * ExternalSortBatch to make decisions about whether to spill or not.
+ *
+ * @param recordCount
+ * @return
+ */
+ public static long memoryNeeded(final int recordCount) {
+ // We need 4 bytes (SV4) for each record.
+ // The memory allocator will round this to the next
+ // power of 2.
+
+ return BaseAllocator.nextPowerOfTwo(recordCount * 4);
+ }
+
+ /**
+ * Given two regions within the selection vector 4 (a left and a right), merge
+ * the two regions to produce a combined output region in the auxiliary
+ * selection vector.
+ *
+ * @param leftStart
+ * @param rightStart
+ * @param rightEnd
+ * @param outStart
+ * @return
+ */
+ protected int merge(final int leftStart, final int rightStart, final int rightEnd, final int outStart) {
+ int l = leftStart;
+ int r = rightStart;
+ int o = outStart;
+ while (l < rightStart && r < rightEnd) {
+ if (compare(l, r) <= 0) {
+ aux.set(o++, vector4.get(l++));
+ } else {
+ aux.set(o++, vector4.get(r++));
+ }
+ }
+ while (l < rightStart) {
+ aux.set(o++, vector4.get(l++));
+ }
+ while (r < rightEnd) {
+ aux.set(o++, vector4.get(r++));
+ }
+ assert o == outStart + (rightEnd - leftStart);
+ return o;
+ }
+
+ @Override
+ public SelectionVector4 getSV4() {
+ return vector4;
+ }
+
+ /**
+ * Sort (really, merge) a set of pre-sorted runs to produce a combined
+ * result set. Merging is done in the selection vector, record data does
+ * not move.
+ * <p>
+ * Runs are merge pairwise in multiple passes, providing performance
+ * of O(n * m * log(n)), where n = number of runs, m = number of records
+ * per run.
+ */
+
+ @Override
+ public void sort(final VectorContainer container) {
+ while (runStarts.size() > 1) {
+ final int totalCount = this.vector4.getTotalCount();
+
+ // check if we're cancelled/failed recently
+ if (!context.shouldContinue()) {
+ return; }
+
+ int outIndex = 0;
+ final Queue<Integer> newRunStarts = Queues.newLinkedBlockingQueue();
+ newRunStarts.add(outIndex);
+ final int size = runStarts.size();
+ for (int i = 0; i < size / 2; i++) {
+ final int left = runStarts.poll();
+ final int right = runStarts.poll();
+ Integer end = runStarts.peek();
+ if (end == null) {
+ end = totalCount;
+ }
+ outIndex = merge(left, right, end, outIndex);
+ if (outIndex < vector4.getTotalCount()) {
+ newRunStarts.add(outIndex);
+ }
+ }
+ if (outIndex < totalCount) {
+ copyRun(outIndex, totalCount);
+ }
+ @SuppressWarnings("resource")
+ final SelectionVector4 tmp = aux.createNewWrapperCurrent(desiredRecordBatchCount);
+ aux.clear();
+ aux = vector4.createNewWrapperCurrent(desiredRecordBatchCount);
+ vector4.clear();
+ vector4 = tmp.createNewWrapperCurrent(desiredRecordBatchCount);
+ tmp.clear();
+ runStarts = newRunStarts;
+ }
+ aux.clear();
+ }
+
+ private void copyRun(final int start, final int end) {
+ for (int i = start; i < end; i++) {
+ aux.set(i, vector4.get(i));
+ }
+ }
+
+ @Override
+ public void swap(final int sv0, final int sv1) {
+ final int tmp = vector4.get(sv0);
+ vector4.set(sv0, vector4.get(sv1));
+ vector4.set(sv1, tmp);
+ }
+
+ @Override
+ public int compare(final int leftIndex, final int rightIndex) {
+ final int sv1 = vector4.get(leftIndex);
+ final int sv2 = vector4.get(rightIndex);
+ compares++;
+ try {
+ return doEval(sv1, sv2);
+ } catch (SchemaChangeException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public void clear() {
+ if(vector4 != null) {
+ vector4.clear();
+ }
+
+ if(aux != null) {
+ aux.clear();
+ }
+ }
+
+ public abstract void doSetup(@Named("context") FragmentContext context, @Named("incoming") VectorContainer incoming, @Named("outgoing") RecordBatch outgoing) throws SchemaChangeException;
+ public abstract int doEval(@Named("leftIndex") int leftIndex, @Named("rightIndex") int rightIndex) throws SchemaChangeException;
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSorter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSorter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSorter.java
new file mode 100644
index 0000000..0d04b7e
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MSorter.java
@@ -0,0 +1,40 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import org.apache.drill.exec.compile.TemplateClassDefinition;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+
+/**
+ * In-memory sorter. Takes a list of batches as input, produces a selection
+ * vector 4, with sorted results, as output.
+ */
+
+public interface MSorter {
+ public void setup(FragmentContext context, BufferAllocator allocator, SelectionVector4 vector4, VectorContainer hyperBatch, int outputBatchSize) throws SchemaChangeException;
+ public void sort(VectorContainer container);
+ public SelectionVector4 getSV4();
+
+ public static TemplateClassDefinition<MSorter> TEMPLATE_DEFINITION = new TemplateClassDefinition<MSorter>(MSorter.class, MSortTemplate.class);
+
+ public void clear();
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSort.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSort.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSort.java
new file mode 100644
index 0000000..c3e2dbe
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/MergeSort.java
@@ -0,0 +1,167 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.util.LinkedList;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.physical.impl.sort.SortRecordBatchBuilder;
+import org.apache.drill.exec.physical.impl.xsort.managed.ExternalSortBatch.SortResults;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.record.selection.SelectionVector4;
+
+/**
+ * Wrapper around the "MSorter" (in memory merge sorter). As batches have
+ * arrived to the sort, they have been individually sorted and buffered
+ * in memory. At the completion of the sort, we detect that no batches
+ * were spilled to disk. In this case, we can merge the in-memory batches
+ * using an efficient memory-based approach implemented here.
+ * <p>
+ * Since all batches are in memory, we don't want to use the usual merge
+ * algorithm as that makes a copy of the original batches (which were read
+ * from a spill file) to produce an output batch. Instead, we want to use
+ * the in-memory batches as-is. To do this, we use a selection vector 4
+ * (SV4) as a global index into the collection of batches. The SV4 uses
+ * the upper two bytes as the batch index, and the lower two as an offset
+ * of a record within the batch.
+ * <p>
+ * The merger ("M Sorter") populates the SV4 by scanning the set of
+ * in-memory batches, searching for the one with the lowest value of the
+ * sort key. The batch number and offset are placed into the SV4. The process
+ * continues until all records from all batches have an entry in the SV4.
+ * <p>
+ * The actual implementation uses an iterative merge to perform the above
+ * efficiently.
+ * <p>
+ * A sort can only do a single merge. So, we do not attempt to share the
+ * generated class; we just generate it internally and discard it at
+ * completion of the merge.
+ */
+
+public class MergeSort implements SortResults {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MergeSort.class);
+
+ private SortRecordBatchBuilder builder;
+ private MSorter mSorter;
+ private final FragmentContext context;
+ private final BufferAllocator oAllocator;
+ private SelectionVector4 sv4;
+ private final OperatorCodeGenerator opCg;
+ private int batchCount;
+
+ public MergeSort(FragmentContext context, BufferAllocator allocator, OperatorCodeGenerator opCg) {
+ this.context = context;
+ this.oAllocator = allocator;
+ this.opCg = opCg;
+ }
+
+ /**
+ * Merge the set of in-memory batches to produce a single logical output in the given
+ * destination container, indexed by an SV4.
+ *
+ * @param batchGroups the complete set of in-memory batches
+ * @param batch the record batch (operator) for the sort operator
+ * @param destContainer the vector container for the sort operator
+ * @return the sv4 for this operator
+ */
+
+ public SelectionVector4 merge(LinkedList<BatchGroup.InputBatch> batchGroups, VectorAccessible batch,
+ VectorContainer destContainer) {
+
+ // Add the buffered batches to a collection that MSorter can use.
+ // The builder takes ownership of the batches and will release them if
+ // an error occurs.
+
+ builder = new SortRecordBatchBuilder(oAllocator);
+ for (BatchGroup.InputBatch group : batchGroups) {
+ RecordBatchData rbd = new RecordBatchData(group.getContainer(), oAllocator);
+ rbd.setSv2(group.getSv2());
+ builder.add(rbd);
+ }
+ batchGroups.clear();
+
+ // Generate the msorter.
+
+ try {
+ builder.build(context, destContainer);
+ sv4 = builder.getSv4();
+ mSorter = opCg.createNewMSorter(batch);
+ mSorter.setup(context, oAllocator, sv4, destContainer, sv4.getCount());
+ } catch (SchemaChangeException e) {
+ throw UserException.unsupportedError(e)
+ .message("Unexpected schema change - likely code error.")
+ .build(logger);
+ }
+
+ // For testing memory-leaks, inject exception after mSorter finishes setup
+ ExternalSortBatch.injector.injectUnchecked(context.getExecutionControls(), ExternalSortBatch.INTERRUPTION_AFTER_SETUP);
+ mSorter.sort(destContainer);
+
+ // sort may have prematurely exited due to should continue returning false.
+ if (!context.shouldContinue()) {
+ return null;
+ }
+
+ // For testing memory-leak purpose, inject exception after mSorter finishes sorting
+ ExternalSortBatch.injector.injectUnchecked(context.getExecutionControls(), ExternalSortBatch.INTERRUPTION_AFTER_SORT);
+ sv4 = mSorter.getSV4();
+
+ destContainer.buildSchema(SelectionVectorMode.FOUR_BYTE);
+ return sv4;
+ }
+
+ /**
+ * The SV4 provides a built-in iterator that returns a virtual set of record
+ * batches so that the downstream operator need not consume the entire set
+ * of accumulated batches in a single step.
+ */
+
+ @Override
+ public boolean next() {
+ boolean more = sv4.next();
+ if (more) { batchCount++; }
+ return more;
+ }
+
+ @Override
+ public void close() {
+ if (builder != null) {
+ builder.clear();
+ builder.close();
+ }
+ if (mSorter != null) {
+ mSorter.clear();
+ }
+ }
+
+ @Override
+ public int getBatchCount() {
+ return batchCount;
+ }
+
+ @Override
+ public int getRecordCount() {
+ return sv4.getTotalCount();
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/OperatorCodeGenerator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/OperatorCodeGenerator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/OperatorCodeGenerator.java
new file mode 100644
index 0000000..57846db
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/OperatorCodeGenerator.java
@@ -0,0 +1,259 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.calcite.rel.RelFieldCollation.Direction;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.ErrorCollector;
+import org.apache.drill.common.expression.ErrorCollectorImpl;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.logical.data.Order.Ordering;
+import org.apache.drill.exec.compile.sig.GeneratorMapping;
+import org.apache.drill.exec.compile.sig.MappingSet;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.expr.ClassGenerator;
+import org.apache.drill.exec.expr.ClassGenerator.HoldingContainer;
+import org.apache.drill.exec.expr.CodeGenerator;
+import org.apache.drill.exec.expr.ExpressionTreeMaterializer;
+import org.apache.drill.exec.expr.fn.FunctionGenerationHelper;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.config.ExternalSort;
+import org.apache.drill.exec.physical.config.Sort;
+import org.apache.drill.exec.physical.impl.xsort.SingleBatchSorter;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.VectorAccessible;
+import org.apache.drill.exec.vector.CopyUtil;
+
+import com.sun.codemodel.JConditional;
+import com.sun.codemodel.JExpr;
+
+/**
+ * Generates and manages the data-specific classes for this operator.
+ * <p>
+ * Several of the code generation methods take a batch, but the methods
+ * are called for many batches, and generate code only for the first one.
+ * Better would be to generate code from a schema; but Drill is not set
+ * up for that at present.
+ */
+
+public class OperatorCodeGenerator {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(OperatorCodeGenerator.class);
+
+ protected static final MappingSet MAIN_MAPPING = new MappingSet((String) null, null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
+ protected static final MappingSet LEFT_MAPPING = new MappingSet("leftIndex", null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
+ protected static final MappingSet RIGHT_MAPPING = new MappingSet("rightIndex", null, ClassGenerator.DEFAULT_SCALAR_MAP, ClassGenerator.DEFAULT_SCALAR_MAP);
+
+ private static final GeneratorMapping COPIER_MAPPING = new GeneratorMapping("doSetup", "doCopy", null, null);
+ private static final MappingSet COPIER_MAPPING_SET = new MappingSet(COPIER_MAPPING, COPIER_MAPPING);
+
+ private final FragmentContext context;
+ @SuppressWarnings("unused")
+ private BatchSchema schema;
+
+ /**
+ * A single PriorityQueueCopier instance is used for 2 purposes:
+ * 1. Merge sorted batches before spilling
+ * 2. Merge sorted batches when all incoming data fits in memory
+ */
+
+ private PriorityQueueCopier copier;
+ private final Sort popConfig;
+
+ /**
+ * Generated sort operation used to sort each incoming batch according to
+ * the sort criteria specified in the {@link ExternalSort} definition of
+ * this operator.
+ */
+
+ private SingleBatchSorter sorter;
+
+ public OperatorCodeGenerator(FragmentContext context, Sort popConfig) {
+ this.context = context;
+ this.popConfig = popConfig;
+ }
+
+ public void setSchema(BatchSchema schema) {
+ close();
+ this.schema = schema;
+ }
+
+ public void close() {
+ closeCopier();
+ sorter = null;
+ }
+
+ public void closeCopier() {
+ if (copier == null) {
+ return; }
+ try {
+ copier.close();
+ copier = null;
+ } catch (IOException e) {
+ throw UserException.dataWriteError(e)
+ .message("Failure while flushing spilled data")
+ .build(logger);
+ }
+ }
+
+ public PriorityQueueCopier getCopier(VectorAccessible batch) {
+ if (copier == null) {
+ copier = generateCopier(batch);
+ }
+ return copier;
+ }
+
+ private PriorityQueueCopier generateCopier(VectorAccessible batch) {
+ // Generate the copier code and obtain the resulting class
+
+ CodeGenerator<PriorityQueueCopier> cg = CodeGenerator.get(PriorityQueueCopier.TEMPLATE_DEFINITION, context.getFunctionRegistry(), context.getOptions());
+ ClassGenerator<PriorityQueueCopier> g = cg.getRoot();
+ cg.plainJavaCapable(true);
+ // Uncomment out this line to debug the generated code.
+// cg.saveCodeForDebugging(true);
+
+ generateComparisons(g, batch);
+
+ g.setMappingSet(COPIER_MAPPING_SET);
+ CopyUtil.generateCopies(g, batch, true);
+ g.setMappingSet(MAIN_MAPPING);
+ return getInstance(cg);
+ }
+
+ public MSorter createNewMSorter(VectorAccessible batch) {
+ return createNewMSorter(popConfig.getOrderings(), batch, MAIN_MAPPING, LEFT_MAPPING, RIGHT_MAPPING);
+ }
+
+ private MSorter createNewMSorter(List<Ordering> orderings, VectorAccessible batch, MappingSet mainMapping, MappingSet leftMapping, MappingSet rightMapping) {
+ CodeGenerator<MSorter> cg = CodeGenerator.get(MSorter.TEMPLATE_DEFINITION, context.getFunctionRegistry(), context.getOptions());
+ cg.plainJavaCapable(true);
+
+ // Uncomment out this line to debug the generated code.
+// cg.saveCodeForDebugging(true);
+ ClassGenerator<MSorter> g = cg.getRoot();
+ g.setMappingSet(mainMapping);
+
+ for (Ordering od : orderings) {
+ // first, we rewrite the evaluation stack for each side of the comparison.
+ ErrorCollector collector = new ErrorCollectorImpl();
+ final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector, context.getFunctionRegistry());
+ if (collector.hasErrors()) {
+ throw UserException.unsupportedError()
+ .message("Failure while materializing expression. " + collector.toErrorString())
+ .build(logger);
+ }
+ g.setMappingSet(leftMapping);
+ HoldingContainer left = g.addExpr(expr, ClassGenerator.BlkCreateMode.FALSE);
+ g.setMappingSet(rightMapping);
+ HoldingContainer right = g.addExpr(expr, ClassGenerator.BlkCreateMode.FALSE);
+ g.setMappingSet(mainMapping);
+
+ // next we wrap the two comparison sides and add the expression block for the comparison.
+ LogicalExpression fh =
+ FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+ context.getFunctionRegistry());
+ HoldingContainer out = g.addExpr(fh, ClassGenerator.BlkCreateMode.FALSE);
+ JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
+
+ if (od.getDirection() == Direction.ASCENDING) {
+ jc._then()._return(out.getValue());
+ }else{
+ jc._then()._return(out.getValue().minus());
+ }
+ g.rotateBlock();
+ }
+
+ g.rotateBlock();
+ g.getEvalBlock()._return(JExpr.lit(0));
+
+ return getInstance(cg);
+ }
+
+ public SingleBatchSorter getSorter(VectorAccessible batch) {
+ if (sorter == null) {
+ sorter = createNewSorter(batch);
+ }
+ return sorter;
+ }
+
+ private SingleBatchSorter createNewSorter(VectorAccessible batch) {
+ CodeGenerator<SingleBatchSorter> cg = CodeGenerator.get(
+ SingleBatchSorter.TEMPLATE_DEFINITION, context.getFunctionRegistry(),
+ context.getOptions());
+ ClassGenerator<SingleBatchSorter> g = cg.getRoot();
+ cg.plainJavaCapable(true);
+ // Uncomment out this line to debug the generated code.
+// cg.saveCodeForDebugging(true);
+
+ generateComparisons(g, batch);
+ return getInstance(cg);
+ }
+
+ private <T> T getInstance(CodeGenerator<T> cg) {
+ try {
+ return context.getImplementationClass(cg);
+ } catch (ClassTransformationException e) {
+ throw UserException.unsupportedError(e)
+ .message("Code generation error - likely code error.")
+ .build(logger);
+ } catch (IOException e) {
+ throw UserException.resourceError(e)
+ .message("IO Error during code generation.")
+ .build(logger);
+ }
+ }
+
+ protected void generateComparisons(ClassGenerator<?> g, VectorAccessible batch) {
+ g.setMappingSet(MAIN_MAPPING);
+
+ for (Ordering od : popConfig.getOrderings()) {
+ // first, we rewrite the evaluation stack for each side of the comparison.
+ ErrorCollector collector = new ErrorCollectorImpl();
+ final LogicalExpression expr = ExpressionTreeMaterializer.materialize(od.getExpr(), batch, collector, context.getFunctionRegistry());
+ if (collector.hasErrors()) {
+ throw UserException.unsupportedError()
+ .message("Failure while materializing expression. " + collector.toErrorString())
+ .build(logger);
+ }
+ g.setMappingSet(LEFT_MAPPING);
+ HoldingContainer left = g.addExpr(expr, ClassGenerator.BlkCreateMode.FALSE);
+ g.setMappingSet(RIGHT_MAPPING);
+ HoldingContainer right = g.addExpr(expr, ClassGenerator.BlkCreateMode.FALSE);
+ g.setMappingSet(MAIN_MAPPING);
+
+ // next we wrap the two comparison sides and add the expression block for the comparison.
+ LogicalExpression fh =
+ FunctionGenerationHelper.getOrderingComparator(od.nullsSortHigh(), left, right,
+ context.getFunctionRegistry());
+ HoldingContainer out = g.addExpr(fh, ClassGenerator.BlkCreateMode.FALSE);
+ JConditional jc = g.getEvalBlock()._if(out.getValue().ne(JExpr.lit(0)));
+
+ if (od.getDirection() == Direction.ASCENDING) {
+ jc._then()._return(out.getValue());
+ }else{
+ jc._then()._return(out.getValue().minus());
+ }
+ g.rotateBlock();
+ }
+
+ g.rotateBlock();
+ g.getEvalBlock()._return(JExpr.lit(0));
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/300e9349/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopier.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopier.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopier.java
new file mode 100644
index 0000000..2657bb8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/xsort/managed/PriorityQueueCopier.java
@@ -0,0 +1,40 @@
+/*
+ * 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.drill.exec.physical.impl.xsort.managed;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.drill.exec.compile.TemplateClassDefinition;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.record.VectorAccessible;
+
+public interface PriorityQueueCopier extends AutoCloseable {
+ public void setup(FragmentContext context, BufferAllocator allocator, VectorAccessible hyperBatch,
+ List<BatchGroup> batchGroups, VectorAccessible outgoing) throws SchemaChangeException;
+
+ public int next(int targetRecordCount);
+
+ public final static TemplateClassDefinition<PriorityQueueCopier> TEMPLATE_DEFINITION =
+ new TemplateClassDefinition<>(PriorityQueueCopier.class, PriorityQueueCopierTemplate.class);
+
+ @Override
+ abstract public void close() throws IOException; // specify this to leave out the Exception
+}