Skip to content

Commit b6c217a

Browse files
authored
Pipe: Fix delete inclusion aliases and metrics maps (#18002)
1 parent aa2acd6 commit b6c217a

4 files changed

Lines changed: 52 additions & 6 deletions

File tree

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/processor/PipeProcessorMetrics.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@
3333
import org.slf4j.Logger;
3434
import org.slf4j.LoggerFactory;
3535

36-
import java.util.HashMap;
3736
import java.util.Map;
3837
import java.util.Objects;
3938
import java.util.concurrent.ConcurrentHashMap;
@@ -45,7 +44,7 @@ public class PipeProcessorMetrics implements IMetricSet {
4544
@SuppressWarnings("java:S3077")
4645
private volatile AbstractMetricService metricService;
4746

48-
private final Map<String, PipeProcessorSubtask> processorMap = new HashMap<>();
47+
private final Map<String, PipeProcessorSubtask> processorMap = new ConcurrentHashMap<>();
4948

5049
private final Map<String, Rate> tabletRateMap = new ConcurrentHashMap<>();
5150

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/sink/PipeDataRegionSinkMetrics.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,6 @@
3535
import org.slf4j.Logger;
3636
import org.slf4j.LoggerFactory;
3737

38-
import java.util.HashMap;
3938
import java.util.Map;
4039
import java.util.Objects;
4140
import java.util.concurrent.ConcurrentHashMap;
@@ -47,7 +46,7 @@ public class PipeDataRegionSinkMetrics implements IMetricSet {
4746
@SuppressWarnings("java:S3077")
4847
private volatile AbstractMetricService metricService;
4948

50-
private final Map<String, PipeSinkSubtask> sinkMap = new HashMap<>();
49+
private final Map<String, PipeSinkSubtask> sinkMap = new ConcurrentHashMap<>();
5150

5251
private final Map<String, Rate> tabletRateMap = new ConcurrentHashMap<>();
5352

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/options/PipeInclusionOptions.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -125,7 +125,7 @@ public class PipeInclusionOptions {
125125
Arrays.asList(
126126
"data.delete",
127127
"schema.database.drop",
128-
"schema.timeseries.ordinary.delete",
128+
"schema.timeseries.ordinary.drop",
129129
"schema.timeseries.view.drop",
130130
"schema.timeseries.template.drop",
131131
"schema.timeseries.template.unset",
@@ -140,7 +140,7 @@ public class PipeInclusionOptions {
140140
new HashSet<>(
141141
Arrays.asList(
142142
"schema.database.drop",
143-
"schema.timeseries.ordinary.delete",
143+
"schema.timeseries.ordinary.drop",
144144
"schema.timeseries.view.drop",
145145
"schema.timeseries.template.drop",
146146
"schema.timeseries.template.unset",
Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.commons.pipe.datastructure.options;
21+
22+
import org.apache.iotdb.commons.path.PartialPath;
23+
24+
import org.junit.Assert;
25+
import org.junit.Test;
26+
27+
import java.util.Set;
28+
29+
public class PipeInclusionOptionsTest {
30+
31+
@Test
32+
public void testDeleteAliasIncludesOrdinaryTimeseriesDrop() throws Exception {
33+
final Set<PartialPath> options = PipeInclusionOptions.parseOptions("delete");
34+
35+
Assert.assertTrue(options.contains(new PartialPath("schema.timeseries.ordinary.drop")));
36+
Assert.assertFalse(options.contains(new PartialPath("schema.timeseries.ordinary.delete")));
37+
Assert.assertTrue(PipeInclusionOptions.optionsAreAllLegal("delete", true, false));
38+
}
39+
40+
@Test
41+
public void testSchemaDeleteAliasIncludesOrdinaryTimeseriesDrop() throws Exception {
42+
final Set<PartialPath> options = PipeInclusionOptions.parseOptions("schema.delete");
43+
44+
Assert.assertTrue(options.contains(new PartialPath("schema.timeseries.ordinary.drop")));
45+
Assert.assertFalse(options.contains(new PartialPath("schema.timeseries.ordinary.delete")));
46+
Assert.assertTrue(PipeInclusionOptions.optionsAreAllLegal("schema.delete", true, false));
47+
}
48+
}

0 commit comments

Comments
 (0)