Skip to content

Commit feaeaa6

Browse files
CaideyipiCaideyipi
authored
Fix partial insert handling for null measurements (#17879)
* Fix partial insert handling for null measurements * Fix partial insert handling for missing columns * Fix table insert schema validation for partial insert * Fix partial insert null column handling * Fix failed measurement WAL serialization tests * Fix all-failed tablet test column reuse
1 parent 69c0b18 commit feaeaa6

52 files changed

Lines changed: 3330 additions & 545 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeTypeConversionISessionIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -324,7 +324,7 @@ private void prepareTypeConversionTest(
324324
generateMeasurementSchemas();
325325

326326
// Generate createTimeSeries in sender and receiver
327-
String uuid = "bcdedit";
327+
String uuid = "bcdedit" + Long.toHexString(System.nanoTime());
328328
for (Pair<MeasurementSchema, MeasurementSchema> pair : measurementSchemas) {
329329
createTimeSeries(uuid, pair.left.getMeasurementName(), pair.left.getType().name(), senderEnv);
330330
createTimeSeries(

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

Lines changed: 91 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@
5454
import java.util.Map;
5555
import java.util.Objects;
5656
import java.util.function.BiConsumer;
57+
import java.util.function.IntPredicate;
5758
import java.util.stream.Collectors;
5859
import java.util.stream.IntStream;
5960

@@ -115,7 +116,9 @@ public void markAsNeedToReport() {
115116
//////////////////////////// parse ////////////////////////////
116117

117118
protected void parse(final InsertRowNode insertRowNode) {
118-
final int originColumnSize = insertRowNode.getMeasurements().length;
119+
final String[] originColumnNameStringList = insertRowNode.getMeasurements();
120+
final int originColumnSize =
121+
originColumnNameStringList == null ? 0 : originColumnNameStringList.length;
119122
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
120123

121124
// The full path is always cached when device path is deserialized
@@ -127,14 +130,25 @@ protected void parse(final InsertRowNode insertRowNode) {
127130
final List<Integer> rowIndexList = generateRowIndexList(originTimestampColumn);
128131
this.timestampColumn = rowIndexList.stream().mapToLong(i -> originTimestampColumn[i]).toArray();
129132

133+
final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas();
134+
final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories();
135+
final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes();
136+
final Object[] originValues = insertRowNode.getValues();
137+
130138
generateColumnIndexMapper(
131-
insertRowNode.getMeasurements(), originColumnIndex2FilteredColumnIndexMapperList);
139+
originColumnNameStringList, originColumnIndex2FilteredColumnIndexMapperList);
132140

133141
final int filteredColumnSize =
134-
Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
135-
.filter(Objects::nonNull)
136-
.toArray()
137-
.length;
142+
compactColumnIndexMapper(
143+
originColumnIndex2FilteredColumnIndexMapperList,
144+
i ->
145+
isValidOriginColumn(
146+
originColumnNameStringList,
147+
originMeasurementSchemaList,
148+
originValueDataTypes,
149+
i)
150+
&& originValues != null
151+
&& i < originValues.length);
138152

139153
this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
140154
this.columnNameStringList = new String[filteredColumnSize];
@@ -143,19 +157,15 @@ protected void parse(final InsertRowNode insertRowNode) {
143157
this.valueColumns = new Object[filteredColumnSize];
144158
this.nullValueColumnBitmaps = new BitMap[filteredColumnSize];
145159

146-
final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas();
147-
final String[] originColumnNameStringList = insertRowNode.getMeasurements();
148-
final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories();
149-
final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes();
150-
final Object[] originValues = insertRowNode.getValues();
151-
152160
for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
153161
if (originColumnIndex2FilteredColumnIndexMapperList[i] != null) {
154162
final int filteredColumnIndex = originColumnIndex2FilteredColumnIndexMapperList[i];
155163
this.measurementSchemaList[filteredColumnIndex] = originMeasurementSchemaList[i];
156164
this.columnNameStringList[filteredColumnIndex] = originColumnNameStringList[i];
157165
this.valueColumnTypes[filteredColumnIndex] =
158-
originColumnCategories != null && originColumnCategories[i] != null
166+
originColumnCategories != null
167+
&& i < originColumnCategories.length
168+
&& originColumnCategories[i] != null
159169
? originColumnCategories[i].toTsFileColumnType()
160170
: ColumnCategory.FIELD;
161171
this.valueColumnDataTypes[filteredColumnIndex] = originValueDataTypes[i];
@@ -190,7 +200,9 @@ protected void parse(final InsertRowNode insertRowNode) {
190200
}
191201

192202
protected void parse(final InsertTabletNode insertTabletNode) throws IllegalPathException {
193-
final int originColumnSize = insertTabletNode.getMeasurements().length;
203+
final String[] originColumnNameStringList = insertTabletNode.getMeasurements();
204+
final int originColumnSize =
205+
originColumnNameStringList == null ? 0 : originColumnNameStringList.length;
194206
final Integer[] originColumnIndex2FilteredColumnIndexMapperList = new Integer[originColumnSize];
195207

196208
// The full path is always cached when device path is deserialized
@@ -202,14 +214,28 @@ protected void parse(final InsertTabletNode insertTabletNode) throws IllegalPath
202214
final List<Integer> rowIndexList = generateRowIndexList(originTimestampColumn);
203215
this.timestampColumn = rowIndexList.stream().mapToLong(i -> originTimestampColumn[i]).toArray();
204216

217+
final MeasurementSchema[] originMeasurementSchemaList =
218+
insertTabletNode.getMeasurementSchemas();
219+
final TsTableColumnCategory[] originColumnCategories = insertTabletNode.getColumnCategories();
220+
final TSDataType[] originValueColumnDataTypes = insertTabletNode.getDataTypes();
221+
final Object[] originValueColumns = insertTabletNode.getColumns();
222+
final BitMap[] originBitMapList = insertTabletNode.getBitMaps();
223+
205224
generateColumnIndexMapper(
206-
insertTabletNode.getMeasurements(), originColumnIndex2FilteredColumnIndexMapperList);
225+
originColumnNameStringList, originColumnIndex2FilteredColumnIndexMapperList);
207226

208227
final int filteredColumnSize =
209-
Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
210-
.filter(Objects::nonNull)
211-
.toArray()
212-
.length;
228+
compactColumnIndexMapper(
229+
originColumnIndex2FilteredColumnIndexMapperList,
230+
i ->
231+
isValidOriginColumn(
232+
originColumnNameStringList,
233+
originMeasurementSchemaList,
234+
originValueColumnDataTypes,
235+
i)
236+
&& originValueColumns != null
237+
&& i < originValueColumns.length
238+
&& originValueColumns[i] != null);
213239

214240
this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
215241
this.columnNameStringList = new String[filteredColumnSize];
@@ -218,21 +244,15 @@ protected void parse(final InsertTabletNode insertTabletNode) throws IllegalPath
218244
this.valueColumns = new Object[filteredColumnSize];
219245
this.nullValueColumnBitmaps = new BitMap[filteredColumnSize];
220246

221-
final MeasurementSchema[] originMeasurementSchemaList =
222-
insertTabletNode.getMeasurementSchemas();
223-
final String[] originColumnNameStringList = insertTabletNode.getMeasurements();
224-
final TsTableColumnCategory[] originColumnCategories = insertTabletNode.getColumnCategories();
225-
final TSDataType[] originValueColumnDataTypes = insertTabletNode.getDataTypes();
226-
final Object[] originValueColumns = insertTabletNode.getColumns();
227-
final BitMap[] originBitMapList = insertTabletNode.getBitMaps();
228-
229247
for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
230248
if (originColumnIndex2FilteredColumnIndexMapperList[i] != null) {
231249
final int filteredColumnIndex = originColumnIndex2FilteredColumnIndexMapperList[i];
232250
this.measurementSchemaList[filteredColumnIndex] = originMeasurementSchemaList[i];
233251
this.columnNameStringList[filteredColumnIndex] = originColumnNameStringList[i];
234252
this.valueColumnTypes[filteredColumnIndex] =
235-
originColumnCategories != null && originColumnCategories[i] != null
253+
originColumnCategories != null
254+
&& i < originColumnCategories.length
255+
&& originColumnCategories[i] != null
236256
? originColumnCategories[i].toTsFileColumnType()
237257
: ColumnCategory.FIELD;
238258
this.valueColumnDataTypes[filteredColumnIndex] = originValueColumnDataTypes[i];
@@ -298,10 +318,9 @@ protected void parse(final Tablet tablet, final boolean isAligned) {
298318
originMeasurementList, originColumnIndex2FilteredColumnIndexMapperList);
299319

300320
final int filteredColumnSize =
301-
Arrays.stream(originColumnIndex2FilteredColumnIndexMapperList)
302-
.filter(Objects::nonNull)
303-
.toArray()
304-
.length;
321+
compactColumnIndexMapper(
322+
originColumnIndex2FilteredColumnIndexMapperList,
323+
i -> isValidOriginColumn(originMeasurementSchemaList, i));
305324

306325
this.measurementSchemaList = new MeasurementSchema[filteredColumnSize];
307326
this.columnNameStringList = new String[filteredColumnSize];
@@ -373,6 +392,46 @@ protected abstract void generateColumnIndexMapper(
373392
final String[] originMeasurementList,
374393
final Integer[] originColumnIndex2FilteredColumnIndexMapperList);
375394

395+
private static int compactColumnIndexMapper(
396+
final Integer[] originColumnIndex2FilteredColumnIndexMapperList,
397+
final IntPredicate columnValidator) {
398+
int filteredCount = 0;
399+
for (int i = 0; i < originColumnIndex2FilteredColumnIndexMapperList.length; i++) {
400+
if (originColumnIndex2FilteredColumnIndexMapperList[i] != null && columnValidator.test(i)) {
401+
originColumnIndex2FilteredColumnIndexMapperList[i] = filteredCount++;
402+
} else {
403+
originColumnIndex2FilteredColumnIndexMapperList[i] = null;
404+
}
405+
}
406+
return filteredCount;
407+
}
408+
409+
private static boolean isValidOriginColumn(
410+
final String[] originColumnNameStringList,
411+
final MeasurementSchema[] originMeasurementSchemaList,
412+
final TSDataType[] originValueDataTypes,
413+
final int index) {
414+
return originColumnNameStringList != null
415+
&& index < originColumnNameStringList.length
416+
&& originColumnNameStringList[index] != null
417+
&& originMeasurementSchemaList != null
418+
&& index < originMeasurementSchemaList.length
419+
&& originMeasurementSchemaList[index] != null
420+
&& originMeasurementSchemaList[index].getType() != null
421+
&& originValueDataTypes != null
422+
&& index < originValueDataTypes.length
423+
&& originValueDataTypes[index] != null;
424+
}
425+
426+
private static boolean isValidOriginColumn(
427+
final List<IMeasurementSchema> originMeasurementSchemaList, final int index) {
428+
return originMeasurementSchemaList != null
429+
&& index < originMeasurementSchemaList.size()
430+
&& originMeasurementSchemaList.get(index) != null
431+
&& originMeasurementSchemaList.get(index).getMeasurementName() != null
432+
&& originMeasurementSchemaList.get(index).getType() != null;
433+
}
434+
376435
private List<Integer> generateRowIndexList(final long[] originTimestampColumn) {
377436
final int rowCount = originTimestampColumn.length;
378437
if (Objects.isNull(sourceEvent) || !sourceEvent.shouldParseTime()) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/transform/statement/PipeConvertedInsertRowStatement.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,7 +107,13 @@ protected boolean checkAndCastDataType(int columnIndex, TSDataType dataType) {
107107

108108
@Override
109109
public void transferType(ZoneId zoneId) throws QueryProcessException {
110+
if (measurementSchemas == null) {
111+
return;
112+
}
110113
for (int i = 0; i < measurementSchemas.length; i++) {
114+
if (!isColumnPresent(i) || dataTypes == null || i >= dataTypes.length) {
115+
continue;
116+
}
111117
// null when time series doesn't exist
112118
if (measurementSchemas[i] == null) {
113119
if (!IoTDBDescriptor.getInstance().getConfig().isEnablePartialInsert()) {
@@ -126,6 +132,9 @@ public void transferType(ZoneId zoneId) throws QueryProcessException {
126132

127133
// parse string value to specific type
128134
dataTypes[i] = measurementSchemas[i].getType();
135+
if (values == null || i >= values.length || values[i] == null) {
136+
continue;
137+
}
129138
try {
130139
values[i] = ValueConverter.parse(values[i].toString(), dataTypes[i]);
131140
} catch (Exception e) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -548,6 +548,9 @@ private static long sizeOfBinary(final Binary binary) {
548548

549549
public static long sizeOfColumns(
550550
final Object[] columns, final MeasurementSchema[] measurementSchemas) {
551+
if (Objects.isNull(columns)) {
552+
return 0L;
553+
}
551554
// Directly calculate if measurementSchemas are absent
552555
if (Objects.isNull(measurementSchemas)) {
553556
return RamUsageEstimator.shallowSizeOf(columns)
@@ -559,7 +562,10 @@ public static long sizeOfColumns(
559562
RamUsageEstimator.alignObjectSize(
560563
NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF * columns.length);
561564
for (int i = 0; i < columns.length; i++) {
562-
if (measurementSchemas[i] == null || measurementSchemas[i].getType() == null) {
565+
if (columns[i] == null
566+
|| i >= measurementSchemas.length
567+
|| measurementSchemas[i] == null
568+
|| measurementSchemas[i].getType() == null) {
563569
continue;
564570
}
565571
switch (measurementSchemas[i].getType()) {
@@ -611,6 +617,9 @@ private static long getNumBytesUnknownObject(final Object obj) {
611617

612618
public static long sizeOfValues(
613619
final Object[] values, final MeasurementSchema[] measurementSchemas) {
620+
if (Objects.isNull(values)) {
621+
return 0L;
622+
}
614623
// Directly calculate if measurementSchemas are absent
615624
if (Objects.isNull(measurementSchemas)) {
616625
return RamUsageEstimator.shallowSizeOf(values)
@@ -622,7 +631,9 @@ public static long sizeOfValues(
622631
RamUsageEstimator.alignObjectSize(
623632
NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF * values.length);
624633
for (int i = 0; i < values.length; i++) {
625-
if (measurementSchemas[i] == null || measurementSchemas[i].getType() == null) {
634+
if (i >= measurementSchemas.length
635+
|| measurementSchemas[i] == null
636+
|| measurementSchemas[i].getType() == null) {
626637
size += NUM_BYTES_OBJECT_HEADER;
627638
continue;
628639
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/TabletStatementConverter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -395,7 +395,7 @@ private static Pair<Object[], Long> readValuesFromBufferWithMemory(
395395
for (int i = 0; i < columns; i++) {
396396
final boolean isValueColumnsNotNull =
397397
BytesUtils.byteToBool(ReadWriteIOUtils.readByte(byteBuffer));
398-
if (isValueColumnsNotNull && types[i] == null) {
398+
if (types[i] == null) {
399399
continue;
400400
}
401401

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/NormalSchemaFetcher.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -358,7 +358,9 @@ void processNormalTimeSeries(
358358
schemaComputationWithAutoCreationList.stream()
359359
.map(
360360
o -> {
361-
TSDataType[] dataTypes = new TSDataType[o.getMeasurements().length];
361+
final String[] measurements = o.getMeasurements();
362+
TSDataType[] dataTypes =
363+
new TSDataType[measurements == null ? 0 : measurements.length];
362364
for (int i = 0, length = dataTypes.length; i < length; i++) {
363365
dataTypes[i] = o.getDataType(i);
364366
}

0 commit comments

Comments
 (0)