Skip to content

Commit 6b140fb

Browse files
loserwang1024loserwang1024
committed
fix test
1 parent e259210 commit 6b140fb

16 files changed

Lines changed: 77 additions & 40 deletions

File tree

‎fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherFilterITCase.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.fluss.client.table.scanner.ScanRecord;
2525
import org.apache.fluss.config.ConfigOptions;
2626
import org.apache.fluss.metadata.TableBucket;
27+
import org.apache.fluss.metadata.TableInfo;
2728
import org.apache.fluss.predicate.Predicate;
2829
import org.apache.fluss.predicate.PredicateBuilder;
2930
import org.apache.fluss.record.LogRecordBatchStatisticsTestUtils;
@@ -49,6 +50,7 @@
4950
import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
5051
import static org.apache.fluss.record.TestData.DATA1_TABLE_INFO;
5152
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
53+
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
5254
import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
5355
import static org.apache.fluss.record.TestData.TEST_SCHEMA_GETTER;
5456
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newProduceLogRequest;
@@ -118,7 +120,18 @@ protected void setup() throws Exception {
118120
new RemoteFileDownloader(1),
119121
LogRecordReadContext.SchemaResolution.TARGET);
120122
logFetcher.registerTable(
121-
new TableScanSpec(DATA1_TABLE_INFO, null, recordBatchFilter), TEST_SCHEMA_GETTER);
123+
new TableScanSpec(
124+
TableInfo.of(
125+
DATA1_TABLE_PATH,
126+
tableId,
127+
1,
128+
DATA1_TABLE_DESCRIPTOR,
129+
DEFAULT_REMOTE_DATA_DIR,
130+
System.currentTimeMillis(),
131+
System.currentTimeMillis()),
132+
null,
133+
recordBatchFilter),
134+
TEST_SCHEMA_GETTER);
122135
}
123136

124137
@AfterEach

‎fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import static org.apache.fluss.record.TestData.DATA2;
6464
import static org.apache.fluss.record.TestData.DATA2_ROW_TYPE;
6565
import static org.apache.fluss.record.TestData.DATA2_SCHEMA;
66+
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
6667
import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newProduceLogRequest;
6768
import static org.apache.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject;
6869
import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
@@ -72,6 +73,7 @@
7273
public class LogFetcherITCase extends ClientToServerITCaseBase {
7374
private LogFetcher logFetcher;
7475
private long tableId;
76+
private TableInfo tableInfo;
7577
private final int bucketId0 = 0;
7678
private final int bucketId1 = 1;
7779
private LogScannerStatus logScannerStatus;
@@ -109,8 +111,16 @@ protected void setup() throws Exception {
109111
TestingScannerMetricGroup.newInstance(),
110112
new RemoteFileDownloader(1),
111113
LogRecordReadContext.SchemaResolution.TARGET);
112-
logFetcher.registerTable(
113-
new TableScanSpec(DATA1_TABLE_INFO, null, null), clientSchemaGetter);
114+
tableInfo =
115+
TableInfo.of(
116+
DATA1_TABLE_PATH,
117+
tableId,
118+
1,
119+
DATA1_TABLE_DESCRIPTOR,
120+
DEFAULT_REMOTE_DATA_DIR,
121+
System.currentTimeMillis(),
122+
System.currentTimeMillis());
123+
logFetcher.registerTable(new TableScanSpec(tableInfo, null, null), clientSchemaGetter);
114124
}
115125

116126
@Test
@@ -292,8 +302,7 @@ void testFetchWhenDestinationIsNullInMetadata() throws Exception {
292302
new RemoteFileDownloader(1),
293303
LogRecordReadContext.SchemaResolution.TARGET);
294304

295-
logFetcher.registerTable(
296-
new TableScanSpec(DATA1_TABLE_INFO, null, null), clientSchemaGetter);
305+
logFetcher.registerTable(new TableScanSpec(tableInfo, null, null), clientSchemaGetter);
297306
// send fetches to fetch data, should have no available fetch.
298307
logFetcher.sendFetches();
299308
assertThat(logFetcher.hasAvailableFetches()).isFalse();
@@ -332,8 +341,7 @@ void testFetchWithInvalidTableOrPartitions() throws Exception {
332341
TestingScannerMetricGroup.newInstance(),
333342
new RemoteFileDownloader(1),
334343
LogRecordReadContext.SchemaResolution.TARGET);
335-
logFetcher.registerTable(
336-
new TableScanSpec(DATA1_TABLE_INFO, null, null), clientSchemaGetter);
344+
logFetcher.registerTable(new TableScanSpec(tableInfo, null, null), clientSchemaGetter);
337345

338346
ExecutorService executor = Executors.newSingleThreadExecutor();
339347
Future<?> future =

