IGNITE-21084 Round up timestamp-to-wait-after-DDL to the closest mill…

…isecond (#2954)

modules/catalog/src/main/java/org/apache/ignite/internal/catalog/CatalogManagerImpl.java

@@ -330,9 +330,10 @@ private CompletableFuture<Void> saveUpdateAndWaitForActivation(UpdateProducer up
                     Catalog catalog = catalogByVer.get(newVersion);
 
                     HybridTimestamp activationTs = HybridTimestamp.hybridTimestamp(catalog.time());
-                    HybridTimestamp clusterWideEnsuredActivationTs = activationTs.addPhysicalTime(
-                            HybridTimestamp.maxClockSkew()
-                    );
+                    HybridTimestamp clusterWideEnsuredActivationTs = activationTs.addPhysicalTime(HybridTimestamp.maxClockSkew())
+                            // Rounding up to the closest millisecond to account for possibility of HLC.now() having different
+                            // logical parts on different nodes of the cluster (see IGNITE-21084).
+                            .roundUpToPhysicalTick();
                     // TODO: this addition has to be removed when IGNITE-20378 is implemented.
                     HybridTimestamp tsSafeForRoReadingInPastOptimization = clusterWideEnsuredActivationTs.addPhysicalTime(
                             partitionIdleSafeTimePropagationPeriodMsSupplier.getAsLong()

modules/core/src/main/java/org/apache/ignite/internal/hlc/HybridTimestamp.java

@@ -267,4 +267,17 @@ public HybridTimestamp subtractPhysicalTime(long millis) {
     public static long maxClockSkew() {
         return CLOCK_SKEW;
     }
+
+    /**
+     * Returns a result of rounding this timestamp up 'to its physical part': that is, if the logical part is zero, the timestamp is
+     * returned as is; if it's non-zero, a new timestamp is returned that has physical part equal to the physical part of this
+     * timestamp plus one, and the logical part is zero.
+     */
+    public HybridTimestamp roundUpToPhysicalTick() {
+        if (getLogical() == 0) {
+            return this;
+        } else {
+            return new HybridTimestamp(getPhysical() + 1, 0);
+        }
+    }
 }

modules/core/src/test/java/org/apache/ignite/internal/hlc/HybridTimestampTest.java

@@ -144,4 +144,18 @@ void subtractPhysicalTimeThrowsIfUnderflowHappens() {
         );
         assertThat(ex.getMessage(), startsWith("Time is out of bounds: "));
     }
+
+    @Test
+    void roundUpLeavesTsUnchangedWhenLogicalPartIsZero() {
+        var ts = new HybridTimestamp(1, 0);
+
+        assertThat(ts.roundUpToPhysicalTick(), is(ts));
+    }
+
+    @Test
+    void roundUpIncrementsPhysicalPartWhenLogicalPartIsNonZero() {
+        var ts = new HybridTimestamp(1, 1);
+
+        assertThat(ts.roundUpToPhysicalTick(), is(new HybridTimestamp(2, 0)));
+    }
 }

modules/runner/src/testFixtures/java/org/apache/ignite/internal/ClusterPerClassIntegrationTest.java

@@ -277,6 +277,10 @@ protected static List<List<Object>> sql(String sql, Object... args) {
         return sql(null, sql, args);
     }
 
+    protected static List<List<Object>> sql(int nodeIndex, String sql, Object... args) {
+        return sql(nodeIndex, null, sql, args);
+    }
+
     /**
      * Run SQL on given Ignite instance with given transaction and parameters.
      *
@@ -296,7 +300,12 @@ public static List<List<Object>> sql(Ignite node, @Nullable Transaction tx, Stri
     }
 
     protected static List<List<Object>> sql(@Nullable Transaction tx, String sql, Object... args) {
-        IgniteImpl node = CLUSTER.node(0);
+        return sql(0, tx, sql, args);
+    }
+
+    protected static List<List<Object>> sql(int nodeIndex, @Nullable Transaction tx, String sql, Object[] args) {
+        IgniteImpl node = CLUSTER.node(nodeIndex);
+
         if (!AWAIT_INDEX_AVAILABILITY.get()) {
             return sql(node, tx, sql, args);
         } else {

modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItMixedQueriesTest.java

@@ -26,9 +26,12 @@
 
 import java.util.Arrays;
 import java.util.List;
+import org.apache.ignite.internal.app.IgniteImpl;
 import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 /**
  * Group of tests that still has not been sorted out. It’s better to avoid extending this class with new tests.
@@ -207,32 +210,35 @@ public void testNotExistsConditionWithSubquery() {
     /**
      * Verifies that table modification events are passed to a calcite schema modification listener.
      */
-    @Test
-    public void testIgniteSchemaAwaresAlterTableCommand() {
+    @ParameterizedTest
+    @ValueSource(ints = {0, 1})
+    public void testIgniteSchemaAwaresAlterTableCommand(int nodeToExecuteSelectsIndex) {
+        IgniteImpl nodeToExecuteSelects = CLUSTER.node(nodeToExecuteSelectsIndex);
+
         String selectAllQry = "select * from test_tbl";
 
-        sql("drop table if exists test_tbl");
-        sql("create table test_tbl(id int primary key, val varchar)");
+        sql(0, "drop table if exists test_tbl");
+        sql(0, "create table test_tbl(id int primary key, val varchar)");
 
-        assertQuery(selectAllQry).columnNames("ID", "VAL").check();
+        assertQuery(nodeToExecuteSelects, selectAllQry).columnNames("ID", "VAL").check();
 
-        sql("alter table test_tbl add column new_col int");
+        sql(0, "alter table test_tbl add column new_col int");
 
-        assertQuery(selectAllQry).columnNames("ID", "VAL", "NEW_COL").check();
+        assertQuery(nodeToExecuteSelects, selectAllQry).columnNames("ID", "VAL", "NEW_COL").check();
 
         // column with such name already exists
-        assertThrows(Exception.class, () -> sql("alter table test_tbl add column new_col int"));
+        assertThrows(Exception.class, () -> sql(0, "alter table test_tbl add column new_col int"));
 
-        assertQuery(selectAllQry).columnNames("ID", "VAL", "NEW_COL").check();
+        assertQuery(nodeToExecuteSelects, selectAllQry).columnNames("ID", "VAL", "NEW_COL").check();
 
-        sql("alter table test_tbl drop column new_col");
+        sql(0, "alter table test_tbl drop column new_col");
 
-        assertQuery(selectAllQry).columnNames("ID", "VAL").check();
+        assertQuery(nodeToExecuteSelects, selectAllQry).columnNames("ID", "VAL").check();
 
         // column with such name is not exists
-        assertThrows(Exception.class, () -> sql("alter table test_tbl drop column new_col"));
+        assertThrows(Exception.class, () -> sql(0, "alter table test_tbl drop column new_col"));
 
-        assertQuery(selectAllQry).columnNames("ID", "VAL").check();
+        assertQuery(nodeToExecuteSelects, selectAllQry).columnNames("ID", "VAL").check();
     }
 
     /** Quantified predicates test. */

modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/BaseSqlIntegrationTest.java

@@ -38,6 +38,7 @@
 import org.apache.ignite.sql.IgniteSql;
 import org.apache.ignite.table.Table;
 import org.apache.ignite.tx.IgniteTransactions;
+import org.jetbrains.annotations.Nullable;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 /**
@@ -55,7 +56,11 @@ public class BaseSqlIntegrationTest extends ClusterPerClassIntegrationTest {
      * @return Instance of QueryChecker.
      */
     protected static QueryChecker assertQuery(String qry) {
-        return assertQuery(null, qry);
+        return assertQuery((InternalTransaction) null, qry);
+    }
+
+    protected static QueryChecker assertQuery(IgniteImpl node, String qry) {
+        return assertQuery(node, null, qry);
     }
 
     /**
@@ -65,9 +70,11 @@ protected static QueryChecker assertQuery(String qry) {
      * @param qry Query to execute.
      * @return Instance of QueryChecker.
      */
-    protected static QueryChecker assertQuery(InternalTransaction tx, String qry) {
-        IgniteImpl node = CLUSTER.aliveNode();
+    protected static QueryChecker assertQuery(@Nullable InternalTransaction tx, String qry) {
+        return assertQuery(CLUSTER.aliveNode(), tx, qry);
+    }
 
+    private static QueryChecker assertQuery(IgniteImpl node, @Nullable InternalTransaction tx, String qry) {
         return queryCheckerFactory.create(node.queryEngine(), node.transactions(), tx, qry);
     }