This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b1e4862219 [codegen] Support normalized key for non-compact timestamp
(#8759)
b1e4862219 is described below
commit b1e48622194fa34d9353d461a8a31948e531cd9b
Author: Vova Kolmakov <[email protected]>
AuthorDate: Tue Jul 21 12:19:02 2026 +0700
[codegen] Support normalized key for non-compact timestamp (#8759)
---
.../apache/paimon/codegen/SortCodeGenerator.scala | 6 +-
.../java/org/apache/paimon/utils/SortUtil.java | 11 +-
.../paimon/codegen/NormalizedKeyComputerTest.java | 174 +++++++++++++++++++++
3 files changed, 185 insertions(+), 6 deletions(-)
diff --git
a/paimon-codegen/src/main/scala/org/apache/paimon/codegen/SortCodeGenerator.scala
b/paimon-codegen/src/main/scala/org/apache/paimon/codegen/SortCodeGenerator.scala
index 77040bd3d5..6ed773fb3b 100644
---
a/paimon-codegen/src/main/scala/org/apache/paimon/codegen/SortCodeGenerator.scala
+++
b/paimon-codegen/src/main/scala/org/apache/paimon/codegen/SortCodeGenerator.scala
@@ -410,9 +410,7 @@ class SortCodeGenerator(val input: RowType, val sortSpec:
SortSpec) {
t.getTypeRoot match {
case _ if TypeUtils.isPrimitive(t) => true
case VARCHAR | CHAR | VARBINARY | BINARY | DATE | TIME_WITHOUT_TIME_ZONE
=> true
- case TIMESTAMP_WITHOUT_TIME_ZONE =>
- // TODO: support normalize key for non-compact timestamp
- Timestamp.isCompact(t.asInstanceOf[TimestampType].getPrecision)
+ case TIMESTAMP_WITHOUT_TIME_ZONE => true
case DECIMAL =>
Decimal.isCompact(t.asInstanceOf[DecimalType].getPrecision)
case _ => false
}
@@ -430,6 +428,8 @@ class SortCodeGenerator(val input: RowType, val sortSpec:
SortSpec) {
case TIMESTAMP_WITHOUT_TIME_ZONE
if Timestamp.isCompact(t.asInstanceOf[TimestampType].getPrecision) =>
8
+ // non-compact timestamp: millisecond (8) + nanoOfMillisecond (4)
+ case TIMESTAMP_WITHOUT_TIME_ZONE => 12
case DATE => 4
case TIME_WITHOUT_TIME_ZONE => 4
case DECIMAL if
Decimal.isCompact(t.asInstanceOf[DecimalType].getPrecision) => 8
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/SortUtil.java
b/paimon-common/src/main/java/org/apache/paimon/utils/SortUtil.java
index b726532a31..5b1b6c4982 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/SortUtil.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/SortUtil.java
@@ -164,11 +164,16 @@ public class SortUtil {
}
}
- /** Support the compact precision TimestampData. */
+ /** Support TimestampData of any precision. */
public static void putTimestampNormalizedKey(
Timestamp value, MemorySegment target, int offset, int numBytes) {
- assert value.getNanoOfMillisecond() == 0;
- putLongNormalizedKey(value.getMillisecond(), target, offset, numBytes);
+ // millisecond occupies the most significant 8 bytes
+ putLongNormalizedKey(value.getMillisecond(), target, offset,
Math.min(numBytes, 8));
+ if (numBytes > 8) {
+ // nanoOfMillisecond is in [0, 999_999], so an unsigned key
preserves order
+ putUnsignedIntegerNormalizedKey(
+ value.getNanoOfMillisecond(), target, offset + 8, numBytes
- 8);
+ }
}
public static void putUnsignedIntegerNormalizedKey(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/codegen/NormalizedKeyComputerTest.java
b/paimon-core/src/test/java/org/apache/paimon/codegen/NormalizedKeyComputerTest.java
new file mode 100644
index 0000000000..770b19be57
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/codegen/NormalizedKeyComputerTest.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.codegen;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.data.serializer.AbstractRowDataSerializer;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.sort.BinaryInMemorySortBuffer;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.utils.MutableObjectIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.paimon.codegen.CodeGenUtils.newNormalizedKeyComputer;
+import static org.apache.paimon.codegen.CodeGenUtils.newRecordComparator;
+import static org.apache.paimon.types.DataTypes.TIMESTAMP;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests the {@link NormalizedKeyComputer} generated by {@link
SortCodeGenerator} for timestamps,
+ * covering non-compact precisions (sub-millisecond ordering).
+ */
+class NormalizedKeyComputerTest {
+
+ private static final int NON_COMPACT_PRECISION = 9;
+
+ @Test
+ public void testCompactTimestampKeyMetadata() {
+ NormalizedKeyComputer computer = timestampKeyComputer(3);
+ // 1 null-aware byte + 8 bytes millisecond
+ assertThat(computer.getNumKeyBytes()).isEqualTo(9);
+ assertThat(computer.isKeyFullyDetermines()).isTrue();
+ }
+
+ @Test
+ public void testNonCompactTimestampKeyMetadata() {
+ for (int precision : new int[] {4, 6, 9}) {
+ NormalizedKeyComputer computer = timestampKeyComputer(precision);
+ // 1 null-aware byte + 8 bytes millisecond + 4 bytes
nanoOfMillisecond
+ assertThat(computer.getNumKeyBytes()).as("precision %d",
precision).isEqualTo(13);
+ assertThat(computer.isKeyFullyDetermines()).as("precision %d",
precision).isTrue();
+ }
+ }
+
+ @Test
+ public void testSubMillisecondOrdering() {
+ NormalizedKeyComputer computer =
timestampKeyComputer(NON_COMPACT_PRECISION);
+ Timestamp lo = Timestamp.fromEpochMillis(1000, 111_111);
+ Timestamp hi = Timestamp.fromEpochMillis(1000, 222_222);
+
+ assertThat(normalizedCompare(computer, lo, hi)).isNegative();
+ assertThat(normalizedCompare(computer, hi, lo)).isPositive();
+ assertThat(normalizedCompare(computer, lo, lo)).isZero();
+ }
+
+ @Test
+ public void testCompareKeyMatchesCompareToForAllPairs() {
+ NormalizedKeyComputer computer =
timestampKeyComputer(NON_COMPACT_PRECISION);
+ List<Timestamp> values = timestampSpread();
+ for (Timestamp a : values) {
+ for (Timestamp b : values) {
+ assertThat(sign(normalizedCompare(computer, a, b)))
+ .as("compare(%s, %s)", a, b)
+ .isEqualTo(sign(a.compareTo(b)));
+ }
+ }
+ }
+
+ @Test
+ public void testEndToEndSortByNonCompactTimestamp() throws Exception {
+ List<Timestamp> values = new ArrayList<>();
+ for (Timestamp value : timestampSpread()) {
+ // duplicate each value so equal normalized keys are exercised
+ values.add(value);
+ values.add(value);
+ }
+ Collections.shuffle(values, new Random(42));
+
+ List<Timestamp> expected = new ArrayList<>(values);
+ expected.sort(Timestamp::compareTo);
+
+
assertThat(sortThroughBuffer(values)).containsExactlyElementsOf(expected);
+ }
+
+ private static NormalizedKeyComputer timestampKeyComputer(int precision) {
+ return newNormalizedKeyComputer(
+ Collections.singletonList(TIMESTAMP(precision)), new int[]
{0});
+ }
+
+ private static int normalizedCompare(NormalizedKeyComputer computer,
Timestamp a, Timestamp b) {
+ int numKeyBytes = computer.getNumKeyBytes();
+ MemorySegment segA = MemorySegment.wrap(new byte[numKeyBytes]);
+ MemorySegment segB = MemorySegment.wrap(new byte[numKeyBytes]);
+ computer.putKey(GenericRow.of(a), segA, 0);
+ computer.putKey(GenericRow.of(b), segB, 0);
+ return computer.compareKey(segA, 0, segB, 0);
+ }
+
+ private static List<Timestamp> timestampSpread() {
+ List<Timestamp> values = new ArrayList<>();
+ long[] milliseconds = {-86_400_000L, -1000L, -1L, 0L, 1L, 1000L,
1_600_000_000_000L};
+ int[] nanos = {0, 1, 500_000, 999_999};
+ for (long millisecond : milliseconds) {
+ for (int nano : nanos) {
+ values.add(Timestamp.fromEpochMillis(millisecond, nano));
+ }
+ }
+ return values;
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static List<Timestamp> sortThroughBuffer(List<Timestamp> values)
throws Exception {
+ List<DataType> fieldTypes =
Collections.singletonList(TIMESTAMP(NON_COMPACT_PRECISION));
+ BinaryRowSerializer serializer = new BinaryRowSerializer(1);
+ MemorySegmentPool pool =
+ new HeapMemorySegmentPool(1024 * 1024,
MemorySegmentPool.DEFAULT_PAGE_SIZE);
+ BinaryInMemorySortBuffer buffer =
+ BinaryInMemorySortBuffer.createBuffer(
+ newNormalizedKeyComputer(fieldTypes, new int[] {0}),
+ (AbstractRowDataSerializer) serializer,
+ newRecordComparator(fieldTypes, new int[] {0}),
+ pool);
+
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ for (Timestamp value : values) {
+ writer.reset();
+ writer.writeTimestamp(0, value, NON_COMPACT_PRECISION);
+ writer.complete();
+ assertThat(buffer.write(row)).isTrue();
+ }
+
+ List<Timestamp> sorted = new ArrayList<>();
+ MutableObjectIterator<BinaryRow> iterator = buffer.sortedIterator();
+ BinaryRow reuse = serializer.createInstance();
+ BinaryRow next;
+ while ((next = iterator.next(reuse)) != null) {
+ sorted.add(next.getTimestamp(0, NON_COMPACT_PRECISION));
+ }
+ buffer.clear();
+ return sorted;
+ }
+
+ private static int sign(int value) {
+ return Integer.compare(value, 0);
+ }
+}
