diff --git a/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php b/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
index 0bb72e7d59ae296c0d6b7d57679ad6ebeab2e50f..881e2fbcf62cd632373ebbe5beb6a3f3b4ef02ba 100644
--- a/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
+++ b/core/modules/migrate/src/Plugin/migrate/source/SqlBase.php
@@ -303,11 +303,17 @@ protected function initializeIterator() {
       }
       // 2. If we are using high water marks, also include rows above the mark.
       //    But, include all rows if the high water mark is not set.
-      if ($this->getHighWaterProperty() && ($high_water = $this->getHighWater())) {
+      if ($this->getHighWaterProperty()) {
         $high_water_field = $this->getHighWaterField();
-        $conditions->condition($high_water_field, $high_water, '>');
+        $high_water = $this->getHighWater();
+        if ($high_water) {
+          $conditions->condition($high_water_field, $high_water, '>');
+          $condition_added = TRUE;
+        }
+        // Always sort by the high water field, to ensure that the first run
+        // (before we have a high water value) also has the results in a
+        // consistent order.
         $this->query->orderBy($high_water_field);
-        $condition_added = TRUE;
       }
       if ($condition_added) {
         $this->query->condition($conditions);
diff --git a/core/modules/migrate/tests/src/Kernel/SqlBaseTest.php b/core/modules/migrate/tests/src/Kernel/SqlBaseTest.php
index faf064d21f7242000ccb9712b6f08f7c6fc1ef22..7115860c39dd0134b53937a1e9f0e60e89d68b58 100644
--- a/core/modules/migrate/tests/src/Kernel/SqlBaseTest.php
+++ b/core/modules/migrate/tests/src/Kernel/SqlBaseTest.php
@@ -7,9 +7,12 @@
 
 namespace Drupal\Tests\migrate\Kernel;
 
+use Drupal\Core\Database\Query\ConditionInterface;
+use Drupal\Core\Database\Query\SelectInterface;
 use Drupal\migrate\Exception\RequirementsException;
-use Drupal\migrate\Plugin\migrate\source\TestSqlBase;
 use Drupal\Core\Database\Database;
+use Drupal\migrate\Plugin\migrate\source\SqlBase;
+use Drupal\migrate\Plugin\MigrationInterface;
 
 /**
  * Tests the functionality of SqlBase.
@@ -18,11 +21,28 @@
  */
 class SqlBaseTest extends MigrateTestBase {
 
+  /**
+   * The (probably mocked) migration under test.
+   *
+   * @var \Drupal\migrate\Plugin\MigrationInterface
+   */
+  protected $migration;
+
+  /**
+   * {@inheritdoc}
+   */
+  protected function setUp() {
+    parent::setUp();
+
+    $this->migration = $this->getMock(MigrationInterface::class);
+    $this->migration->method('id')->willReturn('fubar');
+  }
+
   /**
    * Tests different connection types.
    */
   public function testConnectionTypes() {
-    $sql_base = new TestSqlBase();
+    $sql_base = new TestSqlBase([], $this->migration);
 
     // Verify that falling back to the default 'migrate' connection (defined in
     // the base class) works.
@@ -53,7 +73,7 @@ public function testConnectionTypes() {
     $this->assertSame($sql_base->getDatabase()->getKey(), $key);
 
     // Now test we can have SqlBase create the connection from an info array.
-    $sql_base = new TestSqlBase();
+    $sql_base = new TestSqlBase([], $this->migration);
 
     $target = 'test_db_target2';
     $key = 'test_migrate_connection2';
@@ -81,7 +101,7 @@ public function testConnectionTypes() {
     $this->assertSame($sql_base->getDatabase()->getKey(), $key);
 
     // Now test we can have SqlBase create the connection from an info array.
-    $sql_base = new TestSqlBase();
+    $sql_base = new TestSqlBase([], $this->migration);
 
     $target = 'test_state_db_target2';
     $key = 'test_state_migrate_connection2';
@@ -107,9 +127,55 @@ public function testConnectionTypes() {
     $sql_base->getDatabase();
   }
 
-}
+  /**
+   * Tests that SqlBase respects high-water values.
+   *
+   * @param mixed $high_water
+   *   (optional) The high-water value to set.
+   * @param array $query_result
+   *   (optional) The expected query results.
+   *
+   * @dataProvider highWaterDataProvider
+   */
+  public function testHighWater($high_water = NULL, array $query_result = []) {
+    $configuration = [
+      'high_water_property' => [
+        'name' => 'order',
+      ],
+    ];
+    $source = new TestSqlBase($configuration, $this->migration);
+
+    if ($high_water) {
+      $source->getHighWaterStorage()->set($this->migration->id(), $high_water);
+    }
+
+    $query_result = new \ArrayIterator($query_result);
 
-namespace Drupal\migrate\Plugin\migrate\source;
+    $query = $this->getMock(SelectInterface::class);
+    $query->method('execute')->willReturn($query_result);
+    $query->expects($this->atLeastOnce())->method('orderBy')->with('order', 'ASC');
+
+    $condition_group = $this->getMock(ConditionInterface::class);
+    $query->method('orConditionGroup')->willReturn($condition_group);
+
+    $source->setQuery($query);
+    $source->rewind();
+  }
+
+  /**
+   * Data provider for ::testHighWater().
+   *
+   * @return array
+   *   The scenarios to test.
+   */
+  public function highWaterDataProvider() {
+    return [
+      'no high-water value set' => [],
+      'high-water value set' => [33],
+    ];
+  }
+
+}
 
 /**
  * A dummy source to help with testing SqlBase.
@@ -118,11 +184,23 @@ public function testConnectionTypes() {
  */
 class TestSqlBase extends SqlBase {
 
+  /**
+   * The query to execute.
+   *
+   * @var \Drupal\Core\Database\Query\SelectInterface
+   */
+  protected $query;
+
   /**
    * Overrides the constructor so we can create one easily.
+   *
+   * @param array $configuration
+   *   The plugin instance configuration.
+   * @param \Drupal\migrate\Plugin\MigrationInterface $migration
+   *   (optional) The migration being run.
    */
-  public function __construct() {
-    $this->state = \Drupal::state();
+  public function __construct(array $configuration = [], MigrationInterface $migration = NULL) {
+    parent::__construct($configuration, 'sql_base', [], $migration, \Drupal::state());
   }
 
   /**
@@ -156,6 +234,25 @@ public function fields() {}
   /**
    * {@inheritdoc}
    */
-  public function query() {}
+  public function query() {
+    return $this->query;
+  }
+
+  /**
+   * Sets the query to execute.
+   *
+   * @param \Drupal\Core\Database\Query\SelectInterface $query
+   *   The query to execute.
+   */
+  public function setQuery(SelectInterface $query) {
+    $this->query = $query;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function getHighWaterStorage() {
+    return parent::getHighWaterStorage();
+  }
 
 }