Skip to content

Commit a7f3958

Browse files
CaideyipiCaideyipi
committed
Improve object column parser compatibility
1 parent eb4409d commit a7f3958

9 files changed

Lines changed: 425 additions & 10 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/TsFileInsertionEventParser.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryBlock;
3434
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
3535
import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
36+
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
3637
import org.apache.iotdb.db.utils.datastructure.PatternTreeMapFactory;
3738
import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
3839

@@ -45,6 +46,7 @@
4546

4647
import java.io.File;
4748
import java.io.IOException;
49+
import java.util.List;
4850

4951
public abstract class TsFileInsertionEventParser implements AutoCloseable {
5052

@@ -78,6 +80,8 @@ public abstract class TsFileInsertionEventParser implements AutoCloseable {
7880

7981
protected Iterable<TabletInsertionEvent> tabletInsertionIterable;
8082

83+
protected final boolean objectPathsOnly;
84+
8185
protected TsFileInsertionEventParser(
8286
final File tsFile,
8387
final String pipeName,
@@ -91,6 +95,38 @@ protected TsFileInsertionEventParser(
9195
final boolean skipIfNoPrivileges,
9296
final PipeInsertionEvent sourceEvent,
9397
final boolean isWithMod) {
98+
this(
99+
tsFile,
100+
pipeName,
101+
creationTime,
102+
treePattern,
103+
tablePattern,
104+
startTime,
105+
endTime,
106+
pipeTaskMeta,
107+
entity,
108+
skipIfNoPrivileges,
109+
sourceEvent,
110+
null,
111+
false,
112+
isWithMod);
113+
}
114+
115+
protected TsFileInsertionEventParser(
116+
final File tsFile,
117+
final String pipeName,
118+
final long creationTime,
119+
final TreePattern treePattern,
120+
final TablePattern tablePattern,
121+
final long startTime,
122+
final long endTime,
123+
final PipeTaskMeta pipeTaskMeta,
124+
final IAuditEntity entity,
125+
final boolean skipIfNoPrivileges,
126+
final PipeInsertionEvent sourceEvent,
127+
final TsFileResource tsFileResource,
128+
final boolean objectPathsOnly,
129+
final boolean isWithMod) {
94130
this.pipeName = pipeName;
95131
this.creationTime = creationTime;
96132
this.entity = entity;
@@ -107,6 +143,7 @@ protected TsFileInsertionEventParser(
107143

108144
this.pipeTaskMeta = pipeTaskMeta;
109145
this.sourceEvent = sourceEvent;
146+
this.objectPathsOnly = objectPathsOnly;
110147

111148
this.allocatedMemoryBlockForTablet =
112149
PipeDataNodeResourceManager.memory()
@@ -130,6 +167,8 @@ protected TsFileInsertionEventParser(
130167
*/
131168
public abstract Iterable<TabletInsertionEvent> toTabletInsertionEvents();
132169

170+
public void drainGeneratedObjectColumnModEntriesTo(final List<ModEntry> modEntries) {}
171+
133172
/**
134173
* Record parse start time when hasNext() is called for the first time and returns true. Should be
135174
* called in Iterator.hasNext() when it's the first call.

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/query/TsFileInsertionEventQueryParser.java‎

Lines changed: 88 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,7 @@ public TsFileInsertionEventQueryParser(
9090
final long endTime,
9191
final PipeInsertionEvent sourceEvent)
9292
throws IOException, IllegalPathException {
93-
this(null, 0, tsFile, pattern, startTime, endTime, null, sourceEvent, false);
93+
this(null, 0, tsFile, pattern, startTime, endTime, null, sourceEvent, false, false);
9494
}
9595

9696
public TsFileInsertionEventQueryParser(
@@ -104,6 +104,31 @@ public TsFileInsertionEventQueryParser(
104104
final PipeInsertionEvent sourceEvent,
105105
final boolean isWithMod)
106106
throws IOException, IllegalPathException {
107+
this(
108+
pipeName,
109+
creationTime,
110+
tsFile,
111+
pattern,
112+
startTime,
113+
endTime,
114+
pipeTaskMeta,
115+
sourceEvent,
116+
isWithMod,
117+
false);
118+
}
119+
120+
public TsFileInsertionEventQueryParser(
121+
final String pipeName,
122+
final long creationTime,
123+
final File tsFile,
124+
final TreePattern pattern,
125+
final long startTime,
126+
final long endTime,
127+
final PipeTaskMeta pipeTaskMeta,
128+
final PipeInsertionEvent sourceEvent,
129+
final boolean isWithMod,
130+
final boolean objectPathsOnly)
131+
throws IOException, IllegalPathException {
107132
this(
108133
pipeName,
109134
creationTime,
@@ -116,7 +141,8 @@ public TsFileInsertionEventQueryParser(
116141
null,
117142
false,
118143
null,
119-
isWithMod);
144+
isWithMod,
145+
objectPathsOnly);
120146
}
121147

122148
public TsFileInsertionEventQueryParser(
@@ -133,6 +159,37 @@ public TsFileInsertionEventQueryParser(
133159
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
134160
final boolean isWithMod)
135161
throws IOException, IllegalPathException {
162+
this(
163+
pipeName,
164+
creationTime,
165+
tsFile,
166+
pattern,
167+
startTime,
168+
endTime,
169+
pipeTaskMeta,
170+
sourceEvent,
171+
entity,
172+
skipIfNoPrivileges,
173+
deviceIsAlignedMap,
174+
isWithMod,
175+
false);
176+
}
177+
178+
public TsFileInsertionEventQueryParser(
179+
final String pipeName,
180+
final long creationTime,
181+
final File tsFile,
182+
final TreePattern pattern,
183+
final long startTime,
184+
final long endTime,
185+
final PipeTaskMeta pipeTaskMeta,
186+
final PipeInsertionEvent sourceEvent,
187+
final IAuditEntity entity,
188+
final boolean skipIfNoPrivileges,
189+
final Map<IDeviceID, Boolean> deviceIsAlignedMap,
190+
final boolean isWithMod,
191+
final boolean objectPathsOnly)
192+
throws IOException, IllegalPathException {
136193
super(
137194
tsFile,
138195
pipeName,
@@ -145,6 +202,8 @@ public TsFileInsertionEventQueryParser(
145202
entity,
146203
skipIfNoPrivileges,
147204
sourceEvent,
205+
null,
206+
objectPathsOnly,
148207
isWithMod);
149208

150209
try {
@@ -157,7 +216,7 @@ public TsFileInsertionEventQueryParser(
157216
.forceAllocateForTabletWithRetry(currentModifications.ramBytesUsed());
158217

159218
final PipeTsFileResourceManager tsFileResourceManager = PipeDataNodeResourceManager.tsfile();
160-
final Map<IDeviceID, List<String>> deviceMeasurementsMap;
219+
Map<IDeviceID, List<String>> deviceMeasurementsMap;
161220

162221
// TsFileReader is not thread-safe, so we need to create it here and close it later.
163222
long memoryRequiredInBytes =
@@ -197,6 +256,10 @@ public TsFileInsertionEventQueryParser(
197256
memoryRequiredInBytes +=
198257
PipeMemoryWeightUtil.memoryOfIDeviceID2StrList(deviceMeasurementsMap);
199258
}
259+
if (objectPathsOnly) {
260+
deviceMeasurementsMap =
261+
filterObjectMeasurementsMap(deviceMeasurementsMap, measurementDataTypeMap);
262+
}
200263
allocatedMemoryBlock =
201264
PipeDataNodeResourceManager.memory().forceAllocate(memoryRequiredInBytes);
202265

@@ -373,9 +436,10 @@ private Map<IDeviceID, List<String>> readFilteredDeviceMeasurementsMap(
373436
final Map<IDeviceID, List<String>> result = new HashMap<>();
374437

375438
for (final IDeviceID device : devices) {
376-
tsFileSequenceReader
377-
.readDeviceMetadata(device)
378-
.values()
439+
tsFileSequenceReader.readDeviceMetadata(device).values().stream()
440+
.filter(
441+
timeseriesMetadata ->
442+
!objectPathsOnly || timeseriesMetadata.getTsDataType() == TSDataType.OBJECT)
379443
.forEach(
380444
timeseriesMetadata ->
381445
result
@@ -386,6 +450,24 @@ private Map<IDeviceID, List<String>> readFilteredDeviceMeasurementsMap(
386450
return result;
387451
}
388452

453+
private Map<IDeviceID, List<String>> filterObjectMeasurementsMap(
454+
final Map<IDeviceID, List<String>> deviceMeasurementsMap,
455+
final Map<String, TSDataType> measurementDataTypeMap) {
456+
final Map<IDeviceID, List<String>> result = new HashMap<>();
457+
for (final Map.Entry<IDeviceID, List<String>> entry : deviceMeasurementsMap.entrySet()) {
458+
final List<String> measurements = new ArrayList<>();
459+
for (final String measurement : entry.getValue()) {
460+
if (measurementDataTypeMap.get(entry.getKey() + "." + measurement) == TSDataType.OBJECT) {
461+
measurements.add(measurement);
462+
}
463+
}
464+
if (!measurements.isEmpty()) {
465+
result.put(entry.getKey(), measurements);
466+
}
467+
}
468+
return result;
469+
}
470+
389471
@Override
390472
public Iterable<TabletInsertionEvent> toTabletInsertionEvents() {
391473
if (tabletInsertionIterable == null) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tsfile/parser/scan/TsFileInsertionEventScanParser.java‎

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,6 +123,35 @@ public TsFileInsertionEventScanParser(
123123
final PipeInsertionEvent sourceEvent,
124124
final boolean isWithMod)
125125
throws IOException, IllegalPathException {
126+
this(
127+
pipeName,
128+
creationTime,
129+
tsFile,
130+
pattern,
131+
startTime,
132+
endTime,
133+
pipeTaskMeta,
134+
entity,
135+
skipIfNoPrivileges,
136+
sourceEvent,
137+
isWithMod,
138+
false);
139+
}
140+
141+
public TsFileInsertionEventScanParser(
142+
final String pipeName,
143+
final long creationTime,
144+
final File tsFile,
145+
final TreePattern pattern,
146+
final long startTime,
147+
final long endTime,
148+
final PipeTaskMeta pipeTaskMeta,
149+
final IAuditEntity entity,
150+
final boolean skipIfNoPrivileges,
151+
final PipeInsertionEvent sourceEvent,
152+
final boolean isWithMod,
153+
final boolean objectPathsOnly)
154+
throws IOException, IllegalPathException {
126155
super(
127156
tsFile,
128157
pipeName,
@@ -135,6 +164,8 @@ public TsFileInsertionEventScanParser(
135164
entity,
136165
skipIfNoPrivileges,
137166
sourceEvent,
167+
null,
168+
objectPathsOnly,
138169
isWithMod);
139170

140171
this.startTime = startTime;
@@ -181,6 +212,19 @@ public TsFileInsertionEventScanParser(
181212
final PipeInsertionEvent sourceEvent,
182213
final boolean isWithMod)
183214
throws IOException, IllegalPathException {
215+
this(tsFile, pattern, startTime, endTime, pipeTaskMeta, sourceEvent, isWithMod, false);
216+
}
217+
218+
public TsFileInsertionEventScanParser(
219+
final File tsFile,
220+
final TreePattern pattern,
221+
final long startTime,
222+
final long endTime,
223+
final PipeTaskMeta pipeTaskMeta,
224+
final PipeInsertionEvent sourceEvent,
225+
final boolean isWithMod,
226+
final boolean objectPathsOnly)
227+
throws IOException, IllegalPathException {
184228
this(
185229
null,
186230
0,
@@ -192,7 +236,8 @@ public TsFileInsertionEventScanParser(
192236
null,
193237
false,
194238
sourceEvent,
195-
isWithMod);
239+
isWithMod,
240+
objectPathsOnly);
196241
}
197242

198243
@Override
@@ -416,8 +461,12 @@ private boolean putValueToColumns(final BatchData data, final Tablet tablet, fin
416461
case TEXT:
417462
case BLOB:
418463
case STRING:
464+
case OBJECT:
419465
PipeTabletUtils.putValue(
420466
tablet, rowIndex, i, tablet.getSchemas().get(i).getType(), Binary.EMPTY_VALUE);
467+
break;
468+
default:
469+
break;
421470
}
422471
PipeTabletUtils.markNullValue(tablet, rowIndex, i);
423472
continue;
@@ -469,6 +518,7 @@ private boolean putValueToColumns(final BatchData data, final Tablet tablet, fin
469518
case TEXT:
470519
case BLOB:
471520
case STRING:
521+
case OBJECT:
472522
final Binary binary = primitiveType.getBinary();
473523
PipeTabletUtils.putValue(
474524
tablet,
@@ -524,6 +574,7 @@ private boolean putValueToColumns(final BatchData data, final Tablet tablet, fin
524574
case TEXT:
525575
case BLOB:
526576
case STRING:
577+
case OBJECT:
527578
final Binary binary = data.getBinary();
528579
PipeTabletUtils.putValue(
529580
tablet,
@@ -773,6 +824,11 @@ private boolean filterChunk(
773824
return true;
774825
}
775826

827+
if (objectPathsOnly && chunkHeader.getDataType() != TSDataType.OBJECT) {
828+
tsFileSequenceReader.position(nextMarkerOffset);
829+
return true;
830+
}
831+
776832
// Skip the chunk if it is fully deleted by mods
777833
if (!currentModifications.isEmpty()) {
778834
final Statistics statistics =

0 commit comments

Comments
 (0)