This is an automated email from the ASF dual-hosted git repository.
voonhous 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 9a30ecd60e15 fix(spark): preserve the Avro fixed-size decimal width in
the Spark row write support (#19512)
9a30ecd60e15 is described below
commit 9a30ecd60e1572fca61a32af00ea7bc27deef611
Author: Y Ethan Guo <[email protected]>
AuthorDate: Wed Aug 5 20:32:58 2026 -0700
fix(spark): preserve the Avro fixed-size decimal width in the Spark row
write support (#19512)
* fix(spark): preserve the Avro fixed-size decimal width in the Spark row
write support
The Spark row writer sized decimal FIXED_LEN_BYTE_ARRAY columns from
Decimal.minBytesForPrecision(), discarding an Avro fixed(N) size wider than the
precision-minimal width and diverging from the Avro write path. Honor the
declared fixed size from the resolved HoodieSchema at both the schema converter
and the value writer, resolving the padding buffer once per column.
* Guard against an Avro fixed size smaller than the precision-minimal
decimal width
Fail fast with a diagnosable message if a decimal fixed(N) declares fewer
bytes than minBytesForPrecision, instead of an opaque negative-length
Arrays.fill later.
* Unit-test decimalFixedLen directly for hudi-spark-client coverage
Make decimalFixedLen package-private and cover its branches (min-width
fallback and honored Avro fixed size) from hudi-spark-client's own test module,
where the full write support cannot be constructed.
* Extract decimal sign-extension padding into a testable static helper
Move the fixed-length padding out of the makeWriter lambda into
padDecimalToFixedLength so it can be unit-tested directly from
hudi-spark-client (covering the exact-width, positive-pad, and negative
sign-extension cases), where the full write support cannot be constructed.
* Trim redundant Javadocs on the decimal helpers
Drop the padDecimalToFixedLength Javadoc and reduce decimalFixedLen's to a
single line.
* Rename decimalFixedLen to resolveDecimalByteLength and use assertFalse
for the clustering check
Address review nits: resolveDecimalByteLength reads more clearly at the
call sites than decimalFixedLen, and assertFalse(isEmpty()) is more natural
than assertTrue(!isEmpty()).
---
.../storage/row/HoodieRowParquetWriteSupport.java | 63 ++++++---
.../row/TestHoodieRowParquetWriteSupport.java | 35 +++++
.../row/TestHoodieInternalRowParquetWriter.java | 36 +++++
.../TestHoodieSparkMergeOnReadTableCompaction.java | 155 +++++++++++++++++++++
4 files changed, 268 insertions(+), 21 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
index 42d80b7d279a..c14a75d314d6 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowParquetWriteSupport.java
@@ -482,6 +482,34 @@ public class HoodieRowParquetWriteSupport extends
WriteSupport<InternalRow> {
}
}
+ /**
+ * Fixed-length byte width for a decimal column: honor the declared Avro
{@code fixed} size when
+ * present (it may be wider than the precision-minimal width), otherwise the
precision-minimal width.
+ */
+ static int resolveDecimalByteLength(HoodieSchema resolvedSchema, int
precision) {
+ if (resolvedSchema instanceof HoodieSchema.Decimal) {
+ HoodieSchema.Decimal decimalSchema = (HoodieSchema.Decimal)
resolvedSchema;
+ if (decimalSchema.isFixed()) {
+ int fixedSize = decimalSchema.getFixedSize();
+ ValidationUtils.checkArgument(fixedSize >=
Decimal.minBytesForPrecision()[precision],
+ () -> String.format("Avro fixed size %s is too small for decimal
precision %s (need >= %s bytes)",
+ fixedSize, precision,
Decimal.minBytesForPrecision()[precision]));
+ return fixedSize;
+ }
+ }
+ return Decimal.minBytesForPrecision()[precision];
+ }
+
+ static byte[] padDecimalToFixedLength(byte[] unscaledBytes, int numBytes,
byte[] paddingBuffer) {
+ if (unscaledBytes.length == numBytes) {
+ return unscaledBytes;
+ }
+ byte signByte = (unscaledBytes[0] < 0) ? (byte) -1 : (byte) 0;
+ Arrays.fill(paddingBuffer, 0, numBytes - unscaledBytes.length, signByte);
+ System.arraycopy(unscaledBytes, 0, paddingBuffer, numBytes -
unscaledBytes.length, unscaledBytes.length);
+ return paddingBuffer;
+ }
+
private ValueWriter makeWriter(HoodieSchema schema, DataType dataType) {
HoodieSchema resolvedSchema = schema == null ? null :
schema.getNonNullType();
@@ -550,28 +578,21 @@ public class HoodieRowParquetWriteSupport extends
WriteSupport<InternalRow> {
consumeGroup(() -> variantWriter.accept(row, ordinal));
};
} else if (dataType instanceof DecimalType) {
+ int precision = ((DecimalType) dataType).precision();
+ ValidationUtils.checkArgument(precision <= DecimalType.MAX_PRECISION(),
+ () -> String.format("Decimal precision %s exceeds max precision %s",
precision, DecimalType.MAX_PRECISION()));
+ int scale = ((DecimalType) dataType).scale();
+ // Honor the declared Avro `fixed` size so the
row/bulk-insert/clustering/compaction write
+ // path preserves the schema width instead of narrowing to the
precision-minimal one.
+ int numBytes = resolveDecimalByteLength(resolvedSchema, precision);
+ // Padding buffer resolved once per column, not per record: reuse the
shared decimalBuffer when
+ // it is wide enough, otherwise allocate one dedicated buffer here (an
over-allocated Avro fixed
+ // size can exceed the shared buffer's capacity).
+ byte[] paddingBuffer = numBytes <= decimalBuffer.length ? decimalBuffer
: new byte[numBytes];
return (row, ordinal) -> {
- int precision = ((DecimalType) dataType).precision();
- ValidationUtils.checkArgument(precision <= DecimalType.MAX_PRECISION(),
- () -> String.format("Decimal precision %s exceeds max precision
%s", precision, DecimalType.MAX_PRECISION()));
- int scale = ((DecimalType) dataType).scale();
byte[] bytes = row.getDecimal(ordinal, precision,
scale).toJavaBigDecimal().unscaledValue().toByteArray();
- int numBytes = Decimal.minBytesForPrecision()[precision];
- byte[] fixedLengthBytes;
- if (bytes.length == numBytes) {
- // If the length of the underlying byte array of the unscaled
`BigInteger` happens to be
- // `numBytes`, just reuse it, so that we don't bother copying it to
`decimalBuffer`.
- fixedLengthBytes = bytes;
- } else {
- // Otherwise, the length must be less than `numBytes`. In this case
we copy contents of
- // the underlying bytes with padding sign bytes to `decimalBuffer`
to form the result
- // fixed-length byte array.
- byte signByte = (bytes[0] < 0) ? (byte) -1 : (byte) 0;
- Arrays.fill(decimalBuffer, 0, numBytes - bytes.length, signByte);
- System.arraycopy(bytes, 0, decimalBuffer, numBytes - bytes.length,
bytes.length);
- fixedLengthBytes = decimalBuffer;
- }
- recordConsumer.addBinary(Binary.fromReusedByteArray(fixedLengthBytes,
0, numBytes));
+ recordConsumer.addBinary(Binary.fromReusedByteArray(
+ padDecimalToFixedLength(bytes, numBytes, paddingBuffer), 0,
numBytes));
};
} else if (dataType instanceof ArrayType
&& resolvedSchema != null
@@ -830,7 +851,7 @@ public class HoodieRowParquetWriteSupport extends
WriteSupport<InternalRow> {
return Types
.primitive(FIXED_LEN_BYTE_ARRAY, repetition)
.as(LogicalTypeAnnotation.decimalType(scale, precision))
- .length(Decimal.minBytesForPrecision()[precision])
+ .length(resolveDecimalByteLength(resolvedSchema, precision))
.named(structField.name());
} else if (dataType instanceof ArrayType
&& resolvedSchema != null
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupport.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupport.java
index 0c39e99bdf89..6ec4a4cd2300 100644
---
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupport.java
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/storage/row/TestHoodieRowParquetWriteSupport.java
@@ -18,16 +18,21 @@
package org.apache.hudi.io.storage.row;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.schema.HoodieSchemaType;
import org.apache.hudi.testutils.HoodieClientTestBase;
+import org.apache.spark.sql.types.Decimal;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
import java.util.TimeZone;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
/**
* Coverage for {@link
HoodieRowParquetWriteSupport#resolveSessionLocalTimeZone()}.
@@ -42,6 +47,36 @@ class TestHoodieRowParquetWriteSupport extends
HoodieClientTestBase {
private static final String SESSION_LOCAL_TIME_ZONE_KEY =
"spark.sql.session.timeZone";
+ @Test
+ void testResolveDecimalByteLength() {
+ int minWidth = Decimal.minBytesForPrecision()[20];
+ // A non-decimal schema falls back to the precision-minimal width.
+ assertEquals(minWidth,
+
HoodieRowParquetWriteSupport.resolveDecimalByteLength(HoodieSchema.create(HoodieSchemaType.STRING),
20));
+ // A bytes-backed decimal (no declared fixed size) also falls back to the
minimum.
+ assertEquals(minWidth,
+
HoodieRowParquetWriteSupport.resolveDecimalByteLength(HoodieSchema.createDecimal(20,
2), 20));
+ // An Avro fixed decimal wider than the minimum is honored.
+ assertEquals(10,
+ HoodieRowParquetWriteSupport.resolveDecimalByteLength(
+ HoodieSchema.createDecimal("dec", null, null, 20, 2, 10), 20));
+ }
+
+ @Test
+ void testPadDecimalToFixedLength() {
+ byte[] buffer = new byte[16];
+ // Already the full width: returned as-is, no copy into the buffer.
+ byte[] exact = new byte[] {1, 2, 3, 4};
+ assertSame(exact,
HoodieRowParquetWriteSupport.padDecimalToFixedLength(exact, 4, buffer));
+ // Positive magnitude: left-padded with zero sign bytes.
+ byte[] positive = HoodieRowParquetWriteSupport.padDecimalToFixedLength(new
byte[] {0x12, 0x34}, 4, buffer);
+ assertArrayEquals(new byte[] {0, 0, 0x12, 0x34}, Arrays.copyOf(positive,
4));
+ // Negative magnitude: left-padded with 0xFF sign bytes.
+ byte[] negative = HoodieRowParquetWriteSupport.padDecimalToFixedLength(
+ new byte[] {(byte) 0xFF, (byte) 0x80}, 4, buffer);
+ assertArrayEquals(new byte[] {(byte) 0xFF, (byte) 0xFF, (byte) 0xFF,
(byte) 0x80}, Arrays.copyOf(negative, 4));
+ }
+
@Test
void testResolveSessionLocalTimeZoneWithoutOverride() {
String expected = TimeZone.getDefault().getID();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
index c6719dac403b..04fd0e98324c 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/io/storage/row/TestHoodieInternalRowParquetWriter.java
@@ -36,12 +36,16 @@ import org.apache.hudi.testutils.SparkDatasetTestUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.hadoop.metadata.FileMetaData;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -126,6 +130,38 @@ public class TestHoodieInternalRowParquetWriter extends
HoodieSparkClientTestHar
});
}
+ @Test
+ void testDecimalFixedLenWidthFromAvroSchema() {
+ // The row-writer sizes decimal FIXED_LEN columns from the Avro schema: an
Avro fixed(10) for
+ // decimal(20,2) keeps its declared width (10, wider than the
precision-minimal 9), while a bytes
+ // decimal keeps the precision-minimal width (9).
+ assertEquals(10, decimalParquetTypeLength(decimalRecordSchema(
+
"{\"type\":\"fixed\",\"name\":\"dec_fixed\",\"size\":10,\"logicalType\":\"decimal\",\"precision\":20,\"scale\":2}")),
+ "Avro fixed(10) decimal must stay FIXED_LEN_BYTE_ARRAY(10)");
+ assertEquals(9, decimalParquetTypeLength(decimalRecordSchema(
+
"{\"type\":\"bytes\",\"logicalType\":\"decimal\",\"precision\":20,\"scale\":2}")),
+ "bytes decimal keeps the precision-minimal FIXED_LEN width (9)");
+ }
+
+ private static String decimalRecordSchema(String decType) {
+ return
"{\"type\":\"record\",\"name\":\"rec\",\"fields\":[{\"name\":\"dec\",\"type\":"
+ decType + "}]}";
+ }
+
+ private int decimalParquetTypeLength(String avroSchemaJson) {
+ StructType structType = new StructType().add("dec",
DataTypes.createDecimalType(20, 2), false);
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(avroSchemaJson)
+ .build();
+ HoodieRowParquetWriteSupport writeSupport =
HoodieRowParquetWriteSupport.getHoodieRowParquetWriteSupport(
+ storageConf.unwrap(), structType, Option.empty(), config);
+ MessageType parquetSchema =
writeSupport.init(writeSupport.getHadoopConf()).getSchema();
+ PrimitiveType dec = parquetSchema.getType("dec").asPrimitiveType();
+ assertEquals(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
dec.getPrimitiveTypeName(),
+ "decimal must be encoded as FIXED_LEN_BYTE_ARRAY");
+ return dec.getTypeLength();
+ }
+
private HoodieRowParquetWriteSupport
getWriteSupport(HoodieWriteConfig.Builder writeConfigBuilder, Configuration
hadoopConf, boolean parquetWriteLegacyFormatEnabled) {
writeConfigBuilder.withStorageConfig(HoodieStorageConfig.newBuilder().parquetWriteLegacyFormat(String.valueOf(parquetWriteLegacyFormatEnabled)).build());
HoodieWriteConfig writeConfig = writeConfigBuilder.build();
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
index be856ae3bb2c..9a190185e98b 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableCompaction.java
@@ -28,19 +28,27 @@ import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.PartialUpdateAvroPayload;
import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
import org.apache.hudi.common.util.CompactionUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.config.HoodieCleanConfig;
+import org.apache.hudi.config.HoodieClusteringConfig;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieIndexConfig;
import org.apache.hudi.config.HoodieLayoutConfig;
@@ -52,6 +60,8 @@ import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.table.action.commit.SparkBucketIndexPartitioner;
import org.apache.hudi.table.action.rollback.RollbackUtils;
@@ -59,17 +69,28 @@ import org.apache.hudi.table.storage.HoodieStorageLayout;
import org.apache.hudi.testutils.HoodieMergeOnReadTestUtils;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+import org.apache.avro.Conversions;
+import org.apache.avro.LogicalTypes;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericFixed;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
import org.apache.spark.api.java.JavaRDD;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
+import java.math.BigDecimal;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -81,6 +102,7 @@ import static
org.apache.hudi.common.table.HoodieTableMetaClient.METAFOLDER_NAME
import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -442,4 +464,137 @@ public class TestHoodieSparkMergeOnReadTableCompaction
extends SparkClientFuncti
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());
+
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants().containsInstant(compactionInstant));
+ List<StoragePath> compactedBaseFiles = latestBaseFilePaths(config,
DECIMAL_PARTITION);
+ assertEquals(2, compactedBaseFiles.size(), "expected two compacted base
files");
+ for (StoragePath path : compactedBaseFiles) {
+ assertDecimalFixedLen(path, EXPECTED_DECIMAL_FIXED_LEN);
+ }
+ assertEquals(tableSchema,
+ new
TableSchemaResolver(HoodieTableMetaClient.reload(metaClient)).getTableSchema(false),
+ "table schema in commit metadata must not change after compaction");
+
+ // clustering rewrites the two compacted groups through the same Spark
row-writer path
+ String clusteringInstant = (String)
client.scheduleClustering(Option.empty()).get();
+ HoodieWriteMetadata<JavaRDD<WriteStatus>> clusterMetadata =
client.cluster(clusteringInstant, true);
+ List<HoodieWriteStat> clusterStats = clusterMetadata.getWriteStats().get();
+ assertFalse(clusterStats.isEmpty(), "clustering should write at least one
base file");
+ for (HoodieWriteStat stat : clusterStats) {
+ assertDecimalFixedLen(new StoragePath(metaClient.getBasePath(),
stat.getPath()),
+ EXPECTED_DECIMAL_FIXED_LEN);
+ }
+ assertEquals(tableSchema,
+ new
TableSchemaResolver(HoodieTableMetaClient.reload(metaClient)).getTableSchema(false),
+ "table schema in commit metadata must not change after clustering");
+ }
+
+ private List<HoodieRecord> buildDecimalRecords(int startKey, int count, long
ts, BigDecimal decValue) {
+ Schema schema = new Schema.Parser().parse(FIXED10_DECIMAL_SCHEMA);
+ Schema decSchema = schema.getField("dec").schema();
+ Conversions.DecimalConversion decimalConversion = new
Conversions.DecimalConversion();
+ LogicalTypes.Decimal decimalType = LogicalTypes.decimal(20, 2);
+ BigDecimal scaledValue = decValue.setScale(2);
+ List<HoodieRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ String key = "key_" + (startKey + i);
+ GenericRecord rec = new GenericData.Record(schema);
+ rec.put("_row_key", key);
+ rec.put("partition_path", DECIMAL_PARTITION);
+ rec.put("ts", ts);
+ GenericFixed fixed = decimalConversion.toFixed(scaledValue, decSchema,
decimalType);
+ rec.put("dec", fixed);
+ records.add(new HoodieAvroRecord<>(new HoodieKey(key, DECIMAL_PARTITION),
+ new OverwriteWithLatestAvroPayload(rec, ts)));
+ }
+ return records;
+ }
+
+ private List<StoragePath> latestBaseFilePaths(HoodieWriteConfig config,
String partition) {
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(config, context(),
metaClient);
+ hoodieTable.getHoodieView().sync();
+ return hoodieTable.getBaseFileOnlyView().getLatestBaseFiles(partition)
+ .map(HoodieBaseFile::getStoragePath).collect(Collectors.toList());
+ }
+
+ private void assertDecimalFixedLen(StoragePath baseFilePath, int
expectedLen) {
+ MessageType parquetSchema = ParquetUtils.readMetadata(hoodieStorage(),
baseFilePath)
+ .getFileMetaData().getSchema();
+ PrimitiveType decType = parquetSchema.getType("dec").asPrimitiveType();
+ assertEquals(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY,
decType.getPrimitiveTypeName(),
+ "decimal must be encoded as FIXED_LEN_BYTE_ARRAY");
+ assertEquals(expectedLen, decType.getTypeLength(),
+ "Avro fixed(10) decimal(20,2) must stay FIXED_LEN_BYTE_ARRAY(10), not
narrow to 9");
+ }
}
