Skip to content

Commit 1575c07

Browse files
xx789633xx789633
committed
fix all lake tests
1 parent 50b9ce9 commit 1575c07

File tree

4 files changed

+66
-29
lines changed

4 files changed

+66
-29
lines changed

‎fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -124,14 +124,15 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
124124

125125
TableDescriptor descriptor =
126126
TableDescriptor.builder()
127-
.schema(org.apache.fluss.metadata.Schema.newBuilder()
128-
.column("c1", DataTypes.INT())
129-
.column("c2", DataTypes.STRING())
130-
.column("c3", DataTypes.STRING())
131-
.build())
127+
.schema(
128+
org.apache.fluss.metadata.Schema.newBuilder()
129+
.column("c1", DataTypes.INT())
130+
.column("c2", DataTypes.STRING())
131+
.column("c3", DataTypes.STRING())
132+
.build())
132133
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
133134
.build();
134-
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor,1L, 1L);
135+
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
135136

136137
Table icebergTable = icebergCatalog.loadTable(toIceberg(tablePath));
137138

@@ -212,7 +213,11 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitionedTable
212213
}
213214

214215
private LakeWriter<IcebergWriteResult> createLakeWriter(
215-
TablePath tablePath, int bucket, @Nullable String partition, @Nullable Long partitionId, TableInfo tableInfo)
216+
TablePath tablePath,
217+
int bucket,
218+
@Nullable String partition,
219+
@Nullable Long partitionId,
220+
TableInfo tableInfo)
216221
throws IOException {
217222
return icebergLakeTieringFactory.createLakeWriter(
218223
new WriterInitContext() {

‎fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/tiering/LanceLakeWriter.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -56,9 +56,7 @@ public LanceLakeWriter(Configuration options, WriterInitContext writerInitContex
5656

5757
this.arrowWriter =
5858
LanceDatasetAdapter.getArrowWriter(
59-
schema.get(),
60-
batchSize,
61-
writerInitContext.tableInfo().getRowType());
59+
schema.get(), batchSize, writerInitContext.tableInfo().getRowType());
6260

6361
WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
6462
Callable<List<FragmentMetadata>> fragmentCreator =

‎fluss-lake/fluss-lake-lance/src/test/java/com/alibaba/fluss/lake/lance/tiering/LanceTieringTest.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ void testTieringWriteTable(boolean isPartitioned) throws Exception {
105105
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
106106
.customProperties(customProperties)
107107
.build();
108-
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor,1L, 1L);
108+
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
109109

110110
List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
111111
SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
@@ -250,10 +250,7 @@ private LakeCommitter<LanceWriteResult, LanceCommittable> createLakeCommitter(
250250
}
251251

252252
private LakeWriter<LanceWriteResult> createLakeWriter(
253-
TablePath tablePath,
254-
int bucket,
255-
@Nullable String partition,
256-
TableInfo tableInfo)
253+
TablePath tablePath, int bucket, @Nullable String partition, TableInfo tableInfo)
257254
throws IOException {
258255
return lanceLakeTieringFactory.createLakeWriter(
259256
new WriterInitContext() {

‎fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/tiering/PaimonTieringTest.java‎

Lines changed: 51 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,16 @@
1717

1818
package org.apache.fluss.lake.paimon.tiering;
1919

20+
import org.apache.fluss.config.ConfigOptions;
2021
import org.apache.fluss.config.Configuration;
2122
import org.apache.fluss.lake.committer.CommittedLakeSnapshot;
2223
import org.apache.fluss.lake.committer.LakeCommitter;
2324
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
2425
import org.apache.fluss.lake.writer.LakeWriter;
2526
import org.apache.fluss.lake.writer.WriterInitContext;
2627
import org.apache.fluss.metadata.TableBucket;
28+
import org.apache.fluss.metadata.TableDescriptor;
29+
import org.apache.fluss.metadata.TableInfo;
2730
import org.apache.fluss.metadata.TablePath;
2831
import org.apache.fluss.record.ChangeType;
2932
import org.apache.fluss.record.GenericRecord;
@@ -118,6 +121,17 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr
118121
isPartitioned ? "partitioned" : "non_partitioned"));
119122
createTable(
120123
tablePath, isPrimaryKeyTable, isPartitioned, isPrimaryKeyTable ? bucketNum : null);
124+
TableDescriptor descriptor =
125+
TableDescriptor.builder()
126+
.schema(
127+
org.apache.fluss.metadata.Schema.newBuilder()
128+
.column("c1", org.apache.fluss.types.DataTypes.INT())
129+
.column("c2", org.apache.fluss.types.DataTypes.STRING())
130+
.column("c3", org.apache.fluss.types.DataTypes.STRING())
131+
.build())
132+
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
133+
.build();
134+
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
121135

122136
List<PaimonWriteResult> paimonWriteResults = new ArrayList<>();
123137
SimpleVersionedSerializer<PaimonWriteResult> writeResultSerializer =
@@ -148,7 +162,7 @@ void testTieringWriteTable(boolean isPrimaryKeyTable, boolean isPartitioned) thr
148162
for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
149163
String partition = entry.getValue();
150164
try (LakeWriter<PaimonWriteResult> lakeWriter =
151-
createLakeWriter(tablePath, bucket, partition, entry.getKey())) {
165+
createLakeWriter(tablePath, bucket, partition, entry.getKey(), tableInfo)) {
152166
Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
153167
Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
154168
isPrimaryKeyTable
@@ -233,6 +247,19 @@ void testMultiPartitionTiering() throws Exception {
233247
// Test multiple partitions: region + year
234248
TablePath tablePath = TablePath.of("paimon", "test_multi_partition");
235249
createMultiPartitionTable(tablePath);
250+
TableDescriptor descriptor =
251+
TableDescriptor.builder()
252+
.schema(
253+
org.apache.fluss.metadata.Schema.newBuilder()
254+
.column("c1", org.apache.fluss.types.DataTypes.INT())
255+
.column("c2", org.apache.fluss.types.DataTypes.STRING())
256+
.column("region", org.apache.fluss.types.DataTypes.STRING())
257+
.column("year", org.apache.fluss.types.DataTypes.STRING())
258+
.build())
259+
.partitionedBy("region", "year")
260+
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
261+
.build();
262+
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
236263

237264
Map<String, List<LogRecord>> recordsByPartition = new HashMap<>();
238265
List<PaimonWriteResult> paimonWriteResults = new ArrayList<>();
@@ -253,7 +280,7 @@ void testMultiPartitionTiering() throws Exception {
253280
for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
254281
String partition = entry.getValue();
255282
try (LakeWriter<PaimonWriteResult> lakeWriter =
256-
createLakeWriter(tablePath, bucket, partition, entry.getKey())) {
283+
createLakeWriter(tablePath, bucket, partition, entry.getKey(), tableInfo)) {
257284
List<LogRecord> logRecords =
258285
genLogTableRecordsForMultiPartition(partition, bucket, 3);
259286
recordsByPartition.put(partition, logRecords);
@@ -295,7 +322,20 @@ void testThreePartitionTiering() throws Exception {
295322
// Test three partitions: region + year + month
296323
TablePath tablePath = TablePath.of("paimon", "test_three_partition");
297324
createThreePartitionTable(tablePath);
298-
325+
TableDescriptor descriptor =
326+
TableDescriptor.builder()
327+
.schema(
328+
org.apache.fluss.metadata.Schema.newBuilder()
329+
.column("c1", org.apache.fluss.types.DataTypes.INT())
330+
.column("c2", org.apache.fluss.types.DataTypes.STRING())
331+
.column("region", org.apache.fluss.types.DataTypes.STRING())
332+
.column("year", org.apache.fluss.types.DataTypes.STRING())
333+
.column("month", org.apache.fluss.types.DataTypes.STRING())
334+
.build())
335+
.partitionedBy("region", "year", "month")
336+
.property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
337+
.build();
338+
TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
299339
Map<String, List<LogRecord>> recordsByPartition = new HashMap<>();
300340
List<PaimonWriteResult> paimonWriteResults = new ArrayList<>();
301341
Map<TableBucket, Long> tableBucketOffsets = new HashMap<>();
@@ -313,7 +353,7 @@ void testThreePartitionTiering() throws Exception {
313353
for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
314354
String partition = entry.getValue();
315355
try (LakeWriter<PaimonWriteResult> lakeWriter =
316-
createLakeWriter(tablePath, bucket, partition, entry.getKey())) {
356+
createLakeWriter(tablePath, bucket, partition, entry.getKey(), tableInfo)) {
317357
List<LogRecord> logRecords =
318358
genLogTableRecordsForMultiPartition(
319359
partition, bucket, 2); // Use same method
@@ -639,7 +679,11 @@ private CloseableIterator<InternalRow> getPaimonRowsThreePartition(
639679
}
640680

641681
private LakeWriter<PaimonWriteResult> createLakeWriter(
642-
TablePath tablePath, int bucket, @Nullable String partition, @Nullable Long partitionId)
682+
TablePath tablePath,
683+
int bucket,
684+
@Nullable String partition,
685+
@Nullable Long partitionId,
686+
TableInfo tableInfo)
643687
throws IOException {
644688
return paimonLakeTieringFactory.createLakeWriter(
645689
new WriterInitContext() {
@@ -661,15 +705,8 @@ public String partition() {
661705
}
662706

663707
@Override
664-
public Map<String, String> customProperties() {
665-
// don't care about table custom properties for Paimon lake writer
666-
return new HashMap<>();
667-
}
668-
669-
@Override
670-
public org.apache.fluss.metadata.Schema schema() {
671-
throw new UnsupportedOperationException(
672-
"The lake writer in Paimon currently uses paimonCatalog to determine the schema.");
708+
public TableInfo tableInfo() {
709+
return tableInfo;
673710
}
674711
});
675712
}

0 commit comments

Comments
 (0)