diff --git a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java index f60bb996703ec8..339feb97c9b9f1 100644 --- a/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java +++ b/fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonColumnValue.java @@ -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 DORIS_TIME_ZONE_ALIASES; + + static { + Map 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 arrayValues; + private List mapKeys; + private List mapValues; + private List 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; @@ -70,7 +97,7 @@ public void setOffsetRow(InternalRow record) { } public void setTimeZone(String timeZone) { - this.timeZone = timeZone; + this.timeZone = resolveTimeZone(timeZone); } @Override @@ -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(); } @@ -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 @@ -171,28 +201,41 @@ public byte[] getBytes() { @Override public void unpackArray(List 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 keys, List 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 @@ -200,9 +243,57 @@ public void unpackStruct(List structFieldIndex, List value 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 cache, int cacheIndex, DataGetters childRecord, int childIndex, + ColumnType childDorisType, DataType childPaimonType) { + while (cache.size() <= cacheIndex) { + cache.add(null); + } + 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 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; } } diff --git a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java index 639a2e87f30c18..20062ff2f6e58a 100644 --- a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java +++ b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonColumnValueTest.java @@ -21,21 +21,33 @@ import org.apache.doris.common.jni.vec.ColumnValue; import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericMap; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.Timestamp; import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.types.ArrayType; import org.apache.paimon.types.BigIntType; import org.apache.paimon.types.DataType; import org.apache.paimon.types.IntType; +import org.apache.paimon.types.LocalZonedTimestampType; +import org.apache.paimon.types.MapType; import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; import org.apache.paimon.types.VarCharType; import org.junit.Assert; import org.junit.Test; +import java.lang.reflect.Field; +import java.time.Instant; +import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; public class PaimonColumnValueTest { private final RowType paimonStructType = RowType.of( @@ -80,10 +92,228 @@ public void testUnpackCompleteStruct() { Assert.assertEquals(100L, values.get(2).getLong()); } + @Test + public void testReuseArrayElementsAcrossRows() { + ArrayType paimonArrayType = new ArrayType(new IntType()); + ColumnType dorisArrayType = ColumnType.parseType("a", "array"); + PaimonColumnValue arrayValue = new PaimonColumnValue( + GenericRow.of(new GenericArray(new int[] {1, 2})), 0, + dorisArrayType, paimonArrayType, "UTC"); + + List firstValues = new ArrayList<>(); + arrayValue.unpackArray(firstValues); + arrayValue.setOffsetRow(GenericRow.of(new GenericArray(new int[] {3, 4, 5}))); + List secondValues = new ArrayList<>(); + arrayValue.unpackArray(secondValues); + + Assert.assertSame(firstValues.get(0), secondValues.get(0)); + Assert.assertSame(firstValues.get(1), secondValues.get(1)); + Assert.assertEquals(Arrays.asList(3, 4, 5), getInts(secondValues)); + + arrayValue.setOffsetRow(GenericRow.of(new GenericArray(new int[] {6}))); + List thirdValues = new ArrayList<>(); + arrayValue.unpackArray(thirdValues); + Assert.assertSame(firstValues.get(0), thirdValues.get(0)); + Assert.assertEquals(Collections.singletonList(6), getInts(thirdValues)); + } + + @Test + public void testReuseMapEntriesAcrossRows() { + MapType paimonMapType = new MapType(new IntType(), new BigIntType()); + ColumnType dorisMapType = ColumnType.parseType("m", "map"); + PaimonColumnValue mapValue = new PaimonColumnValue( + GenericRow.of(new GenericMap(linkedMap(1, 10L, 2, 20L))), 0, + dorisMapType, paimonMapType, "UTC"); + + List firstKeys = new ArrayList<>(); + List firstValues = new ArrayList<>(); + mapValue.unpackMap(firstKeys, firstValues); + mapValue.setOffsetRow(GenericRow.of(new GenericMap(linkedMap(3, 30L, 4, 40L)))); + List secondKeys = new ArrayList<>(); + List secondValues = new ArrayList<>(); + mapValue.unpackMap(secondKeys, secondValues); + + Assert.assertSame(firstKeys.get(0), secondKeys.get(0)); + Assert.assertSame(firstKeys.get(1), secondKeys.get(1)); + Assert.assertSame(firstValues.get(0), secondValues.get(0)); + Assert.assertSame(firstValues.get(1), secondValues.get(1)); + Assert.assertEquals(Arrays.asList(3, 4), getInts(secondKeys)); + Assert.assertEquals(Arrays.asList(30L, 40L), getLongs(secondValues)); + } + + @Test + public void testReuseProjectedStructFieldsAcrossRows() { + PaimonColumnValue structValue = createStructValue(); + List projectedFields = Arrays.asList(0, 2); + List firstValues = new ArrayList<>(); + structValue.unpackStruct(projectedFields, firstValues); + + structValue.setOffsetRow(createStructRow(20, "y", 200L)); + List secondValues = new ArrayList<>(); + structValue.unpackStruct(projectedFields, secondValues); + + Assert.assertSame(firstValues.get(0), secondValues.get(0)); + Assert.assertSame(firstValues.get(1), secondValues.get(1)); + Assert.assertEquals(20, secondValues.get(0).getInt()); + Assert.assertEquals(200L, secondValues.get(1).getLong()); + } + + @Test + public void testReuseNestedArrayElementsAcrossRows() { + ArrayType paimonNestedArrayType = new ArrayType(new ArrayType(new IntType())); + ColumnType dorisNestedArrayType = ColumnType.parseType("a", "array>"); + PaimonColumnValue arrayValue = new PaimonColumnValue( + GenericRow.of(new GenericArray(new Object[] {new GenericArray(new int[] {1, 2})})), 0, + dorisNestedArrayType, paimonNestedArrayType, "UTC"); + + List firstOuterValues = new ArrayList<>(); + arrayValue.unpackArray(firstOuterValues); + List firstInnerValues = new ArrayList<>(); + firstOuterValues.get(0).unpackArray(firstInnerValues); + + arrayValue.setOffsetRow( + GenericRow.of(new GenericArray(new Object[] {new GenericArray(new int[] {3, 4})}))); + List secondOuterValues = new ArrayList<>(); + arrayValue.unpackArray(secondOuterValues); + List secondInnerValues = new ArrayList<>(); + secondOuterValues.get(0).unpackArray(secondInnerValues); + + Assert.assertSame(firstOuterValues.get(0), secondOuterValues.get(0)); + Assert.assertSame(firstInnerValues.get(0), secondInnerValues.get(0)); + Assert.assertSame(firstInnerValues.get(1), secondInnerValues.get(1)); + Assert.assertEquals(Arrays.asList(3, 4), getInts(secondInnerValues)); + } + + @Test + public void testNestedArrayCacheRetainsOnlyCurrentShape() throws Exception { + int outerSize = 4; + int innerSize = 32; + ArrayType paimonNestedArrayType = new ArrayType(new ArrayType(new IntType())); + ColumnType dorisNestedArrayType = ColumnType.parseType("a", "array>"); + PaimonColumnValue arrayValue = new PaimonColumnValue( + nestedArrayRow(outerSize, innerSize, 0, -1), 0, + dorisNestedArrayType, paimonNestedArrayType, "UTC"); + + consumeNestedArray(arrayValue); + Assert.assertEquals(outerSize + innerSize, retainedCachedValues(arrayValue)); + + // Moving the large child to position 1 makes position 0 a smaller, non-null array. + arrayValue.setOffsetRow(nestedArrayRow(outerSize, innerSize, 1, -1)); + consumeNestedArray(arrayValue); + Assert.assertEquals(outerSize + innerSize, retainedCachedValues(arrayValue)); + + // Moving it again makes position 1 null, which must release that position's descendants. + arrayValue.setOffsetRow(nestedArrayRow(outerSize, innerSize, 2, 1)); + consumeNestedArray(arrayValue); + Assert.assertEquals(outerSize + innerSize, retainedCachedValues(arrayValue)); + } + + @Test + public void testTimestampConversionsPreserveBoundaryValues() { + Timestamp timestamp = Timestamp.fromEpochMillis(-1, 999_999); + ColumnType dorisTimestampType = ColumnType.parseType("t", "datetimev2(9)"); + + PaimonColumnValue timestampValue = new PaimonColumnValue( + GenericRow.of(timestamp), 0, dorisTimestampType, new TimestampType(9), "UTC"); + Assert.assertEquals( + LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999_999_999), + timestampValue.getDateTime()); + Assert.assertEquals( + LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999_999_999), + timestampValue.getTimeStampTz()); + + PaimonColumnValue localZonedValue = new PaimonColumnValue( + GenericRow.of(Timestamp.fromInstant(Instant.parse("2024-03-10T10:30:00.123456789Z"))), 0, + dorisTimestampType, new LocalZonedTimestampType(9), "America/Los_Angeles"); + Assert.assertEquals( + LocalDateTime.of(2024, 3, 10, 3, 30, 0, 123_456_789), + localZonedValue.getDateTime()); + + localZonedValue.setTimeZone("Asia/Shanghai"); + Assert.assertEquals( + LocalDateTime.of(2024, 3, 10, 18, 30, 0, 123_456_789), + localZonedValue.getDateTime()); + + localZonedValue.setTimeZone("CST"); + Assert.assertEquals( + LocalDateTime.of(2024, 3, 10, 18, 30, 0, 123_456_789), + localZonedValue.getDateTime()); + } + + private InternalRow nestedArrayRow(int outerSize, int innerSize, int populatedIndex, int nullIndex) { + Object[] outerValues = new Object[outerSize]; + for (int i = 0; i < outerSize; i++) { + if (i == nullIndex) { + outerValues[i] = null; + } else if (i == populatedIndex) { + outerValues[i] = new GenericArray(new int[innerSize]); + } else { + outerValues[i] = new GenericArray(new int[0]); + } + } + return GenericRow.of(new GenericArray(outerValues)); + } + + private void consumeNestedArray(PaimonColumnValue arrayValue) { + List outerValues = new ArrayList<>(); + arrayValue.unpackArray(outerValues); + for (ColumnValue outerValue : outerValues) { + if (!outerValue.isNull()) { + outerValue.unpackArray(new ArrayList<>()); + } + } + } + + private int retainedCachedValues(PaimonColumnValue value) throws Exception { + int retained = 0; + for (String fieldName : Arrays.asList("arrayValues", "mapKeys", "mapValues", "structValues")) { + Field field = PaimonColumnValue.class.getDeclaredField(fieldName); + field.setAccessible(true); + List children = (List) field.get(value); + if (children == null) { + continue; + } + for (Object child : children) { + if (child != null) { + retained++; + retained += retainedCachedValues((PaimonColumnValue) child); + } + } + } + return retained; + } + private PaimonColumnValue createStructValue() { - GenericRow nestedRow = GenericRow.of(10, BinaryString.fromString("x"), 100L); + return new PaimonColumnValue(createStructRow(10, "x", 100L), 0, + dorisStructType, paimonStructType, "UTC"); + } + + private InternalRow createStructRow(int intValue, String stringValue, long longValue) { + GenericRow nestedRow = GenericRow.of(intValue, BinaryString.fromString(stringValue), longValue); RowType outerType = RowType.of(new DataType[] {paimonStructType}, new String[] {"s"}); - InternalRow outerRow = new InternalRowSerializer(outerType).toBinaryRow(GenericRow.of(nestedRow)); - return new PaimonColumnValue(outerRow, 0, dorisStructType, paimonStructType, "UTC"); + return new InternalRowSerializer(outerType).toBinaryRow(GenericRow.of(nestedRow)); + } + + private List getInts(List values) { + List result = new ArrayList<>(); + for (ColumnValue value : values) { + result.add(value.getInt()); + } + return result; + } + + private List getLongs(List values) { + List result = new ArrayList<>(); + for (ColumnValue value : values) { + result.add(value.getLong()); + } + return result; + } + + private Map linkedMap(int key1, long value1, int key2, long value2) { + Map result = new LinkedHashMap<>(); + result.put(key1, value1); + result.put(key2, value2); + return result; } } diff --git a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java index d43f251f58deb1..fd7d6958dd3ae1 100644 --- a/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java +++ b/fe/be-java-extensions/paimon-scanner/src/test/java/org/apache/doris/paimon/PaimonJniScannerTest.java @@ -52,6 +52,16 @@ public void testConstructorAcceptsEmptyProjection() { new PaimonJniScanner(128, createBaseParams()); } + @Test + public void testConstructorAcceptsDorisShortTimeZoneForNonTemporalProjection() { + Map params = createBaseParams(); + params.put("required_fields", "id"); + params.put("columns_types", "int"); + params.put("time_zone", "EST"); + + new PaimonJniScanner(128, params); + } + @Test public void testIOManagerOptionHelpers() throws Exception { Map params = createBaseParams();