@@ -1191,27 +1191,21 @@ void testStreamingReadPartitionPushDownWithInExpr() throws Exception {
11911191 .collect (Collectors .toList ());
11921192 waitUntilAllBucketFinishSnapshot (admin , tablePath , Arrays .asList ("2025" , "2026" , "2027" ));
11931193
1194- String plan =
1195- tEnv .explainSql ("select * from partitioned_table_in where c in ('2025','2026')" );
1194+ String query1 = "select * from partitioned_table_in where c in ('2025','2026')" ;
1195+ String plan = tEnv .explainSql (query1 );
11961196 assertThat (plan )
11971197 .contains (
11981198 "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_in, filter=[OR(=(c, _UTF-16LE'2025'), =(c, _UTF-16LE'2026'))]]], fields=[a, b, c])" );
11991199
1200- org .apache .flink .util .CloseableIterator <Row > rowIter =
1201- tEnv .executeSql ("select * from partitioned_table_in where c in ('2025','2026')" )
1202- .collect ();
1203-
1200+ org .apache .flink .util .CloseableIterator <Row > rowIter = tEnv .executeSql (query1 ).collect ();
12041201 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
12051202
1206- plan = tEnv .explainSql ("select * from partitioned_table_in where c ='2025' or c ='2026'" );
1203+ String query2 = "select * from partitioned_table_in where c ='2025' or c ='2026'" ;
1204+ plan = tEnv .explainSql (query2 );
12071205 assertThat (plan )
12081206 .contains (
12091207 "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_in, filter=[OR(=(c, _UTF-16LE'2025':VARCHAR(2147483647) CHARACTER SET \" UTF-16LE\" ), =(c, _UTF-16LE'2026':VARCHAR(2147483647) CHARACTER SET \" UTF-16LE\" ))]]], fields=[a, b, c])" );
1210-
1211- rowIter =
1212- tEnv .executeSql ("select * from partitioned_table_in where c ='2025' or c ='2026'" )
1213- .collect ();
1214-
1208+ rowIter = tEnv .executeSql (query2 ).collect ();
12151209 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
12161210 }
12171211
@@ -1240,7 +1234,6 @@ void testStreamingReadWithCombinedFiltersAndInExpr() throws Exception {
12401234 expectedRowValues .add (String .format ("+I[%d, 2026, %d]" , i , i * 100 ));
12411235 }
12421236 }
1243- writeRows (conn , tablePath , rows , false );
12441237
12451238 for (int i = 0 ; i < 10 ; i ++) {
12461239 rows .add (row (i , "v" + i , "2027" , i * 100 ));
@@ -1249,20 +1242,15 @@ void testStreamingReadWithCombinedFiltersAndInExpr() throws Exception {
12491242 writeRows (conn , tablePath , rows , false );
12501243 waitUntilAllBucketFinishSnapshot (admin , tablePath , Arrays .asList ("2025" , "2026" , "2027" ));
12511244
1252- String plan =
1253- tEnv . explainSql (
1254- "select a,c,d from combined_filters_table_in where c in ('2025','2026') and d % 200 = 0" );
1245+ String query1 =
1246+ "select a,c,d from combined_filters_table_in where c in ('2025','2026') and d % 200 = 0" ;
1247+ String plan = tEnv . explainSql ( query1 );
12551248 assertThat (plan )
12561249 .contains (
12571250 "TableSourceScan(table=[[testcatalog, defaultdb, combined_filters_table_in, filter=[OR(=(c, _UTF-16LE'2025'), =(c, _UTF-16LE'2026'))], project=[a, c, d]]], fields=[a, c, d])" );
12581251
12591252 // test column filter、partition filter and flink runtime filter
1260- org .apache .flink .util .CloseableIterator <Row > rowIter =
1261- tEnv .executeSql (
1262- "select a,c,d from combined_filters_table_in where c in ('2025','2026') "
1263- + "and d % 200 = 0" )
1264- .collect ();
1265-
1253+ org .apache .flink .util .CloseableIterator <Row > rowIter = tEnv .executeSql (query1 ).collect ();
12661254 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
12671255
12681256 rowIter =
@@ -1293,41 +1281,35 @@ void testStreamingReadPartitionPushDownWithLikeExpr() throws Exception {
12931281 .collect (Collectors .toList ());
12941282 waitUntilAllBucketFinishSnapshot (admin , tablePath , Arrays .asList ("2025" , "2026" , "3026" ));
12951283
1296- String plan = tEnv .explainSql ("select * from partitioned_table_like where c like '202%'" );
1284+ String query1 = "select * from partitioned_table_like where c like '202%'" ;
1285+ String plan = tEnv .explainSql (query1 );
12971286 assertThat (plan )
12981287 .contains (
12991288 "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_like, filter=[LIKE(c, _UTF-16LE'202%')]]], fields=[a, b, c])" );
13001289
1301- org .apache .flink .util .CloseableIterator <Row > rowIter =
1302- tEnv .executeSql ("select * from partitioned_table_like where c like '202%'" )
1303- .collect ();
1290+ org .apache .flink .util .CloseableIterator <Row > rowIter = tEnv .executeSql (query1 ).collect ();
13041291
13051292 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
13061293 expectedRowValues =
13071294 allData .stream ()
13081295 .filter (s -> s .contains ("2026" ) || s .contains ("3026" ))
13091296 .collect (Collectors .toList ());
1310- plan = tEnv .explainSql ("select * from partitioned_table_like where c like '%026'" );
1297+ String query2 = "select * from partitioned_table_like where c like '%026'" ;
1298+ plan = tEnv .explainSql (query2 );
13111299 assertThat (plan )
13121300 .contains (
13131301 "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_like, filter=[LIKE(c, _UTF-16LE'%026')]]], fields=[a, b, c])" );
1314-
1315- rowIter =
1316- tEnv .executeSql ("select * from partitioned_table_like where c like '%026'" )
1317- .collect ();
1318-
1302+ rowIter = tEnv .executeSql (query2 ).collect ();
13191303 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
13201304
13211305 expectedRowValues =
13221306 allData .stream ().filter (s -> s .contains ("3026" )).collect (Collectors .toList ());
1323- plan = tEnv .explainSql ("select * from partitioned_table_like where c like '%3026%'" );
1307+ String query3 = "select * from partitioned_table_like where c like '%3026%'" ;
1308+ plan = tEnv .explainSql (query3 );
13241309 assertThat (plan )
13251310 .contains (
13261311 "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_like, filter=[LIKE(c, _UTF-16LE'%3026%')]]], fields=[a, b, c])" );
1327-
1328- rowIter =
1329- tEnv .executeSql ("select * from partitioned_table_like where c like '%3026%'" )
1330- .collect ();
1312+ rowIter = tEnv .executeSql (query3 ).collect ();
13311313
13321314 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
13331315 }
@@ -1353,22 +1335,17 @@ void testStreamingReadPartitionComplexPushDown() throws Exception {
13531335 waitUntilAllBucketFinishSnapshot (
13541336 admin , tablePath , Arrays .asList ("2025$1" , "2025$2" , "2026$1" ));
13551337
1356- String plan =
1357- tEnv . explainSql (
1358- "select * from partitioned_table_complex where a = 3 \n "
1359- + " and (c in ('2026') or d like '%1%') "
1360- + " and b like '%v3%'" );
1338+ String query =
1339+ "select * from partitioned_table_complex where a = 3 \n "
1340+ + " and (c in ('2026') or d like '%1%') "
1341+ + " and b like '%v3%'" ;
1342+ String plan = tEnv . explainSql ( query );
13611343 assertThat (plan )
13621344 .contains (
13631345 "Calc(select=[3 AS a, b, c, d], where=[((a = 3) AND LIKE(b, '%v3%'))])\n "
13641346 + "+- TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_complex, filter=[OR(=(c, _UTF-16LE'2026'), LIKE(d, _UTF-16LE'%1%'))]]], fields=[a, b, c, d])" );
13651347
1366- org .apache .flink .util .CloseableIterator <Row > rowIter =
1367- tEnv .executeSql (
1368- "select * from partitioned_table_complex where a = 3\n "
1369- + " and (c in ('2026') or d like '%1%') "
1370- + " and b like '%v3%'" )
1371- .collect ();
1348+ org .apache .flink .util .CloseableIterator <Row > rowIter = tEnv .executeSql (query ).collect ();
13721349
13731350 assertResultsIgnoreOrder (rowIter , expectedRowValues , true );
13741351 }
0 commit comments