Skip to content

[dev/1.3] Batch aligned memtable column expansion to reduce region lock contention #18830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 3 commits into
base: dev/1.3
Choose a base branch
from
+410 −37
Open

[dev/1.3] Batch aligned memtable column expansion to reduce region lock contention #18830

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b73aae5
Batch aligned memtable column expansion to reduce region lock contention
Caideyipi Oct 10, 2026
d4b21c8
Keep aligned writes without schema expansion on a single pass
Caideyipi Oct 12, 2026
36efb53
Preserve implicit nulls and WAL serialization for lazy aligned columns
Caideyipi Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 25 additions & 14 deletions ...n/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1083,12 +1083,12 @@ public List<Integer> buildColumnIndexList(List<IMeasurementSchema> schemaList) {
}

/**
* Check metadata of columns and return array that mapping existed metadata to index of data
* column.
* Register missing measurements in insertion order, then reorder values and bitmaps into the
* memchunk's column order.
*
* @param schemaListInInsertPlan Contains all existed schema in InsertPlan. If some timeseries
* have been deleted, there will be null in its slot.
* @return columnIndexArray: schemaList[i] is schema of columns[columnIndexArray[i]]
* @return values and bitmaps aligned with this memchunk's schema list
*/
private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
List<IMeasurementSchema> schemaListInInsertPlan, Object[] columnValues, BitMap[] bitMaps) {
Expand All @@ -1097,20 +1097,16 @@ private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
for (int i = 0; i < schemaListInInsertPlan.size(); i++) {
IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i);
if (measurementSchema != null) {
Integer index = this.measurementIndexMap.get(measurementSchema.getMeasurementId());
// Index is null means this measurement was not in this AlignedTVList before.
// We need to extend a new column in AlignedMemChunk and AlignedTVList.
// And the reorderedColumnValues should extend one more column for the new measurement
Integer index = measurementIndexMap.get(measurementSchema.getMeasurementId());
if (index == null) {
index = this.list.getTsDataTypes().size();
this.measurementIndexMap.put(schemaListInInsertPlan.get(i).getMeasurementId(), index);
this.schemaList.add(schemaListInInsertPlan.get(i));
this.list.extendColumn(schemaListInInsertPlan.get(i).getType());
reorderedColumnValues =
Arrays.copyOf(reorderedColumnValues, reorderedColumnValues.length + 1);
// Keep writes without new measurements on a single-pass path. Only the first missing
// measurement triggers a batch extension and one resize of the reordered arrays.
extendColumnsInInsertPlan(schemaListInInsertPlan, i);
reorderedColumnValues = Arrays.copyOf(reorderedColumnValues, schemaList.size());
if (reorderedBitMaps != null) {
reorderedBitMaps = Arrays.copyOf(reorderedBitMaps, reorderedBitMaps.length + 1);
reorderedBitMaps = Arrays.copyOf(reorderedBitMaps, schemaList.size());
}
index = measurementIndexMap.get(measurementSchema.getMeasurementId());
}
reorderedColumnValues[index] = columnValues[i];
if (bitMaps != null) {
Expand All @@ -1121,6 +1117,21 @@ private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
return new Pair<>(reorderedColumnValues, reorderedBitMaps);
}

private void extendColumnsInInsertPlan(
List<IMeasurementSchema> schemaListInInsertPlan, int startIndex) {
List<TSDataType> newDataTypes = new ArrayList<>();
for (int i = startIndex; i < schemaListInInsertPlan.size(); i++) {
IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i);
if (measurementSchema != null
&& !measurementIndexMap.containsKey(measurementSchema.getMeasurementId())) {
measurementIndexMap.put(measurementSchema.getMeasurementId(), schemaList.size());
schemaList.add(measurementSchema);
newDataTypes.add(measurementSchema.getType());
}
}
list.extendColumns(newDataTypes);
}

public long[] getAnySatisfiedTimestamp(
List<List<TimeRange>> deletionList, List<BitMap> bitMaps, Filter globalTimeFilter) {
BitMap columnHasNonNullValue = new BitMap(schemaList.size());
Expand Down
67 changes: 44 additions & 23 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
Expand Down Expand Up @@ -497,22 +498,30 @@ private TsPrimitiveType getAlignedValueByValueIndex(
}

public void extendColumn(TSDataType dataType) {
List<Object> columnValue = new ArrayList<>();
for (int i = 0; i < timestamps.size(); i++) {
columnValue.add(null);
extendColumns(Collections.singletonList(dataType));
}

/** Add columns with implicit nulls for existing rows, resizing accounting arrays once. */
public void extendColumns(List<TSDataType> newDataTypes) {
if (newDataTypes.isEmpty()) {
return;
}
if (bitMaps != null) {
bitMaps.add(null);
for (TSDataType dataType : newDataTypes) {
List<Object> columnValue = new ArrayList<>(timestamps.size());
for (int i = 0; i < timestamps.size(); i++) {
// New columns contain only implicit nulls until their first non-null write.
columnValue.add(null);
}
if (bitMaps != null) {
bitMaps.add(null);
}
values.add(columnValue);
dataTypes.add(dataType);
}
this.values.add(columnValue);
this.dataTypes.add(dataType);
materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, dataTypes.size());
// Recompute the full-width memory estimate only once for the entire batch.
refreshArrayMemCostWithoutIndex();

long[] tmpValueChunkRawSize = memoryBinaryChunkSize;
memoryBinaryChunkSize = new long[dataTypes.size()];
System.arraycopy(
tmpValueChunkRawSize, 0, memoryBinaryChunkSize, 0, tmpValueChunkRawSize.length);
memoryBinaryChunkSize = Arrays.copyOf(memoryBinaryChunkSize, dataTypes.size());
materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, dataTypes.size());
}

/**
Expand Down Expand Up @@ -617,12 +626,16 @@ public boolean isNullValue(int unsortedRowIndex, int columnIndex) {
if (columnIndex < 0 || columnIndex >= values.size() || values.get(columnIndex) == null) {
return true;
}
int arrayIndex = unsortedRowIndex / ARRAY_SIZE;
// A lazy value block contains only implicit nulls, even when a bitmap already exists.
if (values.get(columnIndex).get(arrayIndex) == null) {
return true;
}
if (bitMaps == null
|| bitMaps.get(columnIndex) == null
|| bitMaps.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) == null) {
return values.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) == null;
|| bitMaps.get(columnIndex).get(arrayIndex) == null) {
return false;
}
int arrayIndex = unsortedRowIndex / ARRAY_SIZE;
int elementIndex = unsortedRowIndex % ARRAY_SIZE;
List<BitMap> columnBitMaps = bitMaps.get(columnIndex);
return columnBitMaps.get(arrayIndex).isMarked(elementIndex);
Expand Down Expand Up @@ -1748,8 +1761,14 @@ public int serializedSize() {
case TEXT:
case BLOB:
case STRING:
List<Object> columnValues = values.get(columnIndex);
for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
size += ReadWriteIOUtils.sizeToWrite(getBinaryByValueIndex(rowIdx, columnIndex));
Binary value =
columnValues == null || columnValues.get(rowIdx / ARRAY_SIZE) == null
? null
: getBinaryByValueIndex(rowIdx, columnIndex);
// Null blocks and null Binary entries use an empty Binary placeholder in the WAL.
size += value == null ? Integer.BYTES : ReadWriteIOUtils.sizeToWrite(value);
}
break;
case FLOAT:
Expand Down Expand Up @@ -1796,12 +1815,14 @@ public void serializeToWAL(IWALByteBufferView buffer) {
for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) {
int arrayIndex = rowIndex / ARRAY_SIZE;
int elementIndex = rowIndex % ARRAY_SIZE;
// Keep the WAL layout unchanged without materializing lazy all-null value blocks.
Object valueArray = columnValues == null ? null : columnValues.get(arrayIndex);
// value
switch (dataTypes.get(columnIndex)) {
case TEXT:
case BLOB:
case STRING:
Binary valueT = ((Binary[]) columnValues.get(arrayIndex))[elementIndex];
Binary valueT = valueArray == null ? null : ((Binary[]) valueArray)[elementIndex];
// In some scenario, the Binary in AlignedTVList will be null if this field is empty in
// current row. We need to handle this scenario to get rid of NPE. See the similar issue
// here: https://github.com/apache/iotdb/pull/9884
Expand All @@ -1815,25 +1836,25 @@ public void serializeToWAL(IWALByteBufferView buffer) {
}
break;
case FLOAT:
float valueF = ((float[]) columnValues.get(arrayIndex))[elementIndex];
float valueF = valueArray == null ? 0 : ((float[]) valueArray)[elementIndex];
buffer.putFloat(valueF);
break;
case INT32:
case DATE:
int valueI = ((int[]) columnValues.get(arrayIndex))[elementIndex];
int valueI = valueArray == null ? 0 : ((int[]) valueArray)[elementIndex];
buffer.putInt(valueI);
break;
case INT64:
case TIMESTAMP:
long valueL = ((long[]) columnValues.get(arrayIndex))[elementIndex];
long valueL = valueArray == null ? 0 : ((long[]) valueArray)[elementIndex];
buffer.putLong(valueL);
break;
case DOUBLE:
double valueD = ((double[]) columnValues.get(arrayIndex))[elementIndex];
double valueD = valueArray == null ? 0 : ((double[]) valueArray)[elementIndex];
buffer.putDouble(valueD);
break;
case BOOLEAN:
boolean valueB = ((boolean[]) columnValues.get(arrayIndex))[elementIndex];
boolean valueB = valueArray != null && ((boolean[]) valueArray)[elementIndex];
WALWriteUtils.write(valueB, buffer);
break;
default:
Expand Down
Loading
Loading