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 2f8a7252c383 perf(trino): cache decimal Avro schema in
HudiAvroSerializer instead … (#19483)
2f8a7252c383 is described below
commit 2f8a7252c38379f50b8b970f2d841975ddf090a9
Author: voonhous <[email protected]>
AuthorDate: Mon Aug 3 23:23:58 2026 +0800
perf(trino): cache decimal Avro schema in HudiAvroSerializer instead …
(#19483)
* perf(trino): cache decimal Avro schema in HudiAvroSerializer instead of
parsing per value
AvroDecimalConverter built and JSON-parsed an Avro schema for every
decimal cell on the record read path. Cache the schemas by (precision,
scale) and build them with LogicalTypes on miss. Also cache
buildRecordInPage field positions per record schema instead of doing a
name lookup per record per column, and make writeRow's anonymous-field
name fallback lazy.
Fixes #19361
* address review: replace bit-packed decimal cache key with precision * 100
+ scale
---
.../trino/plugin/hudi/util/HudiAvroSerializer.java | 53 ++++++--
.../plugin/hudi/util/TestHudiAvroSerializer.java | 148 +++++++++++++++++++++
2 files changed, 190 insertions(+), 11 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 0d9bc9f5978b..fe7ddfe42515 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
@@ -42,6 +42,7 @@ 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;
@@ -59,6 +60,7 @@ 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;
@@ -118,6 +120,11 @@ public class HudiAvroSerializer
// constructor, which maps page channel i to record position
channelToFieldPosition[i].
private final Schema schema;
private final int[] channelToFieldPosition;
+ // Single-entry cache for buildRecordInPage: all records of a split share
one schema instance,
+ // so an identity check makes the per-record, per-column field-name lookup
a one-time cost.
+ // Prefilled (hidden/synthesized) columns are not fields of the record
schema; they get -1.
+ private Schema positionsCacheSchema;
+ private int[] positionsCache;
public HudiAvroSerializer(List<HiveColumnHandle> columnHandles,
PrefilledColumnValues prefilledColumnValues)
{
@@ -170,21 +177,36 @@ public class HudiAvroSerializer
public void buildRecordInPage(PageBuilder pageBuilder, IndexedRecord
record)
{
pageBuilder.declarePosition();
- int blockSeq = 0;
- for (int channel = 0; channel < columnTypes.size(); channel++,
blockSeq++) {
- BlockBuilder output = pageBuilder.getBlockBuilder(blockSeq);
- HiveColumnHandle columnHandle = columnHandles.get(channel);
- if (prefilledColumnValues.isPrefilled(columnHandle)) {
- prefilledColumnValues.appendTo(columnHandle, output);
+ // Record may not be projected, get field positions from its own schema
+ int[] fieldPositions = fieldPositionsFor(record.getSchema());
+ for (int channel = 0; channel < columnTypes.size(); channel++) {
+ BlockBuilder output = pageBuilder.getBlockBuilder(channel);
+ int fieldPosition = fieldPositions[channel];
+ if (fieldPosition < 0) {
+ prefilledColumnValues.appendTo(columnHandles.get(channel),
output);
}
else {
- // Record may not be projected, get index from it
- int fieldPosInSchema =
getFieldFromSchema(columnHandle.getName(), record.getSchema()).pos();
- appendTo(columnTypes.get(channel),
record.get(fieldPosInSchema), output);
+ appendTo(columnTypes.get(channel), record.get(fieldPosition),
output);
}
}
}
+ private int[] fieldPositionsFor(Schema recordSchema)
+ {
+ if (positionsCacheSchema != recordSchema) {
+ int[] positions = new int[columnHandles.size()];
+ for (int channel = 0; channel < columnHandles.size(); channel++) {
+ HiveColumnHandle columnHandle = columnHandles.get(channel);
+ positions[channel] =
prefilledColumnValues.isPrefilled(columnHandle)
+ ? -1
+ : getFieldFromSchema(columnHandle.getName(),
recordSchema).pos();
+ }
+ positionsCache = positions;
+ positionsCacheSchema = recordSchema;
+ }
+ return positionsCache;
+ }
+
public static void appendTo(Type type, Object value, BlockBuilder output)
{
if (value == null) {
@@ -493,7 +515,9 @@ public class HudiAvroSerializer
output.buildEntry(fieldBuilders -> {
for (int index = 0; index < fields.size(); index++) {
RowType.Field field = fields.get(index);
- appendTo(field.getType(),
record.get(field.getName().orElse("field" + index)), fieldBuilders.get(index));
+ int fieldIndex = index;
+ String fieldName = field.getName().orElseGet(() -> "field" +
fieldIndex);
+ appendTo(field.getType(), record.get(fieldName),
fieldBuilders.get(index));
}
});
}
@@ -524,10 +548,17 @@ 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)
{
- Schema schema = new
Schema.Parser().parse(format("{\"type\":\"bytes\",\"logicalType\":\"decimal\",\"precision\":%d,\"scale\":%d}",
precision, scale));
+ // 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/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
new file mode 100644
index 000000000000..116d50d115e8
--- /dev/null
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestHudiAvroSerializer.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi.util;
+
+import io.trino.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiBaseFile;
+import io.trino.spi.Page;
+import io.trino.spi.PageBuilder;
+import io.trino.spi.SplitWeight;
+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.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.generic.GenericData;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+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()
+ {
+ 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);
+
+ BlockBuilder blockBuilder = type.createBlockBuilder(null, 1);
+ HudiAvroSerializer.appendTo(type, fixed, blockBuilder);
+ Block block = blockBuilder.build();
+
+ assertThat(type.getLong(block, 0)).isEqualTo(12345L);
+ }
+
+ @Test
+ public void testBuildRecordInPage()
+ {
+ // Schema field order (b, a) deliberately differs from projection
order (a, b, pk_int),
+ // so correct output proves positions are resolved from the record's
schema.
+ Schema schema = recordSchema("rec1");
+ HudiAvroSerializer serializer = new
HudiAvroSerializer(projectedColumns(), prefilledValues());
+ PageBuilder pageBuilder = new PageBuilder(List.of(BIGINT, VARCHAR,
INTEGER));
+
+ serializer.buildRecordInPage(pageBuilder, record(schema, 1L, "one"));
+ // Second record with the same schema instance exercises the cached
field positions
+ serializer.buildRecordInPage(pageBuilder, record(schema, 2L, "two"));
+ // A schema instance with the opposite field order must invalidate the
cache; reusing the
+ // stale positions would swap the a and b values
+ serializer.buildRecordInPage(pageBuilder,
record(reversedRecordSchema("rec2"), 3L, "three"));
+
+ Page page = pageBuilder.build();
+ assertThat(page.getPositionCount()).isEqualTo(3);
+ for (int position = 0; position < 3; position++) {
+ assertThat(BIGINT.getLong(page.getBlock(0),
position)).isEqualTo(position + 1);
+ assertThat(INTEGER.getInt(page.getBlock(2),
position)).isEqualTo(42);
+ }
+ assertThat(VARCHAR.getSlice(page.getBlock(1),
0).toStringUtf8()).isEqualTo("one");
+ assertThat(VARCHAR.getSlice(page.getBlock(1),
1).toStringUtf8()).isEqualTo("two");
+ assertThat(VARCHAR.getSlice(page.getBlock(1),
2).toStringUtf8()).isEqualTo("three");
+ }
+
+ private static byte[] unscaledBytes(String decimal)
+ {
+ return new BigDecimal(decimal).unscaledValue().toByteArray();
+ }
+
+ private static Schema recordSchema(String name)
+ {
+ return SchemaBuilder.record(name).fields()
+ .name("b").type().stringType().noDefault()
+ .name("a").type().longType().noDefault()
+ .endRecord();
+ }
+
+ private static Schema reversedRecordSchema(String name)
+ {
+ return SchemaBuilder.record(name).fields()
+ .name("a").type().longType().noDefault()
+ .name("b").type().stringType().noDefault()
+ .endRecord();
+ }
+
+ private static GenericData.Record record(Schema schema, long a, String b)
+ {
+ GenericData.Record record = new GenericData.Record(schema);
+ record.put("a", a);
+ record.put("b", b);
+ return record;
+ }
+
+ private static List<HiveColumnHandle> projectedColumns()
+ {
+ return List.of(
+ HiveColumnHandle.createBaseColumn("a", 0, HiveType.HIVE_LONG,
BIGINT, HiveColumnHandle.ColumnType.REGULAR, Optional.empty()),
+ HiveColumnHandle.createBaseColumn("b", 1,
HiveType.HIVE_STRING, VARCHAR, HiveColumnHandle.ColumnType.REGULAR,
Optional.empty()),
+ HiveColumnHandle.createBaseColumn("pk_int", -1,
HiveType.HIVE_INT, INTEGER, HiveColumnHandle.ColumnType.PARTITION_KEY,
Optional.empty()));
+ }
+
+ private static PrefilledColumnValues prefilledValues()
+ {
+ HudiBaseFile baseFile = new
HudiBaseFile("s3://bucket/table/file1.parquet", "file1.parquet", 1234,
1700000000123L, 0, 1234);
+ HudiSplit split = new HudiSplit(
+ baseFile,
+ List.of(),
+ "001",
+ TupleDomain.all(),
+ List.of(new HivePartitionKey("pk_int", "42")),
+ SplitWeight.standard());
+ return PrefilledColumnValues.create(split);
+ }
+}
