This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 014358a3e6c8 [HUDI-18075] Cap files per batch in cloud incremental
source (#18076)
014358a3e6c8 is described below
commit 014358a3e6c8a954e3d72eeaa7033e2aa18adc43
Author: Lokesh Jain <[email protected]>
AuthorDate: Mon Jul 13 18:09:20 2026 +0530
[HUDI-18075] Cap files per batch in cloud incremental source (#18076)
The cloud incremental source only limits bytes per batch, so with
small files the driver can hold unbounded file metadata and OOM.
Adds hoodie.streamer.source.cloud.data.num.files.per.batch to cap
files per sync. Applied deterministically via a window predicate so
the checkpoint advances consistently with the emitted batch.
---
.../hudi/utilities/config/CloudSourceConfig.java | 9 ++
.../sources/helpers/CloudDataFetcher.java | 10 +-
.../sources/helpers/IncrSourceHelper.java | 129 ++++++++++++--------
.../sources/TestS3EventsHoodieIncrSource.java | 36 ++++++
.../sources/helpers/TestIncrSourceHelper.java | 131 +++++++++++++++++++++
5 files changed, 267 insertions(+), 48 deletions(-)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
index 6ffd5ad5c7ae..46393d64fa32 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/CloudSourceConfig.java
+++
b/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_000L)
+ .markAdvanced()
+ .sinceVersion("1.4.0")
+ .withDocumentation("Maximum number of files to consume per sync round
when using the cloud incremental source. "
+ + "This limit is applied in addition to the byte-based source limit
to prevent driver OOM when processing "
+ + "many small files. The sync will process files up to whichever
limit is reached first.");
+
public static final ConfigProperty<Integer> MAX_FETCH_TIME_PER_SYNC_SECS =
ConfigProperty
.key(STREAMER_CONFIG_PREFIX +
"source.cloud.meta.max.fetch.time.per.sync.secs")
.defaultValue(60)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java
index 396b3f67dbce..e6a77eaabf81 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudDataFetcher.java
@@ -23,6 +23,7 @@ import org.apache.hudi.common.table.checkpoint.Checkpoint;
import org.apache.hudi.common.table.checkpoint.StreamerCheckpointV1;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
@@ -43,6 +44,7 @@ import static
org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.DATAFILE_FORMAT;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.ENABLE_EXISTS_CHECK;
import static
org.apache.hudi.utilities.config.CloudSourceConfig.SOURCE_MAX_BYTES_PER_PARTITION;
+import static
org.apache.hudi.utilities.config.CloudSourceConfig.SOURCE_MAX_FILES_PER_SYNC;
import static
org.apache.hudi.utilities.config.HoodieIncrSourceConfig.SOURCE_FILE_FORMAT;
/**
@@ -101,10 +103,14 @@ public class CloudDataFetcher implements Serializable {
log.info("Adding filter string to Dataset: {}", filter);
Dataset<Row> filteredSourceData =
queryInfoDatasetPair.getRight().filter(filter);
- log.info("Adjusting end checkpoint:{} based on sourceLimit :{}",
queryInfo.getEndInstant(), sourceLimit);
+ long numFilesLimit = props.getLong(SOURCE_MAX_FILES_PER_SYNC.key(),
SOURCE_MAX_FILES_PER_SYNC.defaultValue());
+ ValidationUtils.checkArgument(numFilesLimit >= 1,
+ SOURCE_MAX_FILES_PER_SYNC.key() + " must be >= 1, got: " +
numFilesLimit);
+ log.info("Adjusting end checkpoint:{} based on sourceLimit:{} and
numFilesLimit:{}",
+ queryInfo.getEndInstant(), sourceLimit, numFilesLimit);
Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> checkPointAndDataset
=
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(
- filteredSourceData, sourceLimit, queryInfo,
cloudObjectIncrCheckpoint);
+ filteredSourceData, sourceLimit, numFilesLimit, queryInfo,
cloudObjectIncrCheckpoint);
if (!checkPointAndDataset.getRight().isPresent()) {
log.info("Empty source, returning endpoint:{}",
checkPointAndDataset.getLeft());
return Pair.of(Option.empty(), new
StreamerCheckpointV1(checkPointAndDataset.getLeft().toString()));
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
index 3428098f112d..148583cbd9cd 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/IncrSourceHelper.java
@@ -33,6 +33,7 @@ import org.apache.hudi.common.table.timeline.TimelineUtils;
import
org.apache.hudi.common.table.timeline.TimelineUtils.HollowCommitHandling;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer;
@@ -64,6 +65,7 @@ import static
org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
import static
org.apache.hudi.utilities.config.HoodieIncrSourceConfig.MISSING_CHECKPOINT_STRATEGY;
import static
org.apache.hudi.utilities.config.HoodieIncrSourceConfig.READ_LATEST_INSTANT_ON_MISSING_CKPT;
import static org.apache.spark.sql.functions.col;
+import static org.apache.spark.sql.functions.row_number;
import static org.apache.spark.sql.functions.sum;
@Slf4j
@@ -71,6 +73,7 @@ public class IncrSourceHelper {
public static final String DEFAULT_START_TIMESTAMP =
HoodieTimeline.INIT_INSTANT_TS;
private static final String CUMULATIVE_COLUMN_NAME = "cumulativeSize";
+ private static final String CUMULATIVE_COUNT_COLUMN_NAME = "cumulativeCount";
/**
* When hollow commits are found while using incremental source with {@link
HoodieDeltaStreamer},
@@ -259,9 +262,25 @@ public class IncrSourceHelper {
* @param queryInfo Query Info
* @return end instants along with filtered rows.
*/
- public static Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>>
filterAndGenerateCheckpointBasedOnSourceLimit(Dataset<Row> sourceData,
+ @VisibleForTesting
+ static Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>>
filterAndGenerateCheckpointBasedOnSourceLimit(Dataset<Row> sourceData,
long sourceLimit, QueryInfo queryInfo,
CloudObjectIncrCheckpoint
cloudObjectIncrCheckpoint) {
+ return filterAndGenerateCheckpointBasedOnSourceLimit(sourceData,
sourceLimit, Long.MAX_VALUE, queryInfo, cloudObjectIncrCheckpoint);
+ }
+
+ /**
+ * Adjust the source dataset to size based batch based on last checkpoint
key.
+ *
+ * @param sourceData Source dataset
+ * @param sourceLimit Max number of bytes to be read from source
+ * @param numFilesLimit Max number of files to be read from source in this
batch
+ * @param queryInfo Query Info
+ * @return end instants along with filtered rows.
+ */
+ public static Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>>
filterAndGenerateCheckpointBasedOnSourceLimit(Dataset<Row> sourceData,
+
long sourceLimit, long numFilesLimit,
QueryInfo queryInfo,
+
CloudObjectIncrCheckpoint
cloudObjectIncrCheckpoint) {
if (sourceData.isEmpty()) {
// There is no file matching the prefix.
CloudObjectIncrCheckpoint updatedCheckpoint =
@@ -272,54 +291,72 @@ public class IncrSourceHelper {
}
// Let's persist the dataset to avoid triggering the dag repeatedly
sourceData.persist(StorageLevel.MEMORY_AND_DISK());
- // Set ordering in query to enable batching
- Dataset<Row> orderedDf = QueryRunner.applyOrdering(sourceData,
queryInfo.getOrderByColumns());
- Option<String> lastCheckpoint =
Option.of(cloudObjectIncrCheckpoint.getCommit());
- Option<String> lastCheckpointKey =
Option.ofNullable(cloudObjectIncrCheckpoint.getKey());
- Option<String> concatenatedKey = lastCheckpoint.flatMap(checkpoint ->
lastCheckpointKey.map(key -> checkpoint + key));
-
- // Filter until last checkpoint key
- if (concatenatedKey.isPresent()) {
- orderedDf = orderedDf.withColumn("commit_key",
- functions.concat(functions.col(queryInfo.getOrderColumn()),
functions.col(queryInfo.getKeyColumn())));
- // Apply incremental filter
- orderedDf =
orderedDf.filter(functions.col("commit_key").gt(concatenatedKey.get())).drop("commit_key");
- // If there are no more files where commit_key is greater than
lastCheckpointCommit#lastCheckpointKey
- if (orderedDf.isEmpty()) {
- log.info("Empty ordered source, returning endpoint: {}",
queryInfo.getEndInstant());
- sourceData.unpersist();
- // queryInfo.getEndInstant() represents source table's last completed
instant
- // If current checkpoint is c1#abc and queryInfo.getEndInstant() is
c1, return c1#abc.
- // If current checkpoint is c1#abc and queryInfo.getEndInstant() is
c2, return c2.
- CloudObjectIncrCheckpoint updatedCheckpoint =
-
queryInfo.getEndInstant().equals(cloudObjectIncrCheckpoint.getCommit())
- ? cloudObjectIncrCheckpoint
- : new CloudObjectIncrCheckpoint(queryInfo.getEndInstant(),
null);
- return Pair.of(updatedCheckpoint, Option.empty());
+ try {
+ // Set ordering in query to enable batching
+ Dataset<Row> orderedDf = QueryRunner.applyOrdering(sourceData,
queryInfo.getOrderByColumns());
+ Option<String> lastCheckpoint =
Option.of(cloudObjectIncrCheckpoint.getCommit());
+ Option<String> lastCheckpointKey =
Option.ofNullable(cloudObjectIncrCheckpoint.getKey());
+ Option<String> concatenatedKey = lastCheckpoint.flatMap(checkpoint ->
lastCheckpointKey.map(key -> checkpoint + key));
+
+ // Filter until last checkpoint key
+ if (concatenatedKey.isPresent()) {
+ orderedDf = orderedDf.withColumn("commit_key",
+ functions.concat(functions.col(queryInfo.getOrderColumn()),
functions.col(queryInfo.getKeyColumn())));
+ // Apply incremental filter
+ orderedDf =
orderedDf.filter(functions.col("commit_key").gt(concatenatedKey.get())).drop("commit_key");
+ // If there are no more files where commit_key is greater than
lastCheckpointCommit#lastCheckpointKey
+ if (orderedDf.isEmpty()) {
+ log.info("Empty ordered source, returning endpoint: {}",
queryInfo.getEndInstant());
+ // queryInfo.getEndInstant() represents source table's last
completed instant
+ // If current checkpoint is c1#abc and queryInfo.getEndInstant() is
c1, return c1#abc.
+ // If current checkpoint is c1#abc and queryInfo.getEndInstant() is
c2, return c2.
+ CloudObjectIncrCheckpoint updatedCheckpoint =
+
queryInfo.getEndInstant().equals(cloudObjectIncrCheckpoint.getCommit())
+ ? cloudObjectIncrCheckpoint
+ : new CloudObjectIncrCheckpoint(queryInfo.getEndInstant(),
null);
+ return Pair.of(updatedCheckpoint, Option.empty());
+ }
}
- }
- // Limit based on sourceLimit
- 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));
-
- Row row = null;
- if (collectedRows.isEmpty()) {
- // If the first element itself exceeds limits then return first element
- log.info("First object exceeding source limit: {} bytes", sourceLimit);
- row = aggregatedData.select(queryInfo.getOrderColumn(),
queryInfo.getKeyColumn(), CUMULATIVE_COLUMN_NAME).first();
- collectedRows = aggregatedData.limit(1);
- } else {
- // Get the last row and form composite key
- row = collectedRows.select(queryInfo.getOrderColumn(),
queryInfo.getKeyColumn(), CUMULATIVE_COLUMN_NAME).orderBy(
- col(queryInfo.getOrderColumn()).desc(),
col(queryInfo.getKeyColumn()).desc()).first();
+ // 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.
+ // Note: Window.orderBy() without partitionBy() triggers Spark's
SinglePartition exchange —
+ // the entire filteredSourceData is held on one executor for the window
pass. This is
+ // intentional (running sum and row_number() require a total ordering),
but it bounds how
+ // large numFilesLimit can usefully go. A follow-up could compute the
prefix via
+ // mapPartitions + driver-merge to avoid the single-partition collapse.
+ WindowSpec windowSpec = Window.orderBy(col(queryInfo.getOrderColumn()),
col(queryInfo.getKeyColumn()));
+ 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
+ .filter(col(CUMULATIVE_COLUMN_NAME).leq(sourceLimit)
+ .and(col(CUMULATIVE_COUNT_COLUMN_NAME).leq(numFilesLimit)));
+
+ Row row;
+ if (collectedRows.isEmpty()) {
+ // The first file itself exceeds the byte limit. Since numFilesLimit
>= 1 is enforced at
+ // the call site and row_number() starts at 1, the files limit can
never exclude the first
+ // file. We deliberately take one file even when it exceeds the byte
limit, matching
+ // pre-existing behavior.
+ log.info("First object exceeds source limit: {} bytes", sourceLimit);
+ collectedRows =
aggregatedData.filter(col(CUMULATIVE_COUNT_COLUMN_NAME).equalTo(1));
+ row = collectedRows.select(queryInfo.getOrderColumn(),
queryInfo.getKeyColumn(), CUMULATIVE_COLUMN_NAME).first();
+ } else {
+ // Get the last row and form composite key
+ row = collectedRows.select(queryInfo.getOrderColumn(),
queryInfo.getKeyColumn(), CUMULATIVE_COLUMN_NAME).orderBy(
+ col(queryInfo.getOrderColumn()).desc(),
col(queryInfo.getKeyColumn()).desc()).first();
+ }
+ log.info("Processed batch size: {} bytes",
row.get(row.fieldIndex(CUMULATIVE_COLUMN_NAME)));
+ // Drop the helper count column so returned dataset shape matches
pre-existing contract
+ // (only the source columns plus cumulativeSize).
+ return Pair.of(new CloudObjectIncrCheckpoint(row.getString(0),
row.getString(1)),
+ Option.of(collectedRows.drop(CUMULATIVE_COUNT_COLUMN_NAME)));
+ } finally {
+ sourceData.unpersist();
}
- log.info("Processed batch size: {} bytes",
row.get(row.fieldIndex(CUMULATIVE_COLUMN_NAME)));
- sourceData.unpersist();
- return Pair.of(new CloudObjectIncrCheckpoint(row.getString(0),
row.getString(1)), Option.of(collectedRows));
}
/**
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
index 4218aaca26d9..42274c859717 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestS3EventsHoodieIncrSource.java
@@ -347,6 +347,42 @@ public class TestS3EventsHoodieIncrSource extends
S3EventsHoodieIncrSourceHarnes
Assertions.assertEquals(numPartitions,
argumentCaptorForMetrics.getAllValues());
}
+ @Test
+ public void testFilesLimitCheckpointConsistency() throws IOException {
+ String commitTimeForWrites = "2";
+ String commitTimeForReads = "1";
+
+ writeS3MetadataRecords(commitTimeForReads);
+ writeS3MetadataRecords(commitTimeForWrites);
+
+ // 5 files all within the same commit with 100 bytes each
+ List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file1.json", 100L, "1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file2.json", 100L, "1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file3.json", 100L, "1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file4.json", 100L, "1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file5.json", 100L, "1"));
+
+ Dataset<Row> inputDs = generateDataset(filePathSizeAndCommitTime);
+
+ setMockQueryRunner(inputDs);
+ when(mockCloudObjectsSelectorCommon.loadAsDataset(Mockito.any(),
Mockito.any(), Mockito.any(), Mockito.eq(schemaProvider),
Mockito.anyInt())).thenReturn(Option.empty());
+ when(sourceProfileSupplier.getSourceProfile()).thenReturn(null);
+
+ // With a large byte limit all 5 files fit, but
SOURCE_MAX_FILES_PER_SYNC=3 caps processing to
+ // the first 3 files. The checkpoint must be recalculated to reflect the
last file actually
+ // processed (file3), not the byte-limit checkpoint (file5).
+ TypedProperties propsWithFilesLimit = setProps(READ_UPTO_LATEST_COMMIT);
+
propsWithFilesLimit.setProperty(CloudSourceConfig.SOURCE_MAX_FILES_PER_SYNC.key(),
"3");
+ readAndAssert(READ_UPTO_LATEST_COMMIT, Option.of(commitTimeForReads),
10000L,
+ "1#path/to/file3.json", propsWithFilesLimit);
+
+ // Without a files limit, all 5 files are processed and the checkpoint
points to the last file.
+ TypedProperties propsWithoutFilesLimit = setProps(READ_UPTO_LATEST_COMMIT);
+ readAndAssert(READ_UPTO_LATEST_COMMIT, Option.of(commitTimeForReads),
10000L,
+ "1#path/to/file5.json", propsWithoutFilesLimit);
+ }
+
@Test
public void testUnsupportedCheckpoint() {
TypedProperties typedProperties = setProps(READ_UPTO_LATEST_COMMIT);
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
index 49c4ffcb0bfc..cb4e6a8708ed 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestIncrSourceHelper.java
@@ -286,6 +286,137 @@ class TestIncrSourceHelper extends
SparkClientFunctionalTestHarness {
assertTrue(!result.getRight().isPresent());
}
+ /**
+ * Files-limit must produce a contiguous prefix of the ordered set even when
the source dataset
+ * is spread across many Spark partitions. Spans 50 files across 8
partitions; iterates 5 syncs
+ * with numFilesLimit = 10 and asserts every file is consumed exactly once
in order.
+ */
+ @Test
+ void testFilesLimitContiguousAcrossManyPartitions() {
+ List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
+ for (int i = 0; i < 50; i++) {
+ // zero-pad so lexicographic key order matches insertion order
+
filePathSizeAndCommitTime.add(Triple.of(String.format("path/to/file%02d.json",
i), 100L, "commit1"));
+ }
+ Dataset<Row> inputDs = generateDataset(filePathSizeAndCommitTime, 8);
+
+ QueryInfo queryInfo = new QueryInfo(
+ QUERY_TYPE_INCREMENTAL_OPT_VAL(), "commit0", "commit0",
+ "commit1", "_hoodie_commit_time",
+ "s3.object.key", "s3.object.size");
+
+ CloudObjectIncrCheckpoint cursor = new
CloudObjectIncrCheckpoint("commit0", null);
+ List<String> consumedKeys = new ArrayList<>();
+ long byteLimit = 1_000_000L;
+ long filesLimit = 10L;
+ for (int batch = 0; batch < 5; batch++) {
+ Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> result =
+
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(inputDs,
byteLimit, filesLimit, queryInfo, cursor);
+ assertTrue(result.getRight().isPresent(), "batch " + batch + "
unexpectedly empty");
+ List<Row> rows = result.getRight().get()
+ .select("s3.object.key")
+ .collectAsList();
+ assertEquals(filesLimit, rows.size(), "batch " + batch + " did not
respect filesLimit");
+ for (Row r : rows) {
+ consumedKeys.add(r.getString(0));
+ }
+ cursor = result.getKey();
+ }
+ assertEquals(50, consumedKeys.size());
+ // Verify keys are returned in the exact order produced; if .limit() on an
unordered dataset
+ // had been used, a multi-partition input would yield duplicates or gaps
under any
+ // non-deterministic order.
+ for (int i = 0; i < 50; i++) {
+ assertEquals(String.format("path/to/file%02d.json", i),
consumedKeys.get(i),
+ "file at position " + i + " is out of order");
+ }
+ }
+
+ /**
+ * Files-limit hits mid-commit. After the truncated batch, next sync must
resume from the
+ * exact next file in the same commit without skipping or re-reading.
+ */
+ @Test
+ void testFilesLimitCrossesCommitBoundary() {
+ List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
+ // commit1: 3 files; commit2: 3 files
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file1.json", 100L,
"commit1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file2.json", 100L,
"commit1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file3.json", 100L,
"commit1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file4.json", 100L,
"commit2"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file5.json", 100L,
"commit2"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file6.json", 100L,
"commit2"));
+ Dataset<Row> inputDs = generateDataset(filePathSizeAndCommitTime, 4);
+
+ QueryInfo queryInfo = new QueryInfo(
+ QUERY_TYPE_INCREMENTAL_OPT_VAL(), "commit0", "commit0",
+ "commit2", "_hoodie_commit_time",
+ "s3.object.key", "s3.object.size");
+
+ // First batch: filesLimit = 4 truncates at file4 (first file of commit2)
+ Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> first =
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(
+ inputDs, 1_000_000L, 4L, queryInfo, new
CloudObjectIncrCheckpoint("commit0", null));
+ assertEquals("commit2#path/to/file4.json", first.getKey().toString());
+ assertEquals(4, first.getRight().get().count());
+
+ // Second batch: resume from the prior checkpoint; remaining files are
file5 and file6
+ Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> second =
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(
+ inputDs, 1_000_000L, 4L, queryInfo, first.getKey());
+ assertEquals("commit2#path/to/file6.json", second.getKey().toString());
+ assertEquals(2, second.getRight().get().count());
+ }
+
+ /**
+ * Files-limit alone (when bytes-limit is non-binding) still selects a
contiguous prefix.
+ */
+ @Test
+ void testFilesLimitBindingOverByteLimit() {
+ List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+
filePathSizeAndCommitTime.add(Triple.of(String.format("path/to/file%02d.json",
i), 100L, "commit1"));
+ }
+ Dataset<Row> inputDs = generateDataset(filePathSizeAndCommitTime, 4);
+
+ QueryInfo queryInfo = new QueryInfo(
+ QUERY_TYPE_INCREMENTAL_OPT_VAL(), "commit0", "commit0",
+ "commit1", "_hoodie_commit_time",
+ "s3.object.key", "s3.object.size");
+
+ Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> result =
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(
+ inputDs, 1_000_000L, 3L, queryInfo, new
CloudObjectIncrCheckpoint("commit0", null));
+ assertEquals("commit1#path/to/file02.json", result.getKey().toString());
+ assertEquals(3, result.getRight().get().count());
+ }
+
+ /**
+ * Files-limit larger than dataset: byte-limit drives the checkpoint as if
no files-limit
+ * existed. Guards against the previous size-based recalculation branch
silently breaking
+ * the byte-only path.
+ */
+ @Test
+ void testFilesLimitLargerThanAvailable() {
+ List<Triple<String, Long, String>> filePathSizeAndCommitTime = new
ArrayList<>();
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file1.json", 100L,
"commit1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file2.json", 100L,
"commit1"));
+ filePathSizeAndCommitTime.add(Triple.of("path/to/file3.json", 100L,
"commit1"));
+ Dataset<Row> inputDs = generateDataset(filePathSizeAndCommitTime, 2);
+
+ QueryInfo queryInfo = new QueryInfo(
+ QUERY_TYPE_INCREMENTAL_OPT_VAL(), "commit0", "commit0",
+ "commit1", "_hoodie_commit_time",
+ "s3.object.key", "s3.object.size");
+
+ Pair<CloudObjectIncrCheckpoint, Option<Dataset<Row>>> result =
IncrSourceHelper.filterAndGenerateCheckpointBasedOnSourceLimit(
+ inputDs, 1_000_000L, 1000L, queryInfo, new
CloudObjectIncrCheckpoint("commit0", null));
+ assertEquals("commit1#path/to/file3.json", result.getKey().toString());
+ assertEquals(3, result.getRight().get().count());
+ }
+
+ private Dataset<Row> generateDataset(List<Triple<String, Long, String>>
filePathSizeAndCommitTime, int numPartitions) {
+ JavaRDD<String> testRdd =
jsc.parallelize(getSampleS3ObjectKeys(filePathSizeAndCommitTime),
numPartitions);
+ return spark().read().json(testRdd);
+ }
+
private HoodieRecord generateS3EventMetadata(String commitTime, String
bucketName, String objectKey, Long objectSize) {
String partitionPath = bucketName;
HoodieSchema schema = S3_METADATA_SCHEMA;
