wombatu-kun commented on code in PR #19115:
URL: https://github.com/apache/hudi/pull/19115#discussion_r3497541715
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestSparkSampleWritesUtils.java:
##########
@@ -99,9 +106,63 @@ public void overwriteRecordSizeEstimateForEmptyTable() {
.build();
String commitTime = HoodieTestDataGenerator.getCommitTimeAtUTC(1);
+ // Input spans multiple source partitions; the sample-writes table must
still be non-partitioned.
JavaRDD<HoodieRecord> records =
jsc().parallelize(dataGen.generateInserts(commitTime, 2000), 2);
Option<HoodieWriteConfig> writeConfigOpt =
SparkSampleWritesUtils.getWriteConfigWithRecordSizeEstimate(jsc(),
Option.of(records), originalWriteConfig);
assertTrue(writeConfigOpt.isPresent());
- assertEquals(779.0,
writeConfigOpt.get().getCopyOnWriteRecordSizeEstimate(), 10.0);
+ assertEquals(337.0,
writeConfigOpt.get().getCopyOnWriteRecordSizeEstimate(), 10.0);
+ assertSampleWritesNonPartitioned();
+ }
+
+ @Test
+ void sampleWritesAreNonPartitionedEvenForManyPartitionInput() throws
IOException {
+ int recordsPerPartition = 50;
+ String[] partitionPaths = IntStream.range(0, 20)
+ .mapToObj(i -> String.format("year=2024/month=01/day=%02d", i + 1))
+ .toArray(String[]::new);
+ HoodieTestDataGenerator manyPartitionGen = new
HoodieTestDataGenerator(partitionPaths);
+
+ TypedProperties props = new TypedProperties();
+ props.put(HoodieStreamerConfig.SAMPLE_WRITES_ENABLED.key(), "true");
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withProperties(props)
+ .forTable("foo")
+ .withPath(basePath())
+ .withSchema(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA)
+ .build();
+
+ String commitTime = HoodieTestDataGenerator.getCommitTimeAtUTC(1);
+ List<HoodieRecord> allRecords = new ArrayList<>();
+ for (String partition : partitionPaths) {
+
allRecords.addAll(manyPartitionGen.generateInsertsForPartition(commitTime,
recordsPerPartition, partition));
+ }
+ JavaRDD<HoodieRecord> records = jsc().parallelize(allRecords, 4);
+
+ Option<HoodieWriteConfig> writeConfigOpt =
SparkSampleWritesUtils.getWriteConfigWithRecordSizeEstimate(jsc(),
Option.of(records), writeConfig);
+ assertTrue(writeConfigOpt.isPresent(), "Sample write should produce a
record-size estimate.");
+ assertSampleWritesNonPartitioned();
+ }
+
+ /**
+ * Fails if the sample-writes folder contains any source-partition
subdirectory, i.e. the
+ * sample write was not flattened into a single non-partitioned file.
+ */
+ private void assertSampleWritesNonPartitioned() throws IOException {
+ Path sampleWritesPath = new Path(basePath(),
".hoodie/.aux/.sample_writes");
Review Comment:
+1
--
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]
