Skip to content

[opt](paimon) Reduce JNI read object allocations #66244

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
Gabriel39 wants to merge 2 commits into
base: master
Choose a base branch
from
+354 −23
Open

[opt](paimon) Reduce JNI read object allocations #66244

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
48a06b0
[opt](paimon) Reduce JNI read object allocations
Gabriel39 Jul 29, 2026
4459f2e
[fix](paimon) Preserve timezone aliases and bound nested caches
Gabriel39 Jul 30, 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
131 changes: 111 additions & 20 deletions ...va-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,20 +38,47 @@
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class PaimonColumnValue implements ColumnValue {
private static final Logger LOG = LoggerFactory.getLogger(PaimonColumnValue.class);
private static final Map<String, String> DORIS_TIME_ZONE_ALIASES;

static {
Map<String, String> aliases = new HashMap<>(ZoneId.SHORT_IDS);
// The scanner cannot depend on FE's TimeUtils, so keep its accepted aliases and CST
// interpretation identical at this JNI boundary.
aliases.put("CST", "Asia/Shanghai");
aliases.put("PRC", "Asia/Shanghai");
aliases.put("UTC", "UTC");
aliases.put("GMT", "UTC");
DORIS_TIME_ZONE_ALIASES = Collections.unmodifiableMap(aliases);
}

private int idx;
private DataGetters record;
private ColumnType dorisType;
private DataType dataType;
private String timeZone;
private ZoneId timeZone;
// Keep these caches lazy so scalar columns do not pay for complex-type reuse bookkeeping.
private List<PaimonColumnValue> arrayValues;
private List<PaimonColumnValue> mapKeys;

@github-actions github-actions Bot Jul 30, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These caches belong to the scanner-wide root value, which PaimonJniScanner reassigns to every projected field. For stable a ARRAY<INT> of size N followed by empty b ARRAY<INT>, b trims this same arrayValues to zero, so the next row recreates all N wrappers for a; MAP has the same issue. A later null scalar is worse: isNull() clears all four cache families. Thus ordinary multi-column scans can still allocate O(rows * elements) wrappers despite stable shapes. Please scope the root cache graph per projected column (while keeping recursive pruning within each graph) and add mixed-column tests covering different collection sizes and a later null scalar.

private List<PaimonColumnValue> mapValues;
private List<PaimonColumnValue> structValues;

public PaimonColumnValue() {
}

public PaimonColumnValue(DataGetters record, int idx, ColumnType columnType, DataType dataType, String timeZone) {
this(record, idx, columnType, dataType, resolveTimeZone(timeZone));
}

private PaimonColumnValue(
DataGetters record, int idx, ColumnType columnType, DataType dataType, ZoneId timeZone) {
this.idx = idx;
this.record = record;
this.dorisType = columnType;
Expand All @@ -70,7 +97,7 @@ public void setOffsetRow(InternalRow record) {
}

public void setTimeZone(String timeZone) {
this.timeZone = timeZone;
this.timeZone = resolveTimeZone(timeZone);
}

@Override
Expand Down Expand Up @@ -142,8 +169,8 @@ public LocalDate getDate() {
public LocalDateTime getDateTime() {
Timestamp ts = record.getTimestamp(idx, dorisType.getPrecision());
if (dataType instanceof LocalZonedTimestampType) {
return ts.toLocalDateTime().atZone(ZoneId.of("UTC"))
.withZoneSameInstant(ZoneId.of(timeZone)).toLocalDateTime();
// Paimon stores TIMESTAMP_LTZ as an epoch instant, so convert it directly in the cached session zone.
return LocalDateTime.ofInstant(ts.toInstant(), timeZone);
} else {
return ts.toLocalDateTime();
}
Expand All @@ -152,15 +179,18 @@ public LocalDateTime getDateTime() {
@Override
public LocalDateTime getTimeStampTz() {
Timestamp ts = record.getTimestamp(idx, dorisType.getPrecision());
LocalDateTime v = ts.toInstant()
.atZone(ZoneId.of("UTC"))
.toLocalDateTime();
return v;
// Timestamp's local representation is identical to converting its epoch instant in UTC.
return ts.toLocalDateTime();
}

@Override
public boolean isNull() {
return record.isNullAt(idx);
boolean isNull = record.isNullAt(idx);
if (isNull) {
// A null complex value has no live descendants; release wrappers retained by its prior row.
clearChildCaches();
}
return isNull;
}

@Override
Expand All @@ -171,38 +201,99 @@ public byte[] getBytes() {
@Override
public void unpackArray(List<ColumnValue> values) {
InternalArray recordArray = record.getArray(idx);
if (arrayValues == null) {
arrayValues = new ArrayList<>();
}
ColumnType elementDorisType = dorisType.getChildTypes().get(0);
DataType elementPaimonType = ((ArrayType) dataType).getElementType();
for (int i = 0; i < recordArray.size(); i++) {
PaimonColumnValue arrayColumnValue = new PaimonColumnValue((DataGetters) recordArray, i,
dorisType.getChildTypes().get(0), ((ArrayType) dataType).getElementType(), timeZone);
values.add(arrayColumnValue);
values.add(reuseColumnValue(arrayValues, i, (DataGetters) recordArray, i,
elementDorisType, elementPaimonType));
}
trimCache(arrayValues, recordArray.size());
}

@Override
public void unpackMap(List<ColumnValue> keys, List<ColumnValue> values) {
InternalMap map = record.getMap(idx);
if (mapKeys == null) {
mapKeys = new ArrayList<>();
mapValues = new ArrayList<>();
}
InternalArray key = map.keyArray();
ColumnType keyDorisType = dorisType.getChildTypes().get(0);
DataType keyPaimonType = ((MapType) dataType).getKeyType();
for (int i = 0; i < key.size(); i++) {
PaimonColumnValue keyColumnValue = new PaimonColumnValue((DataGetters) key, i,
dorisType.getChildTypes().get(0), ((MapType) dataType).getKeyType(), timeZone);
keys.add(keyColumnValue);
keys.add(reuseColumnValue(mapKeys, i, (DataGetters) key, i,
keyDorisType, keyPaimonType));
}
trimCache(mapKeys, key.size());
InternalArray value = map.valueArray();
ColumnType valueDorisType = dorisType.getChildTypes().get(1);
DataType valuePaimonType = ((MapType) dataType).getValueType();
for (int i = 0; i < value.size(); i++) {
PaimonColumnValue valueColumnValue = new PaimonColumnValue((DataGetters) value, i,
dorisType.getChildTypes().get(1), ((MapType) dataType).getValueType(), timeZone);
values.add(valueColumnValue);
values.add(reuseColumnValue(mapValues, i, (DataGetters) value, i,
valueDorisType, valuePaimonType));
}
trimCache(mapValues, value.size());
}

@Override
public void unpackStruct(List<Integer> structFieldIndex, List<ColumnValue> values) {
RowType rowType = (RowType) dataType;
// Projection entries are original child indexes, so the binary row must keep the full RowType arity.
InternalRow row = record.getRow(idx, rowType.getFieldCount());
if (structValues == null) {
structValues = new ArrayList<>();
}
for (int i : structFieldIndex) {
values.add(new PaimonColumnValue(row, i, dorisType.getChildTypes().get(i),
rowType.getFields().get(i).type(), timeZone));
values.add(reuseColumnValue(structValues, i, row, i, dorisType.getChildTypes().get(i),
rowType.getFields().get(i).type()));
}
}

private PaimonColumnValue reuseColumnValue(
List<PaimonColumnValue> cache, int cacheIndex, DataGetters childRecord, int childIndex,
ColumnType childDorisType, DataType childPaimonType) {
while (cache.size() <= cacheIndex) {
cache.add(null);
Comment thread
Gabriel39 marked this conversation as resolved.
}
PaimonColumnValue value = cache.get(cacheIndex);
if (value == null) {
value = new PaimonColumnValue(
childRecord, childIndex, childDorisType, childPaimonType, timeZone);
cache.set(cacheIndex, value);
} else {
// VectorColumn consumes unpacked values synchronously, before this parent advances to another value.
value.reset(childRecord, childIndex, childDorisType, childPaimonType, timeZone);
}
return value;
}

private void reset(
DataGetters record, int idx, ColumnType dorisType, DataType dataType, ZoneId timeZone) {
this.record = record;
this.idx = idx;
this.dorisType = dorisType;
this.dataType = dataType;
this.timeZone = timeZone;
}

private static ZoneId resolveTimeZone(String timeZone) {
return ZoneId.of(timeZone, DORIS_TIME_ZONE_ALIASES);
}

private static void trimCache(List<PaimonColumnValue> cache, int liveSize) {
if (cache.size() > liveSize) {
// Retain only wrappers addressable by the current container, not its historical maximum.
cache.subList(liveSize, cache.size()).clear();
}
}

private void clearChildCaches() {
arrayValues = null;
mapKeys = null;
mapValues = null;
structValues = null;
}
}
Loading
Loading