Skip to content

(IcebergIO) bugfix: wire writeProperties through table create request, not DataWriteBuilder #39645

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
clairemcginty wants to merge 1 commit into
base: master
Choose a base branch
from
Open

(IcebergIO) bugfix: wire writeProperties through table create request, not DataWriteBuilder #39645

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 4 additions & 20 deletions sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@
package org.apache.beam.sdk.io.iceberg;

import java.io.IOException;
import java.util.Map;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.metrics.Metrics;
import org.apache.iceberg.DataFile;
Expand All @@ -35,7 +34,6 @@
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.parquet.Parquet;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand All @@ -56,22 +54,11 @@ class RecordWriter {
catalog.loadTable(destination.getTableIdentifier()),
destination.getFileFormat(),
filename,
partitionKey,
null);
partitionKey);
}

RecordWriter(Table table, FileFormat fileFormat, String filename, StructLike partitionKey)
throws IOException {
this(table, fileFormat, filename, partitionKey, null);
}

RecordWriter(
Table table,
FileFormat fileFormat,
String filename,
StructLike partitionKey,
@Nullable Map<String, String> writeProperties)
throws IOException {
this.table = table;
this.fileFormat = fileFormat;

Expand Down Expand Up @@ -104,17 +91,14 @@ class RecordWriter {
.build();
break;
case PARQUET:
Parquet.DataWriteBuilder parquetBuilder =
icebergDataWriter =
Parquet.writeData(outputFile)
.forTable(table)
.createWriterFunc(GenericParquetWriter::create)
.withPartition(partitionKey)
.withKeyMetadata(keyMetadata)
.overwrite();
if (writeProperties != null && !writeProperties.isEmpty()) {
parquetBuilder.setAll(writeProperties);
}
icebergDataWriter = parquetBuilder.build();
Comment on lines -114 to -117

@ahmedabu98 ahmedabu98 Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removing this makes writeProperties a no-op if the table already exists

.overwrite()
.build();
break;
case ORC:
throw new UnsupportedOperationException("ORC file format not currently supported.");
Expand Down
8 changes: 5 additions & 3 deletions sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/RecordWriterManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,8 +202,7 @@ private RecordWriter createWriter(PartitionKey partitionKey) {
table,
icebergDestination.getFileFormat(),
filePrefix + "_" + stateToken + "_" + recordIndex,
partitionKey,
writeProperties);
partitionKey);
openWriters++;
return writer;
} catch (IOException e) {
Expand Down Expand Up @@ -311,8 +310,11 @@ private Table loadOrCreateTable(IcebergDestination destination, Schema dataSchem
SortOrder sortOrder = createConfig != null ? createConfig.getSortOrder() : SortOrder.unsorted();
Map<String, String> tableProperties =
createConfig != null && createConfig.getTableProperties() != null
? createConfig.getTableProperties()
? Maps.newHashMap(createConfig.getTableProperties())
: Maps.newHashMap();
if (writeProperties != null) {
tableProperties.putAll(writeProperties);
}
Comment on lines +315 to +317

@ahmedabu98 ahmedabu98 Aug 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

writeProperties are supposed to be execution-scoped. They're not meant to be persisted in the actual table's properties


// Create namespace if it does not exist yet
if (!namespace.isEmpty() && catalog instanceof SupportsNamespaces) {
Expand Down
8 changes: 5 additions & 3 deletions .../io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WritePartitionedRowsToFiles.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,8 +141,7 @@ public void processElement(
.addExtension(String.format("%s-%s", filePrefix, UUID.randomUUID()));

RecordWriter writer =
new RecordWriter(
table, destination.getFileFormat(), fileName, partitionData, writeProperties);
new RecordWriter(table, destination.getFileFormat(), fileName, partitionData);
try {
for (Row row : element.getValue()) {
Record record = IcebergUtils.beamRowToIcebergRecord(table.schema(), row);
Expand Down Expand Up @@ -194,8 +193,11 @@ private Table loadOrCreateTable(
createConfig != null ? createConfig.getSortOrder() : SortOrder.unsorted();
Map<String, String> tableProperties =
createConfig != null && createConfig.getTableProperties() != null
? createConfig.getTableProperties()
? Maps.newHashMap(createConfig.getTableProperties())
: Maps.newHashMap();
if (writeProperties != null) {
tableProperties.putAll(writeProperties);
}

// Create namespace if it does not exist yet
if (!namespace.isEmpty() && catalog instanceof SupportsNamespaces) {
Expand Down
106 changes: 106 additions & 0 deletions sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergIOWriteTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,9 +88,13 @@
import org.apache.iceberg.data.IcebergGenerics;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.parquet.Parquet;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.hamcrest.Matchers;
import org.joda.time.Duration;
import org.joda.time.Instant;
Expand Down Expand Up @@ -829,4 +833,106 @@ public void testCreateTableWithPartitionSpecAndSortOrder() {
List<Record> writtenRecords = ImmutableList.copyOf(IcebergGenerics.read(table).build());
assertThat(writtenRecords, Matchers.containsInAnyOrder(TestFixtures.FILE1SNAPSHOT1.toArray()));
}

@Test
public void testWriteWithParquetProperties() throws Exception {
TableIdentifier tableId =
TableIdentifier.of(
"default", "parquet_props_" + Long.toString(UUID.randomUUID().hashCode(), 16));

Schema beamSchema = IcebergUtils.icebergSchemaToBeamSchema(TestFixtures.SCHEMA);

Map<String, String> catalogProps =
ImmutableMap.<String, String>builder()
.put("type", CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP)
.put("warehouse", warehouse.location)
.build();

IcebergCatalogConfig catalog =
IcebergCatalogConfig.builder()
.setCatalogName("name")
.setCatalogProperties(catalogProps)
.build();

testPipeline
.apply("Records To Add", Create.of(TestFixtures.asRows(TestFixtures.FILE1SNAPSHOT1)))
.setRowSchema(beamSchema)
.apply(
"Append To Table",
writeTransform(catalog, tableId)
.withWriteProperties(
ImmutableMap.of("write.parquet.bloom-filter-enabled.column.data", "true")));

testPipeline.run().waitUntilFinish();

Table table = warehouse.loadTable(tableId);

List<Record> writtenRecords = ImmutableList.copyOf(IcebergGenerics.read(table).build());
assertThat(writtenRecords, Matchers.containsInAnyOrder(TestFixtures.FILE1SNAPSHOT1.toArray()));

// verify bloom filter is present on 'data' column in written parquet files
try (CloseableIterable<org.apache.iceberg.FileScanTask> tasks = table.newScan().planFiles()) {
for (org.apache.iceberg.FileScanTask task : tasks) {
String path = task.file().path().toString();
try (ParquetFileReader reader =
ParquetFileReader.open(
org.apache.parquet.hadoop.util.HadoopInputFile.fromPath(
new org.apache.hadoop.fs.Path(path),
new org.apache.hadoop.conf.Configuration()))) {
for (BlockMetaData block : reader.getFooter().getBlocks()) {
for (ColumnChunkMetaData col : block.getColumns()) {
boolean hasBloom = col.getBloomFilterOffset() > 0;
if (col.getPath().toDotString().equals("data")) {
assertTrue("Expected bloom filter on column 'data', but none was found", hasBloom);
} else {
assertFalse(
"Expected no bloom filter on column '" + col.getPath().toDotString() + "'",
hasBloom);
}
}
}
}
}
}
}

@Test
public void testWriteWithTableProperties() throws Exception {
TableIdentifier tableId =
TableIdentifier.of(
"default", "table_props_" + Long.toString(UUID.randomUUID().hashCode(), 16));

Schema beamSchema = IcebergUtils.icebergSchemaToBeamSchema(TestFixtures.SCHEMA);

Map<String, String> catalogProps =
ImmutableMap.<String, String>builder()
.put("type", CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP)
.put("warehouse", warehouse.location)
.build();

IcebergCatalogConfig catalog =
IcebergCatalogConfig.builder()
.setCatalogName("name")
.setCatalogProperties(catalogProps)
.build();

testPipeline
.apply("Records To Add", Create.of(TestFixtures.asRows(TestFixtures.FILE1SNAPSHOT1)))
.setRowSchema(beamSchema)
.apply(
"Append To Table",
writeTransform(catalog, tableId)
.withWriteProperties(
ImmutableMap.of("write.data.path", warehouse.location + "/custom_data_path")));

testPipeline.run().waitUntilFinish();

Table table = warehouse.loadTable(tableId);

List<Record> writtenRecords = ImmutableList.copyOf(IcebergGenerics.read(table).build());
assertThat(writtenRecords, Matchers.containsInAnyOrder(TestFixtures.FILE1SNAPSHOT1.toArray()));

assertEquals(
warehouse.location + "/custom_data_path", table.properties().get("write.data.path"));
}
}
69 changes: 0 additions & 69 deletions ...java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,6 @@
import org.apache.beam.sdk.values.WindowedValues;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
Expand All @@ -79,10 +77,6 @@
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.DateTimeUtil;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.apache.parquet.hadoop.util.HadoopInputFile;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
Expand Down Expand Up @@ -1306,67 +1300,4 @@ public void testFileIOSurvivesAcrossBundles() throws IOException {
assertTrue(
"Bundle 2 should produce data files", bundle2.getSerializableDataFiles().containsKey(dest));
}

@Test
public void testWritePropertiesAppliedToParquetFiles() throws IOException {
Schema bloomSchema =
Schema.builder().addInt32Field("colWithBf").addInt32Field("colWithoutBf").build();
org.apache.iceberg.Schema icebergBloomSchema =
IcebergUtils.beamSchemaToIcebergSchema(bloomSchema);

TableIdentifier tableId = TableIdentifier.of("default", "test_write_properties");
warehouse.createTable(tableId, icebergBloomSchema);

Map<String, String> writeProperties =
ImmutableMap.of(
"write.parquet.bloom-filter-enabled.column.colWithBf", "true",
"write.parquet.bloom-filter-enabled.column.colWithoutBf", "false");

IcebergDestination destination =
IcebergDestination.builder()
.setTableIdentifier(tableId)
.setFileFormat(FileFormat.PARQUET)
.build();
WindowedValue<IcebergDestination> dest = WindowedValues.valueInGlobalWindow(destination);

RecordWriterManager writerManager =
new RecordWriterManager(catalogConfig, "test_bloom", Long.MAX_VALUE, 3, writeProperties);
for (int i = 0; i < 10; i++) {
Row row = Row.withSchema(bloomSchema).addValues(i, 100 + i).build();
assertTrue(writerManager.write(dest, row));
}
writerManager.close();

List<SerializableDataFile> dataFiles = writerManager.getSerializableDataFiles().get(dest);
assertEquals(1, dataFiles.size());

String dataFilePath = dataFiles.get(0).getPath();
assertNotNull(dataFilePath);

try (ParquetFileReader reader =
ParquetFileReader.open(
HadoopInputFile.fromPath(new Path(dataFilePath), new Configuration()))) {
List<BlockMetaData> blocks = reader.getFooter().getBlocks();
assertFalse("Parquet file should have at least one row group", blocks.isEmpty());

for (int i = 0; i < blocks.size(); i++) {
BlockMetaData block = blocks.get(i);
assertEquals("Each row group should have 2 columns", 2, block.getColumns().size());

for (ColumnChunkMetaData col : block.getColumns()) {
boolean hasBloomFilter = col.getBloomFilterOffset() > 0;
String colName = col.getPath().toDotString();
if (colName.equals("colWithBf")) {
assertTrue(
"Column 'colWithBf' in row group " + i + " should have a bloom filter",
hasBloomFilter);
} else if (colName.equals("colWithoutBf")) {
assertFalse(
"Column 'colWithoutBf' in row group " + i + " should not have a bloom filter",
hasBloomFilter);
}
}
}
}
}
}
Loading