You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2020/08/13 19:27:38 UTC
[iceberg] branch master updated: Add Avro and Parquet delete file
writers (#1327)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 50f4ecc Add Avro and Parquet delete file writers (#1327)
50f4ecc is described below
commit 50f4ecca7711e69f63589fea828d26230fac8d59
Author: Ryan Blue <bl...@apache.org>
AuthorDate: Thu Aug 13 12:27:26 2020 -0700
Add Avro and Parquet delete file writers (#1327)
---
.../main/java/org/apache/iceberg/FileMetadata.java | 2 +-
.../java/org/apache/iceberg/MetadataColumns.java | 9 +
.../main/java/org/apache/iceberg/avro/Avro.java | 220 ++++++++++++++++++++-
.../java/org/apache/iceberg/deletes/Deletes.java | 11 +-
.../iceberg/deletes/EqualityDeleteWriter.java | 84 ++++++++
.../org/apache/iceberg/deletes/PositionDelete.java | 89 +++++++++
.../iceberg/deletes/PositionDeleteWriter.java | 83 ++++++++
.../apache/iceberg/avro/TestAvroDeleteWriters.java | 205 +++++++++++++++++++
.../iceberg/data/parquet/BaseParquetReaders.java | 0
.../iceberg/data/parquet/BaseParquetWriter.java | 1 +
.../data/parquet/GenericParquetReaders.java | 0
.../iceberg/data/parquet/GenericParquetWriter.java | 0
.../java/org/apache/iceberg/parquet/Parquet.java | 176 ++++++++++++++++-
.../iceberg/parquet/ParquetValueWriters.java | 21 ++
14 files changed, 888 insertions(+), 13 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/FileMetadata.java b/core/src/main/java/org/apache/iceberg/FileMetadata.java
index 688cd8f..72d3ad5 100644
--- a/core/src/main/java/org/apache/iceberg/FileMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/FileMetadata.java
@@ -30,7 +30,7 @@ import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.util.ByteBuffers;
-class FileMetadata {
+public class FileMetadata {
private FileMetadata() {
}
diff --git a/core/src/main/java/org/apache/iceberg/MetadataColumns.java b/core/src/main/java/org/apache/iceberg/MetadataColumns.java
index 2927fc4..2ae9e55 100644
--- a/core/src/main/java/org/apache/iceberg/MetadataColumns.java
+++ b/core/src/main/java/org/apache/iceberg/MetadataColumns.java
@@ -31,11 +31,20 @@ public class MetadataColumns {
private MetadataColumns() {
}
+ // IDs Integer.MAX_VALUE - (1-100) are used for metadata columns
public static final NestedField FILE_PATH = NestedField.required(
Integer.MAX_VALUE - 1, "_file", Types.StringType.get(), "Path of the file in which a row is stored");
public static final NestedField ROW_POSITION = NestedField.required(
Integer.MAX_VALUE - 2, "_pos", Types.LongType.get(), "Ordinal position of a row in the source data file");
+ // IDs Integer.MAX_VALUE - (101-200) are used for reserved columns
+ public static final NestedField DELETE_FILE_PATH = NestedField.required(
+ Integer.MAX_VALUE - 101, "file_path", Types.StringType.get(), "Path of a file in which a deleted row is stored");
+ public static final NestedField DELETE_FILE_POS = NestedField.required(
+ Integer.MAX_VALUE - 102, "pos", Types.LongType.get(), "Ordinal position of a deleted row in the data file");
+ public static final int DELETE_FILE_ROW_FIELD_ID = Integer.MAX_VALUE - 103;
+ public static final String DELETE_FILE_ROW_DOC = "Deleted row values";
+
private static final Map<String, NestedField> META_COLUMNS = ImmutableMap.of(
FILE_PATH.name(), FILE_PATH,
ROW_POSITION.name(), ROW_POSITION);
diff --git a/core/src/main/java/org/apache/iceberg/avro/Avro.java b/core/src/main/java/org/apache/iceberg/avro/Avro.java
index 24716e3..7c6c7a1 100644
--- a/core/src/main/java/org/apache/iceberg/avro/Avro.java
+++ b/core/src/main/java/org/apache/iceberg/avro/Avro.java
@@ -20,10 +20,14 @@
package org.apache.iceberg.avro;
import java.io.IOException;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.function.BiFunction;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.avro.Conversions;
import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
@@ -31,15 +35,26 @@ import org.apache.avro.file.CodecFactory;
import org.apache.avro.generic.GenericData;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.Encoder;
import org.apache.avro.specific.SpecificData;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDelete;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.util.ArrayUtil;
import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION;
import static org.apache.iceberg.TableProperties.AVRO_COMPRESSION_DEFAULT;
@@ -82,11 +97,11 @@ public class Avro {
public static class WriteBuilder {
private final OutputFile file;
+ private final Map<String, String> config = Maps.newHashMap();
+ private final Map<String, String> metadata = Maps.newLinkedHashMap();
private org.apache.iceberg.Schema schema = null;
private String name = "table";
- private Map<String, String> config = Maps.newHashMap();
- private Map<String, String> metadata = Maps.newLinkedHashMap();
- private Function<Schema, DatumWriter<?>> createWriterFunc = GenericAvroWriter::new;
+ private Function<Schema, DatumWriter<?>> createWriterFunc = null;
private boolean overwrite;
private WriteBuilder(OutputFile file) {
@@ -156,11 +171,208 @@ public class Avro {
Preconditions.checkNotNull(schema, "Schema is required");
Preconditions.checkNotNull(name, "Table name is required and cannot be null");
+ Function<Schema, DatumWriter<?>> writerFunc;
+ if (createWriterFunc != null) {
+ writerFunc = createWriterFunc;
+ } else {
+ writerFunc = GenericAvroWriter::new;
+ }
+
// add the Iceberg schema to keyValueMetadata
meta("iceberg.schema", SchemaParser.toJson(schema));
return new AvroFileAppender<>(
- AvroSchemaUtil.convert(schema, name), file, createWriterFunc, codec(), metadata, overwrite);
+ AvroSchemaUtil.convert(schema, name), file, writerFunc, codec(), metadata, overwrite);
+ }
+ }
+
+ public static DeleteWriteBuilder writeDeletes(OutputFile file) {
+ return new DeleteWriteBuilder(file);
+ }
+
+ public static class DeleteWriteBuilder {
+ private final WriteBuilder appenderBuilder;
+ private final String location;
+ private Function<Schema, DatumWriter<?>> createWriterFunc = null;
+ private org.apache.iceberg.Schema rowSchema;
+ private PartitionSpec spec;
+ private StructLike partition;
+ private EncryptionKeyMetadata keyMetadata = null;
+ private int[] equalityFieldIds = null;
+
+ private DeleteWriteBuilder(OutputFile file) {
+ this.appenderBuilder = write(file);
+ this.location = file.location();
+ }
+
+ public DeleteWriteBuilder forTable(Table table) {
+ rowSchema(table.schema());
+ withSpec(table.spec());
+ setAll(table.properties());
+ return this;
+ }
+
+ public DeleteWriteBuilder set(String property, String value) {
+ appenderBuilder.set(property, value);
+ return this;
+ }
+
+ public DeleteWriteBuilder setAll(Map<String, String> properties) {
+ appenderBuilder.setAll(properties);
+ return this;
+ }
+
+ public DeleteWriteBuilder meta(String property, String value) {
+ appenderBuilder.meta(property, value);
+ return this;
+ }
+
+ public DeleteWriteBuilder meta(Map<String, String> properties) {
+ appenderBuilder.meta(properties);
+ return this;
+ }
+
+ public DeleteWriteBuilder overwrite() {
+ return overwrite(true);
+ }
+
+ public DeleteWriteBuilder overwrite(boolean enabled) {
+ appenderBuilder.overwrite(enabled);
+ return this;
+ }
+
+ public DeleteWriteBuilder createWriterFunc(Function<Schema, DatumWriter<?>> writerFunction) {
+ this.createWriterFunc = writerFunction;
+ return this;
+ }
+
+ public DeleteWriteBuilder rowSchema(org.apache.iceberg.Schema newRowSchema) {
+ this.rowSchema = newRowSchema;
+ return this;
+ }
+
+ public DeleteWriteBuilder withSpec(PartitionSpec newSpec) {
+ this.spec = newSpec;
+ return this;
+ }
+
+ public DeleteWriteBuilder withPartition(StructLike key) {
+ this.partition = key;
+ return this;
+ }
+
+ public DeleteWriteBuilder withKeyMetadata(EncryptionKeyMetadata metadata) {
+ this.keyMetadata = metadata;
+ return this;
+ }
+
+ public DeleteWriteBuilder equalityFieldIds(List<Integer> fieldIds) {
+ this.equalityFieldIds = ArrayUtil.toIntArray(fieldIds);
+ return this;
+ }
+
+ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
+ this.equalityFieldIds = fieldIds;
+ return this;
+ }
+
+ public <T> EqualityDeleteWriter<T> buildEqualityWriter() throws IOException {
+ Preconditions.checkState(rowSchema != null, "Cannot create equality delete file without a schema`");
+ Preconditions.checkState(equalityFieldIds != null, "Cannot create equality delete file without delete field ids");
+ Preconditions.checkState(createWriterFunc != null,
+ "Cannot create equality delete file unless createWriterFunc is set");
+
+ meta("delete-type", "equality");
+ meta("delete-field-ids", IntStream.of(equalityFieldIds)
+ .mapToObj(Objects::toString)
+ .collect(Collectors.joining(", ")));
+
+ // the appender uses the row schema without extra columns
+ appenderBuilder.schema(rowSchema);
+ appenderBuilder.createWriterFunc(createWriterFunc);
+
+ return new EqualityDeleteWriter<>(
+ appenderBuilder.build(), FileFormat.AVRO, location, spec, partition, keyMetadata, equalityFieldIds);
+ }
+
+ 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) {
+ Preconditions.checkState(createWriterFunc != null,
+ "Cannot create delete file with deletes rows unless createWriterFunc is set");
+
+ // 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.createWriterFunc(
+ avroSchema -> new PositionAndRowDatumWriter<>(createWriterFunc.apply(avroSchema)));
+
+ } else {
+ appenderBuilder.schema(new org.apache.iceberg.Schema(
+ MetadataColumns.DELETE_FILE_PATH,
+ MetadataColumns.DELETE_FILE_POS));
+
+ appenderBuilder.createWriterFunc(ignored -> new PositionDatumWriter());
+ }
+
+ return new PositionDeleteWriter<>(
+ appenderBuilder.build(), FileFormat.AVRO, location, spec, partition, keyMetadata);
+ }
+ }
+
+ /**
+ * A {@link DatumWriter} implementation that wraps another to produce position deletes.
+ */
+ private static class PositionDatumWriter implements DatumWriter<PositionDelete<?>> {
+ private static final ValueWriter<Object> PATH_WRITER = ValueWriters.strings();
+ private static final ValueWriter<Long> POS_WRITER = ValueWriters.longs();
+
+ @Override
+ public void setSchema(Schema schema) {
+ }
+
+ @Override
+ public void write(PositionDelete<?> delete, Encoder out) throws IOException {
+ PATH_WRITER.write(delete.path(), out);
+ POS_WRITER.write(delete.pos(), out);
+ }
+ }
+
+ /**
+ * A {@link DatumWriter} implementation that wraps another to produce position deletes with row data.
+ *
+ * @param <D> the type of datum written as a deleted row
+ */
+ private static class PositionAndRowDatumWriter<D> implements DatumWriter<PositionDelete<D>> {
+ private static final ValueWriter<Object> PATH_WRITER = ValueWriters.strings();
+ private static final ValueWriter<Long> POS_WRITER = ValueWriters.longs();
+ private final DatumWriter<D> rowWriter;
+
+ private PositionAndRowDatumWriter(DatumWriter<D> rowWriter) {
+ this.rowWriter = rowWriter;
+ }
+
+ @Override
+ public void setSchema(Schema schema) {
+ Schema.Field rowField = schema.getField("row");
+ if (rowField != null) {
+ rowWriter.setSchema(rowField.schema());
+ }
+ }
+
+ @Override
+ public void write(PositionDelete<D> delete, Encoder out) throws IOException {
+ PATH_WRITER.write(delete.path(), out);
+ POS_WRITER.write(delete.pos(), out);
+ rowWriter.write(delete.row(), out);
}
}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/Deletes.java b/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
index 2a18198..8ee2879 100644
--- a/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
+++ b/core/src/main/java/org/apache/iceberg/deletes/Deletes.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Set;
import java.util.function.Function;
import org.apache.iceberg.Accessor;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.io.CloseableGroup;
@@ -44,12 +45,14 @@ import org.apache.iceberg.util.StructLikeSet;
public class Deletes {
private static final Schema POSITION_DELETE_SCHEMA = new Schema(
- Types.NestedField.required(1, "file_path", Types.StringType.get(), "Data file location of the deleted row"),
- Types.NestedField.required(2, "pos", Types.LongType.get(), "Row position in the data file of the deleted row")
+ MetadataColumns.DELETE_FILE_PATH,
+ MetadataColumns.DELETE_FILE_POS
);
- private static final Accessor<StructLike> FILENAME_ACCESSOR = POSITION_DELETE_SCHEMA.accessorForField(1);
- private static final Accessor<StructLike> POSITION_ACCESSOR = POSITION_DELETE_SCHEMA.accessorForField(2);
+ private static final Accessor<StructLike> FILENAME_ACCESSOR = POSITION_DELETE_SCHEMA
+ .accessorForField(MetadataColumns.DELETE_FILE_PATH.fieldId());
+ private static final Accessor<StructLike> POSITION_ACCESSOR = POSITION_DELETE_SCHEMA
+ .accessorForField(MetadataColumns.DELETE_FILE_POS.fieldId());
private Deletes() {
}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java b/core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java
new file mode 100644
index 0000000..1838ef7
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/deletes/EqualityDeleteWriter.java
@@ -0,0 +1,84 @@
+/*
+ * 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.deletes;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class EqualityDeleteWriter<T> implements Closeable {
+ private final FileAppender<T> appender;
+ private final FileFormat format;
+ private final String location;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+ private final ByteBuffer keyMetadata;
+ private final int[] equalityFieldIds;
+ private DeleteFile deleteFile = null;
+
+ public EqualityDeleteWriter(FileAppender<T> appender, FileFormat format, String location,
+ PartitionSpec spec, StructLike partition, EncryptionKeyMetadata keyMetadata,
+ int... equalityFieldIds) {
+ this.appender = appender;
+ this.format = format;
+ this.location = location;
+ this.spec = spec;
+ this.partition = partition;
+ this.keyMetadata = keyMetadata != null ? keyMetadata.buffer() : null;
+ this.equalityFieldIds = equalityFieldIds;
+ }
+
+ public void deleteAll(Iterable<T> rows) {
+ appender.addAll(rows);
+ }
+
+ public void delete(T row) {
+ appender.add(row);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (deleteFile == null) {
+ appender.close();
+ this.deleteFile = FileMetadata.deleteFileBuilder(spec)
+ .ofEqualityDeletes(equalityFieldIds)
+ .withFormat(format)
+ .withPath(location)
+ .withPartition(partition)
+ .withEncryptionKeyMetadata(keyMetadata)
+ .withFileSizeInBytes(appender.length())
+ .withMetrics(appender.metrics())
+ .build();
+ }
+ }
+
+ public DeleteFile toDeleteFile() {
+ Preconditions.checkState(deleteFile != null, "Cannot create delete file from unclosed writer");
+ return deleteFile;
+ }
+}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java b/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java
new file mode 100644
index 0000000..1a2e849
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java
@@ -0,0 +1,89 @@
+/*
+ * 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.deletes;
+
+import org.apache.iceberg.StructLike;
+
+@SuppressWarnings("checkstyle:OverloadMethodsDeclarationOrder")
+public class PositionDelete<R> implements StructLike {
+ static <T> PositionDelete<T> create() {
+ return new PositionDelete<>();
+ }
+
+ private CharSequence path;
+ private long pos;
+ private R row;
+
+ public PositionDelete<R> set(CharSequence newPath, long newPos, R newRow) {
+ this.path = newPath;
+ this.pos = newPos;
+ this.row = newRow;
+ return this;
+ }
+
+ @Override
+ public int size() {
+ return 3;
+ }
+
+ public CharSequence path() {
+ return path;
+ }
+
+ public long pos() {
+ return pos;
+ }
+
+ public R row() {
+ return row;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T> T get(int colPos, Class<T> javaClass) {
+ switch (colPos) {
+ case 0:
+ return (T) path;
+ case 1:
+ return (T) (Long) pos;
+ case 2:
+ return (T) row;
+ default:
+ throw new IllegalArgumentException("No column at position " + colPos);
+ }
+ }
+
+ @Override
+ public <T> void set(int colPos, T value) {
+ switch (colPos) {
+ case 0:
+ this.path = (CharSequence) value;
+ break;
+ case 1:
+ this.pos = (Long) value;
+ break;
+ case 2:
+ this.row = (R) value;
+ break;
+ default:
+ throw new IllegalArgumentException("No column at position " + colPos);
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
new file mode 100644
index 0000000..b03a3e6
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/deletes/PositionDeleteWriter.java
@@ -0,0 +1,83 @@
+/*
+ * 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.deletes;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileMetadata;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class PositionDeleteWriter<T> implements Closeable {
+ private final FileAppender<StructLike> appender;
+ private final FileFormat format;
+ private final String location;
+ private final PartitionSpec spec;
+ private final StructLike partition;
+ private final ByteBuffer keyMetadata;
+ private final PositionDelete<T> delete;
+ private DeleteFile deleteFile = null;
+
+ public PositionDeleteWriter(FileAppender<StructLike> appender, FileFormat format, String location,
+ PartitionSpec spec, StructLike partition, EncryptionKeyMetadata keyMetadata) {
+ this.appender = appender;
+ this.format = format;
+ this.location = location;
+ this.spec = spec;
+ this.partition = partition;
+ this.keyMetadata = keyMetadata != null ? keyMetadata.buffer() : null;
+ this.delete = PositionDelete.create();
+ }
+
+ public void delete(CharSequence path, long pos) {
+ delete(path, pos, null);
+ }
+
+ public void delete(CharSequence path, long pos, T row) {
+ appender.add(delete.set(path, pos, row));
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (deleteFile == null) {
+ appender.close();
+ this.deleteFile = FileMetadata.deleteFileBuilder(spec)
+ .ofPositionDeletes()
+ .withFormat(format)
+ .withPath(location)
+ .withPartition(partition)
+ .withEncryptionKeyMetadata(keyMetadata)
+ .withFileSizeInBytes(appender.length())
+ .withMetrics(appender.metrics())
+ .build();
+ }
+ }
+
+ public DeleteFile toDeleteFile() {
+ Preconditions.checkState(deleteFile != null, "Cannot create delete file from unclosed writer");
+ return deleteFile;
+ }
+}
diff --git a/core/src/test/java/org/apache/iceberg/avro/TestAvroDeleteWriters.java b/core/src/test/java/org/apache/iceberg/avro/TestAvroDeleteWriters.java
new file mode 100644
index 0000000..dc61139
--- /dev/null
+++ b/core/src/test/java/org/apache/iceberg/avro/TestAvroDeleteWriters.java
@@ -0,0 +1,205 @@
+/*
+ * 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.avro;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.avro.DataReader;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestAvroDeleteWriters {
+ private static final Schema SCHEMA = new Schema(
+ NestedField.required(1, "id", Types.LongType.get()),
+ NestedField.optional(2, "data", Types.StringType.get()));
+
+ private List<Record> records;
+
+ @Rule
+ public TemporaryFolder temp = new TemporaryFolder();
+
+ @Before
+ public void createDeleteRecords() {
+ GenericRecord record = GenericRecord.create(SCHEMA);
+
+ ImmutableList.Builder<Record> builder = ImmutableList.builder();
+ builder.add(record.copy(ImmutableMap.of("id", 1L, "data", "a")));
+ builder.add(record.copy(ImmutableMap.of("id", 2L, "data", "b")));
+ builder.add(record.copy(ImmutableMap.of("id", 3L, "data", "c")));
+ builder.add(record.copy(ImmutableMap.of("id", 4L, "data", "d")));
+ builder.add(record.copy(ImmutableMap.of("id", 5L, "data", "e")));
+
+ this.records = builder.build();
+ }
+
+ @Test
+ public void testEqualityDeleteWriter() throws IOException {
+ File deleteFile = temp.newFile();
+
+ OutputFile out = Files.localOutput(deleteFile);
+ EqualityDeleteWriter<Record> deleteWriter = Avro.writeDeletes(out)
+ .createWriterFunc(DataWriter::create)
+ .overwrite()
+ .rowSchema(SCHEMA)
+ .withSpec(PartitionSpec.unpartitioned())
+ .equalityFieldIds(1)
+ .buildEqualityWriter();
+
+ try (EqualityDeleteWriter<Record> writer = deleteWriter) {
+ writer.deleteAll(records);
+ }
+
+ DeleteFile metadata = deleteWriter.toDeleteFile();
+ Assert.assertEquals("Format should be Avro", FileFormat.AVRO, metadata.format());
+ Assert.assertEquals("Should be equality deletes", FileContent.EQUALITY_DELETES, metadata.content());
+ Assert.assertEquals("Record count should be correct", records.size(), metadata.recordCount());
+ Assert.assertEquals("Partition should be empty", 0, metadata.partition().size());
+ Assert.assertNull("Key metadata should be null", metadata.keyMetadata());
+
+ List<Record> deletedRecords;
+ try (AvroIterable<Record> reader = Avro.read(out.toInputFile())
+ .project(SCHEMA)
+ .createReaderFunc(DataReader::create)
+ .build()) {
+ deletedRecords = Lists.newArrayList(reader);
+ }
+
+ Assert.assertEquals("Deleted records should match expected", records, deletedRecords);
+ }
+
+ @Test
+ public void testPositionDeleteWriter() throws IOException {
+ File deleteFile = temp.newFile();
+
+ Schema deleteSchema = new Schema(
+ MetadataColumns.DELETE_FILE_PATH,
+ MetadataColumns.DELETE_FILE_POS,
+ NestedField.optional(MetadataColumns.DELETE_FILE_ROW_FIELD_ID, "row", SCHEMA.asStruct()));
+
+ String deletePath = "s3://bucket/path/file.parquet";
+ GenericRecord posDelete = GenericRecord.create(deleteSchema);
+ List<Record> expectedDeleteRecords = Lists.newArrayList();
+
+ OutputFile out = Files.localOutput(deleteFile);
+ PositionDeleteWriter<Record> deleteWriter = Avro.writeDeletes(out)
+ .createWriterFunc(DataWriter::create)
+ .overwrite()
+ .rowSchema(SCHEMA)
+ .withSpec(PartitionSpec.unpartitioned())
+ .buildPositionWriter();
+
+ try (PositionDeleteWriter<Record> writer = deleteWriter) {
+ for (int i = 0; i < records.size(); i += 1) {
+ int pos = i * 3 + 2;
+ writer.delete(deletePath, pos, records.get(i));
+ expectedDeleteRecords.add(posDelete.copy(ImmutableMap.of(
+ "file_path", deletePath,
+ "pos", (long) pos,
+ "row", records.get(i))));
+ }
+ }
+
+ DeleteFile metadata = deleteWriter.toDeleteFile();
+ Assert.assertEquals("Format should be Avro", FileFormat.AVRO, metadata.format());
+ Assert.assertEquals("Should be position deletes", FileContent.POSITION_DELETES, metadata.content());
+ Assert.assertEquals("Record count should be correct", records.size(), metadata.recordCount());
+ Assert.assertEquals("Partition should be empty", 0, metadata.partition().size());
+ Assert.assertNull("Key metadata should be null", metadata.keyMetadata());
+
+ List<Record> deletedRecords;
+ try (AvroIterable<Record> reader = Avro.read(out.toInputFile())
+ .project(deleteSchema)
+ .createReaderFunc(DataReader::create)
+ .build()) {
+ deletedRecords = Lists.newArrayList(reader);
+ }
+
+ Assert.assertEquals("Deleted records should match expected", expectedDeleteRecords, deletedRecords);
+ }
+
+ @Test
+ public void testPositionDeleteWriterWithEmptyRow() throws IOException {
+ File deleteFile = temp.newFile();
+
+ Schema deleteSchema = new Schema(
+ MetadataColumns.DELETE_FILE_PATH,
+ MetadataColumns.DELETE_FILE_POS);
+
+ String deletePath = "s3://bucket/path/file.parquet";
+ GenericRecord posDelete = GenericRecord.create(deleteSchema);
+ List<Record> expectedDeleteRecords = Lists.newArrayList();
+
+ OutputFile out = Files.localOutput(deleteFile);
+ PositionDeleteWriter<Void> deleteWriter = Avro.writeDeletes(out)
+ .createWriterFunc(DataWriter::create)
+ .overwrite()
+ .withSpec(PartitionSpec.unpartitioned())
+ .buildPositionWriter();
+
+ try (PositionDeleteWriter<Void> writer = deleteWriter) {
+ for (int i = 0; i < records.size(); i += 1) {
+ int pos = i * 3 + 2;
+ writer.delete(deletePath, pos, null);
+ expectedDeleteRecords.add(posDelete.copy(ImmutableMap.of(
+ "file_path", deletePath,
+ "pos", (long) pos)));
+ }
+ }
+
+ DeleteFile metadata = deleteWriter.toDeleteFile();
+ Assert.assertEquals("Format should be Avro", FileFormat.AVRO, metadata.format());
+ Assert.assertEquals("Should be position deletes", FileContent.POSITION_DELETES, metadata.content());
+ Assert.assertEquals("Record count should be correct", records.size(), metadata.recordCount());
+ Assert.assertEquals("Partition should be empty", 0, metadata.partition().size());
+ Assert.assertNull("Key metadata should be null", metadata.keyMetadata());
+
+ List<Record> deletedRecords;
+ try (AvroIterable<Record> reader = Avro.read(out.toInputFile())
+ .project(deleteSchema)
+ .createReaderFunc(DataReader::create)
+ .build()) {
+ deletedRecords = Lists.newArrayList(reader);
+ }
+
+ Assert.assertEquals("Deleted records should match expected", expectedDeleteRecords, deletedRecords);
+ }
+}
diff --git a/data/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java b/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java
similarity index 100%
rename from data/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java
rename to parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetReaders.java
diff --git a/data/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java b/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java
similarity index 99%
rename from data/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java
rename to parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java
index 282a92d..91fd952 100644
--- a/data/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java
+++ b/parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java
@@ -59,6 +59,7 @@ public abstract class BaseParquetWriter<T> {
@Override
public ParquetValueWriter<?> message(MessageType message, List<ParquetValueWriter<?>> fieldWriters) {
+
return struct(message.asGroupType(), fieldWriters);
}
diff --git a/data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java b/parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java
similarity index 100%
rename from data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java
rename to parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java
diff --git a/data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java b/parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java
similarity index 100%
rename from data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java
rename to parquet/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
index fa6c80d..6a04542 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
@@ -21,15 +21,27 @@ package org.apache.iceberg.parquet;
import java.io.IOException;
import java.util.Collection;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetadataColumns;
import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.deletes.PositionDeleteWriter;
+import org.apache.iceberg.encryption.EncryptionKeyMetadata;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.hadoop.HadoopInputFile;
@@ -39,9 +51,13 @@ import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.parquet.ParquetValueWriters.PositionDeleteStructWriter;
+import org.apache.iceberg.parquet.ParquetValueWriters.StructWriter;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types.NestedField;
+import org.apache.iceberg.util.ArrayUtil;
import org.apache.parquet.HadoopReadOptions;
import org.apache.parquet.ParquetReadOptions;
import org.apache.parquet.avro.AvroReadSupport;
@@ -81,11 +97,11 @@ public class Parquet {
public static class WriteBuilder {
private final OutputFile file;
+ private final Map<String, String> metadata = Maps.newLinkedHashMap();
+ private final Map<String, String> config = Maps.newLinkedHashMap();
private Schema schema = null;
private String name = "table";
private WriteSupport<?> writeSupport = null;
- private Map<String, String> metadata = Maps.newLinkedHashMap();
- private Map<String, String> config = Maps.newLinkedHashMap();
private Function<MessageType, ParquetValueWriter<?>> createWriterFunc = null;
private MetricsConfig metricsConfig = MetricsConfig.getDefault();
private ParquetFileWriter.Mode writeMode = ParquetFileWriter.Mode.CREATE;
@@ -250,6 +266,158 @@ public class Parquet {
}
}
+ public static DeleteWriteBuilder writeDeletes(OutputFile file) {
+ return new DeleteWriteBuilder(file);
+ }
+
+ public static class DeleteWriteBuilder extends WriteBuilder {
+ private final WriteBuilder appenderBuilder;
+ private final String location;
+ private Function<MessageType, ParquetValueWriter<?>> createWriterFunc = null;
+ private Schema rowSchema = null;
+ private PartitionSpec spec = null;
+ private StructLike partition = null;
+ private EncryptionKeyMetadata keyMetadata = null;
+ private int[] equalityFieldIds = null;
+
+ private DeleteWriteBuilder(OutputFile file) {
+ super(file);
+ this.appenderBuilder = write(file);
+ this.location = file.location();
+ }
+
+ public DeleteWriteBuilder forTable(Table table) {
+ schema(table.schema());
+ setAll(table.properties());
+ metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ return this;
+ }
+
+ public DeleteWriteBuilder set(String property, String value) {
+ appenderBuilder.set(property, value);
+ return this;
+ }
+
+ public DeleteWriteBuilder setAll(Map<String, String> properties) {
+ appenderBuilder.setAll(properties);
+ return this;
+ }
+
+ public DeleteWriteBuilder meta(String property, String value) {
+ appenderBuilder.meta(property, value);
+ return this;
+ }
+
+ public DeleteWriteBuilder overwrite() {
+ return overwrite(true);
+ }
+
+ public DeleteWriteBuilder overwrite(boolean enabled) {
+ appenderBuilder.overwrite(enabled);
+ return this;
+ }
+
+ public DeleteWriteBuilder metricsConfig(MetricsConfig newMetricsConfig) {
+ // TODO: keep full metrics for position delete file columns
+ appenderBuilder.metricsConfig(newMetricsConfig);
+ return this;
+ }
+
+ public DeleteWriteBuilder createWriterFunc(Function<MessageType, ParquetValueWriter<?>> newCreateWriterFunc) {
+ this.createWriterFunc = newCreateWriterFunc;
+ return this;
+ }
+
+ public DeleteWriteBuilder rowSchema(Schema newSchema) {
+ this.rowSchema = newSchema;
+ return this;
+ }
+
+ public DeleteWriteBuilder withSpec(PartitionSpec newSpec) {
+ this.spec = newSpec;
+ return this;
+ }
+
+ public DeleteWriteBuilder withPartition(StructLike key) {
+ this.partition = key;
+ return this;
+ }
+
+ public DeleteWriteBuilder withKeyMetadata(EncryptionKeyMetadata metadata) {
+ this.keyMetadata = metadata;
+ return this;
+ }
+
+ public DeleteWriteBuilder equalityFieldIds(List<Integer> fieldIds) {
+ this.equalityFieldIds = ArrayUtil.toIntArray(fieldIds);
+ return this;
+ }
+
+ public DeleteWriteBuilder equalityFieldIds(int... fieldIds) {
+ this.equalityFieldIds = fieldIds;
+ return this;
+ }
+
+ public <T> EqualityDeleteWriter<T> buildEqualityWriter() throws IOException {
+ Preconditions.checkState(rowSchema != null, "Cannot create equality delete file without a schema`");
+ Preconditions.checkState(equalityFieldIds != null, "Cannot create equality delete file without delete field ids");
+ Preconditions.checkState(createWriterFunc != null,
+ "Cannot create equality delete file unless createWriterFunc is set");
+
+ meta("delete-type", "equality");
+ meta("delete-field-ids", IntStream.of(equalityFieldIds)
+ .mapToObj(Objects::toString)
+ .collect(Collectors.joining(", ")));
+
+ // the appender uses the row schema without extra columns
+ appenderBuilder.schema(rowSchema);
+ appenderBuilder.createWriterFunc(createWriterFunc);
+
+ return new EqualityDeleteWriter<>(
+ appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata, equalityFieldIds);
+ }
+
+
+ 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) {
+ Preconditions.checkState(createWriterFunc != null,
+ "Cannot create delete file with deletes rows unless createWriterFunc is set");
+
+ // 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.createWriterFunc(parquetSchema -> {
+ ParquetValueWriter<?> writer = createWriterFunc.apply(parquetSchema);
+ if (writer instanceof StructWriter) {
+ return new PositionDeleteStructWriter<T>((StructWriter<?>) writer);
+ } 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.createWriterFunc(parquetSchema ->
+ new PositionDeleteStructWriter<T>((StructWriter<?>) GenericParquetWriter.buildWriter(parquetSchema)));
+ }
+
+ return new PositionDeleteWriter<>(
+ appenderBuilder.build(), FileFormat.PARQUET, location, spec, partition, keyMetadata);
+ }
+ }
+
private static class ParquetWriteBuilder<T> extends ParquetWriter.Builder<T, ParquetWriteBuilder<T>> {
private Map<String, String> keyValueMetadata = Maps.newHashMap();
private Map<String, String> config = Maps.newHashMap();
@@ -300,6 +468,7 @@ public class Parquet {
public static class ReadBuilder {
private final InputFile file;
+ private final Map<String, String> properties = Maps.newHashMap();
private Long start = null;
private Long length = null;
private Schema schema = null;
@@ -309,7 +478,6 @@ public class Parquet {
private Function<MessageType, ParquetValueReader<?>> readerFunc = null;
private boolean filterRecords = true;
private boolean caseSensitive = true;
- private Map<String, String> properties = Maps.newHashMap();
private boolean callInit = false;
private boolean reuseContainers = false;
private int maxRecordsPerBatch = 10000;
@@ -426,7 +594,7 @@ public class Parquet {
ParquetReadOptions options = optionsBuilder.build();
if (batchedReaderFunc != null) {
- return new VectorizedParquetReader(file, schema, options, batchedReaderFunc, nameMapping, filter,
+ return new VectorizedParquetReader<>(file, schema, options, batchedReaderFunc, nameMapping, filter,
reuseContainers, caseSensitive, maxRecordsPerBatch);
} else {
return new org.apache.iceberg.parquet.ParquetReader<>(
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java
index c433267..ca14e9f 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java
@@ -22,11 +22,13 @@ package org.apache.iceberg.parquet;
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.apache.avro.util.Utf8;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.types.TypeUtil;
@@ -488,4 +490,23 @@ public class ParquetValueWriters {
protected abstract Object get(S struct, int index);
}
+
+ public static class PositionDeleteStructWriter<R> extends StructWriter<PositionDelete<R>> {
+ public PositionDeleteStructWriter(StructWriter<?> replacedWriter) {
+ super(Arrays.asList(replacedWriter.writers));
+ }
+
+ @Override
+ protected Object get(PositionDelete<R> delete, int index) {
+ switch (index) {
+ case 0:
+ return delete.path();
+ case 1:
+ return delete.pos();
+ case 2:
+ return delete.row();
+ }
+ throw new IllegalArgumentException("Cannot get value for invalid index: " + index);
+ }
+ }
}