diff --git a/be/src/runtime/exec-env.cc b/be/src/runtime/exec-env.cc
index 22049b1..7d7fea4 100644
--- a/be/src/runtime/exec-env.cc
+++ b/be/src/runtime/exec-env.cc
@@ -165,16 +165,46 @@
 
 namespace {
 using namespace impala;
-// Helper method to hand off cluster membership update to the frontend instance.
+/// Helper method to forward cluster membership updates to the frontend.
+///
+/// The frontend uses cluster membership information to determine whether it expects the
+/// scheduler to assign local or remote reads. It also uses the number of executors to
+/// determine the join type (partitioned vs broadcast). For the default executor group, we
+/// assume that local reads are preferred and will include the hostnames and IP addresses
+/// in the update to the frontend. For non-default executor groups, we assume that we will
+/// read data remotely and will only send the number of executors in the largest healthy
+/// group.
 void SendClusterMembershipToFrontend(
     ClusterMembershipMgr::SnapshotPtr& snapshot, Frontend* frontend) {
   TUpdateExecutorMembershipRequest update_req;
-  for (const auto& it : snapshot->current_backends) {
-    const TBackendDescriptor& backend = it.second;
-    if (backend.is_executor) {
-      update_req.hostnames.insert(backend.address.hostname);
-      update_req.ip_addresses.insert(backend.ip_address);
-      update_req.num_executors++;
+  const ExecutorGroup* group = nullptr;
+  bool is_default_group = false;
+  auto default_it =
+      snapshot->executor_groups.find(ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME);
+  if (default_it != snapshot->executor_groups.end()) {
+    is_default_group = true;
+    group = &(default_it->second);
+  } else {
+    int max_num_executors = 0;
+    // Find largest healthy group.
+    for (const auto& it : snapshot->executor_groups) {
+      if (!it.second.IsHealthy()) continue;
+      int num_executors = it.second.NumExecutors();
+      if (num_executors > max_num_executors) {
+        max_num_executors = num_executors;
+        group = &(it.second);
+      }
+    }
+  }
+  if (group) {
+    for (const auto& backend : group->GetAllExecutorDescriptors()) {
+      if (backend.is_executor) {
+        if (is_default_group) {
+          update_req.hostnames.insert(backend.address.hostname);
+          update_req.ip_addresses.insert(backend.ip_address);
+        }
+        update_req.num_executors++;
+      }
     }
   }
   Status status = frontend->UpdateExecutorMembership(update_req);
diff --git a/be/src/service/impala-server.cc b/be/src/service/impala-server.cc
index 6d0a20a..c79d708 100644
--- a/be/src/service/impala-server.cc
+++ b/be/src/service/impala-server.cc
@@ -249,6 +249,11 @@
     "contains at least that number of executors for the group will it be considered "
     "healthy for admission. Currently only a single group may be specified.");
 
+DEFINE_int32(num_expected_executors, 20, "The number of executors that are expected to "
+    "be available for the execution of a single query. This value is used during "
+    "planning if no executors have started yet. Once a healthy executor group has "
+    "started, its size is used instead.");
+
 // TODO: can we automatically choose a startup grace period based on the max admission
 // control queue timeout + some margin for error?
 DEFINE_int64(shutdown_grace_period_s, 120, "Shutdown startup grace period in seconds. "
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index 56f6a65..c5eb39e 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -82,6 +82,7 @@
 DECLARE_bool(unlock_zorder_sort);
 DECLARE_string(blacklisted_tables);
 DECLARE_bool(simplify_check_on_show_tables);
+DECLARE_int32(num_expected_executors);
 
 namespace impala {
 
@@ -167,6 +168,7 @@
   cfg.__set_unlock_zorder_sort(FLAGS_unlock_zorder_sort);
   cfg.__set_blacklisted_tables(FLAGS_blacklisted_tables);
   cfg.__set_simplify_check_on_show_tables(FLAGS_simplify_check_on_show_tables);
+  cfg.__set_num_expected_executors(FLAGS_num_expected_executors);
   RETURN_IF_ERROR(SerializeThriftMsg(jni_env, &cfg, cfg_bytes));
   return Status::OK();
 }
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index af29d2c..7fc47d5 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -145,4 +145,6 @@
   60: required bool simplify_check_on_show_tables
 
   61: required bool mt_dop_auto_fallback
+
+  62: required i32 num_expected_executors
 }
diff --git a/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java b/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
index 3482d6b..e852a14 100644
--- a/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
+++ b/fe/src/main/java/org/apache/impala/util/ExecutorMembershipSnapshot.java
@@ -20,6 +20,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.impala.service.BackendConfig;
 import org.apache.impala.thrift.TNetworkAddress;
 import org.apache.impala.thrift.TUpdateExecutorMembershipRequest;
 import com.google.common.collect.Sets;
@@ -35,30 +36,50 @@
   private static AtomicReference<ExecutorMembershipSnapshot> cluster_ =
       new AtomicReference<ExecutorMembershipSnapshot>(new ExecutorMembershipSnapshot());
 
-  // The set of hosts that are members of the cluster given by hostname.
+  // The set of hosts that are members of the cluster given by hostname. Can be empty when
+  // using executor groups, in which case the planner shall assume that all reads are
+  // remote.
   private final Set<String> hostnames_;
 
-  // The set of hosts that are members of the cluster given by IP address.
+  // The set of hosts that are members of the cluster given by IP address. Can be empty
+  // when using executor groups, in which case the planner shall assume that all reads are
+  // remote.
   private final Set<String> ipAddresses_;
 
-  // The number of executor nodes of the cluster.  Normally, this will be equal to
-  // hostnames_.size(), except in the test minicluster where there are multiple
-  // impalad's running on a single host.
+  // The number of executor nodes of the cluster.
+  //
+  // When not using executor groups, this value reflects the number of executors in the
+  // cluster. It will be equal to hostnames_.size(), except in the test minicluster where
+  // multiple impalads are running on a single host.
+
+  // When using executor groups, this value reflects the number of executors in the
+  // largest healthy group. If all group become unhealthy, the backend will not send a
+  // membership update and this value will retain the last non-zero value. This allows the
+  // planner to work on the assumption that a healthy executor group of the same size will
+  // eventually come online to execute queries.
   private final int numExecutors_;
 
-  // Used only to construct the initial ExecutorMembershipSnapshot. Before we get the
-  // first snapshot, assume one node (the localhost) to mimic Scheduler.
+  // Used only to construct the initial ExecutorMembershipSnapshot.
   private ExecutorMembershipSnapshot() {
     hostnames_ = Sets.newHashSet();
     ipAddresses_ = Sets.newHashSet();
-    numExecutors_ = 1;
+    // We use 0 for the number of executors to indicate that no update from the
+    // ClusterMembershipManager has arrived yet and we will return the value
+    // '-num_expected_executors' in numExecutors().
+    numExecutors_ = 0;
   }
 
   // Construct a new snapshot based on the TUpdateExecutorMembershipRequest.
   private ExecutorMembershipSnapshot(TUpdateExecutorMembershipRequest request) {
     hostnames_ = request.getHostnames();
     ipAddresses_ = request.getIp_addresses();
-    numExecutors_ = request.getNum_executors();
+    // If the updates snapshot does not contain any executors we fall back to the previous
+    // value. This can happen if no healthy executor groups are currently online.
+    if (request.getNum_executors() > 0) {
+      numExecutors_ = request.getNum_executors();
+    } else {
+      numExecutors_ = cluster_.get().numExecutors_;
+    }
   }
 
   // Determine whether a host, given either by IP address or hostname, is a member of
@@ -68,8 +89,15 @@
     return ipAddresses_.contains(host) || hostnames_.contains(host);
   }
 
-  // The number of nodes in this snapshot.
-  public int numExecutors() { return numExecutors_; }
+  // Return the number of executors that should be used for planning. If no executors have
+  // been registered so far, this method will return a configurable default to allow the
+  // planner to operated based on the expected number of executors.
+  public int numExecutors() {
+    if (numExecutors_ == 0) {
+      return BackendConfig.INSTANCE.getBackendCfg().num_expected_executors;
+    }
+    return numExecutors_;
+  }
 
   // Atomically update the singleton snapshot instance.  After the update completes,
   // all calls to getCluster() will return the new snapshot.
diff --git a/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java b/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java
new file mode 100644
index 0000000..5f6d3ba
--- /dev/null
+++ b/fe/src/test/java/org/apache/impala/planner/ClusterSizeTest.java
@@ -0,0 +1,95 @@
+// 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.impala.planner;
+
+import org.apache.impala.common.FrontendTestBase;
+import org.apache.impala.common.ImpalaException;
+import org.apache.impala.testutil.TestUtils;
+import org.apache.impala.thrift.TQueryCtx;
+import org.apache.impala.thrift.TUpdateExecutorMembershipRequest;
+import org.apache.impala.util.ExecutorMembershipSnapshot;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import org.junit.Test;
+import com.google.common.collect.Sets;
+
+/**
+ * Unit tests to make sure that the planner handles changes to the cluster size correctly.
+ */
+public class ClusterSizeTest extends FrontendTestBase {
+
+  /**
+   * Retrieves the explain string for query statement 'stmt' from the frontend.
+   */
+  private String getExplainString(String stmt) {
+    String ret = "";
+    try {
+      TQueryCtx queryCtx = TestUtils.createQueryContext(
+          "functional", System.getProperty("user.name"));
+      queryCtx.client_request.setStmt(stmt);
+      ret = frontend_.getExplainString(queryCtx);
+    } catch (ImpalaException e) {
+      fail(e.getMessage());
+    }
+    return ret;
+  }
+
+  /**
+   * Sends an update to the ExecutorMembershipSnapshot containing the specified number of
+   * executors. The host list will only contain localhost.
+   */
+  private void setNumExecutors(int num) {
+    TUpdateExecutorMembershipRequest updateReq = new TUpdateExecutorMembershipRequest();
+    updateReq.setIp_addresses(Sets.newHashSet("127.0.0.1"));
+    updateReq.setHostnames(Sets.newHashSet("localhost"));
+    updateReq.setNum_executors(num);
+    ExecutorMembershipSnapshot.update(updateReq);
+  }
+
+  /**
+   * IMPALA-9151: Tests that the planner selects the correct join strategy based on the
+   * number of executors in the cluster.
+   */
+  @Test
+  public void testChangeClusterSize() {
+    final String query = "select * from alltypes a inner join alltypes b on a.id = b.id";
+    final String broadcast_exchange = ":EXCHANGE [BROADCAST]";
+    final String hash_exchange = ":EXCHANGE [HASH(b.id)]";
+
+    // By default no executors are registered and the planner falls back to the value of
+    // -num_expected_executors, which is 20 by default.
+    assertTrue(getExplainString(query).contains(hash_exchange));
+
+    // Adding a single executor will make the planner switch to a broadcast join.
+    setNumExecutors(1);
+    assertTrue(getExplainString(query).contains(broadcast_exchange));
+
+    // Adding two or more executors will make the planner switch to a partitioned hash
+    // join.
+    for (int n = 2; n < 5; ++n) {
+      setNumExecutors(n);
+      assertTrue(getExplainString(query).contains(hash_exchange));
+    }
+
+    // If the backend reports a single executor, the planner should fall back to a
+    // broadcast join.
+    setNumExecutors(1);
+    assertTrue(getExplainString(query).contains(broadcast_exchange));
+  }
+}
diff --git a/tests/custom_cluster/test_coordinators.py b/tests/custom_cluster/test_coordinators.py
index 0bf5164..613fcf9 100644
--- a/tests/custom_cluster/test_coordinators.py
+++ b/tests/custom_cluster/test_coordinators.py
@@ -298,3 +298,15 @@
                      "pool default-pool. Queued reason: Waiting for executors to " \
                      "start. Only DDL queries can currently run."
     assert expected_error in str(result)
+
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args(cluster_size=1, num_exclusive_coordinators=1,
+                                    impalad_args="-num_expected_executors=10")
+  def test_num_expected_executors_flag(self):
+    """Verifies that the '-num_expected_executors' flag is effective."""
+    client = self.cluster.impalads[0].service.create_beeswax_client()
+    client.execute("set explain_level=2")
+    ret = client.execute("explain select * from functional.alltypes a inner join "
+                         "functional.alltypes b on a.id = b.id;")
+    num_hosts = "hosts=10 instances=10"
+    assert num_hosts in str(ret)
diff --git a/tests/custom_cluster/test_executor_groups.py b/tests/custom_cluster/test_executor_groups.py
index 1bb070a..9b696bb 100644
--- a/tests/custom_cluster/test_executor_groups.py
+++ b/tests/custom_cluster/test_executor_groups.py
@@ -431,3 +431,77 @@
     self.coordinator.service.wait_for_metric_value(
       "admission-controller.executor-group.num-queries-executing.{0}".format(
         self._group_name(group_names[0])), 0, timeout=30)
+
+  @pytest.mark.execute_serially
+  def test_join_strategy_single_executor(self):
+    """Tests that the planner picks the correct join strategy based on the current cluster
+    size. This test uses an executor group with a minimum size of 1."""
+    TABLE = "functional.alltypes"
+    QUERY = "explain select * from %s a inner join %s b on a.id = b.id" % (TABLE, TABLE)
+
+    # Predicates to assert that a certain join type was picked.
+    def assert_broadcast_join():
+      ret = self.execute_query_expect_success(self.client, QUERY)
+      assert ":EXCHANGE [BROADCAST]" in str(ret)
+
+    def assert_hash_join():
+      ret = self.execute_query_expect_success(self.client, QUERY)
+      assert ":EXCHANGE [HASH(b.id)]" in str(ret)
+
+    # Without any executors we default to a hash join.
+    assert_hash_join()
+
+    # Add a healthy executor group of size 1 and observe that we switch to broadcast join.
+    self._add_executor_group("group1", min_size=1, num_executors=1)
+    assert_broadcast_join()
+
+    # Add another executor to the same group and observe that with two executors we pick a
+    # partitioned hash join.
+    self._add_executor_group("group1", min_size=1, num_executors=1)
+    assert_hash_join()
+
+    # Kill an executor. The group remains healthy but its size decreases and we revert
+    # back to a broadcast join.
+    self.cluster.impalads[-1].kill()
+    self.coordinator.service.wait_for_metric_value("cluster-membership.backends.total", 2,
+                                                   timeout=20)
+    assert_broadcast_join()
+
+    # Kill a second executor. The group becomes unhealthy but we cache its last healthy
+    # size and will continue to pick a broadcast join.
+    self.cluster.impalads[-2].kill()
+    self.coordinator.service.wait_for_metric_value("cluster-membership.backends.total", 1,
+                                                   timeout=20)
+    assert_broadcast_join()
+
+  @pytest.mark.execute_serially
+  def test_join_strategy_multiple_executors(self):
+    """Tests that the planner picks the correct join strategy based on the current cluster
+    size. This test uses an executor group which requires multiple executors to be
+    healthy."""
+    TABLE = "functional.alltypes"
+    QUERY = "explain select * from %s a inner join %s b on a.id = b.id" % (TABLE, TABLE)
+
+    # Predicate to assert that the planner decided on a hash join.
+    def assert_hash_join():
+      ret = self.execute_query_expect_success(self.client, QUERY)
+      assert ":EXCHANGE [HASH(b.id)]" in str(ret)
+
+    # Without any executors we default to a hash join.
+    assert_hash_join()
+
+    # Adding an unhealthy group will not affect the planner's decision.
+    self._add_executor_group("group1", min_size=2, num_executors=1)
+    assert_hash_join()
+
+    # Adding a second executor makes the group healthy (note that the resulting join
+    # strategy is the same though).
+    self._add_executor_group("group1", min_size=2, num_executors=1)
+    assert_hash_join()
+
+    # Kill an executor. The unhealthy group does not affect the planner's decision, even
+    # though only one executor is now online.
+    self.cluster.impalads[-1].kill()
+    self.coordinator.service.wait_for_metric_value("cluster-membership.backends.total", 2,
+                                                   timeout=20)
+    assert_hash_join()