You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/11/26 15:16:49 UTC
[GitHub] [iceberg] openinx opened a new pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
openinx opened a new pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836
This is an PR which is divided from this : https://github.com/apache/iceberg/pull/1818.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532935047
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -42,10 +47,25 @@
public class GenericAppenderFactory implements FileAppenderFactory<Record> {
private final Schema schema;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
private final Map<String, String> config = Maps.newHashMap();
- public GenericAppenderFactory(Schema schema) {
+ public GenericAppenderFactory(Schema schema, PartitionSpec spec) {
+ this(schema, spec, null, schema, null);
Review comment:
Why not set `eqDeleteRowSchema` to null since `equalityFieldIds` is null?
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532935529
##########
File path: data/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
##########
@@ -73,13 +74,12 @@ public DataFile writeFile(StructLike partition, List<Record> records) throws IOE
Preconditions.checkNotNull(table, "table not set");
File file = tmp.newFile();
Assert.assertTrue(file.delete());
- return appendToLocalFile(table, file, fileFormat, partition, records);
+ return appendToLocalFile(table, file, fileFormat, partition, records, table.spec());
}
- private static DataFile appendToLocalFile(
- Table table, File file, FileFormat format, StructLike partition, List<Record> records)
- throws IOException {
- FileAppender<Record> appender = new GenericAppenderFactory(table.schema()).newAppender(
+ private static DataFile appendToLocalFile(Table table, File file, FileFormat format, StructLike partition,
+ List<Record> records, PartitionSpec spec) throws IOException {
+ FileAppender<Record> appender = new GenericAppenderFactory(table.schema(), spec).newAppender(
Review comment:
Could we keep the constructor this was using before so we don't need to change any tests that only use `newAppender`?
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533841383
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,37 +384,31 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter()
+ throws IOException {
Preconditions.checkState(equalityFieldIds == null, "Cannot create position delete file using delete field ids");
meta("delete-type", "position");
if (rowSchema != null && createWriterFunc != null) {
// the appender uses the row schema wrapped with position fields
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS,
- NestedField.optional(
- MetadataColumns.DELETE_FILE_ROW_FIELD_ID, "row", rowSchema.asStruct(),
- MetadataColumns.DELETE_FILE_ROW_DOC)));
+ appenderBuilder.schema(DeleteSchemaUtil.posDeleteSchema(rowSchema));
appenderBuilder.createWriterFunc(parquetSchema -> {
ParquetValueWriter<?> writer = createWriterFunc.apply(parquetSchema);
if (writer instanceof StructWriter) {
- return new PositionDeleteStructWriter<T>((StructWriter<?>) writer);
+ return new PositionDeleteStructWriter<T>((StructWriter<?>) writer, pathTransformFunc);
} else {
throw new UnsupportedOperationException("Cannot wrap writer for position deletes: " + writer.getClass());
}
});
} else {
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS));
+ appenderBuilder.schema(DeleteSchemaUtil.pathPosSchema());
appenderBuilder.createWriterFunc(parquetSchema ->
- new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema)));
+ new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema),
+ t -> t));
Review comment:
It's good to use `Function.identity()` here.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533731795
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +115,88 @@ public GenericAppenderFactory setAll(Map<String, String> properties) {
throw new UncheckedIOException(e);
}
}
+
+ @Override
+ public org.apache.iceberg.io.DataWriter<Record> newDataWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ return new org.apache.iceberg.io.DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
Review comment:
Sounds good, thanks for explaining it to me.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533104998
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +115,88 @@ public GenericAppenderFactory setAll(Map<String, String> properties) {
throw new UncheckedIOException(e);
}
}
+
+ @Override
+ public org.apache.iceberg.io.DataWriter<Record> newDataWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ return new org.apache.iceberg.io.DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
Review comment:
I did not move it to be an default implementation for `newDataWriter` because the `DataWriter` construction will need the the partition `spec`. It's not friendly to expose an extra method `spec()` in `FileAppenderFactory` ( so that those subclass could pass the correct `spec` object to `DataWriter`).
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533841204
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,37 +384,31 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter()
+ throws IOException {
Preconditions.checkState(equalityFieldIds == null, "Cannot create position delete file using delete field ids");
meta("delete-type", "position");
if (rowSchema != null && createWriterFunc != null) {
// the appender uses the row schema wrapped with position fields
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS,
- NestedField.optional(
- MetadataColumns.DELETE_FILE_ROW_FIELD_ID, "row", rowSchema.asStruct(),
- MetadataColumns.DELETE_FILE_ROW_DOC)));
+ appenderBuilder.schema(DeleteSchemaUtil.posDeleteSchema(rowSchema));
appenderBuilder.createWriterFunc(parquetSchema -> {
ParquetValueWriter<?> writer = createWriterFunc.apply(parquetSchema);
if (writer instanceof StructWriter) {
- return new PositionDeleteStructWriter<T>((StructWriter<?>) writer);
+ return new PositionDeleteStructWriter<T>((StructWriter<?>) writer, pathTransformFunc);
} else {
throw new UnsupportedOperationException("Cannot wrap writer for position deletes: " + writer.getClass());
}
});
} else {
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS));
+ appenderBuilder.schema(DeleteSchemaUtil.pathPosSchema());
appenderBuilder.createWriterFunc(parquetSchema ->
- new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema)));
+ new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema),
+ t -> t));
Review comment:
We shouldn't pass `pathTransformFunc` here, because in this path we will use `GenericParquetWriter` (Rather than `FlinkParquetWriter` or `SparkParquetWriter`) to write the `PositionDelete`, if convert the path `CharSequence` to `StringData`, the `GenericParquetWriter` could not find the correct writer to write values.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#issuecomment-736977560
All check passed, let me merge this PR, thanks @rdblue for reviewing.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532153703
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
In my mind, there are two ways to solve the problem, the first one is the current version: Just use the completed scheme which has included the `file_path` field and `pos` field to build the flink/spark struct writer, tradeoff
is exposing few details to upper, such as we need to convert the `CharSequence` to flink's `StringData` before we write the `file_path` value (see the `ParquetValueWriters.PathPosAccessor`). The current version is easy to implement.
Another way is: call the `row` schema's `createWriterFunc` in parquet visitor when building the whole struct writer by traversing the columns. In that visitor, we could add the `row` prefix to the fieldNames and finally the ColumnDescriptor won't be messed up.
That seems to be more complex when compared to the first version. The good thing is hiding all the constructing details inside the builder.
IMO, I'd prefer to use the current version because the `FlinkAppenderFactory` and `SparkAppenderFactory` has already hidden the details for upper layer. People won't likely to write their own `AppenderFactory`, even if they want to write their own it will be easy when following the current flink/spark appender factory.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533100045
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -42,10 +47,25 @@
public class GenericAppenderFactory implements FileAppenderFactory<Record> {
private final Schema schema;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
private final Map<String, String> config = Maps.newHashMap();
- public GenericAppenderFactory(Schema schema) {
+ public GenericAppenderFactory(Schema schema, PartitionSpec spec) {
+ this(schema, spec, null, schema, null);
Review comment:
OK, it's more reasonable to set the `eqDeleteRowSchema` to be null when `equalityFieldIds` is null.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533738345
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,37 +384,31 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter()
+ throws IOException {
Preconditions.checkState(equalityFieldIds == null, "Cannot create position delete file using delete field ids");
meta("delete-type", "position");
if (rowSchema != null && createWriterFunc != null) {
// the appender uses the row schema wrapped with position fields
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS,
- NestedField.optional(
- MetadataColumns.DELETE_FILE_ROW_FIELD_ID, "row", rowSchema.asStruct(),
- MetadataColumns.DELETE_FILE_ROW_DOC)));
+ appenderBuilder.schema(DeleteSchemaUtil.posDeleteSchema(rowSchema));
appenderBuilder.createWriterFunc(parquetSchema -> {
ParquetValueWriter<?> writer = createWriterFunc.apply(parquetSchema);
if (writer instanceof StructWriter) {
- return new PositionDeleteStructWriter<T>((StructWriter<?>) writer);
+ return new PositionDeleteStructWriter<T>((StructWriter<?>) writer, pathTransformFunc);
} else {
throw new UnsupportedOperationException("Cannot wrap writer for position deletes: " + writer.getClass());
}
});
} else {
- appenderBuilder.schema(new org.apache.iceberg.Schema(
- MetadataColumns.DELETE_FILE_PATH,
- MetadataColumns.DELETE_FILE_POS));
+ appenderBuilder.schema(DeleteSchemaUtil.pathPosSchema());
appenderBuilder.createWriterFunc(parquetSchema ->
- new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema)));
+ new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema),
+ t -> t));
Review comment:
Shouldn't this pass `pathTransformFunc` as well?
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532953568
##########
File path: core/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableTestBase;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.StructLikeSet;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public abstract class TestAppenderFactory<T> extends TableTestBase {
+ private static final int FORMAT_V2 = 2;
+
+ private final FileFormat format;
+ private final boolean partitioned;
+
+ private PartitionKey partition = null;
+
+ @Parameterized.Parameters(name = "FileFormat={0}, Partitioned={1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ new Object[] {"avro", false},
+ new Object[] {"avro", true},
+ new Object[] {"parquet", false},
+ new Object[] {"parquet", true}
+ };
+ }
+
+
+ public TestAppenderFactory(String fileFormat, boolean partitioned) {
+ super(FORMAT_V2);
+ this.format = FileFormat.valueOf(fileFormat.toUpperCase(Locale.ENGLISH));
+ this.partitioned = partitioned;
+ }
+
+ @Before
+ public void setupTable() throws Exception {
+ this.tableDir = temp.newFolder();
+ Assert.assertTrue(tableDir.delete()); // created by table create
+
+ this.metadataDir = new File(tableDir, "metadata");
+
+ if (partitioned) {
+ this.table = create(SCHEMA, SPEC);
+ } else {
+ this.table = create(SCHEMA, PartitionSpec.unpartitioned());
+ }
+ this.partition = createPartitionKey();
+
+ table.updateProperties()
+ .defaultFormat(format)
+ .commit();
+ }
+
+ protected abstract FileAppenderFactory<T> createAppenderFactory(List<Integer> equalityFieldIds,
+ Schema eqDeleteSchema,
+ Schema posDeleteRowSchema);
+
+ protected abstract T createRow(Integer id, String data);
+
+ protected abstract StructLikeSet expectedRowSet(Iterable<T> records) throws IOException;
+
+ protected abstract StructLikeSet actualRowSet(String... columns) throws IOException;
+
+ private OutputFileFactory createFileFactory() {
+ return new OutputFileFactory(table.spec(), format, table.locationProvider(), table.io(),
+ table.encryption(), 1, 1);
+ }
+
+ private PartitionKey createPartitionKey() {
+ if (table.spec().isUnpartitioned()) {
+ return null;
+ }
+
+ Record record = GenericRecord.create(table.spec().schema()).copy(ImmutableMap.of("data", "aaa"));
+
+ PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema());
+ partitionKey.partition(record);
+
+ return partitionKey;
+ }
+
+ private List<T> testRowSet() {
+ return Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(2, "bbb"),
+ createRow(3, "ccc"),
+ createRow(4, "ddd"),
+ createRow(5, "eee")
+ );
+ }
+
+ private DataFile prepareDataFile(List<T> rowSet, FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory outputFileFactory) throws IOException {
+ DataWriter<T> writer = appenderFactory.newDataWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (DataWriter<T> closeableWriter = writer) {
+ for (T row : rowSet) {
+ closeableWriter.add(row);
+ }
+ }
+
+ return writer.toDataFile();
+ }
+
+ @Test
+ public void testDataWriter() throws IOException {
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(null, null, null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ Assert.assertEquals("Should have the expected records.", expectedRowSet(rowSet), actualRowSet("*"));
+ }
+
+ @Test
+ public void testEqDeleteWriter() throws IOException {
+ List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().findField("id").fieldId());
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(equalityFieldIds,
+ table.schema().select("id"), null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ List<T> deletes = Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(3, "bbb"),
+ createRow(5, "ccc")
+ );
+ EqualityDeleteWriter<T> eqDeleteWriter =
+ appenderFactory.newEqDeleteWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (EqualityDeleteWriter<T> closeableWriter = eqDeleteWriter) {
+ closeableWriter.deleteAll(deletes);
+ }
+
+ table.newRowDelta()
+ .addDeletes(eqDeleteWriter.toDeleteFile())
+ .commit();
+
+ List<T> expected = Lists.newArrayList(
+ createRow(2, "bbb"),
+ createRow(4, "ddd")
+ );
+ Assert.assertEquals("Should have the expected records", expectedRowSet(expected), actualRowSet("*"));
+ }
+
+ @Test
+ public void testPosDeleteWriter() throws IOException {
+ // Initialize FileAppenderFactory without pos-delete row schema.
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(null, null, null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(
+ Pair.of(dataFile.path(), 0L),
+ Pair.of(dataFile.path(), 2L),
+ Pair.of(dataFile.path(), 4L)
+ );
+
+ PositionDeleteWriter<T> eqDeleteWriter =
+ appenderFactory.newPosDeleteWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (PositionDeleteWriter<T> closeableWriter = eqDeleteWriter) {
+ for (Pair<CharSequence, Long> delete : deletes) {
+ closeableWriter.delete(delete.first(), delete.second());
+ }
+ }
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .addDeletes(eqDeleteWriter.toDeleteFile())
+ .validateDataFilesExist(eqDeleteWriter.referencedDataFiles())
+ .validateDeletedFiles()
+ .commit();
+
+ List<T> expected = Lists.newArrayList(
+ createRow(2, "bbb"),
+ createRow(4, "ddd")
+ );
+ Assert.assertEquals("Should have the expected records", expectedRowSet(expected), actualRowSet("*"));
Review comment:
Similar to above, I think this should check that only the path and position columns are written to the file and that they are the expected values. The test below should check that the `row` column is present and set correctly for each row.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx merged pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx merged pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532930957
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +115,88 @@ public GenericAppenderFactory setAll(Map<String, String> properties) {
throw new UncheckedIOException(e);
}
}
+
+ @Override
+ public org.apache.iceberg.io.DataWriter<Record> newDataWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ return new org.apache.iceberg.io.DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
Review comment:
Could this be the default implementation of `newDataWriter`? It looks the same across all of the implementations.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533139850
##########
File path: spark/src/main/java/org/apache/iceberg/spark/source/RowDataRewriter.java
##########
@@ -98,7 +98,7 @@ public RowDataRewriter(Table table, PartitionSpec spec, boolean caseSensitive,
task, schema, schema, nameMapping, io.value(), encryptionManager.value(), caseSensitive);
StructType structType = SparkSchemaUtil.convert(schema);
- SparkAppenderFactory appenderFactory = new SparkAppenderFactory(properties, schema, structType);
+ SparkAppenderFactory appenderFactory = new SparkAppenderFactory(properties, schema, structType, spec);
Review comment:
Here I use the constructor that has `spec` argument because I believe we will use the `DataWriter` to append records once we switch to the `RollingFileWriter`. https://github.com/apache/iceberg/pull/1818/files#diff-fc9a9fd84d24c607fd85e053b08a559f56dd2dd2a46f1341c528e7a0269f873cR263.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532151892
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
Let me provide more details, assume that we want to build a pos delete writer with row schema <id long, data string>.
The straightforward way is to build a struct writer like this:
```java
PositionDeleteStructWriter
|
|----------> StringWriter (file_path field)
|----------> LongWriter (pos field)
|
|
|----------> StructWriter (row field)
|----------> LongWriter (id field)
|----------> StringWriter (data field)
```
The question is how to build the PositionDeleteStructWriter, if we just provide a function to create the `row` field's `StructWriter` for flink, and then concat the <StringWriter, LongWriter, StructWriter> to PositionDeleteStructWriter constructor,
then the problem is: the leaf node `ColumnWriter` will have an incorrect `ColumnDescriptor`:
```java
PositionDeleteStructWriter
|
|----------> StringWriter {<file_path>,type,maxRep,maxDef}
|----------> LongWriter {<pos>,type,maxRep,maxDef}
|
|
|----------> StructWriter
|----------> LongWriter {<id>, type, maxRep, maxDef}
|----------> StringWriter {<data>,type, maxRep, maxDef}
```
In fact, for this example, the correct ColumnWriter should be:
```java
PositionDeleteStructWriter
|
|----------> StringWriter {<file_path>,type,maxRep,maxDef}
|----------> LongWriter {<pos>,type,maxRep,maxDef}
|
|
|----------> StructWriter
|----------> LongWriter {<row,id>, type, maxRep, maxDef}
|----------> StringWriter {<row,data>,type, maxRep, maxDef}
```
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532935529
##########
File path: data/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
##########
@@ -73,13 +74,12 @@ public DataFile writeFile(StructLike partition, List<Record> records) throws IOE
Preconditions.checkNotNull(table, "table not set");
File file = tmp.newFile();
Assert.assertTrue(file.delete());
- return appendToLocalFile(table, file, fileFormat, partition, records);
+ return appendToLocalFile(table, file, fileFormat, partition, records, table.spec());
}
- private static DataFile appendToLocalFile(
- Table table, File file, FileFormat format, StructLike partition, List<Record> records)
- throws IOException {
- FileAppender<Record> appender = new GenericAppenderFactory(table.schema()).newAppender(
+ private static DataFile appendToLocalFile(Table table, File file, FileFormat format, StructLike partition,
+ List<Record> records, PartitionSpec spec) throws IOException {
+ FileAppender<Record> appender = new GenericAppenderFactory(table.schema(), spec).newAppender(
Review comment:
Could we keep the constructor this was using before so we don't need to change any tests that only use `newAppender`? There are 4 files just here that don't appear like they need to change just to add the spec that won't be used.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532940763
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
I'm reviewing it now. I just wanted to comment yesterday to let you know about the Parquet problem.
For the amount of work in Parquet, it shouldn't be too difficult for just this, but we need to support field IDs more generally over there and that is going to be a big project with all of the little things that need to be 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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532931393
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -95,4 +115,88 @@ public GenericAppenderFactory setAll(Map<String, String> properties) {
throw new UncheckedIOException(e);
}
}
+
+ @Override
+ public org.apache.iceberg.io.DataWriter<Record> newDataWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ return new org.apache.iceberg.io.DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<Record> newEqDeleteWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(config);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(file.encryptingOutputFile())
+ .createWriterFunc(DataWriter::create)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(config)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(file.encryptingOutputFile())
+ .createWriterFunc(GenericParquetWriter::buildWriter)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(config)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<Record> newPosDeleteWriter(EncryptedOutputFile file, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(config);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(file.encryptingOutputFile())
+ .createWriterFunc(DataWriter::create)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(config)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(file.encryptingOutputFile())
+ .createWriterFunc(GenericParquetWriter::buildWriter)
+ .withPartition(partition)
+ .overwrite()
+ .setAll(config)
+ .metricsConfig(metricsConfig)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(file.keyMetadata())
+ .buildPositionWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
Review comment:
I think it should be "unsupported" rather than "unknown" because ORC is known, but not supported.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533737203
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -281,6 +280,7 @@ public static DeleteWriteBuilder writeDeletes(OutputFile file) {
private StructLike partition = null;
private EncryptionKeyMetadata keyMetadata = null;
private int[] equalityFieldIds = null;
+ private Function<CharSequence, ?> pathTransformFunc = t -> t;
Review comment:
Nit: I think it would be better to use `Function.identity()`.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532565894
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
> we get a better Parquet API that can load columns by ID rather than by path.
Sounds like it's the correct direction to solve this issue, will that need much work to accomplish (I'm not quite sure).
BTW, do you have other concern about this patch ?
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532947494
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,42 +378,40 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter(ParquetValueWriters.PathPosAccessor<?, ?> accessor)
Review comment:
I'd rather not add this to the build method. Nothing distinguishes it from other options, so I think we should add transformations as configuration methods, like we do for `equalityFieldIds`.
I'm also thinking that it would be good to have a more light-weight way to add these transforms. Rather than an additional accessor that has two abstract methods, why not just register functions? It could look like this:
```java
Avro.writeDeletes(outFile)
...
.transformPaths(StringData::fromString)
.buildPositionWriter();
```
That way it's easier to use a method reference rather than creating a class. And nothing actually needs to transform pos yet, so we can just leave that out.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532265981
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
Sorry, I also hit the column descriptor problem when I tried to update this. I have it working for Avro, but not for Parquet. I should have commented with the results of my work so you didn't try it as well.
I agree with you that this would be really difficult to solve and it is probably not worth it. We'll just have to go with the existing solution for Parquet until we get a better Parquet API that can load columns by ID rather than by path.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532951221
##########
File path: core/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableTestBase;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.StructLikeSet;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public abstract class TestAppenderFactory<T> extends TableTestBase {
+ private static final int FORMAT_V2 = 2;
+
+ private final FileFormat format;
+ private final boolean partitioned;
+
+ private PartitionKey partition = null;
+
+ @Parameterized.Parameters(name = "FileFormat={0}, Partitioned={1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ new Object[] {"avro", false},
+ new Object[] {"avro", true},
+ new Object[] {"parquet", false},
+ new Object[] {"parquet", true}
+ };
+ }
+
+
+ public TestAppenderFactory(String fileFormat, boolean partitioned) {
+ super(FORMAT_V2);
+ this.format = FileFormat.valueOf(fileFormat.toUpperCase(Locale.ENGLISH));
+ this.partitioned = partitioned;
+ }
+
+ @Before
+ public void setupTable() throws Exception {
+ this.tableDir = temp.newFolder();
+ Assert.assertTrue(tableDir.delete()); // created by table create
+
+ this.metadataDir = new File(tableDir, "metadata");
+
+ if (partitioned) {
+ this.table = create(SCHEMA, SPEC);
+ } else {
+ this.table = create(SCHEMA, PartitionSpec.unpartitioned());
+ }
+ this.partition = createPartitionKey();
+
+ table.updateProperties()
+ .defaultFormat(format)
+ .commit();
+ }
+
+ protected abstract FileAppenderFactory<T> createAppenderFactory(List<Integer> equalityFieldIds,
+ Schema eqDeleteSchema,
+ Schema posDeleteRowSchema);
+
+ protected abstract T createRow(Integer id, String data);
+
+ protected abstract StructLikeSet expectedRowSet(Iterable<T> records) throws IOException;
+
+ protected abstract StructLikeSet actualRowSet(String... columns) throws IOException;
+
+ private OutputFileFactory createFileFactory() {
+ return new OutputFileFactory(table.spec(), format, table.locationProvider(), table.io(),
+ table.encryption(), 1, 1);
+ }
+
+ private PartitionKey createPartitionKey() {
+ if (table.spec().isUnpartitioned()) {
+ return null;
+ }
+
+ Record record = GenericRecord.create(table.spec().schema()).copy(ImmutableMap.of("data", "aaa"));
Review comment:
I think this could just be `table.schema()`. No need to go through the spec to get the table schema.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532953195
##########
File path: core/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableTestBase;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.StructLikeSet;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public abstract class TestAppenderFactory<T> extends TableTestBase {
+ private static final int FORMAT_V2 = 2;
+
+ private final FileFormat format;
+ private final boolean partitioned;
+
+ private PartitionKey partition = null;
+
+ @Parameterized.Parameters(name = "FileFormat={0}, Partitioned={1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ new Object[] {"avro", false},
+ new Object[] {"avro", true},
+ new Object[] {"parquet", false},
+ new Object[] {"parquet", true}
+ };
+ }
+
+
+ public TestAppenderFactory(String fileFormat, boolean partitioned) {
+ super(FORMAT_V2);
+ this.format = FileFormat.valueOf(fileFormat.toUpperCase(Locale.ENGLISH));
+ this.partitioned = partitioned;
+ }
+
+ @Before
+ public void setupTable() throws Exception {
+ this.tableDir = temp.newFolder();
+ Assert.assertTrue(tableDir.delete()); // created by table create
+
+ this.metadataDir = new File(tableDir, "metadata");
+
+ if (partitioned) {
+ this.table = create(SCHEMA, SPEC);
+ } else {
+ this.table = create(SCHEMA, PartitionSpec.unpartitioned());
+ }
+ this.partition = createPartitionKey();
+
+ table.updateProperties()
+ .defaultFormat(format)
+ .commit();
+ }
+
+ protected abstract FileAppenderFactory<T> createAppenderFactory(List<Integer> equalityFieldIds,
+ Schema eqDeleteSchema,
+ Schema posDeleteRowSchema);
+
+ protected abstract T createRow(Integer id, String data);
+
+ protected abstract StructLikeSet expectedRowSet(Iterable<T> records) throws IOException;
+
+ protected abstract StructLikeSet actualRowSet(String... columns) throws IOException;
+
+ private OutputFileFactory createFileFactory() {
+ return new OutputFileFactory(table.spec(), format, table.locationProvider(), table.io(),
+ table.encryption(), 1, 1);
+ }
+
+ private PartitionKey createPartitionKey() {
+ if (table.spec().isUnpartitioned()) {
+ return null;
+ }
+
+ Record record = GenericRecord.create(table.spec().schema()).copy(ImmutableMap.of("data", "aaa"));
+
+ PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema());
+ partitionKey.partition(record);
+
+ return partitionKey;
+ }
+
+ private List<T> testRowSet() {
+ return Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(2, "bbb"),
+ createRow(3, "ccc"),
+ createRow(4, "ddd"),
+ createRow(5, "eee")
+ );
+ }
+
+ private DataFile prepareDataFile(List<T> rowSet, FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory outputFileFactory) throws IOException {
+ DataWriter<T> writer = appenderFactory.newDataWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (DataWriter<T> closeableWriter = writer) {
+ for (T row : rowSet) {
+ closeableWriter.add(row);
+ }
+ }
+
+ return writer.toDataFile();
+ }
+
+ @Test
+ public void testDataWriter() throws IOException {
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(null, null, null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ Assert.assertEquals("Should have the expected records.", expectedRowSet(rowSet), actualRowSet("*"));
+ }
+
+ @Test
+ public void testEqDeleteWriter() throws IOException {
+ List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().findField("id").fieldId());
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(equalityFieldIds,
+ table.schema().select("id"), null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ List<T> deletes = Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(3, "bbb"),
+ createRow(5, "ccc")
+ );
+ EqualityDeleteWriter<T> eqDeleteWriter =
+ appenderFactory.newEqDeleteWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (EqualityDeleteWriter<T> closeableWriter = eqDeleteWriter) {
+ closeableWriter.deleteAll(deletes);
+ }
+
+ table.newRowDelta()
+ .addDeletes(eqDeleteWriter.toDeleteFile())
+ .commit();
+
+ List<T> expected = Lists.newArrayList(
+ createRow(2, "bbb"),
+ createRow(4, "ddd")
+ );
+ Assert.assertEquals("Should have the expected records", expectedRowSet(expected), actualRowSet("*"));
Review comment:
Instead of reading from the table, I would rather see a test that the equality delete file contains the expected row data. In this case, it should not contain the `data` column. I would like to see that checked. And it would be good to add a case where the whole original row is written to the file.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532952011
##########
File path: core/src/test/java/org/apache/iceberg/io/TestAppenderFactory.java
##########
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.io;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableTestBase;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.StructLikeSet;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+@RunWith(Parameterized.class)
+public abstract class TestAppenderFactory<T> extends TableTestBase {
+ private static final int FORMAT_V2 = 2;
+
+ private final FileFormat format;
+ private final boolean partitioned;
+
+ private PartitionKey partition = null;
+
+ @Parameterized.Parameters(name = "FileFormat={0}, Partitioned={1}")
+ public static Object[] parameters() {
+ return new Object[][] {
+ new Object[] {"avro", false},
+ new Object[] {"avro", true},
+ new Object[] {"parquet", false},
+ new Object[] {"parquet", true}
+ };
+ }
+
+
+ public TestAppenderFactory(String fileFormat, boolean partitioned) {
+ super(FORMAT_V2);
+ this.format = FileFormat.valueOf(fileFormat.toUpperCase(Locale.ENGLISH));
+ this.partitioned = partitioned;
+ }
+
+ @Before
+ public void setupTable() throws Exception {
+ this.tableDir = temp.newFolder();
+ Assert.assertTrue(tableDir.delete()); // created by table create
+
+ this.metadataDir = new File(tableDir, "metadata");
+
+ if (partitioned) {
+ this.table = create(SCHEMA, SPEC);
+ } else {
+ this.table = create(SCHEMA, PartitionSpec.unpartitioned());
+ }
+ this.partition = createPartitionKey();
+
+ table.updateProperties()
+ .defaultFormat(format)
+ .commit();
+ }
+
+ protected abstract FileAppenderFactory<T> createAppenderFactory(List<Integer> equalityFieldIds,
+ Schema eqDeleteSchema,
+ Schema posDeleteRowSchema);
+
+ protected abstract T createRow(Integer id, String data);
+
+ protected abstract StructLikeSet expectedRowSet(Iterable<T> records) throws IOException;
+
+ protected abstract StructLikeSet actualRowSet(String... columns) throws IOException;
+
+ private OutputFileFactory createFileFactory() {
+ return new OutputFileFactory(table.spec(), format, table.locationProvider(), table.io(),
+ table.encryption(), 1, 1);
+ }
+
+ private PartitionKey createPartitionKey() {
+ if (table.spec().isUnpartitioned()) {
+ return null;
+ }
+
+ Record record = GenericRecord.create(table.spec().schema()).copy(ImmutableMap.of("data", "aaa"));
+
+ PartitionKey partitionKey = new PartitionKey(table.spec(), table.schema());
+ partitionKey.partition(record);
+
+ return partitionKey;
+ }
+
+ private List<T> testRowSet() {
+ return Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(2, "bbb"),
+ createRow(3, "ccc"),
+ createRow(4, "ddd"),
+ createRow(5, "eee")
+ );
+ }
+
+ private DataFile prepareDataFile(List<T> rowSet, FileAppenderFactory<T> appenderFactory,
+ OutputFileFactory outputFileFactory) throws IOException {
+ DataWriter<T> writer = appenderFactory.newDataWriter(outputFileFactory.newOutputFile(), format, partition);
+ try (DataWriter<T> closeableWriter = writer) {
+ for (T row : rowSet) {
+ closeableWriter.add(row);
+ }
+ }
+
+ return writer.toDataFile();
+ }
+
+ @Test
+ public void testDataWriter() throws IOException {
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(null, null, null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ Assert.assertEquals("Should have the expected records.", expectedRowSet(rowSet), actualRowSet("*"));
+ }
+
+ @Test
+ public void testEqDeleteWriter() throws IOException {
+ List<Integer> equalityFieldIds = Lists.newArrayList(table.schema().findField("id").fieldId());
+ FileAppenderFactory<T> appenderFactory = createAppenderFactory(equalityFieldIds,
+ table.schema().select("id"), null);
+ OutputFileFactory outputFileFactory = createFileFactory();
+
+ List<T> rowSet = testRowSet();
+ DataFile dataFile = prepareDataFile(rowSet, appenderFactory, outputFileFactory);
+
+ table.newRowDelta()
+ .addRows(dataFile)
+ .commit();
+
+ List<T> deletes = Lists.newArrayList(
+ createRow(1, "aaa"),
+ createRow(3, "bbb"),
Review comment:
If the data here is `bbb` instead of `ccc` on purpose, then could you add a comment that this is testing that just `id` is used for comparison?
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533129659
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,42 +378,40 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter(ParquetValueWriters.PathPosAccessor<?, ?> accessor)
Review comment:
I like the idea about registering a light-weight function to convert the `CharSequence` to `StringData`.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r531474478
##########
File path: flink/src/main/java/org/apache/iceberg/flink/sink/FlinkAppenderFactory.java
##########
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.flink.sink;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
+import org.apache.iceberg.flink.data.FlinkParquetWriters;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.DeleteSchemaUtil;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class FlinkAppenderFactory implements FileAppenderFactory<RowData>, Serializable {
+ private final Schema schema;
+ private final RowType flinkSchema;
+ private final Map<String, String> props;
+ private final PartitionSpec spec;
+ private final int[] equalityFieldIds;
+ private final Schema eqDeleteRowSchema;
+ private final Schema posDeleteRowSchema;
+
+ private RowType eqDeleteFlinkSchema = null;
+ private RowType posDeleteFlinkSchema = null;
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props, PartitionSpec spec) {
+ this(schema, flinkSchema, props, spec, null, schema, null);
+ }
+
+ public FlinkAppenderFactory(Schema schema, RowType flinkSchema, Map<String, String> props,
+ PartitionSpec spec, int[] equalityFieldIds,
+ Schema eqDeleteRowSchema, Schema posDeleteRowSchema) {
+ this.schema = schema;
+ this.flinkSchema = flinkSchema;
+ this.props = props;
+ this.spec = spec;
+ this.equalityFieldIds = equalityFieldIds;
+ this.eqDeleteRowSchema = eqDeleteRowSchema;
+ this.posDeleteRowSchema = posDeleteRowSchema;
+ }
+
+ private RowType lazyEqDeleteFlinkSchema() {
+ if (eqDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(eqDeleteRowSchema, "Equality delete row schema shouldn't be null");
+ this.eqDeleteFlinkSchema = FlinkSchemaUtil.convert(eqDeleteRowSchema);
+ }
+ return eqDeleteFlinkSchema;
+ }
+
+ private RowType lazyPosDeleteFlinkSchema() {
+ if (posDeleteFlinkSchema == null) {
+ Preconditions.checkNotNull(posDeleteRowSchema, "Pos-delete row schema shouldn't be null");
+ this.posDeleteFlinkSchema = FlinkSchemaUtil.convert(posDeleteRowSchema);
+ }
+ return this.posDeleteFlinkSchema;
+ }
+
+ @Override
+ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat format) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .createWriterFunc(ignore -> new FlinkAvroWriter(flinkSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+ .setAll(props)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(flinkSchema, msgType))
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .schema(schema)
+ .overwrite()
+ .build();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public DataWriter<RowData> newDataWriter(EncryptedOutputFile file, FileFormat format, StructLike partition) {
+ return new DataWriter<>(
+ newAppender(file.encryptingOutputFile(), format), format,
+ file.encryptingOutputFile().location(), spec, partition, file.keyMetadata());
+ }
+
+ @Override
+ public EqualityDeleteWriter<RowData> newEqDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyEqDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ case PARQUET:
+ return Parquet.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(msgType -> FlinkParquetWriters.buildWriter(lazyEqDeleteFlinkSchema(), msgType))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .metricsConfig(metricsConfig)
+ .rowSchema(eqDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .equalityFieldIds(equalityFieldIds)
+ .buildEqualityWriter();
+
+ default:
+ throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ @Override
+ public PositionDeleteWriter<RowData> newPosDeleteWriter(EncryptedOutputFile outputFile, FileFormat format,
+ StructLike partition) {
+ MetricsConfig metricsConfig = MetricsConfig.fromProperties(props);
+ try {
+ switch (format) {
+ case AVRO:
+ return Avro.writeDeletes(outputFile.encryptingOutputFile())
+ .createWriterFunc(ignore -> new FlinkAvroWriter(lazyPosDeleteFlinkSchema()))
+ .withPartition(partition)
+ .overwrite()
+ .setAll(props)
+ .rowSchema(posDeleteRowSchema)
+ .withSpec(spec)
+ .withKeyMetadata(outputFile.keyMetadata())
+ .buildPositionWriter();
+
+ case PARQUET:
+ RowType flinkPosDeleteSchema = FlinkSchemaUtil.convert(DeleteSchemaUtil.posDeleteSchema(posDeleteRowSchema));
Review comment:
@rdblue Here I use the parquet writer which is built from <path, pos, row> schema, rather than the writer built from `row` schema.
I tried to hide all the <path, pos, row> write logic inside the `Parquet#buildPositionWriter`, but failed to make it work. I passed a function to this builder by creating a parquet writer built by `rowSchema`, and construct a PositionDeleteStructWriter with a path string writer, a pos long writer and the row struct writer, it does not work because the `currentPath` for [here](https://github.com/apache/iceberg/blob/f950a3e63c98e88b1f8905faacf33a1d5d31e4c0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java#L126) has been messed up. Still figure out how to make this work.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r532948716
##########
File path: data/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
##########
@@ -73,13 +74,12 @@ public DataFile writeFile(StructLike partition, List<Record> records) throws IOE
Preconditions.checkNotNull(table, "table not set");
File file = tmp.newFile();
Assert.assertTrue(file.delete());
- return appendToLocalFile(table, file, fileFormat, partition, records);
+ return appendToLocalFile(table, file, fileFormat, partition, records, table.spec());
}
- private static DataFile appendToLocalFile(
- Table table, File file, FileFormat format, StructLike partition, List<Record> records)
- throws IOException {
- FileAppender<Record> appender = new GenericAppenderFactory(table.schema()).newAppender(
+ private static DataFile appendToLocalFile(Table table, File file, FileFormat format, StructLike partition,
+ List<Record> records, PartitionSpec spec) throws IOException {
+ FileAppender<Record> appender = new GenericAppenderFactory(table.schema(), spec).newAppender(
Review comment:
I think I've counted at least 10 files that would not need to change if we kept the original constructor.
----------------------------------------------------------------
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.
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
[GitHub] [iceberg] openinx commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533841622
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,37 +384,31 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter()
+ throws IOException {
Review comment:
I can revert 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.
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
[GitHub] [iceberg] rdblue commented on a change in pull request #1836: Core: Add data and delete writers in FileAppenderFactory.
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1836:
URL: https://github.com/apache/iceberg/pull/1836#discussion_r533737655
##########
File path: parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
##########
@@ -379,37 +384,31 @@ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
}
-
- public <T> PositionDeleteWriter<T> buildPositionWriter() throws IOException {
+ public <T> PositionDeleteWriter<T> buildPositionWriter()
+ throws IOException {
Review comment:
Does this line need to change? I'm fine removing the empty line, but I think `throws` can still fit on the previous line.
----------------------------------------------------------------
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.
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