You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by pa...@apache.org on 2018/04/23 14:15:47 UTC
[1/8] drill git commit: DRILL-6289: Cluster view should show more
relevant information
Repository: drill
Updated Branches:
refs/heads/master cc606dbb9 -> 8d562d076
DRILL-6289: Cluster view should show more relevant information
Protobuf change to carry HTTP port info
Allow CORS for access to remote Drillbit metrics
Cross-origin resource sharing (CORS) is required to ensure that the WebServer is able serve REST calls for status pages.
Materialize relevant metrics
1. Heap memory (incl usage)
2. Heap memory (incl usage)
3. Average System Load (last 1 min)
4. Option to view from other nodes (pop out)
5. Added Glyphicons
Update System Table and related tests
1. Updated System Table to show HTTP port
2. Updated unit tests
Skip updating remote bit info when HTTPS (SSL) or Authentication is enabled.
Default CpuGaugeSet is public; Added Gauges
* CPU Utiization by Drill
* Uptime
Show ALL Buttons, but do HTTPS Check
Reduce power button to icon
Allowing CORS for /status/metrics only
Accounting for situations when JVM does not report Process CPU Load
i.e. returned value is negative.
See https://docs.oracle.com/javase/7/docs/jre/api/management/extension/com/sun/management/OperatingSystemMXBean.html#getProcessCpuLoad()
Addressed shutdown security conditions
Added C++ Client Protobuf
Added steps for Protobuf generation to protocol/readme.txt
This closes #1203
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/ae1f8389
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/ae1f8389
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/ae1f8389
Branch: refs/heads/master
Commit: ae1f838912f51244346d25912339fd650aa83dea
Parents: cc606db
Author: Kunal Khatua <kk...@maprtech.com>
Authored: Sun Mar 25 22:03:12 2018 -0700
Committer: Parth Chandra <pa...@apache.org>
Committed: Fri Apr 20 16:30:40 2018 -0700
----------------------------------------------------------------------
.../apache/drill/exec/metrics/CpuGaugeSet.java | 101 +++++++++
.../apache/drill/exec/metrics/DrillMetrics.java | 1 +
.../client/src/protobuf/Coordination.pb.cc | 66 +++++-
.../client/src/protobuf/Coordination.pb.h | 34 ++-
.../client/src/protobuf/UserBitShared.pb.cc | 15 +-
.../client/src/protobuf/UserBitShared.pb.h | 7 +-
.../org/apache/drill/exec/server/Drillbit.java | 11 +-
.../drill/exec/server/rest/DrillRoot.java | 4 +
.../drill/exec/server/rest/WebServer.java | 15 +-
.../drill/exec/store/sys/DrillbitIterator.java | 2 +
.../java-exec/src/main/resources/rest/index.ftl | 218 +++++++++++++++----
.../work/metadata/TestMetadataProvider.java | 9 +-
protocol/readme.txt | 11 +-
.../drill/exec/proto/CoordinationProtos.java | 106 ++++++++-
.../exec/proto/SchemaCoordinationProtos.java | 7 +
.../exec/proto/beans/DrillbitEndpoint.java | 22 ++
protocol/src/main/protobuf/Coordination.proto | 1 +
17 files changed, 546 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/common/src/main/java/org/apache/drill/exec/metrics/CpuGaugeSet.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/exec/metrics/CpuGaugeSet.java b/common/src/main/java/org/apache/drill/exec/metrics/CpuGaugeSet.java
new file mode 100644
index 0000000..a652e3c
--- /dev/null
+++ b/common/src/main/java/org/apache/drill/exec/metrics/CpuGaugeSet.java
@@ -0,0 +1,101 @@
+/*
+ * 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.metrics;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.util.HashMap;
+import java.util.Map;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Metric;
+import com.codahale.metrics.MetricSet;
+import com.sun.management.OperatingSystemMXBean;
+
+/**
+ * Creates a Cpu GaugeSet
+ */
+@SuppressWarnings("restriction")
+public class CpuGaugeSet implements MetricSet {
+
+ private OperatingSystemMXBean osMXBean;
+ private RuntimeMXBean rtMXBean;
+
+ public CpuGaugeSet() {
+ this.osMXBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean();
+ this.rtMXBean = ManagementFactory.getRuntimeMXBean();
+ }
+
+ @Override
+ public Map<String, Metric> getMetrics() {
+ final Map<String, Metric> metric = new HashMap<>(1);
+ metric.put("os.load.avg", new OperatingSystemLoad(osMXBean));
+ metric.put("drillbit.load.avg", new DrillbitProcessLoad(osMXBean));
+ metric.put("drillbit.uptime", new DrillbitUptime(rtMXBean));
+ return metric;
+ }
+}
+
+/**
+ * Creating an AverageSystemLoad Gauge
+ */
+@SuppressWarnings("restriction")
+final class OperatingSystemLoad implements Gauge<Double> {
+ private OperatingSystemMXBean osMXBean;
+ public OperatingSystemLoad(OperatingSystemMXBean osBean) {
+ this.osMXBean = osBean;
+ }
+
+ @Override
+ public Double getValue() {
+ return osMXBean.getSystemLoadAverage();
+ }
+
+}
+
+/**
+ * Creating an AverageDrillbitLoad Gauge
+ */
+@SuppressWarnings("restriction")
+final class DrillbitProcessLoad implements Gauge<Double> {
+ private OperatingSystemMXBean osMXBean;
+ public DrillbitProcessLoad(OperatingSystemMXBean osBean) {
+ this.osMXBean = osBean;
+ }
+
+ @Override
+ public Double getValue() {
+ return osMXBean.getProcessCpuLoad();
+ }
+
+}
+/**
+ * Creating an DrillbitUptime Gauge
+ */
+final class DrillbitUptime implements Gauge<Long> {
+ private RuntimeMXBean rtMXBean;
+ public DrillbitUptime(RuntimeMXBean osBean) {
+ this.rtMXBean = osBean;
+ }
+
+ @Override
+ public Long getValue() {
+ return rtMXBean.getUptime();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java b/common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java
index 46222ab..4c67f44 100644
--- a/common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java
+++ b/common/src/main/java/org/apache/drill/exec/metrics/DrillMetrics.java
@@ -60,6 +60,7 @@ public final class DrillMetrics {
REGISTRY.registerAll(new BufferPoolMetricSet(ManagementFactory.getPlatformMBeanServer()));
REGISTRY.registerAll(new MemoryUsageGaugeSet());
REGISTRY.registerAll(new ThreadStatesGaugeSet());
+ REGISTRY.registerAll(new CpuGaugeSet());
register("fd.usage", new FileDescriptorRatioGauge());
}
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/contrib/native/client/src/protobuf/Coordination.pb.cc
----------------------------------------------------------------------
diff --git a/contrib/native/client/src/protobuf/Coordination.pb.cc b/contrib/native/client/src/protobuf/Coordination.pb.cc
index 14b3103..31ffcce 100644
--- a/contrib/native/client/src/protobuf/Coordination.pb.cc
+++ b/contrib/native/client/src/protobuf/Coordination.pb.cc
@@ -41,7 +41,7 @@ void protobuf_AssignDesc_Coordination_2eproto() {
"Coordination.proto");
GOOGLE_CHECK(file != NULL);
DrillbitEndpoint_descriptor_ = file->message_type(0);
- static const int DrillbitEndpoint_offsets_[7] = {
+ static const int DrillbitEndpoint_offsets_[8] = {
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, address_),
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, user_port_),
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, control_port_),
@@ -49,6 +49,7 @@ void protobuf_AssignDesc_Coordination_2eproto() {
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, roles_),
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, version_),
GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, state_),
+ GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(DrillbitEndpoint, http_port_),
};
DrillbitEndpoint_reflection_ =
new ::google::protobuf::internal::GeneratedMessageReflection(
@@ -136,21 +137,22 @@ void protobuf_AddDesc_Coordination_2eproto() {
GOOGLE_PROTOBUF_VERIFY_VERSION;
::google::protobuf::DescriptorPool::InternalAddGeneratedFile(
- "\n\022Coordination.proto\022\004exec\"\367\001\n\020DrillbitE"
+ "\n\022Coordination.proto\022\004exec\"\212\002\n\020DrillbitE"
"ndpoint\022\017\n\007address\030\001 \001(\t\022\021\n\tuser_port\030\002 "
"\001(\005\022\024\n\014control_port\030\003 \001(\005\022\021\n\tdata_port\030\004"
" \001(\005\022\032\n\005roles\030\005 \001(\0132\013.exec.Roles\022\017\n\007vers"
"ion\030\006 \001(\t\022+\n\005state\030\007 \001(\0162\034.exec.Drillbit"
- "Endpoint.State\"<\n\005State\022\013\n\007STARTUP\020\000\022\n\n\006"
- "ONLINE\020\001\022\r\n\tQUIESCENT\020\002\022\013\n\007OFFLINE\020\003\"i\n\024"
- "DrillServiceInstance\022\n\n\002id\030\001 \001(\t\022\033\n\023regi"
- "strationTimeUTC\030\002 \001(\003\022(\n\010endpoint\030\003 \001(\0132"
- "\026.exec.DrillbitEndpoint\"\227\001\n\005Roles\022\027\n\tsql"
- "_query\030\001 \001(\010:\004true\022\032\n\014logical_plan\030\002 \001(\010"
- ":\004true\022\033\n\rphysical_plan\030\003 \001(\010:\004true\022\033\n\rj"
- "ava_executor\030\004 \001(\010:\004true\022\037\n\021distributed_"
- "cache\030\005 \001(\010:\004trueB3\n\033org.apache.drill.ex"
- "ec.protoB\022CoordinationProtosH\001", 590);
+ "Endpoint.State\022\021\n\thttp_port\030\010 \001(\005\"<\n\005Sta"
+ "te\022\013\n\007STARTUP\020\000\022\n\n\006ONLINE\020\001\022\r\n\tQUIESCENT"
+ "\020\002\022\013\n\007OFFLINE\020\003\"i\n\024DrillServiceInstance\022"
+ "\n\n\002id\030\001 \001(\t\022\033\n\023registrationTimeUTC\030\002 \001(\003"
+ "\022(\n\010endpoint\030\003 \001(\0132\026.exec.DrillbitEndpoi"
+ "nt\"\227\001\n\005Roles\022\027\n\tsql_query\030\001 \001(\010:\004true\022\032\n"
+ "\014logical_plan\030\002 \001(\010:\004true\022\033\n\rphysical_pl"
+ "an\030\003 \001(\010:\004true\022\033\n\rjava_executor\030\004 \001(\010:\004t"
+ "rue\022\037\n\021distributed_cache\030\005 \001(\010:\004trueB3\n\033"
+ "org.apache.drill.exec.protoB\022Coordinatio"
+ "nProtosH\001", 609);
::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
"Coordination.proto", &protobuf_RegisterTypes);
DrillbitEndpoint::default_instance_ = new DrillbitEndpoint();
@@ -204,6 +206,7 @@ const int DrillbitEndpoint::kDataPortFieldNumber;
const int DrillbitEndpoint::kRolesFieldNumber;
const int DrillbitEndpoint::kVersionFieldNumber;
const int DrillbitEndpoint::kStateFieldNumber;
+const int DrillbitEndpoint::kHttpPortFieldNumber;
#endif // !_MSC_VER
DrillbitEndpoint::DrillbitEndpoint()
@@ -230,6 +233,7 @@ void DrillbitEndpoint::SharedCtor() {
roles_ = NULL;
version_ = const_cast< ::std::string*>(&::google::protobuf::internal::kEmptyString);
state_ = 0;
+ http_port_ = 0;
::memset(_has_bits_, 0, sizeof(_has_bits_));
}
@@ -289,6 +293,7 @@ void DrillbitEndpoint::Clear() {
}
}
state_ = 0;
+ http_port_ = 0;
}
::memset(_has_bits_, 0, sizeof(_has_bits_));
mutable_unknown_fields()->Clear();
@@ -412,6 +417,22 @@ bool DrillbitEndpoint::MergePartialFromCodedStream(
} else {
goto handle_uninterpreted;
}
+ if (input->ExpectTag(64)) goto parse_http_port;
+ break;
+ }
+
+ // optional int32 http_port = 8;
+ case 8: {
+ if (::google::protobuf::internal::WireFormatLite::GetTagWireType(tag) ==
+ ::google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT) {
+ parse_http_port:
+ DO_((::google::protobuf::internal::WireFormatLite::ReadPrimitive<
+ ::google::protobuf::int32, ::google::protobuf::internal::WireFormatLite::TYPE_INT32>(
+ input, &http_port_)));
+ set_has_http_port();
+ } else {
+ goto handle_uninterpreted;
+ }
if (input->ExpectAtEnd()) return true;
break;
}
@@ -479,6 +500,11 @@ void DrillbitEndpoint::SerializeWithCachedSizes(
7, this->state(), output);
}
+ // optional int32 http_port = 8;
+ if (has_http_port()) {
+ ::google::protobuf::internal::WireFormatLite::WriteInt32(8, this->http_port(), output);
+ }
+
if (!unknown_fields().empty()) {
::google::protobuf::internal::WireFormat::SerializeUnknownFields(
unknown_fields(), output);
@@ -535,6 +561,11 @@ void DrillbitEndpoint::SerializeWithCachedSizes(
7, this->state(), target);
}
+ // optional int32 http_port = 8;
+ if (has_http_port()) {
+ target = ::google::protobuf::internal::WireFormatLite::WriteInt32ToArray(8, this->http_port(), target);
+ }
+
if (!unknown_fields().empty()) {
target = ::google::protobuf::internal::WireFormat::SerializeUnknownFieldsToArray(
unknown_fields(), target);
@@ -594,6 +625,13 @@ int DrillbitEndpoint::ByteSize() const {
::google::protobuf::internal::WireFormatLite::EnumSize(this->state());
}
+ // optional int32 http_port = 8;
+ if (has_http_port()) {
+ total_size += 1 +
+ ::google::protobuf::internal::WireFormatLite::Int32Size(
+ this->http_port());
+ }
+
}
if (!unknown_fields().empty()) {
total_size +=
@@ -642,6 +680,9 @@ void DrillbitEndpoint::MergeFrom(const DrillbitEndpoint& from) {
if (from.has_state()) {
set_state(from.state());
}
+ if (from.has_http_port()) {
+ set_http_port(from.http_port());
+ }
}
mutable_unknown_fields()->MergeFrom(from.unknown_fields());
}
@@ -672,6 +713,7 @@ void DrillbitEndpoint::Swap(DrillbitEndpoint* other) {
std::swap(roles_, other->roles_);
std::swap(version_, other->version_);
std::swap(state_, other->state_);
+ std::swap(http_port_, other->http_port_);
std::swap(_has_bits_[0], other->_has_bits_[0]);
_unknown_fields_.Swap(&other->_unknown_fields_);
std::swap(_cached_size_, other->_cached_size_);
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/contrib/native/client/src/protobuf/Coordination.pb.h
----------------------------------------------------------------------
diff --git a/contrib/native/client/src/protobuf/Coordination.pb.h b/contrib/native/client/src/protobuf/Coordination.pb.h
index e52c70c..07bbc8e 100644
--- a/contrib/native/client/src/protobuf/Coordination.pb.h
+++ b/contrib/native/client/src/protobuf/Coordination.pb.h
@@ -202,6 +202,13 @@ class DrillbitEndpoint : public ::google::protobuf::Message {
inline ::exec::DrillbitEndpoint_State state() const;
inline void set_state(::exec::DrillbitEndpoint_State value);
+ // optional int32 http_port = 8;
+ inline bool has_http_port() const;
+ inline void clear_http_port();
+ static const int kHttpPortFieldNumber = 8;
+ inline ::google::protobuf::int32 http_port() const;
+ inline void set_http_port(::google::protobuf::int32 value);
+
// @@protoc_insertion_point(class_scope:exec.DrillbitEndpoint)
private:
inline void set_has_address();
@@ -218,6 +225,8 @@ class DrillbitEndpoint : public ::google::protobuf::Message {
inline void clear_has_version();
inline void set_has_state();
inline void clear_has_state();
+ inline void set_has_http_port();
+ inline void clear_has_http_port();
::google::protobuf::UnknownFieldSet _unknown_fields_;
@@ -228,9 +237,10 @@ class DrillbitEndpoint : public ::google::protobuf::Message {
::google::protobuf::int32 data_port_;
int state_;
::std::string* version_;
+ ::google::protobuf::int32 http_port_;
mutable int _cached_size_;
- ::google::protobuf::uint32 _has_bits_[(7 + 31) / 32];
+ ::google::protobuf::uint32 _has_bits_[(8 + 31) / 32];
friend void protobuf_AddDesc_Coordination_2eproto();
friend void protobuf_AssignDesc_Coordination_2eproto();
@@ -744,6 +754,28 @@ inline void DrillbitEndpoint::set_state(::exec::DrillbitEndpoint_State value) {
state_ = value;
}
+// optional int32 http_port = 8;
+inline bool DrillbitEndpoint::has_http_port() const {
+ return (_has_bits_[0] & 0x00000080u) != 0;
+}
+inline void DrillbitEndpoint::set_has_http_port() {
+ _has_bits_[0] |= 0x00000080u;
+}
+inline void DrillbitEndpoint::clear_has_http_port() {
+ _has_bits_[0] &= ~0x00000080u;
+}
+inline void DrillbitEndpoint::clear_http_port() {
+ http_port_ = 0;
+ clear_has_http_port();
+}
+inline ::google::protobuf::int32 DrillbitEndpoint::http_port() const {
+ return http_port_;
+}
+inline void DrillbitEndpoint::set_http_port(::google::protobuf::int32 value) {
+ set_has_http_port();
+ http_port_ = value;
+}
+
// -------------------------------------------------------------------
// DrillServiceInstance
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/contrib/native/client/src/protobuf/UserBitShared.pb.cc
----------------------------------------------------------------------
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.cc b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
index 4350082..c5b2527 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.cc
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.cc
@@ -749,7 +749,7 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
"agmentState\022\013\n\007SENDING\020\000\022\027\n\023AWAITING_ALL"
"OCATION\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FINISHED\020\003\022\r\n\t"
"CANCELLED\020\004\022\n\n\006FAILED\020\005\022\032\n\026CANCELLATION_"
- "REQUESTED\020\006*\227\006\n\020CoreOperatorType\022\021\n\rSING"
+ "REQUESTED\020\006*\302\006\n\020CoreOperatorType\022\021\n\rSING"
"LE_SENDER\020\000\022\024\n\020BROADCAST_SENDER\020\001\022\n\n\006FIL"
"TER\020\002\022\022\n\016HASH_AGGREGATE\020\003\022\r\n\tHASH_JOIN\020\004"
"\022\016\n\nMERGE_JOIN\020\005\022\031\n\025HASH_PARTITION_SENDE"
@@ -769,10 +769,12 @@ void protobuf_AddDesc_UserBitShared_2eproto() {
"ASE_SUB_SCAN\020!\022\n\n\006WINDOW\020\"\022\024\n\020NESTED_LOO"
"P_JOIN\020#\022\021\n\rAVRO_SUB_SCAN\020$\022\021\n\rPCAP_SUB_"
"SCAN\020%\022\022\n\016KAFKA_SUB_SCAN\020&\022\021\n\rKUDU_SUB_S"
- "CAN\020\'*g\n\nSaslStatus\022\020\n\014SASL_UNKNOWN\020\000\022\016\n"
- "\nSASL_START\020\001\022\024\n\020SASL_IN_PROGRESS\020\002\022\020\n\014S"
- "ASL_SUCCESS\020\003\022\017\n\013SASL_FAILED\020\004B.\n\033org.ap"
- "ache.drill.exec.protoB\rUserBitSharedH\001", 5078);
+ "CAN\020\'\022\013\n\007FLATTEN\020(\022\020\n\014LATERAL_JOIN\020)\022\n\n\006"
+ "UNNEST\020**g\n\nSaslStatus\022\020\n\014SASL_UNKNOWN\020\000"
+ "\022\016\n\nSASL_START\020\001\022\024\n\020SASL_IN_PROGRESS\020\002\022\020"
+ "\n\014SASL_SUCCESS\020\003\022\017\n\013SASL_FAILED\020\004B.\n\033org"
+ ".apache.drill.exec.protoB\rUserBitSharedH"
+ "\001", 5121);
::google::protobuf::MessageFactory::InternalRegisterGeneratedFile(
"UserBitShared.proto", &protobuf_RegisterTypes);
UserCredentials::default_instance_ = new UserCredentials();
@@ -933,6 +935,9 @@ bool CoreOperatorType_IsValid(int value) {
case 37:
case 38:
case 39:
+ case 40:
+ case 41:
+ case 42:
return true;
default:
return false;
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/contrib/native/client/src/protobuf/UserBitShared.pb.h
----------------------------------------------------------------------
diff --git a/contrib/native/client/src/protobuf/UserBitShared.pb.h b/contrib/native/client/src/protobuf/UserBitShared.pb.h
index 5a49bf5..1a69906 100644
--- a/contrib/native/client/src/protobuf/UserBitShared.pb.h
+++ b/contrib/native/client/src/protobuf/UserBitShared.pb.h
@@ -243,11 +243,14 @@ enum CoreOperatorType {
AVRO_SUB_SCAN = 36,
PCAP_SUB_SCAN = 37,
KAFKA_SUB_SCAN = 38,
- KUDU_SUB_SCAN = 39
+ KUDU_SUB_SCAN = 39,
+ FLATTEN = 40,
+ LATERAL_JOIN = 41,
+ UNNEST = 42
};
bool CoreOperatorType_IsValid(int value);
const CoreOperatorType CoreOperatorType_MIN = SINGLE_SENDER;
-const CoreOperatorType CoreOperatorType_MAX = KUDU_SUB_SCAN;
+const CoreOperatorType CoreOperatorType_MAX = UNNEST;
const int CoreOperatorType_ARRAYSIZE = CoreOperatorType_MAX + 1;
const ::google::protobuf::EnumDescriptor* CoreOperatorType_descriptor();
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
index e7bdd2e..93e79e9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/Drillbit.java
@@ -180,7 +180,7 @@ public class Drillbit implements AutoCloseable {
if (profileStoreProvider != storeProvider) {
profileStoreProvider.start();
}
- final DrillbitEndpoint md = engine.start();
+ DrillbitEndpoint md = engine.start();
manager.start(md, engine.getController(), engine.getDataConnectionCreator(), coord, storeProvider, profileStoreProvider);
@SuppressWarnings("resource")
final DrillbitContext drillbitContext = manager.getContext();
@@ -189,11 +189,14 @@ public class Drillbit implements AutoCloseable {
drillbitContext.getOptionManager().init();
javaPropertiesToSystemOptions();
manager.getContext().getRemoteFunctionRegistry().init(context.getConfig(), storeProvider, coord);
- registrationHandle = coord.register(md);
webServer.start();
-
+ //Discovering HTTP port (in case of port hunting)
+ if (webServer.isRunning()) {
+ int httpPort = getWebServerPort();
+ md = md.toBuilder().setHttpPort(httpPort).build();
+ }
+ registrationHandle = coord.register(md);
// Must start the RM after the above since it needs to read system options.
-
drillbitContext.startRM();
Runtime.getRuntime().addShutdownHook(new ShutdownThread(this, new StackTrace()));
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRoot.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRoot.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRoot.java
index ef439f6..50d99b0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRoot.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/DrillRoot.java
@@ -421,6 +421,7 @@ public static class ClusterInfo {
public static class DrillbitInfo implements Comparable<DrillbitInfo> {
private final String address;
+ private final String httpPort;
private final String userPort;
private final String controlPort;
private final String dataPort;
@@ -432,6 +433,7 @@ public static class DrillbitInfo implements Comparable<DrillbitInfo> {
@JsonCreator
public DrillbitInfo(DrillbitEndpoint drillbit, boolean current, boolean versionMatch) {
this.address = drillbit.getAddress();
+ this.httpPort = String.valueOf(drillbit.getHttpPort());
this.userPort = String.valueOf(drillbit.getUserPort());
this.controlPort = String.valueOf(drillbit.getControlPort());
this.dataPort = String.valueOf(drillbit.getDataPort());
@@ -443,6 +445,8 @@ public static class DrillbitInfo implements Comparable<DrillbitInfo> {
public String getAddress() { return address; }
+ public String getHttpPort() { return httpPort; }
+
public String getUserPort() { return userPort; }
public String getControlPort() { return controlPort; }
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
index e7aa332..f08ec8e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/WebServer.java
@@ -158,7 +158,14 @@ public class WebServer implements AutoCloseable {
final int selectors = config.getInt(ExecConstants.HTTP_JETTY_SERVER_SELECTORS);
final QueuedThreadPool threadPool = new QueuedThreadPool(2, 2, 60000);
embeddedJetty = new Server(threadPool);
- embeddedJetty.setHandler(createServletContextHandler(authEnabled));
+ ServletContextHandler webServerContext = createServletContextHandler(authEnabled);
+ //Allow for Other Drillbits to make REST calls
+ FilterHolder filterHolder = new FilterHolder(CrossOriginFilter.class);
+ filterHolder.setInitParameter("allowedOrigins", "*");
+ //Allowing CORS for metrics only
+ webServerContext.addFilter(filterHolder, "/status/metrics", null);
+ embeddedJetty.setHandler(webServerContext);
+
ServerConnector connector = createConnector(port, acceptors, selectors);
threadPool.setMaxThreads(1 + connector.getAcceptors() + connector.getSelectorManager().getSelectorCount());
embeddedJetty.addConnector(connector);
@@ -296,12 +303,16 @@ public class WebServer implements AutoCloseable {
}
public int getPort() {
- if (embeddedJetty == null || embeddedJetty.getConnectors().length != 1) {
+ if (!isRunning()) {
throw new UnsupportedOperationException("Http is not enabled");
}
return ((ServerConnector)embeddedJetty.getConnectors()[0]).getPort();
}
+ public boolean isRunning() {
+ return (embeddedJetty != null && embeddedJetty.getConnectors().length == 1);
+ }
+
private ServerConnector createConnector(int port, int acceptors, int selectors) throws Exception {
final ServerConnector serverConnector;
if (config.getBoolean(ExecConstants.HTTP_ENABLE_SSL)) {
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/DrillbitIterator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/DrillbitIterator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/DrillbitIterator.java
index 5629b3f..a50a1c0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/DrillbitIterator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/sys/DrillbitIterator.java
@@ -38,6 +38,7 @@ public class DrillbitIterator implements Iterator<Object> {
public int user_port;
public int control_port;
public int data_port;
+ public int http_port;
public boolean current;
public String version;
public String state;
@@ -54,6 +55,7 @@ public class DrillbitIterator implements Iterator<Object> {
DrillbitInstance i = new DrillbitInstance();
i.current = isCurrent(ep);
i.hostname = ep.getAddress();
+ i.http_port = ep.getHttpPort();
i.user_port = ep.getUserPort();
i.control_port = ep.getControlPort();
i.data_port = ep.getDataPort();
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/main/resources/rest/index.ftl
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/rest/index.ftl b/exec/java-exec/src/main/resources/rest/index.ftl
index e3904a2..5259e44 100644
--- a/exec/java-exec/src/main/resources/rest/index.ftl
+++ b/exec/java-exec/src/main/resources/rest/index.ftl
@@ -60,12 +60,20 @@
<thead>
<tr>
<th>#</th>
- <th>Address</th>
+ <th title="Drillbits in Cluster" style="cursor: help;">Address <span class="glyphicon glyphicon-info-sign" style="font-size:100%"/></th>
+ <th title="Heap Memory in use (as percent of Total Heap)" style="cursor: help;">Heap Memory Usage <span class="glyphicon glyphicon-info-sign" style="font-size:100%"/></th>
+ <th title="Estimated Direct Memory ACTIVELY in use (as percent of Peak Usage)" style="cursor: help;">Direct Memory Usage <span class="glyphicon glyphicon-info-sign" style="font-size:100%"/></th>
+ <th title="Current CPU usage by Drill" style="cursor: help;">CPU Usage<span class="glyphicon glyphicon-info-sign" style="font-size:100%"/></th>
+ <th title="Average load on the system in the last 1 minute" style="cursor: help;">Avg Sys Load <span class="glyphicon glyphicon-info-sign" style="font-size:100%"/></th>
<th>User Port</th>
<th>Control Port</th>
<th>Data Port</th>
<th>Version</th>
<th>Status</th>
+ <th>Uptime</th>
+ <#if (model.shouldShowAdminInfo() || !model.isAuthEnabled()) >
+ <th>Shutdown</th>
+ </#if>
</tr>
</thead>
<tbody>
@@ -75,9 +83,16 @@
<td>${i}</td>
<td id="address" >${drillbit.getAddress()}<#if drillbit.isCurrent()>
<span class="label label-info" id="current">Current</span>
+ <#else>
+ <a onclick="popOutRemoteDbitUI('${drillbit.getAddress()}','${drillbit.getHttpPort()}');" style="cursor:pointer;color:blue" title="Open in new window"><span class="glyphicon glyphicon-new-window"/></a>
</#if>
</td>
- <td id="port" >${drillbit.getUserPort()}</td>
+ <td id="httpPort" style="display:none">${drillbit.getHttpPort()}</td>
+ <td class="heap">Not Available</td>
+ <td class="direct">Not Available</td>
+ <td class="bitload"><span class="label label-info" id="NA">Not Available</span></td>
+ <td class="avgload"><span class="label label-info" id="NA">Not Available</span></td>
+ <td id="port">${drillbit.getUserPort()}</td>
<td>${drillbit.getControlPort()}</td>
<td>${drillbit.getDataPort()}</td>
<td>
@@ -87,11 +102,15 @@
</span>
</td>
<td id="status" >${drillbit.getState()}</td>
- <#if (model.shouldShowAdminInfo() || !model.isAuthEnabled()) && drillbit.isCurrent() >
+ <td class="uptime" >Not Available</td>
<td>
- <button type="button" id="shutdown" onClick="shutdown($(this));"> SHUTDOWN </button>
- </td>
+ <#if ( model.shouldShowAdminInfo() && ( drillbit.isCurrent() || ( !model.isAuthEnabled() && location.protocol != "https" ))) >
+ <button type="button" id="shutdown" onClick="shutdown($(this), '${drillbit.getAddress()}:${drillbit.getHttpPort()}');">
+ <#else>
+ <button type="button" id="shutdown" title="Drillbit cannot be shutdown remotely" disabled="true" style="opacity:0.5;cursor:not-allowed;">
</#if>
+ <span class="glyphicon glyphicon-off"></span></button>
+ </td>
<td id="queriesCount"> </td>
</tr>
<#assign i = i + 1>
@@ -110,11 +129,15 @@
<tbody>
<tr>
<td>Client to Bit Encryption</td>
- <td class="list-value">${model.isUserEncryptionEnabled()?string("Enabled", "Disabled")}</td>
+ <td class="list-value">${model.isUserEncryptionEnabled()?string("Enabled", "Disabled")}
+ <#if model.isBitEncryptionEnabled()><span class="glyphicon glyphicon-lock"/></#if>
+ </td>
</tr>
<tr>
<td>Bit to Bit Encryption</td>
- <td class="list-value">${model.isBitEncryptionEnabled()?string("Enabled", "Disabled")}</td>
+ <td class="list-value">${model.isBitEncryptionEnabled()?string("Enabled", "Disabled")}
+ <#if model.isBitEncryptionEnabled()><span class="glyphicon glyphicon-lock"/></#if>
+ </td>
</tr>
</tbody>
</table>
@@ -194,23 +217,13 @@
</div>
</div>
<script charset="utf-8">
+ var updateRemoteInfo = <#if (model.isAuthEnabled())>false<#else>true</#if>;
var refreshTime = 10000;
var refresh = getRefreshTime();
- var portNum = 0;
- var port = getPortNum();
var timeout;
var size = $("#size").html();
-
- function getPortNum() {
- var port = $.ajax({
- type: 'GET',
- url: '/portNum',
- dataType: "json",
- complete: function(data) {
- portNum = data.responseJSON["port"];
- }
- });
- }
+ reloadMetrics();
+ setInterval(reloadMetrics, refreshTime);
function getRefreshTime() {
$.ajax({
@@ -255,6 +268,7 @@
else {
if (status_map[key] == "ONLINE") {
$("#row-"+i).find("#status").text(status_map[key]);
+ $("#row-"+i).find("#shutdown").prop('disabled',false).css('opacity',1.0);
}
else {
if ($("#row-"+i).find("#current").html() == "Current") {
@@ -268,7 +282,7 @@
function fillQueryCount(row_id) {
var requestPath = "/queriesCount";
var url = getRequestUrl(requestPath);
- var result = $.ajax({
+ var result = $.ajax({
type: 'GET',
url: url,
complete: function(data) {
@@ -278,33 +292,155 @@
}
});
}
- <#if model.shouldShowAdminInfo() || !model.isAuthEnabled()>
- function shutdown(button) {
- if (confirm("Are you sure you want to shutdown Drillbit running on " + location.host + " node?")) {
- var requestPath = "/gracefulShutdown";
- var url = getRequestUrl(requestPath);
- var result = $.ajax({
- type: 'POST',
- url: url,
- contentType : 'text/plain',
- error: function (request, textStatus, errorThrown) {
- alert(errorThrown);
- },
- success: function(data) {
- alert(data["response"]);
- button.prop('disabled',true).css('opacity',0.5);
- }
- });
- }
+ <#if (model.shouldShowAdminInfo() || !model.isAuthEnabled()) >
+ function shutdown(button,host) {
+ if (confirm("Are you sure you want to shutdown Drillbit running on " + host + " node?")) {
+ var url = location.protocol + "//" + host + "/gracefulShutdown";
+ var result = $.ajax({
+ type: 'POST',
+ url: url,
+ contentType : 'text/plain',
+ error: function (request, textStatus, errorThrown) {
+ alert(errorThrown);
+ },
+ success: function(data) {
+ alert(data.responseJSON["response"]);
+ button.prop('disabled',true).css('opacity',0.5);
+ }
+ });
+ }
}
- </#if>
+ </#if>
+
+ function popOutRemoteDbitUI(dbitHost, dbitPort) {
+ var dbitWebUIUrl = location.protocol+'//'+ dbitHost+':'+dbitPort;
+ var tgtWindow = 'dbit_'+dbitHost;
+ window.open(dbitWebUIUrl, tgtWindow);
+ }
+
function getRequestUrl(requestPath) {
var protocol = location.protocol;
var host = location.host;
var url = protocol + "//" + host + requestPath;
return url;
}
- </script>
+
+ //Iterates through all the nodes for update
+ function reloadMetrics() {
+ for (i = 1; i <= size; i++) {
+ var address = $("#row-"+i).find("#address").contents().get(0).nodeValue.trim();
+ var httpPort = $("#row-"+i).find("#httpPort").contents().get(0).nodeValue.trim();
+ updateMetricsHtml(address, httpPort, i);
+ }
+ }
+
+ //Update memory
+ function updateMetricsHtml(drillbit,webport,idx) {
+ /* NOTE: For remote drillbits:
+ If Authentication or SSL is enabled; we'll assume we don't have valid certificates
+ */
+ var remoteHost = location.protocol+"//"+drillbit+":"+webport;
+ if ( !updateRemoteInfo || (location.protocol == "https" && remoteHost != location.host) ) {
+ return;
+ }
+ //
+ var result = $.ajax({
+ type: 'GET',
+ url: location.protocol+"//"+drillbit+":"+webport+"/status/metrics",
+ dataType: "json",
+ error: function(data) {
+ resetMetricsHtml(idx);
+ },
+ complete: function(data) {
+ if (typeof data.responseJSON == 'undefined') {
+ resetMetricsHtml(idx);
+ return;
+ }
+ var metrics = data.responseJSON['gauges'];
+ //Memory
+ var usedHeap = metrics['heap.used'].value;
+ var maxHeap = metrics['heap.max'].value;
+ var usedDirect = metrics['drill.allocator.root.used'].value;
+ var peakDirect = metrics['drill.allocator.root.peak'].value;
+ var heapUsage = computeMemUsage(usedHeap,maxHeap);
+ var directUsage = computeMemUsage(usedDirect,peakDirect);
+ var rowElem = document.getElementById("row-"+idx);
+ var heapElem = rowElem.getElementsByClassName("heap")[0];
+ heapElem.innerHTML = heapUsage;
+ var directElem = rowElem.getElementsByClassName("direct")[0];
+ directElem.innerHTML = directUsage;
+ //DrillbitLoad
+ var dbitLoad = metrics['drillbit.load.avg'].value;
+ var dbitLoadElem = rowElem.getElementsByClassName("bitload")[0];
+ if (dbitLoad >= 0) {
+ dbitLoadElem.innerHTML = parseFloat(Math.round(dbitLoad * 10000)/100).toFixed(2) + "%";
+ } else {
+ dbitLoadElem.innerHTML = "Not Available";
+ }
+ //AvgSysLoad
+ var avgSysLoad = metrics['os.load.avg'].value;
+ var sysLoadElem = rowElem.getElementsByClassName("avgload")[0];
+ sysLoadElem.innerHTML = avgSysLoad;
+ //Uptime
+ var uptimeValue = metrics['drillbit.uptime'].value;
+ var uptimeElem = rowElem.getElementsByClassName("uptime")[0];
+ uptimeElem.innerHTML = elapsedTime(uptimeValue);
+ }
+ });
+ }
+
+ function resetMetricsHtml(idx) {
+ $.each(["heap","direct","bitload","avgload","uptime"], function(i, key) {
+ var resetElem = document.getElementById("row-"+idx).getElementsByClassName(key)[0];
+ resetElem.innerHTML = "Not Available";
+ });
+ };
+
+ //Compute Usage
+ function computeMemUsage(used, max) {
+ var percent = 0;
+ if ( max > 0) {
+ var percent = Math.round((100 * used / max), 2);
+ }
+ var usage = bytesInGB(used, 2) + "GB (" + Math.max(0, percent) + "% of "+ bytesInGB(max, 2) +"GB)";
+ return usage;
+ }
+
+ function bytesInGB(byteVal, decimal) {
+ var scale = Math.pow(10,decimal);
+ return Math.round(scale * byteVal / 1073741824)/scale;
+ }
+
+ //Calculate Uptime
+ function elapsedTime(valueInMsec) {
+ var elapsedTime = "";
+ var d, h, m, s;
+ s = Math.floor(valueInMsec / 1000);
+ m = Math.floor(s / 60);
+ s = s % 60;
+ h = Math.floor(m / 60);
+ m = m % 60;
+ d = Math.floor(h / 24);
+ h = h % 24;
+ var detailCount = 0;
+ if (!isNaN(d) && d > 0) {
+ elapsedTime = d+"d ";
+ detailCount += 1;
+ }
+ if (!isNaN(h) && h > 0) {
+ elapsedTime = elapsedTime + h+"h ";
+ detailCount += 1;
+ }
+ if (!isNaN(m) && m > 0 && detailCount < 2) {
+ elapsedTime = elapsedTime + m+"m ";
+ detailCount += 1;
+ }
+ if (s > 0 && detailCount < 2) {
+ elapsedTime = elapsedTime + s+"s";
+ }
+ return elapsedTime;
+};
+ </script>
</#macro>
<@page_html/>
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/exec/java-exec/src/test/java/org/apache/drill/exec/work/metadata/TestMetadataProvider.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/work/metadata/TestMetadataProvider.java b/exec/java-exec/src/test/java/org/apache/drill/exec/work/metadata/TestMetadataProvider.java
index 57edfa9..4574425 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/work/metadata/TestMetadataProvider.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/work/metadata/TestMetadataProvider.java
@@ -248,7 +248,7 @@ public class TestMetadataProvider extends BaseTestQuery {
assertEquals(RequestStatus.OK, resp.getStatus());
List<ColumnMetadata> columns = resp.getColumnsList();
- assertEquals(117, columns.size());
+ assertEquals(118, columns.size());
// too many records to verify the output.
}
@@ -261,7 +261,7 @@ public class TestMetadataProvider extends BaseTestQuery {
assertEquals(RequestStatus.OK, resp.getStatus());
List<ColumnMetadata> columns = resp.getColumnsList();
- assertEquals(5, columns.size());
+ assertEquals(6, columns.size());
verifyColumn("sys", "drillbits", "user_port", columns);
verifyColumn("sys", "drillbits", "control_port", columns);
@@ -281,7 +281,7 @@ public class TestMetadataProvider extends BaseTestQuery {
assertEquals(RequestStatus.OK, resp.getStatus());
List<ColumnMetadata> columns = resp.getColumnsList();
- assertEquals(3, columns.size());
+ assertEquals(4, columns.size());
verifyColumn("sys", "drillbits", "user_port", columns);
verifyColumn("sys", "drillbits", "control_port", columns);
@@ -302,11 +302,12 @@ public class TestMetadataProvider extends BaseTestQuery {
assertEquals(RequestStatus.OK, resp.getStatus());
List<ColumnMetadata> columns = resp.getColumnsList();
- assertEquals(3, columns.size());
+ assertEquals(4, columns.size());
verifyColumn("sys", "drillbits", "user_port", columns);
verifyColumn("sys", "drillbits", "control_port", columns);
verifyColumn("sys", "drillbits", "data_port", columns);
+ verifyColumn("sys", "drillbits", "http_port", columns);
}
/** Helper method to verify schema contents */
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/protocol/readme.txt
----------------------------------------------------------------------
diff --git a/protocol/readme.txt b/protocol/readme.txt
index 9fdaf19..db07425 100644
--- a/protocol/readme.txt
+++ b/protocol/readme.txt
@@ -24,4 +24,13 @@ download it from http://code.google.com/p/protobuf/downloads/list.
2. In protocol dir, run "mvn process-sources -P proto-compile" or "mvn clean install -P proto-compile".
-3. Check in the new/updated files.
\ No newline at end of file
+3. Check in the new/updated files.
+
+---------------------------------------------------------------
+If changes are made to the DrillClient's protobuf, you would need to regenerate the sources for the C++ client as well.
+Steps for regenerating the sources are available https://github.com/apache/drill/blob/master/contrib/native/client/
+
+You can use any of the following platforms specified in the above location to regenerate the protobuf sources:
+readme.linux : Regenerating on Linux
+readme.macos : Regenerating on MacOS
+readme.win.txt : Regenerating on Windows
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/protocol/src/main/java/org/apache/drill/exec/proto/CoordinationProtos.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/CoordinationProtos.java b/protocol/src/main/java/org/apache/drill/exec/proto/CoordinationProtos.java
index 1b10e1e..8d14ae9 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/CoordinationProtos.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/CoordinationProtos.java
@@ -111,6 +111,16 @@ public final class CoordinationProtos {
* <code>optional .exec.DrillbitEndpoint.State state = 7;</code>
*/
org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint.State getState();
+
+ // optional int32 http_port = 8;
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ boolean hasHttpPort();
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ int getHttpPort();
}
/**
* Protobuf type {@code exec.DrillbitEndpoint}
@@ -212,6 +222,11 @@ public final class CoordinationProtos {
}
break;
}
+ case 64: {
+ bitField0_ |= 0x00000080;
+ httpPort_ = input.readInt32();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -524,6 +539,22 @@ public final class CoordinationProtos {
return state_;
}
+ // optional int32 http_port = 8;
+ public static final int HTTP_PORT_FIELD_NUMBER = 8;
+ private int httpPort_;
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public boolean hasHttpPort() {
+ return ((bitField0_ & 0x00000080) == 0x00000080);
+ }
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public int getHttpPort() {
+ return httpPort_;
+ }
+
private void initFields() {
address_ = "";
userPort_ = 0;
@@ -532,6 +563,7 @@ public final class CoordinationProtos {
roles_ = org.apache.drill.exec.proto.CoordinationProtos.Roles.getDefaultInstance();
version_ = "";
state_ = org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint.State.STARTUP;
+ httpPort_ = 0;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -566,6 +598,9 @@ public final class CoordinationProtos {
if (((bitField0_ & 0x00000040) == 0x00000040)) {
output.writeEnum(7, state_.getNumber());
}
+ if (((bitField0_ & 0x00000080) == 0x00000080)) {
+ output.writeInt32(8, httpPort_);
+ }
getUnknownFields().writeTo(output);
}
@@ -603,6 +638,10 @@ public final class CoordinationProtos {
size += com.google.protobuf.CodedOutputStream
.computeEnumSize(7, state_.getNumber());
}
+ if (((bitField0_ & 0x00000080) == 0x00000080)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeInt32Size(8, httpPort_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -738,6 +777,8 @@ public final class CoordinationProtos {
bitField0_ = (bitField0_ & ~0x00000020);
state_ = org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint.State.STARTUP;
bitField0_ = (bitField0_ & ~0x00000040);
+ httpPort_ = 0;
+ bitField0_ = (bitField0_ & ~0x00000080);
return this;
}
@@ -798,6 +839,10 @@ public final class CoordinationProtos {
to_bitField0_ |= 0x00000040;
}
result.state_ = state_;
+ if (((from_bitField0_ & 0x00000080) == 0x00000080)) {
+ to_bitField0_ |= 0x00000080;
+ }
+ result.httpPort_ = httpPort_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -839,6 +884,9 @@ public final class CoordinationProtos {
if (other.hasState()) {
setState(other.getState());
}
+ if (other.hasHttpPort()) {
+ setHttpPort(other.getHttpPort());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -1266,6 +1314,39 @@ public final class CoordinationProtos {
return this;
}
+ // optional int32 http_port = 8;
+ private int httpPort_ ;
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public boolean hasHttpPort() {
+ return ((bitField0_ & 0x00000080) == 0x00000080);
+ }
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public int getHttpPort() {
+ return httpPort_;
+ }
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public Builder setHttpPort(int value) {
+ bitField0_ |= 0x00000080;
+ httpPort_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional int32 http_port = 8;</code>
+ */
+ public Builder clearHttpPort() {
+ bitField0_ = (bitField0_ & ~0x00000080);
+ httpPort_ = 0;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:exec.DrillbitEndpoint)
}
@@ -2765,21 +2846,22 @@ public final class CoordinationProtos {
descriptor;
static {
java.lang.String[] descriptorData = {
- "\n\022Coordination.proto\022\004exec\"\367\001\n\020DrillbitE" +
+ "\n\022Coordination.proto\022\004exec\"\212\002\n\020DrillbitE" +
"ndpoint\022\017\n\007address\030\001 \001(\t\022\021\n\tuser_port\030\002 " +
"\001(\005\022\024\n\014control_port\030\003 \001(\005\022\021\n\tdata_port\030\004" +
" \001(\005\022\032\n\005roles\030\005 \001(\0132\013.exec.Roles\022\017\n\007vers" +
"ion\030\006 \001(\t\022+\n\005state\030\007 \001(\0162\034.exec.Drillbit" +
- "Endpoint.State\"<\n\005State\022\013\n\007STARTUP\020\000\022\n\n\006" +
- "ONLINE\020\001\022\r\n\tQUIESCENT\020\002\022\013\n\007OFFLINE\020\003\"i\n\024" +
- "DrillServiceInstance\022\n\n\002id\030\001 \001(\t\022\033\n\023regi" +
- "strationTimeUTC\030\002 \001(\003\022(\n\010endpoint\030\003 \001(\0132" +
- "\026.exec.DrillbitEndpoint\"\227\001\n\005Roles\022\027\n\tsql",
- "_query\030\001 \001(\010:\004true\022\032\n\014logical_plan\030\002 \001(\010" +
- ":\004true\022\033\n\rphysical_plan\030\003 \001(\010:\004true\022\033\n\rj" +
- "ava_executor\030\004 \001(\010:\004true\022\037\n\021distributed_" +
- "cache\030\005 \001(\010:\004trueB3\n\033org.apache.drill.ex" +
- "ec.protoB\022CoordinationProtosH\001"
+ "Endpoint.State\022\021\n\thttp_port\030\010 \001(\005\"<\n\005Sta" +
+ "te\022\013\n\007STARTUP\020\000\022\n\n\006ONLINE\020\001\022\r\n\tQUIESCENT" +
+ "\020\002\022\013\n\007OFFLINE\020\003\"i\n\024DrillServiceInstance\022" +
+ "\n\n\002id\030\001 \001(\t\022\033\n\023registrationTimeUTC\030\002 \001(\003" +
+ "\022(\n\010endpoint\030\003 \001(\0132\026.exec.DrillbitEndpoi",
+ "nt\"\227\001\n\005Roles\022\027\n\tsql_query\030\001 \001(\010:\004true\022\032\n" +
+ "\014logical_plan\030\002 \001(\010:\004true\022\033\n\rphysical_pl" +
+ "an\030\003 \001(\010:\004true\022\033\n\rjava_executor\030\004 \001(\010:\004t" +
+ "rue\022\037\n\021distributed_cache\030\005 \001(\010:\004trueB3\n\033" +
+ "org.apache.drill.exec.protoB\022Coordinatio" +
+ "nProtosH\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -2791,7 +2873,7 @@ public final class CoordinationProtos {
internal_static_exec_DrillbitEndpoint_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_exec_DrillbitEndpoint_descriptor,
- new java.lang.String[] { "Address", "UserPort", "ControlPort", "DataPort", "Roles", "Version", "State", });
+ new java.lang.String[] { "Address", "UserPort", "ControlPort", "DataPort", "Roles", "Version", "State", "HttpPort", });
internal_static_exec_DrillServiceInstance_descriptor =
getDescriptor().getMessageTypes().get(1);
internal_static_exec_DrillServiceInstance_fieldAccessorTable = new
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/protocol/src/main/java/org/apache/drill/exec/proto/SchemaCoordinationProtos.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaCoordinationProtos.java b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaCoordinationProtos.java
index ad5897c..eb29ecc 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaCoordinationProtos.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaCoordinationProtos.java
@@ -50,6 +50,8 @@ public final class SchemaCoordinationProtos
output.writeString(6, message.getVersion(), false);
if(message.hasState())
output.writeEnum(7, message.getState().getNumber(), false);
+ if(message.hasHttpPort())
+ output.writeInt32(8, message.getHttpPort(), false);
}
public boolean isInitialized(org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint message)
{
@@ -111,6 +113,9 @@ public final class SchemaCoordinationProtos
case 7:
builder.setState(org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint.State.valueOf(input.readEnum()));
break;
+ case 8:
+ builder.setHttpPort(input.readInt32());
+ break;
default:
input.handleUnknownField(number, this);
}
@@ -158,6 +163,7 @@ public final class SchemaCoordinationProtos
case 5: return "roles";
case 6: return "version";
case 7: return "state";
+ case 8: return "httpPort";
default: return null;
}
}
@@ -176,6 +182,7 @@ public final class SchemaCoordinationProtos
fieldMap.put("roles", 5);
fieldMap.put("version", 6);
fieldMap.put("state", 7);
+ fieldMap.put("httpPort", 8);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/protocol/src/main/java/org/apache/drill/exec/proto/beans/DrillbitEndpoint.java
----------------------------------------------------------------------
diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/beans/DrillbitEndpoint.java b/protocol/src/main/java/org/apache/drill/exec/proto/beans/DrillbitEndpoint.java
index 2239c5c..0e642c5 100644
--- a/protocol/src/main/java/org/apache/drill/exec/proto/beans/DrillbitEndpoint.java
+++ b/protocol/src/main/java/org/apache/drill/exec/proto/beans/DrillbitEndpoint.java
@@ -86,6 +86,7 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
private Roles roles;
private String version;
private State state;
+ private int httpPort;
public DrillbitEndpoint()
{
@@ -185,6 +186,19 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
return this;
}
+ // httpPort
+
+ public int getHttpPort()
+ {
+ return httpPort;
+ }
+
+ public DrillbitEndpoint setHttpPort(int httpPort)
+ {
+ this.httpPort = httpPort;
+ return this;
+ }
+
// java serialization
public void readExternal(ObjectInput in) throws IOException
@@ -261,6 +275,9 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
case 7:
message.state = State.valueOf(input.readEnum());
break;
+ case 8:
+ message.httpPort = input.readInt32();
+ break;
default:
input.handleUnknownField(number, this);
}
@@ -291,6 +308,9 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
if(message.state != null)
output.writeEnum(7, message.state.number, false);
+
+ if(message.httpPort != 0)
+ output.writeInt32(8, message.httpPort, false);
}
public String getFieldName(int number)
@@ -304,6 +324,7 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
case 5: return "roles";
case 6: return "version";
case 7: return "state";
+ case 8: return "httpPort";
default: return null;
}
}
@@ -324,6 +345,7 @@ public final class DrillbitEndpoint implements Externalizable, Message<DrillbitE
__fieldMap.put("roles", 5);
__fieldMap.put("version", 6);
__fieldMap.put("state", 7);
+ __fieldMap.put("httpPort", 8);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/ae1f8389/protocol/src/main/protobuf/Coordination.proto
----------------------------------------------------------------------
diff --git a/protocol/src/main/protobuf/Coordination.proto b/protocol/src/main/protobuf/Coordination.proto
index 0bbb223..f872233 100644
--- a/protocol/src/main/protobuf/Coordination.proto
+++ b/protocol/src/main/protobuf/Coordination.proto
@@ -18,6 +18,7 @@ message DrillbitEndpoint{
OFFLINE = 3;
}
optional State state = 7;
+ optional int32 http_port = 8;
}
message DrillServiceInstance{
[4/8] drill git commit: DRILL-6339: Add a new option to disable TopN
(for testing)
Posted by pa...@apache.org.
DRILL-6339: Add a new option to disable TopN (for testing)
This closes #1219
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/69ec019c
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/69ec019c
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/69ec019c
Branch: refs/heads/master
Commit: 69ec019c85f2fffb2b6f2087e9fd59b0a7cf80e0
Parents: c79dbe9
Author: Ben-Zvi <bb...@mapr.com>
Authored: Tue Apr 17 17:05:50 2018 -0700
Committer: Parth Chandra <pa...@apache.org>
Committed: Fri Apr 20 16:53:37 2018 -0700
----------------------------------------------------------------------
.../apache/drill/exec/planner/physical/PlannerSettings.java | 1 +
.../apache/drill/exec/planner/physical/PushLimitToTopN.java | 6 ++++++
.../apache/drill/exec/server/options/SystemOptionManager.java | 1 +
exec/java-exec/src/main/resources/drill-module.conf | 1 +
4 files changed, 9 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/69ec019c/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
index 1271ff9..8b5c878 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
@@ -55,6 +55,7 @@ public class PlannerSettings implements Context{
public static final OptionValidator EXCHANGE = new BooleanValidator("planner.disable_exchanges");
public static final OptionValidator HASHAGG = new BooleanValidator("planner.enable_hashagg");
public static final OptionValidator STREAMAGG = new BooleanValidator("planner.enable_streamagg");
+ public static final OptionValidator TOPN = new BooleanValidator("planner.enable_topn");
public static final OptionValidator HASHJOIN = new BooleanValidator("planner.enable_hashjoin");
public static final OptionValidator MERGEJOIN = new BooleanValidator("planner.enable_mergejoin");
public static final OptionValidator NESTEDLOOPJOIN = new BooleanValidator("planner.enable_nestedloopjoin");
http://git-wip-us.apache.org/repos/asf/drill/blob/69ec019c/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PushLimitToTopN.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PushLimitToTopN.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PushLimitToTopN.java
index 12f7c3e..66126ec 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PushLimitToTopN.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PushLimitToTopN.java
@@ -31,6 +31,12 @@ public class PushLimitToTopN extends Prule{
}
@Override
+ public boolean matches(RelOptRuleCall call) {
+ return PrelUtil.getPlannerSettings(call.getPlanner()).getOptions()
+ .getOption(PlannerSettings.TOPN.getOptionName()).bool_val;
+ }
+
+ @Override
public void onMatch(RelOptRuleCall call) {
final LimitPrel limit = (LimitPrel) call.rel(0);
final SingleMergeExchangePrel smex = (SingleMergeExchangePrel) call.rel(1);
http://git-wip-us.apache.org/repos/asf/drill/blob/69ec019c/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 d18acb7..63f1dc2 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
@@ -80,6 +80,7 @@ public class SystemOptionManager extends BaseOptionManager implements AutoClosea
new OptionDefinition(PlannerSettings.EXCHANGE),
new OptionDefinition(PlannerSettings.HASHAGG),
new OptionDefinition(PlannerSettings.STREAMAGG),
+ new OptionDefinition(PlannerSettings.TOPN, new OptionMetaData(OptionValue.AccessibleScopes.ALL, false, true)),
new OptionDefinition(PlannerSettings.HASHJOIN),
new OptionDefinition(PlannerSettings.MERGEJOIN),
new OptionDefinition(PlannerSettings.NESTEDLOOPJOIN),
http://git-wip-us.apache.org/repos/asf/drill/blob/69ec019c/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 9af09bc..60fd639 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -483,6 +483,7 @@ drill.exec.options: {
planner.enable_nestedloopjoin: true,
planner.enable_nljoin_for_scalar_only: true,
planner.enable_streamagg: true,
+ planner.enable_topn: true,
planner.enable_type_inference: true,
planner.enable_unionall_distribute: false,
planner.filter.max_selectivity_estimate_factor: 1.0,
[3/8] drill git commit: DRILL-6334: Minor code cleanup
Posted by pa...@apache.org.
DRILL-6334: Minor code cleanup
This closes #1213
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/8e116c22
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/8e116c22
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/8e116c22
Branch: refs/heads/master
Commit: 8e116c220c782bd678290f38e93d6c5d391a9076
Parents: 71785a3
Author: Paul Rogers <pr...@cloudera.com>
Authored: Mon Apr 16 18:24:07 2018 -0700
Committer: Parth Chandra <pa...@apache.org>
Committed: Fri Apr 20 16:53:31 2018 -0700
----------------------------------------------------------------------
.../org/apache/drill/exec/ops/FragmentContext.java | 1 +
.../drill/exec/record/vector/TestValueVector.java | 15 +--------------
.../java/org/apache/drill/test/BaseTestQuery.java | 6 +++++-
.../apache/drill/test/rowSet/test/RowSetTest.java | 2 --
.../exec/exception/OversizedAllocationException.java | 2 ++
.../apache/drill/common/expression/SchemaPath.java | 1 +
6 files changed, 10 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
index ceda8b8..6a0e103 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/FragmentContext.java
@@ -158,6 +158,7 @@ public interface FragmentContext extends UdfUtilities, AutoCloseable {
DrillBuf replace(DrillBuf old, int newSize);
+ @Override
DrillBuf getManagedBuffer();
DrillBuf getManagedBuffer(int size);
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
index ba4c79c..52a1585 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/record/vector/TestValueVector.java
@@ -49,7 +49,6 @@ import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.record.MaterializedField;
-import org.apache.drill.exec.record.VectorAccessibleUtilities;
import org.apache.drill.exec.vector.BaseValueVector;
import org.apache.drill.exec.vector.BitVector;
import org.apache.drill.exec.vector.NullableFloat4Vector;
@@ -67,16 +66,15 @@ import org.apache.drill.exec.vector.complex.RepeatedMapVector;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.junit.experimental.categories.Category;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import io.netty.buffer.DrillBuf;
-import org.junit.experimental.categories.Category;
@Category(VectorTest.class)
public class TestValueVector extends ExecTest {
- //private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestValueVector.class);
private final static String EMPTY_SCHEMA_PATH = "";
@@ -103,7 +101,6 @@ public class TestValueVector extends ExecTest {
@Category(UnlikelyTest.class)
public void testFixedVectorReallocation() {
final MaterializedField field = MaterializedField.create(EMPTY_SCHEMA_PATH, UInt4Holder.TYPE);
- @SuppressWarnings("resource")
final UInt4Vector vector = new UInt4Vector(field, allocator);
// edge case 1: buffer size = max value capacity
final int expectedValueCapacity = BaseValueVector.MAX_ALLOCATION_SIZE / 4;
@@ -161,7 +158,6 @@ public class TestValueVector extends ExecTest {
@Category(UnlikelyTest.class)
public void testVariableVectorReallocation() {
final MaterializedField field = MaterializedField.create(EMPTY_SCHEMA_PATH, UInt4Holder.TYPE);
- @SuppressWarnings("resource")
final VarCharVector vector = new VarCharVector(field, allocator);
// edge case 1: value count = MAX_VALUE_ALLOCATION
final int expectedAllocationInBytes = BaseValueVector.MAX_ALLOCATION_SIZE;
@@ -254,7 +250,6 @@ public class TestValueVector extends ExecTest {
final DrillBuf newBuf = allocator.buffer(size);
final DrillBuf writeBuf = newBuf;
for(final DrillBuf buffer : buffers) {
- @SuppressWarnings("resource")
final DrillBuf readBuf = (DrillBuf) buffer.slice();
final int nBytes = readBuf.readableBytes();
final byte[] bytes = new byte[nBytes];
@@ -270,7 +265,6 @@ public class TestValueVector extends ExecTest {
final MaterializedField field = MaterializedField.create(EMPTY_SCHEMA_PATH, RepeatedIntHolder.TYPE);
// Create a new value vector.
- @SuppressWarnings("resource")
final RepeatedIntVector vector1 = new RepeatedIntVector(field, allocator);
// Populate the vector.
@@ -326,7 +320,6 @@ the interface to load has changed
final MaterializedField field = MaterializedField.create(EMPTY_SCHEMA_PATH, VarCharHolder.TYPE);
// Create a new value vector for 1024 variable length strings.
- @SuppressWarnings("resource")
final VarCharVector vector1 = new VarCharVector(field, allocator);
final VarCharVector.Mutator mutator = vector1.getMutator();
vector1.allocateNew(1024 * 10, 1024);
@@ -343,9 +336,7 @@ the interface to load has changed
// Combine the backing buffers so we can load them into a new vector.
final DrillBuf[] buffers1 = vector1.getBuffers(false);
- @SuppressWarnings("resource")
final DrillBuf buffer1 = combineBuffers(allocator, buffers1);
- @SuppressWarnings("resource")
final VarCharVector vector2 = new VarCharVector(field, allocator);
vector2.load(vector1.getMetadata(), buffer1);
@@ -368,7 +359,6 @@ the interface to load has changed
final MaterializedField field = MaterializedField.create(EMPTY_SCHEMA_PATH, NullableVarCharHolder.TYPE);
// Create a new value vector for 1024 nullable variable length strings.
- @SuppressWarnings("resource")
final NullableVarCharVector vector1 = new NullableVarCharVector(field, allocator);
final NullableVarCharVector.Mutator mutator = vector1.getMutator();
vector1.allocateNew(1024 * 10, 1024);
@@ -684,7 +674,6 @@ the interface to load has changed
}
for (int i = 0; i < valueVectors.length; i++) {
- @SuppressWarnings("resource")
final ValueVector vv = valueVectors[i];
final int vvCapacity = vv.getValueCapacity();
@@ -730,7 +719,6 @@ the interface to load has changed
*
* @param test test function to execute
*/
- @SuppressWarnings("resource")
private void testVectors(VectorVerifier test) throws Exception {
final MaterializedField[] fields = {
MaterializedField.create(EMPTY_SCHEMA_PATH, UInt4Holder.TYPE),
@@ -790,7 +778,6 @@ the interface to load has changed
@Test
public void testVectorCanLoadEmptyBuffer() throws Exception {
- @SuppressWarnings("resource")
final DrillBuf empty = allocator.getEmpty();
testVectors(new VectorVerifier() {
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java b/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
index 10cd94c..d594542 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/BaseTestQuery.java
@@ -284,6 +284,7 @@ public class BaseTestQuery extends ExecTest {
public static void closeClient() throws Exception {
if (client != null) {
client.close();
+ client = null;
}
if (bits != null) {
@@ -292,13 +293,16 @@ public class BaseTestQuery extends ExecTest {
bit.close();
}
}
+ bits = null;
}
- if(serviceSet != null) {
+ if (serviceSet != null) {
serviceSet.close();
+ serviceSet = null;
}
if (allocator != null) {
allocator.close();
+ allocator = null;
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/test/RowSetTest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/test/RowSetTest.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/test/RowSetTest.java
index 92b9266..d5ea3dc 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/test/RowSetTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/test/RowSetTest.java
@@ -430,7 +430,6 @@ public class RowSetTest extends SubOperatorTest {
// Verify that the map accessor's value count was set.
- @SuppressWarnings("resource")
MapVector mapVector = (MapVector) actual.container().getValueVector(1).getValueVector();
assertEquals(actual.rowCount(), mapVector.getAccessor().getValueCount());
@@ -574,7 +573,6 @@ public class RowSetTest extends SubOperatorTest {
// Verify that the map accessor's value count was set.
- @SuppressWarnings("resource")
RepeatedMapVector mapVector = (RepeatedMapVector) actual.container().getValueVector(1).getValueVector();
assertEquals(3, mapVector.getAccessor().getValueCount());
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/exec/vector/src/main/java/org/apache/drill/exec/exception/OversizedAllocationException.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/exception/OversizedAllocationException.java b/exec/vector/src/main/java/org/apache/drill/exec/exception/OversizedAllocationException.java
index 9142801..2a708e0 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/exception/OversizedAllocationException.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/exception/OversizedAllocationException.java
@@ -28,6 +28,8 @@ import org.apache.drill.common.exceptions.DrillRuntimeException;
*
*/
public class OversizedAllocationException extends DrillRuntimeException {
+ private static final long serialVersionUID = 1L;
+
public OversizedAllocationException() {
super();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8e116c22/logical/src/main/java/org/apache/drill/common/expression/SchemaPath.java
----------------------------------------------------------------------
diff --git a/logical/src/main/java/org/apache/drill/common/expression/SchemaPath.java b/logical/src/main/java/org/apache/drill/common/expression/SchemaPath.java
index 9ea57b1..e64d210 100644
--- a/logical/src/main/java/org/apache/drill/common/expression/SchemaPath.java
+++ b/logical/src/main/java/org/apache/drill/common/expression/SchemaPath.java
@@ -42,6 +42,7 @@ import com.google.common.base.Preconditions;
public class SchemaPath extends LogicalExpressionBase {
+ // AKA "Wildcard": expand all columns
public static final String DYNAMIC_STAR = "**";
public static final SchemaPath STAR_COLUMN = getSimplePath(DYNAMIC_STAR);
[5/8] drill git commit: DRILL-6338: Do not skip license maven plugin
when formatting generated protobuf files.
Posted by pa...@apache.org.
DRILL-6338: Do not skip license maven plugin when formatting generated protobuf files.
This closes #1215
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/c79dbe98
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/c79dbe98
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/c79dbe98
Branch: refs/heads/master
Commit: c79dbe985135754bc70e97420aa5fc456225d3a8
Parents: 8e116c2
Author: Timothy Farkas <ti...@apache.org>
Authored: Tue Apr 17 11:57:40 2018 -0700
Committer: Parth Chandra <pa...@apache.org>
Committed: Fri Apr 20 16:53:37 2018 -0700
----------------------------------------------------------------------
.travis.yml | 3 +--
protocol/pom.xml | 4 ++++
2 files changed, 5 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/c79dbe98/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index f04607f..4ca09b2 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -22,5 +22,4 @@ cache:
directories:
- "$HOME/.m2"
install: MAVEN_OPTS="-Xms1G -Xmx1G" mvn install --batch-mode -DskipTests=true -Dmaven.javadoc.skip=true -Dmaven.source.skip=true
-script: mvn install -Drat.skip=false -Dlicense.skip=false -DexcludedGroups="org.apache.drill.categories.SlowTest,org.apache.drill.categories.UnlikelyTest,org.apache.drill
-.categories.SecurityTest" -DforkCount=1 -DmemoryMb=2560 -DdirectMemoryMb=4608
+script: mvn install -Drat.skip=false -Dlicense.skip=false -DexcludedGroups="org.apache.drill.categories.SlowTest,org.apache.drill.categories.UnlikelyTest,org.apache.drill.categories.SecurityTest" -DforkCount=1 -DmemoryMb=2560 -DdirectMemoryMb=4608
http://git-wip-us.apache.org/repos/asf/drill/blob/c79dbe98/protocol/pom.xml
----------------------------------------------------------------------
diff --git a/protocol/pom.xml b/protocol/pom.xml
index 5f22b92..aaef446 100644
--- a/protocol/pom.xml
+++ b/protocol/pom.xml
@@ -55,6 +55,9 @@
<activation>
<activeByDefault>false</activeByDefault>
</activation>
+ <properties>
+ <license.skip>false</license.skip>
+ </properties>
<build>
<plugins>
<plugin>
@@ -151,6 +154,7 @@
<artifactId>license-maven-plugin</artifactId>
<executions>
<execution>
+ <id>proto-format</id>
<phase>process-sources</phase>
<goals>
<goal>format</goal>
[8/8] drill git commit: DRILL-6324: Unnest - Add tests with real
Unnest and real Lateral.
Posted by pa...@apache.org.
DRILL-6324: Unnest - Add tests with real Unnest and real Lateral.
Code cleanup, more comments, fix license headers, and more logging.
Refactor Unnest to allow setting in incoming batch after construction
fix compilation after rebase
This closes #1223
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/8d562d07
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/8d562d07
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/8d562d07
Branch: refs/heads/master
Commit: 8d562d07638bd5839fa6f27ce6973358abd7d3c7
Parents: 51bf1dd
Author: Parth Chandra <pa...@apache.org>
Authored: Wed Mar 7 13:53:14 2018 +0530
Committer: Parth Chandra <pa...@apache.org>
Committed: Sun Apr 22 22:20:57 2018 -0700
----------------------------------------------------------------------
.../drill/exec/ops/OperatorMetricRegistry.java | 2 +
.../drill/exec/physical/config/UnnestPOP.java | 18 +-
.../drill/exec/physical/impl/unnest/Unnest.java | 31 +-
.../impl/unnest/UnnestBatchCreator.java | 5 +-
.../exec/physical/impl/unnest/UnnestImpl.java | 76 +-
.../physical/impl/unnest/UnnestRecordBatch.java | 186 +++--
.../exec/record/AbstractSingleRecordBatch.java | 128 +---
.../AbstractTableFunctionRecordBatch.java | 64 ++
.../exec/record/AbstractUnaryRecordBatch.java | 167 ++++
.../exec/record/TableFunctionContract.java | 38 +
.../exec/store/mock/MockLateralJoinPOP.java | 72 --
.../impl/unnest/MockLateralJoinBatch.java | 32 +-
.../impl/unnest/TestUnnestCorrectness.java | 77 +-
.../TestUnnestWithLateralCorrectness.java | 761 +++++++++++++++++++
14 files changed, 1281 insertions(+), 376 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorMetricRegistry.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorMetricRegistry.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorMetricRegistry.java
index c1c0537..d731ca4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorMetricRegistry.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ops/OperatorMetricRegistry.java
@@ -25,6 +25,7 @@ import org.apache.drill.exec.physical.impl.flatten.FlattenRecordBatch;
import org.apache.drill.exec.physical.impl.join.HashJoinBatch;
import org.apache.drill.exec.physical.impl.mergereceiver.MergingRecordBatch;
import org.apache.drill.exec.physical.impl.partitionsender.PartitionSenderRootExec;
+import org.apache.drill.exec.physical.impl.unnest.UnnestRecordBatch;
import org.apache.drill.exec.physical.impl.unorderedreceiver.UnorderedReceiverBatch;
import org.apache.drill.exec.physical.impl.xsort.ExternalSortBatch;
import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
@@ -54,6 +55,7 @@ public class OperatorMetricRegistry {
register(CoreOperatorType.FLATTEN_VALUE, FlattenRecordBatch.Metric.class);
register(CoreOperatorType.MERGE_JOIN_VALUE, JoinBatchMemoryManager.Metric.class);
register(CoreOperatorType.LATERAL_JOIN_VALUE, JoinBatchMemoryManager.Metric.class);
+ register(CoreOperatorType.UNNEST_VALUE, UnnestRecordBatch.Metric.class);
}
private static void register(final int operatorType, final Class<? extends MetricDef> metricDef) {
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
index 2fb2b51..23e1d14 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.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
@@ -14,7 +14,7 @@
* 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.config;
import com.fasterxml.jackson.annotation.JsonCreator;
@@ -40,16 +40,12 @@ public class UnnestPOP extends AbstractSingle {
private SchemaPath column;
- @JsonProperty("lateral")
- private LateralContract lateral ;
-
@JsonCreator
public UnnestPOP(
@JsonProperty("child") PhysicalOperator child, // Operator with incoming record batch
@JsonProperty("column") SchemaPath column) {
super(child);
this.column = column;
- this.lateral = null;
}
@@ -70,19 +66,9 @@ public class UnnestPOP extends AbstractSingle {
@Override
public PhysicalOperator getNewWithChild(PhysicalOperator child) {
UnnestPOP unnest = new UnnestPOP(child, column);
- unnest.setLateral(this.getLateral());
return unnest;
}
- @JsonProperty("lateral")
- public LateralContract getLateral() {
- return lateral;
- }
-
- public void setLateral(LateralContract l) {
- this.lateral = l;
- }
-
@Override
public int getOperatorType() {
return UNNEST_VALUE;
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
index 6e2e4c4..77a2ffa 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.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,7 +17,6 @@
*/
package org.apache.drill.exec.physical.impl.unnest;
-import org.apache.drill.exec.compile.TemplateClassDefinition;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.base.LateralContract;
@@ -27,18 +26,42 @@ import org.apache.drill.exec.vector.complex.RepeatedValueVector;
import java.util.List;
+/**
+ * Placeholder for future unnest implementation that may require code generation. Current implementation does not
+ * require any
+ * @see UnnestImpl
+ */
public interface Unnest {
- TemplateClassDefinition<Unnest> TEMPLATE_DEFINITION = new TemplateClassDefinition<Unnest>(Unnest.class, UnnestImpl.class);
+ //TemplateClassDefinition<Unnest> TEMPLATE_DEFINITION = new TemplateClassDefinition<Unnest>(Unnest.class, UnnestImpl
+ // .class);
void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers,
LateralContract lateral) throws SchemaChangeException;
- int unnestRecords(int recordCount, int firstOutputIndex);
+ /**
+ * Performs the actual unnest operation.
+ * @param recordCount
+ * @return number of values in output
+ */
+ int unnestRecords(int recordCount);
+ /**
+ * Set the field to be unnested
+ * @param repeatedColumn
+ */
void setUnnestField(RepeatedValueVector repeatedColumn);
+
+ /**
+ * Set the maximum number of values allowed in the output.
+ * @param outputCount
+ */
void setOutputCount(int outputCount);
RepeatedValueVector getUnnestField();
+ /**
+ * Reset the index at which the incoming vector is being processed. Called every
+ * time a new batch comes in.
+ */
void resetGroupIndex();
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
index 5048d29..b37d2c2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.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
@@ -32,7 +32,6 @@ public class UnnestBatchCreator implements BatchCreator<UnnestPOP> {
@Override
public UnnestRecordBatch getBatch(ExecutorFragmentContext context, UnnestPOP config, List<RecordBatch> children)
throws ExecutionSetupException {
- Preconditions.checkArgument(children.size() == 1);
- return new UnnestRecordBatch(config, children.get(0), context);
+ return new UnnestRecordBatch(config, context);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
index d6f7f81..06713a5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.physical.impl.unnest;
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
@@ -31,11 +32,12 @@ import org.slf4j.LoggerFactory;
import java.util.List;
+import static org.apache.drill.exec.record.BatchSchema.SelectionVectorMode.NONE;
+
/**
* Contains the actual unnest operation. Unnest is a simple transfer operation in this impelementation.
* For use as a table function, we will need to change the logic of the unnest method to operate on
* more than one row at a time and remove any dependence on Lateral
- * {@link org.apache.drill.exec.physical.impl.flatten.FlattenTemplate}.
* This class follows the pattern of other operators that generate code at runtime. Normally this class
* would be abstract and have placeholders for doSetup and doEval. Unnest however, doesn't require code
* generation so we can simply implement the code in a simple class that looks similar to the code gen
@@ -44,8 +46,6 @@ import java.util.List;
public class UnnestImpl implements Unnest {
private static final Logger logger = LoggerFactory.getLogger(UnnestImpl.class);
- private static final int OUTPUT_ROW_COUNT = ValueVector.MAX_ROW_COUNT;
-
private ImmutableList<TransferPair> transfers;
private LateralContract lateral; // corresponding lateral Join (or other operator implementing the Lateral Contract)
private SelectionVectorMode svMode;
@@ -56,7 +56,7 @@ public class UnnestImpl implements Unnest {
* The output batch limit starts at OUTPUT_ROW_COUNT, but may be decreased
* if records are found to be large.
*/
- private int outputLimit = OUTPUT_ROW_COUNT;
+ private int outputLimit = ValueVector.MAX_ROW_COUNT;
// The index in the unnest column that is being processed.We start at zero and continue until
@@ -82,52 +82,44 @@ public class UnnestImpl implements Unnest {
}
@Override
- public final int unnestRecords(final int recordCount, final int firstOutputIndex) {
- switch (svMode) {
- case FOUR_BYTE:
- throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
-
- case TWO_BYTE:
- throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
-
- case NONE:
- if (innerValueIndex == -1) {
- innerValueIndex = 0;
- }
-
- // Current record being processed in the incoming record batch. We could keep
- // track of it ourselves, but it is better to check with the Lateral Join and get the
- // current record being processed thru the Lateral Join Contract.
- final int currentRecord = lateral.getRecordIndex();
- final int innerValueCount = accessor.getInnerValueCountAt(currentRecord);
- final int count = Math.min(Math.min(innerValueCount, outputLimit), recordCount);
-
- for (TransferPair t : transfers) {
- t.splitAndTransfer(innerValueIndex, count);
- }
- innerValueIndex += count;
- return count;
-
- default:
- throw new UnsupportedOperationException();
+ public final int unnestRecords(final int recordCount) {
+ Preconditions.checkArgument(svMode == NONE, "Unnest does not support selection vector inputs.");
+ if (innerValueIndex == -1) {
+ innerValueIndex = 0;
}
- }
- @Override public final void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing,
+ // Current record being processed in the incoming record batch. We could keep
+ // track of it ourselves, but it is better to check with the Lateral Join and get the
+ // current record being processed thru the Lateral Join Contract.
+ final int currentRecord = lateral.getRecordIndex();
+ final int innerValueCount = accessor.getInnerValueCountAt(currentRecord);
+ final int count = Math.min(Math.min(innerValueCount, outputLimit), recordCount);
+
+ logger.debug("Unnest: currentRecord: {}, innerValueCount: {}, record count: {}, output limit: {}", innerValueCount,
+ recordCount, outputLimit);
+ for (TransferPair t : transfers) {
+ t.splitAndTransfer(innerValueIndex, count);
+ }
+ innerValueIndex += count;
+ return count;
+
+ }
+
+ @Override
+ public final void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing,
List<TransferPair> transfers, LateralContract lateral) throws SchemaChangeException {
this.svMode = incoming.getSchema().getSelectionVectorMode();
- switch (svMode) {
- case FOUR_BYTE:
- throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
- case TWO_BYTE:
- throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+ if (svMode == NONE) {
+ this.transfers = ImmutableList.copyOf(transfers);
+ this.lateral = lateral;
+ } else {
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
}
- this.transfers = ImmutableList.copyOf(transfers);
- this.lateral = lateral;
}
- @Override public void resetGroupIndex() {
+ @Override
+ public void resetGroupIndex() {
this.innerValueIndex = 0;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
index 50dc99c..576594e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.physical.impl.unnest;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.expression.FieldReference;
@@ -29,12 +30,16 @@ import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.expr.ClassGenerator;
import org.apache.drill.exec.expr.CodeGenerator;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.MetricDef;
import org.apache.drill.exec.physical.base.LateralContract;
import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.record.AbstractSingleRecordBatch;
+import org.apache.drill.exec.record.AbstractTableFunctionRecordBatch;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.CloseableRecordBatch;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatchMemoryManager;
import org.apache.drill.exec.record.RecordBatchSizer;
import org.apache.drill.exec.record.TransferPair;
import org.apache.drill.exec.record.TypedFieldId;
@@ -50,7 +55,7 @@ import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK;
import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK_NEW_SCHEMA;
// TODO - handle the case where a user tries to unnest a scalar, should just return the column as is
-public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
+public class UnnestRecordBatch extends AbstractTableFunctionRecordBatch<UnnestPOP> {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnnestRecordBatch.class);
private Unnest unnest;
@@ -66,27 +71,40 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
private IterOutcome nextState = OK;
private int remainderIndex = 0;
private int recordCount;
- private long outputBatchSize;
- private LateralContract lateral;
private MaterializedField unnestFieldMetadata;
-
-
+ private final UnnestMemoryManager memoryManager;
+
+ public enum Metric implements MetricDef {
+ INPUT_BATCH_COUNT,
+ AVG_INPUT_BATCH_BYTES,
+ AVG_INPUT_ROW_BYTES,
+ INPUT_RECORD_COUNT,
+ OUTPUT_BATCH_COUNT,
+ AVG_OUTPUT_BATCH_BYTES,
+ AVG_OUTPUT_ROW_BYTES,
+ OUTPUT_RECORD_COUNT;
+
+ @Override
+ public int metricId() {
+ return ordinal();
+ }
+ }
/**
* Memory manager for Unnest. Estimates the batch size exactly like we do for Flatten.
*/
- private class UnnestMemoryManager {
- private final int outputRowCount;
- private static final int OFFSET_VECTOR_WIDTH = 4;
- private static final int WORST_CASE_FRAGMENTATION_FACTOR = 1;
- private static final int MAX_NUM_ROWS = ValueVector.MAX_ROW_COUNT;
- private static final int MIN_NUM_ROWS = 1;
-
- private UnnestMemoryManager(RecordBatch incoming, long outputBatchSize, SchemaPath unnestColumn) {
+ private class UnnestMemoryManager extends RecordBatchMemoryManager {
+
+ private UnnestMemoryManager(int outputBatchSize) {
+ super(outputBatchSize);
+ }
+
+ @Override
+ public void update() {
// Get sizing information for the batch.
- RecordBatchSizer sizer = new RecordBatchSizer(incoming);
+ setRecordBatchSizer(new RecordBatchSizer(incoming));
- final TypedFieldId typedFieldId = incoming.getValueVectorId(unnestColumn);
+ final TypedFieldId typedFieldId = incoming.getValueVectorId(popConfig.getColumn());
final MaterializedField field = incoming.getSchema().getColumn(typedFieldId.getFieldIds()[0]);
// Get column size of unnest column.
@@ -97,31 +115,39 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
// Average rowWidth of single element in the unnest list.
// subtract the offset vector size from column data size.
final int avgRowWidthSingleUnnestEntry = RecordBatchSizer
- .safeDivide(columnSize.netSize - (OFFSET_VECTOR_WIDTH * columnSize.valueCount), columnSize.elementCount);
+ .safeDivide(columnSize.getTotalNetSize() - (getOffsetVectorWidth() * columnSize.getValueCount()), columnSize
+ .getElementCount());
// Average rowWidth of outgoing batch.
final int avgOutgoingRowWidth = avgRowWidthSingleUnnestEntry;
// Number of rows in outgoing batch
- outputRowCount = Math.max(MIN_NUM_ROWS, Math.min(MAX_NUM_ROWS,
- RecordBatchSizer.safeDivide((outputBatchSize / WORST_CASE_FRAGMENTATION_FACTOR), avgOutgoingRowWidth)));
+ final int outputBatchSize = getOutputBatchSize();
+ // Number of rows in outgoing batch
+ setOutputRowCount(outputBatchSize, avgOutgoingRowWidth);
- logger.debug(
- "unnest incoming batch sizer : {}, outputBatchSize : {}," + "avgOutgoingRowWidth : {}, outputRowCount : {}",
- sizer, outputBatchSize, avgOutgoingRowWidth, outputRowCount);
- }
+ setOutgoingRowWidth(avgOutgoingRowWidth);
- public int getOutputRowCount() {
- return outputRowCount;
+ // Limit to lower bound of total number of rows possible for this batch
+ // i.e. all rows fit within memory budget.
+ setOutputRowCount(Math.min(columnSize.getElementCount(), getOutputRowCount()));
+
+ logger.debug("incoming batch size : {}", getRecordBatchSizer());
+
+ logger.debug("output batch size : {}, avg outgoing rowWidth : {}, output rowCount : {}",
+ outputBatchSize, avgOutgoingRowWidth, getOutputRowCount());
+
+ updateIncomingStats();
}
+
}
- public UnnestRecordBatch(UnnestPOP pop, RecordBatch incoming, FragmentContext context) throws OutOfMemoryException {
- super(pop, context, incoming);
- this.lateral = pop.getLateral();
+ public UnnestRecordBatch(UnnestPOP pop, FragmentContext context) throws OutOfMemoryException {
+ super(pop, context);
// get the output batch size from config.
- outputBatchSize = context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
+ int configuredBatchSize = (int) context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
+ memoryManager = new UnnestMemoryManager(configuredBatchSize);
}
@Override
@@ -129,16 +155,21 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
return recordCount;
}
-
- @Override
protected void killIncoming(boolean sendUpstream) {
+ // Kill may be received from an operator downstream of the corresponding lateral, or from
+ // a limit that is in a subqueruy between unnest and lateral. In the latter case, unnest has to handle the limit.
+ // In the former case, Lateral will handle most of the kill handling.
+
+ Preconditions.checkNotNull(lateral);
// Do not call kill on incoming. Lateral Join has the responsibility for killing incoming
if (context.getExecutorState().isFailed() || lateral.getLeftOutcome() == IterOutcome.STOP) {
+ logger.debug("Kill received. Stopping all processing");
nextState = IterOutcome.NONE ;
} else {
// if we have already processed the record, then kill from a limit has no meaning.
// if, however, we have values remaining to be emitted, and limit has been reached,
// we abandon the remainder and send an empty batch with EMIT.
+ logger.debug("Kill received from subquery. Stopping processing of current input row.");
if(hasRemainder) {
nextState = IterOutcome.EMIT;
}
@@ -150,6 +181,8 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
@Override
public IterOutcome innerNext() {
+ Preconditions.checkNotNull(lateral);
+
// Short circuit if record batch has already sent all data and is done
if (state == BatchState.DONE) {
return IterOutcome.NONE;
@@ -160,7 +193,7 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
}
if (hasRemainder) {
- return handleRemainder();
+ return doWork();
}
// We do not need to call next() unlike the other operators.
@@ -174,6 +207,7 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
try {
stats.startSetup();
hasRemainder = true; // next call to next will handle the actual data.
+ logger.debug("First batch received");
schemaChanged(); // checks if schema has changed (redundant in this case becaause it has) AND saves the
// current field metadata for check in subsequent iterations
setupNewSchema();
@@ -211,7 +245,7 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
}
- @Override
+ @Override
public VectorContainer getOutgoingContainer() {
return this.container;
}
@@ -240,32 +274,37 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
@Override
protected IterOutcome doWork() {
- final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
- popConfig.getColumn());
- unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
+ Preconditions.checkNotNull(lateral);
+ memoryManager.update();
+ unnest.setOutputCount(memoryManager.getOutputRowCount());
final int incomingRecordCount = incoming.getRecordCount();
final int currentRecord = lateral.getRecordIndex();
- // we call this in setupSchema, but we also need to call it here so we have a reference to the appropriate vector
+ // We call this in setupSchema, but we also need to call it here so we have a reference to the appropriate vector
// inside of the the unnest for the current batch
setUnnestVector();
- //expected output count is the num of values in the unnest colum array for the current record
+ //Expected output count is the num of values in the unnest colum array for the current record
final int childCount =
- incomingRecordCount == 0 ? 0 : unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord);
+ incomingRecordCount == 0 ? 0 : unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord) - remainderIndex;
- // unnest the data
- final int outputRecords = childCount == 0 ? 0 : unnest.unnestRecords(childCount, 0);
+ // Unnest the data
+ final int outputRecords = childCount == 0 ? 0 : unnest.unnestRecords(childCount);
- // Keep track of any spill over into another batch. HAppens only if you artificially set the output batch
+ logger.debug("{} values out of {} were processed.", outputRecords, childCount);
+ // Keep track of any spill over into another batch. Happens only if you artificially set the output batch
// size for unnest to a low number
if (outputRecords < childCount) {
hasRemainder = true;
- remainderIndex = outputRecords;
- this.recordCount = remainderIndex;
+ remainderIndex += outputRecords;
+ logger.debug("Output spilled into new batch. IterOutcome: OK.");
} else {
- this.recordCount = outputRecords;
+ hasRemainder = false;
+ remainderIndex = 0;
+ logger.debug("IterOutcome: EMIT.");
}
+ this.recordCount = outputRecords;
+ memoryManager.updateOutgoingStats(outputRecords);
// If the current incoming record has spilled into two batches, we return
// IterOutcome.OK so that the Lateral Join can keep calling next() until the
// entire incoming recods has been unnested. If the entire records has been
@@ -274,25 +313,6 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
}
- private IterOutcome handleRemainder() {
- final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
- popConfig.getColumn());
- unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
- final int currentRecord = lateral.getRecordIndex();
- final int remainingRecordCount =
- unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord) - remainderIndex;
- final int projRecords = unnest.unnestRecords(remainingRecordCount, 0);
- if (projRecords < remainingRecordCount) {
- this.recordCount = projRecords;
- this.remainderIndex += projRecords;
- } else {
- this.hasRemainder = false;
- this.remainderIndex = 0;
- this.recordCount = remainingRecordCount;
- }
- return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
- }
-
/**
* The data layout is the same for the actual data within a repeated field, as it is in a scalar vector for
* the same sql type. For example, a repeated int vector has a vector of offsets into a regular int vector to
@@ -333,12 +353,15 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
}
@Override protected boolean setupNewSchema() throws SchemaChangeException {
+ Preconditions.checkNotNull(lateral);
container.clear();
+ recordCount = 0;
final List<TransferPair> transfers = Lists.newArrayList();
- final NamedExpression unnestExpr =
- new NamedExpression(popConfig.getColumn(), new FieldReference(popConfig.getColumn()));
- final FieldReference fieldReference = unnestExpr.getRef();
+ //TODO: fixthis once planner changes are done
+ final FieldReference fieldReference =
+ new FieldReference(SchemaPath.getSimplePath(popConfig.getColumn().toString() + "_flat"));
+
final TransferPair transferPair = getUnnestFieldTransferPair(fieldReference);
final ValueVector unnestVector = transferPair.getTo();
@@ -364,13 +387,44 @@ public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
final TypedFieldId fieldId = incoming.getValueVectorId(popConfig.getColumn());
final MaterializedField thisField = incoming.getSchema().getColumn(fieldId.getFieldIds()[0]);
final MaterializedField prevField = unnestFieldMetadata;
+ Preconditions.checkNotNull(thisField);
unnestFieldMetadata = thisField;
// isEquivalent may return false if the order of the fields has changed. This usually does not
// happen but if it does we end up throwing a spurious schema change exeption
if (prevField == null || !prevField.isEquivalent(thisField)) {
+ logger.debug("Schema changed");
return true;
}
return false;
}
+ private void updateStats() {
+ if(memoryManager.getRecordBatchSizer() == null) {
+ return;
+ }
+ stats.setLongStat(Metric.INPUT_BATCH_COUNT, memoryManager.getNumIncomingBatches());
+ stats.setLongStat(Metric.AVG_INPUT_BATCH_BYTES, memoryManager.getAvgInputBatchSize());
+ stats.setLongStat(Metric.AVG_INPUT_ROW_BYTES, memoryManager.getAvgInputRowWidth());
+ stats.setLongStat(Metric.INPUT_RECORD_COUNT, memoryManager.getTotalInputRecords());
+ stats.setLongStat(Metric.OUTPUT_BATCH_COUNT, memoryManager.getNumOutgoingBatches());
+ stats.setLongStat(Metric.AVG_OUTPUT_BATCH_BYTES, memoryManager.getAvgOutputBatchSize());
+ stats.setLongStat(Metric.AVG_OUTPUT_ROW_BYTES, memoryManager.getAvgOutputRowWidth());
+ stats.setLongStat(Metric.OUTPUT_RECORD_COUNT, memoryManager.getTotalOutputRecords());
+
+ logger.debug("input: batch count : {}, avg batch bytes : {}, avg row bytes : {}, record count : {}",
+ memoryManager.getNumIncomingBatches(), memoryManager.getAvgInputBatchSize(),
+ memoryManager.getAvgInputRowWidth(), memoryManager.getTotalInputRecords());
+
+ logger.debug("output: batch count : {}, avg batch bytes : {}, avg row bytes : {}, record count : {}",
+ memoryManager.getNumOutgoingBatches(), memoryManager.getAvgOutputBatchSize(),
+ memoryManager.getAvgOutputRowWidth(), memoryManager.getTotalOutputRecords());
+
+ }
+
+ @Override
+ public void close() {
+ updateStats();
+ super.close();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
index 5929993..1a04b40 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractSingleRecordBatch.java
@@ -18,138 +18,26 @@
package org.apache.drill.exec.record;
import org.apache.drill.exec.exception.OutOfMemoryException;
-import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.physical.base.PhysicalOperator;
-import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
-import org.apache.drill.exec.vector.SchemaChangeCallBack;
-public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> extends AbstractRecordBatch<T> {
+/**
+ * Implements an AbstractUnaryRecordBatch where the inoming record batch is known at the time of creation
+ * @param <T>
+ */
+public abstract class AbstractSingleRecordBatch<T extends PhysicalOperator> extends AbstractUnaryRecordBatch<T> {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(new Object() {}.getClass().getEnclosingClass());
protected final RecordBatch incoming;
- protected boolean outOfMemory = false;
- protected SchemaChangeCallBack callBack = new SchemaChangeCallBack();
public AbstractSingleRecordBatch(T popConfig, FragmentContext context, RecordBatch incoming) throws OutOfMemoryException {
- super(popConfig, context, false);
+ super(popConfig, context);
this.incoming = incoming;
}
@Override
- protected void killIncoming(boolean sendUpstream) {
- incoming.kill(sendUpstream);
- }
-
- @Override
- public IterOutcome innerNext() {
- // Short circuit if record batch has already sent all data and is done
- if (state == BatchState.DONE) {
- return IterOutcome.NONE;
- }
-
- IterOutcome upstream = next(incoming);
- if (state != BatchState.FIRST && upstream == IterOutcome.OK && incoming.getRecordCount() == 0) {
- do {
- for (final VectorWrapper<?> w : incoming) {
- w.clear();
- }
- } while ((upstream = next(incoming)) == IterOutcome.OK && incoming.getRecordCount() == 0);
- }
- if ((state == BatchState.FIRST) && upstream == IterOutcome.OK) {
- upstream = IterOutcome.OK_NEW_SCHEMA;
- }
- switch (upstream) {
- case NONE:
- if (state == BatchState.FIRST) {
- return handleNullInput();
- }
- return upstream;
- case NOT_YET:
- case STOP:
- if (state == BatchState.FIRST) {
- container.buildSchema(SelectionVectorMode.NONE);
- }
- return upstream;
- case OUT_OF_MEMORY:
- return upstream;
- case OK_NEW_SCHEMA:
- if (state == BatchState.FIRST) {
- state = BatchState.NOT_FIRST;
- }
- try {
- stats.startSetup();
- if (!setupNewSchema()) {
- upstream = IterOutcome.OK;
- }
- } catch (SchemaChangeException ex) {
- kill(false);
- logger.error("Failure during query", ex);
- context.getExecutorState().fail(ex);
- return IterOutcome.STOP;
- } finally {
- stats.stopSetup();
- }
- // fall through.
- case OK:
- assert state != BatchState.FIRST : "First batch should be OK_NEW_SCHEMA";
- container.zeroVectors();
- IterOutcome out = doWork();
-
- // since doWork method does not know if there is a new schema, it will always return IterOutcome.OK if it was successful.
- // But if upstream is IterOutcome.OK_NEW_SCHEMA, we should return that
- if (out != IterOutcome.OK) {
- upstream = out;
- }
-
- if (outOfMemory) {
- outOfMemory = false;
- return IterOutcome.OUT_OF_MEMORY;
- }
-
- // Check if schema has changed
- if (callBack.getSchemaChangedAndReset()) {
- return IterOutcome.OK_NEW_SCHEMA;
- }
-
- return upstream; // change if upstream changed, otherwise normal.
- default:
- throw new UnsupportedOperationException();
- }
- }
-
- @Override
- public BatchSchema getSchema() {
- if (container.hasSchema()) {
- return container.getSchema();
- }
-
- return null;
- }
-
- protected abstract boolean setupNewSchema() throws SchemaChangeException;
- protected abstract IterOutcome doWork();
-
- /**
- * Default behavior to handle NULL input (aka FAST NONE): incoming return NONE before return a OK_NEW_SCHEMA:
- * This could happen when the underneath Scan operators do not produce any batch with schema.
- *
- * <p>
- * Notice that NULL input is different from input with an empty batch. In the later case, input provides
- * at least a batch, thought it's empty.
- *</p>
- *
- * <p>
- * This behavior could be override in each individual operator, if the operator's semantics is to
- * inject a batch with schema.
- *</p>
- *
- * @return IterOutcome.NONE.
- */
- protected IterOutcome handleNullInput() {
- container.buildSchema(SelectionVectorMode.NONE);
- container.setRecordCount(0);
- return IterOutcome.NONE;
+ protected RecordBatch getIncoming() {
+ return incoming;
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractTableFunctionRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractTableFunctionRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractTableFunctionRecordBatch.java
new file mode 100644
index 0000000..57d542c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractTableFunctionRecordBatch.java
@@ -0,0 +1,64 @@
+/*
+ * 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.record;
+
+import com.google.common.base.Preconditions;
+import org.apache.drill.exec.exception.OutOfMemoryException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+
+/**
+ * Implements AbstractUnaryRecodBatch for operators that do not have an incoming record batch available at creation
+ * time; the input is typically set up a few steps after creation. Table functions and operators like Unnest that
+ * require input before they can produce output fall into this category.
+ * Table functions can be associated with a Lateral operator in which case they simultaneously operate on the
+ * same row as the Lateral operator. In this case the LateralContract member is not null and the table function uses the
+ * lateral contract to keep in sync with the Lateral operator.
+ * @param <T>
+ */
+public abstract class AbstractTableFunctionRecordBatch<T extends PhysicalOperator> extends
+ AbstractUnaryRecordBatch<T> implements TableFunctionContract{
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(new Object() {}.getClass().getEnclosingClass());
+
+ protected RecordBatch incoming;
+ protected LateralContract lateral;
+
+ public AbstractTableFunctionRecordBatch(T popConfig, FragmentContext context) throws OutOfMemoryException {
+ super(popConfig, context);
+ lateral = null;
+ }
+
+ @Override
+ protected RecordBatch getIncoming() {
+ return incoming;
+ }
+
+ public void setIncoming(RecordBatch incoming) {
+ Preconditions.checkArgument(this.incoming == null, "Incoming is already set. setIncoming cannot be called "
+ + "multiple times.");
+ this.incoming = incoming;
+ }
+
+ public void setIncoming(LateralContract incoming) {
+ setIncoming(incoming.getIncoming());
+ lateral = incoming;
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.java
new file mode 100644
index 0000000..e941405
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/AbstractUnaryRecordBatch.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.record;
+
+import com.google.common.base.Preconditions;
+import org.apache.drill.exec.exception.OutOfMemoryException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.vector.SchemaChangeCallBack;
+
+/**
+ * The base class for operators that have a single input. The concrete implementations provide the
+ * input by implementing the getIncoming() method
+ * Known implementations: AbstractSingleRecordBatch and AbstractTableFunctionRecordBatch.
+ * @see org.apache.drill.exec.record.AbstractRecordBatch
+ * @see org.apache.drill.exec.record.AbstractSingleRecordBatch
+ * @see org.apache.drill.exec.record.AbstractTableFunctionRecordBatch
+ * @param <T>
+ */
+public abstract class AbstractUnaryRecordBatch<T extends PhysicalOperator> extends AbstractRecordBatch<T> {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(new Object() {}.getClass().getEnclosingClass());
+
+ protected boolean outOfMemory = false;
+ protected SchemaChangeCallBack callBack = new SchemaChangeCallBack();
+
+ public AbstractUnaryRecordBatch(T popConfig, FragmentContext context) throws OutOfMemoryException {
+ super(popConfig, context, false);
+ }
+
+ protected abstract RecordBatch getIncoming();
+
+ @Override
+ protected void killIncoming(boolean sendUpstream) {
+ final RecordBatch incoming = getIncoming();
+ incoming.kill(sendUpstream);
+ }
+
+ @Override
+ public IterOutcome innerNext() {
+ final RecordBatch incoming = getIncoming();
+ // Short circuit if record batch has already sent all data and is done
+ if (state == BatchState.DONE) {
+ return IterOutcome.NONE;
+ }
+
+ IterOutcome upstream = next(incoming);
+ if (state != BatchState.FIRST && upstream == IterOutcome.OK && incoming.getRecordCount() == 0) {
+ do {
+ for (final VectorWrapper<?> w : incoming) {
+ w.clear();
+ }
+ } while ((upstream = next(incoming)) == IterOutcome.OK && incoming.getRecordCount() == 0);
+ }
+ if ((state == BatchState.FIRST) && upstream == IterOutcome.OK) {
+ upstream = IterOutcome.OK_NEW_SCHEMA;
+ }
+ switch (upstream) {
+ case NONE:
+ if (state == BatchState.FIRST) {
+ return handleNullInput();
+ }
+ return upstream;
+ case NOT_YET:
+ case STOP:
+ if (state == BatchState.FIRST) {
+ container.buildSchema(SelectionVectorMode.NONE);
+ }
+ return upstream;
+ case OUT_OF_MEMORY:
+ return upstream;
+ case OK_NEW_SCHEMA:
+ if (state == BatchState.FIRST) {
+ state = BatchState.NOT_FIRST;
+ }
+ try {
+ stats.startSetup();
+ if (!setupNewSchema()) {
+ upstream = IterOutcome.OK;
+ }
+ } catch (SchemaChangeException ex) {
+ kill(false);
+ logger.error("Failure during query", ex);
+ context.getExecutorState().fail(ex);
+ return IterOutcome.STOP;
+ } finally {
+ stats.stopSetup();
+ }
+ // fall through.
+ case OK:
+ assert state != BatchState.FIRST : "First batch should be OK_NEW_SCHEMA";
+ container.zeroVectors();
+ IterOutcome out = doWork();
+
+ // since doWork method does not know if there is a new schema, it will always return IterOutcome.OK if it was successful.
+ // But if upstream is IterOutcome.OK_NEW_SCHEMA, we should return that
+ if (out != IterOutcome.OK) {
+ upstream = out;
+ }
+
+ if (outOfMemory) {
+ outOfMemory = false;
+ return IterOutcome.OUT_OF_MEMORY;
+ }
+
+ // Check if schema has changed
+ if (callBack.getSchemaChangedAndReset()) {
+ return IterOutcome.OK_NEW_SCHEMA;
+ }
+
+ return upstream; // change if upstream changed, otherwise normal.
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ public BatchSchema getSchema() {
+ if (container.hasSchema()) {
+ return container.getSchema();
+ }
+
+ return null;
+ }
+
+ protected abstract boolean setupNewSchema() throws SchemaChangeException;
+ protected abstract IterOutcome doWork();
+
+ /**
+ * Default behavior to handle NULL input (aka FAST NONE): incoming return NONE before return a OK_NEW_SCHEMA:
+ * This could happen when the underneath Scan operators do not produce any batch with schema.
+ *
+ * <p>
+ * Notice that NULL input is different from input with an empty batch. In the later case, input provides
+ * at least a batch, thought it's empty.
+ *</p>
+ *
+ * <p>
+ * This behavior could be override in each individual operator, if the operator's semantics is to
+ * inject a batch with schema.
+ *</p>
+ *
+ * @return IterOutcome.NONE.
+ */
+ protected IterOutcome handleNullInput() {
+ container.buildSchema(SelectionVectorMode.NONE);
+ container.setRecordCount(0);
+ return IterOutcome.NONE;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/record/TableFunctionContract.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/TableFunctionContract.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/TableFunctionContract.java
new file mode 100644
index 0000000..1bcd9f4
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/TableFunctionContract.java
@@ -0,0 +1,38 @@
+/*
+ * 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.record;
+
+import org.apache.drill.exec.physical.base.LateralContract;
+
+public interface TableFunctionContract {
+
+ /**
+ * sets the incoming record batch for the operator. The innerNext() method requires one of the setIncoming methods
+ * to be called before it is called.
+ * @param incoming
+ */
+ void setIncoming(RecordBatch incoming);
+
+ /**
+ * sets the incoming Lateral Contract for the operator. The innerNext() method requires one of the setIncoming methods
+ * to be called before it is called.
+ * @param incoming
+ */
+ void setIncoming(LateralContract incoming);
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
deleted file mode 100644
index 6cc1b60..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * 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.store.mock;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-import org.apache.drill.exec.physical.EndpointAffinity;
-import org.apache.drill.exec.physical.base.AbstractStore;
-import org.apache.drill.exec.physical.base.PhysicalOperator;
-import org.apache.drill.exec.physical.base.Store;
-import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
-
-import java.util.Collections;
-import java.util.List;
-
-@JsonTypeName("mock-lj")
-public class MockLateralJoinPOP extends AbstractStore {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockLateralJoinPOP.class);
-
- @JsonCreator
- public MockLateralJoinPOP(@JsonProperty("child") PhysicalOperator child) {
- super(child);
- }
-
- @Override
- public int getMaxWidth() {
- return 1;
- }
-
- @Override
- public List<EndpointAffinity> getOperatorAffinity() {
- return Collections.emptyList();
- }
-
- @Override
- public void applyAssignments(List<DrillbitEndpoint> endpoints) {
-
- }
-
- @Override
- public Store getSpecificStore(PhysicalOperator child, int minorFragmentId) {
- return new MockLateralJoinPOP(child);
- }
-
- @Override
- protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
- return new MockLateralJoinPOP(child);
- }
-
- @Override
- public int getOperatorType() {
- throw new UnsupportedOperationException();
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
index 98c9a9e..0bf995a 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
@@ -1,20 +1,20 @@
/*
-* 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.
-*/
+ * 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.unnest;
import org.apache.drill.common.expression.SchemaPath;
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
index 1a92ba0..02e4da7 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
@@ -1,20 +1,20 @@
/*
-* 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.
-*/
+ * 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.unnest;
import org.apache.drill.categories.OperatorTest;
@@ -25,12 +25,13 @@ import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.LateralContract;
import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.physical.impl.MockRecordBatch;
import org.apache.drill.exec.physical.rowSet.impl.TestResultSetLoaderMapArray;
import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.record.TupleMetadata;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
import org.apache.drill.exec.record.VectorContainer;
import org.apache.drill.exec.store.mock.MockStorePOP;
import org.apache.drill.exec.vector.ValueVector;
@@ -39,7 +40,7 @@ import org.apache.drill.exec.vector.complex.MapVector;
import org.apache.drill.test.SubOperatorTest;
import org.apache.drill.test.rowSet.RowSet;
import org.apache.drill.test.rowSet.RowSetBuilder;
-import org.apache.drill.test.rowSet.SchemaBuilder;
+import org.apache.drill.test.rowSet.schema.SchemaBuilder;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -270,9 +271,10 @@ import static org.junit.Assert.assertTrue;
@Test
public void testUnnestLimitBatchSize() {
- final int limitedOutputBatchSize = 1024;
- final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
- final int inputBatchSize = 1024+1;
+ final int limitedOutputBatchSize = 1023; // one less than the power of two. See RecordBatchMemoryManager
+ // .adjustOutputRowCount
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // (num rows+1) * size of int
+ final int inputBatchSize = 1023+1;
// single record batch with single row. The unnest column has one
// more record than the batch size we want in the output
Object[][] data = new Object[1][1];
@@ -325,9 +327,10 @@ import static org.junit.Assert.assertTrue;
// similar to previous test; we split a record across more than one batch.
// but we also set a limit less than the size of the batch so only one batch gets output.
- final int limitedOutputBatchSize = 1024;
- final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
- final int inputBatchSize = 1024+1;
+ final int limitedOutputBatchSize = 1023; // one less than the power of two. See RecordBatchMemoryManager
+ // .adjustOutputRowCount
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // (num rows+1) * size of int
+ final int inputBatchSize = 1023+1;
// single record batch with single row. The unnest column has one
// more record than the batch size we want in the output
Object[][] data = new Object[1][1];
@@ -376,15 +379,14 @@ import static org.junit.Assert.assertTrue;
// Limit sends a kill. Unnest has exactly one record batch for a record when
// the kill is sent. This test is actually useless since it tests the behaviour of
// the mock lateral which doesn't send kill at all if it gets an EMIT. We expect limit
- // to do the same so let's keep the test to demonstrate the expected behaviour.
+ // to do so, so let's keep the test to demonstrate the expected behaviour.
public void testUnnestKillFromLimitSubquery2() {
// similar to previous test but the size of the array fits exactly into the record batch;
-
-
- final int limitedOutputBatchSize = 1024;
- final int limitedOutputBatchSizeBytes = 1024*4; // num rows * size of int
- final int inputBatchSize = 1024;
+ final int limitedOutputBatchSize = 1023; // one less than the power of two. See RecordBatchMemoryManager
+ // .adjustOutputRowCount
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // (num rows+1) * size of int
+ final int inputBatchSize = 1023;
// single record batch with single row. The unnest column has one
// more record than the batch size we want in the output
Object[][] data = new Object[1][1];
@@ -486,10 +488,10 @@ import static org.junit.Assert.assertTrue;
// Create data
ArrayList<RowSet.SingleRowSet> rowSets = new ArrayList<>();
+ int rowNumber = 0;
int batchNum = 0;
for ( Object[] recordBatch : data) {
RowSetBuilder rowSetBuilder = fixture.rowSetBuilder(incomingSchemas[batchNum]);
- int rowNumber = 0;
for ( Object rowData : recordBatch) {
rowSetBuilder.addRow(++rowNumber, rowData);
}
@@ -516,12 +518,13 @@ import static org.junit.Assert.assertTrue;
final MockLateralJoinBatch lateralJoinBatch =
new MockLateralJoinBatch(fixture.getFragmentContext(), operatorContext, incomingMockBatch);
- // set pointer to Lateral in unnest pop config
- unnestPopConfig.setLateral(lateralJoinBatch);
// setup Unnest record batch
final UnnestRecordBatch unnestBatch =
- new UnnestRecordBatch(unnestPopConfig, incomingMockBatch, fixture.getFragmentContext());
+ new UnnestRecordBatch(unnestPopConfig, fixture.getFragmentContext());
+
+ // set pointer to Lateral in unnest pop config
+ unnestBatch.setIncoming((LateralContract) lateralJoinBatch);
// set backpointer to lateral join in unnest
lateralJoinBatch.setUnnest(unnestBatch);
@@ -620,9 +623,9 @@ import static org.junit.Assert.assertTrue;
TupleMetadata schema = new SchemaBuilder()
.add("rowNum", TypeProtos.MinorType.INT)
.addMapArray("unnestColumn")
- .add("colA", TypeProtos.MinorType.INT)
- .addArray("colB", TypeProtos.MinorType.VARCHAR)
- .buildMap()
+ .add("colA", TypeProtos.MinorType.INT)
+ .addArray("colB", TypeProtos.MinorType.VARCHAR)
+ .resumeSchema()
.buildSchema();
return schema;
}
[6/8] drill git commit: DRILL-6324: Unnest - Initial Implementation
Posted by pa...@apache.org.
DRILL-6324: Unnest - Initial Implementation
- Based on Flatten
- Implement unnestRecords in UnnestTemplate
- Remove unnecessary code inherited from Flatten/Project. Add schema change handling.
- Fix build failure after rebase since RecordBatchSizer used by UNNEST was relocated to a different package
- Add unit tests
- Handling of input row splitting across multiple batches. Also do not kill incoming in killIncoming.
- Schema change generated by Unnest
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/51bf1dd3
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/51bf1dd3
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/51bf1dd3
Branch: refs/heads/master
Commit: 51bf1dd3f6f5054e1922f0b22b0364ef1c0c6e1f
Parents: 69ec019
Author: Parth Chandra <pa...@apache.org>
Authored: Thu Feb 8 10:43:13 2018 +0530
Committer: Parth Chandra <pa...@apache.org>
Committed: Sun Apr 22 22:20:48 2018 -0700
----------------------------------------------------------------------
.../physical/base/AbstractPhysicalVisitor.java | 6 +
.../exec/physical/base/PhysicalVisitor.java | 2 +
.../drill/exec/physical/config/UnnestPOP.java | 90 +++
.../drill/exec/physical/impl/unnest/Unnest.java | 44 ++
.../impl/unnest/UnnestBatchCreator.java | 38 +
.../exec/physical/impl/unnest/UnnestImpl.java | 134 ++++
.../physical/impl/unnest/UnnestRecordBatch.java | 376 ++++++++++
.../exec/store/mock/MockLateralJoinPOP.java | 72 ++
.../exec/physical/impl/MockRecordBatch.java | 6 +-
.../impl/unnest/MockLateralJoinBatch.java | 237 +++++++
.../impl/unnest/TestUnnestCorrectness.java | 691 +++++++++++++++++++
11 files changed, 1693 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
index 66affeb..86a31c1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractPhysicalVisitor.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.config.StreamingAggregate;
import org.apache.drill.exec.physical.config.Trace;
import org.apache.drill.exec.physical.config.UnionAll;
+import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.physical.config.UnorderedReceiver;
import org.apache.drill.exec.physical.config.Values;
import org.apache.drill.exec.physical.config.WindowPOP;
@@ -175,6 +176,11 @@ public abstract class AbstractPhysicalVisitor<T, X, E extends Throwable> impleme
}
@Override
+ public T visitUnnest(UnnestPOP unnest, X value) throws E {
+ return visitOp(unnest, value);
+ }
+
+ @Override
public T visitIteratorValidator(IteratorValidator op, X value) throws E {
return visitOp(op, value);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
index 8480a07..5c926ad 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/PhysicalVisitor.java
@@ -34,6 +34,7 @@ import org.apache.drill.exec.physical.config.Sort;
import org.apache.drill.exec.physical.config.StreamingAggregate;
import org.apache.drill.exec.physical.config.Trace;
import org.apache.drill.exec.physical.config.UnionAll;
+import org.apache.drill.exec.physical.config.UnnestPOP;
import org.apache.drill.exec.physical.config.UnorderedReceiver;
import org.apache.drill.exec.physical.config.Values;
import org.apache.drill.exec.physical.config.WindowPOP;
@@ -76,6 +77,7 @@ public interface PhysicalVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
public RETURN visitSingleSender(SingleSender op, EXTRA value) throws EXCEP;
public RETURN visitWindowFrame(WindowPOP op, EXTRA value) throws EXCEP;
public RETURN visitProducerConsumer(ProducerConsumer op, EXTRA value) throws EXCEP;
+ public RETURN visitUnnest(UnnestPOP unnest, EXTRA value) throws EXCEP;
public RETURN visitIteratorValidator(IteratorValidator op, EXTRA value) throws EXCEP;
}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
new file mode 100644
index 0000000..2fb2b51
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/config/UnnestPOP.java
@@ -0,0 +1,90 @@
+/*******************************************************************************
+ * 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.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.google.common.collect.Iterators;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.AbstractSingle;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.PhysicalVisitor;
+import org.apache.drill.exec.proto.beans.CoreOperatorType;
+
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.drill.exec.proto.UserBitShared.CoreOperatorType.UNNEST_VALUE;
+import static org.apache.drill.exec.proto.beans.CoreOperatorType.UNNEST;
+
+@JsonTypeName("unnest")
+public class UnnestPOP extends AbstractSingle {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnnestPOP.class);
+
+ private SchemaPath column;
+
+ @JsonProperty("lateral")
+ private LateralContract lateral ;
+
+ @JsonCreator
+ public UnnestPOP(
+ @JsonProperty("child") PhysicalOperator child, // Operator with incoming record batch
+ @JsonProperty("column") SchemaPath column) {
+ super(child);
+ this.column = column;
+ this.lateral = null;
+ }
+
+
+ @Override
+ public Iterator<PhysicalOperator> iterator() {
+ return Iterators.singletonIterator(child);
+ }
+
+ public SchemaPath getColumn() {
+ return column;
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E {
+ return physicalVisitor.visitUnnest(this, value);
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ UnnestPOP unnest = new UnnestPOP(child, column);
+ unnest.setLateral(this.getLateral());
+ return unnest;
+ }
+
+ @JsonProperty("lateral")
+ public LateralContract getLateral() {
+ return lateral;
+ }
+
+ public void setLateral(LateralContract l) {
+ this.lateral = l;
+ }
+
+ @Override
+ public int getOperatorType() {
+ return UNNEST_VALUE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
new file mode 100644
index 0000000..6e2e4c4
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/Unnest.java
@@ -0,0 +1,44 @@
+/**
+ * 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.unnest;
+
+import org.apache.drill.exec.compile.TemplateClassDefinition;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+
+import java.util.List;
+
+public interface Unnest {
+ TemplateClassDefinition<Unnest> TEMPLATE_DEFINITION = new TemplateClassDefinition<Unnest>(Unnest.class, UnnestImpl.class);
+
+ void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing, List<TransferPair> transfers,
+ LateralContract lateral) throws SchemaChangeException;
+
+ int unnestRecords(int recordCount, int firstOutputIndex);
+
+ void setUnnestField(RepeatedValueVector repeatedColumn);
+ void setOutputCount(int outputCount);
+
+ RepeatedValueVector getUnnestField();
+
+ void resetGroupIndex();
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
new file mode 100644
index 0000000..5048d29
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestBatchCreator.java
@@ -0,0 +1,38 @@
+/**
+ * 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.unnest;
+
+import com.google.common.base.Preconditions;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
+import org.apache.drill.exec.physical.config.FlattenPOP;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.physical.impl.flatten.FlattenRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+
+import java.util.List;
+
+public class UnnestBatchCreator implements BatchCreator<UnnestPOP> {
+ @Override
+ public UnnestRecordBatch getBatch(ExecutorFragmentContext context, UnnestPOP config, List<RecordBatch> children)
+ throws ExecutionSetupException {
+ Preconditions.checkArgument(children.size() == 1);
+ return new UnnestRecordBatch(config, children.get(0), context);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
new file mode 100644
index 0000000..d6f7f81
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestImpl.java
@@ -0,0 +1,134 @@
+/*
+ * 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.unnest;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * Contains the actual unnest operation. Unnest is a simple transfer operation in this impelementation.
+ * For use as a table function, we will need to change the logic of the unnest method to operate on
+ * more than one row at a time and remove any dependence on Lateral
+ * {@link org.apache.drill.exec.physical.impl.flatten.FlattenTemplate}.
+ * This class follows the pattern of other operators that generate code at runtime. Normally this class
+ * would be abstract and have placeholders for doSetup and doEval. Unnest however, doesn't require code
+ * generation so we can simply implement the code in a simple class that looks similar to the code gen
+ * templates used by other operators but does not implement the doSetup and doEval methods.
+ */
+public class UnnestImpl implements Unnest {
+ private static final Logger logger = LoggerFactory.getLogger(UnnestImpl.class);
+
+ private static final int OUTPUT_ROW_COUNT = ValueVector.MAX_ROW_COUNT;
+
+ private ImmutableList<TransferPair> transfers;
+ private LateralContract lateral; // corresponding lateral Join (or other operator implementing the Lateral Contract)
+ private SelectionVectorMode svMode;
+ private RepeatedValueVector fieldToUnnest;
+ private RepeatedValueVector.RepeatedAccessor accessor;
+
+ /**
+ * The output batch limit starts at OUTPUT_ROW_COUNT, but may be decreased
+ * if records are found to be large.
+ */
+ private int outputLimit = OUTPUT_ROW_COUNT;
+
+
+ // The index in the unnest column that is being processed.We start at zero and continue until
+ // InnerValueCount is reached or if the batch limit is reached
+ // this allows for groups to be written between batches if we run out of space, for cases where we have finished
+ // a batch on the boundary it will be set to 0
+ private int innerValueIndex = 0;
+
+ @Override
+ public void setUnnestField(RepeatedValueVector unnestField) {
+ this.fieldToUnnest = unnestField;
+ this.accessor = RepeatedValueVector.RepeatedAccessor.class.cast(unnestField.getAccessor());
+ }
+
+ @Override
+ public RepeatedValueVector getUnnestField() {
+ return fieldToUnnest;
+ }
+
+ @Override
+ public void setOutputCount(int outputCount) {
+ outputLimit = outputCount;
+ }
+
+ @Override
+ public final int unnestRecords(final int recordCount, final int firstOutputIndex) {
+ switch (svMode) {
+ case FOUR_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+
+ case TWO_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+
+ case NONE:
+ if (innerValueIndex == -1) {
+ innerValueIndex = 0;
+ }
+
+ // Current record being processed in the incoming record batch. We could keep
+ // track of it ourselves, but it is better to check with the Lateral Join and get the
+ // current record being processed thru the Lateral Join Contract.
+ final int currentRecord = lateral.getRecordIndex();
+ final int innerValueCount = accessor.getInnerValueCountAt(currentRecord);
+ final int count = Math.min(Math.min(innerValueCount, outputLimit), recordCount);
+
+ for (TransferPair t : transfers) {
+ t.splitAndTransfer(innerValueIndex, count);
+ }
+ innerValueIndex += count;
+ return count;
+
+ default:
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override public final void setup(FragmentContext context, RecordBatch incoming, RecordBatch outgoing,
+ List<TransferPair> transfers, LateralContract lateral) throws SchemaChangeException {
+
+ this.svMode = incoming.getSchema().getSelectionVectorMode();
+ switch (svMode) {
+ case FOUR_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+ case TWO_BYTE:
+ throw new UnsupportedOperationException("Unnest does not support selection vector inputs.");
+ }
+ this.transfers = ImmutableList.copyOf(transfers);
+ this.lateral = lateral;
+ }
+
+ @Override public void resetGroupIndex() {
+ this.innerValueIndex = 0;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
new file mode 100644
index 0000000..50dc99c
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/unnest/UnnestRecordBatch.java
@@ -0,0 +1,376 @@
+/*
+ * 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.unnest;
+
+import com.google.common.collect.Lists;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.FieldReference;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.logical.data.NamedExpression;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.exception.OutOfMemoryException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.expr.ClassGenerator;
+import org.apache.drill.exec.expr.CodeGenerator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.record.AbstractSingleRecordBatch;
+import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.RecordBatchSizer;
+import org.apache.drill.exec.record.TransferPair;
+import org.apache.drill.exec.record.TypedFieldId;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.complex.RepeatedMapVector;
+import org.apache.drill.exec.vector.complex.RepeatedValueVector;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK;
+import static org.apache.drill.exec.record.RecordBatch.IterOutcome.OK_NEW_SCHEMA;
+
+// TODO - handle the case where a user tries to unnest a scalar, should just return the column as is
+public class UnnestRecordBatch extends AbstractSingleRecordBatch<UnnestPOP> {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(UnnestRecordBatch.class);
+
+ private Unnest unnest;
+ private boolean hasRemainder = false; // set to true if there is data left over for the current row AND if we want
+ // to keep processing it. Kill may be called by a limit in a subquery that
+ // requires us to stop processing thecurrent row, but not stop processing
+ // the data.
+ // In some cases we need to return a predetermined state from a call to next. These are:
+ // 1) Kill is called due to an error occurring in the processing of the query. IterOutcome should be NONE
+ // 2) Kill is called by LIMIT to stop processing of the current row (This occurs when the LIMIT is part of a subquery
+ // between UNNEST and LATERAL. Iteroutcome should be EMIT
+ // 3) Kill is called by LIMIT downstream from LATERAL. IterOutcome should be NONE
+ private IterOutcome nextState = OK;
+ private int remainderIndex = 0;
+ private int recordCount;
+ private long outputBatchSize;
+ private LateralContract lateral;
+ private MaterializedField unnestFieldMetadata;
+
+
+
+ /**
+ * Memory manager for Unnest. Estimates the batch size exactly like we do for Flatten.
+ */
+ private class UnnestMemoryManager {
+ private final int outputRowCount;
+ private static final int OFFSET_VECTOR_WIDTH = 4;
+ private static final int WORST_CASE_FRAGMENTATION_FACTOR = 1;
+ private static final int MAX_NUM_ROWS = ValueVector.MAX_ROW_COUNT;
+ private static final int MIN_NUM_ROWS = 1;
+
+ private UnnestMemoryManager(RecordBatch incoming, long outputBatchSize, SchemaPath unnestColumn) {
+ // Get sizing information for the batch.
+ RecordBatchSizer sizer = new RecordBatchSizer(incoming);
+
+ final TypedFieldId typedFieldId = incoming.getValueVectorId(unnestColumn);
+ final MaterializedField field = incoming.getSchema().getColumn(typedFieldId.getFieldIds()[0]);
+
+ // Get column size of unnest column.
+ RecordBatchSizer.ColumnSize columnSize = RecordBatchSizer
+ .getColumn(incoming.getValueAccessorById(field.getValueClass(), typedFieldId.getFieldIds()).getValueVector(),
+ field.getName());
+
+ // Average rowWidth of single element in the unnest list.
+ // subtract the offset vector size from column data size.
+ final int avgRowWidthSingleUnnestEntry = RecordBatchSizer
+ .safeDivide(columnSize.netSize - (OFFSET_VECTOR_WIDTH * columnSize.valueCount), columnSize.elementCount);
+
+ // Average rowWidth of outgoing batch.
+ final int avgOutgoingRowWidth = avgRowWidthSingleUnnestEntry;
+
+ // Number of rows in outgoing batch
+ outputRowCount = Math.max(MIN_NUM_ROWS, Math.min(MAX_NUM_ROWS,
+ RecordBatchSizer.safeDivide((outputBatchSize / WORST_CASE_FRAGMENTATION_FACTOR), avgOutgoingRowWidth)));
+
+ logger.debug(
+ "unnest incoming batch sizer : {}, outputBatchSize : {}," + "avgOutgoingRowWidth : {}, outputRowCount : {}",
+ sizer, outputBatchSize, avgOutgoingRowWidth, outputRowCount);
+ }
+
+ public int getOutputRowCount() {
+ return outputRowCount;
+ }
+ }
+
+
+ public UnnestRecordBatch(UnnestPOP pop, RecordBatch incoming, FragmentContext context) throws OutOfMemoryException {
+ super(pop, context, incoming);
+ this.lateral = pop.getLateral();
+ // get the output batch size from config.
+ outputBatchSize = context.getOptions().getOption(ExecConstants.OUTPUT_BATCH_SIZE_VALIDATOR);
+ }
+
+ @Override
+ public int getRecordCount() {
+ return recordCount;
+ }
+
+
+ @Override
+ protected void killIncoming(boolean sendUpstream) {
+ // Do not call kill on incoming. Lateral Join has the responsibility for killing incoming
+ if (context.getExecutorState().isFailed() || lateral.getLeftOutcome() == IterOutcome.STOP) {
+ nextState = IterOutcome.NONE ;
+ } else {
+ // if we have already processed the record, then kill from a limit has no meaning.
+ // if, however, we have values remaining to be emitted, and limit has been reached,
+ // we abandon the remainder and send an empty batch with EMIT.
+ if(hasRemainder) {
+ nextState = IterOutcome.EMIT;
+ }
+ }
+ hasRemainder = false; // whatever the case, we need to stop processing the current row.
+ }
+
+
+ @Override
+ public IterOutcome innerNext() {
+
+ // Short circuit if record batch has already sent all data and is done
+ if (state == BatchState.DONE) {
+ return IterOutcome.NONE;
+ }
+
+ if (nextState == IterOutcome.NONE || nextState == IterOutcome.EMIT) {
+ return nextState;
+ }
+
+ if (hasRemainder) {
+ return handleRemainder();
+ }
+
+ // We do not need to call next() unlike the other operators.
+ // When unnest's innerNext is called, the LateralJoin would have already
+ // updated the incoming vector.
+ // We do, however, need to call doWork() to do the actual work.
+ // We also need to handle schema build if it is the first batch
+
+ if ((state == BatchState.FIRST)) {
+ state = BatchState.NOT_FIRST;
+ try {
+ stats.startSetup();
+ hasRemainder = true; // next call to next will handle the actual data.
+ schemaChanged(); // checks if schema has changed (redundant in this case becaause it has) AND saves the
+ // current field metadata for check in subsequent iterations
+ setupNewSchema();
+ } catch (SchemaChangeException ex) {
+ kill(false);
+ logger.error("Failure during query", ex);
+ context.getExecutorState().fail(ex);
+ return IterOutcome.STOP;
+ } finally {
+ stats.stopSetup();
+ }
+ return IterOutcome.OK_NEW_SCHEMA;
+ } else {
+ assert state != BatchState.FIRST : "First batch should be OK_NEW_SCHEMA";
+ container.zeroVectors();
+
+ // Check if schema has changed
+ if (lateral.getRecordIndex() == 0 && schemaChanged()) {
+ hasRemainder = true; // next call to next will handle the actual data.
+ try {
+ setupNewSchema();
+ } catch (SchemaChangeException ex) {
+ kill(false);
+ logger.error("Failure during query", ex);
+ context.getExecutorState().fail(ex);
+ return IterOutcome.STOP;
+ }
+ return OK_NEW_SCHEMA;
+ }
+ if (lateral.getRecordIndex() == 0) {
+ unnest.resetGroupIndex();
+ }
+ return doWork();
+ }
+
+ }
+
+ @Override
+ public VectorContainer getOutgoingContainer() {
+ return this.container;
+ }
+
+ @SuppressWarnings("resource") private void setUnnestVector() {
+ final TypedFieldId typedFieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final MaterializedField field = incoming.getSchema().getColumn(typedFieldId.getFieldIds()[0]);
+ final RepeatedValueVector vector;
+ final ValueVector inVV =
+ incoming.getValueAccessorById(field.getValueClass(), typedFieldId.getFieldIds()).getValueVector();
+
+ if (!(inVV instanceof RepeatedValueVector)) {
+ if (incoming.getRecordCount() != 0) {
+ throw UserException.unsupportedError().message("Unnest does not support inputs of non-list values.")
+ .build(logger);
+ }
+ // Inherited from FLATTEN. When does this happen???
+ //when incoming recordCount is 0, don't throw exception since the type being seen here is not solid
+ logger.error("setUnnestVector cast failed and recordcount is 0, create empty vector anyway.");
+ vector = new RepeatedMapVector(field, oContext.getAllocator(), null);
+ } else {
+ vector = RepeatedValueVector.class.cast(inVV);
+ }
+ unnest.setUnnestField(vector);
+ }
+
+ @Override
+ protected IterOutcome doWork() {
+ final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
+ popConfig.getColumn());
+ unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
+ final int incomingRecordCount = incoming.getRecordCount();
+ final int currentRecord = lateral.getRecordIndex();
+ // we call this in setupSchema, but we also need to call it here so we have a reference to the appropriate vector
+ // inside of the the unnest for the current batch
+ setUnnestVector();
+
+ //expected output count is the num of values in the unnest colum array for the current record
+ final int childCount =
+ incomingRecordCount == 0 ? 0 : unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord);
+
+ // unnest the data
+ final int outputRecords = childCount == 0 ? 0 : unnest.unnestRecords(childCount, 0);
+
+ // Keep track of any spill over into another batch. HAppens only if you artificially set the output batch
+ // size for unnest to a low number
+ if (outputRecords < childCount) {
+ hasRemainder = true;
+ remainderIndex = outputRecords;
+ this.recordCount = remainderIndex;
+ } else {
+ this.recordCount = outputRecords;
+ }
+
+ // If the current incoming record has spilled into two batches, we return
+ // IterOutcome.OK so that the Lateral Join can keep calling next() until the
+ // entire incoming recods has been unnested. If the entire records has been
+ // unnested, we return EMIT and any blocking operators in the pipeline will
+ // unblock.
+ return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
+ }
+
+ private IterOutcome handleRemainder() {
+ final UnnestMemoryManager unnestMemoryManager = new UnnestMemoryManager(incoming, outputBatchSize,
+ popConfig.getColumn());
+ unnest.setOutputCount(unnestMemoryManager.getOutputRowCount());
+ final int currentRecord = lateral.getRecordIndex();
+ final int remainingRecordCount =
+ unnest.getUnnestField().getAccessor().getInnerValueCountAt(currentRecord) - remainderIndex;
+ final int projRecords = unnest.unnestRecords(remainingRecordCount, 0);
+ if (projRecords < remainingRecordCount) {
+ this.recordCount = projRecords;
+ this.remainderIndex += projRecords;
+ } else {
+ this.hasRemainder = false;
+ this.remainderIndex = 0;
+ this.recordCount = remainingRecordCount;
+ }
+ return hasRemainder ? IterOutcome.OK : IterOutcome.EMIT;
+ }
+
+ /**
+ * The data layout is the same for the actual data within a repeated field, as it is in a scalar vector for
+ * the same sql type. For example, a repeated int vector has a vector of offsets into a regular int vector to
+ * represent the lists. As the data layout for the actual values in the same in the repeated vector as in the
+ * scalar vector of the same type, we can avoid making individual copies for the column being unnested, and just
+ * use vector copies between the inner vector of the repeated field to the resulting scalar vector from the unnest
+ * operation. This is completed after we determine how many records will fit (as we will hit either a batch end, or
+ * the end of one of the other vectors while we are copying the data of the other vectors alongside each new unnested
+ * value coming out of the repeated field.)
+ */
+ @SuppressWarnings("resource") private TransferPair getUnnestFieldTransferPair(FieldReference reference) {
+ final TypedFieldId fieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final Class<?> vectorClass = incoming.getSchema().getColumn(fieldId.getFieldIds()[0]).getValueClass();
+ final ValueVector unnestField = incoming.getValueAccessorById(vectorClass, fieldId.getFieldIds()).getValueVector();
+
+ TransferPair tp = null;
+ if (unnestField instanceof RepeatedMapVector) {
+ tp = ((RepeatedMapVector) unnestField)
+ .getTransferPairToSingleMap(reference.getAsNamePart().getName(), oContext.getAllocator());
+ } else if (!(unnestField instanceof RepeatedValueVector)) {
+ if (incoming.getRecordCount() != 0) {
+ throw UserException.unsupportedError().message("Unnest does not support inputs of non-list values.")
+ .build(logger);
+ }
+ logger.error("Cannot cast {} to RepeatedValueVector", unnestField);
+ //when incoming recordCount is 0, don't throw exception since the type being seen here is not solid
+ final ValueVector vv = new RepeatedMapVector(unnestField.getField(), oContext.getAllocator(), null);
+ tp = RepeatedValueVector.class.cast(vv)
+ .getTransferPair(reference.getAsNamePart().getName(), oContext.getAllocator());
+ } else {
+ final ValueVector vvIn = RepeatedValueVector.class.cast(unnestField).getDataVector();
+ // vvIn may be null because of fast schema return for repeated list vectors
+ if (vvIn != null) {
+ tp = vvIn.getTransferPair(reference.getAsNamePart().getName(), oContext.getAllocator());
+ }
+ }
+ return tp;
+ }
+
+ @Override protected boolean setupNewSchema() throws SchemaChangeException {
+ container.clear();
+ final List<TransferPair> transfers = Lists.newArrayList();
+
+ final NamedExpression unnestExpr =
+ new NamedExpression(popConfig.getColumn(), new FieldReference(popConfig.getColumn()));
+ final FieldReference fieldReference = unnestExpr.getRef();
+ final TransferPair transferPair = getUnnestFieldTransferPair(fieldReference);
+
+ final ValueVector unnestVector = transferPair.getTo();
+ transfers.add(transferPair);
+ container.add(unnestVector);
+ logger.debug("Added transfer for unnest expression.");
+ container.buildSchema(SelectionVectorMode.NONE);
+
+ this.unnest = new UnnestImpl();
+ unnest.setup(context, incoming, this, transfers, lateral);
+ setUnnestVector();
+ return true;
+ }
+
+ /**
+ * Compares the schema of the unnest column in the current incoming with the schema of
+ * the unnest column in the previous incoming.
+ * Also saves the schema for comparison in future iterations
+ *
+ * @return true if the schema has changed, false otherwise
+ */
+ private boolean schemaChanged() {
+ final TypedFieldId fieldId = incoming.getValueVectorId(popConfig.getColumn());
+ final MaterializedField thisField = incoming.getSchema().getColumn(fieldId.getFieldIds()[0]);
+ final MaterializedField prevField = unnestFieldMetadata;
+ unnestFieldMetadata = thisField;
+ // isEquivalent may return false if the order of the fields has changed. This usually does not
+ // happen but if it does we end up throwing a spurious schema change exeption
+ if (prevField == null || !prevField.isEquivalent(thisField)) {
+ return true;
+ }
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
new file mode 100644
index 0000000..6cc1b60
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/mock/MockLateralJoinPOP.java
@@ -0,0 +1,72 @@
+/*
+ * 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.store.mock;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.exec.physical.EndpointAffinity;
+import org.apache.drill.exec.physical.base.AbstractStore;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+
+import java.util.Collections;
+import java.util.List;
+
+@JsonTypeName("mock-lj")
+public class MockLateralJoinPOP extends AbstractStore {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockLateralJoinPOP.class);
+
+ @JsonCreator
+ public MockLateralJoinPOP(@JsonProperty("child") PhysicalOperator child) {
+ super(child);
+ }
+
+ @Override
+ public int getMaxWidth() {
+ return 1;
+ }
+
+ @Override
+ public List<EndpointAffinity> getOperatorAffinity() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void applyAssignments(List<DrillbitEndpoint> endpoints) {
+
+ }
+
+ @Override
+ public Store getSpecificStore(PhysicalOperator child, int minorFragmentId) {
+ return new MockLateralJoinPOP(child);
+ }
+
+ @Override
+ protected PhysicalOperator getNewWithChild(PhysicalOperator child) {
+ return new MockLateralJoinPOP(child);
+ }
+
+ @Override
+ public int getOperatorType() {
+ throw new UnsupportedOperationException();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
index f9af887..9ed9848 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/MockRecordBatch.java
@@ -37,7 +37,7 @@ public class MockRecordBatch implements CloseableRecordBatch {
private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(MockRecordBatch.class);
// These resources are owned by this RecordBatch
- private VectorContainer container;
+ protected VectorContainer container;
private int currentContainerIndex;
private int currentOutcomeIndex;
private boolean isDone;
@@ -46,8 +46,8 @@ public class MockRecordBatch implements CloseableRecordBatch {
private final List<VectorContainer> allTestContainers;
private final List<IterOutcome> allOutcomes;
private final FragmentContext context;
- private final OperatorContext oContext;
- private final BufferAllocator allocator;
+ protected final OperatorContext oContext;
+ protected final BufferAllocator allocator;
public MockRecordBatch(FragmentContext context, OperatorContext oContext,
List<VectorContainer> testContainers, List<IterOutcome> iterOutcomes,
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.java
new file mode 100644
index 0000000..98c9a9e
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/MockLateralJoinBatch.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.unnest;
+
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.CloseableRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TypedFieldId;
+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.vector.ValueVector;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * A mock lateral join implementation for testing unnest. This ignores all the other input and
+ * simply puts the unnest output into a results hypervector.
+ * Since Unnest returns an empty batch when it encounters a schema change, this implementation
+ * will also output an empty batch when it sees a schema change from unnest
+ */
+public class MockLateralJoinBatch implements LateralContract, CloseableRecordBatch {
+
+ private RecordBatch incoming;
+
+ private int recordIndex = 0;
+ private RecordBatch unnest;
+ private int unnestLimit = -1; // Unnest will EMIT if the number of records cross this limit
+
+ private boolean isDone;
+ private IterOutcome currentLeftOutcome = IterOutcome.NOT_YET;
+
+ private final FragmentContext context;
+ private final OperatorContext oContext;
+
+ private List<ValueVector> resultList = new ArrayList<>();
+
+ public MockLateralJoinBatch(FragmentContext context, OperatorContext oContext, RecordBatch incoming) {
+ this.context = context;
+ this.oContext = oContext;
+ this.incoming = incoming;
+ this.isDone = false;
+ }
+
+ @Override public RecordBatch getIncoming() {
+ return incoming; // don't need this
+ }
+
+ @Override public int getRecordIndex() {
+ return recordIndex;
+ }
+
+ @Override public IterOutcome getLeftOutcome() {
+ return currentLeftOutcome;
+ }
+
+ public void moveToNextRecord() {
+ recordIndex++;
+ }
+
+ public void reset() {
+ recordIndex = 0;
+ }
+
+ public void setUnnest(RecordBatch unnest){
+ this.unnest = unnest;
+ }
+
+ public void setUnnestLimit(int limit){
+ this.unnestLimit = limit;
+ }
+
+ public RecordBatch getUnnest() {
+ return unnest;
+ }
+
+ public IterOutcome next() {
+
+ IterOutcome currentOutcome = incoming.next();
+ currentLeftOutcome = currentOutcome;
+ recordIndex = 0;
+
+ switch (currentOutcome) {
+ case OK_NEW_SCHEMA:
+ // Nothing to do for this.
+ case OK:
+ IterOutcome outcome;
+ // consume all the outout from unnest until EMIT or end of
+ // incoming data
+ int unnestCount = 0; // number of values unnested for current record
+ while (recordIndex < incoming.getRecordCount()) {
+ outcome = unnest.next();
+ if (outcome == IterOutcome.OK_NEW_SCHEMA) {
+ // setup schema does nothing (this is just a place holder)
+ setupSchema();
+ // however unnest is also expected to return an empty batch
+ // which we will add to our output
+ }
+ // We put each batch output from unnest into a hypervector
+ // the calling test can match this against the baseline
+ //unnestCount +=
+ // unnest.getOutgoingContainer().hasRecordCount() ? unnest.getOutgoingContainer().getRecordCount() : 0;
+ unnestCount += addBatchToHyperContainer(unnest);
+ if (outcome == IterOutcome.EMIT) {
+ // reset unnest count
+ unnestCount = 0;
+ moveToNextRecord();
+ }
+ // Pretend that an operator somewhere between lateral and unnest
+ // wants to terminate processing of the record.
+ if(unnestLimit > 0 && unnestCount >= unnestLimit) {
+ unnest.kill(true);
+ }
+ }
+ return currentOutcome;
+ case NONE:
+ case STOP:
+ case OUT_OF_MEMORY:
+ isDone = true;
+ return currentOutcome;
+ case NOT_YET:
+ return currentOutcome;
+ default:
+ throw new UnsupportedOperationException("This state is not supported");
+ }
+ }
+
+ @Override public WritableBatch getWritableBatch() {
+ return null;
+ }
+
+ public List<ValueVector> getResultList() {
+ return resultList;
+ }
+
+ @Override
+ public void close() throws Exception {
+
+ }
+
+ @Override public int getRecordCount() {
+ return 0;
+ }
+
+ @Override
+ public SelectionVector2 getSelectionVector2() {
+ return null;
+ }
+
+ @Override
+ public SelectionVector4 getSelectionVector4() {
+ return null;
+ }
+
+ @Override
+ public FragmentContext getContext() {
+ return context;
+ }
+
+ @Override public BatchSchema getSchema() {
+ return null;
+ }
+
+ @Override public void kill(boolean sendUpstream) {
+ unnest.kill(sendUpstream);
+ }
+
+ @Override public VectorContainer getOutgoingContainer() {
+ return null;
+ }
+
+ @Override public TypedFieldId getValueVectorId(SchemaPath path) {
+ return null;
+ }
+
+ @Override public VectorWrapper<?> getValueAccessorById(Class<?> clazz, int... ids) {
+ return null;
+ }
+
+ private void setupSchema(){
+ // Nothing to do in this test
+ return;
+ }
+
+ public boolean isCompleted() {
+ return isDone;
+ }
+
+ // returns number of records added to output hyper container
+ private int addBatchToHyperContainer(RecordBatch inputBatch) {
+ int count = 0;
+ final RecordBatchData batchCopy = new RecordBatchData(inputBatch, oContext.getAllocator());
+ boolean success = false;
+ try {
+ for (VectorWrapper<?> w : batchCopy.getContainer()) {
+ ValueVector vv = w.getValueVector();
+ count += vv.getAccessor().getValueCount();
+ resultList.add(vv);
+ }
+ success = true;
+ } finally {
+ if (!success) {
+ batchCopy.clear();
+ }
+ }
+ return count;
+ }
+
+
+ @Override public Iterator<VectorWrapper<?>> iterator() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/51bf1dd3/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
new file mode 100644
index 0000000..1a92ba0
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestCorrectness.java
@@ -0,0 +1,691 @@
+/*
+* 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.unnest;
+
+import org.apache.drill.categories.OperatorTest;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.PathSegment;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.physical.impl.MockRecordBatch;
+import org.apache.drill.exec.physical.rowSet.impl.TestResultSetLoaderMapArray;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.TupleMetadata;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.store.mock.MockStorePOP;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.exec.vector.complex.MapVector;
+import org.apache.drill.test.SubOperatorTest;
+import org.apache.drill.test.rowSet.RowSet;
+import org.apache.drill.test.rowSet.RowSetBuilder;
+import org.apache.drill.test.rowSet.SchemaBuilder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static junit.framework.TestCase.fail;
+import static org.junit.Assert.assertTrue;
+
+@Category(OperatorTest.class) public class TestUnnestCorrectness extends SubOperatorTest {
+
+
+ // Operator Context for mock batch
+ public static OperatorContext operatorContext;
+
+ // use MockLateralJoinPop for MockRecordBatch ??
+ public static PhysicalOperator mockPopConfig;
+
+
+ @BeforeClass public static void setUpBeforeClass() throws Exception {
+ mockPopConfig = new MockStorePOP(null);
+ operatorContext = fixture.newOperatorContext(mockPopConfig);
+ }
+
+ @AfterClass public static void tearDownAfterClass() throws Exception {
+ operatorContext.close();
+ }
+
+ @Test
+ public void testUnnestFixedWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new int[] {1, 2},
+ (Object) new int[] {3, 4, 5}},
+ { (Object) new int[] {6, 7, 8, 9},
+ (Object) new int[] {10, 11, 12, 13, 14}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ Integer[][] baseline = {{}, {1, 2}, {3, 4, 5}, {6, 7, 8, 9}, {10, 11, 12, 13, 14}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestVarWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new String[] {"", "zero"},
+ (Object) new String[] {"one", "two", "three"}},
+ { (Object) new String[] {"four", "five", "six", "seven"},
+ (Object) new String[] {"eight", "nine", "ten", "eleven", "twelve"}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ String[][] baseline = {{}, {"", "zero"}, {"one", "two", "three"}, {"four", "five", "six", "seven"},
+ {"eight", "nine", "ten", "eleven", "twelve"}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestMapColumn() {
+
+ Object[][] data = getMapData();
+
+ // Create input schema
+ TupleMetadata incomingSchema = getRepeatedMapSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ Object[][] baseline = getMapBaseline();
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestEmptyList() {
+
+ Object[][] data = {
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ },
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ // All subsequent batches are also empty
+ String[][] baseline = {{}, {}, {}, {}, {}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestMultipleNewSchemaIncoming() {
+
+ // Schema changes in incoming have no effect on unnest unless the type of the
+ // unnest column itself has changed
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new String[] {"9"}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema, incomingSchema};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ String[][] baseline = {{}, {"0", "1"}, {"2", "3", "4"}, {"5", "6" }, {"9"} };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestSchemaChange() {
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new int[] {9}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema1 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata incomingSchema2 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema1, incomingSchema1, incomingSchema2};
+
+ // First batch in baseline is an empty batch corresponding to OK_NEW_SCHEMA
+ // Another empty batch introduced by the schema change in the last batch
+ Object[][] baseline = {{}, {"0", "1"}, {"2", "3", "4"}, {"5", "6" }, {}, {9} };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestLimitBatchSize() {
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
+ final int inputBatchSize = 1024+1;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[3][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ baseline[2] = new Integer[1];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+ baseline[2][0] = limitedOutputBatchSize;
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has more than one record batch for a record when
+ // the kill is sent.
+ public void testUnnestKillFromLimitSubquery1() {
+
+ // similar to previous test; we split a record across more than one batch.
+ // but we also set a limit less than the size of the batch so only one batch gets output.
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4*1; // num rows * size of int
+ final int inputBatchSize = 1024+1;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[3][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ baseline[2] = new Integer[1];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+ baseline[2] = new Integer[] {}; // because of kill the next batch is an empty batch
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, 100, -1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has exactly one record batch for a record when
+ // the kill is sent. This test is actually useless since it tests the behaviour of
+ // the mock lateral which doesn't send kill at all if it gets an EMIT. We expect limit
+ // to do the same so let's keep the test to demonstrate the expected behaviour.
+ public void testUnnestKillFromLimitSubquery2() {
+
+ // similar to previous test but the size of the array fits exactly into the record batch;
+
+
+ final int limitedOutputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*4; // num rows * size of int
+ final int inputBatchSize = 1024;
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][] baseline = new Integer[2][];
+ baseline[0] = new Integer[] {};
+ baseline[1] = new Integer[limitedOutputBatchSize];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[1][i] = i;
+ }
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, 100, -1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+
+ @Test
+ public void testUnnestNonArrayColumn() {
+
+ Object[][] data = {
+ { (Object) new Integer (1),
+ (Object) new Integer (3)},
+ { (Object) new Integer (6),
+ (Object) new Integer (10)}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .add("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ // We expect an Exception
+ Integer[][] baseline = {};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (UserException e) {
+ return; // succeeded
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+
+ // test unnest for various input conditions without invoking kill
+ private <T> void testUnnest(
+ TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ T[][] data,
+ T[][] baseline ) throws Exception{
+ testUnnest(incomingSchemas, iterOutcomes, -1, -1, data, baseline);
+ }
+
+ // test unnest for various input conditions optionally invoking kill. if the kill or killBatch
+ // parameter is greater than 0 then the record batch is sent a kill after that many batches have been processed
+ private <T> void testUnnest( TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ int unnestLimit, // kill unnest after every 'unnestLimit' number of values in every record
+ int execKill, // number of batches after which to kill the execution (!)
+ T[][] data,
+ T[][] baseline) throws Exception {
+
+ // Get the incoming container with dummy data for LJ
+ final List<VectorContainer> incomingContainer = new ArrayList<>(data.length);
+
+ // Create data
+ ArrayList<RowSet.SingleRowSet> rowSets = new ArrayList<>();
+
+ int batchNum = 0;
+ for ( Object[] recordBatch : data) {
+ RowSetBuilder rowSetBuilder = fixture.rowSetBuilder(incomingSchemas[batchNum]);
+ int rowNumber = 0;
+ for ( Object rowData : recordBatch) {
+ rowSetBuilder.addRow(++rowNumber, rowData);
+ }
+ RowSet.SingleRowSet rowSet = rowSetBuilder.build();
+ rowSets.add(rowSet);
+ incomingContainer.add(rowSet.container());
+ batchNum++;
+ }
+
+ // Get the unnest POPConfig
+ final UnnestPOP unnestPopConfig = new UnnestPOP(null, new SchemaPath(new PathSegment.NameSegment("unnestColumn")));
+
+ // Get the IterOutcomes for LJ
+ final List<RecordBatch.IterOutcome> outcomes = new ArrayList<>(iterOutcomes.length);
+ for(RecordBatch.IterOutcome o : iterOutcomes) {
+ outcomes.add(o);
+ }
+
+ // Create incoming MockRecordBatch
+ final MockRecordBatch incomingMockBatch =
+ new MockRecordBatch(fixture.getFragmentContext(), operatorContext, incomingContainer, outcomes,
+ incomingContainer.get(0).getSchema());
+
+ final MockLateralJoinBatch lateralJoinBatch =
+ new MockLateralJoinBatch(fixture.getFragmentContext(), operatorContext, incomingMockBatch);
+
+ // set pointer to Lateral in unnest pop config
+ unnestPopConfig.setLateral(lateralJoinBatch);
+
+ // setup Unnest record batch
+ final UnnestRecordBatch unnestBatch =
+ new UnnestRecordBatch(unnestPopConfig, incomingMockBatch, fixture.getFragmentContext());
+
+ // set backpointer to lateral join in unnest
+ lateralJoinBatch.setUnnest(unnestBatch);
+ lateralJoinBatch.setUnnestLimit(unnestLimit);
+
+ // Simulate the pipeline by calling next on the incoming
+ List<ValueVector> results = null;
+ int batchesProcessed = 0;
+ try {
+ while (!isTerminal(lateralJoinBatch.next())) {
+ batchesProcessed++;
+ if (batchesProcessed == execKill) {
+ lateralJoinBatch.getContext().getExecutorState().fail(new DrillException("Testing failure of execution."));
+ lateralJoinBatch.kill(true);
+ }
+ // else nothing to do
+ }
+
+ // Check results against baseline
+ results = lateralJoinBatch.getResultList();
+
+ int i = 0;
+ for (ValueVector vv : results) {
+ int valueCount = vv.getAccessor().getValueCount();
+ if (valueCount != baseline[i].length) {
+ fail("Test failed in validating unnest output. Value count mismatch.");
+ }
+ for (int j = 0; j < valueCount; j++) {
+
+ if (vv instanceof MapVector) {
+ if (!compareMapBaseline((Object) baseline[i][j], vv.getAccessor().getObject(j))) {
+ fail("Test failed in validating unnest(Map) output. Value mismatch");
+ }
+ } else if (vv instanceof VarCharVector) {
+ Object val = vv.getAccessor().getObject(j);
+ if (((String) baseline[i][j]).compareTo(val.toString()) != 0) {
+ fail("Test failed in validating unnest output. Value mismatch. Baseline value[]" + i + "][" + j + "]"
+ + ": " + baseline[i][j] + " VV.getObject(j): " + val);
+ }
+ } else {
+ Object val = vv.getAccessor().getObject(j);
+ if (!baseline[i][j].equals(val)) {
+ fail(
+ "Test failed in validating unnest output. Value mismatch. Baseline value[" + i + "][" + j + "]" + ": "
+ + baseline[i][j] + " VV.getObject(j): " + val);
+ }
+ }
+ }
+ i++;
+
+ }
+
+ assertTrue(((MockLateralJoinBatch) lateralJoinBatch).isCompleted());
+
+ } catch (UserException e) {
+ throw e; // Valid exception
+ } catch (Exception e) {
+ fail("Test failed in validating unnest output. Exception : " + e.getMessage());
+ } finally {
+ // Close all the resources for this test case
+ unnestBatch.close();
+ lateralJoinBatch.close();
+ incomingMockBatch.close();
+
+ if (results != null) {
+ for (ValueVector vv : results) {
+ vv.clear();
+ }
+ }
+ for(RowSet.SingleRowSet rowSet: rowSets) {
+ rowSet.clear();
+ }
+ }
+
+ }
+
+ /**
+ * Build a schema with a repeated map -
+ *
+ * {
+ * rowNum,
+ * mapColumn : [
+ * {
+ * colA,
+ * colB : [
+ * varcharCol
+ * ]
+ * }
+ * ]
+ * }
+ *
+ * @see TestResultSetLoaderMapArray TestResultSetLoaderMapArray for similar schema and data
+ * @return TupleMetadata corresponding to the schema
+ */
+ private TupleMetadata getRepeatedMapSchema() {
+ TupleMetadata schema = new SchemaBuilder()
+ .add("rowNum", TypeProtos.MinorType.INT)
+ .addMapArray("unnestColumn")
+ .add("colA", TypeProtos.MinorType.INT)
+ .addArray("colB", TypeProtos.MinorType.VARCHAR)
+ .buildMap()
+ .buildSchema();
+ return schema;
+ }
+
+ private Object[][] getMapData( ) {
+
+ Object[][] d = {
+ {
+ new Object[] {},
+ new Object[] {
+ new Object[] {11, new String[] {"1.1.1", "1.1.2" }},
+ new Object[] {12, new String[] {"1.2.1", "1.2.2" }}
+ },
+
+ new Object[] {
+ new Object[] {21, new String[] {"2.1.1", "2.1.2" }},
+ new Object[] {22, new String[] {}},
+ new Object[] {23, new String[] {"2.3.1", "2.3.2" }}
+ }
+ },
+ {
+ new Object[] {
+ new Object[] {31, new String[] {"3.1.1", "3.1.2" }},
+ new Object[] {32, new String[] {"3.2.1", "3.2.2" }}
+ }
+ }
+ };
+
+ return d;
+ }
+
+ private Object[][] getMapBaseline() {
+
+ Object[][] d = {
+ new Object[] {}, // Empty record batch returned by OK_NEW_SCHEMA
+ new Object[] {}, // First incoming batch is empty
+ new Object[] {
+ "{\"colA\":11,\"colB\":[\"1.1.1\",\"1.1.2\"]}",
+ "{\"colA\":12,\"colB\":[\"1.2.1\",\"1.2.2\"]}"
+ },
+ new Object[] {
+ "{\"colA\":21,\"colB\":[\"2.1.1\",\"2.1.2\"]}",
+ "{\"colA\":22,\"colB\":[]}",
+ "{\"colA\":23,\"colB\":[\"2.3.1\",\"2.3.2\"]}"
+ },
+ new Object[] {
+ "{\"colA\":31,\"colB\":[\"3.1.1\",\"3.1.2\"]}",
+ "{\"colA\":32,\"colB\":[\"3.2.1\",\"3.2.2\"]}"
+ }
+ };
+ return d;
+ }
+
+ private boolean compareMapBaseline(Object baselineValue, Object vector) {
+ String vv = vector.toString();
+ String b = (String)baselineValue;
+ return vv.equalsIgnoreCase(b);
+ }
+
+ private boolean isTerminal(RecordBatch.IterOutcome outcome) {
+ return (outcome == RecordBatch.IterOutcome.NONE || outcome == RecordBatch.IterOutcome.STOP) || (outcome
+ == RecordBatch.IterOutcome.OUT_OF_MEMORY);
+ }
+
+}
+
[2/8] drill git commit: DRILL-6333: Fixed Quotation marks
Posted by pa...@apache.org.
DRILL-6333: Fixed Quotation marks
Initial step to making the source-code ready for Javadoc generation
This closes #1229
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/71785a36
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/71785a36
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/71785a36
Branch: refs/heads/master
Commit: 71785a365a99c9fccfa5cebb9acce67afd7e3e26
Parents: ae1f838
Author: Kunal Khatua <kk...@maprtech.com>
Authored: Fri Apr 20 11:01:51 2018 -0700
Committer: Parth Chandra <pa...@apache.org>
Committed: Fri Apr 20 16:31:38 2018 -0700
----------------------------------------------------------------------
drill-yarn/USAGE.md | 42 ++++++++++----------
.../apache/drill/yarn/scripts/TestScripts.java | 2 +-
.../org/apache/drill/exec/ExecConstants.java | 4 +-
.../exec/expr/fn/impl/CryptoFunctions.java | 2 +-
.../impl/xsort/managed/ExternalSortBatch.java | 4 +-
.../rpc/user/InboundImpersonationManager.java | 8 ++--
.../vector/accessor/writer/package-info.java | 18 ++++-----
7 files changed, 40 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/drill-yarn/USAGE.md
----------------------------------------------------------------------
diff --git a/drill-yarn/USAGE.md b/drill-yarn/USAGE.md
index db7fb4b..5e0efc1 100644
--- a/drill-yarn/USAGE.md
+++ b/drill-yarn/USAGE.md
@@ -2,7 +2,7 @@
## Introduction
-Drill's YARN integration launches your Drill cluster through YARN. Drill becomes a long-running application with YARN. When you launch Drill, YARN automatically deploys (“localizes”) Drill software onto each node, avoiding the need to pre-install Drill on each node. Resource management is simplified because YARN is now aware of resources dedicated to Drill.
+Drill's YARN integration launches your Drill cluster through YARN. Drill becomes a long-running application with YARN. When you launch Drill, YARN automatically deploys ("localizes") Drill software onto each node, avoiding the need to pre-install Drill on each node. Resource management is simplified because YARN is now aware of resources dedicated to Drill.
The discussion below assumes that you are familiar with the main YARN concepts: the Resource Manager (RM), Node Manager (NM) and so on.
@@ -12,7 +12,7 @@ The discussion also assumes that you already have a working Drill cluster that y
Drill, by design, aggressively uses all available resources to run queries at optimal speed. When running Drill under YARN, you inform YARN of the resources that Drill will consume. These settings are descriptive, not proscriptive. That is, Drill does not limit itself to the YARN settings; instead the YARN settings inform YARN of the resources that Drill will consume so that YARN does not over-allocate those same resources to other tasks.
-All YARN distributions provide settings for memory and CPU (called “vcores” by YARN). Some distributions also provide disk settings.
+All YARN distributions provide settings for memory and CPU (called "vcores" by YARN). Some distributions also provide disk settings.
For memory, you first configure Drill’s memory as described below, then you inform YARN of the Drill configuration.
@@ -44,13 +44,13 @@ This section walks you through the steps needed to create a basic Drill cluster.
### The Client Machine
-YARN works by launching an application using a “client” application. For Drill, this is the Drill-on-YARN client. The client can run on any machine that has both the Drill and Hadoop software. The client machine need not be part of your YARN cluster; any host from which you currently launch YARN jobs can be the client. All the steps that follow are done on the client machine. When running Drill outside of YARN, you must install Drill on every node of your cluster. With YARN, you need install Drill only on the client machine; Drill-on-YARN automatically deploys (“localizes”) Drill to the worker nodes.
+YARN works by launching an application using a "client" application. For Drill, this is the Drill-on-YARN client. The client can run on any machine that has both the Drill and Hadoop software. The client machine need not be part of your YARN cluster; any host from which you currently launch YARN jobs can be the client. All the steps that follow are done on the client machine. When running Drill outside of YARN, you must install Drill on every node of your cluster. With YARN, you need install Drill only on the client machine; Drill-on-YARN automatically deploys ("localizes") Drill to the worker nodes.
When running Drill without YARN, many users place their configuration files and custom code within the Drill distribution directory. When running under YARN, all your configuration and custom code resides in the site directory; you should not change anything in the Drill install. (This allows Drill-on-YARN to upload your original Drill install archive without rebuilding it.)
### Create a Master Directory
-To localize Drill files, the client tool requires a copy of the original Drill distribution archive and the location of your site directory. For ease of discussion, we assume all these components reside in a single “master directory” described as `$MASTER_DIR`. On the client machine, do the following to create the master directory:
+To localize Drill files, the client tool requires a copy of the original Drill distribution archive and the location of your site directory. For ease of discussion, we assume all these components reside in a single "master directory" described as `$MASTER_DIR`. On the client machine, do the following to create the master directory:
```
export MASTER_DIR=/path/to/master/dir
mkdir $MASTER_DIR
@@ -96,7 +96,7 @@ Your master directory should now contain the original Drill archive along with a
### Create the Site Directory
-The site directory contains your site-specific files for Drill. If you are converting an existing Drill install, see the “Site Directory” section later in this guide.
+The site directory contains your site-specific files for Drill. If you are converting an existing Drill install, see the "Site Directory" section later in this guide.
1. Create the site directory within your master directory:
```
@@ -188,7 +188,7 @@ Note that in the above, each configuration setting was shown separately. In your
### Drillbit Cluster Configuration
-Drill-on-YARN uses the concept of a “cluster group” of drill-bits to describe the set of drill-bits to launch. A group can be one of three kinds:
+Drill-on-YARN uses the concept of a "cluster group" of drill-bits to describe the set of drill-bits to launch. A group can be one of three kinds:
* Basic: launches drill-bits anywhere in the YARN cluster where a container is available.
* Labeled: Uses YARN labels to identify the set of nodes that should run Drill.
@@ -232,7 +232,7 @@ Drill copies your archive onto your distributed file system (such as HDFS) in a
```
Drill can read the connection information from your Hadoop configuration files (`$HADOOP_HOME/etc/hadoop/core-site.xml`). Or you can specify a connection directly in the Drill cluster configuration file using the connection attribute.
-Then, choose a DFS file system location. Drill uses “`/user/drill`” by default.
+Then, choose a DFS file system location. Drill uses "`/user/drill`" by default.
### Hadoop Location
@@ -353,7 +353,7 @@ More typically you will decide the set of hosts that are to run Drill. Perhaps y
#### Labeled Hosts
-Drill-on-YARN can handle node placement directly without the use of labeled queues. You use the “labeled” pool type. Then, set the drillbit-label-expr property to a YARN label expression that matches the nodes on which Drill should run. You will most often care only about Drillbit placement. Finally, indicate the number of Drillbits to run on the selected nodes.
+Drill-on-YARN can handle node placement directly without the use of labeled queues. You use the "labeled" pool type. Then, set the drillbit-label-expr property to a YARN label expression that matches the nodes on which Drill should run. You will most often care only about Drillbit placement. Finally, indicate the number of Drillbits to run on the selected nodes.
(Need example)
@@ -383,7 +383,7 @@ Application Master URL: http://10.250.50.31:8048/
```
The first line confirms which cluster is starting by displaying the cluster name from your configuration file. The next line shows YARN’s application ID and tracks the job status from Accepted to Running. Once the job starts, you’ll see YARN’s job tracking URL along with Drill-on-YARN’s web UI url. Use this URL to visit the web UI described below.
-Once the application starts, the Drill-on-YARN writes an “appid” file into your master directory:
+Once the application starts, the Drill-on-YARN writes an "appid" file into your master directory:
```
ls $MASTER_DIR
…
@@ -426,7 +426,7 @@ You can stop the Drill cluster from the command line:
```
$DRILL_HOME/bin/drill-on-yarn.sh stop
```
-Note that this command is “forceful”, it kills any in-flight queries. The output tracks the shutdown and displays the final YARN application status:
+Note that this command is "forceful", it kills any in-flight queries. The output tracks the shutdown and displays the final YARN application status:
```
Stopping Application ID: application_1462842354064_0001
Stopping...
@@ -485,7 +485,7 @@ Drill Cluster Status: the state of the Drill cluster, one of:
* LIVE: Normal state: shows that your Drill cluster is running.
* ENDING: The cluster is in the process of shutting down.
-There is no “ENDED” state: once the cluster is shut down, the AM itself exists and the web UI is no longer available.
+There is no "ENDED" state: once the cluster is shut down, the AM itself exists and the web UI is no longer available.
**Target Drillbit Count**: The number of Drillbits to run in the cluster. The actual number may be less if Drillbits have not yet started, or if YARN cannot allocate enough containers.
@@ -499,7 +499,7 @@ There is no “ENDED” state: once the cluster is shut down, the AM itself exis
### Configuration Page
-The configuration page shows the complete set of configuration values used for the current run. The values come from your own configuration along with Drill-provided defaults. Use this page to diagnose configuration-related issues. Names are shown in fully-expanded form. That is the name “drill.yarn.http.port” refers to the parameter defined as follows in your configuration file:
+The configuration page shows the complete set of configuration values used for the current run. The values come from your own configuration along with Drill-provided defaults. Use this page to diagnose configuration-related issues. Names are shown in fully-expanded form. That is the name "drill.yarn.http.port" refers to the parameter defined as follows in your configuration file:
```
drill.yarn:
http: {
@@ -518,11 +518,11 @@ The Drillbits page lists all drillbits in all states.
**Host**: The host name or IP address on which the Drillbit runs. If the Drillbit is in normal operating state, this field is also a hyperlink to the Web UI for the drillbit.
-**State**: The operating state of the Drillbit. The normal state is “Running.” The drillbit passes through a number of states as YARN allocates a container and launches a process, as the AM waits for the Drillbit to become registered in ZooKeeper, and so on. Similarly, the Drillbit passes through a different set of states during shutdown. Use this value to diagnose problems.
+**State**: The operating state of the Drillbit. The normal state is "Running." The drillbit passes through a number of states as YARN allocates a container and launches a process, as the AM waits for the Drillbit to become registered in ZooKeeper, and so on. Similarly, the Drillbit passes through a different set of states during shutdown. Use this value to diagnose problems.
-If the Drillbit is in a live state, then this field shows an “[X]” link that you can use to kill this particular Drillbit. Use this if the Drillbit has startup problems or seems unresponsive. During the shut-down process, the kill link disappears and is replaced with a “(Cancelled)” note.
+If the Drillbit is in a live state, then this field shows an "[X]" link that you can use to kill this particular Drillbit. Use this if the Drillbit has startup problems or seems unresponsive. During the shut-down process, the kill link disappears and is replaced with a "(Cancelled)" note.
-**ZK State**: The ZooKeeper handshake state. Normal state is “START_ACK”, meaning that the Drillbit has registered with ZooKeeper. This state is useful when diagnosing problems.
+**ZK State**: The ZooKeeper handshake state. Normal state is "START_ACK", meaning that the Drillbit has registered with ZooKeeper. This state is useful when diagnosing problems.
**Container ID**: The YARN-assigned container ID for the Drillbit task. The ID is a link, it takes you to the YARN Node Manager UI for the Drillbit task.
@@ -552,7 +552,7 @@ Drill clusters coordinate using ZooKeeper, so you must assign each cluster a dis
### Create a New Site Directory
-Create a new site directory for your new cluster. Let’s say that your new cluster has the name “second”. Using the same structure as above, create a new site directory under your master directory:
+Create a new site directory for your new cluster. Let’s say that your new cluster has the name "second". Using the same structure as above, create a new site directory under your master directory:
```
export SECOND_SITE=$MASTER_DIR/second
mkdir $SECOND_SITE
@@ -649,7 +649,7 @@ Drill-on-YARN creates a tighter coupling between Drill and Hadoop than did previ
### Migrating the `$DRILL_HOME/conf/drill-env.sh` Script
-Prior to Drill 1.8, the `drill-env.sh` script contained Drill defaults, distribution-specific settings, and configuration specific to your application (“site”.) In Drill 1.8, the Drill and distribution settings are moved to other locations. The site-specific settings change in format to allow YARN to override them. The following section details the changes you must make if you reuse a `drill-env.sh` file from a prior release. (If you create a new file, you can skip this section.)
+Prior to Drill 1.8, the `drill-env.sh` script contained Drill defaults, distribution-specific settings, and configuration specific to your application ("site".) In Drill 1.8, the Drill and distribution settings are moved to other locations. The site-specific settings change in format to allow YARN to override them. The following section details the changes you must make if you reuse a `drill-env.sh` file from a prior release. (If you create a new file, you can skip this section.)
At the end of this process, your file should contain just two lines for memory settings, plus any additional custom settings you may have added.
@@ -667,7 +667,7 @@ Replace them with the following two lines.
export DRILL_MAX_DIRECT_MEMORY=${DRILL_MAX_DIRECT_MEMORY:-"8G"}
export DRILL_HEAP=${DRILL_HEAP:-"4G"}
```
-Copy the actual values from the old lines to the new ones (e.g. the “8G” and “4G” values.) Those are the values that Drill when use if you launch it outside of YARN. The new lines ensure that these values are replaced by those set by Drill-on-YARN when running under YARN.
+Copy the actual values from the old lines to the new ones (e.g. the "8G" and "4G" values.) Those are the values that Drill when use if you launch it outside of YARN. The new lines ensure that these values are replaced by those set by Drill-on-YARN when running under YARN.
If you omit this change, then Drill will ignore your memory settings in Drill-on-YARN, resulting in a potential mismatch between the Drill memory settings and the amount of memory requested from YARN.
@@ -759,7 +759,7 @@ Revised instructions:
If you are not using Drill-on-YARN, set a new environment variable in `drill-env.sh`:
```
-export DRILL_JAVA_LIB_PATH=”<directory>”
+export DRILL_JAVA_LIB_PATH="<directory>"
```
If you install the library yourself, either set `DRILL_JAVA_LIB_PATH` as above, or set the following in `drill-on-yarn.conf`:
```
@@ -927,7 +927,7 @@ The first step is to create the master archive of your Drill files. Do the follo
cd $MASTER_DIR
tar -czf archive-name.tar.gz $DRILL_HOME
```
-Replace “archive-name” with the name you chose above.
+Replace "archive-name" with the name you chose above.
### Configure Drill-on-YARN to Use the Archive
@@ -937,5 +937,5 @@ drill.yarn.drill-install.client-path: "/path/to/archive-name.tar.gz"
```
YARN expects that, when extracting the master directory, that it creates a directory called archive-name that contains the Drill directories conf, jars, and so on. However, if archive-name is different than the name of the $DRILL_HOME directory, simply configure the correct name of the expanded folder:
```
-drill.yarn.drill-install.dir-name: “your-dir-name”
+drill.yarn.drill-install.dir-name: "your-dir-name"
```
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/drill-yarn/src/test/java/org/apache/drill/yarn/scripts/TestScripts.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/test/java/org/apache/drill/yarn/scripts/TestScripts.java b/drill-yarn/src/test/java/org/apache/drill/yarn/scripts/TestScripts.java
index e8f30a6..5f6b5bb 100644
--- a/drill-yarn/src/test/java/org/apache/drill/yarn/scripts/TestScripts.java
+++ b/drill-yarn/src/test/java/org/apache/drill/yarn/scripts/TestScripts.java
@@ -1214,7 +1214,7 @@ public class TestScripts {
* <li>drill-conf — Wrapper for sqlline, uses drill config to find Drill.
* Seems this one needs fixing to use a config other than the hard-coded
* $DRILL_HOME/conf location.</li>
- * <li>drill-embedded — Starts a drill “embedded” in SqlLine, using a local
+ * <li>drill-embedded — Starts a drill "embedded" in SqlLine, using a local
* ZK.</li>
* <li>drill-localhost — Wrapper for sqlline, uses a local ZK.</li>
* </ul>
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/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 77fa211..42fd7e0 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
@@ -541,8 +541,8 @@ public final class ExecConstants {
* Impersonation policy format:
* [
* {
- * proxy_principals : { users : [“...”], groups : [“...”] },
- * target_principals : { users : [“...”], groups : [“...”] }
+ * proxy_principals : { users : ["..."], groups : ["..."] },
+ * target_principals : { users : ["..."], groups : ["..."] }
* },
* ...
* ]
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/CryptoFunctions.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/CryptoFunctions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/CryptoFunctions.java
index f3f3e7b..68c4753 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/CryptoFunctions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/CryptoFunctions.java
@@ -263,7 +263,7 @@ public class CryptoFunctions {
/**
* aes_encrypt()/ aes_decrypt(): implement encryption and decryption of data using the official AES (Advanced Encryption Standard) algorithm,
- * previously known as “Rijndael.” AES_ENCRYPT() encrypts the string str using the key string key_str and returns a
+ * previously known as "Rijndael." AES_ENCRYPT() encrypts the string str using the key string key_str and returns a
* binary string containing the encrypted output.
* Usage: SELECT aes_encrypt( 'encrypted_text', 'my_secret_key' ) AS aes FROM (VALUES(1));
*/
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/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
index ea53a80..9964e60 100644
--- 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
@@ -109,10 +109,10 @@ import org.apache.drill.exec.vector.complex.AbstractContainerVector;
* <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 “first generation” files.
+ * <dd>Maximum number of batches to add to "first generation" 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 “first generation” files.
+ * <dd>Minimum number of batches to add to "first generation" 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
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/InboundImpersonationManager.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/InboundImpersonationManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/InboundImpersonationManager.java
index 60d2938..4e2b02c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/InboundImpersonationManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/InboundImpersonationManager.java
@@ -44,12 +44,12 @@ import java.util.Set;
* Impersonation policies format:
* [
* {
- * proxy_principals : { users : [“...”], groups : [“...”] },
- * target_principals : { users : [“...”], groups : [“...”] }
+ * proxy_principals : { users : ["..."], groups : ["..."] },
+ * target_principals : { users : ["..."], groups : ["..."] }
* },
* {
- * proxy_principals : { users : [“...”], groups : [“...”] },
- * target_principals : { users : [“...”], groups : [“...”] }
+ * proxy_principals : { users : ["..."], groups : ["..."] },
+ * target_principals : { users : ["..."], groups : ["..."] }
* },
* ...
* ]
http://git-wip-us.apache.org/repos/asf/drill/blob/71785a36/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/package-info.java
----------------------------------------------------------------------
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/package-info.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/package-info.java
index accf5ab..5e827fa 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/package-info.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/package-info.java
@@ -18,7 +18,7 @@
/**
* Implementation of the vector writers. The code will make much more sense if
* we start with a review of Drill’s complex vector data model. Drill has 38+
- * data (“minor”) types. Drill also has three cardinalities (“modes”). The
+ * data ("minor") types. Drill also has three cardinalities ("modes"). The
* result is over 120+ different vector types. Then, when you add maps, repeated
* maps, lists and repeated lists, you rapidly get an explosion of types that
* the writer code must handle.
@@ -35,13 +35,13 @@
* <p>
* A repeated map, a list, a repeated list and any array (repeated) scalar all
* are array-like. Nullable and required modes are identical (single values),
- * but a nullable has an additional is-set (“bit”) vector.
+ * but a nullable has an additional is-set ("bit") vector.
* <p>
* The writers (and readers) borrow concepts from JSON and relational theory
* to simplify the problem:
* <p>
* <ul>
- * <li>Both the top-level row, and a Drill map are “tuples” and are treated
+ * <li>Both the top-level row, and a Drill map are "tuples" and are treated
* similarly in the model.</li>
* <li>All non-map, non-list (that is, scalar) data types are treated
* uniformly.</li>
@@ -106,8 +106,8 @@
* This data model is similar to; but has important differences from, the prior,
* generated, readers and writers.
* <p>
- * The object layer is new: it is the simplest way to model the three “object
- * types.” An app using this code would use just the leaf scalar readers and
+ * The object layer is new: it is the simplest way to model the three "object
+ * types." An app using this code would use just the leaf scalar readers and
* writers.
*
* <h4>Writer Performance</h4>
@@ -126,7 +126,7 @@
* size and performs the needed reallocation. To avoid multiple doublings, the
* writer computes the needed new size and allocates that size directly.</li>
* <li>Vector reallocation is improved to eliminate zeroing the new half of the
- * buffer, data is left “garbage-filled.”</li>
+ * buffer, data is left "garbage-filled."</li>
* <li>If the vector would grow beyond 16 MB, then overflow is triggered, via a
* listener, which causes the buffer to be replaced. The write then
* continues.</li>
@@ -134,9 +134,9 @@
* `OffsetVectorWriter`. This writer caches the last write position so that each
* array write needs a single offset update, rather than the read and write as
* in previous code.</li>
- * <li>The writers keep track of the “last write position” and perform
- * “fill-empties” work if the new write position is more than one position
- * behind the last write. All types now correctly support “fill-empties”
+ * <li>The writers keep track of the "last write position" and perform
+ * "fill-empties" work if the new write position is more than one position
+ * behind the last write. All types now correctly support "fill-empties"
* (before, only nullable types did so reliably.)</li>
* <li>Null handling is done by an additional writer layer that wraps the
* underlying data writer. This avoids the need for a special nullable writer:
[7/8] drill git commit: DRILL-6324: Unnest - Add tests with real
Unnest and real Lateral.
Posted by pa...@apache.org.
http://git-wip-us.apache.org/repos/asf/drill/blob/8d562d07/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestWithLateralCorrectness.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestWithLateralCorrectness.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestWithLateralCorrectness.java
new file mode 100644
index 0000000..4c6b4c0
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/unnest/TestUnnestWithLateralCorrectness.java
@@ -0,0 +1,761 @@
+/*
+ * 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.unnest;
+
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.drill.categories.OperatorTest;
+import org.apache.drill.common.exceptions.DrillException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.base.LateralContract;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.LateralJoinPOP;
+import org.apache.drill.exec.physical.config.UnnestPOP;
+import org.apache.drill.exec.physical.impl.MockRecordBatch;
+import org.apache.drill.exec.physical.impl.join.LateralJoinBatch;
+import org.apache.drill.exec.physical.impl.sort.RecordBatchData;
+import org.apache.drill.exec.physical.rowSet.impl.TestResultSetLoaderMapArray;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.VectorContainer;
+import org.apache.drill.exec.store.mock.MockStorePOP;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.exec.vector.complex.MapVector;
+import org.apache.drill.test.SubOperatorTest;
+import org.apache.drill.test.rowSet.RowSet;
+import org.apache.drill.test.rowSet.RowSetBuilder;
+import org.apache.drill.test.rowSet.schema.SchemaBuilder;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static junit.framework.TestCase.fail;
+
+@Category(OperatorTest.class)
+public class TestUnnestWithLateralCorrectness extends SubOperatorTest {
+
+
+ // Operator Context for mock batch
+ public static OperatorContext operatorContext;
+
+ public static PhysicalOperator mockPopConfig;
+ public static LateralJoinPOP ljPopConfig;
+
+
+ @BeforeClass public static void setUpBeforeClass() throws Exception {
+ mockPopConfig = new MockStorePOP(null);
+ ljPopConfig = new LateralJoinPOP(null, null, JoinRelType.FULL);
+ operatorContext = fixture.newOperatorContext(mockPopConfig);
+ }
+
+ @AfterClass public static void tearDownAfterClass() throws Exception {
+ operatorContext.close();
+ }
+
+ @Test
+ public void testUnnestFixedWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new int[] {1, 2},
+ (Object) new int[] {3, 4, 5}},
+ { (Object) new int[] {6, 7, 8, 9},
+ (Object) new int[] {10, 11, 12, 13, 14}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ Integer[][][] baseline = {
+ {
+ {1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 4, 4}, //rowNum
+ {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14} // unnestColumn_flat
+ }
+ };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestVarWidthColumn() {
+
+ Object[][] data = {
+ { (Object) new String[] {"", "zero"},
+ (Object) new String[] {"one", "two", "three"}},
+ { (Object) new String[] {"four", "five", "six", "seven"},
+ (Object) new String[] {"eight", "nine", "ten", "eleven", "twelve"}}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ Object[][][] baseline = {
+ {
+ {1, 1, 2, 2, 2, 3, 3, 3, 3, 4, 4, 4, 4, 4}, // rowNum
+ {"", "zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine", "ten", "eleven",
+ "twelve"} // unnestColumn_flat
+ }
+ };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Ignore("RecordBatchSizer throws Exception in RecordBatchSizer.expandMap")
+ @Test
+ public void testUnnestMapColumn() {
+
+ Object[][] data = getMapData();
+
+ // Create input schema
+ TupleMetadata incomingSchema = getRepeatedMapSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ Object[][][] baseline = getMapBaseline();
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestEmptyList() {
+
+ Object[][] data = {
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ },
+ { (Object) new String[] {},
+ (Object) new String[] {}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema};
+
+ // All batches are empty
+ String[][][] baseline = {{{}}};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestMultipleNewSchemaIncoming() {
+
+ // Schema changes in incoming have no effect on unnest unless the type of the
+ // unnest column itself has changed
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new String[] {"9"}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata[] incomingSchemas = {incomingSchema, incomingSchema, incomingSchema};
+
+ Object[][][] baseline = {
+ {
+ {1, 1, 2, 2, 2, 3, 3, 4},
+ {"0", "1", "2", "3", "4", "5", "6", "9"}
+ }
+ };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Test
+ public void testUnnestSchemaChange() {
+ Object[][] data = {
+ {
+ (Object) new String[] {"0", "1"},
+ (Object) new String[] {"2", "3", "4"}
+ },
+ {
+ (Object) new String[] {"5", "6" },
+ },
+ {
+ (Object) new int[] {9}
+ }
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema1 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.VARCHAR).buildSchema();
+ TupleMetadata incomingSchema2 = new SchemaBuilder()
+ .add("someColumn", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema1, incomingSchema1, incomingSchema2};
+
+ Object[][][] baseline = {
+ {
+ {1, 1, 2, 2, 2, 3, 3},
+ {"0", "1", "2", "3", "4", "5", "6"}
+ },
+ {
+ {4},
+ {9}
+ }
+ };
+
+ RecordBatch.IterOutcome[] iterOutcomes = {
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA,
+ RecordBatch.IterOutcome.OK,
+ RecordBatch.IterOutcome.OK_NEW_SCHEMA};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+ @Ignore ("Batch limits need to be sync'd with tthe record batch sizer. Fix once the calulations are stabilized")
+ @Test
+ public void testUnnestLimitBatchSize() {
+
+ final int limitedOutputBatchSize = 1024;
+ final int inputBatchSize = 1024+1;
+ final int limitedOutputBatchSizeBytes = 1024*(4 + 4 + 4 * inputBatchSize); // num rows * (size of int + size of
+ // int + size of int * num entries in
+ // array)
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+ Integer[][][] baseline = new Integer[2][2][];
+ baseline[0][0] = new Integer[limitedOutputBatchSize];
+ baseline[0][1] = new Integer[limitedOutputBatchSize];
+ baseline[1][0] = new Integer[1];
+ baseline[1][1] = new Integer[1];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[0][0][i] = 1;
+ baseline[0][1][i] = i;
+ }
+ baseline[1][0][0] = 1; // row Num
+ baseline[1][1][0] = limitedOutputBatchSize; // value
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has more than one record batch for a record when
+ // the kill is sent.
+ public void testUnnestKillFromLimitSubquery1() {
+
+ // similar to previous test; we split a record across more than one batch.
+ // but we also set a limit less than the size of the batch so only one batch gets output.
+ final int limitedOutputBatchSize = 1024;
+ final int inputBatchSize = 1024+1;
+ final int limitedOutputBatchSizeBytes = 1024*(4 + 4 + 4 * inputBatchSize); // num rows * (size of int + size of
+ // int + size of int * num entries in
+ // array)
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+
+ // because of kill we only get one batch back
+ Integer[][][] baseline = new Integer[1][2][];
+ baseline[0][0] = new Integer[limitedOutputBatchSize];
+ baseline[0][1] = new Integer[limitedOutputBatchSize];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[0][0][i] = 1;
+ baseline[0][1][i] = i;
+ }
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, -1, 1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+ @Test
+ // Limit sends a kill. Unnest has exactly one record batch for a record when
+ // the kill is sent. This test is actually useless since it tests the behaviour of
+ // lateral which doesn't send kill at all if it gets an EMIT. We expect limit
+ // to do so, so let's keep the test to demonstrate the expected behaviour.
+ public void testUnnestKillFromLimitSubquery2() {
+
+ // similar to previous test but the size of the array fits exactly into the record batch;
+
+ final int limitedOutputBatchSize = 1024;
+ final int inputBatchSize = 1024;
+ final int limitedOutputBatchSizeBytes = 1024*(4 + 4 + 4 * inputBatchSize); // num rows * (size of int + size of
+ // int + size of int * num entries in
+ // array)
+ // single record batch with single row. The unnest column has one
+ // more record than the batch size we want in the output
+ Object[][] data = new Object[1][1];
+
+ for (int i = 0; i < data.length; i++) {
+ for (int j = 0; j < data[i].length; j++) {
+ data[i][j] = new int[inputBatchSize];
+ for (int k =0; k < inputBatchSize; k++) {
+ ((int[])data[i][j])[k] = k;
+ }
+ }
+ }
+
+ // because of kill we only get one batch back
+ Integer[][][] baseline = new Integer[1][2][];
+ baseline[0][0] = new Integer[limitedOutputBatchSize];
+ baseline[0][1] = new Integer[limitedOutputBatchSize];
+ for (int i = 0; i < limitedOutputBatchSize; i++) {
+ baseline[0][0][i] = 1;
+ baseline[0][1][i] = i;
+ }
+
+ // Create input schema
+ TupleMetadata incomingSchema = new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .addArray("unnestColumn", TypeProtos.MinorType.INT).buildSchema();
+
+ TupleMetadata[] incomingSchemas = {incomingSchema};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK};
+
+ final long outputBatchSize = fixture.getFragmentContext().getOptions().getOption(ExecConstants
+ .OUTPUT_BATCH_SIZE_VALIDATOR);
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, limitedOutputBatchSizeBytes);
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, -1, 1, data, baseline); // Limit of 100 values for unnest.
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ } finally {
+ fixture.getFragmentContext().getOptions().setLocalOption(ExecConstants.OUTPUT_BATCH_SIZE, outputBatchSize);
+ }
+
+ }
+
+
+ @Test
+ public void testUnnestNonArrayColumn() {
+
+ Object[][] data = {
+ { (Object) new Integer (1),
+ (Object) new Integer (3)},
+ { (Object) new Integer (6),
+ (Object) new Integer (10)}
+ };
+
+ // Create input schema
+ TupleMetadata incomingSchema =
+ new SchemaBuilder()
+ .add("rowNumber", TypeProtos.MinorType.INT)
+ .add("unnestColumn", TypeProtos.MinorType.INT)
+ .buildSchema();
+ TupleMetadata[] incomingSchemas = { incomingSchema, incomingSchema };
+
+ // We expect an Exception
+ Integer[][][] baseline = {};
+
+ RecordBatch.IterOutcome[] iterOutcomes = {RecordBatch.IterOutcome.OK_NEW_SCHEMA, RecordBatch.IterOutcome.OK};
+
+ try {
+ testUnnest(incomingSchemas, iterOutcomes, data, baseline);
+ } catch (UserException|UnsupportedOperationException e) {
+ return; // succeeded
+ } catch (Exception e) {
+ fail("Failed due to exception: " + e.getMessage());
+ }
+
+ }
+
+
+ // test unnest for various input conditions without invoking kill
+ private <T> void testUnnest(
+ TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ T[][] data,
+ T[][][] baseline ) throws Exception{
+ testUnnest(incomingSchemas, iterOutcomes, -1, -1, data, baseline);
+ }
+
+ // test unnest for various input conditions optionally invoking kill. if the kill or killBatch
+ // parameter is greater than 0 then the record batch is sent a kill after that many batches have been processed
+ private <T> void testUnnest( TupleMetadata[] incomingSchemas,
+ RecordBatch.IterOutcome[] iterOutcomes,
+ int unnestLimit, // kill unnest after every 'unnestLimit' number of values in every record
+ int execKill, // number of batches after which to kill the execution (!)
+ T[][] data,
+ T[][][] baseline) throws Exception {
+
+ // Get the incoming container with dummy data for LJ
+ final List<VectorContainer> incomingContainer = new ArrayList<>(data.length);
+
+ // Create data
+ ArrayList<RowSet.SingleRowSet> rowSets = new ArrayList<>();
+ int rowNumber = 0;
+ int batchNum = 0;
+ for ( Object[] recordBatch : data) {
+ RowSetBuilder rowSetBuilder = fixture.rowSetBuilder(incomingSchemas[batchNum]);
+ for ( Object rowData : recordBatch) {
+ rowSetBuilder.addRow(++rowNumber, rowData);
+ }
+ RowSet.SingleRowSet rowSet = rowSetBuilder.build();
+ rowSets.add(rowSet);
+ incomingContainer.add(rowSet.container());
+ batchNum++;
+ }
+
+ // Get the unnest POPConfig
+ final UnnestPOP unnestPopConfig = new UnnestPOP(null, SchemaPath.getCompoundPath("unnestColumn"));
+
+ // Get the IterOutcomes for LJ
+ final List<RecordBatch.IterOutcome> outcomes = new ArrayList<>(iterOutcomes.length);
+ for(RecordBatch.IterOutcome o : iterOutcomes) {
+ outcomes.add(o);
+ }
+
+ // Create incoming MockRecordBatch
+ final MockRecordBatch incomingMockBatch =
+ new MockRecordBatch(fixture.getFragmentContext(), operatorContext, incomingContainer, outcomes,
+ incomingContainer.get(0).getSchema());
+
+ // setup Unnest record batch
+ final UnnestRecordBatch unnestBatch =
+ new UnnestRecordBatch(unnestPopConfig, fixture.getFragmentContext());
+
+ final LateralJoinBatch lateralJoinBatch =
+ new LateralJoinBatch(ljPopConfig, fixture.getFragmentContext(), incomingMockBatch, unnestBatch);
+
+ // set pointer to Lateral in unnest
+ unnestBatch.setIncoming((LateralContract) lateralJoinBatch);
+
+ // Simulate the pipeline by calling next on the incoming
+
+ // results is an array ot batches, each batch being an array of output vectors.
+ List<List<ValueVector> > resultList = new ArrayList<>();
+ List<List<ValueVector> > results = null;
+ int batchesProcessed = 0;
+ try{
+ try {
+ while (!isTerminal(lateralJoinBatch.next())) {
+ if (lateralJoinBatch.getRecordCount() > 0) {
+ addBatchToResults(resultList, lateralJoinBatch);
+ }
+ batchesProcessed++;
+ if (batchesProcessed == execKill) {
+ lateralJoinBatch.getContext().getExecutorState().fail(new DrillException("Testing failure of execution."));
+ lateralJoinBatch.kill(true);
+ }
+ // else nothing to do
+ }
+ } catch (UserException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new Exception ("Test failed to execute lateralJoinBatch.next() because: " + e.getMessage());
+ }
+
+ // Check results against baseline
+ results = resultList;
+
+ int batchIndex = 0;
+ int vectorIndex = 0;
+ //int valueIndex = 0;
+ for ( List<ValueVector> batch: results) {
+ int vectorCount= batch.size();
+ if (vectorCount!= baseline[batchIndex].length+1) { // baseline does not include the original unnest column
+ fail("Test failed in validating unnest output. Batch column count mismatch.");
+ }
+ for (ValueVector vv : batch) {
+ if(vv.getField().getName().equals("unnestColumn")) {
+ continue; // skip the original input column
+ }
+ int valueCount = vv.getAccessor().getValueCount();
+ if (valueCount!= baseline[batchIndex][vectorIndex].length) {
+ fail("Test failed in validating unnest output. Value count mismatch in batch number " + (batchIndex+1) +""
+ + ".");
+ }
+
+ for (int valueIndex = 0; valueIndex < valueCount; valueIndex++) {
+ if (vv instanceof MapVector) {
+ if (!compareMapBaseline(baseline[batchIndex][vectorIndex][valueIndex], vv
+ .getAccessor()
+ .getObject(valueIndex))) {
+ fail("Test failed in validating unnest(Map) output. Value mismatch");
+ }
+ } else if (vv instanceof VarCharVector) {
+ Object val = vv.getAccessor().getObject(valueIndex);
+ if (((String) baseline[batchIndex][vectorIndex][valueIndex]).compareTo(val.toString()) != 0) {
+ fail("Test failed in validating unnest output. Value mismatch. Baseline value[]" + vectorIndex + "][" + valueIndex
+ + "]" + ": " + baseline[vectorIndex][valueIndex] + " VV.getObject(valueIndex): " + val);
+ }
+ } else {
+ Object val = vv.getAccessor().getObject(valueIndex);
+ if (!baseline[batchIndex][vectorIndex][valueIndex].equals(val)) {
+ fail("Test failed in validating unnest output. Value mismatch. Baseline value[" + vectorIndex + "][" + valueIndex
+ + "]" + ": "
+ + ((Object[])baseline[batchIndex][vectorIndex])[valueIndex] + " VV.getObject(valueIndex): " + val);
+ }
+ }
+ }
+ vectorIndex++;
+ }
+ vectorIndex=0;
+ batchIndex++;
+ }
+ } catch (UserException e) {
+ throw e; // Valid exception
+ } catch (Exception e) {
+ fail("Test failed. Exception : " + e.getMessage());
+ } finally {
+ // Close all the resources for this test case
+ unnestBatch.close();
+ lateralJoinBatch.close();
+ incomingMockBatch.close();
+
+ if (results != null) {
+ for (List<ValueVector> batch : results) {
+ for (ValueVector vv : batch) {
+ vv.clear();
+ }
+ }
+ }
+ for(RowSet.SingleRowSet rowSet: rowSets) {
+ rowSet.clear();
+ }
+ }
+
+ }
+
+ /**
+ * Build a schema with a repeated map -
+ *
+ * {
+ * rowNum,
+ * mapColumn : [
+ * {
+ * colA,
+ * colB : [
+ * varcharCol
+ * ]
+ * }
+ * ]
+ * }
+ *
+ * @see TestResultSetLoaderMapArray TestResultSetLoaderMapArray for similar schema and data
+ * @return TupleMetadata corresponding to the schema
+ */
+ private TupleMetadata getRepeatedMapSchema() {
+ TupleMetadata schema = new SchemaBuilder()
+ .add("rowNum", TypeProtos.MinorType.INT)
+ .addMapArray("unnestColumn")
+ .add("colA", TypeProtos.MinorType.INT)
+ .addArray("colB", TypeProtos.MinorType.VARCHAR)
+ .resumeSchema()
+ .buildSchema();
+ return schema;
+ }
+
+ private Object[][] getMapData( ) {
+
+ Object[][] d = {
+ {
+ new Object[] {},
+ new Object[] {
+ new Object[] {11, new String[] {"1.1.1", "1.1.2" }},
+ new Object[] {12, new String[] {"1.2.1", "1.2.2" }}
+ },
+
+ new Object[] {
+ new Object[] {21, new String[] {"2.1.1", "2.1.2" }},
+ new Object[] {22, new String[] {}},
+ new Object[] {23, new String[] {"2.3.1", "2.3.2" }}
+ }
+ },
+ {
+ new Object[] {
+ new Object[] {31, new String[] {"3.1.1", "3.1.2" }},
+ new Object[] {32, new String[] {"3.2.1", "3.2.2" }}
+ }
+ }
+ };
+
+ return d;
+ }
+
+ private Object[][][] getMapBaseline() {
+
+ Object[][][] d = {
+ {
+ {2,2,3,3,3,4,4},
+ {
+ "{\"colA\":11,\"colB\":[\"1.1.1\",\"1.1.2\"]}",
+ "{\"colA\":12,\"colB\":[\"1.2.1\",\"1.2.2\"]}",
+ "{\"colA\":21,\"colB\":[\"2.1.1\",\"2.1.2\"]}",
+ "{\"colA\":22,\"colB\":[]}",
+ "{\"colA\":23,\"colB\":[\"2.3.1\",\"2.3.2\"]}",
+ "{\"colA\":31,\"colB\":[\"3.1.1\",\"3.1.2\"]}",
+ "{\"colA\":32,\"colB\":[\"3.2.1\",\"3.2.2\"]}"
+ }
+ }
+ };
+ return d;
+ }
+
+ private boolean compareMapBaseline(Object baselineValue, Object vector) {
+ String vv = vector.toString();
+ String b = (String)baselineValue;
+ return vv.equalsIgnoreCase(b);
+ }
+
+ private int addBatchToResults(List<List<ValueVector> > resultList, RecordBatch inputBatch) {
+ int count = 0;
+ final RecordBatchData batchCopy = new RecordBatchData(inputBatch, operatorContext.getAllocator());
+ boolean success = false;
+ try {
+ count = batchCopy.getRecordCount();
+ resultList.add(batchCopy.getVectors());
+ success = true;
+ } finally {
+ if (!success) {
+ batchCopy.clear();
+ }
+ }
+ return count;
+ }
+
+ private boolean isTerminal(RecordBatch.IterOutcome outcome) {
+ return (outcome == RecordBatch.IterOutcome.NONE || outcome == RecordBatch.IterOutcome.STOP) || (outcome
+ == RecordBatch.IterOutcome.OUT_OF_MEMORY);
+ }
+
+}
+