This is an automated email from the ASF dual-hosted git repository. amashenkov pushed a commit to branch ignite-22303 in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 1c6fae8a3e499eee2d97e6d8261934f1b6e033a4 Author: amashenkov <[email protected]> AuthorDate: Tue Jul 2 15:28:14 2024 +0300 Minor --- .../ignite/internal/sql/engine/exec/ExecutionServiceImpl.java | 4 +--- .../ignite/internal/sql/engine/schema/SqlSchemaManager.java | 10 +++++++++- .../internal/sql/engine/schema/SqlSchemaManagerImpl.java | 7 +++++-- .../internal/sql/engine/framework/PredefinedSchemaManager.java | 5 +++-- 4 files changed, 18 insertions(+), 8 deletions(-) diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java index 7f1dd6b840..dd16b168e4 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java @@ -320,9 +320,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve QueryTransactionWrapper txWrapper = txContext.getOrStartImplicit(plan.type() != SqlQueryType.DML); - if (!sqlSchemaManager.isActualSchemaVersion(plan.catalogVersion(), txWrapper.unwrap().startTimestamp().longValue())) { - throw new ConcurrentSchemaModificationException(); - } + sqlSchemaManager.ensureActualSchemaVersion(plan.catalogVersion(), txWrapper.unwrap().startTimestamp().longValue()); AsyncCursor<InternalSqlRow> dataCursor = queryManager.execute(txWrapper.unwrap(), plan); diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java index b637e8198a..e7a7744f6e 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManager.java @@ -19,6 +19,7 @@ package org.apache.ignite.internal.sql.engine.schema; import java.util.concurrent.CompletableFuture; import org.apache.calcite.schema.SchemaPlus; +import org.apache.ignite.internal.sql.engine.exec.ConcurrentSchemaModificationException; /** * Sql schemas operations interface. @@ -50,5 +51,12 @@ public interface SqlSchemaManager { */ CompletableFuture<Void> schemaReadyFuture(int catalogVersion); - boolean isActualSchemaVersion(int catalogVersion, long timestamp); + /** + * Validates catalog version is an actual version at the given timestamp. + * + * @param catalogVersion Catalog version. + * @param timestamp Timestamp. + * @throws ConcurrentSchemaModificationException If the given catalog version is outdated. + */ + void ensureActualSchemaVersion(int catalogVersion, long timestamp) throws ConcurrentSchemaModificationException; } diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java index 0ef9b27b52..57e22f76a4 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java @@ -50,6 +50,7 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor; import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor; import org.apache.ignite.internal.lang.IgniteInternalException; import org.apache.ignite.internal.schema.DefaultValueGenerator; +import org.apache.ignite.internal.sql.engine.exec.ConcurrentSchemaModificationException; import org.apache.ignite.internal.sql.engine.schema.IgniteIndex.Type; import org.apache.ignite.internal.sql.engine.trait.IgniteDistribution; import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions; @@ -407,7 +408,9 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { } @Override - public boolean isActualSchemaVersion(int catalogVersion, long timestamp) { - return catalogManager.activeCatalogVersion(timestamp) == catalogVersion; + public void ensureActualSchemaVersion(int catalogVersion, long timestamp) throws ConcurrentSchemaModificationException { + if (catalogManager.activeCatalogVersion(timestamp) > catalogVersion) { + throw new ConcurrentSchemaModificationException(); + } } } diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java index dde60cfd5a..d71870c013 100644 --- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/PredefinedSchemaManager.java @@ -28,6 +28,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import org.apache.calcite.schema.SchemaPlus; import org.apache.calcite.tools.Frameworks; +import org.apache.ignite.internal.sql.engine.exec.ConcurrentSchemaModificationException; import org.apache.ignite.internal.sql.engine.schema.IgniteSchema; import org.apache.ignite.internal.sql.engine.schema.IgniteTable; import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager; @@ -97,7 +98,7 @@ public class PredefinedSchemaManager implements SqlSchemaManager { } @Override - public boolean isActualSchemaVersion(int catalogVersion, long timestamp) { - return true; + public void ensureActualSchemaVersion(int catalogVersion, long timestamp) throws ConcurrentSchemaModificationException { + // No-op. } }
