Skip to content

Commit dd861a5

Browse files
DaZuiZuiDaZuiZui
committed
Fix LIKE with dynamic pattern expressions
1 parent da7056f commit dd861a5

10 files changed

Lines changed: 132 additions & 19 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,17 @@ public void setUp() {
5757
@Override
5858
protected void setupConfig() {
5959
super.setupConfig();
60+
senderEnv
61+
.getConfig()
62+
.getCommonConfig()
63+
.setSchemaReplicationFactor(1)
64+
.setDataReplicationFactor(1);
6065
receiverEnv.getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false);
66+
receiverEnv
67+
.getConfig()
68+
.getCommonConfig()
69+
.setSchemaReplicationFactor(1)
70+
.setDataReplicationFactor(1);
6171
}
6272

6373
@Test
@@ -72,7 +82,7 @@ public void testReceiverAutoCreateSchemaDisabledWithSpecialTimeSeries() throws E
7282
"create pipe test with source ('inclusion'='all','source.realtime.mode'='stream','source.realtime.enable'='true') "
7383
+ "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s');",
7484
receiverEnv.getDataNodeWrapper(0).getIpAndPortString());
75-
final String createDatabaseSql = "create database root.test.sg;";
85+
final String createDatabaseSql = "create database root.test;";
7686
final String createFirstTimeSeriesSql =
7787
"create timeseries root.test.sg.`1~!@#$%^&*()_+=:'\"/|[]{}`.`~!@#$%^&*()_+=:'\"/|[]{}` float;";
7888
final String insertFirstSql =

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LogicalPlanVisitor.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -555,6 +555,7 @@ public PlanNode visitLoadFile(
555555
loadTsFileStatement.getResources(),
556556
isTableModel,
557557
loadTsFileStatement.getDatabase(),
558+
loadTsFileStatement.getDatabaseLevel(),
558559
loadTsFileStatement.isNeedDecode4TimeColumn());
559560
}
560561

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,7 @@ public class LoadSingleTsFileNode extends WritePlanNode {
5959
private final TsFileResource resource;
6060
private final boolean isTableModel;
6161
private final String database;
62+
private final int databaseLevel;
6263
private final boolean deleteAfterLoad;
6364
private final long writePointCount;
6465
private boolean needDecodeTsFile;
@@ -70,6 +71,7 @@ public LoadSingleTsFileNode(
7071
final TsFileResource resource,
7172
final boolean isTableModel,
7273
final String database,
74+
final int databaseLevel,
7375
final boolean deleteAfterLoad,
7476
final long writePointCount,
7577
final boolean needDecodeTsFile) {
@@ -78,6 +80,7 @@ public LoadSingleTsFileNode(
7880
this.resource = resource;
7981
this.isTableModel = isTableModel;
8082
this.database = database;
83+
this.databaseLevel = databaseLevel;
8184
this.deleteAfterLoad = deleteAfterLoad;
8285
this.writePointCount = writePointCount;
8386
this.needDecodeTsFile = needDecodeTsFile;
@@ -175,6 +178,10 @@ public String getDatabase() {
175178
return database;
176179
}
177180

181+
public int getDatabaseLevel() {
182+
return databaseLevel;
183+
}
184+
178185
@Override
179186
public TRegionReplicaSet getRegionReplicaSet() {
180187
return null;
@@ -258,6 +265,7 @@ public boolean equals(Object o) {
258265
&& Objects.equals(resource, loadSingleTsFileNode.resource)
259266
&& Objects.equals(isTableModel, loadSingleTsFileNode.isTableModel)
260267
&& Objects.equals(database, loadSingleTsFileNode.database)
268+
&& Objects.equals(databaseLevel, loadSingleTsFileNode.databaseLevel)
261269
&& Objects.equals(needDecodeTsFile, loadSingleTsFileNode.needDecodeTsFile)
262270
&& Objects.equals(deleteAfterLoad, loadSingleTsFileNode.deleteAfterLoad)
263271
&& Objects.equals(localRegionReplicaSet, loadSingleTsFileNode.localRegionReplicaSet);
@@ -270,6 +278,7 @@ public int hashCode() {
270278
resource,
271279
isTableModel,
272280
database,
281+
databaseLevel,
273282
needDecodeTsFile,
274283
deleteAfterLoad,
275284
localRegionReplicaSet);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadTsFileNode.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,18 +46,21 @@ public class LoadTsFileNode extends WritePlanNode {
4646
private final List<TsFileResource> resources;
4747
private final List<Boolean> isTableModel;
4848
private final String database;
49+
private final int databaseLevel;
4950
private final boolean needDecode4TimeColumn;
5051

5152
public LoadTsFileNode(
5253
final PlanNodeId id,
5354
final List<TsFileResource> resources,
5455
final List<Boolean> isTableModel,
5556
final String database,
57+
final int databaseLevel,
5658
final boolean needDecode4TimeColumn) {
5759
super(id);
5860
this.resources = resources;
5961
this.isTableModel = isTableModel;
6062
this.database = database;
63+
this.databaseLevel = databaseLevel;
6164
this.needDecode4TimeColumn = needDecode4TimeColumn;
6265
}
6366

@@ -126,6 +129,7 @@ private List<WritePlanNode> splitByPartitionForTreeModel(Analysis analysis) {
126129
resources.get(i),
127130
isTableModel.get(i),
128131
database,
132+
databaseLevel,
129133
statement.isDeleteAfterLoad(),
130134
statement.getWritePointCount(i),
131135
needDecode4TimeColumn));
@@ -149,6 +153,7 @@ private List<WritePlanNode> splitByPartitionForTableModel(
149153
resources.get(i),
150154
isTableModel.get(i),
151155
database,
156+
databaseLevel,
152157
statement.isDeleteAfterLoad(),
153158
statement.getWritePointCount(i),
154159
needDecode4TimeColumn));
@@ -170,11 +175,12 @@ public boolean equals(Object o) {
170175
LoadTsFileNode loadTsFileNode = (LoadTsFileNode) o;
171176
return Objects.equals(resources, loadTsFileNode.resources)
172177
&& Objects.equals(database, loadTsFileNode.database)
178+
&& Objects.equals(databaseLevel, loadTsFileNode.databaseLevel)
173179
&& Objects.equals(isTableModel, loadTsFileNode.isTableModel);
174180
}
175181

176182
@Override
177183
public int hashCode() {
178-
return Objects.hash(resources, database, isTableModel);
184+
return Objects.hash(resources, database, databaseLevel, isTableModel);
179185
}
180186
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/PredicatePushIntoMetadataChecker.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,10 @@ public Boolean visitIsNotNullPredicate(final IsNotNullPredicate node, final Void
195195

196196
@Override
197197
public Boolean visitLikePredicate(final LikePredicate node, final Void context) {
198-
return node.getValue().accept(this, context);
198+
return node.getValue() instanceof SymbolReference
199+
&& node.getValue().accept(this, context)
200+
&& node.getPattern() instanceof StringLiteral
201+
&& (!node.getEscape().isPresent() || node.getEscape().get() instanceof StringLiteral);
199202
}
200203

201204
@Override

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/schema/ConvertSchemaPredicateToFilterVisitor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -113,7 +113,8 @@ public SchemaFilter visitIsNotNullPredicate(
113113
final LikePredicate node, final Context context) {
114114
// TODO: Support stringLiteral like tag/attr?
115115
if (!(node.getValue() instanceof SymbolReference)
116-
|| !(node.getPattern() instanceof StringLiteral)) {
116+
|| !(node.getPattern() instanceof StringLiteral)
117+
|| (node.getEscape().isPresent() && !(node.getEscape().get() instanceof StringLiteral))) {
117118
return null;
118119
}
119120
return wrapTagOrAttributeFilter(

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1401,6 +1401,7 @@ public RelationPlan visitLoadTsFile(final LoadTsFile node, final Void context) {
14011401
node.getResources(),
14021402
isTableModel,
14031403
node.getDatabase(),
1404+
node.getDatabaseLevel(),
14041405
node.isNeedDecode4TimeColumn()),
14051406
analysis.getRootScope(),
14061407
Collections.emptyList(),

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java‎

Lines changed: 56 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,12 @@
3030
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
3131
import org.apache.iotdb.commons.consensus.DataRegionId;
3232
import org.apache.iotdb.commons.consensus.index.ProgressIndex;
33+
import org.apache.iotdb.commons.exception.IllegalPathException;
3334
import org.apache.iotdb.commons.exception.IoTDBException;
3435
import org.apache.iotdb.commons.partition.DataPartition;
3536
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
3637
import org.apache.iotdb.commons.partition.StorageExecutor;
38+
import org.apache.iotdb.commons.path.PartialPath;
3739
import org.apache.iotdb.commons.service.metric.MetricService;
3840
import org.apache.iotdb.commons.service.metric.enums.Metric;
3941
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -180,7 +182,8 @@ public void start() {
180182
final LoadSingleTsFileNode node = tsFileNodeList.get(i);
181183
final String filePath = node.getTsFileResource().getTsFilePath();
182184

183-
partitionFetcher.setDatabase(getPartitionQueryDatabase(node, isGeneratedByPipe));
185+
partitionFetcher.setDatabase(
186+
getPartitionQueryDatabase(node, isGeneratedByPipe), node.getDatabaseLevel());
184187

185188
boolean isLoadSingleTsFileSuccess = true;
186189
boolean shouldRemoveFileFromLoadingSet = false;
@@ -634,7 +637,38 @@ private void convertFailedTsFilesToTabletsAndRetry() {
634637

635638
static String getPartitionQueryDatabase(
636639
final LoadSingleTsFileNode node, final boolean isGeneratedByPipe) {
637-
return node.isTableModel() || isGeneratedByPipe ? node.getDatabase() : null;
640+
if (node.isTableModel()) {
641+
return node.getDatabase();
642+
}
643+
if (!isGeneratedByPipe) {
644+
return null;
645+
}
646+
return node.getDatabase() != null
647+
? node.getDatabase()
648+
: inferDatabaseName(node.getTsFileResource().getDevices(), node.getDatabaseLevel());
649+
}
650+
651+
private static String inferDatabaseName(final Set<IDeviceID> devices, final int databaseLevel) {
652+
if (devices == null || devices.isEmpty()) {
653+
return null;
654+
}
655+
return inferDatabaseName(devices.iterator().next(), databaseLevel);
656+
}
657+
658+
private static String inferDatabaseName(final IDeviceID deviceID, final int databaseLevel) {
659+
try {
660+
final String[] deviceNodes = new PartialPath(deviceID).getNodes();
661+
final int databaseNodesLength = databaseLevel + 1;
662+
if (deviceNodes.length < databaseNodesLength) {
663+
return null;
664+
}
665+
final String[] databaseNodes = new String[databaseNodesLength];
666+
System.arraycopy(deviceNodes, 0, databaseNodes, 0, databaseNodesLength);
667+
return new PartialPath(databaseNodes).getFullPath();
668+
} catch (final IllegalPathException e) {
669+
LOGGER.warn("Failed to infer database name from device {}.", deviceID, e);
670+
return null;
671+
}
638672
}
639673

640674
private LoadTsFileStatement buildRetryTreeLoadStatement(
@@ -847,13 +881,15 @@ private void clear() {
847881
private static class DataPartitionBatchFetcher {
848882
private final IPartitionFetcher fetcher;
849883
private String database;
884+
private int databaseLevel;
850885

851886
public DataPartitionBatchFetcher(IPartitionFetcher fetcher) {
852887
this.fetcher = fetcher;
853888
}
854889

855-
public void setDatabase(String database) {
890+
public void setDatabase(String database, int databaseLevel) {
856891
this.database = database;
892+
this.databaseLevel = databaseLevel;
857893
}
858894

859895
public List<TRegionReplicaSet> queryDataPartition(
@@ -869,14 +905,17 @@ public List<TRegionReplicaSet> queryDataPartition(
869905
replicaSets.addAll(
870906
subSlotList.stream()
871907
.map(
872-
pair ->
873-
// database is an explicit database hint for table-model loads and
874-
// pipe-generated tree-model loads.
875-
database != null
876-
? dataPartition.getDataRegionReplicaSetForWriting(
877-
pair.left, pair.right, database)
878-
: dataPartition.getDataRegionReplicaSetForWriting(
879-
pair.left, pair.right))
908+
pair -> {
909+
// database is an explicit database hint for table-model loads and
910+
// pipe-generated tree-model loads. When a pipe-generated tree-model load
911+
// only carries database-level, infer the database from the device.
912+
final String queryDatabase =
913+
database != null ? database : inferDatabaseName(pair.left, databaseLevel);
914+
return queryDatabase != null
915+
? dataPartition.getDataRegionReplicaSetForWriting(
916+
pair.left, pair.right, queryDatabase)
917+
: dataPartition.getDataRegionReplicaSetForWriting(pair.left, pair.right);
918+
})
880919
.collect(Collectors.toList()));
881920
}
882921
return replicaSets;
@@ -895,9 +934,12 @@ private List<DataPartitionQueryParam> toQueryParam(
895934
DataPartitionQueryParam queryParam =
896935
new DataPartitionQueryParam(entry.getKey(), new ArrayList<>(entry.getValue()));
897936
// database is an explicit database hint for table-model loads and
898-
// pipe-generated tree-model loads.
899-
if (database != null) {
900-
queryParam.setDatabaseName(database);
937+
// pipe-generated tree-model loads. When a pipe-generated tree-model load
938+
// only carries database-level, infer the database from the device.
939+
final String queryDatabase =
940+
database != null ? database : inferDatabaseName(entry.getKey(), databaseLevel);
941+
if (queryDatabase != null) {
942+
queryParam.setDatabaseName(queryDatabase);
901943
}
902944
return queryParam;
903945
})

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,11 +40,14 @@ public class LoadTsFileNodeTest {
4040
public void testLoadSingleTsFileNode() {
4141
TsFileResource resource = new TsFileResource(new File("1"));
4242
String database = "root.db";
43+
int databaseLevel = 1;
4344
LoadSingleTsFileNode node =
44-
new LoadSingleTsFileNode(new PlanNodeId(""), resource, false, database, true, 0L, false);
45+
new LoadSingleTsFileNode(
46+
new PlanNodeId(""), resource, false, database, databaseLevel, true, 0L, false);
4547
Assert.assertTrue(node.isDeleteAfterLoad());
4648
Assert.assertEquals(resource, node.getTsFileResource());
4749
Assert.assertEquals(database, node.getDatabase());
50+
Assert.assertEquals(databaseLevel, node.getDatabaseLevel());
4851
Assert.assertNull(node.getLocalRegionReplicaSet());
4952
Assert.assertNull(node.getRegionReplicaSet());
5053
Assert.assertEquals(Collections.emptyList(), node.getChildren());

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/AnalyzerTest.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -671,6 +671,43 @@ public void expressionTest() {
671671
assertNull(deviceTableScanNode.getPushDownPredicate());
672672
assertFalse(deviceTableScanNode.getTimePredicate().isPresent());
673673

674+
sql = "SELECT * FROM table1 WHERE tag1 like 'A' || '%'";
675+
context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
676+
analysis = analyzeSQL(sql, metadata, context);
677+
symbolAllocator = new SymbolAllocator();
678+
logicalQueryPlan =
679+
new TableLogicalPlanner(
680+
context, metadata, sessionInfo, symbolAllocator, WarningCollector.NOOP)
681+
.plan(analysis);
682+
rootNode = logicalQueryPlan.getRootNode();
683+
684+
// Like with a non-literal pattern is evaluated by the filter operator.
685+
assertTrue(rootNode.getChildren().get(0) instanceof FilterNode);
686+
filterNode = (FilterNode) rootNode.getChildren().get(0);
687+
assertTrue(filterNode.getPredicate().toString().contains("LIKE"));
688+
assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof DeviceTableScanNode);
689+
deviceTableScanNode = (DeviceTableScanNode) rootNode.getChildren().get(0).getChildren().get(0);
690+
assertNull(deviceTableScanNode.getPushDownPredicate());
691+
assertFalse(deviceTableScanNode.getTimePredicate().isPresent());
692+
693+
sql = "SELECT * FROM table1 WHERE tag1 like concat('A', '%')";
694+
context = new MPPQueryContext(sql, queryId, sessionInfo, null, null);
695+
analysis = analyzeSQL(sql, metadata, context);
696+
symbolAllocator = new SymbolAllocator();
697+
logicalQueryPlan =
698+
new TableLogicalPlanner(
699+
context, metadata, sessionInfo, symbolAllocator, WarningCollector.NOOP)
700+
.plan(analysis);
701+
rootNode = logicalQueryPlan.getRootNode();
702+
703+
assertTrue(rootNode.getChildren().get(0) instanceof FilterNode);
704+
filterNode = (FilterNode) rootNode.getChildren().get(0);
705+
assertTrue(filterNode.getPredicate().toString().contains("LIKE"));
706+
assertTrue(rootNode.getChildren().get(0).getChildren().get(0) instanceof DeviceTableScanNode);
707+
deviceTableScanNode = (DeviceTableScanNode) rootNode.getChildren().get(0).getChildren().get(0);
708+
assertNull(deviceTableScanNode.getPushDownPredicate());
709+
assertFalse(deviceTableScanNode.getTimePredicate().isPresent());
710+
674711
// 3. in / not in
675712
sql =
676713
"SELECT *, s1/2, s2+1, s2*3, s1+s2, s2%1 FROM table1 WHERE tag1 in ('A', 'B') and tag2 not in ('A', 'C')";

0 commit comments

Comments
 (0)