nsivabalan commented on code in PR #11396:
URL: https://github.com/apache/hudi/pull/11396#discussion_r1635826968
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/QueryInfo.java:
##########
@@ -33,6 +37,7 @@ public class QueryInfo {
private final String previousInstant;
private final String startInstant;
private final String endInstant;
+ private final String predicateFilter;
Review Comment:
can we add some java docs on each entity here. We keep adding more and more.
some of them might be obvious.but good to have some docs.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/QueryInfo.java:
##########
@@ -43,10 +48,32 @@ public QueryInfo(
String startInstant, String endInstant,
String orderColumn, String keyColumn,
String limitColumn) {
+ this(
+ queryType,
+ previousInstant,
+ startInstant,
+ endInstant,
+ StringUtils.EMPTY_STRING,
+ orderColumn,
+ keyColumn,
+ limitColumn
+ );
+ }
+
+ public QueryInfo(
+ String queryType,
+ String previousInstant,
+ String startInstant,
+ String endInstant,
+ String predicateFilter,
Review Comment:
can we do new arg as Option<String> ?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java:
##########
@@ -367,8 +371,51 @@ public void
testHoodieIncrSourceWithDataSourceOptions(HoodieTableType tableType)
}
}
+ @Test
+ public void testPartitionPruningInHoodieIncrSource()
+ throws IOException {
+ this.tableType = MERGE_ON_READ;
+ metaClient = getHoodieMetaClient(storageConf(), basePath());
+ HoodieWriteConfig writeConfig = getConfigBuilder(basePath(), metaClient)
+
.withArchivalConfig(HoodieArchivalConfig.newBuilder().archiveCommitsWith(10,
12).build())
+
.withCleanConfig(HoodieCleanConfig.newBuilder().retainCommits(9).build())
+ .withCompactionConfig(
+ HoodieCompactionConfig.newBuilder()
+ .withScheduleInlineCompaction(true)
+ .withMaxNumDeltaCommitsBeforeCompaction(1)
+ .build())
+
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(true).build())
+ .build();
+ List<Pair<String, List<HoodieRecord>>> inserts = new ArrayList<>();
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig)) {
+ inserts.add(writeRecordsForPartition(writeClient, BULK_INSERT, "100",
DEFAULT_PARTITION_PATHS[0]));
+ inserts.add(writeRecordsForPartition(writeClient, BULK_INSERT, "200",
DEFAULT_PARTITION_PATHS[1]));
+ inserts.add(writeRecordsForPartition(writeClient, BULK_INSERT, "300",
DEFAULT_PARTITION_PATHS[2]));
+ // Go over all possible test cases to assert behaviour.
+ getArgsForPartitionPruningInHoodieIncrSource().forEach(argumentsStream
-> {
+ Object[] arguments = argumentsStream.get();
+ String checkpointToPullFromHoodieInstant = (String) arguments[0];
+ int maxRowsPerSnapshotBatch = (int) arguments[1];
+ String expectedCheckpointHoodieInstant = (String) arguments[2];
+ int expectedCount = (int) arguments[3];
+ int expectedFileParallelism = (int) arguments[4];
Review Comment:
should we name this var in line w/ the actual argument
"expectedRDDPartitions"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
