nsivabalan commented on code in PR #18076:
URL: https://github.com/apache/hudi/pull/18076#discussion_r3548924114
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java:
##########
@@ -166,6 +166,15 @@ public class CloudSourceConfig extends HoodieConfig {
.sinceVersion("0.14.1")
.withDocumentation("specify this value in bytes, to coalesce partitions
of source dataset not greater than specified limit");
+ public static final ConfigProperty<Long> SOURCE_MAX_FILES_PER_SYNC =
ConfigProperty
+ .key(STREAMER_CONFIG_PREFIX + "source.cloud.data.max.files.per.sync")
+ .defaultValue(10_000_000L)
Review Comment:
10M is too high right. can we set something like 10k.
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java:
##########
@@ -299,27 +318,37 @@ public static Pair<CloudObjectIncrCheckpoint,
Option<Dataset<Row>>> filterAndGen
}
}
- // Limit based on sourceLimit
+ // Compute cumulative size and cumulative row count over the same ordered
window so that
+ // both the byte-based and files-based limits select a contiguous prefix
of the ordered set.
+ // Applying both predicates in a single window pass keeps the result
deterministic across
+ // executors and avoids a post-hoc limit() on an unordered dataset.
WindowSpec windowSpec = Window.orderBy(col(queryInfo.getOrderColumn()),
col(queryInfo.getKeyColumn()));
- // Add the 'cumulativeSize' column with running sum of 'limitColumn'
- Dataset<Row> aggregatedData = orderedDf.withColumn(CUMULATIVE_COLUMN_NAME,
- sum(col(queryInfo.getLimitColumn())).over(windowSpec));
- Dataset<Row> collectedRows =
aggregatedData.filter(col(CUMULATIVE_COLUMN_NAME).leq(sourceLimit));
+ Dataset<Row> aggregatedData = orderedDf
+ .withColumn(CUMULATIVE_COLUMN_NAME,
sum(col(queryInfo.getLimitColumn())).over(windowSpec))
+ .withColumn(CUMULATIVE_COUNT_COLUMN_NAME,
row_number().over(windowSpec));
+ Dataset<Row> collectedRows = aggregatedData
Review Comment:
can we rename `collectedRows` -> `trimmedRows`
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java:
##########
@@ -166,6 +166,15 @@ public class CloudSourceConfig extends HoodieConfig {
.sinceVersion("0.14.1")
.withDocumentation("specify this value in bytes, to coalesce partitions
of source dataset not greater than specified limit");
+ public static final ConfigProperty<Long> SOURCE_MAX_FILES_PER_SYNC =
ConfigProperty
+ .key(STREAMER_CONFIG_PREFIX + "source.cloud.data.max.files.per.sync")
+ .defaultValue(10_000_000L)
+ .markAdvanced()
+ .sinceVersion("1.4.0")
Review Comment:
1.3.0 ?
--
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]
