You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "pvary (via GitHub)" <gi...@apache.org> on 2026/02/15 15:48:36 UTC
[PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
pvary opened a new pull request, #15328:
URL: https://github.com/apache/iceberg/pull/15328
Part of: #12298
Implementation of the new API: #12774
SparkFormatModel and related changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819457255
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -754,30 +738,37 @@ private static PositionDeleteWriter<Record> positionDeletesWriter(
StructLike partition,
Schema rowSchema)
throws IOException {
- switch (format) {
- case AVRO:
- return Avro.writeDeletes(outputFile)
- .createWriterFunc(DataWriter::create)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- case PARQUET:
- return Parquet.writeDeletes(outputFile)
- .createWriterFunc(GenericParquetWriter::create)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- case ORC:
- return ORC.writeDeletes(outputFile)
- .createWriterFunc(GenericOrcWriter::buildWriter)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- default:
- throw new UnsupportedOperationException("Unsupported file format: " + format);
+ if (rowSchema == null) {
+ return FormatModelRegistry.<Record>positionDeleteWriteBuilder(
+ format, EncryptedFiles.plainAsEncryptedOutput(outputFile))
+ .partition(partition)
+ .spec(spec)
+ .build();
+ } else {
+ return switch (format) {
+ case AVRO ->
+ Avro.writeDeletes(outputFile)
+ .createWriterFunc(DataWriter::create)
+ .withPartition(partition)
+ .rowSchema(rowSchema)
Review Comment:
Just to double check: what was the rationale for not plumbing a schema passed to the registered builder via `schema` in for this purpose? Couldn't we have the `FileWriterBuilderImpl` return a `PositionDeleteWriter<PositionDelete<InternalRow>>`?
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -754,30 +738,37 @@ private static PositionDeleteWriter<Record> positionDeletesWriter(
StructLike partition,
Schema rowSchema)
throws IOException {
- switch (format) {
- case AVRO:
- return Avro.writeDeletes(outputFile)
- .createWriterFunc(DataWriter::create)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- case PARQUET:
- return Parquet.writeDeletes(outputFile)
- .createWriterFunc(GenericParquetWriter::create)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- case ORC:
- return ORC.writeDeletes(outputFile)
- .createWriterFunc(GenericOrcWriter::buildWriter)
- .withPartition(partition)
- .rowSchema(rowSchema)
- .withSpec(spec)
- .buildPositionWriter();
- default:
- throw new UnsupportedOperationException("Unsupported file format: " + format);
+ if (rowSchema == null) {
+ return FormatModelRegistry.<Record>positionDeleteWriteBuilder(
+ format, EncryptedFiles.plainAsEncryptedOutput(outputFile))
+ .partition(partition)
+ .spec(spec)
+ .build();
+ } else {
+ return switch (format) {
+ case AVRO ->
+ Avro.writeDeletes(outputFile)
+ .createWriterFunc(DataWriter::create)
+ .withPartition(partition)
+ .rowSchema(rowSchema)
Review Comment:
Just to double check: what was the rationale for not plumbing a schema passed to the registered builder via `schema` in for this purpose? Couldn't we have the `FileWriterBuilderImpl` return a `PositionDeleteWriter<PositionDelete<InternalRow>>`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2821217537
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -681,6 +680,13 @@ public CloseableIterable<Record> reader(
return positionDeletesReader(inputFile, format, spec);
}
+ @Override
+ public PositionDeleteWriter<Record> writer(
Review Comment:
Removed. Probably something remaining from previous iterations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2816719209
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
+ try (DataWriter<InternalRow> writer =
+ FormatModelRegistry.dataWriteBuilder(
+ FileFormat.PARQUET,
+ InternalRow.class,
+ EncryptedFiles.plainAsEncryptedOutput(Files.localOutput(dataFile)))
+ .schema(SCHEMA)
+ .engineSchema(SparkSchemaUtil.convert(SCHEMA))
Review Comment:
Probably we could create a new builderWriter method to avoid the need to set the engine schema for Spark when it is not needed, like this:
```
public static <T> ParquetValueWriter<T> buildWriter(
Schema icebergSchema, StructType dfSchema, MessageType type) {
return (ParquetValueWriter<T>)
ParquetWithSparkSchemaVisitor.visit(
dfSchema != null ? dfSchema : SparkSchemaUtil.convert(icebergSchema),
type,
new WriteBuilder(type));
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#issuecomment-3916422025
we mentioned this PR in the [Iceberg <> Spark community sync](https://docs.google.com/document/d/19nno1RoPznbbxKOZZddZNHHafa7XULjbN6RPExdr2n4/edit?tab=t.0#heading=h.r977qio1wsv2) today. Would be great to get some more 👀 on it!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2822960079
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java:
##########
@@ -75,10 +77,27 @@
public class SparkParquetWriters {
private SparkParquetWriters() {}
- @SuppressWarnings("unchecked")
public static <T> ParquetValueWriter<T> buildWriter(StructType dfSchema, MessageType type) {
+ return buildWriter(null, type, dfSchema);
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> ParquetValueWriter<T> buildWriter(
+ Schema icebergSchema, MessageType type, StructType dfSchema) {
+ return (ParquetValueWriter<T>)
+ ParquetWithSparkSchemaVisitor.visit(
+ dfSchema != null ? dfSchema : SparkSchemaUtil.convert(icebergSchema),
+ type,
+ new WriteBuilder(type));
+ }
+
+ public static <T> ParquetValueWriter<T> buildWriter(
+ StructType dfSchema, MessageType type, Schema icebergSchema) {
return (ParquetValueWriter<T>)
- ParquetWithSparkSchemaVisitor.visit(dfSchema, type, new WriteBuilder(type));
+ ParquetWithSparkSchemaVisitor.visit(
+ dfSchema != null ? dfSchema : SparkSchemaUtil.convert(icebergSchema),
+ type,
+ new WriteBuilder(type));
}
Review Comment:
I’ve given this quite a bit of thought. On the caller side we use the order`icebergSchema`, `fileSchema`, `engineSchema`, and I believe this is the most logical ordering. If anyone feels strongly otherwise, I’m happy to adjust it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2822548412
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java:
##########
@@ -75,10 +77,27 @@
public class SparkParquetWriters {
private SparkParquetWriters() {}
- @SuppressWarnings("unchecked")
public static <T> ParquetValueWriter<T> buildWriter(StructType dfSchema, MessageType type) {
+ return buildWriter(null, type, dfSchema);
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> ParquetValueWriter<T> buildWriter(
+ Schema icebergSchema, MessageType type, StructType dfSchema) {
+ return (ParquetValueWriter<T>)
+ ParquetWithSparkSchemaVisitor.visit(
+ dfSchema != null ? dfSchema : SparkSchemaUtil.convert(icebergSchema),
+ type,
+ new WriteBuilder(type));
+ }
+
+ public static <T> ParquetValueWriter<T> buildWriter(
+ StructType dfSchema, MessageType type, Schema icebergSchema) {
return (ParquetValueWriter<T>)
- ParquetWithSparkSchemaVisitor.visit(dfSchema, type, new WriteBuilder(type));
+ ParquetWithSparkSchemaVisitor.visit(
+ dfSchema != null ? dfSchema : SparkSchemaUtil.convert(icebergSchema),
+ type,
+ new WriteBuilder(type));
}
Review Comment:
is it intentional here to have 2 functions with different signature ordering? might be confusing
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,110 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
Review Comment:
```suggestion
this.format = deleteFileFormat;
```
`this.format` is used in `newPositionDeleteWriter`, so i think it should be the delete file format
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,110 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
this.writeProperties = writeProperties != null ? writeProperties : ImmutableMap.of();
+ this.positionDeleteRowSchema = null;
+ this.useDeprecatedPositionDeleteWriter = false;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(dataSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(equalityDeleteSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- boolean withRow =
- positionDeleteSparkType().getFieldIndex(DELETE_FILE_ROW_FIELD_NAME).isDefined();
- if (withRow) {
- // SparkAvroWriter accepts just the Spark type of the row ignoring the path and pos
- StructField rowField = positionDeleteSparkType().apply(DELETE_FILE_ROW_FIELD_NAME);
- StructType positionDeleteRowSparkType = (StructType) rowField.dataType();
- builder.createWriterFunc(ignored -> new SparkAvroWriter(positionDeleteRowSparkType));
- }
-
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(msgType -> SparkParquetWriters.buildWriter(dataSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(equalityDeleteSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType));
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- private StructType dataSparkType() {
- if (dataSparkType == null) {
- Preconditions.checkNotNull(dataSchema(), "Data schema must not be null");
- this.dataSparkType = SparkSchemaUtil.convert(dataSchema());
- }
-
- return dataSparkType;
- }
-
- private StructType equalityDeleteSparkType() {
- if (equalityDeleteSparkType == null) {
- Preconditions.checkNotNull(
- equalityDeleteRowSchema(), "Equality delete schema must not be null");
- this.equalityDeleteSparkType = SparkSchemaUtil.convert(equalityDeleteRowSchema());
- }
-
- return equalityDeleteSparkType;
- }
-
private StructType positionDeleteSparkType() {
if (positionDeleteSparkType == null) {
// wrap the optional row schema into the position delete schema containing path and position
- Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema());
+ Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema);
this.positionDeleteSparkType = SparkSchemaUtil.convert(positionDeleteSchema);
}
return positionDeleteSparkType;
}
+ @Override
+ public PositionDeleteWriter<InternalRow> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (!useDeprecatedPositionDeleteWriter) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.12.0.");
+ Map<String, String> properties = table == null ? ImmutableMap.of() : table.properties();
+ MetricsConfig metricsConfig =
+ table == null
+ ? MetricsConfig.forPositionDelete()
+ : MetricsConfig.forPositionDelete(table);
+
+ try {
+ return switch (format) {
+ case AVRO ->
+ Avro.writeDeletes(file)
+ .createWriterFunc(
+ ignored ->
+ new SparkAvroWriter(
+ (StructType)
+ positionDeleteSparkType()
+ .apply(DELETE_FILE_ROW_FIELD_NAME)
+ .dataType()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .rowSchema(positionDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+ case ORC ->
+ ORC.writeDeletes(file)
+ .createWriterFunc(SparkOrcWriter::new)
+ .transformPaths(path -> UTF8String.fromString(path.toString()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .rowSchema(positionDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+ case PARQUET ->
+ Parquet.writeDeletes(file)
+ .createWriterFunc(
+ msgType ->
+ SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType))
+ .transformPaths(path -> UTF8String.fromString(path.toString()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .metricsConfig(metricsConfig)
Review Comment:
```suggestion
.metricsConfig(metricsConfig)
.withPartition(partition)
.overwrite()
```
called twice
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -75,18 +88,26 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- positionDeleteRowSchema);
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = positionDeleteSparkType;
+ this.table = table;
+ this.format = dataFileFormat;
Review Comment:
```suggestion
this.format = deleteFileFormat;
```
`this.format` is used in `newPositionDeleteWriter`, so i think it should be the delete file format
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#issuecomment-3921375671
Merged to main.
Thanks @huaxingao, @kevinjqliu, @rdblue for the reviews!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#issuecomment-3910880300
> can we run the benchmarks for spark to see how the benchmarks turns out to be post these : https://github.com/apache/iceberg/tree/main/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark ?
Added some new tests for Parquet (`readUsingRegistryReader`, `readWithProjectionUsingRegistryReader`, `readUsingRegistryReader`, `readWithProjectionUsingRegistryReader`, `writeUsingRegistryWriter`, `writeUsingRegistryWriter`):
```
Benchmark Mode Cnt Score Error Units
SparkParquetReadersFlatDataBenchmark.readUsingIcebergReader ss 5 0.311 ± 0.005 s/op
SparkParquetReadersFlatDataBenchmark.readUsingIcebergReaderUnsafe ss 5 0.396 ± 0.018 s/op
SparkParquetReadersFlatDataBenchmark.readUsingRegistryReader ss 5 0.326 ± 0.049 s/op
SparkParquetReadersFlatDataBenchmark.readUsingSparkReader ss 5 0.408 ± 0.008 s/op
SparkParquetReadersFlatDataBenchmark.readWithProjectionUsingIcebergReader ss 5 0.185 ± 0.018 s/op
SparkParquetReadersFlatDataBenchmark.readWithProjectionUsingIcebergReaderUnsafe ss 5 0.363 ± 0.018 s/op
SparkParquetReadersFlatDataBenchmark.readWithProjectionUsingRegistryReader ss 5 0.213 ± 0.026 s/op
SparkParquetReadersFlatDataBenchmark.readWithProjectionUsingSparkReader ss 5 0.273 ± 0.019 s/op
SparkParquetReadersNestedDataBenchmark.readUsingIcebergReader ss 5 0.184 ± 0.018 s/op
SparkParquetReadersNestedDataBenchmark.readUsingIcebergReaderUnsafe ss 5 0.219 ± 0.026 s/op
SparkParquetReadersNestedDataBenchmark.readUsingRegistryReader ss 5 0.179 ± 0.035 s/op
SparkParquetReadersNestedDataBenchmark.readUsingSparkReader ss 5 0.223 ± 0.015 s/op
SparkParquetReadersNestedDataBenchmark.readWithProjectionUsingIcebergReader ss 5 0.077 ± 0.010 s/op
SparkParquetReadersNestedDataBenchmark.readWithProjectionUsingIcebergReaderUnsafe ss 5 0.137 ± 0.007 s/op
SparkParquetReadersNestedDataBenchmark.readWithProjectionUsingRegistryReader ss 5 0.080 ± 0.006 s/op
SparkParquetReadersNestedDataBenchmark.readWithProjectionUsingSparkReader ss 5 0.103 ± 0.003 s/op
SparkParquetWritersFlatDataBenchmark.writeUsingIcebergWriter ss 5 2.602 ± 0.064 s/op
SparkParquetWritersFlatDataBenchmark.writeUsingRegistryWriter ss 5 2.593 ± 0.074 s/op
SparkParquetWritersFlatDataBenchmark.writeUsingSparkWriter ss 5 2.594 ± 0.054 s/op
SparkParquetWritersNestedDataBenchmark.writeUsingIcebergWriter ss 5 1.559 ± 0.022 s/op
SparkParquetWritersNestedDataBenchmark.writeUsingRegistryWriter ss 5 1.569 ± 0.043 s/op
SparkParquetWritersNestedDataBenchmark.writeUsingSparkWriter ss 5 1.595 ± 0.046 s/op
```
The differences are barely noticeable in any direction. There should not be any real difference as the resulting readers and writers are using the same code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2822928500
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,110 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
this.writeProperties = writeProperties != null ? writeProperties : ImmutableMap.of();
+ this.positionDeleteRowSchema = null;
+ this.useDeprecatedPositionDeleteWriter = false;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(dataSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(equalityDeleteSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- boolean withRow =
- positionDeleteSparkType().getFieldIndex(DELETE_FILE_ROW_FIELD_NAME).isDefined();
- if (withRow) {
- // SparkAvroWriter accepts just the Spark type of the row ignoring the path and pos
- StructField rowField = positionDeleteSparkType().apply(DELETE_FILE_ROW_FIELD_NAME);
- StructType positionDeleteRowSparkType = (StructType) rowField.dataType();
- builder.createWriterFunc(ignored -> new SparkAvroWriter(positionDeleteRowSparkType));
- }
-
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(msgType -> SparkParquetWriters.buildWriter(dataSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(equalityDeleteSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType));
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- private StructType dataSparkType() {
- if (dataSparkType == null) {
- Preconditions.checkNotNull(dataSchema(), "Data schema must not be null");
- this.dataSparkType = SparkSchemaUtil.convert(dataSchema());
- }
-
- return dataSparkType;
- }
-
- private StructType equalityDeleteSparkType() {
- if (equalityDeleteSparkType == null) {
- Preconditions.checkNotNull(
- equalityDeleteRowSchema(), "Equality delete schema must not be null");
- this.equalityDeleteSparkType = SparkSchemaUtil.convert(equalityDeleteRowSchema());
- }
-
- return equalityDeleteSparkType;
- }
-
private StructType positionDeleteSparkType() {
if (positionDeleteSparkType == null) {
// wrap the optional row schema into the position delete schema containing path and position
- Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema());
+ Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema);
this.positionDeleteSparkType = SparkSchemaUtil.convert(positionDeleteSchema);
}
return positionDeleteSparkType;
}
+ @Override
+ public PositionDeleteWriter<InternalRow> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (!useDeprecatedPositionDeleteWriter) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.warn("Position deletes with deleted rows are deprecated and will be removed in 1.12.0.");
+ Map<String, String> properties = table == null ? ImmutableMap.of() : table.properties();
+ MetricsConfig metricsConfig =
+ table == null
+ ? MetricsConfig.forPositionDelete()
+ : MetricsConfig.forPositionDelete(table);
+
+ try {
+ return switch (format) {
+ case AVRO ->
+ Avro.writeDeletes(file)
+ .createWriterFunc(
+ ignored ->
+ new SparkAvroWriter(
+ (StructType)
+ positionDeleteSparkType()
+ .apply(DELETE_FILE_ROW_FIELD_NAME)
+ .dataType()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .rowSchema(positionDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+ case ORC ->
+ ORC.writeDeletes(file)
+ .createWriterFunc(SparkOrcWriter::new)
+ .transformPaths(path -> UTF8String.fromString(path.toString()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .rowSchema(positionDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+ case PARQUET ->
+ Parquet.writeDeletes(file)
+ .createWriterFunc(
+ msgType ->
+ SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType))
+ .transformPaths(path -> UTF8String.fromString(path.toString()))
+ .setAll(properties)
+ .setAll(writeProperties)
+ .metricsConfig(metricsConfig)
+ .withPartition(partition)
+ .overwrite()
+ .metricsConfig(metricsConfig)
Review Comment:
Sorry, I have seen this too late.
Created another PR: #15356
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819484024
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,105 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
this.writeProperties = writeProperties != null ? writeProperties : ImmutableMap.of();
+ this.positionDeleteRowSchema = null;
+ this.useDeprecatedPositionDeleteWriter = false;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(dataSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(equalityDeleteSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- boolean withRow =
- positionDeleteSparkType().getFieldIndex(DELETE_FILE_ROW_FIELD_NAME).isDefined();
- if (withRow) {
- // SparkAvroWriter accepts just the Spark type of the row ignoring the path and pos
- StructField rowField = positionDeleteSparkType().apply(DELETE_FILE_ROW_FIELD_NAME);
- StructType positionDeleteRowSparkType = (StructType) rowField.dataType();
- builder.createWriterFunc(ignored -> new SparkAvroWriter(positionDeleteRowSparkType));
- }
-
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(msgType -> SparkParquetWriters.buildWriter(dataSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(equalityDeleteSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType));
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- private StructType dataSparkType() {
- if (dataSparkType == null) {
- Preconditions.checkNotNull(dataSchema(), "Data schema must not be null");
- this.dataSparkType = SparkSchemaUtil.convert(dataSchema());
- }
-
- return dataSparkType;
- }
-
- private StructType equalityDeleteSparkType() {
- if (equalityDeleteSparkType == null) {
- Preconditions.checkNotNull(
- equalityDeleteRowSchema(), "Equality delete schema must not be null");
- this.equalityDeleteSparkType = SparkSchemaUtil.convert(equalityDeleteRowSchema());
- }
-
- return equalityDeleteSparkType;
- }
-
private StructType positionDeleteSparkType() {
if (positionDeleteSparkType == null) {
// wrap the optional row schema into the position delete schema containing path and position
- Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema());
+ Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema);
this.positionDeleteSparkType = SparkSchemaUtil.convert(positionDeleteSchema);
}
return positionDeleteSparkType;
}
+ @Override
+ public PositionDeleteWriter<InternalRow> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (!useDeprecatedPositionDeleteWriter) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.info(
+ "Deprecated feature used. Position delete row schema is used to create the position delete writer.");
Review Comment:
The API is already marked. This method is fine as long as we don't have a row schema for position deletes.
I'd update the message here to be more direct, rather than being two sentences: "Position deletes with deleted rows are deprecated and will be removed in 2.0.0". Is that the right version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819443502
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
+ try (DataWriter<InternalRow> writer =
+ FormatModelRegistry.dataWriteBuilder(
+ FileFormat.PARQUET,
+ InternalRow.class,
+ EncryptedFiles.plainAsEncryptedOutput(Files.localOutput(dataFile)))
+ .schema(SCHEMA)
+ .engineSchema(SparkSchemaUtil.convert(SCHEMA))
Review Comment:
What's the downside to fixing this now? It seems like this is going to lead to adding code only to risk not removing it in the future. Since we have an easy way to default the write schema, it seems like that's what we should do.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2822998084
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -75,18 +88,26 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- positionDeleteRowSchema);
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = positionDeleteSparkType;
+ this.table = table;
+ this.format = dataFileFormat;
Review Comment:
Added the fix to #15356
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819434495
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersFlatDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
Review Comment:
I'm okay with this since it isn't production code. It's unlikely that this is going to cause problems cherry-picking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819459902
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -681,6 +680,13 @@ public CloseableIterable<Record> reader(
return positionDeletesReader(inputFile, format, spec);
}
+ @Override
+ public PositionDeleteWriter<Record> writer(
Review Comment:
Why was this added?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2816677351
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersFlatDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
Review Comment:
These tests were failing with Spark 4.1, but probably doesn't worth to create a new PR for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2821214538
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
+ try (DataWriter<InternalRow> writer =
+ FormatModelRegistry.dataWriteBuilder(
+ FileFormat.PARQUET,
+ InternalRow.class,
+ EncryptedFiles.plainAsEncryptedOutput(Files.localOutput(dataFile)))
+ .schema(SCHEMA)
+ .engineSchema(SparkSchemaUtil.convert(SCHEMA))
Review Comment:
Added
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersFlatDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
Review Comment:
Updated
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2823000927
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,110 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
Review Comment:
Added the fix to #15356
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819437281
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersFlatDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
Review Comment:
Same here as in the read benchmarks. I'd prefer to replace the existing benchmark implementation once we validate that the new way of calling the same writer implementation has no perf regression. I think of them as the same write implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #15328:
URL: https://github.com/apache/iceberg/pull/15328
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#issuecomment-3921925768
Note that https://github.com/apache/iceberg/pull/15356 contains a follow up fix to address the comments above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2812696652
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,105 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
this.writeProperties = writeProperties != null ? writeProperties : ImmutableMap.of();
+ this.positionDeleteRowSchema = null;
+ this.useDeprecatedPositionDeleteWriter = false;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(dataSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(equalityDeleteSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- boolean withRow =
- positionDeleteSparkType().getFieldIndex(DELETE_FILE_ROW_FIELD_NAME).isDefined();
- if (withRow) {
- // SparkAvroWriter accepts just the Spark type of the row ignoring the path and pos
- StructField rowField = positionDeleteSparkType().apply(DELETE_FILE_ROW_FIELD_NAME);
- StructType positionDeleteRowSparkType = (StructType) rowField.dataType();
- builder.createWriterFunc(ignored -> new SparkAvroWriter(positionDeleteRowSparkType));
- }
-
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(msgType -> SparkParquetWriters.buildWriter(dataSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(equalityDeleteSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType));
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- private StructType dataSparkType() {
- if (dataSparkType == null) {
- Preconditions.checkNotNull(dataSchema(), "Data schema must not be null");
- this.dataSparkType = SparkSchemaUtil.convert(dataSchema());
- }
-
- return dataSparkType;
- }
-
- private StructType equalityDeleteSparkType() {
- if (equalityDeleteSparkType == null) {
- Preconditions.checkNotNull(
- equalityDeleteRowSchema(), "Equality delete schema must not be null");
- this.equalityDeleteSparkType = SparkSchemaUtil.convert(equalityDeleteRowSchema());
- }
-
- return equalityDeleteSparkType;
- }
-
private StructType positionDeleteSparkType() {
if (positionDeleteSparkType == null) {
// wrap the optional row schema into the position delete schema containing path and position
- Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema());
+ Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema);
this.positionDeleteSparkType = SparkSchemaUtil.convert(positionDeleteSchema);
}
return positionDeleteSparkType;
}
+ @Override
+ public PositionDeleteWriter<InternalRow> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (!useDeprecatedPositionDeleteWriter) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.info(
+ "Deprecated feature used. Position delete row schema is used to create the position delete writer.");
Review Comment:
nit: should we mark this as `@deprecated`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2818589609
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
+ try (DataWriter<InternalRow> writer =
+ FormatModelRegistry.dataWriteBuilder(
+ FileFormat.PARQUET,
+ InternalRow.class,
+ EncryptedFiles.plainAsEncryptedOutput(Files.localOutput(dataFile)))
+ .schema(SCHEMA)
+ .engineSchema(SparkSchemaUtil.convert(SCHEMA))
Review Comment:
Sounds like a good idea: having a helper that auto-fills engine schema when it’s not provided would make call sites simpler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2818600430
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
Review Comment:
Should we file an issue to track the underlying Spark 4.1 test failure, so we can fix the root cause later?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2818752121
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
.schema(SCHEMA)
.build()) {
writer.addAll(rows);
}
}
+
+ @Benchmark
+ @Threads(1)
+ public void writeUsingRegistryWriter() throws IOException {
+ try (DataWriter<InternalRow> writer =
+ FormatModelRegistry.dataWriteBuilder(
+ FileFormat.PARQUET,
+ InternalRow.class,
+ EncryptedFiles.plainAsEncryptedOutput(Files.localOutput(dataFile)))
+ .schema(SCHEMA)
+ .engineSchema(SparkSchemaUtil.convert(SCHEMA))
Review Comment:
Let’s consider this in a follow-up PR. Approving to unblock the current change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2816678137
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
Review Comment:
These tests were failing with Spark 4.1, but probably doesn't worth to create a new PR for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#issuecomment-3909401711
can we run the benchmarks for spark to see how the benchmarks turns out to be : https://github.com/apache/iceberg/tree/main/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2819432126
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetReadersFlatDataBenchmark.java:
##########
@@ -167,6 +169,21 @@ public void readUsingSparkReader(Blackhole blackhole) throws IOException {
}
}
+ @Benchmark
+ @Threads(1)
+ public void readUsingRegistryReader(Blackhole blackHole) throws IOException {
Review Comment:
Do we need to test the direct method vs the registry method? I would expect this to replace the current `readUsingIcebergReaderUnsafe` implementation since this is the same reader implementation. We should make sure that there is not a regression by running these benchmarks (for which it would be fine to leave this method here) but I don't want to accumulate essentially dead code testing the same thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2818626266
##########
spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/data/parquet/SparkParquetWritersNestedDataBenchmark.java:
##########
@@ -121,10 +126,28 @@ public void writeUsingSparkWriter() throws IOException {
.set("spark.sql.parquet.outputTimestampType", "TIMESTAMP_MICROS")
.set("spark.sql.caseSensitive", "false")
.set("spark.sql.parquet.fieldId.write.enabled", "false")
+ .set("spark.sql.parquet.variant.annotateLogicalType.enabled", "false")
Review Comment:
I think both "true" and "false" is ok as well. The issue was that the config was not set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core, Data, Spark: Moving Spark to use the new FormatModel API [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #15328:
URL: https://github.com/apache/iceberg/pull/15328#discussion_r2821193540
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkFileWriterFactory.java:
##########
@@ -105,119 +126,105 @@ class SparkFileWriterFactory extends BaseFileWriterFactory<InternalRow> {
super(
table,
dataFileFormat,
+ InternalRow.class,
dataSchema,
dataSortOrder,
deleteFileFormat,
equalityFieldIds,
equalityDeleteRowSchema,
equalityDeleteSortOrder,
- ImmutableMap.of());
+ writeProperties,
+ useOrConvert(dataSparkType, dataSchema),
+ useOrConvert(equalityDeleteSparkType, equalityDeleteRowSchema));
- this.dataSparkType = dataSparkType;
- this.equalityDeleteSparkType = equalityDeleteSparkType;
- this.positionDeleteSparkType = null;
+ this.table = table;
+ this.format = dataFileFormat;
this.writeProperties = writeProperties != null ? writeProperties : ImmutableMap.of();
+ this.positionDeleteRowSchema = null;
+ this.useDeprecatedPositionDeleteWriter = false;
}
static Builder builderFor(Table table) {
return new Builder(table);
}
- @Override
- protected void configureDataWrite(Avro.DataWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(dataSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Avro.DeleteWriteBuilder builder) {
- builder.createWriterFunc(ignored -> new SparkAvroWriter(equalityDeleteSparkType()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Avro.DeleteWriteBuilder builder) {
- boolean withRow =
- positionDeleteSparkType().getFieldIndex(DELETE_FILE_ROW_FIELD_NAME).isDefined();
- if (withRow) {
- // SparkAvroWriter accepts just the Spark type of the row ignoring the path and pos
- StructField rowField = positionDeleteSparkType().apply(DELETE_FILE_ROW_FIELD_NAME);
- StructType positionDeleteRowSparkType = (StructType) rowField.dataType();
- builder.createWriterFunc(ignored -> new SparkAvroWriter(positionDeleteRowSparkType));
- }
-
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(Parquet.DataWriteBuilder builder) {
- builder.createWriterFunc(msgType -> SparkParquetWriters.buildWriter(dataSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(equalityDeleteSparkType(), msgType));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(Parquet.DeleteWriteBuilder builder) {
- builder.createWriterFunc(
- msgType -> SparkParquetWriters.buildWriter(positionDeleteSparkType(), msgType));
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureDataWrite(ORC.DataWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configureEqualityDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.setAll(writeProperties);
- }
-
- @Override
- protected void configurePositionDelete(ORC.DeleteWriteBuilder builder) {
- builder.createWriterFunc(SparkOrcWriter::new);
- builder.transformPaths(path -> UTF8String.fromString(path.toString()));
- builder.setAll(writeProperties);
- }
-
- private StructType dataSparkType() {
- if (dataSparkType == null) {
- Preconditions.checkNotNull(dataSchema(), "Data schema must not be null");
- this.dataSparkType = SparkSchemaUtil.convert(dataSchema());
- }
-
- return dataSparkType;
- }
-
- private StructType equalityDeleteSparkType() {
- if (equalityDeleteSparkType == null) {
- Preconditions.checkNotNull(
- equalityDeleteRowSchema(), "Equality delete schema must not be null");
- this.equalityDeleteSparkType = SparkSchemaUtil.convert(equalityDeleteRowSchema());
- }
-
- return equalityDeleteSparkType;
- }
-
private StructType positionDeleteSparkType() {
if (positionDeleteSparkType == null) {
// wrap the optional row schema into the position delete schema containing path and position
- Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema());
+ Schema positionDeleteSchema = DeleteSchemaUtil.posDeleteSchema(positionDeleteRowSchema);
this.positionDeleteSparkType = SparkSchemaUtil.convert(positionDeleteSchema);
}
return positionDeleteSparkType;
}
+ @Override
+ public PositionDeleteWriter<InternalRow> newPositionDeleteWriter(
+ EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
+ if (!useDeprecatedPositionDeleteWriter) {
+ return super.newPositionDeleteWriter(file, spec, partition);
+ } else {
+ LOG.info(
+ "Deprecated feature used. Position delete row schema is used to create the position delete writer.");
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org