Skip to content

Commit 67167ee

Browse files
JingsongLiJingsongLi
committed
fix
1 parent 5053f6b commit 67167ee

File tree

6 files changed

+20
-60
lines changed

6 files changed

+20
-60
lines changed

‎paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -47,8 +47,6 @@
4747
import java.util.concurrent.ExecutorService;
4848

4949
import static java.util.Collections.emptyList;
50-
import static org.apache.paimon.table.source.AppendOnlySplitGenerator.CREATION_TIME_COMPARATOR;
51-
import static org.apache.paimon.table.source.AppendOnlySplitGenerator.SEQUENCE_NUMBER_COMPARATOR;
5250

5351
/** Compact manager for {@link AppendOnlyFileStore}. */
5452
public class BucketedAppendCompactManager extends CompactFutureManager {
@@ -80,7 +78,11 @@ public BucketedAppendCompactManager(
8078
boolean ordered,
8179
CompactRewriter rewriter,
8280
@Nullable CompactionMetrics.Reporter metricsReporter) {
83-
this.comparator = ordered ? SEQUENCE_NUMBER_COMPARATOR : CREATION_TIME_COMPARATOR;
81+
this.comparator =
82+
ordered
83+
? Comparator.comparing(DataFileMeta::minSequenceNumber)
84+
: Comparator.comparing(DataFileMeta::creationTime)
85+
.thenComparing(DataFileMeta::fileName);
8486
this.executor = executor;
8587
this.dvMaintainer = dvMaintainer;
8688
this.toCompact = new PriorityQueue<>(comparator);

‎paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java‎

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import org.apache.paimon.data.InternalRow;
2424
import org.apache.paimon.fs.FileIO;
2525
import org.apache.paimon.fs.Path;
26-
import org.apache.paimon.io.DataFileMeta;
2726
import org.apache.paimon.operation.AppendOnlyFileStoreScan;
2827
import org.apache.paimon.operation.BaseAppendFileStoreWrite;
2928
import org.apache.paimon.operation.FileStoreScan;
@@ -46,14 +45,10 @@
4645
import javax.annotation.Nullable;
4746

4847
import java.util.ArrayList;
49-
import java.util.Comparator;
5048
import java.util.List;
5149
import java.util.function.BiConsumer;
5250
import java.util.function.Function;
5351

54-
import static org.apache.paimon.table.source.AppendOnlySplitGenerator.CREATION_TIME_COMPARATOR;
55-
import static org.apache.paimon.table.source.AppendOnlySplitGenerator.SEQUENCE_NUMBER_COMPARATOR;
56-
5752
/** {@link FileStoreTable} for append table. */
5853
public class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
5954

@@ -96,14 +91,9 @@ protected SplitGenerator splitGenerator() {
9691
CoreOptions options = store().options();
9792
long targetSplitSize = options.splitTargetSize();
9893
long openFileCost = options.splitOpenFileCost();
99-
Comparator<DataFileMeta> comparator = CREATION_TIME_COMPARATOR;
100-
if (bucketMode() == BucketMode.HASH_FIXED && options.bucketAppendOrdered()) {
101-
comparator = SEQUENCE_NUMBER_COMPARATOR;
102-
}
10394
return coreOptions().dataEvolutionEnabled()
10495
? new DataEvolutionSplitGenerator(targetSplitSize, openFileCost)
105-
: new AppendOnlySplitGenerator(
106-
targetSplitSize, openFileCost, bucketMode(), comparator);
96+
: new AppendOnlySplitGenerator(targetSplitSize, openFileCost, bucketMode());
10797
}
10898

10999
@Override

‎paimon-core/src/main/java/org/apache/paimon/table/source/AppendOnlySplitGenerator.java‎

Lines changed: 2 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -32,27 +32,15 @@
3232
/** Append only implementation of {@link SplitGenerator}. */
3333
public class AppendOnlySplitGenerator implements SplitGenerator {
3434

35-
public static final Comparator<DataFileMeta> CREATION_TIME_COMPARATOR =
36-
Comparator.comparing(DataFileMeta::creationTime).thenComparing(DataFileMeta::fileName);
37-
38-
public static final Comparator<DataFileMeta> SEQUENCE_NUMBER_COMPARATOR =
39-
Comparator.comparing(DataFileMeta::minSequenceNumber)
40-
.thenComparing(DataFileMeta::fileName);
41-
4235
private final long targetSplitSize;
4336
private final long openFileCost;
4437
private final BucketMode bucketMode;
45-
private final Comparator<DataFileMeta> fileComparator;
4638

4739
public AppendOnlySplitGenerator(
48-
long targetSplitSize,
49-
long openFileCost,
50-
BucketMode bucketMode,
51-
Comparator<DataFileMeta> fileComparator) {
40+
long targetSplitSize, long openFileCost, BucketMode bucketMode) {
5241
this.targetSplitSize = targetSplitSize;
5342
this.openFileCost = openFileCost;
5443
this.bucketMode = bucketMode;
55-
this.fileComparator = fileComparator;
5644
}
5745

5846
@Override
@@ -63,7 +51,7 @@ public boolean alwaysRawConvertible() {
6351
@Override
6452
public List<SplitGroup> splitForBatch(List<DataFileMeta> input) {
6553
List<DataFileMeta> files = new ArrayList<>(input);
66-
files.sort(fileComparator);
54+
files.sort(Comparator.comparing(DataFileMeta::minSequenceNumber));
6755
Function<DataFileMeta, Long> weightFunc = file -> Math.max(file.fileSize(), openFileCost);
6856
return BinPacking.packForOrdered(files, weightFunc, targetSplitSize).stream()
6957
.map(SplitGroup::rawConvertibleGroup)

‎paimon-core/src/test/java/org/apache/paimon/table/source/SplitGeneratorTest.java‎

Lines changed: 3 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,6 @@
3737
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
3838
import static org.apache.paimon.io.DataFileTestUtils.fromMinMax;
3939
import static org.apache.paimon.io.DataFileTestUtils.newFile;
40-
import static org.apache.paimon.table.source.AppendOnlySplitGenerator.SEQUENCE_NUMBER_COMPARATOR;
4140
import static org.assertj.core.api.Assertions.assertThat;
4241

4342
/** Test for {@link AppendOnlySplitGenerator} and {@link MergeTreeSplitGenerator}. */
@@ -77,11 +76,7 @@ public void testAppend() {
7776
newFileFromSequence("6", 101, 61, 100));
7877
assertThat(
7978
toNames(
80-
new AppendOnlySplitGenerator(
81-
40,
82-
2,
83-
BucketMode.HASH_FIXED,
84-
SEQUENCE_NUMBER_COMPARATOR)
79+
new AppendOnlySplitGenerator(40, 2, BucketMode.HASH_FIXED)
8580
.splitForBatch(files)))
8681
.containsExactlyInAnyOrder(
8782
Arrays.asList("1", "2"),
@@ -91,11 +86,7 @@ public void testAppend() {
9186

9287
assertThat(
9388
toNames(
94-
new AppendOnlySplitGenerator(
95-
70,
96-
2,
97-
BucketMode.HASH_FIXED,
98-
SEQUENCE_NUMBER_COMPARATOR)
89+
new AppendOnlySplitGenerator(70, 2, BucketMode.HASH_FIXED)
9990
.splitForBatch(files)))
10091
.containsExactlyInAnyOrder(
10192
Arrays.asList("1", "2", "3"),
@@ -104,11 +95,7 @@ public void testAppend() {
10495

10596
assertThat(
10697
toNames(
107-
new AppendOnlySplitGenerator(
108-
40,
109-
20,
110-
BucketMode.HASH_FIXED,
111-
SEQUENCE_NUMBER_COMPARATOR)
98+
new AppendOnlySplitGenerator(40, 20, BucketMode.HASH_FIXED)
11299
.splitForBatch(files)))
113100
.containsExactlyInAnyOrder(
114101
Arrays.asList("1", "2"),

‎paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkWriteITCase.java‎

Lines changed: 7 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -103,14 +103,12 @@ public void testWriteWithDefaultValue() {
103103
// test partial write
104104
spark.sql("INSERT INTO T (a) VALUES (1), (2)").collectAsList();
105105
List<Row> rows = spark.sql("SELECT * FROM T").collectAsList();
106-
assertThat(rows.stream().map(Row::toString))
107-
.containsExactlyInAnyOrder("[1,2,my_value]", "[2,2,my_value]");
106+
assertThat(rows.toString()).isEqualTo("[[1,2,my_value], [2,2,my_value]]");
108107

109108
// test write with DEFAULT
110109
spark.sql("INSERT INTO T VALUES (3, DEFAULT, DEFAULT)").collectAsList();
111110
rows = spark.sql("SELECT * FROM T").collectAsList();
112-
assertThat(rows.stream().map(Row::toString))
113-
.containsExactlyInAnyOrder("[1,2,my_value]", "[2,2,my_value]", "[3,2,my_value]");
111+
assertThat(rows.toString()).isEqualTo("[[1,2,my_value], [2,2,my_value], [3,2,my_value]]");
114112

115113
// test add column with DEFAULT not support
116114
assertThatThrownBy(() -> spark.sql("ALTER TABLE T ADD COLUMN d INT DEFAULT 5"))
@@ -121,21 +119,16 @@ public void testWriteWithDefaultValue() {
121119
spark.sql("ALTER TABLE T ALTER COLUMN b TYPE STRING").collectAsList();
122120
spark.sql("INSERT INTO T (a) VALUES (4)").collectAsList();
123121
rows = spark.sql("SELECT * FROM T").collectAsList();
124-
assertThat(rows.stream().map(Row::toString))
125-
.containsExactlyInAnyOrder(
126-
"[1,2,my_value]", "[2,2,my_value]", "[3,2,my_value]", "[4,2,my_value]");
122+
assertThat(rows.toString())
123+
.isEqualTo("[[1,2,my_value], [2,2,my_value], [3,2,my_value], [4,2,my_value]]");
127124

128125
// test alter default column
129126
spark.sql("ALTER TABLE T ALTER COLUMN b SET DEFAULT '3'");
130127
spark.sql("INSERT INTO T (a) VALUES (5)").collectAsList();
131128
rows = spark.sql("SELECT * FROM T").collectAsList();
132-
assertThat(rows.stream().map(Row::toString))
133-
.containsExactlyInAnyOrder(
134-
"[1,2,my_value]",
135-
"[2,2,my_value]",
136-
"[3,2,my_value]",
137-
"[4,2,my_value]",
138-
"[5,3,my_value]");
129+
assertThat(rows.toString())
130+
.isEqualTo(
131+
"[[1,2,my_value], [2,2,my_value], [3,2,my_value], [4,2,my_value], [5,3,my_value]]");
139132
}
140133

141134
@Test

‎paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/LanceFormatTest.scala‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,10 +30,10 @@ class LanceFormatTest extends PaimonSparkTestBase {
3030
"CREATE TABLE t (a INT, b STRING, scores ARRAY<DOUBLE>) TBLPROPERTIES ('file.format' = 'lance')")
3131
sql(
3232
"INSERT INTO t VALUES (1, 'a', ARRAY(CAST(90.5 as double), CAST(88.0 as double))), (2, 'b', ARRAY(CAST(90.6 as double), CAST(88.1 as double)))")
33-
assert(sql("SELECT * FROM t LIMIT 1").collect().length == 1)
33+
checkAnswer(sql("SELECT * FROM t LIMIT 1"), Seq(Row(1, "a", Array(90.5, 88.0))))
3434
checkAnswer(
3535
sql("SELECT * FROM t LIMIT 10"),
36-
Row(1, "a", Array(90.5, 88.0)) :: Row(2, "b", Array(90.6, 88.1)) :: Nil)
36+
Seq(Row(1, "a", Array(90.5, 88.0)), Row(2, "b", Array(90.6, 88.1))))
3737

3838
assert(
3939
sql("SELECT file_size_in_bytes FROM `t$files`")

0 commit comments

Comments
 (0)