diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergBucketFunction.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergBucketFunction.java index 7f513814a408..6b1235d6a127 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergBucketFunction.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergBucketFunction.java @@ -13,15 +13,18 @@ */ package io.trino.plugin.iceberg; +import com.google.common.collect.ImmutableList; import io.trino.plugin.iceberg.PartitionTransforms.ValueTransform; import io.trino.spi.Page; import io.trino.spi.block.Block; import io.trino.spi.block.RowBlock; import io.trino.spi.connector.BucketFunction; import io.trino.spi.connector.ConnectorSplit; +import io.trino.spi.type.Type; import io.trino.spi.type.TypeOperators; import java.lang.invoke.MethodHandle; +import java.util.ArrayList; import java.util.List; import java.util.function.ToIntFunction; @@ -32,6 +35,7 @@ import static io.trino.spi.function.InvocationConvention.InvocationReturnConvention.FAIL_ON_NULL; import static io.trino.spi.function.InvocationConvention.simpleConvention; import static io.trino.spi.type.TypeUtils.NULL_HASH_CODE; +import static io.trino.spi.type.TypeUtils.readNativeValue; import static java.util.Objects.requireNonNull; import static java.util.Objects.requireNonNullElse; @@ -40,6 +44,7 @@ public class IcebergBucketFunction { private final int bucketCount; private final List functions; + private final List partitionStructFields; private final boolean singleBucketFunction; @@ -54,6 +59,7 @@ public IcebergBucketFunction(IcebergPartitioningHandle partitioningHandle, TypeO this.functions = partitionFunctions.stream() .map(partitionFunction -> HashFunction.create(partitionFunction, typeOperators)) .collect(toImmutableList()); + this.partitionStructFields = ImmutableList.copyOf(partitioningHandle.partitionStructFields()); this.singleBucketFunction = partitionFunctions.size() == 1 && partitionFunctions.getFirst().transform() == BUCKET && @@ -81,8 +87,7 @@ public int getBucket(Page page, int position) @Override public int applyAsInt(ConnectorSplit split) { - List partitionValues = ((IcebergSplit) split).getPartitionValues() - .orElseThrow(() -> new IllegalArgumentException("Split does not contain partition values")); + List partitionValues = getPartitionValues(((IcebergSplit) split).getPartitionValues()); if (singleBucketFunction) { long bucket = (long) requireNonNullElse(partitionValues.getFirst(), 0L); @@ -99,7 +104,18 @@ public int applyAsInt(ConnectorSplit split) return (int) ((hash & Long.MAX_VALUE) % bucketCount); } - private record HashFunction(List dataPath, ValueTransform valueTransform, MethodHandle hashCodeOperator) + private List getPartitionValues(List partitionBlocks) + { + // using array list because the value could be null + List partitionValues = new ArrayList<>(partitionStructFields.size()); + for (int i = 0; i < partitionStructFields.size(); i++) { + int fieldIndex = partitionStructFields.get(i); + partitionValues.add(readNativeValue(functions.get(i).resultType(), partitionBlocks.get(fieldIndex), 0)); + } + return partitionValues; + } + + private record HashFunction(List dataPath, ValueTransform valueTransform, MethodHandle hashCodeOperator, Type resultType) { private static HashFunction create(IcebergPartitionFunction partitionFunction, TypeOperators typeOperators) { @@ -107,7 +123,8 @@ private static HashFunction create(IcebergPartitionFunction partitionFunction, T return new HashFunction( partitionFunction.dataPath(), columnTransform.valueTransform(), - typeOperators.getHashCodeOperator(columnTransform.type(), simpleConvention(FAIL_ON_NULL, NEVER_NULL))); + typeOperators.getHashCodeOperator(columnTransform.type(), simpleConvention(FAIL_ON_NULL, NEVER_NULL)), + columnTransform.type()); } private HashFunction diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java index fe94a28f4dcd..942cae8669f3 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java @@ -831,7 +831,7 @@ private Optional getTablePartitioning(ConnectorSession return Optional.empty(); } - IcebergPartitioningHandle partitioningHandle = IcebergPartitioningHandle.create(partitionSpec, typeManager, List.of()); + IcebergPartitioningHandle partitioningHandle = IcebergPartitioningHandle.create(partitionSpec, typeManager); Map partitionColumnById = getPartitionColumns(icebergTable, typeManager).stream() .collect(toImmutableMap(IcebergColumnHandle::getId, identity())); @@ -847,8 +847,7 @@ private Optional getTablePartitioning(ConnectorSession return Optional.of(new IcebergTablePartitioning( false, partitioningHandle, - partitionColumns, - IntStream.range(0, partitioningHandle.partitionFunctions().size()).boxed().collect(toImmutableList()))); + partitionColumns)); } private static long getSnapshotIdFromVersion(ConnectorSession session, Table table, ConnectorTableVersion version) @@ -1074,11 +1073,11 @@ public Optional applyPartitioning(ConnectorSession session // Change the index of the top level column to the location in the new partitioning columns newPartitionFunctions.add(function.withTopLevelColumnIndex(newColumnIndex)); // Some partition functions may be dropped so update the struct fields used in split partitioning must be updated - newPartitionStructFields.add(tablePartitioning.partitionStructFields().get(functionIndex)); + newPartitionStructFields.add(tablePartitioning.partitioningHandle().partitionStructFields().get(functionIndex)); } } - IcebergPartitioningHandle newPartitioningHandle = new IcebergPartitioningHandle(false, newPartitionFunctions.build()); + IcebergPartitioningHandle newPartitioningHandle = new IcebergPartitioningHandle(false, newPartitionFunctions.build(), newPartitionStructFields.build()); if (partitioningHandle.isPresent() && !partitioningHandle.get().equals(newPartitioningHandle)) { // todo if bucketing is a power of two, we can adapt the bucketing return Optional.empty(); @@ -1092,8 +1091,7 @@ public Optional applyPartitioning(ConnectorSession session return Optional.of(icebergTableHandle.withTablePartitioning(Optional.of(new IcebergTablePartitioning( true, newPartitioningHandle, - partitioningColumns.stream().map(IcebergColumnHandle.class::cast).collect(toImmutableList()), - newPartitionStructFields.build())))); + partitioningColumns.stream().map(IcebergColumnHandle.class::cast).collect(toImmutableList()))))); } @Override @@ -1588,7 +1586,7 @@ private Optional getWriteLayout(Schema tableSchema, Partit // Do not set partitioningHandle, to let engine determine whether to repartition data or not, on stat-based basis. return Optional.of(new ConnectorTableLayout(partitioningColumnNames)); } - IcebergPartitioningHandle partitioningHandle = IcebergPartitioningHandle.create(partitionSpec, typeManager, List.of()); + IcebergPartitioningHandle partitioningHandle = IcebergPartitioningHandle.create(partitionSpec, typeManager); return Optional.of(new ConnectorTableLayout(partitioningHandle, partitioningColumnNames, true)); } diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java index a3aeb78de8c0..a82f5b0dfe12 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPageSourceProvider.java @@ -284,7 +284,7 @@ public ConnectorPageSource createPageSource( icebergColumns, schema, partitionSpec, - PartitionData.fromJson(split.getPartitionDataJson(), partitionColumnTypes), + PartitionData.fromBlocks(split.getPartitionValues(), partitionColumnTypes, typeManager), split.getDeletes(), dynamicFilter, tableHandle.getUnenforcedPredicate(), @@ -294,7 +294,6 @@ public ConnectorPageSource createPageSource( split.getLength(), split.getFileSize(), split.getFileRecordCount(), - split.getPartitionDataJson(), split.getFileFormat(), getFileIoProperties(connectorTableCredentials), split.getDataSequenceNumber(), @@ -317,7 +316,6 @@ public ConnectorPageSource createPageSource( long length, long fileSize, long fileRecordCount, - String partitionDataJson, IcebergFileFormat fileFormat, Map fileIoProperties, long dataSequenceNumber, @@ -370,7 +368,7 @@ public ConnectorPageSource createPageSource( length, fileSize, partitionSpec.specId(), - partitionDataJson, + PartitionData.toJson(partitionData), fileFormat, tableSchema, requiredColumns, diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPartitioningHandle.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPartitioningHandle.java index eff26d9501cf..5f59702326d5 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPartitioningHandle.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergPartitioningHandle.java @@ -29,27 +29,31 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.collect.ImmutableList.toImmutableList; import static com.google.common.collect.ImmutableSet.toImmutableSet; import static io.trino.plugin.iceberg.TypeConverter.toTrinoType; import static java.nio.charset.StandardCharsets.UTF_8; import static java.util.Objects.requireNonNull; -public record IcebergPartitioningHandle(boolean update, List partitionFunctions) +public record IcebergPartitioningHandle(boolean update, List partitionFunctions, List partitionStructFields) implements ConnectorPartitioningHandle { public IcebergPartitioningHandle { partitionFunctions = ImmutableList.copyOf(requireNonNull(partitionFunctions, "partitioning is null")); + partitionStructFields = ImmutableList.copyOf(partitionStructFields); + checkArgument(partitionFunctions.size() == partitionStructFields.size(), "partitionFunctions and partitionStructFields must have the same size"); } public IcebergPartitioningHandle forUpdate() { - return new IcebergPartitioningHandle(true, partitionFunctions); + return new IcebergPartitioningHandle(true, partitionFunctions, partitionStructFields); } - public static IcebergPartitioningHandle create(PartitionSpec spec, TypeManager typeManager, List partitioningColumns) + public static IcebergPartitioningHandle create(PartitionSpec spec, TypeManager typeManager) { Map> dataPaths = buildDataPaths(spec); List partitionFields = spec.fields().stream() @@ -59,7 +63,7 @@ public static IcebergPartitioningHandle create(PartitionSpec spec, TypeManager t toTrinoType(spec.schema().findType(field.sourceId()), typeManager))) .collect(toImmutableList()); - return new IcebergPartitioningHandle(false, partitionFields); + return new IcebergPartitioningHandle(false, partitionFields, IntStream.range(0, partitionFields.size()).boxed().collect(toImmutableList())); } /** diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplit.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplit.java index b6ea5d65b06f..6bed586ed031 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplit.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplit.java @@ -21,11 +21,11 @@ import io.trino.plugin.iceberg.delete.DeleteFile; import io.trino.spi.HostAddress; import io.trino.spi.SplitWeight; +import io.trino.spi.block.Block; import io.trino.spi.connector.ConnectorSplit; import io.trino.spi.predicate.TupleDomain; import java.util.List; -import java.util.Optional; import java.util.OptionalLong; import static com.google.common.base.MoreObjects.toStringHelper; @@ -46,9 +46,8 @@ public class IcebergSplit private final long fileSize; private final long fileRecordCount; private final IcebergFileFormat fileFormat; - private final Optional> partitionValues; private final int specId; - private final String partitionDataJson; + private final List partitionValues; private final List deletes; private final SplitWeight splitWeight; private final TupleDomain fileStatisticsDomain; @@ -65,7 +64,7 @@ public IcebergSplit( @JsonProperty("fileRecordCount") long fileRecordCount, @JsonProperty("fileFormat") IcebergFileFormat fileFormat, @JsonProperty("specId") int specId, - @JsonProperty("partitionDataJson") String partitionDataJson, + @JsonProperty("partitionValues") List partitionValues, @JsonProperty("deletes") List deletes, @JsonProperty("splitWeight") SplitWeight splitWeight, @JsonProperty("fileStatisticsDomain") TupleDomain fileStatisticsDomain, @@ -79,9 +78,8 @@ public IcebergSplit( fileSize, fileRecordCount, fileFormat, - Optional.empty(), specId, - partitionDataJson, + partitionValues, deletes, splitWeight, fileStatisticsDomain, @@ -97,9 +95,8 @@ public IcebergSplit( long fileSize, long fileRecordCount, IcebergFileFormat fileFormat, - Optional> partitionValues, int specId, - String partitionDataJson, + List partitionValues, List deletes, SplitWeight splitWeight, TupleDomain fileStatisticsDomain, @@ -113,9 +110,8 @@ public IcebergSplit( this.fileSize = fileSize; this.fileRecordCount = fileRecordCount; this.fileFormat = requireNonNull(fileFormat, "fileFormat is null"); - this.partitionValues = requireNonNull(partitionValues, "partitionValues is null"); this.specId = specId; - this.partitionDataJson = requireNonNull(partitionDataJson, "partitionDataJson is null"); + this.partitionValues = ImmutableList.copyOf(partitionValues); this.deletes = ImmutableList.copyOf(requireNonNull(deletes, "deletes is null")); this.splitWeight = requireNonNull(splitWeight, "splitWeight is null"); this.fileStatisticsDomain = requireNonNull(fileStatisticsDomain, "fileStatisticsDomain is null"); @@ -173,20 +169,10 @@ public int getSpecId() return specId; } - /** - * Trino (stack) values of the partition columns. The values are the result of evaluating - * the partition expressions on the partition data. - */ - @JsonIgnore - public Optional> getPartitionValues() - { - return partitionValues; - } - @JsonProperty - public String getPartitionDataJson() + public List getPartitionValues() { - return partitionDataJson; + return partitionValues; } @JsonProperty @@ -227,7 +213,7 @@ public long getRetainedSizeInBytes() + estimatedSizeOf(path) + SIZE_OF_LONG * 4 // start, length, fileSize, fileRecordCount + SIZE_OF_INT // specId - + estimatedSizeOf(partitionDataJson) + + estimatedSizeOf(partitionValues, Block::getRetainedSizeInBytes) + estimatedSizeOf(deletes, DeleteFile::retainedSizeInBytes) + splitWeight.getRetainedSizeInBytes() + fileStatisticsDomain.getRetainedSizeInBytes(IcebergColumnHandle::getRetainedSizeInBytes) diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java index d5e50213e495..9d81375a524e 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergSplitSource.java @@ -36,6 +36,7 @@ import io.trino.plugin.iceberg.delete.DeleteFile; import io.trino.plugin.iceberg.util.DataFileWithDeleteFiles; import io.trino.spi.SplitWeight; +import io.trino.spi.block.Block; import io.trino.spi.connector.ColumnHandle; import io.trino.spi.connector.ConnectorSession; import io.trino.spi.connector.ConnectorSplit; @@ -116,8 +117,10 @@ import static io.trino.plugin.iceberg.IcebergUtil.primitiveFieldTypes; import static io.trino.plugin.iceberg.StructLikeWrapperWithFieldIdToIndex.createStructLikeWrapper; import static io.trino.plugin.iceberg.TypeConverter.toIcebergType; +import static io.trino.plugin.iceberg.TypeConverter.toTrinoType; import static io.trino.spi.type.DateTimeEncoding.packDateTimeWithZone; import static io.trino.spi.type.TimeZoneKey.UTC_KEY; +import static io.trino.spi.type.TypeUtils.writeNativeValue; import static java.lang.Math.clamp; import static java.util.Collections.emptyIterator; import static java.util.Objects.requireNonNull; @@ -725,18 +728,6 @@ static boolean partitionMatchesPredicate( private IcebergSplit toIcebergSplit(FileScanTaskWithDomain taskWithDomain) { FileScanTask task = taskWithDomain.fileScanTask(); - Optional> partitionValues = Optional.empty(); - if (tableHandle.getTablePartitioning().isPresent()) { - PartitionSpec partitionSpec = task.spec(); - StructLike partition = task.file().partition(); - List fields = partitionSpec.fields(); - - partitionValues = Optional.of(tableHandle.getTablePartitioning().get().partitionStructFields().stream() - .map(fieldIndex -> convertIcebergValueToTrino( - partitionSpec.partitionType().field(fields.get(fieldIndex).fieldId()).type(), - partition.get(fieldIndex, Object.class))) - .toList()); - } return new IcebergSplit( task.file().location(), @@ -745,9 +736,8 @@ private IcebergSplit toIcebergSplit(FileScanTaskWithDomain taskWithDomain) task.file().fileSizeInBytes(), task.file().recordCount(), IcebergFileFormat.fromIceberg(task.file().format()), - partitionValues, task.spec().specId(), - PartitionData.toJson(task.file().partition()), + getPartitionBlockValues(task, typeManager), task.deletes().stream() .peek(file -> verifyDeletionVectorReferencesDataFile(task, file)) .map(DeleteFile::fromIceberg) @@ -759,6 +749,21 @@ private IcebergSplit toIcebergSplit(FileScanTaskWithDomain taskWithDomain) task.file().firstRowId() == null ? OptionalLong.empty() : OptionalLong.of(task.file().firstRowId())); } + private static List getPartitionBlockValues(FileScanTask task, TypeManager typeManager) + { + PartitionSpec spec = task.spec(); + StructLike partition = task.file().partition(); + List fields = spec.fields(); + + ImmutableList.Builder partitionValues = ImmutableList.builder(); + for (int fieldIndex = 0; fieldIndex < fields.size(); fieldIndex++) { + Type icebergType = spec.partitionType().field(fields.get(fieldIndex).fieldId()).type(); + Object partitionValue = convertIcebergValueToTrino(icebergType, partition.get(fieldIndex, Object.class)); + partitionValues.add(writeNativeValue(toTrinoType(icebergType, typeManager), partitionValue)); + } + return partitionValues.build(); + } + private static void verifyDeletionVectorReferencesDataFile(FileScanTask task, org.apache.iceberg.DeleteFile deleteFile) { if (deleteFile.format() != FileFormat.PUFFIN || deleteFile.contentOffset() == null || deleteFile.contentSizeInBytes() == null) { diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTablePartitioning.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTablePartitioning.java index 5d5562e50c49..1b804ca6ef6d 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTablePartitioning.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTablePartitioning.java @@ -19,26 +19,17 @@ import java.util.List; import java.util.Optional; -import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; public record IcebergTablePartitioning( boolean active, IcebergPartitioningHandle partitioningHandle, - List partitioningColumns, - List partitionStructFields) + List partitioningColumns) { public IcebergTablePartitioning { requireNonNull(partitioningHandle, "partitioningHandle is null"); partitioningColumns = ImmutableList.copyOf(requireNonNull(partitioningColumns, "partitioningColumns is null")); - partitionStructFields = ImmutableList.copyOf(requireNonNull(partitionStructFields, "partitionStructFields is null")); - checkArgument(partitioningHandle.partitionFunctions().size() == partitionStructFields.size(), "partitioningColumns and partitionStructFields must have the same size"); - } - - public IcebergTablePartitioning activate() - { - return new IcebergTablePartitioning(true, partitioningHandle, partitioningColumns, partitionStructFields); } public Optional toConnectorTablePartitioning() diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/PartitionData.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/PartitionData.java index bd67906fbb8d..d19d1e46d0d9 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/PartitionData.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/PartitionData.java @@ -18,6 +18,8 @@ import com.fasterxml.jackson.databind.DeserializationFeature; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.json.JsonMapper; +import io.trino.spi.block.Block; +import io.trino.spi.type.TypeManager; import org.apache.iceberg.StructLike; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; @@ -26,11 +28,15 @@ import java.io.StringWriter; import java.io.UncheckedIOException; import java.nio.ByteBuffer; +import java.util.List; import java.util.UUID; import static io.trino.plugin.base.util.JsonUtils.jsonFactory; +import static io.trino.plugin.iceberg.IcebergTypes.convertTrinoValueToIceberg; +import static io.trino.plugin.iceberg.TypeConverter.toTrinoType; import static io.trino.spi.type.DecimalType.createDecimalType; import static io.trino.spi.type.Decimals.rescale; +import static io.trino.spi.type.TypeUtils.readNativeValue; import static java.lang.String.format; import static java.util.Objects.requireNonNull; @@ -99,6 +105,26 @@ public static String toJson(StructLike structLike) } } + public static PartitionData fromBlocks(List partitionValues, Type[] types, TypeManager typeManager) + { + if (types.length == 0) { + return new PartitionData(new Object[0]); + } + + Object[] values = new Object[types.length]; + for (int i = 0; i < types.length; i++) { + io.trino.spi.type.Type trinoType = toTrinoType(types[i], typeManager); + Object value = readNativeValue(trinoType, partitionValues.get(i), 0); + if (value == null) { + values[i] = null; + } + else { + values[i] = convertTrinoValueToIceberg(trinoType, value); + } + } + return new PartitionData(values); + } + public static PartitionData fromJson(String partitionDataAsJson, Type[] types) { if (partitionDataAsJson == null) { diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/functions/tablechanges/TableChangesFunctionProcessor.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/functions/tablechanges/TableChangesFunctionProcessor.java index c1db3c3d09f2..53d5d8c0f96e 100644 --- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/functions/tablechanges/TableChangesFunctionProcessor.java +++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/functions/tablechanges/TableChangesFunctionProcessor.java @@ -133,7 +133,6 @@ else if (column.getId() == DATA_CHANGE_ORDINAL_ID) { split.length(), split.fileSize(), split.fileRecordCount(), - split.partitionDataJson(), split.fileFormat(), getFileIoProperties(tableCredentials), 0, diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java index d7be3c34682e..9bd3a382daab 100644 --- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java +++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java @@ -3917,6 +3917,27 @@ public void testApplyFilterWithNonEmptyConstraintPredicate() assertUpdate("DROP TABLE test_apply_functional_constraint"); } + @Test + void testVoidTransformWithMultiplePartitionFields() + { + // test multiple partition fields with void transform in front of the other partition fields + try (TestTable table = newTrinoTable("test_void_transform_", "(a VARCHAR, b BIGINT, c VARCHAR) WITH (partitioning = ARRAY['void(b)', 'c'])")) { + String values = """ + (VARCHAR 'abcd', BIGINT '1', VARCHAR 'x'), + ('abxy', 2, 'x'), + ('xyzd', 3, 'x'), + (NULL, 4, 'xx'), + (NULL, 5, NULL) + """; + assertUpdate("INSERT INTO " + table.getName() + " VALUES " + values, 5); + assertThat(query("TABLE " + table.getName())) + .matches("VALUES " + values); + + assertThat(query("SELECT COUNT(*) FROM \"%s$partitions\"".formatted(table.getName()))) + .matches("VALUES BIGINT '3'"); + } + } + @Test public void testVoidTransform() { diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergNodeLocalDynamicSplitPruning.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergNodeLocalDynamicSplitPruning.java index 9bf6a3d782fd..2b2cbbd07132 100644 --- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergNodeLocalDynamicSplitPruning.java +++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergNodeLocalDynamicSplitPruning.java @@ -78,6 +78,7 @@ import static io.trino.spi.type.DateType.DATE; import static io.trino.spi.type.Decimals.writeShortDecimal; import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.TypeUtils.writeNativeValue; import static io.trino.spi.type.VarcharType.VARCHAR; import static io.trino.testing.TestingHandles.TEST_CATALOG_HANDLE; import static io.trino.type.InternalTypeManager.TESTING_TYPE_MANAGER; @@ -146,7 +147,7 @@ public void testDynamicSplitPruningOnUnpartitionedTable() -1, // invalid; normally known ORC, PartitionSpec.unpartitioned().specId(), - PartitionData.toJson(new PartitionData(new Object[] {})), + ImmutableList.of(), ImmutableList.of(), SplitWeight.standard(), TupleDomain.all(), @@ -208,7 +209,7 @@ public void testDynamicSplitPruningOnUnpartitionedTable() -1, // invalid; normally known ORC, PartitionSpec.unpartitioned().specId(), - PartitionData.toJson(new PartitionData(new Object[] {})), + ImmutableList.of(), ImmutableList.of(), SplitWeight.standard(), TupleDomain.withColumnDomains(ImmutableMap.of(keyColumnHandle, Domain.singleValue(INTEGER, (long) keyColumnValue))), @@ -318,7 +319,7 @@ public void testDynamicSplitPruningWithExplicitPartitionFilter() -1, // invalid; normally known ORC, partitionSpec.specId(), - PartitionData.toJson(new PartitionData(new Object[] {dateColumnValue})), + ImmutableList.of(writeNativeValue(DATE, dateColumnValue)), ImmutableList.of(), SplitWeight.standard(), TupleDomain.all(), @@ -471,7 +472,7 @@ public void testDynamicSplitPruningWithExplicitPartitionFilterPartitionEvolution -1, // invalid; normally known ORC, partitionSpec.specId(), - PartitionData.toJson(new PartitionData(new Object[] {yearColumnValue})), + ImmutableList.of(writeNativeValue(INTEGER, yearColumnValue)), ImmutableList.of(), SplitWeight.standard(), TupleDomain.all(), diff --git a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergPageSourceProvider.java b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergPageSourceProvider.java index 9ee830ee5d0d..fd02127d72ed 100644 --- a/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergPageSourceProvider.java +++ b/plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/TestIcebergPageSourceProvider.java @@ -140,7 +140,6 @@ void testMemoryTrackingWithEqualityDeletes(@TempDir Path tempDir) dataInputFile.length(), dataInputFile.length(), 3, // fileRecordCount - PartitionData.toJson(new PartitionData(new Object[] {})), PARQUET, ImmutableMap.of(), 0L, // dataSequenceNumber