.../src/Plugin/migrate/source/SourcePluginBase.php | 33 ++++++------
 .../migrate/src/Plugin/migrate/source/SqlBase.php  |  3 +-
 .../migrate_sqlbase_count_cache_test.info.yml      |  7 +++
 .../src/Plugin/migrate/source/SqlCountCache.php    | 44 +++++++++++++++
 .../src/Kernel/MigrateSqlSourceCountCacheTest.php  | 62 ++++++++++++++++++++++
 5 files changed, 133 insertions(+), 16 deletions(-)

diff --git a/core/modules/migrate/src/Plugin/migrate/source/SourcePluginBase.php b/core/modules/migrate/src/Plugin/migrate/source/SourcePluginBase.php
index 8d69418262..db7691e8af 100644
--- a/core/modules/migrate/src/Plugin/migrate/source/SourcePluginBase.php
+++ b/core/modules/migrate/src/Plugin/migrate/source/SourcePluginBase.php
@@ -450,29 +450,32 @@ public function count($refresh = FALSE) {
       return -1;
     }
 
-    if (!isset($this->cacheKey)) {
-      $this->cacheKey = hash('sha256', $this->getPluginId());
-    }
+    $count = NULL;
 
     // If a refresh is requested, or we're not caching counts, ask the derived
     // class to get the count from the source.
     if ($refresh || !$this->cacheCounts) {
       $count = $this->doCount();
-      $this->getCache()->set($this->cacheKey, $count);
     }
-    else {
-      // Caching is in play, first try to retrieve a cached count.
-      $cache_object = $this->getCache()->get($this->cacheKey, 'cache');
-      if (is_object($cache_object)) {
-        // Success.
-        $count = $cache_object->data;
+
+    // If count caching is enabled: read from cache or update it.
+    if ($this->cacheCounts) {
+      if (!isset($this->cacheKey)) {
+        $this->cacheKey = hash('sha256', $this->getPluginId());
       }
-      else {
-        // No cached count, ask the derived class to count 'em up, and cache
-        // the result.
-        $count = $this->doCount();
-        $this->getCache()->set($this->cacheKey, $count);
+      if ($count === NULL) {
+        $cache_object = $this->getCache()->get($this->cacheKey, 'cache');
+        if (is_object($cache_object)) {
+          // Early return for a cache hit.
+          return $cache_object->data;
+        }
+        else {
+          $count = $this->doCount();
+        }
       }
+      // No early return, so this was not a cache hit. Either a forced refresh
+      // or a cache miss would lead to this point. Either way, update the cache.
+      $this->getCache()->set($this->cacheKey, $count);
     }
     return $count;
   }
diff --git a/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php b/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
index cf13b46392..cb9cc3ff16 100644
--- a/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
+++ b/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
@@ -385,7 +385,8 @@ protected function fetchNextBatch() {
   /**
    * {@inheritdoc}
    */
-  public function count($refresh = FALSE) {
+  protected function doCount() {
+    // Uses countQuery() because it is faster than the parent implementation.
     return (int) $this->query()->countQuery()->execute()->fetchField();
   }
 
diff --git a/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/migrate_sqlbase_count_cache_test.info.yml b/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/migrate_sqlbase_count_cache_test.info.yml
new file mode 100644
index 0000000000..722bcd5941
--- /dev/null
+++ b/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/migrate_sqlbase_count_cache_test.info.yml
@@ -0,0 +1,7 @@
+type: module
+name: Migrate sqlbase count cache test.
+description: 'Provides a source plugin to test that counts are cached in sql sources.'
+package: Testing
+core: 8.x
+dependencies:
+- drupal:migrate
diff --git a/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/src/Plugin/migrate/source/SqlCountCache.php b/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/src/Plugin/migrate/source/SqlCountCache.php
new file mode 100644
index 0000000000..d8af575e46
--- /dev/null
+++ b/core/modules/migrate/tests/modules/migrate_sqlbase_count_cache_test/src/Plugin/migrate/source/SqlCountCache.php
@@ -0,0 +1,44 @@
+<?php
+
+namespace Drupal\migrate_sqlbase_count_cache_test\Plugin\migrate\source;
+
+use Drupal\migrate\Plugin\migrate\source\SqlBase;
+
+/**
+ * Source plugin for SqlBase count cache test.
+ *
+ * @MigrateSource(
+ *   id = "sql_count_cache",
+ *   source_module = "migrate",
+ * )
+ */
+class SqlCountCache extends SqlBase {
+
+  /**
+   * {@inheritdoc}
+   */
+  public function fields() {
+    return [
+      'id' => t('Id'),
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getIds() {
+    return [
+      'id' => [
+        'type' => 'integer',
+      ],
+    ];
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function query() {
+    return $this->select('source_table', 's')->fields('s', ['id']);
+  }
+
+}
diff --git a/core/modules/migrate/tests/src/Kernel/MigrateSqlSourceCountCacheTest.php b/core/modules/migrate/tests/src/Kernel/MigrateSqlSourceCountCacheTest.php
new file mode 100644
index 0000000000..f370594646
--- /dev/null
+++ b/core/modules/migrate/tests/src/Kernel/MigrateSqlSourceCountCacheTest.php
@@ -0,0 +1,62 @@
+<?php
+
+namespace Drupal\Tests\migrate\Kernel;
+
+/**
+ * Tests SqlBase source count caching.
+ *
+ * @covers \Drupal\migrate_sqlbase_count_cache_test\Plugin\migrate\source\SqlCountCache
+ *
+ * @group migrate
+ */
+class MigrateSqlSourceCountCacheTest extends MigrateSqlSourceTestBase {
+
+  public static $modules = ['migrate_sqlbase_count_cache_test'];
+
+  /**
+   * Tests that the count is cached.
+   *
+   * @dataProvider providerSource
+   */
+  public function testCountCache($source_data, $expected_results) {
+    $plugin = $this->getPlugin([]);
+
+    // Since we don't yet inject the database connection, we need to use a
+    // reflection hack to set it in the plugin instance.
+    $reflector = new \ReflectionObject($plugin);
+    $property = $reflector->getProperty('database');
+    $property->setAccessible(TRUE);
+    $property->setValue($plugin, $this->getDatabase($source_data));
+
+    $count = $plugin->count();
+    $this->assertSame(4, $count);
+    $cache_key = hash('sha256', $plugin->getPluginId());
+    $cached_count = \Drupal::cache('migrate')->get($cache_key, 'cache')->data;
+    $this->assertSame($count, $cached_count);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function providerSource() {
+    return [
+      [
+        'source_data' => [
+          'source_table' => [
+            ['id' => 1],
+            ['id' => 2],
+            ['id' => 3],
+            ['id' => 4],
+          ],
+        ],
+        'expected_result' => [
+          ['id' => 1],
+          ['id' => 2],
+          ['id' => 3],
+          ['id' => 4],
+        ],
+      ],
+    ];
+  }
+
+}