Skip to content

Commit b516462

Browse files
alex-plekhanovalex-plekhanov
committed
IGNITE-27431 SQL Calcite: Optimize scans with filter
1 parent 3534e59 commit b516462

File tree

14 files changed

+391
-104
lines changed

14 files changed

+391
-104
lines changed

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractCacheColumnsScan.java‎

Lines changed: 27 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,16 @@
1717

1818
package org.apache.ignite.internal.processors.query.calcite.exec;
1919

20+
import java.util.Iterator;
2021
import org.apache.calcite.rel.type.RelDataType;
2122
import org.apache.calcite.util.ImmutableBitSet;
2223
import org.apache.ignite.internal.processors.query.calcite.exec.RowHandler.RowFactory;
2324
import org.apache.ignite.internal.processors.query.calcite.schema.CacheTableDescriptor;
2425
import org.jetbrains.annotations.Nullable;
2526

2627
/** */
27-
public abstract class AbstractCacheColumnsScan<Row> extends AbstractCacheScan<Row> {
28+
public abstract class AbstractCacheColumnsScan<TableRow, Row> extends AbstractCacheScan<Row>
29+
implements TableRowIterable<TableRow, Row> {
2830
/** */
2931
protected final CacheTableDescriptor desc;
3032

@@ -34,8 +36,8 @@ public abstract class AbstractCacheColumnsScan<Row> extends AbstractCacheScan<Ro
3436
/** */
3537
protected final RelDataType rowType;
3638

37-
/** Participating columns. */
38-
protected final ImmutableBitSet requiredColumns;
39+
/** Row field to column mapping. */
40+
protected final int[] fieldColMapping;
3941

4042
/** */
4143
AbstractCacheColumnsScan(
@@ -47,9 +49,30 @@ public abstract class AbstractCacheColumnsScan<Row> extends AbstractCacheScan<Ro
4749
super(ectx, desc.cacheContext(), parts);
4850

4951
this.desc = desc;
50-
this.requiredColumns = requiredColumns;
5152

5253
rowType = desc.rowType(ectx.getTypeFactory(), requiredColumns);
5354
factory = ectx.rowHandler().factory(ectx.getTypeFactory(), rowType);
55+
56+
ImmutableBitSet reqCols = requiredColumns == null ? ImmutableBitSet.range(0, rowType.getFieldCount())
57+
: requiredColumns;
58+
59+
fieldColMapping = reqCols.toArray();
5460
}
61+
62+
/** {@inheritDoc} */
63+
@Override public final Iterator<TableRow> tableRowIterator() {
64+
reserve();
65+
66+
try {
67+
return createTableRowIterator();
68+
}
69+
catch (Exception e) {
70+
release();
71+
72+
throw e;
73+
}
74+
}
75+
76+
/** Table row iterator.*/
77+
protected abstract Iterator<TableRow> createTableRowIterator();
5578
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/AbstractCacheScan.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,7 @@ public abstract class AbstractCacheScan<Row> implements Iterable<Row>, AutoClose
100100
}
101101

102102
/** */
103-
private synchronized void reserve() {
103+
protected synchronized void reserve() {
104104
if (reservation != null)
105105
return;
106106

@@ -151,7 +151,7 @@ private synchronized void reserve() {
151151
}
152152

153153
/** */
154-
private synchronized void release() {
154+
protected synchronized void release() {
155155
if (reservation != null)
156156
reservation.release();
157157

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/IndexScan.java‎

Lines changed: 36 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@
5656
/**
5757
* Scan on index.
5858
*/
59-
public class IndexScan<Row> extends AbstractCacheColumnsScan<Row> {
59+
public class IndexScan<Row> extends AbstractCacheColumnsScan<IndexRow, Row> {
6060
/** */
6161
private final GridKernalContext kctx;
6262

@@ -109,7 +109,7 @@ public IndexScan(
109109
for (int i = 0; i < srcRowType.getFieldCount(); i++)
110110
fieldsStoreTypes[i] = typeFactory.getResultClass(srcRowType.getFieldList().get(i).getType());
111111

112-
fieldIdxMapping = fieldToInlinedKeysMapping(srcRowType.getFieldCount());
112+
fieldIdxMapping = fieldToInlinedKeysMapping();
113113

114114
if (!F.isEmpty(ectx.getQryTxEntries())) {
115115
InlineIndexRowHandler rowHnd = idx.segment(0).rowHandler();
@@ -131,7 +131,7 @@ public IndexScan(
131131
* @return Mapping from target row fields to inlined index keys, or {@code null} if inlined index keys
132132
* should not be used.
133133
*/
134-
private int[] fieldToInlinedKeysMapping(int srcFieldsCnt) {
134+
private int[] fieldToInlinedKeysMapping() {
135135
List<InlineIndexKeyType> inlinedKeys = idx.segment(0).rowHandler().inlineIndexKeyTypes();
136136

137137
// Since inline scan doesn't check expire time, allow it only if expired entries are eagerly removed.
@@ -141,7 +141,7 @@ private int[] fieldToInlinedKeysMapping(int srcFieldsCnt) {
141141
// Even if we need some subset of inlined keys we are required to the read full inlined row, since this row
142142
// is also participated in comparison with other rows when cursor processing the next index page.
143143
if (inlinedKeys.size() < idx.segment(0).rowHandler().indexKeyDefinitions().size() ||
144-
inlinedKeys.size() < (requiredColumns == null ? srcFieldsCnt : requiredColumns.cardinality()))
144+
inlinedKeys.size() < fieldColMapping.length)
145145
return null;
146146

147147
for (InlineIndexKeyType keyType : inlinedKeys) {
@@ -153,14 +153,11 @@ private int[] fieldToInlinedKeysMapping(int srcFieldsCnt) {
153153
return null;
154154
}
155155

156-
ImmutableBitSet reqCols = requiredColumns == null ? ImmutableBitSet.range(0, srcFieldsCnt) :
157-
requiredColumns;
158-
159156
int[] fieldIdxMapping = new int[rowType.getFieldCount()];
160157

161-
for (int i = 0, j = reqCols.nextSetBit(0); j != -1; j = reqCols.nextSetBit(j + 1), i++) {
158+
for (int i = 0; i < fieldColMapping.length; i++) {
162159
// j = source field index, i = target field index.
163-
int keyIdx = idxFieldMapping.indexOf(j);
160+
int keyIdx = idxFieldMapping.indexOf(fieldColMapping[i]);
164161

165162
if (keyIdx >= 0 && keyIdx < inlinedKeys.size())
166163
fieldIdxMapping[i] = keyIdx;
@@ -173,14 +170,32 @@ private int[] fieldToInlinedKeysMapping(int srcFieldsCnt) {
173170

174171
/** {@inheritDoc} */
175172
@Override protected Iterator<Row> createIterator() {
173+
return F.iterator(createTableRowIterator(), this::indexRow2Row, true);
174+
}
175+
176+
/** {@inheritDoc} */
177+
@Override protected Iterator<IndexRow> createTableRowIterator() {
176178
RangeIterable<IndexRow> ranges0 = ranges == null ? null : new TransformRangeIterable<>(ranges, this::row2indexRow);
177179

178180
TreeIndex<IndexRow> treeIdx = treeIndex();
179181

180182
if (!txChanges.changedKeysEmpty())
181183
treeIdx = new TxAwareTreeIndexWrapper(treeIdx);
182184

183-
return F.iterator(new TreeIndexIterable<>(treeIdx, ranges0), this::indexRow2Row, true);
185+
return new TreeIndexIterable<>(treeIdx, ranges0).iterator();
186+
}
187+
188+
/** {@inheritDoc} */
189+
@Override public Row enrichRow(IndexRow idxRow, Row row, int[] fieldColMapping) {
190+
try {
191+
if (idxRow.indexPlainRow())
192+
return inlineIndexRow2Row(idxRow, row, fieldColMapping);
193+
else
194+
return desc.toRow(ectx, idxRow.cacheDataRow(), row, fieldColMapping);
195+
}
196+
catch (IgniteCheckedException e) {
197+
throw new IgniteException(e);
198+
}
184199
}
185200

186201
/** */
@@ -221,28 +236,23 @@ protected IndexRow row2indexRow(Row bound) {
221236
}
222237

223238
/** From IndexRow to Row convertor. */
224-
protected Row indexRow2Row(IndexRow row) {
225-
try {
226-
if (row.indexPlainRow())
227-
return inlineIndexRow2Row(row);
228-
else
229-
return desc.toRow(ectx, row.cacheDataRow(), factory, requiredColumns);
230-
}
231-
catch (IgniteCheckedException e) {
232-
throw new IgniteException(e);
233-
}
239+
protected Row indexRow2Row(IndexRow idxRow) {
240+
Row row = factory.create();
241+
242+
return enrichRow(idxRow, row, fieldColMapping);
234243
}
235244

236245
/** */
237-
private Row inlineIndexRow2Row(IndexRow row) {
246+
private Row inlineIndexRow2Row(IndexRow idxRow, Row row, int[] fieldColMapping) {
238247
RowHandler<Row> hnd = ectx.rowHandler();
239248

240-
Row res = factory.create();
241-
242-
for (int i = 0; i < fieldIdxMapping.length; i++)
243-
hnd.set(i, res, TypeUtils.toInternal(ectx, row.key(fieldIdxMapping[i]).key()));
249+
for (int i = 0; i < fieldColMapping.length; i++) {
250+
// Skip not required fields.
251+
if (fieldColMapping[i] >= 0)
252+
hnd.set(i, row, TypeUtils.toInternal(ectx, idxRow.key(fieldIdxMapping[i]).key()));
253+
}
244254

245-
return res;
255+
return row;
246256
}
247257

248258
/** Query context. */

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/LogicalRelImplementor.java‎

Lines changed: 49 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ProjectNode;
6565
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanNode;
6666
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanStorageNode;
67+
import org.apache.ignite.internal.processors.query.calcite.exec.rel.ScanTableRowNode;
6768
import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortAggregateNode;
6869
import org.apache.ignite.internal.processors.query.calcite.exec.rel.SortNode;
6970
import org.apache.ignite.internal.processors.query.calcite.exec.rel.TableSpoolNode;
@@ -116,6 +117,7 @@
116117
import org.apache.ignite.internal.processors.query.calcite.util.Commons;
117118
import org.apache.ignite.internal.processors.query.calcite.util.RexUtils;
118119
import org.apache.ignite.internal.util.typedef.F;
120+
import org.jetbrains.annotations.Nullable;
119121

120122
import static org.apache.calcite.rel.RelDistribution.Type.HASH_DISTRIBUTED;
121123
import static org.apache.ignite.internal.processors.query.calcite.util.TypeUtils.combinedRowType;
@@ -382,7 +384,8 @@ private boolean hasExchange(RelNode rel) {
382384
if (idx != null && !tbl.isIndexRebuildInProgress()) {
383385
Iterable<Row> rowsIter = idx.scan(ctx, grp, ranges, requiredColumns);
384386

385-
return new ScanStorageNode<>(tbl.name() + '.' + idx.name(), ctx, rowType, rowsIter, filters, prj);
387+
return createStorageScan(tbl.name() + '.' + idx.name(), rowType, rowsIter, filters, prj,
388+
requiredColumns, rel.conditionColumns());
386389
}
387390
else {
388391
// Index was invalidated after planning, workaround through table-scan -> sort -> index spool.
@@ -406,8 +409,8 @@ private boolean hasExchange(RelNode rel) {
406409
if (!spoolNodeRequired && projects != null)
407410
rowType = rel.getRowType();
408411

409-
Node<Row> node = new ScanStorageNode<>(tbl.name(), ctx, rowType, rowsIter, filterHasCorrelation ? null : filters,
410-
projNodeRequired ? null : prj);
412+
Node<Row> node = createStorageScan(tbl.name(), rowType, rowsIter, filterHasCorrelation ? null : filters,
413+
projNodeRequired ? null : prj, requiredColumns, rel.conditionColumns());
411414

412415
RelCollation collation = rel.collation();
413416

@@ -560,12 +563,13 @@ private boolean hasExchange(RelNode rel) {
560563
if (idx != null && !tbl.isIndexRebuildInProgress()) {
561564
Iterable<Row> rowsIter = idx.scan(ctx, grp, null, requiredColumns);
562565

563-
return new ScanStorageNode<>(tbl.name() + '.' + idx.name(), ctx, rowType, rowsIter, filters, prj);
566+
return createStorageScan(tbl.name() + '.' + idx.name(), rowType, rowsIter, filters, prj,
567+
requiredColumns, rel.conditionColumns());
564568
}
565569
else {
566570
Iterable<Row> rowsIter = tbl.scan(ctx, grp, requiredColumns);
567571

568-
return new ScanStorageNode<>(tbl.name(), ctx, rowType, rowsIter, filters, prj);
572+
return createStorageScan(tbl.name(), rowType, rowsIter, filters, prj, requiredColumns, rel.conditionColumns());
569573
}
570574
}
571575

@@ -943,4 +947,44 @@ private Node<Row> visit(RelNode rel) {
943947
public <T extends Node<Row>> T go(IgniteRel rel) {
944948
return (T)visit(rel);
945949
}
950+
951+
/** */
952+
private ScanStorageNode<Row> createStorageScan(
953+
String storageName,
954+
RelDataType rowType,
955+
Iterable<Row> rowsIter,
956+
@Nullable Predicate<Row> filter,
957+
@Nullable Function<Row, Row> rowTransformer,
958+
@Nullable ImmutableBitSet requiredColumns,
959+
@Nullable ImmutableBitSet filterColumns
960+
) {
961+
int fieldsCnt = rowType.getFieldCount();
962+
963+
if (filter == null || filterColumns == null || filterColumns.cardinality() == fieldsCnt
964+
|| !(rowsIter instanceof TableRowIterable))
965+
return new ScanStorageNode<>(storageName, ctx, rowType, rowsIter, filter, rowTransformer);
966+
967+
ImmutableBitSet reqCols = requiredColumns == null ? ImmutableBitSet.range(0, fieldsCnt) : requiredColumns;
968+
969+
int[] filterColMapping = reqCols.toArray();
970+
int[] otherColMapping = filterColMapping.clone();
971+
972+
for (int i = 0; i < filterColMapping.length; i++) {
973+
if (filterColumns.get(i))
974+
otherColMapping[i] = -1;
975+
else
976+
filterColMapping[i] = -1;
977+
}
978+
979+
return new ScanTableRowNode<>(
980+
storageName,
981+
ctx,
982+
rowType,
983+
(TableRowIterable<Object, Row>)rowsIter,
984+
filter,
985+
rowTransformer,
986+
filterColMapping,
987+
otherColMapping
988+
);
989+
}
946990
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/SystemViewScan.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,8 @@ public class SystemViewScan<Row, ViewRow> implements Iterable<Row> {
4747
/** */
4848
private final RangeIterable<Row> ranges;
4949

50-
/** Participating colunms. */
51-
private final ImmutableBitSet requiredColumns;
50+
/** Row field to view column mapping. */
51+
protected final int[] fieldColMapping;
5252

5353
/** System view field names (for filtering). */
5454
private final String[] filterableFieldNames;
@@ -66,7 +66,6 @@ public SystemViewScan(
6666
this.ectx = ectx;
6767
this.desc = desc;
6868
this.ranges = ranges;
69-
this.requiredColumns = requiredColumns;
7069

7170
RelDataType rowType = desc.rowType(ectx.getTypeFactory(), requiredColumns);
7271

@@ -83,6 +82,11 @@ public SystemViewScan(
8382
}
8483
}
8584
}
85+
86+
ImmutableBitSet reqCols = requiredColumns == null ? ImmutableBitSet.range(0, rowType.getFieldCount())
87+
: requiredColumns;
88+
89+
fieldColMapping = reqCols.toArray();
8690
}
8791

8892
/** {@inheritDoc} */
@@ -123,6 +127,6 @@ public SystemViewScan(
123127
else
124128
viewIter = view.iterator();
125129

126-
return F.iterator(viewIter, row -> desc.toRow(ectx, row, factory, requiredColumns), true);
130+
return F.iterator(viewIter, row -> desc.toRow(ectx, row, factory.create(), fieldColMapping), true);
127131
}
128132
}

‎modules/calcite/src/main/java/org/apache/ignite/internal/processors/query/calcite/exec/TableRowIterable.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.ignite.internal.processors.query.calcite.exec;
18+
19+
import java.util.Iterator;
20+
21+
/**
22+
* Interface to iterate over raw table data and convert this data to relational node rows.
23+
*
24+
* @param <TableRow>> Raw table row type.
25+
* @param <Row>> Relational node row type.
26+
*/
27+
public interface TableRowIterable<TableRow, Row> extends Iterable<Row> {
28+
/**
29+
* @return Table row iterator.
30+
* */
31+
public Iterator<TableRow> tableRowIterator();
32+
33+
/**
34+
* Enriches {@code nodeRow} with columns from {@code tableRow} *
35+
*
36+
* @param tableRow Table row.
37+
* @param nodeRow Relational node row.
38+
* @param fieldColMapping Mapping from node row fields to table row columns. If column is not requried
39+
* corresponding field has -1 mapped value.
40+
* @return Enriched relational node row.
41+
*/
42+
public Row enrichRow(TableRow tableRow, Row nodeRow, int[] fieldColMapping);
43+
}

0 commit comments

Comments
 (0)