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 c2e884aa20a4 perf(trino): drop the decimal schema cache and memoize
prefilled values (#19495)
c2e884aa20a4 is described below
commit c2e884aa20a48c539750b7b4ac7211f12c41d9fe
Author: voonhous <[email protected]>
AuthorDate: Tue Aug 4 19:26:57 2026 +0800
perf(trino): drop the decimal schema cache and memoize prefilled values
(#19495)
* perf(trino): drop the decimal schema cache and memoize prefilled column
values
Follow-up to #19483, addressing wombatu-kun's review comments.
The decimal schema cache was unnecessary rather than merely mis-keyed.
Avro's
DecimalConversion.fromBytes reads only the scale (it never touches
precision, and
ignores its schema argument), and Decimals.encodeShortScaledValue then calls
setScale to that same scale, which returns the BigDecimal unchanged. The
pair
reduces to new BigInteger(fixed.bytes()).longValueExact(), so
AvroDecimalConverter
and its ConcurrentHashMap are deleted instead of re-keyed.
PrefilledColumnValues resolved every value through
HiveUtil.getPrefilledColumnValue
on each call, and appendTo runs once per prefilled column per record. Every
input is
a split constant, so the resolved value is now memoized per column.
Tests: the decimal test now drives the public appendTo path over scales,
signs and
the widest short decimal rather than the deleted converter; it passes
against both
the old and new implementations. Added repeated resolution of a hive-null
column,
the case a naive memo would get wrong.
* refactor(trino): rename the uncached prefilled resolver to
computeNativeValue
* perf(trino): collapse the prefilled memo hit path to a single hash lookup
Addresses wombatu-kun's review comment on #19495.
containsKey-then-get was two hash lookups per prefilled column per record
on the
buildRecordInPage path. An UNRESOLVED sentinel with getOrDefault does it in
one,
while still distinguishing a not-yet-resolved column from one resolved to
null
(the hive-"\N" convention, and the lenient fallback for a column the split
cannot
provide). The map still stores real nulls, so it stays a HashMap.
* test(trino): build the decimal fixed the way Avro writes it
Addresses wombatu-kun's two review comments on #19495.
The short-decimal cases built the Fixed from
BigDecimal.unscaledValue().toByteArray(),
the minimal two's-complement encoding, and sized the schema to those bytes.
Avro's
DecimalConversion.toFixed instead left-pads to the schema's fixed size with
the sign
byte, so a real decimal(10,2) is always five bytes. The negative cases were
one and
two bytes wide, meaning no case exercised sign extension across padding --
the thing
the decode is most likely to get wrong. The fixture now sizes the schema
from the
precision and runs the value through Avro's own conversion, matching how
TestHudiUtilColumnHandles builds its decimal fixed schema. Cases are
unchanged and
still pass; -0.07 now decodes from FF FF FF FF F9 rather than F9.
Also drops an overreaching claim on the hive-null repeats: a null-check
memo returns
null on every call too, so the repeats do not rule it out. They cover what
they
actually cover, that both read paths keep returning null once the memo is
populated.
---
.../trino/plugin/hudi/util/HudiAvroSerializer.java | 33 +++-------
.../plugin/hudi/util/PrefilledColumnValues.java | 27 ++++++++
.../plugin/hudi/TestPrefilledColumnValues.java | 13 +++-
.../plugin/hudi/util/TestHudiAvroSerializer.java | 74 +++++++++++++++-------
4 files changed, 98 insertions(+), 49 deletions(-)
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
index fe7ddfe42515..8b035c084af6 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
@@ -41,8 +41,6 @@ import io.trino.spi.type.SqlVarbinary;
import io.trino.spi.type.Type;
import io.trino.spi.type.VarbinaryType;
import io.trino.spi.type.VarcharType;
-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.GenericRecord;
@@ -50,6 +48,7 @@ import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.util.Utf8;
import java.math.BigDecimal;
+import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.time.DateTimeException;
import java.time.Instant;
@@ -60,7 +59,6 @@ import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Verify.verify;
@@ -70,7 +68,6 @@ import static
io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static io.trino.spi.StandardErrorCode.NUMERIC_VALUE_OUT_OF_RANGE;
import static io.trino.spi.type.BigintType.BIGINT;
import static io.trino.spi.type.DateType.DATE;
-import static io.trino.spi.type.Decimals.encodeShortScaledValue;
import static io.trino.spi.type.Decimals.writeBigDecimal;
import static io.trino.spi.type.Decimals.writeShortDecimal;
import static io.trino.spi.type.IntegerType.INTEGER;
@@ -108,7 +105,6 @@ public class HudiAvroSerializer
1, // 9 digits after the dot
};
- private static final AvroDecimalConverter DECIMAL_CONVERTER = new
AvroDecimalConverter();
private final PrefilledColumnValues prefilledColumnValues;
private final List<HiveColumnHandle> columnHandles;
@@ -262,8 +258,13 @@ public class HudiAvroSerializer
}
else if (value instanceof GenericData.Fixed fixed) {
verify(decimalType.isShort(), "The type should be
short decimal");
- BigDecimal decimal =
DECIMAL_CONVERTER.convert(decimalType.getPrecision(), decimalType.getScale(),
fixed.bytes());
- type.writeLong(output, encodeShortScaledValue(decimal,
decimalType.getScale()));
+ // Avro stores a decimal as its unscaled value in
big-endian two's complement, which is
+ // exactly what Trino's short decimal holds. Going
through Avro's DecimalConversion and
+ // Decimals.encodeShortScaledValue is a no-op round
trip: DecimalConversion.fromBytes reads
+ // only the scale (it ignores precision, and its
schema argument entirely) to build
+ // BigDecimal(unscaled, scale), and
encodeShortScaledValue then calls setScale to that same
+ // scale, which returns the BigDecimal unchanged,
before taking the unscaled value back out.
+ type.writeLong(output, new
BigInteger(fixed.bytes()).longValueExact());
}
else {
throw new TrinoException(GENERIC_INTERNAL_ERROR,
@@ -544,22 +545,4 @@ public class HudiAvroSerializer
}
});
}
-
- static class AvroDecimalConverter
- {
- private static final Conversions.DecimalConversion
AVRO_DECIMAL_CONVERSION = new Conversions.DecimalConversion();
- // convert() runs once per decimal cell on the record read path, and
building a Schema costs
- // orders of magnitude more than the conversion itself. The
(precision, scale) space is tiny
- // and fixed per column, so cache the schemas globally.
- private static final Map<Integer, Schema> DECIMAL_SCHEMAS = new
ConcurrentHashMap<>();
-
- BigDecimal convert(int precision, int scale, byte[] bytes)
- {
- // The key is unique because precision and scale are at most 38
- Schema schema = DECIMAL_SCHEMAS.computeIfAbsent(
- precision * 100 + scale,
- key -> LogicalTypes.decimal(precision,
scale).addToSchema(Schema.create(Schema.Type.BYTES)));
- return AVRO_DECIMAL_CONVERSION.fromBytes(ByteBuffer.wrap(bytes),
schema, schema.getLogicalType());
- }
- }
}
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java
index 19c6c9d01322..c55d46956c27 100644
---
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java
@@ -22,6 +22,7 @@ import io.trino.spi.block.Block;
import io.trino.spi.block.BlockBuilder;
import io.trino.spi.block.RunLengthEncodedBlock;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.OptionalInt;
@@ -45,11 +46,18 @@ import static io.trino.spi.type.TypeUtils.writeNativeValue;
*/
public class PrefilledColumnValues
{
+ // Absent-marker for the memo, so a not-yet-resolved column is
distinguishable from one resolved to null
+ private static final Object UNRESOLVED = new Object();
+
private final Map<String, HivePartitionKey> partitionKeysByName;
private final String partitionName;
private final String filePath;
private final long fileSize;
private final long fileModifiedTime;
+ // Resolved native value per column name, populated lazily. One instance
belongs to one split and a
+ // split is read by a single driver thread, so a plain HashMap is enough;
it also has to hold nulls,
+ // which a ConcurrentHashMap could not.
+ private final Map<String, Object> resolvedValues = new HashMap<>();
public static PrefilledColumnValues create(HudiSplit hudiSplit)
{
@@ -108,6 +116,25 @@ public class PrefilledColumnValues
}
private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ // Every input to computeNativeValue() is a constant of the split, but
appendTo is called once per
+ // prefilled column per record, and computing re-parses the partition
string each time
+ // ($file_modified_time even formats a timestamp and parses it
straight back). Memoize per column so
+ // each one is resolved once per split. Keyed on the name rather than
the handle because
+ // HiveColumnHandle.hashCode hashes seven fields through a varargs
array, whereas a String caches
+ // its hash. A sentinel rather than a null check, because null is a
legitimate resolved value -- both
+ // for the hive-null convention and for the lenient fallback below --
and getOrDefault keeps the hit
+ // path, the one taken per record, to a single hash lookup.
+ String name = columnHandle.getName();
+ Object value = resolvedValues.getOrDefault(name, UNRESOLVED);
+ if (value == UNRESOLVED) {
+ value = computeNativeValue(columnHandle);
+ resolvedValues.put(name, value);
+ }
+ return value;
+ }
+
+ private Object computeNativeValue(HiveColumnHandle columnHandle)
{
if (!isPrefilled(columnHandle)) {
// Lenient null fill, e.g. for a hidden column Trino defines but
Hudi does not populate.
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestPrefilledColumnValues.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestPrefilledColumnValues.java
index 4ee508cada0b..59442f8cdbe6 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestPrefilledColumnValues.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestPrefilledColumnValues.java
@@ -81,9 +81,20 @@ class TestPrefilledColumnValues
{
// Trino's HivePartitionKey encodes a null partition value as the
literal string "\N"
PrefilledColumnValues values = prefilledValues(new
HivePartitionKey("pk_string", "\\N"));
+ HiveColumnHandle handle = partitionKey("pk_string", VARCHAR,
HiveType.HIVE_STRING);
- Block block = singleValueBlock(values, partitionKey("pk_string",
VARCHAR, HiveType.HIVE_STRING));
+ Block block = singleValueBlock(values, handle);
assertThat(block.isNull(0)).isTrue();
+
+ // Values are resolved once per column and reused, so both read paths
have to keep returning null
+ // for a hive-null column after the first read has populated the memo.
+ BlockBuilder blockBuilder = VARCHAR.createBlockBuilder(null, 2);
+ values.appendTo(handle, blockBuilder);
+ values.appendTo(handle, blockBuilder);
+ Block repeated = blockBuilder.build();
+ assertThat(repeated.isNull(0)).isTrue();
+ assertThat(repeated.isNull(1)).isTrue();
+ assertThat(values.toRleBlock(handle, 1).isNull(0)).isTrue();
}
@Test
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
index 116d50d115e8..514218d48add 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
@@ -25,14 +25,21 @@ import io.trino.spi.block.Block;
import io.trino.spi.block.BlockBuilder;
import io.trino.spi.predicate.TupleDomain;
import io.trino.spi.type.DecimalType;
+import org.apache.avro.Conversions;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.avro.generic.GenericData;
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.MethodSource;
import java.math.BigDecimal;
+import java.math.BigInteger;
import java.util.List;
import java.util.Optional;
+import java.util.stream.Stream;
import static io.trino.spi.type.BigintType.BIGINT;
import static io.trino.spi.type.IntegerType.INTEGER;
@@ -41,32 +48,37 @@ import static org.assertj.core.api.Assertions.assertThat;
class TestHudiAvroSerializer
{
- @Test
- public void testDecimalConverter()
- {
- HudiAvroSerializer.AvroDecimalConverter converter = new
HudiAvroSerializer.AvroDecimalConverter();
-
- assertThat(converter.convert(10, 2,
unscaledBytes("123.45"))).isEqualTo(new BigDecimal("123.45"));
- // Same (precision, scale) again: served from the cached schema
- assertThat(converter.convert(10, 2,
unscaledBytes("-0.07"))).isEqualTo(new BigDecimal("-0.07"));
- // Same precision, different scale, and vice versa: must not collide
in the cache
- assertThat(converter.convert(10, 4,
unscaledBytes("123.4567"))).isEqualTo(new BigDecimal("123.4567"));
- assertThat(converter.convert(18, 2,
unscaledBytes("9999999999999999.99"))).isEqualTo(new
BigDecimal("9999999999999999.99"));
- assertThat(converter.convert(5, 0, unscaledBytes("42"))).isEqualTo(new
BigDecimal("42"));
- }
-
- @Test
- public void testAppendShortDecimalFromAvroFixed()
+ /**
+ * A short decimal is stored in Trino as the unscaled value, which is
exactly what Avro writes into the
+ * fixed bytes, so the read is a plain big-endian two's complement decode.
The cases below pin the parts
+ * that decode gets wrong if it is ever rewritten: sign extension for
negatives, scale 0, and the
+ * full-width value at the maximum short-decimal precision.
+ */
+ @ParameterizedTest
+ @MethodSource("shortDecimals")
+ public void testAppendShortDecimalFromAvroFixed(int precision, int scale,
String value, long expectedUnscaled)
{
- DecimalType type = DecimalType.createDecimalType(10, 2);
- byte[] bytes = unscaledBytes("123.45");
- GenericData.Fixed fixed = new
GenericData.Fixed(Schema.createFixed("fix", null, null, bytes.length), bytes);
+ DecimalType type = DecimalType.createDecimalType(precision, scale);
BlockBuilder blockBuilder = type.createBlockBuilder(null, 1);
- HudiAvroSerializer.appendTo(type, fixed, blockBuilder);
+ HudiAvroSerializer.appendTo(type, avroDecimalFixed(precision, scale,
value), blockBuilder);
Block block = blockBuilder.build();
- assertThat(type.getLong(block, 0)).isEqualTo(12345L);
+ assertThat(type.getLong(block, 0)).isEqualTo(expectedUnscaled);
+ }
+
+ private static Stream<Arguments> shortDecimals()
+ {
+ return Stream.of(
+ Arguments.of(10, 2, "123.45", 12345L),
+ Arguments.of(10, 2, "-0.07", -7L),
+ Arguments.of(10, 2, "0.00", 0L),
+ Arguments.of(10, 4, "123.4567", 1234567L),
+ Arguments.of(5, 0, "42", 42L),
+ Arguments.of(5, 0, "-42", -42L),
+ // Widest short decimal: 18 digits, both signs
+ Arguments.of(18, 2, "9999999999999999.99",
999999999999999999L),
+ Arguments.of(18, 2, "-9999999999999999.99",
-999999999999999999L));
}
@Test
@@ -96,9 +108,25 @@ class TestHudiAvroSerializer
assertThat(VARCHAR.getSlice(page.getBlock(1),
2).toStringUtf8()).isEqualTo("three");
}
- private static byte[] unscaledBytes(String decimal)
+ /**
+ * Encodes the value the way an Avro writer does, via Avro's own
conversion: a fixed sized from the
+ * precision, holding the unscaled value left-padded to that width with
the sign byte (0xFF for
+ * negatives). Building the fixed from the minimal two's-complement
encoding instead would leave the
+ * padding bytes, and so sign extension across them, untested.
+ */
+ private static GenericData.Fixed avroDecimalFixed(int precision, int
scale, String value)
+ {
+ LogicalTypes.Decimal decimalType = LogicalTypes.decimal(precision,
scale);
+ Schema fixedSchema = decimalType.addToSchema(
+ Schema.createFixed("fix", null, null,
decimalFixedSize(precision)));
+ return (GenericData.Fixed) new Conversions.DecimalConversion()
+ .toFixed(new BigDecimal(value), fixedSchema, decimalType);
+ }
+
+ /** Bytes needed to hold the widest unscaled value at this precision, i.e.
the fixed size Avro sizes a decimal to. */
+ private static int decimalFixedSize(int precision)
{
- return new BigDecimal(decimal).unscaledValue().toByteArray();
+ return
BigInteger.TEN.pow(precision).subtract(BigInteger.ONE).toByteArray().length;
}
private static Schema recordSchema(String name)
