@@ -1123,63 +1123,74 @@ void testUnionReadPartitionsExistInPaimonButExpiredInFluss() throws Exception {
11231123 void testPointLookupOnExpiredPartitionReadsFromLake () throws Exception {
11241124 JobClient jobClient = buildTieringJob (execEnv );
11251125
1126- String tableName = "point_lookup_expired_partition_pk_table" ;
1127- TablePath tablePath = TablePath .of (DEFAULT_DB , tableName );
1128- Map <TableBucket , Long > bucketLogEndOffset = new HashMap <>();
1129- Function <String , List <InternalRow >> rowGenerator =
1130- (partition ) ->
1131- Arrays .asList (
1132- row (3 , "string" , partition ), row (30 , "another_string" , partition ));
1133- long tableId =
1134- prepareSimplePKTable (
1135- tablePath , DEFAULT_BUCKET_NUM , true , rowGenerator , bucketLogEndOffset );
1136-
1137- waitUntilBucketSynced (tablePath , tableId , DEFAULT_BUCKET_NUM , true );
1138-
1139- Map <Long , String > partitionNameByIds = waitUntilPartitions (tablePath );
1140- assertThat (partitionNameByIds .size ()).isGreaterThanOrEqualTo (2 );
1141-
1142- // stop tiering so the read is served from the lake snapshot
1143- jobClient .cancel ().get ();
1144-
1145- Iterator <String > partitionIterator = partitionNameByIds .values ().iterator ();
1146- String expiredPartition = partitionIterator .next ();
1147- String livePartition = partitionIterator .next ();
1148-
1149- admin .dropPartition (
1150- tablePath ,
1151- new PartitionSpec (Collections .singletonMap ("c3" , expiredPartition )),
1152- false )
1153- .get ();
1154- retry (
1155- Duration .ofSeconds (60 ),
1156- () ->
1157- assertThat (admin .listPartitionInfos (tablePath ).get ())
1158- .noneMatch (p -> expiredPartition .equals (p .getPartitionName ())));
1159-
1160- List <String > expiredResult =
1161- collectBatchRows (
1162- batchTEnv
1163- .executeSql (
1164- String .format (
1165- "select * from %s where c1 = 3 and c3 = '%s'" ,
1166- tableName , expiredPartition ))
1167- .collect ());
1168- assertThat (expiredResult )
1169- .as ("point query on a lake-only (expired) partition must read from the lake" )
1170- .containsExactly (String .format ("+I[3, string, %s]" , expiredPartition ));
1171-
1172- List <String > liveResult =
1173- collectBatchRows (
1174- batchTEnv
1175- .executeSql (
1176- String .format (
1177- "select * from %s where c1 = 3 and c3 = '%s'" ,
1178- tableName , livePartition ))
1179- .collect ());
1180- assertThat (liveResult )
1181- .as ("point query on a live partition must still return its row" )
1182- .containsExactly (String .format ("+I[3, string, %s]" , livePartition ));
1126+ boolean tieringCancelled = false ;
1127+ try {
1128+ String tableName = "point_lookup_expired_partition_pk_table" ;
1129+ TablePath tablePath = TablePath .of (DEFAULT_DB , tableName );
1130+ Map <TableBucket , Long > bucketLogEndOffset = new HashMap <>();
1131+ Function <String , List <InternalRow >> rowGenerator =
1132+ (partition ) ->
1133+ Arrays .asList (
1134+ row (3 , "string" , partition ),
1135+ row (30 , "another_string" , partition ));
1136+ long tableId =
1137+ prepareSimplePKTable (
1138+ tablePath , DEFAULT_BUCKET_NUM , true , rowGenerator , bucketLogEndOffset );
1139+
1140+ waitUntilBucketSynced (tablePath , tableId , DEFAULT_BUCKET_NUM , true );
1141+
1142+ Map <Long , String > partitionNameByIds = waitUntilPartitions (tablePath );
1143+ assertThat (partitionNameByIds .size ()).isGreaterThanOrEqualTo (2 );
1144+
1145+ // stop tiering so the read is served from the lake snapshot; the per-job MiniCluster
1146+ // shuts down with the job, so cancel exactly once
1147+ jobClient .cancel ().get ();
1148+ tieringCancelled = true ;
1149+
1150+ Iterator <String > partitionIterator = partitionNameByIds .values ().iterator ();
1151+ String expiredPartition = partitionIterator .next ();
1152+ String livePartition = partitionIterator .next ();
1153+
1154+ admin .dropPartition (
1155+ tablePath ,
1156+ new PartitionSpec (Collections .singletonMap ("c3" , expiredPartition )),
1157+ false )
1158+ .get ();
1159+ retry (
1160+ Duration .ofSeconds (60 ),
1161+ () ->
1162+ assertThat (admin .listPartitionInfos (tablePath ).get ())
1163+ .noneMatch (p -> expiredPartition .equals (p .getPartitionName ())));
1164+
1165+ List <String > expiredResult =
1166+ collectBatchRows (
1167+ batchTEnv
1168+ .executeSql (
1169+ String .format (
1170+ "select * from %s where c1 = 3 and c3 = '%s'" ,
1171+ tableName , expiredPartition ))
1172+ .collect ());
1173+ assertThat (expiredResult )
1174+ .as ("point query on a lake-only (expired) partition must read from the lake" )
1175+ .containsExactly (String .format ("+I[3, string, %s]" , expiredPartition ));
1176+
1177+ List <String > liveResult =
1178+ collectBatchRows (
1179+ batchTEnv
1180+ .executeSql (
1181+ String .format (
1182+ "select * from %s where c1 = 3 and c3 = '%s'" ,
1183+ tableName , livePartition ))
1184+ .collect ());
1185+ assertThat (liveResult )
1186+ .as ("point query on a live partition must still return its row" )
1187+ .containsExactly (String .format ("+I[3, string, %s]" , livePartition ));
1188+ } finally {
1189+ // only cancel here if setup failed before the intended cancel above
1190+ if (!tieringCancelled ) {
1191+ jobClient .cancel ().get ();
1192+ }
1193+ }
11831194 }
11841195
11851196 @ Test
0 commit comments