yihua commented on code in PR #19512:
URL: https://github.com/apache/hudi/pull/19512#discussion_r3725273941
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java:
##########
@@ -482,6 +482,20 @@ private void writeFields(InternalRow row, StructType
schema, ValueWriter[] field
}
}
+ private static int decimalFixedLen(HoodieSchema resolvedSchema, int
precision) {
Review Comment:
Done, renamed to `resolveDecimalByteLength`.
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java:
##########
@@ -442,4 +463,137 @@ private void commitToTable(String instant,
List<WriteStatus> writeStatuses) {
client.commitStats(instant, writeStats, Option.empty(),
metaClient.getCommitActionType());
assertTrue(committed);
}
+
+ // Avro fixed(10) decimal(20,2). 10 is wider than the precision-minimal
width (9 for precision
+ // 20), so Spark's own DecimalType->Avro conversion would emit fixed(9); the
declared 10 only
+ // survives if the write path honors the Avro fixed size. Do not derive this
schema from a
+ // DataFrame, which would drop the fixed size.
+ private static final String FIXED10_DECIMAL_SCHEMA =
+ "{\"type\":\"record\",\"name\":\"decimalRec\",\"fields\":["
+ + "{\"name\":\"_row_key\",\"type\":\"string\"},"
+ + "{\"name\":\"partition_path\",\"type\":\"string\"},"
+ + "{\"name\":\"ts\",\"type\":\"long\"},"
+ +
"{\"name\":\"dec\",\"type\":{\"type\":\"fixed\",\"name\":\"decFixed\",\"size\":10,"
+ + "\"logicalType\":\"decimal\",\"precision\":20,\"scale\":2}}]}";
+
+ private static final String DECIMAL_PARTITION = "p1";
+ private static final int EXPECTED_DECIMAL_FIXED_LEN = 10;
+
+ @Test
+ void testDecimalFixedWidthPreservedAfterCompactionAndClustering() throws
Exception {
+ Properties props = getPropertiesForKeyGen(true);
+ Properties rowWriterProps = new Properties();
+ rowWriterProps.put("hoodie.datasource.write.row.writer.enable", "true");
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .forTable("test-decimal-fixed")
+ .withPath(basePath())
+ .withSchema(FIXED10_DECIMAL_SCHEMA)
+ .withParallelism(2, 2)
+ .withPreCombineField("ts")
+ .withProperties(rowWriterProps)
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withMaxNumDeltaCommitsBeforeCompaction(1)
+ .compactionSmallFileSize(0)
+ .withInlineCompaction(false)
+ .build())
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+ .withClusteringMaxNumGroups(10)
+ .withClusteringTargetPartitions(0)
+ .withInlineClustering(false)
+ .withInlineClusteringNumCommits(1)
+ .build())
+ .build();
+ props.putAll(config.getProps());
+
+ metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, props);
+ client = getHoodieWriteClient(config);
+
+ // two insert commits (small-file size 0 forces a fresh file group each)
create two base-file
+ // groups, both written via the Avro path at fixed(10)
+ String instant1 = WriteClientTestUtils.createNewInstantTime();
+ writeData(instant1, buildDecimalRecords(0, 10, 1L, new
BigDecimal("123456789.12")), true);
+ String instant2 = WriteClientTestUtils.createNewInstantTime();
+ writeData(instant2, buildDecimalRecords(10, 10, 1L, new
BigDecimal("223456789.34")), true);
+ // update every key so both groups accumulate log files for compaction to
merge
+ String instant3 = WriteClientTestUtils.createNewInstantTime();
+ writeData(instant3, buildDecimalRecords(0, 20, 2L, new
BigDecimal("323456789.56")), true);
+
+ // precondition: both file groups must carry log files, else compaction is
a no-op and would not
+ // exercise the row-writer merge path
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context(),
metaClient);
+ hoodieTable.getHoodieView().sync();
+ List<FileSlice> latestSlices =
+
hoodieTable.getHoodieView().getLatestFileSlices(DECIMAL_PARTITION).collect(Collectors.toList());
+ assertEquals(2, latestSlices.size(), "expected two file groups before
compaction");
+ assertTrue(latestSlices.stream().allMatch(slice ->
slice.getLogFiles().findAny().isPresent()),
+ "each file group must have log files for compaction to merge");
+ HoodieSchema tableSchema = new
TableSchemaResolver(metaClient).getTableSchema(false);
+
+ // compaction rewrites both base files through the Spark record type
+ String compactionInstant = (String)
client.scheduleCompaction(Option.empty()).get();
+ HoodieWriteMetadata compactionResult = client.compact(compactionInstant);
+ client.commitCompaction(compactionInstant, compactionResult,
Option.empty());
Review Comment:
Done, switched to `assertFalse(clusterStats.isEmpty(), ...)`.
--
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]
