You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by yufeldman <gi...@git.apache.org> on 2016/02/08 03:10:10 UTC
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
GitHub user yufeldman opened a pull request:
https://github.com/apache/drill/pull/368
DRILL-4132 Ability to submit simple type of physical plan directly to…
… EndPoint DrillBit for execution.
There are multiple changes to achieve this:
1. During physical planning split single plan into multiple based on the number of minor fragments of the Leaf Major fragment.
a. Removing exchange operators during planning
b. Producing just root fragments (that will be also leaf fragments)
2. Each fragment can be executed against Drillbit it is assigned to, so to keep locality
Design document can be found in the JIRA: DRILL-4132
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/yufeldman/incubator-drill DRILL-4132
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/drill/pull/368.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #368
----
commit 2a35967396dd66c3e371973fe8baf14f720ed219
Author: Yuliya Feldman <yf...@maprtech.com>
Date: 2016-02-04T22:09:21Z
DRILL-4132 Ability to submit simple type of physical plan directly to EndPoint DrillBit for execution.
There are multiple changes to achieve this:
1. During physical planning split single plan into multiple based on the number of minor fragments of the Leaf Major fragment.
a. Removing exchange operators during planning
b. Producing just root fragments (that will be also leaf fragments)
2. Each fragment can be executed against Drillbit it is assigned to, so to keep locality
Design document can be found in the JIRA: DRILL-4132
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61347587
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java ---
@@ -17,13 +17,27 @@
*/
package org.apache.drill.exec.util;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.fn.impl.DateUtility;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.ExternalSort;
import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.drill.exec.server.options.OptionManager;
public class Utilities {
--- End diff --
I am against the anti-pattern of a generic Utilities class. It won't take too long for us to turn into a junk yard of code :) I would propose that we externalize the functionality here to its relevant utility class like SortUtils or similar.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61349687
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java ---
@@ -419,6 +425,64 @@ private void runPhysicalPlan(final PhysicalPlan plan) throws ExecutionSetupExcep
logger.debug("Fragments running.");
}
+ /**
+ * This is a helper method to run query based on the list of PlanFragment that were planned
+ * at some point of time
+ * @param fragmentsList
+ * @throws ExecutionSetupException
+ */
+ private void runFragment(List<PlanFragment> fragmentsList) throws ExecutionSetupException {
+ // need to set QueryId, MinorFragment for incoming Fragments
+ PlanFragment rootFragment = null;
+ boolean isFirst = true;
+ final List<PlanFragment> planFragments = Lists.newArrayList();
+ final int fragmentsCount = fragmentsList.size();
+ for (PlanFragment myFragment : fragmentsList) {
+ final FragmentHandle handle = myFragment.getHandle();
+ // for split plan number of minor fragments will be always one,
+ // but minor fragmentId may not be 0, as it is one of the minor fragments from split plan
+ final int minorFragment = (fragmentsCount == 1) ? 0 : handle.getMinorFragmentId();
--- End diff --
Even though I understand that a split plan consists of a single minor fragment, I wonder if there is a better way to handle minor fragment numbering. Why do not we just re-number minor fragments during planning/splitting phase at the first place instead of relying on fragmentsCount == 1 check?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61349902
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/QueryPlanFragmentsSplitHelper.java ---
@@ -0,0 +1,141 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class QueryPlanFragmentsSplitHelper {
--- End diff --
or even PlanSplitter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58622650
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(exchange);
+ PhysicalOperator childEx = exchange.getChild().accept(this, iNode);
+ childEx.setOperatorId(Short.MAX_VALUE & exchange.getOperatorId());
+ return childEx;
+ }
+
+ @Override
+ public PhysicalOperator visitGroupScan(GroupScan groupScan, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ PhysicalOperator child = groupScan.getSpecificScan(iNode.getMinorFragmentId());
+ child.setOperatorId(Short.MAX_VALUE & groupScan.getOperatorId());
+ return child;
+ }
+
+ @Override
+ public PhysicalOperator visitSubScan(SubScan subScan, IndexedFragmentNode value) throws ExecutionSetupException {
+ value.addAllocation(subScan);
+ // TODO - implement this
+ return super.visitOp(subScan, value);
+ }
+
+ @Override
+ public PhysicalOperator visitStore(Store store, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ PhysicalOperator child = store.getChild().accept(this, iNode);
+
+ iNode.addAllocation(store);
+
+ try {
+ PhysicalOperator o = store.getSpecificStore(child, iNode.getMinorFragmentId());
+ o.setOperatorId(Short.MAX_VALUE & store.getOperatorId());
+ return o;
+ } catch (PhysicalOperatorSetupException e) {
+ throw new FragmentSetupException("Failure while generating a specific Store materialization.");
--- End diff --
The original PhysicalOperatorSetupException is getting lost.. can you add it here ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58624118
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(exchange);
+ PhysicalOperator childEx = exchange.getChild().accept(this, iNode);
+ childEx.setOperatorId(Short.MAX_VALUE & exchange.getOperatorId());
+ return childEx;
+ }
+
+ @Override
+ public PhysicalOperator visitGroupScan(GroupScan groupScan, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ PhysicalOperator child = groupScan.getSpecificScan(iNode.getMinorFragmentId());
+ child.setOperatorId(Short.MAX_VALUE & groupScan.getOperatorId());
+ return child;
+ }
+
+ @Override
+ public PhysicalOperator visitSubScan(SubScan subScan, IndexedFragmentNode value) throws ExecutionSetupException {
+ value.addAllocation(subScan);
+ // TODO - implement this
+ return super.visitOp(subScan, value);
+ }
+
+ @Override
+ public PhysicalOperator visitStore(Store store, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ PhysicalOperator child = store.getChild().accept(this, iNode);
+
+ iNode.addAllocation(store);
+
+ try {
+ PhysicalOperator o = store.getSpecificStore(child, iNode.getMinorFragmentId());
+ o.setOperatorId(Short.MAX_VALUE & store.getOperatorId());
+ return o;
+ } catch (PhysicalOperatorSetupException e) {
+ throw new FragmentSetupException("Failure while generating a specific Store materialization.");
--- End diff --
good catch, will add
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58753681
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple physical plans from a single logical plan
+ * to be able to run them separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
+ * End goal is to get single major fragment: Screen with chain that ends up with a single minor fragment
+ * from Leaf Exchange. This way each plan can run independently without any exchange involvement
+ * @param options
+ * @param foremanNode - not really applicable
+ * @param queryId
+ * @param reader
+ * @param rootNode
+ * @param planningSet
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ private List<QueryWorkUnit> generateWorkUnits(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ // now we generate all the individual plan fragments and associated assignments. Note, we need all endpoints
+ // assigned before we can materialize, so we start a new loop here rather than utilizing the previous one.
+
+ List<QueryWorkUnit> workUnits = Lists.newArrayList();
+ int plansCount = 0;
+ DrillbitEndpoint[] endPoints = null;
+ long initialAllocation = 0;
+ long maxAllocation = 0;
+
+ final Iterator<Wrapper> iter = planningSet.iterator();
+ while (iter.hasNext()) {
+ Wrapper wrapper = iter.next();
+ Fragment node = wrapper.getNode();
+ boolean isLeafFragment = node.getReceivingExchangePairs().size() == 0;
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ // get all the needed info from leaf fragment
+ if ( (physicalOperatorRoot instanceof Exchange) && isLeafFragment) {
+ // need to get info about
+ // number of minor fragments
+ // assignedEndPoints
+ // allocation
+ plansCount = wrapper.getWidth();
+ initialAllocation = (wrapper.getInitialAllocation() != 0 ) ? wrapper.getInitialAllocation()/plansCount : 0;
+ maxAllocation = (wrapper.getMaxAllocation() != 0 ) ? wrapper.getMaxAllocation()/plansCount : 0;
+ endPoints = new DrillbitEndpoint[plansCount];
+ for (int mfId = 0; mfId < plansCount; mfId++) {
+ endPoints[mfId] = wrapper.getAssignedEndpoint(mfId);
+ }
+ }
+ }
+ if ( plansCount == 0 ) {
+ // no exchange, return list of single QueryWorkUnit
+ workUnits.add(generateWorkUnit(options, foremanNode, queryId, reader, rootNode, planningSet, session, queryContextInfo));
+ return workUnits;
+ }
+
+ for (Wrapper wrapper : planningSet) {
+ Fragment node = wrapper.getNode();
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ if ( physicalOperatorRoot instanceof Exchange ) {
+ // get to 0 MajorFragment
+ continue;
+ }
+ boolean isRootNode = rootNode == node;
+
+ if (isRootNode && wrapper.getWidth() != 1) {
+ throw new ForemanSetupException(String.format("Failure while trying to setup fragment. " +
+ "The root fragment must always have parallelization one. In the current case, the width was set to %d.",
+ wrapper.getWidth()));
+ }
+ // this fragment is always leaf, as we are removing all the exchanges
+ boolean isLeafFragment = true;
+
+ // Create a minorFragment for each major fragment.
+ for (int minorFragmentId = 0; minorFragmentId < plansCount; minorFragmentId++) {
+ // those fragments should be empty
+ List<PlanFragment> fragments = Lists.newArrayList();
+
+ PlanFragment rootFragment = null;
+ FragmentRoot rootOperator = null;
+
+ IndexedFragmentNode iNode = new IndexedFragmentNode(minorFragmentId, wrapper);
+ wrapper.resetAllocation();
+ // two visitors here
+ // 1. To remove exchange
+ // 2. To reset operator IDs as exchanges were removed
+ PhysicalOperator op = physicalOperatorRoot.accept(ExchangeManipulatorMaterializerVisitor.INSTANCE, iNode).
--- End diff --
Why ExchangeRemoverVisitor would not drop any exchange? It does drop in any case. Decision on whether to drop or not is done before that visitor is called
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-215110667
@amansinha100 and @hnfgns - could you please review again? I have addressed comments
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58625882
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple physical plans from a single logical plan
--- End diff --
The comment "from a single logical plan" is not quite accurate since this is taking a POP (Physical Operator) plan and generating another POP plan.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61347199
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/rpc/user/UserClient.java ---
@@ -137,4 +142,9 @@ protected void finalizeConnection(BitToUserHandshake handshake, BasicClientWithC
public ProtobufLengthDecoder getDecoder(BufferAllocator allocator) {
return new UserProtobufLengthDecoder(allocator, OutOfMemoryHandler.DEFAULT_INSTANCE);
}
+
+ public DrillRpcFuture<QueryPlanFragments> submitPlanQuery(
--- End diff --
Should we consider renaming this to planQuery(...) and add a small comment?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58629500
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
--- End diff --
makes sense
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61491214
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,228 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.PlanningSet;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.Wrapper;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple Physical Operator plans from original Physical Operator plan
+ * to be able to run plans separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleParallelizerMultiPlans.class);
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
--- End diff --
Name advice: splitFragments/split.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61516040
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/PlanSplitter.java ---
@@ -0,0 +1,142 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.MemoryAllocationUtilities;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class PlanSplitter {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanSplitter.class);
+
+ private static PlanSplitter s_instance = new PlanSplitter();
+
+ private PlanSplitter() {
+
+ }
+
+ public static PlanSplitter getInstance() {
+ return s_instance;
+ }
+
+ /**
+ * Method to plan the query and return list of fragments
+ * it will return query plan "as is" or split plans based on the req setting: split_plan
+ * @param dContext
+ * @param queryId
+ * @param req
+ * @param connection
+ * @return
+ */
+ public QueryPlanFragments planFragments(DrillbitContext dContext, QueryId queryId,
+ GetQueryPlanFragments req, UserClientConnection connection) {
+ QueryPlanFragments.Builder responseBuilder = QueryPlanFragments.newBuilder();
+ QueryContext queryContext = new QueryContext(connection.getSession(), dContext, queryId);
+
+ responseBuilder.setQueryId(queryId);
+
+ try {
+ responseBuilder.addAllFragments(getFragments(dContext, req, queryContext, queryId));
+ responseBuilder.setStatus(QueryState.COMPLETED);
+ } catch (Exception e) {
+ final String errorMessage = String.format("Failed to produce PlanFragments for query id \"%s\" with "
+ + "request to %s plan", queryId, (req.getSplitPlan() ? "split" : "no split"));
+ DrillPBError error = DrillPBError.newBuilder().setMessage(errorMessage).setErrorType(DrillPBError.ErrorType.PLAN).build();
+
+ responseBuilder.setStatus(QueryState.FAILED);
+ responseBuilder.setError(error);
+ }
+ return responseBuilder.build();
+ }
+
+ private List<PlanFragment> getFragments(final DrillbitContext dContext, final GetQueryPlanFragments req,
+ final QueryContext queryContext, final QueryId queryId) throws Exception {
+ final PhysicalPlan plan;
+ final String query = req.getQuery();
+ switch(req.getType()) {
+ case SQL:
+ final Pointer<String> textPlan = new Pointer<>();
+ plan = DrillSqlWorker.getPlan(queryContext, query, textPlan);
+ break;
+ case PHYSICAL:
+ plan = dContext.getPlanReader().readPhysicalPlan(query);
+ break;
+ default:
+ throw new IllegalStateException("Planning fragments supports only SQL or PHYSICAL QueryType");
+ }
+
+ MemoryAllocationUtilities.setupSortMemoryAllocations(plan, queryContext);
+
+ final PhysicalOperator rootOperator = plan.getSortedOperators(false).iterator().next();
+
+ final Fragment rootFragment = rootOperator.accept(MakeFragmentsVisitor.INSTANCE, null);
+ final SimpleParallelizer parallelizer = new SimpleParallelizerMultiPlans(queryContext);
+
+ List<PlanFragment> fragments = Lists.newArrayList();
+
+ if ( req.getSplitPlan() ) {
--- End diff --
I am trying to leave a door open for other features that may rely on planning a query separately from it's execution. In that case we will just return original query plan.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61624612
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java ---
@@ -321,6 +329,38 @@ public void close() {
return listener.getResults();
}
+ public DrillRpcFuture<QueryPlanFragments> planQuery(QueryType type, String query, boolean isSplitPlan) {
+ GetQueryPlanFragments runQuery = GetQueryPlanFragments.newBuilder().setQuery(query).setType(type).setSplitPlan(isSplitPlan).build();
+ return client.submitPlanQuery(runQuery);
+ }
+
+ public void runQuery(QueryType type, List<PlanFragment> planFragments, UserResultsListener resultsListener)
+ throws RpcException {
+ // QueryType can be only executional
+ checkArgument((QueryType.EXECUTION == type), "Only EXECUTIONAL type query is supported with PlanFragments");
+ // setting Plan on RunQuery will be used for logging purposes and therefore can not be null
+ // since there is no Plan string provided we will create a JsonArray out of individual fragment Plans
+ ArrayNode jsonArray = objectMapper.createArrayNode();
+ for (PlanFragment fragment : planFragments) {
+ try {
+ jsonArray.add(objectMapper.readTree(fragment.getFragmentJson()));
+ } catch (IOException e) {
+ logger.error("Exception while trying to read PlanFragment JSON for %s", fragment.getHandle().getQueryId(), e);
+ throw new RpcException(e);
+ }
+ }
+ final String fragmentsToJsonString;
+ try {
+ fragmentsToJsonString = objectMapper.writeValueAsString(jsonArray);
--- End diff --
It is not that easy, as on DrillClient there is no knowledge of DrillContext and subsequently PhysicalPlanReader.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61346266
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/ExchangeRemoverMaterializer.java ---
@@ -0,0 +1,96 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ * NOTE: this Visitor does NOT set OperatorId, as after Exchange removal all operators need renumbering
+ * Use OperatorIdVisitor on top to set correct OperatorId
+ */
+public class ExchangeRemoverMaterializer extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeRemoverMaterializer.class);
--- End diff --
private
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61515747
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/PlanSplitter.java ---
@@ -0,0 +1,142 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.MemoryAllocationUtilities;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class PlanSplitter {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanSplitter.class);
+
+ private static PlanSplitter s_instance = new PlanSplitter();
+
+ private PlanSplitter() {
+
+ }
+
+ public static PlanSplitter getInstance() {
+ return s_instance;
--- End diff --
not much difference so far.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58628326
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
--- End diff --
Since this new parallelizer is for a specific use case, it would make sense to put it in a separate child package.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61494824
--- Diff: exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java ---
@@ -0,0 +1,350 @@
+/**
+ * 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;
+
+import static org.junit.Assert.*;
+import io.netty.buffer.DrillBuf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.DrillAutoCloseables;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.util.TestTools;
+import org.apache.drill.exec.client.DrillClient;
+import org.apache.drill.exec.client.PrintingResultsListener;
+import org.apache.drill.exec.client.QuerySubmitter.Format;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.QueryData;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserBitShared.QueryType;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.rpc.ConnectionThrottle;
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.user.AwaitableUserResultsListener;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+import org.apache.drill.exec.rpc.user.UserResultsListener;
+import org.apache.drill.exec.util.VectorUtil;
+import org.apache.drill.exec.vector.ValueVector;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Class to test different planning use cases (separate form query execution)
+ *
+ */
+public class DrillSeparatePlanningTest extends BaseTestQuery {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillSeparatePlanningTest.class);
+
+ static final String WORKING_PATH = TestTools.getWorkingPath();
+ static final String TEST_RES_PATH = WORKING_PATH + "/src/test/resources";
+
+ //final String query = "SELECT sales_city, COUNT(*) cnt FROM cp.`region.json` GROUP BY sales_city";
+ //final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+ //final String query = "SELECT o_orderkey, o_custkey FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000 limit 5";
+ //final String query = "SELECT sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000";
+ //final String query = "SELECT o_orderkey FROM dfs.tmp.`multilevel` order by o_orderkey";
+ //final String query = "SELECT dir1, sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 group by dir1 order by dir1";
+ //final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+
+ @Test(timeout=30000)
+ public void testSingleFragmentQuery() throws Exception {
+ final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertEquals(1, planFragments.getFragmentsCount());
+ assertTrue(planFragments.getFragments(0).getLeafFragment());
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentSimpleQuery() throws Exception {
+ final String query = String.format("SELECT o_orderkey FROM dfs_test.`%s/multilevel/json`", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentComplexQuery() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+
+ }
+
+ @Test(timeout=30000)
+ public void testPlanningNoSplit() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ updateTestCluster(2, config);
+
+ List<QueryDataBatch> results = client.runQuery(QueryType.SQL, "alter session set `planner.slice_target`=1");
+ for(QueryDataBatch batch : results) {
+ batch.release();
+ }
+
+ DrillRpcFuture<QueryPlanFragments> queryFragmentsFutures = client.planQuery(QueryType.SQL, query, false);
+
+ final QueryPlanFragments planFragments = queryFragmentsFutures.get();
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ PlanFragment rootFragment = planFragments.getFragments(0);
+ assertFalse(rootFragment.getLeafFragment());
+
+ getCombinedResultsHelper(planFragments);
+
+ client.close();
--- End diff --
BaseTestQuery closes client at the end of suite. We should not close the client. I would be surprised if your unit tests passed as it is 👍
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61491435
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,228 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.PlanningSet;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.Wrapper;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple Physical Operator plans from original Physical Operator plan
+ * to be able to run plans separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleParallelizerMultiPlans.class);
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
--- End diff --
I am not comfortable with this statement. How do you drop exchange for "order by" but still sort?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58623171
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
--- End diff --
My understanding from the design doc was you want to remove UnionExchange and SingleMergeExchange...whereas this is dropping all types of exchanges.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58623402
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(exchange);
+ PhysicalOperator childEx = exchange.getChild().accept(this, iNode);
+ childEx.setOperatorId(Short.MAX_VALUE & exchange.getOperatorId());
--- End diff --
The operator id is set here and then again later with the separate traversal...why not do it only once later ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by jacques-n <gi...@git.apache.org>.
Github user jacques-n commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58633147
--- Diff: protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java ---
@@ -133,6 +133,10 @@ private RpcChannel(int index, int value) {
* <code>PHYSICAL = 3;</code>
*/
PHYSICAL(2, 3),
+ /**
+ * <code>EXECUTIONAL = 4;</code>
+ */
+ EXECUTIONAL(3, 4),
--- End diff --
EXECUTION? e.g. execution plan?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-216724026
+1 if made sure all tests are green.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61492206
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/PlanSplitter.java ---
@@ -0,0 +1,142 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.MemoryAllocationUtilities;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class PlanSplitter {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanSplitter.class);
+
+ private static PlanSplitter s_instance = new PlanSplitter();
+
+ private PlanSplitter() {
+
+ }
+
+ public static PlanSplitter getInstance() {
+ return s_instance;
--- End diff --
This seems a very cheap object used once per query. Why to make it singleton? We should create it in the stack/heap as we need.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61349858
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/QueryPlanFragmentsSplitHelper.java ---
@@ -0,0 +1,141 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class QueryPlanFragmentsSplitHelper {
--- End diff --
rename to QueryPlanSplit[ter or Helper]?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58624039
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
--- End diff --
It is true, that originally I was thinking about just those two types of exchanges, but it feels like it can be more than that. I did not come up with a good set of exchanges that can not be removed - ones that non idempotent. Will look into it more closely
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-216933980
My review comments were previously addressed, so +1 assuming tests pass.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58754414
--- Diff: protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java ---
@@ -133,6 +133,10 @@ private RpcChannel(int index, int value) {
* <code>PHYSICAL = 3;</code>
*/
PHYSICAL(2, 3),
+ /**
+ * <code>EXECUTIONAL = 4;</code>
+ */
+ EXECUTIONAL(3, 4),
--- End diff --
sure again :).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58623695
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeManipulatorMaterializerVisitor.class);
+
+ public static final ExchangeManipulatorMaterializerVisitor INSTANCE = new ExchangeManipulatorMaterializerVisitor();
+
+ private ExchangeManipulatorMaterializerVisitor() {
+
+ }
+
+ @Override
+ public PhysicalOperator visitExchange(Exchange exchange, IndexedFragmentNode iNode) throws ExecutionSetupException {
+ iNode.addAllocation(exchange);
+ PhysicalOperator childEx = exchange.getChild().accept(this, iNode);
+ childEx.setOperatorId(Short.MAX_VALUE & exchange.getOperatorId());
--- End diff --
yes, I can probably skip setting id here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-215850679
Addressed review comments
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61516213
--- Diff: exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java ---
@@ -0,0 +1,350 @@
+/**
+ * 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;
+
+import static org.junit.Assert.*;
+import io.netty.buffer.DrillBuf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.DrillAutoCloseables;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.util.TestTools;
+import org.apache.drill.exec.client.DrillClient;
+import org.apache.drill.exec.client.PrintingResultsListener;
+import org.apache.drill.exec.client.QuerySubmitter.Format;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.QueryData;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserBitShared.QueryType;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.rpc.ConnectionThrottle;
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.user.AwaitableUserResultsListener;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+import org.apache.drill.exec.rpc.user.UserResultsListener;
+import org.apache.drill.exec.util.VectorUtil;
+import org.apache.drill.exec.vector.ValueVector;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Class to test different planning use cases (separate form query execution)
+ *
+ */
+public class DrillSeparatePlanningTest extends BaseTestQuery {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillSeparatePlanningTest.class);
+
+ static final String WORKING_PATH = TestTools.getWorkingPath();
+ static final String TEST_RES_PATH = WORKING_PATH + "/src/test/resources";
+
+ //final String query = "SELECT sales_city, COUNT(*) cnt FROM cp.`region.json` GROUP BY sales_city";
+ //final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+ //final String query = "SELECT o_orderkey, o_custkey FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000 limit 5";
+ //final String query = "SELECT sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000";
+ //final String query = "SELECT o_orderkey FROM dfs.tmp.`multilevel` order by o_orderkey";
+ //final String query = "SELECT dir1, sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 group by dir1 order by dir1";
+ //final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+
+ @Test(timeout=30000)
+ public void testSingleFragmentQuery() throws Exception {
+ final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertEquals(1, planFragments.getFragmentsCount());
+ assertTrue(planFragments.getFragments(0).getLeafFragment());
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentSimpleQuery() throws Exception {
+ final String query = String.format("SELECT o_orderkey FROM dfs_test.`%s/multilevel/json`", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentComplexQuery() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+
+ }
+
+ @Test(timeout=30000)
+ public void testPlanningNoSplit() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ updateTestCluster(2, config);
+
+ List<QueryDataBatch> results = client.runQuery(QueryType.SQL, "alter session set `planner.slice_target`=1");
+ for(QueryDataBatch batch : results) {
+ batch.release();
+ }
+
+ DrillRpcFuture<QueryPlanFragments> queryFragmentsFutures = client.planQuery(QueryType.SQL, query, false);
+
+ final QueryPlanFragments planFragments = queryFragmentsFutures.get();
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ PlanFragment rootFragment = planFragments.getFragments(0);
+ assertFalse(rootFragment.getLeafFragment());
+
+ getCombinedResultsHelper(planFragments);
+
+ client.close();
--- End diff --
tests work both ways - I would not check in w/o successful tests runs :). But, yes, I am removing client.close() since it is done in base class @AfterClass and in my tests I am anyway updating/resetting clients at the beginning of each test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61490732
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java ---
@@ -321,6 +329,38 @@ public void close() {
return listener.getResults();
}
+ public DrillRpcFuture<QueryPlanFragments> planQuery(QueryType type, String query, boolean isSplitPlan) {
+ GetQueryPlanFragments runQuery = GetQueryPlanFragments.newBuilder().setQuery(query).setType(type).setSplitPlan(isSplitPlan).build();
+ return client.submitPlanQuery(runQuery);
+ }
+
+ public void runQuery(QueryType type, List<PlanFragment> planFragments, UserResultsListener resultsListener)
+ throws RpcException {
+ // QueryType can be only executional
+ checkArgument((QueryType.EXECUTION == type), "Only EXECUTIONAL type query is supported with PlanFragments");
+ // setting Plan on RunQuery will be used for logging purposes and therefore can not be null
+ // since there is no Plan string provided we will create a JsonArray out of individual fragment Plans
+ ArrayNode jsonArray = objectMapper.createArrayNode();
+ for (PlanFragment fragment : planFragments) {
+ try {
+ jsonArray.add(objectMapper.readTree(fragment.getFragmentJson()));
+ } catch (IOException e) {
+ logger.error("Exception while trying to read PlanFragment JSON for %s", fragment.getHandle().getQueryId(), e);
+ throw new RpcException(e);
+ }
+ }
+ final String fragmentsToJsonString;
+ try {
+ fragmentsToJsonString = objectMapper.writeValueAsString(jsonArray);
--- End diff --
That's because we are not telling Jackson how to handle cyclicity. Can you try using DrillbitContext#getPlanReader instead?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61346446
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/ExchangeRemoverMaterializer.java ---
@@ -0,0 +1,96 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ * NOTE: this Visitor does NOT set OperatorId, as after Exchange removal all operators need renumbering
+ * Use OperatorIdVisitor on top to set correct OperatorId
+ */
+public class ExchangeRemoverMaterializer extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ExchangeRemoverMaterializer.class);
--- End diff --
oops -will fix
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-205994496
@yufeldman , sorry for the delay in reviewing. I will review the parallelizer related enhancements (the new code since existing parallelizer is not impacted) and related planner changes. I probably won't review the protobuf changes since I am not completely familiar with that code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58629471
--- Diff: protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java ---
@@ -133,6 +133,10 @@ private RpcChannel(int index, int value) {
* <code>PHYSICAL = 3;</code>
*/
PHYSICAL(2, 3),
+ /**
+ * <code>EXECUTIONAL = 4;</code>
+ */
+ EXECUTIONAL(3, 4),
--- End diff --
sure :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-215326874
addressed review comments
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58631107
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple physical plans from a single logical plan
+ * to be able to run them separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
+ * End goal is to get single major fragment: Screen with chain that ends up with a single minor fragment
+ * from Leaf Exchange. This way each plan can run independently without any exchange involvement
+ * @param options
+ * @param foremanNode - not really applicable
+ * @param queryId
+ * @param reader
+ * @param rootNode
+ * @param planningSet
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ private List<QueryWorkUnit> generateWorkUnits(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ // now we generate all the individual plan fragments and associated assignments. Note, we need all endpoints
+ // assigned before we can materialize, so we start a new loop here rather than utilizing the previous one.
+
+ List<QueryWorkUnit> workUnits = Lists.newArrayList();
+ int plansCount = 0;
+ DrillbitEndpoint[] endPoints = null;
+ long initialAllocation = 0;
+ long maxAllocation = 0;
+
+ final Iterator<Wrapper> iter = planningSet.iterator();
+ while (iter.hasNext()) {
+ Wrapper wrapper = iter.next();
+ Fragment node = wrapper.getNode();
+ boolean isLeafFragment = node.getReceivingExchangePairs().size() == 0;
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ // get all the needed info from leaf fragment
+ if ( (physicalOperatorRoot instanceof Exchange) && isLeafFragment) {
+ // need to get info about
+ // number of minor fragments
+ // assignedEndPoints
+ // allocation
+ plansCount = wrapper.getWidth();
+ initialAllocation = (wrapper.getInitialAllocation() != 0 ) ? wrapper.getInitialAllocation()/plansCount : 0;
+ maxAllocation = (wrapper.getMaxAllocation() != 0 ) ? wrapper.getMaxAllocation()/plansCount : 0;
+ endPoints = new DrillbitEndpoint[plansCount];
+ for (int mfId = 0; mfId < plansCount; mfId++) {
+ endPoints[mfId] = wrapper.getAssignedEndpoint(mfId);
--- End diff --
Since the end-point assignments are saved with the plan fragment which will be submitted later (maybe after several minutes or even hours depending on the client), it is possible that due to data movement (e.g rebalancing) the data locality could have changed in the meantime. Is there additional check done at the time the split plan is run ? Is that a concern ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61515592
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,228 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.PlanningSet;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.Wrapper;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple Physical Operator plans from original Physical Operator plan
+ * to be able to run plans separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SimpleParallelizerMultiPlans.class);
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
--- End diff --
After discussing with Aman - in his own words:
Currently, Drill will generate the following type of plan for an ORDER BY:
SingleMergeExchange
|
Sort
|
HashToRandomExchange
|
Scan
This is for global ordering, but if you don't care about that, you can drop the top SingleMergeExchange. The lower HashToRandomExchange is actually only done to divide up the data in some way (we don't currently have RoundRobin exchange), such that a single node does not end up doing all the sort. However, if your data locality is such that data is somewhat uniformly distributed already, you could even drop the lower exchange and do the local Sort at each node or each minor fragment.
So, it sounds like you don't even care about global ordering, so it should be ok to drop..
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61355122
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java ---
@@ -419,6 +425,64 @@ private void runPhysicalPlan(final PhysicalPlan plan) throws ExecutionSetupExcep
logger.debug("Fragments running.");
}
+ /**
+ * This is a helper method to run query based on the list of PlanFragment that were planned
+ * at some point of time
+ * @param fragmentsList
+ * @throws ExecutionSetupException
+ */
+ private void runFragment(List<PlanFragment> fragmentsList) throws ExecutionSetupException {
+ // need to set QueryId, MinorFragment for incoming Fragments
+ PlanFragment rootFragment = null;
+ boolean isFirst = true;
+ final List<PlanFragment> planFragments = Lists.newArrayList();
+ final int fragmentsCount = fragmentsList.size();
+ for (PlanFragment myFragment : fragmentsList) {
+ final FragmentHandle handle = myFragment.getHandle();
+ // for split plan number of minor fragments will be always one,
+ // but minor fragmentId may not be 0, as it is one of the minor fragments from split plan
+ final int minorFragment = (fragmentsCount == 1) ? 0 : handle.getMinorFragmentId();
--- End diff --
good point
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61376393
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java ---
@@ -321,6 +329,38 @@ public void close() {
return listener.getResults();
}
+ public DrillRpcFuture<QueryPlanFragments> planQuery(QueryType type, String query, boolean isSplitPlan) {
+ GetQueryPlanFragments runQuery = GetQueryPlanFragments.newBuilder().setQuery(query).setType(type).setSplitPlan(isSplitPlan).build();
+ return client.submitPlanQuery(runQuery);
+ }
+
+ public void runQuery(QueryType type, List<PlanFragment> planFragments, UserResultsListener resultsListener)
+ throws RpcException {
+ // QueryType can be only executional
+ checkArgument((QueryType.EXECUTION == type), "Only EXECUTIONAL type query is supported with PlanFragments");
+ // setting Plan on RunQuery will be used for logging purposes and therefore can not be null
+ // since there is no Plan string provided we will create a JsonArray out of individual fragment Plans
+ ArrayNode jsonArray = objectMapper.createArrayNode();
+ for (PlanFragment fragment : planFragments) {
+ try {
+ jsonArray.add(objectMapper.readTree(fragment.getFragmentJson()));
+ } catch (IOException e) {
+ logger.error("Exception while trying to read PlanFragment JSON for %s", fragment.getHandle().getQueryId(), e);
+ throw new RpcException(e);
+ }
+ }
+ final String fragmentsToJsonString;
+ try {
+ fragmentsToJsonString = objectMapper.writeValueAsString(jsonArray);
--- End diff --
Unfortunately it creates: Direct self-reference leading to cycle (through reference chain....) if I use List<PlanFragment> directly
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61494244
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/PlanSplitter.java ---
@@ -0,0 +1,142 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.MemoryAllocationUtilities;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class PlanSplitter {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanSplitter.class);
+
+ private static PlanSplitter s_instance = new PlanSplitter();
+
+ private PlanSplitter() {
+
+ }
+
+ public static PlanSplitter getInstance() {
+ return s_instance;
+ }
+
+ /**
+ * Method to plan the query and return list of fragments
+ * it will return query plan "as is" or split plans based on the req setting: split_plan
+ * @param dContext
+ * @param queryId
+ * @param req
+ * @param connection
+ * @return
+ */
+ public QueryPlanFragments planFragments(DrillbitContext dContext, QueryId queryId,
+ GetQueryPlanFragments req, UserClientConnection connection) {
+ QueryPlanFragments.Builder responseBuilder = QueryPlanFragments.newBuilder();
+ QueryContext queryContext = new QueryContext(connection.getSession(), dContext, queryId);
+
+ responseBuilder.setQueryId(queryId);
+
+ try {
+ responseBuilder.addAllFragments(getFragments(dContext, req, queryContext, queryId));
+ responseBuilder.setStatus(QueryState.COMPLETED);
+ } catch (Exception e) {
+ final String errorMessage = String.format("Failed to produce PlanFragments for query id \"%s\" with "
+ + "request to %s plan", queryId, (req.getSplitPlan() ? "split" : "no split"));
+ DrillPBError error = DrillPBError.newBuilder().setMessage(errorMessage).setErrorType(DrillPBError.ErrorType.PLAN).build();
+
+ responseBuilder.setStatus(QueryState.FAILED);
+ responseBuilder.setError(error);
+ }
+ return responseBuilder.build();
+ }
+
+ private List<PlanFragment> getFragments(final DrillbitContext dContext, final GetQueryPlanFragments req,
+ final QueryContext queryContext, final QueryId queryId) throws Exception {
+ final PhysicalPlan plan;
+ final String query = req.getQuery();
+ switch(req.getType()) {
+ case SQL:
+ final Pointer<String> textPlan = new Pointer<>();
+ plan = DrillSqlWorker.getPlan(queryContext, query, textPlan);
+ break;
+ case PHYSICAL:
+ plan = dContext.getPlanReader().readPhysicalPlan(query);
+ break;
+ default:
+ throw new IllegalStateException("Planning fragments supports only SQL or PHYSICAL QueryType");
+ }
+
+ MemoryAllocationUtilities.setupSortMemoryAllocations(plan, queryContext);
+
+ final PhysicalOperator rootOperator = plan.getSortedOperators(false).iterator().next();
+
+ final Fragment rootFragment = rootOperator.accept(MakeFragmentsVisitor.INSTANCE, null);
+ final SimpleParallelizer parallelizer = new SimpleParallelizerMultiPlans(queryContext);
+
+ List<PlanFragment> fragments = Lists.newArrayList();
+
+ if ( req.getSplitPlan() ) {
--- End diff --
Help me understand this please: at this point we for sure know that req is a split request. So why to make this check here again?
If the intention is to allow the client to enable/disable splitting, why do we need this functionality?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61345892
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/client/DrillClient.java ---
@@ -321,6 +329,38 @@ public void close() {
return listener.getResults();
}
+ public DrillRpcFuture<QueryPlanFragments> planQuery(QueryType type, String query, boolean isSplitPlan) {
+ GetQueryPlanFragments runQuery = GetQueryPlanFragments.newBuilder().setQuery(query).setType(type).setSplitPlan(isSplitPlan).build();
+ return client.submitPlanQuery(runQuery);
+ }
+
+ public void runQuery(QueryType type, List<PlanFragment> planFragments, UserResultsListener resultsListener)
+ throws RpcException {
+ // QueryType can be only executional
+ checkArgument((QueryType.EXECUTION == type), "Only EXECUTIONAL type query is supported with PlanFragments");
+ // setting Plan on RunQuery will be used for logging purposes and therefore can not be null
+ // since there is no Plan string provided we will create a JsonArray out of individual fragment Plans
+ ArrayNode jsonArray = objectMapper.createArrayNode();
+ for (PlanFragment fragment : planFragments) {
+ try {
+ jsonArray.add(objectMapper.readTree(fragment.getFragmentJson()));
+ } catch (IOException e) {
+ logger.error("Exception while trying to read PlanFragment JSON for %s", fragment.getHandle().getQueryId(), e);
+ throw new RpcException(e);
+ }
+ }
+ final String fragmentsToJsonString;
+ try {
+ fragmentsToJsonString = objectMapper.writeValueAsString(jsonArray);
--- End diff --
why don't we create & serialize List<String> instead of relying on ArrayNode?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61515624
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java ---
@@ -17,13 +17,27 @@
*/
package org.apache.drill.exec.util;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.fn.impl.DateUtility;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.ExternalSort;
import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.drill.exec.server.options.OptionManager;
public class Utilities {
+
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Utilities.class);
--- End diff --
yeah, sorry - forgot to remove all the imports, etc. will do
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-200495039
@amansinha100 and @hnfgns - could you please review this PR and provide your feedback
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58631424
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple physical plans from a single logical plan
+ * to be able to run them separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
+ * End goal is to get single major fragment: Screen with chain that ends up with a single minor fragment
+ * from Leaf Exchange. This way each plan can run independently without any exchange involvement
+ * @param options
+ * @param foremanNode - not really applicable
+ * @param queryId
+ * @param reader
+ * @param rootNode
+ * @param planningSet
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ private List<QueryWorkUnit> generateWorkUnits(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ // now we generate all the individual plan fragments and associated assignments. Note, we need all endpoints
+ // assigned before we can materialize, so we start a new loop here rather than utilizing the previous one.
+
+ List<QueryWorkUnit> workUnits = Lists.newArrayList();
+ int plansCount = 0;
+ DrillbitEndpoint[] endPoints = null;
+ long initialAllocation = 0;
+ long maxAllocation = 0;
+
+ final Iterator<Wrapper> iter = planningSet.iterator();
+ while (iter.hasNext()) {
+ Wrapper wrapper = iter.next();
+ Fragment node = wrapper.getNode();
+ boolean isLeafFragment = node.getReceivingExchangePairs().size() == 0;
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ // get all the needed info from leaf fragment
+ if ( (physicalOperatorRoot instanceof Exchange) && isLeafFragment) {
+ // need to get info about
+ // number of minor fragments
+ // assignedEndPoints
+ // allocation
+ plansCount = wrapper.getWidth();
+ initialAllocation = (wrapper.getInitialAllocation() != 0 ) ? wrapper.getInitialAllocation()/plansCount : 0;
+ maxAllocation = (wrapper.getMaxAllocation() != 0 ) ? wrapper.getMaxAllocation()/plansCount : 0;
+ endPoints = new DrillbitEndpoint[plansCount];
+ for (int mfId = 0; mfId < plansCount; mfId++) {
+ endPoints[mfId] = wrapper.getAssignedEndpoint(mfId);
--- End diff --
it would be the same concern as with any other application that does splits calculation separately from execution. In this particular case it is no better or worse than use case we are doing this change for.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61521260
--- Diff: exec/java-exec/src/test/java/org/apache/drill/exec/DrillSeparatePlanningTest.java ---
@@ -0,0 +1,350 @@
+/**
+ * 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;
+
+import static org.junit.Assert.*;
+import io.netty.buffer.DrillBuf;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.DrillAutoCloseables;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.util.TestTools;
+import org.apache.drill.exec.client.DrillClient;
+import org.apache.drill.exec.client.PrintingResultsListener;
+import org.apache.drill.exec.client.QuerySubmitter.Format;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.UserBitShared.QueryData;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserBitShared.QueryType;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.record.RecordBatchLoader;
+import org.apache.drill.exec.record.VectorWrapper;
+import org.apache.drill.exec.rpc.ConnectionThrottle;
+import org.apache.drill.exec.rpc.DrillRpcFuture;
+import org.apache.drill.exec.rpc.RpcException;
+import org.apache.drill.exec.rpc.user.AwaitableUserResultsListener;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+import org.apache.drill.exec.rpc.user.UserResultsListener;
+import org.apache.drill.exec.util.VectorUtil;
+import org.apache.drill.exec.vector.ValueVector;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+/**
+ * Class to test different planning use cases (separate form query execution)
+ *
+ */
+public class DrillSeparatePlanningTest extends BaseTestQuery {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DrillSeparatePlanningTest.class);
+
+ static final String WORKING_PATH = TestTools.getWorkingPath();
+ static final String TEST_RES_PATH = WORKING_PATH + "/src/test/resources";
+
+ //final String query = "SELECT sales_city, COUNT(*) cnt FROM cp.`region.json` GROUP BY sales_city";
+ //final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+ //final String query = "SELECT o_orderkey, o_custkey FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000 limit 5";
+ //final String query = "SELECT sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 and o_orderkey > 100 and o_orderkey < 1000";
+ //final String query = "SELECT o_orderkey FROM dfs.tmp.`multilevel` order by o_orderkey";
+ //final String query = "SELECT dir1, sum(o_totalprice) FROM dfs.tmp.`multilevel` where dir0 = 1995 group by dir1 order by dir1";
+ //final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+
+ @Test(timeout=30000)
+ public void testSingleFragmentQuery() throws Exception {
+ final String query = "SELECT * FROM cp.`employee.json` where employee_id > 1 and employee_id < 1000";
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertEquals(1, planFragments.getFragmentsCount());
+ assertTrue(planFragments.getFragments(0).getLeafFragment());
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentSimpleQuery() throws Exception {
+ final String query = String.format("SELECT o_orderkey FROM dfs_test.`%s/multilevel/json`", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+ }
+
+ @Test(timeout=30000)
+ public void testMultiMinorFragmentComplexQuery() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ QueryPlanFragments planFragments = getFragmentsHelper(query);
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ for ( PlanFragment planFragment : planFragments.getFragmentsList()) {
+ assertTrue(planFragment.getLeafFragment());
+ }
+
+ getResultsHelper(planFragments);
+
+ }
+
+ @Test(timeout=30000)
+ public void testPlanningNoSplit() throws Exception {
+ final String query = String.format("SELECT dir0, sum(o_totalprice) FROM dfs_test.`%s/multilevel/json` group by dir0 order by dir0", TEST_RES_PATH);
+
+ updateTestCluster(2, config);
+
+ List<QueryDataBatch> results = client.runQuery(QueryType.SQL, "alter session set `planner.slice_target`=1");
+ for(QueryDataBatch batch : results) {
+ batch.release();
+ }
+
+ DrillRpcFuture<QueryPlanFragments> queryFragmentsFutures = client.planQuery(QueryType.SQL, query, false);
+
+ final QueryPlanFragments planFragments = queryFragmentsFutures.get();
+
+ assertNotNull(planFragments);
+
+ assertTrue((planFragments.getFragmentsCount() > 1));
+
+ PlanFragment rootFragment = planFragments.getFragments(0);
+ assertFalse(rootFragment.getLeafFragment());
+
+ getCombinedResultsHelper(planFragments);
+
+ client.close();
--- End diff --
Thanks for the clarification. Sounds weird/buggish to me that tests are not failing even after closing the client.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61491885
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java ---
@@ -17,13 +17,27 @@
*/
package org.apache.drill.exec.util;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.fn.impl.DateUtility;
+import org.apache.drill.exec.memory.RootAllocatorFactory;
import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.config.ExternalSort;
import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
+import org.apache.drill.exec.server.options.OptionManager;
public class Utilities {
+
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Utilities.class);
--- End diff --
Please revert the changes in this class after migration of the real logic.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on the pull request:
https://github.com/apache/drill/pull/368#issuecomment-215548496
Overall looks good 👍 Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman closed the pull request at:
https://github.com/apache/drill/pull/368
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by yufeldman <gi...@git.apache.org>.
Github user yufeldman commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61523924
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/work/user/PlanSplitter.java ---
@@ -0,0 +1,142 @@
+/**
+ * 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.work.user;
+
+import java.util.List;
+
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.contrib.SimpleParallelizerMultiPlans;
+import org.apache.drill.exec.planner.sql.DrillSqlWorker;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.UserBitShared.DrillPBError;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState;
+import org.apache.drill.exec.proto.UserProtos.GetQueryPlanFragments;
+import org.apache.drill.exec.proto.UserProtos.QueryPlanFragments;
+import org.apache.drill.exec.rpc.user.UserServer.UserClientConnection;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.util.MemoryAllocationUtilities;
+import org.apache.drill.exec.util.Pointer;
+import org.apache.drill.exec.util.Utilities;
+import org.apache.drill.exec.work.QueryWorkUnit;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Helper class to return PlanFragments based on the query plan
+ * or based on split query plan
+ *
+ */
+public class PlanSplitter {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanSplitter.class);
+
+ private static PlanSplitter s_instance = new PlanSplitter();
+
+ private PlanSplitter() {
+
+ }
+
+ public static PlanSplitter getInstance() {
+ return s_instance;
+ }
+
+ /**
+ * Method to plan the query and return list of fragments
+ * it will return query plan "as is" or split plans based on the req setting: split_plan
+ * @param dContext
+ * @param queryId
+ * @param req
+ * @param connection
+ * @return
+ */
+ public QueryPlanFragments planFragments(DrillbitContext dContext, QueryId queryId,
+ GetQueryPlanFragments req, UserClientConnection connection) {
+ QueryPlanFragments.Builder responseBuilder = QueryPlanFragments.newBuilder();
+ QueryContext queryContext = new QueryContext(connection.getSession(), dContext, queryId);
+
+ responseBuilder.setQueryId(queryId);
+
+ try {
+ responseBuilder.addAllFragments(getFragments(dContext, req, queryContext, queryId));
+ responseBuilder.setStatus(QueryState.COMPLETED);
+ } catch (Exception e) {
+ final String errorMessage = String.format("Failed to produce PlanFragments for query id \"%s\" with "
+ + "request to %s plan", queryId, (req.getSplitPlan() ? "split" : "no split"));
+ DrillPBError error = DrillPBError.newBuilder().setMessage(errorMessage).setErrorType(DrillPBError.ErrorType.PLAN).build();
+
+ responseBuilder.setStatus(QueryState.FAILED);
+ responseBuilder.setError(error);
+ }
+ return responseBuilder.build();
+ }
+
+ private List<PlanFragment> getFragments(final DrillbitContext dContext, final GetQueryPlanFragments req,
+ final QueryContext queryContext, final QueryId queryId) throws Exception {
+ final PhysicalPlan plan;
+ final String query = req.getQuery();
+ switch(req.getType()) {
+ case SQL:
+ final Pointer<String> textPlan = new Pointer<>();
+ plan = DrillSqlWorker.getPlan(queryContext, query, textPlan);
+ break;
+ case PHYSICAL:
+ plan = dContext.getPlanReader().readPhysicalPlan(query);
+ break;
+ default:
+ throw new IllegalStateException("Planning fragments supports only SQL or PHYSICAL QueryType");
+ }
+
+ MemoryAllocationUtilities.setupSortMemoryAllocations(plan, queryContext);
+
+ final PhysicalOperator rootOperator = plan.getSortedOperators(false).iterator().next();
+
+ final Fragment rootFragment = rootOperator.accept(MakeFragmentsVisitor.INSTANCE, null);
+ final SimpleParallelizer parallelizer = new SimpleParallelizerMultiPlans(queryContext);
+
+ List<PlanFragment> fragments = Lists.newArrayList();
+
+ if ( req.getSplitPlan() ) {
--- End diff --
I meant list of fragments based on original query and it's physical plan
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58629311
--- Diff: protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java ---
@@ -133,6 +133,10 @@ private RpcChannel(int index, int value) {
* <code>PHYSICAL = 3;</code>
*/
PHYSICAL(2, 3),
+ /**
+ * <code>EXECUTIONAL = 4;</code>
+ */
+ EXECUTIONAL(3, 4),
--- End diff --
Unlike 'Logical' and 'Physical', EXECUTIONAL seems odd since it is not a real word :) If you want to convey that this plan is directly executable (it was already fully planned before), should it just be EXECUTABLE instead ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58624266
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,222 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple physical plans from a single logical plan
+ * to be able to run them separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
+
+ public SimpleParallelizerMultiPlans(QueryContext context) {
+ super(context);
+ }
+
+ /**
+ * Create multiple physical plans from original query planning, it will allow execute them eventually independently
+ * @param options
+ * @param foremanNode
+ * @param queryId
+ * @param activeEndpoints
+ * @param reader
+ * @param rootFragment
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ public List<QueryWorkUnit> getSplitFragments(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ Collection<DrillbitEndpoint> activeEndpoints, PhysicalPlanReader reader, Fragment rootFragment,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ final PlanningSet planningSet = getFragmentsHelper(activeEndpoints, rootFragment);
+
+ return generateWorkUnits(
+ options, foremanNode, queryId, reader, rootFragment, planningSet, session, queryContextInfo);
+ }
+
+ /**
+ * Split plan into multiple plans based on parallelization
+ * Ideally it is applicable only to plans with two major fragments: Screen and UnionExchange
+ * But there could be cases where we can remove even multiple exchanges like in case of "order by"
+ * End goal is to get single major fragment: Screen with chain that ends up with a single minor fragment
+ * from Leaf Exchange. This way each plan can run independently without any exchange involvement
+ * @param options
+ * @param foremanNode - not really applicable
+ * @param queryId
+ * @param reader
+ * @param rootNode
+ * @param planningSet
+ * @param session
+ * @param queryContextInfo
+ * @return
+ * @throws ExecutionSetupException
+ */
+ private List<QueryWorkUnit> generateWorkUnits(OptionList options, DrillbitEndpoint foremanNode, QueryId queryId,
+ PhysicalPlanReader reader, Fragment rootNode, PlanningSet planningSet,
+ UserSession session, QueryContextInformation queryContextInfo) throws ExecutionSetupException {
+
+ // now we generate all the individual plan fragments and associated assignments. Note, we need all endpoints
+ // assigned before we can materialize, so we start a new loop here rather than utilizing the previous one.
+
+ List<QueryWorkUnit> workUnits = Lists.newArrayList();
+ int plansCount = 0;
+ DrillbitEndpoint[] endPoints = null;
+ long initialAllocation = 0;
+ long maxAllocation = 0;
+
+ final Iterator<Wrapper> iter = planningSet.iterator();
+ while (iter.hasNext()) {
+ Wrapper wrapper = iter.next();
+ Fragment node = wrapper.getNode();
+ boolean isLeafFragment = node.getReceivingExchangePairs().size() == 0;
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ // get all the needed info from leaf fragment
+ if ( (physicalOperatorRoot instanceof Exchange) && isLeafFragment) {
+ // need to get info about
+ // number of minor fragments
+ // assignedEndPoints
+ // allocation
+ plansCount = wrapper.getWidth();
+ initialAllocation = (wrapper.getInitialAllocation() != 0 ) ? wrapper.getInitialAllocation()/plansCount : 0;
+ maxAllocation = (wrapper.getMaxAllocation() != 0 ) ? wrapper.getMaxAllocation()/plansCount : 0;
+ endPoints = new DrillbitEndpoint[plansCount];
+ for (int mfId = 0; mfId < plansCount; mfId++) {
+ endPoints[mfId] = wrapper.getAssignedEndpoint(mfId);
+ }
+ }
+ }
+ if ( plansCount == 0 ) {
+ // no exchange, return list of single QueryWorkUnit
+ workUnits.add(generateWorkUnit(options, foremanNode, queryId, reader, rootNode, planningSet, session, queryContextInfo));
+ return workUnits;
+ }
+
+ for (Wrapper wrapper : planningSet) {
+ Fragment node = wrapper.getNode();
+ final PhysicalOperator physicalOperatorRoot = node.getRoot();
+ if ( physicalOperatorRoot instanceof Exchange ) {
+ // get to 0 MajorFragment
+ continue;
+ }
+ boolean isRootNode = rootNode == node;
+
+ if (isRootNode && wrapper.getWidth() != 1) {
+ throw new ForemanSetupException(String.format("Failure while trying to setup fragment. " +
+ "The root fragment must always have parallelization one. In the current case, the width was set to %d.",
+ wrapper.getWidth()));
+ }
+ // this fragment is always leaf, as we are removing all the exchanges
+ boolean isLeafFragment = true;
+
+ // Create a minorFragment for each major fragment.
+ for (int minorFragmentId = 0; minorFragmentId < plansCount; minorFragmentId++) {
+ // those fragments should be empty
+ List<PlanFragment> fragments = Lists.newArrayList();
+
+ PlanFragment rootFragment = null;
+ FragmentRoot rootOperator = null;
+
+ IndexedFragmentNode iNode = new IndexedFragmentNode(minorFragmentId, wrapper);
+ wrapper.resetAllocation();
+ // two visitors here
+ // 1. To remove exchange
+ // 2. To reset operator IDs as exchanges were removed
+ PhysicalOperator op = physicalOperatorRoot.accept(ExchangeManipulatorMaterializerVisitor.INSTANCE, iNode).
--- End diff --
In this case, if the Exchange remover visitor did not drop any exchanges, then the OperatorId visitor should not be called. Can you add a check for that ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by hnfgns <gi...@git.apache.org>.
Github user hnfgns commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r61491072
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/contrib/SimpleParallelizerMultiPlans.java ---
@@ -0,0 +1,228 @@
+/**
+ * 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.planner.fragment.contrib;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.util.DrillStringUtils;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.FragmentRoot;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.fragment.Fragment;
+import org.apache.drill.exec.planner.fragment.PlanningSet;
+import org.apache.drill.exec.planner.fragment.SimpleParallelizer;
+import org.apache.drill.exec.planner.fragment.Wrapper;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+import org.apache.drill.exec.proto.BitControl.PlanFragment;
+import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.proto.ExecProtos.FragmentHandle;
+import org.apache.drill.exec.proto.UserBitShared.QueryId;
+import org.apache.drill.exec.rpc.user.UserSession;
+import org.apache.drill.exec.server.options.OptionList;
+import org.apache.drill.exec.work.QueryWorkUnit;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+
+/**
+ * SimpleParallelizerMultiPlans class is an extension to SimpleParallelizer
+ * to help with getting PlanFragments for split plan.
+ * Split plan is essentially ability to create multiple Physical Operator plans from original Physical Operator plan
+ * to be able to run plans separately.
+ * Moving functionality specific to splitting the plan to this class
+ * allows not to pollute parent class with non-authentic functionality
+ *
+ */
+public class SimpleParallelizerMultiPlans extends SimpleParallelizer {
--- End diff --
Name advise: SplittingParallelizer + documentation seems pretty good here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] drill pull request: DRILL-4132 Ability to submit simple type of ph...
Posted by amansinha100 <gi...@git.apache.org>.
Github user amansinha100 commented on a diff in the pull request:
https://github.com/apache/drill/pull/368#discussion_r58623875
--- Diff: exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/ExchangeManipulatorMaterializerVisitor.java ---
@@ -0,0 +1,97 @@
+/**
+ * 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.planner.fragment;
+
+import java.util.List;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.FragmentSetupException;
+import org.apache.drill.exec.physical.PhysicalOperatorSetupException;
+import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor;
+import org.apache.drill.exec.physical.base.Exchange;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.Store;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.planner.fragment.Materializer.IndexedFragmentNode;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Materializer visitor to remove exchange(s)
+ */
+public class ExchangeManipulatorMaterializerVisitor extends AbstractPhysicalVisitor<PhysicalOperator, Materializer.IndexedFragmentNode, ExecutionSetupException> {
--- End diff --
Minor comment: the name of the class is too verbose...should this just be ExhangeRemover ? Although, it is still a bit confusing since this is removing a PhysicalOperator, not a Prel...in any case let's think of a better name.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---