‎fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -268,14 +268,6 @@ public static LogRecordReadContext createCompactedRowReadContext(
268268
return createCompactedRowReadContext(-1L, target, schemaGetter);
269269
}
270270

271-
/** Creates a LogRecordReadContext for COMPACTED log format with schema evolution support. */
272-
public static LogRecordReadContext createCompactedRowReadContext(
273-
long tableId, RowType rowType, int schemaId, SchemaGetter schemaGetter) {
274-
int[] selectedFields = IntStream.range(0, rowType.getFieldCount()).toArray();
275-
ReadTarget target = new ReadTarget(schemaId, rowType, selectedFields, true);
276-
return createCompactedRowReadContext(tableId, target, schemaGetter);
277-
}
278-
279271
// -------------------------------------------------------------------------
280272
// Constructor
281273
// -------------------------------------------------------------------------
@@ -306,7 +298,7 @@ public LogFormat getLogFormat() {
306298

307299
@Override
308300
public RowType getRowType(int schemaId) {
309-
if (isProjectionPushDowned()) {
301+
if (isProjectionPushDowned() || (target != null && target.schemaId == schemaId)) {
310302
assert target != null;
311303
return target.dataRowType;
312304
}

‎fluss-common/src/test/java/org/apache/fluss/record/DefaultLogRecordBatchTest.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -90,10 +90,7 @@ void testIndexedRowWriteAndReadBatch(byte magic) throws Exception {
9090
assertThat(logRecordBatch.isValid()).isTrue();
9191
assertThat(logRecordBatch.schemaId()).isEqualTo(schemaId);
9292

93-
SchemaGetter schemaGetter =
94-
new TestingSchemaGetter(
95-
new SchemaInfo(
96-
Schema.newBuilder().fromRowType(allRowType).build(), schemaId));
93+
SchemaGetter schemaGetter = new TestingSchemaGetter(schemaId, allRowType);
9794
// verify record.
9895
int i = 0;
9996
try (LogRecordReadContext readContext =

‎fluss-common/src/test/java/org/apache/fluss/record/LogTestBase.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -63,15 +63,18 @@ protected List<IndexedRow> createAllTypeRowDataList() {
6363
}
6464

6565
public static void assertLogRecordsListEquals(
66-
List<MemoryLogRecords> expected, LogRecords actual, RowType rowType) {
66+
List<MemoryLogRecords> expected,
67+
LogRecords actual,
68+
RowType rowType,
69+
SchemaGetter schemaGetter) {
6770
LogRecordsAssert.assertThatLogRecords(actual)
6871
.withSchema(rowType)
69-
.withSchemaGetter(TEST_SCHEMA_GETTER)
72+
.withSchemaGetter(schemaGetter)
7073
.isEqualTo(new ListLogRecords(expected));
7174
}
7275

7376
protected void assertLogRecordsListEquals(List<MemoryLogRecords> expected, LogRecords actual) {
74-
assertLogRecordsListEquals(expected, actual, baseRowType);
77+
assertLogRecordsListEquals(expected, actual, baseRowType, TEST_SCHEMA_GETTER);
7578
}
7679

7780
protected void assertIndexedLogRecordBatchAndRowEquals(

‎fluss-common/src/test/java/org/apache/fluss/record/TestingSchemaGetter.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import org.apache.fluss.metadata.Schema;
2222
import org.apache.fluss.metadata.SchemaGetter;
2323
import org.apache.fluss.metadata.SchemaInfo;
24+
import org.apache.fluss.types.RowType;
2425

2526
import java.util.HashMap;
2627
import java.util.Map;
@@ -31,17 +32,16 @@ public class TestingSchemaGetter implements SchemaGetter {
3132
private SchemaInfo schemaInfo;
3233
private final Map<Integer, Schema> schemaCaches;
3334

34-
public TestingSchemaGetter() {
35-
this.schemaInfo = null;
36-
this.schemaCaches = new HashMap<>();
37-
}
38-
3935
public TestingSchemaGetter(SchemaInfo schemaInfo) {
4036
this.schemaInfo = schemaInfo;
4137
this.schemaCaches = new HashMap<>();
4238
this.schemaCaches.put(schemaInfo.getSchemaId(), schemaInfo.getSchema());
4339
}
4440

41+
public TestingSchemaGetter(int schemaId, RowType rowType) {
42+
this(schemaId, Schema.newBuilder().fromRowType(rowType).build());
43+
}
44+
4545
public TestingSchemaGetter(int schemaId, Schema schema) {
4646
this.schemaInfo = new SchemaInfo(schema, schemaId);
4747
this.schemaCaches = new HashMap<>();

‎fluss-common/src/test/java/org/apache/fluss/testutils/LogRecordsAssert.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
import org.apache.fluss.metadata.SchemaGetter;
2121
import org.apache.fluss.record.LogRecordBatch;
2222
import org.apache.fluss.record.LogRecords;
23-
import org.apache.fluss.record.TestingSchemaGetter;
2423
import org.apache.fluss.types.RowType;
2524

2625
import org.assertj.core.api.AbstractAssert;
@@ -40,7 +39,7 @@ public static LogRecordsAssert assertThatLogRecords(LogRecords actual) {
4039

4140
private RowType rowType;
4241
private boolean assertCheckSum = true;
43-
private SchemaGetter schemaGetter = new TestingSchemaGetter();
42+
private SchemaGetter schemaGetter;
4443

4544
private LogRecordsAssert(LogRecords actual) {
4645
super(actual, LogRecordsAssert.class);

‎fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletMergeModeTest.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ void testDefaultModeAppliesMergeEngine() throws Exception {
209209

210210
assertThatLogRecords(actualLogRecords)
211211
.withSchema(AGG_ROW_TYPE)
212+
.withSchemaGetter(schemaGetter)
212213
.assertCheckSum(true)
213214
.isEqualTo(expectedLogs);
214215
}
@@ -248,6 +249,7 @@ void testOverwriteModeBypassesMergeEngine() throws Exception {
248249

249250
assertThatLogRecords(actualLogRecords)
250251
.withSchema(AGG_ROW_TYPE)
252+
.withSchemaGetter(schemaGetter)
251253
.assertCheckSum(true)
252254
.isEqualTo(expectedLogs);
253255

@@ -299,6 +301,7 @@ void testOverwriteModeForUndoRecoveryScenario() throws Exception {
299301

300302
assertThatLogRecords(actualLogRecords)
301303
.withSchema(AGG_ROW_TYPE)
304+
.withSchemaGetter(schemaGetter)
302305
.assertCheckSum(true)
303306
.isEqualTo(expectedLogs);
304307
}
@@ -321,6 +324,7 @@ void testOverwriteModeWithNewKey() throws Exception {
321324

322325
assertThatLogRecords(actualLogRecords)
323326
.withSchema(AGG_ROW_TYPE)
327+
.withSchemaGetter(schemaGetter)
324328
.assertCheckSum(true)
325329
.isEqualTo(expectedLogs);
326330
}
@@ -351,6 +355,7 @@ void testOverwriteModeWithDelete() throws Exception {
351355

352356
assertThatLogRecords(actualLogRecords)
353357
.withSchema(AGG_ROW_TYPE)
358+
.withSchemaGetter(schemaGetter)
354359
.assertCheckSum(true)
355360
.isEqualTo(expectedLogs);
356361
}
@@ -404,6 +409,7 @@ void testMixedMergeModeOperations() throws Exception {
404409

405410
assertThatLogRecords(actualLogRecords)
406411
.withSchema(AGG_ROW_TYPE)
412+
.withSchemaGetter(schemaGetter)
407413
.assertCheckSum(true)
408414
.isEqualTo(expectedLogs);
409415
}
@@ -442,6 +448,7 @@ void testOverwriteModeWithPartialUpdate() throws Exception {
442448

443449
assertThatLogRecords(actualLogRecords)
444450
.withSchema(AGG_ROW_TYPE)
451+
.withSchemaGetter(schemaGetter)
445452
.assertCheckSum(true)
446453
.isEqualTo(expectedLogs);
447454
}
@@ -489,6 +496,7 @@ void testOverwriteModeWithMultipleKeys() throws Exception {
489496

490497
assertThatLogRecords(actualLogRecords)
491498
.withSchema(AGG_ROW_TYPE)
499+
.withSchemaGetter(schemaGetter)
492500
.assertCheckSum(true)
493501
.isEqualTo(expectedLogs);
494502
}
@@ -528,6 +536,7 @@ void testDefaultMergeModeIsDefault() throws Exception {
528536

529537
assertThatLogRecords(actualLogRecords)
530538
.withSchema(AGG_ROW_TYPE)
539+
.withSchemaGetter(schemaGetter)
531540
.assertCheckSum(true)
532541
.isEqualTo(expectedLogs);
533542
}

‎fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletSchemaEvolutionTest.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,7 @@ void testPartialUpdateAfterAddColumn() throws Exception {
198198

199199
assertThatLogRecords(actualLogRecords)
200200
.withSchema(ROW_TYPE_V1)
201+
.withSchemaGetter(schemaGetter)
201202
.assertCheckSum(true)
202203
.isEqualTo(expectedLogs);
203204
}
@@ -242,6 +243,7 @@ void testPartialDeleteAfterAddColumn() throws Exception {
242243

243244
assertThatLogRecords(actualLogRecords)
244245
.withSchema(ROW_TYPE_V1)
246+
.withSchemaGetter(schemaGetter)
245247
.assertCheckSum(true)
246248
.isEqualTo(expectedLogs);
247249
}

0 commit comments

Comments
 (0)