Skip to content

Commit 47f6265

Browse files
JingsongLiJingsongLi
committed
[core] Introduce manifest entry grouped by partition and bucket
1 parent fe7f16e commit 47f6265

File tree

24 files changed

+943
-145
lines changed

24 files changed

+943
-145
lines changed

‎paimon-api/src/main/java/org/apache/paimon/utils/MathUtils.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818

1919
package org.apache.paimon.utils;
2020

21+
import static org.apache.paimon.utils.Preconditions.checkArgument;
22+
2123
/** Collection of simple mathematical routines. */
2224
public class MathUtils {
2325

@@ -32,6 +34,21 @@ public static int roundDownToPowerOf2(int value) {
3234
return Integer.highestOneBit(value);
3335
}
3436

37+
/**
38+
* Increments the given number up to the closest power of two. If the argument is a power of
39+
* two, it remains unchanged.
40+
*
41+
* @param value The value to round up.
42+
* @return The closest value that is a power of two and large or equal than the given value.
43+
*/
44+
public static int roundUpToPowerOf2(int value) {
45+
checkArgument(value > 0, "The given value must be larger than 0.");
46+
if (value == 1) {
47+
return 2;
48+
}
49+
return Integer.highestOneBit((value - 1) << 1);
50+
}
51+
3552
/**
3653
* Checks whether the given value is a power of two.
3754
*

‎paimon-api/src/test/java/org/apache/paimon/utils/MathUtilsTest.java‎

Lines changed: 191 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,191 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.utils;
20+
21+
import org.junit.jupiter.api.Test;
22+
23+
import static org.assertj.core.api.Assertions.assertThat;
24+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
25+
26+
/** Tests for {@link MathUtils}. */
27+
class MathUtilsTest {
28+
29+
@Test
30+
void testRoundDownToPowerOf2() {
31+
assertThat(MathUtils.roundDownToPowerOf2(0)).isZero();
32+
assertThat(MathUtils.roundDownToPowerOf2(1)).isEqualTo(1);
33+
assertThat(MathUtils.roundDownToPowerOf2(2)).isEqualTo(2);
34+
assertThat(MathUtils.roundDownToPowerOf2(3)).isEqualTo(2);
35+
assertThat(MathUtils.roundDownToPowerOf2(7)).isEqualTo(4);
36+
assertThat(MathUtils.roundDownToPowerOf2(8)).isEqualTo(8);
37+
assertThat(MathUtils.roundDownToPowerOf2(9)).isEqualTo(8);
38+
assertThat(MathUtils.roundDownToPowerOf2(65535)).isEqualTo(32768);
39+
assertThat(MathUtils.roundDownToPowerOf2(65536)).isEqualTo(65536);
40+
assertThat(MathUtils.roundDownToPowerOf2(Integer.MAX_VALUE)).isEqualTo(1 << 30);
41+
}
42+
43+
@Test
44+
void testRoundUpToPowerOf2() {
45+
assertThat(MathUtils.roundUpToPowerOf2(1)).isEqualTo(2);
46+
assertThat(MathUtils.roundUpToPowerOf2(2)).isEqualTo(2);
47+
assertThat(MathUtils.roundUpToPowerOf2(3)).isEqualTo(4);
48+
assertThat(MathUtils.roundUpToPowerOf2(7)).isEqualTo(8);
49+
assertThat(MathUtils.roundUpToPowerOf2(8)).isEqualTo(8);
50+
assertThat(MathUtils.roundUpToPowerOf2(9)).isEqualTo(16);
51+
assertThat(MathUtils.roundUpToPowerOf2(65535)).isEqualTo(65536);
52+
assertThat(MathUtils.roundUpToPowerOf2(65536)).isEqualTo(65536);
53+
assertThat(MathUtils.roundUpToPowerOf2(1 << 30)).isEqualTo(1 << 30);
54+
assertThat(MathUtils.roundUpToPowerOf2((1 << 30) - 1)).isEqualTo(1 << 30);
55+
}
56+
57+
@Test
58+
void testIsPowerOf2() {
59+
assertThat(MathUtils.isPowerOf2(1L)).isTrue();
60+
assertThat(MathUtils.isPowerOf2(2L)).isTrue();
61+
assertThat(MathUtils.isPowerOf2(4L)).isTrue();
62+
assertThat(MathUtils.isPowerOf2(1024L)).isTrue();
63+
assertThat(MathUtils.isPowerOf2(1L << 30)).isTrue();
64+
assertThat(MathUtils.isPowerOf2(1L << 62)).isTrue();
65+
66+
// The current implementation considers 0 a power of 2, which is unconventional.
67+
assertThat(MathUtils.isPowerOf2(0L)).isTrue();
68+
69+
assertThat(MathUtils.isPowerOf2(3L)).isFalse();
70+
assertThat(MathUtils.isPowerOf2(5L)).isFalse();
71+
assertThat(MathUtils.isPowerOf2(100L)).isFalse();
72+
assertThat(MathUtils.isPowerOf2(-1L)).isFalse();
73+
assertThat(MathUtils.isPowerOf2(-2L)).isFalse();
74+
assertThat(MathUtils.isPowerOf2(-4L)).isFalse();
75+
76+
// Long.MIN_VALUE is -2^63, which is a power of 2 in two's complement.
77+
assertThat(MathUtils.isPowerOf2(Long.MIN_VALUE)).isTrue();
78+
assertThat(MathUtils.isPowerOf2(Long.MAX_VALUE)).isFalse();
79+
}
80+
81+
@Test
82+
void testLog2strict() {
83+
assertThat(MathUtils.log2strict(1)).isZero();
84+
assertThat(MathUtils.log2strict(2)).isEqualTo(1);
85+
assertThat(MathUtils.log2strict(8)).isEqualTo(3);
86+
assertThat(MathUtils.log2strict(1024)).isEqualTo(10);
87+
assertThat(MathUtils.log2strict(1 << 30)).isEqualTo(30);
88+
89+
assertThatThrownBy(() -> MathUtils.log2strict(0))
90+
.isInstanceOf(ArithmeticException.class)
91+
.hasMessage("Logarithm of zero is undefined.");
92+
93+
assertThatThrownBy(() -> MathUtils.log2strict(3))
94+
.isInstanceOf(IllegalArgumentException.class)
95+
.hasMessage("The given value 3 is not a power of two.");
96+
97+
assertThatThrownBy(() -> MathUtils.log2strict(5))
98+
.isInstanceOf(IllegalArgumentException.class)
99+
.hasMessage("The given value 5 is not a power of two.");
100+
101+
assertThatThrownBy(() -> MathUtils.log2strict(-2))
102+
.isInstanceOf(IllegalArgumentException.class)
103+
.hasMessage("The given value -2 is not a power of two.");
104+
}
105+
106+
@Test
107+
void testMax() {
108+
assertThat(MathUtils.max(1, 5)).isEqualTo(5);
109+
assertThat(MathUtils.max(5, 1)).isEqualTo(5);
110+
assertThat(MathUtils.max(5, 5)).isEqualTo(5);
111+
assertThat(MathUtils.max(-1, -5)).isEqualTo(-1);
112+
113+
assertThat(MathUtils.max(null, 5)).isEqualTo(5);
114+
assertThat(MathUtils.max(5, null)).isEqualTo(5);
115+
}
116+
117+
@Test
118+
void testMin() {
119+
assertThat(MathUtils.min(1, 5)).isEqualTo(1);
120+
assertThat(MathUtils.min(5, 1)).isEqualTo(1);
121+
assertThat(MathUtils.min(5, 5)).isEqualTo(5);
122+
assertThat(MathUtils.min(-1, -5)).isEqualTo(-5);
123+
124+
assertThat(MathUtils.min(null, 5)).isEqualTo(5);
125+
assertThat(MathUtils.min(5, null)).isEqualTo(5);
126+
}
127+
128+
@Test
129+
void testIncrementSafely() {
130+
assertThat(MathUtils.incrementSafely(0)).isEqualTo(1);
131+
assertThat(MathUtils.incrementSafely(10)).isEqualTo(11);
132+
assertThat(MathUtils.incrementSafely(-1)).isZero();
133+
assertThat(MathUtils.incrementSafely(Integer.MAX_VALUE - 1)).isEqualTo(Integer.MAX_VALUE);
134+
assertThat(MathUtils.incrementSafely(Integer.MAX_VALUE)).isEqualTo(Integer.MAX_VALUE);
135+
}
136+
137+
@Test
138+
void testAddSafely() {
139+
assertThat(MathUtils.addSafely(1, 2)).isEqualTo(3);
140+
assertThat(MathUtils.addSafely(-1, -2)).isEqualTo(-3);
141+
assertThat(MathUtils.addSafely(Integer.MAX_VALUE, 0)).isEqualTo(Integer.MAX_VALUE);
142+
assertThat(MathUtils.addSafely(Integer.MIN_VALUE, 0)).isEqualTo(Integer.MIN_VALUE);
143+
144+
// Overflow
145+
assertThat(MathUtils.addSafely(Integer.MAX_VALUE, 1)).isEqualTo(Integer.MAX_VALUE);
146+
assertThat(MathUtils.addSafely(Integer.MAX_VALUE - 1, 2)).isEqualTo(Integer.MAX_VALUE);
147+
assertThat(MathUtils.addSafely(1, Integer.MAX_VALUE)).isEqualTo(Integer.MAX_VALUE);
148+
149+
// Underflow
150+
assertThat(MathUtils.addSafely(Integer.MIN_VALUE, -1)).isEqualTo(Integer.MAX_VALUE);
151+
assertThat(MathUtils.addSafely(Integer.MIN_VALUE + 1, -2)).isEqualTo(Integer.MAX_VALUE);
152+
assertThat(MathUtils.addSafely(-1, Integer.MIN_VALUE)).isEqualTo(Integer.MAX_VALUE);
153+
}
154+
155+
@Test
156+
void testMultiplySafely() {
157+
assertThat(MathUtils.multiplySafely(2, 3)).isEqualTo(6);
158+
assertThat(MathUtils.multiplySafely(-2, 3)).isEqualTo(-6);
159+
assertThat(MathUtils.multiplySafely(-2, -3)).isEqualTo(6);
160+
assertThat(MathUtils.multiplySafely(Integer.MAX_VALUE, 1)).isEqualTo(Integer.MAX_VALUE);
161+
assertThat(MathUtils.multiplySafely(Integer.MIN_VALUE, 1)).isEqualTo(Integer.MIN_VALUE);
162+
assertThat(MathUtils.multiplySafely(100, 0)).isZero();
163+
164+
// Overflow
165+
assertThat(MathUtils.multiplySafely(Integer.MAX_VALUE, 2)).isEqualTo(Integer.MAX_VALUE);
166+
assertThat(MathUtils.multiplySafely(Integer.MAX_VALUE / 2 + 1, 2))
167+
.isEqualTo(Integer.MAX_VALUE);
168+
169+
// Underflow
170+
assertThat(MathUtils.multiplySafely(Integer.MIN_VALUE, 2)).isEqualTo(Integer.MAX_VALUE);
171+
assertThat(MathUtils.multiplySafely(Integer.MIN_VALUE / 2 - 1, 2))
172+
.isEqualTo(Integer.MAX_VALUE);
173+
assertThat(MathUtils.multiplySafely(Integer.MAX_VALUE, -2)).isEqualTo(Integer.MAX_VALUE);
174+
}
175+
176+
@Test
177+
void testAddSafelyUnderflowReturnsMaxValue() {
178+
// The current implementation of addSafely returns Integer.MAX_VALUE on underflow,
179+
// which might be unexpected. This test verifies that behavior.
180+
assertThat(MathUtils.addSafely(Integer.MIN_VALUE, -1)).isEqualTo(Integer.MAX_VALUE);
181+
assertThat(MathUtils.addSafely(-1, Integer.MIN_VALUE)).isEqualTo(Integer.MAX_VALUE);
182+
}
183+
184+
@Test
185+
void testMultiplySafelyUnderflowReturnsMaxValue() {
186+
// The current implementation of multiplySafely returns Integer.MAX_VALUE on underflow,
187+
// which might be unexpected. This test verifies that behavior.
188+
assertThat(MathUtils.multiplySafely(Integer.MIN_VALUE, 2)).isEqualTo(Integer.MAX_VALUE);
189+
assertThat(MathUtils.multiplySafely(Integer.MAX_VALUE, -2)).isEqualTo(Integer.MAX_VALUE);
190+
}
191+
}

‎paimon-common/src/main/java/org/apache/paimon/data/MultiSegments.java‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.data;
20+
21+
import org.apache.paimon.memory.MemorySegment;
22+
23+
import java.util.ArrayList;
24+
25+
/** A {@link Segments} with limit in last segment. */
26+
public class MultiSegments implements Segments {
27+
28+
private final ArrayList<MemorySegment> segments;
29+
30+
private final int limitInLastSegment;
31+
private final int pageSize;
32+
private final long totalMemorySize;
33+
34+
public MultiSegments(ArrayList<MemorySegment> segments, int limitInLastSegment) {
35+
this.segments = segments;
36+
this.limitInLastSegment = limitInLastSegment;
37+
this.pageSize = segments.get(0).size();
38+
this.totalMemorySize = ((long) segments.size()) * pageSize;
39+
}
40+
41+
public int pageSize() {
42+
return pageSize;
43+
}
44+
45+
public ArrayList<MemorySegment> segments() {
46+
return segments;
47+
}
48+
49+
public int limitInLastSegment() {
50+
return limitInLastSegment;
51+
}
52+
53+
@Override
54+
public long totalMemorySize() {
55+
return totalMemorySize;
56+
}
57+
}

‎paimon-common/src/main/java/org/apache/paimon/data/Segments.java‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -19,26 +19,27 @@
1919
package org.apache.paimon.data;
2020

2121
import org.apache.paimon.memory.MemorySegment;
22+
import org.apache.paimon.utils.MathUtils;
2223

2324
import java.util.ArrayList;
2425

25-
/** Segments with limit in last segment. */
26-
public class Segments {
27-
28-
private final ArrayList<MemorySegment> segments;
29-
30-
private final int limitInLastSegment;
31-
32-
public Segments(ArrayList<MemorySegment> segments, int limitInLastSegment) {
33-
this.segments = segments;
34-
this.limitInLastSegment = limitInLastSegment;
35-
}
36-
37-
public ArrayList<MemorySegment> segments() {
38-
return segments;
39-
}
40-
41-
public int limitInLastSegment() {
42-
return limitInLastSegment;
26+
/** Segments contains multiple {@link MultiSegments}. */
27+
public interface Segments {
28+
29+
long totalMemorySize();
30+
31+
static Segments create(ArrayList<MemorySegment> segments, int limitInLastSegment) {
32+
if (segments.size() == 1) {
33+
MemorySegment segment = segments.get(0);
34+
int roundUp = MathUtils.roundUpToPowerOf2(limitInLastSegment);
35+
if (roundUp < segment.size()) {
36+
MemorySegment newSegment = MemorySegment.allocateHeapMemory(roundUp);
37+
segment.copyTo(0, newSegment, 0, limitInLastSegment);
38+
segment = newSegment;
39+
}
40+
return new SingleSegments(segment, limitInLastSegment);
41+
} else {
42+
return new MultiSegments(segments, limitInLastSegment);
43+
}
4344
}
4445
}

‎paimon-common/src/main/java/org/apache/paimon/data/SimpleCollectingOutputView.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,15 +25,14 @@
2525
import java.io.EOFException;
2626
import java.io.IOException;
2727
import java.util.ArrayList;
28-
import java.util.List;
2928

3029
/**
3130
* The list with the full segments contains at any point all completely full segments, plus the
3231
* segment that is currently filled.
3332
*/
3433
public class SimpleCollectingOutputView extends AbstractPagedOutputView {
3534

36-
private final List<MemorySegment> fullSegments;
35+
private final ArrayList<MemorySegment> fullSegments;
3736

3837
private final MemorySegmentSource memorySource;
3938

@@ -46,16 +45,22 @@ public SimpleCollectingOutputView(MemorySegmentSource memSource, int segmentSize
4645
}
4746

4847
public SimpleCollectingOutputView(
49-
List<MemorySegment> fullSegmentTarget, MemorySegmentSource memSource, int segmentSize) {
48+
ArrayList<MemorySegment> fullSegmentTarget,
49+
MemorySegmentSource memSource,
50+
int segmentSize) {
5051
super(memSource.nextSegment(), segmentSize);
5152
this.segmentSizeBits = MathUtils.log2strict(segmentSize);
5253
this.fullSegments = fullSegmentTarget;
5354
this.memorySource = memSource;
5455
this.fullSegments.add(getCurrentSegment());
5556
}
5657

58+
public ArrayList<MemorySegment> fullSegments() {
59+
return fullSegments;
60+
}
61+
5762
public void reset() {
58-
if (this.fullSegments.size() != 0) {
63+
if (!this.fullSegments.isEmpty()) {
5964
throw new IllegalStateException("The target list still contains memory segments.");
6065
}
6166

0 commit comments

Comments
 (0)