You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/09/14 17:08:43 UTC
(iceberg) branch main updated: MR: Pass partition constants to the format model read builder (#18037)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 2cac9eeb68 MR: Pass partition constants to the format model read builder (#18037)
2cac9eeb68 is described below
commit 2cac9eeb68da76094e60e2510f92753ac6cae50c
Author: Eunbin Son <58...@users.noreply.github.com>
AuthorDate: Tue Sep 15 02:08:37 2026 +0900
MR: Pass partition constants to the format model read builder (#18037)
---
.../iceberg/mr/mapreduce/IcebergInputFormat.java | 5 +++
.../apache/iceberg/mr/TestIcebergInputFormats.java | 51 ++++++++++++++++++++++
2 files changed, 56 insertions(+)
diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
index 033b359850..764648e263 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
@@ -45,6 +45,7 @@ import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
import org.apache.iceberg.data.DeleteFilter;
import org.apache.iceberg.data.GenericDeleteFilter;
+import org.apache.iceberg.data.IdentityPartitionConverters;
import org.apache.iceberg.data.InternalRecordWrapper;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.encryption.EncryptedFiles;
@@ -63,6 +64,7 @@ import org.apache.iceberg.mapping.NameMappingParser;
import org.apache.iceberg.mr.Catalogs;
import org.apache.iceberg.mr.InputFormatConfig;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.PartitionUtil;
import org.apache.iceberg.util.SerializationUtil;
import org.apache.iceberg.util.ThreadPools;
@@ -328,6 +330,9 @@ public class IcebergInputFormat<T> extends InputFormat<Void, T> {
(CloseableIterable<T>)
readBuilder
.project(readSchema)
+ .idToConstant(
+ PartitionUtil.constantsMap(
+ currentTask, IdentityPartitionConverters::convertConstant))
.split(currentTask.start(), currentTask.length())
.caseSensitive(caseSensitive)
.filter(currentTask.residual())
diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
index ce588a7e83..7be03cd7bc 100644
--- a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
+++ b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
@@ -31,6 +31,7 @@ import java.security.PrivilegedAction;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
@@ -47,19 +48,26 @@ import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
import org.apache.iceberg.Parameter;
import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.Parameters;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.TestHelpers.Row;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.FileHelpers;
+import org.apache.iceberg.data.GenericFileWriterFactory;
+import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.mr.mapred.Container;
import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat;
import org.apache.iceberg.mr.mapreduce.IcebergInputFormat;
@@ -298,6 +306,49 @@ public class TestIcebergInputFormats {
}
}
+ @TestTemplate
+ void identityPartitionValuesMissingFromDataFile() throws Exception {
+ Table table = helper.createTable(LOG_SCHEMA, IDENTITY_PARTITION_SPEC);
+
+ // Tables imported with add_files/migrate have data files that do not physically store the
+ // identity partition columns; their values only exist in the manifest entry. See
+ // TableMigrationUtil#listPartition, which reads the partition values from Hive metadata.
+ Schema fileSchema = withColumns("id", "message");
+ Record record = GenericRecord.create(fileSchema);
+ record.setField("id", 1);
+ record.setField("message", "hello");
+
+ DataFile dataFile =
+ writeFileWithSchema(table, fileSchema, Row.of("2020-03-20", "info"), record);
+ table.newAppend().appendFile(dataFile).commit();
+
+ builder.project(LOG_SCHEMA);
+ List<Record> records = testInputFormat.create(builder.conf()).getRecords();
+
+ assertThat(records).hasSize(1);
+ assertThat(records.get(0).getField("date")).isEqualTo("2020-03-20");
+ assertThat(records.get(0).getField("level")).isEqualTo("info");
+ assertThat(records.get(0).getField("id")).isEqualTo(1);
+ assertThat(records.get(0).getField("message")).isEqualTo("hello");
+ }
+
+ private DataFile writeFileWithSchema(
+ Table table, Schema fileSchema, StructLike partition, Record record) throws IOException {
+ OutputFile outputFile =
+ Files.localOutput(temp.resolve("partial-schema-" + UUID.randomUUID()).toFile());
+ DataWriter<Record> writer =
+ new GenericFileWriterFactory.Builder(table)
+ .dataFileFormat(fileFormat)
+ .dataSchema(fileSchema)
+ .build()
+ .newDataWriter(FileHelpers.encrypt(outputFile), table.spec(), partition);
+ try (writer) {
+ writer.write(record);
+ }
+
+ return writer.toDataFile();
+ }
+
@TestTemplate
public void testSnapshotReads() throws Exception {
helper.createUnpartitionedTable();