Skip to content

Commit 1e66878

Browse files
vldpyatkovvldpyatkov
committed
Fixed issue with technical columns access in SQL validation and scan operations.
1 parent 8841db1 commit 1e66878

2 files changed

Lines changed: 128 additions & 5 deletions

File tree

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/schema/CacheTableDescriptorImpl.java‎

Lines changed: 29 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,9 @@ public class CacheTableDescriptorImpl extends NullInitializerExpressionFactory
112112
/** */
113113
private final ImmutableBitSet insertFields;
114114

115+
/** Count of non-technical columns used in the UPDATE source SELECT. */
116+
private final int updateRowFieldCnt;
117+
115118
/** */
116119
private RelDataType tableRowType;
117120

@@ -235,6 +238,9 @@ else if (!F.isEmpty(typeDesc.primaryKeyFields())) {
235238
this.affFields = ImmutableIntList.copyOf(affFields);
236239
this.descriptors = descriptors.toArray(DUMMY);
237240
this.descriptorsMap = descriptorsMap;
241+
this.updateRowFieldCnt = (int)descriptors.stream()
242+
.filter(d -> !TechnicalColumns.isTechnicalFieldNameIgnoreCase(d.name()))
243+
.count();
238244

239245
virtualFields.flip(0, descriptors.size());
240246
insertFields = ImmutableBitSet.fromBitSet(virtualFields);
@@ -245,6 +251,18 @@ else if (!F.isEmpty(typeDesc.primaryKeyFields())) {
245251
return rowType(factory, insertFields);
246252
}
247253

254+
/** {@inheritDoc} */
255+
@Override public RelDataType selectForUpdateRowType(IgniteTypeFactory factory) {
256+
RelDataTypeFactory.Builder b = new RelDataTypeFactory.Builder(factory);
257+
258+
for (CacheColumnDescriptor desc : descriptors) {
259+
if (!TechnicalColumns.isTechnicalFieldNameIgnoreCase(desc.name()))
260+
b.add(desc.name(), desc.logicalType(factory));
261+
}
262+
263+
return b.build();
264+
}
265+
248266
/** {@inheritDoc} */
249267
@Override public GridCacheContext cacheContext() {
250268
return cacheInfo.cacheContext();
@@ -443,7 +461,8 @@ private <Row> ModifyTuple updateTuple(Row row, List<String> updateColList, int o
443461
Object key = Objects.requireNonNull(hnd.get(offset + QueryUtils.KEY_COL, row));
444462
Object val = clone(Objects.requireNonNull(hnd.get(offset + QueryUtils.VAL_COL, row)));
445463

446-
offset += descriptorsMap.size();
464+
// New values start after the source fields; compute dynamically to handle both UPDATE and MERGE.
465+
offset = hnd.columnCount(row) - updateColList.size();
447466

448467
for (int i = 0; i < updateColList.size(); i++) {
449468
final CacheColumnDescriptor desc = Objects.requireNonNull(descriptorsMap.get(updateColList.get(i)));
@@ -473,14 +492,19 @@ private <Row> ModifyTuple mergeTuple(Row row, List<String> updateColList, Execut
473492

474493
int rowColumnsCnt = hnd.columnCount(row);
475494

476-
if (rowColumnsCnt == descriptors.length)
495+
// An empty update column list unambiguously means there is no WHEN MATCHED clause at all (a MERGE
496+
// statement always has at least one WHEN clause), so the row can only originate from the INSERT
497+
// section. Note: the row width alone can't be used to detect this case, since, depending on the
498+
// number of updated columns, it may coincide with the width of a WHEN MATCHED-only row.
499+
if (updateColList.isEmpty())
477500
return insertTuple(row, ectx); // Only WHEN NOT MATCHED clause in MERGE.
478-
else if (rowColumnsCnt == descriptors.length + updateColList.size())
501+
else if (rowColumnsCnt == updateRowFieldCnt + updateColList.size())
479502
return updateTuple(row, updateColList, 0, ectx); // Only WHEN MATCHED clause in MERGE.
480503
else {
481504
// Both WHEN MATCHED and WHEN NOT MATCHED clauses in MERGE.
482-
assert rowColumnsCnt == descriptors.length * 2 + updateColList.size() : "Unexpected columns count: " +
483-
rowColumnsCnt;
505+
// INSERT section has all fields (insertRowType); UPDATE section excludes technical columns.
506+
assert rowColumnsCnt == descriptors.length + updateRowFieldCnt + updateColList.size() :
507+
"Unexpected columns count: " + rowColumnsCnt;
484508

485509
int updateOffset = descriptors.length; // Offset of fields for update statement.
486510

‎modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/TechnicalColumnsScanTest.java‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,14 +23,20 @@
2323
import java.util.List;
2424
import java.util.Set;
2525
import java.util.UUID;
26+
import java.util.concurrent.Callable;
2627
import org.apache.calcite.util.ImmutableBitSet;
28+
import org.apache.ignite.IgniteCheckedException;
29+
import org.apache.ignite.cache.CacheEntry;
2730
import org.apache.ignite.cache.query.SqlFieldsQuery;
2831
import org.apache.ignite.calcite.CalciteQueryEngineConfiguration;
2932
import org.apache.ignite.configuration.IgniteConfiguration;
3033
import org.apache.ignite.configuration.SqlConfiguration;
3134
import org.apache.ignite.configuration.TransactionConfiguration;
3235
import org.apache.ignite.internal.IgniteEx;
36+
import org.apache.ignite.internal.IgniteInternalFuture;
3337
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
38+
import org.apache.ignite.internal.processors.cache.CacheEntryImplEx;
39+
import org.apache.ignite.internal.processors.cache.IgniteCacheProxy;
3440
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
3541
import org.apache.ignite.internal.processors.query.IgniteSQLException;
3642
import org.apache.ignite.internal.processors.query.QueryUtils;
@@ -49,6 +55,7 @@
4955
import org.apache.ignite.internal.processors.query.calcite.schema.IgniteIndex;
5056
import org.apache.ignite.internal.processors.query.calcite.schema.TechnicalColumns;
5157
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
58+
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
5259
import org.apache.ignite.testframework.GridTestUtils;
5360
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
5461
import org.apache.ignite.transactions.Transaction;
@@ -95,6 +102,72 @@ public void testTableScanReturnsTechnicalColumns() throws Exception {
95102
assertTechnicalColumns(tbl.scan(scanCtx.ectx, scanCtx.grp, requiredColumns(tbl)), tbl);
96103
}
97104

105+
/** */
106+
@Test
107+
@SuppressWarnings("unchecked")
108+
public void testScannedTechnicalColumnsCanLockTxEntries() throws Exception {
109+
createAndPopulatePersonTable();
110+
111+
IgniteCacheTable tbl = personTable();
112+
ScanContext scanCtx = scanContext(tbl);
113+
List<Object[]> rows = materialize(tbl.scan(scanCtx.ectx, scanCtx.grp, lockRequiredColumns(tbl)));
114+
List<CacheEntry<Object, Object>> entries = new ArrayList<>();
115+
Integer expSrc = tbl.descriptor().cacheInfo().cacheId();
116+
117+
assertEquals(30, rows.size());
118+
119+
for (Object[] row : rows) {
120+
assertEquals(4, row.length);
121+
assertTrue("Unexpected _VER value [val=" + row[2] + ", cls=" +
122+
(row[2] == null ? null : row[2].getClass()) + ']', row[2] instanceof GridCacheVersion);
123+
assertEquals(expSrc, row[3]);
124+
125+
entries.add(new CacheEntryImplEx<>(row[0], row[1], (GridCacheVersion)row[2]));
126+
}
127+
128+
try (Transaction tx = node.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
129+
assertTrue(node.cache(tbl.descriptor().cacheInfo().name()).unwrap(IgniteCacheProxy.class)
130+
.internalProxy().lockTxEntries(entries, 5_000));
131+
132+
checkInaccessInOtherTx();
133+
134+
sql("UPDATE Person SET age = 42 WHERE id = 2");
135+
136+
tx.commit();
137+
}
138+
139+
List<List<?>> rowsAfterUpdate = sql("SELECT id, name, age FROM Person WHERE id = 2");
140+
141+
assertEquals(1, rowsAfterUpdate.size());
142+
assertEquals(personName(2), rowsAfterUpdate.get(0).get(1));
143+
assertEquals(42, rowsAfterUpdate.get(0).get(2));
144+
}
145+
146+
/**
147+
* Checks that another transaction cannot access the cache.
148+
*
149+
* @throws IgniteCheckedException If failed.
150+
*/
151+
private void checkInaccessInOtherTx() throws IgniteCheckedException {
152+
IgniteInternalFuture<Void> accessFut = GridTestUtils.runAsync(new Callable<Void>() {
153+
@Override public Void call() {
154+
try (Transaction tx = node.transactions().txStart(PESSIMISTIC, READ_COMMITTED, 500, 1)) {
155+
sql("UPDATE Person SET name = 'Charley' WHERE id = 2");
156+
157+
tx.commit();
158+
}
159+
160+
return null;
161+
}
162+
});
163+
164+
GridTestUtils.assertThrowsWithCause(new Callable<Object>() {
165+
@Override public Object call() throws Exception {
166+
return accessFut.get(10_000);
167+
}
168+
}, IgniteTxTimeoutCheckedException.class);
169+
}
170+
98171
/** */
99172
@Test
100173
public void testIndexScanReturnsTechnicalColumns() throws Exception {
@@ -132,6 +205,22 @@ public void testTechnicalColumnsAreHiddenFromSql() throws Exception {
132205
assertTechnicalColumnAccessForbidden("SELECT CASE WHEN _ver IS NOT NULL THEN 1 ELSE 0 END FROM Person");
133206
assertTechnicalColumnAccessForbidden("SELECT CAST(_ver AS VARCHAR) FROM Person");
134207
assertTechnicalColumnAccessForbidden("SELECT id FROM Person WHERE (SELECT _ver FROM Person WHERE id = 1) IS NOT NULL");
208+
209+
// MERGE: technical columns must be forbidden in all clause positions.
210+
assertTechnicalColumnAccessForbidden(
211+
"MERGE INTO Person " +
212+
"USING (SELECT id, _ver FROM Person) AS src ON (Person.id = src.id) " +
213+
"WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (src.id, 'x', 1)");
214+
215+
assertTechnicalColumnAccessForbidden(
216+
"MERGE INTO Person " +
217+
"USING (SELECT 100 AS id) AS src ON (Person._ver IS NOT NULL AND Person.id = src.id) " +
218+
"WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (src.id, 'x', 1)");
219+
220+
assertTechnicalColumnAccessForbidden(
221+
"MERGE INTO Person " +
222+
"USING (SELECT 1 AS id) AS src ON (Person.id = src.id) " +
223+
"WHEN MATCHED THEN UPDATE SET name = CAST(Person._ver AS VARCHAR)");
135224
}
136225

137226
/** */
@@ -202,6 +291,16 @@ private ImmutableBitSet requiredColumns(IgniteCacheTable tbl) {
202291
);
203292
}
204293

294+
/** */
295+
private ImmutableBitSet lockRequiredColumns(IgniteCacheTable tbl) {
296+
return ImmutableBitSet.of(
297+
columnIndex(tbl, QueryUtils.KEY_FIELD_NAME),
298+
columnIndex(tbl, QueryUtils.VAL_FIELD_NAME),
299+
columnIndex(tbl, TechnicalColumns.VER_FIELD_NAME),
300+
columnIndex(tbl, TechnicalColumns.SRC_FIELD_NAME)
301+
);
302+
}
303+
205304
/** */
206305
private int columnIndex(IgniteCacheTable tbl, String name) {
207306
ColumnDescriptor desc = tbl.descriptor().columnDescriptor(name);

0 commit comments

Comments
 (0)