Skip to content

Commit 1d893f1

Browse files
CaideyipiCaideyipi
authored
Support unified pipe pattern inclusion keys (#18016)
1 parent 4d05a85 commit 1d893f1

6 files changed

Lines changed: 199 additions & 114 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@
109109
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_END_TIME_KEY;
110110
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_HISTORY_START_TIME_KEY;
111111
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_EXCLUSION_KEY;
112+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_INCLUSION_KEY;
112113
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_KEY;
113114
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_EXCLUSION_KEY;
114115
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_INCLUSION_KEY;
@@ -121,6 +122,7 @@
121122
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_END_TIME_KEY;
122123
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_HISTORY_START_TIME_KEY;
123124
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_EXCLUSION_KEY;
125+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_INCLUSION_KEY;
124126
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_KEY;
125127
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_EXCLUSION_KEY;
126128
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_INCLUSION_KEY;
@@ -913,6 +915,8 @@ private long calculateTsFileParserMemory(
913915
|| sourceParameters.hasAnyAttributes(
914916
EXTRACTOR_PATH_KEY,
915917
SOURCE_PATH_KEY,
918+
EXTRACTOR_PATH_INCLUSION_KEY,
919+
SOURCE_PATH_INCLUSION_KEY,
916920
EXTRACTOR_PATTERN_INCLUSION_KEY,
917921
SOURCE_PATTERN_INCLUSION_KEY,
918922
EXTRACTOR_PATH_EXCLUSION_KEY,

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@
8585
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODE_STRICT_KEY;
8686
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_ENABLE_KEY;
8787
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_MODS_KEY;
88+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_INCLUSION_KEY;
8889
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_KEY;
8990
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE;
9091
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_FORMAT_KEY;
@@ -119,6 +120,7 @@
119120
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODE_STRICT_KEY;
120121
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_ENABLE_KEY;
121122
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_MODS_KEY;
123+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_INCLUSION_KEY;
122124
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_KEY;
123125
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_FORMAT_KEY;
124126
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_INCLUSION_KEY;
@@ -197,6 +199,8 @@ public void validate(final PipeParameterValidator validator) throws Exception {
197199
SOURCE_PATH_KEY,
198200
EXTRACTOR_PATTERN_KEY,
199201
SOURCE_PATTERN_KEY,
202+
EXTRACTOR_PATH_INCLUSION_KEY,
203+
SOURCE_PATH_INCLUSION_KEY,
200204
EXTRACTOR_PATTERN_INCLUSION_KEY,
201205
SOURCE_PATTERN_INCLUSION_KEY)) {
202206
throw new PipeException(DataNodePipeMessages.THE_PIPE_CANNOT_EXTRACT_TREE_MODEL_DATA);

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/pattern/TreePatternPruningTest.java‎

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ public void testComplexPruning() {
122122
}
123123

124124
@Test
125-
public void testLegacyPatternMultipleRulesRejected() {
125+
public void testLegacyPatternMultipleRulesPreserved() {
126126
final PipeParameters params =
127127
new PipeParameters(
128128
new HashMap<String, String>() {
@@ -131,12 +131,9 @@ public void testLegacyPatternMultipleRulesRejected() {
131131
}
132132
});
133133

134-
try {
135-
TreePattern.parsePipePatternFromSourceParameters(params);
136-
Assert.fail("Should throw PipeException for legacy multi-pattern parameters");
137-
} catch (final PipeException ignored) {
138-
// Expected exception
139-
}
134+
final TreePattern result = TreePattern.parsePipePatternFromSourceParameters(params);
135+
136+
Assert.assertEquals("root.sg.A,root.sg.B", result.getPattern());
140137
}
141138

142139
@Test
@@ -157,7 +154,7 @@ public void testUnionPreservedWhenNotCovered() {
157154
}
158155

159156
@Test
160-
public void testLegacyPathMultipleRulesRejected() {
157+
public void testLegacyPathMultipleRulesPreserved() {
161158
final PipeParameters params =
162159
new PipeParameters(
163160
new HashMap<String, String>() {
@@ -166,12 +163,28 @@ public void testLegacyPathMultipleRulesRejected() {
166163
}
167164
});
168165

169-
try {
170-
TreePattern.parsePipePatternFromSourceParameters(params);
171-
Assert.fail("Should throw PipeException for legacy multi-path parameters");
172-
} catch (final PipeException ignored) {
173-
// Expected exception
174-
}
166+
final TreePattern result = TreePattern.parsePipePatternFromSourceParameters(params);
167+
168+
Assert.assertTrue(result instanceof UnionIoTDBTreePattern);
169+
Assert.assertEquals("root.sg.d1,root.sg.d2", result.getPattern());
170+
}
171+
172+
@Test
173+
public void testLegacyPathMultipleExclusionsPreserved() {
174+
final PipeParameters params =
175+
new PipeParameters(
176+
new HashMap<String, String>() {
177+
{
178+
put(PipeSourceConstant.SOURCE_PATH_KEY, "root.sg.**");
179+
put(PipeSourceConstant.SOURCE_PATH_EXCLUSION_KEY, "root.sg.d1,root.sg.d2");
180+
}
181+
});
182+
183+
final TreePattern result = TreePattern.parsePipePatternFromSourceParameters(params);
184+
185+
Assert.assertTrue(result instanceof WithExclusionIoTDBTreePattern);
186+
Assert.assertEquals(
187+
"INCLUSION(root.sg.**), EXCLUSION(root.sg.d1,root.sg.d2)", result.getPattern());
175188
}
176189

177190
@Test

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSourceConstant.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,8 @@ public class PipeSourceConstant {
5858
public static final String SOURCE_PATTERN_INCLUSION_KEY = "source.pattern.inclusion";
5959
public static final String EXTRACTOR_PATH_KEY = "extractor.path";
6060
public static final String SOURCE_PATH_KEY = "source.path";
61+
public static final String EXTRACTOR_PATH_INCLUSION_KEY = "extractor.path.inclusion";
62+
public static final String SOURCE_PATH_INCLUSION_KEY = "source.path.inclusion";
6163
public static final String EXTRACTOR_PATTERN_FORMAT_KEY = "extractor.pattern.format";
6264
public static final String SOURCE_PATTERN_FORMAT_KEY = "source.pattern.format";
6365
public static final String EXTRACTOR_PATTERN_FORMAT_PREFIX_VALUE = "prefix";

‎iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/datastructure/pattern/TreePattern.java‎

Lines changed: 79 additions & 100 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import java.util.stream.Collectors;
4545

4646
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_EXCLUSION_KEY;
47+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_INCLUSION_KEY;
4748
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATH_KEY;
4849
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_EXCLUSION_KEY;
4950
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_FORMAT_IOTDB_VALUE;
@@ -52,6 +53,7 @@
5253
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_INCLUSION_KEY;
5354
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_PATTERN_KEY;
5455
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_EXCLUSION_KEY;
56+
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_INCLUSION_KEY;
5557
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATH_KEY;
5658
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_EXCLUSION_KEY;
5759
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.SOURCE_PATTERN_FORMAT_KEY;
@@ -164,49 +166,9 @@ public static TreePattern parsePipePatternFromSourceParametersInternal(
164166
final boolean isTreeModelDataAllowedToBeCaptured =
165167
isTreeModelDataAllowToBeCaptured(sourceParameters);
166168

167-
final boolean hasPatternInclusionKey =
168-
sourceParameters.hasAnyAttributes(
169-
EXTRACTOR_PATTERN_INCLUSION_KEY, SOURCE_PATTERN_INCLUSION_KEY);
170-
final boolean hasPatternExclusionKey =
171-
sourceParameters.hasAnyAttributes(
172-
EXTRACTOR_PATTERN_EXCLUSION_KEY, SOURCE_PATTERN_EXCLUSION_KEY);
173-
final boolean hasLegacyPathKey =
174-
sourceParameters.hasAnyAttributes(EXTRACTOR_PATH_KEY, SOURCE_PATH_KEY);
175-
final boolean hasLegacyPatternKey =
176-
sourceParameters.hasAnyAttributes(EXTRACTOR_PATTERN_KEY, SOURCE_PATTERN_KEY);
177-
final boolean usePatternSyntax =
178-
hasPatternInclusionKey
179-
|| (hasPatternExclusionKey && !hasLegacyPathKey && !hasLegacyPatternKey);
180-
181-
if (hasPatternInclusionKey && (hasLegacyPathKey || hasLegacyPatternKey)) {
182-
final String msg =
183-
String.format(
184-
PipeMessages.PATTERN_INCLUSION_CANNOT_BE_USED_WITH_PATTERN_OR_PATH,
185-
SOURCE_PATTERN_INCLUSION_KEY,
186-
SOURCE_PATTERN_KEY,
187-
SOURCE_PATH_KEY);
188-
LOGGER.warn(msg);
189-
throw new PipeException(msg);
190-
}
191-
192169
// 1. Parse INCLUSION patterns into a list
193170
List<TreePattern> inclusionPatterns =
194-
usePatternSyntax
195-
? parseIoTDBPatternList(
196-
sourceParameters.getStringByKeys(
197-
EXTRACTOR_PATTERN_INCLUSION_KEY, SOURCE_PATTERN_INCLUSION_KEY),
198-
isTreeModelDataAllowedToBeCaptured,
199-
true,
200-
SOURCE_PATTERN_INCLUSION_KEY)
201-
: parseLegacyPatternList(
202-
sourceParameters,
203-
isTreeModelDataAllowedToBeCaptured,
204-
EXTRACTOR_PATH_KEY,
205-
SOURCE_PATH_KEY,
206-
EXTRACTOR_PATTERN_KEY,
207-
SOURCE_PATTERN_KEY,
208-
SOURCE_PATH_KEY,
209-
SOURCE_PATTERN_KEY);
171+
parseInclusionPatternList(sourceParameters, isTreeModelDataAllowedToBeCaptured);
210172

211173
// If no inclusion patterns are specified, use default "root.**"
212174
if (inclusionPatterns.isEmpty()) {
@@ -217,35 +179,8 @@ public static TreePattern parsePipePatternFromSourceParametersInternal(
217179
}
218180

219181
// 2. Parse EXCLUSION patterns into a list
220-
if (usePatternSyntax
221-
&& sourceParameters.hasAnyAttributes(
222-
EXTRACTOR_PATH_EXCLUSION_KEY, SOURCE_PATH_EXCLUSION_KEY)) {
223-
final String msg =
224-
String.format(
225-
PipeMessages.PATTERN_INCLUSION_CANNOT_BE_USED_WITH_PATH_EXCLUSION,
226-
SOURCE_PATTERN_INCLUSION_KEY,
227-
SOURCE_PATH_EXCLUSION_KEY);
228-
LOGGER.warn(msg);
229-
throw new PipeException(msg);
230-
}
231-
232182
List<TreePattern> exclusionPatterns =
233-
usePatternSyntax
234-
? parseIoTDBPatternList(
235-
sourceParameters.getStringByKeys(
236-
EXTRACTOR_PATTERN_EXCLUSION_KEY, SOURCE_PATTERN_EXCLUSION_KEY),
237-
isTreeModelDataAllowedToBeCaptured,
238-
true,
239-
SOURCE_PATTERN_EXCLUSION_KEY)
240-
: parseLegacyPatternList(
241-
sourceParameters,
242-
isTreeModelDataAllowedToBeCaptured,
243-
EXTRACTOR_PATH_EXCLUSION_KEY,
244-
SOURCE_PATH_EXCLUSION_KEY,
245-
EXTRACTOR_PATTERN_EXCLUSION_KEY,
246-
SOURCE_PATTERN_EXCLUSION_KEY,
247-
SOURCE_PATH_EXCLUSION_KEY,
248-
SOURCE_PATTERN_EXCLUSION_KEY);
183+
parseExclusionPatternList(sourceParameters, isTreeModelDataAllowedToBeCaptured);
249184

250185
// 3. Optimize the lists: remove redundant patterns (e.g., if "root.**" exists, "root.db" is
251186
// redundant)
@@ -264,6 +199,8 @@ public static TreePattern parsePipePatternFromSourceParametersInternal(
264199
sourceParameters.getStringByKeys(
265200
EXTRACTOR_PATTERN_INCLUSION_KEY,
266201
SOURCE_PATTERN_INCLUSION_KEY,
202+
EXTRACTOR_PATH_INCLUSION_KEY,
203+
SOURCE_PATH_INCLUSION_KEY,
267204
EXTRACTOR_PATH_KEY,
268205
SOURCE_PATH_KEY,
269206
EXTRACTOR_PATTERN_KEY,
@@ -382,49 +319,91 @@ public static TreePattern parsePatternFromString(
382319
parseMultiplePatterns(trimmedPattern, basePatternSupplier));
383320
}
384321

385-
/**
386-
* Helper method to parse legacy pattern parameters into a list of patterns without creating the
387-
* Union object immediately.
388-
*/
389-
private static List<TreePattern> parseLegacyPatternList(
390-
final PipeParameters sourceParameters,
391-
final boolean isTreeModelDataAllowedToBeCaptured,
392-
final String extractorPathKey,
393-
final String sourcePathKey,
394-
final String extractorPatternKey,
395-
final String sourcePatternKey,
396-
final String pathKeyName,
397-
final String patternKeyName) {
322+
private static List<TreePattern> parseInclusionPatternList(
323+
final PipeParameters sourceParameters, final boolean isTreeModelDataAllowedToBeCaptured) {
324+
final List<TreePattern> result = new ArrayList<>();
398325

399-
final String path = sourceParameters.getStringByKeys(extractorPathKey, sourcePathKey);
400-
final String pattern = sourceParameters.getStringByKeys(extractorPatternKey, sourcePatternKey);
326+
addPatternsFromPatternParameterIfPresent(
327+
result,
328+
sourceParameters,
329+
isTreeModelDataAllowedToBeCaptured,
330+
EXTRACTOR_PATTERN_KEY,
331+
SOURCE_PATTERN_KEY,
332+
SOURCE_PATTERN_KEY);
333+
addIoTDBPatternsIfPresent(
334+
result,
335+
sourceParameters,
336+
isTreeModelDataAllowedToBeCaptured,
337+
EXTRACTOR_PATTERN_INCLUSION_KEY,
338+
SOURCE_PATTERN_INCLUSION_KEY,
339+
SOURCE_PATTERN_INCLUSION_KEY);
340+
addIoTDBPatternsIfPresent(
341+
result,
342+
sourceParameters,
343+
isTreeModelDataAllowedToBeCaptured,
344+
EXTRACTOR_PATH_KEY,
345+
SOURCE_PATH_KEY,
346+
SOURCE_PATH_KEY);
347+
addIoTDBPatternsIfPresent(
348+
result,
349+
sourceParameters,
350+
isTreeModelDataAllowedToBeCaptured,
351+
EXTRACTOR_PATH_INCLUSION_KEY,
352+
SOURCE_PATH_INCLUSION_KEY,
353+
SOURCE_PATH_INCLUSION_KEY);
401354

402-
if (path != null && pattern != null) {
403-
final String msg =
404-
String.format(
405-
PipeMessages.PATH_AND_PATTERN_CANNOT_BE_USED_TOGETHER, pathKeyName, patternKeyName);
406-
LOGGER.warn(msg);
407-
throw new PipeException(msg);
408-
}
355+
return result;
356+
}
409357

358+
private static List<TreePattern> parseExclusionPatternList(
359+
final PipeParameters sourceParameters, final boolean isTreeModelDataAllowedToBeCaptured) {
410360
final List<TreePattern> result = new ArrayList<>();
411361

412-
if (path != null) {
362+
addIoTDBPatternsIfPresent(
363+
result,
364+
sourceParameters,
365+
isTreeModelDataAllowedToBeCaptured,
366+
EXTRACTOR_PATTERN_EXCLUSION_KEY,
367+
SOURCE_PATTERN_EXCLUSION_KEY,
368+
SOURCE_PATTERN_EXCLUSION_KEY);
369+
addIoTDBPatternsIfPresent(
370+
result,
371+
sourceParameters,
372+
isTreeModelDataAllowedToBeCaptured,
373+
EXTRACTOR_PATH_EXCLUSION_KEY,
374+
SOURCE_PATH_EXCLUSION_KEY,
375+
SOURCE_PATH_EXCLUSION_KEY);
376+
377+
return result;
378+
}
379+
380+
private static void addIoTDBPatternsIfPresent(
381+
final List<TreePattern> result,
382+
final PipeParameters sourceParameters,
383+
final boolean isTreeModelDataAllowedToBeCaptured,
384+
final String extractorKey,
385+
final String sourceKey,
386+
final String parameterKey) {
387+
final String pattern = sourceParameters.getStringByKeys(extractorKey, sourceKey);
388+
if (pattern != null) {
413389
result.addAll(
414-
parseIoTDBPatternList(path, isTreeModelDataAllowedToBeCaptured, false, pathKeyName));
390+
parseIoTDBPatternList(pattern, isTreeModelDataAllowedToBeCaptured, true, parameterKey));
415391
}
392+
}
416393

394+
private static void addPatternsFromPatternParameterIfPresent(
395+
final List<TreePattern> result,
396+
final PipeParameters sourceParameters,
397+
final boolean isTreeModelDataAllowedToBeCaptured,
398+
final String extractorKey,
399+
final String sourceKey,
400+
final String parameterKey) {
401+
final String pattern = sourceParameters.getStringByKeys(extractorKey, sourceKey);
417402
if (pattern != null) {
418403
result.addAll(
419404
parsePatternsFromPatternParameter(
420-
pattern,
421-
sourceParameters,
422-
isTreeModelDataAllowedToBeCaptured,
423-
false,
424-
patternKeyName));
405+
pattern, sourceParameters, isTreeModelDataAllowedToBeCaptured, true, parameterKey));
425406
}
426-
427-
return result;
428407
}
429408

430409
private static List<TreePattern> parseIoTDBPatternList(

0 commit comments

Comments
 (0)