This is an automated email from the ASF dual-hosted git repository.
vldpyatkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 822971b5e06 IGNITE-28957 Calcite SELECT FOR UPDATE may cause OOM by
materializing the entire result set (#13482)
822971b5e06 is described below
commit 822971b5e0687766ffac8d998287b6811e926acb
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Tue Aug 18 10:33:47 2026 +0300
IGNITE-28957 Calcite SELECT FOR UPDATE may cause OOM by materializing the
entire result set (#13482)
---
.../query/calcite/exec/ExecutionServiceImpl.java | 91 ++++++++++++++++------
.../integration/MemoryQuotasIntegrationTest.java | 29 ++++++-
2 files changed, 92 insertions(+), 28 deletions(-)
diff --git
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
index a0b99c47f65..5ab4e94c773 100644
---
a/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
+++
b/modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/ExecutionServiceImpl.java
@@ -82,6 +82,7 @@ import
org.apache.ignite.internal.processors.query.calcite.exec.rel.Node;
import org.apache.ignite.internal.processors.query.calcite.exec.rel.Outbox;
import
org.apache.ignite.internal.processors.query.calcite.exec.task.AbstractQueryTaskExecutor;
import
org.apache.ignite.internal.processors.query.calcite.exec.task.QueryBlockingTaskExecutor;
+import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.ExecutionNodeMemoryTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.GlobalMemoryTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.IoTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.MemoryTracker;
@@ -89,6 +90,7 @@ import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpIoTr
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.NoOpMemoryTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.PerformanceStatisticsIoTracker;
import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.QueryMemoryTracker;
+import
org.apache.ignite.internal.processors.query.calcite.exec.tracker.RowTracker;
import
org.apache.ignite.internal.processors.query.calcite.message.CalciteErrorMessage;
import
org.apache.ignite.internal.processors.query.calcite.message.MessageService;
import
org.apache.ignite.internal.processors.query.calcite.message.QueryStartRequest;
@@ -132,6 +134,7 @@ import
org.apache.ignite.internal.processors.query.calcite.util.ListFieldsQueryC
import org.apache.ignite.internal.processors.query.running.HeavyQueriesTracker;
import org.apache.ignite.internal.processors.security.SecurityUtils;
import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
+import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -629,23 +632,37 @@ public class ExecutionServiceImpl<Row> extends
AbstractService implements Execut
? U.currentTimeMillis() + waitMs
: waitMs < 0 ? U.currentTimeMillis() : 0L;
- RootQuery<Row> selectQry = qry;
+ MemoryTracker forUpdateMemoryTracker =
QueryMemoryTracker.create(memoryTracker, cfg.getQueryMemoryQuota());
- while (true) {
- FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(selectQry,
plan, userTx, waitMs, lockAcquisitionEndTime);
+ try {
+ RootQuery<Row> selectQry = qry;
+
+ while (true) {
+ FieldsQueryCursor<List<?>> cursor = tryExecuteForUpdate(
+ selectQry,
+ plan,
+ userTx,
+ waitMs,
+ lockAcquisitionEndTime,
+ forUpdateMemoryTracker
+ );
- if (cursor != null)
- return cursor;
+ if (cursor != null)
+ return cursor;
- if (lockAcquisitionEndTime != 0 && U.currentTimeMillis() >=
lockAcquisitionEndTime) {
- throw new IgniteSQLException(
- IgniteResource.INSTANCE.selectForUpdateLockFailed().str(),
- IgniteQueryErrorCode.CONCURRENT_UPDATE);
- }
+ if (lockAcquisitionEndTime != 0 && U.currentTimeMillis() >=
lockAcquisitionEndTime) {
+ throw new IgniteSQLException(
+
IgniteResource.INSTANCE.selectForUpdateLockFailed().str(),
+ IgniteQueryErrorCode.CONCURRENT_UPDATE);
+ }
- // The previous query has already been closed after execution, so
retry with a fresh root query.
- selectQry = qry.retryQuery();
- qryReg.register(selectQry);
+ // The previous query has already been closed after execution,
so retry with a fresh root query.
+ selectQry = qry.retryQuery();
+ qryReg.register(selectQry);
+ }
+ }
+ finally {
+ forUpdateMemoryTracker.reset();
}
}
@@ -678,6 +695,7 @@ public class ExecutionServiceImpl<Row> extends
AbstractService implements Execut
* @param userTx Transaction that acquires the locks.
* @param waitMs Lock wait time in the internal representation.
* @param lockAcquisitionEndTime Absolute lock acquisition deadline in
milliseconds.
+ * @param forUpdateMemoryTracker Memory tracker shared by the inner SELECT
and materialized rows.
* @return Result cursor if all required locks were acquired, or {@code
null} if at least one lock was not acquired.
*/
@Nullable private FieldsQueryCursor<List<?>> tryExecuteForUpdate(
@@ -685,26 +703,51 @@ public class ExecutionServiceImpl<Row> extends
AbstractService implements Execut
SelectForUpdatePlan plan,
GridNearTxLocal userTx,
long waitMs,
- long lockAcquisitionEndTime
+ long lockAcquisitionEndTime,
+ MemoryTracker forUpdateMemoryTracker
) {
+ // Use one quota for both execution-node buffers and the rows retained
by SELECT FOR UPDATE.
+ qry.createMemoryTracker(forUpdateMemoryTracker, 0);
+
// Run the inner SELECT (with _KEY, _VAL, _VER appended) and collect
all rows.
ListFieldsQueryCursor<?> innerCursor = mapAndExecutePlan(qry,
plan.innerPlan());
+ List<List<?>> rows = new ArrayList<>();
+ RowTracker<List<?>> rowTracker = ExecutionNodeMemoryTracker.create(
+ forUpdateMemoryTracker,
+ GridUnsafe.OBJ_REF_SIZE
+ );
- // TODO: IGNITE-28957 SELECT FOR UPDATE may cause OOM by materializing
the entire result set.
- List<List<?>> rows = innerCursor.getAll();
+ try {
+ innerCursor.getAll(row -> {
+ rowTracker.onRowAdded(row);
+ rows.add(row);
+ });
- int userColCnt = plan.userColumnCount();
+ int userColCnt = plan.userColumnCount();
- if (rows.isEmpty())
- return createResultCursor(qry, plan, rows, userColCnt);
+ if (rows.isEmpty())
+ return createResultCursor(qry, plan, rows, userColCnt);
- List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches =
- collectLockBatches(plan, rows);
+ List<Map.Entry<IgniteInternalCache<Object, Object>, Map<Object,
CacheEntry<Object, Object>>>> lockBatches =
+ collectLockBatches(plan, rows);
- if (!tryAcquireLocks(userTx, lockBatches, waitMs,
lockAcquisitionEndTime))
- return null;
+ if (!tryAcquireLocks(userTx, lockBatches, waitMs,
lockAcquisitionEndTime))
+ return null;
- return createResultCursor(qry, plan, rows, userColCnt);
+ return createResultCursor(qry, plan, rows, userColCnt);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteSQLException(e.getMessage(),
U.convertException(e));
+ }
+ catch (IgniteSQLException e) {
+ throw e;
+ }
+ catch (Exception e) {
+ throw new IgniteSQLException(e.getMessage(), e);
+ }
+ finally {
+ rowTracker.reset();
+ }
}
/**
diff --git
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
index 31a2ad82f5b..2d905e457d7 100644
---
a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
+++
b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/MemoryQuotasIntegrationTest.java
@@ -22,13 +22,18 @@ import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.SqlConfiguration;
+import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.calcite.QueryChecker;
import org.apache.ignite.internal.processors.query.calcite.hint.HintDefinition;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.transactions.Transaction;
import org.junit.Test;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
/** */
public class MemoryQuotasIntegrationTest extends AbstractBasicIntegrationTest {
/** */
@@ -44,16 +49,19 @@ public class MemoryQuotasIntegrationTest extends
AbstractBasicIntegrationTest {
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
- return super.getConfiguration(igniteInstanceName).setSqlConfiguration(
- new SqlConfiguration().setQueryEnginesConfiguration(new
CalciteQueryEngineConfiguration()
-
.setGlobalMemoryQuota(GLOBAL_MEM_QUOTA).setQueryMemoryQuota(QRY_MEMORY_QUOTA)));
+ return super.getConfiguration(igniteInstanceName)
+ .setSqlConfiguration(new
SqlConfiguration().setQueryEnginesConfiguration(
+ new CalciteQueryEngineConfiguration()
+ .setGlobalMemoryQuota(GLOBAL_MEM_QUOTA)
+ .setQueryMemoryQuota(QRY_MEMORY_QUOTA)))
+ .setTransactionConfiguration(new
TransactionConfiguration().setTxAwareQueriesEnabled(true));
}
/** {@inheritDoc} */
@Override protected void beforeTest() throws Exception {
super.beforeTest();
- sql("CREATE TABLE tbl (id INT, b VARBINARY) WITH TEMPLATE=REPLICATED");
+ sql("CREATE TABLE tbl (id INT, b VARBINARY) WITH TEMPLATE=REPLICATED,
ATOMICITY=TRANSACTIONAL");
for (int i = 0; i < 1000; i++)
sql("INSERT INTO tbl VALUES (?, ?)", i, new byte[1000]);
@@ -460,6 +468,19 @@ public class MemoryQuotasIntegrationTest extends
AbstractBasicIntegrationTest {
.check();
}
+ /** SELECT FOR UPDATE fails when materialized rows exceed the per-query
memory quota. */
+ @Test
+ public void testSelectForUpdateExceedsMemoryQuota() {
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertThrows(grid(0), "SELECT id, b FROM tbl FOR UPDATE",
+ IgniteSQLException.class, "Query quota exceeded");
+ }
+
+ try (Transaction tx = grid(0).transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertEquals(10, sql(grid(0), "SELECT id, b FROM tbl WHERE id < 10
FOR UPDATE").size());
+ }
+ }
+
/** {@inheritDoc} */
@Override protected QueryChecker assertQuery(String qry) {
return super.assertQuery(qry).withRowsIterator(true);
